Upload scripts/label_captchas.py with huggingface_hub

scripts/label_captchas.py

@@ -1,3 +1,115 @@
-version https://git-lfs.github.com/spec/v1
-oid sha256:3386e054e01fc764c9d728195ca6d9b200094b393a987bedb35cdf9709e102f4
-size 4292
+#!/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()