fix(registration): preserve batch task mailboxes

Use each registration task's bound email_service_id during batch runs instead of falling back to the batch-level mailbox configuration.

This keeps Outlook batch registration aligned with the user-selected inboxes in both parallel and pipeline execution paths, and adds regression coverage for task-bound mailbox resolution.

Author: GRD-Chang

src/web/routes/registration.py

@@ -404,6 +404,10 @@ def _raise_if_cancelled(reason: str = "任务已取消") -> None:
             service_type = EmailServiceType(email_service_type)
             settings = get_settings()
 
+            # 批量任务可能已在创建阶段绑定了独立邮箱服务 ID（例如 Outlook 批量注册）。
+            if not email_service_id and getattr(task, "email_service_id", None):
+                email_service_id = int(task.email_service_id)
+
             # 优先使用数据库中配置的邮箱服务
             if email_service_id:
                 from ...database.models import EmailService as EmailServiceModel
@@ -900,6 +904,21 @@ def update_batch_status(**kwargs):
     return add_batch_log, update_batch_status
 
 
+def _resolve_task_email_service_id(
+    task_uuid: str,
+    fallback_email_service_id: Optional[int],
+) -> Optional[int]:
+    """优先使用任务自身绑定的邮箱服务，缺失时再回退到批量级配置。"""
+    try:
+        with get_db() as db:
+            task = crud.get_registration_task(db, task_uuid)
+            if task and getattr(task, "email_service_id", None):
+                return int(task.email_service_id)
+    except Exception as exc:
+        logger.warning("读取任务绑定邮箱服务失败 %s: %s", task_uuid, exc)
+    return fallback_email_service_id
+
+
 async def run_batch_parallel(
     batch_id: str,
     task_uuids: List[str],
@@ -942,8 +961,9 @@ async def _run_one(idx: int, uuid: str):
                 task_manager.cancel_task(uuid)
                 task_manager.update_status(uuid, "cancelled", error="批量任务已取消")
                 return
+            resolved_email_service_id = _resolve_task_email_service_id(uuid, email_service_id)
             await run_registration_task(
-                uuid, email_service_type, proxy, email_service_config, email_service_id,
+                uuid, email_service_type, proxy, email_service_config, resolved_email_service_id,
                 log_prefix=prefix, batch_id=batch_id,
                 auto_upload_cpa=auto_upload_cpa, cpa_service_ids=cpa_service_ids or [],
                 auto_upload_sub2api=auto_upload_sub2api, sub2api_service_ids=sub2api_service_ids or [],
@@ -1025,8 +1045,9 @@ async def _run_and_release(idx: int, uuid: str, pfx: str):
                 task_manager.cancel_task(uuid)
                 task_manager.update_status(uuid, "cancelled", error="批量任务已取消")
                 return
+            resolved_email_service_id = _resolve_task_email_service_id(uuid, email_service_id)
             await run_registration_task(
-                uuid, email_service_type, proxy, email_service_config, email_service_id,
+                uuid, email_service_type, proxy, email_service_config, resolved_email_service_id,
                 log_prefix=pfx, batch_id=batch_id,
                 auto_upload_cpa=auto_upload_cpa, cpa_service_ids=cpa_service_ids or [],
                 auto_upload_sub2api=auto_upload_sub2api, sub2api_service_ids=sub2api_service_ids or [],

tests/test_outlook_batch_registration.py

@@ -71,3 +71,81 @@ async def fake_run_batch_registration(**kwargs):
     assert created_service_ids == [1, 2, 3]
     assert captured["email_service_type"] == "outlook"
     assert captured["registration_type"] == "parent"
+
+
+def test_resolve_task_email_service_id_prefers_task_binding(monkeypatch):
+    class DummyDb:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+    class DummyTask:
+        email_service_id = 42
+
+    monkeypatch.setattr(registration, "get_db", lambda: DummyDb())
+    monkeypatch.setattr(registration.crud, "get_registration_task", lambda db, task_uuid: DummyTask())
+
+    assert registration._resolve_task_email_service_id("task-1", None) == 42
+
+
+def test_run_batch_pipeline_uses_task_bound_email_service_ids(monkeypatch):
+    captured = []
+
+    class DummyDb:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+    class DummyTask:
+        status = "completed"
+        error_message = None
+
+    async def fake_run_registration_task(
+        task_uuid,
+        email_service_type,
+        proxy,
+        email_service_config,
+        email_service_id=None,
+        **kwargs,
+    ):
+        captured.append((task_uuid, email_service_id))
+
+    monkeypatch.setattr(registration, "run_registration_task", fake_run_registration_task)
+    monkeypatch.setattr(
+        registration,
+        "_resolve_task_email_service_id",
+        lambda task_uuid, fallback: {"task-1": 101, "task-2": 202}[task_uuid],
+    )
+    monkeypatch.setattr(registration, "get_db", lambda: DummyDb())
+    monkeypatch.setattr(registration.crud, "get_registration_task", lambda db, task_uuid: DummyTask())
+    monkeypatch.setattr(registration.task_manager, "init_batch", lambda batch_id, total: None)
+    monkeypatch.setattr(registration.task_manager, "add_batch_log", lambda batch_id, message: None)
+    monkeypatch.setattr(registration.task_manager, "update_batch_status", lambda batch_id, **kwargs: None)
+    monkeypatch.setattr(registration.task_manager, "is_batch_cancelled", lambda batch_id: False)
+    monkeypatch.setattr(registration.task_manager, "cancel_task", lambda task_uuid: None)
+    monkeypatch.setattr(registration.task_manager, "update_status", lambda task_uuid, status, error=None: None)
+    monkeypatch.setattr(registration.random, "randint", lambda a, b: 0)
+
+    asyncio.run(
+        registration.run_batch_pipeline(
+            batch_id="batch-pipeline",
+            task_uuids=["task-1", "task-2"],
+            email_service_type="outlook",
+            proxy=None,
+            email_service_config=None,
+            email_service_id=None,
+            interval_min=0,
+            interval_max=0,
+            concurrency=1,
+            registration_type="child",
+        )
+    )
+
+    try:
+        assert captured == [("task-1", 101), ("task-2", 202)]
+    finally:
+        registration.batch_tasks.pop("batch-pipeline", None)