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

"""真实发放/支付事件通知：业务事务入队，不依赖审批周期或补发历史。"""
from dataclasses import dataclass
from datetime import datetime
import json
from urllib.parse import quote

from sqlalchemy import inspect

from app.core.config import settings
from app.models.bx_models import BxAdvanceApply, BxReimburseApply
from app.models.models import SysUser
from app.models.pl_models import (
    PlFreightSettlement, PlFreightSettlementApproval, PlRemittanceVoucher,
    PlVehicleRechargeApply, PlVehicleRechargeRecord,
)
from app.models.wechat_notice import WechatNotice
from app.services.wechat_notice_events import enqueue_event


@dataclass
class Completion:
    name: str
    number: str
    recipient: str
    occurred_at: datetime
    record_id: str = ''
    page: str = ''


def _document(db, model, biz_id, tenant_no):
    return db.query(model).filter_by(id=biz_id, tenant_no=tenant_no, del_flag=0).first()


def _completion(db, biz_type, biz_id, tenant_no, *, record_id=None):
    if biz_type in ('bx_advance', 'bx_reimburse'):
        advance = biz_type == 'bx_advance'
        row = _document(db, BxAdvanceApply if advance else BxReimburseApply, biz_id, tenant_no)
        if row is None:
            return None
        when = row.issue_time if advance else row.pay_time
        complete = (row.status == 'issued' and row.issue_status == 'issued' if advance else
                    row.status == 'processed' and row.pay_status == 'paid')
        if not complete or not when:
            return None
        domain = 'advance' if advance else 'reimburse'
        return Completion('备用金发放' if advance else '报销支付', row.apply_code or '',
                          row.applicant_id, when,
                          page=f'pages/bx/{domain}/detail?id={quote(str(row.id), safe="")}')
    if biz_type == 'pl_freight':
        row = _document(db, PlFreightSettlement, biz_id, tenant_no)
        if row is None or row.settlement_status != 'confirmed' or not record_id:
            return None
        record = db.query(PlRemittanceVoucher).filter_by(
            id=record_id, settlement_id=biz_id, tenant_no=tenant_no, del_flag=0,
        ).first()
        if record is None or not record.create_time:
            return None
        approval = db.query(PlFreightSettlementApproval).filter(
            PlFreightSettlementApproval.settlement_id == biz_id,
            PlFreightSettlementApproval.tenant_no == tenant_no,
            PlFreightSettlementApproval.del_flag == 0,
            PlFreightSettlementApproval.create_by.isnot(None),
            PlFreightSettlementApproval.create_by != '',
        ).order_by(PlFreightSettlementApproval.step, PlFreightSettlementApproval.create_time,
                   PlFreightSettlementApproval.id).first()
        recipient = approval.create_by if approval else row.create_by
        return Completion('运费发放', row.settlement_no or '', recipient or '', record.create_time, record.id)
    if biz_type == 'pl_recharge':
        row = _document(db, PlVehicleRechargeApply, biz_id, tenant_no)
        if row is None or row.status != 'disbursed':
            return None
        query = db.query(PlVehicleRechargeRecord).filter_by(
            apply_id=biz_id, tenant_no=tenant_no, del_flag=0,
        )
        if record_id:
            query = query.filter(PlVehicleRechargeRecord.id == record_id)
        record = query.first()
        if record is None or not record.create_time:
            return None
        return Completion('充值发放', row.apply_no or '', row.applicant_id, record.create_time, record.id,
                          f'pages/pl/vehicle-recharge-apply/form?id={quote(str(row.id), safe="")}')
    return None


