' PDF财务报表解析主程序Sub ExtractFinancialStatements()    Dim wb As Workbook    Dim wsLog As Worksheet, wsResult As Worksheet    Dim pdfPath As String    Dim acroApp As Object, acroAVDoc As Object, acroPDDoc As Object    Dim startTime As Double    Dim pageCount As Long, tableCount As Long    On Error GoTo ErrorHandler    startTime = Timer    Application.ScreenUpdating = False    Application.Calculation = xlCalculationManual    Set wb = ThisWorkbook    ' 1. 准备工作表    PrepareWorksheets wb    ' 2. 选择PDF文件    pdfPath = SelectPDFFile("请选择财务报表PDF文件")    If pdfPath = "" Then        MsgBox "未选择文件，操作取消", vbInformation        Exit Sub    End If    ' 3. 初始化Acrobat    Set acroApp = CreateObject("AcroExch.App")    If acroApp Is Nothing Then        MsgBox "未找到Adobe Acrobat，请确保已安装Acrobat Pro", vbCritical        Exit Sub    End If    ' 4. 打开PDF文档    Set acroAVDoc = CreateObject("AcroExch.AVDoc")    If Not acroAVDoc.Open(pdfPath, "") Then        MsgBox "无法打开PDF文件", vbCritical        Exit Sub    End If    Set acroPDDoc = acroAVDoc.GetPDDoc    ' 5. 获取PDF信息    pageCount = acroPDDoc.GetNumPages    Debug.Print "PDF页数: " & pageCount    ' 6. 查找财务报表    Dim statementPages As Collection    Set statementPages = FindStatementPages(acroPDDoc)    ' 7. 提取表格数据    tableCount = 0    Dim i As Long    For i = 1 To statementPages.Count        Application.StatusBar = "正在处理第 " & i & "/" & statementPages.Count & " 个财务报表"        Dim pageNum As Long        pageNum = statementPages(i)        ' 提取页面文本        Dim pageText As String        pageText = GetPageText(acroPDDoc, pageNum - 1) ' Acrobat页面从0开始        ' 识别表格类型        Dim statementType As String        statementType = IdentifyStatementType(pageText)        ' 提取表格数据        If statementType <> "未知" Then            Dim tableData As Object            Set tableData = ExtractTableData(acroPDDoc, pageNum - 1, statementType)            If Not tableData Is Nothing Then                ' 写入结果                WriteStatementData wsResult, tableData, statementType, pageNum                tableCount = tableCount + 1            End If        End If    Next i    ' 8. 关闭文档    acroAVDoc.Close True    acroApp.Exit    ' 9. 数据标准化    StandardizeFinancialData wsResult    ' 10. 生成汇总报告    GenerateSummaryReport wb, pdfPath, pageCount, tableCount    ' 11. 清理与报告    Application.ScreenUpdating = True    Application.Calculation = xlCalculationAutomatic    Application.StatusBar = False    Dim elapsedTime As Double    elapsedTime = Timer - startTime    Dim msg As String    msg = "财务报表提取完成！" & vbCrLf & vbCrLf    msg = msg & "统计结果：" & vbCrLf    msg = msg & "PDF总页数：" & pageCount & vbCrLf    msg = msg & "识别报表数：" & tableCount & vbCrLf    msg = msg & "处理耗时：" & Format(elapsedTime, "0.0") & " 秒" & vbCrLf    msg = msg & "平均每表：" & Format(elapsedTime / tableCount, "0.0") & " 秒"    MsgBox msg, vbInformation, "提取完成"    Exit SubErrorHandler:    Application.ScreenUpdating = True    Application.Calculation = xlCalculationAutomatic    Application.StatusBar = False    MsgBox "处理过程中发生错误：" & Err.Description, vbCriticalEnd Sub' 查找财务报表页面Function FindStatementPages(acroPDDoc As Object) As Collection    Dim pages As Collection    Set pages = New Collection    Dim pageCount As Long, i As Long    pageCount = acroPDDoc.GetNumPages    ' 财务报表关键词    Dim statementKeywords As Object    Set statementKeywords = CreateObject("Scripting.Dictionary")    statementKeywords.Add "资产负债表", Array("资产负债表", "财务状况表", "BALANCE SHEET")    statementKeywords.Add "利润表", Array("利润表", "损益表", "INCOME STATEMENT", "PROFIT AND LOSS")    statementKeywords.Add "现金流量表", Array("现金流量表", "CASH FLOW STATEMENT")    ' 搜索每一页    For i = 0 To pageCount - 1        Dim pageText As String        pageText = GetPageText(acroPDDoc, i)        ' 检查是否包含财务报表关键词        Dim key As Variant        For Each key In statementKeywords.Keys            Dim keywords As Variant            keywords = statementKeywords(key)            Dim j As Long            For j = LBound(keywords) To UBound(keywords)                If InStr(1, pageText, keywords(j), vbTextCompare) > 0 Then                    ' 找到财务报表，记录页码                    pages.Add i + 1  ' 转换为1-based页码                    Exit For                End If            Next j        Next key    Next i    Set FindStatementPages = pagesEnd Function' 获取页面文本Function GetPageText(acroPDDoc As Object, pageIndex As Long) As String    Dim jso As Object    Set jso = acroPDDoc.GetJSObject    If Not jso Is Nothing Then        ' 使用Acrobat的JavaScript接口提取文本        GetPageText = jso.getPageNthWords(pageIndex, 0, 10000)    Else        GetPageText = ""    End IfEnd Function' 识别报表类型Function IdentifyStatementType(pageText As String) As String    ' 检查关键词确定报表类型    If InStr(1, pageText, "资产负债表") > 0 Or _       InStr(1, pageText, "BALANCE SHEET") > 0 Then        IdentifyStatementType = "资产负债表"    ElseIf InStr(1, pageText, "利润表") > 0 Or _           InStr(1, pageText, "损益表") > 0 Or _           InStr(1, pageText, "INCOME STATEMENT") > 0 Then        IdentifyStatementType = "利润表"    ElseIf InStr(1, pageText, "现金流量表") > 0 Or _           InStr(1, pageText, "CASH FLOW STATEMENT") > 0 Then        IdentifyStatementType = "现金流量表"    Else        ' 尝试通过内容识别        If InStr(1, pageText, "货币资金") > 0 And _           InStr(1, pageText, "应收账款") > 0 Then            IdentifyStatementType = "资产负债表"        ElseIf InStr(1, pageText, "营业收入") > 0 And _               InStr(1, pageText, "营业成本") > 0 Then            IdentifyStatementType = "利润表"        ElseIf InStr(1, pageText, "经营活动现金流量") > 0 And _               InStr(1, pageText, "投资活动现金流量") > 0 Then            IdentifyStatementType = "现金流量表"        Else            IdentifyStatementType = "未知"        End If    End IfEnd Function' 提取表格数据Function ExtractTableData(acroPDDoc As Object, pageIndex As Long, statementType As String) As Object    Dim jso As Object    Set jso = acroPDDoc.GetJSObject    If jso Is Nothing Then        Set ExtractTableData = Nothing        Exit Function    End If    ' 获取页面所有单词及其位置    Dim words As Object    Set words = jso.getPageNthWords(pageIndex, 0, 10000)    ' 分析页面结构，识别表格    Dim tableData As Object    Set tableData = IdentifyTableStructure(words, statementType)    Set ExtractTableData = tableDataEnd Function' 识别表格结构Function IdentifyTableStructure(words As Object, statementType As String) As Object    ' 这是一个简化的表格识别算法    ' 实际应用需要更复杂的逻辑    Dim tableData As Object    Set tableData = CreateObject("Scripting.Dictionary")    ' 按报表类型设置不同的提取规则    Select Case statementType        Case "资产负债表"            Set tableData = ExtractBalanceSheetStructure(words)        Case "利润表"            Set tableData = ExtractIncomeStatementStructure(words)        Case "现金流量表"            Set tableData = ExtractCashFlowStructure(words)        Case Else            ' 通用表格识别            Set tableData = ExtractGenericTableStructure(words)    End Select    Set IdentifyTableStructure = tableDataEnd Function' 提取资产负债表结构Function ExtractBalanceSheetStructure(words As Object) As Object    Dim tableData As Object    Set tableData = CreateObject("Scripting.Dictionary")    ' 资产负债表常见科目    Dim balanceSheetItems As Object    Set balanceSheetItems = CreateObject("Scripting.Dictionary")    ' 资产类科目    balanceSheetItems.Add "货币资金", ""    balanceSheetItems.Add "交易性金融资产", ""    balanceSheetItems.Add "应收票据", ""    balanceSheetItems.Add "应收账款", ""    balanceSheetItems.Add "预付款项", ""    balanceSheetItems.Add "存货", ""    balanceSheetItems.Add "流动资产合计", ""    balanceSheetItems.Add "固定资产", ""    balanceSheetItems.Add "无形资产", ""    balanceSheetItems.Add "非流动资产合计", ""    balanceSheetItems.Add "资产总计", ""    ' 负债和权益类科目    balanceSheetItems.Add "短期借款", ""    balanceSheetItems.Add "应付票据", ""    balanceSheetItems.Add "应付账款", ""    balanceSheetItems.Add "预收款项", ""    balanceSheetItems.Add "流动负债合计", ""    balanceSheetItems.Add "长期借款", ""    balanceSheetItems.Add "非流动负债合计", ""    balanceSheetItems.Add "负债合计", ""    balanceSheetItems.Add "实收资本", ""    balanceSheetItems.Add "资本公积", ""    balanceSheetItems.Add "未分配利润", ""    balanceSheetItems.Add "所有者权益合计", ""    balanceSheetItems.Add "负债和所有者权益总计", ""    ' 在文本中查找这些科目    Dim i As Long    For i = 0 To words.Count - 1        Dim word As String        word = Trim(words(i))        Dim key As Variant        For Each key In balanceSheetItems.Keys            If InStr(1, word, key) > 0 Then                ' 找到科目，提取后续的数值                balanceSheetItems(key) = ExtractValueAfterWord(words, i)                Exit For            End If        Next key    Next i    Set ExtractBalanceSheetStructure = balanceSheetItemsEnd Function' 提取词语后的数值Function ExtractValueAfterWord(words As Object, wordIndex As Long) As String    Dim value As String    value = ""    ' 向后查找数值    Dim i As Long    For i = wordIndex + 1 To wordIndex + 5        If i < words.Count Then            Dim testWord As String            testWord = Trim(words(i))            ' 检查是否为数值            If IsNumeric(testWord) Then                value = testWord                Exit For            ElseIf InStr(1, testWord, ",") > 0 Then                ' 包含逗号的数值                Dim cleanWord As String                cleanWord = Replace(testWord, ",", "")                If IsNumeric(cleanWord) Then                    value = cleanWord                    Exit For                End If            End If        End If    Next i    ExtractValueAfterWord = valueEnd Function' 写入报表数据Sub WriteStatementData(ws As Worksheet, tableData As Object, statementType As String, pageNum As Long)    Dim lastRow As Long    lastRow = ws.Cells(ws.Rows.Count, "A").End(xlUp).Row + 1    ' 写入表头    If lastRow = 2 Then        ws.Range("A1:E1").Value = Array("报表类型", "科目", "金额", "页码", "提取时间")        ws.Range("A1:E1").Font.Bold = True        ws.Range("A1:E1").Interior.Color = RGB(191, 191, 191)    End If    ' 写入数据    Dim key As Variant, i As Long    i = 0    For Each key In tableData.Keys        Dim value As String        value = tableData(key)        If value <> "" Then            ws.Cells(lastRow + i, 1).Value = statementType            ws.Cells(lastRow + i, 2).Value = key            ws.Cells(lastRow + i, 3).Value = CDbl(value)            ws.Cells(lastRow + i, 4).Value = pageNum            ws.Cells(lastRow + i, 5).Value = Now            ' 设置金额格式            ws.Cells(lastRow + i, 3).NumberFormat = "#,##0.00"            i = i + 1        End If    Next keyEnd Sub' 数据标准化Sub StandardizeFinancialData(ws As Worksheet)    Dim lastRow As Long, i As Long    lastRow = ws.Cells(ws.Rows.Count, "A").End(xlUp).Row    ' 1. 统一计量单位    ' 假设原始数据单位是元，转换为万元    For i = 2 To lastRow        Dim amount As Double        amount = ws.Cells(i, 3).Value        If amount <> 0 Then            ws.Cells(i, 3).Value = amount / 10000        End If    Next i    ' 2. 科目名称标准化    Dim mappingDict As Object    Set mappingDict = CreateObject("Scripting.Dictionary")    ' 建立科目映射    mappingDict.Add "货币资金", "货币资金"    mappingDict.Add "现金及银行存款", "货币资金"    mappingDict.Add "库存现金", "货币资金"    mappingDict.Add "应收账款", "应收账款"    mappingDict.Add "应收帐款", "应收账款"    mappingDict.Add "应收款项", "应收账款"    ' 应用映射    For i = 2 To lastRow        Dim originalName As String        originalName = ws.Cells(i, 2).Value        If mappingDict.Exists(originalName) Then            ws.Cells(i, 2).Value = mappingDict(originalName)        End If    Next i    ' 3. 添加校验列    ws.Cells(1, 6).Value = "校验结果"    For i = 2 To lastRow        Dim statementType As String        statementType = ws.Cells(i, 1).Value        ' 资产负债表校验        If statementType = "资产负债表" Then            Dim accountName As String            accountName = ws.Cells(i, 2).Value            ' 检查资产类科目金额应为正            If InStr(1, accountName, "资产") > 0 And _               Not InStr(1, accountName, "负债") > 0 Then                If ws.Cells(i, 3).Value < 0 Then                    ws.Cells(i, 6).Value = "资产类科目为负"                    ws.Cells(i, 6).Font.Color = RGB(255, 0, 0)                Else                    ws.Cells(i, 6).Value = "正常"                End If            End If        End If    Next iEnd Sub' 生成汇总报告Sub GenerateSummaryReport(wb As Workbook, pdfPath As String, pageCount As Long, tableCount As Long)    Dim wsSummary As Worksheet    On Error Resume Next    Set wsSummary = wb.Worksheets("汇总报告")    If wsSummary Is Nothing Then        Set wsSummary = wb.Worksheets.Add        wsSummary.Name = "汇总报告"    End If    On Error GoTo 0    wsSummary.Cells.Clear    ' 写入汇总信息    With wsSummary        .Range("A1").Value = "PDF财务报表提取汇总报告"        .Range("A1").Font.Bold = True        .Range("A1").Font.Size = 14        .Range("A3").Value = "PDF文件:"        .Range("B3").Value = pdfPath        .Range("A4").Value = "提取时间:"        .Range("B4").Value = Now        .Range("A5").Value = "总页数:"        .Range("B5").Value = pageCount        .Range("A6").Value = "识别报表数:"        .Range("B6").Value = tableCount        .Range("A8").Value = "提取项目统计"        .Range("A8").Font.Bold = True        ' 从提取结果中统计        Dim wsResult As Worksheet        Set wsResult = wb.Worksheets("提取结果")        If Not wsResult Is Nothing Then            Dim lastRow As Long            lastRow = wsResult.Cells(wsResult.Rows.Count, "A").End(xlUp).Row            ' 按报表类型统计            Dim dictTypes As Object            Set dictTypes = CreateObject("Scripting.Dictionary")            Dim i As Long            For i = 2 To lastRow                Dim stype As String                stype = wsResult.Cells(i, 1).Value                If dictTypes.Exists(stype) Then                    dictTypes(stype) = dictTypes(stype) + 1                Else                    dictTypes.Add stype, 1                End If            Next i            ' 写入统计            .Range("A10").Value = "报表类型"            .Range("B10").Value = "科目数量"            .Range("A10:B10").Font.Bold = True            Dim outputRow As Long            outputRow = 11            Dim typeKey As Variant            For Each typeKey In dictTypes.Keys                .Cells(outputRow, 1).Value = typeKey                .Cells(outputRow, 2).Value = dictTypes(typeKey)                outputRow = outputRow + 1            Next typeKey        End If        .Columns.AutoFit    End WithEnd Sub' 准备工作表Sub PrepareWorksheets(wb As Workbook)    ' 结果表    Dim wsResult As Worksheet    On Error Resume Next    Set wsResult = wb.Worksheets("提取结果")    If wsResult Is Nothing Then        Set wsResult = wb.Worksheets.Add        wsResult.Name = "提取结果"    End If    wsResult.Cells.Clear    ' 日志表    Dim wsLog As Worksheet    On Error Resume Next    Set wsLog = wb.Worksheets("处理日志")    If wsLog Is Nothing Then        Set wsLog = wb.Worksheets.Add        wsLog.Name = "处理日志"    End If    wsLog.Cells.Clear    ' 设置日志表头    wsLog.Range("A1:D1").Value = Array("时间", "操作", "结果", "详情")    wsLog.Range("A1:D1").Font.Bold = True    wsLog.Range("A1:D1").Interior.Color = RGB(191, 191, 191)End Sub' 选择PDF文件Function SelectPDFFile(prompt As String) As String    Dim fileDialog As Object    Set fileDialog = Application.FileDialog(msoFileDialogFilePicker)    fileDialog.Title = prompt    fileDialog.AllowMultiSelect = False    fileDialog.Filters.Clear    fileDialog.Filters.Add "PDF文件", "*.pdf"    If fileDialog.Show = -1 Then        SelectPDFFile = fileDialog.SelectedItems(1)    Else        SelectPDFFile = ""    End IfEnd Function

