scripts/label_captchas.py

#!/usr/bin/env python3
"""用 pie-xian command-a-vision 批量识别验证码：3x放大 + 每张2次读取(不一致补第3次) + 多数投票。
用法: label_captchas.py <图片目录> <输出.json> [workers]
输出: {filename: {label, votes:[...], count, agreed}}
"""
import base64
import concurrent.futures
import glob
import json
import os
import re
import ssl
import sys
import tempfile
import urllib.error
import urllib.request
from collections import Counter

from PIL import Image

API = "https://api.pie-xian.com/v1/chat/completions"
KEY = os.environ.get("PIE_XIAN_API_KEY", "sk-Rel8iERHrmu9NDbvQhnhwTowEUnGqVvCFXMatASH8q5O4gGP")
MODEL = "command-a-vision-07-2025"
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/126.0.0.0 Safari/537.36"
PROMPT = "识别这张验证码图片中的字符。只输出字符本身,不要任何解释、标点或额外文字。这是登录验证码,可能包含1-4个数字。"

_CTX = ssl.create_default_context()
_CTX.check_hostname = False
_CTX.verify_mode = ssl.CERT_NONE


def call_vision(big_path):
    b64 = base64.b64encode(open(big_path, "rb").read()).decode()
    body = json.dumps({
        "model": MODEL,
        "messages": [{"role": "user", "content": [
            {"type": "text", "text": PROMPT},
            {"type": "image_url", "image_url": {"url": "data:image/png;base64," + b64}},
        ]}],
        "max_tokens": 20, "temperature": 0,
    }).encode()
    req = urllib.request.Request(API, data=body, headers={
        "Authorization": "Bearer " + KEY, "Content-Type": "application/json", "User-Agent": UA})
    d = json.loads(urllib.request.urlopen(req, timeout=90, context=_CTX).read())
    code = re.sub(r"[^0-9A-Za-z]", "", d["choices"][0]["message"]["content"])
    if not code:
        raise RuntimeError("empty output")
    return code


def make_big(src, bigdir):
    name = os.path.basename(src)
    dst = os.path.join(bigdir, name)
    im = Image.open(src).convert("RGB")
    im = im.resize((im.width * 3, im.height * 3), Image.LANCZOS)
    im.save(dst)
    return dst


def label_one(big):
    votes = []
    for _ in range(2):
        for attempt in range(3):
            try:
                votes.append(call_vision(big))
                break
            except Exception:
                continue
    c = Counter(votes)
    if len(votes) >= 2 and c.most_common(1)[0][1] >= 2:
        label, n = c.most_common(1)[0]
        return {"label": label, "votes": votes, "count": n, "agreed": True}
    for attempt in range(3):
        try:
            votes.append(call_vision(big))
            break
        except Exception:
            continue
    if not votes:
        return {"label": None, "votes": [], "count": 0, "agreed": False, "error": "all reads failed"}
    c = Counter(votes)
    label, n = c.most_common(1)[0]
    return {"label": label, "votes": votes, "count": n, "agreed": n == len(votes)}


def main():
    d = sys.argv[1]
    out = sys.argv[2]
    workers = int(sys.argv[3]) if len(sys.argv) > 3 else 8
    with tempfile.TemporaryDirectory(prefix="sp_big_") as bigdir:
        files = sorted(glob.glob(os.path.join(d, "*.png")))
        print(f"images={len(files)} workers={workers}", flush=True)
        bigs = {f: make_big(f, bigdir) for f in files}
        results = {}
        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
            futs = {ex.submit(label_one, bigs[f]): f for f in files}
            done = 0
            for fut in concurrent.futures.as_completed(futs):
                f = futs[fut]
                try:
                    results[os.path.basename(f)] = fut.result()
                except Exception as e:
                    results[os.path.basename(f)] = {"label": None, "votes": [], "count": 0, "error": str(e)}
                done += 1
                if done % 10 == 0 or done == len(files):
                    print(f"progress {done}/{len(files)}", flush=True)
        with open(out, "w") as fh:
            json.dump(results, fh, ensure_ascii=False, indent=1)
        ok = sum(1 for v in results.values() if v.get("label"))
        agree = sum(1 for v in results.values() if v.get("agreed"))
        print(f"DONE labeled={ok}/{len(files)} full-agree={agree} -> {out}")


if __name__ == "__main__":
    main()