diff --git a/AGENTS.md b/AGENTS.md index 51e7f28..934cbcd 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -33,16 +33,25 @@ uv run prefect # 运行 prefect CLI ## 运行环境 -- **Prefect server**:自托管,通过 Docker Compose 运行,数据持久化。 +- **Prefect server + worker**:自托管,通过 Docker Compose 运行(server + worker 容器),数据持久化。 ```bash - docker compose up -d # 启动 server(UI: http://192.168.124.92:4200) + docker compose up -d # 启动 server(UI: http://192.168.124.92:4200)+ worker docker compose ps # 查看状态 - docker compose logs -f # 查看日志 + docker compose logs -f prefect-worker # 看 worker 日志 docker compose down # 停止 ``` - **API 地址**:`http://localhost:4200/api`(本机)/ `http://192.168.124.92:4200`(局域网)。 - 客户端已配置:`uv run prefect config set PREFECT_API_URL="http://localhost:4200/api"`。 +## 部署模型:Work Pool + Git 拉取(推送即生效) + +- 代码托管在 Gitee:`git@gitee.com:inkreachZLL/inkreach-prefect.git`(分支 `master`)。 +- worker 是 **Docker 容器**(`prefect-worker` 服务,基于 `Dockerfile`),镜像只装依赖(git + uv 按锁文件装),**不烤业务代码**。 +- 每个 flow run 执行前,pull 步骤从 Gitee **clone 最新代码**(token 存 `gitee-token` Secret Block,运行时解析)。 +- **改 flow 代码 → `git push` → 下次 run 自动用新代码**(worker 不重启、不 redeploy)。 +- 改 deployment 配置(cron/参数)才需重跑:`uv run prefect deploy --all`(读取 `prefect.yaml`)。 +- **日常开发流程**:本地开发机 clone Gitee 仓库 → 改代码 → `git push` → 自动生效。 + ## 密钥管理(标准做法) - **不要把密钥写进代码或 pyproject.toml**。 @@ -54,6 +63,7 @@ uv run prefect # 运行 prefect CLI ``` - RIIN / SDS 登录凭据用 `Secret` Block 加密存储,值为 JSON:`{"username":"...","password":"..."}`。 Block 名:`riin-credentials`、`sds-credentials`。flow 在 extract task 内部读取,不外泄。 +- Gitee 拉取令牌:`Secret` Block `gitee-token`(裸 token,pull 步骤用 `oauth2:{{ secret }}@gitee.com` 内嵌)。 - 本地开发备用:`.env`(已 gitignore)含 `DB_DSN`,加载用 `python-dotenv` 或 `os.environ`。 ## 编写/运行工作流 @@ -62,10 +72,13 @@ uv run prefect # 运行 prefect CLI - `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`)、`clients.py`(客户端工厂+顺序分页+重试+渠道字典)。 -- 运行单个 flow:`uv run python flows/riin/.py`(或 `flows/sds/.py`) -- 部署调度:flow 末尾 `flow.serve(name=..., cron=...)`,保持 serve 进程运行。 +- 本地直接跑(调试):`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 模型取代,保留仅作参考。 + ## 参考文档 - Prefect 入门:https://docs.prefect.org.cn/v3/get-started diff --git a/prefect.yaml b/prefect.yaml index acf4d7c..75eb44f 100644 --- a/prefect.yaml +++ b/prefect.yaml @@ -1,78 +1,82 @@ prefect-version: 3.7.8 name: inkreach-prefect - -# 每次 flow run 前:从 Gitee clone 最新代码(token 运行时从 Secret 解析,不落盘) pull: - prefect.deployments.steps.git_clone: - repository: "https://oauth2:{{ prefect.blocks.secret.gitee-token }}@gitee.com/inkreachZLL/inkreach-prefect.git" + repository: https://oauth2:{{ prefect.blocks.secret.gitee-token }}@gitee.com/inkreachZLL/inkreach-prefect.git branch: master - -defaults: - work_pool: - name: inkreach-pool - deployments: -# ---------- RIIN ---------- - name: riin-order-sync entrypoint: flows/riin/order_sync.py:riin_order_sync - + work_pool: + name: inkreach-pool - name: riin-incremental-sync entrypoint: flows/riin/incremental_sync.py:riin_incremental_sync schedules: - - cron: "0 * * * *" - timezone: "Asia/Shanghai" - + - cron: 0 * * * * + timezone: Asia/Shanghai + work_pool: + name: inkreach-pool - name: riin-recent-sync entrypoint: flows/riin/recent_sync.py:riin_recent_sync schedules: - - cron: "30 * * * *" - timezone: "Asia/Shanghai" - + - cron: 30 * * * * + timezone: Asia/Shanghai + work_pool: + name: inkreach-pool - name: riin-refresh-pending entrypoint: flows/riin/refresh_pending.py:riin_refresh_pending schedules: - - cron: "0 */3 * * *" - timezone: "Asia/Shanghai" - + - cron: 0 */3 * * * + timezone: Asia/Shanghai + work_pool: + name: inkreach-pool - name: riin-by-orders-sync entrypoint: flows/riin/by_orders.py:riin_by_orders_sync - + work_pool: + name: inkreach-pool - name: riin-verify entrypoint: flows/riin/verify.py:riin_verify schedules: - - cron: "0 6 * * *" - timezone: "Asia/Shanghai" - -# ---------- SDS ---------- + - cron: 0 6 * * * + timezone: Asia/Shanghai + work_pool: + name: inkreach-pool - name: sds-order-sync entrypoint: flows/sds/order_sync.py:sds_order_sync - + work_pool: + name: inkreach-pool - name: sds-incremental-sync entrypoint: flows/sds/incremental_sync.py:sds_incremental_sync schedules: - - cron: "15 * * * *" - timezone: "Asia/Shanghai" - + - cron: 15 * * * * + timezone: Asia/Shanghai + work_pool: + name: inkreach-pool - name: sds-recent-sync entrypoint: flows/sds/recent_sync.py:sds_recent_sync schedules: - - cron: "45 * * * *" - timezone: "Asia/Shanghai" - + - cron: 45 * * * * + timezone: Asia/Shanghai + work_pool: + name: inkreach-pool - name: sds-refresh-pending entrypoint: flows/sds/refresh_pending.py:sds_refresh_pending schedules: - - cron: "0 */2 * * *" - timezone: "Asia/Shanghai" - + - cron: 0 */2 * * * + timezone: Asia/Shanghai + work_pool: + name: inkreach-pool - name: sds-refresh-100-prefix entrypoint: flows/sds/refresh_100_prefix.py:sds_refresh_100_prefix schedules: - - cron: "30 */2 * * *" - timezone: "Asia/Shanghai" - + - cron: 30 */2 * * * + timezone: Asia/Shanghai + work_pool: + name: inkreach-pool - name: sds-verify entrypoint: flows/sds/verify.py:sds_verify schedules: - - cron: "0 7 * * *" - timezone: "Asia/Shanghai" + - cron: 0 7 * * * + timezone: Asia/Shanghai + work_pool: + name: inkreach-pool