Function DetectTableBoundaries(words As Object) As Object    ' 智能识别表格边界    Dim boundaries As Object    Set boundaries = CreateObject("Scripting.Dictionary")    ' 1. 通过文本对齐方式识别列    ' 假设同一列的数字右对齐，文本左对齐    ' 2. 通过数字模式识别数据行    Dim dataRows As Collection    Set dataRows = New Collection    Dim i As Long, inTable As Boolean    inTable = False    For i = 0 To words.Count - 1        Dim word As String        word = Trim(words(i))        ' 检查是否开始表格        If Not inTable Then            ' 检查表格开始标记            If IsTableStartMarker(word, words, i) Then                inTable = True                boundaries.Add "start", i            End If        Else            ' 检查是否结束表格            If IsTableEndMarker(word, words, i) Then                inTable = False                boundaries.Add "end", i                Exit For            End If        End If    Next i    Set DetectTableBoundaries = boundariesEnd Function

Function ProcessMergedCells(tableData As Object) As Object    ' 处理合并单元格    Dim result As Object    Set result = CreateObject("Scripting.Dictionary")    Dim keys() As String    keys = GetKeys(tableData)    Dim i As Long, currentKey As String    currentKey = ""    For i = 0 To UBound(keys)        Dim key As String, value As String        key = keys(i)        value = tableData(key)        ' 检查是否为空值        If value = "" Then            ' 合并到上一个有值的单元格            If currentKey <> "" Then                ' 可以记录合并关系                result.Add key & "_merged_to", currentKey            End If        Else            result.Add key, value            currentKey = key        End If    Next i    Set ProcessMergedCells = resultEnd Function

