import json import os import re from collections import defaultdict from dataclasses import dataclass from datetime import date, datetime, timedelta from decimal import Decimal from pathlib import Path from typing import Any from urllib.parse import urlparse import psycopg2 import requests from openpyxl import load_workbook 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_STORE_FIELD_NAME = "店铺名称" STORE_FIELD_CANDIDATES = ["业务店铺名称", "店铺名称", "门店名称", "店铺", "名称"] SUPPORTED_PROJECTS = ["大眼睛", "羽心堂", "MMMJ", "EM", "春回盏", "BROKEN ZOO"] @dataclass(frozen=True) class MetricConfig: name: str description: str sql_template: str field_suffix: str METRIC_CONFIGS = { "sales": MetricConfig( name="sales", description="销售额", field_suffix="总销售额", sql_template=""" SET TIMEZONE = 'Asia/Shanghai'; WITH base_data AS ( SELECT f.支付时间::DATE AS 统计日期, f.店铺主键, s.店铺名称 AS 原始店铺名称, p.品牌名称, f.业务模式, f.实付分摊金额 FROM analytics_layer.fact_sales f LEFT JOIN analytics_layer.dim_shop s ON f.店铺主键 = s.店铺主键 LEFT JOIN analytics_layer.dim_product p ON f.商品主键 = p.商品主键 AND f.统一规格编码 = p.统一规格编码 WHERE f.支付时间 >= %(start_ts)s AND f.支付时间 < %(end_ts_exclusive)s ), brand_split AS ( SELECT 统计日期, 店铺主键, 原始店铺名称, 业务模式, CASE WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE '羽心堂' THEN 原始店铺名称 || '_羽心堂' WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE '春回盏' THEN 原始店铺名称 || '_春回盏' WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'Mama Mujer' THEN 原始店铺名称 || '_Mama Mujer' WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'ERBA MIGINA' THEN 原始店铺名称 || '_ERBA MIGINA' WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'BROKEN ZOO' THEN 原始店铺名称 || '_BROKEN ZOO' WHEN 店铺主键 IN (1,28,29) THEN 原始店铺名称 || '_其他' WHEN 店铺主键 IN (3,9) AND 品牌名称 LIKE '羽心堂' THEN 原始店铺名称 || '_羽心堂' WHEN 店铺主键 IN (3,9) AND 品牌名称 LIKE 'Mama Mujer' THEN 原始店铺名称 || '_Mama Mujer' WHEN 店铺主键 IN (3,9) THEN 原始店铺名称 || '_其他' WHEN 店铺主键 IS NULL OR 原始店铺名称 IS NULL THEN '分销/SEEDING' ELSE 原始店铺名称 END AS 店铺名称, 实付分摊金额 FROM base_data ), shop_brand_detail AS ( SELECT 统计日期, 店铺主键, 店铺名称, SUM(实付分摊金额) AS 总销售额 FROM brand_split GROUP BY 1,2,3 ) SELECT 统计日期, 店铺名称, 总销售额 FROM shop_brand_detail WHERE NOT ( 店铺主键 IN (1,9,28,29) AND 店铺名称 LIKE '%%_其他' ) UNION ALL -- 店铺合计（使用 dim_shop 店铺名） SELECT 统计日期, 原始店铺名称 AS 店铺名称, SUM(实付分摊金额) FROM base_data WHERE 店铺主键 IN (1,28,29,3) GROUP BY 1,2 UNION ALL -- 权益金（充值购物金单独统计） SELECT 统计日期, 原始店铺名称 || '_权益金' AS 店铺名称, SUM(实付分摊金额) AS 总销售额 FROM base_data WHERE 店铺主键 IN (5,11,14,22,26) AND 业务模式 = '充值购物金' GROUP BY 1,2 ORDER BY 统计日期 DESC, 总销售额 DESC; """, ), "refund": MetricConfig( name="refund", description="退款额", field_suffix="退款额", sql_template=""" SET TIMEZONE = 'Asia/Shanghai'; WITH base_data AS ( SELECT f.完结时间::DATE AS 统计日期, f.店铺主键, f.平台订单号, s.店铺名称 AS 原始店铺名称, p.品牌名称, f.退款金额 FROM analytics_layer.fact_returns f LEFT JOIN analytics_layer.dim_shop s ON f.店铺主键 = s.店铺主键 LEFT JOIN analytics_layer.dim_product p ON f.商品主键 = p.商品主键 AND f.统一规格编码 = p.统一规格编码 WHERE f.完结时间 >= %(start_ts)s AND f.完结时间 < %(end_ts_exclusive)s ), recharge_orders AS ( SELECT DISTINCT 平台订单号, 店铺主键 FROM analytics_layer.fact_sales WHERE 业务模式 = '充值购物金' ), brand_split AS ( SELECT 统计日期, 店铺主键, 原始店铺名称, CASE WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE '羽心堂' THEN 原始店铺名称 || '_羽心堂' WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE '春回盏' THEN 原始店铺名称 || '_春回盏' WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'Mama Mujer' THEN 原始店铺名称 || '_Mama Mujer' WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'ERBA MIGINA' THEN 原始店铺名称 || '_ERBA MIGINA' WHEN 店铺主键 IN (1,28,29) AND 品牌名称 LIKE 'BROKEN ZOO' THEN 原始店铺名称 || '_BROKEN ZOO' WHEN 店铺主键 IN (1,28,29) THEN 原始店铺名称 || '_其他' WHEN 店铺主键 IN (3,9) AND 品牌名称 LIKE '羽心堂' THEN 原始店铺名称 || '_羽心堂' WHEN 店铺主键 IN (3,9) AND 品牌名称 LIKE 'Mama Mujer' THEN 原始店铺名称 || '_Mama Mujer' WHEN 店铺主键 IN (3,9) THEN 原始店铺名称 || '_其他' WHEN 店铺主键 IS NULL OR 原始店铺名称 IS NULL THEN '分销/SEEDING' ELSE 原始店铺名称 END AS 店铺名称, 退款金额 FROM base_data ), shop_brand_detail AS ( SELECT 统计日期, 店铺主键, 店铺名称, SUM(退款金额) AS 总退款金额 FROM brand_split GROUP BY 1,2,3 ) SELECT 统计日期, 店铺名称, 总退款金额 FROM shop_brand_detail WHERE NOT ( 店铺主键 IN (1,9,28,29) AND 店铺名称 LIKE '%%_其他' ) UNION ALL -- 店铺合计 SELECT 统计日期, 原始店铺名称 AS 店铺名称, SUM(退款金额) AS 总退款金额 FROM base_data WHERE 店铺主键 IN (1,28,29,3) GROUP BY 1,2 UNION ALL -- 权益金退款 SELECT r.统计日期, r.原始店铺名称 || '_权益金' AS 店铺名称, SUM(r.退款金额) AS 总退款金额 FROM base_data r JOIN recharge_orders ro ON r.平台订单号 = ro.平台订单号 AND r.店铺主键 = ro.店铺主键 WHERE r.店铺主键 IN (5,11,14,22,26) GROUP BY 1,2 ORDER BY 统计日期 DESC, 总退款金额 DESC; """, ), } @dataclass(frozen=True) class MetricRow: stat_date: date bi_store_name: str amount: Decimal @dataclass(frozen=True) class TableInfo: table_id: str name: str @dataclass(frozen=True) class SyncResult: sql_only_stores: list[str] feishu_only_stores: list[str] @dataclass(frozen=True) class MetricSyncArtifacts: created_fields: set[str] @dataclass(frozen=True) class DailySyncConfig: base_url: str mapping_file: Path store_field_name: str table_map_file: str dry_run: bool skip_missing_tables: bool chat_id: str = "" report_image_path: str = "" group_webhook: str = "" @dataclass class DailySyncContext: client: "FeishuBitableClient" tables: list["TableInfo"] mapping: dict[str, str] table_overrides: dict[str, str] config: DailySyncConfig def log_step(message: str) -> None: get_logger().info(message) def log_debug(message: str) -> None: get_logger().debug(message) def preview_list(values: list[Any], limit: int = 20) -> str: if len(values) <= limit: return str(values) return f"{values[:limit]} ... 共{len(values)}项" def load_dotenv(env_file: Path = ENV_FILE) -> None: if not env_file.exists(): log_step(f"未找到 .env 文件，跳过加载: {env_file}") return log_step(f"加载环境变量: {env_file}") for raw_line in env_file.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) os.environ.setdefault(key.strip(), value.strip()) setup_logging(force=True) def get_base_url() -> str: base_url = os.getenv("FEISHU_BASE_URL") if not base_url: raise ValueError("缺少环境变量 FEISHU_BASE_URL") return base_url def get_mapping_file() -> Path: return Path(os.getenv("PROJECT_MAPPING_FILE", str(DEFAULT_MAPPING_FILE))) def get_store_field_name() -> str: return os.getenv("FEISHU_STORE_FIELD_NAME", DEFAULT_STORE_FIELD_NAME) def get_default_start_date() -> str: return os.getenv("SYNC_START_DATE", DEFAULT_START_DATE) def get_default_end_date() -> str: return (date.today() - timedelta(days=1)).isoformat() def get_group_webhook_url() -> str: candidates = [ "FEISHU_BOT_WEBHOOK", "FEISHU_ROBOT_WEBHOOK", "FEISHU_WEBHOOK", "LARK_BOT_WEBHOOK", "LARK_ROBOT_WEBHOOK", "GROUP_WEBHOOK", ] for key in candidates: value = os.getenv(key) if value: return value return "" def get_feishu_chat_id() -> str: return os.getenv("FEISHU_CHAT_ID", "") def get_feishu_report_image_path() -> str: return os.getenv("FEISHU_REPORT_IMAGE_PATH", "") def resolve_store_field_name(fields: list[dict[str, Any]], preferred_field_name: str) -> str: field_names = [str(item.get("field_name") or "") for item in fields] if preferred_field_name in field_names: return preferred_field_name for candidate in STORE_FIELD_CANDIDATES: if candidate in field_names: return candidate raise ValueError(f"未找到店铺名称字段: preferred={preferred_field_name}, 可用字段={field_names}") def validate_date(value: str) -> date: return datetime.strptime(value, "%Y-%m-%d").date() def normalize_text(value: str) -> str: return re.sub(r"[\s_\-（）()]+", "", value or "").lower() def classify_project(store_name: str) -> str: name = store_name or "" prioritized_rules = [ ("MMMJ", ["Mama", "mmmj", "MMMJ"]), ("羽心堂", ["羽心堂"]), ("EM", ["EM", "ERBA MIGINA"]), ("春回盏", ["春回盏"]), ("BROKEN ZOO", ["BROKEN ZOO"]), ] for project, keywords in prioritized_rules: if any(keyword in name for keyword in keywords): return project if "大眼睛" in name or "大西瓜籽妈妈" in name: return "大眼睛" return "" def get_project_table_aliases(project: str) -> list[str]: aliases = { "大眼睛": ["大眼睛", "大眼睛买买买"], "羽心堂": ["羽心堂"], "MMMJ": ["MMMJ", "Mama Mujer", "MAMA MUJER"], "EM": ["EM", "ERBA MIGINA"], "春回盏": ["春回盏"], "BROKEN ZOO": ["BROKEN ZOO", "BROKENZOO", "葆福颂"], } return aliases.get(project, [project]) def load_store_mapping(path: Path) -> dict[str, str]: log_step(f"加载店铺映射表: {path}") wb = load_workbook(path, data_only=True) ws = wb.active mapping: dict[str, str] = {} for row in ws.iter_rows(min_row=2, values_only=True): if not row: continue if len(row) >= 3: _, business_name, bi_name = row[:3] elif len(row) >= 2: business_name, bi_name = row[:2] else: continue if not business_name or not bi_name: continue mapping[str(bi_name).strip()] = str(business_name).strip() log_step(f"店铺映射加载完成: {len(mapping)} 条") return mapping def resolve_business_store_name(bi_store_name: str, store_mapping: dict[str, str]) -> str: return store_mapping.get(bi_store_name, "") def get_db_connection() -> psycopg2.extensions.connection: required = [ "ANALYTICS_DB_HOST", "ANALYTICS_DB_PORT", "ANALYTICS_DB_USER", "ANALYTICS_DB_PASSWORD", "ANALYTICS_DB_NAME", ] missing = [key for key in required if not os.getenv(key)] if missing: raise ValueError(f"缺少数据库环境变量: {', '.join(missing)}") log_step( f"连接 PostgreSQL: host={os.environ['ANALYTICS_DB_HOST']} port={os.environ['ANALYTICS_DB_PORT']} db={os.environ['ANALYTICS_DB_NAME']}" ) return psycopg2.connect( host=os.environ["ANALYTICS_DB_HOST"], port=os.environ["ANALYTICS_DB_PORT"], user=os.environ["ANALYTICS_DB_USER"], password=os.environ["ANALYTICS_DB_PASSWORD"], dbname=os.environ["ANALYTICS_DB_NAME"], ) def fetch_metric_rows(start_date: date, end_date: date, metric_config: MetricConfig) -> list[MetricRow]: start_ts = f"{start_date.isoformat()} 00:00:00" end_ts_exclusive = f"{(end_date + timedelta(days=1)).isoformat()} 00:00:00" log_step(f"开始查询{metric_config.description}: {start_ts} -> {end_ts_exclusive} (右开区间)") with get_db_connection() as conn: with conn.cursor() as cur: cur.execute(metric_config.sql_template, {"start_ts": start_ts, "end_ts_exclusive": end_ts_exclusive}) rows = cur.fetchall() log_step(f"{metric_config.description}查询完成: {len(rows)} 行") return [ MetricRow(stat_date=row[0], bi_store_name=(row[1] or "").strip(), amount=row[2] or Decimal("0")) for row in rows ] def build_project_metrics( rows: list[MetricRow], store_mapping: dict[str, str], metric_config: MetricConfig, ) -> dict[str, dict[tuple[int, int], dict[date, dict[str, Decimal]]]]: log_step(f"开始按项目/月/日期聚合{metric_config.description}") grouped: dict[str, dict[tuple[int, int], dict[date, dict[str, Decimal]]]] = defaultdict( lambda: defaultdict(lambda: defaultdict(dict)) ) for row in rows: business_store_name = resolve_business_store_name(row.bi_store_name, store_mapping) if not business_store_name: continue project = classify_project(business_store_name) if not project: continue month_key = (row.stat_date.year, row.stat_date.month) current_amount = grouped[project][month_key][row.stat_date].get(business_store_name, Decimal("0")) grouped[project][month_key][row.stat_date][business_store_name] = current_amount + row.amount project_count = len(grouped) month_count = sum(len(months) for months in grouped.values()) log_step(f"{metric_config.description}聚合完成: 项目={project_count}，项目月份组合={month_count}") return grouped def build_target_project_months( start_date: date, end_date: date, grouped_metrics: dict[str, dict[tuple[int, int], dict[date, dict[str, Decimal]]]], ) -> dict[str, dict[tuple[int, int], dict[date, dict[str, Decimal]]]]: result = { project: {month_key: dict(date_map) for month_key, date_map in month_dict.items()} for project, month_dict in grouped_metrics.items() } current = start_date while current <= end_date: month_key = (current.year, current.month) for project in SUPPORTED_PROJECTS: project_months = result.setdefault(project, {}) metrics_by_date = project_months.setdefault(month_key, {}) metrics_by_date.setdefault(current, {}) current += timedelta(days=1) return result def build_daily_sync_context(config: DailySyncConfig) -> DailySyncContext: mapping = load_store_mapping(config.mapping_file) table_overrides = load_table_name_overrides(config.table_map_file) if table_overrides: log_step(f"加载表名映射完成: {len(table_overrides)} 条") client = FeishuBitableClient(config.base_url) tables = client.list_tables() return DailySyncContext( client=client, tables=tables, mapping=mapping, table_overrides=table_overrides, config=config, ) def get_feishu_token() -> str: app_id = os.getenv("FEISHU_APP_ID") app_secret = os.getenv("FEISHU_APP_SECRET") if not app_id or not app_secret: raise ValueError("缺少飞书环境变量 FEISHU_APP_ID / FEISHU_APP_SECRET") log_step("开始获取飞书 tenant_access_token") resp = requests.post( "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal", json={"app_id": app_id, "app_secret": app_secret}, timeout=30, ) resp.raise_for_status() payload = resp.json() if payload.get("code") != 0: raise RuntimeError(f"获取飞书 token 失败: {payload}") log_step("飞书 tenant_access_token 获取成功") return payload["tenant_access_token"] class FeishuBitableClient: def __init__(self, base_url: str) -> None: self.base_url = base_url self.app_token = self.extract_app_token(base_url) log_step(f"解析飞书 Base URL 成功: app_token={self.app_token}") self.token = get_feishu_token() self._fields_cache: dict[str, list[dict[str, Any]]] = {} self._records_cache: dict[str, list[dict[str, Any]]] = {} self._tables_cache: list[TableInfo] | None = None self.session = requests.Session() self.session.headers.update( { "Authorization": f"Bearer {self.token}", "Content-Type": "application/json; charset=utf-8", } ) @staticmethod def extract_app_token(base_url: str) -> str: path = urlparse(base_url).path.strip("/") match = re.search(r"base/([^/]+)", path) if not match: raise ValueError(f"无法从链接中解析 app_token: {base_url}") return match.group(1) def request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]: url = f"https://open.feishu.cn{path}" params = kwargs.get("params") if params: log_debug(f"请求飞书接口: {method} {path} params={params}") else: log_debug(f"请求飞书接口: {method} {path}") resp = self.session.request(method, url, timeout=60, **kwargs) try: payload = resp.json() except ValueError: resp.raise_for_status() raise RuntimeError(f"飞书接口返回了非 JSON 响应: status={resp.status_code}, body={resp.text[:500]}") if resp.status_code >= 400: error = payload.get("error") or {} raise RuntimeError( f"飞书接口 HTTP {resp.status_code} {path}: code={payload.get('code')} msg={payload.get('msg')} " f"log_id={error.get('log_id', '')} payload={payload}" ) if payload.get("code") != 0: error = payload.get("error") or {} raise RuntimeError( f"飞书接口失败 {path}: code={payload.get('code')} msg={payload.get('msg')} " f"log_id={error.get('log_id', '')} payload={payload}" ) return payload.get("data") or {} def list_tables(self) -> list[TableInfo]: if self._tables_cache is not None: log_debug(f"命中数据表缓存: 表数={len(self._tables_cache)}") return self._tables_cache log_step("开始获取多维表格数据表列表") items: list[TableInfo] = [] page_token = "" while True: params: dict[str, Any] = {} if page_token: params["page_token"] = page_token data = self.request("GET", f"/open-apis/bitable/v1/apps/{self.app_token}/tables", params=params) for item in data.get("items") or []: items.append(TableInfo(table_id=item["table_id"], name=item["name"])) if not data.get("has_more", False): break page_token = data.get("page_token") or "" if not page_token: break log_debug(f"飞书返回的数据表名称: {preview_list([item.name for item in items])}") log_step(f"数据表列表获取完成: {len(items)} 个") self._tables_cache = items return items def list_fields(self, table_id: str) -> list[dict[str, Any]]: if table_id in self._fields_cache: log_debug(f"命中字段缓存: table_id={table_id}, 字段数={len(self._fields_cache[table_id])}") return self._fields_cache[table_id] log_debug(f"开始获取字段列表: table_id={table_id}") items: list[dict[str, Any]] = [] page_token = "" seen_tokens: set[str] = set() page_index = 0 while True: page_index += 1 params = {"page_size": 100} if page_token: params["page_token"] = page_token if page_token in seen_tokens: raise RuntimeError(f"字段列表分页出现重复 page_token，已中断: table_id={table_id}, page_token={page_token}") seen_tokens.add(page_token) log_debug(f"获取字段列表分页: table_id={table_id}, page={page_index}, page_token={page_token or ''}") data = self.request( "GET", f"/open-apis/bitable/v1/apps/{self.app_token}/tables/{table_id}/fields", params=params, ) items.extend(data.get("items") or []) has_more = data.get("has_more", False) next_page_token = data.get("page_token") or "" log_debug( f"字段列表分页完成: table_id={table_id}, page={page_index}, 累计字段数={len(items)}, " f"has_more={has_more}, next_page_token={next_page_token or ''}" ) if not has_more: break if not next_page_token: raise RuntimeError(f"字段列表分页返回 has_more=True 但 page_token 为空: table_id={table_id}, 已获取字段数={len(items)}") page_token = next_page_token if page_index >= 100: raise RuntimeError(f"字段列表分页超过 100 页，已中断: table_id={table_id}") self._fields_cache[table_id] = items log_debug(f"飞书返回的字段名称: table_id={table_id}, fields={preview_list([item.get('field_name', '') for item in items])}") log_debug(f"字段列表获取完成: table_id={table_id}, 字段数={len(items)}") return items def ensure_number_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": 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 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]}" ) 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: log_debug(f"命中记录缓存: table_id={table_id}, 记录数={len(self._records_cache[table_id])}") return self._records_cache[table_id] log_debug(f"开始获取记录列表: table_id={table_id}") items: list[dict[str, Any]] = [] page_token = "" page_index = 0 while True: page_index += 1 params = {"page_size": 500} if page_token: params["page_token"] = page_token log_debug(f"获取记录列表分页: table_id={table_id}, page={page_index}, page_token={page_token or ''}") data = self.request( "GET", f"/open-apis/bitable/v1/apps/{self.app_token}/tables/{table_id}/records", params=params, ) items.extend(data.get("items") or []) page_token = data.get("page_token") or "" log_debug( f"记录列表分页完成: table_id={table_id}, page={page_index}, " f"累计记录数={len(items)}, next_page_token={page_token or ''}" ) if not page_token: break log_debug(f"记录列表获取完成: table_id={table_id}, 记录数={len(items)}") self._records_cache[table_id] = items return items def update_record(self, table_id: str, record_id: str, fields: dict[str, Any]) -> None: log_debug(f"更新记录: table_id={table_id}, record_id={record_id}, fields={list(fields.keys())}") self.request( "PUT", f"/open-apis/bitable/v1/apps/{self.app_token}/tables/{table_id}/records/{record_id}", json={"fields": fields}, ) cached_records = self._records_cache.get(table_id) if cached_records is not None: for item in cached_records: if item.get("record_id") == record_id: item.setdefault("fields", {}).update(fields) break def load_table_name_overrides(path_text: str) -> dict[str, str]: if not path_text: return {} path = Path(path_text) if not path.exists(): raise FileNotFoundError(f"未找到表名映射文件: {path}") payload = json.loads(path.read_text(encoding="utf-8")) return {str(key): str(value) for key, value in payload.items()} def send_feishu_sync_card( webhook_url: str, title: str, base_url: str, start_date: date, end_date: date, ) -> None: if not webhook_url: log_step("未配置飞书群机器人 webhook，跳过群消息推送") return card = { "msg_type": "interactive", "card": { "config": {"wide_screen_mode": True}, "header": { "template": "blue", "title": {"tag": "plain_text", "content": title}, }, "elements": [ { "tag": "div", "text": { "tag": "lark_md", "content": ( f"**同步范围**\n{start_date.isoformat()} 至 {end_date.isoformat()}\n\n" f"**多维表格**\n[点击打开多维表格]({base_url})" ), }, } ], }, } log_step("开始推送飞书群卡片消息") resp = requests.post(webhook_url, json=card, timeout=30) resp.raise_for_status() payload = resp.json() if payload.get("code") not in (0, None): raise RuntimeError(f"飞书群消息推送失败: {payload}") log_step("飞书群卡片消息推送完成") def upload_feishu_image(image_path: str, token: str | None = None) -> str: if not image_path: return "" path = Path(image_path) if not path.exists(): raise FileNotFoundError(f"未找到预览图文件: {path}") token = token or get_feishu_token() headers = {"Authorization": f"Bearer {token}"} with path.open("rb") as image_file: files = {"image": (path.name, image_file, "application/octet-stream")} data = {"image_type": "message"} log_step(f"开始上传飞书消息图片: {path}") resp = requests.post("https://open.feishu.cn/open-apis/im/v1/images", headers=headers, data=data, files=files, timeout=60) resp.raise_for_status() payload = resp.json() if payload.get("code") != 0: raise RuntimeError(f"飞书图片上传失败: {payload}") image_key = ((payload.get("data") or {}).get("image_key")) or "" if not image_key: raise RuntimeError(f"飞书图片上传成功但未返回 image_key: {payload}") log_step(f"飞书消息图片上传完成: image_key={image_key}") return image_key def send_feishu_app_card( chat_id: str, title: str, base_url: str, start_date: date, end_date: date, image_path: str = "", token: str | None = None, ) -> None: if not chat_id: log_step("未配置 FEISHU_CHAT_ID，跳过应用机器人卡片推送") return token = token or get_feishu_token() image_key = upload_feishu_image(image_path, token=token) if image_path else "" elements: list[dict[str, Any]] = [ { "tag": "div", "text": { "tag": "lark_md", "content": ( f"**同步范围**\n{start_date.isoformat()} 至 {end_date.isoformat()}\n\n" f"**多维表格**\n[{title}]({base_url})" ), }, } ] if image_key: elements.append({"tag": "img", "img_key": image_key, "alt": {"tag": "plain_text", "content": "报表预览图"}}) elements.append( { "tag": "action", "actions": [ { "tag": "button", "text": {"tag": "plain_text", "content": "打开仪表盘"}, "type": "primary", "url": base_url, } ], } ) message = { "receive_id": chat_id, "msg_type": "interactive", "content": json.dumps( { "config": {"wide_screen_mode": True}, "header": {"template": "blue", "title": {"tag": "plain_text", "content": title}}, "elements": elements, }, ensure_ascii=False, ), } headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json; charset=utf-8", } log_step(f"开始通过应用机器人发送群卡片: chat_id={chat_id}") resp = requests.post( "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id", headers=headers, json=message, timeout=60, ) try: payload = resp.json() except ValueError: resp.raise_for_status() raise RuntimeError(f"应用机器人发送群卡片失败，返回非 JSON: status={resp.status_code}, body={resp.text[:500]}") if resp.status_code >= 400: raise RuntimeError(f"应用机器人发送群卡片失败: status={resp.status_code}, payload={payload}") if payload.get("code") != 0: raise RuntimeError(f"应用机器人发送群卡片失败: {payload}") log_step("应用机器人群卡片发送完成") def find_target_table( tables: list[TableInfo], project: str, year: int, month: int, overrides: dict[str, str] | None = None, ) -> TableInfo: log_step(f"开始匹配目标表: 项目={project}, 月份={year:04d}-{month:02d}") overrides = overrides or {} override_key = f"{project}:{year:04d}-{month:02d}" if override_key in overrides: target_name = overrides[override_key] for table in tables: if table.name == target_name: log_step(f"命中表名映射: {override_key} -> {table.name}") return table raise ValueError(f"表名映射命中了 {target_name}，但飞书中不存在该表") normalized_projects = [normalize_text(alias) for alias in get_project_table_aliases(project)] month_tokens = { str(month), f"{month}月", f"{month:02d}月", f"{year:04d}{month:02d}", f"{year:04d}年{month:02d}月", f"{year:04d}-{month:02d}", f"{year:04d}_{month:02d}", f"{year:04d}.{month:02d}", } log_debug(f"项目表名别名: 项目={project}, aliases={get_project_table_aliases(project)}") candidates = [] project_related_tables = [] for table in tables: normalized_name = normalize_text(table.name) matched_project = any(alias in normalized_name for alias in normalized_projects) if matched_project: project_related_tables.append(table.name) if not matched_project: continue if any(normalize_text(token) in normalized_name for token in month_tokens): candidates.append(table) if len(candidates) == 1: log_step(f"匹配到目标表: 项目={project}, 月份={year:04d}-{month:02d}, 表={candidates[0].name}") return candidates[0] if not candidates: all_table_names = [table.name for table in tables] raise ValueError( f"未找到匹配表: 项目={project}, 月份={year:04d}-{month:02d}, 项目相关表={project_related_tables or '[]'}, 全部表={all_table_names}" ) raise ValueError( f"匹配到多个表，请提供 table_map_file 指定: 项目={project}, 月份={year:04d}-{month:02d}, 表={', '.join(t.name for t in candidates)}" ) def decimal_to_number(value: Decimal) -> float: return float(value.quantize(Decimal("0.01"))) def get_metric_field_name(stat_date: date, metric_config: MetricConfig) -> str: return stat_date.strftime(f"%m-%d{metric_config.field_suffix}") def parse_number(value: Any) -> float | None: if value in (None, ""): return None if isinstance(value, bool): return None if isinstance(value, (int, float)): return round(float(value), 2) if isinstance(value, str): text = value.strip().replace(",", "") if not text: return None try: return round(float(text), 2) except ValueError: return None return None def iter_dates(start_date: date, end_date: date) -> list[date]: dates = [] current = start_date while current <= end_date: dates.append(current) current = date.fromordinal(current.toordinal() + 1) return dates def build_group_dates_by_month(start_date: date, end_date: date) -> dict[tuple[int, int], list[date]]: grouped: dict[tuple[int, int], list[date]] = defaultdict(list) for stat_date in iter_dates(start_date, end_date): grouped[(stat_date.year, stat_date.month)].append(stat_date) return grouped def build_field_group_payload( fields: list[dict[str, Any]], 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: field_id = str(field.get("field_id") or "").strip() field_name = str(field.get("field_name") or "").strip() if not field_id: continue if field_name == store_field_name: continue 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: groups.append({"name": "销售额/退款额", "children": other_fields}) return groups 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( client: FeishuBitableClient, table: TableInfo, store_field_name: str, allowed_field_names: set[str] | None, dry_run: bool, ) -> None: log_step(f"开始检查字段编组: 表={table.name}") fields = client.list_fields(table.table_id) desired_groups = build_field_group_payload(fields, store_field_name, allowed_field_names) if not desired_groups: log_step(f"跳过字段编组: 表={table.name}, 原因=没有可分组字段") return desired_summary = {group["name"]: len(group.get("children") or []) for group in desired_groups} log_step(f"目标字段编组: 表={table.name}, summary={desired_summary}") if dry_run: print(f"[DRY-RUN] 表={table.name} 将创建/更新字段编组: {desired_summary}") return try: client.create_field_groups(table.table_id, desired_groups) log_step(f"字段编组处理完成: 表={table.name}") except Exception as exc: error_text = str(exc) log_step(f"字段编组创建失败: 表={table.name}, error={error_text}") if "99991672" in error_text or "base:field_group:create" in error_text: log_step( f"跳过字段编组创建: 表={table.name}, " "原因=应用缺少 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: try: client.create_field_groups(table.table_id, [group]) log_step( f"单个字段编组创建成功: 表={table.name}, group={group.get('name')}, " f"字段数={len(group.get('children') or [])}" ) except Exception as single_exc: log_step( f"单个字段编组创建失败: 表={table.name}, group={group.get('name')}, error={single_exc}" ) return raise def sync_field_groups_for_range( start_date: date, end_date: date, base_url: str, store_field_name: str, table_map_file: str, dry_run: bool, skip_missing_tables: bool, ) -> None: log_step(f"开始同步字段编组: {start_date.isoformat()} -> {end_date.isoformat()}") context = build_daily_sync_context( DailySyncConfig( base_url=base_url, mapping_file=get_mapping_file(), store_field_name=store_field_name, table_map_file=table_map_file, dry_run=dry_run, skip_missing_tables=skip_missing_tables, ) ) 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 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, force_full_rebuild: bool = False, allowed_field_names_override: set[str] | None = None, ) -> None: day_mode = "完整编组" if force_full_rebuild or stat_date.day == 1 else "增量编组" log_step(f"开始同步字段编组: 日期={stat_date.isoformat()}, 模式={day_mode}") allowed_field_names: set[str] | None = None 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_incremental_field_names_for_day(stat_date) log_step(f"增量编组目标字段: 日期={stat_date.isoformat()}, fields={sorted(allowed_field_names)}") for project in SUPPORTED_PROJECTS: 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 ensure_field_groups_for_table( client=context.client, table=table, store_field_name=context.config.store_field_name, allowed_field_names=allowed_field_names, dry_run=context.config.dry_run, ) log_step(f"字段编组同步完成: 日期={stat_date.isoformat()}, 模式={day_mode}") def sync_field_groups_for_day( stat_date: date, base_url: str, store_field_name: str, table_map_file: str, dry_run: bool, skip_missing_tables: bool, ) -> None: context = build_daily_sync_context( DailySyncConfig( base_url=base_url, mapping_file=get_mapping_file(), store_field_name=store_field_name, table_map_file=table_map_file, dry_run=dry_run, skip_missing_tables=skip_missing_tables, ) ) sync_field_groups_for_day_with_context(context, stat_date) def sync_metric_table_for_month( client: FeishuBitableClient, table: TableInfo, store_field_name: str, metrics_by_date: dict[date, dict[str, Decimal]], dry_run: bool, metric_config: MetricConfig, ) -> tuple[SyncResult, set[str]]: log_step(f"开始同步数据表: {table.name}") table_fields = client.list_fields(table.table_id) store_field_name = resolve_store_field_name(table_fields, store_field_name) log_step(f"识别到店铺字段: 表={table.name}, field={store_field_name}") records = client.list_records(table.table_id) record_by_store: dict[str, dict[str, Any]] = {} preview_store_fields: list[str] = [] for item in records: fields = item.get("fields", {}) store_name = fields.get(store_field_name) if isinstance(store_name, list): store_name = store_name[0] if store_name else "" if not store_name: continue normalized_store_name = str(store_name).strip() normalized_store_key = normalize_text(normalized_store_name) preview_store_fields.append(normalized_store_name) record_by_store[normalized_store_key] = item log_debug(f"飞书返回的记录店铺名称预览: table_id={table.table_id}, stores={preview_list(preview_store_fields)}") feishu_store_names = set(record_by_store) sql_store_names: set[str] = set() sql_store_names_raw: set[str] = set() for store_metrics in metrics_by_date.values(): sql_store_names_raw.update(store_metrics.keys()) sql_store_names.update(normalize_text(name) for name in store_metrics.keys()) all_store_names = feishu_store_names | sql_store_names sql_name_map = {normalize_text(name): name for name in sql_store_names_raw} feishu_name_map = {normalize_text(name): name for name in preview_store_fields} sql_only_stores = sorted(sql_name_map[name] for name in (sql_store_names - feishu_store_names)) feishu_only_stores = sorted(feishu_name_map[name] for name in (feishu_store_names - sql_store_names)) log_debug(f"SQL 聚合出的店铺名称: 表={table.name}, stores={preview_list(sorted(sql_store_names_raw))}") log_step(f"待处理店铺数: 表={table.name}, 店铺数={len(all_store_names)}, 日期数={len(metrics_by_date)}") log_step( f"店铺匹配情况: 表={table.name}, SQL有但飞书无={preview_list(sql_only_stores)}, " f"飞书有但SQL无={preview_list(feishu_only_stores)}" ) created_fields: set[str] = set() for stat_date in sorted(metrics_by_date): field_name = get_metric_field_name(stat_date, metric_config) if dry_run: print(f"[DRY-RUN] 表 {table.name} 将新增字段: {field_name}") else: if client.ensure_number_field(table.table_id, field_name): created_fields.add(field_name) for store_key in sorted(all_store_names): store_name = sql_name_map.get(store_key) or feishu_name_map.get(store_key) or store_key amount = None for raw_name, raw_amount in metrics_by_date[stat_date].items(): if normalize_text(raw_name) == store_key: amount = raw_amount break field_value = decimal_to_number(amount) if amount is not None else None existing = record_by_store.get(store_key) if dry_run: print( f"[DRY-RUN] 表={table.name} 店铺={store_name} 字段={field_name} 值={field_value if field_value is not None else ''}" ) continue if existing: client.update_record(table.table_id, existing["record_id"], {field_name: field_value}) continue log_debug(f"跳过新增记录: 表={table.name}, 店铺={store_name}, 字段={field_name}, 原因=飞书中不存在该店铺行") log_step(f"数据表同步完成: {table.name}") return SyncResult(sql_only_stores=sql_only_stores, feishu_only_stores=feishu_only_stores), created_fields def run_metric_sync_with_context( context: DailySyncContext, metric_name: str, start_date: date, end_date: date, ) -> MetricSyncArtifacts: metric_config = METRIC_CONFIGS[metric_name] metric_rows = fetch_metric_rows(start_date, end_date, metric_config) grouped_metrics = build_project_metrics(metric_rows, context.mapping, metric_config) grouped_metrics = build_target_project_months(start_date, end_date, grouped_metrics) created_fields: set[str] = set() unmatched_summary: list[tuple[str, str, str, SyncResult]] = [] for project in sorted(grouped_metrics): for (year, month), metrics_by_date in sorted(grouped_metrics[project].items()): try: table = find_target_table(context.tables, project, year, month, context.table_overrides) except ValueError as exc: if not context.config.skip_missing_tables: raise log_step(f"跳过: {exc}") continue log_step(f"同步项目={project} 月份={year:04d}-{month:02d} -> 表={table.name}") sync_result, created = sync_metric_table_for_month( client=context.client, table=table, store_field_name=context.config.store_field_name, metrics_by_date=metrics_by_date, dry_run=context.config.dry_run, metric_config=metric_config, ) created_fields.update(created) unmatched_summary.append((project, f"{year:04d}-{month:02d}", table.name, sync_result)) log_step("未匹配店铺汇总开始") for project, month, table_name, result in unmatched_summary: log_step( f"未匹配店铺汇总: 项目={project}, 月份={month}, 表={table_name}, " f"SQL有但飞书无={result.sql_only_stores}, 飞书有但SQL无={result.feishu_only_stores}" ) return MetricSyncArtifacts(created_fields=created_fields) def run_metric_sync( metric_name: str, start_date: date, end_date: date, base_url: str, mapping_file: Path, store_field_name: str, table_map_file: str, dry_run: bool, skip_missing_tables: bool, ) -> None: context = build_daily_sync_context( DailySyncConfig( base_url=base_url, mapping_file=mapping_file, store_field_name=store_field_name, table_map_file=table_map_file, dry_run=dry_run, skip_missing_tables=skip_missing_tables, ) ) run_metric_sync_with_context(context, metric_name, start_date, end_date) def get_sales_field_name(stat_date: date) -> str: return get_metric_field_name(stat_date, METRIC_CONFIGS["sales"]) def get_refund_field_name(stat_date: date) -> str: return get_metric_field_name(stat_date, METRIC_CONFIGS["refund"]) 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, ) -> 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}, " f"退款字段={refund_field_name in field_names}, GMV字段={gmv_field_name in field_names}" ) if sales_field_name not in field_names or refund_field_name not in field_names: log_step(f"跳过 GMV 计算: 表={table.name}, 原因=缺少销售额或退款额字段") return set() created_fields: set[str] = set() if not dry_run: 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 skipped_empty_count = 0 for item in records: record_id = item["record_id"] record_fields = item.get("fields", {}) store_name = record_fields.get(store_field_name) if isinstance(store_name, list): store_name = store_name[0] if store_name else "" store_name = str(store_name or "").strip() if store_name: preview_stores.append(store_name) sales_value = parse_number(record_fields.get(sales_field_name)) refund_value = parse_number(record_fields.get(refund_field_name)) if sales_value is None and refund_value is None: 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"{summary_gmv_field_name}={summary_gmv_value}" ) else: 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_fields def run_gmv_sync_with_context( context: DailySyncContext, start_date: date, end_date: date, ) -> MetricSyncArtifacts: created_fields: set[str] = set() for stat_date in iter_dates(start_date, end_date): for project in SUPPORTED_PROJECTS: try: table = find_target_table(context.tables, project, stat_date.year, stat_date.month, context.table_overrides) except ValueError as exc: if not context.config.skip_missing_tables: raise log_step(f"跳过: {exc}") continue created = sync_gmv_table_for_day( client=context.client, table=table, stat_date=stat_date, preferred_store_field_name=context.config.store_field_name, dry_run=context.config.dry_run, ) created_fields.update(created) return MetricSyncArtifacts(created_fields=created_fields) def run_gmv_sync( start_date: date, end_date: date, base_url: str, store_field_name: str, dry_run: bool, skip_missing_tables: bool, ) -> None: context = build_daily_sync_context( DailySyncConfig( base_url=base_url, mapping_file=get_mapping_file(), store_field_name=store_field_name, table_map_file="", dry_run=dry_run, skip_missing_tables=skip_missing_tables, ) ) 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() 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) 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}" ) 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()}") def send_sync_notification( context: DailySyncContext, start_date: date, end_date: date, ) -> None: if context.config.dry_run: return if context.config.chat_id: send_feishu_app_card( chat_id=context.config.chat_id, title="GMV数据同步", base_url=context.config.base_url, start_date=start_date, end_date=end_date, image_path=context.config.report_image_path, token=context.client.token, ) return send_feishu_sync_card( webhook_url=context.config.group_webhook, title="GMV数据同步", base_url=context.config.base_url, start_date=start_date, end_date=end_date, )