diff --git a/flows/monitor/alert_failure.py b/flows/monitor/alert_failure.py
index 870a08b..4f7c975 100644
--- a/flows/monitor/alert_failure.py
+++ b/flows/monitor/alert_failure.py
@@ -102,14 +102,11 @@ def wecom_alert_failure(event: dict | str | None = None) -> dict:
ended = detail.get("end_time") or detail.get("expected_start_time")
occurred = occurred or ended
- lines = [
- f"🔴 **流程失败** {flow_name or resource_name or '未知'}",
- ]
- if run_name and run_name != flow_name:
- lines.append(f"> Run：{run_name}")
+ lines = [f"## 🔴 流程失败：{flow_name or resource_name or '未知'}"]
+ if run_name:
+ lines.append(f"> Run「{run_name}」 · {event_name.split('.')[-1]} · {_fmt_ts(occurred)}")
if state_msg:
lines.append(f"> 错误：{state_msg[:150].replace(chr(10), ' ')}")
- lines.append(f"> {event_name} · {_fmt_ts(occurred)}")
if run_id:
lines.append(f"> [查看详情]({UI_BASE}/runs/flow-run/{run_id})")
else:
diff --git a/flows/monitor/daily_report.py b/flows/monitor/daily_report.py
index a7770d1..41d5e4d 100644
--- a/flows/monitor/daily_report.py
+++ b/flows/monitor/daily_report.py
@@ -27,7 +27,7 @@ def _affected_units(diff: dict) -> int:
def _system_lines(name: str, report: dict, goods: dict, lines: list[str]) -> bool:
- """单系统的核验两段:①数量是否一致(订单+商品) ②状态不符多少单。返回是否需关注。"""
+ """单系统一行结论:数量(远程 vs 本地,一致只显示一个数)+商品;状态不符进引用行。"""
vr = (report.get("verifyReport") or {}) if isinstance(report, dict) else {}
days_r = vr.get("days") or []
post = vr.get("post")
@@ -36,35 +36,27 @@ def _system_lines(name: str, report: dict, goods: dict, lines: list[str]) -> boo
else:
remote_t = sum(d.get("remote", 0) for d in days_r)
local_t = sum(d.get("local", 0) for d in days_r)
- # 数量一致只看总量;状态差异属第二段,不混入数量结论
- ok = remote_t == local_t
+ ok = remote_t == local_t # 数量一致只看总量;状态差异属第二段,不混入数量结论
- if ok:
- lines.append(f"**{name} 订单** 远程 {remote_t:,} · 本地 {local_t:,} ✅")
- else:
- lines.append(f"**{name} 订单** 远程 {remote_t:,} · 本地 {local_t:,} ⚠️ 不一致")
+ g = goods or {}
+ zero = g.get("zeroGoodsOrders", 0)
+ goods_txt = f"{g.get('goods', 0):,} 行" + (f"（缺 {zero:,}）" if zero else "")
+ order_txt = f"订单 {remote_t:,} ✅" if ok else f"订单 远程 {remote_t:,} / 本地 {local_t:,} ⚠️"
+ lines.append(f"**{name}** {order_txt} · 商品 {goods_txt} {'✅' if not zero else '⚠️'}")
diff_days = [d for d in days_r if not d.get("ok")]
diff_units = sum(_affected_units(d.get("diff")) for d in diff_days)
if diff_units:
- lines.append(f"> 核验时发现约 {diff_units:,} 单状态不符（{len(diff_days)} 天），已自动刷新修复")
-
- g = goods or {}
- zero = g.get("zeroGoodsOrders", 0)
- if zero:
- lines.append(f"**{name} 商品** {g.get('goods', 0):,} 行 · {zero:,} 单缺商品 ⚠️")
- return True
- lines.append(f"**{name} 商品** {g.get('goods', 0):,} 行 · 无缺漏 ✅")
- return not ok
+ lines.append(f"> 状态不符约 {diff_units:,} 单（{len(diff_days)} 天），已自动刷新修复")
+ return (not ok) or bool(zero)
def build_daily_message(
days: int, window: str, riin: dict, sds: dict,
goods: dict, ocr_today: dict, ocr_pending: int, failures: list[dict],
) -> tuple[str, bool]:
- """拼装日报 markdown,返回 (消息, 是否需关注)。"""
- lines = [f"📊 **数据核验日报** {datetime.now().strftime('%m-%d %H:%M')}"
- f"（近 {days} 天，{window}）", ""]
+ """拼装日报 markdown,返回 (消息, 是否需关注)。单行结论 + 引用详情的紧凑排版。"""
+ lines = [f"## 📊 数据核验日报 {datetime.now():%m-%d %H:%M}（近 {days} 天 {window}）"]
attention = False
for name, report in (("RIIN", riin), ("SDS", sds)):
@@ -73,30 +65,26 @@ def build_daily_message(
except Exception as err: # noqa: BLE001
lines.append(f"**{name}** 核验结果解析失败：{err}")
attention = True
- lines.append("")
processed, failed = (ocr_today or {}).get("processed", 0), (ocr_today or {}).get("failed", 0)
- lines.append(f"**OCR 今日** 已处理 {processed:,} 单 · 识别失败 {failed:,} 单" + (" ✅" if not failed else ""))
+ ocr_txt = f"**OCR 今日** 处理 {processed:,} · 识别失败 {failed:,}" + (" ✅" if not failed else " ⚠️")
if ocr_pending:
- lines.append(f"> 累计待处理 {ocr_pending:,} 单")
- lines.append("")
+ ocr_txt += f"（累计待处理 {ocr_pending:,}）"
+ lines.append(ocr_txt)
if failures:
attention = True
- lines.append(f"**失败任务(24h)** 共 {len(failures)} 次 🔴")
+ lines.append(f"**失败任务(24h)** {len(failures)} 次 🔴")
for f in failures[:6]:
start = (f.get("start") or "")[5:] # MM-DD HH:MM
- lines.append(f"> {f.get('flow', '?')} · run「{f.get('name', '?')}」{f.get('state', '?')} · {start}")
+ lines.append(f"> {f.get('flow', '?')}「{f.get('name', '?')}」{f.get('state', '?')} · {start}")
if len(failures) > 6:
lines.append(f"> …等共 {len(failures)} 次")
else:
- lines.append("**失败任务(24h)** 0 次 ✅")
+ lines.append("**失败任务(24h)** 0 ✅")
- lines.append("")
- if attention:
- lines.append("⚠️ **存在需关注项，见上**")
- else:
- lines.append("✅ **数据全部一致，无异常**")
+ lines.append('⚠️ 存在需关注项，见上' if attention
+ else '✅ 数据全部一致，无异常')
return "\n".join(lines), attention
diff --git a/flows/monitor/device_check.py b/flows/monitor/device_check.py
index afe2a6b..014a25a 100644
--- a/flows/monitor/device_check.py
+++ b/flows/monitor/device_check.py
@@ -291,11 +291,11 @@ def _check_services() -> dict:
def _mark(pct: float | None, threshold: int) -> str:
- """百分比展示:超阈值加粗+🔴(WeCom aibot markdown 不支持 font 标签和表格)。"""
+ """百分比展示:超阈值用 warning 色加粗+🔴。"""
if pct is None:
return "—"
if pct >= threshold:
- return f"**{pct:.0f}%**🔴"
+ return f'**{pct:.0f}%** 🔴'
return f"{pct:.0f}%"
@@ -343,104 +343,94 @@ def device_check(alarm_threshold: int = 80) -> dict:
print(f"[device-check] 磁盘检查失败: {err}")
threshold = alarm_threshold
- header = f"🖥 **设备日报** {datetime.now():%m-%d %H:%M}（过去 24h · 采样 {agg['count']} 次"
+ header = f"## 🖥 设备日报 {datetime.now():%m-%d %H:%M}（24h 采样 {agg['count']} 次"
if agg.get("firstTs"):
first = datetime.fromisoformat(agg["firstTs"])
if first > datetime.now() - timedelta(hours=23):
header += f" · 自 {first:%m-%d %H:%M} 起"
- lines = [header + "）", ""]
+ lines = [header + "）"]
# ---- 内存 / swap ----
mem, swap = agg.get("mem"), agg.get("swap")
if mem:
- lines.append(f"**内存** 平均 {mem['avg']:.0f}% · 峰值 {_mark(mem['peak'], threshold)} · 最低 {mem['min']:.0f}%"
- + (f"（共 {_gb((agg.get('memTotal') or 0) * (1 << 30))}）" if agg.get("memTotal") else ""))
+ total_txt = f"（共 {_gb((agg.get('memTotal') or 0) * (1 << 30))}）" if agg.get("memTotal") else ""
+ lines.append(f"**内存** 均 {mem['avg']:.0f}% · 峰 {_mark(mem['peak'], threshold)} · 低 {mem['min']:.0f}%{total_txt}")
else:
try:
now_mem = _check_memory()
- lines.append(f"**内存** 当前 {_mark(now_mem['pct'], threshold)}（{now_mem['used']}/{now_mem['total']}，无采样数据）")
+ lines.append(f"**内存** 当前 {_mark(now_mem['pct'], threshold)}（{now_mem['used']}/{now_mem['total']}，无采样）")
mem = {"peak": now_mem["pct"]}
except Exception: # noqa: BLE001
lines.append("**内存** 检查失败")
mem = None
if swap:
- lines.append(f"**交换内存** 平均 {swap['avg']:.0f}% · 峰值 {_mark(swap['peak'], threshold)} · 最低 {swap['min']:.0f}%"
- + (f"（共 {_gb((agg.get('swapTotal') or 0) * (1 << 30))}）" if agg.get("swapTotal") else ""))
+ total_txt = f"（共 {_gb((agg.get('swapTotal') or 0) * (1 << 30))}）" if agg.get("swapTotal") else ""
+ lines.append(f"**交换内存** 均 {swap['avg']:.0f}% · 峰 {_mark(swap['peak'], threshold)} · 低 {swap['min']:.0f}%{total_txt}")
else:
lines.append("**交换内存** 无采样数据")
swap = None
# ---- 磁盘:总体 / 根目录 / home ----
- lines.append("**磁盘**")
if overall:
- lines.append(f"> 总体 已用 {_gb(overall['used'])} · 剩余 {_gb(overall['free'])}（{_mark(overall['pct'], threshold)}）")
+ lines.append(f"**磁盘** 总体 已用 {_gb(overall['used'])} · 剩 {_gb(overall['free'])}"
+ f"（{_mark(overall['pct'], threshold)}）")
+ else:
+ lines.append("**磁盘** 检查不可用")
label_map = {"/": "root", "/home": "home"}
for m in mounts:
g = growth.get(label_map.get(m["label"], ""))
- growth_txt = ""
- if g and g.get("dir"):
- growth_txt = f"｜增长最快 {g['dir']} {_fmt_growth(g['growth'])}"
- elif m["label"] in ("/", "/home"):
- growth_txt = "｜增长数据不足" if not g else ""
- lines.append(f"> {m['label']} 已用 {_gb(m['used'])} · 剩余 {_gb(m['free'])}"
- f"（{_mark(m['pct'], threshold)}）{growth_txt}")
- if not mounts:
- lines.append("> 磁盘检查不可用")
+ growth_txt = f" · 增长最快 {g['dir']} {_fmt_growth(g['growth'])}" if g and g.get("dir") else ""
+ lines.append(f"> {m['label']} {_mark(m['pct'], threshold)} · 已用 {_gb(m['used'])} · 剩 {_gb(m['free'])}{growth_txt}")
# ---- GPU ----
gpu_stats = agg.get("gpus") or []
if gpu_stats:
for g in gpu_stats:
- name = f"{g['name']}（GPU{g['index']}）" if g.get("name") else f"GPU{g['index']}"
+ name = g.get("name") or f"GPU{g['index']}"
util = g.get("util")
- util_txt = f"利用率 平均 {util['avg']:.0f}% · 峰值 {util['peak']:.0f}% · 最低 {util['min']:.0f}%" if util else "利用率 无数据"
- mem_peak = g.get("memPeakUsed")
- mem_total = g.get("memTotal")
+ util_txt = (f"利用率 均 {util['avg']:.0f}% · 峰 {util['peak']:.0f}% · 低 {util['min']:.0f}%"
+ if util else "利用率 无数据")
+ mem_peak, mem_total = g.get("memPeakUsed"), g.get("memTotal")
if mem_peak is not None and mem_total:
- mem_pct = mem_peak / mem_total * 100
- mem_txt = f"显存峰值 {_gb(mem_peak * (1 << 30))}/{_gb(mem_total * (1 << 30))}（{_mark(mem_pct, threshold)}）"
+ mem_txt = (f"显存峰 {_gb(mem_peak * (1 << 30))}/{_gb(mem_total * (1 << 30))}"
+ f"（{_mark(mem_peak / mem_total * 100, threshold)}）")
else:
mem_txt = "显存 无数据"
temp_peak = g.get("tempPeak")
- if temp_peak is not None:
- temp_txt = f"温度峰值 {temp_peak:.0f}°C" + ("🔴 过热" if temp_peak >= GPU_TEMP_WARN else " 正常 ✅")
- else:
- temp_txt = "温度 无数据"
- lines.append(f"**GPU** {name}｜{util_txt}｜{mem_txt}｜{temp_txt}")
+ temp_txt = (f"温度峰 {temp_peak:.0f}°C" + (" 🔴过热" if temp_peak >= GPU_TEMP_WARN else " ✅")
+ if temp_peak is not None else "温度 无数据")
+ lines.append(f"**GPU** {name} {util_txt} ｜ {mem_txt} ｜ {temp_txt}")
else:
gpus_now = _check_gpus()
if isinstance(gpus_now, list) and gpus_now:
for g in gpus_now:
- lines.append(f"**GPU** {g['name']}｜利用率 {g['utilPct']:.0f}%｜显存 {g['memUsed']}/{g['memTotal']}"
- f"（无采样数据）")
+ lines.append(f"**GPU** {g['name']} 利用率 {g['utilPct']:.0f}% · 显存 {g['memUsed']}/{g['memTotal']}（无采样）")
else:
lines.append(f"**GPU** {gpus_now if isinstance(gpus_now, str) else '未检测到'}")
- # ---- Docker / 失败任务 / 探活 ----
+ # ---- Docker / 探活 / 失败任务 ----
docker = checks.get("docker") or {}
issues = docker.get("issues") or []
+ svc = checks.get("services") or {}
if docker.get("running") is None:
- lines.append(f"**Docker** {issues[0]['issue'] if issues else '不可用'}")
+ docker_txt = issues[0]["issue"] if issues else "不可用"
else:
- docker_line = f"**Docker** 运行 {docker['running']}/{docker.get('total', '?')}"
- if docker.get("stopped"):
- docker_line += f" · 已停用 {docker['stopped']}"
- docker_line += f" · 异常 **{len(issues)}**🔴" if issues else " · 异常 0"
- lines.append(docker_line)
+ docker_txt = f"运行 {docker['running']}/{docker.get('total', '?')} · 异常 {len(issues)}" + ("🔴" if issues else "")
+ if svc:
+ docker_txt += f" ｜ OCR {'✅' if svc.get('ocr') else '🔴'} ｜ PG {'✅' if svc.get('pg') else '🔴'}"
+ lines.append(f"**Docker** {docker_txt}")
for it in issues[:8]:
lines.append(f"> 🔴 {it['name']}：{it['issue']}")
failures = checks.get("prefectFailures") or []
- lines.append(f"**失败任务(24h)** {len(failures)} 次" + (" ✅" if not failures else "🔴"))
- for f in failures[:5]:
- lines.append(f"> 🔴 {f['flow']} ·「{f['name']}」{f['state']} · {f['start']}")
- if len(failures) > 5:
- lines.append(f"> …共 {len(failures)} 次")
-
- svc = checks.get("services") or {}
- if svc:
- lines.append(f"**探活** OCR {'✅' if svc.get('ocr') else '🔴'}｜PG {'✅' if svc.get('pg') else '🔴'}"
- f"｜负载 {os.getloadavg()[0]:.1f}")
+ if failures:
+ lines.append(f"**失败任务(24h)** {len(failures)} 次 🔴")
+ for f in failures[:5]:
+ lines.append(f"> {f['flow']}「{f['name']}」{f['state']} · {(f.get('start') or '')[5:]}")
+ if len(failures) > 5:
+ lines.append(f"> …共 {len(failures)} 次")
+ else:
+ lines.append("**失败任务(24h)** 0 ✅")
mem_peak_for_alarm = mem.get("peak") if mem else None
swap_peak_for_alarm = swap.get("peak") if swap else None
@@ -452,7 +442,8 @@ def device_check(alarm_threshold: int = 80) -> dict:
or any((g.get("tempPeak") or 0) >= GPU_TEMP_WARN for g in gpu_stats)
or bool(issues) or bool(failures) or not svc.get("ocr") or not svc.get("pg")
)
- lines += ["", ("🔴 **有项目需要处理，见上**" if alarms else "✅ **一切正常**")]
+ lines.append('🔴 有项目需要处理，见上' if alarms
+ else '✅ 一切正常')
content = "\n".join(lines)
sent = send_markdown(content)
diff --git a/flows/shared/wecom.py b/flows/shared/wecom.py
index c77c131..5cde34a 100644
--- a/flows/shared/wecom.py
+++ b/flows/shared/wecom.py
@@ -1,58 +1,44 @@
-"""企业微信智能机器人通知层(wecom-aibot-sdk,WebSocket 按需连接)。
+"""企微群机器人通知层(消息推送 webhook,HTTP POST,无状态)。
-连接模式:每次发消息 建立→等认证→发送→断开,无状态,契合 flow run 级部署模型。
-凭据存 Prefect Secret Block `wecom-aibot-credentials`,JSON:
- {"bot_id": "...", "secret": "...", "chat_id": "..."}
-chat_id 为目标群会话 ID(用 scripts/wecom_capture_chatid.py 一次性捕获)。
+文档:https://developer.work.weixin.qq.com/document/path/99110
+凭据存 Prefect Secret Block `wecom-webhook`(值为完整 webhook URL,含 key);
+本地调试可用环境变量 WECOM_WEBHOOK_URL 覆盖。
+markdown 支持:1-6 级标题/加粗/链接/引用/(不支持表格);
+单条 ≤4096 字节,超长自动按行分段;频率限制 20 条/分钟。
所有函数容错——发送失败只 print 警告并返回 False,绝不中断 flow(对齐 artifacts.py 风格)。
"""
from __future__ import annotations
-import asyncio
-import json
+import os
-from prefect.blocks.system import Secret
+import httpx
-SECRET_BLOCK = "wecom-aibot-credentials"
-MAX_CONTENT_BYTES = 4000 # WeCom markdown 单条约 4096 字节上限,留余量
+WEBHOOK_BLOCK = "wecom-webhook"
+MAX_MARKDOWN_BYTES = 4000 # 官方上限 4096,留余量
+MAX_TEXT_BYTES = 2000 # text 类型上限 2048,留余量
-def load_credentials() -> dict[str, str]:
- value = Secret.load(SECRET_BLOCK).get()
- if isinstance(value, str):
- value = json.loads(value)
- missing = [k for k in ("bot_id", "secret", "chat_id") if not value.get(k)]
- if missing:
- raise ValueError(f"Secret Block {SECRET_BLOCK} 缺少字段: {missing}")
- return value
+def load_webhook_url() -> str:
+ url = os.environ.get("WECOM_WEBHOOK_URL", "").strip()
+ if not url:
+ from prefect.blocks.system import Secret
+
+ url = (Secret.load(WEBHOOK_BLOCK).get() or "").strip()
+ if not url:
+ raise ValueError(f"webhook 地址为空(Secret Block {WEBHOOK_BLOCK} 或 WECOM_WEBHOOK_URL)")
+ return url
-async def _send_bodies(creds: dict[str, str], bodies: list[dict]) -> None:
- from wecom_aibot_sdk import WSClient
-
- ws = WSClient(bot_id=creds["bot_id"], secret=creds["secret"])
- auth_ok = asyncio.Event()
- ws.on("authenticated", lambda: auth_ok.set())
- await asyncio.wait_for(ws.connect(), timeout=15)
- try:
- # connect() 只等到认证帧发出;认证结果由 authenticated 事件确认
- await asyncio.wait_for(auth_ok.wait(), timeout=10)
- for body in bodies:
- await ws.send_message(creds["chat_id"], body)
- finally:
- await ws.disconnect()
-
-
-def _split_markdown(content: str) -> list[str]:
- """按字节预算切段,优先在换行处切,避免拆开表格行。"""
- if len(content.encode("utf-8")) <= MAX_CONTENT_BYTES:
+def _split_bytes(content: str, limit: int) -> list[str]:
+ """按字节预算切段,优先在换行处切。"""
+ if len(content.encode("utf-8")) <= limit:
return [content]
parts: list[str] = []
cur = ""
for line in content.splitlines(keepends=True):
- if cur and len((cur + line).encode("utf-8")) > MAX_CONTENT_BYTES:
+ if cur and len((cur + line).encode("utf-8")) > limit:
parts.append(cur)
cur = line
else:
@@ -62,26 +48,36 @@ def _split_markdown(content: str) -> list[str]:
return parts
-def send_markdown(content: str) -> bool:
- """发送 markdown 消息到目标群(超长自动分段)。失败返回 False,不抛异常。"""
+def _post(url: str, payload: dict) -> bool:
+ resp = httpx.post(url, json=payload, timeout=15)
try:
- creds = load_credentials()
- parts = _split_markdown(content)
- asyncio.run(_send_bodies(creds, [{"msgtype": "markdown", "markdown": {"content": p}} for p in parts]))
- if len(parts) > 1:
- print(f"[wecom] markdown 已发送 ({len(parts)} 段)")
- return True
- except Exception as err: # noqa: BLE001
- print(f"[wecom] markdown 发送失败: {err}")
+ data = resp.json()
+ except ValueError:
+ data = {}
+ if resp.status_code != 200 or data.get("errcode") != 0:
+ print(f"[wecom] 发送失败: HTTP {resp.status_code} {data or resp.text[:100]}")
return False
+ return True
+
+
+def _send(content: str, msgtype: str, limit: int) -> bool:
+ try:
+ url = load_webhook_url()
+ parts = _split_bytes(content, limit)
+ ok = all(_post(url, {"msgtype": msgtype, msgtype: {"content": part}}) for part in parts)
+ if ok and len(parts) > 1:
+ print(f"[wecom] {msgtype} 已发送 ({len(parts)} 段)")
+ return ok
+ except Exception as err: # noqa: BLE001
+ print(f"[wecom] {msgtype} 发送失败: {err}")
+ return False
+
+
+def send_markdown(content: str) -> bool:
+ """发送 markdown 到群(超长自动分段)。失败返回 False,不抛异常。"""
+ return _send(content, "markdown", MAX_MARKDOWN_BYTES)
def send_text(content: str) -> bool:
- """发送纯文本消息到目标群。失败返回 False,不抛异常。"""
- try:
- creds = load_credentials()
- asyncio.run(_send_bodies(creds, [{"msgtype": "text", "text": {"content": content}}]))
- return True
- except Exception as err: # noqa: BLE001
- print(f"[wecom] text 发送失败: {err}")
- return False
+ """发送纯文本到群。失败返回 False,不抛异常。"""
+ return _send(content, "text", MAX_TEXT_BYTES)
diff --git a/pyproject.toml b/pyproject.toml
index 12c7a7c..ff3d8d1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -10,7 +10,6 @@ dependencies = [
"httpx",
"pandas",
"pydantic>=2.0",
- "wecom-aibot-sdk>=1.0.8",
"psutil>=7.2.2",
]
diff --git a/scripts/api_db_diff.py b/scripts/api_db_diff.py
new file mode 100644
index 0000000..4f7de9d
--- /dev/null
+++ b/scripts/api_db_diff.py
@@ -0,0 +1,167 @@
+"""一次性对账:RIIN/SDS 远程订单量 vs PG Orders,按月 + 按状态(未完结)对比。
+
+口径:
+- RIIN:远程 count API(time_type=3 商家推送时间) vs 本地 orderedAt(=pushTime)。
+- SDS:远程 statistics(payDate 窗口) vs 本地 orderedAt(=gmtPay),排除"售后"。
+运行:uv run python scripts/api_db_diff.py
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import datetime, timedelta
+
+from dotenv import load_dotenv
+
+load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env"))
+
+import psycopg2 # noqa: E402
+
+from api.riin.models import OrderPageRequest as RiinReq # noqa: E402
+from api.sds.models import OrderQueryParams # noqa: E402
+from flows.shared import ( # noqa: E402
+ create_riin_client, create_sds_client, format_date, start_of_day,
+)
+
+RIIN_FINISHED = {"已发货", "已关闭", "已退款"}
+SDS_FINISHED = {"已完成", "已取消"}
+
+# count API 返回字段(camelCase 原始 key)→ 状态标签
+RIIN_COUNT_MAP = {
+ "assignedSuceessed": "已分配",
+ "factoryAuditing": "工厂审核中",
+ "inProducing": "生产中",
+ "shipped": "已发货",
+ "closed": "已关闭",
+ "refunding": "退款中",
+ "refunded": "已退款",
+ "rejecting": "拒审中",
+ "rejectedToShop": "已退回店铺",
+}
+
+
+def month_windows() -> list[tuple[datetime, datetime, str]]:
+ """2025-01 起到当月的连续月窗口。"""
+ out = []
+ cur = datetime(2025, 1, 1)
+ now = datetime.now()
+ while cur < now:
+ nxt = (cur.replace(day=28) + timedelta(days=4)).replace(day=1)
+ out.append((cur, min(nxt, now), cur.strftime("%Y-%m")))
+ cur = nxt
+ return out
+
+
+def riin_count(client, start: datetime, end: datetime, retries: int = 4) -> dict[str, int]:
+ """直调 count 接口并解析原始 JSON(严格模型在月窗口下会因 dayRangeGroup=null 校验失败)。
+
+ 大月份窗口服务端聚合慢,超时按 5/10/20/40s 退避重试。
+ """
+ import time
+
+ req = RiinReq(time_type=3, start_time=format_date(start), end_time=format_date(end))
+ body = client._build_order_page_body(req)
+ body.pop("pageIndex", None)
+ body.pop("pageSize", None)
+ for attempt in range(retries):
+ try:
+ data = client.post("distribution/order/list/count", body=body)
+ break
+ except Exception as err: # noqa: BLE001
+ if attempt == retries - 1 or "timed out" not in str(err).lower() and "timeout" not in str(err).lower():
+ raise
+ wait = 5 * (2 ** attempt)
+ print(f"[RIIN count {start:%Y-%m}] 超时,{wait}s 后重试 ({attempt + 1}/{retries})")
+ time.sleep(wait)
+ raw = data.get("data") or {}
+ return {RIIN_COUNT_MAP[k]: v for k, v in raw.items()
+ if k in RIIN_COUNT_MAP and v}
+
+
+def sds_status(client, start: datetime | None, end: datetime | None) -> dict[str, int]:
+ params = OrderQueryParams(
+ pay_date_begin=format_date(start) if start else None,
+ pay_date_end=format_date(end) if end else None,
+ )
+ stats = client.get_status(params)
+ out: dict[str, int] = {}
+ for s in stats:
+ name = getattr(s, "name", None)
+ if not name or name == "售后":
+ continue
+ out[name] = out.get(name, 0) + (getattr(s, "orderNum", 0) or 0)
+ return out
+
+
+def main() -> None:
+ months = month_windows()
+ conn = psycopg2.connect(os.environ["DB_DSN"])
+ cur = conn.cursor()
+
+ # ---- DB 按月(orderedAt) ----
+ cur.execute("""SELECT date_trunc('month',"orderedAt"), 'RIIN', COUNT(DISTINCT "orderNo")
+ FROM "Orders" WHERE "systemName"='RIIN' AND "orderedAt" IS NOT NULL GROUP BY 1""")
+ db_riin_month = {r[0].strftime("%Y-%m"): r[2] for r in cur.fetchall()}
+ cur.execute("""SELECT date_trunc('month',"orderedAt"), COUNT(DISTINCT "orderNo")
+ FROM "Orders" WHERE "systemName"='SDS' AND "orderedAt" IS NOT NULL GROUP BY 1""")
+ db_sds_month = {r[0].strftime("%Y-%m"): r[1] for r in cur.fetchall()}
+
+ # ---- DB 分状态(全时) ----
+ cur.execute("""SELECT "systemStatus", COUNT(DISTINCT "orderNo") FROM "Orders"
+ WHERE "systemName"='RIIN' GROUP BY 1""")
+ db_riin_status = {r[0] or "(空)": r[1] for r in cur.fetchall()}
+ cur.execute("""SELECT "systemStatus", COUNT(DISTINCT "orderNo") FROM "Orders"
+ WHERE "systemName"='SDS' GROUP BY 1""")
+ db_sds_status = {r[0] or "(空)": r[1] for r in cur.fetchall()}
+
+ riin = create_riin_client()
+ sds = create_sds_client()
+ try:
+ # ---- RIIN 远程按月(含状态明细) ----
+ remote_riin_month: dict[str, int] = {}
+ riin_status_sum: dict[str, int] = {}
+ for start, end, label in months:
+ m = riin_count(riin, start, end)
+ remote_riin_month[label] = sum(m.values())
+ for k, v in m.items():
+ riin_status_sum[k] = riin_status_sum.get(k, 0) + v
+ # ---- SDS 远程按月(含状态明细) ----
+ remote_sds_month: dict[str, int] = {}
+ sds_status_sum: dict[str, int] = {}
+ for start, end, label in months:
+ m = sds_status(sds, start, end)
+ remote_sds_month[label] = sum(m.values())
+ for k, v in m.items():
+ sds_status_sum[k] = sds_status_sum.get(k, 0) + v
+
+ # 全时状态计数 = 按月求和(RIIN 全时窗口 count 接口会超时)
+ remote_riin_all = riin_status_sum
+ remote_sds_all = sds_status_sum
+ finally:
+ riin.close()
+ sds.close()
+
+ print("=" * 72)
+ print("按月对比(RIIN 远程=推送时间窗口计数 / 库=orderedAt;SDS 远程=付款时间 / 库=orderedAt)")
+ print(f"{'月份':10}{'RIIN远程':>12}{'RIIN库':>12}{'差':>10} {'SDS远程':>12}{'SDS库':>12}{'差':>10}")
+ for _, _, label in months:
+ rr, dr = remote_riin_month.get(label, 0), db_riin_month.get(label, 0)
+ rs, ds = remote_sds_month.get(label, 0), db_sds_month.get(label, 0)
+ mark = " ←" if abs(rr - dr) > 0 or abs(rs - ds) > 0 else ""
+ print(f"{label:10}{rr:>12,}{dr:>12,}{dr - rr:>10,} {rs:>12,}{ds:>12,}{ds - rs:>10,}{mark}")
+
+ print("=" * 72)
+ print("全时状态对比(远程 2025-01-01 ~ 明天 vs 库全量)")
+ for title, remote, local in (("RIIN", remote_riin_all, db_riin_status),
+ ("SDS", remote_sds_all, db_sds_status)):
+ print(f"\n[{title}] 远程合计 {sum(remote.values()):,} / 库合计 {sum(local.values()):,}")
+ for st in sorted(set(remote) | set(local)):
+ r, l = remote.get(st, 0), local.get(st, 0)
+ mark = " ← 差异" if r != l else ""
+ print(f" {st:10} 远程 {r:>10,} 库 {l:>10,}{mark}")
+
+ conn.close()
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/wecom_capture_chatid.py b/scripts/wecom_capture_chatid.py
deleted file mode 100644
index 65a508d..0000000
--- a/scripts/wecom_capture_chatid.py
+++ /dev/null
@@ -1,116 +0,0 @@
-"""一次性工具:捕获目标群的 chat_id 并写回 Secret Block / 发测试消息。
-
-用法:
- # 1. 初始化 Secret Block(凭据可从环境变量 WECOM_BOT_ID / WECOM_SECRET 读取)
- uv run python scripts/wecom_capture_chatid.py --init
-
- # 2. 监听模式:运行后在目标群里 @机器人 说任意一句话,即可捕获 chat_id
- uv run python scripts/wecom_capture_chatid.py --listen
-
- # 3. 捕获到 chat_id 后写回 Secret Block(替换 )
- uv run python scripts/wecom_capture_chatid.py --save-chatid
-
- # 4. 验证收发链路
- uv run python scripts/wecom_capture_chatid.py --test "hello"
-"""
-
-from __future__ import annotations
-
-import argparse
-import asyncio
-import json
-import os
-import sys
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-from prefect.blocks.system import Secret # noqa: E402
-
-SECRET_BLOCK = "wecom-aibot-credentials"
-
-
-def _read_block() -> dict[str, str]:
- value = Secret.load(SECRET_BLOCK).get()
- if isinstance(value, str):
- value = json.loads(value)
- return value
-
-
-def _write_block(creds: dict[str, str]) -> None:
- Secret(value=json.dumps(creds, ensure_ascii=False)).save(SECRET_BLOCK, overwrite=True)
- shown = {k: (v[:8] + "..." if k == "secret" and v else v) for k, v in creds.items()}
- print(f"已保存 Secret Block {SECRET_BLOCK}: {shown}")
-
-
-def do_init() -> None:
- creds = {
- "bot_id": os.environ.get("WECOM_BOT_ID", ""),
- "secret": os.environ.get("WECOM_SECRET", ""),
- "chat_id": "",
- }
- missing = [k for k in ("bot_id", "secret") if not creds[k]]
- if missing:
- sys.exit(f"缺少环境变量: {missing},请先 export WECOM_BOT_ID=... WECOM_SECRET=...")
- _write_block(creds)
-
-
-def do_save_chatid(chatid: str) -> None:
- creds = _read_block()
- creds["chat_id"] = chatid
- _write_block(creds)
-
-
-def do_test(message: str) -> None:
- from flows.shared.wecom import send_markdown
-
- ok = send_markdown(f"✅ **机器人通知链路正常**\n\n> {message}")
- sys.exit(0 if ok else 1)
-
-
-async def _listen(creds: dict[str, str]) -> None:
- from wecom_aibot_sdk import WSClient
-
- ws = WSClient(bot_id=creds["bot_id"], secret=creds["secret"])
-
- def on_any(frame) -> None:
- body = frame.get("body") or {}
- chatid = body.get("chatid") or ""
- sender = (body.get("from") or {}).get("userid", "")
- msgtype = body.get("msgtype", "")
- text = ((body.get("text") or {}).get("content") or "")[:50]
- print(f"收到帧: msgtype={msgtype} chatid={chatid} from={sender} text={text}")
- if chatid:
- print(f"\n*** chat_id = {chatid} ***\n")
- print(f"写回命令: uv run python scripts/wecom_capture_chatid.py --save-chatid {chatid}")
-
- ws.on("message", on_any)
- ws.on("event.enter_chat", on_any)
- ws.on("authenticated", lambda: print("已认证,等待消息(请在目标群 @机器人 发一句话)...Ctrl+C 退出"))
- await ws.connect()
- await asyncio.Event().wait()
-
-
-def main() -> None:
- parser = argparse.ArgumentParser(description=__doc__)
- parser.add_argument("--init", action="store_true", help="初始化 Secret Block(需 WECOM_BOT_ID/WECOM_SECRET 环境变量)")
- parser.add_argument("--listen", action="store_true", help="监听消息以捕获 chat_id")
- parser.add_argument("--save-chatid", metavar="CHATID", help="把捕获的 chat_id 写回 Secret Block")
- parser.add_argument("--test", metavar="MSG", help="向已保存的 chat_id 发送测试消息")
- args = parser.parse_args()
-
- if args.init:
- do_init()
- elif args.save_chatid:
- do_save_chatid(args.save_chatid)
- elif args.test:
- do_test(args.test)
- elif args.listen:
- # Secret.load 是 sync_compatible,须在进入事件循环前同步调用
- creds = _read_block()
- asyncio.run(_listen(creds))
- else:
- parser.print_help()
-
-
-if __name__ == "__main__":
- main()
diff --git a/uv.lock b/uv.lock
index 5b6638b..f79e013 100644
--- a/uv.lock
+++ b/uv.lock
@@ -745,7 +745,6 @@ dependencies = [
{ name = "psutil" },
{ name = "psycopg2-binary" },
{ name = "pydantic" },
- { name = "wecom-aibot-sdk" },
]
[package.metadata]
@@ -757,7 +756,6 @@ requires-dist = [
{ name = "psutil", specifier = ">=7.2.2" },
{ name = "psycopg2-binary" },
{ name = "pydantic", specifier = ">=2.0" },
- { name = "wecom-aibot-sdk", specifier = ">=1.0.8" },
]
[[package]]
@@ -2241,20 +2239,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/be/4d/2d0d67834092e354d2b0498f014a41249a89556bc406cf86f3e1557bb463/websockets-16.1.1-py3-none-any.whl", hash = "sha256:6abbd3e82c731c8e531714466acd5d87b5e88ac3243465337ba71d68e23ae7e3", size = 173814, upload-time = "2026-07-17T22:51:04.184Z" },
]
-[[package]]
-name = "wecom-aibot-sdk"
-version = "1.0.8"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
- { name = "cryptography" },
- { name = "httpx" },
- { name = "websockets" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/45/72/e14f62362ecee2bae1fb13e7f48e000a39514c883f2b49568294624e0241/wecom_aibot_sdk-1.0.8.tar.gz", hash = "sha256:dd8ac2917e0b23a190a718b2c13f717acd242af487e2bf98a5ca40a117c133e6", size = 64863, upload-time = "2026-06-02T09:43:50.569Z" }
-wheels = [
- { url = "https://files.pythonhosted.org/packages/a8/82/8e342b64b29c52d9e92fa6f3596d53df5d2dd3dab871493ce0f871b01914/wecom_aibot_sdk-1.0.8-py3-none-any.whl", hash = "sha256:9577742c1b9d76232ece675342871e26c3f5a9d9617c7f860f951e2c6265c184", size = 28069, upload-time = "2026-06-02T09:43:49.316Z" },
-]
-
[[package]]
name = "whenever"
version = "0.10.3"