shanxia2.0/backend/app/services/wechat_cert_notices.py

"""证件到期周一、周三、周五汇总：扫描后入队，发送前复核真实证件与收件权限。"""
from __future__ import annotations

from datetime import date, datetime
import json
from zoneinfo import ZoneInfo

from sqlalchemy import and_
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.db_auth import DbUserContext
from app.models.models import SysRole, SysUser, SysUserRole
from app.models.pl_models import PlCert, PlDocumentReminder
from app.models.wechat_notice import WechatNotice
from app.services.pl_data_scope import apply_pl_scope
from app.services.web_auth_roles import resolve_web_role_for_user
from app.services.wechat_notice_events import enqueue_event


SHANGHAI = ZoneInfo("Asia/Shanghai")


def _local_time(now: datetime | None) -> datetime:
    # 无时区输入遵循业务本地时间；生产默认始终取上海时区。
    value = now or datetime.now(SHANGHAI)
    return value.replace(tzinfo=SHANGHAI) if value.tzinfo is None else value.astimezone(SHANGHAI)


def _recipients(db: Session):
    return (
        db.query(SysUser)
        .join(SysUserRole, and_(
            SysUserRole.user_id == SysUser.id,
            SysUserRole.tenant_no == SysUser.tenant_no,
            SysUserRole.del_flag == 0,
        ))
        .join(SysRole, and_(
            SysRole.id == SysUserRole.role_id,
            SysRole.tenant_no == SysUser.tenant_no,
            SysRole.del_flag == 0,
            SysRole.status == "1",
        ))
        .filter(
            SysUser.status == "1", SysUser.del_flag == 0,
            SysRole.role_name == settings.WECHAT_OA_CERT_NOTICE_ROLE_NAME,
        )
        .distinct()
        .populate_existing()
    )


def _is_notice_time(now: datetime) -> bool:
    # 门户预警仍每日扫描，微信只在上海时间周一、周三、周五10点（默认）之后发送。
    return now.weekday() in (0, 2, 4) and (now.hour, now.minute) >= (
        settings.WECHAT_OA_CERT_REMINDER_HOUR, settings.WECHAT_OA_CERT_REMINDER_MINUTE,
    )


def _active_cert_counts(db: Session, user: SysUser, today: date) -> tuple[int, int]:
    query = db.query(PlCert.id, PlCert.expire_date, PlCert.remind_days).join(
        PlDocumentReminder, and_(
            PlDocumentReminder.cert_id == PlCert.id,
            PlDocumentReminder.tenant_no == PlCert.tenant_no,
        ),
    ).filter(
        PlCert.tenant_no == user.tenant_no,
        PlCert.del_flag == 0,
        PlCert.status == "valid",
        PlCert.expire_date.isnot(None),
        PlDocumentReminder.tenant_no == user.tenant_no,
        PlDocumentReminder.del_flag == 0,
        PlDocumentReminder.status == "pending",
    )
    context = DbUserContext(
        id=user.id, name=user.nick_name or user.user_no, dept_id=user.dept_id or "",
        # 已校验本租户接收角色，因此不会走角色解析的默认租户回退。
        role=resolve_web_role_for_user(db, user.id, user.tenant_no), tenant_no=user.tenant_no,
    )
    query = apply_pl_scope(query, PlDocumentReminder, context, db, user.tenant_no)
    # 不使用历史 days_remaining；重复 pending 按证件去重，包含仍未处理的已过期证件。
    upcoming, expired = 0, 0
    for _, expiry, remind_days in query.distinct().all():
        days = (expiry - today).days
        if days > remind_days:
            continue
        # 与提醒列表统计一致：当天到期也纳入已到期数量。
        if days <= 0:
            expired += 1
        else:
            upcoming += 1
    return upcoming, expired


def _payload(now: datetime, upcoming: int, expired: int) -> dict:
    parts = []
    if upcoming:
        parts.append(f"{upcoming}项即将到期")
    if expired:
        parts.append(f"{expired}项已到期")
    title = "，".join(parts)
    if len(title) > 20:
        title = f"{upcoming + expired}项证件到期提醒"
    return {
        "thing5": {"value": title},
        "time6": {"value": now.strftime("%Y-%m-%d %H:%M:%S")},
        "thing3": {"value": "系统默认发出"},
    }


def enqueue_daily_cert_notices(db: Session, *, now: datetime | None = None) -> int:
    """每日扫描后调用，仅通知日入队；事务由调用者提交，按租户/收件人/上海日期幂等。"""
    if not settings.WECHAT_OA_ENABLED or not settings.WECHAT_OA_CERT_REMINDER_ENABLED:
        return 0
    local_now = _local_time(now)
    if not _is_notice_time(local_now):
        return 0
    db.flush()
    created = 0
    for recipient in _recipients(db).all():
        upcoming, expired = _active_cert_counts(db, recipient, local_now.date())
        if not upcoming and not expired:
            continue
        notice = enqueue_event(
            db, tenant_no=recipient.tenant_no, business_type="pl_cert_expiry",
            business_id=recipient.id, event_id=local_now.date().isoformat(),
            kind="cert_daily", recipient_id=recipient.id,
            template_id=settings.WECHAT_OA_TEMPLATE_CERT_REMINDER,
            payload=_payload(local_now, upcoming, expired),
        )
        if notice is not None:
            created += 1
    return created


def refresh_cert_notice(
    db: Session, notice: WechatNotice, *, now: datetime | None = None,
) -> bool:
    """发送前复核当日、收件人、角色及证件状态，time6 使用实际发送时间。"""
    if not settings.WECHAT_OA_ENABLED or not settings.WECHAT_OA_CERT_REMINDER_ENABLED:
        return False
    local_now = _local_time(now)
    if not _is_notice_time(local_now):
        return False
    if (
        notice.kind != "cert_daily" or notice.business_type != "pl_cert_expiry"
        or notice.business_id != notice.recipient_id
        or notice.cycle_id != local_now.date().isoformat()
    ):
        return False
    db.flush()
    recipient = _recipients(db).filter(
        SysUser.id == notice.recipient_id, SysUser.tenant_no == notice.tenant_no,
    ).first()
    if recipient is None:
        return False
    upcoming, expired = _active_cert_counts(db, recipient, local_now.date())
    if not upcoming and not expired:
        return False
    notice.payload = json.dumps(_payload(local_now, upcoming, expired), ensure_ascii=False)
    return True