diff --git a/src/api/routes/ingest.py b/src/api/routes/ingest.py
index b3efc3f..a4d6bb1 100644
--- a/src/api/routes/ingest.py
+++ b/src/api/routes/ingest.py
@@ -33,6 +33,19 @@ _background_tasks: set[asyncio.Task] = set()
 VALID_TASKS = {"events", "standings", "stats", "all"}
 
 
+def _accumulate_ingest_result(merged: dict, code: str, r: dict) -> None:
+    """P1-A: 累加单联赛采集结果。联赛级计数读 r["leagues"][code],顶层读 total_*。"""
+    merged["total_inserted"] += r.get("total_inserted", 0)
+    merged["total_updated"] += r.get("total_updated", 0)
+    merged["errors"].extend(r.get("errors", []))
+    # 联赛级计数必须来自 leagues[code],而非顶层 r.get("inserted")
+    league_r = r.get("leagues", {}).get(code, {})
+    acc = merged["leagues"].setdefault(code, {"inserted": 0, "updated": 0, "errors": []})
+    acc["inserted"] += league_r.get("inserted", 0)
+    acc["updated"] += league_r.get("updated", 0)
+    acc["errors"].extend(r.get("errors", []))
+
+
 def _spawn(coro) -> None:
     """启动后台采集任务;异常已在任务内记录到系统日志。"""
     task = asyncio.create_task(coro)
@@ -105,13 +118,7 @@ async def _run_bzzoiro(job_id: str, task: str, leagues: list[str], req: IngestBz
                             session, leagues=[code],
                             date_from=req.date_from, date_to=req.date_to, status=st,
                         )
-                        merged["total_inserted"] += r.get("total_inserted", 0)
-                        merged["total_updated"] += r.get("total_updated", 0)
-                        merged["errors"].extend(r.get("errors", []))
-                        acc = merged["leagues"].setdefault(code, {"inserted": 0, "updated": 0, "errors": []})
-                        acc["inserted"] += r.get("inserted", 0)
-                        acc["updated"] += r.get("updated", 0)
-                        acc["errors"].extend(r.get("errors", []))
+                        _accumulate_ingest_result(merged, code, r)
             logger.info(
                 "bzzoiro 比赛采集完成: 新增 %d, 更新 %d, 联赛 %d 个, 状态 %s",
                 merged["total_inserted"], merged["total_updated"], len(merged["leagues"]), statuses,
diff --git a/tests/test_p1_a_ingest_league_counts.py b/tests/test_p1_a_ingest_league_counts.py
new file mode 100644
index 0000000..ce03140
--- /dev/null
+++ b/tests/test_p1_a_ingest_league_counts.py
@@ -0,0 +1,67 @@
+"""P1-A 回归测试: ingest 联赛级 inserted/updated 必须读 leagues[code],而非顶层 r.get("inserted")。
+
+运行: pytest tests/test_p1_a_ingest_league_counts.py -v
+(纯函数测试,无 DB/网络依赖。)
+"""
+from src.api.routes.ingest import _accumulate_ingest_result
+
+
+class TestAccumulateIngestResult:
+    """P1-A: _accumulate_ingest_result 联赛级计数必须来自 r["leagues"][code]。"""
+
+    def _merged(self):
+        return {"leagues": {}, "total_inserted": 0, "total_updated": 0, "errors": []}
+
+    def test_league_counts_read_from_leagues_key(self):
+        """核心: 联赛级 inserted/updated 应来自 leagues[code],而非顶层 inserted/updated。"""
+        merged = self._merged()
+        r = {
+            # 顶层无 inserted/updated 键(只有 total_*)
+            "total_inserted": 5,
+            "total_updated": 2,
+            "errors": [],
+            "leagues": {"E0": {"inserted": 3, "updated": 1, "rows": 4, "errors": []}},
+        }
+        _accumulate_ingest_result(merged, "E0", r)
+
+        # 顶层总计
+        assert merged["total_inserted"] == 5
+        assert merged["total_updated"] == 2
+        # 联赛级计数来自 leagues["E0"],而非顶层
+        assert merged["leagues"]["E0"]["inserted"] == 3, "联赛 inserted 必须来自 leagues[code]"
+        assert merged["leagues"]["E0"]["updated"] == 1, "联赛 updated 必须来自 leagues[code]"
+
+    def test_does_not_read_top_level_inserted(self):
+        """防御: 若 r 误含顶层 inserted 键,不得影响联赛级计数。"""
+        merged = self._merged()
+        r = {
+            "total_inserted": 5,
+            "total_updated": 2,
+            "inserted": 999,  # 错误的顶层键(旧代码可能读这个)
+            "updated": 999,
+            "errors": [],
+            "leagues": {"E0": {"inserted": 3, "updated": 1}},
+        }
+        _accumulate_ingest_result(merged, "E0", r)
+        # 必须忽略顶层 inserted/updated,使用 leagues["E0"]
+        assert merged["leagues"]["E0"]["inserted"] == 3
+        assert merged["leagues"]["E0"]["updated"] == 1
+
+    def test_missing_league_key_defaults_to_zero(self):
+        """r["leagues"] 无该 code 时,默认 0 不抛错。"""
+        merged = self._merged()
+        r = {"total_inserted": 1, "total_updated": 0, "errors": [], "leagues": {}}
+        _accumulate_ingest_result(merged, "E0", r)
+        assert merged["leagues"]["E0"]["inserted"] == 0
+        assert merged["total_inserted"] == 1
+
+    def test_multiple_calls_accumulate(self):
+        """多次调用应累加到同一联赛。"""
+        merged = self._merged()
+        r1 = {"total_inserted": 3, "total_updated": 1, "errors": [], "leagues": {"E0": {"inserted": 3, "updated": 1}}}
+        r2 = {"total_inserted": 2, "total_updated": 0, "errors": [], "leagues": {"E0": {"inserted": 2, "updated": 0}}}
+        _accumulate_ingest_result(merged, "E0", r1)
+        _accumulate_ingest_result(merged, "E0", r2)
+        assert merged["leagues"]["E0"]["inserted"] == 5
+        assert merged["leagues"]["E0"]["updated"] == 1
+        assert merged["total_inserted"] == 5