chuanqi/cq2_sdk/scripts/test_jfbym_live.py

"""Live-test JfbymSliderSolver with the REAL token from docs/passwd.txt.

Downloads jfbym's own sample slider image (from the 22222 product page)
and runs solve_distance against it. Verifies:
  1. token is valid (not 10003)
  2. account has balance (not 10002)
  3. response parses to an int distance (not 10007)

Cost: 0.008 yuan per run (one 22222 call).
"""
import asyncio
import base64
import sys
from pathlib import Path

_HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(_HERE.parent / "src"))

from cq2_sdk.env_loader import load_env
load_env(_HERE.parent / ".env")

import httpx

from cq2_sdk.login.jfbym import JfbymError, JfbymSliderSolver

# 官方示例图 (来自 /test/286.html 产品页, 单图滑块 22222 的例图)
SAMPLES = [
    "https://www.jfbym.com/static/pic/hk/dthk/4.bmp",
    "https://obs.jfbym.com/public/upload/22222_1.png",
    "https://obs.jfbym.com/public/upload/22222_2.png",
]


async def main() -> int:
    solver = JfbymSliderSolver()  # token from .env
    if not solver.token:
        print("FAIL: CQ2_JFBYM_TOKEN not set")
        return 1
    print(f"token: {solver.token[:8]}...{solver.token[-4:]}")

    async with httpx.AsyncClient(timeout=15, verify=False) as c:
        img = None
        for url in SAMPLES:
            try:
                r = await c.get(url)
                if r.status_code == 200 and len(r.content) > 1000:
                    img = base64.b64encode(r.content).decode()
                    print(f"sample image: {url} ({len(r.content)} bytes)")
                    break
            except Exception as e:
                print(f"  [skip] {url}: {type(e).__name__}")
        if not img:
            print("FAIL: could not download any sample image")
            return 1

    print("\ncalling jfbym customApi (type 22222, costs 0.008 yuan)...")
    try:
        distance = await solver.solve_distance(img)
        print(f"[OK] recognized distance = {distance} px")
        print("     token valid + balance OK + response parses correctly")
    except JfbymError as e:
        print(f"[FAIL] jfbym code={e.code}: {e.msg}")
        if e.code == 10002:
            print("     → 需要充值 https://console.jfbym.com/own/charge/index.html")
        elif e.code == 10003:
            print("     → token 无效, 检查 docs/passwd.txt 是否复制完整")
        return 1
    finally:
        await solver.close()
    return 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))