# AGENTS.md ETL 工作流项目,使用 [Prefect](https://docs.prefect.org.cn/v3/get-started) 编排。 任务:REST API → PostgreSQL(Extract / Transform / Load)。 ## 设计理念(简述) - **高内聚、低耦合、关注点分离**。 - `api/` 是纯客户端层(httpx + pydantic,**禁止 import prefect**),封装 RIIN/SDS 两个外部系统的 REST API,可独立复用/测试。 - `flows/` 是编排层(依赖 `api/`,使用 prefect),负责重试/缓存/并发/调度/密钥。 - 共享行为(token 注入、401 重登、分页、批量、缓存)放 `api/_common/`,用组合式 helper,不过度抽象。 - 新增系统是加法式:加一个 `api/<系统>/` + 一个 flow,不碰旧代码。 - 详细架构说明见 [`docs/dev/systematic-design.md`](docs/dev/systematic-design.md)。 ## 依赖管理 用 [uv](https://docs.astral.sh/uv/) 管理依赖(锁定文件 `uv.lock`,Python `>=3.12`):`uv sync` 装依赖,`uv add/remove` 增删包,一切命令用 `uv run ...` 前缀,不要直接用系统 `python` / `pip`。提交时一并提交 `pyproject.toml` 和 `uv.lock`;不要新建 requirements.txt,不要提交 `.venv/`。 ## 部署流程 - **Prefect server + worker**:自托管,通过 Docker Compose 运行(server + worker 容器),数据持久化。 ```bash docker compose up -d # 启动 server(UI: http://192.168.124.92:4200)+ worker docker compose ps # 查看状态 docker compose logs -f prefect-worker # 看 worker 日志 docker compose down # 停止 ``` - **API 地址**:`http://localhost:4200/api`(本机)/ `http://192.168.124.92:4200`(局域网);客户端已配置 `PREFECT_API_URL`。 - 代码托管在自托管 Gitea:`ssh://git@gitea.inkreach.cc:2222/inkreachhub/inkreach-prefect.git`。分支:`main`(主干,worker 拉取)、`develop`(开发)、`deploy`(部署);旧远程 Gitee(`git@gitee.com:inkreachZLL/inkreach-prefect.git`)已退役保留。 - worker 是 **Docker 容器**(`prefect-worker` 服务,基于 `Dockerfile`),镜像只装依赖(git + uv 按锁文件装),**不烤业务代码**;每个 flow run 执行前,pull 步骤从 Gitea **clone 最新代码**(token 存 `gitea-token` Secret Block,运行时解析)。 - **改 flow 代码 → `git push` → 下次 run 自动用新代码**(worker 不重启、不 redeploy);改 deployment 配置(cron/参数)才需重跑 `uv run prefect deploy --all`(读取 `prefect.yaml`)。 - **日常开发流程**:clone Gitea 仓库 → 改代码 → `git push` → 自动生效。 ## 密钥管理(标准做法) - **不要把密钥写进代码或 pyproject.toml**。 - 数据库凭据用 Prefect Block 加密存储,Block 名:`pg-inkreach-dmp`(`SqlAlchemyConnector` 类型)。 ```python from prefect_sqlalchemy import SqlAlchemyConnector connector = SqlAlchemyConnector.load("pg-inkreach-dmp") engine = connector.get_engine() ``` - RIIN / SDS 登录凭据用 `Secret` Block 加密存储,值为 JSON:`{"username":"...","password":"..."}`。 Block 名:`riin-credentials`、`sds-credentials`。flow 在 extract task 内部读取,不外泄。 - Gitea 拉取令牌:`Secret` Block `gitea-token`(裸 token,pull 步骤用 `oauth2:{{ secret }}@gitea.inkreach.cc` 内嵌)。 - 本地开发备用:`.env`(已 gitignore)含 `DB_DSN`,加载用 `python-dotenv` 或 `os.environ`。 ## 编写/运行工作流 - 同步任务(对齐 inkreach-bi-auth-mng 的 sync tasks)放在 `flows/riin/` 和 `flows/sds/`,每系统各 6 个: - `order_sync`(按时间窗口)、`incremental_sync`(按 DB 最近下单时间增量,无参)、 `recent_sync`(近 N 天)、`refresh_pending`(刷新未完结)、`by_orders`(RIIN 指定单号)/ `refresh_100_prefix`(SDS 100 前缀)、`verify`(逐天核验修复)。 - 共享层在 `flows/shared/`:`mappers.py`(RIIN/SDS→统一 Order/Goods)、`db.py`(PG upsert `SyncDB`,写并发控制:advisory 锁串行 + 分批提交 + lock_timeout + 死锁重试)、`clients.py`(客户端工厂+顺序分页+重试+渠道字典)。 - 本地直接跑(调试):`uv run python flows/riin/.py` - 部署到 work pool:`uv run prefect deploy --all`(按 `prefect.yaml`,全部 12 个 deployment 挂 `inkreach-pool`)。 - 手动触发一次:`uv run prefect deployment run '/'` - 写入目标:PG 的 `Orders` / `Goods` / `Merchant` 表(主键 `Orders=("orderNo","createdAt")`、`Goods=("orderNo","productionNo")`),`createdAt` 写入前以库中已有值覆盖以保证幂等。 > 注:`serve_all.py` 为旧进程式部署,已被 work pool 模型取代,保留仅作参考。 ## 企微机器人告警/上报(monitor) - 发送层 `flows/shared/wecom.py`:`send_markdown/send_text`,群机器人 webhook HTTP POST([文档](https://developer.work.weixin.qq.com/document/path/99110)),全容错。 凭据在 `Secret` Block `wecom-webhook`(值为完整 webhook URL);markdown 支持 1-6 级标题/引用/``(不支持表格),单条 ≤4096 字节超长自动分段,限 20 条/分钟。 - `flows/monitor/alert_failure.py`(wecom-alert):prefect.yaml 里声明的事件触发器,任意 flow run `Failed/Crashed` 自动运行并发群报警,零代码侵入,新 flow 自动覆盖。 - `flows/monitor/daily_report.py`(daily-report,18:30):近 7 天 RIIN/SDS 核验(subflow 复用 verify,先报订单数量/商品行数是否一致,再报状态不符单数"已自动刷新修复",verify 有差异时修复后重数即 `verifyReport.post`)+OCR 当日处理量(按 `Orders.ocrProcessedAt` 盖章,`db.count_ocr_today`)+近 24h 失败任务明细(workflow/run/次数/时间)。 - `flows/monitor/device_check.py`(device-check,09:00):按 24h 采样报内存/swap/GPU 的平均·峰值·最低 + 磁盘 总体/根目录/home 余量与"增长最快目录"(du 快照对比)+ GPU 温度;另查 Docker 容器异常(docker.sock)、Prefect 失败、OCR/PG 探活。阈值 80% 报警。 - **`metrics-agent` 侧车容器**(docker-compose 服务,非 Prefect flow):`scripts/metrics_agent.py` 每 5 分钟采样内存/swap/GPU 写 `data/metrics/metrics-YYYY-MM-DD.jsonl`,每日 7 点后 du 快照 `data/metrics/disk-*.json`(实际占块;容错个别文件消失的 rc=1);保留采样 14 天/快照 7 份。device-check 经 `/data/metrics:ro` 读取,采样缺失自动降级为瞬时值。 - worker 容器已挂载 `/:/host:ro`、`/var/run/docker.sock`、GPU、`./data/metrics:ro`(见 docker-compose.yml),勿移除,否则对应检查降级;镜像含 tzdata(`ENV TZ=Asia/Shanghai`)。 ## 参考文档 - Prefect 入门:https://docs.prefect.org.cn/v3/get-started - API 驱动 ETL 示例:https://docs.prefect.org.cn/v3/examples/run-api-sourced-etl - 密钥存储:https://docs.prefect.org.cn/v3/how-to-guides/configuration/store-secrets