Spaces:
Runtime error
Runtime error
| """ | |
| 本地定时同步脚本。 | |
| 由 Windows 任务计划程序每天 09:00 自动调用。 | |
| 拉取各广告平台数据 → 写入 Google Sheets → 保存本地 CSV 备份。 | |
| 目前包含: Meta（HuggingFace 无法连接）、Criteo | |
| """ | |
| import sys | |
| import os | |
| import logging | |
| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parent | |
| sys.path.insert(0, str(ROOT)) | |
| os.chdir(ROOT) | |
| from dotenv import load_dotenv | |
| load_dotenv(ROOT / ".env") | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format="%(asctime)s [%(levelname)s] %(message)s", | |
| handlers=[ | |
| logging.StreamHandler(), | |
| logging.FileHandler(ROOT / "data" / "sync.log", encoding="utf-8"), | |
| ], | |
| ) | |
| log = logging.getLogger("local-sync") | |
| def _sync_platform(name, fetch_fn, config_key, csv_filename): | |
| from src.loader import load_config, write_api_data_to_sheets | |
| try: | |
| config = load_config(ROOT / "config.yaml") | |
| p_cfg = config.get(config_key, {}) | |
| if not p_cfg.get("enabled"): | |
| log.info(f"{name} 未启用，跳过") | |
| return | |
| days_back = p_cfg.get("days_back", 90) | |
| log.info(f"拉取最近 {days_back} 天的 {name} 数据...") | |
| df = fetch_fn(days_back=days_back) | |
| if df.empty: | |
| log.warning(f"未拉取到 {name} 数据，跳过") | |
| return | |
| date_min = df["date"].min().strftime("%Y-%m-%d") | |
| date_max = df["date"].max().strftime("%Y-%m-%d") | |
| log.info(f"{name}: {len(df)} 条数据 ({date_min} ~ {date_max})") | |
| csv_path = ROOT / "data" / csv_filename | |
| csv_path.parent.mkdir(parents=True, exist_ok=True) | |
| df.to_csv(csv_path, index=False) | |
| log.info(f"本地 CSV 已保存: {csv_path}") | |
| try: | |
| result = write_api_data_to_sheets(df, config_path=ROOT / "config.yaml") | |
| log.info(f"{name} 写入 Google Sheets: {result['added']} 行新数据") | |
| except Exception as e: | |
| log.error(f"{name} 写入 Google Sheets 失败: {e}") | |
| except Exception as e: | |
| log.error(f"{name} 同步失败（已跳过，不影响其他平台）: {e}") | |
| def main(): | |
| log.info("=" * 50) | |
| log.info("本地定时同步开始") | |
| from src.meta_api import fetch_meta_data | |
| from src.criteo_api import fetch_criteo_data | |
| from src.bing_api import fetch_bing_data | |
| _sync_platform("Meta", fetch_meta_data, "meta_api", "meta_api_data.csv") | |
| _sync_platform("Criteo", fetch_criteo_data, "criteo_api", "criteo_api_data.csv") | |
| _sync_platform("Bing", fetch_bing_data, "bing_api", "bing_api_data.csv") | |
| log.info("本地定时同步完成") | |
| log.info("=" * 50) | |
| if __name__ == "__main__": | |
| try: | |
| main() | |
| except Exception as e: | |
| log.exception(f"同步异常: {e}") | |
| sys.exit(1) | |