虎牙自动注册持久化成功流水与批次，支持失败续跑与随时导出

将注册批次/条目/成功记录落库，成功一个写入一条流水；服务中断可恢复并续跑失败项，换电脑也能导出 txt。同时修复 pending 被误判为运行中导致页面卡住的问题。

web/backend/services/huya_register_runner.py

@@ -1,4 +1,4 @@
-"""虎牙自动注册批次执行器。"""
+"""虎牙自动注册批次执行器（支持持久化与失败续跑）。"""
 
 from __future__ import annotations
 
@@ -15,10 +15,22 @@ from core.huya.cookie_utils import normalize_huya_cookie
 from core.sms_provider import SmsLine
 
 from ..database import SessionLocal
-from ..models import ProxyConfig as ProxyConfigModel
+from ..models import (
+    HuyaRegisterBatch as HuyaRegisterBatchModel,
+    HuyaRegisterItem as HuyaRegisterItemModel,
+    HuyaRegisterSuccessLog,
+    ProxyConfig as ProxyConfigModel,
+)
 from .huya_service import upsert_huya_cookie
 
 
+# 运行中状态：服务中断后视为可续跑
+RUNNING_ITEM_STATUSES = frozenset({"sending", "waiting", "changing", "logging"})
+# 可重试状态：失败 / 停止 / 中断后的运行态 / 尚未开始
+RETRYABLE_ITEM_STATUSES = frozenset({"error", "stopped", "pending"} | RUNNING_ITEM_STATUSES)
+TERMINAL_BATCH_STATUSES = frozenset({"finished", "stopped", "error", "interrupted"})
+
+
 def _now() -> datetime:
     return datetime.now(timezone.utc)
 
@@ -32,7 +44,7 @@ def _cookie_preview(cookie: str) -> str:
 
 @dataclass
 class HuyaRegisterItemState:
-    """单个手机号在批次中的状态。"""
+    """单个手机号在批次中的状态（内存镜像）。"""
 
     line: int
     phone: str
@@ -53,6 +65,7 @@ class HuyaRegisterItemState:
     cookie_preview: str = ""
     started_at: datetime | None = None
     finished_at: datetime | None = None
