Spaces:

BinbomJared
/

DM-Agent

Runtime error

DM-Agent / src /p3_config.py

wen Qiu

refactor: align column names with updated Sheets headers

8d0d365 8 months ago

13.8 kB

from __future__ import annotations

"""
P3 配置数据加载器。
从 Google Sheets（数据分析工作流 — 依据数据汇总）动态读取：
  - 月度目标（预算、营收目标、ROAS 目标、漏斗预算拆分）
  - 渠道预算分配
  - KPI 与 Benchmark 映射
  - 历史实绩参考
复用 loader.py 的 Sheets 认证逻辑，加内存缓存（TTL 可配）。
"""
import logging
import re
import time
from pathlib import Path
from typing import Any

from .loader import load_config

log = logging.getLogger("ad-agent")

_cache: dict[str, Any] = {}
_cache_ts: float = 0.0
_DEFAULT_TTL = 600  # 10 minutes


def _clean_money(val: str) -> float | None:
    """清洗 $1,234.56 / $1.2M 等格式为 float。"""
    if not val or val in ("—", "待定", ""):
        return None
    s = str(val).strip()
    s = re.sub(r'[$,\s]', '', s)
    try:
        return float(s)
    except ValueError:
        return None


def _clean_pct(val: str) -> float | None:
    """清洗 22.70% 为 0.227，0.14% 为 0.0014。带 % 号的一律除以 100。"""
    if not val or val in ("—", "待定", ""):
        return None
    s = str(val).strip()
    has_pct = '%' in s
    s = s.replace('%', '')
    try:
        v = float(s)
        return v / 100.0 if has_pct else v
    except ValueError:
        return None


def _clean_float(val: str) -> float | None:
    if not val or val in ("—", "待定", ""):
        return None
    s = str(val).strip()
    s = re.sub(r'[$,%\s]', '', s)
    try:
        return float(s)
    except ValueError:
        return None


_MONTH_MAP = {
    "1月": 1, "2月": 2, "3月": 3, "4月": 4, "5月": 5, "6月": 6,
    "7月": 7, "8月": 8, "9月": 9, "10月": 10, "11月": 11, "12月": 12,
}


def _parse_month(raw: str) -> int | None:
    s = str(raw).strip()
    if s in _MONTH_MAP:
        return _MONTH_MAP[s]
    try:
        v = int(s)
        return v if 1 <= v <= 12 else None
    except ValueError:
        return None


def _get_config_spreadsheet(config: dict):
    """
    获取 P3 配置 Sheets 的 gspread spreadsheet 对象。
    独立创建 gspread client（复用同样的认证凭证），直接 open_by_key。
    """
    import gspread
    from google.oauth2.service_account import Credentials
    import json
    import os

    p3_cfg = config.get("p3_config_sheet", {})
    sheet_id = p3_cfg.get("spreadsheet_id", "")
    if not sheet_id:
        raise RuntimeError("未配置 p3_config_sheet.spreadsheet_id")

    scopes = [
        "https://www.googleapis.com/auth/spreadsheets",
        "https://www.googleapis.com/auth/drive",
    ]

    creds = None
    creds_json = os.environ.get("GOOGLE_CREDENTIALS", "").strip()
    if creds_json:
        info = json.loads(creds_json)
        creds = Credentials.from_service_account_info(info, scopes=scopes)
    else:
        gs_cfg = config.get("google_sheets", {})
        creds_file = gs_cfg.get("credentials_file", "credentials.json")
        creds_path = Path(creds_file)
        if not creds_path.is_absolute():
            creds_path = Path(__file__).resolve().parent.parent / creds_path
        if creds_path.exists():
            creds = Credentials.from_service_account_file(str(creds_path), scopes=scopes)
        else:
            raise RuntimeError(f"凭证文件 {creds_path} 不存在，且未设置 GOOGLE_CREDENTIALS 环境变量")

    gc = gspread.authorize(creds)
    return gc.open_by_key(sheet_id)


