fix: prevent duplicate data when Export table and platform sheets overlap

Root cause: _load_from_google_sheets was reading both the Export sheet
(which contains all platforms) AND individual platform sheets (Meta,
Google, Criteo, Bing written by API sync), then concatenating them.
This caused campaigns present in both sources to be double-counted,
inflating Cost/Revenue figures (e.g. US Conversion showing $407K
instead of the correct $307K).

Fix: When a platform has its own API-synced sheet, exclude that
platform's rows from the Export sheet before merging. The API sheet
data takes priority as it's more current and accurately formatted.

Made-with: Cursor

src/loader.py

@@ -104,7 +104,6 @@ def _load_from_google_sheets(config: dict[str, Any]) -> pd.DataFrame | None:
         gc = gspread.authorize(creds)
         sh = gc.open_by_url(url)
 
-        all_dfs = []
         platform_sheets = gs_cfg.get("platform_sheets", ["Meta", "Google", "Criteo", "Bing"])
         main_sheet = gs_cfg.get("worksheet", "Export")
 
@@ -117,14 +116,14 @@ def _load_from_google_sheets(config: dict[str, Any]) -> pd.DataFrame | None:
             col_cfg.get("date_idx", 4): "date",
         }
 
-        for sheet_name in [main_sheet] + platform_sheets:
+        def _read_one_sheet(ws_name: str) -> pd.DataFrame | None:
             try:
-                ws = sh.worksheet(sheet_name)
+                ws = sh.worksheet(ws_name)
                 records = ws.get_all_values()
                 if len(records) < 2:
-                    continue
+                    return None
                 header = records[0]
-                seen = {}
+                seen: dict[str, int] = {}
                 unique_header = []
                 for col in header:
                     if col in seen:
@@ -139,17 +138,40 @@ def _load_from_google_sheets(config: dict[str, Any]) -> pd.DataFrame | None:
                     if idx < len(sdf.columns):
                         rename[sdf.columns[idx]] = name
                 sdf = sdf.rename(columns=rename)
-                log.info(f"从 Sheet [{sheet_name}] 读取 {len(sdf)} 行")
-                all_dfs.append(sdf)
+                log.info(f"从 Sheet [{ws_name}] 读取 {len(sdf)} 行")
+                return sdf
             except Exception:
-                pass
+                return None
 
-        if not all_dfs:
+        main_df = _read_one_sheet(main_sheet)
+
+        api_platform_dfs: dict[str, pd.DataFrame] = {}
+        for ps in platform_sheets:
+            ps_df = _read_one_sheet(ps)
+            if ps_df is not None and not ps_df.empty:
+                api_platform_dfs[ps] = ps_df
+
+        if main_df is None and not api_platform_dfs:
             log.warning("Google Sheets 所有工作表均无数据")
             return None
 
+        if main_df is not None and api_platform_dfs:
+            platforms_with_api_data = set(api_platform_dfs.keys())
+            if "platform" in main_df.columns:
+                main_keep = main_df[~main_df["platform"].astype(str).str.strip().isin(platforms_with_api_data)]
+                main_drop_count = len(main_df) - len(main_keep)
+                if main_drop_count:
+                    log.info(f"Export 表中排除 {main_drop_count} 行已有分平台 API 数据的平台（{', '.join(platforms_with_api_data)}），避免重复")
+                all_dfs = [main_keep] + list(api_platform_dfs.values())
+            else:
+                all_dfs = [main_df] + list(api_platform_dfs.values())
+        elif main_df is not None:
+            all_dfs = [main_df]
+        else:
+            all_dfs = list(api_platform_dfs.values())
+
         df = pd.concat(all_dfs, ignore_index=True)
-        log.info(f"从 Google Sheets 共读取 {len(df)} 行数据")
+        log.info(f"从 Google Sheets 共读取 {len(df)} 行数据（去重后）")
         return df
 
     except Exception as e: