★ FEATURED ARTICLE 简介面向Python开发者与数据管理人员的实战资料包围绕“爬虫技术抓取MySQL数据”这一主题系统讲解数据库连接、SQL查询、结果处理及数据入库的完整流程。配套源码包含MySQL连接封装、SQLite辅助工具、核心服务模块与API接口并附有详细说明文档和监控配置样例可帮助读者理解pymysql参数化查询防注入、批量插入性能优化等关键知识点。压缩包共17个文件以6个Python脚本和4个Markdown文档为主兼有service服务配置、db数据文件、json配置及Dockerfile容器化部署文件整体仅76KB轻量实用适合入门参考与实际工程改造。已有306人学习下载内容覆盖从环境搭建、连接配置到数据抓取入库与性能监控的常见场景按目录结构即可快速定位所需模块。1. 把网页数据抓进 MySQL这套采集方案到底解决什么问题真正跑过“mysql 爬虫”项目的人都有体会爬虫本身不难写难的是数据抓下来之后怎么稳定、不重、可续地落进 MySQL。这个资源对应的是一套典型的采集工程拆解——用 requests 发请求用 lxml 解析 HTML再用 MySQL 同时承担任务登记、去重标记和最终存储三个角色。它不是一本讲爬虫原理的书更像是一个可以直接照抄的落地脚手架建表 SQL、翻页循环、XPath 解析、增量入库、断点续抓按顺序拼起来就是一条能挂在服务器上每天跑的采集链路。适合两类人一类是刚接触爬虫、正愁数据怎么存进数据库的初学者另一类是已经在用 MySQL 存采集结果但被重复数据、连接爆掉、页面改版搞到头疼的工程师。下面从数据库设计开始一步步拆完整套流程。2. 抓取前置准备为什么选 requestslxmlMySQL 表该怎么建2.1 为什么不用 Scrapy而用 requests lxml pymysql一提到爬虫很多人的第一反应是上 Scrapy。Scrapy 自带调度器、下载中间件、Item Pipeline功能确实完整但回到“我需要把几十个列表页抓完解析字段再存进 MySQL”这个需求上Scrapy 经常是把事情做重了改一个解析规则要改 spider 类再重启任务断点续抓要依赖额外扩展项目结构也有一定学习成本。我一般会先评估任务量。如果目标页面总量是几百到几万个并发要求不高用 requests 加 lxml 的脚本方案会更直观理由有三个控制粒度更细。翻页、解析、入库都是自己写的函数想在哪一步打日志、加重试、加延时直接改代码就行不用猜框架在背后帮你做了什么。数据库连接是透明的。连接池、事务提交都在代码里显式控制出问题时能直接看到 pymysql 在哪一步抛异常不用隔着一层框架排查。部署成本低。一个 main 函数一张任务表扔到 crontab 里定时跑就能完成任务。当然如果采集规模到了百万级页面、需要多机分布式执行再迁移到 Scrapy 也不迟。这套资源的价值在于先把单机采集链路跑通后续真要扩展任务表的结构也能平滑对接分布式调度。2.2 三张核心表抓取任务表、结果数据表、请求日志表把 MySQL 放进抓取流程不是简单地给爬虫准备一个 insert 语句。我习惯把库拆成三张表各管一件事表名角色关键字段fetch_task抓取任务与状态管理url、status、retry_countfetch_detail最终业务数据content_md5 唯一键、业务字段crawl_log请求日志方便回查url、http_status、cost_ms先看抓取任务表。它存的是每个待抓页面的 URL 和处理状态是整个流程的“账本”。CREATE TABLE IF NOT EXISTS fetch_task ( id BIGINT PRIMARY KEY AUTO_INCREMENT, url VARCHAR(500) NOT NULL, source VARCHAR(50) NOT NULL DEFAULT , category VARCHAR(50) NOT NULL DEFAULT , status ENUM(pending,running,done,failed) NOT NULL DEFAULT pending, retry_count INT NOT NULL DEFAULT 0, last_error VARCHAR(500) NOT NULL DEFAULT , created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_url (url(255)) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;status 字段是整个断点续抓的核心。pending 表示待抓running 表示正在处理done 表示完成failed 表示抓取失败。retry_count 记录失败次数避免一个坏 URL 被反复尝试。url 上加了长度为 255 的前缀唯一索引重复的种子 URL 直接插不进去避免任务表自己先膨胀。接着是结果表。CREATE TABLE IF NOT EXISTS fetch_detail ( id BIGINT PRIMARY KEY AUTO_INCREMENT, task_id BIGINT NOT NULL, url VARCHAR(500) NOT NULL, title VARCHAR(200) NOT NULL DEFAULT , price DECIMAL(10,2) NOT NULL DEFAULT 0, publish_time VARCHAR(50) NOT NULL DEFAULT , category VARCHAR(50) NOT NULL DEFAULT , content_md5 CHAR(32) NOT NULL, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_md5 (content_md5) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;这里有一个值得注意的设计唯一键没用 url而是用 content_md5。原因很实际——同一个内容可能通过不同入口 URL 进入比如带 UT 参数、带来源标识url 会变但页面内容不变。用内容摘要做唯一键二次抓取时就能自动识别“这不是新数据”从源头挡住重复入库。publish_time 用 VARCHAR 而不是 DATETIME是因为很多站点的时间格式不规范直接入库会报错先存字符串保留原始表现后续需要排序再处理。crawl_log 表不展开 DDL核心字段就是 url、http_status、cost_ms、抓取时间。它的用途是事后排查页面卡住了看日志表里最后几条记录就能定位到具体 URL而不是靠猜。2.3 连接池为什么每次请求都新建连接会出事很多爬虫脚本翻车不是挂在反爬而是挂在 MySQL 连接上。如果每个详情页都 pymysql.connect() 一次短任务跑完没问题一旦任务量上来高频握手会拖慢速度更危险的是连接异常断开后MySQL 服务端可能还保留着这条连接直到 wait_timeout 到期。跑几万条数据之后“Too many connections” 几乎是必然结果。常见做法是用 DBUtils 的 PooledDB 维护一个连接池让几个连接反复复用。from dbutils.pooled_db import PooledDB import pymysql POOL PooledDB( creatorpymysql, maxconnections10, mincached2, maxcached5, blockingTrue, host127.0.0.1, port3306, usercrawler_user, passwordyour_pass, databasecrawler_db, charsetutf8mb4, autocommitTrue ) def get_conn(): return POOL.connection()参数意义很直接maxconnections 是连接池上限10 个连接对单机采集足够blockingTrue 表示连接不够用时让请求排队等待而不是直接抛异常charset 必须用 utf8mb4因为很多网页里有冷门汉字或 emojiutf8 会有概率写不进去。这里设置 autocommitTrue 是为了让单条写操作省去手动提交的麻烦。但第 3 章做批量入库时我会把它配合事务一起用两者的侧重点不同。3. 抓取主流程翻页循环、XPath 解析与增量入库3.1 从任务表取种子 URL组装翻页循环抓取开始前先从 fetch_task 表里捞一批 pending 的任务。这里有个细节一次不要取太多取 20 条左右就够了。因为任务会在处理过程中被更新状态分批取相当于给整个流程加了检查点中间崩了重启没处理完的任务还是 pending不会被漏掉。def load_pending_tasks(conn, limit20): with conn.cursor() as cur: cur.execute( SELECT id, url, category FROM fetch_task WHERE statuspending AND retry_count 3 LIMIT %s, (limit,) ) return cur.fetchall()retry_count 3 是硬上限。有些 URL 本身已经失效比如页面被删除、域名跳转异常反复重试只会把时间耗在无效请求上还会提高触发反爬的概率。三次抓不到就标记 failed留给人工确认。拿到种子 URL 之后处理翻页。大多数信息站点的列表页都是 ?pageN 这种结构翻到最后一页会出现两种情况一是返回 404二是返回 200 但页面里没有有效条目。所以循环停止条件不能只看 HTTP 状态码还要看解析出来的条目数。def crawl_list(session, base_url, max_page5): all_items [] for page in range(1, max_page 1): url f{base_url}?page{page} html fetch_html(session, url) if html is None: break page_items parse_list_page(html) if not page_items: break all_items.extend(page_items) time.sleep(random.uniform(0.5, 1.5)) return all_items逐行看逻辑fetch_html 负责发请求并返回可解析的 HTML 文本parse_list_page 解析当页条目返回字典列表。列表为空就 break说明已经翻到最后一页。sleep 加随机区间而不是固定值是为了让请求间隔更像真实访问节奏。固定间隔一旦被统计出来就是反爬系统里最显眼的特征。3.2 fetch_html 的请求与编码处理列表页请求是整个链路的入口Header 和编码处理都在这一步做先把这个函数补全。import requests def fetch_html(session, url): headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Accept-Language: zh-CN,zh;q0.9, } try: resp session.get(url, headersheaders, timeout10) if resp.status_code ! 200: return None resp.encoding resp.apparent_encoding return resp.text except requests.RequestException: return None两点要说明。第一session 用 requests.Session() 创建后传入Session 会复用底层 TCP 连接比每次 requests.get 重新建连快不少。第二resp.encoding resp.apparent_encoding 是关键requests 默认会从响应头取编码但很多站点响应头里没有 charset直接从 HTML meta 里推断才准。这一行不写第 4 章要讲的乱码问题会大面积出现。timeout10 也是必须的。有些站点单页响应慢几十秒不返回如果不设超时脚本会一直挂在请求上整个任务队列被一个坏页面堵死。3.3 XPath 解析与字段清洗拿到 HTML 文本后用 lxml 转成可查询的结构。解析的重点是容错页面里偶尔会有字段缺失的项不能因为一个卡片少了个价格就让整个任务崩溃。from lxml import etree import re def parse_list_page(html): root etree.HTML(html) cards root.xpath(//div[contains(class,item)]) rows [] for card in cards: title (card.xpath(.//a[contains(class,title)]/text()) or [])[0].strip() price_text (card.xpath(.//span[contains(class,price)]/text()) or [])[0].strip() href (card.xpath(.//a[contains(class,title)]/href) or [])[0] if not title or not href: continue rows.append({ title: title, price: parse_price(price_text), href: href }) return rows def parse_price(text): m re.search(r(\d(?:\.\d{1,2})?), text.replace(,, )) return float(m.group(1)) if m else 0.0这里面有两个容易踩的细节。第一个是 XPath 表达式末尾的 or []如果某个卡片没有 title 节点xpath 返回空列表直接下标取 [0] 会抛 IndexError用空列表兜底后取到空字符串。第二个是 title 和 href 都为空时直接 continue这种不完整的卡片多半是广告位或特殊样式没有入库价值。parse_price 的正则只匹配数字和小数点把“价格面议”“¥1,200/月”里的有效数字提取出来提取失败就返回 0.0。价格这类字段不要试图用 if else 去穷举所有格式正则加默认值就够用。3.4 批量入库与 ON DUPLICATE KEY UPDATE解析完成后数据要写进 fetch_detail。这里最忌讳的做法是一条一条 insert几千条数据会产生几千次网络往返。正确姿势是拼成批量参数用 executemany 一次性交给 MySQL 执行。import hashlib def save_details(conn, rows): sql INSERT INTO fetch_detail (task_id, url, title, price, publish_time, category, content_md5) VALUES (%s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE titleVALUES(title), priceVALUES(price), updated_atNOW() data [] for r in rows: md5 hashlib.md5( (r[title] r[publish_time]).encode(utf-8) ).hexdigest() data.append(( r[task_id], r[url][:500], r[title][:200], r[price], r[publish_time][:50], r[category][:50], md5 )) with conn.cursor() as cur: cur.executemany(sql, data) conn.commit()ON DUPLICATE KEY UPDATE 是这套方案的精华。插入时如果 content_md5 已经存在MySQL 不会报错也不会新增重复行而是原地更新 title 和 price相当于把“重新抓取”自动降级为“数据刷新”。第二次、第三次跑同一个任务表里始终只有一条最新记录。content_md5 的生成规则是 title 加 publish_time 拼起来做摘要。这是内容去重的简化做法够用且快。如果要更严格可以把正文关键字段也拼进去。关于 VALUES() 函数有个提醒MySQL 8.0.20 起标记为废弃建议用行别名语法。但 5.7 和 8.0 早期版本依然支持兼容性没问题。如果数据是纯新增场景去掉 ON DUPLICATE KEY UPDATE 那三行也能用只是无法处理重跑时的去重。task_id、url 这些字段为什么要截断因为表结构里字段有长度限制超长字符串直接入库会报 Data too long。解析出来的数据不像手工录入那么规整入库前做一次边界裁剪是批量写入的基本习惯。4. 常见抓取问题与排查五个高频坑的记录4.1 标题乱码问号还是 Unicode 转义现象抓回来的标题在 MySQL 里显示成“??????”或者一整串反斜杠转义字符英文正常中文全乱。原因requests 请求时没有正确识别网页编码。很多服务器响应头里不写 charsetrequests 会默认用 ISO-8859-1 或 UTF-8 去解码遇到 GBK 编码的网页就会解出乱码。另一种情况是页面里嵌了 JavaScript 动态数据抓回来的是 unicode 转义串而不是可见中文。解决统一在 fetch_html 里加 resp.encoding resp.apparent_encoding让 requests 根据网页内容自动推断编码。如果发现 apparent_encoding 也不准可以从 HTML 源码里手动匹配 meta charset。入库前再用 print 抽样打印几条确认标题是正常中文再放量跑。4.2 请求返回 403 或 429反爬识别的信号现象脚本跑前几分钟一切正常突然连续返回 403 Forbidden任务表里瞬间躺了一大批 failed 记录。原因请求频率太高或 User-Agent 特征太明显。站点统计到同一 IP 短时间高频访问触发了访问频率限制。429 状态码表示请求过多你的请求被限流了。解决降低单页抓取频率把 sleep 间隔从 0.1 秒调到 1 秒以上并且用随机区间打散节奏。User-Agent 不要只用默认的 python-requests换成常见浏览器的 UA并在任务开始时随机抽一个。再配合 Referer 填充让请求看起来像是从列表页点进详情的正常跳转。这里不建议一上来就想着换出口 IP那属于高风险操作先把频率和请求头做好大部分站点都能过。4.3 页面结构升级XPath 突然解析不到数据现象某天开始入库行数直接掉到零日志里 HTTP 状态码都正常但 parse_list_page 返回的 rows 一直是空列表。原因网站前端改版了class 名换掉或 DOM 层级变化。XPath 表达式写的是死路径页面结构一变路径就断了。问题在于它不报错只是静默地返回空结果不盯日志很难发现。解决在解析函数里统计空数据比例。如果连续三条列表页都没解析到任何条目说明不是自然翻页到底而是解析器失效直接发告警或者把当前 HTML 存到本地文件等排查。XPath 写法上尽量用稳定的属性比如>def process_task(conn, task_id): with conn.cursor() as cur: cur.execute( UPDATE fetch_task SET statusrunning, updated_atNOW() WHERE id%s AND statuspending, (task_id,) ) conn.commit()先把它标记为 running再去抓取和入库。这一步放在请求前而不是请求后就是为了防止重复占用如果两个进程同时取到同一个任务只有先更新成功的那一个能继续处理另一个 UPDATE 影响行数为 0自动放弃。任务完成后标记 donedef mark_done(conn, task_id): with conn.cursor() as cur: cur.execute( UPDATE fetch_task SET statusdone, updated_atNOW() WHERE id%s, (task_id,) ) conn.commit()脚本重启后只要把残留的 running 状态重置回 pending 即可def recover_interrupted(conn): with conn.cursor() as cur: cur.execute( UPDATE fetch_task SET statuspending WHERE statusrunning ) conn.commit()这一步放在 main 函数最开头。为什么 running 状态要重置而不是直接当成失败因为进程被强杀时任务可能只跑了一半数据库没来得及写入但它也不是确定失败。重置回 pending让它重新走一遍处理逻辑最安全。5.2 用 SQL 做采集完整性验证任务跑完或者跑了一部分时不要只盯着成功日志用 SQL 抽样对比更直观。SELECT COUNT(*) AS total, COUNT(DISTINCT content_md5) AS unique_cnt FROM fetch_detail;total 和 unique_cnt 一致说明没有重复数据不一致就去查分组排名靠前的 md5。SELECT status, COUNT(*) AS cnt FROM fetch_task GROUP BY status;这个语句一次性看到任务全景pending 剩多少、done 多少、failed 多少。failed 比例超过 5% 就该检查站点是否改版或者反爬升级而不是继续放量跑。上线的最后一步我习惯做一次空跑验证先把任务表清空或备份只放十条真实目标 URL跑完整链路确认解析条数正确、第二次重跑不产生重复行、状态表里所有任务都变成 done再放全量数据。从那以后我每次上线新的采集任务都强制先走这一遍空跑流程看起来多花十分钟实际省掉的是排查几千条脏数据的半天时间。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
