1016 lines
38 KiB
Python
1016 lines
38 KiB
Python
"""虎牙自动注册批次执行器（支持持久化与从停止处继续）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import threading
|
||
import uuid
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timezone
|
||
from typing import Optional
|
||
|
||
from core.douyu.proxy_fetcher import ProxyFetcher
|
||
from core.huya.auto_register import HuyaAutoRegisterResult, register_huya_with_sms_line
|
||
from core.huya.cookie_utils import normalize_huya_cookie
|
||
from core.sms_provider import SmsLine
|
||
|
||
from ..database import SessionLocal
|
||
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"})
|
||
# 续跑（continue）：只跑「还没跑完」的，不重试已明确失败(error)、不重跑成功(success)
|
||
# - pending: 队列里还没轮到
|
||
# - stopped: 点停止时尚未开始或被中断
|
||
# - running 中间态: 进程中断时卡在半路
|
||
RESUME_ITEM_STATUSES = frozenset({"pending", "stopped"} | RUNNING_ITEM_STATUSES)
|
||
# 仅重试明确失败（可选模式）
|
||
RETRY_FAILED_ITEM_STATUSES = frozenset({"error"})
|
||
TERMINAL_BATCH_STATUSES = frozenset({"finished", "stopped", "error", "interrupted"})
|
||
|
||
|
||
def _now() -> datetime:
|
||
return datetime.now(timezone.utc)
|
||
|
||
|
||
def _cookie_preview(cookie: str) -> str:
|
||
normalized = normalize_huya_cookie(cookie or "")
|
||
if not normalized:
|
||
return ""
|
||
return normalized[:50] + "..." if len(normalized) > 50 else normalized
|
||
|
||
|
||
@dataclass
|
||
class HuyaRegisterItemState:
|
||
"""单个手机号在批次中的状态（内存镜像）。"""
|
||
|
||
line: int
|
||
phone: str
|
||
provider: str
|
||
sms_url: str = ""
|
||
status: str = "pending"
|
||
message: str = "等待开始"
|
||
code: str = ""
|
||
change_code: str = ""
|
||
attempts: int = 0
|
||
change_attempts: int = 0
|
||
account_id: int | None = None
|
||
username: str = ""
|
||
uid: str = ""
|
||
password: str = ""
|
||
password_changed: bool = False
|
||
cookie: str = ""
|
||
cookie_preview: str = ""
|
||
started_at: datetime | None = None
|
||
finished_at: datetime | None = None
|
||
db_id: int | None = None
|
||
|
||
def to_dict(self) -> dict:
|
||
return {
|
||
"line": self.line,
|
||
"phone": self.phone,
|
||
"provider": self.provider,
|
||
"sms_url": self.sms_url,
|
||
"status": self.status,
|
||
"message": self.message,
|
||
"code": self.code,
|
||
"change_code": self.change_code,
|
||
"attempts": self.attempts,
|
||
"change_attempts": self.change_attempts,
|
||
"account_id": self.account_id,
|
||
"username": self.username,
|
||
"uid": self.uid,
|
||
"password": self.password,
|
||
"password_changed": self.password_changed,
|
||
"cookie": self.cookie,
|
||
"cookie_preview": self.cookie_preview,
|
||
"started_at": self.started_at,
|
||
"finished_at": self.finished_at,
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class HuyaRegisterBatch:
|
||
"""自动注册批次内存快照。"""
|
||
|
||
batch_id: str
|
||
tag: str
|
||
created_by: int
|
||
concurrency: int
|
||
wait_seconds: float
|
||
poll_interval: float
|
||
items: list[HuyaRegisterItemState]
|
||
password_prefix: str = "hy"
|
||
fixed_password: str = ""
|
||
use_proxy: bool = False
|
||
status: str = "pending"
|
||
message: str = "等待开始"
|
||
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:
|
||
# 记为 stopped 而非 error，以便「继续未完成」能接着跑
|
||
item.status = "stopped"
|
||
item.message = "服务中断，可继续"
|
||
item.finished_at = now
|
||
elif item.status == "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:
|
||
"""在后台线程中批量执行虎牙手机号自动注册。"""
|
||
|
||
def __init__(
|
||
self,
|
||
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()
|
||
|
||
def _create_proxy_fetcher(self) -> ProxyFetcher | None:
|
||
"""按需创建 API 代理获取器。"""
|
||
if not self.batch.use_proxy or not self.proxy_config or not self.proxy_config.enabled:
|
||
return None
|
||
if not self.proxy_config.api_url:
|
||
return None
|
||
|
||
wl_platform = "xiequ"
|
||
wl_credentials = None
|
||
if self.proxy_config.whitelist_enabled:
|
||
wl_platform = getattr(self.proxy_config, "whitelist_platform", None) or "xiequ"
|
||
wl_credentials = getattr(self.proxy_config, "whitelist_credentials", None)
|
||
if not wl_credentials and self.proxy_config.whitelist_uid and self.proxy_config.whitelist_ukey:
|
||
wl_credentials = {
|
||
"uid": self.proxy_config.whitelist_uid,
|
||
"ukey": self.proxy_config.whitelist_ukey,
|
||
}
|
||
|
||
return ProxyFetcher(
|
||
api_url=self.proxy_config.api_url,
|
||
whitelist_platform=wl_platform,
|
||
whitelist_credentials=wl_credentials,
|
||
stop_event=self._stop,
|
||
)
|
||
|
||
def stop(self):
|
||
self._stop.set()
|
||
with self._lock:
|
||
if self.batch.status == "running":
|
||
self.batch.message = "正在停止"
|
||
self._persist_batch_meta()
|
||
|
||
def snapshot(self) -> dict:
|
||
with self._lock:
|
||
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")
|
||
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_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 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 ""
|
||
finally:
|
||
db.close()
|
||
|
||
def _resolve_proxy(self) -> tuple[dict[str, str] | None, str]:
|
||
"""为单个手机号解析代理；返回代理字典和错误消息。"""
|
||
if not self.batch.use_proxy:
|
||
return None, ""
|
||
if not self.proxy_config or not self.proxy_config.enabled:
|
||
return None, "已开启代理，但代理配置未启用"
|
||
|
||
if self.proxy_config.http or self.proxy_config.https:
|
||
proxy_url = self.proxy_config.http or self.proxy_config.https
|
||
return {"http": proxy_url, "https": proxy_url}, ""
|
||
|
||
if self._shared_proxy_fetcher:
|
||
proxy_url = self._shared_proxy_fetcher.fetch_new_proxy(max_attempts=3)
|
||
if proxy_url:
|
||
return {"http": proxy_url, "https": proxy_url}, ""
|
||
return None, "获取代理失败"
|
||
|
||
return None, "已开启代理，但未配置静态代理或代理 API"
|
||
|
||
def _run_one(self, index: int, item: SmsLine):
|
||
if self._stop.is_set():
|
||
self._set_item(index, status="stopped", message="已停止", finished_at=_now())
|
||
return
|
||
|
||
proxies, proxy_error = self._resolve_proxy()
|
||
if proxy_error:
|
||
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,
|
||
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,
|
||
poll_interval=self.batch.poll_interval,
|
||
password_prefix=self.batch.password_prefix,
|
||
fixed_password=self.batch.fixed_password,
|
||
proxies=proxies,
|
||
stop_event=self._stop,
|
||
)
|
||
|
||
account_id = None
|
||
username = result.username
|
||
uid = result.uid
|
||
message = result.message
|
||
status = result.status
|
||
cookie = result.cookie if result.success else ""
|
||
full_cookie = cookie
|
||
if result.success:
|
||
self._set_item(
|
||
index,
|
||
status="logging",
|
||
message="保存账号密码",
|
||
code=result.code,
|
||
change_code=result.change_code,
|
||
attempts=result.attempts,
|
||
change_attempts=result.change_attempts,
|
||
username=result.username,
|
||
uid=result.uid,
|
||
password=result.password,
|
||
password_changed=result.password_changed,
|
||
)
|
||
try:
|
||
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
|
||
|
||
# 先把完整 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:
|
||
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:
|
||
ok, msg = self._shared_proxy_fetcher.warmup_whitelist()
|
||
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 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):
|
||
future.result()
|
||
except Exception as exc:
|
||
with self._lock:
|
||
self.batch.status = "error"
|
||
self.batch.message = f"批次执行异常: {exc}"
|
||
self.batch.finished_at = _now()
|
||
self._persist_batch_meta()
|
||
return
|
||
|
||
with self._lock:
|
||
if self._stop.is_set():
|
||
self.batch.status = "stopped"
|
||
self.batch.message = "批次已停止"
|
||
else:
|
||
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()
|
||
self._runners: dict[str, HuyaRegisterRunner] = {}
|
||
|
||
def create(
|
||
self,
|
||
sms_lines: list[SmsLine],
|
||
tag: str,
|
||
created_by: int,
|
||
concurrency: int,
|
||
wait_seconds: float,
|
||
poll_interval: float,
|
||
password_prefix: str = "hy",
|
||
fixed_password: str = "",
|
||
use_proxy: bool = False,
|
||
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=concurrency,
|
||
wait_seconds=wait_seconds,
|
||
poll_interval=poll_interval,
|
||
password_prefix=password_prefix,
|
||
fixed_password=fixed_password,
|
||
use_proxy=bool(use_proxy),
|
||
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,
|
||
*,
|
||
mode: str = "continue",
|
||
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:
|
||
"""继续批次。
|
||
|
||
mode:
|
||
- continue（默认）：从停止处往下跑，只处理 pending/stopped/中断中，
|
||
**不重试已失败(error)**，不重跑已成功。
|
||
- retry_failed：只重试 error。
|
||
- all_unfinished：error + pending + stopped + 中断中（旧行为）。
|
||
"""
|
||
with self._lock:
|
||
existing = self._runners.get(batch_id)
|
||
if existing and existing.batch.status == "running":
|
||
raise RuntimeError("批次正在运行中，请先停止或等待完成")
|
||
|
||
mode = (mode or "continue").strip().lower()
|
||
if mode not in {"continue", "retry_failed", "all_unfinished"}:
|
||
raise ValueError("mode 仅支持 continue / retry_failed / all_unfinished")
|
||
|
||
if mode == "continue":
|
||
target_statuses = RESUME_ITEM_STATUSES
|
||
empty_msg = "没有未完成条目可继续（已成功/已失败的不会自动重跑）"
|
||
run_label = "继续未完成"
|
||
elif mode == "retry_failed":
|
||
target_statuses = RETRY_FAILED_ITEM_STATUSES
|
||
empty_msg = "没有失败条目可重试"
|
||
run_label = "重试失败"
|
||
else:
|
||
target_statuses = RESUME_ITEM_STATUSES | RETRY_FAILED_ITEM_STATUSES
|
||
empty_msg = "没有可继续的未完成/失败条目"
|
||
run_label = "续跑未完成"
|
||
|
||
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):
|
||
status = row.status or "pending"
|
||
if status == "success":
|
||
continue
|
||
if status not in target_statuses:
|
||
continue
|
||
row.status = "pending"
|
||
row.message = "等待继续" if mode == "continue" else "等待重试"
|
||
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(empty_msg)
|
||
|
||
batch_row.status = "running"
|
||
batch_row.message = f"{run_label}（{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:
|
||
pass
|
||
|
||
runner = HuyaRegisterRunner(
|
||
batch=batch,
|
||
sms_lines=sms_lines,
|
||
proxy_config=proxy_config if batch.use_proxy else None,
|
||
item_indices=retry_indices,
|
||
)
|
||
runner.batch.status = "running"
|
||
runner.batch.message = f"{run_label}（{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()
|