chatgpt-register/ops/scan.py

"""Low-frequency token scan for local pool files."""

from __future__ import annotations

import argparse
import json
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import asdict, dataclass
from datetime import datetime
from pathlib import Path

from curl_cffi import requests
from platforms.chatgpt.fingerprint import OPENAI_FINGERPRINT_PROFILE, build_browser_headers

from .common import DEFAULT_POOL_DIR

DEFAULT_TIMEOUT = 15
DEFAULT_WORKERS = 5
DEFAULT_TRANSPORT_MAX_ATTEMPTS = 3
VALIDATE_URL = "https://chatgpt.com/backend-api/wham/usage"
RESPONSES_VALIDATE_URL = "https://chatgpt.com/backend-api/codex/responses"
PROBE_FINGERPRINT_PROFILE = OPENAI_FINGERPRINT_PROFILE
THREAD_LOCAL = threading.local()
PROJECT_DIR = Path(__file__).resolve().parents[1]
DEFAULT_OUTPUT_DIR = PROJECT_DIR / "logs"
RESPONSES_PROBE_PAYLOAD = {
    "model": "gpt-5.4",
    "instructions": "Return exactly OK.",
    "input": [{"role": "user", "content": "ping"}],
    "stream": True,
    "store": False,
    "text": {"verbosity": "low"},
}


@dataclass(frozen=True)
class ScanResult:
    file: str
    category: str
    status_code: int | None
    detail: str


def now_iso() -> str:
    return datetime.now().astimezone().isoformat(timespec="seconds")


def compact_text(value: str, limit: int = 240) -> str:
    return " ".join(str(value or "").split())[:limit]


def get_session() -> requests.Session:
    session = getattr(THREAD_LOCAL, "session", None)
    if session is None:
        session = requests.Session()
        THREAD_LOCAL.session = session
    return session


def reset_session() -> None:
    session = getattr(THREAD_LOCAL, "session", None)
    if session is None:
        return
    try:
        session.close()
    except Exception:
        pass
    THREAD_LOCAL.session = None


def is_transient_transport_error(exc: Exception) -> bool:
    message = str(exc or "").lower()
    markers = (
        "connection closed abruptly",
        "connection timed out",
        "connection reset",
        "connection refused",
        "tls connect error",
        "recv failure",
        "send failure",
        "http/2 stream",
        "operation timed out",
        "unexpected eof",
        "tls handshake timeout",
        "eof",
        "curl: (7)",
        "curl: (28)",
        "curl: (35)",
        "curl: (52)",
        "curl: (55)",
        "curl: (56)",
        "curl: (16)",
        "nghttp2",
    )
    return any(marker in message for marker in markers)


def iter_token_files(token_dir: Path, limit: int | None = None) -> list[Path]:
    files = sorted(path for path in token_dir.glob("*.json") if path.is_file())
    if limit is not None and limit >= 0:
        return files[:limit]
    return files


def _load_token_payload(path: Path) -> tuple[dict[str, object] | None, ScanResult | None]:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError as exc:
        return None, ScanResult(file=path.name, category="missing", status_code=None, detail=f"missing_file: {exc}")
    except Exception as exc:
        return None, ScanResult(file=path.name, category="suspicious", status_code=None, detail=f"invalid_json: {exc}")

    if not isinstance(payload, dict):
        return None, ScanResult(file=path.name, category="suspicious", status_code=None, detail="invalid_json: token record must be object")
    return payload, None


def _extract_credentials(path: Path, payload: dict[str, object]) -> tuple[str, str] | ScanResult:
    access_token = str(payload.get("access_token") or "").strip()
    account_id = str(payload.get("account_id") or "").strip()
    if not access_token or not account_id:
        return ScanResult(
            file=path.name,
            category="suspicious",
            status_code=None,
            detail="missing access_token or account_id",
        )
    return access_token, account_id


def _request_with_retry(
    method: str,
    url: str,
    *,
    headers: dict[str, str],
    json_body: object | None,
    proxy: str | None,
    timeout: int,
) -> ScanResult | object:
    proxies = {"http": proxy, "https": proxy} if proxy else None
    last_exc: Exception | None = None
    response = None
    for attempt in range(1, DEFAULT_TRANSPORT_MAX_ATTEMPTS + 1):
        try:
            request_fn = getattr(get_session(), method.lower())
            response = request_fn(
                url,
                headers=headers,
                json=json_body,
                proxies=proxies,
                impersonate="chrome",
                timeout=timeout,
            )
            break
        except Exception as exc:
            last_exc = exc
            if not is_transient_transport_error(exc):
                return ScanResult(file="", category="suspicious", status_code=None, detail=f"request_error: {exc}")
            reset_session()
            if attempt >= DEFAULT_TRANSPORT_MAX_ATTEMPTS:
                return ScanResult(file="", category="transport_error", status_code=None, detail=f"transport_error: {exc}")
    if response is None:
        return ScanResult(file="", category="transport_error", status_code=None, detail=f"transport_error: {last_exc or 'request failed'}")
    return response


def _probe_usage_path(path: Path, access_token: str, account_id: str, proxy: str | None, timeout: int) -> ScanResult:
    response = _request_with_retry(
        "GET",
        VALIDATE_URL,
        headers=build_browser_headers(
            access_token=access_token,
            account_id=account_id,
            accept="application/json",
            content_type="application/json",
        ),
        json_body=None,
        proxy=proxy,
        timeout=timeout,
    )
    if isinstance(response, ScanResult):
        return ScanResult(file=path.name, category=response.category, status_code=response.status_code, detail=response.detail)

    detail = compact_text(response.text)
    if response.status_code == 200:
        return ScanResult(file=path.name, category="normal", status_code=200, detail=detail)
    if response.status_code == 401:
        return ScanResult(file=path.name, category="invalid", status_code=401, detail=detail)
    if response.status_code == 429:
        return ScanResult(file=path.name, category="rate_limited", status_code=429, detail=detail)
    if int(response.status_code or 0) >= 500:
        return ScanResult(file=path.name, category="service_error", status_code=int(response.status_code), detail=detail)
    return ScanResult(file=path.name, category="suspicious", status_code=response.status_code, detail=detail)


