183 lines
5.6 KiB
Python
183 lines
5.6 KiB
Python
"""虎牙基础业务服务。"""
|
|
|
|
import re
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..huya_defaults import HUYA_CONFIG_DEFAULTS, HUYA_CONFIG_FIELDS
|
|
from ..models import HuyaAccount, HuyaConfig, HuyaTask
|
|
|
|
|
|
SUPPORTED_TASK_TYPES = {
|
|
"get_bind_qr": "获取绑定二维码",
|
|
"query_points": "一键查询积分",
|
|
"open_elite_book": "开通精英宝典",
|
|
"recharge_points": "充值积分",
|
|
"query_game_name": "一键查询游戏名",
|
|
"query_exchange_records": "一键查询兑换记录",
|
|
"confirm_bind": "确认绑定",
|
|
"refresh_goods": "刷新商品列表",
|
|
}
|
|
|
|
|
|
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:
|
|
normalized = huya_config_value(field, getattr(config, field, None))
|
|
if getattr(config, field, None) != normalized:
|
|
setattr(config, field, normalized)
|
|
changed = True
|
|
return changed
|
|
|
|
|
|
def cookie_value(cookie: str, key: str) -> str:
|
|
"""从 Cookie 文本中提取指定 key。"""
|
|
match = re.search(rf"(?:^|;\s*){re.escape(key)}=([^;]+)", cookie or "")
|
|
return match.group(1).strip() if match else ""
|
|
|
|
|
|
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
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
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
|
|
|
|
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="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 = "updated"
|
|
account.updated_at = datetime.now(timezone.utc)
|
|
|
|
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()
|
|
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)
|