废话不多说 , 直接上源码

import csv  # 用于读写 CSV 文件
import json  # 用于解析 JSON 数据
import requests  # 用于发送 HTTP 请求
from lxml import etree  # 用于解析 HTML 文档# 定义 DataScraper 类，实现数据抓取功能
class DataScraper:# 初始化函数def __init__(self):# 定义一个字典，用于存储页面名称和页面英文名称的对应关系self.pagename_type = {"业绩报表": "RPT_LICO_FN_CPD","业绩快报": "RPT_FCI_PERFORMANCEE","业绩预告": "RPT_PUBLIC_OP_NEWPREDICT","预约披露时间": "RPT_PUBLIC_BS_APPOIN","资产负债表": "RPT_DMSK_FN_BALANCE","利润表": "RPT_DMSK_FN_INCOME","现金流量表": "RPT_DMSK_FN_CASHFLOW"}self.pagename_en = {"业绩报表": "yjbb","业绩快报": "yjkb","业绩预告": "yjyg","预约披露时间": "yysj","资产负债表": "zcfz","利润表": "lrb","现金流量表": "xjll"}# 定义一个列表，用于存储英文列名self.en_list = []# 定义一个抓取数据的 URLself.url = 'https://datacenter-web.eastmoney.com/api/data/v1/get'# 定义请求头self.headers = {'Accept': '*/*','Accept-Language': 'zh-CN,zh;q=0.9','Connection': 'closed','Referer': 'https://data.eastmoney.com/','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','sec-ch-ua': '"Google Chrome";v="111", "Not(A:Brand";v="8", "Chromium";v="111"','sec-ch-ua-mobile': '?0','sec-ch-ua-platform': '"Windows"'}# 定义一个函数，用于获取指定页面的表格数据def get_table(self, page):# 定义一个参数字典，用于传递抓取参数params = {'sortTypes': '-1,-1',  # 用于指定排序方式，这里设置为降序'reportName': self.table_type,  # 用于指定要抓取的报表类型'columns': 'ALL',  # 用于指定要抓取的列名'filter': f'(REPORT_DATE=\'{self.timePoint}\')'  # 用于指定查询条件，这里设置为查询指定日期范围的数据}# 如果要抓取的报表类型为 "RPT_LICO_FN_CPD"，则需要修改查询条件if self.table_type in ['RPT_LICO_FN_CPD']:params['filter'] = f'(REPORTDATE=\'{self.timePoint}\')'# 添加分页参数params['pageNumber'] = str(page)# 使用 requests 库发送 GET 请求，获取数据response = requests.get(url=self.url, params=params, headers=self.headers)# 使用 json 模块解析响应数据data = json.loads(response.text)# 如果数据抓取成功，则返回数据if data['result']:return data['result']['data']else:return# 定义一个函数，用于获取指定页面的表头信息def get_header(self, all_en_list):# 创建一个空列表，用于存储中文列名ch_list = []# 定义一个页面的 URLurl = f'https://data.eastmoney.com/bbsj/{self.pagename_en[self.pagename]}.html'# 使用 requests 库发送 GET 请求，获取页面数据response = requests.get(url)# 使用 lxml 模块解析 HTML 文档res = etree.HTML(response.text)# 遍历所有英文列名，获取对应的中文列名for en in all_en_list:ch = ''.join([i.strip() for i in res.xpath(f'//div[@class="dataview"]//table[1]//th[@data-field="{en}"]//text()')])# 如果找到了中文列名，则将其添加到列表中if ch:ch_list.append(ch)# 将英文列名添加到列表中self.en_list.append(en)# 返回中文列名列表return ch_list# 定义一个函数，用于将抓取的数据写到 CSV 文件中def write_header(self, table_data):# 打开一个 CSV 文件，用于写数据with open(self.filename, 'w', encoding='utf-8', newline='') as f:# 创建一个 CSV 写器writer = csv.writer(f)# 获取表头信息headers = self.get_header(list(table_data[0].keys()))# 使用 CSV 写器写表头writer.writerow(headers)# 定义一个函数，用于将抓取的数据写到 CSV 文件中def write_table(self, table_data):# 打开一个 CSV 文件，用于追加写数据with open(self.filename, 'a', encoding='utf-8', newline='') as csvfile:# 创建一个 CSV 写器writer = csv.writer(csvfile)# 遍历所有数据，并将其写到 CSV 文件中for item in table_data:# 创建一个空列表，用于存储一行的数据row = []# 遍历所有列名，获取对应的数据for key in item.keys():# 如果列名在英文列名列表中，则将其添加到列表中if key in self.en_list:row.append(str(item[key]))# 使用 CSV 写器写一行数据print(row)writer.writerow(row)# 定义一个函数，用于获取时间列表def get_timeList(self):# 定义一个请求头，用于模拟浏览器请求headers = {'Referer': 'https://data.eastmoney.com/bbsj/202312.html'}# 使用 requests 库发送 GET 请求，获取页面数据response = requests.get('https://data.eastmoney.com/bbsj/202312.html', headers=headers)# 使用 lxml 模块解析 HTML 文档res = etree.HTML(response.text)# 使用xpath获取时间列表return res.xpath('//*[@id="filter_date"]//option/text()')# 定义一个运行函数，用于启动数据抓取程序def run(self):# 使用 get_timeList 函数获取时间列表self.timeList = self.get_timeList()# 遍历时间列表，打印时间for index, value in enumerate(self.timeList):if (index + 1) % 5 == 0:print(value)else:print(value, end=' ; ')# 输入要抓取的时间点self.timePoint = str(input('\n请选择时间（可选项如上）:'))# 输入要抓取的报表类型self.pagename = str(input('请输入报表类型（业绩报表;业绩快报;业绩预告;预约披露时间;资产负债表;利润表；现金流量表）:'))# 判断输入的时间点和报表类型是否正确assert self.timePoint in self.timeList, '时间输入错误'assert self.pagename in list(self.pagename_type.keys()), '报表类型输入错误'# 根据输入的报表类型获取对应的表格类型self.table_type = self.pagename_type[self.pagename]self.filename = f'{self.pagename}_{self.timePoint}.csv'self.write_header(self.get_table(1))page = 1while True:table = self.get_table(page)if table:self.write_table(table)else:breakpage += 1if __name__ == '__main__':scraper = DataScraper()scraper.run()