Sub ValidateFinancialData(ws As Worksheet)    ' 验证财务数据    Dim lastRow As Long, i As Long    lastRow = ws.Cells(ws.Rows.Count, "A").End(xlUp).Row    ' 检查资产负债表平衡    Dim totalAssets As Double, totalLiabilities As Double    totalAssets = 0    totalLiabilities = 0    For i = 2 To lastRow        If ws.Cells(i, 1).Value = "资产负债表" Then            Dim account As String, amount As Double            account = ws.Cells(i, 2).Value            amount = ws.Cells(i, 3).Value            ' 累加资产            If InStr(1, account, "资产") > 0 And _               Not InStr(1, account, "负债") > 0 And _               Not InStr(1, account, "权益") > 0 Then                totalAssets = totalAssets + amount            End If            ' 累加负债和权益            If InStr(1, account, "负债") > 0 Or _               InStr(1, account, "权益") > 0 Then                totalLiabilities = totalLiabilities + amount            End If        End If    Next i    ' 检查平衡    Dim diff As Double    diff = Abs(totalAssets - totalLiabilities)    If diff < 0.01 Then        Debug.Print "资产负债表平衡"    Else        Debug.Print "资产负债表不平衡，差异: " & diff    End IfEnd Sub

import pandas as pdimport numpy as npimport pdfplumberfrom typing import Dict, List, Optional, Tuple, Anyimport loggingimport refrom datetime import datetimeimport jsonimport osfrom pathlib import Pathimport warningswarnings.filterwarnings('ignore')# 配置日志logging.basicConfig(    level=logging.INFO,    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',    handlers=[        logging.FileHandler('financial_statement_parser.log', encoding='utf-8'),        logging.StreamHandler()    ])logger = logging.getLogger(__name__)class FinancialStatementParser:    """财务报表解析器"""    def __init__(self, config_path: str = 'config/financial_config.json'):        """        初始化解析器        参数:            config_path: 配置文件路径        """        with open(config_path, 'r', encoding='utf-8') as f:            self.config = json.load(f)        # 科目映射配置        self.account_mapping = self.config.get('account_mapping', {})        # 关键词配置        self.keywords = self.config.get('keywords', {})        # 数据处理        self.results = []        self.errors = []        logger.info("财务报表解析器初始化完成")    def parse_pdf(self, pdf_path: str) -> Dict:        """        解析PDF文件        参数:            pdf_path: PDF文件路径        返回:            解析结果字典        """        try:            logger.info(f"开始解析PDF: {pdf_path}")            result = {                'file_path': pdf_path,                'company_name': '',                'report_year': '',                'statements': {},                'metadata': {                    'parse_time': datetime.now().isoformat(),                    'total_pages': 0                }            }            # 打开PDF            with pdfplumber.open(pdf_path) as pdf:                result['metadata']['total_pages'] = len(pdf.pages)                # 提取公司名称和年份                result['company_name'], result['report_year'] = self._extract_metadata(pdf)                # 查找财务报表页面                statement_pages = self._find_statement_pages(pdf)                # 提取财务报表数据                for statement_type, page_nums in statement_pages.items():                    logger.info(f"提取 {statement_type}: 页面 {page_nums}")                    statement_data = self._extract_statement_data(pdf, page_nums, statement_type)                    if statement_data:                        result['statements'][statement_type] = statement_data            logger.info(f"PDF解析完成: {pdf_path}")            return result        except Exception as e:            logger.error(f"解析PDF失败 {pdf_path}: {e}")            self.errors.append({                'file': pdf_path,                'error': str(e)            })            return {}    def _extract_metadata(self, pdf) -> Tuple[str, str]:        """提取元数据（公司名称、报告年份）"""        company_name = ''        report_year = ''        # 检查前几页        for i, page in enumerate(pdf.pages[:5]):            text = page.extract_text()            if not text:                continue            # 提取公司名称            if not company_name:                # 查找常见公司名称模式                name_patterns = [                    r'公司名称[:：]\s*([^\n\r]+)',                    r'(\S+有限公司)',                    r'(\S+股份有限公司)',                    r'(\S+集团)'                ]                for pattern in name_patterns:                    match = re.search(pattern, text)                    if match:                        company_name = match.group(1).strip()                        break            # 提取报告年份            if not report_year:                year_patterns = [                    r'(\d{4})年[\s\S]{0,20}(年度报告|财务报告)',                    r'报告期间[:：]\s*(\d{4})',                    r'截至(\d{4})年'                ]                for pattern in year_patterns:                    match = re.search(pattern, text)                    if match:                        report_year = match.group(1)                        break            if company_name and report_year:                break        return company_name, report_year    def _find_statement_pages(self, pdf) -> Dict[str, List[int]]:        """查找财务报表页面"""        statement_pages = {            'balance_sheet': [],            'income_statement': [],            'cash_flow': []        }        # 财务报表关键词        statement_keywords = {            'balance_sheet': ['资产负债表', '财务状况表', 'BALANCE SHEET'],            'income_statement': ['利润表', '损益表', 'INCOME STATEMENT', 'PROFIT AND LOSS'],            'cash_flow': ['现金流量表', 'CASH FLOW STATEMENT']        }        # 搜索每一页        for i, page in enumerate(pdf.pages):            text = page.extract_text()            if not text:                continue            # 检查是否包含财务报表关键词            for stmt_type, keywords in statement_keywords.items():                for keyword in keywords:                    if keyword in text:                        # 找到财务报表                        statement_pages[stmt_type].append(i + 1)  # 1-based页码                        break        return statement_pages    def _extract_statement_data(self, pdf, page_nums: List[int], statement_type: str) -> List[Dict]:        """提取财务报表数据"""        statement_data = []        for page_num in page_nums:            page = pdf.pages[page_num - 1]  # 转换为0-based            # 提取表格            tables = page.extract_tables({                'vertical_strategy': 'lines',  # 使用线检测                'horizontal_strategy': 'lines',                'explicit_vertical_lines': [],                'explicit_horizontal_lines': [],                'snap_tolerance': 3,                'join_tolerance': 3,                'edge_min_length': 3,                'min_words_vertical': 3,                'min_words_horizontal': 1,                'keep_blank_chars': False,                'text_tolerance': 3,                'text_x_tolerance': 3,                'text_y_tolerance': 3,                'intersection_tolerance': 3,                'intersection_x_tolerance': 3,                'intersection_y_tolerance': 3,            })            if not tables:                # 如果没有检测到表格，尝试文本提取                text_data = self._extract_from_text(page, statement_type)                if text_data:                    statement_data.extend(text_data)                continue            # 处理检测到的表格            for table in tables:                table_data = self._process_table(table, statement_type, page_num)                if table_data:                    statement_data.extend(table_data)        return statement_data    def _process_table(self, table: List[List], statement_type: str, page_num: int) -> List[Dict]:        """处理表格数据"""        if not table or len(table) < 2:            return []        # 识别表头        header_row = self._identify_header(table)        # 处理数据行        data_rows = []        for i, row in enumerate(table):            if i <= header_row:                continue            # 跳过空行            if not any(cell for cell in row if cell and str(cell).strip()):                continue            # 处理行数据            row_data = self._process_row(row, table[header_row], statement_type, page_num, i)            if row_data:                data_rows.append(row_data)        return data_rows    def _identify_header(self, table: List[List]) -> int:        """识别表头行"""        if len(table) < 2:            return 0        # 检查前两行        for i in range(min(3, len(table))):            row = table[i]            # 检查是否包含表头关键词            header_keywords = ['项目', '科目', 'ITEM', 'DESCRIPTION', '名称']            row_text = ' '.join(str(cell) for cell in row if cell)            for keyword in header_keywords:                if keyword in row_text:                    return i        return 0    def _process_row(self, row: List, header: List, statement_type: str, page_num: int, row_num: int) -> Optional[Dict]:        """处理单行数据"""        if len(row) != len(header):            # 调整行长度            if len(row) < len(header):                row = list(row) + [''] * (len(header) - len(row))            else:                row = row[:len(header)]        # 查找科目列        account_col = -1        for i, col in enumerate(header):            if col and any(keyword in str(col) for keyword in ['项目', '科目', 'ITEM', 'DESCRIPTION']):                account_col = i                break        if account_col == -1:            # 如果没有找到科目列，假设第一列是科目            account_col = 0        # 获取科目名称        account = row[account_col]        if not account or not str(account).strip():            return None        account = str(account).strip()        # 标准化科目名称        account = self._standardize_account_name(account, statement_type)        # 提取金额        amounts = []        for i, (cell, header_cell) in enumerate(zip(row, header)):            if i == account_col:                continue            if cell and str(cell).strip():                # 尝试解析金额                amount = self._parse_amount(str(cell))                if amount is not None:                    amounts.append({                        'column': i,                        'header': str(header_cell) if header_cell else f'Col{i}',                        'amount': amount                    })        if not amounts:            return None        # 创建数据记录        record = {            'statement_type': statement_type,            'account': account,            'page_num': page_num,            'row_num': row_num + 1,  # 1-based            'amounts': amounts,            'raw_data': row        }        return record    def _standardize_account_name(self, account: str, statement_type: str) -> str:        """标准化科目名称"""        # 清理科目名称        account = re.sub(r'\s+', ' ', account)  # 合并多个空格        account = account.strip()        # 移除常见前缀后缀        account = re.sub(r'^[一二三四五六七八九十]、\s*', '', account)        account = re.sub(r'^[0-9]+[\.、]\s*', '', account)        account = re.sub(r'\s*[:：]$', '', account)        # 应用科目映射        for pattern, standard_name in self.account_mapping.get(statement_type, {}).items():            if re.search(pattern, account, re.IGNORECASE):                return standard_name        return account    def _parse_amount(self, amount_str: str) -> Optional[float]:        """解析金额字符串"""        if not amount_str or not str(amount_str).strip():            return None        amount_str = str(amount_str).strip()        # 处理常见金额格式        # 1. 移除括号（表示负数）        is_negative = False        if amount_str.startswith('(') and amount_str.endswith(')'):            amount_str = amount_str[1:-1]            is_negative = True        # 2. 移除货币符号、千分位分隔符        amount_str = re.sub(r'[¥￥$,，\s]', '', amount_str)        # 3. 处理中文数字单位        unit_multiplier = 1        if '亿' in amount_str:            unit_multiplier = 100000000            amount_str = amount_str.replace('亿', '')        elif '万' in amount_str:            unit_multiplier = 10000            amount_str = amount_str.replace('万', '')        elif '千' in amount_str:            unit_multiplier = 1000            amount_str = amount_str.replace('千', '')        # 4. 尝试转换为数字        try:            # 处理百分数            if '%' in amount_str:                amount_str = amount_str.replace('%', '')                value = float(amount_str) / 100            else:                value = float(amount_str)            # 应用单位乘数            value *= unit_multiplier            # 应用符号            if is_negative:                value = -value            return value        except ValueError:            # 尝试其他格式            try:                # 处理科学计数法                if 'E' in amount_str.upper():                    return float(amount_str)            except:                pass            return None    def _extract_from_text(self, page, statement_type: str) -> List[Dict]:        """从文本中提取数据"""        text = page.extract_text()        if not text:            return []        # 按行分割文本        lines = text.split('\n')        data = []        for i, line in enumerate(lines):            line = line.strip()            if not line:                continue            # 查找科目和金额模式            patterns = [                # 模式1: 科目 金额                r'^([^\d]+?)\s+([-\(]?\d[\d,\.]*[\)]?)(?:\s|$)',                # 模式2: 科目: 金额                r'^([^:]+?):\s*([-\(]?\d[\d,\.]*[\)]?)(?:\s|$)',            ]            for pattern in patterns:                match = re.match(pattern, line)                if match:                    account = match.group(1).strip()                    amount_str = match.group(2).strip()                    # 标准化科目                    account = self._standardize_account_name(account, statement_type)                    # 解析金额                    amount = self._parse_amount(amount_str)                    if amount is not None:                        data.append({                            'statement_type': statement_type,                            'account': account,                            'amount': amount,                            'source': 'text',                            'line_num': i + 1                        })                    break        return data    def validate_data(self, data: Dict) -> List[str]:        """验证财务数据"""        errors = []        # 检查必要数据        if not data.get('statements'):            errors.append("未找到财务报表数据")            return errors        statements = data['statements']        # 验证资产负债表        if 'balance_sheet' in statements:            bs_errors = self._validate_balance_sheet(statements['balance_sheet'])            errors.extend(bs_errors)        # 验证利润表        if 'income_statement' in statements:            is_errors = self._validate_income_statement(statements['income_statement'])            errors.extend(is_errors)        # 验证现金流量表        if 'cash_flow' in statements:            cf_errors = self._validate_cash_flow(statements['cash_flow'])            errors.extend(cf_errors)        return errors    def _validate_balance_sheet(self, data: List[Dict]) -> List[str]:        """验证资产负债表"""        errors = []        # 计算资产总计和负债权益总计        total_assets = 0        total_liabilities_equity = 0        asset_keywords = ['资产总计', '总资产', 'TOTAL ASSETS']        liability_keywords = ['负债和所有者权益总计', '负债合计', 'TOTAL LIABILITIES']        for record in data:            account = record['account']            # 累加资产总计            if any(keyword in account for keyword in asset_keywords):                for amount_info in record['amounts']:                    total_assets += amount_info['amount']            # 累加负债权益总计            if any(keyword in account for keyword in liability_keywords):                for amount_info in record['amounts']:                    total_liabilities_equity += amount_info['amount']        # 检查平衡        if total_assets and total_liabilities_equity:            diff = abs(total_assets - total_liabilities_equity)            if diff > 0.01:                errors.append(f"资产负债表不平衡: 资产={total_assets:.2f}, 负债权益={total_liabilities_equity:.2f}, 差异={diff:.2f}")        return errors    def export_to_excel(self, data: Dict, output_path: Optional[str] = None) -> str:        """        导出到Excel        参数:            data: 解析的数据            output_path: 输出文件路径        返回:            输出文件路径        """        if not data or not data.get('statements'):            logger.warning("无数据可导出")            return ""        if output_path is None:            company = data.get('company_name', '未知公司')            year = data.get('report_year', '未知年份')            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')            output_path = f'{company}_{year}_财务报表_{timestamp}.xlsx'        with pd.ExcelWriter(output_path, engine='openpyxl') as writer:            # 1. 元数据            metadata = {                '项目': ['公司名称', '报告年份', '解析时间', 'PDF文件', '总页数'],                '值': [                    data.get('company_name', ''),                    data.get('report_year', ''),                    data.get('metadata', {}).get('parse_time', ''),                    data.get('file_path', ''),                    data.get('metadata', {}).get('total_pages', 0)                ]            }            metadata_df = pd.DataFrame(metadata)            metadata_df.to_excel(writer, sheet_name='元数据', index=False)            # 2. 财务报表数据            statements = data.get('statements', {})            for stmt_type, stmt_data in statements.items():                if not stmt_data:                    continue                # 转换为DataFrame                rows = []                for record in stmt_data:                    for amount_info in record.get('amounts', []):                        rows.append({                            '报表类型': self._get_statement_name(stmt_type),                            '科目': record.get('account', ''),                            '金额': amount_info.get('amount', 0),                            '期间': amount_info.get('header', ''),                            '页码': record.get('page_num', ''),                            '行号': record.get('row_num', '')                        })                if rows:                    df = pd.DataFrame(rows)                    # 按报表类型选择工作表名称                    sheet_name = self._get_sheet_name(stmt_type)                    df.to_excel(writer, sheet_name=sheet_name, index=False)            # 3. 验证结果            validation_errors = self.validate_data(data)            if validation_errors:                errors_df = pd.DataFrame({'验证错误': validation_errors})                errors_df.to_excel(writer, sheet_name='验证结果', index=False)        logger.info(f"数据已导出: {output_path}")        return output_path    def _get_statement_name(self, stmt_type: str) -> str:        """获取报表类型名称"""        mapping = {            'balance_sheet': '资产负债表',            'income_statement': '利润表',            'cash_flow': '现金流量表'        }        return mapping.get(stmt_type, stmt_type)    def _get_sheet_name(self, stmt_type: str) -> str:        """获取工作表名称"""        mapping = {            'balance_sheet': '资产负债表',            'income_statement': '利润表',            'cash_flow': '现金流量表'        }        name = mapping.get(stmt_type, stmt_type)        # 确保工作表名称不超过31个字符        return name[:31]    def batch_process(self, folder_path: str) -> List[Dict]:        """        批量处理文件夹中的PDF文件        参数:            folder_path: 文件夹路径        返回:            处理结果列表        """        folder_path = Path(folder_path)        if not folder_path.exists():            logger.error(f"文件夹不存在: {folder_path}")            return []        # 获取所有PDF文件        pdf_files = list(folder_path.glob("**/*.pdf"))        logger.info(f"找到 {len(pdf_files)} 个PDF文件")        # 处理每个文件        all_results = []        for pdf_file in pdf_files:            result = self.parse_pdf(str(pdf_file))            if result:                all_results.append(result)                # 导出单个文件结果                output_path = f"output/{pdf_file.stem}_解析结果.xlsx"                self.export_to_excel(result, output_path)        # 生成合并报告        if all_results:            self._generate_consolidated_report(all_results)        logger.info(f"批量处理完成: 成功 {len(all_results)}/{len(pdf_files)}")        return all_results    def _generate_consolidated_report(self, results: List[Dict]):        """生成合并报告"""        if not results:            return        # 提取关键数据        consolidated_data = []        for result in results:            company = result.get('company_name', '未知公司')            year = result.get('report_year', '未知年份')            # 提取关键财务指标            key_metrics = self._extract_key_metrics(result)            for metric, value in key_metrics.items():                consolidated_data.append({                    '公司': company,                    '年份': year,                    '指标': metric,                    '数值': value                })        # 导出到Excel        if consolidated_data:            df = pd.DataFrame(consolidated_data)            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')            output_path = f'output/财务数据对比_{timestamp}.xlsx'            with pd.ExcelWriter(output_path, engine='openpyxl') as writer:                df.to_excel(writer, sheet_name='对比数据', index=False)                # 创建数据透视表                pivot = df.pivot_table(                    index='公司',                    columns=['年份', '指标'],                    values='数值',                    aggfunc='sum'                )                pivot.to_excel(writer, sheet_name='数据透视')            logger.info(f"合并报告已生成: {output_path}")# 使用示例def main():    """主程序示例"""    # 1. 创建解析器    config = {        "account_mapping": {            "balance_sheet": {                "货币资金|现金及银行存款": "货币资金",                "应收账款|应收帐款": "应收账款",                "固定资产|固定资产净额": "固定资产",                "资产总计|总资产": "资产总计",                "负债合计|总负债": "负债合计",                "所有者权益合计|净资产": "所有者权益合计"            },            "income_statement": {                "营业收入|主营业务收入": "营业收入",                "营业成本|主营业务成本": "营业成本",                "净利润|税后利润": "净利润"            }        },        "keywords": {            "balance_sheet": ["资产负债表", "财务状况表"],            "income_statement": ["利润表", "损益表"],            "cash_flow": ["现金流量表"]        }    }    # 保存配置文件    os.makedirs('config', exist_ok=True)    with open('config/financial_config.json', 'w', encoding='utf-8') as f:        json.dump(config, f, ensure_ascii=False, indent=2)    parser = FinancialStatementParser('config/financial_config.json')    # 2. 解析单个PDF    pdf_path = "data/茅台_2023年报.pdf"    if os.path.exists(pdf_path):        result = parser.parse_pdf(pdf_path)        if result:            print(f"\n解析完成!")            print(f"公司: {result.get('company_name')}")            print(f"年份: {result.get('report_year')}")            print(f"找到报表: {', '.join(result.get('statements', {}).keys())}")            # 导出到Excel            excel_path = parser.export_to_excel(result)            print(f"结果已导出: {excel_path}")    else:        print(f"文件不存在: {pdf_path}")        print("尝试批量处理...")        # 3. 批量处理        folder_path = "data/annual_reports"        if os.path.exists(folder_path):            results = parser.batch_process(folder_path)            print(f"\n批量处理完成: {len(results)} 个文件")        else:            print(f"文件夹不存在: {folder_path}")if __name__ == "__main__":    main()

