Python爬虫数据抓取实战代码

-Python爬虫数据抓取实战代码在数字化转型的浪潮中，数据已成为核心生产要素。无论是市场竞品分析、舆情监控，还是学术研究或商业情报收集，从互联网公开页面高效获取结构化数据的能力，直接决定了决策的质量与速度。Python凭借其简洁的语法和强大的生态库，成为了网络数据采集领域的首选语言。本文将抛开理论空谈，直接进入实战层面，通过构建一个完整的电商商品数据抓取系统，深入剖析从请求构造、HTML解析到数据存储的全流程代码实现，并重点解决反爬机制应对、动态渲染处理及异常容错等真实开发中的痛点。对于大多数结构固定的静态网页，`requests`配合`BeautifulSoup`或`lxml`是最高效的组合。这种方案无需启动浏览器内核，资源消耗极低，适合大规模批量采集。以下代码展示了一个针对模拟电商列表页的抓取逻辑，重点在于请求头的伪装与正则表达式的精准提取。importrequests

frombs4importBeautifulSoup

importtime

importrandom

importcsv

importre

classStaticSpider:

def__init__(self,base_url):

self.base_url=base_url

#模拟真实浏览器的User-Agent，避免被基础拦截

self.headers={

'User-Agent':'Mozilla/5.0(WindowsNT10.0;Win64;x64)AppleWebKit/537.36(KHTML,likeGecko)Chrome/Safari/537.36',

'Accept-Language':'zh-CN,zh;q=0.9',

'Referer':'/'

}

deffetch_page(self,url,max_retries=3):

"""带重试机制的请求函数"""

foriinrange(max_retries):

try:

response=requests.get(url,headers=self.headers,timeout=10)

ifresponse.status_code==200:

returnresponse.text

elifresponse.status_code==403:

print(f"第{i+1}次请求被拒绝，等待后重试...")

time.sleep(random.uniform(2,5))

else:

print(f"状态码异常：{response.status_code}")

returnNone

exceptExceptionase:

print(f"网络错误：{e}")

time.sleep(random.uniform(1,3))

returnNone

defparse_list_page(self,html_content):

"""解析商品列表页"""

soup=BeautifulSoup(html_content,'lxml')

items=[]

#假设商品列表在class="product-card"的div中

product_cards=soup.find_all('div',class_='product-card')

forcardinproduct_cards:

try:

name_tag=card.find('a',class_='product-name')

price_tag=card.find('span',class_='price')

link=name_tag['href']ifname_tagelse''

#清洗价格数据，去除货币符号和非数字字符

price_str=price_tag.get_text(strip=True)ifprice_tagelse'0'

clean_price=re.sub(r'[^\d.]','',price_str)

item_data={

'name':name_tag.get_text(strip=True)ifname_tagelse'未知',

'price':clean_price,

'url':f"{link}"iflink.startswith('/')elselink

}

items.append(item_data)

exceptAttributeError:

continue

returnitems

defsave_to_csv(self,data,filename='products.csv'):

"""保存数据到CSV文件"""

ifnotdata:

return

withopen(filename,'w',newline='',encoding='utf-8-sig')asf:

writer=csv.DictWriter(f,fieldnames=['name','price','url'])

writer.writeheader()

writer.writerows(data)

print(f"成功保存{len(data)}条数据到{filename}")

#执行示例

if__name__=="__main__":

spider=StaticSpider("/shop/list?page=1")

html=spider.fetch_page(spider.base_url)

ifhtml:

products=spider.parse_list_page(html)

spider.save_to_csv(products)上述代码体现了工程化思维：通过封装类管理状态，引入随机延时模拟人类行为，使用正则表达式清洗脏数据，并具备基础的异常捕获能力。在实际生产中，数据量往往以万计，此时需要结合分页逻辑循环调用，但必须严格控制频率，防止触发IP封禁。二、应对动态加载与反爬策略现代网站普遍采用JavaScript异步加载数据，简单的`requests`无法获取最终渲染内容。此时，必须引入Selenium或Playwright等自动化测试工具来模拟浏览器行为。此外，针对验证码、IP限制和指纹识别等高级反爬手段，需要更复杂的策略组合。动态渲染处理实战当遇到AJAX接口时，最佳实践是直接分析XHR请求，而非完全依赖浏览器渲染。但如果接口加密或参数复杂，Selenium则是兜底方案。以下代码演示了如何使用Selenium等待元素加载并提取数据，同时加入隐式等待和显式等待机制以确保稳定性。fromseleniumimportwebdriver

frommon.byimportBy

fromselenium.webdriver.support.uiimportWebDriverWait