def _read_sheet(sh, sheet_name: str) -> list[list[str]]:
    try:
        ws = sh.worksheet(sheet_name)
        return ws.get_all_values()
    except Exception as e:
        log.warning(f"读取 Sheet [{sheet_name}] 失败: {e}")
        return []


def load_monthly_targets(sh) -> dict[str, dict[int, dict]]:
    """
    解析「月度目标」sheet。
    返回 {"US": {1: {...}, 2: {...}}, "CA": {1: {...}, ...}}
    """
    rows = _read_sheet(sh, "月度目标")
    if not rows:
        return {}

    result: dict[str, dict[int, dict]] = {"US": {}, "CA": {}}
    section = None

    for row in rows:
        first = str(row[0]).strip() if row else ""
        second = str(row[1]).strip() if len(row) > 1 else ""

        if "US 月度预算目标" in second:
            section = "US"
            continue
        if "CA 月度预算目标" in second:
            section = "CA"
            continue
        if first in ("月份", "", "全年合计") or not section:
            continue

        month = _parse_month(first)
        if month is None:
            continue

        if section == "US":
            result["US"][month] = {
                "spf_target": _clean_money(row[1] if len(row) > 1 else ""),
                "dm_target": _clean_money(row[2] if len(row) > 2 else ""),
                "dm_budget": _clean_money(row[3] if len(row) > 3 else ""),
                "target_roas": _clean_float(row[4] if len(row) > 4 else ""),
                "awareness_budget": _clean_money(row[5] if len(row) > 5 else ""),
                "consideration_budget": _clean_money(row[6] if len(row) > 6 else ""),
                "conversion_budget": _clean_money(row[7] if len(row) > 7 else ""),
            }
        elif section == "CA":
            result["CA"][month] = {
                "spf_target": _clean_money(row[1] if len(row) > 1 else ""),
                "dm_target": _clean_money(row[2] if len(row) > 2 else ""),
                "dm_budget": _clean_money(row[3] if len(row) > 3 else ""),
                "target_roas": _clean_float(row[4] if len(row) > 4 else ""),
                "cad_usd_rate": _clean_float(row[5] if len(row) > 5 else ""),
            }

    return result


def load_channel_budgets(sh) -> dict[str, dict]:
    """
    解析「渠道与漏斗」sheet。
    返回渠道预算分配和漏斗配比。
    """
    rows = _read_sheet(sh, "渠道与漏斗")
    if not rows:
        return {}

    result: dict[str, Any] = {"US": {"channel_budgets": {}, "funnel_split": {}}}
    section = None

    for row in rows:
        first = str(row[0]).strip() if row else ""
        second = str(row[1]).strip() if len(row) > 1 else ""

        if "US 渠道预算分配" in second:
            section = "us_channel"
            continue
        if "US 漏斗预算配比" in second:
            section = "us_funnel"
            continue
        if "CA 漏斗预算配比" in second:
            section = "ca_funnel"
            continue
        if "预算来源结构" in second or "预算来源说明" in second or "口径定义" in second:
            section = None
            continue

        if first in ("月份", "", "全年合计") or not section:
            continue

        month = _parse_month(first)

        if section == "us_channel" and month:
            channels = {}
            channel_names = ["Google", "Meta", "TikTok", "Criteo", "Bing"]
            for i, ch in enumerate(channel_names):
                val = _clean_money(row[i + 1] if len(row) > i + 1 else "")
                if val is not None:
                    channels[ch] = val
            if channels:
                result["US"]["channel_budgets"][month] = channels

        elif section == "us_funnel" and month:
            result["US"]["funnel_split"][month] = {
                "awareness": _clean_money(row[1] if len(row) > 1 else ""),
                "consideration": _clean_money(row[2] if len(row) > 2 else ""),
                "conversion": _clean_money(row[3] if len(row) > 3 else ""),
                "total": _clean_money(row[4] if len(row) > 4 else ""),
                "awareness_pct": _clean_pct(row[5] if len(row) > 5 else ""),
                "consideration_pct": _clean_pct(row[6] if len(row) > 6 else ""),
                "conversion_pct": _clean_pct(row[7] if len(row) > 7 else ""),
            }

    return result