+    db_id: int | None = None
 
     def to_dict(self) -> dict:
         return {
@@ -97,6 +110,274 @@ class HuyaRegisterBatch:
     created_at: datetime = field(default_factory=_now)
     started_at: datetime | None = None
     finished_at: datetime | None = None
+    db_id: int | None = None
+
+
+def _item_from_db(row: HuyaRegisterItemModel) -> HuyaRegisterItemState:
+    cookie = normalize_huya_cookie(row.cookie or "")
+    exposed = "" if row.password_changed else cookie
+    return HuyaRegisterItemState(
+        line=row.line,
+        phone=row.phone or "",
+        provider=row.provider or "",
+        sms_url=row.sms_url or "",
+        status=row.status or "pending",
+        message=row.message or "",
+        code=row.code or "",
+        change_code=row.change_code or "",
+        attempts=int(row.attempts or 0),
+        change_attempts=int(row.change_attempts or 0),
+        account_id=row.account_id,
+        username=row.username or "",
+        uid=row.uid or "",
+        password=row.password or "",
+        password_changed=bool(row.password_changed),
+        cookie=exposed,
+        cookie_preview=_cookie_preview(exposed),
+        started_at=row.started_at,
+        finished_at=row.finished_at,
+        db_id=row.id,
+    )
+
+
+def _batch_from_db(
+    batch_row: HuyaRegisterBatchModel,
+    item_rows: list[HuyaRegisterItemModel],
+) -> HuyaRegisterBatch:
+    items = [_item_from_db(row) for row in sorted(item_rows, key=lambda x: x.line)]
+    return HuyaRegisterBatch(
+        batch_id=batch_row.batch_id,
+        tag=batch_row.tag or "",
+        created_by=batch_row.created_by,
+        concurrency=int(batch_row.concurrency or 1),
+        wait_seconds=float(batch_row.wait_seconds or 180),
+        poll_interval=float(batch_row.poll_interval or 5),
+        items=items,
+        password_prefix=batch_row.password_prefix or "hy",
+        fixed_password=batch_row.fixed_password or "",
+        use_proxy=bool(batch_row.use_proxy),
+        status=batch_row.status or "pending",
+        message=batch_row.message or "",
+        created_at=batch_row.created_at or _now(),
+        started_at=batch_row.started_at,
+        finished_at=batch_row.finished_at,
+        db_id=batch_row.id,
+    )
+
+
+def snapshot_from_batch(batch: HuyaRegisterBatch) -> dict:
+    """从内存批次生成 API 快照。"""
+    total = len(batch.items)
+    success = sum(1 for item in batch.items if item.status == "success")
+    failed = sum(1 for item in batch.items if item.status == "error")
+    stopped = sum(1 for item in batch.items if item.status == "stopped")
+    running = sum(1 for item in batch.items if item.status in RUNNING_ITEM_STATUSES)
+    return {
+        "batch_id": batch.batch_id,
+        "status": batch.status,
+        "message": batch.message,
+        "tag": batch.tag,
+        "created_by": batch.created_by,
+        "concurrency": batch.concurrency,
+        "wait_seconds": batch.wait_seconds,
+        "poll_interval": batch.poll_interval,
+        "password_prefix": batch.password_prefix,
+        "use_proxy": batch.use_proxy,
+        "total": total,
+        "success_count": success,
+        "failed_count": failed,
+        "stopped_count": stopped,
+        "running_count": running,
+        "created_at": batch.created_at,
+        "started_at": batch.started_at,
+        "finished_at": batch.finished_at,
+        "items": [item.to_dict() for item in batch.items],
+    }
+
+
+def load_batch_snapshot(batch_id: str, *, recover_interrupted: bool = True) -> dict | None:
+    """从数据库加载批次详情；若服务中断则标记为 interrupted。"""
+    db = SessionLocal()
+    try:
+        batch_row = (
+            db.query(HuyaRegisterBatchModel)
+            .filter(HuyaRegisterBatchModel.batch_id == batch_id)
+            .first()
+        )
+        if not batch_row:
+            return None
+        item_rows = (
+            db.query(HuyaRegisterItemModel)
+            .filter(HuyaRegisterItemModel.batch_id == batch_id)
+            .order_by(HuyaRegisterItemModel.line.asc())
+            .all()
+        )
+        # 无内存 runner 时，running/pending 都视为中断（避免「准备续跑」永久卡住）
+        if recover_interrupted and batch_row.status in {"running", "pending"}:
+            now = _now()
+            batch_row.status = "interrupted"
+            batch_row.message = "服务中断或未真正启动，可从失败/未完成项续跑"
+            batch_row.finished_at = now
+            for item in item_rows:
+                if item.status in RUNNING_ITEM_STATUSES:
+                    item.status = "error"
+                    item.message = "服务中断，可续跑"
+                    item.finished_at = now
+                elif item.status == "pending":
+                    # 保留 pending 供续跑，仅更新提示
+                    item.message = item.message or "等待续跑"
+            _refresh_batch_counts(batch_row, item_rows)
+            db.commit()
+            db.refresh(batch_row)
+            item_rows = (
+                db.query(HuyaRegisterItemModel)
+                .filter(HuyaRegisterItemModel.batch_id == batch_id)
+                .order_by(HuyaRegisterItemModel.line.asc())
+                .all()
+            )
+        batch = _batch_from_db(batch_row, item_rows)
+        return snapshot_from_batch(batch)
+    finally:
+        db.close()
+
+
+def list_batch_summaries(limit: int = 50, live_batch_ids: set[str] | None = None) -> list[dict]:
+    """列出最近的注册批次摘要。live_batch_ids 中的 running 保持运行中。"""
+    live = live_batch_ids or set()
+    db = SessionLocal()
+    try:
+        rows = (
+            db.query(HuyaRegisterBatchModel)
+            .order_by(HuyaRegisterBatchModel.id.desc())
+            .limit(max(1, min(int(limit or 50), 200)))
+            .all()
+        )
+        result = []
+        for row in rows:
+            status = row.status or "pending"
+            message = row.message or ""
+            # 无内存 runner 且状态仍是 running/pending，展示为 interrupted
+            if status in {"running", "pending"} and row.batch_id not in live:
+                status = "interrupted"
+                message = message or "服务中断或未真正启动，可从失败/未完成项续跑"
+            running_count = 0
+            if status == "running":
+                running_count = max(
+                    0,
+                    int(row.total or 0) - int(row.success_count or 0) - int(row.failed_count or 0) - int(row.stopped_count or 0),
+                )
+            result.append({
+                "batch_id": row.batch_id,
+                "status": status,
+                "message": message,
+                "tag": row.tag or "",
+                "created_by": row.created_by,
+                "concurrency": int(row.concurrency or 1),
+                "wait_seconds": float(row.wait_seconds or 180),
+                "poll_interval": float(row.poll_interval or 5),
+                "password_prefix": row.password_prefix or "hy",
+                "use_proxy": bool(row.use_proxy),
+                "total": int(row.total or 0),
+                "success_count": int(row.success_count or 0),
+                "failed_count": int(row.failed_count or 0),
+                "stopped_count": int(row.stopped_count or 0),
+                "running_count": running_count,
+                "created_at": row.created_at,
+                "started_at": row.started_at,
+                "finished_at": row.finished_at,
+                "items": [],
+            })
+        return result
+    finally:
+        db.close()
+
+
+def _refresh_batch_counts(batch_row: HuyaRegisterBatchModel, items: list[HuyaRegisterItemModel]):
+    batch_row.total = len(items)
+    batch_row.success_count = sum(1 for item in items if item.status == "success")
+    batch_row.failed_count = sum(1 for item in items if item.status == "error")
+    batch_row.stopped_count = sum(1 for item in items if item.status == "stopped")
+
+
+def format_success_export_line(username: str, uid: str, password: str, phone: str, sms_url: str) -> str:
+    """统一成功导出格式。"""
+    account = (username or uid or "").strip()
+    return f"{account}----{password or ''}----{phone or ''}----{sms_url or ''}"
+
+
+def export_success_logs_text(
+    *,
+    batch_id: str | None = None,
+    tag: str | None = None,
+    limit: int = 5000,
+) -> str:
+    """从成功流水表导出 txt。"""
+    db = SessionLocal()
+    try:
+        query = db.query(HuyaRegisterSuccessLog).order_by(HuyaRegisterSuccessLog.id.asc())
+        if batch_id:
+            query = query.filter(HuyaRegisterSuccessLog.batch_id == batch_id)
+        if tag:
+            query = query.filter(HuyaRegisterSuccessLog.tag == tag)
+        rows = query.limit(max(1, min(int(limit or 5000), 20000))).all()
+        lines = [
+            format_success_export_line(
+                row.username or "",
+                row.uid or "",
+                row.password or "",
+                row.phone or "",
+                row.sms_url or "",
+            )
+            for row in rows
+            if (row.username or row.uid) and row.password
+        ]
+        return "\n".join(lines)
+    finally:
+        db.close()
+
+
+def list_success_logs(
+    *,
+    batch_id: str | None = None,
+    tag: str | None = None,
+    limit: int = 200,
+) -> list[dict]:
+    """列出成功流水（含密码，供管理端展示/导出）。"""
+    db = SessionLocal()
+    try:
+        query = db.query(HuyaRegisterSuccessLog).order_by(HuyaRegisterSuccessLog.id.desc())
+        if batch_id:
+            query = query.filter(HuyaRegisterSuccessLog.batch_id == batch_id)
+        if tag:
+            query = query.filter(HuyaRegisterSuccessLog.tag == tag)
+        rows = query.limit(max(1, min(int(limit or 200), 2000))).all()
+        return [
+            {
+                "id": row.id,
+                "batch_id": row.batch_id or "",
+                "item_id": row.item_id,
+                "account_id": row.account_id,
+                "phone": row.phone or "",
+                "username": row.username or "",
+                "uid": row.uid or "",
+                "password": row.password or "",
+                "sms_url": row.sms_url or "",
+                "tag": row.tag or "",
+                "provider": row.provider or "",
+                "created_by": row.created_by,
+                "created_at": row.created_at,
+                "export_line": format_success_export_line(
+                    row.username or "",
+                    row.uid or "",
+                    row.password or "",
+                    row.phone or "",
+                    row.sms_url or "",
+                ),
+            }
+            for row in rows
+        ]
+    finally:
+        db.close()
 
 
 class HuyaRegisterRunner:
@@ -107,10 +388,13 @@ class HuyaRegisterRunner:
         batch: HuyaRegisterBatch,
         sms_lines: list[SmsLine],
         proxy_config: Optional[ProxyConfigModel] = None,
+        *,
+        item_indices: list[int] | None = None,
     ):
         self.batch = batch
         self.sms_lines = sms_lines
         self.proxy_config = proxy_config
