#!/usr/bin/env python3
"""PHPWind 验证码手动标注 Web 服务。
显示验证码大图，输入数字回车保存并跳下一张。标签存 JSON。
用法: label_server.py <图片目录> <标签文件> [端口] [host]
"""
import glob
import io
import json
import os
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse
from PIL import Image
IMG_DIR = os.path.abspath(sys.argv[1])
LABEL_FILE = sys.argv[2]
PORT = int(sys.argv[3]) if len(sys.argv) > 3 else 8000
HOST = sys.argv[4] if len(sys.argv) > 4 else "0.0.0.0"
_upscale_cache = {}
def load_labels():
if os.path.exists(LABEL_FILE):
return json.load(open(LABEL_FILE, encoding="utf-8"))
return {}
def save_labels(labels):
json.dump(labels, open(LABEL_FILE, "w", encoding="utf-8"), ensure_ascii=False, indent=1)
def all_images():
return sorted(os.path.basename(f) for f in glob.glob(os.path.join(IMG_DIR, "*.png")))
def upscaled_bytes(name):
if name in _upscale_cache:
return _upscale_cache[name]
p = os.path.join(IMG_DIR, name)
if not os.path.exists(p):
return None
im = Image.open(p).convert("RGB")
im = im.resize((im.width * 3, im.height * 3), Image.LANCZOS)
buf = io.BytesIO()
im.save(buf, "PNG")
_upscale_cache[name] = buf.getvalue()
return _upscale_cache[name]
PAGE = """
PHPWind 验证码标注
PHPWind 验证码标注
加载中...
"""
class Handler(BaseHTTPRequestHandler):
labels = load_labels()
def _send(self, code, body, ctype):
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(body)
def _json(self, obj):
self._send(200, json.dumps(obj).encode(), "application/json")
def do_GET(self):
u = urlparse(self.path)
if u.path == "/":
self._send(200, PAGE.encode(), "text/html; charset=utf-8")
elif u.path == "/api/images":
total = all_images()
labeled = set(Handler.labels.keys())
self._json({"images": [n for n in total if n not in labeled], "total": len(total), "labeled": len(labeled)})
elif u.path == "/api/stats":
self._json({"labeled": len(Handler.labels), "total": len(all_images())})
elif u.path.startswith("/img/"):
name = os.path.basename(u.path[len("/img/"):])
data = upscaled_bytes(name)
if data is None:
self._send(404, b"not found", "text/plain")
else:
self._send(200, data, "image/png")
else:
self._send(404, b"not found", "text/plain")
def do_POST(self):
u = urlparse(self.path)
if u.path == "/api/label":
length = int(self.headers.get("Content-Length", 0))
try:
data = json.loads(self.rfile.read(length))
except Exception:
data = {}
name, label = data.get("name", ""), str(data.get("label", "")).strip()
if name and label:
Handler.labels[name] = {"label": label, "votes": [label], "count": 1, "agreed": True}
save_labels(Handler.labels)
self._json({"ok": True, "labeled": len(Handler.labels)})
else:
self._send(404, b"not found", "text/plain")
def log_message(self, fmt, *args):
pass
def main():
os.makedirs(IMG_DIR, exist_ok=True)
srv = ThreadingHTTPServer((HOST, PORT), Handler)
print(f"标注服务已启动: http://{HOST}:{PORT} 图片目录={IMG_DIR} 标签文件={LABEL_FILE}", flush=True)
print(f"待标图片数: {len(all_images())} 已标: {len(Handler.labels)}", flush=True)
srv.serve_forever()
if __name__ == "__main__":
main()