OurCodesAreAllRight/FilesReadSystem
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增强 Word 文档 AI 解析和模板填充功能

Browse Source
This commit is contained in:
zzz
2026-04-10 09:48:57 +08:00
parent 7f67fa89de
commit bedf1af9c0
13 changed files with 2285 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

160
backend/app/services/llm_service.py
View File

@@ -65,7 +65,17 @@ class LLMService:
return response.json()
except httpx.HTTPStatusError as e:
logger.error(f"LLM API 请求失败: {e.response.status_code} - {e.response.text}")
error_detail = e.response.text
logger.error(f"LLM API 请求失败: {e.response.status_code} - {error_detail}")
# 尝试解析错误信息
try:
import json
err_json = json.loads(error_detail)
err_code = err_json.get("error", {}).get("code", "unknown")
err_msg = err_json.get("error", {}).get("message", "unknown")
logger.error(f"API 错误码: {err_code}, 错误信息: {err_msg}")
except:
pass
raise
except Exception as e:
logger.error(f"LLM API 调用异常: {str(e)}")
@@ -328,6 +338,154 @@ Excel 数据概览:
"analysis": None
}
async def chat_with_images(
self,
text: str,
images: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: Optional[int] = None
) -> Dict[str, Any]:
"""
调用视觉模型 API（支持图片输入）
Args:
text: 文本内容
images: 图片列表，每项包含 base64 编码和 mime_type
格式: [{"base64": "...", "mime_type": "image/png"}, ...]
temperature: 温度参数
max_tokens: 最大 token 数
Returns:
Dict[str, Any]: API 响应结果
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
# 构建图片内容
image_contents = []
for img in images:
image_contents.append({
"type": "image_url",
"image_url": {
"url": f"data:{img['mime_type']};base64,{img['base64']}"
}
})
# 构建消息
messages = [
{
"role": "user",
"content": [
{
"type": "text",
"text": text
},
*image_contents
]
}
]
payload = {
"model": self.model_name,
"messages": messages,
"temperature": temperature
}
if max_tokens:
payload["max_tokens"] = max_tokens
try:
async with httpx.AsyncClient(timeout=120.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
error_detail = e.response.text
logger.error(f"视觉模型 API 请求失败: {e.response.status_code} - {error_detail}")
# 尝试解析错误信息
try:
import json
err_json = json.loads(error_detail)
err_code = err_json.get("error", {}).get("code", "unknown")
err_msg = err_json.get("error", {}).get("message", "unknown")
logger.error(f"API 错误码: {err_code}, 错误信息: {err_msg}")
logger.error(f"请求模型: {self.model_name}, base_url: {self.base_url}")
except:
pass
raise
except Exception as e:
logger.error(f"视觉模型 API 调用异常: {str(e)}")
raise
async def analyze_images(
self,
images: List[Dict[str, str]],
user_prompt: str = ""
) -> Dict[str, Any]:
"""
分析图片内容（使用视觉模型）
Args:
images: 图片列表，每项包含 base64 编码和 mime_type
user_prompt: 用户提示词
Returns:
Dict[str, Any]: 分析结果
"""
prompt = f"""你是一个专业的视觉分析专家。请分析以下图片内容。
{user_prompt if user_prompt else "请详细描述图片中的内容，包括文字、数据、图表、流程等所有可见信息。"}
请按照以下 JSON 格式输出：
{{
"description": "图片内容的详细描述",
"text_content": "图片中的文字内容（如有）",
"data_extracted": {{"键": "值"}} // 如果图片中有表格或数据
}}
如果图片不包含有用信息，请返回空的描述。"""
try:
response = await self.chat_with_images(
text=prompt,
images=images,
temperature=0.1,
max_tokens=4000
)
content = self.extract_message_content(response)
# 解析 JSON
import json
try:
result = json.loads(content)
return {
"success": True,
"analysis": result,
"model": self.model_name
}
except json.JSONDecodeError:
return {
"success": True,
"analysis": {"description": content},
"model": self.model_name
}
except Exception as e:
logger.error(f"图片分析失败: {str(e)}")
return {
"success": False,
"error": str(e),
"analysis": None
}
# 全局单例
llm_service = LLMService()

349
backend/app/services/template_fill_service.py
View File

@@ -11,6 +11,7 @@ from app.core.database import mongodb
from app.services.llm_service import llm_service
from app.core.document_parser import ParserFactory
from app.services.markdown_ai_service import markdown_ai_service
from app.services.word_ai_service import word_ai_service
logger = logging.getLogger(__name__)
@@ -173,16 +174,106 @@ class TemplateFillService:
if source_file_paths:
for file_path in source_file_paths:
try:
file_ext = file_path.lower().split('.')[-1]
# 对于 Word 文档，优先使用 AI 解析
if file_ext == 'docx':
# 使用 AI 深度解析 Word 文档
ai_result = await word_ai_service.parse_word_with_ai(
file_path=file_path,
user_hint="请提取文档中的所有结构化数据，包括表格、键值对等"
)
if ai_result.get("success"):
# AI 解析成功，转换为 SourceDocument 格式
# 注意：word_ai_service 返回的是顶层数据，不是 {"data": {...}} 包装
parse_type = ai_result.get("type", "unknown")
# 构建 structured_data
doc_structured = {
"ai_parsed": True,
"parse_type": parse_type,
"tables": [],
"key_values": ai_result.get("key_values", {}) if "key_values" in ai_result else {},
"list_items": ai_result.get("list_items", []) if "list_items" in ai_result else [],
"summary": ai_result.get("summary", "") if "summary" in ai_result else ""
}
# 如果 AI 返回了表格数据
if parse_type == "table_data":
headers = ai_result.get("headers", [])
rows = ai_result.get("rows", [])
if headers and rows:
doc_structured["tables"] = [{
"headers": headers,
"rows": rows
}]
doc_structured["columns"] = headers
doc_structured["rows"] = rows
logger.info(f"AI 表格数据: {len(headers)} 列, {len(rows)} 行")
elif parse_type == "structured_text":
tables = ai_result.get("tables", [])
if tables:
doc_structured["tables"] = tables
logger.info(f"AI 结构化文本提取到 {len(tables)} 个表格")
# 获取摘要内容
content_text = doc_structured.get("summary", "") or ai_result.get("description", "")
source_docs.append(SourceDocument(
doc_id=file_path,
filename=file_path.split("/")[-1] if "/" in file_path else file_path.split("\\")[-1],
doc_type="docx",
content=content_text,
structured_data=doc_structured
))
logger.info(f"AI 解析 Word 文档: {file_path}, type={parse_type}, tables={len(doc_structured.get('tables', []))}")
continue # 跳过后续的基础解析
# 基础解析（Excel 或非 AI 解析的 Word）
parser = ParserFactory.get_parser(file_path)
result = parser.parse(file_path)
if result.success:
# result.data 的结构取决于解析器类型:
# - Excel 单 sheet: {columns: [...], rows: [...], row_count, column_count}
# - Excel 多 sheet: {sheets: {sheet_name: {columns, rows, ...}}}
# - Word/TXT: {content: "...", structured_data: {...}}
# - Word: {content: "...", paragraphs: [...], tables: [...], structured_data: {...}}
doc_data = result.data if result.data else {}
doc_content = doc_data.get("content", "") if isinstance(doc_data, dict) else ""
doc_structured = doc_data if isinstance(doc_data, dict) and "rows" in doc_data or isinstance(doc_data, dict) and "sheets" in doc_data else {}
# 根据文档类型确定 structured_data 的内容
if "sheets" in doc_data:
# Excel 多 sheet 格式
doc_structured = doc_data
elif "rows" in doc_data:
# Excel 单 sheet 格式
doc_structured = doc_data
elif "tables" in doc_data:
# Word 文档格式（已有表格）
doc_structured = doc_data
elif "paragraphs" in doc_data:
# Word 文档只有段落，没有表格 - 尝试 AI 二次解析
unstructured = doc_data
ai_result = await word_ai_service.parse_word_with_ai(
file_path=file_path,
user_hint="请提取文档中的所有结构化信息"
)
if ai_result.get("success"):
parse_type = ai_result.get("type", "text")
doc_structured = {
"ai_parsed": True,
"parse_type": parse_type,
"tables": ai_result.get("tables", []) if "tables" in ai_result else [],
"key_values": ai_result.get("key_values", {}) if "key_values" in ai_result else {},
"list_items": ai_result.get("list_items", []) if "list_items" in ai_result else [],
"summary": ai_result.get("summary", "") if "summary" in ai_result else "",
"content": doc_content
}
logger.info(f"AI 二次解析 Word 段落文档: type={parse_type}")
else:
doc_structured = unstructured
else:
doc_structured = {}
source_docs.append(SourceDocument(
doc_id=file_path,
@@ -321,11 +412,13 @@ class TemplateFillService:
# ========== 步骤3: 尝试解析 JSON ==========
# 3a. 尝试直接解析整个字符串
parsed_confidence = 0.5 # 默认置信度
try:
result = json.loads(json_text)
extracted_values = self._extract_values_from_json(result)
extracted_values, parsed_confidence = self._extract_values_from_json(result)
if extracted_values:
logger.info(f"✅ 直接解析成功，得到 {len(extracted_values)} 个值")
confidence = parsed_confidence if parsed_confidence > 0 else 0.8 # 成功提取，提高置信度
logger.info(f"✅ 直接解析成功，得到 {len(extracted_values)} 个值，置信度: {confidence}")
else:
logger.warning(f"直接解析成功但未提取到值")
except json.JSONDecodeError as e:
@@ -337,9 +430,10 @@ class TemplateFillService:
if fixed_json:
try:
result = json.loads(fixed_json)
extracted_values = self._extract_values_from_json(result)
extracted_values, parsed_confidence = self._extract_values_from_json(result)
if extracted_values:
logger.info(f"✅ 修复后解析成功，得到 {len(extracted_values)} 个值")
confidence = parsed_confidence if parsed_confidence > 0 else 0.7
logger.info(f"✅ 修复后解析成功，得到 {len(extracted_values)} 个值，置信度: {confidence}")
except json.JSONDecodeError as e2:
logger.warning(f"修复后仍然失败: {e2}")
@@ -347,10 +441,15 @@ class TemplateFillService:
if not extracted_values:
extracted_values = self._extract_values_by_regex(cleaned)
if extracted_values:
logger.info(f"✅ 正则提取成功，得到 {len(extracted_values)} 个值")
confidence = 0.6 # 正则提取置信度
logger.info(f"✅ 正则提取成功，得到 {len(extracted_values)} 个值，置信度: {confidence}")
else:
# 最后的备选：使用旧的文本提取
extracted_values = self._extract_values_from_text(cleaned, field.name)
if extracted_values:
confidence = 0.5
else:
confidence = 0.3 # 最后备选
# 如果仍然没有提取到值
if not extracted_values:
@@ -483,30 +582,27 @@ class TemplateFillService:
doc_content += " | ".join(str(cell) for cell in row) + "\n"
row_count += 1
elif doc.structured_data and doc.structured_data.get("tables"):
# Markdown 表格格式: {tables: [{headers: [...], rows: [...]}]}
# Word 文档的表格格式 - 直接输出完整表格，让 LLM 理解
tables = doc.structured_data.get("tables", [])
for table in tables:
for table_idx, table in enumerate(tables):
if isinstance(table, dict):
headers = table.get("headers", [])
rows = table.get("rows", [])
if rows and headers:
doc_content += f"\n【文档: {doc.filename} - 表格】\n"
doc_content += " | ".join(str(h) for h in headers) + "\n"
for row in rows:
table_rows = table.get("rows", [])
if table_rows:
doc_content += f"\n【文档: {doc.filename} - 表格{table_idx + 1}，共 {len(table_rows)} 行】\n"
# 输出表头
if table_rows and isinstance(table_rows[0], list):
doc_content += "表头: " + " | ".join(str(cell) for cell in table_rows[0]) + "\n"
# 输出所有数据行
for row_idx, row in enumerate(table_rows[1:], start=1): # 跳过表头
if isinstance(row, list):
doc_content += " | ".join(str(cell) for cell in row) + "\n"
row_count += 1
# 如果有标题结构，也添加上下文
if doc.structured_data.get("titles"):
titles = doc.structured_data.get("titles", [])
doc_content += f"\n【文档章节结构】\n"
for title in titles[:20]: # 限制前20个标题
doc_content += f"{'#' * title.get('level', 1)} {title.get('text', '')}\n"
# 如果没有提取到表格内容，使用纯文本
if not doc_content.strip():
doc_content = doc.content[:5000] if doc.content else ""
doc_content += "行" + str(row_idx) + ": " + " | ".join(str(cell) for cell in row) + "\n"
row_count += 1
elif doc.content:
doc_content = doc.content[:5000]
# 普通文本内容（Word 段落、纯文本等）
content_preview = doc.content[:8000] if doc.content else ""
if content_preview:
doc_content = f"\n【文档: {doc.filename} ({doc.doc_type})】\n{content_preview}"
row_count = len(content_preview.split('\n'))
if doc_content:
doc_context = f"【文档: {doc.filename} ({doc.doc_type})】\n{doc_content}"
@@ -614,8 +710,20 @@ class TemplateFillService:
logger.info(f"读取 Excel 表头: {df.shape}, 列: {list(df.columns)[:10]}")
# 如果 DataFrame 列为空或只有默认索引，尝试其他方式
if len(df.columns) == 0 or (len(df.columns) == 1 and df.columns[0] == 0):
# 如果 DataFrame 列为空或只有默认索引（0, 1, 2... 或 Unnamed: 0, Unnamed: 1...），尝试其他方式
needs_reparse = len(df.columns) == 0
if not needs_reparse:
# 检查是否所有列都是自动生成的（0, 1, 2... 或 Unnamed: 0, Unnamed: 1...）
auto_generated_count = 0
for col in df.columns:
col_str = str(col)
if col_str in ['0', '1', '2'] or col_str.startswith('Unnamed'):
auto_generated_count += 1
# 如果超过50%的列是自动生成的，认为表头解析失败
if auto_generated_count >= len(df.columns) * 0.5:
needs_reparse = True
if needs_reparse:
logger.warning(f"表头解析结果异常，重新解析: {df.columns}")
# 尝试读取整个文件获取列信息
df_full = pd.read_excel(file_path, header=None)
@@ -656,16 +764,26 @@ class TemplateFillService:
doc = Document(file_path)
for table_idx, table in enumerate(doc.tables):
for row_idx, row in enumerate(table.rows):
rows = list(table.rows)
if len(rows) < 2:
continue # 跳过少于2行的表格（需要表头+数据）
# 第一行是表头，用于识别字段位置
header_cells = [cell.text.strip() for cell in rows[0].cells]
logger.info(f"Word 表格 {table_idx} 表头: {header_cells}")
# 从第二行开始是数据行（比赛模板格式：字段名 | 提示词 | 填写值）
for row_idx, row in enumerate(rows[1:], start=1):
cells = [cell.text.strip() for cell in row.cells]
# 假设第一列是字段名
# 第一列是字段名
if cells and cells[0]:
field_name = cells[0]
# 第二列是提示词
hint = cells[1] if len(cells) > 1 else ""
# 跳过空行或标题行
if field_name and field_name not in ["", "字段名", "名称", "项目"]:
# 跳过空行或明显是表头的行
if field_name and field_name not in ["", "字段名", "名称", "项目", "序号", "编号"]:
fields.append(TemplateField(
cell=f"T{table_idx}R{row_idx}",
name=field_name,
@@ -673,9 +791,10 @@ class TemplateFillService:
required=True,
hint=hint
))
logger.info(f" 提取字段: {field_name}, hint: {hint}")
except Exception as e:
logger.error(f"从Word提取字段失败: {str(e)}")
logger.error(f"从Word提取字段失败: {str(e)}", exc_info=True)
return fields
@@ -783,23 +902,101 @@ class TemplateFillService:
logger.info(f"从文档 {doc.filename} 提取到 {len(values)} 个值")
break
# 处理 Markdown 表格格式: {tables: [{headers: [...], rows: [...]}]}
# 处理 Word 文档表格格式: {tables: [{headers: [...], rows: [[]], ...}]}
elif structured.get("tables"):
tables = structured.get("tables", [])
for table in tables:
if isinstance(table, dict):
headers = table.get("headers", [])
rows = table.get("rows", [])
values = self._extract_column_values(rows, headers, field_name)
if values:
all_values.extend(values)
logger.info(f"从 Markdown 表格提取到 {len(values)} 个值")
break
# AI 返回格式: {headers: [...], rows: [[row1], [row2], ...]}
# 原始 Word 格式: {rows: [[header], [row1], [row2], ...]}
table_rows = table.get("rows", [])
headers = table.get("headers", []) # AI 返回的 headers
if not headers and table_rows:
# 原始格式：第一个元素是表头
headers = table_rows[0] if table_rows else []
data_rows = table_rows[1:] if len(table_rows) > 1 else []
else:
# AI 返回格式：headers 和 rows 分开
data_rows = table_rows
if headers and data_rows:
values = self._extract_values_from_docx_table(data_rows, headers, field_name)
if values:
all_values.extend(values)
logger.info(f"从 Word 表格提取到 {len(values)} 个值")
break
if all_values:
break
return all_values
def _extract_values_from_docx_table(self, table_rows: List, header: List, field_name: str) -> List[str]:
"""
从 Word 文档表格中提取指定列的值
Args:
table_rows: 表格所有行（包括表头）
header: 表头行（可能是真正的表头，也可能是第一行数据）
field_name: 要提取的字段名
Returns:
值列表
"""
if not table_rows or len(table_rows) < 2:
return []
# 第一步：尝试在 header（第一行）中查找匹配列
target_col_idx = None
for col_idx, col_name in enumerate(header):
col_str = str(col_name).strip()
if field_name.lower() in col_str.lower() or col_str.lower() in field_name.lower():
target_col_idx = col_idx
break
# 如果 header 中没找到，尝试在 table_rows[1]（第二行）中查找
# 这是因为有时第一行是数据而不是表头
if target_col_idx is None and len(table_rows) > 1:
second_row = table_rows[1]
if isinstance(second_row, list):
for col_idx, col_name in enumerate(second_row):
col_str = str(col_name).strip()
if field_name.lower() in col_str.lower() or col_str.lower() in field_name.lower():
target_col_idx = col_idx
logger.info(f"在第二行找到匹配列: {field_name} @ 列{col_idx}, header={header}")
break
if target_col_idx is None:
logger.warning(f"未找到匹配列: {field_name}, 表头: {header}")
return []
# 确定从哪一行开始提取数据
# 如果 header 是表头（包含 field_name），则从 table_rows[1] 开始提取
# 如果 header 是数据（不包含 field_name），则从 table_rows[2] 开始提取
header_contains_field = any(
field_name.lower() in str(col).strip().lower() or str(col).strip().lower() in field_name.lower()
for col in header
)
if header_contains_field:
# header 是表头，从第二行开始提取
data_start_idx = 1
else:
# header 是数据，从第三行开始提取（跳过表头和第一行数据）
data_start_idx = 2
# 提取值
values = []
for row_idx, row in enumerate(table_rows[data_start_idx:], start=data_start_idx):
if isinstance(row, list) and target_col_idx < len(row):
val = str(row[target_col_idx]).strip() if row[target_col_idx] else ""
values.append(val)
elif isinstance(row, dict):
val = str(row.get(target_col_idx, "")).strip()
values.append(val)
logger.info(f"从 Word 表格列 {target_col_idx} 提取到 {len(values)} 个值: {values[:3]}")
return values
def _extract_column_values(self, rows: List, columns: List, field_name: str) -> List[str]:
"""
从 rows 和 columns 中提取指定列的值
@@ -839,27 +1036,37 @@ class TemplateFillService:
return values
def _extract_values_from_json(self, result) -> List[str]:
def _extract_values_from_json(self, result) -> tuple:
"""
从解析后的 JSON 对象/数组中提取值数组
从解析后的 JSON 对象/数组中提取值数组和置信度
Args:
result: json.loads() 返回的对象
Returns:
值列表
(值列表, 置信度) 元组
"""
# 提取置信度
confidence = 0.5
if isinstance(result, dict) and "confidence" in result:
try:
conf = float(result["confidence"])
if 0 <= conf <= 1:
confidence = conf
except (ValueError, TypeError):
pass
if isinstance(result, dict):
# 优先找 values 数组
if "values" in result and isinstance(result["values"], list):
vals = [str(v).strip() for v in result["values"] if v and str(v).strip()]
if vals:
return vals
return vals, confidence
# 尝试找 value 字段
if "value" in result:
val = str(result["value"]).strip()
if val:
return [val]
return [val], confidence
# 尝试找任何数组类型的键
for key in result.keys():
val = result[key]
@@ -867,14 +1074,14 @@ class TemplateFillService:
if all(isinstance(v, (str, int, float, bool)) or v is None for v in val):
vals = [str(v).strip() for v in val if v is not None and str(v).strip()]
if vals:
return vals
return vals, confidence
elif isinstance(val, (str, int, float, bool)):
return [str(val).strip()]
return [str(val).strip()], confidence
elif isinstance(result, list):
vals = [str(v).strip() for v in result if v is not None and str(v).strip()]
if vals:
return vals
return []
return vals, confidence
return [], confidence
def _fix_json(self, json_text: str) -> str:
"""
@@ -1189,14 +1396,15 @@ class TemplateFillService:
json_text = cleaned[json_start:]
try:
result = json.loads(json_text)
values = self._extract_values_from_json(result)
values, parsed_conf = self._extract_values_from_json(result)
if values:
conf = result.get("confidence", parsed_conf) if isinstance(result, dict) else parsed_conf
return FillResult(
field=field.name,
values=values,
value=values[0] if values else "",
source=f"AI分析: {doc.filename}",
confidence=result.get("confidence", 0.8)
confidence=max(conf, 0.8) # 最低0.8
)
except json.JSONDecodeError:
# 尝试修复 JSON
@@ -1204,14 +1412,15 @@ class TemplateFillService:
if fixed:
try:
result = json.loads(fixed)
values = self._extract_values_from_json(result)
values, parsed_conf = self._extract_values_from_json(result)
if values:
conf = result.get("confidence", parsed_conf) if isinstance(result, dict) else parsed_conf
return FillResult(
field=field.name,
values=values,
value=values[0] if values else "",
source=f"AI分析: {doc.filename}",
confidence=result.get("confidence", 0.8)
confidence=max(conf, 0.8) # 最低0.8
)
except json.JSONDecodeError:
pass
@@ -1242,6 +1451,8 @@ class TemplateFillService:
try:
import pandas as pd
content_sample = ""
# 读取 Excel 内容检查是否为空
if file_type in ["xlsx", "xls"]:
df = pd.read_excel(file_path, header=None)
@@ -1265,9 +1476,35 @@ class TemplateFillService:
content_sample = df.iloc[:10].to_string() if len(df) >= 10 else df.to_string()
else:
content_sample = df.to_string()
else:
elif file_type == "docx":
# Word 文档：尝试使用 docx_parser 提取内容
try:
from docx import Document
doc = Document(file_path)
# 提取段落文本
paragraphs = [p.text.strip() for p in doc.paragraphs if p.text.strip()]
tables_text = ""
# 提取表格
if doc.tables:
for table in doc.tables:
for row in table.rows:
row_text = " | ".join(cell.text.strip() for cell in row.cells)
tables_text += row_text + "\n"
content_sample = f"【段落】\n{' '.join(paragraphs[:20])}\n\n【表格】\n{tables_text}"
logger.info(f"Word 文档内容预览: {len(content_sample)} 字符")
except Exception as e:
logger.warning(f"读取 Word 文档失败: {str(e)}")
content_sample = ""
else:
logger.warning(f"不支持的文件类型进行 AI 表头生成: {file_type}")
return None
# 调用 AI 生成表头
prompt = f"""你是一个专业的表格设计助手。请为以下空白表格生成合适的表头字段。