def load_kpi_config(sh) -> dict[str, Any]:
    """
    解析「KPI与Benchmark」sheet。
    返回各优化目标的 KPI 映射 + Benchmark 倒推 + 历史参考。
    """
    rows = _read_sheet(sh, "KPI与Benchmark")
    if not rows:
        return {}

    result: dict[str, Any] = {
        "objective_kpi": {},
        "benchmark_chain": {},
        "reference_2025": {},
    }
    section = None

    for row in rows:
        first = str(row[0]).strip() if row else ""
        second = str(row[1]).strip() if len(row) > 1 else ""

        if ("各广告目的" in second or "各优化目标" in second) and "核心指标映射" in second:
            section = "kpi_map"
            continue
        if "Benchmark 倒推" in second:
            section = "benchmark_chain"
            continue
        if "全年实绩参考" in second:
            section = "reference"
            continue

        if first in ("广告目的", "优化目标", "指标", "") or not section:
            continue

        if section == "kpi_map" and first:
            result["objective_kpi"][first] = {
                "core_kpi": str(row[1]).strip() if len(row) > 1 else "",
                "compute_method": str(row[2]).strip() if len(row) > 2 else "",
                "funnel_level": str(row[3]).strip() if len(row) > 3 else "",
            }

        elif section == "benchmark_chain" and first:
            result["benchmark_chain"][first] = {
                "formula": str(row[1]).strip() if len(row) > 1 else "",
                "us_example": _clean_float(row[2] if len(row) > 2 else ""),
                "ca_example": _clean_float(row[3] if len(row) > 3 else ""),
            }

        elif section == "reference" and first:
            result["reference_2025"][first] = {
                "US": _clean_float(row[1] if len(row) > 1 else ""),
                "CA": _clean_float(row[2] if len(row) > 2 else ""),
            }

    return result


def load_historical_benchmarks(sh) -> dict[str, Any]:
    """
    解析「历史实绩」sheet。
    返回渠道×漏斗实绩 + Campaign 类型 ROAS + 视频内容指标。
    """
    rows = _read_sheet(sh, "历史实绩")
    if not rows:
        return {}

    result: dict[str, Any] = {
        "us_campaign_roas": {},
        "ca_channel_funnel": {},
        "us_video_metrics": {},
    }
    section = None

    for row in rows:
        first = str(row[0]).strip() if row else ""
        second = str(row[1]).strip() if len(row) > 1 else ""

        if "US 2025 Campaign Type ROAS" in second:
            section = "us_roas"
            continue
        if "CA 2025 渠道" in second:
            section = "ca_channel"
            continue
        if "US 2025 视频内容" in second:
            section = "us_video"
            continue

        if first in ("Campaign 类型", "渠道", "指标", "") or not section:
            continue

        if section == "us_roas" and first:
            result["us_campaign_roas"][first] = {
                "avg_roas": _clean_float(row[1] if len(row) > 1 else ""),
                "range": str(row[2]).strip() if len(row) > 2 else "",
            }

        elif section == "ca_channel" and first:
            platform = str(row[0]).strip()
            funnel = str(row[1]).strip() if len(row) > 1 else ""
            key = f"{platform}_{funnel}"
            result["ca_channel_funnel"][key] = {
                "spend": _clean_money(row[2] if len(row) > 2 else ""),
                "roas": _clean_float(row[3] if len(row) > 3 else ""),
                "cpc": _clean_money(row[4] if len(row) > 4 else ""),
                "ctr": _clean_pct(row[5] if len(row) > 5 else ""),
                "cpm": _clean_money(row[6] if len(row) > 6 else ""),
            }

        elif section == "us_video" and first:
            result["us_video_metrics"][first] = {
                "avg": _clean_pct(row[1] if len(row) > 1 else ""),
                "new_product": _clean_pct(row[2] if len(row) > 2 else ""),
                "regular": _clean_pct(row[3] if len(row) > 3 else ""),
                "promo": _clean_pct(row[4] if len(row) > 4 else ""),
            }

    return result


