调整表结构

project_bitable_common.py

@@ -3,7 +3,7 @@ import os
 import re
 from collections import defaultdict
 from dataclasses import dataclass
-from datetime import date, datetime, timedelta
+from datetime import date, datetime, timedelta, timezone
 from decimal import Decimal
 from pathlib import Path
 from typing import Any
@@ -19,7 +19,7 @@ from logging_config import get_logger, setup_logging
 BASE_DIR = Path(__file__).resolve().parent
 ENV_FILE = BASE_DIR / ".env"
 DEFAULT_MAPPING_FILE = BASE_DIR / "店铺映射表.xlsx"
-DEFAULT_START_DATE = "2026-02-25"
+DEFAULT_START_DATE = "2026-04-01"
 DEFAULT_STORE_FIELD_NAME = "店铺名称"
 STORE_FIELD_CANDIDATES = ["业务店铺名称", "店铺名称", "门店名称", "店铺", "名称"]
 SUPPORTED_PROJECTS = ["大眼睛", "羽心堂", "MMMJ", "EM", "春回盏", "BROKEN ZOO"]
@@ -717,6 +717,42 @@ class FeishuBitableClient:
         log_step(f"字段创建完成: table_id={table_id}, field={field_name}")
         return True
 
+    def ensure_text_field(self, table_id: str, field_name: str) -> bool:
+        log_debug(f"检查文本字段是否存在: table_id={table_id}, field={field_name}")
+        fields = self.list_fields(table_id)
+        if any(item.get("field_name") == field_name for item in fields):
+            log_debug(f"文本字段已存在，跳过创建: table_id={table_id}, field={field_name}")
+            return False
+        log_step(f"开始创建文本字段: table_id={table_id}, field={field_name}")
+        self.request(
+            "POST",
+            f"/open-apis/bitable/v1/apps/{self.app_token}/tables/{table_id}/fields",
+            json={"field_name": field_name, "type": 1},
+        )
+        self._fields_cache.pop(table_id, None)
+        log_step(f"文本字段创建完成: table_id={table_id}, field={field_name}")
+        return True
+
+    def ensure_date_field(self, table_id: str, field_name: str) -> bool:
+        log_debug(f"检查日期字段是否存在: table_id={table_id}, field={field_name}")
+        fields = self.list_fields(table_id)
+        if any(item.get("field_name") == field_name for item in fields):
+            log_debug(f"日期字段已存在，跳过创建: table_id={table_id}, field={field_name}")
+            return False
+        log_step(f"开始创建日期字段: table_id={table_id}, field={field_name}")
+        self.request(
+            "POST",
+            f"/open-apis/bitable/v1/apps/{self.app_token}/tables/{table_id}/fields",
+            json={
+                "field_name": field_name,
+                "type": 5,
+                "property": {"date_formatter": "yyyy-MM-dd", "auto_fill": False},
+            },
+        )
+        self._fields_cache.pop(table_id, None)
+        log_step(f"日期字段创建完成: table_id={table_id}, field={field_name}")
+        return True
+
     def create_field_groups(self, table_id: str, field_groups: list[dict[str, Any]]) -> dict[str, Any]:
         log_step(
             f"开始创建字段编组: table_id={table_id}, "
@@ -775,6 +811,18 @@ class FeishuBitableClient:
                     item.setdefault("fields", {}).update(fields)
                     break
 
+    def create_record(self, table_id: str, fields: dict[str, Any]) -> dict[str, Any]:
+        log_debug(f"新增记录: table_id={table_id}, fields={list(fields.keys())}")
+        data = self.request(
+            "POST",
+            f"/open-apis/bitable/v1/apps/{self.app_token}/tables/{table_id}/records",
+            json={"fields": fields},
+        )
+        record = data.get("record") or {}
+        if table_id in self._records_cache and record:
+            self._records_cache[table_id].append(record)
+        return record
+
 
 def load_table_name_overrides(path_text: str) -> dict[str, str]:
     if not path_text:
@@ -1531,33 +1579,117 @@ def run_daily_data_sync_for_date(context: DailySyncContext, stat_date: date) ->
 
 
 def run_daily_sync_for_date(context: DailySyncContext, stat_date: date) -> None:
-    log_step(f"开始处理日期: {stat_date.isoformat()}")
+    log_step(f"开始处理日期(旧流程+新表结构): {stat_date.isoformat()}")
-    created_fields: set[str] = set()
+    sales_rows = fetch_metric_rows(stat_date, stat_date, METRIC_CONFIGS["sales"])
-    for metric_name in ("sales", "refund"):
+    refund_rows = fetch_metric_rows(stat_date, stat_date, METRIC_CONFIGS["refund"])
-        artifacts = run_metric_sync_with_context(context, metric_name, stat_date, stat_date)
+    grouped_sales = build_project_metrics(sales_rows, context.mapping, METRIC_CONFIGS["sales"])
-        created_fields.update(artifacts.created_fields)
+    grouped_refund = build_project_metrics(refund_rows, context.mapping, METRIC_CONFIGS["refund"])
-    gmv_artifacts = run_gmv_sync_with_context(context, stat_date, stat_date)
-    created_fields.update(gmv_artifacts.created_fields)
 
-    should_sync_field_groups = stat_date.day == 1 or bool(created_fields)
+    date_text = stat_date.isoformat()
-    if should_sync_field_groups:
+    shanghai_tz = timezone(timedelta(hours=8))
-        if stat_date.day == 1:
+    date_value_ms = int(datetime(stat_date.year, stat_date.month, stat_date.day, tzinfo=shanghai_tz).timestamp() * 1000)
-            sync_reason = "每月1号完整编组"
+    header_date = "日期"
-        else:
+    header_metric = "指标"
-            sync_reason = f"新增字段={sorted(created_fields)}"
+    metric_labels = (
-        log_step(
+        ("销售金额", "sales"),
-            f"触发字段编组同步: 日期={stat_date.isoformat()}, "
+        ("退款金额", "refund"),
-            f"原因={sync_reason}"
+        ("GMV", "gmv"),
+        ("汇总GMV", "summary_gmv"),
+    )
+
+    for project in SUPPORTED_PROJECTS:
+        month_key = (stat_date.year, stat_date.month)
+        sales_by_store = grouped_sales.get(project, {}).get(month_key, {}).get(stat_date, {})
+        refund_by_store = grouped_refund.get(project, {}).get(month_key, {}).get(stat_date, {})
+        mapped_project_stores = sorted(
+            store_name for store_name in set(context.mapping.values()) if classify_project(store_name) == project
         )
-        sync_field_groups_for_day_with_context(
+        all_stores = sorted(set(mapped_project_stores) | set(sales_by_store.keys()) | set(refund_by_store.keys()))
-            context=context,
+        gmv_by_store = {
-            stat_date=stat_date,
+            store_name: sales_by_store.get(store_name, Decimal("0")) - refund_by_store.get(store_name, Decimal("0"))
-            force_full_rebuild=stat_date.day == 1,
+            for store_name in all_stores
-            allowed_field_names_override=None if stat_date.day == 1 else created_fields,
+        }
-        )
+        metric_store_map: dict[str, dict[str, float]] = {
-    else:
+            "sales": {store_name: decimal_to_number(amount) for store_name, amount in sales_by_store.items()},
-        log_step(f"跳过字段编组同步: 日期={stat_date.isoformat()}, 原因=当天无新增字段")
+            "refund": {store_name: decimal_to_number(amount) for store_name, amount in refund_by_store.items()},
-    log_step(f"日期处理完成: {stat_date.isoformat()}")
+            "gmv": {store_name: decimal_to_number(amount) for store_name, amount in gmv_by_store.items()},
+            "summary_gmv": {},
+        }
+        try:
+            table = find_target_table(context.tables, project, stat_date.year, stat_date.month, context.table_overrides)
+        except ValueError as exc:
+            if context.config.skip_missing_tables:
+                log_step(f"跳过: {exc}")
+                continue
+            raise
+        if context.config.dry_run:
+            for metric_label, metric_key in metric_labels:
+                print(
+                    f"[DRY-RUN] 表={table.name} 日期={date_text} 指标={metric_label} "
+                    f"店铺数={len(metric_store_map[metric_key])}"
+                )
+            continue
+
+        context.client.ensure_date_field(table.table_id, header_date)
+        context.client.ensure_text_field(table.table_id, header_metric)
+        for store_name in all_stores:
+            context.client.ensure_number_field(table.table_id, store_name)
+        records = context.client.list_records(table.table_id)
+        record_index: dict[tuple[str, str], str] = {}
+        summary_record_id: str | None = None
+        for item in records:
+            fields = item.get("fields", {})
+            row_date_value = fields.get(header_date)
+            row_date = ""
+            if isinstance(row_date_value, (int, float)):
+                row_date = datetime.fromtimestamp(float(row_date_value) / 1000).date().isoformat()
+            elif isinstance(row_date_value, str):
+                row_date = row_date_value.strip()
+            row_metric = str(fields.get(header_metric) or "").strip()
+            if row_metric == "汇总GMV" and not summary_record_id:
+                summary_record_id = item["record_id"]
+            if row_date and row_metric:
+                record_index[(row_date, row_metric)] = item["record_id"]
+
+        for metric_label, metric_key in metric_labels:
+            row_fields = {
+                header_date: date_value_ms,
+                header_metric: metric_label,
+            }
+            row_fields.update(metric_store_map[metric_key])
+            if metric_label == "汇总GMV":
+                summary_gmv_values: dict[str, float] = {}
+                for item in records:
+                    fields = item.get("fields", {})
+                    row_metric = str(fields.get(header_metric) or "").strip()
+                    if row_metric != "GMV":
+                        continue
+                    row_date_value = fields.get(header_date)
+                    row_date: date | None = None
+                    if isinstance(row_date_value, (int, float)):
+                        row_date = datetime.fromtimestamp(float(row_date_value) / 1000).date()
+                    elif isinstance(row_date_value, str):
+                        try:
+                            row_date = validate_date(row_date_value.strip())
+                        except ValueError:
+                            row_date = None
+                    if row_date is None:
+                        continue
+                    if row_date.year != stat_date.year or row_date.month != stat_date.month or row_date > stat_date:
+                        continue
+                    for store_name in all_stores:
+                        store_amount = parse_number(fields.get(store_name)) or 0.0
+                        summary_gmv_values[store_name] = round(summary_gmv_values.get(store_name, 0.0) + store_amount, 2)
+                row_fields.update(summary_gmv_values)
+                record_id = summary_record_id
+            else:
+                record_id = record_index.get((date_text, metric_label))
+            if record_id:
+                context.client.update_record(table.table_id, record_id, row_fields)
+            else:
+                context.client.create_record(table.table_id, row_fields)
+        log_step(f"表处理完成: {table.name}, 日期={date_text}, 指标行=销售金额/退款金额/GMV，汇总GMV=月内GMV累加")
+    log_step(f"日期处理完成(旧流程+新表结构): {stat_date.isoformat()}")
 
 
 def send_sync_notification(