Python: 6 小时爬完上交所和深交所的年报问询函

Stata 连享会   主页 || 视频 || 推文

温馨提示： 定期 清理浏览器缓存，可以获得最佳浏览体验。

✌ 课程详情： https://gitee.com/arlionn/Course   |   lianxh.cn

✨ 课程主页： https://gitee.com/arlionn/Course

作者: 许梦洁 (Frankfurt School of Finance and Management)
E-mail: m.xu@fs.de

目录

昨天开组会的时候导师说想搞年报问询函的研究，但是买数据库太贵了。我说放着我来 ( ‵▽′)ψ。

一、任务描述

• 分别从上交所和深交所的官网上爬取年报问询函的记录

二、解决思路

• 解析网页获取全部的年报问询函列表及相应的文件链接
• 打开第一步获取的文件链接，读取 PDF 数据，并直接转成 TXT 格式的文字

三、网页分析

以上交所网站为例：

1. 数据包位置

上交所的年报问询列表存储在响应的一个 JSON 文件里，只要用 Python3 发送请求并截取这个 JSON 包就可以直接获取该页的列表啦。

2. 翻页

由于网站结构简单，所以翻页也很好实现。对比第一页的参数和第二页的参数可以发现：翻页的关键在于 pageHelp.pageNo 和 pageHelp.beginPage 这两个参数上，所以虽然一共有 88 页的问询函，只要每次 post 不同的页数就可以得到相应的页面。

3. 下载PDF

通过前两步的分析我们已经拿到了上交所问询函的所有列表：

显然上交所网站里已经直接给出了每个问询函 PDF 的下载链接(即表格最后一列)，直接打开这个链接就可以读取保存相应的问询函。

4. 深交所网页和上交所网页的区别

深交所网站比较好的是直接给出了问询函列表并且同时给出了函件内容和公司回复文档的文件编码：

举个栗子：加入函件内容的文件编码是 CDD00080753986.pdf ，那么给这个编码套上外衣就是可以打开的 PDF 地址了：

http://reportdocs.static.szse.cn/UpFiles/fxklwxhj/ + 文件编码+ ?random=0.42680171432249325

最后得到的URL地址为：http://reportdocs.static.szse.cn/UpFiles/fxklwxhj/CDD00080753986.pdf?random=0.42680171432249325

小伙伴们可以试试打开是不是正确的问询函件。

四、PDF转TXT

在Python3 里用经典的 Pdfminer3k 包即可解决，这里直接改了大佬的代码ᕕ( ᐛ )ᕗ：Pythonscrapy的简书：python3 在线读取pdf

五、核心代码

1. 获取上交所问询函列表

本来应该规规矩矩地用 Requests 包的 post 方法上传参数的，但是我太懒了啊哈哈哈就直接迭代了丑丑的 URL。

import requests

def downlourl(currentpage):
    url = "http://query.sse.com.cn/commonSoaQuery.do?siteId=28&sqlId=BS_GGLL&extGGLX=&stockcode=&channelId=10743%2C10744%2C10012&extGGDL=&order=createTime%7Cdesc%2Cstockcode%7Casc&isPagination=true&pageHelp.pageSize=15&pageHelp.pageNo=" + repr(currentpage) + "&pageHelp.beginPage=" + repr(currentpage) +"&pageHelp.cacheSize=1"
    return(url)

headers = {
    'Referer':'http://www.sse.com.cn/disclosure/credibility/supervision/inquiries/',
    'User-Agent':'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36'
}

