spider_work/jiwu_spider/jiwu_core.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/8/18 19:19
"""集物星球爬虫核心库：签名、脚本自登录拿 sessionId、通用请求（curl_cffi 过 TLS 风控）。

 三大关键点：
     1. 请求头签名 X_SIG_1 = base64(HMAC_SHA256(key, msg))，msg 拼接顺序见 _sign。
     2. 必须用 curl_cffi（impersonate=chrome）发请求——服务端按 TLS/JA3 指纹识别客户端，
        python requests 的指纹会被判为非真机，接口静默返回 code:200 空壳（data.result=[]）；
        curl_cffi 才能拿到真正的 code:1 数据。
     3. sessionId 由客户端自造、regLogin 上报绑定，服务端沿用该值；请求头 X_SID、body.sessionId、
        签名串里的 sessionId 三处必须同值。业务接口成功码是 code==1。
 """
import sys
import time
import json
import hmac
import base64
import random
import hashlib

from curl_cffi import requests as creq
from loguru import logger
from tenacity import retry, stop_after_attempt, wait_fixed

BASE_URL = "https://api.jiwuplanet.com"
API_SIGN_SECRET_KEY = "EahycFpDpewggO2rksvQwTlnhCbTHFx6"  # 生产环境签名密钥（逆向所得）
X_VERSION = "2.12.2"  # App 版本号
X_PLATFORM = "150"  # 平台常量
IMPERSONATE = "chrome"  # curl_cffi TLS 指纹，过服务端风控的关键
NO_PROXY = {"http": None, "https": None}  # 显式直连，忽略系统 HTTP(S)_PROXY(127.0.0.1:7890)
BIZ_OK_CODE = 1  # 业务接口成功码（登录/查询成功均为 1）
PRICE_DIVISOR = 1000000  # 金额原始值 ÷ 此 = 元（实测 amount=99000000 → App 显示 ¥99.00）

# 脚本自登录账号（无人值守续期用；账密为运行必需，如需可挪到 application.yml）
LOGIN_ACCOUNT = {
    "appVersions": X_VERSION, "deviceNumber": "40b03f4bec96675f",
    "identificationNumber": "19521500850", "ip": "10.1.10.1", "language": "ZH",
    "loginPassword": "pass2022", "loginType": "ACCOUNT", "mobile": "19521500850",
    "mobileCode": "86", "platformInfo": "Google Pixel 5", "platformType": "ANDROID_USER",
    "systemBusinessType": 6, "terminalVersions": "11",
}

SESSION_TTL = 1800  # sessionId 复用时长(秒)，超时重新登录
_session = {"sid": None, "ts": 0.0}  # 登录态 sessionId 内存缓存
_anon = {"sid": None}  # 免登录接口用的匿名 sid（随机自造、不经 regLogin，进程内复用；被风控则轮换）


def after_log(retry_state):
    """tenacity 重试回调，记录每次尝试的结果。

    Args:
        retry_state: tenacity 传入的 RetryCallState，args[0] 约定为 log。
    """
    log = retry_state.args[0] if retry_state.args else logger
    if retry_state.outcome.failed:
        log.warning(f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} Times")
    else:
        log.info(f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} succeeded")


def canonical_body(body_obj: dict) -> str:
    """把请求体规范化为被签名/被发送的紧凑串（等价 App 端 Gson 处理）。

    Args:
        body_obj (dict): 原始请求体字典。

    Returns:
        str: 递归按 key 排序、紧凑、非 ASCII 不转义的 JSON 串；既用于签名也用于实际发送。
    """
    return json.dumps(body_obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)


def to_yuan(raw) -> float | None:
    """把接口原始金额换算成元（÷PRICE_DIVISOR，保留 2 位），供入库前统一转换。

    Args:
        raw (int | float | None): 接口原始金额。

    Returns:
        float | None: 元（保留 2 位）；raw 为空返回 None。
    """
    return round(raw / PRICE_DIVISOR, 2) if isinstance(raw, (int, float)) else None


