【RAG实战】基于TextIn打造上市公司财务报表智能问答系统

#定义一个check_world函数def check_word(sentence, word):if word in sentence:#如果关键词word在文本中，返回true，否则返回falsereturn Trueelse:return False

def download_report(code): # 启动Edge浏览器并加载选项browser = webdriver.Edge()url = 'http://www.cninfo.com.cn/new/commonUrl?url=disclosure/list/notice#sse'browser.get(url)browser.maximize_window()#输入时间#注：这段有没有都无所谓（）因为我发现就算写了他也不会给我执行这段操作，但是因为网站的自动检索年报的范围就是我需要的范围，所以没差（）browser.find_element_by_xpath('//*[@id="main"]/div[2]/div[1]/div[2]/div[1]/div[2]/form/div[1]/div/div/input[1]').send_keys('2022-12-31')browser.find_element_by_xpath('//*[@id="main"]/div[2]/div[1]/div[2]/div[1]/div[2]/form/div[1]/div/div/input[2]').send_keys('2022-06-15')#browser.find_element_by_xpath("//body").click()#browser.find_element_by_xpath("//body").click()    #输入年报，将检索范围锁定在年报中#第一段用于点击分类按钮browser.find_element_by_xpath('//*[@id="main"]/div[2]/div[1]/div[2]/div[1]/div[2]/form/div[2]/div[3]/div/div/span/button').click()#第二段用于点击年报选项browser.find_element_by_xpath('/html/body/div[6]/div[1]/label[1]/span[1]/span').click()#输入代码#第一段用于点击输入框browser.find_element_by_xpath('/html/body/div[2]/div/div[2]/div[1]/div[2]/div[1]/div[2]/form/div[2]/div[1]/div/div/div/div/input').send_keys(code)#这一段用于暂停页面操作，等待等待元素加载完成后继续执行操作（我设置的是三秒）time.sleep(3)#这一段用于点击搜索按钮，使用了更为稳健的模拟鼠标操作button_element = browser.find_element_by_xpath('/html/body/div[2]/div/div[2]/div[1]/div[2]/div[1]/div[2]/div[1]/button')actions = ActionChains(browser)actions.move_to_element(button_element).click().perform()# 等待页面元素加载完成time.sleep(3)     #进入公告#写了一个try，因为有时候可能对应的代码没有公司try:#调用check_word函数，判断要进入具体页面的报告是摘要还是年报word = "摘要"#获取页面中第二个xpath对应的text内容（如600000号股票就会返回“上海浦东发展银行股份有限公司2022年年度报告（全文）”）browser_text = browser.find_element_by_xpath('/html/body/div[2]/div/div[2]/div[1]/div[1]/div[2]/div/div[3]/table/tbody/tr[2]/td[3]/div/span/a').text  #如果是摘要，返回true，执行第一个xpath#注：有的页面可能有三个按钮，但是前两个按钮中一个有一个是年报或者年报修订版（确信）if check_word(browser_text, word):browser.find_element_by_xpath('/html/body/div[2]/div/div[2]/div[1]/div[1]/div[2]/div/div[3]/table/tbody/tr[1]/td[3]/div/span/a').click() #上一步操作后会打开一个新的页面，我们要获取新页面的url用于下载，将browser变更为新页面window_handles = browser.window_handleslatest_window_handle = window_handles[-1]browser.switch_to.window(latest_window_handle) #当不含有摘要，返回false，点击第二个xpathelse:browser.find_element_by_xpath('/html/body/div[2]/div/div[2]/div[1]/div[1]/div[2]/div/div[3]/table/tbody/tr[2]/td[3]/div/span/a').click()window_handles = browser.window_handleslatest_window_handle = window_handles[-1]browser.switch_to.window(latest_window_handle) #获取网页urlbrowser_url = browser.current_url #当所属公司code无法搜索出年报，那么这家公司可能是退市了，输出没有找到年报except Exception as e:print("没有找到",code,"的对应年报")browser.quit()#跳出方法return#这一段用于检测code有没有对应的公司，如果没有公司，那么点击搜索按钮只会停留在原来的页面上。这时执行上文获取的url就会下载错误的年报。#这里我用了暴力的解决方法，直接查看原有页面的前两个xpath具体页面的url内容，然后ban掉他们！这里在使用的时候一定要记得检查当日的前两位urlif browser_url == "http://www.cninfo.com.cn/new/disclosure/detail?stockCode=688669&announcementId=1217087254&orgId=gfbj0833817&announcementTime=2023-06-17" or browser_url == "http://www.cninfo.com.cn/new/disclosure/detail?stockCode=603825&announcementId=1217085098&orgId=9900024448&announcementTime=2023-06-17":print("没有",code,"对应的公司")browser.quit()else: browser2 = webdriver.Edge()browser2.get(browser_url)#进入url的页面，点击下载按钮，下载年报。其中，如果网不好+文件大的情况，就需要将time.sleep（10）的参数调大，不然下不完，网站就关了browser2.find_element_by_xpath('/html/body/div[1]/div/div[1]/div[3]/div[1]/button').click()time.sleep(10)print("已成功下载",code,"公司的年报") #关闭网站，防止资源浪费browser.quit()browser2.quit()

# 贵州茅台code = "600519"download_report(code)

import requests
class CommonOcr(object):def __init__(self, img_path):# 请登录后前往 “工作台-账号设置-开发者信息” 查看 x-ti-app-id 和 x-ti-secret-codeself._app_id = '0c88509xxxx'self._secret_code = '3017d8ccxxxx'self._img_path = img_path
def get_file_content(self):with open(self._img_path, 'rb') as fp:return fp.read()
def recognize(self):# 通用文档解析url = 'https://api.textin.com/ai/service/v1/pdf_to_markdown'head = {}try:image = self.get_file_content()head['x-ti-app-id'] = self._app_idhead['x-ti-secret-code'] = self._secret_coderesult = requests.post(url, data=image, headers=head)return result.textexcept Exception as e:return e

if __name__ == "__main__":file_path = r'PDF测试文档.pdf'response = CommonOcr(file_path)output = response.recognize()    print(output)

import requestsimport jsonimport timefrom langchain.vectorstores import FAISSfrom langchain_core.documents import Documentfrom langchain.text_splitter import CharacterTextSplitterfrom langchain.embeddings import HuggingFaceEmbeddings

def get_all_docs(input):if isinstance(input, str):result = json.loads(input)result = result.get('result', {})markdown = result.get('markdown')metadata = {"source": file_path}documents = [(Document(page_content=markdown, metadata=metadata))]return documentsif __name__ == "__main__":    file_path = r'2023年度贵州茅台财务报告.pdf'response = CommonOcr(file_path)output = response.recognize()
documents = get_all_docs(output)# print(documents)text_splitter = CharacterTextSplitter(separator="\n\n", chunk_size=2048)    texts = text_splitter.split_documents(documents)local_model_name = 'shibing624_text2vec-base-chinese'    embeddings = HuggingFaceEmbeddings(model_name=local_model_name)db = FAISS.from_documents(texts, embeddings)faiss_index = "vectors_db/hln_tb_faiss_index"db.save_local(faiss_index)# db = FAISS.load_local(faiss_index, embeddings)
while True:question = input("请输入问题: ").replace(" ", "")if question == "stop":breakstart_time = time.time()docs = db.similarity_search(question, k=10)print(docs)for doc in docs:print(doc)        print(f"本次回答共耗时：{time.time() - start_time}")