使用Python解析五大主流文档从PDF到EPUB的全攻略

import fitz  # pymupdf

# 打开pdf文件
doc = fitz.open('example.pdf')

# 提取所有文本
all_text = ""
for page in doc:
    all_text += page.get_text()

# 提取页面中的图像
for page_num in range(len(doc)):
    page = doc.load_page(page_num)
    image_list = page.get_images()
    
    for img_index, img in enumerate(image_list):
        xref = img[0]
        pix = fitz.pixmap(doc, xref)
        if pix.n - pix.alpha > 3:  # 检查是否为rgb图像
            pix = fitz.pixmap(fitz.csrgb, pix)
        pix.save(f"page_{page_num}_img_{img_index}.png")

# 提取带格式的文本（保留位置信息）
for page in doc:
    blocks = page.get_text("dict")["blocks"]
    for block in blocks:
        if "lines" in block:  # 文本块
            for line in block["lines"]:
                for span in line["spans"]:
                    print(f"文本: {span['text']}")
                    print(f"字体: {span['font']}")
                    print(f"大小: {span['size']}")
                    print(f"位置: {span['bbox']}")

doc.close()

import pdfplumber

with pdfplumber.open('example.pdf') as pdf:
    # 提取第一页的文本
    first_page = pdf.pages[0]
    text = first_page.extract_text()
    print(text)
    
    # 提取表格
    tables = first_page.extract_tables()
    for table in tables:
        for row in table:
            print(row)
    
    # 可视化文本位置（调试用）
    im = first_page.to_image()
    im.debug_tablefinder().show()

from unstructured.partition.pdf import partition_pdf

# 解析pdf并自动识别元素类型
elements = partition_pdf(
    filename="example.pdf",
    strategy="auto",  # 自动选择解析策略
    infer_table_structure=true,  # 推断表格结构
    include_page_breaks=true  # 包含分页符
)

# 查看识别出的元素类型
for element in elements:
    print(f"类型: {type(element).__name__}")
    print(f"文本: {str(element)[:100]}...")
    print("-" * 50)
    
    # 不同类型的元素有不同的属性
    if hasattr(element, 'category'):
        print(f"分类: {element.category}")

import fitz

def extract_chinese_pdf(pdf_path):
    doc = fitz.open(pdf_path)
    
    # 方法1：尝试直接提取
    text = ""
    for page in doc:
        text += page.get_text()
    
    # 如果提取的中文是乱码，尝试指定编码
    if "乱码" in text or len(text.strip()) < 10:
        print("检测到可能的编码问题，尝试其他方法...")
        
        # 方法2：使用ocr（需要安装额外的依赖）
        # 这里展示思路，实际需要安装pytesseract和pillow
        # import pytesseract
        # from pil import image
        # 
        # for page_num in range(len(doc)):
        #     pix = doc[page_num].get_pixmap()
        #     img = image.frombytes("rgb", [pix.width, pix.height], pix.samples)
        #     text += pytesseract.image_to_string(img, lang='chi_sim')
    
    return text

import markdown

# 基本使用：将markdown转换为html
md_text = """
# 标题一

这是一个段落，包含**加粗**和*斜体*文本。

- 列表项1
- 列表项2

[这是一个链接](https://example.com)
"""

html_output = markdown.markdown(md_text)
print(html_output)

# 使用扩展功能
html_with_extensions = markdown.markdown(
    md_text,
    extensions=[
        'toc',           # 目录生成
        'tables',        # 表格支持
        'fenced_code',   # 代码块
        'footnotes'      # 脚注
    ]
)

# 解析为ast（抽象语法树）
from markdown.extensions import extension
from markdown.treeprocessors import treeprocessor

class linkcollector(treeprocessor):
    def run(self, root):
        links = []
        for element in root.iter():
            if element.tag == 'a':
                links.append(element.get('href'))
        return links

class myextension(extension):
    def extendmarkdown(self, md):
        md.treeprocessors.register(linkcollector(md), 'linkcollector', 15)

md = markdown.markdown(extensions=[myextension()])
result = md.convert(md_text)
print(f"找到的链接: {md.treeprocessors['linkcollector'].run(md.parser.root)}")

from markdown_it import markdownit
from markdown_it.tree import syntaxtreenode

# 创建解析器实例
md = markdownit()

# 解析markdown
tokens = md.parse(md_text)

# 遍历语法标记
for token in tokens:
    if token.type == 'heading_open' and token.tag == 'h1':
        print("找到一个一级标题")
    elif token.type == 'inline':
        print(f"文本内容: {token.content}")

