Fix SSE route dependency and align architecture docs

This commit is contained in:
ash66
2026-05-18 16:32:42 +08:00
parent 86b9ac806a
commit 3f69cad404
149 changed files with 4786 additions and 5957 deletions

View File

@@ -1,4 +1,4 @@
"""Word文档解析 - 使用python-docx"""
"""Provide service-layer logic for docx parser."""
from docx import Document
from docx.enum.text import WD_ALIGN_PARAGRAPH
@@ -6,27 +6,29 @@ from typing import List, Dict, Optional
from dataclasses import dataclass, field
from loguru import logger
import re
# Keep service responsibilities explicit so downstream behavior stays predictable.
@dataclass
class DocxParagraph:
"""段落内容"""
"""Represent the Docx Paragraph type."""
text: str
level: int = 0 # 标题级别0表示正文
level: int = 0 # Keep service responsibilities explicit so downstream behavior stays predictable.
is_list: bool = False
list_number: Optional[str] = None
@dataclass
class DocxTable:
"""表格内容"""
"""Represent the Docx Table type."""
rows: List[List[str]]
markdown: str = ""
@dataclass
class DocxDocumentContent:
"""Word文档完整内容"""
"""Represent the Docx Document Content type."""
file_path: str
paragraphs: List[DocxParagraph]
tables: List[DocxTable]
@@ -35,21 +37,14 @@ class DocxDocumentContent:
class DocxParser:
"""Word文档解析器 - 基于python-docx"""
"""Provide the Docx Parser parser."""
def __init__(self):
"""Initialize the Docx Parser instance."""
self.document = None
def parse(self, file_path: str) -> DocxDocumentContent:
"""
解析Word文档
Args:
file_path: Word文档路径
Returns:
DocxDocumentContent: 解析后的文档内容
"""
"""Handle parse for the Docx Parser instance."""
logger.info(f"开始解析Word文档: {file_path}")
try:
@@ -60,16 +55,16 @@ class DocxParser:
tables=[]
)
# 提取文档元数据
# Keep service responsibilities explicit so downstream behavior stays predictable.
doc_content.metadata = self._extract_metadata()
# 提取段落
# Keep service responsibilities explicit so downstream behavior stays predictable.
doc_content.paragraphs = self._extract_paragraphs()
# 提取表格
# Keep service responsibilities explicit so downstream behavior stays predictable.
doc_content.tables = self._extract_tables()
# 生成Markdown格式文本
# Keep service responsibilities explicit so downstream behavior stays predictable.
doc_content.markdown_text = self._generate_markdown(doc_content)
logger.success(f"Word文档解析完成{len(doc_content.paragraphs)}个段落")
@@ -81,7 +76,7 @@ class DocxParser:
raise
def _extract_metadata(self) -> Dict[str, str]:
"""提取文档元数据"""
"""Handle extract metadata for this module for the Docx Parser instance."""
metadata = {}
try:
core_props = self.document.core_properties
@@ -98,7 +93,7 @@ class DocxParser:
return metadata
def _extract_paragraphs(self) -> List[DocxParagraph]:
"""提取所有段落"""
"""Handle extract paragraphs for this module for the Docx Parser instance."""
paragraphs = []
for para in self.document.paragraphs:
@@ -106,10 +101,10 @@ class DocxParser:
if not text:
continue
# 判断标题级别
# Keep service responsibilities explicit so downstream behavior stays predictable.
level = self._get_paragraph_level(para)
# 判断是否是列表项
# Keep service responsibilities explicit so downstream behavior stays predictable.
is_list, list_number = self._detect_list_item(para)
paragraph = DocxParagraph(
@@ -123,66 +118,61 @@ class DocxParser:
return paragraphs
def _get_paragraph_level(self, para) -> int:
"""
判断段落标题级别
Returns:
int: 标题级别0表示正文
"""
# 方法1检查段落样式
"""Handle get paragraph level for this module for the Docx Parser instance."""
# Keep service responsibilities explicit so downstream behavior stays predictable.
style_name = para.style.name if para.style else ""
if "Heading" in style_name or "标题" in style_name:
# 从样式名称中提取级别
# Keep service responsibilities explicit so downstream behavior stays predictable.
match = re.search(r'Heading\s*(\d)|标题\s*(\d)', style_name)
if match:
level = int(match.group(1) or match.group(2))
return level
# 方法2检查段落格式字号
# 标题通常字号较大
# Keep service responsibilities explicit so downstream behavior stays predictable.
# Keep service responsibilities explicit so downstream behavior stays predictable.
if para.paragraph_format:
# 可以根据字号判断,这里简化处理
# Keep service responsibilities explicit so downstream behavior stays predictable.
pass
# 方法3根据内容模式判断法规文档特征
# Keep service responsibilities explicit so downstream behavior stays predictable.
text = para.text.strip()
# 第一章、第X章 -> 二级标题
# Keep service responsibilities explicit so downstream behavior stays predictable.
if re.match(r'^第[一二三四五六七八九十百]+章\s', text):
return 2
# 第X节 -> 三级标题
# Keep service responsibilities explicit so downstream behavior stays predictable.
elif re.match(r'^第[一二三四五六七八九十百]+节\s', text):
return 3
# 第X条 -> 四级标题
# Keep service responsibilities explicit so downstream behavior stays predictable.
elif re.match(r'^第[一二三四五六七八九十百]+条\s', text):
return 4
return 0 # 正文
return 0 # Keep service responsibilities explicit so downstream behavior stays predictable.
def _detect_list_item(self, para) -> tuple[bool, Optional[str]]:
"""检测是否是列表项"""
"""Handle detect list item for this module for the Docx Parser instance."""
text = para.text.strip()
# 数字列表1.、2.、1、[1]等
# Keep service responsibilities explicit so downstream behavior stays predictable.
if re.match(r'^[\d]+[.、)\]]\s', text):
match = re.match(r'^([\d]+[.、)\]])\s', text)
return True, match.group(1) if match else None
# 中文数字列表:一、二、(一)等
# Keep service responsibilities explicit so downstream behavior stays predictable.
if re.match(r'^[一二三四五六七八九十]+[、.)]\s', text):
match = re.match(r'^([一二三四五六七八九十]+[、.)])\s', text)
return True, match.group(1) if match else None
# 检查段落格式中的列表编号
# Keep service responsibilities explicit so downstream behavior stays predictable.
if para.paragraph_format and hasattr(para.paragraph_format, 'left_indent'):
# 有缩进的可能是列表项
# Keep service responsibilities explicit so downstream behavior stays predictable.
pass
return False, None
def _extract_tables(self) -> List[DocxTable]:
"""提取所有表格"""
"""Handle extract tables for this module for the Docx Parser instance."""
tables = []
for table in self.document.tables:
@@ -193,7 +183,7 @@ class DocxParser:
cells.append(cell.text.strip())
rows.append(cells)
# 转换为Markdown表格
# Keep service responsibilities explicit so downstream behavior stays predictable.
markdown = self._table_to_markdown(rows)
table_content = DocxTable(rows=rows, markdown=markdown)
@@ -202,34 +192,34 @@ class DocxParser:
return tables
def _table_to_markdown(self, rows: List[List[str]]) -> str:
"""将表格转换为Markdown格式"""
"""Handle table to markdown for this module for the Docx Parser instance."""
if not rows or len(rows) < 1:
return ""
lines = []
# 表头
# Keep service responsibilities explicit so downstream behavior stays predictable.
if len(rows) >= 1:
header = rows[0]
lines.append("| " + " | ".join(cell for cell in header) + " |")
lines.append("| " + " | ".join("---" for _ in header) + " |")
# 数据行
# Keep service responsibilities explicit so downstream behavior stays predictable.
for row in rows[1:]:
lines.append("| " + " | ".join(cell for cell in row) + " |")
return "\n".join(lines)
def _generate_markdown(self, doc_content: DocxDocumentContent) -> str:
"""生成Markdown格式文本"""
"""Handle generate markdown for this module for the Docx Parser instance."""
lines = []
# 文档标题
# Keep service responsibilities explicit so downstream behavior stays predictable.
title = doc_content.metadata.get("title", "")
if title:
lines.append(f"# {title}\n")
else:
# 从第一个段落获取标题(如果是标题样式)
# Keep service responsibilities explicit so downstream behavior stays predictable.
for para in doc_content.paragraphs[:5]:
if para.level == 1:
lines.append(f"# {para.text}\n")
@@ -237,29 +227,29 @@ class DocxParser:
else:
lines.append(f"# {doc_content.file_path}\n")
# 元数据信息
# Keep service responsibilities explicit so downstream behavior stays predictable.
lines.append("\n## 文档信息\n")
for key, value in doc_content.metadata.items():
if value:
lines.append(f"- **{key}**: {value}")
# 正文内容
# Keep service responsibilities explicit so downstream behavior stays predictable.
lines.append("\n## 正文\n")
table_index = 0
for para in doc_content.paragraphs:
if para.level > 0:
# 标题
# Keep service responsibilities explicit so downstream behavior stays predictable.
prefix = "#" * para.level
lines.append(f"\n{prefix} {para.text}\n")
elif para.is_list:
# 列表项
# Keep service responsibilities explicit so downstream behavior stays predictable.
lines.append(f"- {para.text}")
else:
# 正文
# Keep service responsibilities explicit so downstream behavior stays predictable.
lines.append(para.text)
# 添加表格
# Keep service responsibilities explicit so downstream behavior stays predictable.
if doc_content.tables:
lines.append("\n## 表格\n")
for i, table in enumerate(doc_content.tables):
@@ -269,18 +259,18 @@ class DocxParser:
return "\n".join(lines)
def parse_to_markdown(self, file_path: str) -> str:
"""直接解析并返回Markdown文本"""
"""Parse to markdown for the Docx Parser instance."""
doc_content = self.parse(file_path)
return doc_content.markdown_text
def parse_docx(file_path: str) -> DocxDocumentContent:
"""便捷函数解析Word文档"""
"""Parse docx."""
parser = DocxParser()
return parser.parse(file_path)
def parse_docx_to_markdown(file_path: str) -> str:
"""便捷函数解析Word并返回Markdown"""
"""Parse docx to markdown."""
parser = DocxParser()
return parser.parse_to_markdown(file_path)