diff --git a/DOCKER_PREFECT_USAGE.md b/DOCKER_PREFECT_USAGE.md index 5dfe212..c484472 100644 --- a/DOCKER_PREFECT_USAGE.md +++ b/DOCKER_PREFECT_USAGE.md @@ -1,40 +1,60 @@ -# Docker + Prefect +# Docker + Prefect 使用说明 -## 1. 构建镜像 +当前方案包含两条定时任务： + +- `08:00` 执行数据统计并写入飞书多维表格 +- `09:00` 推送结果到飞书群 + +调度配置写在 [prefect.yaml](/Users/idss/YXT_Project/gmv-analytics/prefect.yaml) 中。 + +## 1. 创建 Docker 镜像 + +在项目目录执行： ```bash cd /Users/idss/YXT_Project/gmv-analytics docker build -t gmv-analytics-prefect:latest . ``` -## 2. 单次执行 +镜像内容来自： -```bash -docker run --rm \ - --env-file /Users/idss/YXT_Project/gmv-analytics/.env \ - -e PROJECT_MAPPING_FILE=/app/店铺映射表.xlsx \ - gmv-analytics-prefect:latest \ - python prefect_gmv_scheduler.py -``` +- [Dockerfile](/Users/idss/YXT_Project/gmv-analytics/Dockerfile) +- [docker-entrypoint.sh](/Users/idss/YXT_Project/gmv-analytics/docker-entrypoint.sh) -## 3. 启动每天 09:00 的 Prefect 调度 +容器启动后会： -这会连接 `.env` 中已经配置的 `PREFECT_API_URL`，并在 Prefect 中启动一个 deployment。 - -```bash -docker run --rm \ - --env-file /Users/idss/YXT_Project/gmv-analytics/.env \ - -e PROJECT_MAPPING_FILE=/app/店铺映射表.xlsx \ - gmv-analytics-prefect:latest -``` - -容器默认执行： +- 加载 `.env` +- 自动把 `PROJECT_MAPPING_FILE` 修正为容器内的 `/app/店铺映射表.xlsx` +- 默认执行： ```bash python prefect_gmv_scheduler.py --serve ``` -也可以直接用 Compose： +## 2. 启动 Docker 容器 + +### 方式 A：直接用 `docker run` + +```bash +docker run -d \ + --name gmv-prefect-scheduler \ + --env-file /Users/idss/YXT_Project/gmv-analytics/.env \ + -e PROJECT_MAPPING_FILE=/app/店铺映射表.xlsx \ + gmv-analytics-prefect:latest +``` + +停止容器： + +```bash +docker stop gmv-prefect-scheduler +docker rm gmv-prefect-scheduler +``` + +### 方式 B：用 `docker compose` + +项目里已经有 [docker-compose.yaml](/Users/idss/YXT_Project/gmv-analytics/docker-compose.yaml)。 + +启动： ```bash cd /Users/idss/YXT_Project/gmv-analytics @@ -47,40 +67,172 @@ docker compose up -d --build docker compose down ``` -## 4. 如果你是用 Prefect worker 跑容器 +## 3. 注册 Prefect 调度 -先保证你的 Prefect worker 已经连到同一个 `PREFECT_API_URL`。 +推荐方式是使用 [prefect.yaml](/Users/idss/YXT_Project/gmv-analytics/prefect.yaml) 注册 deployment： -推荐命令示例： +```bash +cd /Users/idss/YXT_Project/gmv-analytics +prefect deploy --all +``` + +这会向 `.env` 中的 `PREFECT_API_URL` 注册两个 deployment： + +- `gmv-daily-data-sync-8am` +- `gmv-daily-notify-9am` + +前提条件： + +- `.env` 中已经配置好 `PREFECT_API_URL` +- Prefect API 服务可访问 +- 已经有可用的 work pool + 当前 [prefect.yaml](/Users/idss/YXT_Project/gmv-analytics/prefect.yaml) 默认使用： + `default-agent-pool` + +如果还没有 worker，可以在宿主机启动一个： ```bash prefect worker start --pool default-agent-pool --type process ``` -如果你后面要改成真正的 Docker work pool，再在 Prefect UI 中把 deployment 的基础设施切到 Docker，并指定镜像： +## 4. 自动执行是怎么发生的 -```text -gmv-analytics-prefect:latest +如果你走的是 `prefect deploy --all` 方案： + +1. `prefect.yaml` 注册两个 deployment +2. Prefect Server/Cloud 按 cron 触发： + - `08:00`：`gmv-daily-data-sync-8am` + - `09:00`：`gmv-daily-notify-9am` +3. Prefect worker 拉起执行 +4. 代码入口在 [prefect_gmv_scheduler.py](/Users/idss/YXT_Project/gmv-analytics/prefect_gmv_scheduler.py) + +如果你走的是容器内 `--serve` 方案： + +1. 容器常驻运行 `python prefect_gmv_scheduler.py --serve` +2. 代码里会本地注册并维持两个定时调度 +3. 到时间后容器内部直接执行 flow + +建议： + +- 正式环境优先用 `prefect.yaml + prefect deploy --all` +- `--serve` 更适合本地或过渡期 + +## 5. 单次手动执行 + +只跑数据同步： + +```bash +python /Users/idss/YXT_Project/gmv-analytics/prefect_gmv_scheduler.py --mode sync ``` -## 5. 常用覆盖参数 +只跑飞书通知： -单次 dry-run： +```bash +python /Users/idss/YXT_Project/gmv-analytics/prefect_gmv_scheduler.py --mode notify +``` + +先同步再通知： + +```bash +python /Users/idss/YXT_Project/gmv-analytics/prefect_gmv_scheduler.py --mode all +``` + +如果要在 Docker 里手动执行： ```bash docker run --rm \ --env-file /Users/idss/YXT_Project/gmv-analytics/.env \ -e PROJECT_MAPPING_FILE=/app/店铺映射表.xlsx \ gmv-analytics-prefect:latest \ - python prefect_gmv_scheduler.py --dry-run + python prefect_gmv_scheduler.py --mode sync ``` -指定时间范围： +## 6. Docker 中的日志 + +现在日志分成两层： + +### 6.1 容器标准输出日志 + +这是你最先看到的日志来源。 + +如果用 `docker run`： ```bash -docker run --rm \ - --env-file /Users/idss/YXT_Project/gmv-analytics/.env \ - -e PROJECT_MAPPING_FILE=/app/店铺映射表.xlsx \ - gmv-analytics-prefect:latest \ - python prefect_gmv_scheduler.py --start-date 2026-03-01 --end-date 2026-03-12 +docker logs -f gmv-prefect-scheduler ``` + +如果用 `docker compose`： + +```bash +cd /Users/idss/YXT_Project/gmv-analytics +docker compose logs -f +``` + +这些日志包含： + +- Prefect flow 启动/完成 +- 每天数据同步流程 +- 飞书接口日志 +- 错误堆栈 + +### 6.2 容器内本地文件日志 + +项目已经接了统一日志模块 [logging_config.py](/Users/idss/YXT_Project/gmv-analytics/logging_config.py)，所以日志还会写入容器内文件： + +- 默认目录：`/app/logs` +- 默认文件：`/app/logs/gmv_sync.log` + +进入容器查看： + +```bash +docker exec -it gmv-prefect-scheduler sh +tail -f /app/logs/gmv_sync.log +``` + +如果是 compose： + +```bash +docker exec -it gmv-prefect-scheduler sh +tail -f /app/logs/gmv_sync.log +``` + +### 6.3 日志级别 + +默认是 `INFO`。 +如果要看更细的调试日志，在 `.env` 里加： + +```bash +LOG_LEVEL=DEBUG +``` + +可选日志配置还有： + +```bash +LOG_DIR=/app/logs +LOG_FILE_NAME=gmv_sync.log +LOG_MAX_BYTES=10485760 +LOG_BACKUP_COUNT=7 +``` + +## 7. 现在最推荐的执行方式 + +如果你要的是“长期稳定自动执行”，建议这样： + +1. 构建镜像 +2. 准备好 `.env` +3. 启动 Prefect API +4. 启动 worker +5. 执行： + +```bash +cd /Users/idss/YXT_Project/gmv-analytics +prefect deploy --all +``` + +如果你只是希望一个容器自己在里面定时跑，也可以直接： + +```bash +docker compose up -d --build +``` + +这时它会直接跑 `--serve`，由容器自己维持调度。 diff --git a/__pycache__/project_bitable_common.cpython-312.pyc b/__pycache__/project_bitable_common.cpython-312.pyc index 95f58ec..695cb01 100644 Binary files a/__pycache__/project_bitable_common.cpython-312.pyc and b/__pycache__/project_bitable_common.cpython-312.pyc differ diff --git a/field_group_state.json b/field_group_state.json new file mode 100644 index 0000000..c34d3f1 --- /dev/null +++ b/field_group_state.json @@ -0,0 +1,34 @@ +{ + "XpA0biRh4arbSTsEAcuc7U1Znhh": { + "tblIdcsE2RP8b0V4": { + "汇总金额": "fgkKA4sTDeMhBDQ", + "GMV编组": "fgOuepoxtRQ9OxX", + "销售额/退款额": "fgScyMddEHsCkKp" + }, + "tblqQ5MtqfVDgBx7": { + "汇总金额": "fgkRBorMX90vTeA", + "GMV编组": "fg76wQNHbh2m7E2", + "销售额/退款额": "fg6vKipfag0BSxi" + }, + "tbljWa2kxC1GAFLO": { + "汇总金额": "fgF9Nj9fizqmrnK", + "GMV编组": "fg4HnQJQhxdwcyv", + "销售额/退款额": "fgoSAxK2ABO8C7e" + }, + "tblFnLz4YNVDKHhx": { + "汇总金额": "fghdR8NiF0ZMMi6", + "GMV编组": "fgR3uN6iTbJngoS", + "销售额/退款额": "fgdc1C0NkN5MwFr" + }, + "tblCo2dD4TWFebBv": { + "汇总金额": "fgLvnxp0hEpz6M4", + "GMV编组": "fgItrizttYa2tUO", + "销售额/退款额": "fg3GucwB5LnbJtj" + }, + "tblBuzDGVzRPlrlU": { + "汇总金额": "fgoFVymkm5pgKjQ", + "GMV编组": "fgwTPpWsXRy5q1b", + "销售额/退款额": "fg7sYpOSjx4DpAY" + } + } +} \ No newline at end of file diff --git a/prefect.yaml b/prefect.yaml new file mode 100644 index 0000000..3e67d78 --- /dev/null +++ b/prefect.yaml @@ -0,0 +1,50 @@ +name: gmv-analytics +prefect-version: "3.0.0" + +build: [] +push: [] +pull: [] + +deployments: + - name: gmv-daily-data-sync-8am + version: "1.0.0" + description: "每天上午 8 点执行 GMV 数据同步并写入飞书多维表格。" + entrypoint: prefect_gmv_scheduler.py:gmv_daily_data_sync_flow + parameters: + start_date: null + end_date: null + dry_run: false + skip_missing_tables: false + table_map_file: "" + tags: + - gmv + - feishu + - bitable + - sync + schedules: + - cron: "0 16 50 * *" + timezone: "Asia/Shanghai" + active: true + work_pool: + name: default-agent-pool + + - name: gmv-daily-notify-9am + version: "1.0.0" + description: "每天上午 9 点推送 GMV 数据同步结果到飞书群。" + entrypoint: prefect_gmv_scheduler.py:gmv_daily_notify_flow + parameters: + start_date: null + end_date: null + dry_run: false + skip_missing_tables: false + table_map_file: "" + tags: + - gmv + - feishu + - notify + schedules: + - cron: "0 17 30 * *" + timezone: "Asia/Shanghai" + active: true + work_pool: + name: default-agent-pool diff --git a/prefect_gmv_scheduler.py b/prefect_gmv_scheduler.py index ac8505e..b7ed1b0 100644 --- a/prefect_gmv_scheduler.py +++ b/prefect_gmv_scheduler.py @@ -1,4 +1,5 @@ import argparse +import json import os from dataclasses import dataclass from datetime import date @@ -42,6 +43,9 @@ from project_bitable_common import ( validate_date, ) +BASE_DIR = Path(__file__).resolve().parent +DEFAULT_PREFECT_SYNC_STATE_FILE = BASE_DIR / "prefect_sync_state.json" + @dataclass(frozen=True) class SyncConfig: @@ -58,6 +62,12 @@ class SyncConfig: skip_missing_tables: bool +@dataclass(frozen=True) +class SyncWindow: + start_date: date + end_date: date + + def load_sync_config( start_date: str | None = None, end_date: str | None = None, @@ -81,6 +91,68 @@ def load_sync_config( ) +def get_prefect_sync_state_file() -> Path: + return Path(os.getenv("PREFECT_SYNC_STATE_FILE", str(DEFAULT_PREFECT_SYNC_STATE_FILE))) + + +def load_prefect_sync_state() -> dict[str, str]: + path = get_prefect_sync_state_file() + if not path.exists(): + return {} + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except Exception as exc: + log_step(f"读取 Prefect 同步状态失败，已忽略: path={path}, error={exc}") + return {} + return {str(key): str(value) for key, value in payload.items() if value is not None} + + +def save_prefect_sync_state(state: dict[str, str]) -> None: + path = get_prefect_sync_state_file() + path.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8") + + +def resolve_sync_window( + config: SyncConfig, + explicit_start_date: str | None, + explicit_end_date: str | None, + mode: str, +) -> SyncWindow: + if explicit_start_date or explicit_end_date: + start_date_value = validate_date(explicit_start_date or config.start_date) + end_date_value = validate_date(explicit_end_date or config.end_date) + if end_date_value < start_date_value: + raise ValueError("end-date 不能早于 start-date") + return SyncWindow(start_date=start_date_value, end_date=end_date_value) + + state = load_prefect_sync_state() + configured_start_date = validate_date(config.start_date) + yesterday = validate_date(config.end_date) + + if mode == "notify": + last_sync_start = state.get("last_sync_start_date") + last_sync_end = state.get("last_sync_end_date") + if last_sync_start and last_sync_end: + start_date_value = validate_date(last_sync_start) + end_date_value = validate_date(last_sync_end) + if end_date_value < start_date_value: + raise ValueError("prefect_sync_state.json 中的日期范围无效") + return SyncWindow(start_date=start_date_value, end_date=end_date_value) + return SyncWindow(start_date=configured_start_date, end_date=yesterday) + + if state.get("bootstrap_completed") == "true": + return SyncWindow(start_date=yesterday, end_date=yesterday) + return SyncWindow(start_date=configured_start_date, end_date=yesterday) + + +def mark_sync_window_completed(window: SyncWindow) -> None: + state = load_prefect_sync_state() + state["bootstrap_completed"] = "true" + state["last_sync_start_date"] = window.start_date.isoformat() + state["last_sync_end_date"] = window.end_date.isoformat() + save_prefect_sync_state(state) + + @task(name="notify-feishu") def notify_feishu(context: SyncConfig, start_date_text: str, end_date_text: str) -> None: context_obj = build_daily_sync_context( @@ -99,8 +171,8 @@ def notify_feishu(context: SyncConfig, start_date_text: str, end_date_text: str) send_sync_notification(context_obj, validate_date(start_date_text), validate_date(end_date_text)) -@flow(name="gmv-daily-sync") -def gmv_daily_sync_flow( +@flow(name="gmv-daily-data-sync") +def gmv_daily_data_sync_flow( start_date: str | None = None, end_date: str | None = None, dry_run: bool = False, @@ -114,11 +186,7 @@ def gmv_daily_sync_flow( skip_missing_tables=skip_missing_tables, table_map_file=table_map_file, ) - - start_date_value = validate_date(config.start_date) - end_date_value = validate_date(config.end_date) - if end_date_value < start_date_value: - raise ValueError("end-date 不能早于 start-date") + sync_window = resolve_sync_window(config, start_date, end_date, mode="sync") context = build_daily_sync_context( DailySyncConfig( base_url=config.base_url, @@ -133,27 +201,50 @@ def gmv_daily_sync_flow( ) ) - log_step(f"Prefect 调度开始: {config.start_date} -> {config.end_date}") - try: - for stat_date in iter_dates(start_date_value, end_date_value): - log_step(f"Prefect 开始处理日期: {stat_date.isoformat()}") - run_daily_sync_for_date(context, stat_date) - except Exception: - notify_feishu.with_options(name="notify-feishu-failed")(config, config.start_date, config.end_date) - raise + log_step(f"Prefect 数据同步开始: {sync_window.start_date.isoformat()} -> {sync_window.end_date.isoformat()}") + for stat_date in iter_dates(sync_window.start_date, sync_window.end_date): + log_step(f"Prefect 开始处理日期: {stat_date.isoformat()}") + run_daily_sync_for_date(context, stat_date) + if not config.dry_run: + mark_sync_window_completed(sync_window) + log_step("Prefect 数据同步完成") - notify_feishu(config, config.start_date, config.end_date) - log_step("Prefect 调度完成") + +@flow(name="gmv-daily-notify") +def gmv_daily_notify_flow( + start_date: str | None = None, + end_date: str | None = None, + dry_run: bool = False, + skip_missing_tables: bool = False, + table_map_file: str = "", +) -> None: + config = load_sync_config( + start_date=start_date, + end_date=end_date, + dry_run=dry_run, + skip_missing_tables=skip_missing_tables, + table_map_file=table_map_file, + ) + sync_window = resolve_sync_window(config, start_date, end_date, mode="notify") + log_step(f"Prefect 飞书通知开始: {sync_window.start_date.isoformat()} -> {sync_window.end_date.isoformat()}") + notify_feishu(config, sync_window.start_date.isoformat(), sync_window.end_date.isoformat()) + log_step("Prefect 飞书通知完成") def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser(description="Prefect 调度：每天上午 9 点执行 GMV 数据同步。") + parser = argparse.ArgumentParser(description="Prefect 调度：8点执行数据同步，9点执行飞书通知。") parser.add_argument("--start-date", default="", help="开始日期，格式 YYYY-MM-DD，默认读取 .env 中的 SYNC_START_DATE") parser.add_argument("--end-date", default="", help="结束日期，格式 YYYY-MM-DD，默认昨天") parser.add_argument("--table-map-file", default="", help="可选，项目到表名的映射 JSON 文件路径") parser.add_argument("--dry-run", action="store_true", help="只输出将要更新的内容，不调用飞书写接口") parser.add_argument("--skip-missing-tables", action="store_true", help="找不到对应月份的飞书表时跳过") - parser.add_argument("--serve", action="store_true", help="启动 Prefect deployment，并按每天上午 9 点定时执行") + parser.add_argument("--serve", action="store_true", help="启动 Prefect deployments，按 8 点同步、9 点通知 定时执行") + parser.add_argument( + "--mode", + choices=("sync", "notify", "all"), + default="all", + help="单次执行模式：sync=只跑数据同步，notify=只发飞书通知，all=先同步再通知", + ) return parser.parse_args() @@ -167,15 +258,29 @@ def main() -> None: "table_map_file": args.table_map_file, } if args.serve: - gmv_daily_sync_flow.serve( - name="gmv-daily-sync-9am", - cron="0 9 * * *", + gmv_daily_data_sync_flow.serve( + name="gmv-daily-data-sync-8am", + cron="0 8 * * *", timezone="Asia/Shanghai", parameters=flow_kwargs, tags=["gmv", "feishu", "bitable"], ) + gmv_daily_notify_flow.serve( + name="gmv-daily-notify-9am", + cron="0 9 * * *", + timezone="Asia/Shanghai", + parameters=flow_kwargs, + tags=["gmv", "feishu", "notify"], + ) return - gmv_daily_sync_flow(**flow_kwargs) + if args.mode == "sync": + gmv_daily_data_sync_flow(**flow_kwargs) + return + if args.mode == "notify": + gmv_daily_notify_flow(**flow_kwargs) + return + gmv_daily_data_sync_flow(**flow_kwargs) + gmv_daily_notify_flow(**flow_kwargs) if __name__ == "__main__": diff --git a/project_bitable_common.py b/project_bitable_common.py index 51a52b1..a97592a 100644 --- a/project_bitable_common.py +++ b/project_bitable_common.py @@ -124,7 +124,7 @@ UNION ALL SELECT 统计日期, CASE - WHEN 业务模式 IN ('三方计算','三方分销') + WHEN 业务模式 IN ('三方结算','三方分销') THEN 原始店铺名称 || '_三方' ELSE 原始店铺名称 || '_自营' END AS 店铺名称, @@ -664,44 +664,24 @@ class FeishuBitableClient: self.request( "POST", f"/open-apis/bitable/v1/apps/{self.app_token}/tables/{table_id}/fields", - json={"field_name": field_name, "type": 2}, + json={"field_name": field_name, "type": 2, "property": {"formatter": "0.00"}}, ) self._fields_cache.pop(table_id, None) log_step(f"字段创建完成: table_id={table_id}, field={field_name}") return True - def try_list_field_groups(self, table_id: str) -> list[dict[str, Any]] | None: - path = f"/open-apis/bitable/v1/apps/{self.app_token}/tables/{table_id}/field_groups" - url = f"https://open.feishu.cn{path}" - log_debug(f"尝试读取字段编组: table_id={table_id}") - resp = self.session.request("GET", url, timeout=60) - try: - payload = resp.json() - except ValueError: - log_debug(f"字段编组接口返回非 JSON，跳过读取: table_id={table_id}, status={resp.status_code}") - return None - if resp.status_code == 404: - log_debug(f"字段编组读取接口不可用: table_id={table_id}, payload={payload}") - return None - if resp.status_code >= 400 or payload.get("code") != 0: - log_debug(f"字段编组读取失败，跳过比对: table_id={table_id}, payload={payload}") - return None - data = payload.get("data") or {} - items = data.get("items") or data.get("field_groups") or [] - log_debug(f"字段编组读取成功: table_id={table_id}, 编组数={len(items)}") - return items - - def create_field_groups(self, table_id: str, field_groups: list[dict[str, Any]]) -> None: + def create_field_groups(self, table_id: str, field_groups: list[dict[str, Any]]) -> dict[str, Any]: log_step( f"开始创建字段编组: table_id={table_id}, " f"groups={[(group.get('name') or '', len(group.get('children') or [])) for group in field_groups]}" ) - self.request( + data = self.request( "POST", f"/open-apis/bitable/v1/apps/{self.app_token}/tables/{table_id}/field_groups", json={"field_groups": field_groups}, ) log_step(f"字段编组创建请求完成: table_id={table_id}") + return data def list_records(self, table_id: str) -> list[dict[str, Any]]: if table_id in self._records_cache: @@ -1000,6 +980,7 @@ def build_field_group_payload( store_field_name: str, allowed_field_names: set[str] | None = None, ) -> list[dict[str, Any]]: + summary_fields: list[dict[str, str]] = [] gmv_fields: list[dict[str, str]] = [] other_fields: list[dict[str, str]] = [] for field in fields: @@ -1012,11 +993,16 @@ def build_field_group_payload( if allowed_field_names is not None and field_name not in allowed_field_names: continue child = {"type": "field", "id": field_id} + if field_name == get_summary_gmv_field_name(): + summary_fields.append(child) + continue if "gmv" in field_name.lower(): gmv_fields.append(child) else: other_fields.append(child) groups: list[dict[str, Any]] = [] + if summary_fields: + groups.append({"name": "汇总金额", "children": summary_fields}) if gmv_fields: groups.append({"name": "GMV编组", "children": gmv_fields}) if other_fields: @@ -1024,22 +1010,13 @@ def build_field_group_payload( return groups -def normalize_field_groups(field_groups: list[dict[str, Any]]) -> dict[str, set[str]]: - normalized: dict[str, set[str]] = {} - for group in field_groups: - name = str(group.get("name") or "").strip() - if not name: - continue - normalized[name] = { - str(child.get("id") or "").strip() - for child in (group.get("children") or []) - if str(child.get("id") or "").strip() - } - return normalized - - -def get_field_group_order(field_groups: list[dict[str, Any]]) -> list[str]: - return [str(group.get("name") or "").strip() for group in field_groups if str(group.get("name") or "").strip()] +def get_incremental_field_names_for_day(stat_date: date) -> set[str]: + return { + get_summary_gmv_field_name(), + get_gmv_field_name(stat_date), + get_sales_field_name(stat_date), + get_refund_field_name(stat_date), + } def ensure_field_groups_for_table( @@ -1057,21 +1034,6 @@ def ensure_field_groups_for_table( return desired_summary = {group["name"]: len(group.get("children") or []) for group in desired_groups} log_step(f"目标字段编组: 表={table.name}, summary={desired_summary}") - existing_groups = client.try_list_field_groups(table.table_id) - if existing_groups is not None: - existing_normalized = normalize_field_groups(existing_groups) - desired_normalized = normalize_field_groups(desired_groups) - existing_order = get_field_group_order(existing_groups) - desired_order = get_field_group_order(desired_groups) - if existing_normalized == desired_normalized and existing_order == desired_order: - log_step(f"字段编组已存在且一致，跳过创建: 表={table.name}") - return - log_debug( - f"字段编组存在差异: 表={table.name}, " - f"existing={ {name: len(ids) for name, ids in existing_normalized.items()} }, " - f"desired={ {name: len(ids) for name, ids in desired_normalized.items()} }, " - f"existing_order={existing_order}, desired_order={desired_order}" - ) if dry_run: print(f"[DRY-RUN] 表={table.name} 将创建/更新字段编组: {desired_summary}") return @@ -1087,6 +1049,12 @@ def ensure_field_groups_for_table( "原因=应用缺少 base:field_group:create 权限，请在飞书开放平台开通后重试" ) return + if "1254117" in error_text or "The name of field group is duplicated" in error_text: + log_step( + f"跳过字段编组创建: 表={table.name}, " + "原因=字段编组已存在同名分组，按幂等场景处理" + ) + return if "1254122" in error_text or "Child belongs to multiple field groups" in error_text: log_step(f"检测到字段已被其他编组占用，尝试逐个创建编组: 表={table.name}") for group in desired_groups: @@ -1143,6 +1111,31 @@ def sync_field_groups_for_range( log_step("字段编组同步完成") +def rebuild_field_groups_for_range_with_context( + context: DailySyncContext, + start_date: date, + end_date: date, +) -> None: + log_step(f"开始重建字段编组: {start_date.isoformat()} -> {end_date.isoformat()}") + for (year, month), _ in sorted(build_group_dates_by_month(start_date, end_date).items()): + for project in SUPPORTED_PROJECTS: + try: + table = find_target_table(context.tables, project, year, month, context.table_overrides) + except ValueError as exc: + if context.config.skip_missing_tables: + log_step(f"跳过字段编组重建: {exc}") + continue + raise + ensure_field_groups_for_table( + client=context.client, + table=table, + store_field_name=context.config.store_field_name, + allowed_field_names=None, + dry_run=context.config.dry_run, + ) + log_step("字段编组重建完成") + + def sync_field_groups_for_day_with_context( context: DailySyncContext, stat_date: date, @@ -1155,11 +1148,7 @@ def sync_field_groups_for_day_with_context( if allowed_field_names_override is not None: allowed_field_names = set(allowed_field_names_override) elif not force_full_rebuild and stat_date.day != 1: - allowed_field_names = { - get_gmv_field_name(stat_date), - get_sales_field_name(stat_date), - get_refund_field_name(stat_date), - } + allowed_field_names = get_incremental_field_names_for_day(stat_date) log_step(f"增量编组目标字段: 日期={stat_date.isoformat()}, fields={sorted(allowed_field_names)}") for project in SUPPORTED_PROJECTS: try: @@ -1355,19 +1344,24 @@ def get_gmv_field_name(stat_date: date) -> str: return stat_date.strftime("%m-%d GMV") +def get_summary_gmv_field_name() -> str: + return "汇总GMV" + + def sync_gmv_table_for_day( client: FeishuBitableClient, table: TableInfo, stat_date: date, preferred_store_field_name: str, dry_run: bool, -) -> bool: +) -> set[str]: log_step(f"开始计算 GMV: 表={table.name}, 日期={stat_date.isoformat()}") fields = client.list_fields(table.table_id) store_field_name = resolve_store_field_name(fields, preferred_store_field_name) sales_field_name = get_sales_field_name(stat_date) refund_field_name = get_refund_field_name(stat_date) gmv_field_name = get_gmv_field_name(stat_date) + summary_gmv_field_name = get_summary_gmv_field_name() field_names = [str(item.get("field_name") or "") for item in fields] log_debug( f"GMV 字段检查: 表={table.name}, 店铺字段={store_field_name}, 销售字段={sales_field_name in field_names}, " @@ -1375,10 +1369,20 @@ def sync_gmv_table_for_day( ) if sales_field_name not in field_names or refund_field_name not in field_names: log_step(f"跳过 GMV 计算: 表={table.name}, 原因=缺少销售额或退款额字段") - return False - created_field = False + return set() + created_fields: set[str] = set() if not dry_run: - created_field = client.ensure_number_field(table.table_id, gmv_field_name) + if client.ensure_number_field(table.table_id, gmv_field_name): + created_fields.add(gmv_field_name) + if client.ensure_number_field(table.table_id, summary_gmv_field_name): + created_fields.add(summary_gmv_field_name) + fields = client.list_fields(table.table_id) + field_names = [str(item.get("field_name") or "") for item in fields] + month_gmv_field_names = sorted( + field_name + for field_name in field_names + if field_name.endswith(" GMV") and field_name != summary_gmv_field_name + ) records = client.list_records(table.table_id) preview_stores = [] updated_count = 0 @@ -1398,18 +1402,27 @@ def sync_gmv_table_for_day( skipped_empty_count += 1 continue gmv_value = round((sales_value or 0.0) - (refund_value or 0.0), 2) + summary_gmv_value = round( + sum(parse_number(record_fields.get(name)) or 0.0 for name in month_gmv_field_names if name != gmv_field_name) + gmv_value, + 2, + ) if dry_run: print( f"[DRY-RUN] 表={table.name} 店铺={store_name} 字段={gmv_field_name} " f"销售额={sales_value if sales_value is not None else ''} " - f"退款额={refund_value if refund_value is not None else ''} GMV={gmv_value}" + f"退款额={refund_value if refund_value is not None else ''} GMV={gmv_value} " + f"{summary_gmv_field_name}={summary_gmv_value}" ) else: - client.update_record(table.table_id, record_id, {gmv_field_name: gmv_value}) + client.update_record( + table.table_id, + record_id, + {gmv_field_name: gmv_value, summary_gmv_field_name: summary_gmv_value}, + ) updated_count += 1 log_debug(f"飞书记录店铺预览: 表={table.name}, stores={preview_list(preview_stores)}") log_step(f"GMV 计算完成: 表={table.name}, 更新记录数={updated_count}, 双空跳过={skipped_empty_count}") - return created_field + return created_fields def run_gmv_sync_with_context( @@ -1434,8 +1447,7 @@ def run_gmv_sync_with_context( preferred_store_field_name=context.config.store_field_name, dry_run=context.config.dry_run, ) - if created: - created_fields.add(get_gmv_field_name(stat_date)) + created_fields.update(created) return MetricSyncArtifacts(created_fields=created_fields) @@ -1460,6 +1472,17 @@ def run_gmv_sync( run_gmv_sync_with_context(context, start_date, end_date) +def run_daily_data_sync_for_date(context: DailySyncContext, stat_date: date) -> None: + log_step(f"开始处理日期: {stat_date.isoformat()}") + created_fields: set[str] = set() + for metric_name in ("sales", "refund"): + artifacts = run_metric_sync_with_context(context, metric_name, stat_date, stat_date) + created_fields.update(artifacts.created_fields) + gmv_artifacts = run_gmv_sync_with_context(context, stat_date, stat_date) + created_fields.update(gmv_artifacts.created_fields) + log_step(f"日期处理完成: {stat_date.isoformat()}") + + def run_daily_sync_for_date(context: DailySyncContext, stat_date: date) -> None: log_step(f"开始处理日期: {stat_date.isoformat()}") created_fields: set[str] = set() @@ -1471,9 +1494,13 @@ def run_daily_sync_for_date(context: DailySyncContext, stat_date: date) -> None: should_sync_field_groups = stat_date.day == 1 or bool(created_fields) if should_sync_field_groups: + if stat_date.day == 1: + sync_reason = "每月1号完整编组" + else: + sync_reason = f"新增字段={sorted(created_fields)}" log_step( f"触发字段编组同步: 日期={stat_date.isoformat()}, " - f"原因={'每月1号完整编组' if stat_date.day == 1 else f'新增字段={sorted(created_fields)}'}" + f"原因={sync_reason}" ) sync_field_groups_for_day_with_context( context=context, diff --git a/店铺映射表.xlsx b/店铺映射表.xlsx index b30ff3e..46b3dcf 100644 Binary files a/店铺映射表.xlsx and b/店铺映射表.xlsx differ