在本文中，我們將介紹如何使用Python編寫一個簡單的數(shù)據(jù)抓取器，用于爬取東方財富網(wǎng)上的各類財務(wù)報表數(shù)據(jù)。我們將利用requests和lxml庫進(jìn)行數(shù)據(jù)請求和解析，并將抓取到的數(shù)據(jù)保存到CSV文件中。
1. 準(zhǔn)備工作
首先，確保你已經(jīng)安裝了以下Python庫：
pip install requests
pip install lxml
2. 創(chuàng)建數(shù)據(jù)抓取器
我們將創(chuàng)建一個名為DataScraper的類，用于封裝所有數(shù)據(jù)抓取相關(guān)的方法。數(shù)據(jù)抓取器的主要功能包括：
- 獲取報表數(shù)據(jù)
- 解析并提取表頭信息
- 將數(shù)據(jù)寫入CSV文件
2.1 初始化
在DataScraper類的__init__方法中，我們將初始化一些必要的屬性，如報表類型、報表名稱等。此外，我們還需要設(shè)置請求URL和請求頭，以便稍后進(jìn)行數(shù)據(jù)請求。
class DataScraper:
def __init__(self):
self.pagename_type = {
# ...
}
self.pagename_en = {
# ...
}
self.en_list = []
self.url = 'https://datacenter-web.eastmoney.com/api/data/v1/get'
self.headers = {
# ...
}
2.2 獲取報表數(shù)據(jù)
我們定義一個名為get_table的方法，用于向東方財富網(wǎng)發(fā)送請求并獲取報表數(shù)據(jù)。傳入頁數(shù)作為參數(shù)，返回當(dāng)前頁的報表數(shù)據(jù)。
def get_table(self, page):
# ...
2.3 解析表頭
在抓取數(shù)據(jù)之前，我們需要解析表頭信息。我們創(chuàng)建一個名為get_header的方法，傳入一個包含所有英文表頭的列表。該方法將請求報表頁面，使用lxml庫解析HTML，并提取中文表頭信息。
def get_header(self, all_en_list):
# ...
2.4 寫入表頭
接下來，我們創(chuàng)建一個名為write_header的方法，用于將解析到的表頭信息寫入CSV文件。在該方法中，我們首先調(diào)用get_header方法獲取表頭信息，然后使用csv.writer將其寫入CSV文件。
def write_header(self, table_data):
# ...
2.5 寫入報表數(shù)據(jù)
定義一個名為write_table的方法，用于將抓取到的報表數(shù)據(jù)逐行寫入CSV文件。在該方法中，我們遍歷抓取到的數(shù)據(jù)，并將每一行數(shù)據(jù)寫入CSV文件。
def write_table(self, table_data):
# ...
2.6 獲取時間列表
為了讓用戶選擇爬取的報表時間，我們定義一個名為get_timeList的方法。該方法將發(fā)送請求到東方財富網(wǎng)，解析并提取可選的時間列表。
def get_timeList(self):
# ...
3 使用數(shù)據(jù)抓取器
在創(chuàng)建好DataScraper類之后，我們可以使用以下代碼來實例化它并爬取所需的報表數(shù)據(jù)：文章來源：http://www.zghlxwxcb.cn/news/detail-481487.html
if __name__ == '__main__':
scraper = DataScraper()
timeList = scraper.get_timeList()
for index, value in enumerate(timeList):
if (index + 1) % 5 == 0:
print(value)
else:
print(value, end=' ; ')
timePoint = str(input('\n請選擇時間（可選項如上）:'))
pagename = str(input('請輸入報表類型（業(yè)績報表;業(yè)績快報;業(yè)績預(yù)告;預(yù)約披露時間;資產(chǎn)負(fù)債表;利潤表；現(xiàn)金流量表）:'))
# 校驗輸入
assert timePoint in timeList, '時間輸入錯誤'
assert pagename in list(scraper.pagename_type.keys()), '報表類型輸入錯誤'
table_type = scraper.pagename_type[pagename]
filename = f'{pagename}_{timePoint}.csv'
# 寫入表頭
scraper.write_header(scraper.get_table(1))
# 循環(huán)遍歷所有頁數(shù)
page = 1
while True:
table = scraper.get_table(page)
if table:
scraper.write_table(table)
else:
break
page += 1
4 完整代碼及結(jié)果截圖
import csv
import json
import requests
from lxml import etree
class DataScraper:
def __init__(self):
self.pagename_type = {
"業(yè)績報表": "RPT_LICO_FN_CPD",
"業(yè)績快報": "RPT_FCI_PERFORMANCEE",
"業(yè)績預(yù)告": "RPT_PUBLIC_OP_NEWPREDICT",
"預(yù)約披露時間": "RPT_PUBLIC_BS_APPOIN",
"資產(chǎn)負(fù)債表": "RPT_DMSK_FN_BALANCE",
"利潤表": "RPT_DMSK_FN_INCOME",
"現(xiàn)金流量表": "RPT_DMSK_FN_CASHFLOW"
}
self.pagename_en = {
"業(yè)績報表": "yjbb",
"業(yè)績快報": "yjkb",
"業(yè)績預(yù)告": "yjyg",
"預(yù)約披露時間": "yysj",
"資產(chǎn)負(fù)債表": "zcfz",
"利潤表": "lrb",
"現(xiàn)金流量表": "xjll"
}
self.en_list = []
self.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}\')'
}
if self.table_type in ['RPT_LICO_FN_CPD']:
params['filter'] = f'(REPORTDATE=\'{self.timePoint}\')'
params['pageNumber'] = str(page)
response = requests.get(url=self.url, params=params, headers=self.headers)
data = json.loads(response.text)
if data['result']:
return data['result']['data']
else:
return
def get_header(self, all_en_list):
ch_list = []
url = f'https://data.eastmoney.com/bbsj/{self.pagename_en[self.pagename]}.html'
response = requests.get(url)
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
def write_header(self, table_data):
with open(self.filename, 'w', encoding='utf-8', newline='') as f:
writer = csv.writer(f)
headers = self.get_header(list(table_data[0].keys()))
writer.writerow(headers)
def write_table(self, table_data):
with open(self.filename, 'a', encoding='utf-8', newline='') as csvfile:
writer = csv.writer(csvfile)
for item in table_data:
row = []
for key in item.keys():
if key in self.en_list:
row.append(str(item[key]))
print(row)
writer.writerow(row)
def get_timeList(self):
headers = {
'Referer': 'https://data.eastmoney.com/bbsj/202206.html',
}
response = requests.get('https://data.eastmoney.com/bbsj/202206.html', headers=headers)
res = etree.HTML(response.text)
return res.xpath('//*[@id="filter_date"]//option/text()')
def run(self):
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('請輸入報表類型（業(yè)績報表;業(yè)績快報;業(yè)績預(yù)告;預(yù)約披露時間;資產(chǎn)負(fù)債表;利潤表；現(xiàn)金流量表）:'))
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 = 1
while True:
table = self.get_table(page)
if table:
self.write_table(table)
else:
break
page += 1
if __name__ == '__main__':
scraper = DataScraper()
scraper.run()
關(guān)于更多東方財富網(wǎng)、巨潮網(wǎng)、中國知網(wǎng)的爬蟲，歡迎來參觀我的github倉庫文章來源地址http://www.zghlxwxcb.cn/news/detail-481487.html
到了這里，關(guān)于Python 爬取財務(wù)報表的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
