diff --git a/__pycache__/project_bitable_common.cpython-312.pyc b/__pycache__/project_bitable_common.cpython-312.pyc index 695cb01..47c2077 100644 Binary files a/__pycache__/project_bitable_common.cpython-312.pyc and b/__pycache__/project_bitable_common.cpython-312.pyc differ diff --git a/project_bitable_common.py b/project_bitable_common.py index 4fc8197..f1bff57 100644 --- a/project_bitable_common.py +++ b/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()}") - 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()}") + sales_rows = fetch_metric_rows(stat_date, stat_date, METRIC_CONFIGS["sales"]) + refund_rows = fetch_metric_rows(stat_date, stat_date, METRIC_CONFIGS["refund"]) + grouped_sales = build_project_metrics(sales_rows, context.mapping, METRIC_CONFIGS["sales"]) + grouped_refund = build_project_metrics(refund_rows, context.mapping, METRIC_CONFIGS["refund"]) - 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"原因={sync_reason}" + date_text = stat_date.isoformat() + shanghai_tz = timezone(timedelta(hours=8)) + date_value_ms = int(datetime(stat_date.year, stat_date.month, stat_date.day, tzinfo=shanghai_tz).timestamp() * 1000) + header_date = "日期" + header_metric = "指标" + metric_labels = ( + ("销售金额", "sales"), + ("退款金额", "refund"), + ("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( - context=context, - stat_date=stat_date, - force_full_rebuild=stat_date.day == 1, - allowed_field_names_override=None if stat_date.day == 1 else created_fields, - ) - else: - log_step(f"跳过字段编组同步: 日期={stat_date.isoformat()}, 原因=当天无新增字段") - log_step(f"日期处理完成: {stat_date.isoformat()}") + all_stores = sorted(set(mapped_project_stores) | set(sales_by_store.keys()) | set(refund_by_store.keys())) + gmv_by_store = { + store_name: sales_by_store.get(store_name, Decimal("0")) - refund_by_store.get(store_name, Decimal("0")) + for store_name in all_stores + } + metric_store_map: dict[str, dict[str, float]] = { + "sales": {store_name: decimal_to_number(amount) for store_name, amount in sales_by_store.items()}, + "refund": {store_name: decimal_to_number(amount) for store_name, amount in refund_by_store.items()}, + "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(