# 转换为语法树
tree = syntaxtreenode(tokens)
for node in tree.walk():
    if node.type == 'heading' and node.tag == 'h1':
        print(f"标题: {node.children[0].content}")

# 基础读取
def read_txt_basic(filepath):
    with open(filepath, 'r', encoding='utf-8') as f:
        content = f.read()
    return content

# 处理大文件（逐行读取）
def process_large_txt(filepath):
    with open(filepath, 'r', encoding='utf-8') as f:
        for line_number, line in enumerate(f, 1):
            # 处理每一行
            processed_line = line.strip()
            if processed_line:  # 跳过空行
                print(f"行 {line_number}: {processed_line[:50]}...")

# 自动检测编码
import chardet

def read_txt_with_encoding_detection(filepath):
    with open(filepath, 'rb') as f:
        raw_data = f.read()
        result = chardet.detect(raw_data)
        encoding = result['encoding']
        confidence = result['confidence']
        
        print(f"检测到编码: {encoding} (置信度: {confidence})")
        
        try:
            return raw_data.decode(encoding)
        except unicodedecodeerror:
            # 如果检测失败，尝试常见编码
            for enc in ['utf-8', 'gbk', 'gb2312', 'latin-1']:
                try:
                    return raw_data.decode(enc)
                except unicodedecodeerror:
                    continue
            raise

# 使用pandas处理结构化文本数据
import pandas as pd

def process_tabular_txt(filepath):
    # 读取csv/tsv文件
    try:
        # 尝试用逗号分隔
        df = pd.read_csv(filepath, encoding='utf-8')
    except:
        try:
            # 尝试用制表符分隔
            df = pd.read_csv(filepath, sep='\t', encoding='utf-8')
        except:
            # 尝试自动检测分隔符
            with open(filepath, 'r') as f:
                first_line = f.readline()
            
            if ',' in first_line:
                df = pd.read_csv(filepath, sep=',', encoding='utf-8')
            elif '\t' in first_line:
                df = pd.read_csv(filepath, sep='\t', encoding='utf-8')
            else:
                df = pd.read_csv(filepath, delim_whitespace=true, encoding='utf-8')
    
    # 数据分析示例
    print(f"数据形状: {df.shape}")
    print(f"列名: {df.columns.tolist()}")
    print(f"前5行:\n{df.head()}")
    
    return df

from docx import document
from docx.document import document as docdocument

def read_docx(filepath):
    # 打开文档
    doc = document(filepath)
    
    # 提取所有段落
    full_text = []
    for paragraph in doc.paragraphs:
        if paragraph.text.strip():  # 跳过空段落
            full_text.append(paragraph.text)
            print(f"段落: {paragraph.text[:50]}...")
    
    # 提取表格数据
    tables_data = []
    for table in doc.tables:
        table_data = []
        for row in table.rows:
            row_data = [cell.text for cell in row.cells]
            table_data.append(row_data)
        tables_data.append(table_data)
        print(f"表格找到，有{len(table.rows)}行{len(table.columns)}列")
    
    # 提取样式信息
    styled_elements = []
    for paragraph in doc.paragraphs:
        style_info = {
            'text': paragraph.text,
            'style': paragraph.style.name,
            'runs': []
        }
        
        # 获取运行级别的格式
        for run in paragraph.runs:
            run_info = {
                'text': run.text,
                'bold': run.bold,
                'italic': run.italic,
                'underline': run.underline,
                'font_name': run.font.name,
                'font_size': run.font.size
            }
            style_info['runs'].append(run_info)
        
        if style_info['runs']:
            styled_elements.append(style_info)
    
    # 处理列表
    lists = []
    for paragraph in doc.paragraphs:
        if paragraph.style.name.startswith('list'):
            lists.append({
                'text': paragraph.text,
                'style': paragraph.style.name,
                'level': get_list_level(paragraph.style.name)
            })
    
    return {
        'full_text': '\n'.join(full_text),
        'tables': tables_data,
        'styled_elements': styled_elements,
        'lists': lists
    }

def get_list_level(style_name):
    """获取列表缩进级别"""
    if '1' in style_name:
        return 1
    elif '2' in style_name:
        return 2
    elif '3' in style_name:
        return 3
    else:
        return 0

# 处理旧版.doc文件（需要额外的库）
def read_doc_file(filepath):
    # 注意：python-docx只能处理.docx文件
    # 处理.doc文件需要安装antiword或使用其他方法
    
    # 方法1：使用libreoffice转换（需要系统安装libreoffice）
    # import subprocess
    # subprocess.run(['libreoffice', '--headless', '--convert-to', 'docx', filepath])
    
    # 方法2：使用pywin32（仅windows）
    # import win32com.client
    # word = win32com.client.dispatch("word.application")
    # doc = word.documents.open(filepath)
    # text = doc.content.text
    # doc.close()
    # word.quit()
    
    print("处理.doc文件需要额外的工具")
    return none

