live-hub-py/web/backend/routers/huya/cookies.py

"""虎牙 Cookie 查看、导出与清除。"""

from fastapi import APIRouter

from ._common import (
    Depends,
    HTTPException,
    HuyaAccount,
    Query,
    Session,
    StreamingResponse,
    User,
    _can_view_huya_cookie,
    _cookie_out,
    _cookie_out_light,
    _has_huya_perm,
    _require_huya_perm,
    _visible_huya_accounts_query,
    csv,
    defer,
    get_current_user,
    get_db,
    io,
    normalize_huya_cookie,
    or_,
    rebuild_web_cookie_from_raw,
    record_audit,
)

router = APIRouter()


@router.get("/cookies")
def list_cookies(
    search: str = Query(""),
    page: int | None = Query(None, ge=1),
    page_size: int = Query(20, ge=1, le=200),
    include_cookie: bool = Query(True),
    db: Session = Depends(get_db),
    current: User = Depends(get_current_user),
):
    """查看当前用户可见的虎牙 Cookie。"""
    should_include_cookie = include_cookie and _can_view_huya_cookie(current)
    query = _visible_huya_accounts_query(db, current).filter(HuyaAccount.cookie != "")
    search_text = (search or "").strip()
    if search_text:
        pattern = f"%{search_text}%"
        query = query.outerjoin(User, HuyaAccount.assigned_to == User.id).filter(
            or_(
                HuyaAccount.uid.ilike(pattern),
                HuyaAccount.yyuid.ilike(pattern),
                HuyaAccount.username.ilike(pattern),
                HuyaAccount.nickname.ilike(pattern),
                User.username.ilike(pattern),
            )
        )
    total = None
    if page is not None:
        total = query.order_by(None).count()
    query = query.order_by(HuyaAccount.updated_at.desc())
    if page is not None:
        query = query.offset((page - 1) * page_size).limit(page_size)
    if not should_include_cookie:
        query = query.options(
            defer(HuyaAccount.cookie), defer(HuyaAccount.account_password)
        )
    accounts = query.all()
    result = (
        [_cookie_out(account, include_cookie=True) for account in accounts]
        if should_include_cookie
        else [_cookie_out_light(account) for account in accounts]
    )
    if page is not None:
        return {
            "items": result,
            "total": total or 0,
            "page": page,
            "page_size": page_size,
        }
    return result


@router.get("/cookies/summary")
def cookies_summary(
    db: Session = Depends(get_db),
    current: User = Depends(get_current_user),
):
    """虎牙 Cookie 统计，避免前端为了卡片统计拉全量 Cookie。"""
    query = _visible_huya_accounts_query(db, current).filter(HuyaAccount.cookie != "")
    total = query.count()
    assigned_count = query.filter(HuyaAccount.assigned_to.isnot(None)).count()
    return {
        "total": total,
        "assigned_count": assigned_count,
        "unassigned_count": max(0, total - assigned_count),
    }


@router.get("/cookies/export")
def export_cookies(
    format: str = "csv",
    db: Session = Depends(get_db),
    current: User = Depends(get_current_user),
):
    """导出虎牙 Cookie，支持 csv 和 custom 格式。"""
    _require_huya_perm(current, "huya:cookie:export")
    accounts = (
        _visible_huya_accounts_query(db, current)
        .filter(HuyaAccount.cookie != "")
        .order_by(HuyaAccount.updated_at.desc())
        .all()
    )
    if format == "custom":
        lines = []
        for account in accounts:
            username = account.username or account.nickname or account.uid or ""
            lines.append(f"{username}----{normalize_huya_cookie(account.cookie or '')}")
        content = "\r\n".join(lines)
        filename = "huya_cookies_custom.txt"
        media = "text/plain"
    else:
        output = io.StringIO()
        writer = csv.writer(output)
        writer.writerow(["账号", "UID", "YYUID", "Cookie", "时间"])
        for account in accounts:
            username = account.nickname or account.username or account.uid or ""
            writer.writerow(
                [
                    username,
                    account.uid or "",
                    account.yyuid or "",
                    normalize_huya_cookie(account.cookie or ""),
                    account.updated_at.isoformat() if account.updated_at else "",
                ]
            )
        content = output.getvalue()
        filename = "huya_cookies.csv"
        media = "text/csv"
    return StreamingResponse(
        iter([content]),
        media_type=media,
        headers={"Content-Disposition": f"attachment; filename={filename}"},
    )


@router.get("/cookies/{account_id}")
def get_cookie(
    account_id: int,
    db: Session = Depends(get_db),
    current: User = Depends(get_current_user),
):
    """获取单条虎牙 Cookie 详情，供复制操作按需读取完整敏感字段。"""
    account = (
        _visible_huya_accounts_query(db, current)
        .filter(HuyaAccount.id == account_id, HuyaAccount.cookie != "")
        .first()
    )
    if not account:
        raise HTTPException(status_code=404, detail="记录不存在")
    return _cookie_out(account, include_cookie=_can_view_huya_cookie(current))


@router.delete("/cookies/batch")
def delete_cookies_batch(
    account_ids: str = "",
    db: Session = Depends(get_db),
    current: User = Depends(get_current_user),
):
    """批量清除虎牙 Cookie，保留账号记录。"""
    _require_huya_perm(current, "huya:cookie:export")
    ids = [int(x) for x in account_ids.split(",") if x.strip().isdigit()]
    if not ids:
        raise HTTPException(status_code=400, detail="无效的账号ID")
    accounts = (
        _visible_huya_accounts_query(db, current).filter(HuyaAccount.id.in_(ids)).all()
    )
    for account in accounts:
        account.cookie = ""
        account.status = "invalid"
    db.commit()
    return {
        "message": f"已清除 {len(accounts)} 条虎牙 Cookie",
        "deleted": len(accounts),
        "success": True,
    }


@router.delete("/cookies/{account_id}")
def delete_cookie(
    account_id: int,
    db: Session = Depends(get_db),
    current: User = Depends(get_current_user),
):
    """清除单个虎牙 Cookie，保留账号记录。"""
    _require_huya_perm(current, "huya:cookie:export")
    account = (
        _visible_huya_accounts_query(db, current)
        .filter(HuyaAccount.id == account_id)
        .first()
    )
    if not account:
        raise HTTPException(status_code=404, detail="记录不存在")
    account.cookie = ""
    account.status = "invalid"
    db.commit()
    return {"message": "已清除", "success": True}


@router.post("/cookies/{account_id}/rebuild")
def rebuild_web_cookie(
    account_id: int,
    db: Session = Depends(get_db),
    current: User = Depends(get_current_user),
):
    """从 App 原始凭据重建网页运行 Cookie，不发起登录或支付。"""
    if not (
        _has_huya_perm(current, "huya:import")
        or _has_huya_perm(current, "huya:task")
        or _has_huya_perm(current, "huya:cookie:export")
    ):
        raise HTTPException(status_code=403, detail="无权重建虎牙 Cookie")
    account = (
        _visible_huya_accounts_query(db, current)
        .filter(HuyaAccount.id == account_id)
        .first()
    )
    if not account:
        raise HTTPException(status_code=404, detail="记录不存在")
    try:
        rebuild_web_cookie_from_raw(db, account)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    record_audit(
        db,
        current,
        action="huya:cookie:rebuild",
        target=f"huya_account:{account.id}",
        detail={"account_id": account.id},
    )
    db.commit()
    return _cookie_out(account, include_cookie=_can_view_huya_cookie(current))