+        self.item_indices = item_indices  # None 表示跑全部
         self._lock = threading.Lock()
         self._stop = threading.Event()
         self._shared_proxy_fetcher = self._create_proxy_fetcher()
@@ -145,54 +429,123 @@ class HuyaRegisterRunner:
         with self._lock:
             if self.batch.status == "running":
                 self.batch.message = "正在停止"
+                self._persist_batch_meta()
 
     def snapshot(self) -> dict:
         with self._lock:
-            total = len(self.batch.items)
+            return snapshot_from_batch(self.batch)
+
+    def _persist_batch_meta(self):
+        """把批次汇总写回数据库。"""
+        if not self.batch.db_id:
+            return
+        db = SessionLocal()
+        try:
+            row = db.query(HuyaRegisterBatchModel).filter(HuyaRegisterBatchModel.id == self.batch.db_id).first()
+            if not row:
+                return
+            row.status = self.batch.status
+            row.message = self.batch.message
+            row.started_at = self.batch.started_at
+            row.finished_at = self.batch.finished_at
+            row.concurrency = self.batch.concurrency
+            row.wait_seconds = int(self.batch.wait_seconds)
+            row.poll_interval = int(self.batch.poll_interval)
+            row.password_prefix = self.batch.password_prefix
+            row.fixed_password = self.batch.fixed_password
+            row.use_proxy = self.batch.use_proxy
+            row.tag = self.batch.tag
+            items = (
+                db.query(HuyaRegisterItemModel)
+                .filter(HuyaRegisterItemModel.batch_id == self.batch.batch_id)
+                .all()
+            )
+            _refresh_batch_counts(row, items)
+            # 同步内存计数到 batch 对象侧的 status 字段已足够；counts 以 DB items 为准
             success = sum(1 for item in self.batch.items if item.status == "success")
             failed = sum(1 for item in self.batch.items if item.status == "error")
             stopped = sum(1 for item in self.batch.items if item.status == "stopped")