from ebooklib import epub
import html2text

def read_epub(filepath):
    # 打开epub文件
    book = epub.read_epub(filepath)
    
    # 获取书籍元数据
    metadata = {
        'title': book.get_metadata('dc', 'title'),
        'creator': book.get_metadata('dc', 'creator'),
        'publisher': book.get_metadata('dc', 'publisher'),
        'date': book.get_metadata('dc', 'date'),
        'language': book.get_metadata('dc', 'language'),
        'identifier': book.get_metadata('dc', 'identifier')
    }
    
    print(f"书名: {metadata['title']}")
    print(f"作者: {metadata['creator']}")
    
    # 提取所有文本内容
    h = html2text.html2text()
    h.ignore_links = false
    h.ignore_images = false
    
    full_text = []
    toc_items = []
    
    # 处理目录
    for item in book.toc:
        if isinstance(item, tuple):
            # 处理嵌套目录项
            section, subsections = item
            toc_items.append({
                'title': section.title,
                'href': section.href
            })
        else:
            toc_items.append({
                'title': item.title,
                'href': item.href
            })
    
    # 按章节读取内容
    for item in book.get_items():
        if item.get_type() == ebooklib.item_document:
            # 获取章节内容（html格式）
            content = item.get_content().decode('utf-8')
            
            # 转换为纯文本
            text_content = h.handle(content)
            
            # 清理文本
            cleaned_text = clean_epub_text(text_content)
            
            if cleaned_text.strip():
                full_text.append({
                    'title': item.get_name(),
                    'content': cleaned_text,
                    'raw_html': content[:500] + '...'  # 保存部分html供参考
                })
    
    # 按目录顺序组织内容
    organized_content = organize_by_toc(full_text, toc_items)
    
    return {
        'metadata': metadata,
        'toc': toc_items,
        'content': organized_content,
        'full_text': '\n\n'.join([item['content'] for item in full_text])
    }

def clean_epub_text(text):
    """清理epub文本中的多余空白和标记"""
    lines = text.split('\n')
    cleaned_lines = []
    
    for line in lines:
        line = line.strip()
        if line and not line.startswith('#' * 4):  # 跳过html2text的标题标记
            cleaned_lines.append(line)
    
    return '\n'.join(cleaned_lines)

def organize_by_toc(content_items, toc):
    """根据目录组织内容"""
    organized = []
    
    for toc_item in toc:
        # 查找对应章节
        for content_item in content_items:
            if toc_item['href'] in content_item['title']:
                organized.append({
                    'toc_title': toc_item['title'],
                    'content_title': content_item['title'],
                    'content': content_item['content']
                })
                break
    
    return organized