def sync_completion_notification(db, biz_type, biz_id, tenant_no, *, actor_id='',
                                 event_id=None, occurred_at=None):
    """只由真实完成/撤销调用。充值显式UUID标识本次关联，防止同秒换绑复用事件。"""
    # 默认禁用且尚未安装通知表时，不给原业务增加迁移依赖。
    # 已安装的队列仍取消旧事件，防止停用期间撤销/换绑后恢复发送旧通知。
    if not settings.WECHAT_OA_ENABLED and not inspect(db.connection()).has_table(WechatNotice.__tablename__):
        return None
    db.flush()
    snapshot = _completion(db, biz_type, biz_id, tenant_no,
                           record_id=event_id if biz_type == 'pl_freight' else None)
    if biz_type in ('bx_advance', 'bx_reimburse'):
        event_id = 'first-completion'
    # 完成状态取消或新关联产生时，即使总开关关闭，也要取消原待发通知。
    pending = db.query(WechatNotice).filter_by(
        tenant_no=tenant_no, business_type=biz_type, business_id=biz_id,
        kind='completion', del_flag=0,
    ).filter(WechatNotice.status.in_(('pending', 'sending'))).all()
    for notice in pending:
        if snapshot is None or (event_id and notice.cycle_id != event_id):
            notice.status, notice.last_error = 'cancelled', '发放或支付记录已撤销或变更'
            # 独立失效标记不被旧worker的瞬时失败重试覆盖；保留原记录ID供审计。
            notice.result = 'revoked'
    if snapshot is None or not event_id or not settings.WECHAT_OA_ENABLED:
        return None
    # 充值记录后续挂申请时，真实处理时间是本次挂接时间，而不是旧记录创建时间。
    when = occurred_at if biz_type == 'pl_recharge' else snapshot.occurred_at
    if not when:
        return None
    number = snapshot.number.strip()
    payload = {
        'thing3': {'value': snapshot.name},
        'thing4': {'value': number},
        'thing5': {'value': '支付完成' if biz_type == 'bx_reimburse' else '发放完成'},
        'time12': {'value': when.strftime('%Y-%m-%d %H:%M:%S')},
    }
    notice = enqueue_event(
        db, tenant_no=tenant_no, business_type=biz_type, business_id=biz_id,
        event_id=event_id, kind='completion', recipient_id=snapshot.recipient,
        template_id=settings.WECHAT_OA_TEMPLATE_COMPLETION, payload=payload,
        page=snapshot.page, actor_id=actor_id,
        invalid_error='单号为空或超过20字符' if not number or len(number) > 20 else None,
    )
    if notice is not None:
        notice.task_key = snapshot.record_id
        db.flush()
    return notice


def valid_completion_notice(db, notice) -> bool:
    """发送前验证本租户实际完成记录与原申请人；旧审批周期不参与判断。"""
    if notice.kind != 'completion' or notice.status == 'cancelled' or notice.del_flag:
        return False
    if notice.appid != settings.WECHAT_OA_APPID:
        return False
    # 领取发送后可能发生撤销；重读持久化状态，不信任worker内存中的旧sending对象。
    current = db.query(WechatNotice.status, WechatNotice.cycle_id, WechatNotice.task_key, WechatNotice.result).filter_by(
        id=notice.id, tenant_no=notice.tenant_no, del_flag=0,
    ).first()
    if (not current or current.status == 'cancelled' or current.result == 'revoked'
            or current.cycle_id != notice.cycle_id or current.task_key != notice.task_key):
        return False
    if not db.query(SysUser.id).filter_by(
        id=notice.recipient_id, tenant_no=notice.tenant_no, status='1', del_flag=0,
    ).first():
        return False
    snapshot = _completion(db, notice.business_type, notice.business_id,
                           notice.tenant_no, record_id=notice.task_key)
    if snapshot is None or snapshot.recipient != notice.recipient_id:
        return False
    try:
        payload = json.loads(notice.payload)
        if payload['thing4']['value'] != snapshot.number.strip():
            return False
        if notice.business_type != 'pl_recharge':
            return payload['time12']['value'] == snapshot.occurred_at.strftime('%Y-%m-%d %H:%M:%S')
        return bool(notice.task_key)
    except (KeyError, TypeError, ValueError):
        return False