def detect_tables_with_dl(pdf_path: str):    """使用深度学习检测表格"""    try:        import cv2        import numpy as np        from paddleocr import PaddleOCR        # 使用PaddleOCR进行表格识别        ocr = PaddleOCR(use_angle_cls=True, lang='ch')        # 将PDF转换为图片        images = convert_pdf_to_images(pdf_path)        tables = []        for img in images:            # 表格结构识别            result = ocr.ocr(np.array(img), cls=True)            # 解析表格结构            table_structure = parse_table_structure(result)            tables.append(table_structure)        return tables    except ImportError:        logger.warning("未安装深度学习库，使用传统方法")        return []

def intelligent_data_validation(data: Dict) -> Dict:    """智能数据校验"""    validation_results = {        'warnings': [],        'errors': [],        'suggestions': []    }    # 检查数据一致性    statements = data.get('statements', {})    # 检查资产负债表平衡    if 'balance_sheet' in statements:        bs_data = statements['balance_sheet']        # 计算关键比率        ratios = calculate_financial_ratios(statements)        # 检查异常值        for ratio_name, ratio_value in ratios.items():            if ratio_name == 'debt_to_equity' and ratio_value > 3:                validation_results['warnings'].append(f"资产负债率异常高: {ratio_value:.2f}")            elif ratio_name == 'current_ratio' and ratio_value < 1:                validation_results['warnings'].append(f"流动比率异常低: {ratio_value:.2f}")    # 检查趋势合理性    if has_historical_data(data):        trends = analyze_financial_trends(data)        for trend_name, trend_value in trends.items():            if abs(trend_value) > 0.5:  # 变化超过50%                validation_results['suggestions'].append(f"{trend_name}变化较大: {trend_value:.1%}")    return validation_results

