345 lines
11 KiB
Python
345 lines
11 KiB
Python
"""虎牙基础业务服务。"""
|
|
|
|
import csv
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from urllib.parse import urlparse
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from core.huya.cookie_utils import cookie_value, normalize_huya_cookie
|
|
|
|
from ..huya_defaults import HUYA_CONFIG_DEFAULTS, HUYA_CONFIG_FIELDS
|
|
from ..models import HuyaAccount, HuyaConfig, HuyaTask
|
|
|
|
|
|
SUPPORTED_TASK_TYPES = {
|
|
"get_bind_qr": "获取绑定二维码",
|
|
"query_points": "一键查询积分",
|
|
"query_game_name": "一键查询游戏名",
|
|
"query_exchange_records": "一键查询兑换记录",
|
|
"confirm_bind": "确认绑定",
|
|
"refresh_goods": "刷新商品列表",
|
|
"refresh_recharge_goods": "刷新充值商品列表",
|
|
"exchange_goods": "兑换商品",
|
|
"create_recharge_order": "生成支付二维码",
|
|
}
|
|
|
|
|
|
def huya_config_value(field: str, value: str | None) -> str:
|
|
"""读取配置值；空值自动回退到当前活动默认配置。"""
|
|
text = str(value or "").strip()
|
|
return text or HUYA_CONFIG_DEFAULTS[field]
|
|
|
|
|
|
def apply_huya_config_defaults(config: HuyaConfig) -> bool:
|
|
"""补齐虎牙配置默认值，返回是否发生变更。"""
|
|
changed = False
|
|
for field in HUYA_CONFIG_FIELDS:
|
|
if field == "bind_act_id" and str(getattr(config, field, "") or "").strip() == "17096":
|
|
setattr(config, field, HUYA_CONFIG_DEFAULTS[field])
|
|
changed = True
|
|
continue
|
|
normalized = huya_config_value(field, getattr(config, field, None))
|
|
if getattr(config, field, None) != normalized:
|
|
setattr(config, field, normalized)
|
|
changed = True
|
|
return changed
|
|
|
|
|
|
def _looks_like_huya_cookie(value: str) -> bool:
|
|
"""判断文本是否像虎牙 Cookie。"""
|
|
return "udb_" in value or "yyuid=" in value
|
|
|
|
|
|
def parse_huya_cookie_line(line: str) -> dict | None:
|
|
"""解析单行虎牙 CK，兼容纯 CK、账号----密码----CK、CK----手机号。"""
|
|
raw = (line or "").strip()
|
|
if not raw:
|
|
return None
|
|
|
|
parts = [part.strip() for part in raw.split("----")]
|
|
username_hint = ""
|
|
game_phone = ""
|
|
|
|
if len(parts) == 1:
|
|
cookie = raw
|
|
elif _looks_like_huya_cookie(parts[0]):
|
|
cookie = parts[0]
|
|
game_phone = parts[1] if len(parts) >= 2 else ""
|
|
elif _looks_like_huya_cookie(parts[-1]):
|
|
cookie = parts[-1]
|
|
username_hint = parts[0]
|
|
else:
|
|
return None
|
|
|
|
if not _looks_like_huya_cookie(cookie):
|
|
return None
|
|
cookie = normalize_huya_cookie(cookie)
|
|
|
|
yyuid = cookie_value(cookie, "yyuid")
|
|
uid = cookie_value(cookie, "udb_uid") or yyuid
|
|
username = cookie_value(cookie, "udb_passport") or cookie_value(cookie, "username")
|
|
if not username and username_hint:
|
|
username = username_hint
|
|
|
|
if not uid and not yyuid:
|
|
return None
|
|
|
|
return {
|
|
"uid": uid,
|
|
"yyuid": yyuid or uid,
|
|
"username": username or uid or yyuid,
|
|
"cookie": cookie,
|
|
"game_phone": game_phone,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class HuyaPasswordLine:
|
|
"""虎牙账号密码导入行。"""
|
|
|
|
username: str
|
|
password: str
|
|
cookie: str = ""
|
|
phone: str = ""
|
|
sms_url: str = ""
|
|
|
|
|
|
def _looks_like_url(value: str) -> bool:
|
|
"""判断文本是否像 URL。"""
|
|
parsed = urlparse(str(value or "").strip())
|
|
return parsed.scheme in {"http", "https"} and bool(parsed.netloc)
|
|
|
|
|
|
def split_huya_password_line(line: str) -> HuyaPasswordLine | None:
|
|
"""拆分虎牙账号密码行，兼容账号----密码----手机号----验证码链接。"""
|
|
raw = (line or "").strip()
|
|
if not raw or raw.startswith("#"):
|
|
return None
|
|
|
|
if "----" in raw:
|
|
parts = [part.strip() for part in raw.split("----")]
|
|
elif "|" in raw:
|
|
parts = [part.strip() for part in raw.split("|")]
|
|
elif "\t" in raw:
|
|
parts = [part.strip() for part in raw.split("\t")]
|
|
elif "," in raw:
|
|
parts = [part.strip() for part in next(csv.reader([raw]))]
|
|
else:
|
|
parts = raw.split()
|
|
|
|
if len(parts) < 2:
|
|
return None
|
|
|
|
username = parts[0].strip()
|
|
password = parts[1].strip()
|
|
cookie = ""
|
|
phone = ""
|
|
sms_url = ""
|
|
extra = [part.strip() for part in parts[2:] if part.strip()]
|
|
if len(extra) >= 2 and _looks_like_url(extra[1]):
|
|
phone = extra[0]
|
|
sms_url = extra[1]
|
|
cookie = "----".join(extra[2:])
|
|
elif extra and _looks_like_url(extra[-1]) and len(extra) >= 2:
|
|
phone = extra[-2]
|
|
sms_url = extra[-1]
|
|
cookie = "----".join(extra[:-2])
|
|
elif extra:
|
|
cookie = "----".join(extra)
|
|
if not username or not password:
|
|
return None
|
|
return HuyaPasswordLine(
|
|
username=username,
|
|
password=password,
|
|
cookie=cookie,
|
|
phone=phone,
|
|
sms_url=sms_url,
|
|
)
|
|
|
|
|
|
def import_huya_password_accounts(db: Session, text: str, tag: str = "") -> tuple[int, int]:
|
|
"""导入虎牙账号密码，返回 (导入/更新数, 跳过数)。"""
|
|
created_or_updated = 0
|
|
skipped = 0
|
|
tag = (tag or "").strip()
|
|
|
|
for line in (text or "").splitlines():
|
|
parsed = split_huya_password_line(line)
|
|
if not parsed:
|
|
if line.strip():
|
|
skipped += 1
|
|
continue
|
|
|
|
username = parsed.username
|
|
password = parsed.password
|
|
cookie = parsed.cookie
|
|
account = db.query(HuyaAccount).filter(HuyaAccount.username == username).first()
|
|
if account is None:
|
|
account = HuyaAccount(
|
|
uid="",
|
|
yyuid="",
|
|
username=username,
|
|
account_password=password,
|
|
cookie=normalize_huya_cookie(cookie) if cookie else "",
|
|
game_phone=parsed.phone,
|
|
tag=tag,
|
|
status="password_imported",
|
|
)
|
|
db.add(account)
|
|
else:
|
|
account.account_password = password
|
|
if cookie:
|
|
account.cookie = normalize_huya_cookie(cookie)
|
|
if parsed.phone:
|
|
account.game_phone = parsed.phone
|
|
if tag:
|
|
account.tag = tag
|
|
if not account.cookie:
|
|
account.status = "password_imported"
|
|
account.updated_at = datetime.now(timezone.utc)
|
|
created_or_updated += 1
|
|
|
|
if created_or_updated:
|
|
db.commit()
|
|
return created_or_updated, skipped
|
|
|
|
|
|
def _upsert_huya_account(db: Session, parsed: dict, tag: str = "", status: str | None = None) -> HuyaAccount:
|
|
"""按 uid/yyuid 新增或更新虎牙账号。"""
|
|
account = None
|
|
if parsed["uid"]:
|
|
account = db.query(HuyaAccount).filter(HuyaAccount.uid == parsed["uid"]).first()
|
|
if account is None and parsed["yyuid"]:
|
|
account = db.query(HuyaAccount).filter(HuyaAccount.yyuid == parsed["yyuid"]).first()
|
|
|
|
if account is None:
|
|
account = HuyaAccount(
|
|
uid=parsed["uid"],
|
|
yyuid=parsed["yyuid"],
|
|
username=parsed["username"],
|
|
cookie=parsed["cookie"],
|
|
game_phone=parsed["game_phone"],
|
|
tag=tag,
|
|
status=status or "imported",
|
|
)
|
|
db.add(account)
|
|
else:
|
|
account.uid = parsed["uid"] or account.uid
|
|
account.yyuid = parsed["yyuid"] or account.yyuid
|
|
account.username = parsed["username"] or account.username
|
|
account.cookie = parsed["cookie"]
|
|
account.game_phone = parsed["game_phone"] or account.game_phone
|
|
if tag:
|
|
account.tag = tag
|
|
account.status = status or "updated"
|
|
account.updated_at = datetime.now(timezone.utc)
|
|
return account
|
|
|
|
|
|
def save_huya_login_cookie_to_account(
|
|
db: Session,
|
|
account: HuyaAccount,
|
|
cookie: str,
|
|
tag: str = "",
|
|
username_hint: str = "",
|
|
) -> HuyaAccount:
|
|
"""把登录成功后的 Cookie 回填到指定虎牙账号。"""
|
|
line = f"{username_hint}----{cookie}" if username_hint else cookie
|
|
parsed = parse_huya_cookie_line(line)
|
|
if not parsed:
|
|
raise ValueError("登录成功但 Cookie 中没有识别到虎牙 uid")
|
|
|
|
account.uid = parsed["uid"] or account.uid
|
|
account.yyuid = parsed["yyuid"] or account.yyuid
|
|
account.username = parsed["username"] or account.username
|
|
account.cookie = parsed["cookie"]
|
|
account.game_phone = parsed["game_phone"] or account.game_phone
|
|
if tag:
|
|
account.tag = tag
|
|
account.status = "login_success"
|
|
account.updated_at = datetime.now(timezone.utc)
|
|
db.commit()
|
|
db.refresh(account)
|
|
return account
|
|
|
|
|
|
def upsert_huya_cookie(db: Session, cookie: str, tag: str = "", username_hint: str = "") -> HuyaAccount:
|
|
"""保存单条登录得到的虎牙 Cookie。"""
|
|
line = f"{username_hint}----{cookie}" if username_hint else cookie
|
|
parsed = parse_huya_cookie_line(line)
|
|
if not parsed:
|
|
raise ValueError("登录成功但 Cookie 中没有识别到虎牙 uid")
|
|
account = _upsert_huya_account(db, parsed, tag=(tag or "").strip(), status="login_success")
|
|
db.commit()
|
|
db.refresh(account)
|
|
return account
|
|
|
|
|
|
def import_huya_cookies(db: Session, text: str, tag: str = "") -> tuple[int, int]:
|
|
"""导入虎牙 Cookie，返回 (成功数, 跳过数)。"""
|
|
created_or_updated = 0
|
|
skipped = 0
|
|
tag = (tag or "").strip()
|
|
|
|
for line in (text or "").splitlines():
|
|
parsed = parse_huya_cookie_line(line)
|
|
if not parsed:
|
|
skipped += 1
|
|
continue
|
|
|
|
_upsert_huya_account(db, parsed, tag=tag)
|
|
created_or_updated += 1
|
|
|
|
if created_or_updated:
|
|
db.commit()
|
|
return created_or_updated, skipped
|
|
|
|
|
|
def ensure_huya_config(db: Session) -> HuyaConfig:
|
|
"""获取单条虎牙配置，不存在则创建。"""
|
|
config = db.query(HuyaConfig).first()
|
|
if config:
|
|
if apply_huya_config_defaults(config):
|
|
db.commit()
|
|
db.refresh(config)
|
|
return config
|
|
config = HuyaConfig(**HUYA_CONFIG_DEFAULTS)
|
|
db.add(config)
|
|
db.commit()
|
|
db.refresh(config)
|
|
return config
|
|
|
|
|
|
def create_planned_tasks(
|
|
db: Session,
|
|
account_ids: list[int],
|
|
task_type: str,
|
|
created_by: int,
|
|
payload: dict | None = None,
|
|
) -> tuple[str, int]:
|
|
"""创建虎牙任务记录，等待后台执行器消费。"""
|
|
if task_type not in SUPPORTED_TASK_TYPES:
|
|
raise ValueError("不支持的任务类型")
|
|
|
|
batch_id = uuid.uuid4().hex[:12]
|
|
payload = payload or {}
|
|
accounts = db.query(HuyaAccount).filter(HuyaAccount.id.in_(account_ids)).all()
|
|
if task_type in {"refresh_goods", "refresh_recharge_goods", "exchange_goods", "create_recharge_order"} and accounts:
|
|
# 全局快照、单笔兑换和单笔支付二维码都使用一个选中的 CK 即可。
|
|
accounts = accounts[:1]
|
|
for account in accounts:
|
|
db.add(HuyaTask(
|
|
batch_id=batch_id,
|
|
account_id=account.id,
|
|
task_type=task_type,
|
|
status="planned",
|
|
message="任务已创建，等待执行",
|
|
result={"payload": payload} if payload else None,
|
|
created_by=created_by,
|
|
))
|
|
db.commit()
|
|
return batch_id, len(accounts)
|