with open('上交所列表.txt',"a") as f:
    for page in range(89):
        r = requests.get(downlourl(page), headers=headers)
        for i in r.json()['result']:
            f.write('\t''.join([i['cmsOpDate'],i['docTitle'],i['stockcode'],i['extWTFL'],i['extGSJC'],i['docType'],i['createTime'],i['docURL']])+'\n')
        print('完成爬取第%d页'%page)  

2. 爬取PDF并直接转为TXT

import pandas as pd
import time
from urllib.request import urlopen
from urllib.request import Request
from urllib.request import quote
import requests
from pdfminer.converter import PDFPageAggregator
from pdfminer.layout import LTTextBoxHorizontal, LAParams
from pdfminer.pdfinterp import PDFResourceManager, PDFPageInterpreter
from pdfminer.pdfinterp import PDFTextExtractionNotAllowed
from pdfminer.pdfparser import PDFParser, PDFDocument


data = pd.read_table('/Users/mengjiexu/深交所回复列表.txt',header=None,encoding='utf8',delim_whitespace=True)
data.columns=['函件编码','函件类型']


函件编码 = data.loc[:,'函件编码']
函件类型 = data.loc[:,'函件类型']

headers = {'content-type': 'application/json',
           'Accept-Encoding': 'gzip, deflate',
           'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:53.0) Gecko/20100101 Firefox/53.0'}

baseurl = "http://reportdocs.static.szse.cn/UpFiles/fxklwxhj/"

def parse(docucode):
    # 打开在线PDF文档
    _path = baseurl + quote(docucode) +"?random=0.3006649122149502"
    request = Request(url=_path, headers=headers)  # 随机从user_agent列表中抽取一个元素
    fp = urlopen(request)
    # 读取本地文件
    # path = './2015.pdf'
    # fp = open(path, 'rb')
    # 用文件对象来创建一个pdf文档分析器
    praser_pdf = PDFParser(fp)
    # 创建一个PDF文档
    doc = PDFDocument()
    # 连接分析器 与文档对象
    praser_pdf.set_document(doc)
    doc.set_parser(praser_pdf)
    # 提供初始化密码doc.initialize("123456")
    # 如果没有密码 就创建一个空的字符串
    doc.initialize()
    # 检测文档是否提供txt转换，不提供就忽略
    if not doc.is_extractable:
        raise PDFTextExtractionNotAllowed
    else:
        # 创建PDf资源管理器 来管理共享资源
        rsrcmgr = PDFResourceManager()
        # 创建一个PDF参数分析器
        laparams = LAParams()
        # 创建聚合器
        device = PDFPageAggregator(rsrcmgr, laparams=laparams)
        # 创建一个PDF页面解释器对象
        interpreter = PDFPageInterpreter(rsrcmgr, device)
        # 循环遍历列表，每次处理一页的内容
        # doc.get_pages() 获取page列表
        for page in doc.get_pages():
            # 使用页面解释器来读取
            interpreter.process_page(page)
            # 使用聚合器获取内容
            layout = device.get_result()
            # 这里layout是一个LTPage对象 里面存放着 这个page解析出的各种对象 一般包括LTTextBox, LTFigure, LTImage, LTTextBoxHorizontal 等等 想要获取文本就获得对象的text属性，
            for out in layout:
                # 判断是否含有get_text()方法，图片之类的就没有
                # if ``hasattr(out,"get_text"):
                docname = "/Users/mengjiexu/罗党论/年报问询函/深交所回复/"+str(docucode).split('.')[0]+'.txt'
                with open(docname,'a') as f:
                    if isinstance(out, LTTextBoxHorizontal):
                        results = out.get_text()
                        print(results)
                        f.write(results)


for i in range(len(函件编码)):
    函件名称 = (函件编码[i] + '.' + 函件类型[i])
    print(函件名称)
    开始爬取时间 = "这是第%d个公告"%i
    print(开始爬取时间)
    print(time.strftime('%Y.%m.%d.%H:%M:%S',time.localtime(time.time())))
    if 函件类型[i]=="pdf":
        parse(函件名称)
        print(函件名称 + "爬取成功")
    else:
        with open("/Users/mengjiexu/深交所回复/%s"%函件名称,'wb') as f:
            _path = baseurl + quote(函件名称) +"?random=0.3006649122149502"
            request = requests.get(url=_path, headers=headers)  # 随机从user_agent列表中抽取一个元素
            f.write(request.content)
    结束爬取时间 = time.strftime('%Y.%m.%d.%H:%M:%S', time.localtime(time.time()))
    print(结束爬取时间)
    print("第%d个公告爬取完成" % i)

3. 遍历文件夹中所有TXT和DOC文件并生成列表

参考了 Lynn大神的博客 |･ω･｀)

import os
import docx2txt
from openpyxl import Workbook

content_list = []

wb = Workbook()
sheet = wb.active
sheet['A1'].value = '公告编码'
sheet['A2'].value = '公告内容'

def readdocx(filepath):
    content = docx2txt.process(filepath)  #打开传进来的路径
    docucode = filepath.split('/')[-1]
    content_list.append([docucode.split('.')[0],content])
    content_list.append([docucode.split('.')[0],content])

def readtxt(filepath):
    content = open(filepath, "r").read()     #打开传进来的路径
    docucode = filepath.split('/')[-1]
    content_list.append([docucode.split('.')[0],content])

def eachFile(filepath):
    pathDir = os.listdir(filepath) #获取当前路径下的文件名，返回List
    for s in pathDir:
        newDir=os.path.join(filepath,s)#将文件命加入到当前文件路径后面
        if os.path.isfile(newDir) :         #如果是文件
            doctype = os.path.splitext(newDir)[1]
            if doctype == ".txt":  #判断是否是txt
                readtxt(newDir)
            elif doctype == ".docx":
                readdocx(newDir)
            else:
                pass
        else:
            eachFile(newDir) #如果不是文件，递归这个文件夹的路径


eachFile("/Users/上交所txt/")
a = 1
for doc in content_list:
    sheet['A%d'%a].value = doc[0]
    print(doc[0])
    sheet['B%d'%a].value = doc[1]
    a += 1
wb.save('上交所问询函.xlsx')

六、最终爬取结果

主要参考链接

温馨提示： 文中链接在微信中无法生效。请点击底部「阅读原文」

1. Python文件处理：递归批处理文件夹子目录内所有 txt 数据
2. Python 爬虫怎么处理 json 内容
3. Python3 在线读取 PDF

相关课程

连享会-直播课 上线了！
http://lianxh.duanshu.com

免费公开课：

• 直击面板数据模型 - 连玉君，时长：1小时40分钟
• Stata 33 讲 - 连玉君, 每讲 15 分钟.
• 部分直播课 课程资料下载 (PPT，dofiles等)

课程一览

支持回看，所有课程可以随时购买观看。

Note: 部分课程的资料，PPT 等可以前往 连享会-直播课 主页查看，下载。

关于我们

• Stata连享会 由中山大学连玉君老师团队创办，定期分享实证分析经验。直播间 有很多视频课程，可以随时观看。
• 连享会-主页 和 知乎专栏，300+ 推文，实证分析不再抓狂。
• 公众号推文分类： 计量专题 | 分类推文 | 资源工具。推文分成 内生性 | 空间计量 | 时序面板 | 结果输出 | 交乘调节 五类，主流方法介绍一目了然：DID, RDD, IV, GMM, FE, Probit 等。
• 公众号关键词搜索/回复 功能已经上线。大家可以在公众号左下角点击键盘图标，输入简要关键词，以便快速呈现历史推文，获取工具软件和数据下载。常见关键词：课程, 直播, 视频, 客服, 模型设定, 研究设计, stata, plus, 绘图, 编程, 面板, 论文重现, 可视化, RDD, DID, PSM, 合成控制法 等

连享会小程序：扫一扫，看推文，看视频……

扫码加入连享会微信群，提问交流更方便

✏ 连享会学习群-常见问题解答汇总：
✨ https://gitee.com/arlionn/WD