def gen_session_id() -> str:
    """自造一个 32 位小写 hex 会话标识（服务端沿用此值作会话）。

    Returns:
        str: 32 位十六进制字符串。
    """
    return hashlib.md5(f"{time.time()}-{random.random()}".encode()).hexdigest()


def _sign(path: str, body_str: str, session_id: str) -> dict:
    """按签名算法生成带 X_SIG_1 的完整请求头。

    Args:
        path (str): URL 路径（encodedPath，不含 host/query）。
        body_str (str): 规范化后的请求体字符串。
        session_id (str): 会话标识，与 body 内 sessionId 一致。

    Returns:
        dict: 含 6 个 X_* 头及基础头的请求头字典。
    """
    x_time = str(int(time.time()))  # 秒级时间戳
    x_rand = str(random.randint(100000, 999999))  # 6 位随机数
    msg = path + X_VERSION + X_PLATFORM + session_id + x_time + x_rand + body_str
    sig = base64.b64encode(
        hmac.new(API_SIGN_SECRET_KEY.encode(), msg.encode(), hashlib.sha256).digest()
    ).decode()
    return {
        "X_VERSION": X_VERSION, "X_TIME": x_time, "X_SID": session_id,
        "X_RAND": x_rand, "X_PLATFORM": X_PLATFORM, "X_SIG_1": sig,
        "Content-Type": "application/json;charset=UTF-8",
        "User-Agent": "okhttp/3.12.10", "Host": "api.jiwuplanet.com",
    }


@retry(stop=stop_after_attempt(3), wait=wait_fixed(2), after=after_log)
def _post(log, path: str, body_obj: dict, session_id: str) -> dict:
    """用 curl_cffi 发一次带签名的 POST 请求（带重试）。

    Args:
        log: 日志对象（首参，供 after_log 取用）。
        path (str): 接口路径。
        body_obj (dict): 请求体（须已含 sessionId）。
        session_id (str): 会话标识。

    Returns:
        dict: 响应 JSON。

    Raises:
        RuntimeError: HTTP 状态非 200 时抛出以触发重试。
    """
    body_str = canonical_body(body_obj)
    headers = _sign(path, body_str, session_id)
    r = creq.post(BASE_URL + path, headers=headers, data=body_str.encode("utf-8"),
                  timeout=20, impersonate=IMPERSONATE, proxies=NO_PROXY)
    if r.status_code != 200:
        log.error(f"请求失败 HTTP {r.status_code}: {path}")
        raise RuntimeError(f"HTTP {r.status_code}")
    return r.json()


def login(log) -> str:
    """脚本自登录：自造 sessionId 调 regLogin 绑定，返回可用会话标识并写入缓存。

    Args:
        log: 日志对象。

    Returns:
        str: 登录成功后可用的 sessionId；失败返回空串。
    """
    sid = gen_session_id()
    body = dict(LOGIN_ACCOUNT)
    body["sessionId"] = sid
    try:
        j = _post(log, "/acct/user/regLogin", body, sid)
    except Exception as e:
        log.error(f"登录请求异常: {e}")
        return ""
    if j.get("code") != BIZ_OK_CODE:
        log.error(f"登录失败: code={j.get('code')} msg={j.get('message')}")
        return ""
    data = j.get("data") or {}
    ret_sid = data.get("sessionId") or sid
    _session["sid"] = ret_sid
    _session["ts"] = time.time()
    log.success(f"登录成功 nickName={data.get('nickName')} userId={data.get('userId')} sid={ret_sid}")
    return ret_sid


def ensure_session(log) -> str:
    """取可用 sessionId：内存缓存未过期直接用，否则重新登录。

    Args:
        log: 日志对象。

    Returns:
        str: 可用的 sessionId；登录失败返回空串。
    """
    if _session["sid"] and (time.time() - _session["ts"] < SESSION_TTL):
        return _session["sid"]
    return login(log)


