"""从飞书「自动化账户」表同步待投放账户配置到 DB。 表格: https://w42nne6hzg.feishu.cn/sheets/D8f4sKEb2hfBnNttX1ucZTklnHf 读取 tab: 自动化账户 (sheet_id=y3uCcz) 规则: - 表头在第 2 行。 - 仅当「是否自动化执行」为「是」时启用/更新账户。 - 「否」会禁用已有的 ad_creation_account_config,不删除历史配置。 - 「初始出价」支持固定值 0.35 或范围 0.28-0.31。 - 「预算（单广告）」为空时使用模板默认预算;不限制/不限/- 按腾讯不限预算 0 传递;数字按元转换为分。 - 「素材来源」为空默认历史素材;支持历史素材、AI生成素材、外部合作素材。 - 「生成失败是否回退历史素材」为空/否默认不回退。 - 「出价方式」为空默认稳定成本;填「最大转化量」时使用「最大转化量出价」作为控制成本。 - 「投放版位」为空使用投放模板默认;填 AIM+ 时使用智能版位。 - 「年龄」支持 30-66+ 或 45-66;为空使用 45-66。 - 「地域」支持省市中文名逗号分隔;填 不限 时不传地域定向;为空使用投放模板。 - 「日期」是飞书配置生效日期,每日只同步当天日期的行。 """ from __future__ import annotations import argparse import json import re import sys from datetime import date, datetime from pathlib import Path from typing import Any from zoneinfo import ZoneInfo import httpx from dotenv import load_dotenv _HERE = Path(__file__).parent load_dotenv(_HERE / ".env") sys.path.insert(0, str(_HERE.parent.parent)) sys.path.insert(0, str(_HERE)) from configure_creation_accounts import ( # noqa: E402 AccountConfigInput, DEFAULT_AGE_MAX, DEFAULT_AGE_MIN, parse_bid, parse_budget_fen, set_account_automation_enabled, upsert_account_config, ) from tools.account_material_strategy import ( # noqa: E402 normalize_material_source, parse_bool_flag, ) from tools.delivery_config import ( # noqa: E402 BID_MODE_MAX_CONVERSION, parse_bid_scene, parse_geo_location_config, parse_placement_config, ) from tools.feishu_doc import ( # noqa: E402 FEISHU_BASE_URL, _auth_headers, _get_tenant_token, ) DEFAULT_SPREADSHEET_TOKEN = "D8f4sKEb2hfBnNttX1ucZTklnHf" DEFAULT_SHEET_ID = "y3uCcz" DEFAULT_RANGE = "A1:N300" HEADER_ROW_INDEX = 1 DATA_START_INDEX = 2 REQUIRED_COLUMNS = { "日期": "config_date", "账户id": "account_id", "人群包": "audience_name", "初始出价": "bid", "预算（单广告）": "budget", "是否自动化执行": "enabled", } OPTIONAL_COLUMNS = { "素材来源": "material_source", "生成失败是否回退历史素材": "ai_fallback_to_history", "出价方式": "bid_scene", "最大转化量出价": "max_conversion_bid", "投放版位": "placement", "版位": "placement", "年龄": "age", "地域": "geo_location", "投放地域": "geo_location", } def _plain_value(value: Any) -> str: """将飞书单元格值（可能是富文本 list/dict）转为纯文本字符串""" if value is None: return "" if isinstance(value, list): parts: list[str] = [] for item in value: if isinstance(item, dict): parts.append(str(item.get("text") or item.get("link") or "")) else: parts.append(str(item)) return "".join(parts).strip() return str(value).strip() def _fetch_rows(spreadsheet_token: str, sheet_id: str, cell_range: str) -> list[list[Any]]: """读取飞书表格指定范围的原始行数据""" token = _get_tenant_token() url = ( f"{FEISHU_BASE_URL}/sheets/v2/spreadsheets/" f"{spreadsheet_token}/values/{sheet_id}!{cell_range}" ) resp = httpx.get(url, headers=_auth_headers(token), timeout=30) resp.raise_for_status() data = resp.json() if data.get("code") != 0: raise RuntimeError(f"读取飞书配置表失败:{json.dumps(data, ensure_ascii=False)[:500]}") return ((data.get("data") or {}).get("valueRange") or {}).get("values") or [] def _header_map(rows: list[list[Any]]) -> dict[str, int]: """解析表头行，返回 列名→列索引 映射；缺少必需列时抛异常""" if len(rows) <= HEADER_ROW_INDEX: raise RuntimeError("飞书配置表缺少表头行") headers = [_plain_value(v) for v in rows[HEADER_ROW_INDEX]] mapping = {name: idx for idx, name in enumerate(headers) if name} missing = [name for name in REQUIRED_COLUMNS if name not in mapping] if missing: raise RuntimeError(f"飞书配置表缺少列:{missing}; 当前表头={headers}") return mapping def _cell(row: list[Any], idx: int) -> str: """安全取行内指定列的纯文本值，越界返回空字符串""" return _plain_value(row[idx]) if idx < len(row) else "" def _parse_account_id(raw: str) -> int | None: """解析账户 ID 文本为 int；空值返回 None，非法值抛 ValueError""" text = raw.strip() if not text: return None # 飞书数字可能读成 "84545647.0"。 try: return int(float(text)) except Exception as e: raise ValueError(f"账户id非法:{raw}") from e def _parse_config_date(raw: str) -> date: text = raw.strip().replace("/", "-") for fmt in ("%Y%m%d", "%Y-%m-%d"): try: return datetime.strptime(text, fmt).date() except ValueError: continue raise ValueError(f"日期非法:{raw}; 请输入 YYYYMMDD 或 YYYY-MM-DD") def _parse_age_range(raw: str) -> tuple[int, int]: text = raw.strip() if not text: return DEFAULT_AGE_MIN, DEFAULT_AGE_MAX matched = re.fullmatch(r"(\d{1,3})\s*[-~至]\s*(\d{1,3})\+?", text) if not matched: raise ValueError(f"年龄格式非法:{raw}; 示例 30-66+") age_min, age_max = (int(value) for value in matched.groups()) if age_min > age_max: raise ValueError(f"年龄范围非法:{raw}") return age_min, age_max def sync_from_feishu( spreadsheet_token: str = DEFAULT_SPREADSHEET_TOKEN, sheet_id: str = DEFAULT_SHEET_ID, cell_range: str = DEFAULT_RANGE, config_date: str | None = None, dry_run: bool = False, strict: bool = True, ) -> dict[str, Any]: """从飞书「自动化账户」表同步账户配置到 DB，返回统计信息（enabled/disabled/skipped/errors）""" rows = _fetch_rows(spreadsheet_token, sheet_id, cell_range) mapping = _header_map(rows) target_config_date = ( _parse_config_date(config_date) if config_date else datetime.now(ZoneInfo("Asia/Shanghai")).date() ) stats = { "config_date": target_config_date.isoformat(), "enabled": 0, "disabled": 0, "skipped": 0, "date_skipped": 0, "errors": 0, } enabled_records: list[AccountConfigInput] = [] disabled_accounts: list[int] = [] for row_num, row in enumerate(rows[DATA_START_INDEX:], start=DATA_START_INDEX + 1): try: account_id = _parse_account_id(_cell(row, mapping["账户id"])) if account_id is None: stats["skipped"] += 1 continue row_config_date = _parse_config_date(_cell(row, mapping["日期"])) if row_config_date != target_config_date: stats["date_skipped"] += 1 continue enabled_text = _cell(row, mapping["是否自动化执行"]) enabled = enabled_text == "是" if not enabled: disabled_accounts.append(account_id) stats["disabled"] += 1 continue audience_name = _cell(row, mapping["人群包"]) bid_text = _cell(row, mapping["初始出价"]) budget_text = _cell(row, mapping["预算（单广告）"]) material_source_text = ( _cell(row, mapping["素材来源"]) if "素材来源" in mapping else "" ) fallback_text = ( _cell(row, mapping["生成失败是否回退历史素材"]) if "生成失败是否回退历史素材" in mapping else "" ) bid_scene_text = ( _cell(row, mapping["出价方式"]) if "出价方式" in mapping else "" ) max_conversion_bid_text = ( _cell(row, mapping["最大转化量出价"]) if "最大转化量出价" in mapping else "" ) placement_col = "投放版位" if "投放版位" in mapping else "版位" placement_text = ( _cell(row, mapping[placement_col]) if placement_col in mapping else "" ) age_text = _cell(row, mapping["年龄"]) if "年龄" in mapping else "" geo_col = "地域" if "地域" in mapping else "投放地域" geo_text = _cell(row, mapping[geo_col]) if geo_col in mapping else "" bid_min, bid_max, bid_amount = parse_bid(bid_text) bid_scene = parse_bid_scene(bid_scene_text) custom_cost_cap_fen = None if bid_scene == BID_MODE_MAX_CONVERSION: if not max_conversion_bid_text: raise ValueError("最大转化量出价不能为空") max_bid_min, max_bid_max, max_bid_amount = parse_bid(max_conversion_bid_text) if max_bid_min != max_bid_max or max_bid_amount is None: raise ValueError("最大转化量出价必须是固定值") custom_cost_cap_fen = max_bid_amount placement = parse_placement_config(placement_text) age_min, age_max = _parse_age_range(age_text) geo_location = parse_geo_location_config(geo_text) enabled_records.append(AccountConfigInput( account_id=account_id, audience_name=audience_name, bid_min_fen=bid_min, bid_max_fen=bid_max, bid_amount_fen=bid_amount, daily_budget_fen=parse_budget_fen(budget_text), enabled=True, material_source=normalize_material_source(material_source_text), ai_fallback_to_history=parse_bool_flag(fallback_text), bid_scene=bid_scene, custom_cost_cap_fen=custom_cost_cap_fen, automatic_site_enabled=( placement.automatic_site_enabled if placement is not None else None ), site_set=(placement.site_set if placement is not None else None), age_min=age_min, age_max=age_max, location_types=( geo_location.location_types if geo_location is not None else None ), region_ids=(geo_location.region_ids if geo_location is not None else None), config_date=row_config_date, )) stats["enabled"] += 1 except Exception as e: stats["errors"] += 1 print(f"row={row_num} ERROR {e}") if strict and stats["errors"]: print("sync summary:", json.dumps(stats, ensure_ascii=False)) raise RuntimeError(f"飞书配置表存在 {stats['errors']} 行错误,停止本轮自动化") for account_id in disabled_accounts: set_account_automation_enabled( account_id, False, config_date=target_config_date, dry_run=dry_run, ) for record in enabled_records: upsert_account_config(record, dry_run=dry_run) print("sync summary:", json.dumps(stats, ensure_ascii=False)) return stats def main() -> int: """命令行入口：解析参数并执行同步，存在错误行且未放行时返回 1""" parser = argparse.ArgumentParser() parser.add_argument("--spreadsheet-token", default=DEFAULT_SPREADSHEET_TOKEN) parser.add_argument("--sheet-id", default=DEFAULT_SHEET_ID) parser.add_argument("--range", default=DEFAULT_RANGE) parser.add_argument("--config-date", help="只同步该配置日期,格式 YYYYMMDD 或 YYYY-MM-DD") parser.add_argument("--dry-run", action="store_true") parser.add_argument("--allow-row-errors", action="store_true") args = parser.parse_args() stats = sync_from_feishu( spreadsheet_token=args.spreadsheet_token, sheet_id=args.sheet_id, cell_range=args.range, config_date=args.config_date, dry_run=args.dry_run, strict=not args.allow_row_errors, ) return 1 if stats["errors"] and not args.allow_row_errors else 0 if __name__ == "__main__": raise SystemExit(main())