|
|
@@ -0,0 +1,1333 @@
|
|
|
+"""把一轮开跑收成卡片,写入自有库,再交给发送。
|
|
|
+
|
|
|
+不改结论行的字段,不重新定义公式。取数、规则和发送都调用已经存在的函数。
|
|
|
+"""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import asyncio
|
|
|
+import hashlib
|
|
|
+from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
|
|
|
+from dataclasses import dataclass, replace
|
|
|
+from datetime import date, datetime
|
|
|
+from pathlib import Path
|
|
|
+from typing import Any
|
|
|
+
|
|
|
+from asyncmy.errors import IntegrityError
|
|
|
+from loguru import logger
|
|
|
+
|
|
|
+from post_clear_check.config import (
|
|
|
+ Settings,
|
|
|
+ describe_exception,
|
|
|
+ push_log_suffix,
|
|
|
+ push_verdicts,
|
|
|
+ resolve_business_date,
|
|
|
+ secret_values,
|
|
|
+ shanghai_wall_now,
|
|
|
+)
|
|
|
+from post_clear_check.feishu import (
|
|
|
+ CARD_SUMMARY,
|
|
|
+ FeishuError,
|
|
|
+ OutboundCard,
|
|
|
+ SendResult,
|
|
|
+ SummaryOwner,
|
|
|
+ send_card,
|
|
|
+ send_summary,
|
|
|
+)
|
|
|
+from post_clear_check.gate import (
|
|
|
+ NO_HEDGE_END_TIME,
|
|
|
+ ConclusionLine,
|
|
|
+ GateScan,
|
|
|
+ QueryGateway,
|
|
|
+ ReadyRun,
|
|
|
+ TrsBlocked,
|
|
|
+ daycut_line,
|
|
|
+ scan_gate,
|
|
|
+ trs_blocked_line,
|
|
|
+)
|
|
|
+from post_clear_check.published import daycut_mismatches
|
|
|
+from post_clear_check.chain import (
|
|
|
+ CHAIN_GATE,
|
|
|
+ CHAIN_ORDER,
|
|
|
+ CHAIN_UNASSIGNED,
|
|
|
+ CHAINS,
|
|
|
+ GATE_MISSING_TABLE,
|
|
|
+ OWNER_BATCH,
|
|
|
+ OWNER_FUND,
|
|
|
+ OWNER_MASTER,
|
|
|
+ OWNER_USER,
|
|
|
+ SIDE_NONE,
|
|
|
+ hang,
|
|
|
+ hang_by_side,
|
|
|
+ owner_of,
|
|
|
+ unassign,
|
|
|
+)
|
|
|
+from post_clear_check.issue_common import is_population_bucket
|
|
|
+from post_clear_check.issue_reports import check_issue_reports
|
|
|
+from post_clear_check.order_cell import CheckLine
|
|
|
+from post_clear_check.order_decimal import VERDICT_CORRECT, VERDICT_ERROR, VERDICT_SUSPECT
|
|
|
+from post_clear_check.progress import Stage, WatchedGateway
|
|
|
+
|
|
|
+# 规则版本只哈希这些模块。编排和飞书变化不重算已核对的批次。链路登记表决定挂链和依据,算进规则。
|
|
|
+_RULE_FILE_NAMES = frozenset(
|
|
|
+ {
|
|
|
+ "order_cell.py",
|
|
|
+ "order_decimal.py",
|
|
|
+ "order_rules.py",
|
|
|
+ "hedge_link.py",
|
|
|
+ "reader.py",
|
|
|
+ "columns.py",
|
|
|
+ "chain.py",
|
|
|
+ "fund_chain.py",
|
|
|
+ }
|
|
|
+)
|
|
|
+_RULE_PREFIXES = ("issue_", "sales_", "trs_")
|
|
|
+_CLIP_MARK = "…已截断"
|
|
|
+_VERDICT_LABEL = {VERDICT_CORRECT: "正确", VERDICT_ERROR: "错误", VERDICT_SUSPECT: "可疑"}
|
|
|
+from post_clear_check.order_rules import check_order
|
|
|
+from post_clear_check.reader import ReadRequest, ReadResult, read_rows
|
|
|
+from post_clear_check.sales_reports import check_sales_reports
|
|
|
+from post_clear_check.trs_reports import check_trs_reports
|
|
|
+
|
|
|
+NOTIFY_PENDING = "pending"
|
|
|
+NOTIFY_SENT = "sent"
|
|
|
+NOTIFY_FAILED = "failed"
|
|
|
+
|
|
|
+# 归属种类,与链路登记表一致。
|
|
|
+CARD_MASTER = OWNER_MASTER
|
|
|
+CARD_FUND = OWNER_FUND
|
|
|
+CARD_USER = OWNER_USER
|
|
|
+CARD_BATCH = OWNER_BATCH
|
|
|
+
|
|
|
+_KIND_RANK = {CARD_MASTER: 0, CARD_FUND: 1, CARD_USER: 2, CARD_BATCH: 3}
|
|
|
+# 结论行每次 executemany 的行数。只影响进度推进的粒度,单条语句长度由驱动按 1MB 切分。
|
|
|
+_LINE_BATCH = 2000
|
|
|
+_VERDICTS = frozenset({VERDICT_CORRECT, VERDICT_ERROR, VERDICT_SUSPECT})
|
|
|
+_KINDS = frozenset(set(_KIND_RANK))
|
|
|
+
|
|
|
+# 自有库语句。业务连接不执行这些写入。
|
|
|
+_FIND_RUN_SQL = (
|
|
|
+ "SELECT id FROM check_run "
|
|
|
+ "WHERE market = %s AND trade_date = %s AND issue_batch_id = %s "
|
|
|
+ "AND sales_batch_id = %s AND trs_batch_id = %s AND issue_end_time = %s "
|
|
|
+ "AND sales_end_time = %s AND trs_end_time = %s AND rule_pack_version = %s"
|
|
|
+)
|
|
|
+_INSERT_RUN_SQL = (
|
|
|
+ "INSERT INTO check_run ("
|
|
|
+ "market, trade_date, business_date, issue_batch_id, sales_batch_id, trs_batch_id, "
|
|
|
+ "issue_end_time, sales_end_time, trs_end_time, issue_applied_revision, "
|
|
|
+ "sales_applied_revision, trs_applied_revision, rule_pack_version, "
|
|
|
+ "correct_count, error_count, suspect_count, create_time"
|
|
|
+ ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
|
|
|
+)
|
|
|
+_INSERT_RESULT_SQL = (
|
|
|
+ "INSERT INTO check_result ("
|
|
|
+ "run_id, card_kind, chain_name, biz_key, verdict, dedupe_key, notify_status, create_time, update_time"
|
|
|
+ ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)"
|
|
|
+)
|
|
|
+_INSERT_LINE_SQL = (
|
|
|
+ "INSERT INTO check_line ("
|
|
|
+ "result_id, formula_text, calc_text, calc_value, check_text, check_value, basis, verdict, "
|
|
|
+ "report_code, master_order_no, fund_account, user_account"
|
|
|
+ ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
|
|
|
+)
|
|
|
+_SELECT_UNSENT_SQL = (
|
|
|
+ "SELECT id, card_kind, chain_name, biz_key, verdict, notify_status FROM check_result "
|
|
|
+ "WHERE run_id = %s AND notify_status IN (%s, %s) ORDER BY id"
|
|
|
+)
|
|
|
+# 概况按这条结果现编归属名单。明细和已有概况都要读到,发送时再丢掉概况行。
|
|
|
+_SELECT_RUN_RESULTS_SQL = (
|
|
|
+ "SELECT card_kind, chain_name, biz_key, verdict FROM check_result "
|
|
|
+ "WHERE run_id = %s ORDER BY id"
|
|
|
+)
|
|
|
+_SELECT_LINES_SQL = (
|
|
|
+ "SELECT formula_text, calc_text, calc_value, check_text, check_value, basis, verdict, "
|
|
|
+ "report_code, master_order_no, fund_account, user_account "
|
|
|
+ "FROM check_line WHERE result_id = %s ORDER BY id"
|
|
|
+)
|
|
|
+_UPDATE_NOTIFY_SQL = "UPDATE check_result SET notify_status = %s, update_time = %s WHERE id = %s"
|
|
|
+_SELECT_DEDUPE_SQL = "SELECT dedupe_key FROM check_result WHERE run_id = %s"
|
|
|
+_ADD_COUNTS_SQL = (
|
|
|
+ "UPDATE check_run SET correct_count = correct_count + %s, error_count = error_count + %s, "
|
|
|
+ "suspect_count = suspect_count + %s WHERE id = %s"
|
|
|
+)
|
|
|
+_FIND_SNAPSHOT_SQL = (
|
|
|
+ "SELECT id FROM check_run WHERE market = %s AND business_date = %s "
|
|
|
+ "AND issue_batch_id = 0 AND sales_batch_id = 0 AND trs_batch_id = 0"
|
|
|
+)
|
|
|
+_SILENCE_SNAPSHOT_SQL = (
|
|
|
+ "UPDATE check_result SET notify_status = %s, update_time = %s "
|
|
|
+ "WHERE run_id = %s AND notify_status IN (%s, %s)"
|
|
|
+)
|
|
|
+_BATCH_NO_SQL = "SELECT batch_no FROM db_issue.clearing_batch WHERE id = %s"
|
|
|
+
|
|
|
+_OrderFn = Callable[[ReadResult], Mapping[str, Sequence[CheckLine]]]
|
|
|
+_ReportFn = Callable[[Mapping[str, Any]], Sequence[CheckLine]]
|
|
|
+_SendFn = Callable[[Settings, OutboundCard], Awaitable[SendResult]]
|
|
|
+
|
|
|
+
|
|
|
+def rule_pack_version() -> str:
|
|
|
+ """当前规则模块源文件的 sha256。文件没变则版本不变,已核对的轮次不重算。"""
|
|
|
+
|
|
|
+ root = Path(__file__).resolve().parent
|
|
|
+ files = {
|
|
|
+ path.name: path.read_bytes()
|
|
|
+ for path in root.glob("*.py")
|
|
|
+ if path.name in _RULE_FILE_NAMES or path.name.startswith(_RULE_PREFIXES)
|
|
|
+ }
|
|
|
+ return rule_pack_version_from(files)
|
|
|
+
|
|
|
+
|
|
|
+def rule_pack_version_from(files: Mapping[str, bytes]) -> str:
|
|
|
+ """按文件名排序哈希。测试用内存字节,不读磁盘。"""
|
|
|
+
|
|
|
+ digest = hashlib.sha256()
|
|
|
+ for name in sorted(files):
|
|
|
+ digest.update(name.encode("utf-8"))
|
|
|
+ digest.update(b"\0")
|
|
|
+ digest.update(files[name])
|
|
|
+ digest.update(b"\0")
|
|
|
+ return digest.hexdigest()
|
|
|
+
|
|
|
+
|
|
|
+@dataclass(frozen=True)
|
|
|
+class CardDraft:
|
|
|
+ """还没落库的一张卡:一条链路、一个归属对象。没有结论行的卡不会被构造出来。"""
|
|
|
+
|
|
|
+ chain_name: str
|
|
|
+ card_kind: str
|
|
|
+ biz_key: str
|
|
|
+ verdict: str
|
|
|
+ lines: tuple[CheckLine, ...]
|
|
|
+ # 门禁快照用。表名或「TRS未完成」变化时另开一张卡,不改三端结束时间。
|
|
|
+ snapshot_token: str = ""
|
|
|
+
|
|
|
+
|
|
|
+@dataclass(frozen=True)
|
|
|
+class StoredCard:
|
|
|
+ """自有库里待发送或待重发的一张卡。"""
|
|
|
+
|
|
|
+ id: int
|
|
|
+ card_kind: str
|
|
|
+ chain_name: str
|
|
|
+ biz_key: str
|
|
|
+ verdict: str
|
|
|
+ notify_status: str
|
|
|
+ lines: tuple[CheckLine, ...]
|
|
|
+
|
|
|
+
|
|
|
+def summarize(lines: Sequence[CheckLine]) -> str:
|
|
|
+ """有 error 就是 error,否则有 suspect 就是 suspect,否则是 correct。"""
|
|
|
+
|
|
|
+ return _worst_verdict(line.verdict for line in lines)
|
|
|
+
|
|
|
+
|
|
|
+def _worst_verdict(verdicts: Iterable[str]) -> str:
|
|
|
+ """明细卡和概况卡共用:错误高于可疑,可疑高于正确。"""
|
|
|
+
|
|
|
+ found = set(verdicts)
|
|
|
+ if not found <= _VERDICTS:
|
|
|
+ raise ValueError("卡片结论不在 correct、error、suspect 之内")
|
|
|
+ if VERDICT_ERROR in found:
|
|
|
+ return VERDICT_ERROR
|
|
|
+ if VERDICT_SUSPECT in found:
|
|
|
+ return VERDICT_SUSPECT
|
|
|
+ return VERDICT_CORRECT
|
|
|
+
|
|
|
+
|
|
|
+def _summary_drafts(details: Sequence[tuple[str, str, str, str]]) -> tuple[CardDraft, ...]:
|
|
|
+ """每条有明细的链路一张概况。没有结论行。biz_key 用链路代码,便于幂等。
|
|
|
+
|
|
|
+ details 的四列是链路、归属、主键、明细卡总结论。
|
|
|
+ """
|
|
|
+
|
|
|
+ grouped: dict[str, list[str]] = {}
|
|
|
+ for chain_name, _kind, _biz_key, verdict in details:
|
|
|
+ grouped.setdefault(chain_name, []).append(verdict)
|
|
|
+ unknown = sorted(code for code in grouped if code not in CHAIN_ORDER)
|
|
|
+ if unknown:
|
|
|
+ logger.error("概况跳过未登记链路:{}", "、".join(unknown))
|
|
|
+ drafts: list[CardDraft] = []
|
|
|
+ for chain_name in sorted(grouped, key=lambda code: CHAIN_ORDER.get(code, len(CHAIN_ORDER))):
|
|
|
+ if chain_name not in CHAIN_ORDER:
|
|
|
+ continue
|
|
|
+ drafts.append(
|
|
|
+ CardDraft(
|
|
|
+ chain_name,
|
|
|
+ CARD_SUMMARY,
|
|
|
+ chain_name,
|
|
|
+ _worst_verdict(grouped[chain_name]),
|
|
|
+ (),
|
|
|
+ )
|
|
|
+ )
|
|
|
+ return tuple(drafts)
|
|
|
+
|
|
|
+
|
|
|
+def _attach_summaries(cards: Sequence[CardDraft]) -> tuple[CardDraft, ...]:
|
|
|
+ """概况排在明细前面,这样新轮次的主键也是概况在先。已带概况的再收一次不会重复。"""
|
|
|
+
|
|
|
+ details = tuple(card for card in cards if card.card_kind != CARD_SUMMARY)
|
|
|
+ rows = tuple((card.chain_name, card.card_kind, card.biz_key, card.verdict) for card in details)
|
|
|
+ return _summary_drafts(rows) + details
|
|
|
+
|
|
|
+
|
|
|
+def _pending_order(cards: Sequence[StoredCard]) -> tuple[StoredCard, ...]:
|
|
|
+ """先发概况,概况之间按链路登记顺序;再按主键升序发明细。旧轮次补插的概况主键更大,不能靠主键排序。"""
|
|
|
+
|
|
|
+ summaries = [card for card in cards if card.card_kind == CARD_SUMMARY]
|
|
|
+ details = [card for card in cards if card.card_kind != CARD_SUMMARY]
|
|
|
+ summaries.sort(key=lambda card: CHAIN_ORDER.get(card.chain_name, len(CHAIN_ORDER)))
|
|
|
+ return tuple(summaries) + tuple(details)
|
|
|
+
|
|
|
+
|
|
|
+def dedupe_key(
|
|
|
+ ready: ReadyRun,
|
|
|
+ card_kind: str,
|
|
|
+ biz_key: str,
|
|
|
+ *,
|
|
|
+ chain_name: str,
|
|
|
+ snapshot_token: str = "",
|
|
|
+) -> str:
|
|
|
+ """按库表把一轮和一张卡拼成幂等键。放不下或业务键里有分隔符时改存摘要。
|
|
|
+
|
|
|
+ 链路代码紧跟卡片种类:同一主订单上的两条链路是两张卡,幂等键必须不同。
|
|
|
+ """
|
|
|
+
|
|
|
+ parts = [
|
|
|
+ ready.trade_date.isoformat(),
|
|
|
+ card_kind,
|
|
|
+ chain_name,
|
|
|
+ biz_key,
|
|
|
+ str(ready.issue_batch_id),
|
|
|
+ str(ready.sales_batch_id),
|
|
|
+ str(ready.trs_batch_id),
|
|
|
+ _clock_text(ready.issue_end_time),
|
|
|
+ _clock_text(ready.sales_end_time),
|
|
|
+ _clock_text(ready.trs_end_time),
|
|
|
+ rule_pack_version(),
|
|
|
+ ]
|
|
|
+ # 快照令牌只加在门禁卡上,避免普通卡的幂等键因空字段变化。
|
|
|
+ if snapshot_token:
|
|
|
+ parts.append(snapshot_token)
|
|
|
+ # 分隔符不出现在订单号和账号里。业务键若自带这个字符,拼接会撞车,改走摘要。
|
|
|
+ raw = "\x1f".join(parts)
|
|
|
+ if "\x1f" not in biz_key and "\x1f" not in snapshot_token and len(raw) <= 191:
|
|
|
+ return raw
|
|
|
+ return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
|
|
+
|
|
|
+
|
|
|
+def assemble_cards(
|
|
|
+ order_lines: Mapping[str, Sequence[CheckLine]],
|
|
|
+ issue_lines: Sequence[CheckLine],
|
|
|
+ sales_lines: Sequence[CheckLine],
|
|
|
+ trs_lines: Sequence[CheckLine],
|
|
|
+ *,
|
|
|
+ business_date: date | None = None,
|
|
|
+ missing_tables: Sequence[str] = (),
|
|
|
+ gate_lines: Sequence[CheckLine] = (),
|
|
|
+) -> tuple[CardDraft, ...]:
|
|
|
+ """按「链路 + 归属 + 主键」收卡。归属按链路登记的优先顺序取第一个非空键。
|
|
|
+
|
|
|
+ 没挂链、链路要求的归属键为空的行进未归链路,结论为错误。缺表进核对门禁。
|
|
|
+ """
|
|
|
+
|
|
|
+ buckets: dict[tuple[str, str, str], list[CheckLine]] = {}
|
|
|
+ loose: list[CheckLine] = []
|
|
|
+
|
|
|
+ def add(chain_name: str, kind: str, biz_key: str, line: CheckLine) -> None:
|
|
|
+ buckets.setdefault((chain_name, kind, biz_key), []).append(line)
|
|
|
+
|
|
|
+ def place(line: CheckLine) -> None:
|
|
|
+ if line.chain_name == CHAIN_UNASSIGNED:
|
|
|
+ loose.append(line)
|
|
|
+ return
|
|
|
+ chain = CHAINS.get(line.chain_name)
|
|
|
+ if chain is None:
|
|
|
+ loose.append(unassign(line, "没有挂上链路"))
|
|
|
+ return
|
|
|
+ owner = owner_of(line)
|
|
|
+ if owner is None:
|
|
|
+ loose.append(unassign(line, f"{chain.name}要求的归属键为空"))
|
|
|
+ return
|
|
|
+ add(chain.code, owner[0], owner[1], line)
|
|
|
+
|
|
|
+ for order_no in sorted(order_lines):
|
|
|
+ for line in order_lines[order_no]:
|
|
|
+ place(line)
|
|
|
+ for line in tuple(issue_lines) + tuple(sales_lines) + tuple(trs_lines):
|
|
|
+ place(line)
|
|
|
+ if loose or missing_tables:
|
|
|
+ if business_date is None:
|
|
|
+ raise ValueError("没有归属的结论行缺少营业日")
|
|
|
+ day = business_date.isoformat()
|
|
|
+ for line in loose:
|
|
|
+ add(CHAIN_UNASSIGNED, CARD_BATCH, day, line)
|
|
|
+ for table in missing_tables:
|
|
|
+ add(CHAIN_GATE, CARD_BATCH, day, _missing_table_line(table))
|
|
|
+ if gate_lines:
|
|
|
+ if business_date is None:
|
|
|
+ raise ValueError("核对门禁结论行缺少营业日")
|
|
|
+ day = business_date.isoformat()
|
|
|
+ for line in gate_lines:
|
|
|
+ add(CHAIN_GATE, CARD_BATCH, day, line)
|
|
|
+ drafts: list[CardDraft] = []
|
|
|
+ # 同一归属对象的几条链路相邻,群里按主键能连着看。
|
|
|
+ order = sorted(buckets, key=lambda item: (_KIND_RANK[item[1]], item[2], CHAIN_ORDER[item[0]]))
|
|
|
+ for chain_name, kind, biz_key in order:
|
|
|
+ rows = tuple(buckets[(chain_name, kind, biz_key)])
|
|
|
+ if not rows:
|
|
|
+ continue
|
|
|
+ drafts.append(CardDraft(chain_name, kind, biz_key, summarize(rows), rows))
|
|
|
+ return tuple(drafts)
|
|
|
+
|
|
|
+
|
|
|
+def assemble_batch_cards(lines: Sequence[ConclusionLine]) -> tuple[CardDraft, ...]:
|
|
|
+ """批次门禁一条一个业务键。同一个批次号上的多条放在同一张卡,归核对门禁。"""
|
|
|
+
|
|
|
+ buckets: dict[str, list[ConclusionLine]] = {}
|
|
|
+ for line in lines:
|
|
|
+ buckets.setdefault(line.biz_key, []).append(line)
|
|
|
+ drafts: list[CardDraft] = []
|
|
|
+ for biz_key in sorted(buckets):
|
|
|
+ source = tuple(buckets[biz_key])
|
|
|
+ rows = tuple(_from_conclusion(line) for line in source)
|
|
|
+ if not rows:
|
|
|
+ continue
|
|
|
+ drafts.append(
|
|
|
+ CardDraft(
|
|
|
+ CHAIN_GATE,
|
|
|
+ CARD_BATCH,
|
|
|
+ biz_key,
|
|
|
+ summarize(rows),
|
|
|
+ rows,
|
|
|
+ snapshot_token=_snapshot_token(source),
|
|
|
+ )
|
|
|
+ )
|
|
|
+ return tuple(drafts)
|
|
|
+
|
|
|
+
|
|
|
+def split_batch_runs(lines: Sequence[ConclusionLine]) -> tuple[tuple[ReadyRun, tuple[CardDraft, ...]], ...]:
|
|
|
+ """还没有完成批次的门禁行,按市场和营业日合成一轮。"""
|
|
|
+
|
|
|
+ grouped: dict[tuple[str, date, date], list[ConclusionLine]] = {}
|
|
|
+ for line in lines:
|
|
|
+ grouped.setdefault(_batch_identity(line), []).append(line)
|
|
|
+ found: list[tuple[ReadyRun, tuple[CardDraft, ...]]] = []
|
|
|
+ for market, trade_date, business_date in sorted(grouped):
|
|
|
+ ready = _synthetic_run(market, trade_date, business_date)
|
|
|
+ found.append((ready, assemble_batch_cards(grouped[(market, trade_date, business_date)])))
|
|
|
+ return tuple(found)
|
|
|
+
|
|
|
+
|
|
|
+def collect_order_lines(result: ReadResult) -> dict[str, list[CheckLine]]:
|
|
|
+ """每张主单单独跑订单规则,按主订单方向挂到买入执行或卖出执行。sources 是逻辑表到来源。
|
|
|
+
|
|
|
+ 只跑核对人口。报表行引用的往日主订单是参照数据,执行公式已在它的交易日核过,不再每天重复出卡。
|
|
|
+ """
|
|
|
+
|
|
|
+ found: dict[str, list[CheckLine]] = {}
|
|
|
+ for order_no in sorted(result.orders):
|
|
|
+ bucket = result.orders[order_no]
|
|
|
+ if not isinstance(bucket, Mapping) or not is_population_bucket(bucket):
|
|
|
+ continue
|
|
|
+ # 订单结论的主订单编号以字典键为准,不看正文里有没有写。
|
|
|
+ lines = [
|
|
|
+ replace(line, master_order_no=str(order_no))
|
|
|
+ for line in check_order(bucket, result.sources)
|
|
|
+ ]
|
|
|
+ found[str(order_no)] = hang_by_side(lines, {str(order_no): bucket}, SIDE_NONE)
|
|
|
+ return found
|
|
|
+
|
|
|
+
|
|
|
+class SqlRunStore:
|
|
|
+ """自有库上的三张表。一轮的插入包在同一个事务里。"""
|
|
|
+
|
|
|
+ def __init__(self, connection: Any) -> None:
|
|
|
+ self._connection = connection
|
|
|
+
|
|
|
+ async def find_run_id(self, ready: ReadyRun) -> int | None:
|
|
|
+ """按唯一键找已经写过的一轮。没有则返回空。"""
|
|
|
+
|
|
|
+ rows = await self._query(_FIND_RUN_SQL, _run_key(ready))
|
|
|
+ if not rows:
|
|
|
+ return None
|
|
|
+ return int(rows[0][0])
|
|
|
+
|
|
|
+ async def save_new_run(
|
|
|
+ self,
|
|
|
+ ready: ReadyRun,
|
|
|
+ cards: Sequence[CardDraft],
|
|
|
+ now: datetime,
|
|
|
+ *,
|
|
|
+ tick: Callable[[int], None] | None = None,
|
|
|
+ ) -> int | None:
|
|
|
+ """插入一轮和它的卡片。唯一键冲突时整段回滚,返回空。tick 每写一条结论行回报一次,只用于进度显示。"""
|
|
|
+
|
|
|
+ correct, error, suspect = _counts(cards)
|
|
|
+ await self._connection.begin()
|
|
|
+ try:
|
|
|
+ run_id = await self._insert_run(ready, correct, error, suspect, now)
|
|
|
+ if run_id is None:
|
|
|
+ await self._connection.rollback()
|
|
|
+ return None
|
|
|
+ for card in cards:
|
|
|
+ await self._insert_card(run_id, ready, card, now, tick)
|
|
|
+ await self._connection.commit()
|
|
|
+ except Exception:
|
|
|
+ await self._rollback_quietly()
|
|
|
+ raise
|
|
|
+ return run_id
|
|
|
+
|
|
|
+ async def add_missing_cards(
|
|
|
+ self,
|
|
|
+ run_id: int,
|
|
|
+ ready: ReadyRun,
|
|
|
+ cards: Sequence[CardDraft],
|
|
|
+ now: datetime,
|
|
|
+ ) -> None:
|
|
|
+ """快照范围变了才追加卡。已有幂等键的卡不复制结论行。"""
|
|
|
+
|
|
|
+ existing = await self._query(_SELECT_DEDUPE_SQL, (run_id,))
|
|
|
+ known = {str(row[0]) for row in existing}
|
|
|
+ fresh = [card for card in cards if _card_dedupe(ready, card) not in known]
|
|
|
+ if not fresh:
|
|
|
+ return
|
|
|
+ correct, error, suspect = _counts(fresh)
|
|
|
+ await self._connection.begin()
|
|
|
+ try:
|
|
|
+ for card in fresh:
|
|
|
+ await self._insert_card(run_id, ready, card, now, None)
|
|
|
+ await self._execute(_ADD_COUNTS_SQL, (correct, error, suspect, run_id))
|
|
|
+ await self._connection.commit()
|
|
|
+ except Exception:
|
|
|
+ await self._rollback_quietly()
|
|
|
+ raise
|
|
|
+
|
|
|
+ async def silence_snapshots(self, market: str, business_date: date, now: datetime) -> None:
|
|
|
+ """真实轮次已经落下后,未发出的门禁快照不再补发。"""
|
|
|
+
|
|
|
+ rows = await self._query(_FIND_SNAPSHOT_SQL, (market, business_date))
|
|
|
+ for row in rows:
|
|
|
+ run_id = int(row[0])
|
|
|
+ await self._connection.begin()
|
|
|
+ try:
|
|
|
+ await self._execute(
|
|
|
+ _SILENCE_SNAPSHOT_SQL,
|
|
|
+ (NOTIFY_SENT, now, run_id, NOTIFY_PENDING, NOTIFY_FAILED),
|
|
|
+ )
|
|
|
+ await self._connection.commit()
|
|
|
+ except Exception:
|
|
|
+ await self._rollback_quietly()
|
|
|
+ raise
|
|
|
+
|
|
|
+ async def unsent_cards(self, run_id: int) -> tuple[StoredCard, ...]:
|
|
|
+ """pending 和 failed,按主键升序。已 sent 的不返回。"""
|
|
|
+
|
|
|
+ rows = await self._query(_SELECT_UNSENT_SQL, (run_id, NOTIFY_PENDING, NOTIFY_FAILED))
|
|
|
+ cards: list[StoredCard] = []
|
|
|
+ for row in rows:
|
|
|
+ result_id = int(row[0])
|
|
|
+ chain_name = str(row[2])
|
|
|
+ line_rows = await self._query(_SELECT_LINES_SQL, (result_id,))
|
|
|
+ cards.append(
|
|
|
+ StoredCard(
|
|
|
+ id=result_id,
|
|
|
+ card_kind=str(row[1]),
|
|
|
+ chain_name=chain_name,
|
|
|
+ biz_key=str(row[3]),
|
|
|
+ verdict=str(row[4]),
|
|
|
+ notify_status=str(row[5]),
|
|
|
+ # 链路记在卡上,读回的结论行沿用卡的链路,与落库前一致。
|
|
|
+ lines=tuple(replace(_line_from_row(item), chain_name=chain_name) for item in line_rows),
|
|
|
+ )
|
|
|
+ )
|
|
|
+ return tuple(cards)
|
|
|
+
|
|
|
+ async def detail_rows(self, run_id: int) -> tuple[tuple[str, str, str, str], ...]:
|
|
|
+ """这一轮全部卡片的归属四列:种类、链路、主键、总结论。含概况行,调用方自己丢掉。"""
|
|
|
+
|
|
|
+ rows = await self._query(_SELECT_RUN_RESULTS_SQL, (run_id,))
|
|
|
+ return tuple((str(row[0]), str(row[1]), str(row[2]), str(row[3])) for row in rows)
|
|
|
+
|
|
|
+ async def ensure_summaries(self, run_id: int, ready: ReadyRun, now: datetime) -> None:
|
|
|
+ """旧轮次没有概况卡时补插。不改三个计数,不复制明细的结论行。
|
|
|
+
|
|
|
+ 和明细写在不同的事务里:补插失败不会把已经落下的明细卷走。
|
|
|
+ 幂等键撞上说明另一个进程刚补过,回滚后继续发。
|
|
|
+ """
|
|
|
+
|
|
|
+ rows = await self.detail_rows(run_id)
|
|
|
+ have = {chain for kind, chain, _biz, _verdict in rows if kind == CARD_SUMMARY}
|
|
|
+ details = tuple(
|
|
|
+ (chain, kind, biz, verdict)
|
|
|
+ for kind, chain, biz, verdict in rows
|
|
|
+ if kind != CARD_SUMMARY
|
|
|
+ )
|
|
|
+ fresh = [card for card in _summary_drafts(details) if card.chain_name not in have]
|
|
|
+ if not fresh:
|
|
|
+ return
|
|
|
+ await self._connection.begin()
|
|
|
+ try:
|
|
|
+ for card in fresh:
|
|
|
+ await self._insert_card(run_id, ready, card, now, None)
|
|
|
+ await self._connection.commit()
|
|
|
+ except IntegrityError as exc:
|
|
|
+ await self._rollback_quietly()
|
|
|
+ if _duplicate(exc):
|
|
|
+ return
|
|
|
+ raise
|
|
|
+ except Exception:
|
|
|
+ await self._rollback_quietly()
|
|
|
+ raise
|
|
|
+ logger.info("补插链路概况 {} 张", len(fresh))
|
|
|
+
|
|
|
+ async def set_notify(self, result_id: int, status: str, now: datetime) -> None:
|
|
|
+ """把一张卡改成 sent 或 failed。只按主键更新。"""
|
|
|
+
|
|
|
+ if status not in {NOTIFY_SENT, NOTIFY_FAILED}:
|
|
|
+ raise ValueError("发送状态只能是 sent 或 failed")
|
|
|
+ await self._connection.begin()
|
|
|
+ try:
|
|
|
+ await self._execute(_UPDATE_NOTIFY_SQL, (status, now, result_id))
|
|
|
+ await self._connection.commit()
|
|
|
+ except Exception:
|
|
|
+ await self._rollback_quietly()
|
|
|
+ raise
|
|
|
+
|
|
|
+ async def _insert_run(
|
|
|
+ self,
|
|
|
+ ready: ReadyRun,
|
|
|
+ correct: int,
|
|
|
+ error: int,
|
|
|
+ suspect: int,
|
|
|
+ now: datetime,
|
|
|
+ ) -> int | None:
|
|
|
+ args = (
|
|
|
+ ready.market,
|
|
|
+ ready.trade_date,
|
|
|
+ ready.business_date,
|
|
|
+ ready.issue_batch_id,
|
|
|
+ ready.sales_batch_id,
|
|
|
+ ready.trs_batch_id,
|
|
|
+ ready.issue_end_time,
|
|
|
+ ready.sales_end_time,
|
|
|
+ ready.trs_end_time,
|
|
|
+ ready.issue_applied_revision,
|
|
|
+ ready.sales_applied_revision,
|
|
|
+ ready.trs_applied_revision,
|
|
|
+ rule_pack_version(),
|
|
|
+ correct,
|
|
|
+ error,
|
|
|
+ suspect,
|
|
|
+ now,
|
|
|
+ )
|
|
|
+ async with self._connection.cursor() as cursor:
|
|
|
+ try:
|
|
|
+ await cursor.execute(_INSERT_RUN_SQL, args)
|
|
|
+ except IntegrityError as exc:
|
|
|
+ if _duplicate(exc):
|
|
|
+ return None
|
|
|
+ raise
|
|
|
+ return int(cursor.lastrowid)
|
|
|
+
|
|
|
+ async def _insert_card(
|
|
|
+ self,
|
|
|
+ run_id: int,
|
|
|
+ ready: ReadyRun,
|
|
|
+ card: CardDraft,
|
|
|
+ now: datetime,
|
|
|
+ tick: Callable[[int], None] | None,
|
|
|
+ ) -> None:
|
|
|
+ result_args = (
|
|
|
+ run_id,
|
|
|
+ card.card_kind,
|
|
|
+ card.chain_name,
|
|
|
+ card.biz_key,
|
|
|
+ card.verdict,
|
|
|
+ _card_dedupe(ready, card),
|
|
|
+ NOTIFY_PENDING,
|
|
|
+ now,
|
|
|
+ now,
|
|
|
+ )
|
|
|
+ async with self._connection.cursor() as cursor:
|
|
|
+ await cursor.execute(_INSERT_RESULT_SQL, result_args)
|
|
|
+ result_id = int(cursor.lastrowid)
|
|
|
+ rows = [
|
|
|
+ (
|
|
|
+ result_id,
|
|
|
+ line.formula_text,
|
|
|
+ line.calc_text,
|
|
|
+ line.calc_value,
|
|
|
+ line.check_text,
|
|
|
+ line.check_value,
|
|
|
+ line.basis,
|
|
|
+ line.verdict,
|
|
|
+ line.report_code,
|
|
|
+ line.master_order_no,
|
|
|
+ line.fund_account,
|
|
|
+ line.user_account,
|
|
|
+ )
|
|
|
+ for line in card.lines
|
|
|
+ ]
|
|
|
+ # 结论行批量写入:驱动把同一条 INSERT 合并成多行 VALUES,并按语句长度自动再切。
|
|
|
+ # 先按块调用,只为让进度按块推进;一张卡仍在本轮同一个事务里。
|
|
|
+ for start in range(0, len(rows), _LINE_BATCH):
|
|
|
+ chunk = rows[start : start + _LINE_BATCH]
|
|
|
+ async with self._connection.cursor() as cursor:
|
|
|
+ await cursor.executemany(_INSERT_LINE_SQL, chunk)
|
|
|
+ if tick is not None:
|
|
|
+ tick(len(chunk))
|
|
|
+
|
|
|
+ async def _query(self, sql: str, args: tuple[Any, ...]) -> list[tuple[Any, ...]]:
|
|
|
+ async with self._connection.cursor() as cursor:
|
|
|
+ await cursor.execute(sql, args)
|
|
|
+ if cursor.description is None:
|
|
|
+ return []
|
|
|
+ return list(await cursor.fetchall())
|
|
|
+
|
|
|
+ async def _execute(self, sql: str, args: tuple[Any, ...]) -> None:
|
|
|
+ async with self._connection.cursor() as cursor:
|
|
|
+ await cursor.execute(sql, args)
|
|
|
+
|
|
|
+ async def _rollback_quietly(self) -> None:
|
|
|
+ try:
|
|
|
+ await self._connection.rollback()
|
|
|
+ except Exception as exc:
|
|
|
+ logger.error("回滚自有库事务失败:{}", describe_exception(exc))
|
|
|
+
|
|
|
+
|
|
|
+async def issue_batch_no(gateway: QueryGateway, batch_id: int) -> str:
|
|
|
+ """卡片头上的批次号取发行完成批次。读不到时用批次 id,不编一个号。"""
|
|
|
+
|
|
|
+ if isinstance(batch_id, bool) or not isinstance(batch_id, int) or batch_id <= 0:
|
|
|
+ return "0"
|
|
|
+ rows = await gateway.execute(_BATCH_NO_SQL, (batch_id,))
|
|
|
+ if not rows:
|
|
|
+ return str(batch_id)
|
|
|
+ value = rows[0][0]
|
|
|
+ if isinstance(value, bytes):
|
|
|
+ value = value.decode("utf-8", errors="replace")
|
|
|
+ if not isinstance(value, str):
|
|
|
+ return str(batch_id)
|
|
|
+ text = value.strip().splitlines()[0] if value.strip() else ""
|
|
|
+ return text or str(batch_id)
|
|
|
+
|
|
|
+
|
|
|
+async def run_once(
|
|
|
+ business: QueryGateway,
|
|
|
+ own: Any,
|
|
|
+ settings: Settings,
|
|
|
+ *,
|
|
|
+ store: SqlRunStore | None = None,
|
|
|
+ scan: Callable[[QueryGateway], Awaitable[GateScan]] | None = None,
|
|
|
+ read: Callable[[QueryGateway, ReadRequest], Awaitable[ReadResult]] | None = None,
|
|
|
+ orders: _OrderFn | None = None,
|
|
|
+ issue: _ReportFn | None = None,
|
|
|
+ sales: _ReportFn | None = None,
|
|
|
+ trs: _ReportFn | None = None,
|
|
|
+ sender: _SendFn | None = None,
|
|
|
+ today: date | None = None,
|
|
|
+) -> None:
|
|
|
+ """做一轮。没有开跑且没有批次级结论行就返回。today 只给测试注入上海今天。"""
|
|
|
+
|
|
|
+ read_fn = read or read_rows
|
|
|
+ order_fn = orders or collect_order_lines
|
|
|
+ issue_fn = issue or check_issue_reports
|
|
|
+ sales_fn = sales or check_sales_reports
|
|
|
+ trs_fn = trs or check_trs_reports
|
|
|
+ send_fn = sender or _send_outbound
|
|
|
+ owned = store or SqlRunStore(own)
|
|
|
+ # 没配置营业日时只核上海时区的今天。注入的扫描只在测试里使用,不改它的入参。
|
|
|
+ day = resolve_business_date(settings)
|
|
|
+ wall_today = today or shanghai_wall_now().date()
|
|
|
+ logger.info("本轮只核对营业日 {}", day.isoformat())
|
|
|
+ async with Stage("检查三端清算批次") as stage:
|
|
|
+ watched = WatchedGateway(business, stage)
|
|
|
+ if scan is None:
|
|
|
+ scanned = await scan_gate(watched, day)
|
|
|
+ else:
|
|
|
+ scanned = await scan(watched)
|
|
|
+ stage.summary = f"{watched.count} 条查询,可核对 {len(scanned.ready)} 轮"
|
|
|
+ _log_trs_blocked(scanned)
|
|
|
+ historical = day < wall_today
|
|
|
+ if not scanned.ready and not scanned.lines and not (historical and scanned.blocked):
|
|
|
+ # 当天被 TRS 挡住已经另有日志,不再写成「没有开跑」。
|
|
|
+ if not scanned.blocked:
|
|
|
+ logger.info("没有开跑,本轮结束")
|
|
|
+ return
|
|
|
+ for ready in scanned.ready:
|
|
|
+ matching = tuple(
|
|
|
+ line
|
|
|
+ for line in scanned.checks
|
|
|
+ if line.market == ready.market
|
|
|
+ and line.trade_date == ready.trade_date
|
|
|
+ and line.business_date == ready.business_date
|
|
|
+ )
|
|
|
+ await _process_ready(
|
|
|
+ business,
|
|
|
+ settings,
|
|
|
+ owned,
|
|
|
+ ready,
|
|
|
+ read_fn,
|
|
|
+ order_fn,
|
|
|
+ issue_fn,
|
|
|
+ sales_fn,
|
|
|
+ trs_fn,
|
|
|
+ send_fn,
|
|
|
+ matching,
|
|
|
+ )
|
|
|
+ await owned.silence_snapshots(ready.market, ready.business_date, _now())
|
|
|
+ for ready, cards in split_batch_runs(scanned.lines):
|
|
|
+ normalized = tuple(item for item in (_normalize_card(card) for card in cards) if item is not None)
|
|
|
+ await _process_existing_or_save(
|
|
|
+ settings,
|
|
|
+ owned,
|
|
|
+ ready,
|
|
|
+ normalized,
|
|
|
+ send_fn,
|
|
|
+ batch_no=None,
|
|
|
+ allow_new_cards=True,
|
|
|
+ )
|
|
|
+ if historical:
|
|
|
+ for blocked in scanned.blocked:
|
|
|
+ await _store_trs_blocked(settings, owned, blocked, send_fn)
|
|
|
+
|
|
|
+
|
|
|
+async def _process_ready(
|
|
|
+ business: QueryGateway,
|
|
|
+ settings: Settings,
|
|
|
+ store: SqlRunStore,
|
|
|
+ ready: ReadyRun,
|
|
|
+ read_fn: Callable[[QueryGateway, ReadRequest], Awaitable[ReadResult]],
|
|
|
+ order_fn: _OrderFn,
|
|
|
+ issue_fn: _ReportFn,
|
|
|
+ sales_fn: _ReportFn,
|
|
|
+ trs_fn: _ReportFn,
|
|
|
+ send_fn: _SendFn,
|
|
|
+ gate_checks: Sequence[ConclusionLine] = (),
|
|
|
+) -> None:
|
|
|
+ """已有同一轮次时不取数、不重算。新轮次先算完再一次性落库。"""
|
|
|
+
|
|
|
+ existing = await store.find_run_id(ready)
|
|
|
+ if existing is not None:
|
|
|
+ # 唯一键已在:不重新计算。已送达的卡跳过,失败和未送达的用原行重发。
|
|
|
+ logger.info("轮次已存在,不重新计算")
|
|
|
+ batch_no = await issue_batch_no(business, ready.issue_batch_id)
|
|
|
+ await _send_unsent(settings, store, existing, ready, send_fn, batch_no)
|
|
|
+ return
|
|
|
+ async with Stage("从业务库取数") as stage:
|
|
|
+ watched = WatchedGateway(business, stage)
|
|
|
+ result = await read_fn(watched, _request(ready))
|
|
|
+ stage.summary = f"{watched.count} 条查询,主订单 {len(result.orders)} 张"
|
|
|
+ payload = _payload(result)
|
|
|
+ # 规则是纯计算,放到线程里跑,事件循环空出来刷新进度。线程内不 await,输入已全部取齐。
|
|
|
+ async with Stage("核对订单和三端报表", total=4) as stage:
|
|
|
+ stage.note("订单规则")
|
|
|
+ order_lines = await asyncio.to_thread(order_fn, result)
|
|
|
+ stage.advance()
|
|
|
+ stage.note("发行端报表")
|
|
|
+ issue_lines = await asyncio.to_thread(issue_fn, payload)
|
|
|
+ stage.advance()
|
|
|
+ stage.note("销售端报表")
|
|
|
+ sales_lines = await asyncio.to_thread(sales_fn, payload)
|
|
|
+ stage.advance()
|
|
|
+ stage.note("TRS 报表")
|
|
|
+ trs_lines = await asyncio.to_thread(trs_fn, payload)
|
|
|
+ stage.advance()
|
|
|
+ gate_lines = tuple(_from_conclusion(line) for line in gate_checks)
|
|
|
+ compared = daycut_line(ready, daycut_mismatches(result.fund_accounts, result.missing_tables))
|
|
|
+ if compared is not None:
|
|
|
+ gate_lines = gate_lines + (_from_conclusion(compared),)
|
|
|
+ cards = assemble_cards(
|
|
|
+ order_lines,
|
|
|
+ issue_lines,
|
|
|
+ sales_lines,
|
|
|
+ trs_lines,
|
|
|
+ business_date=ready.business_date,
|
|
|
+ missing_tables=result.missing_tables,
|
|
|
+ gate_lines=gate_lines,
|
|
|
+ )
|
|
|
+ normalized = _attach_summaries(
|
|
|
+ tuple(item for item in (_normalize_card(card) for card in cards) if item is not None)
|
|
|
+ )
|
|
|
+ detail_count = sum(1 for card in normalized if card.card_kind != CARD_SUMMARY)
|
|
|
+ stage.summary = (
|
|
|
+ f"明细 {detail_count} 张,概况 {len(normalized) - detail_count} 张,"
|
|
|
+ f"结论行 {sum(len(card.lines) for card in normalized)} 条"
|
|
|
+ )
|
|
|
+ await _process_existing_or_save(settings, store, ready, normalized, send_fn, business=business)
|
|
|
+
|
|
|
+
|
|
|
+async def _process_existing_or_save(
|
|
|
+ settings: Settings,
|
|
|
+ store: SqlRunStore,
|
|
|
+ ready: ReadyRun,
|
|
|
+ cards: Sequence[CardDraft],
|
|
|
+ send_fn: _SendFn,
|
|
|
+ *,
|
|
|
+ business: QueryGateway | None = None,
|
|
|
+ batch_no: str | None = None,
|
|
|
+ allow_new_cards: bool = False,
|
|
|
+) -> None:
|
|
|
+ """先落库。并发下唯一键冲突则改为重发已有卡,不再插一行结论。概况和明细一起落,计数不含概况。"""
|
|
|
+
|
|
|
+ cards = _attach_summaries(cards)
|
|
|
+ existing = await store.find_run_id(ready)
|
|
|
+ if existing is not None:
|
|
|
+ logger.info("轮次已存在,不重新计算")
|
|
|
+ # 门禁快照的表名变了才追加卡。已完成轮次不在这里补结论。
|
|
|
+ if allow_new_cards:
|
|
|
+ await store.add_missing_cards(existing, ready, cards, _now())
|
|
|
+ shown = batch_no
|
|
|
+ if shown is None and business is not None:
|
|
|
+ shown = await issue_batch_no(business, ready.issue_batch_id)
|
|
|
+ await _send_unsent(settings, store, existing, ready, send_fn, shown)
|
|
|
+ return
|
|
|
+ total_lines = sum(len(card.lines) for card in cards)
|
|
|
+ async with Stage("写入自有库", total=total_lines) as stage:
|
|
|
+ run_id = await store.save_new_run(ready, cards, _now(), tick=stage.advance)
|
|
|
+ detail_count = sum(1 for card in cards if card.card_kind != CARD_SUMMARY)
|
|
|
+ stage.summary = f"明细 {detail_count} 张,概况 {len(cards) - detail_count} 张,结论行 {total_lines} 条"
|
|
|
+ if run_id is None:
|
|
|
+ found = await store.find_run_id(ready)
|
|
|
+ if found is None:
|
|
|
+ return
|
|
|
+ logger.info("轮次已存在,不重新计算")
|
|
|
+ run_id = found
|
|
|
+ if allow_new_cards:
|
|
|
+ await store.add_missing_cards(run_id, ready, cards, _now())
|
|
|
+ shown = batch_no
|
|
|
+ if shown is None and business is not None:
|
|
|
+ shown = await issue_batch_no(business, ready.issue_batch_id)
|
|
|
+ await _send_unsent(settings, store, run_id, ready, send_fn, shown)
|
|
|
+
|
|
|
+
|
|
|
+async def _send_unsent(
|
|
|
+ settings: Settings,
|
|
|
+ store: SqlRunStore,
|
|
|
+ run_id: int,
|
|
|
+ ready: ReadyRun,
|
|
|
+ send_fn: _SendFn,
|
|
|
+ batch_no: str | None,
|
|
|
+) -> None:
|
|
|
+ """先补齐并发送概况,再按主键升序发送明细。成功改 sent,失败改 failed,不新插结论行。"""
|
|
|
+
|
|
|
+ async with Stage("发送飞书卡片") as stage:
|
|
|
+ stage.note("补齐链路概况")
|
|
|
+ await store.ensure_summaries(run_id, ready, _now())
|
|
|
+ stage.note("读取待发送的卡片")
|
|
|
+ cards = _pending_order(await store.unsent_cards(run_id))
|
|
|
+ rows = await store.detail_rows(run_id)
|
|
|
+ owners = _owners_by_chain(rows)
|
|
|
+ stage.total = len(cards) or None
|
|
|
+ allowed = push_verdicts(settings)
|
|
|
+ tally = {NOTIFY_SENT: 0, NOTIFY_FAILED: 0, "skipped": 0}
|
|
|
+ for card in cards:
|
|
|
+ stage.note(f"{card.chain_name} {card.biz_key}")
|
|
|
+ tally[
|
|
|
+ await _send_one(
|
|
|
+ settings,
|
|
|
+ store,
|
|
|
+ card,
|
|
|
+ ready,
|
|
|
+ send_fn,
|
|
|
+ batch_no,
|
|
|
+ allowed=allowed,
|
|
|
+ owners=owners.get(card.chain_name, ()),
|
|
|
+ )
|
|
|
+ ] += 1
|
|
|
+ stage.advance()
|
|
|
+ stage.summary = (
|
|
|
+ f"已发送 {tally[NOTIFY_SENT]} 张,失败 {tally[NOTIFY_FAILED]} 张,"
|
|
|
+ f"未推送 {tally['skipped']} 张(无结论行、无归属或不在推送配置内)"
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+async def _send_outbound(settings: Settings, card: OutboundCard) -> SendResult:
|
|
|
+ """明细走原版式。概况另走概况标题和正文,避免空结论行走明细的「没有结论行,不发送」。"""
|
|
|
+
|
|
|
+ if card.card_kind == CARD_SUMMARY:
|
|
|
+ return await send_summary(settings, card)
|
|
|
+ return await send_card(settings, card)
|
|
|
+
|
|
|
+
|
|
|
+def _owners_by_chain(
|
|
|
+ rows: Sequence[tuple[str, str, str, str]],
|
|
|
+) -> dict[str, tuple[SummaryOwner, ...]]:
|
|
|
+ """明细卡按链路收成概况用的归属。概况行自己不进名单。正文只列主键,不带结论行。"""
|
|
|
+
|
|
|
+ grouped: dict[str, list[SummaryOwner]] = {}
|
|
|
+ for kind, chain_name, biz_key, verdict in rows:
|
|
|
+ if kind == CARD_SUMMARY:
|
|
|
+ continue
|
|
|
+ grouped.setdefault(chain_name, []).append(SummaryOwner(kind, biz_key, verdict))
|
|
|
+ return {chain_name: tuple(items) for chain_name, items in grouped.items()}
|
|
|
+
|
|
|
+
|
|
|
+async def _send_one(
|
|
|
+ settings: Settings,
|
|
|
+ store: SqlRunStore,
|
|
|
+ card: StoredCard,
|
|
|
+ ready: ReadyRun,
|
|
|
+ send_fn: _SendFn,
|
|
|
+ batch_no: str | None,
|
|
|
+ *,
|
|
|
+ allowed: frozenset[str] | None,
|
|
|
+ owners: Sequence[SummaryOwner] = (),
|
|
|
+) -> str:
|
|
|
+ """发一张卡。成功改 sent,失败改 failed;没发时返回 skipped。返回值只给进度汇总用。
|
|
|
+
|
|
|
+ 概况卡不看推送配置:配置成只推错误时,正确链路的概况仍然先发。过滤只留在明细卡上。
|
|
|
+ """
|
|
|
+
|
|
|
+ summary = card.card_kind == CARD_SUMMARY
|
|
|
+ if summary:
|
|
|
+ if not owners:
|
|
|
+ return "skipped"
|
|
|
+ elif not card.lines:
|
|
|
+ return "skipped"
|
|
|
+ elif allowed is not None and card.verdict not in allowed:
|
|
|
+ label = _VERDICT_LABEL.get(card.verdict, card.verdict)
|
|
|
+ logger.info(
|
|
|
+ "飞书卡片不推送,{} {} {},结论 {} 不在推送配置内",
|
|
|
+ card.chain_name,
|
|
|
+ card.card_kind,
|
|
|
+ card.biz_key,
|
|
|
+ label,
|
|
|
+ )
|
|
|
+ return "skipped"
|
|
|
+ shown = batch_no if batch_no is not None else card.biz_key
|
|
|
+ outbound = OutboundCard(
|
|
|
+ card_kind=card.card_kind,
|
|
|
+ chain_name=card.chain_name,
|
|
|
+ biz_key=card.biz_key,
|
|
|
+ business_date=ready.business_date,
|
|
|
+ trade_date=ready.trade_date,
|
|
|
+ batch_no=shown,
|
|
|
+ lines=card.lines,
|
|
|
+ record_no=str(card.id),
|
|
|
+ resend=card.notify_status == NOTIFY_FAILED,
|
|
|
+ owners=tuple(owners) if summary else (),
|
|
|
+ )
|
|
|
+ try:
|
|
|
+ result = await send_fn(settings, outbound)
|
|
|
+ except FeishuError as exc:
|
|
|
+ logger.error("飞书卡片未能发送:{}{}", exc, push_log_suffix(settings))
|
|
|
+ await store.set_notify(card.id, NOTIFY_FAILED, _now())
|
|
|
+ return NOTIFY_FAILED
|
|
|
+ except Exception as exc:
|
|
|
+ logger.error(
|
|
|
+ "飞书卡片未能发送:{}{}",
|
|
|
+ describe_exception(exc, *secret_values(settings)),
|
|
|
+ push_log_suffix(settings),
|
|
|
+ )
|
|
|
+ await store.set_notify(card.id, NOTIFY_FAILED, _now())
|
|
|
+ return NOTIFY_FAILED
|
|
|
+ status = NOTIFY_SENT if result.ok else NOTIFY_FAILED
|
|
|
+ await store.set_notify(card.id, status, _now())
|
|
|
+ return status
|
|
|
+
|
|
|
+
|
|
|
+def _card_dedupe(ready: ReadyRun, card: CardDraft) -> str:
|
|
|
+ return dedupe_key(
|
|
|
+ ready,
|
|
|
+ card.card_kind,
|
|
|
+ card.biz_key,
|
|
|
+ chain_name=card.chain_name,
|
|
|
+ snapshot_token=card.snapshot_token,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _missing_table_line(table: str) -> CheckLine:
|
|
|
+ """1146 单独成行,归核对门禁。不改各张报表的公式。"""
|
|
|
+
|
|
|
+ line = CheckLine(
|
|
|
+ formula_text=GATE_MISSING_TABLE,
|
|
|
+ calc_text="读取该表时返回 1146",
|
|
|
+ calc_value=None,
|
|
|
+ check_text=table,
|
|
|
+ check_value=None,
|
|
|
+ basis="1146",
|
|
|
+ verdict=VERDICT_ERROR,
|
|
|
+ )
|
|
|
+ return hang((line,), CHAIN_GATE, SIDE_NONE)[0]
|
|
|
+
|
|
|
+
|
|
|
+def _snapshot_token(lines: Sequence[ConclusionLine]) -> str:
|
|
|
+ """发行或销售加上有行的表名。集合变化则幂等键变化。"""
|
|
|
+
|
|
|
+ bits: list[str] = []
|
|
|
+ for line in lines:
|
|
|
+ if not line.side_name and not line.report_tables:
|
|
|
+ continue
|
|
|
+ tables = ",".join(line.report_tables)
|
|
|
+ bits.append(f"{line.side_name}:{tables}")
|
|
|
+ return "|".join(sorted(bits))
|
|
|
+
|
|
|
+
|
|
|
+def _log_trs_blocked(scanned: GateScan) -> None:
|
|
|
+ """每次跳过都写明是 TRS 未完成,不和批次还没跑完混成同一句。"""
|
|
|
+
|
|
|
+ for blocked in scanned.blocked:
|
|
|
+ logger.info(
|
|
|
+ "市场 {} 交易日 {} 营业日 {} 有对冲,TRS 批次未完成,发行和销售本轮不核对",
|
|
|
+ blocked.market,
|
|
|
+ blocked.trade_date.isoformat(),
|
|
|
+ blocked.business_date.isoformat(),
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+async def _store_trs_blocked(
|
|
|
+ settings: Settings,
|
|
|
+ store: SqlRunStore,
|
|
|
+ blocked: TrsBlocked,
|
|
|
+ send_fn: _SendFn,
|
|
|
+) -> None:
|
|
|
+ """早于今天的挡住记一张错误卡。同一市场同一交易日只保留一轮。"""
|
|
|
+
|
|
|
+ line = trs_blocked_line(blocked)
|
|
|
+ draft = CardDraft(
|
|
|
+ CHAIN_GATE,
|
|
|
+ CARD_BATCH,
|
|
|
+ line.biz_key,
|
|
|
+ VERDICT_ERROR,
|
|
|
+ (_from_conclusion(line),),
|
|
|
+ snapshot_token="TRS未完成",
|
|
|
+ )
|
|
|
+ ready = _synthetic_run(blocked.market, blocked.trade_date, blocked.business_date)
|
|
|
+ normalized = _normalize_card(draft)
|
|
|
+ if normalized is None:
|
|
|
+ return
|
|
|
+ await _process_existing_or_save(
|
|
|
+ settings,
|
|
|
+ store,
|
|
|
+ ready,
|
|
|
+ (normalized,),
|
|
|
+ send_fn,
|
|
|
+ batch_no=None,
|
|
|
+ allow_new_cards=True,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _from_conclusion(line: ConclusionLine) -> CheckLine:
|
|
|
+ """门禁结论行挂核对门禁。依据换成研发设计里的门禁口径,原说明挪到代入式后面。"""
|
|
|
+
|
|
|
+ row = CheckLine(
|
|
|
+ formula_text=line.formula_text,
|
|
|
+ calc_text=line.calc_text,
|
|
|
+ calc_value=line.calc_value,
|
|
|
+ check_text=line.check_text,
|
|
|
+ check_value=line.check_value,
|
|
|
+ basis=line.basis,
|
|
|
+ verdict=line.verdict,
|
|
|
+ )
|
|
|
+ return hang((row,), CHAIN_GATE, SIDE_NONE)[0]
|
|
|
+
|
|
|
+
|
|
|
+def _batch_identity(line: ConclusionLine) -> tuple[str, date, date]:
|
|
|
+ """市场和营业日来自结论行字段。没有交易日时用营业日。"""
|
|
|
+
|
|
|
+ if line.business_date is None:
|
|
|
+ raise ValueError("批次结论行没有营业日")
|
|
|
+ market = line.market
|
|
|
+ if len(market) > 8:
|
|
|
+ raise ValueError("批次结论行的市场超过 8 位")
|
|
|
+ trade_date = line.trade_date or line.business_date
|
|
|
+ return market, trade_date, line.business_date
|
|
|
+
|
|
|
+
|
|
|
+def _synthetic_run(market: str, trade_date: date, business_date: date) -> ReadyRun:
|
|
|
+ """门禁还没有完成批次。三端 id 用 0,结束时间用哨兵,避免和已完成轮次撞键。"""
|
|
|
+
|
|
|
+ return ReadyRun(
|
|
|
+ market=market,
|
|
|
+ trade_date=trade_date,
|
|
|
+ business_date=business_date,
|
|
|
+ issue_batch_id=0,
|
|
|
+ sales_batch_id=0,
|
|
|
+ trs_batch_id=0,
|
|
|
+ issue_end_time=NO_HEDGE_END_TIME,
|
|
|
+ sales_end_time=NO_HEDGE_END_TIME,
|
|
|
+ trs_end_time=NO_HEDGE_END_TIME,
|
|
|
+ issue_applied_revision=0,
|
|
|
+ sales_applied_revision=0,
|
|
|
+ trs_applied_revision=0,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _normalize_card(card: CardDraft) -> CardDraft | None:
|
|
|
+ if card.card_kind not in _KINDS:
|
|
|
+ raise ValueError("卡片种类不在四种之内")
|
|
|
+ chain = CHAINS.get(card.chain_name)
|
|
|
+ if chain is None:
|
|
|
+ raise ValueError("卡片链路不在登记表之内")
|
|
|
+ if card.card_kind not in chain.owners:
|
|
|
+ raise ValueError("卡片归属不在该链路登记的归属之内")
|
|
|
+ stored, original = _fit_biz_key(card.biz_key)
|
|
|
+ raw_lines = card.lines
|
|
|
+ if original and raw_lines:
|
|
|
+ first = raw_lines[0]
|
|
|
+ raw_lines = (replace(first, formula_text=f"原业务键:{original}\n{first.formula_text}"),) + tuple(raw_lines[1:])
|
|
|
+ lines = tuple(_normalize_line(line) for line in raw_lines)
|
|
|
+ if not lines:
|
|
|
+ return None
|
|
|
+ return CardDraft(
|
|
|
+ card.chain_name,
|
|
|
+ card.card_kind,
|
|
|
+ stored,
|
|
|
+ summarize(lines),
|
|
|
+ lines,
|
|
|
+ snapshot_token=card.snapshot_token,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _normalize_line(line: CheckLine) -> CheckLine:
|
|
|
+ if line.verdict not in _VERDICTS:
|
|
|
+ raise ValueError("结论不在 correct、error、suspect 之内")
|
|
|
+ # 正文不是幂等键。超长时按列宽截断并标明,避免这一行写不进去把整轮退回重试。
|
|
|
+ return replace(
|
|
|
+ line,
|
|
|
+ formula_text=_clip(line.formula_text, 500),
|
|
|
+ calc_text=_clip(line.calc_text, 1000),
|
|
|
+ check_text=_clip(line.check_text, 1000),
|
|
|
+ basis=_clip(line.basis, 500),
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _fit_biz_key(value: str) -> tuple[str, str]:
|
|
|
+ """放得下时原键为空。超长时入库键稳定,原键写进卡片正文。"""
|
|
|
+
|
|
|
+ if len(value) <= 64:
|
|
|
+ return value, ""
|
|
|
+ digest = hashlib.sha256(value.encode("utf-8")).hexdigest()
|
|
|
+ return f"{value[:39]}~{digest[:24]}", value
|
|
|
+
|
|
|
+
|
|
|
+def _clip(text: str, limit: int) -> str:
|
|
|
+ if len(text) <= limit:
|
|
|
+ return text
|
|
|
+ mark = _CLIP_MARK
|
|
|
+ if limit <= len(mark):
|
|
|
+ return text[:limit]
|
|
|
+ return text[: limit - len(mark)] + mark
|
|
|
+
|
|
|
+
|
|
|
+def _counts(cards: Sequence[CardDraft]) -> tuple[int, int, int]:
|
|
|
+ """按明细卡片数计,不按结论行数,也不把链路概况算进去。"""
|
|
|
+
|
|
|
+ correct = error = suspect = 0
|
|
|
+ for card in cards:
|
|
|
+ if card.card_kind == CARD_SUMMARY:
|
|
|
+ continue
|
|
|
+ if card.verdict == VERDICT_CORRECT:
|
|
|
+ correct += 1
|
|
|
+ elif card.verdict == VERDICT_ERROR:
|
|
|
+ error += 1
|
|
|
+ elif card.verdict == VERDICT_SUSPECT:
|
|
|
+ suspect += 1
|
|
|
+ else:
|
|
|
+ raise ValueError("卡片结论不在 correct、error、suspect 之内")
|
|
|
+ return correct, error, suspect
|
|
|
+
|
|
|
+
|
|
|
+def _request(ready: ReadyRun) -> ReadRequest:
|
|
|
+ return ReadRequest(
|
|
|
+ market=ready.market,
|
|
|
+ trade_date=ready.trade_date,
|
|
|
+ business_date=ready.business_date,
|
|
|
+ issue_batch_id=ready.issue_batch_id,
|
|
|
+ sales_batch_id=ready.sales_batch_id,
|
|
|
+ trs_batch_id=ready.trs_batch_id,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _payload(result: ReadResult) -> dict[str, Any]:
|
|
|
+ """把取数结果交给报表规则。没有读到的来源不在这里补。"""
|
|
|
+
|
|
|
+ return {
|
|
|
+ "market": result.market,
|
|
|
+ "trade_date": result.trade_date,
|
|
|
+ "business_date": result.business_date,
|
|
|
+ "orders": result.orders,
|
|
|
+ "fund_accounts": result.fund_accounts,
|
|
|
+ "trs_users": result.trs_users,
|
|
|
+ "sources": result.sources,
|
|
|
+ "loose_reports": result.loose_reports,
|
|
|
+ "unmapped_ledgers": result.unmapped_ledgers,
|
|
|
+ # 票据已卖出合计用的往日订单,不在 orders 里。
|
|
|
+ "note_orders": result.note_orders,
|
|
|
+ "note_copies": result.note_copies,
|
|
|
+ "latest_prices": result.latest_prices,
|
|
|
+ "missing_tables": result.missing_tables,
|
|
|
+ "locations": result.locations,
|
|
|
+ "stages": result.stages,
|
|
|
+ "issue_stages": result.issue_stages,
|
|
|
+ "baseline_kind": result.baseline_kind,
|
|
|
+ "baseline_notes": result.baseline_notes,
|
|
|
+ "live_notes_loaded": result.live_notes_loaded,
|
|
|
+ "live_note_subs": result.live_note_subs,
|
|
|
+ "live_note_masters": result.live_note_masters,
|
|
|
+ "live_topups": result.live_topups,
|
|
|
+ "live_extensions": result.live_extensions,
|
|
|
+ "live_positions": result.live_positions,
|
|
|
+ "cleared_positions": result.cleared_positions,
|
|
|
+ "sales_note_loaded": result.sales_note_loaded,
|
|
|
+ "sales_current_notes": result.sales_current_notes,
|
|
|
+ "sales_cleared_notes": result.sales_cleared_notes,
|
|
|
+ "sales_fund_loaded": result.sales_fund_loaded,
|
|
|
+ "sales_orders": result.sales_orders,
|
|
|
+ "sales_settle_day": result.sales_settle_day,
|
|
|
+ "sales_baseline_state": result.sales_baseline_state,
|
|
|
+ "trs_steps_loaded": result.trs_steps_loaded,
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+def _run_key(ready: ReadyRun) -> tuple[Any, ...]:
|
|
|
+ return (
|
|
|
+ ready.market,
|
|
|
+ ready.trade_date,
|
|
|
+ ready.issue_batch_id,
|
|
|
+ ready.sales_batch_id,
|
|
|
+ ready.trs_batch_id,
|
|
|
+ ready.issue_end_time,
|
|
|
+ ready.sales_end_time,
|
|
|
+ ready.trs_end_time,
|
|
|
+ rule_pack_version(),
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _line_from_row(row: tuple[Any, ...]) -> CheckLine:
|
|
|
+ report_code = str(row[7]) if len(row) > 7 else ""
|
|
|
+ master_order_no = str(row[8]) if len(row) > 8 else ""
|
|
|
+ fund_account = str(row[9]) if len(row) > 9 else ""
|
|
|
+ user_account = str(row[10]) if len(row) > 10 else ""
|
|
|
+ return CheckLine(
|
|
|
+ formula_text=str(row[0]),
|
|
|
+ calc_text=str(row[1]),
|
|
|
+ calc_value=row[2],
|
|
|
+ check_text=str(row[3]),
|
|
|
+ check_value=row[4],
|
|
|
+ basis=str(row[5]),
|
|
|
+ verdict=str(row[6]),
|
|
|
+ report_code=report_code,
|
|
|
+ master_order_no=master_order_no,
|
|
|
+ fund_account=fund_account,
|
|
|
+ user_account=user_account,
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+def _clock_text(value: datetime) -> str:
|
|
|
+ return value.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
+
|
|
|
+
|
|
|
+def _now() -> datetime:
|
|
|
+ """自有库的创建时间用上海墙钟,不跟容器的 UTC。"""
|
|
|
+
|
|
|
+ return shanghai_wall_now()
|
|
|
+
|
|
|
+
|
|
|
+def _duplicate(exc: IntegrityError) -> bool:
|
|
|
+ args = getattr(exc, "args", ())
|
|
|
+ if not args:
|
|
|
+ return False
|
|
|
+ return args[0] == 1062
|