def get_anon_sid() -> str:
    """取免登录接口用的匿名 sessionId：随机自造、不经 regLogin，进程内复用。

    Returns:
        str: 32 位十六进制匿名会话标识。
    """
    if not _anon["sid"]:
        _anon["sid"] = gen_session_id()
    return _anon["sid"]


def do_request(log, path: str, body: dict, need_auth: bool = False) -> dict | None:
    """通用业务请求：自动带 sessionId + 签名。

    登录态最小化：**默认走免登录**（need_auth=False，用随机未绑定的匿名 sid 签名，不调 regLogin）；
    仅服务端强制要求登录的接口（商家列表 hotRecommend、购买记录 publicity/user/group/pager、拆卡报告 gift/report）才传 need_auth=True。

    Args:
        log: 日志对象。
        path (str): 接口路径。
        body (dict): 业务请求参数（不含 sessionId，内部自动补）。
        need_auth (bool, optional): 是否需要真实登录会话。False=匿名免登录；True=regLogin 会话。Defaults to False。

    Returns:
        dict | None: 成功返回响应 JSON（code==1）；失败返回 None。
    """
    if not need_auth:  # 免登录：匿名 sid，被风控则轮换匿名 sid 下轮再来（始终不登录）
        sid = get_anon_sid()
        b = dict(body)
        b["sessionId"] = sid
        try:
            j = _post(log, path, b, sid)
        except Exception as e:
            log.warning(f"请求异常 {path}: {e}")
            return None
        if j.get("code") == BIZ_OK_CODE:
            return j
        log.warning(f"{path} 匿名请求返回 code={j.get('code')}，轮换匿名 sid")
        _anon["sid"] = None
        return None

    for attempt in range(2):  # 需登录：最多两次（首次 + 会话失效重登后再试）
        sid = ensure_session(log)
        if not sid:
            return None
        b = dict(body)
        b["sessionId"] = sid
        try:
            j = _post(log, path, b, sid)
        except Exception as e:
            log.warning(f"请求异常 {path}: {e}")
            return None
        if j.get("code") == BIZ_OK_CODE:
            return j
        log.warning(f"{path} 返回 code={j.get('code')}，判为会话失效，第 {attempt + 1} 次，清缓存重登")
        _session["sid"] = None
    return None


if __name__ == "__main__":
    # 自测：登录 + 在售 + 购买记录
    logger.remove()
    logger.add(sys.stderr, level="INFO", format="[{time:HH:mm:ss}] {level} {message}")
    _log = logger

    _log.info("===== 自测1：脚本自登录 =====")
    _sid = ensure_session(_log)
    _log.info(f"拿到 sessionId: {_sid}")

    _log.info("===== 自测2：在售列表 index/top =====")
    _j = do_request(_log, "/search/app/index/top",
                    {"currentPage": "1", "limit": "10", "productType": "3", "systemBusinessType": 5})
    if _j:
        _d = _j["data"]
        _log.info(f"code={_j['code']} total={_d.get('totalCount')} 本页={len(_d.get('records', []))}")
        if _d.get("records"):
            _r = _d["records"][0]
            _log.info(
                f"首条: {_r.get('goodsName')} | {_r.get('corpInfoName')}({_r.get('corpInfoId')}) | 价={_r.get('highestPrice')}")

    _log.info("===== 自测3：购买记录 玩家维度（需登录）=====")
    _j = do_request(_log, "/order/merchant/app/query/gift/publicity/user/group/pager",
                    {"currentPage": "1", "giftBusinessName": "", "goodsId": "1660823", "limit": "5"},
                    need_auth=True)
    if _j:
        _d = _j["data"]
        _log.info(f"code={_j['code']} total={_d.get('totalCount')} 本页={len(_d.get('records', []))}")
        for _rr in _d.get("records", [])[:3]:
            _log.info(f"买家={_rr.get('userNick')} 份数={_rr.get('count')} uid={_rr.get('userId')}")