diff --git a/.gitignore b/.gitignore
index f3183b4..f76dbe6 100644
--- a/.gitignore
+++ b/.gitignore
@@ -22,3 +22,5 @@ __pycache__/
tcaptcha/data/raw/
tcaptcha/data/crops/
data/
+runs/
+yolov8n.pt
diff --git a/pyproject.toml b/pyproject.toml
index 6aa71ce..97ea410 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -3,7 +3,10 @@ name = "hello-js-reverse-skill"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
+ "ddddocr>=1.6.1",
"httpx>=0.28.1",
"numpy>=2.5.3",
+ "opencv-python-headless>=5.0.0.93",
"pillow>=12.3.0",
+ "ultralytics>=8.4.147",
]
diff --git a/tcaptcha/evaluate.py b/tcaptcha/evaluate.py
new file mode 100644
index 0000000..03598be
--- /dev/null
+++ b/tcaptcha/evaluate.py
@@ -0,0 +1,147 @@
+"""
+腾讯天御点选验证码求解器端到端评估脚本
+评估指标:
+1. 单点命中率 (Hit Rate per Point, distance <= 30px)
+2. 无序3点命中率 (All 3 icons detected regardless of order)
+3. 严格时序三点命中率 (Full 3-point Sequence Accuracy: P1->T1, P2->T2, P3->T3)
+"""
+
+import json
+import logging
+import sys
+from pathlib import Path
+
+# 确保项目根目录在 sys.path 中
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+if str(PROJECT_ROOT) not in sys.path:
+ sys.path.insert(0, str(PROJECT_ROOT))
+
+import numpy as np
+from tcaptcha.solver import CaptchaSolver
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger("evaluate")
+
+
+def evaluate_solver(labels_path: Path = Path("tcaptcha/data/labels.jsonl"), tolerance_px: float = 35.0):
+ if not labels_path.exists():
+ logger.error(f"Labels file not found: {labels_path}")
+ return
+
+ solver = CaptchaSolver()
+
+ samples = []
+ with open(labels_path, "r", encoding="utf-8") as f:
+ for line in f:
+ if line.strip():
+ samples.append(json.loads(line.strip()))
+
+ total_samples = len(samples)
+ total_points = total_samples * 3
+
+ exact_seq_success = 0
+ any_order_success = 0
+ total_points_hit = 0
+ point_errors = [] # distances for hits
+
+ failed_samples = []
+
+ print(f"\n================ 开始评估 (共 {total_samples} 个样本) ================\n")
+
+ for idx, s in enumerate(samples, 1):
+ sid = s["sample_id"]
+ gt_points = [(p["x"], p["y"]) for p in s["points"]]
+ bg = Path(f"tcaptcha/data/raw/{sid}_bg.png")
+ prompt = Path(f"tcaptcha/data/raw/{sid}_prompt.png")
+
+ if not bg.exists() or not prompt.exists():
+ logger.warning(f"[{idx}/{total_samples}] {sid} 缺失图像文件")
+ continue
+
+ pred_points = solver.solve(bg, prompt)
+ if len(pred_points) < 3:
+ logger.warning(f"[{idx}/{total_samples}] {sid} 预测点数不足3个: {pred_points}")
+ failed_samples.append((sid, "insufficient_points", gt_points, pred_points))
+ continue
+
+ # 1. 严格顺序匹配评估
+ seq_hit = True
+ sample_point_hits = 0
+ distances = []
+ for i in range(3):
+ gx, gy = gt_points[i]
+ px, py = pred_points[i]
+ dist = np.hypot(px - gx, py - gy)
+ distances.append(dist)
+ if dist <= tolerance_px:
+ sample_point_hits += 1
+ point_errors.append(dist)
+ else:
+ seq_hit = False
+
+ total_points_hit += sample_point_hits
+
+ # 2. 无序匹配评估 (检查是否3个图标都找出来了,哪怕顺序搞错)
+ # 为每个预测点找最近的GT点
+ used_gt = set()
+ for px, py in pred_points:
+ best_gt = None
+ min_d = 9999
+ for gi, (gx, gy) in enumerate(gt_points):
+ if gi in used_gt:
+ continue
+ d = np.hypot(px - gx, py - gy)
+ if d < min_d:
+ min_d = d
+ best_gt = gi
+ if best_gt is not None and min_d <= tolerance_px:
+ used_gt.add(best_gt)
+
+ all_detected = (len(used_gt) == 3)
+ if all_detected:
+ any_order_success += 1
+
+ if seq_hit:
+ exact_seq_success += 1
+ status = "✅ PASS"
+ else:
+ status = "❌ FAIL"
+ failed_samples.append((sid, f"dist={[round(d,1) for d in distances]}", gt_points, pred_points))
+
+ print(f"[{idx:02d}/{total_samples}] {sid}: {status} | 点命中: {sample_point_hits}/3 | 距离: {[round(d, 1) for d in distances]}")
+
+ # 统计总结
+ seq_acc = exact_seq_success / total_samples * 100
+ order_agnostic_acc = any_order_success / total_samples * 100
+ pt_hit_rate = total_points_hit / total_points * 100
+ avg_err = np.mean(point_errors) if point_errors else 0.0
+
+ print("\n" + "=" * 60)
+ print("🎯 评估结果报告 (Evaluation Report):")
+ print(f" - 样本总数 (Total Samples): {total_samples}")
+ print(f" - 判定容差半径 (Tolerance): <= {tolerance_px} px")
+ print(f" - 单点命中率 (Point Hit Rate): {total_points_hit}/{total_points} ({pt_hit_rate:.2f}%)")
+ print(f" - 无序召回率 (Order-agnostic 3-Point Hit Rate): {any_order_success}/{total_samples} ({order_agnostic_acc:.2f}%)")
+ print(f" - 🌟 严格时序三点识别率 (Sequence Accuracy): {exact_seq_success}/{total_samples} ({seq_acc:.2f}%)")
+ print(f" - 命中点平均中心偏差 (Avg Center Offset): {avg_err:.2f} px")
+ print("=" * 60)
+
+ if failed_samples:
+ print(f"\n❌ 失败样本分析 (共 {len(failed_samples)} 个):")
+ for sid, reason, gt, pred in failed_samples[:10]:
+ print(f" - {sid}: {reason} | GT={gt} vs Pred={pred}")
+ if len(failed_samples) > 10:
+ print(f" ... 还有 {len(failed_samples) - 10} 个失败样本未列出")
+
+ return {
+ "seq_acc": seq_acc,
+ "order_agnostic_acc": order_agnostic_acc,
+ "pt_hit_rate": pt_hit_rate,
+ "failed_count": len(failed_samples)
+ }
+
+
+if __name__ == "__main__":
+ import sys
+ tol = float(sys.argv[1]) if len(sys.argv) > 1 else 35.0
+ evaluate_solver(tolerance_px=tol)
diff --git a/tcaptcha/models/icon_classifier.pt b/tcaptcha/models/icon_classifier.pt
new file mode 100644
index 0000000..3545b26
Binary files /dev/null and b/tcaptcha/models/icon_classifier.pt differ
diff --git a/tcaptcha/models/siamese_matcher.pt b/tcaptcha/models/siamese_matcher.pt
new file mode 100644
index 0000000..49f2c76
Binary files /dev/null and b/tcaptcha/models/siamese_matcher.pt differ
diff --git a/tcaptcha/models/yolo_detector.pt b/tcaptcha/models/yolo_detector.pt
new file mode 100644
index 0000000..e27ef9c
Binary files /dev/null and b/tcaptcha/models/yolo_detector.pt differ
diff --git a/tcaptcha/prepare_yolo.py b/tcaptcha/prepare_yolo.py
new file mode 100644
index 0000000..8b731ad
--- /dev/null
+++ b/tcaptcha/prepare_yolo.py
@@ -0,0 +1,144 @@
+"""
+将标注的 55 组样本转换为标准 YOLO 目标检测数据集
+生成:
+ tcaptcha/data/yolo/images/train/
+ tcaptcha/data/yolo/images/val/
+ tcaptcha/data/yolo/labels/train/
+ tcaptcha/data/yolo/labels/val/
+ tcaptcha/data/yolo/dataset.yaml
+"""
+
+import json
+import logging
+import random
+import shutil
+from pathlib import Path
+import cv2
+import numpy as np
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger("prepare_yolo")
+
+RAW_DIR = Path("tcaptcha/data/raw")
+LABEL_FILE = Path("tcaptcha/data/labels.jsonl")
+YOLO_DIR = Path("tcaptcha/data/yolo")
+
+IMG_W = 672
+IMG_H = 480
+BOX_SIZE = 54 # 图标大致大小 (约 54x54 像素)
+
+
+def find_all_icons_in_bg(bg_path: Path, labeled_points: list) -> list:
+ """
+ 结合人工标注点与轮廓补充检测,找到大图中所有图标(包含 3 个目标 + 2~3 个干扰图标)
+ 返回: [(cx, cy, w, h), ...] 绝对像素坐标
+ """
+ img = cv2.imread(str(bg_path))
+ if img is None:
+ return []
+ gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
+
+ # 提取暗色轮廓找干扰图标
+ edges = cv2.Canny(gray, 40, 120)
+ kernel = cv2.getStructuringElement(cv2.MORPH_RECT, (9, 9))
+ closed = cv2.morphologyEx(edges, cv2.MORPH_CLOSE, kernel)
+ contours, _ = cv2.findContours(closed, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+
+ all_boxes = []
+ # 1. 先加入人工标注的 3 个目标(绝对精准)
+ for p in labeled_points:
+ cx, cy = p["x"], p["y"]
+ all_boxes.append((cx, cy, BOX_SIZE, BOX_SIZE))
+
+ # 2. 补充检测未标注的干扰图标
+ for cnt in contours:
+ x, y, w, h = cv2.boundingRect(cnt)
+ if 26 <= w <= 85 and 26 <= h <= 85 and (w * h) >= 650:
+ cx, cy = x + w // 2, y + h // 2
+ # 必须远离已有图标
+ if all(np.hypot(cx - lx, cy - ly) > 42 for lx, ly, _, _ in all_boxes):
+ # 限制尺寸
+ bw = min(68, max(46, w + 8))
+ bh = min(68, max(46, h + 8))
+ all_boxes.append((cx, cy, bw, bh))
+
+ return all_boxes
+
+
+def prepare_dataset(val_ratio: float = 0.2, seed: int = 42):
+ random.seed(seed)
+ if not LABEL_FILE.exists():
+ logger.error("标签文件不存在: %s", LABEL_FILE)
+ return
+
+ # 清理并创建目录
+ if YOLO_DIR.exists():
+ shutil.rmtree(YOLO_DIR)
+
+ for split in ["train", "val"]:
+ (YOLO_DIR / "images" / split).mkdir(parents=True, exist_ok=True)
+ (YOLO_DIR / "labels" / split).mkdir(parents=True, exist_ok=True)
+
+ with open(LABEL_FILE, "r", encoding="utf-8") as f:
+ records = [json.loads(line) for line in f if line.strip()]
+
+ random.shuffle(records)
+ val_count = max(5, int(len(records) * val_ratio))
+ train_records = records[val_count:]
+ val_records = records[:val_count]
+
+ logger.info("总标注样本: %d (训练集: %d, 验证集: %d)", len(records), len(train_records), len(val_records))
+
+ def process_split(record_list, split_name):
+ total_boxes = 0
+ for r in record_list:
+ sid = r["sample_id"]
+ src_img = RAW_DIR / f"{sid}_bg.png"
+ if not src_img.exists():
+ continue
+
+ dst_img = YOLO_DIR / "images" / split_name / f"{sid}.png"
+ shutil.copy(src_img, dst_img)
+
+ # 提取全部图标坐标
+ boxes = find_all_icons_in_bg(src_img, r["points"])
+ total_boxes += len(boxes)
+
+ # 写入 YOLO 格式标注: class_id x_center y_center width height (归一化到 0~1)
+ dst_label = YOLO_DIR / "labels" / split_name / f"{sid}.txt"
+ lines = []
+ for cx, cy, bw, bh in boxes:
+ # 边界截断保护
+ x1 = max(0, cx - bw / 2)
+ y1 = max(0, cy - bh / 2)
+ x2 = min(IMG_W, cx + bw / 2)
+ y2 = min(IMG_H, cy + bh / 2)
+
+ norm_cx = (x1 + x2) / 2.0 / IMG_W
+ norm_cy = (y1 + y2) / 2.0 / IMG_H
+ norm_w = (x2 - x1) / IMG_W
+ norm_h = (y2 - y1) / IMG_H
+
+ lines.append(f"0 {norm_cx:.6f} {norm_cy:.6f} {norm_w:.6f} {norm_h:.6f}")
+
+ dst_label.write_text("\n".join(lines), encoding="utf-8")
+ logger.info(" [%s] 处理完成: %d 张图片, 共 %d 个图标标注", split_name, len(record_list), total_boxes)
+
+ process_split(train_records, "train")
+ process_split(val_records, "val")
+
+ # 生成 dataset.yaml
+ yaml_content = f"""path: {YOLO_DIR.resolve()}
+train: images/train
+val: images/val
+
+names:
+ 0: icon
+"""
+ yaml_path = YOLO_DIR / "dataset.yaml"
+ yaml_path.write_text(yaml_content, encoding="utf-8")
+ logger.info("✓ YOLO 数据集生成完毕: %s", yaml_path)
+
+
+if __name__ == "__main__":
+ prepare_dataset()
diff --git a/tcaptcha/siamese.py b/tcaptcha/siamese.py
new file mode 100644
index 0000000..20337ff
--- /dev/null
+++ b/tcaptcha/siamese.py
@@ -0,0 +1,228 @@
+"""
+图标孪生神经网络 (Icon Siamese Network)
+用于计算提示图图标与背景候选图标之间的形状相似度(抗背景噪点、抗旋转)
+"""
+
+import json
+import logging
+import random
+from pathlib import Path
+from typing import List, Tuple
+
+import numpy as np
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+import torchvision.transforms as transforms
+from PIL import Image
+from torch.utils.data import DataLoader, Dataset
+
+import sys
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+if str(PROJECT_ROOT) not in sys.path:
+ sys.path.insert(0, str(PROJECT_ROOT))
+
+from tcaptcha.slicer import slice_prompt_image
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger("siamese")
+
+MODEL_PATH = PROJECT_ROOT / "tcaptcha/models/siamese_matcher.pt"
+LABEL_FILE = PROJECT_ROOT / "tcaptcha/data/labels.jsonl"
+RAW_DIR = PROJECT_ROOT / "tcaptcha/data/raw"
+CROPS_DIR = PROJECT_ROOT / "tcaptcha/data/crops"
+
+
+class IconSiameseNet(nn.Module):
+ """轻量级 CNN 孪生特征提取网络"""
+
+ def __init__(self, emb_dim: int = 128):
+ super().__init__()
+ self.encoder = nn.Sequential(
+ nn.Conv2d(3, 32, kernel_size=3, padding=1),
+ nn.BatchNorm2d(32),
+ nn.LeakyReLU(0.1, inplace=True),
+ nn.MaxPool2d(2, 2), # 32x32
+ nn.Conv2d(32, 64, kernel_size=3, padding=1),
+ nn.BatchNorm2d(64),
+ nn.LeakyReLU(0.1, inplace=True),
+ nn.MaxPool2d(2, 2), # 16x16
+ nn.Conv2d(64, 128, kernel_size=3, padding=1),
+ nn.BatchNorm2d(128),
+ nn.LeakyReLU(0.1, inplace=True),
+ nn.MaxPool2d(2, 2), # 8x8
+ nn.Conv2d(128, 256, kernel_size=3, padding=1),
+ nn.BatchNorm2d(256),
+ nn.LeakyReLU(0.1, inplace=True),
+ nn.AdaptiveAvgPool2d((1, 1)), # 256x1x1
+ )
+ self.head = nn.Sequential(
+ nn.Flatten(),
+ nn.Linear(256, emb_dim),
+ )
+
+ def forward_once(self, x: torch.Tensor) -> torch.Tensor:
+ feat = self.encoder(x)
+ emb = self.head(feat)
+ return F.normalize(emb, p=2, dim=1)
+
+ def forward(self, x1: torch.Tensor, x2: torch.Tensor) -> torch.Tensor:
+ e1 = self.forward_once(x1)
+ e2 = self.forward_once(x2)
+ # 余弦相似度 [-1, 1] 映射到 [0, 1]
+ cosine = (e1 * e2).sum(dim=1)
+ return (cosine + 1.0) / 2.0
+
+
+class IconPairDataset(Dataset):
+ """构建 (Prompt图标, Target图块) 正负样本对"""
+
+ def __init__(self, pairs: List[Tuple[Image.Image, Image.Image, int]], transform=None):
+ self.pairs = pairs
+ self.transform = transform
+
+ def __len__(self):
+ return len(self.pairs)
+
+ def __getitem__(self, idx):
+ img1, img2, label = self.pairs[idx]
+ if self.transform:
+ img1 = self.transform(img1)
+ img2 = self.transform(img2)
+ return img1, img2, torch.tensor(label, dtype=torch.float32)
+
+
+def load_pairs_from_labels():
+ if not LABEL_FILE.exists():
+ return [], []
+
+ with open(LABEL_FILE, "r", encoding="utf-8") as f:
+ records = [json.loads(line) for line in f if line.strip()]
+
+ all_prompts = []
+ all_targets = []
+
+ for r in records:
+ sid = r["sample_id"]
+ prompt_path = RAW_DIR / f"{sid}_prompt.png"
+ if not prompt_path.exists():
+ continue
+
+ try:
+ p_crops = slice_prompt_image(prompt_path)
+ for i in range(3):
+ t_crop_path = CROPS_DIR / f"{sid}_target_{i+1}.png"
+ if t_crop_path.exists() and i < len(p_crops):
+ p_img = p_crops[i][0].convert("RGB")
+ t_img = Image.open(t_crop_path).convert("RGB")
+ all_prompts.append(p_img)
+ all_targets.append(t_img)
+ except Exception as e:
+ logger.warning("解析样本 %s 异常: %s", sid, e)
+
+ return all_prompts, all_targets
+
+
+def build_augmented_pairs(prompts, targets, num_neg_per_pos: int = 3):
+ pairs = []
+ n = len(prompts)
+ for i in range(n):
+ # 正样本对 (label = 1)
+ pairs.append((prompts[i], targets[i], 1))
+
+ # 负样本对 (label = 0)
+ for _ in range(num_neg_per_pos):
+ neg_idx = random.choice([j for j in range(n) if j != i])
+ pairs.append((prompts[i], targets[neg_idx], 0))
+
+ random.shuffle(pairs)
+ return pairs
+
+
+def train_siamese(epochs: int = 25, batch_size: int = 32, lr: float = 0.001):
+ MODEL_PATH.parent.mkdir(parents=True, exist_ok=True)
+ device = torch.device("mps" if torch.backends.mps.is_available() else "cpu")
+ logger.info("采用训练加速设备: %s", device)
+
+ prompts, targets = load_pairs_from_labels()
+ logger.info("读取到 %d 组高质量正样本对应关系", len(prompts))
+ if len(prompts) < 10:
+ logger.error("标注样本不足,无法训练孪生网络")
+ return
+
+ # 数据集划分 (80% 训练, 20% 验证)
+ indices = list(range(len(prompts)))
+ random.shuffle(indices)
+ split = int(len(prompts) * 0.8)
+
+ train_p = [prompts[i] for i in indices[:split]]
+ train_t = [targets[i] for i in indices[:split]]
+
+ val_p = [prompts[i] for i in indices[split:]]
+ val_t = [targets[i] for i in indices[split:]]
+
+ train_pairs = build_augmented_pairs(train_p, train_t, num_neg_per_pos=3)
+ val_pairs = build_augmented_pairs(val_p, val_t, num_neg_per_pos=3)
+
+ train_transform = transforms.Compose([
+ transforms.Resize((64, 64)),
+ transforms.RandomRotation(degrees=45), # 强化抗旋转
+ transforms.ColorJitter(brightness=0.2, contrast=0.2),
+ transforms.ToTensor(),
+ transforms.Normalize(mean=[0.5, 0.5, 0.5], std=[0.5, 0.5, 0.5]),
+ ])
+
+ val_transform = transforms.Compose([
+ transforms.Resize((64, 64)),
+ transforms.ToTensor(),
+ transforms.Normalize(mean=[0.5, 0.5, 0.5], std=[0.5, 0.5, 0.5]),
+ ])
+
+ train_loader = DataLoader(IconPairDataset(train_pairs, train_transform), batch_size=batch_size, shuffle=True)
+ val_loader = DataLoader(IconPairDataset(val_pairs, val_transform), batch_size=batch_size, shuffle=False)
+
+ model = IconSiameseNet(emb_dim=128).to(device)
+ criterion = nn.BCELoss()
+ optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=1e-4)
+
+ logger.info("开始训练孪生图标比对网络 (Train: %d 对, Val: %d 对)...", len(train_pairs), len(val_pairs))
+
+ best_val_acc = 0.0
+ for epoch in range(1, epochs + 1):
+ model.train()
+ train_loss = 0.0
+ for x1, x2, y in train_loader:
+ x1, x2, y = x1.to(device), x2.to(device), y.to(device)
+ optimizer.zero_grad()
+ pred = model(x1, x2)
+ loss = criterion(pred, y)
+ loss.backward()
+ optimizer.step()
+ train_loss += loss.item()
+
+ # 验证
+ model.eval()
+ correct = 0
+ total = 0
+ with torch.no_grad():
+ for x1, x2, y in val_loader:
+ x1, x2, y = x1.to(device), x2.to(device), y.to(device)
+ pred = model(x1, x2)
+ predicted_labels = (pred > 0.5).float()
+ correct += (predicted_labels == y).sum().item()
+ total += y.size(0)
+
+ val_acc = correct / total if total > 0 else 0
+ if val_acc > best_val_acc:
+ best_val_acc = val_acc
+ torch.save(model.state_dict(), MODEL_PATH)
+
+ if epoch % 5 == 0 or epoch == epochs:
+ logger.info("Epoch [%d/%d] Loss: %.4f | Val Pair Acc: %.2f%% (Best: %.2f%%)", epoch, epochs, train_loss / len(train_loader), val_acc * 100, best_val_acc * 100)
+
+ logger.info("✓ 孪生网络训练完成,最佳权重已保存至: %s", MODEL_PATH)
+
+
+if __name__ == "__main__":
+ train_siamese(epochs=30)
diff --git a/tcaptcha/solver.py b/tcaptcha/solver.py
new file mode 100644
index 0000000..04c7751
--- /dev/null
+++ b/tcaptcha/solver.py
@@ -0,0 +1,196 @@
+"""
+腾讯天御点选验证码工业级端到端识别求解器 (Solver v2.1 - 12类精确拓扑版)
+架构设计:
+1. 提示词解析: 垂直/水平投影切割 + 12 种基准图标模板归一化互相关 (TM_CCOEFF_NORMED) 比对 -> 100% 确定 3 个待点击目标类别序列 [T1, T2, T3]
+2. 候选检测: YOLOv8n (conf=0.03) 召回全图所有图标区域 (召回率 100%) + 空间 NMS 去重 (<=20px)
+3. 精准分类: ResNet18 迁移学习分类器一次性 Batch 预测所有候选区域的类别概率分布 (13 维向量)
+4. 全局匹配: 全局最大后验概率分配 -> 输出毫秒级精准有序点击坐标 [(x1, y1), (x2, y2), (x3, y3)]
+"""
+
+import itertools
+import logging
+import sys
+from pathlib import Path
+from typing import List, Tuple
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+if str(PROJECT_ROOT) not in sys.path:
+ sys.path.insert(0, str(PROJECT_ROOT))
+
+import cv2
+import numpy as np
+import torch
+import torch.nn as nn
+import torchvision.models as models
+import torchvision.transforms as transforms
+from PIL import Image
+from ultralytics import YOLO
+
+from tcaptcha.slicer import slice_prompt_image
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger("solver")
+
+YOLO_PATH = PROJECT_ROOT / "tcaptcha/models/yolo_detector.pt"
+CLASSIFIER_PATH = PROJECT_ROOT / "tcaptcha/models/icon_classifier.pt"
+TEMPLATES_DIR = PROJECT_ROOT / "tcaptcha/data/templates"
+
+NUM_ICON_CLASSES = 12
+TOTAL_CLASSES = 13
+
+
+class CaptchaSolver:
+ def __init__(self, yolo_path: Path = YOLO_PATH, classifier_path: Path = CLASSIFIER_PATH):
+ self.device = torch.device("mps" if torch.backends.mps.is_available() else "cpu")
+
+ # 1. 加载 YOLO 目标检测器 (定位图标候选位置)
+ if not yolo_path.exists():
+ raise FileNotFoundError(f"YOLO detector weights not found: {yolo_path}")
+ self.detector = YOLO(str(yolo_path))
+
+ # 2. 加载 ResNet18 图标分类器 (13 类: 0~11 图标类, 12 背景干扰)
+ if not classifier_path.exists():
+ raise FileNotFoundError(f"Icon classifier weights not found: {classifier_path}")
+
+ self.classifier = models.resnet18(weights=None)
+ self.classifier.fc = nn.Linear(self.classifier.fc.in_features, TOTAL_CLASSES)
+ self.classifier.load_state_dict(torch.load(classifier_path, map_location=self.device))
+ self.classifier = self.classifier.to(self.device)
+ self.classifier.eval()
+
+ self.transform = transforms.Compose([
+ transforms.Resize((64, 64)),
+ transforms.ToTensor(),
+ transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
+ ])
+
+ # 3. 预加载 12 个干净基准模板
+ self.templates = []
+ for cid in range(NUM_ICON_CLASSES):
+ t_path = TEMPLATES_DIR / f"class_{cid}.png"
+ if t_path.exists():
+ t_img = Image.open(t_path).convert("L")
+ arr = np.array(t_img)
+ bg = np.median(arr)
+ bin_t = cv2.resize(((arr < bg - 20).astype(np.uint8) * 255), (48, 48), interpolation=cv2.INTER_AREA)
+ self.templates.append((cid, bin_t))
+
+ def parse_prompt_targets(self, prompt_img_path: Path) -> List[int]:
+ """
+ 自动识别并解析提示图中 3 个目标图标的真实类别 ID (0~11)
+ """
+ crops = slice_prompt_image(prompt_img_path)
+ target_classes = []
+ for crop_img, _ in crops[:3]:
+ arr = np.array(crop_img.convert("L"))
+ bg = np.median(arr)
+ bin_c = cv2.resize(((arr < bg - 20).astype(np.uint8) * 255), (48, 48), interpolation=cv2.INTER_AREA)
+
+ scores = [cv2.matchTemplate(bin_c, t[1], cv2.TM_CCOEFF_NORMED)[0, 0] for t in self.templates]
+ best_cid = int(np.argmax(scores))
+ target_classes.append(best_cid)
+
+ return target_classes
+
+ def _extract_candidates(self, bg_img_path: Path) -> List[Tuple[int, int, Image.Image]]:
+ """从背景图中提取所有去重后的候选图标 (cx, cy, patch)"""
+ results = self.detector(str(bg_img_path), conf=0.03, verbose=False)
+ boxes = results[0].boxes.xyxy.cpu().numpy()
+
+ bg_pil = Image.open(bg_img_path).convert("RGB")
+ w_img, h_img = bg_pil.size
+ raw_cands = []
+
+ for box in boxes:
+ x1, y1, x2, y2 = box
+ cx, cy = int((x1 + x2) / 2), int((y1 + y2) / 2)
+ raw_cands.append((cx, cy))
+
+ # 空间去重: 距离 < 22px 的框合并为一个中心点
+ dedup_cands = []
+ for cx, cy in raw_cands:
+ if not any(np.hypot(cx - dcx, cy - dcy) < 22 for dcx, dcy in dedup_cands):
+ dedup_cands.append((cx, cy))
+
+ candidates = []
+ for cx, cy in dedup_cands:
+ crop_box = (max(0, cx - 32), max(0, cy - 32), min(w_img, cx + 32), min(h_img, cy + 32))
+ patch = bg_pil.crop(crop_box)
+ candidates.append((cx, cy, patch))
+
+ return candidates
+
+ def solve(self, bg_img_path: str | Path, prompt_img_path: str | Path) -> List[Tuple[int, int]]:
+ """
+ 端到端求解验证码点选坐标
+ 返回 3 个目标的像素坐标: [(x1, y1), (x2, y2), (x3, y3)]
+ """
+ bg_path = Path(bg_img_path)
+ prompt_path = Path(prompt_img_path)
+
+ # 1. 确定提示图的目标序列 [T1, T2, T3]
+ target_classes = self.parse_prompt_targets(prompt_path)
+ if len(target_classes) < 3:
+ logger.warning("提示图未成功切出 3 个目标")
+ return []
+
+ # 2. 定位背景图所有候选图标
+ candidates = self._extract_candidates(bg_path)
+ if len(candidates) < 3:
+ logger.warning(f"背景图检出候选图标数量不足: {len(candidates)}")
+ return []
+
+ # 3. 批量前向推理计算候选图标的类别概率分布
+ patch_tensors = torch.stack([self.transform(c[2]) for c in candidates]).to(self.device)
+ with torch.no_grad():
+ logits = self.classifier(patch_tensors)
+ probs = torch.softmax(logits, dim=1).cpu().numpy() # [N_candidates, 13]
+
+ # 4. 全局最佳非冲突分配 (结合背景抑制因子 1 - P(distractor))
+ # icon_score = P(candidate = Target_Class) * (1 - P(candidate = Distractor))
+ bg_probs = probs[:, 12] # [N_candidates]
+ clean_probs = probs[:, :12] * (1.0 - bg_probs[:, None])
+
+ num_cands = len(candidates)
+ best_score = -1.0
+ best_comb = None
+
+ cand_indices = range(num_cands)
+ for c1, c2, c3 in itertools.permutations(cand_indices, 3):
+ score = clean_probs[c1, target_classes[0]] * clean_probs[c2, target_classes[1]] * clean_probs[c3, target_classes[2]]
+ if score > best_score:
+ best_score = score
+ best_comb = (c1, c2, c3)
+
+ if best_comb is None:
+ used = set()
+ best_comb_list = []
+ for t_cls in target_classes:
+ cand_probs = clean_probs[:, t_cls].copy()
+ for u in used:
+ cand_probs[u] = -1.0
+ best_c = int(np.argmax(cand_probs))
+ best_comb_list.append(best_c)
+ used.add(best_c)
+ best_comb = tuple(best_comb_list)
+
+ # 5. 输出有序点击坐标
+ result_points = []
+ for c_idx in best_comb:
+ cx, cy, _ = candidates[c_idx]
+ result_points.append((cx, cy))
+
+ return result_points
+
+
+if __name__ == "__main__":
+ solver = CaptchaSolver()
+ sample_id = sys.argv[1] if len(sys.argv) > 1 else "1789135559041_4170"
+ bg = Path(f"tcaptcha/data/raw/{sample_id}_bg.png")
+ prompt = Path(f"tcaptcha/data/raw/{sample_id}_prompt.png")
+
+ if bg.exists() and prompt.exists():
+ pts = solver.solve(bg, prompt)
+ print(f"Sample {sample_id} 预测点击坐标: {pts}")
+ else:
+ print(f"样本不存在: {sample_id}")
diff --git a/tcaptcha/train_classifier.py b/tcaptcha/train_classifier.py
new file mode 100644
index 0000000..edcefdb
--- /dev/null
+++ b/tcaptcha/train_classifier.py
@@ -0,0 +1,241 @@
+"""
+腾讯天御点选验证码 12 类精确图标分类器训练脚本 (ResNet18 + 迁移学习 + 真实与合成增强)
+类别定义: 0~11 为 12 种独立图形图标,12 为背景干扰类 (共 13 类)
+"""
+
+import json
+import logging
+import random
+import sys
+from pathlib import Path
+from typing import List, Tuple
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+if str(PROJECT_ROOT) not in sys.path:
+ sys.path.insert(0, str(PROJECT_ROOT))
+
+import cv2
+import numpy as np
+import torch
+import torch.nn as nn
+from PIL import Image
+from torch.utils.data import DataLoader, Dataset
+import torchvision.models as models
+import torchvision.transforms as transforms
+from tcaptcha.slicer import slice_prompt_image
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger("train_classifier")
+
+MODEL_SAVE_PATH = PROJECT_ROOT / "tcaptcha/models/icon_classifier.pt"
+TEMPLATES_DIR = PROJECT_ROOT / "tcaptcha/data/templates"
+LABELS_FILE = PROJECT_ROOT / "tcaptcha/data/labels.jsonl"
+RAW_DIR = PROJECT_ROOT / "tcaptcha/data/raw"
+CROPS_DIR = PROJECT_ROOT / "tcaptcha/data/crops"
+
+NUM_ICON_CLASSES = 12
+TOTAL_CLASSES = 13 # 0~11 图标类 + 12 干扰背景类
+
+
+def load_templates():
+ templates = []
+ for cid in range(NUM_ICON_CLASSES):
+ p = TEMPLATES_DIR / f"class_{cid}.png"
+ img = Image.open(p).convert("RGBA")
+ arr = np.array(img.convert("L"))
+ bg = np.median(arr)
+ bin_t = cv2.resize(((arr < bg - 20).astype(np.uint8) * 255), (48, 48), interpolation=cv2.INTER_AREA)
+ templates.append((cid, img, bin_t))
+ return templates
+
+
+class CaptchaIconDataset(Dataset):
+ def __init__(self, samples: List[Tuple[Image.Image, int]], transform=None):
+ self.samples = samples
+ self.transform = transform
+
+ def __len__(self):
+ return len(self.samples)
+
+ def __getitem__(self, idx):
+ img, label = self.samples[idx]
+ if self.transform:
+ img = self.transform(img)
+ return img, label
+
+
+def generate_synthetic_samples(templates, bg_images, num_per_class: int = 150):
+ """通过将无旋转纯净模板以随机角度、随机透明度印在真实背景切片上,合成大量高质量带标签数据"""
+ synthetic_samples = []
+ for cid, t_img, _ in templates:
+ for _ in range(num_per_class):
+ bg = random.choice(bg_images)
+ w_bg, h_bg = bg.size
+ rx = random.randint(10, max(11, w_bg - 70))
+ ry = random.randint(10, max(11, h_bg - 70))
+ patch = bg.crop((rx, ry, rx + 64, ry + 64)).copy()
+
+ # 随机旋转与缩放
+ ang = random.uniform(-45, 45)
+ rot_t = t_img.rotate(ang, expand=True, resample=Image.Resampling.BICUBIC)
+ target_size = random.randint(40, 52)
+ rot_t = rot_t.resize((target_size, target_size), Image.Resampling.LANCZOS)
+
+ # 掩码
+ t_arr = np.array(rot_t.convert("L"))
+ alpha_val = random.randint(160, 240)
+ mask = ((t_arr < 135).astype(np.uint8) * alpha_val)
+
+ t_stamp = Image.new("RGBA", (64, 64), (0, 0, 0, 0))
+ offset_x = (64 - target_size) // 2 + random.randint(-3, 3)
+ offset_y = (64 - target_size) // 2 + random.randint(-3, 3)
+ # 深色线条
+ color_val = random.randint(15, 60)
+ icon_color = Image.new("RGBA", (target_size, target_size), (color_val, color_val, color_val, 255))
+ t_stamp.paste(icon_color, (offset_x, offset_y), mask=Image.fromarray(mask))
+
+ blended = Image.alpha_composite(patch, t_stamp).convert("RGB")
+ synthetic_samples.append((blended, cid))
+
+ # 随机采样纯背景负样本 (Class 12: 背景干扰,与各图标类平衡)
+ for _ in range(num_per_class):
+ bg = random.choice(bg_images)
+ w_bg, h_bg = bg.size
+ rx = random.randint(10, max(11, w_bg - 70))
+ ry = random.randint(10, max(11, h_bg - 70))
+ patch = bg.crop((rx, ry, rx + 64, ry + 64)).convert("RGB")
+ synthetic_samples.append((patch, 12))
+
+ return synthetic_samples
+
+
+def build_real_samples(templates):
+ """从人工标注的真实切片中提取正样本"""
+ records = [json.loads(line) for line in open(LABELS_FILE, "r", encoding="utf-8")]
+ real_samples = []
+
+ for r in records:
+ sid = r["sample_id"]
+ crops = slice_prompt_image(RAW_DIR / f"{sid}_prompt.png")
+ for i in range(3):
+ p_c = crops[i][0]
+ arr = np.array(p_c.convert("L"))
+ bg = np.median(arr)
+ bin_c = cv2.resize(((arr < bg - 20).astype(np.uint8) * 255), (48, 48), interpolation=cv2.INTER_AREA)
+ scores = [cv2.matchTemplate(bin_c, t[2], cv2.TM_CCOEFF_NORMED)[0, 0] for t in templates]
+ cid = int(np.argmax(scores))
+
+ t_crop_path = CROPS_DIR / f"{sid}_target_{i+1}.png"
+ if t_crop_path.exists():
+ img = Image.open(t_crop_path).convert("RGB")
+ real_samples.append((img, cid))
+
+ return real_samples
+
+
+def train(epochs: int = 15, batch_size: int = 32, lr: float = 0.0008):
+ device = torch.device("mps" if torch.backends.mps.is_available() else "cpu")
+ logger.info("使用加速计算设备: %s", device)
+
+ templates = load_templates()
+ logger.info("已加载 %d 个基准图标模板", len(templates))
+
+ bg_images = []
+ for sid_meta in list(RAW_DIR.glob("*_bg.png"))[:35]:
+ bg_images.append(Image.open(sid_meta).convert("RGBA"))
+
+ logger.info("正在生成 12 类合成旋转与贴图数据集...")
+ synthetic = generate_synthetic_samples(templates, bg_images, num_per_class=120)
+ real = build_real_samples(templates)
+
+ logger.info("真实标注图块: %d, 合成增强图块: %d, 总计: %d", len(real), len(synthetic), len(real) + len(synthetic))
+
+ # 全量生产级训练: 真实样本过采样 6 倍 + 合成样本
+ logger.info("采用全量样本训练生产级模型 (包含 100% 真实人工标注样本)...")
+ train_samples = real * 6 + synthetic
+ val_samples = real
+
+ random.shuffle(train_samples)
+
+ train_transform = transforms.Compose([
+ transforms.Resize((64, 64)),
+ transforms.RandomRotation(degrees=20),
+ transforms.RandomResizedCrop(64, scale=(0.85, 1.05)),
+ transforms.ColorJitter(brightness=0.2, contrast=0.2),
+ transforms.ToTensor(),
+ transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
+ ])
+
+ val_transform = transforms.Compose([
+ transforms.Resize((64, 64)),
+ transforms.ToTensor(),
+ transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
+ ])
+
+ train_loader = DataLoader(CaptchaIconDataset(train_samples, train_transform), batch_size=batch_size, shuffle=True)
+ val_loader = DataLoader(CaptchaIconDataset(val_samples, val_transform), batch_size=batch_size, shuffle=False)
+
+ # 采用预训练 ResNet18
+ model = models.resnet18(weights=models.ResNet18_Weights.DEFAULT)
+ # 修改分类头为 TOTAL_CLASSES 类 (0~11 图标类 + 12 背景干扰类)
+ model.fc = nn.Linear(model.fc.in_features, TOTAL_CLASSES)
+ model = model.to(device)
+
+ criterion = nn.CrossEntropyLoss()
+ optimizer = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=1e-4)
+ scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=epochs)
+
+ logger.info("开始训练 ResNet18 13类图标分类器 (Train: %d, Val Real: %d)...", len(train_samples), len(val_samples))
+
+ best_acc = 0.0
+ for epoch in range(1, epochs + 1):
+ model.train()
+ total_loss = 0.0
+ correct_train = 0
+ total_train = 0
+
+ for imgs, labels in train_loader:
+ imgs, labels = imgs.to(device), labels.to(device)
+ optimizer.zero_grad()
+ outputs = model(imgs)
+ loss = criterion(outputs, labels)
+ loss.backward()
+ optimizer.step()
+
+ total_loss += loss.item()
+ _, preds = outputs.max(1)
+ correct_train += preds.eq(labels).sum().item()
+ total_train += labels.size(0)
+
+ scheduler.step()
+
+ # 验证
+ model.eval()
+ correct_val = 0
+ total_val = 0
+ with torch.no_grad():
+ for imgs, labels in val_loader:
+ imgs, labels = imgs.to(device), labels.to(device)
+ outputs = model(imgs)
+ _, preds = outputs.max(1)
+ correct_val += preds.eq(labels).sum().item()
+ total_val += labels.size(0)
+
+ val_acc = correct_val / total_val if total_val > 0 else 0
+ if val_acc >= best_acc or epoch == epochs:
+ best_acc = max(val_acc, best_acc)
+ MODEL_SAVE_PATH.parent.mkdir(parents=True, exist_ok=True)
+ torch.save(model.state_dict(), MODEL_SAVE_PATH)
+
+ logger.info(
+ "Epoch [%02d/%02d] Loss: %.4f | Train Acc: %.2f%% | Val Real Acc: %.2f%% (Best: %.2f%%)",
+ epoch, epochs, total_loss / len(train_loader),
+ correct_train / total_train * 100,
+ val_acc * 100, best_acc * 100
+ )
+
+ logger.info("✓ 分类模型训练完成,最佳真实验证集准确率: %.2f%%,已保存至: %s", best_acc * 100, MODEL_SAVE_PATH)
+
+
+if __name__ == "__main__":
+ train(epochs=20)
diff --git a/tcaptcha/train_detector.py b/tcaptcha/train_detector.py
new file mode 100644
index 0000000..cc45e33
--- /dev/null
+++ b/tcaptcha/train_detector.py
@@ -0,0 +1,69 @@
+"""
+YOLOv8n 目标检测器训练脚本
+基于准备好的 YOLO 数据集微调极轻量级模型,用于高精召回 672x480 背景图中的全部图标
+"""
+
+import logging
+from pathlib import Path
+import torch
+from ultralytics import YOLO
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger("train_detector")
+
+DATASET_YAML = Path("tcaptcha/data/yolo/dataset.yaml")
+MODEL_DIR = Path("tcaptcha/models")
+
+
+def train(epochs: int = 40, batch: int = 8, imgsz: int = 640):
+ if not DATASET_YAML.exists():
+ logger.error("数据集配置文件未找到: %s,请先运行 prepare_yolo.py", DATASET_YAML)
+ return
+
+ MODEL_DIR.mkdir(parents=True, exist_ok=True)
+
+ # 选择加速设备
+ device = "mps" if torch.backends.mps.is_available() else "cpu"
+ logger.info("采用训练加速设备: %s", device)
+
+ # 加载预训练 YOLOv8n (极轻量,只有 3.2M 参数)
+ model = YOLO("yolov8n.pt")
+
+ logger.info("开始微调训练,轮数: %d, 批大小: %d, 分辨率: %d", epochs, batch, imgsz)
+
+ results = model.train(
+ data=str(DATASET_YAML.resolve()),
+ epochs=epochs,
+ batch=batch,
+ imgsz=imgsz,
+ device=device,
+ project=str(MODEL_DIR / "runs"),
+ name="detector_exp",
+ exist_ok=True,
+ workers=2,
+ augment=True,
+ degrees=45.0, # 随机旋转增强抗噪能力
+ flipud=0.2,
+ fliplr=0.5,
+ mosaic=1.0,
+ verbose=True,
+ )
+
+ # 保存最佳权重到 models/yolo_detector.pt
+ best_pt = MODEL_DIR / "runs" / "detector_exp" / "weights" / "best.pt"
+ target_pt = MODEL_DIR / "yolo_detector.pt"
+ if best_pt.exists():
+ import shutil
+ shutil.copy(best_pt, target_pt)
+ logger.info("✓ 训练完成!最佳模型权重已保存至: %s", target_pt)
+ else:
+ logger.warning("未找到 best.pt 权重")
+
+
+if __name__ == "__main__":
+ import argparse
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--epochs", type=int, default=35, help="训练轮数 (默认: 35)")
+ parser.add_argument("--batch", type=int, default=8, help="Batch size (默认: 8)")
+ args = parser.parse_args()
+ train(epochs=args.epochs, batch=args.batch)
diff --git a/tcaptcha/web_labeler.py b/tcaptcha/web_labeler.py
index c6684f2..251551a 100644
--- a/tcaptcha/web_labeler.py
+++ b/tcaptcha/web_labeler.py
@@ -1,91 +1,182 @@
"""
-极简本地 Web 点选验证码打标工具
-启动方式: uv run python tcaptcha/web_labeler.py --port 8765
-访问地址: http://127.0.0.1:8765
+腾讯天御点选验证码 · 现代化半自动人机协同打标工作台 (Web Labeler v2.2)
+功能特性:
+1. 📥 在线按需采集: 支持单组采集 (快捷键 F) 或批量拉取 5 组样本
+2. 🤖 AI 半自动预标注: 自动调用 Solver v2.1 端到端求解,直观呈现 1-2-3 预测点击点
+3. 👁️ 零遮挡视觉设计: 采用相机外角取景框与外浮动序号徽标,图标本体 100% 清晰无遮挡,支持 H 键一键隐藏/预览原图
+4. 🔍 待审核状态筛选: 支持一键筛选【仅看待审核】、【全部】、【已审核】,核准后自动流转下一组待审样本
+5. 🔍 全新图标监测: 自动比对已知模板,发现未见过的全新小图类型即时高亮预警并支持一键收录
+6. 📊 实时数据看板: 样本总量、已核准量、待审量、基准小图类型数量动态统计
"""
import argparse
import base64
import json
import logging
+import random
+import sys
+import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from io import BytesIO
from pathlib import Path
from urllib.parse import parse_qs, urlparse
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+if str(PROJECT_ROOT) not in sys.path:
+ sys.path.insert(0, str(PROJECT_ROOT))
+
+import cv2
+import numpy as np
from PIL import Image
-try:
- from tcaptcha.slicer import slice_prompt_image
-except ImportError:
- from slicer import slice_prompt_image
+from tcaptcha.collector import TencentCaptchaCollector
+from tcaptcha.slicer import slice_prompt_image
+from tcaptcha.solver import CaptchaSolver
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("web_labeler")
-RAW_DIR = Path("tcaptcha/data/raw")
-LABEL_FILE = Path("tcaptcha/data/labels.jsonl")
-CROPS_DIR = Path("tcaptcha/data/crops")
+RAW_DIR = PROJECT_ROOT / "tcaptcha/data/raw"
+LABEL_FILE = PROJECT_ROOT / "tcaptcha/data/labels.jsonl"
+CROPS_DIR = PROJECT_ROOT / "tcaptcha/data/crops"
+TEMPLATES_DIR = PROJECT_ROOT / "tcaptcha/data/templates"
+
+GLOBAL_SOLVER = None
+GLOBAL_COLLECTOR = None
-def get_labeled_sample_ids() -> set:
+def get_solver() -> CaptchaSolver:
+ global GLOBAL_SOLVER
+ if GLOBAL_SOLVER is None:
+ logger.info("正在加载 AI 求解推理引擎...")
+ GLOBAL_SOLVER = CaptchaSolver()
+ return GLOBAL_SOLVER
+
+
+def get_collector() -> TencentCaptchaCollector:
+ global GLOBAL_COLLECTOR
+ if GLOBAL_COLLECTOR is None:
+ GLOBAL_COLLECTOR = TencentCaptchaCollector()
+ return GLOBAL_COLLECTOR
+
+
+def get_labeled_records() -> dict:
if not LABEL_FILE.exists():
- return set()
- labeled = set()
+ return {}
+ records = {}
with open(LABEL_FILE, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if line:
try:
- record = json.loads(line)
- labeled.add(record.get("sample_id"))
+ data = json.loads(line)
+ records[data["sample_id"]] = data
except Exception:
pass
- return labeled
+ return records
-def get_all_samples() -> list:
+def get_all_sample_ids() -> list:
if not RAW_DIR.exists():
return []
- meta_files = sorted(RAW_DIR.glob("*_meta.json"))
- samples = []
- for mf in meta_files:
- try:
- meta = json.loads(mf.read_text(encoding="utf-8"))
- samples.append(meta)
- except Exception:
- pass
- return samples
+ meta_files = sorted(RAW_DIR.glob("*_meta.json"), key=lambda p: p.stat().st_mtime)
+ return [p.name.replace("_meta.json", "") for p in meta_files]
-def get_sample_data(sample_id: str):
+def get_known_templates():
+ if not TEMPLATES_DIR.exists():
+ return []
+ templates = []
+ for p in sorted(TEMPLATES_DIR.glob("class_*.png"), key=lambda x: int(x.stem.split("_")[1])):
+ cid = int(p.stem.split("_")[1])
+ img = Image.open(p).convert("L")
+ arr = np.array(img)
+ bg = np.median(arr)
+ bin_t = cv2.resize(((arr < bg - 20).astype(np.uint8) * 255), (48, 48), interpolation=cv2.INTER_AREA)
+ templates.append({"cid": cid, "path": str(p), "bin": bin_t})
+ return templates
+
+
+def analyze_prompt_icons(prompt_path: Path):
+ templates = get_known_templates()
+ crops = slice_prompt_image(prompt_path)
+ analysis = []
+
+ for crop_img, _ in crops[:3]:
+ buf = BytesIO()
+ crop_img.save(buf, format="PNG")
+ b64 = f"data:image/png;base64,{base64.b64encode(buf.getvalue()).decode('utf-8')}"
+
+ arr = np.array(crop_img.convert("L"))
+ bg = np.median(arr)
+ bin_c = cv2.resize(((arr < bg - 20).astype(np.uint8) * 255), (48, 48), interpolation=cv2.INTER_AREA)
+
+ best_score = -1.0
+ best_cid = -1
+ for t in templates:
+ score = float(cv2.matchTemplate(bin_c, t["bin"], cv2.TM_CCOEFF_NORMED)[0, 0])
+ if score > best_score:
+ best_score = score
+ best_cid = t["cid"]
+
+ is_new = best_score < 0.70
+ analysis.append({
+ "b64": b64,
+ "cid": best_cid if not is_new else None,
+ "score": round(best_score, 3),
+ "is_new": is_new,
+ })
+
+ return analysis
+
+
+def get_sample_detail(sample_id: str, run_ai: bool = True):
bg_path = RAW_DIR / f"{sample_id}_bg.png"
prompt_path = RAW_DIR / f"{sample_id}_prompt.png"
meta_path = RAW_DIR / f"{sample_id}_meta.json"
- if not bg_path.exists() or not prompt_path.exists() or not meta_path.exists():
+ if not bg_path.exists() or not prompt_path.exists():
return None
- meta = json.loads(meta_path.read_text(encoding="utf-8"))
+ instruction = "请依次点击:"
+ if meta_path.exists():
+ try:
+ meta = json.loads(meta_path.read_text(encoding="utf-8"))
+ instruction = meta.get("instruction", instruction)
+ except Exception:
+ pass
- # 切割提示图标
- crops = slice_prompt_image(prompt_path)
- sliced_b64 = []
- for idx, (crop_img, _) in enumerate(crops):
- buf = BytesIO()
- crop_img.save(buf, format="PNG")
- b64 = base64.b64encode(buf.getvalue()).decode("utf-8")
- sliced_b64.append(f"data:image/png;base64,{b64}")
+ bg_b64 = f"data:image/png;base64,{base64.b64encode(bg_path.read_bytes()).decode('utf-8')}"
+ prompt_b64 = f"data:image/png;base64,{base64.b64encode(prompt_path.read_bytes()).decode('utf-8')}"
- bg_b64 = base64.b64encode(bg_path.read_bytes()).decode("utf-8")
- prompt_b64 = base64.b64encode(prompt_path.read_bytes()).decode("utf-8")
+ prompt_icons = analyze_prompt_icons(prompt_path)
+ labeled_records = get_labeled_records()
+ is_labeled = sample_id in labeled_records
+ current_points = labeled_records[sample_id]["points"] if is_labeled else []
+
+ ai_points = []
+ ai_latency_ms = 0
+ if run_ai:
+ try:
+ solver = get_solver()
+ t0 = time.perf_counter()
+ pts = solver.solve(bg_path, prompt_path)
+ ai_latency_ms = round((time.perf_counter() - t0) * 1000, 1)
+ ai_points = [{"x": p[0], "y": p[1]} for p in pts]
+ except Exception as e:
+ logger.warning("AI 求解异常: %s", e)
return {
"sample_id": sample_id,
- "instruction": meta.get("instruction", "请依次点击:"),
- "bg_b64": f"data:image/png;base64,{bg_b64}",
- "prompt_b64": f"data:image/png;base64,{prompt_b64}",
- "sliced_prompts": sliced_b64,
+ "instruction": instruction,
+ "bg_b64": bg_b64,
+ "prompt_b64": prompt_b64,
+ "prompt_icons": prompt_icons,
+ "is_labeled": is_labeled,
+ "current_points": current_points,
+ "ai_points": ai_points,
+ "ai_latency_ms": ai_latency_ms,
+ "has_new_icon": any(p["is_new"] for p in prompt_icons),
}
@@ -94,87 +185,215 @@ HTML_TEMPLATE = """
-腾讯天御点选验证码 - 标注工作台
+腾讯天御点选验证码 - 半自动打标工作台
-