from __future__ import annotations """ 数据加载：支持 Google Sheets（优先）和本地 Excel 两种数据源。 处理编码异常列名、过滤汇总行、从 Date 派生 year/week/month。 """ import logging import os from pathlib import Path from typing import Any import pandas as pd import yaml log = logging.getLogger("ad-agent") def _project_root() -> Path: return Path(__file__).resolve().parent.parent def load_config(config_path: Path | None = None) -> dict[str, Any]: path = config_path or (_project_root() / "config.yaml") with open(path, "r", encoding="utf-8") as f: return yaml.safe_load(f) # ───────────────────────────────────────────── # Google Sheets 数据源 # ───────────────────────────────────────────── def _build_google_creds(config: dict[str, Any]): """ 构建 Google API 凭证，支持三种认证方式（按优先级）： 1. GOOGLE_CREDENTIALS 环境变量（服务账号 JSON 字符串，云端部署推荐） 2. GOOGLE_OAUTH_REFRESH_TOKEN 环境变量（OAuth refresh token，GitHub Actions 等场景） 3. credentials.json 本地文件（本地开发） """ from google.oauth2.service_account import Credentials as SACredentials from google.oauth2.credentials import Credentials as OAuthCredentials import json scopes = [ "https://www.googleapis.com/auth/spreadsheets", "https://www.googleapis.com/auth/drive", ] creds_json = os.environ.get("GOOGLE_CREDENTIALS", "").strip() if creds_json: info = json.loads(creds_json) creds = SACredentials.from_service_account_info(info, scopes=scopes) log.info("使用环境变量 GOOGLE_CREDENTIALS（服务账号）认证") return creds refresh_token = os.environ.get("GOOGLE_OAUTH_REFRESH_TOKEN", "").strip() client_id = os.environ.get("GOOGLE_OAUTH_CLIENT_ID", "").strip() client_secret = os.environ.get("GOOGLE_OAUTH_CLIENT_SECRET", "").strip() if refresh_token and client_id and client_secret: creds = OAuthCredentials( token=None, refresh_token=refresh_token, client_id=client_id, client_secret=client_secret, token_uri="https://oauth2.googleapis.com/token", ) log.info("使用 OAuth refresh token 认证") return creds 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 = _project_root() / creds_path if creds_path.exists(): creds = SACredentials.from_service_account_file(str(creds_path), scopes=scopes) log.info(f"使用凭证文件 {creds_path} 认证") return creds return None def _load_from_google_sheets(config: dict[str, Any]) -> pd.DataFrame | None: """ 从 Google Sheets 分平台表（Meta/Google/Criteo/Bing）读取 API 同步数据。 单一数据源，不读 Export 表，避免多源合并导致的重复。 """ gs_cfg = config.get("google_sheets", {}) if not gs_cfg.get("enabled"): return None url = gs_cfg.get("spreadsheet_url", "").strip() if not url: log.warning("Google Sheets 已启用但未配置 spreadsheet_url，回退到本地 Excel") return None try: import gspread creds = _build_google_creds(config) if creds is None: log.warning("无可用 Google 认证凭证，回退到本地 Excel") return None gc = gspread.authorize(creds) sh = gc.open_by_url(url) platform_sheets = gs_cfg.get("platform_sheets", ["Meta", "Google", "Criteo", "Bing"]) col_cfg = config.get("columns", {}) idx_rename = { col_cfg.get("country_idx", 0): "country", col_cfg.get("platform_idx", 1): "platform", col_cfg.get("campaign_type_idx", 2): "campaign_type", col_cfg.get("campaign_name_idx", 3): "campaign_name", col_cfg.get("date_idx", 4): "date", } def _read_one_sheet(ws_name: str) -> pd.DataFrame | None: try: ws = sh.worksheet(ws_name) records = ws.get_all_values() if len(records) < 2: return None header = records[0] seen: dict[str, int] = {} unique_header = [] for col in header: if col in seen: seen[col] += 1 unique_header.append(f"{col}_{seen[col]}") else: seen[col] = 0 unique_header.append(col) sdf = pd.DataFrame(records[1:], columns=unique_header) rename = {} for idx, name in idx_rename.items(): if idx < len(sdf.columns): rename[sdf.columns[idx]] = name sdf = sdf.rename(columns=rename) log.info(f"从 Sheet [{ws_name}] 读取 {len(sdf)} 行") return sdf except Exception: return None all_dfs: list[pd.DataFrame] = [] for ps in platform_sheets: ps_df = _read_one_sheet(ps) if ps_df is not None and not ps_df.empty: all_dfs.append(ps_df) if not all_dfs: log.warning("Google Sheets 所有分平台工作表均无数据") return None df = pd.concat(all_dfs, ignore_index=True) log.info(f"从 Google Sheets 分平台表共读取 {len(df)} 行数据（{', '.join(platform_sheets)}）") return df except Exception as e: log.error(f"Google Sheets 读取失败: {e}，回退到本地 Excel") return None # ───────────────────────────────────────────── # 本地 Excel 数据源 # ───────────────────────────────────────────── def _load_from_excel(config: dict[str, Any]) -> pd.DataFrame: """从 data 目录读取最新的 .xlsx 文件。""" root = _project_root() data_dir = Path(config.get("data_dir", "data")) if not data_dir.is_absolute(): data_dir = root / data_dir files = sorted( (f for f in data_dir.glob("*.xlsx") if not f.name.startswith("~$")), key=lambda p: p.stat().st_mtime, reverse=True, ) if not files: raise FileNotFoundError(f"在 {data_dir} 中未找到 .xlsx 文件") sheet = config.get("excel", {}).get("sheet", 0) df = pd.read_excel(files[0], sheet_name=sheet, engine="openpyxl") log.info(f"从本地 Excel 读取: {files[0].name}（{len(df)} 行）") return df # ───────────────────────────────────────────── # 写回 Google Sheets # ───────────────────────────────────────────── def _get_spreadsheet(config: dict[str, Any]): """获取 gspread spreadsheet 对象（读写权限）。""" import gspread gs_cfg = config.get("google_sheets", {}) url = gs_cfg.get("spreadsheet_url", "").strip() if not url: raise RuntimeError("未配置 spreadsheet_url") creds = _build_google_creds(config) if creds is None: raise RuntimeError("无可用 Google 认证凭证（需要 GOOGLE_CREDENTIALS、GOOGLE_OAUTH_REFRESH_TOKEN 或 credentials.json）") gc = gspread.authorize(creds) return gc.open_by_url(url) def _get_or_create_worksheet(sh, sheet_name: str, header: list[str]): """获取或创建指定名称的 worksheet，确保有表头。""" import gspread try: ws = sh.worksheet(sheet_name) except gspread.WorksheetNotFound: ws = sh.add_worksheet(title=sheet_name, rows=1000, cols=len(header)) ws.append_row(header, value_input_option="RAW") log.info(f"创建新 Sheet: {sheet_name}") return ws def _build_header(config: dict[str, Any]) -> list[str]: """根据 config 构建标准表头。""" col_cfg = config.get("columns", {}) dim_count = 5 if col_cfg.get("ad_id_idx") is not None: dim_count = max(dim_count, col_cfg["ad_id_idx"] + 1) dims = [""] * dim_count dims[col_cfg.get("country_idx", 0)] = "Country" dims[col_cfg.get("platform_idx", 1)] = "Platform" dims[col_cfg.get("campaign_type_idx", 2)] = "Campaign Type" dims[col_cfg.get("campaign_name_idx", 3)] = "Campaign Name" dims[col_cfg.get("date_idx", 4)] = "Date" if col_cfg.get("adset_name_idx") is not None: dims[col_cfg["adset_name_idx"]] = "Ad Set Name" if col_cfg.get("ad_name_idx") is not None: dims[col_cfg["ad_name_idx"]] = "Ad Name" if col_cfg.get("ad_id_idx") is not None: dims[col_cfg["ad_id_idx"]] = "Ad ID" return dims + config.get("metrics", []) def write_api_data_to_sheets( df: pd.DataFrame, config_path: Path | None = None, ) -> dict[str, int]: """ 将 API 拉取的数据按平台写入不同的 Google Sheets worksheet。 每个平台一个 sheet（如 Meta、Google、Criteo），自动创建不存在的 sheet。 按 (country, date, campaign_name, ad_name) 去重，只追加新行。 """ config = load_config(config_path) sh = _get_spreadsheet(config) col_cfg = config.get("columns", {}) country_idx = col_cfg.get("country_idx", 0) date_idx = col_cfg.get("date_idx", 4) campaign_name_idx = col_cfg.get("campaign_name_idx", 3) metrics_list = config.get("metrics", []) header = _build_header(config) key_indices = [country_idx, date_idx, campaign_name_idx] platforms = df["platform"].unique() if "platform" in df.columns else [] total_added = 0 for platform in platforms: p_df = df[df["platform"] == platform] ws = _get_or_create_worksheet(sh, platform, header) all_values = ws.get_all_values() col_order = all_values[0] if all_values else header existing_rows = all_values[1:] if len(all_values) > 1 else [] existing_keys = set() for row in existing_rows: if len(row) > max(key_indices): key = tuple(row[i].strip() for i in key_indices) existing_keys.add(key) new_rows = [] for _, row_data in p_df.iterrows(): dt = row_data.get("date") date_str = dt.strftime("%Y-%m-%d") if hasattr(dt, "strftime") else str(dt).split(" ")[0] campaign_name = str(row_data.get("campaign_name", "")) ad_name = str(row_data.get("ad_name", "")) country = str(row_data.get("country", "")) key = (country, date_str, campaign_name, ad_name) if key in existing_keys: continue out_row = [""] * len(col_order) dim_map = { col_cfg.get("country_idx", 0): str(row_data.get("country", "")), col_cfg.get("platform_idx", 1): platform, col_cfg.get("campaign_type_idx", 2): str(row_data.get("campaign_type", "")), campaign_name_idx: campaign_name, date_idx: date_str, } adset_name_idx = col_cfg.get("adset_name_idx") ad_name_idx = col_cfg.get("ad_name_idx") ad_id_idx = col_cfg.get("ad_id_idx") if adset_name_idx is not None: dim_map[adset_name_idx] = str(row_data.get("adset_name", "")) if ad_name_idx is not None: dim_map[ad_name_idx] = ad_name if ad_id_idx is not None: dim_map[ad_id_idx] = str(row_data.get("ad_id", "")) for idx, val in dim_map.items(): if idx < len(out_row): out_row[idx] = val for col_name in col_order: if col_name in metrics_list and col_name in row_data.index: val = row_data[col_name] if val is None or (isinstance(val, float) and val != val): continue col_idx = col_order.index(col_name) v = float(val) out_row[col_idx] = str(int(v)) if v == int(v) else str(v) new_rows.append(out_row) existing_keys.add(key) if new_rows: ws.append_rows(new_rows, value_input_option="USER_ENTERED") log.info(f"写入 Sheet [{platform}]: {len(new_rows)} 行新数据") total_added += len(new_rows) return {"added": total_added} # ───────────────────────────────────────────── # 统一加载入口 # ───────────────────────────────────────────── _data_source: str = "unknown" def load_raw(config_path: Path | None = None) -> pd.DataFrame: """优先 Google Sheets 分平台表，回退本地 Excel。""" global _data_source config = load_config(config_path) df = _load_from_google_sheets(config) if df is not None: _data_source = "sheets" else: df = _load_from_excel(config) _data_source = "excel" return df def normalize_columns(df: pd.DataFrame, config: dict[str, Any]) -> pd.DataFrame: """用列索引将编码异常的列重命名为可读字段名。""" col_cfg = config.get("columns", {}) idx_map = { col_cfg.get("country_idx", 0): "country", col_cfg.get("platform_idx", 1): "platform", col_cfg.get("campaign_type_idx", 2): "campaign_type", col_cfg.get("campaign_name_idx", 3): "campaign_name", col_cfg.get("date_idx", 4): "date", } rename = {} for idx, name in idx_map.items(): if idx < len(df.columns): rename[df.columns[idx]] = name return df.rename(columns=rename) def _clean_numeric(s: pd.Series) -> pd.Series: """ 清洗 Google Sheets 格式化字符串。 带 % 的值先标记，去掉符号后除以 100 还原为小数。 """ raw = s.astype(str) is_pct = raw.str.contains('%', na=False) cleaned = raw.str.replace(r'[$,%\s]', '', regex=True) values = pd.to_numeric(cleaned, errors="coerce") values[is_pct] = values[is_pct] / 100.0 return values def load_daily_data(config_path: Path | None = None) -> pd.DataFrame: """ 加载并清洗数据，只保留日度明细行。 自动从 date 列派生 year、week（ISO week）、month。 兼容本地 Excel（数值）和 Google Sheets（格式化字符串）。 """ config = load_config(config_path) df = load_raw(config_path) df = normalize_columns(df, config) # Excel 层级结构中，子行的维度列为空字符串，需要前向填充 # API 写入的 Sheets 分平台表每行都有完整字段，不需要 ffill if _data_source == "excel": for dim in ["country", "platform", "campaign_type", "campaign_name"]: if dim in df.columns: df[dim] = df[dim].replace("", pd.NA).ffill() df["date"] = pd.to_datetime(df["date"], format="mixed", errors="coerce") df = df[df["date"].notna()].copy() df = df[df["country"].notna() & (~df["country"].astype(str).str.contains("Total|Applied filters", na=False))].copy() df = df[df["platform"].notna() & (df["platform"].astype(str).str.strip() != "Total")].copy() df = df[df["campaign_type"].notna() & (df["campaign_type"].astype(str).str.strip() != "Total")].copy() df = df[df["campaign_name"].notna() & (df["campaign_name"].astype(str).str.strip() != "Total")].copy() df["year"] = df["date"].dt.isocalendar().year.astype(int) df["week"] = df["date"].dt.isocalendar().week.astype(int) df["month"] = df["date"].dt.month metrics = config.get("metrics", []) for col in metrics: if col in df.columns: df[col] = _clean_numeric(df[col]) df = df.reset_index(drop=True) df = _fix_campaign_type_from_name(df) dedup_cols = ["date", "platform", "campaign_name", "country"] if all(c in df.columns for c in dedup_cols): before = len(df) df = df.drop_duplicates(subset=dedup_cols, keep="last") dropped = before - len(df) if dropped: log.info(f"去重删除 {dropped} 行重复数据") df = df.reset_index(drop=True) return df _VALID_FUNNELS = {"Awareness", "Consideration", "Conversion"} def _fix_campaign_type_from_name(df: pd.DataFrame) -> pd.DataFrame: """根据 campaign_name 中的第二段修正 campaign_type（如名称含明确漏斗标识）。""" if "campaign_name" not in df.columns or "campaign_type" not in df.columns: return df parts = df["campaign_name"].astype(str).str.split("_", n=2) name_funnel = parts.str[1] mask = name_funnel.isin(_VALID_FUNNELS) & (name_funnel != df["campaign_type"]) if mask.any(): log.info(f"根据 campaign_name 修正 {mask.sum()} 行 campaign_type") df.loc[mask, "campaign_type"] = name_funnel[mask] return df # 兼容旧接口 load_weekly_data = load_daily_data def get_available_weeks(df: pd.DataFrame) -> list[tuple[int, int]]: """返回可用的 (year, week) 列表，按时间排序。""" pairs = df[["year", "week"]].drop_duplicates().sort_values(["year", "week"]) return list(pairs.itertuples(index=False, name=None)) def get_latest_week(df: pd.DataFrame) -> tuple[int, int]: """返回数据中最新一周的 (year, week)。""" weeks = get_available_weeks(df) if not weeks: raise ValueError("数据中无可用周数据") return weeks[-1] def get_available_months(df: pd.DataFrame) -> list[tuple[int, int]]: """返回可用的 (year, month) 列表，按时间排序。""" cal_year = df["date"].dt.year pairs = pd.DataFrame({"year": cal_year, "month": df["month"]}).drop_duplicates().sort_values(["year", "month"]) return list(pairs.itertuples(index=False, name=None))