coolapk_poll.py

#!/usr/bin/env python3
import argparse
import gzip
import json
import time
import urllib.error
import urllib.request

from coolapk_token import generate_token


DEFAULT_DEVICE = (
    "AZmV2N4UzN0UmZ3kDOzEzYgsjMwAjL2IjMwUjMuE0MRFEI7MkMxITM4AjMyAyOp1GZlJFI7kWbvFWaYByO"
    "gsDI7AyOzYGO3okVq1GWOlEez8WYLlkWKVWbllzX3pUTjFTcjx2aPVFR"
)
DEFAULT_URL = (
    "https://api.coolapk.com/v6/page/dataList?"
    "url=%2Fpage%3Furl%3DV11_VERTICAL_TOPIC&title=%E8%AF%9D%E9%A2%98&subTitle&page=1"
)
DEFAULT_UA = (
    "Dalvik/2.1.0 (Linux; U; Android 15; 22081212C Build/AQ3A.250226.002) "
    "(#Build; Redmi; 22081212C; AQ3A.250226.002; HyperOS_3.0; 3.0.1.0) "
    "+CoolMarket/16.0.3-2602111-universal"
)


def build_headers(
    token: str,
    device: str,
    cookie: str,
    app_version: str,
    app_code: int,
    user_agent: str,
) -> dict:
    headers = {
        "User-Agent": user_agent,
        "Connection": "Keep-Alive",
        "Accept-Encoding": "gzip",
        "X-Requested-With": "XMLHttpRequest",
        "X-Sdk-Int": "35",
        "X-Sdk-Locale": "zh-CN",
        "X-App-Id": "com.coolapk.market",
        "X-App-Token": token,
        "X-App-Version": app_version,
        "X-App-Code": str(app_code),
        "X-Api-Version": "16",
        "X-App-Device": device,
        "X-Dark-Mode": "0",
        "X-App-Channel": "coolapk",
        "X-App-Mode": "universal",
        "X-App-Supported": str(app_code),
    }
    if cookie:
        headers["Cookie"] = cookie
    return headers


def decode_body(resp: urllib.response.addinfourl, raw: bytes) -> str:
    if resp.headers.get("Content-Encoding", "").lower() == "gzip":
        raw = gzip.decompress(raw)
    return raw.decode("utf-8", errors="replace")


def generate_token_compatible(args: argparse.Namespace) -> tuple[str, int, int]:
    # pyca/bcrypt rejects some salts generated by the app algorithm.
    # Try a small forward window so the poller can run continuously.
    if args.ts is not None:
        token = generate_token(
            apk_path=args.apk,
            device=args.device,
            ts=args.ts,
            version_code=args.version_code,
            package=args.package,
        )
        return token, args.ts, 0

    start_ts = int(time.time())
    last_err = None
    for offset in range(args.max_ahead + 1):
        ts = start_ts + offset
        try:
            token = generate_token(
                apk_path=args.apk,
                device=args.device,
                ts=ts,
                version_code=args.version_code,
                package=args.package,
            )
            return token, ts, offset
        except ValueError as exc:
            if "Invalid salt" in str(exc):
                last_err = exc
                continue
            raise
    raise RuntimeError(f"could not generate token in +0..+{args.max_ahead}s window: {last_err}")


def run_once(args: argparse.Namespace) -> None:
    token, ts, offset = generate_token_compatible(args)
    headers = build_headers(
        token=token,
        device=args.device,
        cookie=args.cookie,
        app_version=args.app_version,
        app_code=args.version_code,
        user_agent=args.user_agent,
    )
    req = urllib.request.Request(args.url, headers=headers, method="GET")

    with urllib.request.urlopen(req, timeout=args.timeout) as resp:
        body = decode_body(resp, resp.read())
        http_code = resp.getcode()

    try:
        payload = json.loads(body)
        status = payload.get("status")
        message = payload.get("message", "")
        data = payload.get("data")
        size = len(data) if isinstance(data, list) else (1 if data is not None else 0)
        keys = ",".join(list(payload.keys())[:8])
        print(
            f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] "
            f"http={http_code} api_status={status} items={size} ts={ts} offset={offset}s "
            f"keys={keys} message={message}"
        )
        if args.print_json:
            print(json.dumps(payload, ensure_ascii=False, indent=2))
    except json.JSONDecodeError:
        snippet = body[:300].replace("\n", " ")
        print(
            f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] "
            f"http={http_code} non_json_preview={snippet}"
        )


def main() -> None:
    parser = argparse.ArgumentParser(description="CoolApk periodic poller with dynamic X-App-Token")
    parser.add_argument("--apk", default="coolapk.apk", help="Path to coolapk.apk")
    parser.add_argument("--url", default=DEFAULT_URL, help="API URL to request")
    parser.add_argument("--device", default=DEFAULT_DEVICE, help="X-App-Device")
    parser.add_argument(
        "--cookie",
        default="ddid=21b52710-79a8-43a9-a9b0-a773b106d08f",
        help="Optional Cookie header",
    )
    parser.add_argument("--package", default="com.coolapk.market", help="Package name")
    parser.add_argument("--app-version", default="16.0.3", help="X-App-Version")
    parser.add_argument("--version-code", type=int, default=2602111, help="X-App-Code")
    parser.add_argument("--user-agent", default=DEFAULT_UA, help="User-Agent")
    parser.add_argument("--timeout", type=int, default=20, help="Request timeout seconds")
    parser.add_argument("--interval", type=int, default=60, help="Interval seconds for loop mode")
    parser.add_argument("--count", type=int, default=0, help="Request count; 0 means infinite")
    parser.add_argument("--ts", type=int, default=None, help="Fixed timestamp; disables auto window")
    parser.add_argument("--max-ahead", type=int, default=10, help="Auto token timestamp window in seconds")
    parser.add_argument("--once", action="store_true", help="Run one request and exit")
    parser.add_argument("--print-json", action="store_true", help="Print full JSON response")
    args = parser.parse_args()

    if args.once:
        args.count = 1
    if args.interval < 1:
        raise ValueError("--interval must be >= 1")
    if args.max_ahead < 0:
        raise ValueError("--max-ahead must be >= 0")

    done = 0
    while True:
        start = time.time()
        try:
            run_once(args)
        except urllib.error.HTTPError as exc:
            body = exc.read().decode("utf-8", errors="replace")[:300].replace("\n", " ")
            print(
                f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] "
                f"http_error={exc.code} reason={exc.reason} body={body}"
            )
        except Exception as exc:
            print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] error={exc}")

        done += 1
        if args.count and done >= args.count:
            break

        elapsed = time.time() - start
        sleep_s = max(0.0, args.interval - elapsed)
        time.sleep(sleep_s)


if __name__ == "__main__":
    main()