yml2213
60ae14d54a
- 题目图片自动下载: 所有题型原图统一落盘至 scratch/captcha_debug，文字点选自动生成带编号红圈标注图便于肉眼核验 - 字形严格一致性校验: 在 SolveResult 增加 text_match_ok 检查，对图字不符或混淆干扰题直接拒答换题，避免错答引发 errorCode=12 - 防秒换与拟人延时: 遇到非预期题目换题等待 2.8~4.2s (对齐 HAR 真实抓包)，求解失败等待 2.5~3.8s，作答前模拟交互 5.5~7.5s - 代理容错增强: 在 OneMsAdapter 捕获代理断流与鉴权异常，支持动态轮换新代理重试
369 lines
17 KiB
Python
369 lines
17 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
腾讯天御『文字点选』验证码求解器 (Tencent Turing Text-Click Solver)
|
||
|
||
适用题型: subcapclass = 2404, click_cfg.data_type = ["DynAnswerType_POS"] 且提示为
|
||
纯文字指令, 例如: "请依次点击：虫 呈 般"
|
||
|
||
与 tcaptcha.solver.CaptchaSolver (图标点选 2408: 提示是 img_index=0 的图标图) 的区别:
|
||
* 本题型 **没有提示图** (img_index=0 返回 0 字节)
|
||
* 背景图里只出现与提示完全相同的 3 个汉字 (随机位置), 不含额外干扰字
|
||
* 部分题目会做 **斜切(italic)变形**, 但实测不影响颜色分割
|
||
|
||
算法管线:
|
||
1. HSV 颜色分割 (腾讯点选汉字固定为橙黄色, H≈20 S≈185 V≈228) 提取字形掩膜
|
||
2. 形态学闭运算 + 膨胀连通同一汉字的多个偏旁, 连通域得到汉字包围盒
|
||
3. 掩膜 -> 白底黑字 -> 放大, 交给 OCR 集成(ddddocr + PP-OCRv4)做 CTC 前向
|
||
4. 取 CTC 每一时间步上目标字的概率最大值 -> (字形, 目标字) 相似度矩阵
|
||
5. 对 3! 个全排列求最优指派, 并用归一化后验概率给出本次求解的置信度
|
||
(置信度低的题目直接刷新会话, 避免提交明显不可靠的答案)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import itertools
|
||
import logging
|
||
import re
|
||
from typing import Dict, List, Optional, Sequence, Tuple
|
||
|
||
import cv2
|
||
import numpy as np
|
||
from PIL import Image
|
||
|
||
from .ocr_engines import EnsembleOcr
|
||
|
||
logger = logging.getLogger("tcaptcha_glyph")
|
||
|
||
# 腾讯点选汉字颜色 (实测 H≈20.5 S≈200 V≈232)。
|
||
# 关键是 **V(亮度)** 而不是色相: 暖色草地/沙地的 H 会落在 24~28、S 也不低,
|
||
# 但 V 只有 120~156, 而汉字 V≈230。早期版本用 V>=145 会把草地并进字形掩膜,
|
||
# 导致 3 个字被背景吞成 1~2 个候选 (见图 proxy/fail_2、fail_3)。
|
||
HSV_LOWER = (12, 145, 190)
|
||
HSV_UPPER = (34, 255, 255)
|
||
|
||
MIN_GLYPH_AREA = 700 # 单个汉字掩膜最小像素数
|
||
MERGE_KERNEL = 13 # 膨胀核, 用于把偏旁/离散笔画合并为同一汉字
|
||
SCALE = 3 # OCR 输入放大倍数
|
||
PAD = 24 # OCR 输入白边
|
||
|
||
_CJK_RE = re.compile(r"[\u4e00-\u9fff]")
|
||
# 滑块/图标题型的模板串关键词: 出现即说明提示里没有目标汉字
|
||
_NON_TARGET_HINTS = ("滑块", "拖动", "拼图", "请依次", "点击", "完成")
|
||
|
||
|
||
def parse_instruction(instruction: str) -> List[str]:
|
||
"""从 '请依次点击：虫 呈 般' 中解析出目标汉字序列 ['虫','呈','般']
|
||
|
||
注意: 图标点选 (2408) 与滑块 (2402/2401) 的 instruction 只有 '请依次点击：' /
|
||
'拖动下方滑块完成拼图' 这类无目标文字的模板串, 此时必须返回空列表, 否则会把
|
||
'请依次点击' 本身当成目标字。
|
||
"""
|
||
if not instruction:
|
||
return []
|
||
parts = re.split(r"[：:]", instruction, maxsplit=1)
|
||
if len(parts) == 2:
|
||
return _CJK_RE.findall(parts[1])
|
||
# 没有分隔符: 只有当整体就是一个短目标字序列 (如 '虫 呈 般') 时才接受
|
||
if any(h in instruction for h in _NON_TARGET_HINTS):
|
||
return []
|
||
chars = _CJK_RE.findall(instruction)
|
||
# 文字点选恒为 3~4 个目标字, 超过该数量必然是模板句
|
||
return chars if 0 < len(chars) <= 4 else []
|
||
|
||
|
||
class Glyph:
|
||
"""一个已定位的候选汉字字形"""
|
||
|
||
__slots__ = ("x0", "y0", "x1", "y1", "cx", "cy", "mask", "ocr_text", "confidence")
|
||
|
||
def __init__(self, x0: int, y0: int, x1: int, y1: int, mask: np.ndarray):
|
||
self.x0, self.y0, self.x1, self.y1 = x0, y0, x1, y1
|
||
self.cx = (x0 + x1) // 2
|
||
self.cy = (y0 + y1) // 2
|
||
self.mask = mask
|
||
self.ocr_text = ""
|
||
self.confidence = 0.0
|
||
|
||
@property
|
||
def point(self) -> Tuple[int, int]:
|
||
return self.cx, self.cy
|
||
|
||
def gray_image(self, thick: int = 0, scale: int = SCALE) -> np.ndarray:
|
||
"""掩膜 -> 白底黑字灰度图 -> 放大 (thick>0 时加粗笔画, 用于多尺度集成)"""
|
||
m = self.mask
|
||
if thick:
|
||
m = cv2.dilate(m, np.ones((3, 3), np.uint8), iterations=thick)
|
||
gray = 255 - m
|
||
h, w = gray.shape
|
||
side = max(h, w) + PAD
|
||
canvas = np.full((side, side), 255, np.uint8)
|
||
canvas[(side - h) // 2:(side - h) // 2 + h, (side - w) // 2:(side - w) // 2 + w] = gray
|
||
return cv2.resize(canvas, (side * scale, side * scale), interpolation=cv2.INTER_LINEAR)
|
||
|
||
# 兼容旧接口
|
||
def ocr_image(self, thick: int = 0, scale: int = SCALE) -> Image.Image:
|
||
return Image.fromarray(self.gray_image(thick=thick, scale=scale))
|
||
|
||
|
||
class SolveResult:
|
||
"""一次求解结果"""
|
||
|
||
__slots__ = ("points", "targets", "glyphs", "score_matrix", "posterior", "assignments")
|
||
|
||
def __init__(self, points, targets, glyphs, score_matrix, posterior, assignments):
|
||
self.points = points
|
||
self.targets = targets
|
||
self.glyphs = glyphs
|
||
self.score_matrix = score_matrix
|
||
self.posterior = posterior
|
||
self.assignments = assignments
|
||
|
||
@property
|
||
def text_match_ok(self) -> bool:
|
||
"""检查指派的每个字形直接 OCR 文本是否与目标汉字一致 (杜绝图字不符或混淆干扰字盲猜提交)"""
|
||
if not self.assignments or len(self.assignments) != len(self.targets):
|
||
return False
|
||
for ti, g in enumerate(self.assignments):
|
||
target_char = self.targets[ti]
|
||
ocr_char = self.glyphs[g].ocr_text
|
||
if ocr_char and target_char not in ocr_char:
|
||
return False
|
||
return True
|
||
|
||
@property
|
||
def ok(self) -> bool:
|
||
return (len(self.points) == len(self.targets)
|
||
and len(self.points) > 0
|
||
and self.text_match_ok)
|
||
|
||
@property
|
||
def min_target_score(self) -> float:
|
||
if not self.assignments or self.score_matrix is None:
|
||
return 0.0
|
||
return min(float(self.score_matrix[g, ti]) for ti, g in enumerate(self.assignments))
|
||
|
||
def describe(self) -> str:
|
||
detail = ", ".join(
|
||
f"{t}<-{self.glyphs[g].ocr_text or '?'}@{self.glyphs[g].point}"
|
||
f"({self.score_matrix[g, ti]:.3f})"
|
||
for ti, (t, g) in enumerate(zip(self.targets, self.assignments))
|
||
)
|
||
status = "ALL_MATCHED" if self.text_match_ok else "MISMATCH"
|
||
return f"status={status} posterior={self.posterior:.3f} [{detail}]"
|
||
|
||
|
||
class TextClickSolver:
|
||
"""腾讯天御文字点选求解器 (颜色分割 + OCR 集成概率指派)"""
|
||
|
||
def __init__(self, recognizer: Optional[EnsembleOcr] = None, min_area: int = MIN_GLYPH_AREA,
|
||
variants: Sequence[Tuple[int, int]] = ((0, SCALE), (1, SCALE)),
|
||
min_glyph_score: float = 0.02):
|
||
self.min_area = min_area
|
||
self.variants = tuple(variants)
|
||
# 颜色分割会在照片的暖色区域误检出少量"伪字形", 用识别得分把它们剔除
|
||
self.min_glyph_score = min_glyph_score
|
||
self._recognizer = recognizer
|
||
|
||
@property
|
||
def recognizer(self) -> EnsembleOcr:
|
||
if self._recognizer is None:
|
||
self._recognizer = EnsembleOcr.default()
|
||
return self._recognizer
|
||
|
||
# ------------------------------------------------------------- detection
|
||
@staticmethod
|
||
def _mask_of(bgr: np.ndarray) -> np.ndarray:
|
||
hsv = cv2.cvtColor(bgr, cv2.COLOR_BGR2HSV)
|
||
mask = cv2.inRange(hsv, np.array(HSV_LOWER, np.uint8), np.array(HSV_UPPER, np.uint8))
|
||
# 闭运算填掉字内高光空洞
|
||
return cv2.morphologyEx(mask, cv2.MORPH_CLOSE,
|
||
cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5)))
|
||
|
||
def detect_glyphs(self, image: bytes | np.ndarray | Image.Image) -> List[Glyph]:
|
||
"""颜色分割定位背景图中的全部汉字"""
|
||
bgr = self._to_bgr(image)
|
||
mask = self._mask_of(bgr)
|
||
dilated = cv2.dilate(mask, cv2.getStructuringElement(cv2.MORPH_ELLIPSE,
|
||
(MERGE_KERNEL, MERGE_KERNEL)))
|
||
count, labels, stats, _ = cv2.connectedComponentsWithStats(dilated)
|
||
|
||
glyphs: List[Glyph] = []
|
||
for i in range(1, count):
|
||
if stats[i, cv2.CC_STAT_AREA] < self.min_area:
|
||
continue
|
||
sub = ((labels == i) & (mask > 0))
|
||
ys, xs = np.nonzero(sub)
|
||
if len(xs) == 0:
|
||
continue
|
||
x0, y0, x1, y1 = int(xs.min()), int(ys.min()), int(xs.max()), int(ys.max())
|
||
w, h = x1 - x0 + 1, y1 - y0 + 1
|
||
# 单字包围盒合理性: 过扁/过瘦/过小的多半是背景误检
|
||
if not (0.35 <= w / h <= 2.8) or w < 24 or h < 24:
|
||
continue
|
||
glyphs.append(Glyph(x0, y0, x1, y1,
|
||
(sub[y0:y1 + 1, x0:x1 + 1] * 255).astype(np.uint8)))
|
||
glyphs.sort(key=lambda g: (g.y0 // 64, g.x0))
|
||
return glyphs
|
||
|
||
def detect_glyphs_relaxed(self, image: bytes | np.ndarray | Image.Image) -> List[Glyph]:
|
||
"""放宽阈值的备用分割, 防止汉字偏暗或面积偏小导致的漏检"""
|
||
bgr = self._to_bgr(image)
|
||
hsv = cv2.cvtColor(bgr, cv2.COLOR_BGR2HSV)
|
||
mask = cv2.inRange(hsv, np.array((8, 85, 135), np.uint8), np.array((45, 255, 255), np.uint8))
|
||
mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (5, 5)))
|
||
dilated = cv2.dilate(mask, cv2.getStructuringElement(cv2.MORPH_ELLIPSE, (MERGE_KERNEL, MERGE_KERNEL)))
|
||
count, labels, stats, _ = cv2.connectedComponentsWithStats(dilated)
|
||
|
||
glyphs: List[Glyph] = []
|
||
for i in range(1, count):
|
||
if stats[i, cv2.CC_STAT_AREA] < 350:
|
||
continue
|
||
sub = ((labels == i) & (mask > 0))
|
||
ys, xs = np.nonzero(sub)
|
||
if len(xs) == 0:
|
||
continue
|
||
x0, y0, x1, y1 = int(xs.min()), int(ys.min()), int(xs.max()), int(ys.max())
|
||
w, h = x1 - x0 + 1, y1 - y0 + 1
|
||
if not (0.28 <= w / h <= 3.5) or w < 18 or h < 18:
|
||
continue
|
||
glyphs.append(Glyph(x0, y0, x1, y1, (sub[y0:y1 + 1, x0:x1 + 1] * 255).astype(np.uint8)))
|
||
glyphs.sort(key=lambda g: (g.y0 // 64, g.x0))
|
||
return glyphs
|
||
|
||
# ---------------------------------------------------------- recognition
|
||
def char_scores(self, glyph: Glyph, targets: Sequence[str]) -> Dict[str, float]:
|
||
"""对单个字形, 返回每个目标字的集成概率 (多预处理变体取最大)"""
|
||
acc = {t: 0.0 for t in targets}
|
||
best_text, best_conf = "", 0.0
|
||
for thick, scale in self.variants:
|
||
gray = glyph.gray_image(thick=thick, scale=scale)
|
||
scores = self.recognizer.scores(gray, targets)
|
||
for t, v in scores.items():
|
||
acc[t] = max(acc[t], v)
|
||
# 用集成里最强单引擎的读数做可读性展示
|
||
try:
|
||
engine = self.recognizer.engines[-1]
|
||
chars, probs = engine.probabilities(glyph.gray_image())
|
||
idx = probs.argmax(axis=1)
|
||
seq, prev = [], -1
|
||
for i in idx:
|
||
if i != 0 and i != prev and i < len(chars):
|
||
seq.append(chars[i])
|
||
prev = i
|
||
best_text = "".join(seq)
|
||
best_conf = float(probs.max(axis=1).mean())
|
||
except Exception: # 展示信息失败不影响求解
|
||
pass
|
||
glyph.ocr_text, glyph.confidence = best_text, best_conf
|
||
return acc
|
||
|
||
# ----------------------------------------------------------------- solve
|
||
def solve(self, image: bytes | np.ndarray | Image.Image, instruction: str,
|
||
retries: int = 1) -> List[Tuple[int, int]]:
|
||
"""返回提示顺序的像素点击坐标; 失败返回空列表 (兼容旧接口)"""
|
||
res = self.solve_ex(image, instruction, retries=retries)
|
||
return res.points if res.ok else []
|
||
|
||
def solve_ex(self, image: bytes | np.ndarray | Image.Image, instruction: str,
|
||
retries: int = 1) -> SolveResult:
|
||
"""完整求解, 附带置信度(归一化后验概率)与命中明细"""
|
||
targets = parse_instruction(instruction)
|
||
empty = SolveResult([], targets, [], np.zeros((0, 0)), 0.0, [])
|
||
if len(targets) < 2:
|
||
logger.warning("提示语解析失败: %r", instruction)
|
||
return empty
|
||
|
||
for attempt in range(retries):
|
||
glyphs = self.detect_glyphs(image)
|
||
if len(glyphs) < len(targets):
|
||
glyphs = self.detect_glyphs_relaxed(image)
|
||
if len(glyphs) < len(targets):
|
||
logger.warning("第 %d 次: 检出字形 %d 个 < 目标 %d 个",
|
||
attempt + 1, len(glyphs), len(targets))
|
||
if attempt + 1 < retries:
|
||
continue
|
||
return empty
|
||
if len(glyphs) > len(targets):
|
||
logger.info("检出字形 %d 个 (目标 %d 个), 进行全局指派", len(glyphs), len(targets))
|
||
|
||
matrix = np.array([[self.char_scores(g, targets)[t] for t in targets]
|
||
for g in glyphs], dtype=np.float64) # (n, m)
|
||
|
||
# 剔除伪字形: 任何目标字都识别不出来的候选不参与指派 (不影响后验置信度)
|
||
row_max = matrix.max(axis=1) if matrix.size else np.zeros(0)
|
||
strong = [i for i in range(len(glyphs)) if row_max[i] >= self.min_glyph_score]
|
||
if len(strong) < len(targets):
|
||
# 得分整体偏低(生僻字/罕见字形)时退化为取得分最高的 m 个
|
||
order = np.argsort(-row_max)[:len(targets)]
|
||
strong = sorted(int(i) for i in order)
|
||
logger.info("可用强字形不足, 退化为得分 Top-%d 候选", len(targets))
|
||
dropped = [i for i in range(len(glyphs)) if i not in strong]
|
||
if dropped:
|
||
logger.info("剔除伪字形 %s", [glyphs[i].point for i in dropped])
|
||
glyphs = [glyphs[i] for i in strong]
|
||
matrix = matrix[strong]
|
||
|
||
# 每个字形对目标字的归一化分布 (尺度无关), 避免某个字形整体概率偏低时被忽略
|
||
eps = 1e-9
|
||
row_sums = matrix.sum(axis=1, keepdims=True)
|
||
norm = (matrix + eps) / (row_sums + eps * matrix.shape[1])
|
||
|
||
best_perm, best_score, second_score = None, -1.0, 0.0
|
||
perms = list(itertools.permutations(range(len(glyphs)), len(targets)))
|
||
probs = []
|
||
for perm in perms:
|
||
p = 1.0
|
||
for t_idx, g_idx in enumerate(perm):
|
||
p *= float(norm[g_idx, t_idx])
|
||
probs.append(p)
|
||
if p > best_score:
|
||
second_score = best_score
|
||
best_score, best_perm = p, perm
|
||
elif p > second_score:
|
||
second_score = p
|
||
total = float(sum(probs)) or 1.0
|
||
posterior = float(best_score / total)
|
||
|
||
if best_perm is None:
|
||
return empty
|
||
logger.info("题目 %s -> %s", "".join(targets),
|
||
", ".join(f"{targets[t]}<-{glyphs[g].ocr_text or '?'}@{glyphs[g].point}"
|
||
f"({matrix[g, t]:.4f})" for t, g in enumerate(best_perm)))
|
||
logger.info(" 后验置信度=%.4f (次优/最优=%.4f)", posterior,
|
||
(second_score / best_score) if best_score > 0 else 1.0)
|
||
|
||
result = SolveResult([glyphs[g].point for g in best_perm], targets, glyphs,
|
||
matrix, posterior, list(best_perm))
|
||
if best_score <= 0:
|
||
logger.warning("所有候选得分均为 0, 判定求解失败")
|
||
return SolveResult([], targets, glyphs, matrix, 0.0, list(best_perm))
|
||
return result
|
||
return empty
|
||
|
||
# ------------------------------------------------------------- utilities
|
||
@staticmethod
|
||
def _to_bgr(image: bytes | np.ndarray | Image.Image) -> np.ndarray:
|
||
if isinstance(image, np.ndarray):
|
||
return image
|
||
if isinstance(image, Image.Image):
|
||
return cv2.cvtColor(np.array(image.convert("RGB")), cv2.COLOR_RGB2BGR)
|
||
arr = np.frombuffer(image, np.uint8)
|
||
bgr = cv2.imdecode(arr, cv2.IMREAD_COLOR)
|
||
if bgr is None:
|
||
bgr = cv2.cvtColor(np.array(Image.open(io.BytesIO(image)).convert("RGB")),
|
||
cv2.COLOR_RGB2BGR)
|
||
return bgr
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
|
||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||
solver = TextClickSolver()
|
||
img_path = sys.argv[1] if len(sys.argv) > 1 else "har/1ms_dianxuan/bg_3.jpg"
|
||
instr = sys.argv[2] if len(sys.argv) > 2 else "请依次点击：虫 呈 般"
|
||
res = solver.solve_ex(open(img_path, "rb").read(), instr)
|
||
print("points:", res.points)
|
||
print("confidence:", res.posterior)
|