-            running = sum(1 for item in self.batch.items if item.status in {"sending", "waiting", "changing", "logging"})
-            return {
-                "batch_id": self.batch.batch_id,
-                "status": self.batch.status,
-                "message": self.batch.message,
-                "tag": self.batch.tag,
-                "created_by": self.batch.created_by,
-                "concurrency": self.batch.concurrency,
-                "wait_seconds": self.batch.wait_seconds,
-                "poll_interval": self.batch.poll_interval,
-                "password_prefix": self.batch.password_prefix,
-                "use_proxy": self.batch.use_proxy,
-                "total": total,
-                "success_count": success,
-                "failed_count": failed,
-                "stopped_count": stopped,
-                "running_count": running,
-                "created_at": self.batch.created_at,
-                "started_at": self.batch.started_at,
-                "finished_at": self.batch.finished_at,
-                "items": [item.to_dict() for item in self.batch.items],
-            }
+            row.success_count = success
+            row.failed_count = failed
+            row.stopped_count = stopped
+            row.total = len(self.batch.items)
+            db.commit()
+        finally:
+            db.close()
+
+    def _persist_item(self, index: int):
+        """把单条 item 状态写回数据库。"""
+        item = self.batch.items[index]
+        if not item.db_id:
+            return
+        db = SessionLocal()
+        try:
+            row = db.query(HuyaRegisterItemModel).filter(HuyaRegisterItemModel.id == item.db_id).first()
+            if not row:
+                return
+            row.status = item.status
+            row.message = (item.message or "")[:512]
+            row.code = item.code or ""
+            row.change_code = item.change_code or ""
+            row.attempts = item.attempts
+            row.change_attempts = item.change_attempts
+            row.account_id = item.account_id
+            row.username = item.username or ""
+            row.uid = item.uid or ""
+            row.password = item.password or ""
+            row.password_changed = bool(item.password_changed)
+            # DB 存完整 cookie；内存暴露受改密控制
+            if item.cookie:
+                row.cookie = normalize_huya_cookie(item.cookie)
+            elif item.password_changed and item.status == "success":
+                # 改密成功时 cookie 可能被前端隐藏，保留 DB 已有值
+                pass
+            row.started_at = item.started_at
+            row.finished_at = item.finished_at
+            db.commit()
+        finally:
+            db.close()
 
     def _set_item(self, index: int, **updates):
         with self._lock:
             item = self.batch.items[index]
             for key, value in updates.items():
                 setattr(item, key, value)
