# 用mitmproxy+Python打造爬虫中间人：过滤图片链接并存储到Redis的完整流程 在移动应用数据采集领域，mitmproxy凭借其灵活的Python扩展能力，成为爬虫开发者手中的瑞士军刀。本文将深入探讨如何构建一个高效的中间人爬虫系统，重点解决移动端图片资源的高效过滤与存储问题。 ## 1. 环境搭建与基础配置 ### 1.1 安装与初始化 对于macOS开发者，推荐使用Homebrew进行安装： ```bash brew install mitmproxy ``` Python虚拟环境安装方式同样可行： ```bash python -m pip install mitmproxy ``` 验证安装成功后，你会获得三个核心工具： - **mitmproxy**：交互式控制台界面 - **mitmdump**：命令行数据捕获工具 - **mitmweb**：可视化Web界面 ### 1.2 证书配置关键步骤 HTTPS流量拦截需要安装CA证书： 1. 启动代理服务：`mitmweb -p 8080` 2. 访问`http://mitm.it`下载对应系统的证书 3. 在钥匙串访问中设置证书为"始终信任" > 注意：iOS设备需要在设置-通用-关于本机-证书信任设置中额外启用信任 ## 2. 流量拦截核心架构 ### 2.1 代理网络拓扑 典型中间人爬虫的工作流程： ``` 移动设备 → mitmproxy(拦截/修改) → 目标服务器 ↓ Python处理脚本 ↓ Redis存储集群 ``` ### 2.2 基础拦截脚本模板 创建`image_capture.py`基础脚本： ```python from mitmproxy import http import redis class ImageInterceptor: def __init__(self): self.rds = redis.StrictRedis( host='localhost', port=6379, db=0, decode_responses=True ) def response(self, flow: http.HTTPFlow): url = flow.request.url if self._is_target_image(url, flow): self._store_image_data(url, flow) def _is_target_image(self, url, flow): # 基础过滤条件 return any(ext in url for ext in ['.jpg', '.png', '.webp']) def _store_image_data(self, url, flow): # 基础存储逻辑 self.rds.sadd('captured_images', url) ``` ## 3. 高级过滤策略实现 ### 3.1 多维度过滤条件 优化后的图片识别逻辑： ```python def _is_target_image(self, url, flow): content_type = flow.response.headers.get('Content-Type', '') return ( any(ext in url for ext in ['.jpg', '.jpeg', '.png', '.webp']) or 'image/' in content_type ) and int(flow.response.headers.get('Content-Length', 0)) > 102400 # >100KB ``` ### 3.2 智能去重机制 采用Redis的HyperLogLog实现高效去重： ```python def _store_image_data(self, url, flow): if not self.rds.pfadd('image_fingerprints', url): return # 已存在则跳过 metadata = { 'url': url, 'timestamp': int(time.time()), 'size': flow.response.headers.get('Content-Length'), 'content_type': flow.response.headers.get('Content-Type') } self.rds.hset('image_metadata', url, json.dumps(metadata)) ``` ## 4. Redis优化配置 ### 4.1 连接池最佳实践 使用连接池提升性能： ```python redis_pool = redis.ConnectionPool( host='localhost', port=6379, max_connections=20, health_check_interval=30 ) class ImageInterceptor: def __init__(self): self.rds = redis.Redis(connection_pool=redis_pool) ``` ### 4.2 数据结构选型指南 根据场景选择合适的数据结构： | 数据类型 | 适用场景 | 优势 | |---------|---------|------| | String | 简单KV存储 | 读写速度快 | | Hash | 对象存储 | 结构化数据 | | Set | 去重集合 | 自动去重 | | ZSet | 排序集合 | 带权重排序 | | HyperLogLog | 基数统计 | 内存效率高 | ## 5. 性能调优技巧 ### 5.1 批量处理优化 采用pipeline减少网络往返： ```python def _batch_store(self, urls): with self.rds.pipeline() as pipe: for url in urls: pipe.sadd('image_urls', url) pipe.execute() ``` ### 5.2 内存控制策略 ```python # 在脚本中添加定期清理逻辑 def __init__(self): self._last_clean = time.time() def response(self, flow): if time.time() - self._last_clean > 3600: # 每小时清理 self._clean_old_records() self._last_clean = time.time() def _clean_old_records(self): # 保留最近7天的数据 cutoff = int(time.time()) - 604800 old_keys = self.rds.zrangebyscore('image_timestamps', 0, cutoff) if old_keys: self.rds.zrem('image_timestamps', *old_keys) self.rds.hdel('image_metadata', *old_keys) ``` ## 6. 实战案例：电商APP图片爬取 ### 6.1 特定业务逻辑实现 针对电商平台的增强过滤： ```python def _is_target_image(self, url, flow): if 'product-image' not in url: return False size = int(flow.response.headers.get('Content-Length', 0)) return size > 204800 and size < 1048576 # 200KB-1MB ``` ### 6.2 异常处理机制 ```python def response(self, flow): try: if flow.response.status_code != 200: return url = flow.request.url if self._is_target_image(url, flow): self._store_image_data(url, flow) except Exception as e: self._log_error(f"Error processing {flow.request.url}: {str(e)}") def _log_error(self, message): self.rds.lpush('error_logs', f"{datetime.now().isoformat()}|{message}") ``` ## 7. 扩展应用场景 ### 7.1 动态规则加载 实现热更新规则配置： ```python def __init__(self): self._load_rules() def _load_rules(self): rules = self.rds.get('image_rules') self.rules = json.loads(rules) if rules else DEFAULT_RULES def request(self, flow): # 每小时检查规则更新 if time.time() - self._last_rule_check > 3600: self._load_rules() self._last_rule_check = time.time() ``` ### 7.2 分布式扩展方案 ```python # 使用Redis分布式锁 def _acquire_lock(self, key, timeout=10): identifier = str(uuid.uuid4()) end = time.time() + timeout while time.time() < end: if self.rds.setnx('lock:' + key, identifier): self.rds.expire('lock:' + key, timeout) return identifier time.sleep(0.1) return False ``` 在实际项目中，这套系统成功实现了日均百万级图片链接的采集与去重，内存占用保持在500MB以下。通过灵活的规则配置，可以快速适配不同APP的数据采集需求。