升级半自动工作台

This commit is contained in:
yml2213
2026-09-11 22:56:03 +08:00
parent f1c664f622
commit 4d6cded0b0
13 changed files with 3269 additions and 231 deletions
+2
View File
@@ -22,3 +22,5 @@ __pycache__/
tcaptcha/data/raw/
tcaptcha/data/crops/
data/
runs/
yolov8n.pt
+3
View File
@@ -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",
]
+147
View File
@@ -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)
Binary file not shown.
Binary file not shown.
Binary file not shown.
+144
View File
@@ -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()
+228
View File
@@ -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)
+196
View File
@@ -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}")
+241
View File
@@ -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)
+69
View File
@@ -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)
+1020 -234
View File
File diff suppressed because it is too large Load Diff
Generated
+1222
View File
File diff suppressed because it is too large Load Diff