首页 文章 精选 留言 我的

精选列表

搜索[Python],共10000篇文章
优秀的个人博客,低调大师

2026 Python采集实战:BeautifulSoup批量抓取多页电商商品数据

单页商品抓取不难,真正容易出问题的是从第1页跑到第50页。 我做采集这些年遇到过不少类似情况:前几页运行正常,到了中间突然超时;某一页结构变化,整个程序直接退出;代理已经失效,脚本却还在对同一个地址反复重试。 所以,多页采集不能只在单页代码外面套一个`for`循环。至少要补上会话复用、有限重试、随机间隔、异常记录和CSV存储。 本文仍使用公开爬虫练习网站演示。实际采集电商平台前,应确认访问授权、服务条款和数据使用边界。 ## **先建立可复用的请求会话** 如果每一页都重新创建连接,会增加额外开销。`requests.Session()`可以复用部分连接及公共请求头: ``` import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry session = requests.Session() session.headers.update({ "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/126.0.0.0 Safari/537.36" ), "Accept-Language": "zh-CN,zh;q=0.9" }) retry = Retry( total=2, connect=2, read=2, status=2, backoff_factor=0.8, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET"] ) adapter = HTTPAdapter(max_retries=retry) session.mount("http://", adapter) session.mount("https://", adapter) ``` 重试次数不是越多越好。我一般先设为2次,用于处理短暂网络波动。遇到403、登录提示或验证码时,应停止并检查权限与访问策略,而不是通过无限重试继续施压。 ## **编写多页商品解析逻辑** 完整示例代码如下: ``` import csv import random import time from urllib.parse import urljoin import requests from bs4 import BeautifulSoup from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry BASE_URL = "https://books.toscrape.com/catalogue/page-{}.html" OUTPUT_FILE = "ecommerce_products.csv" session = requests.Session() session.headers.update({ "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/126.0.0.0 Safari/537.36" ) }) retry = Retry( total=2, backoff_factor=0.8, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET"] ) adapter = HTTPAdapter(max_retries=retry) session.mount("http://", adapter) session.mount("https://", adapter) def fetch_page(page_number): page_url = BASE_URL.format(page_number) response = session.get( page_url, timeout=(5, 15) ) response.raise_for_status() soup = BeautifulSoup(response.text, "html.parser") cards = soup.select("article.product_pod") products = [] for card in cards: title_tag = card.select_one("h3 a") price_tag = card.select_one(".price_color") stock_tag = card.select_one(".instock.availability") if not title_tag or not price_tag: continue products.append({ "页码": page_number, "商品名称": title_tag.get("title", "").strip(), "价格": price_tag.get_text(strip=True), "库存状态": ( stock_tag.get_text(" ", strip=True) if stock_tag else "" ), "详情链接": urljoin( page_url, title_tag.get("href", "") ) }) return products def save_products(products): fieldnames = [ "页码", "商品名称", "价格", "库存状态", "详情链接" ] with open( OUTPUT_FILE, "w", newline="", encoding="utf-8-sig" ) as file: writer = csv.DictWriter( file, fieldnames=fieldnames ) writer.writeheader() writer.writerows(products) def main(): all_products = [] failed_pages = [] for page_number in range(1, 11): try: page_products = fetch_page(page_number) all_products.extend(page_products) print( f"第{page_number}页完成," f"获取{len(page_products)}条" ) except requests.RequestException as error: failed_pages.append(page_number) print(f"第{page_number}页失败:{error}") time.sleep(random.uniform(1.5, 3.0)) save_products(all_products) print(f"共保存{len(all_products)}条商品数据") print(f"失败页码:{failed_pages}") if __name__ == "__main__": main() ``` 这段代码有一个比较实用的设计:某一页失败后,只记录失败页码,不会让整个任务立即退出。后续可以单独补采失败页面,而不必把成功页面全部重跑一遍。 ## **为什么要保留页码字段** 保存结果时,我特意增加了“页码”字段。 商品名称、价格和链接是业务数据,页码则是采集链路数据。出现数量异常时,可以直接查看是哪一页缺失。如果不保存页码,最后只知道总量少了,却很难快速定位断点。 正式任务中还可以继续增加: - 采集时间; - HTTP状态码; - 请求耗时; - 出口标识; - 重试次数。 这些字段未必都要交付给业务人员,但对排查稳定性很有帮助。 ## **代理IP应该在哪一步接入** 如果只是抓取练习网站的10个页面,没有必要为了使用代理而使用代理。 当任务扩大到大量授权页面、不同地区商品展示验证或持续价格监测时,才需要把请求出口纳入设计。我一般会把代理认证地址写入环境变量: ``` import os proxy_url = os.getenv("PROXY_URL") if proxy_url: session.proxies.update({ "http": proxy_url, "https": proxy_url }) ``` 接入极安代理时,也可以沿用这种方式。代理地址变化后只调整运行环境,不改采集逻辑,更适合长期维护。 不过,代理不能替代访问授权,也不能解决所有失败。403可能是权限或页面策略问题,429通常与请求频率有关,连接超时才更需要检查代理存活状态、认证配置和链路响应。 ## **别一开始就把并发调得很高** 不少人为了“提速”,单线程代码刚跑通就直接开20个线程。结果目标站响应变慢、429增加、代理连接数也被打满,最终实际完成速度反而下降。 我的习惯是先用低并发跑一小批页面,记录成功率和平均响应时间,再逐步增加任务量。每次只改一个参数,才能判断性能变化来自哪里。 对于BeautifulSoup多页采集来说,能稳定完成、失败可定位、结果可补采,比短时间内冲出一个很高的请求数更重要。先把单页解析做准,再把多页链路做稳,最后才轮到并发和代理优化。

