200 lines
7.1 KiB
Python
200 lines
7.1 KiB
Python
# 返利班次：领表 4 起草行 / 表 3 只录协议行 → runner 查号+总单+细单+送审 → 写表 4。
|
||
#
|
||
# 崩在表 4 或表 3 的 process_status=3：超过租约可再领，未过期需改回 0。
|
||
# 未捕获异常原因列只写短码。原文进审计表，会话结束后另写。
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from collections import OrderedDict
|
||
from contextlib import nullcontext
|
||
from dataclasses import dataclass
|
||
from typing import Callable
|
||
|
||
from rpa_oracle import ShiftError
|
||
from persist import (
|
||
KIND_FAIL,
|
||
KIND_HOLD,
|
||
KIND_NOT_FOUND,
|
||
KIND_OK,
|
||
AgreementOutcome,
|
||
RebateGroup,
|
||
agreement_ok,
|
||
build_found,
|
||
build_not_found,
|
||
)
|
||
from repo import Claimed
|
||
|
||
HEADER_FAILED = "HEADER_FAILED"
|
||
IMPORT_FAILED = "IMPORT_FAILED"
|
||
SUBMIT_FAILED = "SUBMIT_FAILED"
|
||
PAGE_FAILED = "PAGE_FAILED"
|
||
Runner = Callable[[RebateGroup], AgreementOutcome]
|
||
Header = Callable[[str], str]
|
||
Importer = Callable[[str], object]
|
||
Submitter = Callable[[str], object]
|
||
logger = logging.getLogger("rpa_rebate")
|
||
|
||
|
||
# 一个返利班次的领取、合组和成败计数。
|
||
@dataclass
|
||
class JobSummary:
|
||
claimed: int = 0
|
||
groups: int = 0
|
||
ok: int = 0
|
||
not_found: int = 0
|
||
failed: int = 0
|
||
|
||
|
||
def recorder(
|
||
contract_no: str,
|
||
header: Header,
|
||
importer: Importer,
|
||
submitter: Submitter,
|
||
existing: Callable[[str], str] | None = None,
|
||
) -> AgreementOutcome:
|
||
"""按总单 → 细单新增 → 送审录入一份协议。
|
||
|
||
送审前按合同号查已有协议。已有总单号则直接带回，不再新建。
|
||
existing 返回空串表示确认没有协议。抛错表示读号失败，保持处理中。
|
||
任一步抛错只写短码，原文打日志。业务失败仍写对应中文原因。
|
||
"""
|
||
if existing is not None:
|
||
try:
|
||
found_id = (existing(contract_no) or "").strip()
|
||
except Exception as exc:
|
||
logger.exception("返利查已有协议失败 contract_no=%s", contract_no)
|
||
return AgreementOutcome(KIND_HOLD, None, None, "读号失败", _text(exc))
|
||
if found_id:
|
||
return AgreementOutcome(KIND_OK, found_id, None, None)
|
||
try:
|
||
order_no = (header(contract_no) or "").strip()
|
||
except Exception as exc:
|
||
logger.exception("返利总单未捕获异常 contract_no=%s", contract_no)
|
||
return AgreementOutcome(KIND_FAIL, None, None, HEADER_FAILED, _text(exc))
|
||
if not order_no:
|
||
return AgreementOutcome(KIND_FAIL, None, None, "总单未拿到号")
|
||
try:
|
||
imported = importer(contract_no)
|
||
except Exception as exc:
|
||
logger.exception("返利细单未捕获异常 contract_no=%s", contract_no)
|
||
return AgreementOutcome(KIND_FAIL, None, None, IMPORT_FAILED, _text(exc))
|
||
if not _import_ok(imported):
|
||
return AgreementOutcome(KIND_FAIL, None, None, _import_reason(imported))
|
||
try:
|
||
submitted = submitter(contract_no)
|
||
except Exception as exc:
|
||
logger.exception("返利送审未捕获异常 contract_no=%s", contract_no)
|
||
return AgreementOutcome(KIND_FAIL, None, None, SUBMIT_FAILED, _text(exc))
|
||
if submitted is None or not bool(getattr(submitted, "submitted", False)):
|
||
return AgreementOutcome(KIND_FAIL, None, None, "送审失败")
|
||
agreement_id = (getattr(submitted, "agreement_id", None) or "").strip()
|
||
if not agreement_id:
|
||
return AgreementOutcome(KIND_HOLD, None, None, "送审未读到总单号")
|
||
agreement_dtl_id = (getattr(submitted, "agreement_dtl_id", None) or "").strip() or None
|
||
return AgreementOutcome(KIND_OK, agreement_id, agreement_dtl_id, None)
|
||
|
||
|
||
def _session(repo):
|
||
"""有 session 就借一条，没有就空上下文。页面调用必须在这个块外面。"""
|
||
session = getattr(repo, "session", None)
|
||
if session is None:
|
||
return nullcontext()
|
||
return session()
|
||
|
||
|
||
def run_batch(
|
||
repo,
|
||
*,
|
||
runner: Runner,
|
||
limit: int | None = None,
|
||
) -> JobSummary:
|
||
"""领取待查号行，按合同号合组、调 runner、落表。
|
||
|
||
session 只包领取和每一组的落表。查号和录协议在连接归还之后跑。一组落完再处理下一组。
|
||
runner 抛未捕获异常时原因列只写 PAGE_FAILED。limit 交给领取层。
|
||
"""
|
||
audits: list[ShiftError] = []
|
||
with _session(repo):
|
||
claimed = repo.claim_pending(limit=limit)
|
||
groups = group_claimed(claimed)
|
||
summary = JobSummary(
|
||
claimed=len(claimed.drafts) + len(claimed.protocols),
|
||
groups=len(groups),
|
||
)
|
||
for group in groups:
|
||
try:
|
||
outcome = runner(group)
|
||
except Exception as exc:
|
||
logger.exception("返利班未捕获异常 contract_no=%s", group.contract_no)
|
||
outcome = AgreementOutcome(KIND_FAIL, None, None, PAGE_FAILED, _text(exc))
|
||
if outcome is None:
|
||
outcome = AgreementOutcome(KIND_FAIL, None, None, "协议录入失败")
|
||
if outcome.error_text:
|
||
audits.append(
|
||
ShiftError(
|
||
"返利",
|
||
group.contract_no,
|
||
outcome.reason or PAGE_FAILED,
|
||
outcome.error_text,
|
||
)
|
||
)
|
||
with _session(repo):
|
||
if outcome.kind == KIND_NOT_FOUND:
|
||
repo.save_group(build_not_found(group))
|
||
else:
|
||
repo.save_group(build_found(group, outcome))
|
||
if outcome.kind == KIND_NOT_FOUND:
|
||
summary.not_found += 1
|
||
continue
|
||
if agreement_ok(outcome):
|
||
summary.ok += 1
|
||
else:
|
||
summary.failed += 1
|
||
repo.save_shift_errors(audits)
|
||
return summary
|
||
|
||
|
||
# 把已领取的起草行和只录协议行按合同号收成组，组内顺序跟领取顺序一致。
|
||
def group_claimed(claimed: Claimed) -> list[RebateGroup]:
|
||
groups: OrderedDict[str, RebateGroup] = OrderedDict()
|
||
|
||
# 取出或新建该合同号的组。
|
||
def bucket(contract_no: str) -> RebateGroup:
|
||
number = (contract_no or "").strip()
|
||
if number not in groups:
|
||
groups[number] = RebateGroup(contract_no=number)
|
||
return groups[number]
|
||
|
||
for row in claimed.drafts:
|
||
bucket(row["contract_no"]).drafts.append(row)
|
||
for row in claimed.protocols:
|
||
bucket(row["contract_no"]).protocols.append(row)
|
||
return list(groups.values())
|
||
|
||
|
||
# 取出异常原文。空串不当审计正文。
|
||
def _text(exc: BaseException) -> str | None:
|
||
text = str(exc).strip()
|
||
return text or None
|
||
|
||
|
||
# 细单新增至少成功一行、失败零行、解析成功才算过。
|
||
def _import_ok(result: object) -> bool:
|
||
if result is None:
|
||
return False
|
||
success = int(getattr(result, "success_count", 0) or 0)
|
||
fail = int(getattr(result, "fail_count", 0) or 0)
|
||
parsed = getattr(result, "parsed", True)
|
||
return bool(parsed) and success > 0 and fail == 0
|
||
|
||
|
||
# 取出细单第一条失败原因；没有则写「细单导入失败」。
|
||
def _import_reason(result: object) -> str:
|
||
if result is None:
|
||
return "细单导入失败"
|
||
reasons = getattr(result, "fail_reasons", ()) or ()
|
||
if reasons:
|
||
return str(reasons[0]).strip() or "细单导入失败"
|
||
return "细单导入失败"
|