Python图像还原实战：破解滑块验证码的撕裂图片重组算法

最近在分析某个主流安全平台的滑块验证码时，发现其前端会将验证图片切割成16个碎片传输。这种设计不仅增加了直接爬取的难度，也为自动化识别带来了新的挑战。本文将分享如何通过Python的Pillow库，完整实现这类撕裂图片的还原算法。

1. 理解滑块验证码的图片切割机制

现代滑块验证系统普遍采用前端切割技术来防止直接图片抓取。以某安全平台为例，其验证码图片会被分割成16个不规则的碎片，这些碎片通过不同的网络请求分别加载。这种设计主要有两个目的：

1. 增加直接爬取完整图片的难度 ：需要处理多个异步请求
2. 防止简单的OCR识别 ：切割后的单个碎片无法提供完整信息

通过浏览器开发者工具分析网络请求，我们发现图片碎片具有以下特征：

• 每张碎片尺寸相同（如80×80像素）
• 碎片按4×4网格排列
• 部分碎片可能经过旋转或镜像处理
• 每个碎片有唯一的定位标识符

# 示例：观察到的碎片命名规律
fragment_names = [
    "frag_0_0.png", "frag_0_1.png", "frag_0_2.png", "frag_0_3.png",
    "frag_1_0.png", "frag_1_1.png", "frag_1_2.png", "frag_1_3.png",
    # ... 共16个碎片
]

2. 逆向分析前端还原逻辑

要正确重组图片，首先需要理解前端JavaScript如何处理这些碎片。通过Chrome开发者工具的调试功能，我们可以定位到关键的图片重组函数。

2.1 定位关键JS函数

在浏览器中加载验证码页面，使用以下步骤定位还原逻辑：

1. 在Network面板过滤图片请求（类型：img）
2. 找到任意碎片图片的请求，右键选择"Open in Sources panel"
3. 在Sources面板设置断点，观察图片加载过程
4. 跟踪调用栈，找到负责碎片组合的函数

通常这类函数会具有以下特征：

• 接收碎片URL数组作为输入
• 包含Canvas相关操作
• 有明确的碎片位置计算逻辑

2.2 分析还原算法

通过逆向分析，我们发现典型的还原算法包含以下步骤：

1. 创建画布 ：初始化一个足够大的Canvas元素
2. 碎片排序 ：根据碎片ID确定其在网格中的位置
3. 坐标计算 ：计算每个碎片在最终图片中的(x,y)偏移量
4. 绘制碎片 ：使用Canvas的drawImage方法组合碎片

// 示例：观察到的JS还原逻辑片段
function combineFragments(fragments) {
    const canvas = document.createElement('canvas');
    canvas.width = 320;  // 4*80
    canvas.height = 320; // 4*80
    
    const ctx = canvas.getContext('2d');
    fragments.forEach(frag => {
        const [row, col] = parseFragmentId(frag.id);
        const x = col * 80;
        const y = row * 80;
        ctx.drawImage(frag.image, x, y);
    });
    
    return canvas.toDataURL();
}

3. Python实现图片重组算法

基于前端逻辑分析，我们可以用Python实现等效的图片重组功能。这里使用Pillow库（PIL的分支）来处理图像操作。

3.1 基础环境准备

首先安装必要的依赖：

pip install pillow requests

3.2 核心重组算法实现

from PIL import Image
import os

