commit 679edd1f5859e410b362e9f3b6aac24e8bafba1e Author: BXY0414 <1079823339@qq.com> Date: Fri Mar 13 13:10:20 2026 +0800 refactor: consolidate bitable pipeline and add legacy reconstructed backups diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000..cc5d92e Binary files /dev/null and b/.DS_Store differ diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..4b5a294 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,4 @@ +{ + "python-envs.defaultEnvManager": "ms-python.python:conda", + "python-envs.defaultPackageManager": "ms-python.python:conda" +} \ No newline at end of file diff --git a/__pycache__/project_bitable_common.cpython-312.pyc b/__pycache__/project_bitable_common.cpython-312.pyc new file mode 100644 index 0000000..acefc01 Binary files /dev/null and b/__pycache__/project_bitable_common.cpython-312.pyc differ diff --git a/generate_project_gmv_csv.py b/generate_project_gmv_csv.py new file mode 100644 index 0000000..f4bdb17 --- /dev/null +++ b/generate_project_gmv_csv.py @@ -0,0 +1,46 @@ +import argparse + +from project_bitable_common import ( + DEFAULT_END_DATE, + DEFAULT_START_DATE, + DEFAULT_STORE_FIELD_NAME, + get_base_url, + load_dotenv, + log_step, + run_gmv_sync, + validate_date, +) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="从飞书多维表格读取销售额和退款额，计算 GMV 并写回。") + parser.add_argument("--start-date", default=DEFAULT_START_DATE, help="开始日期，格式 YYYY-MM-DD") + parser.add_argument("--end-date", default=DEFAULT_END_DATE, help="结束日期，格式 YYYY-MM-DD") + parser.add_argument("--base-url", default=get_base_url(), help="飞书多维表格地址") + parser.add_argument("--store-field-name", default=DEFAULT_STORE_FIELD_NAME, help="飞书中的店铺字段名") + parser.add_argument("--dry-run", action="store_true", help="只输出将要更新的内容，不调用飞书写接口") + parser.add_argument("--skip-missing-tables", action="store_true", help="找不到对应月份的飞书表时跳过") + return parser.parse_args() + + +def main() -> None: + load_dotenv() + args = parse_args() + start_date = validate_date(args.start_date) + end_date = validate_date(args.end_date) + if end_date < start_date: + raise ValueError("end-date 不能早于 start-date") + log_step("开始执行 GMV 同步") + run_gmv_sync( + start_date=start_date, + end_date=end_date, + base_url=args.base_url, + store_field_name=args.store_field_name, + dry_run=args.dry_run, + skip_missing_tables=args.skip_missing_tables, + ) + log_step("GMV 同步完成") + + +if __name__ == "__main__": + main() diff --git a/generate_project_refund_csv.py b/generate_project_refund_csv.py new file mode 100644 index 0000000..38baa3f --- /dev/null +++ b/generate_project_refund_csv.py @@ -0,0 +1,53 @@ +import argparse +from pathlib import Path + +from project_bitable_common import ( + DEFAULT_END_DATE, + DEFAULT_START_DATE, + DEFAULT_STORE_FIELD_NAME, + get_base_url, + get_mapping_file, + load_dotenv, + log_step, + run_metric_sync, + validate_date, +) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="从 PostgreSQL 拉取退款额并写入飞书多维表格。") + parser.add_argument("--start-date", default=DEFAULT_START_DATE, help="开始日期，格式 YYYY-MM-DD") + parser.add_argument("--end-date", default=DEFAULT_END_DATE, help="结束日期，格式 YYYY-MM-DD") + parser.add_argument("--base-url", default=get_base_url(), help="飞书多维表格地址") + parser.add_argument("--mapping-file", default=str(get_mapping_file()), help="店铺映射表 xlsx") + parser.add_argument("--store-field-name", default=DEFAULT_STORE_FIELD_NAME, help="飞书中的店铺字段名") + 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="找不到对应月份的飞书表时跳过") + return parser.parse_args() + + +def main() -> None: + load_dotenv() + args = parse_args() + start_date = validate_date(args.start_date) + end_date = validate_date(args.end_date) + if end_date < start_date: + raise ValueError("end-date 不能早于 start-date") + log_step("开始执行退款额同步") + run_metric_sync( + metric_name="refund", + start_date=start_date, + end_date=end_date, + base_url=args.base_url, + mapping_file=Path(args.mapping_file), + store_field_name=args.store_field_name, + table_map_file=args.table_map_file, + dry_run=args.dry_run, + skip_missing_tables=args.skip_missing_tables, + ) + log_step("退款额同步完成") + + +if __name__ == "__main__": + main() diff --git a/generate_project_sales_csv.py b/generate_project_sales_csv.py new file mode 100644 index 0000000..3a652b0 --- /dev/null +++ b/generate_project_sales_csv.py @@ -0,0 +1,53 @@ +import argparse +from pathlib import Path + +from project_bitable_common import ( + DEFAULT_END_DATE, + DEFAULT_START_DATE, + DEFAULT_STORE_FIELD_NAME, + get_base_url, + get_mapping_file, + load_dotenv, + log_step, + run_metric_sync, + validate_date, +) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="从 PostgreSQL 拉取销售额并写入飞书多维表格。") + parser.add_argument("--start-date", default=DEFAULT_START_DATE, help="开始日期，格式 YYYY-MM-DD") + parser.add_argument("--end-date", default=DEFAULT_END_DATE, help="结束日期，格式 YYYY-MM-DD") + parser.add_argument("--base-url", default=get_base_url(), help="飞书多维表格地址") + parser.add_argument("--mapping-file", default=str(get_mapping_file()), help="店铺映射表 xlsx") + parser.add_argument("--store-field-name", default=DEFAULT_STORE_FIELD_NAME, help="飞书中的店铺字段名") + 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="找不到对应月份的飞书表时跳过") + return parser.parse_args() + + +def main() -> None: + load_dotenv() + args = parse_args() + start_date = validate_date(args.start_date) + end_date = validate_date(args.end_date) + if end_date < start_date: + raise ValueError("end-date 不能早于 start-date") + log_step("开始执行销售额同步") + run_metric_sync( + metric_name="sales", + start_date=start_date, + end_date=end_date, + base_url=args.base_url, + mapping_file=Path(args.mapping_file), + store_field_name=args.store_field_name, + table_map_file=args.table_map_file, + dry_run=args.dry_run, + skip_missing_tables=args.skip_missing_tables, + ) + log_step("销售额同步完成") + + +if __name__ == "__main__": + main() diff --git a/project.zip b/project.zip new file mode 100644 index 0000000..3d3505e Binary files /dev/null and b/project.zip differ diff --git a/project_bitable_common.py b/project_bitable_common.py new file mode 100644 index 0000000..aca42cc --- /dev/null +++ b/project_bitable_common.py @@ -0,0 +1,919 @@ +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 + + +BASE_DIR = Path(__file__).resolve().parent +ENV_FILE = BASE_DIR / ".env" +DEFAULT_MAPPING_FILE = BASE_DIR / "店铺映射表.xlsx" +DEFAULT_START_DATE = "2026-03-01" +DEFAULT_END_DATE = "2026-03-01" +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.实付分摊金额 AS 指标金额 + 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 + +SELECT + 统计日期, + 原始店铺名称 AS 店铺名称, + SUM(指标金额) +FROM base_data +WHERE 店铺主键 IN (1,28,29,3) +GROUP BY 1,2 + +UNION ALL + +SELECT + 统计日期, + CASE + WHEN 业务模式 IN ('三方计算','三方分销') + THEN 原始店铺名称 || '_三方' + ELSE 原始店铺名称 || '_自营' + END AS 店铺名称, + SUM(指标金额) +FROM base_data +WHERE 店铺主键 = 1 +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.店铺主键, + s.店铺名称 AS 原始店铺名称, + p.品牌名称, + f.退款金额 AS 指标金额 + 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 +), +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(指标金额) +FROM base_data +WHERE 店铺主键 IN (1,28,29,3) +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] + + +def log_step(message: str) -> None: + now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print(f"[{now}] {message}", flush=True) + + +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()) + + +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 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 business_name, bi_name in ws.iter_rows(min_row=2, values_only=True): + 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 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.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_step(f"请求飞书接口: {method} {path} params={params}") + else: + log_step(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]: + 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_step(f"飞书返回的数据表名称: {preview_list([item.name for item in items])}") + log_step(f"数据表列表获取完成: {len(items)} 个") + return items + + def list_fields(self, table_id: str) -> list[dict[str, Any]]: + if table_id in self._fields_cache: + log_step(f"命中字段缓存: table_id={table_id}, 字段数={len(self._fields_cache[table_id])}") + return self._fields_cache[table_id] + log_step(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_step(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_step( + 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_step(f"飞书返回的字段名称: table_id={table_id}, fields={preview_list([item.get('field_name', '') for item in items])}") + log_step(f"字段列表获取完成: table_id={table_id}, 字段数={len(items)}") + return items + + def ensure_number_field(self, table_id: str, field_name: str) -> None: + log_step(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_step(f"字段已存在，跳过创建: table_id={table_id}, field={field_name}") + return + 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}, + ) + self._fields_cache.pop(table_id, None) + log_step(f"字段创建完成: table_id={table_id}, field={field_name}") + + def list_records(self, table_id: str) -> list[dict[str, Any]]: + log_step(f"开始获取记录列表: table_id={table_id}") + items: list[dict[str, Any]] = [] + page_token = "" + while True: + params = {"page_size": 500} + if page_token: + params["page_token"] = page_token + 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 "" + if not page_token: + break + log_step(f"记录列表获取完成: table_id={table_id}, 记录数={len(items)}") + return items + + def update_record(self, table_id: str, record_id: str, fields: dict[str, Any]) -> None: + log_step(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}, + ) + + +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 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_step(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 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, +) -> SyncResult: + 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_step(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_step(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)}" + ) + + 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: + client.ensure_number_field(table.table_id, 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_step(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) + + +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: + metric_config = METRIC_CONFIGS[metric_name] + mapping = load_store_mapping(mapping_file) + metric_rows = fetch_metric_rows(start_date, end_date, metric_config) + grouped_metrics = build_project_metrics(metric_rows, mapping, metric_config) + grouped_metrics = build_target_project_months(start_date, end_date, grouped_metrics) + table_overrides = load_table_name_overrides(table_map_file) + if table_overrides: + log_step(f"加载表名映射完成: {len(table_overrides)} 条") + client = FeishuBitableClient(base_url) + tables = client.list_tables() + 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(tables, project, year, month, table_overrides) + except ValueError as exc: + if not skip_missing_tables: + raise + log_step(f"跳过: {exc}") + continue + log_step(f"同步项目={project} 月份={year:04d}-{month:02d} -> 表={table.name}") + sync_result = sync_metric_table_for_month( + client=client, + table=table, + store_field_name=store_field_name, + metrics_by_date=metrics_by_date, + dry_run=dry_run, + metric_config=metric_config, + ) + 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}" + ) + + +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 sync_gmv_table_for_day( + client: FeishuBitableClient, + table: TableInfo, + stat_date: date, + preferred_store_field_name: str, + dry_run: bool, +) -> None: + 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) + field_names = [str(item.get("field_name") or "") for item in fields] + log_step( + 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 + if not dry_run: + client.ensure_number_field(table.table_id, 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) + 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}" + ) + else: + client.update_record(table.table_id, record_id, {gmv_field_name: gmv_value}) + updated_count += 1 + log_step(f"飞书记录店铺预览: 表={table.name}, stores={preview_list(preview_stores)}") + log_step(f"GMV 计算完成: 表={table.name}, 更新记录数={updated_count}, 双空跳过={skipped_empty_count}") + + +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: + client = FeishuBitableClient(base_url) + tables = client.list_tables() + for stat_date in iter_dates(start_date, end_date): + for project in SUPPORTED_PROJECTS: + try: + table = find_target_table(tables, project, stat_date.year, stat_date.month) + except ValueError as exc: + if not skip_missing_tables: + raise + log_step(f"跳过: {exc}") + continue + sync_gmv_table_for_day( + client=client, + table=table, + stat_date=stat_date, + preferred_store_field_name=store_field_name, + dry_run=dry_run, + ) + diff --git a/run_project_bitable_pipeline.py b/run_project_bitable_pipeline.py new file mode 100644 index 0000000..4008f22 --- /dev/null +++ b/run_project_bitable_pipeline.py @@ -0,0 +1,74 @@ +import argparse +from pathlib import Path + +from project_bitable_common import ( + DEFAULT_END_DATE, + DEFAULT_START_DATE, + DEFAULT_STORE_FIELD_NAME, + get_base_url, + get_mapping_file, + load_dotenv, + log_step, + run_gmv_sync, + run_metric_sync, + validate_date, +) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="按 销售 -> 退款 -> GMV 顺序执行飞书多维表格同步。") + parser.add_argument("--start-date", default=DEFAULT_START_DATE, help="开始日期，格式 YYYY-MM-DD") + parser.add_argument("--end-date", default=DEFAULT_END_DATE, help="结束日期，格式 YYYY-MM-DD") + parser.add_argument("--base-url", default=get_base_url(), help="飞书多维表格地址") + parser.add_argument("--mapping-file", default=str(get_mapping_file()), help="店铺映射表 xlsx") + parser.add_argument("--store-field-name", default=DEFAULT_STORE_FIELD_NAME, help="飞书中的店铺字段名") + 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="找不到对应月份的飞书表时跳过") + return parser.parse_args() + + +def main() -> None: + load_dotenv() + args = parse_args() + start_date = validate_date(args.start_date) + end_date = validate_date(args.end_date) + if end_date < start_date: + raise ValueError("end-date 不能早于 start-date") + + log_step("流水线开始: 销售 -> 退款 -> GMV") + run_metric_sync( + metric_name="sales", + start_date=start_date, + end_date=end_date, + base_url=args.base_url, + mapping_file=Path(args.mapping_file), + store_field_name=args.store_field_name, + table_map_file=args.table_map_file, + dry_run=args.dry_run, + skip_missing_tables=args.skip_missing_tables, + ) + run_metric_sync( + metric_name="refund", + start_date=start_date, + end_date=end_date, + base_url=args.base_url, + mapping_file=Path(args.mapping_file), + store_field_name=args.store_field_name, + table_map_file=args.table_map_file, + dry_run=args.dry_run, + skip_missing_tables=args.skip_missing_tables, + ) + run_gmv_sync( + start_date=start_date, + end_date=end_date, + base_url=args.base_url, + store_field_name=args.store_field_name, + dry_run=args.dry_run, + skip_missing_tables=args.skip_missing_tables, + ) + log_step("流水线完成") + + +if __name__ == "__main__": + main() diff --git a/店铺映射表.xlsx b/店铺映射表.xlsx new file mode 100644 index 0000000..5af1e52 Binary files /dev/null and b/店铺映射表.xlsx differ