biliTickerBuy/interface/common.py

from __future__ import annotations

import copy
import json
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs, urlparse

import requests


REQUIRED_FIELDS = (
    "detail",
    "count",
    "screen_id",
    "project_id",
    "sku_id",
    "pay_money",
    "buyer_info",
    "buyer",
    "tel",
    "deliver_info",
    "cookies",
)
BUYER_REQUIRED_FIELDS = ("name", "personal_id")
DELIVER_REQUIRED_FIELDS = ("name", "tel", "addr_id", "addr")
COOKIE_REQUIRED_FIELDS = ("name", "value")
SALES_FLAG_NUMBER_MAP = {
    1: "不可售",
    2: "预售",
    3: "停售",
    4: "售罄",
    5: "不可用",
    6: "库存紧张",
    8: "暂时售罄",
    9: "不在白名单",
    101: "未开始",
    102: "已结束",
    103: "未完成",
    105: "下架",
    106: "已取消",
}


def _load_json_file(path: str | Path) -> Any:
    with open(path, "r", encoding="utf-8") as handle:
        return json.load(handle)


def _read_tinydb_value(path: str | Path, key: str) -> Any:
    target = Path(path)
    if not target.exists():
        return None
    try:
        raw = _load_json_file(target)
    except Exception:
        return None
    if isinstance(raw, dict):
        default_group = raw.get("_default")
        if isinstance(default_group, dict):
            for item in default_group.values():
                if isinstance(item, dict) and item.get("key") == key:
                    return item.get("value")
        if raw.get("key") == key:
            return raw.get("value")
    if isinstance(raw, list):
        for item in raw:
            if isinstance(item, dict) and item.get("key") == key:
                return item.get("value")
    return None


def _cookie_store_path(cookies_path: str | Path | None) -> str | None:
    if cookies_path is not None:
        return str(Path(cookies_path))

    package_root = Path(__file__).resolve().parents[1]
    config_path = package_root / "config.json"
    configured = _read_tinydb_value(config_path, "cookies_path")
    if configured:
        return str(Path(configured))
    return str(package_root / "cookies.json")


def _coerce_cookie_store(raw: Any) -> list[dict[str, Any]] | None:
    if raw is None:
        return None
    if isinstance(raw, list):
        return copy.deepcopy(raw)
    if isinstance(raw, dict):
        if isinstance(raw.get("cookie"), list):
            return copy.deepcopy(raw["cookie"])
        default_group = raw.get("_default")
        if isinstance(default_group, dict):
            for item in default_group.values():
                if isinstance(item, dict) and item.get("key") == "cookie":
                    value = item.get("value")
                    if isinstance(value, list):
                        return copy.deepcopy(value)
    return None


def _resolve_cookie_list(
    cookies: list[dict[str, Any]] | dict[str, Any] | None = None,
    *,
    cookies_path: str | Path | None = None,
) -> list[dict[str, Any]] | None:
    if cookies is not None:
        return _coerce_cookie_store(cookies)

    store_path = _cookie_store_path(cookies_path)
    if not store_path:
        return None
    try:
        return _coerce_cookie_store(_load_json_file(store_path))
    except Exception:
        return None


def _cookies_to_header(cookies: list[dict[str, Any]] | None) -> str:
    if not cookies:
        return ""
    parts: list[str] = []
    for cookie in cookies:
        name = cookie.get("name")
        value = cookie.get("value")
        if name and value is not None:
            parts.append("{0}={1}".format(name, value))
    return "; ".join(parts)


def _fetch_username_silently(
    cookies: list[dict[str, Any]] | None,
    *,
    timeout: float = 10.0,
) -> str:
    if not cookies:
        return "Not login"
    headers = {
        "accept": "*/*",
        "accept-language": "zh-CN,zh;q=0.9",
        "referer": "https://show.bilibili.com/",
        "user-agent": (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/126.0.0.0 Safari/537.36"
        ),
        "cookie": _cookies_to_header(cookies),
    }
    try:
        response = requests.get(
            "https://api.bilibili.com/x/web-interface/nav",
            headers=headers,
            timeout=timeout,
        )
        response.raise_for_status()
        payload = response.json()
        data = payload.get("data") or {}
        username = data.get("uname")
        if isinstance(username, str) and username.strip():
            return username.strip()
    except Exception:
        return "Not login"
    return "Not login"


def _deepcopy_dict(data: Any) -> dict[str, Any]:
    if isinstance(data, dict):
        return copy.deepcopy(data)
    raise TypeError("config must be a dict or a json file path")


def _load_config(config_or_path: str | Path | dict[str, Any]) -> dict[str, Any]:
    if isinstance(config_or_path, (str, Path)):
        return _deepcopy_dict(_load_json_file(config_or_path))
    return _deepcopy_dict(config_or_path)


def _extract_project_id(project_input: str | int) -> int:
    if isinstance(project_input, int):
        return project_input

    text = str(project_input).strip()
    if not text:
        raise ValueError("project_input is empty")
    if text.isdigit():
        return int(text)

    parsed = urlparse(text)
    query = parse_qs(parsed.query)
    project_ids = query.get("id", [])
    if project_ids and project_ids[0].isdigit():
        return int(project_ids[0])
    raise ValueError("could not extract project id from input")


def _format_sale_status(ticket: dict[str, Any]) -> str:
    sale_flag_number = ticket.get("sale_flag_number")
    if sale_flag_number in SALES_FLAG_NUMBER_MAP:
        return SALES_FLAG_NUMBER_MAP[sale_flag_number]
    if "clickable" in ticket:
        return "可购买" if ticket.get("clickable") else "不可购买"
    return "未知状态"


def _make_request(
    *,
    cookies: list[dict[str, Any]] | dict[str, Any] | None = None,
    cookies_path: str | Path | None = None,
) -> Any:
    from util.request.BiliRequest import BiliRequest

    return BiliRequest(
        cookies=cookies,
        cookies_config_path=_cookie_store_path(cookies_path),
    )