def combine_fragments(fragment_dir, output_path, grid_size=(4,4), fragment_size=(80,80)):
    """
    组合切割的图片碎片
    
    :param fragment_dir: 碎片图片所在目录
    :param output_path: 输出图片路径
    :param grid_size: 网格尺寸(行,列)
    :param fragment_size: 每个碎片尺寸(宽,高)
    """
    total_width = grid_size[1] * fragment_size[0]
    total_height = grid_size[0] * fragment_size[1]
    
    # 创建空白画布
    combined = Image.new('RGB', (total_width, total_height))
    
    # 遍历网格位置
    for row in range(grid_size[0]):
        for col in range(grid_size[1]):
            # 构造预期文件名（根据实际命名规则调整）
            fragment_path = os.path.join(fragment_dir, f'frag_{row}_{col}.png')
            
            if os.path.exists(fragment_path):
                fragment = Image.open(fragment_path)
                # 计算碎片在组合图中的位置
                x = col * fragment_size[0]
                y = row * fragment_size[1]
                # 将碎片粘贴到组合图
                combined.paste(fragment, (x, y))
            else:
                print(f'警告：缺少碎片 {fragment_path}')
    
    combined.save(output_path)
    return combined

3.3 处理特殊变形情况

某些高级验证系统会对碎片进行额外处理以增加还原难度。以下是几种常见变形及应对方法：

情况1：碎片旋转

# 在粘贴前旋转碎片
if should_rotate(fragment_path):  # 需要实现判断逻辑
    fragment = fragment.rotate(90)  # 旋转90度

情况2：碎片镜像

# 水平镜像
fragment = fragment.transpose(Image.FLIP_LEFT_RIGHT)
# 垂直镜像
fragment = fragment.transpose(Image.FLIP_TOP_BOTTOM)

情况3：碎片重叠

# 使用alpha通道处理重叠
for fragment in fragments:
    combined.alpha_composite(fragment, dest=(x,y))

4. 完整工作流程实现

将上述分析整合为一个完整的图片还原解决方案：

import os
import requests
from PIL import Image
from io import BytesIO

class SliderImageReconstructor:
    def __init__(self, grid_size=(4,4), fragment_size=(80,80)):
        self.grid_size = grid_size
        self.fragment_size = fragment_size
        self.total_width = grid_size[1] * fragment_size[0]
        self.total_height = grid_size[0] * fragment_size[1]
    
    def download_fragment(self, url, save_path=None):
        """下载单个碎片图片"""
        response = requests.get(url, headers={'User-Agent': 'Mozilla/5.0'})
        if response.status_code == 200:
            img = Image.open(BytesIO(response.content))
            if save_path:
                img.save(save_path)
            return img
        return None
    
    def reconstruct_image(self, fragment_urls, output_path):
        """
        从碎片URL列表重建完整图片
        
        :param fragment_urls: 碎片URL字典，格式为 {(row,col): url}
        :param output_path: 输出图片路径
        """
        combined = Image.new('RGB', (self.total_width, self.total_height))
        
        for (row, col), url in fragment_urls.items():
            try:
                fragment = self.download_fragment(url)
                if fragment:
                    x = col * self.fragment_size[0]
                    y = row * self.fragment_size[1]
                    combined.paste(fragment, (x, y))
            except Exception as e:
                print(f'处理碎片({row},{col})出错: {str(e)}')
        
        combined.save(output_path)
        return combined

# 使用示例
if __name__ == '__main__':
    # 示例碎片URL映射 (需要根据实际情况获取)
    fragment_urls = {
        (0,0): 'https://example.com/frag_0_0.png',
        (0,1): 'https://example.com/frag_0_1.png',
        # ... 填充所有16个碎片的URL
    }
    
    reconstructor = SliderImageReconstructor()
    result = reconstructor.reconstruct_image(
        fragment_urls,
        output_path='restored_image.jpg'
    )
    print('图片重组完成，保存为 restored_image.jpg')

5. 常见问题与调试技巧

在实际应用中，可能会遇到以下典型问题：

5.1 碎片顺序错乱

症状 ：重组后的图片明显错位

解决方案 ：

1. 检查碎片命名规则是否与代码假设一致
2. 添加调试输出，打印每个碎片的(row,col)和实际坐标
3. 使用中间可视化检查：