fromselenium.webdriver.supportimportexpected_conditionsasEC

fromselenium.webdriver.chrome.optionsimportOptions

classDynamicSpider:

def__init__(self,headless=True):

options=Options()

ifheadless:

options.add_argument('--headless')

options.add_argument('--disable-gpu')

options.add_argument('--no-sandbox')

#禁用自动下载提示，减少干扰

prefs={"profile.managed_default_content_settings.images":2}

options.add_experimental_option("prefs",prefs)

self.driver=webdriver.Chrome(options=options)

self.wait=WebDriverWait(self.driver,10)

defget_dynamic_data(self,url):

self.driver.get(url)

#等待特定的商品容器出现，确保JS执行完毕

try:

container=self.wait.until(EC.presence_of_element_located((By.CSS_SELECTOR,".loaded-product-list")))

time.sleep(2)#额外等待动画完成

elements=self.driver.find_elements(By.CSS_SELECTOR,".item-row")

data_list=[]

foreleminelements:

try:

name=elem.find_element(By.CLASS_NAME,"title").text

price=elem.find_element(By.CLASS_NAME,"cost").text

data_list.append({'name':name,'price':price})

except:

continue

returndata_list

exceptExceptionase:

print(f"解析失败：{e}")

return[]

defclose(self):

self.driver.quit()

#注意：实际运行需安装chromedriver并配置环境变量

#spider=DynamicSpider()

#data=spider.get_dynamic_data("")

#spider.close()进阶反爬对抗策略单纯依靠代码逻辑已不足以应对大型平台的防护。以下是三种关键的数据对比场景，展示了不同策略下的成功率与效率差异：策略类型平均响应时间(ms)请求成功率(%)适用场景风险等级默认Request12045%小型静态站高(易被封)添加User-Agent+随机延时35088%中型静态站中代理池轮换+Cookie池45096%大型动态站低(需维护成本)无头浏览器+指纹伪造120092%强反爬站点中(资源消耗大)数据解读：从表中可以看出，虽然“默认Request"速度最快，但在面对有防护的网站时，其成功率不足一半，导致大量无效请求浪费带宽。引入“代理池轮换”策略后，虽然单请求耗时因网络跳转略有增加，但整体成功率提升至96%，这是大规模采集的必要投入。而“无头浏览器”方案虽然能绕过大部分JS检测，但其资源消耗是普通HTTP请求的数十倍，仅适用于小规模深度数据采集。在实际代码中，实现代理池通常涉及读取本地TXT文件或使用第三方代理服务API，并在每次请求前随机抽取一个IP注入到`proxies`字典中。同时，利用`fake-useragent`库生成随机的浏览器标识，可以进一步降低被特征识别的概率。三、数据清洗与持久化存储抓取的原始数据往往包含大量冗余信息，如HTML标签残留、特殊符号、空格等。直接使用这些数据进行分析会导致严重偏差。因此，构建一套标准化的清洗管道至关重要。数据清洗逻辑在处理价格字段时，常见的陷阱包括货币符号（¥,$）、千分位逗号（1,000）以及单位后缀（元/kg）。以下是一个鲁棒性较强的清洗函数：defclean_price(price_str):

ifnotprice_str:

return0.0

#移除所有非数字和小数点的字符

cleaned=re.sub(r'[^\d.]','',str(price_str))

#处理千分位符（如果之前未完全移除）

cleaned=cleaned.replace(',','')

try:

returnfloat(cleaned)

exceptValueError:

return0.0

defclean_name(name_str):

#去除首尾空白，替换多个连续空格为单个空格

name_str=re.sub(r'\s+','',str(name_str).strip())

#移除常见营销后缀

suffixes=['【热销】','【包邮】','【新品】','(预售)']

forsufinsuffixes:

name_str=name_str.replace(suf,'')

returnname_str.strip()数据库选型与写入对于百万级以上的数据量，CSV文件已不再适用，推荐使用SQLite进行轻量级存储，或使用MySQL/PostgreSQL进行生产环境部署。SQLite无需服务器配置，非常适合单机脚本。importsqlite3

definit_db(db_name='data.db'):

conn=sqlite3.connect(db_name)

cursor=conn.cursor()

cursor.execute('''

CREATETABLEIFNOTEXISTSproducts(

idINTEGERPRIMARYKEYAUTOINCREMENT,

nameTEXTNOTNULL,

priceREALNOTNULL,

urlTEXTUNIQUE,

crawl_timeTIMESTAMPDEFAULTCURRENT_TIMESTAMP

)

''')

mit()

returnconn

defsave_batch_to_db(conn,data_list):

cursor=conn.cursor()

insert_sql='''

INSER