637
backend/app/services/word_ai_service.py Normal file
View File

@@ -0,0 +1,637 @@
"""
Word 文档 AI 解析服务
使用 LLM (GLM) 对 Word 文档进行深度理解，提取结构化数据
"""
import logging
from typing import Dict, Any, List, Optional
import json
from app.services.llm_service import llm_service
from app.core.document_parser.docx_parser import DocxParser
logger = logging.getLogger(__name__)
class WordAIService:
"""Word 文档 AI 解析服务"""
def __init__(self):
self.llm = llm_service
self.parser = DocxParser()
async def parse_word_with_ai(
self,
file_path: str,
user_hint: str = ""
) -> Dict[str, Any]:
"""
使用 AI 解析 Word 文档，提取结构化数据
适用于从非结构化的 Word 文档中提取表格数据、键值对等信息
Args:
file_path: Word 文件路径
user_hint: 用户提示词，指定要提取的内容类型
Returns:
Dict: 包含结构化数据的解析结果
"""
try:
# 1. 先用基础解析器提取原始内容
parse_result = self.parser.parse(file_path)
if not parse_result.success:
return {
"success": False,
"error": parse_result.error,
"structured_data": None
}
# 2. 获取原始数据
raw_data = parse_result.data
paragraphs = raw_data.get("paragraphs", [])
paragraphs_with_style = raw_data.get("paragraphs_with_style", [])
tables = raw_data.get("tables", [])
content = raw_data.get("content", "")
images_info = raw_data.get("images", {})
metadata = parse_result.metadata or {}
image_count = images_info.get("image_count", 0)
image_descriptions = images_info.get("descriptions", [])
logger.info(f"Word 基础解析完成: {len(paragraphs)} 个段落, {len(tables)} 个表格, {image_count} 张图片")
# 3. 提取图片数据（用于视觉分析）
images_base64 = []
if image_count > 0:
try:
images_base64 = self.parser.extract_images_as_base64(file_path)
logger.info(f"提取到 {len(images_base64)} 张图片的 base64 数据")
except Exception as e:
logger.warning(f"提取图片 base64 失败: {str(e)}")
# 4. 根据内容类型选择 AI 解析策略
# 如果有图片，先分析图片
image_analysis = ""
if images_base64:
image_analysis = await self._analyze_images_with_ai(images_base64, user_hint)
logger.info(f"图片 AI 分析完成: {len(image_analysis)} 字符")
# 优先处理：表格 > (表格+文本) > 纯文本
if tables and len(tables) > 0:
structured_data = await self._extract_tables_with_ai(
tables, paragraphs, image_count, user_hint, metadata, image_analysis
)
elif paragraphs and len(paragraphs) > 0:
structured_data = await self._extract_from_text_with_ai(
paragraphs, content, image_count, image_descriptions, user_hint, image_analysis
)
else:
structured_data = {
"success": True,
"type": "empty",
"message": "文档内容为空"
}
# 添加图片分析结果
if image_analysis:
structured_data["image_analysis"] = image_analysis
return structured_data
except Exception as e:
logger.error(f"AI 解析 Word 文档失败: {str(e)}")
return {
"success": False,
"error": str(e),
"structured_data": None
}
async def _extract_tables_with_ai(
self,
tables: List[Dict],
paragraphs: List[str],
image_count: int,
user_hint: str,
metadata: Dict,
image_analysis: str = ""
) -> Dict[str, Any]:
"""
使用 AI 从 Word 表格和文本中提取结构化数据
Args:
tables: 表格列表
paragraphs: 段落列表
image_count: 图片数量
user_hint: 用户提示
metadata: 文档元数据
image_analysis: 图片 AI 分析结果
Returns:
结构化数据
"""
try:
# 构建表格文本描述
tables_text = self._build_tables_description(tables)
# 构建段落描述
paragraphs_text = "\n".join(paragraphs[:50]) if paragraphs else "（无正文文本）"
if len(paragraphs) > 50:
paragraphs_text += f"\n...（共 {len(paragraphs)} 个段落，仅显示前50个）"
# 图片提示
image_hint = f"注意：此文档包含 {image_count} 张图片/图表。" if image_count > 0 else ""
prompt = f"""你是一个专业的数据提取专家。请从以下 Word 文档的完整内容中提取结构化数据。
【用户需求】
{user_hint if user_hint else "请提取文档中的所有结构化数据，包括表格数据、键值对、列表项等。"}
【文档正文（段落）】
{paragraphs_text}
【文档表格】
{tables_text}
【文档图片信息】
{image_hint}
请按照以下 JSON 格式输出：
{{
"type": "table_data",
"headers": ["列1", "列2", ...],
"rows": [["行1列1", "行1列2", ...], ["行2列1", "行2列2", ...], ...],
"key_values": {{"键1": "值1", "键2": "值2", ...}},
"list_items": ["项1", "项2", ...],
"description": "文档内容描述"
}}
重点：
- 优先从表格中提取结构化数据
- 如果表格中有表头，headers 是表头，rows 是数据行
- 如果文档中有键值对（如 名称: 张三），提取到 key_values 中
- 如果文档中有列表项，提取到 list_items 中
- 图片内容无法直接提取，但请在 description 中说明图片的大致主题（如"包含流程图"、"包含数据图表"等）
"""
messages = [
{"role": "system", "content": "你是一个专业的数据提取助手。请严格按JSON格式输出。"},
{"role": "user", "content": prompt}
]
response = await self.llm.chat(
messages=messages,
temperature=0.1,
max_tokens=50000
)
content = self.llm.extract_message_content(response)
# 解析 JSON
result = self._parse_json_response(content)
if result:
logger.info(f"AI 表格提取成功: {len(result.get('rows', []))} 行数据")
return {
"success": True,
"type": "table_data",
"headers": result.get("headers", []),
"rows": result.get("rows", []),
"description": result.get("description", "")
}
else:
# 如果 AI 返回格式不对，尝试直接解析表格
return self._fallback_table_parse(tables)
except Exception as e:
logger.error(f"AI 表格提取失败: {str(e)}")
return self._fallback_table_parse(tables)
async def _extract_from_text_with_ai(
self,
paragraphs: List[str],
full_text: str,
image_count: int,
image_descriptions: List[str],
user_hint: str,
image_analysis: str = ""
) -> Dict[str, Any]:
"""
使用 AI 从 Word 纯文本中提取结构化数据
Args:
paragraphs: 段落列表
full_text: 完整文本
image_count: 图片数量
image_descriptions: 图片描述列表
user_hint: 用户提示
image_analysis: 图片 AI 分析结果
Returns:
结构化数据
"""
try:
# 限制文本长度
text_preview = full_text[:8000] if len(full_text) > 8000 else full_text
# 图片提示
image_hint = f"\n【文档图片】此文档包含 {image_count} 张图片/图表。" if image_count > 0 else ""
if image_descriptions:
image_hint += "\n" + "\n".join(image_descriptions)
prompt = f"""你是一个专业的数据提取专家。请从以下 Word 文档的完整内容中提取结构化数据。
【用户需求】
{user_hint if user_hint else "请识别并提取文档中的关键信息，包括：表格数据、键值对、列表项等。"}
【文档正文】{image_hint}
{text_preview}
请按照以下 JSON 格式输出：
{{
"type": "structured_text",
"tables": [{{"headers": [...], "rows": [...]}}],
"key_values": {{"键1": "值1", "键2": "值2", ...}},
"list_items": ["项1", "项2", ...],
"summary": "文档内容摘要"
}}
重点：
- 如果文档包含表格数据，提取到 tables 中
- 如果文档包含键值对（如 名称: 张三），提取到 key_values 中
- 如果文档包含列表项，提取到 list_items 中
- 如果文档包含图片，请根据上下文推断图片内容（如"流程图"、"数据折线图"等）并在 description 中说明
- 如果无法提取到结构化数据，至少提供一个详细的摘要
"""
messages = [
{"role": "system", "content": "你是一个专业的数据提取助手。请严格按JSON格式输出。"},
{"role": "user", "content": prompt}
]
response = await self.llm.chat(
messages=messages,
temperature=0.1,
max_tokens=50000
)
content = self.llm.extract_message_content(response)
result = self._parse_json_response(content)
if result:
logger.info(f"AI 文本提取成功: type={result.get('type')}")
return {
"success": True,
"type": result.get("type", "structured_text"),
"tables": result.get("tables", []),
"key_values": result.get("key_values", {}),
"list_items": result.get("list_items", []),
"summary": result.get("summary", ""),
"raw_text_preview": text_preview[:500]
}
else:
return {
"success": True,
"type": "text",
"summary": text_preview[:500],
"raw_text_preview": text_preview[:500]
}
except Exception as e:
logger.error(f"AI 文本提取失败: {str(e)}")
return {
"success": False,
"error": str(e)
}
async def _analyze_images_with_ai(
self,
images: List[Dict[str, str]],
user_hint: str = ""
) -> str:
"""
使用视觉模型分析 Word 文档中的图片
Args:
images: 图片列表，每项包含 base64 和 mime_type
user_hint: 用户提示
Returns:
图片分析结果文本
"""
try:
# 调用 LLM 的视觉分析功能
result = await self.llm.analyze_images(
images=images,
user_prompt=user_hint or "请详细描述图片内容，提取所有文字和数据信息。"
)
if result.get("success"):
analysis = result.get("analysis", {})
if isinstance(analysis, dict):
description = analysis.get("description", "")
text_content = analysis.get("text_content", "")
data_extracted = analysis.get("data_extracted", {})
result_text = f"【图片分析结果】\n{description}"
if text_content:
result_text += f"\n\n【图片中的文字】\n{text_content}"
if data_extracted:
result_text += f"\n\n【提取的数据】\n{json.dumps(data_extracted, ensure_ascii=False)}"
return result_text
else:
return str(analysis)
else:
logger.warning(f"图片 AI 分析失败: {result.get('error')}")
return ""
except Exception as e:
logger.error(f"图片 AI 分析异常: {str(e)}")
return ""
def _build_tables_description(self, tables: List[Dict]) -> str:
"""构建表格的文本描述"""
result = []
for idx, table in enumerate(tables):
rows = table.get("rows", [])
if not rows:
continue
result.append(f"\n--- 表格 {idx + 1} ---")
for row_idx, row in enumerate(rows[:50]): # 限制每表格最多50行
if isinstance(row, list):
result.append(" | ".join(str(cell).strip() for cell in row))
elif isinstance(row, dict):
result.append(str(row))
if len(rows) > 50:
result.append(f"...（共 {len(rows)} 行，仅显示前50行）")
return "\n".join(result) if result else "（无表格内容）"
def _parse_json_response(self, content: str) -> Optional[Dict]:
"""解析 JSON 响应，处理各种格式问题"""
import re
# 清理 markdown 标记
cleaned = content.strip()
cleaned = re.sub(r'^```json\s*', '', cleaned, flags=re.MULTILINE)
cleaned = re.sub(r'^```\s*', '', cleaned, flags=re.MULTILINE)
cleaned = cleaned.strip()
# 找到 JSON 开始位置
json_start = -1
for i, c in enumerate(cleaned):
if c == '{':
json_start = i
break
if json_start == -1:
logger.warning("无法找到 JSON 开始位置")
return None
json_text = cleaned[json_start:]
# 尝试直接解析
try:
return json.loads(json_text)
except json.JSONDecodeError:
pass
# 尝试修复并解析
try:
# 找到闭合括号
depth = 0
end_pos = -1
for i, c in enumerate(json_text):
if c == '{':
depth += 1
elif c == '}':
depth -= 1
if depth == 0:
end_pos = i + 1
break
if end_pos > 0:
fixed = json_text[:end_pos]
# 移除末尾逗号
fixed = re.sub(r',\s*([}]])', r'\1', fixed)
return json.loads(fixed)
except Exception as e:
logger.warning(f"JSON 修复失败: {e}")
return None
def _fallback_table_parse(self, tables: List[Dict]) -> Dict[str, Any]:
"""当 AI 解析失败时，直接解析表格"""
if not tables:
return {
"success": True,
"type": "empty",
"data": {},
"message": "无表格内容"
}
all_rows = []
all_headers = None
for table in tables:
rows = table.get("rows", [])
if not rows:
continue
# 查找真正的表头行（跳过标题行）
header_row_idx = 0
for idx, row in enumerate(rows[:5]): # 只检查前5行
if not isinstance(row, list):
continue
# 如果某一行包含"表"字开头且单元格内容很长，这可能是标题行
first_cell = str(row[0]) if row else ""
if first_cell.startswith("表") and len(first_cell) > 15:
header_row_idx = idx + 1
continue
# 如果某一行有超过3个空单元格，可能是无效行
empty_count = sum(1 for cell in row if not str(cell).strip())
if empty_count > 3:
header_row_idx = idx + 1
continue
# 找到第一行看起来像表头的行（短单元格，大部分有内容）
avg_len = sum(len(str(c)) for c in row) / len(row) if row else 0
if avg_len < 20: # 表头通常比数据行短
header_row_idx = idx
break
if header_row_idx >= len(rows):
continue
# 使用找到的表头行
if rows and isinstance(rows[header_row_idx], list):
headers = rows[header_row_idx]
if all_headers is None:
all_headers = headers
# 数据行（从表头之后开始）
for row in rows[header_row_idx + 1:]:
if isinstance(row, list) and len(row) == len(headers):
all_rows.append(row)
if all_headers and all_rows:
return {
"success": True,
"type": "table_data",
"headers": all_headers,
"rows": all_rows,
"description": "直接从 Word 表格提取"
}
return {
"success": True,
"type": "raw",
"tables": tables,
"message": "表格数据（未AI处理）"
}
async def fill_template_with_ai(
self,
file_path: str,
template_fields: List[Dict[str, Any]],
user_hint: str = ""
) -> Dict[str, Any]:
"""
使用 AI 解析 Word 文档并填写模板
这是主要入口函数，前端调用此函数即可完成：
1. AI 解析 Word 文档
2. 根据模板字段提取数据
3. 返回填写结果
Args:
file_path: Word 文件路径
template_fields: 模板字段列表 [{"name": "字段名", "hint": "提示词"}, ...]
user_hint: 用户提示
Returns:
填写结果
"""
try:
# 1. AI 解析文档
parse_result = await self.parse_word_with_ai(file_path, user_hint)
if not parse_result.get("success"):
return {
"success": False,
"error": parse_result.get("error", "解析失败"),
"filled_data": {},
"source": "ai_parse_failed"
}
# 2. 根据字段类型提取数据
filled_data = {}
extract_details = []
parse_type = parse_result.get("type", "")
if parse_type == "table_data":
# 表格数据：直接匹配列名
headers = parse_result.get("headers", [])
rows = parse_result.get("rows", [])
for field in template_fields:
field_name = field.get("name", "")
values = self._extract_field_from_table(headers, rows, field_name)
filled_data[field_name] = values
extract_details.append({
"field": field_name,
"values": values,
"source": "ai_table_extraction",
"confidence": 0.9 if values else 0.0
})
elif parse_type == "structured_text":
# 结构化文本：尝试从 key_values 和 list_items 提取
key_values = parse_result.get("key_values", {})
list_items = parse_result.get("list_items", [])
for field in template_fields:
field_name = field.get("name", "")
value = key_values.get(field_name, "")
if not value and list_items:
value = list_items[0] if list_items else ""
filled_data[field_name] = [value] if value else []
extract_details.append({
"field": field_name,
"values": [value] if value else [],
"source": "ai_text_extraction",
"confidence": 0.7 if value else 0.0
})
else:
# 其他类型：返回原始解析结果供后续处理
for field in template_fields:
field_name = field.get("name", "")
filled_data[field_name] = []
extract_details.append({
"field": field_name,
"values": [],
"source": "no_ai_data",
"confidence": 0.0
})
# 3. 返回结果
max_rows = max(len(v) for v in filled_data.values()) if filled_data else 1
return {
"success": True,
"filled_data": filled_data,
"fill_details": extract_details,
"ai_parse_result": {
"type": parse_type,
"description": parse_result.get("description", "")
},
"source_doc_count": 1,
"max_rows": max_rows
}
except Exception as e:
logger.error(f"AI 填表失败: {str(e)}")
return {
"success": False,
"error": str(e),
"filled_data": {},
"fill_details": []
}
def _extract_field_from_table(
self,
headers: List[str],
rows: List[List],
field_name: str
) -> List[str]:
"""从表格中提取指定字段的值"""
# 查找匹配的列
target_col_idx = None
for col_idx, header in enumerate(headers):
if field_name.lower() in str(header).lower() or str(header).lower() in field_name.lower():
target_col_idx = col_idx
break
if target_col_idx is None:
return []
# 提取该列所有值
values = []
for row in rows:
if isinstance(row, list) and target_col_idx < len(row):
val = str(row[target_col_idx]).strip()
if val:
values.append(val)
return values
# 全局单例
word_ai_service = WordAIService()