scripts/label_server.py

#!/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 = """<!DOCTYPE html>
<html lang="zh">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=2.0">
<title>PHPWind 验证码标注</title>
<style>
  *{box-sizing:border-box}
  body{font-family:-apple-system,Segoe UI,sans-serif;background:#0f1115;color:#eee;margin:0;padding:10px 8px 20px;text-align:center;-webkit-tap-highlight-color:transparent;overflow-y:auto}
  h1{font-size:17px;margin:0}
  .progress{color:#7a8698;font-size:13px;margin:3px 0}
  .img-area{display:flex;align-items:center;justify-content:center;margin:4px 0}
  .img-area img{display:block;max-width:96vw;max-height:42vh;background:#fff;padding:5px;border-radius:10px;box-shadow:0 2px 10px rgba(0,0,0,.5);object-fit:contain}
  .name{color:#4a5260;font-size:11px;margin:2px 0}
  input{font-size:30px;letter-spacing:10px;width:88%;max-width:280px;text-align:center;background:#1a1d24;border:2px solid #3a3f4a;border-radius:10px;color:#fff;padding:16px 6px;margin:10px 0;-webkit-appearance:none}
  input:focus{outline:none;border-color:#4f8cff;box-shadow:0 0 0 3px rgba(79,140,255,.3)}
  .btns{display:flex;justify-content:center;gap:12px;margin:8px 0}
  button{font-size:16px;padding:12px 24px;background:#2a2f3a;color:#eee;border:none;border-radius:8px;cursor:pointer;min-width:76px}
  button:active{background:#4f8cff;color:#fff;transform:scale(.96)}
  .hint{color:#4a5260;font-size:11px}
  .history{max-height:25vh;overflow-y:auto;text-align:left;margin:8px auto;padding:6px;background:#191c24;border-radius:8px;width:92%;max-width:360px}
  .history .item{font-size:11px;color:#8892a0;padding:2px 4px;border-bottom:1px solid #222;display:flex;gap:8px}
  .history .item span{color:#4f8cff;font-weight:600;font-size:13px}
  .history .empty{color:#4a5260;font-size:11px;text-align:center;padding:6px}
  @media (max-width:480px){
    input{font-size:26px;padding:14px 4px;letter-spacing:6px}
    button{padding:12px 18px;font-size:15px}
    .img-area img{max-height:38vh}
  }
</style>
</head>
<body>
<h1>PHPWind 验证码标注</h1>
<div class="progress" id="prog">加载中...</div>
<div class="img-area"><img id="capimg" src="" alt="captcha"></div>
<div class="name" id="name"></div>
<input id="inp" maxlength="4" autocomplete="off" placeholder="数字" inputmode="numeric" autocapitalize="off">
<div class="btns">
  <button onclick="skip()">跳过</button>
  <button onclick="back()">上一张</button>
</div>
<div class="hint" id="hint"></div>
<div class="history" id="history"><div class="empty">还没有标注记录</div></div>
<script>
let images=[], idx=0, labeledSet=new Set();
let history=[];  // [{name, code}]
async function load(){
  const d=await(await fetch('/api/images')).json();
  images=d.images; idx=0;
  document.getElementById('prog').textContent='已标 '+d.labeled+' / '+d.total;
  show();
}
function show(){
  if(idx>=images.length){
    document.body.innerHTML='<h2 style="padding:40px">全部完成!</h2>';
    return;
  }
  const name=images[idx];
  document.getElementById('capimg').src='/img/'+name;
  document.getElementById('name').textContent=name+(labeledSet.has(name)?' (已标)':'');
  document.getElementById('hint').textContent=idx+' / '+(labeledSet.size+images.length-idx);
  const inp=document.getElementById('inp'); inp.value=''; inp.focus();
  window.scrollTo(0,0);
  renderHistory();
}
function renderHistory(){
  const h=document.getElementById('history');
  if(history.length===0){h.innerHTML='<div class="empty">还没有标注记录</div>'; return;}
  h.innerHTML=history.slice(-20).reverse().map(r=>'<div class="item"><span>&#10003;</span> '+r.name+' &rarr; <span>'+r.code+'</span></div>').join('');
}
async function submit(){
  const name=images[idx]; const v=document.getElementById('inp').value.trim();
  if(!v) return;
  await fetch('/api/label',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({name,label:v})});
  labeledSet.add(name); history.push({name,code:v});
  idx++; refresh();
}
function skip(){ if(idx<images.length){idx++; refresh();} }
function back(){
  if(idx>0){ idx--;
    // 回退时从 history 移除最后一条（如果是刚提交的）
    if(history.length>0 && history[history.length-1].name===images[idx]) history.pop();
    show();
  }
}
async function refresh(){
  const d=await(await fetch('/api/stats')).json();
  document.getElementById('prog').textContent='已标 '+d.labeled+' / '+d.total;
  show();
}
document.getElementById('inp').addEventListener('keydown',e=>{ if(e.key==='Enter'){e.preventDefault();submit();} });
document.addEventListener('keydown',e=>{
  const v=document.getElementById('inp').value;
  if(e.key==='s'&&v==='') skip();
  if(e.key==='b'&&v==='') back();
});
load();
</script>
</body></html>
"""


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()