class universaldocumentparser:
    def __init__(self):
        self.supported_formats = {
            '.pdf': self._parse_pdf,
            '.md': self._parse_markdown,
            '.txt': self._parse_text,
            '.docx': self._parse_docx,
            '.epub': self._parse_epub
        }
    
    def parse(self, filepath):
        import os
        
        # 获取文件扩展名
        _, ext = os.path.splitext(filepath)
        ext = ext.lower()
        
        # 检查是否支持该格式
        if ext not in self.supported_formats:
            raise valueerror(f"不支持的文件格式: {ext}")
        
        # 调用对应的解析函数
        return self.supported_formats[ext](filepath)
    
    def _parse_pdf(self, filepath):
        """解析pdf文件"""
        # 根据需求选择合适的pdf解析器
        try:
            # 首先尝试使用pymupdf（速度快）
            import fitz
            doc = fitz.open(filepath)
            text = ""
            for page in doc:
                text += page.get_text() + "\n"
            doc.close()
            return {'format': 'pdf', 'content': text, 'parser': 'pymupdf'}
        except importerror:
            # 回退到其他解析器
            try:
                import pdfplumber
                with pdfplumber.open(filepath) as pdf:
                    text = ""
                    for page in pdf.pages:
                        text += page.extract_text() + "\n"
                return {'format': 'pdf', 'content': text, 'parser': 'pdfplumber'}
            except importerror:
                raise importerror("请安装pymupdf或pdfplumber以解析pdf文件")
    
    def _parse_markdown(self, filepath):
        """解析markdown文件"""
        import markdown
        with open(filepath, 'r', encoding='utf-8') as f:
            content = f.read()
        
        # 转换为html
        html = markdown.markdown(content)
        
        return {
            'format': 'markdown',
            'raw_content': content,
            'html_content': html
        }
    
    def _parse_text(self, filepath):
        """解析纯文本文件"""
        # 自动检测编码
        import chardet
        
        with open(filepath, 'rb') as f:
            raw_data = f.read()
            result = chardet.detect(raw_data)
            encoding = result['encoding']
        
        with open(filepath, 'r', encoding=encoding) as f:
            content = f.read()
        
        return {
            'format': 'text',
            'encoding': encoding,
            'content': content
        }
    
    def _parse_docx(self, filepath):
        """解析docx文件"""
        from docx import document
        
        doc = document(filepath)
        paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]
        
        # 提取表格
        tables = []
        for table in doc.tables:
            table_data = []
            for row in table.rows:
                table_data.append([cell.text for cell in row.cells])
            tables.append(table_data)
        
        return {
            'format': 'docx',
            'paragraphs': paragraphs,
            'tables': tables,
            'full_text': '\n'.join(paragraphs)
        }
    
    def _parse_epub(self, filepath):
        """解析epub文件"""
        import ebooklib
        from ebooklib import epub
        import html2text
        
        book = epub.read_epub(filepath)
        h = html2text.html2text()
        h.ignore_links = true
        
        # 提取所有文本内容
        text_parts = []
        for item in book.get_items():
            if item.get_type() == ebooklib.item_document:
                content = item.get_content().decode('utf-8')
                text = h.handle(content)
                if text.strip():
                    text_parts.append(text)
        
        full_text = '\n\n'.join(text_parts)
        
        # 提取元数据
        metadata = {}
        for key in ['title', 'creator', 'publisher', 'date']:
            meta = book.get_metadata('dc', key)
            if meta:
                metadata[key] = meta[0][0]
        
        return {
            'format': 'epub',
            'metadata': metadata,
            'content': full_text
        }

# 使用示例
parser = universaldocumentparser()

# 解析各种格式的文件
formats_to_test = ['document.pdf', 'notes.md', 'data.txt', 'report.docx', 'book.epub']

for file in formats_to_test:
    try:
        result = parser.parse(file)
        print(f"成功解析 {file}: {result['format']} 格式")
        print(f"内容预览: {result.get('content', result.get('full_text', ''))[:100]}...")
        print("-" * 50)
    except filenotfounderror:
        print(f"文件不存在: {file}")
    except exception as e:
        print(f"解析 {file} 时出错: {e}")

class efficientdocumentprocessor:
    def __init__(self, chunk_size=1000):
        self.chunk_size = chunk_size  # 每次处理的块大小
    
    def process_large_pdf(self, filepath, callback=none):
        """流式处理大型pdf文件"""
        import fitz
        
        doc = fitz.open(filepath)
        total_pages = len(doc)
        
        for page_num in range(total_pages):
            page = doc.load_page(page_num)
            text = page.get_text()
            
            # 分块处理文本
            chunks = self._split_into_chunks(text)
            
            for chunk in chunks:
                if callback:
                    callback(chunk, page_num + 1)
                else:
                    yield chunk, page_num + 1
        
        doc.close()
    
    def _split_into_chunks(self, text, chunk_size=none):
        """将文本分割为指定大小的块"""
        if chunk_size is none:
            chunk_size = self.chunk_size
        
        chunks = []
        words = text.split()
        
        current_chunk = []
        current_size = 0
        
        for word in words:
            word_size = len(word) + 1  # 加1是空格
            
            if current_size + word_size > chunk_size and current_chunk:
                chunks.append(' '.join(current_chunk))
                current_chunk = [word]
                current_size = word_size
            else:
                current_chunk.append(word)
                current_size += word_size
        
        if current_chunk:
            chunks.append(' '.join(current_chunk))
        
        return chunks

import logging
from functools import wraps

# 配置日志
logging.basicconfig(
    level=logging.info,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getlogger(__name__)

def document_parser_error_handler(func):
    """文档解析器的错误处理装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except filenotfounderror as e:
            logger.error(f"文件未找到: {e.filename}")
            raise
        except permissionerror as e:
            logger.error(f"权限不足: {e.filename}")
            raise
        except unicodedecodeerror as e:
            logger.error(f"编码错误: {e.reason}")
            # 尝试其他编码
            return handle_encoding_error(*args, **kwargs)
        except exception as e:
            logger.exception(f"解析文档时发生未知错误: {e}")
            raise
    return wrapper

@document_parser_error_handler
def safe_document_parse(filepath, parser_func):
    """安全的文档解析函数"""
    return parser_func(filepath)