AGENTS.md

ETL 工作流项目,使用 Prefect 编排。任务: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。

依赖管理

用 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 容器),数据持久化。

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 → 自动生效。
CI/CD 自动部署:develop 合并到 main(push)→ Gitea webhook → 本机 9000 的 inkreach-prefect-webhook 服务(systemd,deploy 账号)→ 自动 git pull + uv sync + prefect deploy --all → 企微群通知。CI 检查:.gitea/workflows/ci.yml,PR 到 develop/main 时跑轻量语法检查(runner 标签 ubuntu-latest,本机 gitea-act-runner)。
部署脚本与密钥在 /opt/inkreach-prefect-deploy/(deploy.sh / rollback.sh / status.sh / webhook.secret / wecom_webhook.txt,均 deploy 账号所有);代码检出在 /opt/inkreach-prefect(deploy 账号所有,deploy.sh 自动 reset 到 origin/main)。

密钥管理(标准做法)

不要把密钥写进代码或 pyproject.toml。

数据库凭据用 Prefect Block 加密存储,Block 名:pg-inkreach-dmp(SqlAlchemyConnector 类型)。

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/<file>.py
部署到 work pool:uv run prefect deploy --all(按 prefect.yaml,全部 12 个 deployment 挂 inkreach-pool)。
手动触发一次:uv run prefect deployment run '<name>/<name>'
写入目标: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(文档),全容错。凭据在 Secret Block wecom-webhook(值为完整 webhook URL);markdown 支持 1-6 级标题/引用/<font color="info|comment|warning">(不支持表格),单条 ≤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

7.1 KiB Raw Blame History