diff --git a/.DS_Store b/.DS_Store index 45bae9d..129834e 100644 Binary files a/.DS_Store and b/.DS_Store differ diff --git a/.gitignore b/.gitignore index 2e5f4a2..c7edfd2 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,4 @@ __pycache__/* created/ 创建编组.zip .DS_Store +logs/ \ No newline at end of file diff --git a/__pycache__/project_bitable_common.cpython-312.pyc b/__pycache__/project_bitable_common.cpython-312.pyc index 0b26afc..95f58ec 100644 Binary files a/__pycache__/project_bitable_common.cpython-312.pyc and b/__pycache__/project_bitable_common.cpython-312.pyc differ diff --git a/logging_config.py b/logging_config.py new file mode 100644 index 0000000..1114458 --- /dev/null +++ b/logging_config.py @@ -0,0 +1,65 @@ +import logging +import os +import sys +from logging.handlers import RotatingFileHandler +from pathlib import Path + + +BASE_DIR = Path(__file__).resolve().parent +DEFAULT_LOG_DIR = BASE_DIR / "logs" +DEFAULT_LOG_FILE_NAME = "gmv_sync.log" +LOGGER_NAME = "gmv_analytics" +LOG_FORMAT = "[%(asctime)s] %(message)s" +DATE_FORMAT = "%Y-%m-%d %H:%M:%S" + +_IS_CONFIGURED = False + + +def _build_handlers() -> list[logging.Handler]: + log_dir = Path(os.getenv("LOG_DIR", str(DEFAULT_LOG_DIR))) + log_file_name = os.getenv("LOG_FILE_NAME", DEFAULT_LOG_FILE_NAME) + log_file_path = log_dir / log_file_name + log_dir.mkdir(parents=True, exist_ok=True) + + formatter = logging.Formatter(LOG_FORMAT, datefmt=DATE_FORMAT) + + console_handler = logging.StreamHandler(sys.stdout) + console_handler.setFormatter(formatter) + + file_handler = RotatingFileHandler( + log_file_path, + maxBytes=int(os.getenv("LOG_MAX_BYTES", str(10 * 1024 * 1024))), + backupCount=int(os.getenv("LOG_BACKUP_COUNT", "7")), + encoding="utf-8", + ) + file_handler.setFormatter(formatter) + return [console_handler, file_handler] + + +def setup_logging(force: bool = False) -> logging.Logger: + global _IS_CONFIGURED + + logger = logging.getLogger(LOGGER_NAME) + if _IS_CONFIGURED and not force: + return logger + + level_name = os.getenv("LOG_LEVEL", "INFO").strip().upper() + level = getattr(logging, level_name, logging.INFO) + handlers = _build_handlers() + + if force: + for handler in list(logger.handlers): + logger.removeHandler(handler) + handler.close() + + logger.setLevel(level) + logger.propagate = False + for handler in handlers: + logger.addHandler(handler) + + _IS_CONFIGURED = True + return logger + + +def get_logger() -> logging.Logger: + return setup_logging() diff --git a/prefect_gmv_scheduler.py b/prefect_gmv_scheduler.py index b4d3265..ac8505e 100644 --- a/prefect_gmv_scheduler.py +++ b/prefect_gmv_scheduler.py @@ -22,9 +22,10 @@ def bootstrap_prefect_env() -> None: bootstrap_prefect_env() from prefect import flow, task -from prefect.logging import get_run_logger from project_bitable_common import ( + DailySyncConfig, + build_daily_sync_context, get_base_url, get_default_end_date, get_default_start_date, @@ -36,10 +37,8 @@ from project_bitable_common import ( iter_dates, load_dotenv, log_step, - run_gmv_sync, - run_metric_sync, - send_feishu_app_card, - send_feishu_sync_card, + run_daily_sync_for_date, + send_sync_notification, validate_date, ) @@ -82,75 +81,22 @@ def load_sync_config( ) -@task(name="sync-sales") -def sync_sales_for_day(stat_date_text: str, config: SyncConfig) -> None: - stat_date = validate_date(stat_date_text) - run_metric_sync( - metric_name="sales", - start_date=stat_date, - end_date=stat_date, - base_url=config.base_url, - mapping_file=Path(config.mapping_file), - store_field_name=config.store_field_name, - table_map_file=config.table_map_file, - dry_run=config.dry_run, - skip_missing_tables=config.skip_missing_tables, - ) - - -@task(name="sync-refund") -def sync_refund_for_day(stat_date_text: str, config: SyncConfig) -> None: - stat_date = validate_date(stat_date_text) - run_metric_sync( - metric_name="refund", - start_date=stat_date, - end_date=stat_date, - base_url=config.base_url, - mapping_file=Path(config.mapping_file), - store_field_name=config.store_field_name, - table_map_file=config.table_map_file, - dry_run=config.dry_run, - skip_missing_tables=config.skip_missing_tables, - ) - - -@task(name="sync-gmv") -def sync_gmv_for_day(stat_date_text: str, config: SyncConfig) -> None: - stat_date = validate_date(stat_date_text) - run_gmv_sync( - start_date=stat_date, - end_date=stat_date, - base_url=config.base_url, - store_field_name=config.store_field_name, - dry_run=config.dry_run, - skip_missing_tables=config.skip_missing_tables, - ) - - @task(name="notify-feishu") -def notify_feishu(title: str, config: SyncConfig) -> None: - if config.dry_run: - log_step("dry-run 模式下跳过飞书群推送") - return - start_date = validate_date(config.start_date) - end_date = validate_date(config.end_date) - if config.chat_id: - send_feishu_app_card( - chat_id=config.chat_id, - title=title, - base_url=config.base_url, - start_date=start_date, - end_date=end_date, - image_path=config.report_image_path, +def notify_feishu(context: SyncConfig, start_date_text: str, end_date_text: str) -> None: + context_obj = build_daily_sync_context( + DailySyncConfig( + base_url=context.base_url, + mapping_file=Path(context.mapping_file), + store_field_name=context.store_field_name, + table_map_file=context.table_map_file, + dry_run=context.dry_run, + skip_missing_tables=context.skip_missing_tables, + chat_id=context.chat_id, + report_image_path=context.report_image_path, + group_webhook=context.group_webhook, ) - return - send_feishu_sync_card( - webhook_url=config.group_webhook, - title=title, - base_url=config.base_url, - start_date=start_date, - end_date=end_date, ) + send_sync_notification(context_obj, validate_date(start_date_text), validate_date(end_date_text)) @flow(name="gmv-daily-sync") @@ -161,7 +107,6 @@ def gmv_daily_sync_flow( skip_missing_tables: bool = False, table_map_file: str = "", ) -> None: - logger = get_run_logger() config = load_sync_config( start_date=start_date, end_date=end_date, @@ -174,22 +119,31 @@ def gmv_daily_sync_flow( end_date_value = validate_date(config.end_date) if end_date_value < start_date_value: raise ValueError("end-date 不能早于 start-date") + context = build_daily_sync_context( + DailySyncConfig( + base_url=config.base_url, + mapping_file=Path(config.mapping_file), + store_field_name=config.store_field_name, + table_map_file=config.table_map_file, + dry_run=config.dry_run, + skip_missing_tables=config.skip_missing_tables, + chat_id=config.chat_id, + report_image_path=config.report_image_path, + group_webhook=config.group_webhook, + ) + ) - logger.info("Prefect 调度开始: %s -> %s", config.start_date, config.end_date) + log_step(f"Prefect 调度开始: {config.start_date} -> {config.end_date}") try: for stat_date in iter_dates(start_date_value, end_date_value): - stat_date_text = stat_date.isoformat() - logger.info("开始处理日期: %s", stat_date_text) - sync_sales_for_day(stat_date_text, config) - sync_refund_for_day(stat_date_text, config) - sync_gmv_for_day(stat_date_text, config) - logger.info("日期处理完成: %s", stat_date_text) + log_step(f"Prefect 开始处理日期: {stat_date.isoformat()}") + run_daily_sync_for_date(context, stat_date) except Exception: - notify_feishu.with_options(name="notify-feishu-failed")("GMV数据同步失败", config) + notify_feishu.with_options(name="notify-feishu-failed")(config, config.start_date, config.end_date) raise - notify_feishu("GMV数据同步", config) - logger.info("Prefect 调度完成") + notify_feishu(config, config.start_date, config.end_date) + log_step("Prefect 调度完成") def parse_args() -> argparse.Namespace: diff --git a/project_bitable_common.py b/project_bitable_common.py index 9a5c503..51a52b1 100644 --- a/project_bitable_common.py +++ b/project_bitable_common.py @@ -13,6 +13,8 @@ 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" @@ -242,9 +244,39 @@ class SyncResult: 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: - now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - print(f"[{now}] {message}", flush=True) + get_logger().info(message) + + +def log_debug(message: str) -> None: + get_logger().debug(message) def preview_list(values: list[Any], limit: int = 20) -> str: @@ -264,6 +296,7 @@ def load_dotenv(env_file: Path = ENV_FILE) -> None: continue key, value = line.split("=", 1) os.environ.setdefault(key.strip(), value.strip()) + setup_logging(force=True) def get_base_url() -> str: @@ -468,6 +501,22 @@ def build_target_project_months( 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") @@ -494,6 +543,8 @@ class FeishuBitableClient: 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( { @@ -514,9 +565,9 @@ class FeishuBitableClient: url = f"https://open.feishu.cn{path}" params = kwargs.get("params") if params: - log_step(f"请求飞书接口: {method} {path} params={params}") + log_debug(f"请求飞书接口: {method} {path} params={params}") else: - log_step(f"请求飞书接口: {method} {path}") + log_debug(f"请求飞书接口: {method} {path}") resp = self.session.request(method, url, timeout=60, **kwargs) try: payload = resp.json() @@ -538,6 +589,9 @@ class FeishuBitableClient: 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 = "" @@ -553,15 +607,16 @@ class FeishuBitableClient: page_token = data.get("page_token") or "" if not page_token: break - log_step(f"飞书返回的数据表名称: {preview_list([item.name for item in items])}") + 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_step(f"命中字段缓存: table_id={table_id}, 字段数={len(self._fields_cache[table_id])}") + log_debug(f"命中字段缓存: table_id={table_id}, 字段数={len(self._fields_cache[table_id])}") return self._fields_cache[table_id] - log_step(f"开始获取字段列表: table_id={table_id}") + log_debug(f"开始获取字段列表: table_id={table_id}") items: list[dict[str, Any]] = [] page_token = "" seen_tokens: set[str] = set() @@ -574,7 +629,7 @@ class FeishuBitableClient: 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 ''}") + 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", @@ -583,7 +638,7 @@ class FeishuBitableClient: items.extend(data.get("items") or []) has_more = data.get("has_more", False) next_page_token = data.get("page_token") or "" - log_step( + log_debug( f"字段列表分页完成: table_id={table_id}, page={page_index}, 累计字段数={len(items)}, " f"has_more={has_more}, next_page_token={next_page_token or ''}" ) @@ -595,16 +650,16 @@ class FeishuBitableClient: 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)}") + 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) -> None: - log_step(f"检查字段是否存在: table_id={table_id}, field={field_name}") + 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_step(f"字段已存在，跳过创建: table_id={table_id}, field={field_name}") - return + 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", @@ -613,26 +668,27 @@ class FeishuBitableClient: ) self._fields_cache.pop(table_id, None) log_step(f"字段创建完成: table_id={table_id}, field={field_name}") + return True def try_list_field_groups(self, table_id: str) -> list[dict[str, Any]] | None: path = f"/open-apis/bitable/v1/apps/{self.app_token}/tables/{table_id}/field_groups" url = f"https://open.feishu.cn{path}" - log_step(f"尝试读取字段编组: table_id={table_id}") + log_debug(f"尝试读取字段编组: table_id={table_id}") resp = self.session.request("GET", url, timeout=60) try: payload = resp.json() except ValueError: - log_step(f"字段编组接口返回非 JSON，跳过读取: table_id={table_id}, status={resp.status_code}") + log_debug(f"字段编组接口返回非 JSON，跳过读取: table_id={table_id}, status={resp.status_code}") return None if resp.status_code == 404: - log_step(f"字段编组读取接口不可用: table_id={table_id}, payload={payload}") + log_debug(f"字段编组读取接口不可用: table_id={table_id}, payload={payload}") return None if resp.status_code >= 400 or payload.get("code") != 0: - log_step(f"字段编组读取失败，跳过比对: table_id={table_id}, payload={payload}") + log_debug(f"字段编组读取失败，跳过比对: table_id={table_id}, payload={payload}") return None data = payload.get("data") or {} items = data.get("items") or data.get("field_groups") or [] - log_step(f"字段编组读取成功: table_id={table_id}, 编组数={len(items)}") + log_debug(f"字段编组读取成功: table_id={table_id}, 编组数={len(items)}") return items def create_field_groups(self, table_id: str, field_groups: list[dict[str, Any]]) -> None: @@ -648,13 +704,19 @@ class FeishuBitableClient: log_step(f"字段编组创建请求完成: table_id={table_id}") def list_records(self, table_id: str) -> list[dict[str, Any]]: - log_step(f"开始获取记录列表: table_id={table_id}") + 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", @@ -662,18 +724,29 @@ class FeishuBitableClient: ) 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_step(f"记录列表获取完成: table_id={table_id}, 记录数={len(items)}") + 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_step(f"更新记录: table_id={table_id}, record_id={record_id}, fields={list(fields.keys())}") + 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]: @@ -727,13 +800,13 @@ def send_feishu_sync_card( log_step("飞书群卡片消息推送完成") -def upload_feishu_image(image_path: str) -> str: +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 = get_feishu_token() + 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")} @@ -758,12 +831,13 @@ def send_feishu_app_card( 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 = get_feishu_token() - image_key = upload_feishu_image(image_path) if image_path else "" + 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", @@ -854,7 +928,7 @@ def find_target_table( f"{year:04d}_{month:02d}", f"{year:04d}.{month:02d}", } - log_step(f"项目表名别名: 项目={project}, aliases={get_project_table_aliases(project)}") + log_debug(f"项目表名别名: 项目={project}, aliases={get_project_table_aliases(project)}") candidates = [] project_related_tables = [] for table in tables: @@ -944,7 +1018,7 @@ def build_field_group_payload( other_fields.append(child) groups: list[dict[str, Any]] = [] if gmv_fields: - groups.append({"name": "GMV", "children": gmv_fields}) + groups.append({"name": "GMV编组", "children": gmv_fields}) if other_fields: groups.append({"name": "销售额/退款额", "children": other_fields}) return groups @@ -992,7 +1066,7 @@ def ensure_field_groups_for_table( if existing_normalized == desired_normalized and existing_order == desired_order: log_step(f"字段编组已存在且一致，跳过创建: 表={table.name}") return - log_step( + log_debug( f"字段编组存在差异: 表={table.name}, " f"existing={ {name: len(ids) for name, ids in existing_normalized.items()} }, " f"desired={ {name: len(ids) for name, ids in desired_normalized.items()} }, " @@ -1040,29 +1114,71 @@ def sync_field_groups_for_range( skip_missing_tables: bool, ) -> None: log_step(f"开始同步字段编组: {start_date.isoformat()} -> {end_date.isoformat()}") - 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() + 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(tables, project, year, month, table_overrides) + table = find_target_table(context.tables, project, year, month, context.table_overrides) except ValueError as exc: - if skip_missing_tables: + if context.config.skip_missing_tables: log_step(f"跳过字段编组同步: {exc}") continue raise ensure_field_groups_for_table( - client=client, + client=context.client, table=table, - store_field_name=store_field_name, - dry_run=dry_run, + 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_gmv_field_name(stat_date), + get_sales_field_name(stat_date), + get_refund_field_name(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, @@ -1071,37 +1187,17 @@ def sync_field_groups_for_day( dry_run: bool, skip_missing_tables: bool, ) -> None: - day_mode = "完整编组" if stat_date.day == 1 else "增量编组" - log_step(f"开始同步字段编组: 日期={stat_date.isoformat()}, 模式={day_mode}") - 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() - allowed_field_names: set[str] | None = None - if stat_date.day != 1: - allowed_field_names = { - get_gmv_field_name(stat_date), - get_sales_field_name(stat_date), - get_refund_field_name(stat_date), - } - log_step(f"增量编组目标字段: 日期={stat_date.isoformat()}, fields={sorted(allowed_field_names)}") - for project in SUPPORTED_PROJECTS: - try: - table = find_target_table(tables, project, stat_date.year, stat_date.month, table_overrides) - except ValueError as exc: - if skip_missing_tables: - log_step(f"跳过字段编组同步: {exc}") - continue - raise - ensure_field_groups_for_table( - client=client, - table=table, + context = build_daily_sync_context( + DailySyncConfig( + base_url=base_url, + mapping_file=get_mapping_file(), store_field_name=store_field_name, - allowed_field_names=allowed_field_names, + table_map_file=table_map_file, dry_run=dry_run, + skip_missing_tables=skip_missing_tables, ) - log_step(f"字段编组同步完成: 日期={stat_date.isoformat()}, 模式={day_mode}") + ) + sync_field_groups_for_day_with_context(context, stat_date) def sync_metric_table_for_month( @@ -1111,7 +1207,7 @@ def sync_metric_table_for_month( metrics_by_date: dict[date, dict[str, Decimal]], dry_run: bool, metric_config: MetricConfig, -) -> SyncResult: +) -> 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) @@ -1130,7 +1226,7 @@ def sync_metric_table_for_month( 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)}") + 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() @@ -1145,19 +1241,21 @@ def sync_metric_table_for_month( 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_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: - client.ensure_number_field(table.table_id, field_name) + 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 @@ -1175,9 +1273,50 @@ def sync_metric_table_for_month( 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_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) + 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( @@ -1191,42 +1330,17 @@ def run_metric_sync( 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}" + 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: @@ -1247,7 +1361,7 @@ def sync_gmv_table_for_day( stat_date: date, preferred_store_field_name: str, dry_run: bool, -) -> None: +) -> bool: 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) @@ -1255,15 +1369,16 @@ def sync_gmv_table_for_day( 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( + 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 + return False + created_field = False if not dry_run: - client.ensure_number_field(table.table_id, gmv_field_name) + created_field = client.ensure_number_field(table.table_id, gmv_field_name) records = client.list_records(table.table_id) preview_stores = [] updated_count = 0 @@ -1292,8 +1407,36 @@ def sync_gmv_table_for_day( 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_debug(f"飞书记录店铺预览: 表={table.name}, stores={preview_list(preview_stores)}") log_step(f"GMV 计算完成: 表={table.name}, 更新记录数={updated_count}, 双空跳过={skipped_empty_count}") + return created_field + + +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, + ) + if created: + created_fields.add(get_gmv_field_name(stat_date)) + return MetricSyncArtifacts(created_fields=created_fields) def run_gmv_sync( @@ -1304,21 +1447,67 @@ def run_gmv_sync( 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, - ) + 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_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: + log_step( + f"触发字段编组同步: 日期={stat_date.isoformat()}, " + f"原因={'每月1号完整编组' if stat_date.day == 1 else f'新增字段={sorted(created_fields)}'}" + ) + 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, + ) diff --git a/run_project_bitable_pipeline.py b/run_project_bitable_pipeline.py index c85b3d1..6c8bb55 100644 --- a/run_project_bitable_pipeline.py +++ b/run_project_bitable_pipeline.py @@ -2,6 +2,8 @@ import argparse from pathlib import Path from project_bitable_common import ( + DailySyncConfig, + build_daily_sync_context, get_default_end_date, get_default_start_date, get_base_url, @@ -13,11 +15,8 @@ from project_bitable_common import ( iter_dates, load_dotenv, log_step, - run_gmv_sync, - run_metric_sync, - send_feishu_app_card, - send_feishu_sync_card, - sync_field_groups_for_day, + run_daily_sync_for_date, + send_sync_notification, validate_date, ) @@ -47,65 +46,22 @@ def main() -> None: raise ValueError("end-date 不能早于 start-date") log_step("流水线开始: 按天执行 销售 -> 退款 -> GMV") + context = build_daily_sync_context( + DailySyncConfig( + 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, + chat_id=args.chat_id, + report_image_path=args.report_image_path, + group_webhook=args.group_webhook, + ) + ) for stat_date in iter_dates(start_date, end_date): - log_step(f"开始处理日期: {stat_date.isoformat()}") - run_metric_sync( - metric_name="sales", - start_date=stat_date, - end_date=stat_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=stat_date, - end_date=stat_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=stat_date, - end_date=stat_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, - ) - sync_field_groups_for_day( - stat_date=stat_date, - base_url=args.base_url, - 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(f"日期处理完成: {stat_date.isoformat()}") - if not args.dry_run: - if args.chat_id: - send_feishu_app_card( - chat_id=args.chat_id, - title="GMV数据同步", - base_url=args.base_url, - start_date=start_date, - end_date=end_date, - image_path=args.report_image_path, - ) - else: - send_feishu_sync_card( - webhook_url=args.group_webhook, - title="GMV数据同步", - base_url=args.base_url, - start_date=start_date, - end_date=end_date, - ) + run_daily_sync_for_date(context, stat_date) + send_sync_notification(context, start_date, end_date) log_step("流水线完成")