# 在重组循环中添加调试标记
for (row, col), url in fragment_urls.items():
    fragment = self.download_fragment(url)
    if fragment:
        # 绘制位置标记
        draw = ImageDraw.Draw(fragment)
        draw.text((5,5), f'{row},{col}', fill='red')
        # ... 粘贴操作

5.2 碎片尺寸不一致

症状 ：Pillow抛出尺寸不匹配异常

解决方案 ：

1. 统一调整碎片尺寸：

fragment = fragment.resize(self.fragment_size)

2. 或者修改重组逻辑适应不同尺寸：

# 动态计算总尺寸
total_width = max((col+1)*frag.width for (row,col),frag in fragments.items())
total_height = max((row+1)*frag.height for (row,col),frag in fragments.items())

5.3 网络请求限制

症状 ：部分碎片下载失败

解决方案 ：

1. 添加重试机制：

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def download_fragment_with_retry(self, url):
    return self.download_fragment(url)

2. 使用代理IP轮询：

proxies = ['http://proxy1:port', 'http://proxy2:port']
current_proxy = 0

def get_with_proxy(url):
    global current_proxy
    proxy = {'http': proxies[current_proxy]}
    current_proxy = (current_proxy + 1) % len(proxies)
    return requests.get(url, proxies=proxy, timeout=10)

6. 高级优化技巧

对于需要处理大量验证码的场景，可以考虑以下优化：

6.1 并行下载碎片

from concurrent.futures import ThreadPoolExecutor

def download_all_fragments(self, fragment_urls):
    """并行下载所有碎片"""
    with ThreadPoolExecutor(max_workers=8) as executor:
        future_to_pos = {
            executor.submit(self.download_fragment, url): (row,col)
            for (row,col), url in fragment_urls.items()
        }
        
        fragments = {}
        for future in concurrent.futures.as_completed(future_to_pos):
            pos = future_to_pos[future]
            try:
                fragments[pos] = future.result()
            except Exception as e:
                print(f'下载碎片{pos}失败: {str(e)}')
        
        return fragments

6.2 缓存已下载碎片

from functools import lru_cache

@lru_cache(maxsize=100)
def download_fragment_cached(self, url):
    """带缓存的碎片下载"""
    return self.download_fragment(url)

6.3 自动识别网格参数

对于不确定切割参数的情况，可以实现自动检测：

def detect_grid_params(self, sample_fragments):
    """通过样本碎片检测网格参数"""
    widths = {img.width for img in sample_fragments.values()}
    heights = {img.height for img in sample_fragments.values()}
    
    if len(widths) != 1 or len(heights) != 1:
        raise ValueError("碎片尺寸不一致")
    
    frag_width = widths.pop()
    frag_height = heights.pop()
    
    # 通过碎片命名推测网格大小
    max_row = max(row for (row,col) in sample_fragments.keys())
    max_col = max(col for (row,col) in sample_fragments.keys())
    
    return {
        'grid_size': (max_row+1, max_col+1),
        'fragment_size': (frag_width, frag_height)
    }

7. 安全与伦理考量

在实现这类技术时，必须注意：

1. 遵守目标网站的服务条款 ：未经授权抓取可能违反规定
2. 限制请求频率 ：避免对目标服务器造成过大负担
3. 明确使用目的 ：仅用于学习研究和授权测试
4. 数据最小化原则 ：只获取必要数据，及时删除不需要的信息

# 良好的爬虫公民应包含的请求头
HEADERS = {
    'User-Agent': 'ResearchBot/1.0 (+https://example.com/bot-info)',
    'Accept-Language': 'en-US,en;q=0.9',
    'Accept-Encoding': 'gzip, deflate',
    'Connection': 'keep-alive',
    'Referer': 'https://example.com/',
    'DNT': '1'  # Do Not Track
}

def ethical_download(self, url):
    """遵守规范的下载方法"""
    response = requests.get(
        url,
        headers=HEADERS,
        timeout=10,
        stream=True
    )
    # 添加延迟避免请求过频
    time.sleep(1)
    return response