248 lines
12 KiB
Python
248 lines
12 KiB
Python
"""审批事务内入队，后台独立发送；不在网络失败时回滚已完成的审批。"""
|
|
from datetime import datetime, timedelta
|
|
import hashlib
|
|
import json
|
|
|
|
from app.core.config import settings
|
|
from app.core.utils import generate_uuid
|
|
from app.models.models import SysUser
|
|
from app.models.wechat_notice import WechatApprovalCycle, WechatBinding, WechatNotice
|
|
from app.services.wechat_oa_client import client as default_client, WechatAPIError, WechatDeliveryUnknown
|
|
|
|
|
|
def _snapshot(db, biz_type, biz_id, tenant_no):
|
|
if biz_type in ("crm_quote", "crm_contract", "pl_freight"):
|
|
from app.services.wechat_notice_crm_sources import get_snapshot
|
|
elif biz_type in ("bx_advance", "bx_reimburse", "pl_recharge"):
|
|
from app.services.wechat_notice_bx_sources import get_snapshot
|
|
else:
|
|
return None
|
|
return get_snapshot(db, biz_type, biz_id, tenant_no)
|
|
|
|
|
|
def _user(db, tenant_no, user_id):
|
|
return db.query(SysUser).filter(
|
|
SysUser.id == user_id, SysUser.tenant_no == tenant_no,
|
|
SysUser.del_flag == 0, SysUser.status == "1",
|
|
).first()
|
|
|
|
|
|
def _thing(value):
|
|
return " ".join(str(value or "").split())[:20]
|
|
|
|
|
|
def _assignment_key(task):
|
|
return json.dumps([task.task_key, task.user_id], ensure_ascii=False)
|
|
|
|
|
|
def _event_key(tenant_no, appid, biz_type, biz_id, cycle_id, kind, task_key, recipient_id, assignment):
|
|
return hashlib.sha256(json.dumps([
|
|
tenant_no, appid, biz_type, biz_id, cycle_id, kind, task_key, recipient_id, assignment,
|
|
], ensure_ascii=False).encode()).hexdigest()
|
|
|
|
|
|
def _enqueue(db, cycle, snapshot, *, kind, recipient_id, task_key="", assignment="", actor_id=""):
|
|
user = _user(db, cycle.tenant_no, recipient_id)
|
|
applicant = _user(db, cycle.tenant_no, snapshot.applicant_id)
|
|
if not user or not applicant:
|
|
return
|
|
# 不截断业务编号，防止通知指向错误单据；不符合模板长度的记录可查失败原因。
|
|
number = str(snapshot.business_no or "").strip()
|
|
event_key = _event_key(
|
|
cycle.tenant_no, settings.WECHAT_OA_APPID, cycle.business_type, cycle.business_id,
|
|
cycle.cycle_id, kind, task_key, recipient_id, assignment,
|
|
)
|
|
if db.query(WechatNotice.id).filter(WechatNotice.event_key == event_key).first():
|
|
return
|
|
if kind == "pending":
|
|
payload = {
|
|
"thing2": {"value": _thing(snapshot.business_name)},
|
|
"character_string6": {"value": number},
|
|
"thing3": {"value": _thing(applicant.nick_name or applicant.user_no)},
|
|
"time4": {"value": cycle.submitted_at.strftime("%Y-%m-%d %H:%M:%S")},
|
|
}
|
|
template = settings.WECHAT_OA_TEMPLATE_PENDING
|
|
else:
|
|
approver = _user(db, cycle.tenant_no, actor_id or snapshot.approver_id)
|
|
if not approver or not snapshot.occurred_at:
|
|
return
|
|
payload = {
|
|
"thing1": {"value": _thing(snapshot.business_name)},
|
|
"character_string2": {"value": number},
|
|
"thing6": {"value": _thing(approver.nick_name or approver.user_no)},
|
|
"time7": {"value": snapshot.occurred_at.strftime("%Y-%m-%d %H:%M:%S")},
|
|
"const5": {"value": "通过" if snapshot.result == "approved" else "打回"},
|
|
}
|
|
template = settings.WECHAT_OA_TEMPLATE_RESULT
|
|
invalid = not number or len(number) > 32 or not template
|
|
db.add(WechatNotice(
|
|
id=generate_uuid(), tenant_no=cycle.tenant_no, appid=settings.WECHAT_OA_APPID,
|
|
business_type=cycle.business_type, business_id=cycle.business_id, cycle_id=cycle.cycle_id,
|
|
event_key=event_key, kind=kind, task_key=task_key, recipient_id=recipient_id,
|
|
result=snapshot.result if kind == "result" else None, template_id=template,
|
|
payload=json.dumps(payload, ensure_ascii=False), page=snapshot.page or "",
|
|
status="failed" if invalid else "pending", last_error="模板或单号不符合发送要求" if invalid else None,
|
|
create_by=actor_id or cycle.create_by, create_dept_id=user.dept_id or "",
|
|
))
|
|
db.flush()
|
|
|
|
|
|
def sync_approval_notifications(db, biz_type, biz_id, tenant_no, *, new_cycle=False, actor_id=""):
|
|
if not settings.WECHAT_OA_ENABLED:
|
|
return
|
|
db.flush()
|
|
cycle = db.query(WechatApprovalCycle).filter_by(
|
|
tenant_no=tenant_no, business_type=biz_type, business_id=biz_id, del_flag=0,
|
|
).with_for_update().first()
|
|
if not cycle and not new_cycle:
|
|
return # 上线前已存在的审批不补发历史通知。
|
|
snapshot = _snapshot(db, biz_type, biz_id, tenant_no)
|
|
if new_cycle:
|
|
db.query(WechatNotice).filter_by(
|
|
tenant_no=tenant_no, business_type=biz_type, business_id=biz_id, status="pending",
|
|
).filter(WechatNotice.kind.in_(("pending", "result"))).update(
|
|
{"status": "cancelled", "last_error": "审批已重新提交"}, synchronize_session="fetch")
|
|
if not snapshot:
|
|
return
|
|
if not cycle:
|
|
cycle = WechatApprovalCycle(id=generate_uuid(), tenant_no=tenant_no, business_type=biz_type, business_id=biz_id)
|
|
db.add(cycle)
|
|
cycle.cycle_id = generate_uuid()
|
|
cycle.submitted_at = datetime.now()
|
|
cycle.assignments = "{}"
|
|
cycle.create_by = actor_id or snapshot.applicant_id
|
|
db.flush()
|
|
if not cycle:
|
|
return
|
|
previous = json.loads(cycle.assignments or "{}")
|
|
current = {}
|
|
if snapshot:
|
|
for task in snapshot.pending:
|
|
key = _assignment_key(task)
|
|
current[key] = previous.get(key) or generate_uuid()
|
|
# 包括已经发出的任务在内，保存当前集合用于转交后重新激活。
|
|
cycle.assignments = json.dumps(current, ensure_ascii=False)
|
|
pending_rows = db.query(WechatNotice).filter_by(
|
|
tenant_no=tenant_no, business_type=biz_type, business_id=biz_id, status="pending",
|
|
).filter(WechatNotice.kind.in_(("pending", "result"))).all()
|
|
for row in pending_rows:
|
|
key = json.dumps([row.task_key, row.recipient_id], ensure_ascii=False)
|
|
valid = row.cycle_id == cycle.cycle_id and snapshot is not None
|
|
valid = valid and (key in current if row.kind == "pending" else snapshot.result == row.result)
|
|
if not valid:
|
|
row.status, row.last_error = "cancelled", "审批状态或处理人已变更"
|
|
db.flush()
|
|
if not snapshot:
|
|
return
|
|
for task in snapshot.pending:
|
|
_enqueue(db, cycle, snapshot, kind="pending", recipient_id=task.user_id,
|
|
task_key=task.task_key, assignment=current[_assignment_key(task)], actor_id=actor_id)
|
|
if snapshot.result in ("approved", "rejected"):
|
|
_enqueue(db, cycle, snapshot, kind="result", recipient_id=snapshot.applicant_id, actor_id=actor_id)
|
|
|
|
|
|
def _valid_now(db, notice):
|
|
if notice.appid != settings.WECHAT_OA_APPID or not _user(db, notice.tenant_no, notice.recipient_id):
|
|
return False
|
|
if notice.kind == "completion":
|
|
from app.services.wechat_completion_notices import valid_completion_notice
|
|
return valid_completion_notice(db, notice)
|
|
if notice.kind == "cert_daily":
|
|
from app.services.wechat_cert_notices import refresh_cert_notice
|
|
return refresh_cert_notice(db, notice)
|
|
if notice.kind not in ("pending", "result"):
|
|
return False
|
|
cycle = db.query(WechatApprovalCycle).filter_by(
|
|
tenant_no=notice.tenant_no, business_type=notice.business_type, business_id=notice.business_id, del_flag=0,
|
|
).first()
|
|
if not cycle or cycle.cycle_id != notice.cycle_id:
|
|
return False
|
|
snapshot = _snapshot(db, notice.business_type, notice.business_id, notice.tenant_no)
|
|
if not snapshot:
|
|
return False
|
|
if notice.kind == "result":
|
|
return snapshot.result == notice.result and snapshot.applicant_id == notice.recipient_id
|
|
key = json.dumps([notice.task_key, notice.recipient_id], ensure_ascii=False)
|
|
assignment = json.loads(cycle.assignments or "{}").get(key)
|
|
if not assignment or notice.event_key != _event_key(
|
|
notice.tenant_no, notice.appid, notice.business_type, notice.business_id,
|
|
notice.cycle_id, notice.kind, notice.task_key, notice.recipient_id, assignment,
|
|
):
|
|
return False
|
|
return any(t.task_key == notice.task_key and t.user_id == notice.recipient_id for t in snapshot.pending)
|
|
|
|
|
|
def _fallback_url(notice):
|
|
base = settings.WECHAT_OA_PUBLIC_BASE_URL.rstrip("/")
|
|
if not base:
|
|
return ""
|
|
if notice.kind == "cert_daily":
|
|
return base + "/#/pl/document-reminder-module/document-reminder"
|
|
if notice.business_type == "pl_freight":
|
|
route = "/pl/freight-fee/freight-approval" if notice.kind == "pending" else "/pl/freight-fee/freight"
|
|
return base + "/#" + route
|
|
if notice.kind == "pending":
|
|
route = {
|
|
"bx_advance": "/bx/advance/approve",
|
|
"bx_reimburse": "/bx/reimburse/approve",
|
|
"pl_recharge": "/vehicle-mgmt/etc-recharge/vehicle-recharge-approval",
|
|
}.get(notice.business_type)
|
|
if route:
|
|
return base + "/#" + route
|
|
# 原生页无 Web 对等详情时回门户；关联小程序配置后优先用 page。
|
|
return base + "/#/"
|
|
|
|
|
|
def deliver_pending(db, *, client=None, limit=20):
|
|
if not settings.WECHAT_OA_ENABLED:
|
|
return 0
|
|
client = client or default_client
|
|
now = datetime.now()
|
|
# 进程在发送与写入结果之间退出：人工核对，禁止自动重发潜在已送达消息。
|
|
db.query(WechatNotice).filter(
|
|
WechatNotice.status == "sending", WechatNotice.claimed_at < now - timedelta(minutes=3),
|
|
).update({"status": "unknown", "last_error": "发送进程中断，结果待核对"}, synchronize_session=False)
|
|
db.commit()
|
|
count = 0
|
|
for _ in range(limit):
|
|
row = db.query(WechatNotice).filter(
|
|
WechatNotice.status == "pending", WechatNotice.del_flag == 0,
|
|
WechatNotice.next_attempt_at <= datetime.now(),
|
|
).order_by(WechatNotice.create_time, WechatNotice.id).with_for_update().first()
|
|
if row is None:
|
|
db.rollback()
|
|
break
|
|
row.status, row.claimed_at = "sending", datetime.now()
|
|
row.attempts += 1
|
|
db.commit() # MySQL 5.7 支持 FOR UPDATE；提交认领后其他进程不能重复认领。
|
|
try:
|
|
db.expire_all()
|
|
if not _valid_now(db, row):
|
|
row.status, row.last_error = "cancelled", "审批状态或接收人已失效"
|
|
else:
|
|
binding = db.query(WechatBinding).filter_by(
|
|
tenant_no=row.tenant_no, user_id=row.recipient_id, appid=row.appid,
|
|
subscribed=1, del_flag=0,
|
|
).first()
|
|
if not binding or not binding.openid:
|
|
row.status, row.last_error = "skipped", "接收人未绑定或未关注服务号"
|
|
else:
|
|
row.message_id = client.send(
|
|
openid=binding.openid, template_id=row.template_id,
|
|
data=json.loads(row.payload), event_key=row.event_key,
|
|
page=row.page, url=_fallback_url(row),
|
|
)
|
|
row.status, row.sent_at, row.last_error = "sent", datetime.now(), None
|
|
count += 1
|
|
except WechatDeliveryUnknown:
|
|
row.status, row.last_error = "unknown", "微信发送结果不确定，请核对发送记录"
|
|
except WechatAPIError as exc:
|
|
row.last_error = str(exc)
|
|
if exc.code in (-1, 45009) and row.attempts < 5:
|
|
row.status = "pending"
|
|
row.next_attempt_at = datetime.now() + timedelta(seconds=min(3600, 60 * 2 ** row.attempts))
|
|
else:
|
|
row.status = "skipped" if exc.code == 43004 else "failed"
|
|
db.commit()
|
|
return count
|