+        self._persist_item(index)
 
-    def _save_cookie(self, result: HuyaAutoRegisterResult) -> tuple[int | None, str, str]:
+    def _save_success(self, index: int, result: HuyaAutoRegisterResult) -> tuple[int | None, str, str]:
+        """成功时：写账号 + 成功流水（成功一个写一条，立即可导出）。"""
         db = SessionLocal()
         try:
+            # upsert_huya_cookie 内部会 commit 一次
             account = upsert_huya_cookie(db, result.cookie, tag=self.batch.tag, username_hint="")
-            account.game_phone = result.phone
+            account.game_phone = result.phone or account.game_phone or ""
             if result.username:
                 account.username = result.username
             if result.password:
                 account.account_password = result.password
+            sms_url = result.sms_url or self.batch.items[index].sms_url or ""
+            if sms_url:
+                account.sms_url = sms_url
             if result.password_changed:
                 account.status = "password_changed"
             account.updated_at = _now()
+
+            item = self.batch.items[index]
+            db.add(HuyaRegisterSuccessLog(
+                batch_id=self.batch.batch_id,
+                item_id=item.db_id,
+                account_id=account.id,
+                phone=result.phone or item.phone,
+                username=result.username or account.username or "",
+                uid=result.uid or account.uid or account.yyuid or "",
+                password=result.password or "",
+                sms_url=sms_url,
+                tag=self.batch.tag,
+                provider=result.provider or item.provider,
+                created_by=self.batch.created_by,
+                created_at=_now(),
+            ))
             db.commit()
             db.refresh(account)
             return account.id, account.username or "", account.uid or account.yyuid or ""
@@ -228,7 +581,24 @@ class HuyaRegisterRunner:
             self._set_item(index, status="error", message=proxy_error, finished_at=_now())
             return
 
