import pandas as pd from bs4 import BeautifulSoup from selenium import webdriver from selenium.webdriver.common.action_chains import ActionChains import time import random import os import glob # def clean_text(html_content): # soup = BeautifulSoup(html_content, 'html.parser') # paragraphs = soup.find_all('p') # lines = [] # for p in paragraphs: # line = ''.join([span.get_text(strip=True) for span in p.find_all('span', recursive=False)]) # lines.append(line) # return '\n'.join(lines).strip() def clean_text(html_content): import re soup = BeautifulSoup(html_content, 'html.parser') # 移除脚本和样式内容 for script_or_style in soup(["script", "style", "o:p"]): script_or_style.decompose() # 提取所有可见文本 text = soup.get_text(strip=True) # 清理和整理文本，去除所有空格 cleaned_text = re.sub(r'\s+', '', text) # 去除所有空格 return cleaned_text def process_table(table_rows,current_url, error_urls): results = { "行政处罚决定书文号": "", "被处罚当事人": "", "主要违法违规事实": "", "行政处罚依据": "", "行政处罚决定": "", "作出处罚决定的机关名称": "", "作出处罚决定的日期": "", "网址":current_url # 新增URL列 } try: if len(table_rows) == 9: results["行政处罚决定书文号"] = clean_text(str(table_rows[0].find_all('td')[1])) person_name = clean_text(str(table_rows[1].find_all('td')[2])) org_name = clean_text(str(table_rows[2].find_all('td')[2])) legal_rep_name = clean_text(str(table_rows[3].find_all('td')[1])) results["被处罚当事人"] = f'"个人姓名": "{person_name}"\n"单位名称": "{org_name}"\n"单位法定代表人（主要负责人）姓名": "{legal_rep_name}"' results["主要违法违规事实"] = clean_text(str(table_rows[4].find_all('td')[1])) results["行政处罚依据"] = clean_text(str(table_rows[5].find_all('td')[1])) results["行政处罚决定"] = clean_text(str(table_rows[6].find_all('td')[1])) results["作出处罚决定的机关名称"] = clean_text(str(table_rows[7].find_all('td')[1])) results["作出处罚决定的日期"] = clean_text(str(table_rows[8].find_all('td')[1])) elif len(table_rows) == 10: results["行政处罚决定书文号"] = clean_text(str(table_rows[0].find_all('td')[1])) person_name = clean_text(str(table_rows[1].find_all('td')[3])) person_org = clean_text(str(table_rows[2].find_all('td')[1])) org_name = clean_text(str(table_rows[3].find_all('td')[2])) legal_rep_name = clean_text(str(table_rows[4].find_all('td')[1])) results["被处罚当事人"] = f'"个人姓名": "{person_name}"\n"个人单位": "{person_org}"\n"单位名称": "{org_name}"\n"单位法定代表人（主要负责人）姓名": "{legal_rep_name}"' results["主要违法违规事实"] = clean_text(str(table_rows[5].find_all('td')[1])) results["行政处罚依据"] = clean_text(str(table_rows[6].find_all('td')[1])) results["行政处罚决定"] = clean_text(str(table_rows[7].find_all('td')[1])) results["作出处罚决定的机关名称"] = clean_text(str(table_rows[8].find_all('td')[1])) results["作出处罚决定的日期"] = clean_text(str(table_rows[9].find_all('td')[1])) elif len(table_rows) == 8: flag = 0 for row in table_rows: cells = row.find_all('td') if len(cells) == 3: flag = 1 results["行政处罚决定书文号"] = clean_text(str(table_rows[0].find_all('td')[1])) if(flag): org_name=clean_text(str(table_rows[1].find_all('td')[2])) name=clean_text(str(table_rows[2].find_all('td')[1])) results["被处罚当事人"] = f'"单位名称": "{org_name}"\n"主要负责人姓名": "{name}"' else: part1 = clean_text(str(table_rows[1].find_all('td')[0])) + ":" + clean_text( str(table_rows[1].find_all('td')[1])) part2 = clean_text(str(table_rows[2].find_all('td')[0])) + ":" + clean_text( str(table_rows[2].find_all('td')[1])) results["被处罚当事人"] = part1 + "\n" + part2 # 使用换行符分隔 results["主要违法违规事实"] = clean_text(str(table_rows[3].find_all('td')[1])) results["行政处罚依据"] = clean_text(str(table_rows[4].find_all('td')[1])) results["行政处罚决定"] = clean_text(str(table_rows[5].find_all('td')[1])) results["作出处罚决定的机关名称"] = clean_text(str(table_rows[6].find_all('td')[1])) results["作出处罚决定的日期"] = clean_text(str(table_rows[7].find_all('td')[1])) elif len(table_rows)==7: flag=0 for row in table_rows: cells = row.find_all('td') if len(cells) == 3: # 检查并拼接具有三个td的行的第二和第三个td内容 name = clean_text(str(cells[1])) + ":" + clean_text(str(cells[2])) flag=1 if(flag): results["被处罚当事人"] = name else: results["被处罚当事人"] = clean_text(str(table_rows[1].find_all_next('td')[1])) results["行政处罚决定书文号"]=clean_text(str(table_rows[0].find_all_next('td')[1])) results["主要违法违规事实"] = clean_text(str(table_rows[2].find_all_next('td')[1])) results["行政处罚依据"] = clean_text(str(table_rows[3].find_all_next('td')[1])) results["行政处罚决定"] = clean_text(str(table_rows[4].find_all_next('td')[1])) results["作出处罚决定的机关名称"] = clean_text(str(table_rows[5].find_all_next('td')[1])) results["作出处罚决定的日期"] = clean_text(str(table_rows[6].find_all_next('td')[1])) else: print(f"Unexpected number of rows in table at URL: {current_url}") error_urls.append(current_url) return None except Exception as e: print(f"Error processing table: {e}") error_urls.append(current_url) return None return results # def create_browser(): #适用docker环境中 # options = webdriver.ChromeOptions() # options.add_argument('--headless') # 使用无头模式 # options.add_argument('--disable-blink-features=AutomationControlled') # options.add_argument( # 'user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36') # options.add_experimental_option('excludeSwitches', ['enable-automation']) # options.add_experimental_option('useAutomationExtension', False) # driver = webdriver.Remote( # command_executor='http://chrome:4444/wd/hub', # options=options # ) # return driver def fetch_data(urls): options = webdriver.ChromeOptions() #适用本地环境 options.add_argument('--headless') # 使用无头模式 options.add_argument('--disable-blink-features=AutomationControlled') options.add_argument( 'user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36') options.add_experimental_option('excludeSwitches', ['enable-automation']) options.add_experimental_option('useAutomationExtension', False) driver = webdriver.Chrome(options=options) driver.execute_cdp_cmd('Page.addScriptToEvaluateOnNewDocument', { 'source': ''' Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); window.navigator.chrome = { runtime: {} }; Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] }); Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] }); ''' }) # driver = create_browser() all_data = pd.DataFrame() error_urls = [] for url in urls: try: driver.get(url) print("Processing URL:", url) random_wait(1, 3) # 随机等待时间 html = driver.page_source soup = BeautifulSoup(html, 'html.parser') selectors = [ '.Section0 .MsoNormalTable, .Section0 .MsoTableGrid', '.Section1 .MsoNormalTable, .Section1 .MsoTableGrid', '.WordSection1 .MsoNormalTable, .WordSection1 .MsoTableGrid', '.Section0 table', '.Section1 table', '.WordSection1 table' ] table = None for selector in selectors: table = soup.select_one(selector) if table: break if table: table_rows = table.find_all('tr') results = process_table(table_rows,url,error_urls) df = pd.DataFrame([results]) all_data = pd.concat([all_data, df], ignore_index=True) else: print(f"No table found for URL: {url}") error_urls.append(url) except Exception as e: print(f"Error processing URL {url}: {e}") error_urls.append(url) driver.quit() if error_urls: with open('error_urls.txt', 'w') as file: for error_url in error_urls: file.write(f"{error_url}\n") print(f"Error URLs have been saved to error_urls.txt") return all_data def random_wait(min_time=1, max_time=3): time.sleep(random.uniform(min_time, max_time)) def create_empty_excel(filename): columns = ["行政处罚决定书文号", "被处罚当事人", "主要违法违规事实", "行政处罚依据", "行政处罚决定", "作出处罚决定的机关名称", "作出处罚决定的日期"] df = pd.DataFrame(columns=columns) df.to_excel(filename, index=False) def process_in_batches(url_files_pattern, output_file_prefix, batch_size=100, max_rows_per_file=10000): url_files = glob.glob(url_files_pattern) urls = [] for url_file in url_files: with open(url_file, 'r') as file: urls.extend([line.strip() for line in file if line.strip()]) total_urls = len(urls) num_batches = (total_urls // batch_size) + (1 if total_urls % batch_size != 0 else 0) file_index = 1 output_filename = f'{output_file_prefix}{file_index}.xlsx' rows_in_file = 0 if not os.path.exists(output_filename): create_empty_excel(output_filename) for batch_num in range(num_batches): start_index = batch_num * batch_size end_index = start_index + batch_size batch_urls = urls[start_index:end_index] print(f"Processing batch {batch_num + 1} of {num_batches}") batch_data = fetch_data(batch_urls) try: existing_data = pd.read_excel(output_filename, sheet_name='Sheet1') combined_data = pd.concat([existing_data, batch_data], ignore_index=True) except FileNotFoundError: combined_data = batch_data with pd.ExcelWriter(output_filename, engine='openpyxl', mode='a', if_sheet_exists='overlay') as writer: combined_data.to_excel(writer, index=False, sheet_name='Sheet1') rows_in_file += batch_data.shape[0] if rows_in_file >= max_rows_per_file: file_index += 1 output_filename = f'{output_file_prefix}{file_index}.xlsx' rows_in_file = 0 if not os.path.exists(output_filename): create_empty_excel(output_filename) # Example usage url_files_pattern = 'url*.txt' # 匹配所有以 'url' 开头的 txt 文件 output_file_prefix = 'output_data' process_in_batches(url_files_pattern, output_file_prefix, batch_size=100) #默认每处理100个url更新一次excel，每个excel存储数据大于1万条时新增一个文件保存。 print("Data has been appended to the existing Excel files.")