mirror of
https://github.com/OLmatter/glm-coding-helper.git
synced 2026-10-07 17:18:45 +08:00
551 lines
21 KiB
Python
551 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import base64
|
|
import time
|
|
import sys
|
|
import threading
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from pathlib import Path
|
|
import tkinter as tk
|
|
from tkinter import ttk
|
|
|
|
sys.setswitchinterval(0.001)
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPTS_DIR = ROOT / "scripts"
|
|
sys.path.append(str(SCRIPTS_DIR / "monitor"))
|
|
sys.path.insert(0, str(ROOT))
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
sys.path.insert(0, str(ROOT / "scripts/tools"))
|
|
|
|
from backend_config import (
|
|
add_backend_args,
|
|
apply_backend_config,
|
|
apply_cli_overrides,
|
|
print_config,
|
|
resolve_backend_config,
|
|
)
|
|
|
|
try:
|
|
from window_helper import capture_browser_window
|
|
from captcha_crop import crop_challenge_image
|
|
except ImportError:
|
|
print("Warning: monitor modules not found.")
|
|
capture_browser_window = None
|
|
crop_challenge_image = None
|
|
|
|
DATASET_DIR = ROOT / "dataset" / "auto_captured"
|
|
BACKEND_CONFIG = resolve_backend_config(source="server")
|
|
apply_backend_config(BACKEND_CONFIG)
|
|
HOST = BACKEND_CONFIG.host
|
|
PORT = BACKEND_CONFIG.port
|
|
|
|
class AppState:
|
|
def __init__(self):
|
|
self.status = "准备就绪"
|
|
self.last_prompt = "无"
|
|
self.last_save = "无"
|
|
self.log_messages = []
|
|
self.gui_update_needed = False
|
|
self.latest_request_ts = 0
|
|
self.recognition_results = {}
|
|
self.auto_click_enabled = False
|
|
self.rush_mode = False
|
|
self.rush_target_time = ""
|
|
self.cached_modal = None # {"img": PIL.Image, "crop_rect": tuple, "ts": float, "density": float}
|
|
|
|
state = AppState()
|
|
|
|
def log_to_gui(msg):
|
|
timestamp = datetime.now().strftime("%H:%M:%S")
|
|
formatted_msg = f"[{timestamp}] {msg}"
|
|
state.log_messages.append(formatted_msg)
|
|
if len(state.log_messages) > 20:
|
|
state.log_messages.pop(0)
|
|
state.gui_update_needed = True
|
|
|
|
def _is_colored_content(pixel):
|
|
r, g, b = pixel[:3]
|
|
if r > 250 and g > 250 and b > 250: return False
|
|
if max(r, g, b) - min(r, g, b) < 8: return False
|
|
return True
|
|
|
|
def get_color_density(img):
|
|
w, h = img.size
|
|
pixels = img.load()
|
|
hits = 0
|
|
step = 4
|
|
for y in range(0, h, step):
|
|
for x in range(0, w, step):
|
|
if _is_colored_content(pixels[x, y]):
|
|
hits += 1
|
|
return hits / ((w/step) * (h/step))
|
|
|
|
def is_white(rgb):
|
|
return rgb[0] > 242 and rgb[1] > 242 and rgb[2] > 242
|
|
|
|
def is_dark(rgb):
|
|
return rgb[0] < 35 and rgb[1] < 35 and rgb[2] < 35
|
|
|
|
def locate_modal(img):
|
|
w, h = img.size
|
|
cx, cy = w // 2, h // 2
|
|
pixels = img.load()
|
|
candidates = []
|
|
row_idx = 0
|
|
for y in range(int(h * 0.15), int(h * 0.85), 8):
|
|
if is_white(pixels[cx, y]) or is_dark(pixels[cx, y]):
|
|
l, r = cx, cx
|
|
check_fn = is_white if is_white(pixels[cx, y]) else is_dark
|
|
while l > 0 and check_fn(pixels[l, y]): l -= 1
|
|
while r < w - 1 and check_fn(pixels[r, y]): r += 1
|
|
if 280 < (r - l) < 700: candidates.append((l, r, y))
|
|
row_idx += 1
|
|
if row_idx % 10 == 0:
|
|
time.sleep(0.001)
|
|
if not candidates:
|
|
for y in range(int(h * 0.15), int(h * 0.85), 8):
|
|
if not is_white(pixels[cx, y]) and pixels[cx, y][0] < 100:
|
|
l, r = cx, cx
|
|
while l > 0 and pixels[l, y][0] < 100: l -= 1
|
|
while r < w - 1 and pixels[r, y][0] < 100: r += 1
|
|
if 280 < (r - l) < 700: candidates.append((l, r, y))
|
|
row_idx += 1
|
|
if row_idx % 10 == 0:
|
|
time.sleep(0.001)
|
|
if not candidates: return None
|
|
from collections import Counter
|
|
counts = Counter([(c[0], c[1]) for c in candidates])
|
|
if not counts: return None
|
|
(l, r), freq = counts.most_common(1)[0]
|
|
if freq < 3: return None
|
|
ys = [c[2] for c in candidates if (c[0], c[1]) == (l, r)]
|
|
return (l, min(ys) - int((r - l) * 0.08), r, max(ys) + int((r - l) * 0.08))
|
|
|
|
import subprocess
|
|
import queue
|
|
|
|
_worker_proc = None
|
|
_worker_lock = threading.Lock()
|
|
_result_queue = queue.Queue()
|
|
|
|
def _reader_thread(proc):
|
|
while True:
|
|
try:
|
|
line = proc.stdout.readline()
|
|
if not line:
|
|
break
|
|
if line.lstrip().startswith(b"{"):
|
|
_result_queue.put(line)
|
|
else:
|
|
try:
|
|
sys.stderr.write(line.decode("utf-8", errors="replace"))
|
|
except Exception:
|
|
pass
|
|
except:
|
|
break
|
|
|
|
def _get_worker():
|
|
global _worker_proc
|
|
with _worker_lock:
|
|
if _worker_proc is None or _worker_proc.poll() is not None:
|
|
worker_script = str(Path(__file__).parent / "captcha_worker.py")
|
|
log_to_gui("启动识别子进程...")
|
|
env = os.environ.copy()
|
|
env["PYTHONIOENCODING"] = "utf-8"
|
|
env["PYTHONUTF8"] = "1"
|
|
_worker_proc = subprocess.Popen(
|
|
[sys.executable, "-u", worker_script],
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
cwd=str(ROOT),
|
|
env=env,
|
|
)
|
|
while not _result_queue.empty():
|
|
_result_queue.get_nowait()
|
|
threading.Thread(target=_reader_thread, args=(_worker_proc,), daemon=True).start()
|
|
def _drain_stderr():
|
|
for line in _worker_proc.stderr:
|
|
try:
|
|
sys.stderr.write(line.decode(errors='replace'))
|
|
except:
|
|
pass
|
|
threading.Thread(target=_drain_stderr, daemon=True).start()
|
|
log_to_gui("识别子进程已启动")
|
|
return _worker_proc
|
|
|
|
def recognize_captcha(image_path, prompt_chars, crop_rect=None):
|
|
try:
|
|
proc = _get_worker()
|
|
req = json.dumps({
|
|
"image_path": str(image_path),
|
|
"chars": list(prompt_chars),
|
|
"crop_rect": list(crop_rect) if crop_rect else None,
|
|
}, ensure_ascii=False) + "\n"
|
|
print(f"[CAPTURE] sending to worker: {image_path} chars={''.join(prompt_chars)}", flush=True)
|
|
proc.stdin.write(req.encode("utf-8"))
|
|
proc.stdin.flush()
|
|
print(f"[CAPTURE] waiting for worker result...", flush=True)
|
|
try:
|
|
result_line = _result_queue.get(timeout=15)
|
|
except queue.Empty:
|
|
print(f"[CAPTURE] worker timeout!", flush=True)
|
|
global _worker_proc
|
|
_worker_proc = None
|
|
return {"error": "worker timeout", "success": False}
|
|
print(f"[CAPTURE] got result: {result_line.decode()[:200]}", flush=True)
|
|
return json.loads(result_line)
|
|
except Exception as e:
|
|
import traceback; traceback.print_exc()
|
|
return {"error": str(e), "success": False}
|
|
|
|
def trigger_auto_capture(chars_text: str, request_ts: int):
|
|
state.latest_request_ts = request_ts
|
|
my_id = request_ts
|
|
t0 = time.time()
|
|
print(f"[CAPTURE] === start: {chars_text} ts={request_ts} ===", flush=True)
|
|
|
|
try:
|
|
state.status = "截图中..."
|
|
state.last_prompt = chars_text
|
|
log_to_gui(f"收到请求: {chars_text}")
|
|
|
|
DATASET_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
best_img = None
|
|
best_crop_rect = None
|
|
|
|
for attempt in range(5):
|
|
if state.latest_request_ts != my_id:
|
|
return
|
|
if not capture_browser_window:
|
|
time.sleep(0.08)
|
|
continue
|
|
screen, rect = capture_browser_window(["Chrome", "Google Chrome", "bigmodel"])
|
|
time.sleep(0.001)
|
|
if not screen:
|
|
continue
|
|
modal_rect = locate_modal(screen)
|
|
time.sleep(0.001)
|
|
if not modal_rect:
|
|
if attempt < 2:
|
|
time.sleep(0.05)
|
|
continue
|
|
else:
|
|
print(f"[CAPTURE] no modal after {attempt+1} attempts, giving up", flush=True)
|
|
state.status = "无弹窗"
|
|
log_to_gui("ERR: 弹窗未出现或已关闭")
|
|
state.recognition_results[str(request_ts)] = {
|
|
"result": {"success": False, "error": "弹窗未出现或已关闭"},
|
|
"timestamp": datetime.now().isoformat(),
|
|
}
|
|
return
|
|
|
|
modal_img = screen.crop(modal_rect)
|
|
image_only, crop_rect = crop_challenge_image(modal_img)
|
|
time.sleep(0.001)
|
|
if image_only and image_only.size[1] >= 250:
|
|
best_img = image_only
|
|
best_crop_rect = crop_rect
|
|
print(f"[CAPTURE] captured on attempt #{attempt+1} in {(time.time()-t0)*1000:.0f}ms", flush=True)
|
|
break
|
|
|
|
if not best_img:
|
|
state.status = "超时退出"
|
|
log_to_gui("ERR: 未截得合格图片")
|
|
state.recognition_results[str(request_ts)] = {
|
|
"result": {"success": False, "error": "未截得合格图片"},
|
|
"timestamp": datetime.now().isoformat(),
|
|
}
|
|
return
|
|
|
|
t1 = time.time()
|
|
|
|
timestamp_str = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
filename = f"{chars_text}_{timestamp_str}.png"
|
|
save_path = DATASET_DIR / filename
|
|
best_img.save(save_path)
|
|
|
|
log_to_gui(f"截图保存: {filename}")
|
|
print(f"[CAPTURE] saved in {(time.time()-t1)*1000:.0f}ms, calling recognize...", flush=True)
|
|
|
|
log_to_gui("开始识别...")
|
|
result = recognize_captcha(str(save_path), list(chars_text), crop_rect=best_crop_rect)
|
|
t2 = time.time()
|
|
print(f"[CAPTURE] total={(t2-t0)*1000:.0f}ms recog={result.get('success')}", flush=True)
|
|
|
|
if result.get("success"):
|
|
coords = result["click_coords"]
|
|
|
|
state.status = f"识别完成: {result['pred_text']} ({result['confidence']})"
|
|
state.last_save = filename
|
|
state.recognition_results[str(request_ts)] = {
|
|
"result": result,
|
|
"image_path": str(save_path),
|
|
"timestamp": datetime.now().isoformat(),
|
|
}
|
|
|
|
log_to_gui(f"识别成功: {result['pred_text']} 置信度:{result['confidence']} 耗时:{(t2-t0)*1000:.0f}ms")
|
|
|
|
else:
|
|
state.status = f"识别失败: {result.get('error', '未知')}"
|
|
log_to_gui(f"识别结果: {json.dumps(result, ensure_ascii=False)}")
|
|
|
|
except Exception as e:
|
|
state.status = "系统出错"
|
|
log_to_gui(f"FATAL: {e}")
|
|
finally:
|
|
state.gui_update_needed = True
|
|
|
|
class CaptchaHandler(BaseHTTPRequestHandler):
|
|
def send_cors_headers(self):
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
|
|
self.send_header("Access-Control-Allow-Headers", "*")
|
|
self.send_header("Access-Control-Allow-Private-Network", "true")
|
|
|
|
def send_json(self, status, payload):
|
|
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_cors_headers()
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_OPTIONS(self):
|
|
self.send_response(200)
|
|
self.send_cors_headers()
|
|
self.end_headers()
|
|
|
|
def do_POST(self):
|
|
path = self.path.rstrip("/")
|
|
try:
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
data = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
|
|
|
|
if path == "/captcha":
|
|
text = str(data.get("text", "")).strip()
|
|
request_ts = int(data.get("ts", time.time() * 1000))
|
|
chars = re.findall(r"[\u4e00-\u9fff]", text or "")
|
|
chars_text = "".join(chars[:3])
|
|
|
|
if chars_text:
|
|
threading.Thread(target=trigger_auto_capture, args=(chars_text, request_ts), daemon=True).start()
|
|
|
|
self.send_json(200, {"status": "processing", "chars": chars_text, "ts": request_ts})
|
|
|
|
elif path == "/captcha_direct":
|
|
text = str(data.get("text", "")).strip()
|
|
img_b64 = data.get("image", "")
|
|
chars = re.findall(r"[\u4e00-\u9fff]", text or "")
|
|
chars_text = "".join(chars[:3])
|
|
|
|
if not chars_text or not img_b64:
|
|
self.send_json(400, {"error": "missing text or image", "success": False})
|
|
return
|
|
|
|
try:
|
|
import io
|
|
from PIL import Image
|
|
img_bytes = base64.b64decode(img_b64.split(",")[-1])
|
|
image = Image.open(io.BytesIO(img_bytes)).convert("RGB")
|
|
log_to_gui(f"[direct] 收到图片: {image.size}, 识别: {chars_text}")
|
|
|
|
DEBUG_DIR = ROOT / "dataset" / "debug_captcha_direct"
|
|
DEBUG_DIR.mkdir(parents=True, exist_ok=True)
|
|
ts_str = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
|
|
debug_path = DEBUG_DIR / f"{chars_text}_{ts_str}.png"
|
|
image.save(debug_path)
|
|
log_to_gui(f"[direct] 调试保存: {debug_path.name} ({len(img_bytes)//1024}KB)")
|
|
|
|
result = recognize_captcha(str(debug_path), list(chars_text), crop_rect=None)
|
|
|
|
if result.get("success"):
|
|
log_to_gui(f"[direct] 识别成功: {result['pred_text']} conf={result['confidence']}")
|
|
state.status = f"识别完成: {result['pred_text']} ({result['confidence']})"
|
|
else:
|
|
log_to_gui(f"[direct] 识别失败: {result.get('error', '?')}")
|
|
state.status = f"识别失败: {result.get('error', '?')}"
|
|
|
|
self.send_json(200, {"success": True, "result": result})
|
|
except Exception as e:
|
|
log_to_gui(f"[direct] 异常: {e}")
|
|
self.send_json(500, {"error": str(e), "success": False})
|
|
|
|
elif path == "/result":
|
|
ts = data.get("ts", "")
|
|
result = state.recognition_results.get(str(ts), {})
|
|
self.send_json(200, {"has_result": bool(result), "result": result})
|
|
|
|
elif path == "/config":
|
|
action = data.get("action")
|
|
if action == "get":
|
|
self.send_json(200, {
|
|
"auto_click": state.auto_click_enabled,
|
|
"rush_mode": state.rush_mode,
|
|
"rush_target": state.rush_target_time,
|
|
"results_count": len(state.recognition_results),
|
|
})
|
|
elif action == "set_auto_click":
|
|
state.auto_click_enabled = data.get("value", True)
|
|
self.send_json(200, {"auto_click": state.auto_click_enabled})
|
|
elif action == "set_rush_mode":
|
|
state.rush_mode = data.get("value", False)
|
|
state.rush_target_time = data.get("target_time", "")
|
|
self.send_json(200, {"rush_mode": state.rush_mode, "target": state.rush_target_time})
|
|
else:
|
|
self.send_json(400, {"error": "unknown action"})
|
|
else:
|
|
self.send_json(404, {"status": "not_found"})
|
|
|
|
except Exception as e:
|
|
self.send_json(400, {"error": str(e)})
|
|
|
|
def do_GET(self):
|
|
path = self.path.rstrip("/")
|
|
if path == "/results":
|
|
results = {}
|
|
for k, v in state.recognition_results.items():
|
|
results[k] = {
|
|
"pred_text": v.get("result", {}).get("pred_text", "?"),
|
|
"confidence": v.get("result", {}).get("confidence", 0),
|
|
"timestamp": v.get("timestamp", ""),
|
|
"coords": v.get("abs_coords", []),
|
|
}
|
|
self.send_json(200, {"count": len(results), "results": results})
|
|
elif path == "/health":
|
|
self.send_json(
|
|
200,
|
|
{
|
|
"status": "ok",
|
|
"queue": len(state.recognition_results),
|
|
"backend": BACKEND_CONFIG.to_dict(),
|
|
},
|
|
)
|
|
else:
|
|
self.send_json(404, {})
|
|
|
|
def log_message(self, *args): pass
|
|
|
|
def run_gui():
|
|
root = tk.Tk()
|
|
root.title("Captcha Server v2 - Auto Rush Mode")
|
|
root.geometry("480x420")
|
|
root.attributes("-topmost", True)
|
|
root.configure(bg="#f0f2f5")
|
|
|
|
style = ttk.Style()
|
|
style.configure("TLabel", background="#f0f2f5", font=("Microsoft YaHei UI", 10))
|
|
style.configure("Status.TLabel", font=("Microsoft YaHei UI", 12, "bold"), foreground="#1890ff")
|
|
style.configure("Success.TLabel", font=("Microsoft YaHei UI", 11, "bold"), foreground="#52c41a")
|
|
|
|
main_frame = ttk.Frame(root, padding="15")
|
|
main_frame.pack(fill=tk.BOTH, expand=True)
|
|
|
|
row = 0
|
|
ttk.Label(main_frame, text="系统状态:").grid(row=row, column=0, sticky=tk.W, pady=2)
|
|
lbl_status = ttk.Label(main_frame, text="准备就绪", style="Status.TLabel")
|
|
lbl_status.grid(row=row, column=1, sticky=tk.W, pady=2); row += 1
|
|
|
|
ttk.Label(main_frame, text="当前提示:").grid(row=row, column=0, sticky=tk.W, pady=2)
|
|
lbl_prompt = ttk.Label(main_frame, text="无")
|
|
lbl_prompt.grid(row=row, column=1, sticky=tk.W, pady=2); row += 1
|
|
|
|
ttk.Label(main_frame, text="识别结果:").grid(row=row, column=0, sticky=tk.W, pady=2)
|
|
lbl_result = ttk.Label(main_frame, text="--", style="Success.TLabel")
|
|
lbl_result.grid(row=row, column=1, sticky=tk.W, pady=2); row += 1
|
|
|
|
ttk.Label(main_frame, text="最后截图:").grid(row=row, column=0, sticky=tk.W, pady=2)
|
|
lbl_save = ttk.Label(main_frame, text="无", wraplength=320)
|
|
lbl_save.grid(row=row, column=1, sticky=tk.W, pady=2); row += 1
|
|
|
|
var_auto_click = tk.BooleanVar(value=False)
|
|
cb_auto = ttk.Checkbutton(main_frame, text="自动点击验证码 (pyautogui)", variable=var_auto_click)
|
|
cb_auto.grid(row=row, column=0, columnspan=2, sticky=tk.W, pady=5); row += 1
|
|
|
|
var_rush = tk.BooleanVar(value=False)
|
|
cb_rush = ttk.Checkbutton(main_frame, text="抢购模式 (10点卡点)", variable=var_rush)
|
|
cb_rush.grid(row=row, column=0, columnspan=2, sticky=tk.W, pady=2); row += 1
|
|
|
|
log_box = tk.Text(main_frame, height=12, width=58, font=("Consolas", 9), bg="#ffffff", relief=tk.FLAT)
|
|
log_box.grid(row=row, column=0, columnspan=2, pady=8); row += 1
|
|
|
|
def update_gui():
|
|
try:
|
|
if state.gui_update_needed:
|
|
lbl_status.config(text=state.status)
|
|
lbl_prompt.config(text=state.last_prompt)
|
|
lbl_save.config(text=state.last_save)
|
|
res = state.recognition_results
|
|
latest = max(res.keys(), key=lambda k: res[k]["timestamp"], default=None) if res else None
|
|
if latest:
|
|
r = res[latest]["result"]
|
|
lbl_result.config(text=f"{r.get('pred_text','?')} (conf={r.get('confidence',0):.0%})")
|
|
else:
|
|
lbl_result.config(text="--")
|
|
log_box.delete('1.0', tk.END)
|
|
log_box.insert(tk.END, "\n".join(state.log_messages))
|
|
log_box.see(tk.END)
|
|
state.gui_update_needed = False
|
|
except Exception:
|
|
pass
|
|
root.after(200, update_gui)
|
|
|
|
def toggle_auto_click(*args):
|
|
state.auto_click_enabled = var_auto_click.get()
|
|
log_to_gui(f"自动点击: {'开启' if state.auto_click_enabled else '关闭'}")
|
|
|
|
def toggle_rush(*args):
|
|
state.rush_mode = var_rush.get()
|
|
log_to_gui(f"抢购模式: {'开启' if state.rush_mode else '关闭'}")
|
|
|
|
var_auto_click.trace_add("write", toggle_auto_click)
|
|
var_rush.trace_add("write", toggle_rush)
|
|
|
|
def start_server():
|
|
try:
|
|
server = HTTPServer((HOST, PORT), CaptchaHandler)
|
|
server.serve_forever()
|
|
except Exception as e:
|
|
log_to_gui(f"Server Error: {e}")
|
|
|
|
threading.Thread(target=start_server, daemon=True).start()
|
|
|
|
def preload_worker():
|
|
try:
|
|
log_to_gui("预热识别子进程...")
|
|
_get_worker()
|
|
log_to_gui("识别子进程就绪！")
|
|
except Exception as e:
|
|
log_to_gui(f"子进程启动失败: {e}")
|
|
|
|
threading.Thread(target=preload_worker, daemon=True).start()
|
|
|
|
root.after(100, update_gui)
|
|
root.mainloop()
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
global BACKEND_CONFIG, HOST, PORT
|
|
parser = argparse.ArgumentParser(description="Start the CNCAPTCHA GUI backend.")
|
|
add_backend_args(parser)
|
|
args = parser.parse_args(argv)
|
|
apply_cli_overrides(args)
|
|
BACKEND_CONFIG = resolve_backend_config(source="server-cli")
|
|
apply_backend_config(BACKEND_CONFIG)
|
|
HOST = BACKEND_CONFIG.host
|
|
PORT = BACKEND_CONFIG.port
|
|
print_config(BACKEND_CONFIG)
|
|
run_gui()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|