-        self._set_item(index, status="sending", message="注册并改密", started_at=_now(), finished_at=None)
+        self._set_item(
+            index,
+            status="sending",
+            message="注册并改密",
+            started_at=_now(),
+            finished_at=None,
+            code="",
+            change_code="",
+            attempts=0,
+            change_attempts=0,
+            account_id=None,
+            username="",
+            uid="",
+            password="",
+            password_changed=False,
+            cookie="",
+            cookie_preview="",
+        )
         result = register_huya_with_sms_line(
             item,
             wait_seconds=self.batch.wait_seconds,
@@ -245,6 +615,7 @@ class HuyaRegisterRunner:
         message = result.message
         status = result.status
         cookie = result.cookie if result.success else ""
+        full_cookie = cookie
         if result.success:
             self._set_item(
                 index,
@@ -260,38 +631,71 @@ class HuyaRegisterRunner:
                 password_changed=result.password_changed,
             )
             try:
-                account_id, username, uid = self._save_cookie(result)
+                account_id, username, uid = self._save_success(index, result)
             except Exception as exc:
                 status = "error"
                 cookie = ""
+                full_cookie = ""
                 message = f"账号保存失败: {exc}"
 
         exposed_cookie = "" if result.password_changed else cookie
 
-        self._set_item(
-            index,
-            status=status,
-            message=message,
-            code=result.code,
-            change_code=result.change_code,
-            attempts=result.attempts,
-            change_attempts=result.change_attempts,
-            account_id=account_id,
-            username=username,
-            uid=uid,
-            password=result.password,
-            password_changed=result.password_changed,
-            cookie=normalize_huya_cookie(exposed_cookie),
-            cookie_preview=_cookie_preview(exposed_cookie),
-            finished_at=_now(),
-        )
+        # 先把完整 cookie 写内存供 persist，再把暴露值用于展示
+        with self._lock:
+            mem = self.batch.items[index]
+            mem.status = status
+            mem.message = message
+            mem.code = result.code
+            mem.change_code = result.change_code
+            mem.attempts = result.attempts
+            mem.change_attempts = result.change_attempts
+            mem.account_id = account_id
+            mem.username = username
+            mem.uid = uid
+            mem.password = result.password
+            mem.password_changed = result.password_changed
+            mem.cookie = normalize_huya_cookie(full_cookie)
+            mem.cookie_preview = _cookie_preview(exposed_cookie)
+            mem.finished_at = _now()
+            # 展示用 cookie 在 to_dict 时再处理：成功且改密则隐藏
+            if result.password_changed and status == "success":
+                # to_dict 使用 mem.cookie；这里保持 DB 有完整 cookie，API 隐藏
+                pass
+
+        self._persist_item(index)
+
+        # API 快照中改密成功不暴露 cookie
+        with self._lock:
+            mem = self.batch.items[index]
+            if result.password_changed and status == "success":
+                mem.cookie = ""
+                mem.cookie_preview = ""
+            else:
+                mem.cookie = normalize_huya_cookie(exposed_cookie)
+                mem.cookie_preview = _cookie_preview(exposed_cookie)
+
+    def mark_running(self, message: str = "批次运行中"):
+        """在启动线程前立刻标记 running，避免前端看到 pending 误判/卡住。"""
+        with self._lock:
+            self.batch.status = "running"
+            self.batch.message = message
+            self.batch.started_at = self.batch.started_at or _now()
+            self.batch.finished_at = None
+            self._persist_batch_meta()
 
     def run(self):
         """线程入口。"""
         with self._lock:
-            self.batch.status = "running"
-            self.batch.message = "批次运行中"
-            self.batch.started_at = _now()
+            if self.batch.status != "running":
+                self.batch.status = "running"
+                self.batch.message = "批次运行中"
+                self.batch.started_at = self.batch.started_at or _now()
+                self.batch.finished_at = None
+                self._persist_batch_meta()
+
+        indices = self.item_indices
+        if indices is None:
+            indices = list(range(len(self.sms_lines)))
 
         try:
             if self._shared_proxy_fetcher:
@@ -299,13 +703,15 @@ class HuyaRegisterRunner:
                 if not ok:
                     with self._lock:
                         self.batch.message = f"代理白名单预热失败: {msg}"
+                        self._persist_batch_meta()
 
             with ThreadPoolExecutor(max_workers=self.batch.concurrency) as executor:
                 futures = []
-                for index, item in enumerate(self.sms_lines):
+                for index in indices:
                     if self._stop.is_set():
                         self._set_item(index, status="stopped", message="已停止", finished_at=_now())
                         continue
+                    item = self.sms_lines[index]
                     futures.append(executor.submit(self._run_one, index, item))
 
                 for future in as_completed(futures):
@@ -315,6 +721,7 @@ class HuyaRegisterRunner:
                 self.batch.status = "error"
                 self.batch.message = f"批次执行异常: {exc}"
                 self.batch.finished_at = _now()
+                self._persist_batch_meta()
             return
 
         with self._lock:
@@ -325,10 +732,11 @@ class HuyaRegisterRunner:
                 self.batch.status = "finished"
                 self.batch.message = "批次已完成"
             self.batch.finished_at = _now()
+            self._persist_batch_meta()
 
 
 class HuyaRegisterRegistry:
-    """管理自动注册批次。"""
+    """管理自动注册批次（内存 runner + DB 持久化）。"""
 
     def __init__(self):
         self._lock = threading.Lock()
@@ -348,29 +756,229 @@ class HuyaRegisterRegistry:
         proxy_config: Optional[ProxyConfigModel] = None,
     ) -> HuyaRegisterRunner:
         batch_id = uuid.uuid4().hex[:12]