def load_all_p3_config(config_path: Path | None = None) -> dict[str, Any]:
    """
    一次性加载所有 P3 配置数据，带内存缓存。
    """
    global _cache, _cache_ts

    config = load_config(config_path)
    p3_cfg = config.get("p3_config_sheet", {})
    ttl = p3_cfg.get("cache_ttl_minutes", 10) * 60

    now = time.time()
    if _cache and (now - _cache_ts) < ttl:
        return _cache

    try:
        sh = _get_config_spreadsheet(config)
    except Exception as e:
        if _cache:
            log.warning(f"连接 Sheets 失败({e})，使用过期缓存")
            return _cache
        raise RuntimeError(
            f"无法连接 P3 配置 Google Sheets: {e}\n"
            "P3 配置复用与广告数据读写相同的 Google 服务账号。请确认:\n"
            "  1. 现有的 Google Sheets 数据读写功能可正常工作（credentials.json 或 GOOGLE_CREDENTIALS 已配置）\n"
            "  2. 该服务账号已被授权访问 P3 配置 Sheets（ID: 12Ex-5SXHOS2-j7UMyP2_UR9NV0QJ1cZE1u7W5griOMw）\n"
            "  3. config.yaml 中 p3_config_sheet.spreadsheet_id 配置正确"
        ) from e

    result = {
        "monthly_targets": load_monthly_targets(sh),
        "channel_budgets": load_channel_budgets(sh),
        "kpi_config": load_kpi_config(sh),
        "historical": load_historical_benchmarks(sh),
    }

    empty_sections = [k for k, v in result.items() if not v]
    if empty_sections:
        log.warning(f"以下 Sheets 数据为空（可能 sheet 名不匹配或无数据）: {empty_sections}")

    _cache = result
    _cache_ts = now
    log.info("P3 配置数据已从 Google Sheets 加载")
    return result


