Python从CSV读取URL爬取数据并导出至目标CSV的问题求助
Python 3.11.2; PyCharm 2022.3.3 (社区版) - Build PC-223.8836.43; 系统:Windows 11 Pro 22H2 22621.1413; Chrome 111.0.5563.65 (正式版) 64位
因编辑框异常,已省略若干失败的中间尝试。
需求求助
需要实现以下功能:
- 读取单列CSV文件
caselist.csv(包含10条URL) - 对每个URL执行爬取脚本,将结果按指定列(
case_title、case_plaintiff、case_defendant、case_number、case_filed、court、case_nature_of_suit、case_cause_of_action、jury_demanded)输出到caselist_output.csv,每行对应一个案例
caselist.csv中的URL列表:
https://dockets.justia.com/docket/alabama/alndce/6:2013cv01516/148887 https://dockets.justia.com/docket/arizona/azdce/2:2010cv02664/572428 https://dockets.justia.com/docket/arkansas/aredce/4:2003cv01507/20369 https://dockets.justia.com/docket/arkansas/aredce/4:2007cv00051/67198 https://dockets.justia.com/docket/arkansas/aredce/4:2007cv01067/69941 https://dockets.justia.com/docket/arkansas/aredce/4:2008cv00172/70993 https://dockets.justia.com/docket/arkansas/aredce/4:2008cv01288/73322 https://dockets.justia.com/docket/arkansas/aredce/4:2008cv01839/73965 https://dockets.justia.com/docket/arkansas/aredce/4:2008cv02513/74818 https://dockets.justia.com/docket/arkansas/aredce/4:2008cv02666/74976
尝试的方案及问题
方案1:基础CSV版本
代码如下:
from bs4 import BeautifulSoup import requests import csv th_fields = { 'case_plaintiff': 'Plaintiff', 'case_defendant': 'Defendant', 'case_number': 'Case Number', 'case_filed': 'Filed', 'court': 'Court', 'case_nature_of_suit': 'Nature of Suit', 'case_cause_of_action': 'Cause of Action', 'jury_demanded': 'Jury Demanded By' } fgtParams = [('div', {'class': 'title-wrapper'})] + [('td', {'data-th': f}) for f in th_fields.values()] with open('caselist.csv') as f: links = [l.strip() for l in f.read().splitlines() if l.strip().startswith('https://dockets.justia.com/docket')] def find_get_text(bsTag, tName='div', tAttrs=None): t = bsTag.find(tName, {} if tAttrs is None else tAttrs) if t: return t.get_text(' ',strip=True) # safer as a conditional def scrape_docketsjustia(djUrl, paramsList=fgtParams): soup = BeautifulSoup((r:=requests.get(djUrl)).content, 'lxml') cases_class = 'wrapper jcard has-padding-30 blocks has-no-bottom-padding' cases = soup.find_all('div', class_=cases_class) # print(f'{len(cases)} cases <{r.status_code} {r.reason}> from {r.url}') return [[find_get_text(c, n, a) for n, a in paramsList] for c in cases] all_ouputs = [] for url in links: all_ouputs += scrape_docketsjustia(url) with open("posts/caselist_output.csv", "w") as f: writer = csv.writer(f) writer.writerow(['case_title', *th_fields]) # [ header row with column names ] writer.writerows(all_ouputs)
问题:运行后无任何输出。
方案2:Pandas版本
初始代码:
import requests from bs4 import BeautifulSoup import pandas as pd # [I just prefer pandas] input_fp = 'caselist.csv' output_fp = 'caselist_output.csv' th_fields = { 'case_plaintiff': 'Plaintiff', 'case_defendant': 'Defendant', 'case_number': 'Case Number', 'case_filed': 'Filed', 'court': 'Court', 'case_nature_of_suit': 'Nature of Suit', 'case_cause_of_action': 'Cause of Action', 'jury_demanded': 'Jury Demanded By' } fgtParams = [('case_title', 'div', {'class': 'title-wrapper'})] + [(k, 'td', {'data-th': f}) for k,f in th_fields.items()] ## function definitions ## def find_get_text(bsTag, tName='div', tAttrs=None): t = bsTag.find(tName, {} if tAttrs is None else tAttrs) if t: return t.get_text(' ',strip=True) def scrape_docketsjustia(djUrl, paramsList=fgtParams): soup = BeautifulSoup((r:=requests.get(djUrl)).content, 'lxml') cases_class = 'wrapper jcard has-padding-30 blocks has-no-bottom-padding' for c in soup.find_all('div', class_=cases_class): return {k:find_get_text(c,n,a) for k,n,a in paramsList} # return {} # just return empty row if cases_class can't be found return {'error_msg': f'no cases <{r.status_code} {r.reason}> from {r.url}'} ## main logic ## ## load list of links # links = list(pd.read_csv(input_fp, header=None)[0]) # [ if you're sure ] links = [l.strip() for l in pd.read_csv(input_fp)[0] # header will get filtered anyway if l.strip().startswith('https://dockets.justia.com/docket/')] # safer ## scrape for each link df = pd.DataFrame([scrape_docketsjustia(u) for u in links]) # df = pd.DataFrame(map(scrape_docketsjustia,links)).dropna(axis='rows') # drop empty rows # df['links'] = links # [ add another column with the links ] ## save scraped data # df.to_csv(output_fp, index=False, header=False) # no column headers df.to_csv(output_fp, index=False)
运行报错:
Traceback (most recent call last): File "C:\Users\cs\AppData\Local\Programs\Python\Python311\Lib\site-packages\pandas\core\indexes\base.py", line 3802, in get_loc return self._engine.get_loc(casted_key) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "pandas_libs\index.pyx", line 138, in pandas._libs.index.IndexEngine.get_loc File "pandas_libs\index.pyx", line 165, in pandas._libs.index.IndexEngine.get_loc File "pandas_libs\hashtable_class_helper.pxi", line 5745, in pandas._libs.hashtable.PyObjectHashTable.get_item File "pandas_libs\hashtable_class_helper.pxi", line 5753, in pandas._libs.hashtable.PyObjectHashTable.get_item KeyError: 0
后续异常:
Traceback (most recent call last): File "C:\Users\cs\PycharmProjects\pythonProject1\solution2.py", line 29, in links = [l.strip() for l in pd.read_csv(input_fp)[0] # header will get filtered anyway ~~~~~~~~~~~~~~~~~~~~~^^^ File "C:\Users\cs\AppData\Local\Programs\Python\Python311\Lib\site-packages\pandas\core\frame.py", line 3807, in __getitem__ indexer = self.columns.get_loc(key) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "C:\Users\cs\AppData\Local\Programs\Python\Python311\Lib\site-packages\pandas\core\indexes\base.py", line 3804, in get_loc raise KeyError(key) from err KeyError: 0
修改读取链接的代码为links = [l.strip() for l in pd.read_csv(input_fp , header=None )[0] if l.strip().startswith('https://dockets.justia.com/docket/')]后,运行仍无输出。
修复后的解决方案
问题根源
- 原代码中的页面元素选择器已失效:Justia的页面结构更新后,原
wrapper jcard has-padding-30 blocks has-no-bottom-padding等类名不再对应案例数据区域 - 无请求头模拟浏览器:直接请求容易被网站反爬机制拦截
- 方案1的输出路径
posts/caselist_output.csv若posts目录不存在会静默失败
修复代码(Pandas版本)
import requests from bs4 import BeautifulSoup import pandas as pd input_fp = 'caselist.csv' output_fp = 'caselist_output.csv' # 适配当前页面结构的字段选择器 field_mapping = { 'case_title': ('h1', {'class': 'case-name'}), 'case_plaintiff': ('div', {'class': 'parties-plaintiff'}), 'case_defendant': ('div', {'class': 'parties-defendant'}), 'case_number': ('div', {'class': 'docket-number'}), 'case_filed': ('div', {'class': 'filed-date'}), 'court': ('div', {'class': 'court-info'}), 'case_nature_of_suit': ('div', {'data-label': 'Nature of Suit'}), 'case_cause_of_action': ('div', {'data-label': 'Cause of Action'}), 'jury_demanded': ('div', {'data-label': 'Jury Demanded By'}) } # 模拟浏览器请求头,避免被反爬 headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36' } def extract_field(soup, selector): """提取指定元素的文本,找不到返回None""" element = soup.find(selector[0], selector[1]) return element.get_text(strip=True) if element else None def scrape_single_case(url): """爬取单个URL的案例数据""" try: response = requests.get(url, headers=headers) response.raise_for_status() # 捕获HTTP错误 soup = BeautifulSoup(response.content, 'lxml') case_data = {} for field, selector in field_mapping.items(): case_data[field] = extract_field(soup, selector) # 添加原URL用于错误排查 case_data['source_url'] = url return case_data except Exception as e: # 记录错误信息 return {'source_url': url, 'error_msg': str(e)} # 读取并过滤URL列表 links = pd.read_csv(input_fp, header=None, squeeze=True).str.strip().tolist() valid_links = [url for url in links if url.startswith('https://dockets.justia.com/docket/')] # 批量爬取数据 case_results = [scrape_single_case(url) for url in valid_links] # 保存结果到CSV,指定编码避免中文乱码 result_df = pd.DataFrame(case_results) result_df.to_csv(output_fp, index=False, encoding='utf-8-sig') print(f"爬取完成,结果已保存至 {output_fp}")
修复说明
- 更新了页面元素选择器,适配当前Justia案件页面的结构
- 添加
User-Agent请求头,模拟浏览器访问,降低被反爬拦截的概率 - 增加异常捕获机制,记录每个URL的爬取错误信息,便于排查问题
- 使用
pd.read_csv的squeeze=True参数简化单列CSV的读取操作 - 指定CSV编码为
utf-8-sig,避免中文内容乱码 - 移除了依赖特定目录的输出路径,直接保存到当前工作目录
内容的提问来源于stack exchange,提问作者PressMeister
相关产品推荐
相关产品推荐

