1. Python正则匹配与网络爬取实战指南正则表达式和网络爬取是Python开发者必须掌握的两项核心技能。我在过去五年的爬虫项目实战中发现90%的数据提取问题都可以通过合理的正则匹配解决而剩余10%的复杂场景往往需要结合XPath或CSS选择器完成。本文将从一个爬虫工程师的角度分享如何高效运用正则表达式进行网页内容抓取与数据清洗。2. 正则表达式基础与Python实现2.1 正则表达式核心语法精要Python通过re模块提供正则支持以下是最常用的元字符及其应用场景import re # 基础匹配模式 pattern r\d{3}-\d{4} # 匹配如123-4567的格式 text 我的电话是123-4567 match re.search(pattern, text) print(match.group()) # 输出: 123-4567关键元字符速查表元字符说明示例.匹配任意字符(除换行)a.c 匹配 abc,a c\d数字字符 [0-9]\d 匹配 123\w单词字符 [a-zA-Z0-9_]\w 匹配 hello_123\s空白字符\s 匹配空格/tab*0次或多次重复a* 匹配 ,a,aaa1次或多次重复\d 匹配 1,12?0次或1次colou?r 匹配 color/colour{m,n}m到n次重复\d{3,5} 匹配 123/12345提示在Python中始终使用原始字符串(r前缀)定义正则模式避免转义字符的混淆2.2 Python re模块高级用法除了基础的search/matchre模块还提供# 分组提取 html div classtitlePython正则指南/div result re.search(rdiv class([^])(.?)/div, html) print(result.groups()) # 输出: (title, Python正则指南) # 非贪婪匹配(重要!) greedy re.search(rdiv.*/div, diva/divdivb/div) print(greedy.group()) # 匹配整个字符串 non_greedy re.search(rdiv.*?/div, diva/divdivb/div) print(non_greedy.group()) # 仅匹配第一个div # 预编译正则对象 phone_re re.compile(r(\d{3})-(\d{4})) matches phone_re.findall(电话:123-4567, 备用:890-1234) print(matches) # [(123,4567), (890,1234)]3. 网络爬取实战技巧3.1 请求处理与反爬策略使用requests库时需要注意的关键点import requests from fake_useragent import UserAgent headers { User-Agent: UserAgent().random, Accept-Language: zh-CN,zh;q0.9 } proxies { http: http://proxy.example.com:8080, https: https://proxy.example.com:8080 } try: response requests.get( https://example.com/api, headersheaders, proxiesproxies, timeout10, verifyFalse # 仅测试环境禁用SSL验证 ) response.raise_for_status() except requests.exceptions.RequestException as e: print(f请求失败: {e})注意实际项目中应该使用会话(Session)对象保持连接并实现自动重试机制3.2 内容解析最佳实践结合正则与BeautifulSoup的混合解析方案from bs4 import BeautifulSoup import re html_doc html div classproduct span classnamePython编程/span span classprice¥59.00/span /div script var data {id:12345,stock:true}; /script /html # 方案1BeautifulSoup提取结构化数据 soup BeautifulSoup(html_doc, html.parser) products [] for div in soup.find_all(div, class_product): name div.find(span, class_name).text price float(div.find(span, class_price).text[1:]) products.append({name: name, price: price}) # 方案2正则提取JSON数据 json_data re.search(rvar data ({.*?});, html_doc, re.DOTALL) if json_data: import json data json.loads(json_data.group(1)) print(data[id]) # 输出: 123454. 实战项目电商价格监控爬虫4.1 系统架构设计graph TD A[爬虫调度中心] -- B[URL队列管理] B -- C{是否有新URL} C --|是| D[下载页面] C --|否| E[等待5分钟] D -- F[内容解析] F -- G[正则提取关键数据] G -- H[数据存储] H -- I[价格变动报警]4.2 核心代码实现import re import sqlite3 from datetime import datetime def parse_product_page(html): 解析电商产品页面 pattern r productId:\s*(\d),\n .*? name:\s*([^]),\n .*? price:\s*([\d.]) matches re.search(pattern, html, re.VERBOSE | re.DOTALL) if not matches: return None return { product_id: matches.group(1), name: matches.group(2), price: float(matches.group(3)), update_time: datetime.now().isoformat() } def save_to_database(data): conn sqlite3.connect(products.db) c conn.cursor() # 检查价格是否变化 c.execute(SELECT price FROM products WHERE product_id? ORDER BY update_time DESC LIMIT 1, (data[product_id],)) last_price c.fetchone() if not last_price or last_price[0] ! data[price]: c.execute(INSERT INTO products (product_id, name, price, update_time) VALUES (?,?,?,?), (data[product_id], data[name], data[price], data[update_time])) conn.commit() if last_price: print(f价格变动: {data[name]} 从 {last_price[0]} 变为 {data[price]}) conn.close()5. 常见问题与性能优化5.1 正则表达式调试技巧使用在线测试工具验证模式Regex101 (支持Python语法)RegExr (可视化解释)Python内置调试方法pattern re.compile(r (?Pyear\d{4})- (?Pmonth\d{2})- (?Pday\d{2}) , re.VERBOSE) match pattern.search(2023-05-15) print(match.groupdict()) # {year: 2023, month: 05, day: 15}5.2 爬虫性能优化方案连接池优化import requests from requests.adapters import HTTPAdapter session requests.Session() adapter HTTPAdapter(pool_connections10, pool_maxsize100, max_retries3) session.mount(http://, adapter) session.mount(https://, adapter)异步IO实现import aiohttp import asyncio async def fetch(session, url): async with session.get(url) as response: return await response.text() async def main(urls): async with aiohttp.ClientSession() as session: tasks [fetch(session, url) for url in urls] return await asyncio.gather(*tasks) urls [https://example.com/page1, https://example.com/page2] results asyncio.run(main(urls))分布式爬虫架构# 使用Scrapy-Redis实现分布式 # settings.py SCHEDULER scrapy_redis.scheduler.Scheduler DUPEFILTER_CLASS scrapy_redis.dupefilter.RFPDupeFilter REDIS_URL redis://:passwordhost:6379 # 启动多个爬虫实例 # scrapy crawl myspider6. 法律合规与道德考量遵守robots.txt协议from urllib.robotparser import RobotFileParser rp RobotFileParser() rp.set_url(https://example.com/robots.txt) rp.read() can_fetch rp.can_fetch(*, https://example.com/private-page) print(f允许爬取: {can_fetch})数据采集原则限制请求频率(建议≥3秒/次)不爬取个人隐私数据遵守网站服务条款对公开API使用认证密钥存储与使用规范敏感数据加密存储注明数据来源不用于商业牟利(除非获得授权)