def invalidate_cache():
    """手动清除缓存。"""
    global _cache, _cache_ts
    _cache = {}
    _cache_ts = 0.0

	from __future__ import annotations

	"""
	P3 配置数据加载器。

	从 Google Sheets（数据分析工作流 — 依据数据汇总）动态读取：
	- 月度目标（预算、营收目标、ROAS 目标、漏斗预算拆分）
	- 渠道预算分配
	- KPI 与 Benchmark 映射
	- 历史实绩参考

	复用 loader.py 的 Sheets 认证逻辑，加内存缓存（TTL 可配）。
	"""
	import logging
	import re
	import time
	from pathlib import Path
	from typing import Any

	from .loader import load_config

	log = logging.getLogger("ad-agent")

	_cache: dict[str, Any] = {}
	_cache_ts: float = 0.0
	_DEFAULT_TTL = 600 # 10 minutes


	def _clean_money(val: str) -> float \| None:
	"""清洗 $1,234.56 / $1.2M 等格式为 float。"""
	if not val or val in ("—", "待定", ""):
	return None
	s = str(val).strip()
	s = re.sub(r'[$,\s]', '', s)
	try:
	return float(s)
	except ValueError:
	return None


	def _clean_pct(val: str) -> float \| None:
	"""清洗 22.70% 为 0.227，0.14% 为 0.0014。带 % 号的一律除以 100。"""
	if not val or val in ("—", "待定", ""):
	return None
	s = str(val).strip()
	has_pct = '%' in s
	s = s.replace('%', '')
	try:
	v = float(s)
	return v / 100.0 if has_pct else v
	except ValueError:
	return None


	def _clean_float(val: str) -> float \| None:
	if not val or val in ("—", "待定", ""):
	return None
	s = str(val).strip()
	s = re.sub(r'[$,%\s]', '', s)
	try:
	return float(s)
	except ValueError:
	return None


	_MONTH_MAP = {
	"1月": 1, "2月": 2, "3月": 3, "4月": 4, "5月": 5, "6月": 6,
	"7月": 7, "8月": 8, "9月": 9, "10月": 10, "11月": 11, "12月": 12,
	}


	def _parse_month(raw: str) -> int \| None:
	s = str(raw).strip()
	if s in _MONTH_MAP:
	return _MONTH_MAP[s]
	try:
	v = int(s)
	return v if 1 <= v <= 12 else None
	except ValueError:
	return None


	def _get_config_spreadsheet(config: dict):
	"""
	获取 P3 配置 Sheets 的 gspread spreadsheet 对象。
	独立创建 gspread client（复用同样的认证凭证），直接 open_by_key。
	"""
	import gspread
	from google.oauth2.service_account import Credentials
	import json
	import os

	p3_cfg = config.get("p3_config_sheet", {})
	sheet_id = p3_cfg.get("spreadsheet_id", "")
	if not sheet_id:
	raise RuntimeError("未配置 p3_config_sheet.spreadsheet_id")

	scopes = [
	"https://www.googleapis.com/auth/spreadsheets",
	"https://www.googleapis.com/auth/drive",
	]

	creds = None
	creds_json = os.environ.get("GOOGLE_CREDENTIALS", "").strip()
	if creds_json:
	info = json.loads(creds_json)
	creds = Credentials.from_service_account_info(info, scopes=scopes)
	else:
	gs_cfg = config.get("google_sheets", {})
	creds_file = gs_cfg.get("credentials_file", "credentials.json")
	creds_path = Path(creds_file)
	if not creds_path.is_absolute():
	creds_path = Path(__file__).resolve().parent.parent / creds_path
	if creds_path.exists():
	creds = Credentials.from_service_account_file(str(creds_path), scopes=scopes)
	else:
	raise RuntimeError(f"凭证文件 {creds_path} 不存在，且未设置 GOOGLE_CREDENTIALS 环境变量")

	gc = gspread.authorize(creds)
	return gc.open_by_key(sheet_id)


	def _read_sheet(sh, sheet_name: str) -> list[list[str]]:
	try:
	ws = sh.worksheet(sheet_name)
	return ws.get_all_values()
	except Exception as e:
	log.warning(f"读取 Sheet [{sheet_name}] 失败: {e}")
	return []


	def load_monthly_targets(sh) -> dict[str, dict[int, dict]]:
	"""
	解析「月度目标」sheet。
	返回 {"US": {1: {...}, 2: {...}}, "CA": {1: {...}, ...}}
	"""
	rows = _read_sheet(sh, "月度目标")
	if not rows:
	return {}

	result: dict[str, dict[int, dict]] = {"US": {}, "CA": {}}
	section = None

	for row in rows:
	first = str(row[0]).strip() if row else ""
	second = str(row[1]).strip() if len(row) > 1 else ""

	if "US 月度预算目标" in second:
	section = "US"
	continue
	if "CA 月度预算目标" in second:
	section = "CA"
	continue
	if first in ("月份", "", "全年合计") or not section:
	continue

	month = _parse_month(first)
	if month is None:
	continue

	if section == "US":
	result["US"][month] = {
	"spf_target": _clean_money(row[1] if len(row) > 1 else ""),
	"dm_target": _clean_money(row[2] if len(row) > 2 else ""),
	"dm_budget": _clean_money(row[3] if len(row) > 3 else ""),
	"target_roas": _clean_float(row[4] if len(row) > 4 else ""),
	"awareness_budget": _clean_money(row[5] if len(row) > 5 else ""),
	"consideration_budget": _clean_money(row[6] if len(row) > 6 else ""),
	"conversion_budget": _clean_money(row[7] if len(row) > 7 else ""),
	}
	elif section == "CA":
	result["CA"][month] = {
	"spf_target": _clean_money(row[1] if len(row) > 1 else ""),
	"dm_target": _clean_money(row[2] if len(row) > 2 else ""),
	"dm_budget": _clean_money(row[3] if len(row) > 3 else ""),
	"target_roas": _clean_float(row[4] if len(row) > 4 else ""),
	"cad_usd_rate": _clean_float(row[5] if len(row) > 5 else ""),
	}

	return result


	def load_channel_budgets(sh) -> dict[str, dict]:
	"""
	解析「渠道与漏斗」sheet。
	返回渠道预算分配和漏斗配比。
	"""
	rows = _read_sheet(sh, "渠道与漏斗")
	if not rows:
	return {}

	result: dict[str, Any] = {"US": {"channel_budgets": {}, "funnel_split": {}}}
	section = None

	for row in rows:
	first = str(row[0]).strip() if row else ""
	second = str(row[1]).strip() if len(row) > 1 else ""

	if "US 渠道预算分配" in second:
	section = "us_channel"
	continue
	if "US 漏斗预算配比" in second:
	section = "us_funnel"
	continue
	if "CA 漏斗预算配比" in second:
	section = "ca_funnel"
	continue
	if "预算来源结构" in second or "预算来源说明" in second or "口径定义" in second:
	section = None
	continue

	if first in ("月份", "", "全年合计") or not section:
	continue

	month = _parse_month(first)

	if section == "us_channel" and month:
	channels = {}
	channel_names = ["Google", "Meta", "TikTok", "Criteo", "Bing"]
	for i, ch in enumerate(channel_names):
	val = _clean_money(row[i + 1] if len(row) > i + 1 else "")
	if val is not None:
	channels[ch] = val
	if channels:
	result["US"]["channel_budgets"][month] = channels

	elif section == "us_funnel" and month:
	result["US"]["funnel_split"][month] = {
	"awareness": _clean_money(row[1] if len(row) > 1 else ""),
	"consideration": _clean_money(row[2] if len(row) > 2 else ""),
	"conversion": _clean_money(row[3] if len(row) > 3 else ""),
	"total": _clean_money(row[4] if len(row) > 4 else ""),
	"awareness_pct": _clean_pct(row[5] if len(row) > 5 else ""),
	"consideration_pct": _clean_pct(row[6] if len(row) > 6 else ""),
	"conversion_pct": _clean_pct(row[7] if len(row) > 7 else ""),
	}

	return result


	def load_kpi_config(sh) -> dict[str, Any]:
	"""
	解析「KPI与Benchmark」sheet。
	返回各优化目标的 KPI 映射 + Benchmark 倒推 + 历史参考。
	"""
	rows = _read_sheet(sh, "KPI与Benchmark")
	if not rows:
	return {}

	result: dict[str, Any] = {
	"objective_kpi": {},
	"benchmark_chain": {},
	"reference_2025": {},
	}
	section = None

	for row in rows:
	first = str(row[0]).strip() if row else ""
	second = str(row[1]).strip() if len(row) > 1 else ""

	if ("各广告目的" in second or "各优化目标" in second) and "核心指标映射" in second:
	section = "kpi_map"
	continue
	if "Benchmark 倒推" in second:
	section = "benchmark_chain"
	continue
	if "全年实绩参考" in second:
	section = "reference"
	continue

	if first in ("广告目的", "优化目标", "指标", "") or not section:
	continue

	if section == "kpi_map" and first:
	result["objective_kpi"][first] = {
	"core_kpi": str(row[1]).strip() if len(row) > 1 else "",
	"compute_method": str(row[2]).strip() if len(row) > 2 else "",
	"funnel_level": str(row[3]).strip() if len(row) > 3 else "",
	}

	elif section == "benchmark_chain" and first:
	result["benchmark_chain"][first] = {
	"formula": str(row[1]).strip() if len(row) > 1 else "",
	"us_example": _clean_float(row[2] if len(row) > 2 else ""),
	"ca_example": _clean_float(row[3] if len(row) > 3 else ""),
	}

	elif section == "reference" and first:
	result["reference_2025"][first] = {
	"US": _clean_float(row[1] if len(row) > 1 else ""),
	"CA": _clean_float(row[2] if len(row) > 2 else ""),
	}

	return result


	def load_historical_benchmarks(sh) -> dict[str, Any]:
	"""
	解析「历史实绩」sheet。
	返回渠道×漏斗实绩 + Campaign 类型 ROAS + 视频内容指标。
	"""
	rows = _read_sheet(sh, "历史实绩")
	if not rows:
	return {}

	result: dict[str, Any] = {
	"us_campaign_roas": {},
	"ca_channel_funnel": {},
	"us_video_metrics": {},
	}
	section = None

	for row in rows:
	first = str(row[0]).strip() if row else ""
	second = str(row[1]).strip() if len(row) > 1 else ""

	if "US 2025 Campaign Type ROAS" in second:
	section = "us_roas"
	continue
	if "CA 2025 渠道" in second:
	section = "ca_channel"
	continue
	if "US 2025 视频内容" in second:
	section = "us_video"
	continue

	if first in ("Campaign 类型", "渠道", "指标", "") or not section:
	continue

	if section == "us_roas" and first:
	result["us_campaign_roas"][first] = {
	"avg_roas": _clean_float(row[1] if len(row) > 1 else ""),
	"range": str(row[2]).strip() if len(row) > 2 else "",
	}

	elif section == "ca_channel" and first:
	platform = str(row[0]).strip()
	funnel = str(row[1]).strip() if len(row) > 1 else ""
	key = f"{platform}_{funnel}"
	result["ca_channel_funnel"][key] = {
	"spend": _clean_money(row[2] if len(row) > 2 else ""),
	"roas": _clean_float(row[3] if len(row) > 3 else ""),
	"cpc": _clean_money(row[4] if len(row) > 4 else ""),
	"ctr": _clean_pct(row[5] if len(row) > 5 else ""),
	"cpm": _clean_money(row[6] if len(row) > 6 else ""),
	}

	elif section == "us_video" and first:
	result["us_video_metrics"][first] = {
	"avg": _clean_pct(row[1] if len(row) > 1 else ""),
	"new_product": _clean_pct(row[2] if len(row) > 2 else ""),
	"regular": _clean_pct(row[3] if len(row) > 3 else ""),
	"promo": _clean_pct(row[4] if len(row) > 4 else ""),
	}

	return result


	def load_all_p3_config(config_path: Path \| None = None) -> dict[str, Any]:
	"""
	一次性加载所有 P3 配置数据，带内存缓存。
	"""
	global _cache, _cache_ts

	config = load_config(config_path)
	p3_cfg = config.get("p3_config_sheet", {})
	ttl = p3_cfg.get("cache_ttl_minutes", 10) * 60

	now = time.time()
	if _cache and (now - _cache_ts) < ttl:
	return _cache

	try:
	sh = _get_config_spreadsheet(config)
	except Exception as e:
	if _cache:
	log.warning(f"连接 Sheets 失败({e})，使用过期缓存")
	return _cache
	raise RuntimeError(
	f"无法连接 P3 配置 Google Sheets: {e}\n"
	"P3 配置复用与广告数据读写相同的 Google 服务账号。请确认:\n"
	" 1. 现有的 Google Sheets 数据读写功能可正常工作（credentials.json 或 GOOGLE_CREDENTIALS 已配置）\n"
	" 2. 该服务账号已被授权访问 P3 配置 Sheets（ID: 12Ex-5SXHOS2-j7UMyP2_UR9NV0QJ1cZE1u7W5griOMw）\n"
	" 3. config.yaml 中 p3_config_sheet.spreadsheet_id 配置正确"
	) from e

	result = {
	"monthly_targets": load_monthly_targets(sh),
	"channel_budgets": load_channel_budgets(sh),
	"kpi_config": load_kpi_config(sh),
	"historical": load_historical_benchmarks(sh),
	}

	empty_sections = [k for k, v in result.items() if not v]
	if empty_sections:
	log.warning(f"以下 Sheets 数据为空（可能 sheet 名不匹配或无数据）: {empty_sections}")

	_cache = result
	_cache_ts = now
	log.info("P3 配置数据已从 Google Sheets 加载")
	return result


	def invalidate_cache():
	"""手动清除缓存。"""
	global _cache, _cache_ts
	_cache = {}
	_cache_ts = 0.0