+        concurrency = max(1, min(int(concurrency or 1), 5))
+        wait_seconds = max(15.0, float(wait_seconds or 180))
+        poll_interval = max(1.0, float(poll_interval or 5))
+        password_prefix = (password_prefix or "hy").strip()[:8] or "hy"
+        fixed_password = (fixed_password or "").strip()
+        tag = (tag or "").strip()
+
+        items = [
+            HuyaRegisterItemState(
+                line=index + 1,
+                phone=item.phone,
+                provider=item.provider,
+                sms_url=item.url,
+                message="等待开始",
+            )
+            for index, item in enumerate(sms_lines)
+        ]
+
+        db = SessionLocal()
+        try:
+            batch_row = HuyaRegisterBatchModel(
+                batch_id=batch_id,
+                tag=tag,
+                created_by=created_by,
+                concurrency=concurrency,
+                wait_seconds=int(wait_seconds),
+                poll_interval=int(poll_interval),
+                password_prefix=password_prefix,
+                fixed_password=fixed_password,
+                use_proxy=bool(use_proxy),
+                status="pending",
+                message="等待开始",
+                total=len(items),
+                success_count=0,
+                failed_count=0,
+                stopped_count=0,
+                created_at=_now(),
+            )
+            db.add(batch_row)
+            db.flush()
+            for item in items:
+                row = HuyaRegisterItemModel(
+                    batch_db_id=batch_row.id,
+                    batch_id=batch_id,
+                    line=item.line,
+                    phone=item.phone,
+                    provider=item.provider,
+                    sms_url=item.sms_url,
+                    status="pending",
+                    message="等待开始",
+                    password="",
+                    cookie="",
+                )
+                db.add(row)
+                db.flush()
+                item.db_id = row.id
+            db.commit()
+            db_id = batch_row.id
+        finally:
+            db.close()
+
         batch = HuyaRegisterBatch(
             batch_id=batch_id,
             tag=tag,
             created_by=created_by,
-            concurrency=max(1, min(int(concurrency or 1), 5)),
-            wait_seconds=max(15.0, float(wait_seconds or 180)),
-            poll_interval=max(1.0, float(poll_interval or 5)),
-            password_prefix=(password_prefix or "hy").strip()[:8] or "hy",
-            fixed_password=(fixed_password or "").strip(),
+            concurrency=concurrency,
+            wait_seconds=wait_seconds,
+            poll_interval=poll_interval,
+            password_prefix=password_prefix,
+            fixed_password=fixed_password,
             use_proxy=bool(use_proxy),
-            items=[
-                HuyaRegisterItemState(line=index + 1, phone=item.phone, provider=item.provider, sms_url=item.url)
-                for index, item in enumerate(sms_lines)
-            ],
+            items=items,
+            db_id=db_id,
+            status="running",
+            message="批次运行中",
+            started_at=_now(),
         )
+        # DB 同步为 running，避免返回 pending 导致前端误判
+        db = SessionLocal()
+        try:
+            row = db.query(HuyaRegisterBatchModel).filter(HuyaRegisterBatchModel.id == db_id).first()
+            if row:
+                row.status = "running"
+                row.message = "批次运行中"
+                row.started_at = batch.started_at
+                db.commit()
+        finally:
+            db.close()
+
         runner = HuyaRegisterRunner(batch=batch, sms_lines=sms_lines, proxy_config=proxy_config)
         with self._lock:
             self._runners[batch_id] = runner
         return runner
 
