1. 项目背景与核心需求最近在做一个图片资源归档项目需要定期从特定网站抓取最新图片。经过多次尝试发现mm131这个站点的资源质量稳定且更新及时但手动下载效率实在太低。于是决定用Python写个自动化脚本既能高效抓取图片又能自动识别新增内容。这个爬虫需要解决三个核心问题首先是高效抓取要能在短时间内完成大量图片下载其次是稳定性要能应对网站的反爬机制最后是可持续性要能自动识别新内容并增量更新。下面我就把实战中总结的经验分享给大家这个方案已经稳定运行了半年多日均抓取上千张图片从没翻车。2. 环境准备与基础配置2.1 必备工具清单工欲善其事必先利其器先准备好这些工具Python 3.8我用的3.9.7版本最稳定Requests库处理HTTP请求BeautifulSoup4解析HTMLlxml加速解析tqdm显示进度条安装命令很简单pip install requests beautifulsoup4 lxml tqdm2.2 反爬应对策略现在网站都有基础反爬措施我们需要做好这些准备用户代理轮换准备10个常见浏览器的User-Agent请求间隔控制随机延迟1-3秒代理IP池可选如果是大规模抓取建议配置Referer设置模拟从站内跳转这是我的headers模板实测有效headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: https://www.mm131.net/ }3. 网页结构分析与数据提取3.1 分页机制解析首先分析列表页规律比如首页是https://www.mm131.net/xinggan/翻页后变成https://www.mm131.net/xinggan/list_1_2.html可以看出分页参数在list_1_2.html中其中1是分类ID2是页码。3.2 图片链接提取技巧通过开发者工具检查元素发现图集链接藏在article标签里article a href/xinggan/1234.html title清纯写真 img srchttps://img.mm131.net/pic/1234/1.jpg /article用BeautifulSoup可以这样提取soup.find_all(article) for item in articles: title item.a[title] url item.a[href]3.3 图片详情页抓取进入图集详情页后发现图片URL有规律https://img.mm131.net/pic/1234/1.jpg https://img.mm131.net/pic/1234/2.jpg ...只需要替换最后的数字就能获取所有图片。4. 核心爬虫代码实现4.1 分页抓取模块先写个分页抓取函数def get_page_list(category, max_page): base_url fhttps://www.mm131.net/{category}/ for page in range(1, max_page1): url f{base_url}list_1_{page}.html if page 1 else base_url response requests.get(url, headersheaders) soup BeautifulSoup(response.text, lxml) parse_album_list(soup) # 解析图集列表 time.sleep(random.uniform(1, 3)) # 随机延迟4.2 图片下载函数下载函数要考虑重试机制def download_image(url, save_path): for retry in range(3): # 最多重试3次 try: resp requests.get(url, headersheaders, streamTrue) if resp.status_code 200: with open(save_path, wb) as f: for chunk in resp: f.write(chunk) return True except Exception as e: print(f下载失败重试 {retry1}/3) time.sleep(2) return False4.3 自动创建目录结构按分类日期自动创建目录def create_folder(category): today datetime.now().strftime(%Y%m%d) path f./downloads/{category}/{today}/ if not os.path.exists(path): os.makedirs(path) return path5. 自动化更新策略5.1 增量更新机制每次运行先读取已下载记录def get_downloaded_ids(): try: with open(downloaded.json, r) as f: return set(json.load(f)) except FileNotFoundError: return set()下载完成后更新记录def update_record(album_id): downloaded get_downloaded_ids() downloaded.add(album_id) with open(downloaded.json, w) as f: json.dump(list(downloaded), f)5.2 定时任务配置用APScheduler设置每日定时运行from apscheduler.schedulers.blocking import BlockingScheduler sched BlockingScheduler() sched.scheduled_job(cron, hour2) # 每天凌晨2点运行 def daily_job(): main() sched.start()6. 异常处理与日志记录6.1 常见错误处理针对不同错误采取不同策略try: response requests.get(url, timeout10) except requests.exceptions.Timeout: print(请求超时) except requests.exceptions.TooManyRedirects: print(重定向过多) except requests.exceptions.RequestException as e: print(f严重错误{e})6.2 日志系统配置使用logging模块记录运行情况import logging logging.basicConfig( filenamespider.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) logging.info(开始下载图集 %s, album_title) logging.warning(图片 %s 下载失败, image_url)7. 性能优化技巧7.1 多线程加速使用线程池提高下载速度from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor(max_workers5) as executor: futures [] for img_url in image_urls: futures.append(executor.submit(download_image, img_url, save_path)) for future in tqdm(as_completed(futures), totallen(futures)): pass # 等待所有任务完成7.2 内存优化对于大文件使用流式下载response requests.get(url, streamTrue) with open(path, wb) as f: for chunk in response.iter_content(chunk_size8192): if chunk: # 过滤keep-alive数据块 f.write(chunk)8. 完整代码示例最后给出整合后的核心代码import os import json import time import random import requests from bs4 import BeautifulSoup from datetime import datetime from tqdm import tqdm from concurrent.futures import ThreadPoolExecutor class MM131Spider: def __init__(self): self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: https://www.mm131.net/ } self.downloaded self.load_records() def run(self, categoryxinggan, max_page3): for page in range(1, max_page1): self.crawl_page(category, page) def crawl_page(self, category, page): url self.build_url(category, page) soup self.get_soup(url) albums soup.find_all(article) with ThreadPoolExecutor(max_workers5) as executor: list(tqdm( executor.map(self.process_album, albums), totallen(albums), descf第{page}页进度 )) def process_album(self, album): album_url album.a[href] if album_url in self.downloaded: return images self.get_album_images(album_url) save_dir self.create_save_dir(album.a[title]) for idx, img_url in enumerate(images, 1): self.download_image(img_url, save_dir, idx) self.update_records(album_url) # 其他辅助方法...这套代码已经处理了大多数异常情况在实际使用中可以根据需要调整线程数、延迟时间等参数。记得要合理控制请求频率建议每秒钟不超过2次请求避免给目标网站造成过大压力。