File size: 4,292 Bytes
6b0e0fc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | #!/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()
|