优秀的个人博客,低调大师

用 Python + 大模型自动审核合同,一天审了200份

![](https://developer.qcloudimg.com/http-save/yehe-11081884/46d54d0bd585a61be2d334caf609a874.jpg) 上个月朋友老陈找我帮忙。他在一家连锁餐饮企业做采购主管,公司最近集中签供应商合同,两百多份堆在桌上,等着他一份一份审。 我到他办公室的时候,他正戴着眼镜,拿着荧光笔在纸质合同上划线。每份合同他要看付款条件、违约责任、知识产权归属这些关键条款,然后在Excel里填一行审核记录。一份合同快的话十五分钟,慢的话半小时。 “我这周每天加班到九点,眼睛都花了。”老陈把荧光笔往桌上一扔,“而且人工审容易漏,上周就有一份合同把违约金比例写成了0.5%,他没看出来,差点出问题。” 两百多份合同,格式大同小异,关键条款就那么几个维度。这不就是让AI干的活嘛。 --- 思路很直接:把合同文本丢给大模型,让它提取关键条款,同时标注潜在风险点。用pandas做批量处理,结果导出Excel,跟老陈原来的工作习惯对接。模型我用的是DeepSeek,长文本处理能力强,合同丢进去不会截断。 关键是要让模型返回JSON,后面处理起来方便。 --- 先装依赖: ``` pip install requests pandas openpyxl ``` 完整代码: ``` import requests import json import pandas as pd import time import os from datetime import datetime # ============ 配置区域 ============ API_KEY = "sk-your-deepseek-api-key" # 替换成你的API Key API_URL = "https://api.deepseek.com/chat/completions" # 要提取和审核的关键维度 REVIEW_PROMPT = """你是一个资深法务助理,请审核以下合同文本,提取关键条款并标注风险点。 请按以下JSON格式返回(不要输出其他内容): { "contract_parties": "合同双方名称", "contract_type": "合同类型(采购/服务/租赁/其他)", "contract_amount": "合同金额(如有)", "payment_terms": "付款条件摘要", "delivery_terms": "交付/履约条件摘要", "liability_clause": "违约责任条款摘要", "termination_clause": "合同终止/解除条件摘要", "ip_ownership": "知识产权归属(如有)", "dispute_resolution": "争议解决方式", "contract_period": "合同期限", "risk_items": [ { "description": "风险描述", "severity": "高/中/低", "suggestion": "修改建议" } ], "overall_risk_level": "高/中/低", "summary": "一句话总结该合同的核心风险" } 审核要点: 1. 付款条件是否对我方不利(如预付比例过高) 2. 违约金是否合理(通常不超过合同金额的30%) 3. 是否存在单方面解除权不对等 4. 知识产权归属是否明确 5. 争议解决方式是否对我方不利(如对方所在地仲裁) 6. 合同期限和续约条件是否合理 合同文本: """ def review_single_contract(contract_text, contract_name=""): """ 审核单份合同 参数: contract_text: 合同全文文本 contract_name: 合同文件名(用于日志) 返回: 解析后的JSON结果字典 """ headers = { "Content-Type": "application/json", "Authorization": f"Bearer {API_KEY}" } # 系统提示词,约束输出格式 system_msg = "你是专业的合同审核助理,严格按照要求的JSON格式输出,不要输出任何多余内容。" payload = { "model": "deepseek-chat", "messages": [ {"role": "system", "content": system_msg}, {"role": "user", "content": REVIEW_PROMPT+contract_text} ], "temperature": 0.1, # 温度调低,合同审核需要稳定准确 "max_tokens": 2000 } try: response = requests.post(API_URL, headers=headers, json=payload, timeout=120) response.raise_for_status() result = response.json() # 提取AI回复内容 content = result["choices"][0]["message"]["content"] # 尝试解析JSON,处理可能的markdown代码块包裹 content = content.strip() ifcontent.startswith("```"): # 去掉 ```json ... ``` 的包裹 content = content.split("```")[1] ifcontent.startswith("json"): content = content[4:] content = content.strip() review_result = json.loads(content) return review_result except json.JSONDecodeError: print(f" [警告] {contract_name} 的返回结果不是有效JSON,尝试修复...") # 简单的修复:尝试提取第一个 { 到最后一个 } 之间的内容 start = content.find("{") end = content.rfind("}") +1 if start!= -1 and end>start: try: return json.loads(content[start:end]) except json.JSONDecodeError: pass return {"error": "JSON解析失败", "raw_response": content} except Exception as e: print(f" [错误] 审核 {contract_name} 时出错:{e}") return {"error": str(e)} def batch_review_contracts(input_dir="./contracts"): """ 批量审核目录下的所有合同文件 参数: input_dir: 合同文本文件存放目录 返回: DataFrame格式的审核结果 """ # 收集所有txt文件 contract_files = [f for f in os.listdir(input_dir) if f.endswith(".txt")] print(f"发现 {len(contract_files)} 份合同文件") all_results = [] for i, filename in enumerate(contract_files, 1): filepath = os.path.join(input_dir, filename) print(f"[{i}/{len(contract_files)}] 正在审核:{filename} ...", end=" ") # 读取合同文本 with open(filepath, "r", encoding="utf-8") as f: contract_text = f.read() # 调用AI审核 result = review_single_contract(contract_text, contract_name=filename) print("完成") # 整理结果 risk_items = result.get("risk_items", []) high_risks = [r for r in risk_items if r.get("severity") == "高"] medium_risks = [r for r in risk_items if r.get("severity") == "中"] row = { "文件名": filename, "合同双方": result.get("contract_parties", ""), "合同类型": result.get("contract_type", ""), "合同金额": result.get("contract_amount", ""), "付款条件": result.get("payment_terms", ""), "违约责任": result.get("liability_clause", ""), "争议解决": result.get("dispute_resolution", ""), "合同期限": result.get("contract_period", ""), "整体风险等级": result.get("overall_risk_level", ""), "高风险项数量": len(high_risks), "中风险项数量": len(medium_risks), "风险详情": "; ".join([ f"[{r.get('severity','')}] {r.get('description','')}" for r in risk_items ]), "修改建议": "; ".join([ r.get("suggestion", "") for r in risk_items if r.get("severity") in ("高", "中") ]), "总结": result.get("summary", ""), "审核时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } all_results.append(row) # API调用间隔,避免触发限流 time.sleep(1) # 转成DataFrame df = pd.DataFrame(all_results) return df def export_to_excel(df, output_path="./contract_review_results.xlsx"): """ 将审核结果导出为格式化的Excel文件 """ with pd.ExcelWriter(output_path, engine="openpyxl") as writer: df.to_excel(writer, sheet_name="审核结果", index=False) # 调整列宽 worksheet = writer.sheets["审核结果"] column_widths = { "A": 20, "B": 25, "C": 10, "D": 15, "E": 30, "F": 30, "G": 20, "H": 15, "I": 12, "J": 12, "K": 12, "L": 50, "M": 50, "N": 40, "O": 18 } for col, width in column_widths.items(): worksheet.column_dimensions[col].width = width print(f"\n审核结果已导出到:{output_path}") def print_summary(df): """打印审核概览""" print("\n"+"="*50) print("合同审核概览") print("="*50) print(f"合同总数:{len(df)}") risk_counts = df["整体风险等级"].value_counts() for level in ["高", "中", "低"]: count = risk_counts.get(level, 0) print(f" {level}风险:{count} 份") # 列出高风险合同 high_risk = df[df["整体风险等级"] == "高"] if len(high_risk) >0: print(f"\n高风险合同清单({len(high_risk)} 份):") for _, rowinhigh_risk.iterrows(): print(f" - {row['文件名']}:{row['总结']}") # ============ 主流程 ============ if __name__ == "__main__": contracts_dir = "./contracts" # 如果目录不存在,创建并放几个示例合同进去 if not os.path.exists(contracts_dir): os.makedirs(contracts_dir) sample1 = """供应商采购合同 合同编号:CG-2024-0156 甲方:北京美味连锁餐饮有限公司 乙方:上海绿源食品供应有限公司 第一条 采购标的 甲方向乙方采购冷冻蔬菜系列产品,具体品种和规格详见附件《产品清单》。 第二条 合同金额 本合同总金额为人民币壹佰贰拾万元整(¥1,200,000.00)。 第三条 付款条件 甲方应在合同签订后3个工作日内预付合同总额的50%作为定金,即人民币陆拾万元整。 剩余50%在最后一批货物验收合格后7个工作日内支付。 第四条 交付条件 乙方应在合同签订后30天内分三批完成全部货物的交付。 交货地点为甲方指定的中央厨房仓库(北京市大兴区XX路XX号)。 第五条 违约责任 若乙方延期交货,每延期一天,应支付合同总额0.5%的违约金。 若甲方延期付款,每延期一天,应支付未付款项0.5%的违约金。 违约金累计上限为合同总额的15%。 第六条 争议解决 因本合同引起的争议,双方应友好协商。协商不成的,提交上海仲裁委员会仲裁。 第七条 合同期限 本合同自双方签字盖章之日起生效,有效期至2025年6月30日。 合同到期前30天,如双方均未提出终止,合同自动续期一年。""" sample2 = """设备租赁合同 合同编号:ZL-2024-0089 出租方(甲方):深圳智厨设备科技有限公司 承租方(乙方):北京美味连锁餐饮有限公司 第一条 租赁标的 甲方将以下厨房设备租赁给乙方使用: - 商用烤箱 x5台 - 商用冰箱 x10台 - 自动洗碗机 x3台 第二条 租金及支付 月租金为人民币叁万伍仟元整(¥35,000.00)。 乙方应在每月5日前支付当月租金。逾期支付需额外支付月租金2%的滞纳金。 第三条 租赁期限 租赁期自2024年7月1日起至2026年6月30日止,共24个月。 租赁期满后,如乙方续租,月租金上调10%。 第四条 设备维护 租赁期间设备的日常维护由乙方负责。 设备出现非人为故障时,由甲方负责维修,维修费用由甲方承担。 因乙方操作不当导致的损坏,维修费用由乙方承担。 第五条 违约责任 乙方提前终止合同的,应支付剩余租期租金总额的30%作为违约金。 甲方提前终止合同的,应退还乙方已付的全部押金,并支付等同于一个月租金的补偿金。 第六条 争议解决 因本合同引起的争议,提交甲方所在地人民法院诉讼解决。 第七条 其他 乙方应妥善保管和使用租赁设备,租赁期满时按原状归还。""" with open(os.path.join(contracts_dir, "采购合同_绿源食品.txt"), "w", encoding="utf-8") as f: f.write(sample1) with open(os.path.join(contracts_dir, "设备租赁合同_智厨科技.txt"), "w", encoding="utf-8") as f: f.write(sample2) print("已创建示例合同文件") # 批量审核 print("开始批量合同审核...\n") df = batch_review_contracts(contracts_dir) # 打印概览 print_summary(df) # 导出Excel export_to_excel(df) ``` --- 运行效果: ``` 发现 2 份合同文件 [1/2] 正在审核:采购合同_绿源食品.txt ... 完成 [2/2] 正在审核:设备租赁合同_智厨科技.txt ... 完成 ================================================== 合同审核概览 ================================================== 合同总数:2 高风险:1 份 中风险:1 份 低风险:0 份 高风险合同清单(1 份): - 采购合同_绿源食品.txt:违约金比例偏高(每日0.5%),争议仲裁地在对方城市 审核结果已导出到:./contract_review_results.xlsx ``` Excel里大概长这样(摘取关键字段): |文件名|整体风险等级|高风险项数量|风险详情| | --- | --- | --- | --- | |采购合同_绿源食品.txt|高|3|[高] 违约金每日0.5%偏高,年化超过180%; [高] 仲裁地为上海(对方所在地); [中] 自动续期条款缺少退出机制| |设备租赁合同_智厨科技.txt|中|1|[高] 提前终止需支付剩余租金30%违约金; [中] 续租租金上调10%未设上限| --- 老陈拿到这个工具之后,把两百多份合同的文本版丢进去,跑了一个下午全审完了。Excel里高风险的合同单独挑出来人工细看,中低风险的基本过一眼就行。 他后来跟我说,还真查出来三份合同有严重问题,其中一份的违约金条款跟范本对不上,差点就签了。 用这个方法审合同有个前提:你得先把合同转成纯文本。如果是扫描件PDF,需要先用OCR工具(比如pytesseract)提取文字。这一步不复杂,但合同量大的话建议提前处理好。

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长,特征是大鼻子、头戴帽子、身穿背带裤,还留着胡子。与他的双胞胎兄弟路易基一起,长年担任任天堂的招牌角色。

Rocky Linux

Rocky Linux

Rocky Linux(中文名:洛基)是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版,作为CentOS稳定版停止维护后与RHEL(Red Hat Enterprise Linux)完全兼容的开源替代方案,由社区拥有并管理,支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性,采用模块化包装和SELinux安全架构,默认包含GNOME桌面环境及XFS文件系统,支持十年生命周期更新。

Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能,例如代码缩略图,Python的插件,代码段等。还可自定义键绑定,菜单和工具栏。Sublime Text 的主要功能包括:拼写检查,书签,完整的 Python API , Goto 功能,即时项目切换,多选择,多窗口等等。Sublime Text 是一个跨平台的编辑器,同时支持Windows、Linux、Mac OS X等操作系统。

WebStorm

WebStorm

WebStorm 是jetbrains公司旗下一款JavaScript 开发工具。目前已经被广大中国JS开发者誉为“Web前端开发神器”、“最强大的HTML5编辑器”、“最智能的JavaScript IDE”等。与IntelliJ IDEA同源,继承了IntelliJ IDEA强大的JS部分的功能。

用户登录
用户注册