def generate_analysis_report(data: Dict, template_path: str) -> str:    """生成分析报告"""    from jinja2 import Template    # 读取模板    with open(template_path, 'r', encoding='utf-8') as f:        template_content = f.read()    # 准备数据    report_data = {        'company_name': data.get('company_name', ''),        'report_year': data.get('report_year', ''),        'extract_time': data.get('metadata', {}).get('parse_time', ''),        'key_metrics': extract_key_metrics(data),        'validation_results': intelligent_data_validation(data),        'charts': generate_charts_data(data)    }    # 渲染报告    template = Template(template_content)    html_report = template.render(**report_data)    # 保存报告    report_path = f"reports/{report_data['company_name']}_{report_data['report_year']}_分析报告.html"    with open(report_path, 'w', encoding='utf-8') as f:        f.write(html_report)    return report_path

def multi_company_comparison(company_data: List[Dict]) -> pd.DataFrame:    """多公司对比分析"""    comparison_data = []    for data in company_data:        company = data.get('company_name')        year = data.get('report_year')        # 提取关键指标        metrics = extract_key_metrics(data)        for metric_name, metric_value in metrics.items():            comparison_data.append({                '公司': company,                '年份': year,                '指标': metric_name,                '数值': metric_value            })    df = pd.DataFrame(comparison_data)    # 创建数据透视表    pivot_table = df.pivot_table(        index='公司',        columns=['年份', '指标'],        values='数值',        aggfunc='sum'    )    # 计算行业平均    industry_avg = pivot_table.mean()    # 计算排名    rankings = pivot_table.rank(ascending=False)    return {        'raw_data': df,        'pivot_table': pivot_table,        'industry_average': industry_avg,        'rankings': rankings    }