+    def retry(
+        self,
+        batch_id: str,
+        proxy_config: Optional[ProxyConfigModel] = None,
+        *,
+        concurrency: int | None = None,
+        wait_seconds: float | None = None,
+        poll_interval: float | None = None,
+        password_prefix: str | None = None,
+        fixed_password: str | None = None,
+        use_proxy: bool | None = None,
+    ) -> HuyaRegisterRunner:
+        """从失败/停止/未完成的条目续跑；成功项跳过。"""
+        with self._lock:
+            existing = self._runners.get(batch_id)
+            if existing and existing.batch.status == "running":
+                raise RuntimeError("批次正在运行中，请先停止或等待完成")
+
+        db = SessionLocal()
+        try:
+            batch_row = (
+                db.query(HuyaRegisterBatchModel)
+                .filter(HuyaRegisterBatchModel.batch_id == batch_id)
+                .first()
+            )
+            if not batch_row:
+                raise ValueError("批次不存在")
+            if batch_row.status == "running":
+                # 无内存 runner 的 running 视为中断，允许续跑
+                batch_row.status = "interrupted"
+                batch_row.message = "服务中断，准备续跑"
+                db.commit()
+
+            item_rows = (
+                db.query(HuyaRegisterItemModel)
+                .filter(HuyaRegisterItemModel.batch_id == batch_id)
+                .order_by(HuyaRegisterItemModel.line.asc())
+                .all()
+            )
+            if not item_rows:
+                raise ValueError("批次没有可运行条目")
+
+            # 可选覆盖运行参数
+            if concurrency is not None:
+                batch_row.concurrency = max(1, min(int(concurrency), 5))
+            if wait_seconds is not None:
+                batch_row.wait_seconds = int(max(15.0, float(wait_seconds)))
+            if poll_interval is not None:
+                batch_row.poll_interval = int(max(1.0, float(poll_interval)))
+            if password_prefix is not None:
+                batch_row.password_prefix = (password_prefix or "hy").strip()[:8] or "hy"
+            if fixed_password is not None:
+                batch_row.fixed_password = (fixed_password or "").strip()
+            if use_proxy is not None:
+                batch_row.use_proxy = bool(use_proxy)
+
+            retry_indices: list[int] = []
+            for idx, row in enumerate(item_rows):
+                if (row.status or "pending") in RETRYABLE_ITEM_STATUSES and row.status != "success":
+                    row.status = "pending"
+                    row.message = "等待续跑"
+                    row.finished_at = None
+                    row.started_at = None
+                    row.code = ""
+                    row.change_code = ""
+                    row.attempts = 0
+                    row.change_attempts = 0
+                    # 保留历史密码等成功字段不覆盖；失败项本来也没有
+                    retry_indices.append(idx)
+
+            if not retry_indices:
+                raise ValueError("没有可续跑的失败/未完成条目")
+
+            batch_row.status = "running"
+            batch_row.message = f"续跑中（{len(retry_indices)} 条）"
+            batch_row.started_at = batch_row.started_at or _now()
+            batch_row.finished_at = None
+            _refresh_batch_counts(batch_row, item_rows)
+            db.commit()
+
+            batch = _batch_from_db(batch_row, item_rows)
+            sms_lines = [
+                SmsLine(phone=item.phone, url=item.sms_url, provider=item.provider, raw=f"{item.phone}----{item.sms_url}")
+                for item in batch.items
+            ]
+        finally:
+            db.close()
+
+        if use_proxy is False:
+            proxy_config = None
+        elif batch.use_proxy and proxy_config is None:
+            # 由调用方传入；若未传则 runner 内会报代理未配置
+            pass
+
+        runner = HuyaRegisterRunner(
+            batch=batch,
+            sms_lines=sms_lines,
+            proxy_config=proxy_config if batch.use_proxy else None,
+            item_indices=retry_indices,
+        )
+        # 内存侧也保持 running，与 DB 一致
+        runner.batch.status = "running"
+        runner.batch.message = f"续跑中（{len(retry_indices)} 条）"
+        with self._lock:
+            self._runners[batch_id] = runner
+        return runner
+
     def get(self, batch_id: str) -> HuyaRegisterRunner | None:
         with self._lock:
             return self._runners.get(batch_id)
 
+    def live_batch_ids(self) -> set[str]:
+        with self._lock:
+            return {
+                batch_id
+                for batch_id, runner in self._runners.items()
+                if runner.batch.status == "running"
+            }
+
+    def list_summaries(self, limit: int = 50) -> list[dict]:
+        return list_batch_summaries(limit=limit, live_batch_ids=self.live_batch_ids())
+
+    def get_snapshot(self, batch_id: str) -> dict | None:
+        """优先内存 runner，否则读库（并处理中断恢复）。"""
+        with self._lock:
+            runner = self._runners.get(batch_id)
+            if runner:
+                return runner.snapshot()
+        return load_batch_snapshot(batch_id, recover_interrupted=True)
+
 
 huya_register_registry = HuyaRegisterRegistry()