diff --git a/docker-compose.yml b/docker-compose.yml index b33760e..79bef4d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -20,6 +20,18 @@ services: TZ: Asia/Shanghai depends_on: - prefect-server + volumes: + # 设备监控(device-check flow):宿主机磁盘使用率 + Docker 容器状态 + - /var/run/docker.sock:/var/run/docker.sock + - /:/host:ro + deploy: + resources: + reservations: + devices: + # GPU 监控(nvidia-smi);依赖宿主机 nvidia-container-toolkit + - driver: nvidia + count: all + capabilities: [gpu] restart: unless-stopped volumes: diff --git a/flows/monitor/__init__.py b/flows/monitor/__init__.py new file mode 100644 index 0000000..57f5173 --- /dev/null +++ b/flows/monitor/__init__.py @@ -0,0 +1 @@ +"""监控上报层:企微机器人异常报警 / 数据核验日报 / 设备健康日报。""" diff --git a/flows/monitor/alert_failure.py b/flows/monitor/alert_failure.py new file mode 100644 index 0000000..ca5b6ca --- /dev/null +++ b/flows/monitor/alert_failure.py @@ -0,0 +1,129 @@ +"""企微机器人异常报警:任意 flow run 失败/崩溃时由 Prefect 事件触发器自动调用。 + +对应 wecom-alert deployment(prefect.yaml 里的 triggers 声明): +- match: prefect.resource.id = prefect.flow-run.* +- expect: prefect.flow-run.Failed / Crashed +- parameters.event = "{{ event }}" + +flow 体全容错——本 flow 自身永不进入 Failed 状态,避免自我触发循环。 +event 可能以 dict 或 str(jinja 渲染结果)传入,均做防御性解析; +解析出 flow run id 后再查 Prefect API 补全权威信息(名称/错误消息/时间)。 +""" + +from __future__ import annotations + +import json +import os +import re +from datetime import datetime +from typing import Any + +import httpx +from prefect import flow + +from flows.shared import send_markdown + +UI_BASE = os.environ.get("PREFECT_UI_URL", "http://192.168.124.92:4200") + + +def _parse_event(event: Any) -> dict: + if isinstance(event, dict): + return event + if isinstance(event, str) and event.strip(): + # jinja "{{ event }}" 可能渲染成 JSON 串或 Python repr,统一抠除关键字段 + rid = re.search(r"prefect\.flow-run\.[0-9a-fA-F-]{36}", event) + ev = re.search(r"prefect\.flow-run\.(?:Failed|Crashed)", event) + name = re.search(r"prefect\.resource\.name['\"]?\s*[:=]\s*['\"]([^'\"]+)", event) + occurred = re.search(r"occurred['\"]?\s*[:=]\s*['\"]([^'\"]+)", event) + return { + "event": ev.group(0) if ev else "", + "resource": { + "prefect.resource.id": rid.group(0) if rid else "", + "prefect.resource.name": name.group(1) if name else "", + }, + "occurred": occurred.group(1) if occurred else "", + } + return {} + + +def _flow_run_id(event: dict) -> str | None: + rid = (event.get("resource") or {}).get("prefect.resource.id", "") + m = re.match(r"^prefect\.flow-run\.(.+)$", rid) + return m.group(1) if m else None + + +def _fetch_run_detail(run_id: str) -> dict: + """查 Prefect API 补全 flow run 权威信息,失败返回空 dict。""" + api_url = os.environ.get("PREFECT_API_URL", "http://localhost:4200/api") + try: + resp = httpx.get(f"{api_url}/flow_runs/{run_id}", timeout=10) + resp.raise_for_status() + return resp.json() + except Exception as err: # noqa: BLE001 + print(f"[wecom-alert] 查询 flow run {run_id} 失败: {err}") + return {} + + +def _fmt_ts(value: Any) -> str: + if not value: + return "—" + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone().strftime("%m-%d %H:%M") + except (ValueError, TypeError): + return str(value)[:19] + + +@flow(name="wecom-alert", log_prints=True) +def wecom_alert_failure(event: dict | str | None = None) -> dict: + """Prefect 事件触发:flow run 失败/崩溃 → 发企微群报警。永不抛异常。""" + result = {"sent": False, "flowName": None, "runId": None} + try: + ev = _parse_event(event) + event_name = ev.get("event", "prefect.flow-run.?") + resource_name = (ev.get("resource") or {}).get("prefect.resource.name", "") + run_id = _flow_run_id(ev) + + flow_name, run_name, state_msg, occurred = "", resource_name, "", ev.get("occurred", "") + if run_id: + detail = _fetch_run_detail(run_id) + flow_name = detail.get("name", "") or resource_name + run_name = detail.get("name", "") + state = detail.get("state") or {} + state_msg = (state.get("message") or "").strip() + ended = detail.get("end_time") or detail.get("expected_start_time") + occurred = occurred or ended + + lines = [ + "🔴 **Prefect 流程失败报警**", + f"- 事件: `{event_name}`", + f"- Flow: {flow_name or resource_name or '未知'}", + ] + if run_name and run_name != flow_name: + lines.append(f"- Run: {run_name}") + if state_msg: + lines.append(f"- 错误: {state_msg[:200].replace(chr(10), ' ')}") + lines.append(f"- 时间: {_fmt_ts(occurred)}") + if run_id: + lines.append(f"- 详情: [Prefect UI]({UI_BASE}/runs/flow-run/{run_id})") + else: + raw = json.dumps(ev, ensure_ascii=False)[:300] if ev else "(事件内容解析失败)" + lines.append(f"- 原始事件: {raw}") + + result["sent"] = send_markdown("\n".join(lines)) + result.update({"flowName": flow_name or None, "runId": run_id}) + print(f"[wecom-alert] 已发送失败报警: {flow_name or resource_name} ({event_name})") + except Exception as err: # noqa: BLE001 + print(f"[wecom-alert] 报警流程异常(已吞掉,避免自我触发): {err}") + return result + + +if __name__ == "__main__": + # 本地手测:模拟一条 Failed 事件 + wecom_alert_failure({ + "event": "prefect.flow-run.Failed", + "resource": { + "prefect.resource.id": "prefect.flow-run.00000000-0000-0000-0000-000000000000", + "prefect.resource.name": "riin-order-sync/manual", + }, + "occurred": datetime.now().astimezone().isoformat(), + }) diff --git a/flows/monitor/daily_report.py b/flows/monitor/daily_report.py new file mode 100644 index 0000000..c4a7481 --- /dev/null +++ b/flows/monitor/daily_report.py @@ -0,0 +1,106 @@ +"""数据核验日报(每日 18:30):近 N 天 RIIN/SDS 订单+商品数据准确性核验并上报企微群。 + +对应 daily-report deployment,cron 30 18 * * *。内容: +1. 复用 riin-verify / sds-verify 逻辑(subflow 调用):远程 vs 本地逐天逐状态计数比对, + 有差异自动拉取修复——与早 6/7 点的每日核验互补,二次确认白天同步质量。 +2. 商品数据完整性:窗口内零商品订单 / 商品无名称 / 全库孤儿商品(本地 SQL)。 +3. OCR 处理状态与当日失败 flow runs 回顾(事件报警的兜底汇总)。 +""" + +from __future__ import annotations + +from datetime import datetime, timedelta + +from prefect import flow + +from flows.monitor.device_check import check_prefect_failures +from flows.riin.verify import riin_verify +from flows.sds.verify import sds_verify +from flows.shared import SyncDB, emit_sync_summary, send_markdown, start_of_day + + +def _verify_section(name: str, report: dict) -> tuple[str, list[str], bool]: + """把 verify flow 返回的 verifyReport 压缩成结论行 + 差异日明细行。""" + days = (report.get("verifyReport") or {}).get("days") or [] + summary = (report.get("verifyReport") or {}).get("summary") or {} + diff_days = [d for d in days if not d.get("ok")] + total, n_diff = summary.get("totalDays", len(days)), summary.get("daysWithDiff", len(diff_days)) + head = f"**{name}**: {total} 天中 {total - n_diff} 天一致" + (" ✅" if not diff_days else f", {n_diff} 天有差异已修复") + detail = [ + f"- {d['date']}: 本地 {d.get('local', '—')} / 远程 {d.get('remote', '—')}," + f"差异 {d.get('diff') or '—'},已补拉 订单 {d.get('orders', 0)} / 商品 {d.get('goods', 0)}" + for d in diff_days + ] + totals = {k: v for k, v in summary.items() if k not in ("totalDays", "daysWithDiff")} + if any(totals.values()): + detail.append(f"- 修复量合计: {totals}") + return head, detail, bool(diff_days) + + +@flow(name="daily-report", log_prints=True) +def daily_report(days: int = 7) -> dict: + """近 N 天数据核验 + 商品完整性 + OCR/失败回顾,发企微群日报。""" + today = start_of_day(datetime.now()) + window_start = today - timedelta(days=days) + window_end = today # 与 verify 的 recent 约定一致:近 N 天、不含今天 + + riin = riin_verify(mode="recent", recent_days=days) + sds = sds_verify(mode="recent", recent_days=days) + + db = SyncDB() + goods = {sys: db.get_goods_integrity(sys, window_start, window_end) for sys in ("RIIN", "SDS")} + ocr = db.count_ocr_status() + + failures = check_prefect_failures(hours=24) + + lines = [f"📊 **数据核验日报** {datetime.now().strftime('%Y-%m-%d %H:%M')}" + f"(近 {days} 天,{window_start:%m-%d} ~ {window_end:%m-%d})", ""] + + attention = False + for name, report in (("RIIN", riin), ("SDS", sds)): + try: + head, detail, had_diff = _verify_section(name, report) + lines.append(head) + lines.extend(detail) + attention = attention or had_diff + except Exception as err: # noqa: BLE001 + lines.append(f"**{name}**: 核验结果解析失败: {err}") + attention = True + + lines.append("**商品完整性**:") + lines.append("| 系统 | 订单 | 商品行 | 零商品订单 | 商品无名称 | 孤儿商品(全库) |") + lines.append("| --- | --- | --- | --- | --- | --- |") + for sys in ("RIIN", "SDS"): + g = goods[sys] + flag = " ⚠️" if g["zeroGoodsOrders"] > 0 else "" + lines.append(f"| {sys} | {g['orders']} | {g['goods']} | {g['zeroGoodsOrders']}{flag} " + f"| {g['goodsNoName']} | {g['orphanGoods']} |") + if g["zeroGoodsOrders"] > 0: + attention = True + + lines.append(f"**OCR(SDS)**: 完成 {ocr.get('done', 0)} / 失败 {ocr.get('error', 0)} / " + f"不支持 {ocr.get('unsupported', 0)} / 待处理 {ocr.get('pending', 0)}") + lines.append(f"**近24h 失败 flow runs**: {len(failures)} 个" + ("" if failures else " ✅")) + for f in failures[:8]: + lines.append(f"- 🔴 {f['name']}({f['state']}) {f['start']} {f['message'][:60]}") + if len(failures) > 8: + lines.append(f"- ...共 {len(failures)} 个,详见 Prefect UI") + if failures: + attention = True + + lines += ["", ("⚠️ **存在需关注项,见上**" if attention else "✅ **数据全部一致,无异常**")] + + sent = send_markdown("\n".join(lines)) + emit_sync_summary("数据核验日报", [ + ("核验窗口", f"近{days}天"), ("需关注", "是" if attention else "否"), + ("RIIN 零商品订单", goods["RIIN"]["zeroGoodsOrders"]), + ("SDS 零商品订单", goods["SDS"]["zeroGoodsOrders"]), + ("OCR 待处理", ocr.get("pending", 0)), ("失败 runs(24h)", len(failures)), + ("已发企微", "是" if sent else "否"), + ]) + return {"attention": attention, "sent": sent, + "goods": goods, "ocr": ocr, "failures24h": len(failures)} + + +if __name__ == "__main__": + daily_report.serve(name="daily-report", cron="30 18 * * *") diff --git a/flows/monitor/device_check.py b/flows/monitor/device_check.py new file mode 100644 index 0000000..8bde042 --- /dev/null +++ b/flows/monitor/device_check.py @@ -0,0 +1,262 @@ +"""设备/系统健康日报(每日一次):内存/磁盘/GPU 使用率 + 系统级错误检查。 + +对应 device-check deployment,每天 09:00(可调)。检查项: +- 内存/swap、宿主机磁盘(经 /:/host:ro 挂载)、GPU(nvidia-smi)——使用率 ≥ 阈值(默认 80%)标 🔴 +- Docker 容器状态(经 /var/run/docker.sock):应运行但退出 / restarting / unhealthy / OOMKilled +- Prefect 近 24h FAILED/CRASHED flow runs、OCR 服务探活、PG 连通性、系统负载 + +依赖 docker-compose 给 prefect-worker 挂载 /:/host:ro、/var/run/docker.sock、GPU; +未挂载时对应检查项自动降级为"不可用",不影响其余检查。 +""" + +from __future__ import annotations + +import os +import subprocess +from datetime import datetime, timedelta + +import httpx +import psutil +from prefect import flow + +from flows.shared import SyncDB, emit_sync_summary, send_markdown + +# (容器内路径, 展示名);/host 为 compose 挂载的宿主机根目录 +DISK_PATHS = [("/host", "/"), ("/host/home", "/home")] +DOCKER_SOCK = "/var/run/docker.sock" + +_OCR_BASE = "http://192.168.124.92:9801" + + +def _gb(nbytes: float) -> str: + return f"{nbytes / (1 << 30):.1f}G" + + +def _check_memory() -> dict: + vm = psutil.virtual_memory() + sw = psutil.swap_memory() + return { + "pct": vm.percent, "used": _gb(vm.used), "total": _gb(vm.total), + "swapPct": sw.percent, "swapUsed": _gb(sw.used), "swapTotal": _gb(sw.total), + } + + +def _check_disks() -> list[dict]: + paths = DISK_PATHS if os.path.isdir(DISK_PATHS[0][0]) else [("/", "/")] + out = [] + for raw, label in paths: + try: + du = psutil.disk_usage(raw) + out.append({"label": label, "pct": du.percent, "used": _gb(du.used), "total": _gb(du.total)}) + except Exception as err: # noqa: BLE001 + out.append({"label": label, "error": str(err)[:80]}) + return out + + +def _check_gpus() -> list[dict] | str: + """nvidia-smi 查 GPU 利用率与显存。容器无 GPU 访问时返回说明文字。""" + try: + proc = subprocess.run( + ["nvidia-smi", "--query-gpu=index,name,utilization.gpu,memory.used,memory.total", + "--format=csv,noheader,nounits"], + capture_output=True, text=True, timeout=15, + ) + if proc.returncode != 0: + return f"nvidia-smi 退出码 {proc.returncode}: {proc.stderr.strip()[:100]}" + gpus = [] + for line in proc.stdout.strip().splitlines(): + parts = [p.strip() for p in line.split(",")] + if len(parts) != 5: + continue + idx, name, util, mem_used, mem_total = parts + mem_pct = float(mem_used) / float(mem_total) * 100 if float(mem_total) > 0 else 0.0 + gpus.append({"name": f"GPU{idx} {name[:24]}", "utilPct": float(util), "memPct": mem_pct, + "memUsed": f"{float(mem_used)/1024:.1f}G", "memTotal": f"{float(mem_total)/1024:.1f}G"}) + return gpus or "未检测到 GPU" + except FileNotFoundError: + return "不可用(容器内无 nvidia-smi,需 compose 配置 GPU)" + except Exception as err: # noqa: BLE001 + return f"检查失败: {str(err)[:80]}" + + +def _inspect_container(client: httpx.Client, cid: str) -> dict: + try: + resp = client.get(f"http://localhost/containers/{cid}/json", timeout=10) + resp.raise_for_status() + return resp.json() + except Exception: # noqa: BLE001 + return {} + + +def _check_docker() -> dict: + """容器状态汇总 + 异常容器列表(应运行但退出/restarting/unhealthy/OOMKilled)。""" + if not os.path.exists(DOCKER_SOCK): + return {"running": None, "issues": [{"name": "Docker", "issue": f"不可用(未挂载 {DOCKER_SOCK})"}]} + with httpx.Client(transport=httpx.HTTPTransport(uds=DOCKER_SOCK), timeout=10) as client: + resp = client.get("http://localhost/containers/json", params={"all": "1"}) + resp.raise_for_status() + containers = resp.json() + running = sum(1 for c in containers if c.get("State") == "running") + issues = [] + stopped = 0 + for c in containers: + name = (c.get("Names") or ["?"])[0].lstrip("/") + state, status = c.get("State", ""), c.get("Status", "") + if state == "restarting": + issues.append({"name": name, "issue": f"反复重启 ({status})"}) + elif state == "running" and "(unhealthy)" in status: + issues.append({"name": name, "issue": f"健康检查失败 ({status})"}) + elif state in ("exited", "dead"): + detail = _inspect_container(client, c["Id"]) + policy = ((detail.get("HostConfig") or {}).get("RestartPolicy") or {}).get("Name", "no") + oom = (detail.get("State") or {}).get("OOMKilled", False) + finished_at = (detail.get("State") or {}).get("FinishedAt", "") + if oom and _exited_recently(finished_at): + issues.append({"name": name, "issue": f"被 OOM 杀死 ({status})"}) + elif policy in ("always", "unless-stopped"): + if _exited_recently(finished_at): + issues.append({"name": name, "issue": f"应常驻但刚退出 ({status}, restart={policy})"}) + else: + stopped += 1 # 数周前人为停用的旧服务,不算异常 + return {"running": running, "total": len(containers), "issues": issues, "stopped": stopped} + + +def _exited_recently(finished_at: str, hours: int = 24) -> bool: + try: + from datetime import timezone + dt = datetime.fromisoformat(finished_at.replace("Z", "+00:00")) + return (datetime.now(timezone.utc) - dt).total_seconds() < hours * 3600 + except (ValueError, TypeError): + return True # 解析失败按新退出处理,宁可多报 + + +def check_prefect_failures(hours: int = 24) -> list[dict]: + api_url = os.environ.get("PREFECT_API_URL", "http://localhost:4200/api") + since = (datetime.now().astimezone() - timedelta(hours=hours)).isoformat() + resp = httpx.post( + f"{api_url}/flow_runs/filter", + json={ + "flow_runs": {"state": {"type": {"any_": ["FAILED", "CRASHED"]}}, "start_time": {"after_": since}}, + "limit": 50, + "sort": "START_TIME_DESC", + }, + timeout=15, + ) + resp.raise_for_status() + out = [] + for r in resp.json(): + start = (r.get("start_time") or "")[:16].replace("T", " ") + out.append({"name": r.get("name", "?"), "state": (r.get("state") or {}).get("type", "?"), + "message": ((r.get("state") or {}).get("message") or "").strip()[:100], "start": start}) + return out + + +def _check_services() -> dict: + from api.ocr import OcrClient + + ocr_ok = False + try: + ocr_ok = OcrClient().health() + except Exception: # noqa: BLE001 + pass + pg_ok = False + try: + pg_ok = bool(SyncDB().ping()) + except Exception: # noqa: BLE001 + pass + return {"ocr": ocr_ok, "pg": pg_ok} + + +def _mark(pct: float | None, threshold: int) -> str: + if pct is None: + return "—" + return f"{pct:.0f}%" + (" 🔴" if pct >= threshold else "") + + +@flow(name="device-check", log_prints=True) +def device_check(alarm_threshold: int = 80) -> dict: + """设备/系统健康日报:检查并直接发企微群。单项检查失败不影响其余项。""" + checks: dict[str, object] = {} + for name, fn in [ + ("memory", _check_memory), ("disks", _check_disks), ("gpus", _check_gpus), + ("docker", _check_docker), ("prefectFailures", check_prefect_failures), ("services", _check_services), + ]: + try: + checks[name] = fn() + except Exception as err: # noqa: BLE001 + checks[name] = None + print(f"[device-check] {name} 检查失败: {err}") + + threshold = alarm_threshold + lines = [f"🖥 **设备健康日报** {datetime.now().strftime('%Y-%m-%d %H:%M')}", ""] + + mem = checks.get("memory") or {} + if mem: + lines.append(f"**内存**: {_mark(mem['pct'], threshold)}({mem['used']}/{mem['total']})" + f" | swap {_mark(mem['swapPct'], threshold)}({mem['swapUsed']}/{mem['swapTotal']})") + else: + lines.append("**内存**: 检查失败") + + lines.append("**磁盘**:") + lines.append("| 挂载点 | 使用率 | 已用/总量 |") + lines.append("| --- | --- | --- |") + for d in checks.get("disks") or []: + if "error" in d: + lines.append(f"| {d['label']} | 检查失败 | {d['error']} |") + else: + lines.append(f"| {d['label']} | {_mark(d['pct'], threshold)} | {d['used']}/{d['total']} |") + + lines.append("**GPU**:") + gpus = checks.get("gpus") + if isinstance(gpus, list) and gpus: + lines.append("| GPU | 利用率 | 显存 |") + lines.append("| --- | --- | --- |") + for g in gpus: + lines.append(f"| {g['name']} | {_mark(g['utilPct'], threshold)} | " + f"{_mark(g['memPct'], threshold)}({g['memUsed']}/{g['memTotal']}) |") + else: + lines.append(f"> {gpus if isinstance(gpus, str) else '未检测到'}") + + docker = checks.get("docker") or {} + issues = docker.get("issues") or [] + if docker.get("running") is None: + lines.append(f"**Docker**: {issues[0]['issue'] if issues else '不可用'}") + else: + lines.append(f"**Docker 容器**: 运行中 {docker['running']}/{docker.get('total', '?')}" + + (f",已停用 {docker['stopped']}" if docker.get("stopped") else "") + + (f",⚠️ 异常 {len(issues)} 个" if issues else ",全部正常")) + for it in issues[:10]: + lines.append(f"- 🔴 {it['name']}: {it['issue']}") + + failures = checks.get("prefectFailures") or [] + lines.append(f"**Prefect 近24h 失败**: {len(failures)} 个 flow run" + + ("" if failures else " ✅")) + for f in failures[:8]: + lines.append(f"- 🔴 {f['name']}({f['state']}) {f['start']} {f['message'][:60]}") + + svc = checks.get("services") or {} + lines.append(f"**服务探活**: OCR {'✅' if svc.get('ocr') else '🔴'} | " + f"PG {'✅' if svc.get('pg') else '🔴'} | 负载 {os.getloadavg()[0]:.1f}") + + alarms = ( + (mem and (mem["pct"] >= threshold or mem["swapPct"] >= threshold)) + or any(d.get("pct", 0) >= threshold for d in (checks.get("disks") or [])) + or (isinstance(gpus, list) and any(g["utilPct"] >= threshold or g["memPct"] >= threshold for g in gpus)) + or bool(issues) or bool(failures) or not svc.get("ocr") or not svc.get("pg") + ) + lines += ["", ("🔴 **有项目需要处理,见上**" if alarms else "✅ **一切正常**")] + + content = "\n".join(lines) + sent = send_markdown(content) + emit_sync_summary("设备健康日报", [ + ("内存", f"{_mark(mem['pct'], threshold)}" if mem else "—"), + ("Docker 异常", len(issues)), + ("Prefect 失败(24h)", len(failures)), + ("OCR", "✅" if svc.get("ocr") else "🔴"), ("PG", "✅" if svc.get("pg") else "🔴"), + ("已发企微", "是" if sent else "否"), + ]) + return {"alarms": bool(alarms), "sent": sent, "dockerIssues": len(issues), "prefectFailures": len(failures)} + + +if __name__ == "__main__": + device_check.serve(name="device-check", cron="0 9 * * *") diff --git a/flows/shared/__init__.py b/flows/shared/__init__.py index 587e485..39cbc25 100644 --- a/flows/shared/__init__.py +++ b/flows/shared/__init__.py @@ -17,6 +17,7 @@ from flows.shared.clients import ( ) from flows.shared.db import SyncDB from flows.shared.mappers import GoodsUpsertInput, OrderUpsertInput +from flows.shared.wecom import send_markdown, send_text __all__ = [ "clients", "db", "mappers", "SyncDB", @@ -26,4 +27,5 @@ __all__ = [ "get_channel_type_map", "refresh_channel_type_map", "resolve_courier_source", "format_date", "start_of_day", "floor_to_hour", "build_stats_rows", "emit_sync_summary", "emit_verify_report", + "send_markdown", "send_text", ] diff --git a/flows/shared/db.py b/flows/shared/db.py index ac3c621..060dcf5 100644 --- a/flows/shared/db.py +++ b/flows/shared/db.py @@ -86,6 +86,51 @@ class SyncDB: row = cur.fetchone() return int(row[0]) if row else 0 + def get_goods_integrity(self, system_name: str, start: datetime, end: datetime) -> dict[str, int]: + """商品数据完整性(与核验同一时间窗,按 orderedAt): + - orders: 窗口内订单数 + - goods: 窗口内商品行数 + - zeroGoodsOrders: 窗口内没有任何商品行的订单数(可疑缺拉) + - goodsNoName: 窗口内商品 itemName 为空的行数 + - orphanGoods: 全库没有对应订单的商品行数(脏数据) + """ + with self._conn() as conn: + with conn.cursor() as cur: + cur.execute( + 'SELECT COUNT(*), ' + 'COALESCE(SUM(CASE WHEN NOT EXISTS (' + ' SELECT 1 FROM "Goods" g WHERE g."orderNo"=o."orderNo" AND g."createdAt"=o."createdAt"' + ') THEN 1 ELSE 0 END), 0) ' + 'FROM "Orders" o WHERE o."systemName"=%s AND o."orderedAt">=%s AND o."orderedAt"<%s', + (system_name, start, end), + ) + orders, zero_goods = cur.fetchone() + cur.execute( + 'SELECT COUNT(*), COALESCE(SUM(CASE WHEN "itemName" IS NULL OR "itemName"=\'\' ' + 'THEN 1 ELSE 0 END), 0) FROM "Goods" g JOIN "Orders" o ' + 'ON g."orderNo"=o."orderNo" AND g."createdAt"=o."createdAt" ' + 'WHERE o."systemName"=%s AND o."orderedAt">=%s AND o."orderedAt"<%s', + (system_name, start, end), + ) + goods, no_name = cur.fetchone() + cur.execute( + 'SELECT COUNT(*) FROM "Goods" g WHERE NOT EXISTS ' + '(SELECT 1 FROM "Orders" o WHERE o."orderNo"=g."orderNo")', + ) + orphan = cur.fetchone()[0] + return { + "orders": int(orders), "goods": int(goods), + "zeroGoodsOrders": int(zero_goods), "goodsNoName": int(no_name), "orphanGoods": int(orphan), + } + + def ping(self) -> str: + """连通性检查,返回 PG 版本号(供监控探活)。""" + with self._conn() as conn: + with conn.cursor() as cur: + cur.execute("SELECT version()") + row = cur.fetchone() + return (row[0] or "").split(",")[0] if row else "" + def get_orders_with_100_prefix(self, system_name: str, statuses: list[str]) -> list[tuple[str, datetime]]: with self._conn() as conn: with conn.cursor() as cur: diff --git a/flows/shared/wecom.py b/flows/shared/wecom.py new file mode 100644 index 0000000..c77c131 --- /dev/null +++ b/flows/shared/wecom.py @@ -0,0 +1,87 @@ +"""企业微信智能机器人通知层(wecom-aibot-sdk,WebSocket 按需连接)。 + +连接模式:每次发消息 建立→等认证→发送→断开,无状态,契合 flow run 级部署模型。 +凭据存 Prefect Secret Block `wecom-aibot-credentials`,JSON: + {"bot_id": "...", "secret": "...", "chat_id": "..."} +chat_id 为目标群会话 ID(用 scripts/wecom_capture_chatid.py 一次性捕获)。 + +所有函数容错——发送失败只 print 警告并返回 False,绝不中断 flow(对齐 artifacts.py 风格)。 +""" + +from __future__ import annotations + +import asyncio +import json + +from prefect.blocks.system import Secret + +SECRET_BLOCK = "wecom-aibot-credentials" +MAX_CONTENT_BYTES = 4000 # WeCom markdown 单条约 4096 字节上限,留余量 + + +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 + + +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: + 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: + parts.append(cur) + cur = line + else: + cur += line + if cur: + parts.append(cur) + return parts + + +def send_markdown(content: str) -> bool: + """发送 markdown 消息到目标群(超长自动分段)。失败返回 False,不抛异常。""" + 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}") + return False + + +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 diff --git a/prefect.yaml b/prefect.yaml index 228910b..cbb4434 100644 --- a/prefect.yaml +++ b/prefect.yaml @@ -87,3 +87,31 @@ deployments: timezone: Asia/Shanghai work_pool: name: inkreach-pool +- name: wecom-alert + entrypoint: flows/monitor/alert_failure.py:wecom_alert_failure + triggers: + - type: event + enabled: true + match: + prefect.resource.id: prefect.flow-run.* + expect: + - prefect.flow-run.Failed + - prefect.flow-run.Crashed + parameters: + event: "{{ event }}" + work_pool: + name: inkreach-pool +- name: daily-report + entrypoint: flows/monitor/daily_report.py:daily_report + schedules: + - cron: 30 18 * * * + timezone: Asia/Shanghai + work_pool: + name: inkreach-pool +- name: device-check + entrypoint: flows/monitor/device_check.py:device_check + schedules: + - cron: 0 9 * * * + timezone: Asia/Shanghai + work_pool: + name: inkreach-pool diff --git a/pyproject.toml b/pyproject.toml index 40cb739..12c7a7c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,8 @@ dependencies = [ "httpx", "pandas", "pydantic>=2.0", + "wecom-aibot-sdk>=1.0.8", + "psutil>=7.2.2", ] [build-system] diff --git a/scripts/wecom_capture_chatid.py b/scripts/wecom_capture_chatid.py new file mode 100644 index 0000000..fd5355e --- /dev/null +++ b/scripts/wecom_capture_chatid.py @@ -0,0 +1,116 @@ +"""一次性工具:捕获目标群的 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}\n\n来自 inkreach-prefect 通知链路。") + 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 16ef435..5b6638b 100644 --- a/uv.lock +++ b/uv.lock @@ -742,8 +742,10 @@ dependencies = [ { name = "pandas" }, { name = "prefect" }, { name = "prefect-sqlalchemy" }, + { name = "psutil" }, { name = "psycopg2-binary" }, { name = "pydantic" }, + { name = "wecom-aibot-sdk" }, ] [package.metadata] @@ -752,8 +754,10 @@ requires-dist = [ { name = "pandas" }, { name = "prefect", specifier = ">=3.4" }, { name = "prefect-sqlalchemy" }, + { name = "psutil", specifier = ">=7.2.2" }, { name = "psycopg2-binary" }, { name = "pydantic", specifier = ">=2.0" }, + { name = "wecom-aibot-sdk", specifier = ">=1.0.8" }, ] [[package]] @@ -1265,6 +1269,34 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/8d/9b/d4b1e644385499c8346fa9b622a3f030dce14cd6ef8a1871c221a17a67e7/prometheus_client-0.25.0-py3-none-any.whl", hash = "sha256:d5aec89e349a6ec230805d0df882f3807f74fd6c1a2fa86864e3c2279059fed1", size = 64154, upload-time = "2026-04-09T19:53:41.324Z" }, ] +[[package]] +name = "psutil" +version = "7.2.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/aa/c6/d1ddf4abb55e93cebc4f2ed8b5d6dbad109ecb8d63748dd2b20ab5e57ebe/psutil-7.2.2.tar.gz", hash = "sha256:0746f5f8d406af344fd547f1c8daa5f5c33dbc293bb8d6a16d80b4bb88f59372", size = 493740, upload-time = "2026-01-28T18:14:54.428Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/51/08/510cbdb69c25a96f4ae523f733cdc963ae654904e8db864c07585ef99875/psutil-7.2.2-cp313-cp313t-macosx_10_13_x86_64.whl", hash = "sha256:2edccc433cbfa046b980b0df0171cd25bcaeb3a68fe9022db0979e7aa74a826b", size = 130595, upload-time = "2026-01-28T18:14:57.293Z" }, + { url = "https://files.pythonhosted.org/packages/d6/f5/97baea3fe7a5a9af7436301f85490905379b1c6f2dd51fe3ecf24b4c5fbf/psutil-7.2.2-cp313-cp313t-macosx_11_0_arm64.whl", hash = "sha256:e78c8603dcd9a04c7364f1a3e670cea95d51ee865e4efb3556a3a63adef958ea", size = 131082, upload-time = "2026-01-28T18:14:59.732Z" }, + { url = "https://files.pythonhosted.org/packages/37/d6/246513fbf9fa174af531f28412297dd05241d97a75911ac8febefa1a53c6/psutil-7.2.2-cp313-cp313t-manylinux2010_x86_64.manylinux_2_12_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:1a571f2330c966c62aeda00dd24620425d4b0cc86881c89861fbc04549e5dc63", size = 181476, upload-time = "2026-01-28T18:15:01.884Z" }, + { url = "https://files.pythonhosted.org/packages/b8/b5/9182c9af3836cca61696dabe4fd1304e17bc56cb62f17439e1154f225dd3/psutil-7.2.2-cp313-cp313t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:917e891983ca3c1887b4ef36447b1e0873e70c933afc831c6b6da078ba474312", size = 184062, upload-time = "2026-01-28T18:15:04.436Z" }, + { url = "https://files.pythonhosted.org/packages/16/ba/0756dca669f5a9300d0cbcbfae9a4c30e446dfc7440ffe43ded5724bfd93/psutil-7.2.2-cp313-cp313t-win_amd64.whl", hash = "sha256:ab486563df44c17f5173621c7b198955bd6b613fb87c71c161f827d3fb149a9b", size = 139893, upload-time = "2026-01-28T18:15:06.378Z" }, + { url = "https://files.pythonhosted.org/packages/1c/61/8fa0e26f33623b49949346de05ec1ddaad02ed8ba64af45f40a147dbfa97/psutil-7.2.2-cp313-cp313t-win_arm64.whl", hash = "sha256:ae0aefdd8796a7737eccea863f80f81e468a1e4cf14d926bd9b6f5f2d5f90ca9", size = 135589, upload-time = "2026-01-28T18:15:08.03Z" }, + { url = "https://files.pythonhosted.org/packages/81/69/ef179ab5ca24f32acc1dac0c247fd6a13b501fd5534dbae0e05a1c48b66d/psutil-7.2.2-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:eed63d3b4d62449571547b60578c5b2c4bcccc5387148db46e0c2313dad0ee00", size = 130664, upload-time = "2026-01-28T18:15:09.469Z" }, + { url = "https://files.pythonhosted.org/packages/7b/64/665248b557a236d3fa9efc378d60d95ef56dd0a490c2cd37dafc7660d4a9/psutil-7.2.2-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:7b6d09433a10592ce39b13d7be5a54fbac1d1228ed29abc880fb23df7cb694c9", size = 131087, upload-time = "2026-01-28T18:15:11.724Z" }, + { url = "https://files.pythonhosted.org/packages/d5/2e/e6782744700d6759ebce3043dcfa661fb61e2fb752b91cdeae9af12c2178/psutil-7.2.2-cp314-cp314t-manylinux2010_x86_64.manylinux_2_12_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:1fa4ecf83bcdf6e6c8f4449aff98eefb5d0604bf88cb883d7da3d8d2d909546a", size = 182383, upload-time = "2026-01-28T18:15:13.445Z" }, + { url = "https://files.pythonhosted.org/packages/57/49/0a41cefd10cb7505cdc04dab3eacf24c0c2cb158a998b8c7b1d27ee2c1f5/psutil-7.2.2-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:e452c464a02e7dc7822a05d25db4cde564444a67e58539a00f929c51eddda0cf", size = 185210, upload-time = "2026-01-28T18:15:16.002Z" }, + { url = "https://files.pythonhosted.org/packages/dd/2c/ff9bfb544f283ba5f83ba725a3c5fec6d6b10b8f27ac1dc641c473dc390d/psutil-7.2.2-cp314-cp314t-win_amd64.whl", hash = "sha256:c7663d4e37f13e884d13994247449e9f8f574bc4655d509c3b95e9ec9e2b9dc1", size = 141228, upload-time = "2026-01-28T18:15:18.385Z" }, + { url = "https://files.pythonhosted.org/packages/f2/fc/f8d9c31db14fcec13748d373e668bc3bed94d9077dbc17fb0eebc073233c/psutil-7.2.2-cp314-cp314t-win_arm64.whl", hash = "sha256:11fe5a4f613759764e79c65cf11ebdf26e33d6dd34336f8a337aa2996d71c841", size = 136284, upload-time = "2026-01-28T18:15:19.912Z" }, + { url = "https://files.pythonhosted.org/packages/e7/36/5ee6e05c9bd427237b11b3937ad82bb8ad2752d72c6969314590dd0c2f6e/psutil-7.2.2-cp36-abi3-macosx_10_9_x86_64.whl", hash = "sha256:ed0cace939114f62738d808fdcecd4c869222507e266e574799e9c0faa17d486", size = 129090, upload-time = "2026-01-28T18:15:22.168Z" }, + { url = "https://files.pythonhosted.org/packages/80/c4/f5af4c1ca8c1eeb2e92ccca14ce8effdeec651d5ab6053c589b074eda6e1/psutil-7.2.2-cp36-abi3-macosx_11_0_arm64.whl", hash = "sha256:1a7b04c10f32cc88ab39cbf606e117fd74721c831c98a27dc04578deb0c16979", size = 129859, upload-time = "2026-01-28T18:15:23.795Z" }, + { url = "https://files.pythonhosted.org/packages/b5/70/5d8df3b09e25bce090399cf48e452d25c935ab72dad19406c77f4e828045/psutil-7.2.2-cp36-abi3-manylinux2010_x86_64.manylinux_2_12_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:076a2d2f923fd4821644f5ba89f059523da90dc9014e85f8e45a5774ca5bc6f9", size = 155560, upload-time = "2026-01-28T18:15:25.976Z" }, + { url = "https://files.pythonhosted.org/packages/63/65/37648c0c158dc222aba51c089eb3bdfa238e621674dc42d48706e639204f/psutil-7.2.2-cp36-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:b0726cecd84f9474419d67252add4ac0cd9811b04d61123054b9fb6f57df6e9e", size = 156997, upload-time = "2026-01-28T18:15:27.794Z" }, + { url = "https://files.pythonhosted.org/packages/8e/13/125093eadae863ce03c6ffdbae9929430d116a246ef69866dad94da3bfbc/psutil-7.2.2-cp36-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:fd04ef36b4a6d599bbdb225dd1d3f51e00105f6d48a28f006da7f9822f2606d8", size = 148972, upload-time = "2026-01-28T18:15:29.342Z" }, + { url = "https://files.pythonhosted.org/packages/04/78/0acd37ca84ce3ddffaa92ef0f571e073faa6d8ff1f0559ab1272188ea2be/psutil-7.2.2-cp36-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:b58fabe35e80b264a4e3bb23e6b96f9e45a3df7fb7eed419ac0e5947c61e47cc", size = 148266, upload-time = "2026-01-28T18:15:31.597Z" }, + { url = "https://files.pythonhosted.org/packages/b4/90/e2159492b5426be0c1fef7acba807a03511f97c5f86b3caeda6ad92351a7/psutil-7.2.2-cp37-abi3-win_amd64.whl", hash = "sha256:eb7e81434c8d223ec4a219b5fc1c47d0417b12be7ea866e24fb5ad6e84b3d988", size = 137737, upload-time = "2026-01-28T18:15:33.849Z" }, + { url = "https://files.pythonhosted.org/packages/8c/c7/7bb2e321574b10df20cbde462a94e2b71d05f9bbda251ef27d104668306a/psutil-7.2.2-cp37-abi3-win_arm64.whl", hash = "sha256:8c233660f575a5a89e6d4cb65d9f938126312bca76d8fe087b947b3a1aaac9ee", size = 134617, upload-time = "2026-01-28T18:15:36.514Z" }, +] + [[package]] name = "psycopg2-binary" version = "2.9.12" @@ -2209,6 +2241,20 @@ 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"