【说明】
通过selenium下载74专业技术服务业中华建集团(600629)、华电重工(601226)、中国汽研(601965)、中衡设计(603017)、华设集团(603018)、国检集团(603060)、中材节能(603126)、建研院(603183)、设计总院(603357)、勘设股份(603458)这10家上市公司含有年报链接等相关信息的网页文件。
【说明】
通过定义get_data(tr)函数提取网页中的证券代码、证券简称、公告链接、公告标题、公告时间等信息，并通过定义parse_table(fname,save=True)函数将数据导入到二维数据表DateFrame中，最后将数据存储为csv文件，得到十家公司包含证券代码、证券简称、公告链接、公告标题、公告时间的csv文件。
【说明】
通过定义filter_words(words,df,include=True)函数筛选年报链接；通过定义keep_version(df)函数保留所需要的年报版本，如修订版等；通过定义filter_nb_10y(df)函数筛选近十年的年报。最终筛选得到十家公司近十年的排除摘要保留修订版的年报的公告链接等信息。
【说明】
通过定义download_pdfs(hrefs,code,years)函数批量下载这十家公司近十年的年报，即公司上市超过十年的下载2013-2022年的年报，公司上市不超过十年的下载上市-2022年的年报。
【说明】
提取十家公司公司资料中公司简称、办公地址、公司网址、公司电子信箱、董秘的姓名、董秘的电话、董秘的电子信箱等相关信息并存储为csv文件，csv文件展示了这十家公司公司简称、办公地址、公司网址、公司电子信箱、董秘的姓名、董秘的电话、董秘的电子信箱的具体内容。
【说明】
通过定义get_th_span(txt)、get_bounds(txt)、parse_key_fin_data(subtxt,keywords)等函数提取十家公司年报中的营业收入以及归属于上市公司股东的净利润的数据并存储为csv文件。
【说明】
绘制出600629华建集团近十年营业收入和归属于上市公司股东净利润的时间序列图，由图10以及图11可以看出，华建集团的营业收入和归属于上市公司股东净利润整体上呈上升趋势，而归属于上市公司股东的净利润在2020年存在大幅的下跌，其主要受疫情的影响。
【说明】
绘制出这十家公司2013年—2022年的营业收入时间序列图，从图12可以看出，专业技术服务业中的这十家公司中华电重工(601226)、华建集团(600629)、华设集团(603018)这三家公司的营业收入变化相对较大，而其他七家公司的营业收入变化相对平稳且与华电重工(601226)、华建集团(600629)、华设集团(603018)这三家公司的营业收入相比处于更低的水平。可见，相比之下华电重工(601226)、华建集团(600629)、华设集团(603018)这三家公司属于成长型的公司，其他七家公司属于价值型公司。
#专业技术服务业
#下载年报
import re
import pandas as pd
import time
import datetime as dt
import requests
import fitz
import os
import numpy as np
import csv
import matplotlib.pyplot as plt
from pylab import mpl
mpl.rcParams['font.sans-serif']=['SimHei']
mpl.rcParams['axes.unicode_minus']=False
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
def get_table_sse(code):
browser = webdriver.Edge() #浏览器驱动
url='http://www.sse.com.cn/disclosure/listedinfo/regular/' #上海证券交易所定期报告页网址
browser.get(url) #打开上述网页
time.sleep(3) #设置执行挂起时间，等待网页加载
browser.set_window_size(1552, 832) #全屏
browser.find_element(By.ID, "inputCode").click() #定位到搜索框
browser.find_element(By.ID, "inputCode").send_keys(code) #输入股票代码
time.sleep(3) #设置执行挂起时间，等待网页加载
selector=".sse_outerItem:nth-child(4) .filter-option-inner-inner"
browser.find_element(By.CSS_SELECTOR, selector).click()
browser.find_element(By.LINK_TEXT, "年报").click() #选择报告类型为年报
time.sleep(3) #设置执行挂起时间
#
selector = "body > div.container.sse_content > div > "
selector += "div.col-lg-9.col-xxl-10 > div > "
selector += "div.sse_colContent.js_regular > "
selector += "div.table-responsive > table"
#
element = browser.find_element(By.CSS_SELECTOR, selector)
table_html=element.get_attribute('innerHTML')
#
fname = f'{code}.html'
f = open(fname,'w',encoding='utf-8')
f.write(table_html)
f.close()
#
browser.quit()
#下载10家公司含有年报链接的网页
codes=[600629,601226,601965,603017,603018,603060,603126,603183,603357,603458]
for code in codes:
get_table_sse(code)
#定义提取网页中证券代码、证券简称、公告链接、公告标题、公告时间的函数
def get_data(tr):
p_td = re.compile('(.*?)', re.DOTALL) #利用正则表达式查找链接的标签
tds = p_td.findall(tr) #将标签添加到tds列表
#
s = tds[0].find('>') + 1 #起始索引
e = tds[0].rfind('<') #结束索引
code = tds[0][s:e]
#
s = tds[1].find('>') + 1
e = tds[1].rfind('<')
name = tds[1][s:e]
#
s = tds[2].find('href="') + 6
e = tds[2].find('.pdf"') + 4
href = 'http://www.sse.com.cn' + tds[2][s:e]
s = tds[2].find('$(this))">') + 10
e = tds[2].find('')
title = tds[2][s:e]
#
date = tds[3].strip() #删除空格
data = [code,name,href,title,date]
return(data)
#定义将数据导入到二维数据表DateFrame中的函数
def parse_table(fname, save=True):
'''
return a Dataframe containing the urls and save it in csv file
:param fname: File name of the table html source,
:type string: str
:param save: Wether save the result into a csv file or not,
:type bool: bool
:return: Dataframe containing the urls,
:rtype: Dataframe
'''
f = open(fname, encoding='utf-8')
html = f.read()
f.close()
#
p = re.compile('(.+?) ', re.DOTALL)
trs = p.findall(html)
#处理空行
trs_new = []
for tr in trs:
if tr.strip() != '':
trs_new.append(tr)
#
data_all = [get_data(tr) for tr in trs_new[1:]]
df = pd.DataFrame({
'code': [d[0] for d in data_all],
'name': [d[1] for d in data_all],
'href': [d[2] for d in data_all],
'title': [d[3] for d in data_all],
'date': [d[4] for d in data_all]
})
#
if save:
df.to_csv(f'{fname[0:-5]}.csv')
return(df)
#将数据存储为csv文件
for code in codes:
f = open(f'{code}.html', encoding='utf-8')
html = f.read()
f.close()
df = parse_table(f'{code}.html')
df.to_csv(f'{code}.csv')
#定义筛选链接的函数
def filter_words(words,df,include=True):
'''
筛选保留年报链接
:param words: 保留或剔除包含关键词列表
:param df: Dataframe
:param include: keep or exclude
:return: a dataframe
:rtype: DataFrame
'''
ls = []
for word in words:
if include:
ls.append([word in f for f in df['title']])
else:
ls.append([word not in f for f in df['title']])
index = []
for r in range(len(df)):
flag = not include
for c in range(len(words)):
if include:
flag = flag or ls[c][r]
else:
flag = flag and ls[c][r]
index.append(flag)
df2 = df[index]
return(df2)
#定义保留所需版本的函数
def keep_version(df):
df_exclude=filter_words(['摘要','全文','独立意见','鉴证报告','可收回金额','评估报告','专项说明'],df,include=False)
df_keep=filter_words(['修订版'],df_exclude,include=True) #保留修订版
df_exclude.sort_values(by='date',ascending=False,inplace=True)
href=df_exclude['href'].to_list()
title=df_exclude['title'].to_list()
date=df_exclude['date'].to_list()
old_index=[]
for i in range(len(df_keep)):
d=df_keep['date'].iloc[i]
for j in range(len(date)):
d2=date[j]
if d2 < d:
old_index.append(j)
break
for j in reversed(old_index):
del date[j], href[j], title[j]
return(href,title,date)
#定义筛选时间的函数
def filter_nb_10y(df):
dt_now=dt.datetime.now()
current_year=dt_now.year
start=f'{current_year-9}-01-01'
end=f'{current_year}-12-31'
date=df['date']
v=[d >= start and d <= end for d in date]
df_new = df[v]
return(df_new)
#定义下载年报的函数
def download_pdf(href,code,year): #单份年报下载
r=requests.get(href,allow_redirects=True)
fname=f'{code}_{year}.pdf'
f=open(fname,'wb')
f.write(r.content)
f.close()
r.close()
#定义批量下载年报的函数
def download_pdfs(hrefs,code,years):
for i in range(len(hrefs)):
href=hrefs[i]
year=years[i]
download_pdf(href,code,year)
time.sleep(30)
return()
#下载10家公司近十年年报
codes=[600629,601226,601965,603017,603018,603060,603126,603183,603357,603458]
for code in codes:
df=pd.read_csv(f'{code}.csv')
href,title,date = keep_version(df)
df1 = pd.DataFrame([href,title,date],index=['href','title','date']).T #转置筛选出来的信息
df1 = filter_nb_10y(df1) #筛选时间
#在df1中添加年报年份列数
def extractFirst6(s):
return s[:4]
df1['year'] = df1['date'].apply(lambda s:extractFirst6(s))
df1['year'] = df1['year'].astype('int')
df1['year'] = df1['year'].map(lambda x: x-1)
hrefs=df1['href']
years=df1['year']
download_pdfs(hrefs,code,years)
#提取公司资料中的相关信息并存储为csv文件
#创建存放信息的表格
df_info = pd.DataFrame(index=['公司简称','办公地址','公司网址','公司电子信箱','董秘的姓名','董秘的电话','董秘的电子信箱'],
columns=[600629,601226,601965,603017,603018,603060,603126,603183,603357,603458])
codes = [600629,601226,601965,603017,603018,603060,603126,603183,603357,603458]
for code in codes:
pdf = fitz.open(f'{code}_2022.pdf')
text = ''
for i in range(20):
page = pdf[i]
text += page.get_text()
#匹配公司简称
p_com = re.compile('(?<=公司简称：).*(?=\n)')
com = p_com.search(text).group()
df_info.loc['公司简称',{code}] = com
#匹配办公地址
p_site = re.compile('(?<=\n)\w*办公地址：?\s?\n?(.*?)\s?(?=\n)',re.DOTALL)
site = p_site.search(text).group(1)
df_info.loc['办公地址',{code}] = site
#匹配公司网址
p_web = re.compile('(?<=\n)\s?公司网址：?\s?\n?(.*?)\s?(?=\n)',re.DOTALL)
web = p_web.search(text).group(1)
df_info.loc['公司网址',{code}] = web
#匹配公司的电子信箱
p_email2 = re.compile('(?<=\n)\s?公司网址?\s?\n?(.*?)(电子信箱?\s?\n?)(.*?)\s?(?=\n)',re.DOTALL)
email2 = p_email2.search(text).group(3)
df_info.loc['公司电子信箱',{code}] = email2
#匹配董秘的姓名
p_name = re.compile('(?<=\n)董事会秘书 ?\s?\n?(证券事务代表?\s?\n?)(姓名?\s?\n?)(.*?)\s?(?=\n)',re.DOTALL)
name = p_name.search(text).group(3)
df_info.loc['董秘的姓名',{code}] = name
#匹配董秘的电话号码
p_number = re.compile('(?<=\n)电话 ?\s?\n?(.*?)\s?(?=\n)',re.DOTALL)
number = p_number.search(text).group(1)
df_info.loc['董秘的电话',{code}] = number
#匹配董秘电子信箱
p_email = re.compile('(?<=\n)电子信箱 ?\s?\n?(.*?)\s?(?=\n)',re.DOTALL)
email = p_email.search(text).group(1)
df_info.loc['董秘的电子信箱',{code}] = email
df_info.to_csv('information.csv')
#定义查找主要会计数据和财务指标所在页码的函数
def get_subtxt(doc,bounds = ('主要会计数据和财务指标','总资产')):
#默认设置为首尾页码
start_pageno = 0
end_pageno = len(doc)-1
#lb为下届，ub为上界
lb,ub = bounds
#获取左界页码
for n in range(len(doc)):
page = doc[n]
txt = page.get_text()
if lb in txt:
start_pageno = n
break
#获取右界页码
for n in range(start_pageno,len(doc)):
if ub in doc[n].get_text():
end_pageno = n
break
#获取小范围内字符串
txt = ''
for n in range(start_pageno,end_pageno+1):
page = doc[n]
txt += page.get_text()
return(txt)
#定义获取表格标题的函数
def get_th_span(txt):
nianfen = '(20\d\d|199\d)\s*?年'
s = f'{nianfen}\s*{nianfen}.*?{nianfen}'
p = re.compile(s,re.DOTALL) #遇到换行符也适用
matchobj = p.search(txt)
#
end = matchobj.end()
year1 = matchobj.group(1)
year2 = matchobj.group(2)
year3 = matchobj.group(3)
#
flag = (int(year1) - int(year2) == 1) and (int(year2) - int(year3) == 1)
#
while (not flag):
matchobj = p.search(txt[end:])
end = matchobj.end()
year1 = matchobj.group(1)
year2 = matchobj.group(2)
year3 = matchobj.group(3)
flag = (int(year1) - int(year2) == 1)
flag = flag and (int(year2) - int(year3) ==1)
return(matchobj.span())
#定义获取表格边界的函数
def get_bounds(txt):
th_span_1st = get_th_span(txt)
end = th_span_1st[1]
th_span_2nd = get_th_span(txt[end:])
th_span_2nd = (end+th_span_2nd[0],end+th_span_2nd[1])
#
s = th_span_1st[1]
e = th_span_2nd[0]-1
#
while (txt[e] not in '0123456789'):
e = e-1
return(s,e)
#定义获取关键词的函数
def get_keywords(txt):
p = re.compile(r'\d+\s+([\u2E80-\u9FFF]+)')
keywords = p.findall(txt)
keywords.insert(0,'营业收入')
return(keywords)
#定义获取所需数据的函数
def parse_key_fin_data(subtxt,keywords):
ss = []
s = 0
for kw in keywords:
n = subtxt.find(kw,s)
ss.append(n)
s = n+len(kw)
ss.append(len(subtxt))
data = []
p = re.compile('\D+(?:\s+\D*)?(?:（.*）|\(.*\))?')
p2 = re.compile('\s')
for n in range(len(ss)-1):
s = ss[n]
e = ss[n+1]
line = subtxt[s:e]
#获取可能换行的账户名称
matchobj = p.search(line)
account_name = p2.sub('',matchobj.group())
#获取三年数据
amnts = line[matchobj.end():].split()
#加上账户名称
amnts.insert(0,account_name)
#追加到总数据
data.append(amnts)
return data
codes = [600629,601226,601965,603017,603018,603060,603126,603183,603357,603458]
for code in codes:
import os
fname = []
#遍历所有pdf文件
def main():
file_path = f'C:/Users/cherish/Documents/Python/nianbao' #文件路径
folders = os.listdir(file_path) #提取所有文件生成列表
for file in folders: #判断文件后缀名是否为pdf
if(file.split('.')[-1] == 'pdf'):
fname.append(file) #将pdf文件名添加到列表中用于遍历
if __name__ == '__main__':
main()
#创建表格存放数据
locals()[f'df_{code}']=pd.DataFrame(index=range(2013,2023),
columns=['营业收入（元）','归属于上市公司股东的净利润(元)'])
for f in fname:
doc = fitz.open(f'{f}')
#解析表格
txt = get_subtxt(doc)
span = get_bounds(txt)
subtxt = txt[span[0]:span[1]]
keywords = get_keywords(subtxt)
data = parse_key_fin_data(subtxt, keywords)
#提取营业收入和归属于上市公司股东的净利润并存放到DataFrame中
revenue = float(data[0][1].replace(',',''))
profit = float(data[1][1].replace(',',''))
#利用正则表达式提取年报所属年份
text = ''
for i in range(20):
page = doc[i]
text += page.get_text()
p_year=re.compile('.*?(\d{4}) .*?年度报告.*?')
year = int(p_year.findall(text)[0])
locals()[f'df_{code}'].loc[year,'营业收入（元）']=revenue
locals()[f'df_{code}'].loc[year,'归属于上市公司股东的净利润(元)']=profit
locals()[f'df_{code}'].to_csv(f'{code}_info.csv')
#提取公司资料中的相关信息并存储为csv文件
#创建存放信息的表格
df_info = pd.DataFrame(index=['公司简称','办公地址','公司网址','公司电子信箱','董秘的姓名','董秘的电话','董秘的电子信箱'],
columns=[600629,601226,601965,603017,603018,603060,603126,603183,603357,603458])
codes = [600629,601226,601965,603017,603018,603060,603126,603183,603357,603458]
for code in codes:
pdf = fitz.open(f'{code}_2022.pdf')
text = ''
for i in range(20):
page = pdf[i]
text += page.get_text()
#匹配公司简称
p_com = re.compile('(?<=公司简称：).*(?=\n)')
com = p_com.search(text).group()
df_info.loc['公司简称',{code}] = com
#匹配办公地址
p_site = re.compile('(?<=\n)\w*办公地址：?\s?\n?(.*?)\s?(?=\n)',re.DOTALL)
site = p_site.search(text).group(1)
df_info.loc['办公地址',{code}] = site
#匹配公司网址
p_web = re.compile('(?<=\n)\s?公司网址：?\s?\n?(.*?)\s?(?=\n)',re.DOTALL)
web = p_web.search(text).group(1)
df_info.loc['公司网址',{code}] = web
#匹配公司的电子信箱
p_email2 = re.compile('(?<=\n)\s?公司网址?\s?\n?(.*?)(电子信箱?\s?\n?)(.*?)\s?(?=\n)',re.DOTALL)
email2 = p_email2.search(text).group(3)
df_info.loc['公司电子信箱',{code}] = email2
#匹配董秘的姓名
p_name = re.compile('(?<=\n)董事会秘书 ?\s?\n?(证券事务代表?\s?\n?)(姓名?\s?\n?)(.*?)\s?(?=\n)',re.DOTALL)
name = p_name.search(text).group(3)
df_info.loc['董秘的姓名',{code}] = name
#匹配董秘的电话号码
p_number = re.compile('(?<=\n)电话 ?\s?\n?(.*?)\s?(?=\n)',re.DOTALL)
number = p_number.search(text).group(1)
df_info.loc['董秘的电话',{code}] = number
#匹配董秘电子信箱
p_email = re.compile('(?<=\n)电子信箱 ?\s?\n?(.*?)\s?(?=\n)',re.DOTALL)
email = p_email.search(text).group(1)
df_info.loc['董秘的电子信箱',{code}] = email
df_info.to_csv('information.csv')