diff --git a/.DS_Store b/.DS_Store index cc5d92e..45bae9d 100644 Binary files a/.DS_Store and b/.DS_Store differ diff --git a/.gitignore b/.gitignore index 1367188..2e5f4a2 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,7 @@ -__pycache__/ +__pycache__/* *.pyc .vscode/ .env +created/ +创建编组.zip +.DS_Store diff --git a/__pycache__/project_bitable_common.cpython-312.pyc b/__pycache__/project_bitable_common.cpython-312.pyc index acefc01..0b26afc 100644 Binary files a/__pycache__/project_bitable_common.cpython-312.pyc and b/__pycache__/project_bitable_common.cpython-312.pyc differ diff --git a/project_bitable_common.py b/project_bitable_common.py index 9f12c57..9a5c503 100644 --- a/project_bitable_common.py +++ b/project_bitable_common.py @@ -614,6 +614,39 @@ class FeishuBitableClient: self._fields_cache.pop(table_id, None) log_step(f"字段创建完成: table_id={table_id}, field={field_name}") + 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}") + 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}") + return None + if resp.status_code == 404: + log_step(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}") + 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)}") + return items + + def create_field_groups(self, table_id: str, field_groups: list[dict[str, Any]]) -> None: + log_step( + f"开始创建字段编组: table_id={table_id}, " + f"groups={[(group.get('name') or '', len(group.get('children') or [])) for group in field_groups]}" + ) + 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}") + def list_records(self, table_id: str) -> list[dict[str, Any]]: log_step(f"开始获取记录列表: table_id={table_id}") items: list[dict[str, Any]] = [] @@ -881,6 +914,196 @@ def iter_dates(start_date: date, end_date: date) -> list[date]: 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]]: + 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 "gmv" in field_name.lower(): + gmv_fields.append(child) + else: + other_fields.append(child) + groups: list[dict[str, Any]] = [] + if gmv_fields: + groups.append({"name": "GMV", "children": gmv_fields}) + if other_fields: + groups.append({"name": "销售额/退款额", "children": other_fields}) + return groups + + +def normalize_field_groups(field_groups: list[dict[str, Any]]) -> dict[str, set[str]]: + normalized: dict[str, set[str]] = {} + for group in field_groups: + name = str(group.get("name") or "").strip() + if not name: + continue + normalized[name] = { + str(child.get("id") or "").strip() + for child in (group.get("children") or []) + if str(child.get("id") or "").strip() + } + return normalized + + +def get_field_group_order(field_groups: list[dict[str, Any]]) -> list[str]: + return [str(group.get("name") or "").strip() for group in field_groups if str(group.get("name") or "").strip()] + + +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}") + existing_groups = client.try_list_field_groups(table.table_id) + if existing_groups is not None: + existing_normalized = normalize_field_groups(existing_groups) + desired_normalized = normalize_field_groups(desired_groups) + existing_order = get_field_group_order(existing_groups) + desired_order = get_field_group_order(desired_groups) + if existing_normalized == desired_normalized and existing_order == desired_order: + log_step(f"字段编组已存在且一致，跳过创建: 表={table.name}") + return + log_step( + 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()} }, " + f"existing_order={existing_order}, desired_order={desired_order}" + ) + 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 "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()}") + 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() + 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) + except ValueError as exc: + if skip_missing_tables: + log_step(f"跳过字段编组同步: {exc}") + continue + raise + ensure_field_groups_for_table( + client=client, + table=table, + store_field_name=store_field_name, + dry_run=dry_run, + ) + log_step("字段编组同步完成") + + +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: + 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, + store_field_name=store_field_name, + allowed_field_names=allowed_field_names, + dry_run=dry_run, + ) + log_step(f"字段编组同步完成: 日期={stat_date.isoformat()}, 模式={day_mode}") + + def sync_metric_table_for_month( client: FeishuBitableClient, table: TableInfo, diff --git a/run_project_bitable_pipeline.py b/run_project_bitable_pipeline.py index e486900..c85b3d1 100644 --- a/run_project_bitable_pipeline.py +++ b/run_project_bitable_pipeline.py @@ -17,6 +17,7 @@ from project_bitable_common import ( run_metric_sync, send_feishu_app_card, send_feishu_sync_card, + sync_field_groups_for_day, validate_date, ) @@ -78,6 +79,14 @@ def main() -> None: 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: