"""批量写入待投放账户三元组配置。 用法: python configure_creation_accounts.py --account-id 84545647 --audience-name 'R50*已转化' --bid 0.35 python configure_creation_accounts.py --csv inputs/accounts.csv CSV 字段: account_id,audience_name,bid """ from __future__ import annotations import argparse import csv import json import sys from dataclasses import dataclass from datetime import date from decimal import Decimal, ROUND_HALF_UP from pathlib import Path from dotenv import load_dotenv _HERE = Path(__file__).parent load_dotenv(_HERE / ".env") sys.path.insert(0, str(_HERE)) DEFAULT_DELIVERY_VERSION = "miniapp_user_growth_v1" DEFAULT_BRAND_KEY = "default_piaoquan" DEFAULT_FEEDBACK_KEY = "miniapp_click_default" DEFAULT_AGE_MIN = 45 DEFAULT_AGE_MAX = 66 DEFAULT_SOURCE_ACCOUNT_ID = 55615440 DEFAULT_BID_SCENE = "average_cost" @dataclass(frozen=True) class AccountConfigInput: account_id: int audience_name: str bid_min_fen: int bid_max_fen: int bid_amount_fen: int | None = None daily_budget_fen: int | None = None enabled: bool = True material_source: str = "history" ai_fallback_to_history: bool = False bid_scene: str = DEFAULT_BID_SCENE custom_cost_cap_fen: int | None = None automatic_site_enabled: bool | None = None site_set: list[str] | None = None age_min: int = DEFAULT_AGE_MIN age_max: int = DEFAULT_AGE_MAX location_types: list[str] | None = None region_ids: list[int] | None = None config_date: date | None = None def _bid_to_fen(raw: str) -> int: """将出价文本（元，可带"元"后缀）转换为分（int），四舍五入到分""" text = str(raw).strip().replace("元", "") if not text: raise ValueError("bid 为空") value = Decimal(text).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) return int((value * 100).to_integral_value(rounding=ROUND_HALF_UP)) def parse_bid(raw: str) -> tuple[int, int, int | None]: """解析出价文本，返回 (bid_min_fen, bid_max_fen, bid_amount_fen)。 支持固定值 "0.35" 或范围 "0.28-0.31"；范围模式 bid_amount 为 None。 """ text = str(raw).strip() if "-" in text: left, right = text.split("-", 1) bid_min = _bid_to_fen(left) bid_max = _bid_to_fen(right) if bid_min > bid_max: raise ValueError(f"出价范围非法:{raw}") return bid_min, bid_max, None bid = _bid_to_fen(text) return bid, bid, bid def parse_budget_fen(raw: object) -> int | None: """解析预算文本（元）为分；空/不限制/不限/- 返回 None（使用模板默认预算）""" text = str(raw or "").strip() if not text: return None if text in {"不限制", "不限", "-"}: return 0 # 飞书里预算按元填,腾讯 API 用分。 return _bid_to_fen(text) def ensure_account_delivery_config_columns() -> None: """Add account-level delivery override columns when deploying old DBs.""" from db.connection import get_connection conn = get_connection() try: with conn.cursor() as cur: cur.execute( """ SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'ad_creation_account_config' AND COLUMN_NAME IN ( 'bid_scene', 'custom_cost_cap_fen', 'automatic_site_enabled', 'site_set_json', 'location_types_json', 'region_ids_json', 'config_date' ) """ ) existing = {row["COLUMN_NAME"] for row in (cur.fetchall() or [])} if "bid_scene" not in existing: cur.execute( """ ALTER TABLE ad_creation_account_config ADD COLUMN bid_scene VARCHAR(50) DEFAULT NULL COMMENT '出价场景:BID_SCENE_NORMAL_AVERAGE/BID_SCENE_NORMAL_MAX' AFTER bid_amount_fen """ ) if "custom_cost_cap_fen" not in existing: cur.execute( """ ALTER TABLE ad_creation_account_config ADD COLUMN custom_cost_cap_fen INT DEFAULT NULL COMMENT '最大转化量控制成本(分),用于 custom_cost_cap' AFTER bid_scene """ ) if "automatic_site_enabled" not in existing: cur.execute( """ ALTER TABLE ad_creation_account_config ADD COLUMN automatic_site_enabled BOOLEAN DEFAULT NULL COMMENT '账户级是否开启智能版位;NULL使用模板默认手动版位' AFTER daily_budget_fen """ ) if "site_set_json" not in existing: cur.execute( """ ALTER TABLE ad_creation_account_config ADD COLUMN site_set_json TEXT DEFAULT NULL COMMENT '账户级投放版位覆盖JSON;NULL使用投放模板' AFTER automatic_site_enabled """ ) if "location_types_json" not in existing: cur.execute( """ ALTER TABLE ad_creation_account_config ADD COLUMN location_types_json TEXT DEFAULT NULL COMMENT '账户级地域类型覆盖JSON;NULL使用投放模板' AFTER site_set_json """ ) if "region_ids_json" not in existing: cur.execute( """ ALTER TABLE ad_creation_account_config ADD COLUMN region_ids_json MEDIUMTEXT DEFAULT NULL COMMENT '账户级地域ID覆盖JSON;NULL使用投放模板,空数组表示不限地域' AFTER location_types_json """ ) if "config_date" not in existing: cur.execute( """ ALTER TABLE ad_creation_account_config ADD COLUMN config_date DATE DEFAULT NULL COMMENT '飞书配置生效日期;仅当天配置参与自动创建/补创意' AFTER region_ids_json """ ) conn.commit() finally: conn.close() def _read_inputs(args: argparse.Namespace) -> list[AccountConfigInput]: """从 CSV 文件或命令行参数读取并校验账户配置输入""" rows: list[AccountConfigInput] = [] if args.csv: with open(args.csv, newline="", encoding="utf-8-sig") as f: reader = csv.DictReader(f) for idx, row in enumerate(reader, start=2): try: rows.append(AccountConfigInput( account_id=int(str(row.get("account_id") or "").strip()), audience_name=str(row.get("audience_name") or "").strip(), bid_min_fen=parse_bid(str(row.get("bid") or ""))[0], bid_max_fen=parse_bid(str(row.get("bid") or ""))[1], bid_amount_fen=parse_bid(str(row.get("bid") or ""))[2], )) except Exception as e: raise ValueError(f"CSV 第 {idx} 行格式错误:{e}") from e else: if args.account_id is None or not args.audience_name or not args.bid: raise ValueError("单账户模式必须提供 --account-id --audience-name --bid") bid_min, bid_max, bid_amount = parse_bid(args.bid) rows.append(AccountConfigInput( account_id=int(args.account_id), audience_name=args.audience_name.strip(), bid_min_fen=bid_min, bid_max_fen=bid_max, bid_amount_fen=bid_amount, )) for row in rows: if not row.audience_name: raise ValueError(f"account_id={row.account_id} audience_name 为空") if row.bid_min_fen <= 0 or row.bid_max_fen <= 0: raise ValueError(f"account_id={row.account_id} bid 必须大于 0") return rows def upsert_account_config(row: AccountConfigInput, dry_run: bool = False) -> None: """将账户配置写入 DB（account_whitelist + ad_creation_account_config），存在则更新""" from db.connection import get_connection from tools.account_material_strategy import ensure_account_material_strategy_columns source_account_id = None if row.audience_name == "泛人群" else DEFAULT_SOURCE_ACCOUNT_ID grant_status = "not_required_no_audience_pack" if row.audience_name == "泛人群" else "pending_resolve" audience_pack_id = None if dry_run: print( f"[dry-run] account={row.account_id} audience={row.audience_name} " f"bid={row.bid_min_fen}-{row.bid_max_fen} source={source_account_id} " f"status={grant_status} material_source={row.material_source} " f"ai_fallback={row.ai_fallback_to_history} bid_scene={row.bid_scene} " f"custom_cost_cap={row.custom_cost_cap_fen} " f"automatic_site={row.automatic_site_enabled} site_set={row.site_set} " f"age={row.age_min}-{row.age_max} regions={row.region_ids} " f"config_date={row.config_date}" ) return ensure_account_material_strategy_columns() ensure_account_delivery_config_columns() conn = get_connection() try: with conn.cursor() as cur: cur.execute( """ SELECT user_token FROM account_whitelist WHERE user_token IS NOT NULL ORDER BY updated_at DESC LIMIT 1 """ ) token_row = cur.fetchone() or {} user_token = token_row.get("user_token") cur.execute( """ INSERT INTO account_whitelist (account_id, account_name, business_line, enabled, remark, created_by, updated_by, audience_pack_id, audience_tier_label, user_token, crowd_package) VALUES (%s, %s, %s, TRUE, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE enabled=TRUE, remark=VALUES(remark), updated_by=VALUES(updated_by), audience_pack_id=VALUES(audience_pack_id), audience_tier_label=VALUES(audience_tier_label), user_token=COALESCE(account_whitelist.user_token, VALUES(user_token)), crowd_package=VALUES(crowd_package) """, ( row.account_id, f"auto-create-{row.account_id}", "auto_put_tencent", "configured from account_id/audience_name/bid triple", "configure_creation_accounts", "configure_creation_accounts", audience_pack_id, row.audience_name, user_token, row.audience_name, ), ) cur.execute( """ INSERT INTO ad_creation_account_config (account_id, enabled, delivery_version, brand_key, feedback_key, material_source, ai_fallback_to_history, audience_name, audience_pack_id, audience_tier_label, bid_min_fen, bid_max_fen, bid_amount_fen, bid_scene, custom_cost_cap_fen, daily_budget_fen, automatic_site_enabled, site_set_json, location_types_json, region_ids_json, age_min, age_max, config_date, audience_source_account_id, audience_grant_status, remark, created_by, updated_by) VALUES (%s, TRUE, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE enabled=TRUE, delivery_version=VALUES(delivery_version), brand_key=VALUES(brand_key), feedback_key=VALUES(feedback_key), material_source=VALUES(material_source), ai_fallback_to_history=VALUES(ai_fallback_to_history), audience_pack_id=IF( ad_creation_account_config.audience_name = VALUES(audience_name), ad_creation_account_config.audience_pack_id, VALUES(audience_pack_id) ), audience_tier_label=VALUES(audience_tier_label), bid_min_fen=VALUES(bid_min_fen), bid_max_fen=VALUES(bid_max_fen), bid_amount_fen=VALUES(bid_amount_fen), bid_scene=VALUES(bid_scene), custom_cost_cap_fen=VALUES(custom_cost_cap_fen), daily_budget_fen=VALUES(daily_budget_fen), automatic_site_enabled=VALUES(automatic_site_enabled), site_set_json=VALUES(site_set_json), location_types_json=VALUES(location_types_json), region_ids_json=VALUES(region_ids_json), age_min=VALUES(age_min), age_max=VALUES(age_max), config_date=VALUES(config_date), audience_source_account_id=IF( ad_creation_account_config.audience_name = VALUES(audience_name), ad_creation_account_config.audience_source_account_id, VALUES(audience_source_account_id) ), audience_grant_status=IF( ad_creation_account_config.audience_name = VALUES(audience_name), ad_creation_account_config.audience_grant_status, VALUES(audience_grant_status) ), audience_verified_at=IF( ad_creation_account_config.audience_name = VALUES(audience_name), ad_creation_account_config.audience_verified_at, NULL ), audience_name=VALUES(audience_name), remark=VALUES(remark), updated_by=VALUES(updated_by) """, ( row.account_id, DEFAULT_DELIVERY_VERSION, DEFAULT_BRAND_KEY, DEFAULT_FEEDBACK_KEY, row.material_source, 1 if row.ai_fallback_to_history else 0, row.audience_name, audience_pack_id, row.audience_name, row.bid_min_fen, row.bid_max_fen, row.bid_amount_fen, row.bid_scene, row.custom_cost_cap_fen, row.daily_budget_fen, None if row.automatic_site_enabled is None else (1 if row.automatic_site_enabled else 0), json.dumps(row.site_set, ensure_ascii=False) if row.site_set is not None else None, json.dumps(row.location_types, ensure_ascii=False) if row.location_types is not None else None, json.dumps(row.region_ids, ensure_ascii=False) if row.region_ids is not None else None, row.age_min, row.age_max, row.config_date, source_account_id, grant_status, "configured from account_id/audience_name/bid triple", "configure_creation_accounts", "configure_creation_accounts", ), ) conn.commit() finally: conn.close() print( f"configured account={row.account_id} audience={row.audience_name} " f"bid={row.bid_min_fen / 100:.2f}-{row.bid_max_fen / 100:.2f} " f"bid_scene={row.bid_scene} custom_cost_cap={row.custom_cost_cap_fen} " f"age={row.age_min}-{row.age_max} regions={row.region_ids} " f"config_date={row.config_date} status={grant_status}" ) def set_account_automation_enabled( account_id: int, enabled: bool, config_date: date | None = None, dry_run: bool = False, ) -> None: """启用/禁用账户的自动化投放开关（更新 ad_creation_account_config.enabled）""" if dry_run: print(f"[dry-run] set account={account_id} enabled={enabled} config_date={config_date}") return from db.connection import get_connection conn = get_connection() try: with conn.cursor() as cur: cur.execute( """ UPDATE ad_creation_account_config SET enabled=%s, config_date=%s, updated_by='configure_creation_accounts' WHERE account_id=%s """, (1 if enabled else 0, config_date, account_id), ) conn.commit() finally: conn.close() print(f"set account={account_id} enabled={enabled}") def main() -> int: """命令行入口：读取输入（CSV 或单账户参数）并逐条写入账户配置""" parser = argparse.ArgumentParser() parser.add_argument("--account-id", type=int) parser.add_argument("--audience-name") parser.add_argument("--bid", help="元,例如 0.35") parser.add_argument("--csv", type=Path) parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() rows = _read_inputs(args) for row in rows: upsert_account_config(row, dry_run=args.dry_run) return 0 if __name__ == "__main__": raise SystemExit(main())