def _probe_responses_path(path: Path, access_token: str, account_id: str, proxy: str | None, timeout: int) -> ScanResult:
    response = _request_with_retry(
        "POST",
        RESPONSES_VALIDATE_URL,
        headers=build_browser_headers(
            access_token=access_token,
            account_id=account_id,
            accept="text/event-stream",
            content_type="application/json",
        ),
        json_body=RESPONSES_PROBE_PAYLOAD,
        proxy=proxy,
        timeout=max(timeout, 20),
    )
    if isinstance(response, ScanResult):
        return ScanResult(file=path.name, category=response.category, status_code=response.status_code, detail=response.detail)

    detail = compact_text(response.text, limit=320)
    if response.status_code == 200:
        if "response.failed" in detail or '"status":"failed"' in detail:
            return ScanResult(file=path.name, category="service_error", status_code=200, detail=detail)
        return ScanResult(file=path.name, category="normal", status_code=200, detail="responses_ok")
    if response.status_code == 401:
        return ScanResult(file=path.name, category="invalid", status_code=401, detail=detail)
    if response.status_code == 429:
        return ScanResult(file=path.name, category="rate_limited", status_code=429, detail=detail)
    if int(response.status_code or 0) >= 500:
        return ScanResult(file=path.name, category="service_error", status_code=int(response.status_code), detail=detail)
    return ScanResult(file=path.name, category="suspicious", status_code=response.status_code, detail=detail)


def classify_token_file(
    path: Path,
    proxy: str | None,
    timeout: int,
    *,
    require_response_path: bool = False,
) -> ScanResult:
    payload, load_error = _load_token_payload(path)
    if load_error is not None:
        return load_error
    assert payload is not None

    credentials = _extract_credentials(path, payload)
    if isinstance(credentials, ScanResult):
        return credentials
    access_token, account_id = credentials

    usage_result = _probe_usage_path(path, access_token, account_id, proxy, timeout)
    if not require_response_path or usage_result.category != "normal":
        return usage_result

    response_result = _probe_responses_path(path, access_token, account_id, proxy, timeout)
    if response_result.category == "normal":
        return ScanResult(file=path.name, category="normal", status_code=200, detail="usage_ok | responses_ok")
    return response_result


def scan_once(
    token_dir: Path,
    proxy: str | None,
    timeout: int,
    workers: int,
    output_dir: Path,
    limit: int | None = None,
) -> dict[str, object]:
    output_dir.mkdir(parents=True, exist_ok=True)
    files = iter_token_files(token_dir, limit=limit)
    results: list[ScanResult] = []
    with ThreadPoolExecutor(max_workers=max(1, workers)) as executor:
        future_map = {executor.submit(classify_token_file, path, proxy, timeout): path for path in files}
        for future in as_completed(future_map):
            results.append(future.result())

    results.sort(key=lambda item: item.file)
    summary = {
        "total": len(results),
        "normal": sum(1 for item in results if item.category == "normal"),
        "invalid": sum(1 for item in results if item.category == "invalid"),
        "rate_limited": sum(1 for item in results if item.category == "rate_limited"),
        "suspicious": sum(1 for item in results if item.category == "suspicious"),
        "service_error": sum(1 for item in results if item.category == "service_error"),
        "transport_error": sum(1 for item in results if item.category == "transport_error"),
        "missing": sum(1 for item in results if item.category == "missing"),
    }
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    report_path = output_dir / f"scan_report_{timestamp}.json"
    payload = {
        "generated_at": now_iso(),
        "token_dir": str(token_dir),
        "proxy": proxy,
        "timeout_seconds": timeout,
        "workers": workers,
        "limit": limit,
        "summary": summary,
        "results": [asdict(item) for item in results],
    }
    report_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    return {
        "summary": summary,
        "report_path": str(report_path),
        "results": payload["results"],
    }


def main() -> None:
    parser = argparse.ArgumentParser(description="Low-frequency scan for local zhuce6 token files")
    parser.add_argument("--token-dir", default=str(DEFAULT_POOL_DIR), help="Token directory, default zhuce6 pool")
    parser.add_argument("--proxy", default=None, help="Optional proxy URL")
    parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT, help="Per-request timeout seconds")
    parser.add_argument("--workers", type=int, default=DEFAULT_WORKERS, help="Concurrent workers")
    parser.add_argument("--output-dir", default=str(DEFAULT_OUTPUT_DIR), help="Report output directory")
    parser.add_argument("--limit", type=int, default=None, help="Optional cap for scanned files")
    args = parser.parse_args()

    token_dir = Path(args.token_dir).expanduser().resolve()
    if not token_dir.is_dir():
        raise SystemExit(f"token directory does not exist: {token_dir}")

    summary = scan_once(
        token_dir=token_dir,
        proxy=str(args.proxy or "").strip() or None,
        timeout=max(1, int(args.timeout)),
        workers=max(1, int(args.workers)),
        output_dir=Path(args.output_dir).expanduser().resolve(),
        limit=args.limit,
    )
    stats = summary["summary"]
    print(f"[scan] dir={token_dir}")
    print(f"[scan] total={stats['total']} normal={stats['normal']} invalid={stats['invalid']} suspicious={stats['suspicious']}")
    print(f"[scan] report={summary['report_path']}")


if __name__ == "__main__":
    main()