567 lines
22 KiB
Python
567 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
金融文档分析整理模块
|
||
在 document_parser 基础上构建结构化分析层：
|
||
- 单文档深度分析（分类/元数据/章节/财务指标/风险因素/术语表）
|
||
- 批量整理目录（扫描+逐份分析+分类聚合+索引+汇总）
|
||
- 多文档并排对比（同维度提取+差异高亮）
|
||
零新外部依赖（复用 PyPDF2/python-docx/openpyxl，正则+启发式分类）。
|
||
"""
|
||
|
||
import re
|
||
import json
|
||
import os as _os
|
||
from pathlib import Path
|
||
from typing import Dict, Any, List, Optional, Tuple
|
||
from datetime import datetime
|
||
|
||
try:
|
||
from document_parser import parse_document, extract_financial_numbers
|
||
except ImportError:
|
||
def parse_document(file_path, **kw):
|
||
return {"error": "document_parser not available"}
|
||
def extract_financial_numbers(text):
|
||
return {}
|
||
|
||
SKILL_DATA_DIR = Path(__file__).parent.parent / "data"
|
||
DOC_INDEX_DIR = SKILL_DATA_DIR / "document_index"
|
||
|
||
|
||
# ==================== 文档分类 ====================
|
||
|
||
DOC_CATEGORIES = [
|
||
("券商研报", ["研究报告", "评级推荐", "目标价", "分析师", "投资建议", "证券研究报告"], 10),
|
||
("公告", ["公告编号", "证券代码", "临时公告", "重大事项"], 9),
|
||
("年报", ["年度报告", "annual report", "年报摘要"], 8),
|
||
("半年报", ["半年度报告", "半年报", "中期报告"], 7),
|
||
("季报", ["季度报告", "一季报", "三季报"], 6),
|
||
("基金合同", ["基金合同", "基金托管协议", "基金管理人"], 5),
|
||
("招募说明书", ["招募说明书", "招募更新"], 4),
|
||
("财务报告", ["财务报告", "财务报表", "审计报告", "合并报表"], 3),
|
||
("通用金融文档", [], 0),
|
||
]
|
||
|
||
SECTION_PATTERNS = {
|
||
"h1": re.compile(r'第[一二三四五六七八九十百零\d]+[章节篇部]\s*[^\n]{0,80}'),
|
||
"h2": re.compile(r'^[一二三四五六七八九十]+\s*[、.．]\s*[^\n]{0,80}', re.MULTILINE),
|
||
"h3": re.compile(r'^\d+[\.．]\s*(?!\d)[^\n]{0,80}', re.MULTILINE),
|
||
"h4": re.compile(r'^[（\(]\d+[）\)]\s*[^\n]{0,80}', re.MULTILINE),
|
||
}
|
||
|
||
FINANCIAL_INDICATOR_PATTERNS = {
|
||
"营业收入": r'营业(?:总)?收入[（(]?万元?[）)]?\s*[:：]?\s*([+-]?\d[\d,.]*)\s*(?:万?元?|亿?)',
|
||
"净利润": r'(?:归属.*)?净利润[（(]?万元?[）)]?\s*[:：]?\s*([+-]?\d[\d,.]*)\s*(?:万?元?|亿?)',
|
||
"总资产": r'总资产[（(]?万元?[）)]?\s*[:：]?\s*([+-]?\d[\d,.]*)\s*(?:万?元?|亿?)',
|
||
"净资产": r'(?:归属.*)?净资产[（(]?万元?[）)]?\s*[:：]?\s*([+-]?\d[\d,.]*)\s*(?:万?元?|亿?)',
|
||
"每股收益_EPS": r'(?:基本)?每股收益[（(]?EPS[）)]?\s*[:：]?\s*([+-]?\d+[\.]?\d*)',
|
||
"市盈率_PE": r'市盈率[（(]?P[Ee][）)]?\s*[:：]?\s*([+-]?\d+[\.]?\d*)',
|
||
"市净率_PB": r'市净率[（(]?P[Bb][）)]?\s*[:：]?\s*([+-]?\d+[\.]?\d*)',
|
||
"净资产收益率_ROE": r'净资产收益率[（(]?ROE[）)]?\s*[:：]?\s*([+-]?\d+[\.]?\d*)\s*%?',
|
||
"毛利率": r'毛利率\s*[:：]?\s*([+-]?\d+[\.]?\d*)\s*%',
|
||
"资产负债率": r'资产负债率\s*[:：]?\s*([+-]?\d+[\.]?\d*)\s*%',
|
||
"同比增速": r'(?:营业[总收]入.*)?同比[增减增速]*[:：]?\s*([+-]?\d+[\.]?\d*)\s*%',
|
||
}
|
||
|
||
|
||
# ==================== 辅助函数 ====================
|
||
|
||
def classify_document(text: str, file_path: str = "") -> Tuple[str, int]:
|
||
"""启发式文档分类。返回 (类别, 置信度 0-100)。"""
|
||
text_head = (text or "")[:3000].lower()
|
||
file_lower = Path(file_path).name.lower() if file_path else ""
|
||
|
||
best_category = "通用金融文档"
|
||
best_score = 0
|
||
max_possible = 60
|
||
|
||
for cat_name, keywords, priority in DOC_CATEGORIES:
|
||
score = 0
|
||
for kw in keywords:
|
||
kw_lower = kw.lower()
|
||
if re.search(kw_lower, text_head):
|
||
score += 15
|
||
if file_lower and re.search(kw_lower, file_lower):
|
||
score += 20
|
||
if score > best_score:
|
||
best_score = score
|
||
best_category = cat_name
|
||
|
||
confidence = min(100, int(best_score / max_possible * 100)) if max_possible > 0 else 10
|
||
return best_category, max(confidence, 10)
|
||
|
||
|
||
def extract_metadata(text: str, file_path: str = "") -> Dict[str, str]:
|
||
"""从文本提取元数据：标题、机构、日期、股票代码。"""
|
||
meta = {"title": "", "institution": "", "date": "", "stock_code": ""}
|
||
text_head = (text or "")[:2000]
|
||
|
||
lines = [l.strip() for l in text_head.split("\n") if l.strip()]
|
||
if lines:
|
||
meta["title"] = lines[0][:120]
|
||
title_match = re.search(r'(?:标题|题目|报告名称)[：:]\s*(.+?)(?:\n|$)', text_head)
|
||
if title_match:
|
||
meta["title"] = title_match.group(1).strip()[:120]
|
||
|
||
inst_patterns = [
|
||
r'([\u4e00-\u9fff]{2,8}(?:证券|基金|期货|信托|银行|保险|资产管理|资本|投资|研究))(?:(?:股份)?有限(?:公司|责任公司))?',
|
||
r'([\u4e00-\u9fff]{2,12}(?:股份)?有限(?:公司|责任公司))',
|
||
r'(?:发布机构|研究机构|出处|来源|作者单位)[：:]\s*([^\n]{2,40})',
|
||
]
|
||
for pat in inst_patterns:
|
||
m = re.search(pat, text_head)
|
||
if m:
|
||
inst = m.group(1).strip()
|
||
if len(inst) >= 4:
|
||
meta["institution"] = inst
|
||
break
|
||
|
||
date_patterns = [
|
||
r'(\d{4})\s*[年\-\/.]\s*(\d{1,2})\s*[月\-\/.]\s*(\d{1,2})\s*日?',
|
||
r'(\d{4})-(\d{2})-(\d{2})',
|
||
r'(?:报告日期|发布日期|公告日期|日期)[：:]\s*(\d{4}[\-\/.]\d{1,2}[\-\/.]\d{1,2})',
|
||
]
|
||
for pat in date_patterns:
|
||
m = re.search(pat, text_head)
|
||
if m:
|
||
if len(m.groups()) >= 3:
|
||
y, mo, d = m.group(1), m.group(2), m.group(3)
|
||
try:
|
||
meta["date"] = f"{int(y)}-{int(mo):02d}-{int(d):02d}"
|
||
except ValueError:
|
||
pass
|
||
elif len(m.groups()) == 1:
|
||
meta["date"] = m.group(1)
|
||
break
|
||
|
||
code_match = re.search(r'(?:股票代码|证券代码|标的代码|code)[：:]\s*(\d{6})', text_head)
|
||
if code_match:
|
||
meta["stock_code"] = code_match.group(1)
|
||
else:
|
||
code_match2 = re.search(r'[\(（](\d{6})[\._\s]?(?:SH|SZ|sh|sz|上证|深证)?[\)）]', text_head)
|
||
if code_match2:
|
||
meta["stock_code"] = code_match2.group(1)
|
||
|
||
return meta
|
||
|
||
|
||
def extract_sections(text: str) -> List[Dict[str, Any]]:
|
||
"""提取文档章节结构。"""
|
||
sections = []
|
||
seen_titles = set()
|
||
lines = text.split("\n")
|
||
|
||
for line_no, line in enumerate(lines):
|
||
line = line.strip()
|
||
if not line or len(line) < 3:
|
||
continue
|
||
if len(line) > 150:
|
||
continue
|
||
|
||
level = None
|
||
title = None
|
||
|
||
for lvl_name, pattern in SECTION_PATTERNS.items():
|
||
m = pattern.match(line)
|
||
if m:
|
||
title = m.group(0).strip()
|
||
if title.replace(" ", "").replace(".", "").replace("．", "").isdigit():
|
||
continue
|
||
level = int(lvl_name[1])
|
||
break
|
||
|
||
if title and level:
|
||
norm = re.sub(r'\s+', '', title)
|
||
if norm in seen_titles:
|
||
continue
|
||
seen_titles.add(norm)
|
||
|
||
preview_start = line_no + 1
|
||
preview_parts = []
|
||
for l in lines[preview_start:preview_start + 5]:
|
||
if l.strip():
|
||
preview_parts.append(l.strip())
|
||
preview = " ".join(preview_parts)[:150]
|
||
|
||
sections.append({
|
||
"title": title,
|
||
"level": level,
|
||
"preview": preview,
|
||
"line": line_no + 1,
|
||
})
|
||
|
||
return sections[:60]
|
||
|
||
|
||
def extract_risk_factors(text: str) -> List[str]:
|
||
"""提取风险因素/风险提示章节条目。"""
|
||
risks = []
|
||
risk_patterns = [
|
||
r'(?:风险因素|风险提示|风险分析|风险管理|风险警示)(?:.{0,500}?)',
|
||
]
|
||
for pat in risk_patterns:
|
||
for m in re.finditer(pat, text):
|
||
start = m.start()
|
||
chunk = text[start:start + 3000]
|
||
|
||
bullet_patterns = [
|
||
r'[•·●◆■]\s*([^\n•·●◆■]{5,200})',
|
||
r'(?:^|\n)\s*\d+[\.、．)]\s*([^\n]{5,200})',
|
||
r'(?:^|\n)\s*[（\(]\d+[）\)]\s*([^\n]{5,200})',
|
||
]
|
||
for bpat in bullet_patterns:
|
||
for bm in re.finditer(bpat, chunk):
|
||
item = bm.group(1).strip()
|
||
if len(item) > 10 and item not in risks:
|
||
risks.append(item[:200])
|
||
if risks:
|
||
break
|
||
if risks:
|
||
break
|
||
if risks:
|
||
break
|
||
|
||
return risks[:20]
|
||
|
||
|
||
def extract_glossary(text: str) -> Dict[str, str]:
|
||
"""提取术语定义：匹配 "XX 指/系/是指/谓/意为" 模式。"""
|
||
glossary = {}
|
||
pattern = re.compile(
|
||
r'([\u4e00-\u9fffA-Za-z]{2,15})\s*(?:指|系|是指|谓|意为|即|又称)\s*(.{5,120}?)(?:[。；;]|$|\n)'
|
||
)
|
||
for m in re.finditer(pattern, text[:5000]):
|
||
term = m.group(1).strip()
|
||
definition = m.group(2).strip()
|
||
if term and definition and term not in glossary:
|
||
glossary[term] = definition[:120]
|
||
|
||
return dict(list(glossary.items())[:30])
|
||
|
||
|
||
def extract_financial_indicators(text: str) -> Dict[str, Any]:
|
||
"""扩展的财务指标提取（比 extract_financial_numbers 更全面）。"""
|
||
indicators = {}
|
||
for key, pattern in FINANCIAL_INDICATOR_PATTERNS.items():
|
||
m = re.search(pattern, text)
|
||
if m:
|
||
try:
|
||
val = m.group(1).replace(",", "")
|
||
indicators[key] = float(val) if '.' in val else int(val)
|
||
except (ValueError, AttributeError):
|
||
pass
|
||
return indicators
|
||
|
||
|
||
# ==================== DocumentAnalyzer 类 ====================
|
||
|
||
class DocumentAnalyzer:
|
||
"""金融文档分析整理器。"""
|
||
|
||
def __init__(self):
|
||
self._cache = {}
|
||
|
||
def analyze(self, file_path: str) -> Dict[str, Any]:
|
||
"""
|
||
单文档深度分析。
|
||
返回：{file_path, file_type, doc_category, category_confidence,
|
||
metadata, summary, sections, financial_indicators,
|
||
risk_factors, glossary, stats}
|
||
"""
|
||
cache_key = _os.path.abspath(file_path)
|
||
if cache_key in self._cache:
|
||
return self._cache[cache_key]
|
||
|
||
result = {
|
||
"file_path": file_path,
|
||
"file_type": Path(file_path).suffix.lower().lstrip("."),
|
||
"doc_category": "通用金融文档",
|
||
"category_confidence": 0,
|
||
"metadata": {},
|
||
"summary": "",
|
||
"sections": [],
|
||
"financial_indicators": {},
|
||
"risk_factors": [],
|
||
"glossary": {},
|
||
"stats": {},
|
||
}
|
||
|
||
parsed = parse_document(file_path)
|
||
if parsed.get("error"):
|
||
result["error"] = parsed["error"]
|
||
return result
|
||
|
||
text_content = parsed.get("text_content", "") or ""
|
||
if not text_content.strip():
|
||
result["error"] = "文档内容为空或无法提取文本"
|
||
return result
|
||
|
||
category, confidence = classify_document(text_content, file_path)
|
||
result["doc_category"] = category
|
||
result["category_confidence"] = confidence
|
||
result["metadata"] = extract_metadata(text_content, file_path)
|
||
result["sections"] = extract_sections(text_content)
|
||
result["financial_indicators"] = extract_financial_indicators(text_content)
|
||
# 合并 document_parser 的简单数字提取
|
||
simple_nums = extract_financial_numbers(text_content)
|
||
for k, v in simple_nums.items():
|
||
if k not in result["financial_indicators"] and k not in ("amounts_yi", "total_yi", "percentages"):
|
||
result["financial_indicators"][k] = v
|
||
result["risk_factors"] = extract_risk_factors(text_content)
|
||
result["glossary"] = extract_glossary(text_content)
|
||
result["summary"] = text_content[:600].replace("\n", " ").strip()
|
||
result["stats"] = {
|
||
"char_count": len(text_content),
|
||
"section_count": len(result["sections"]),
|
||
"indicator_count": len(result["financial_indicators"]),
|
||
"risk_count": len(result["risk_factors"]),
|
||
"analyzed_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
}
|
||
|
||
self._cache[cache_key] = result
|
||
return result
|
||
|
||
def organize_directory(self, dir_path: str, recursive: bool = True) -> Dict[str, Any]:
|
||
"""
|
||
批量整理文档目录。
|
||
返回：{total, categories, index_file, summary_file}
|
||
"""
|
||
dir_p = Path(dir_path)
|
||
if not dir_p.is_dir():
|
||
return {"error": f"目录不存在: {dir_path}"}
|
||
|
||
supported_ext = {".pdf", ".docx", ".doc", ".xlsx", ".xls", ".txt", ".md", ".csv"}
|
||
files = []
|
||
if recursive:
|
||
for f in dir_p.rglob("*"):
|
||
if f.is_file() and f.suffix.lower() in supported_ext:
|
||
files.append(f)
|
||
else:
|
||
for f in dir_p.iterdir():
|
||
if f.is_file() and f.suffix.lower() in supported_ext:
|
||
files.append(f)
|
||
|
||
if not files:
|
||
return {"error": "目录中没有支持的文档文件", "total": 0, "categories": {}}
|
||
|
||
categories: Dict[str, List[Dict]] = {}
|
||
for fp in files:
|
||
try:
|
||
analysis = self.analyze(str(fp))
|
||
cat = analysis.get("doc_category", "通用金融文档")
|
||
if cat not in categories:
|
||
categories[cat] = []
|
||
categories[cat].append({
|
||
"file": fp.name,
|
||
"path": str(fp),
|
||
"type": analysis.get("file_type", ""),
|
||
"metadata": analysis.get("metadata", {}),
|
||
"confidence": analysis.get("category_confidence", 0),
|
||
"indicator_count": len(analysis.get("financial_indicators", {})),
|
||
})
|
||
except Exception as e:
|
||
if "处理错误" not in categories:
|
||
categories["处理错误"] = []
|
||
categories["处理错误"].append({"file": fp.name, "error": str(e)})
|
||
|
||
total = sum(len(v) for v in categories.values())
|
||
|
||
DOC_INDEX_DIR.mkdir(parents=True, exist_ok=True)
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
index_file = DOC_INDEX_DIR / f"index_{timestamp}.json"
|
||
summary_file = DOC_INDEX_DIR / f"summary_{timestamp}.md"
|
||
|
||
index_data = {
|
||
"source_dir": str(dir_p),
|
||
"generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"total_documents": total,
|
||
"category_count": len(categories),
|
||
"categories": {},
|
||
}
|
||
for cat, items in categories.items():
|
||
index_data["categories"][cat] = {"count": len(items), "files": items}
|
||
with open(index_file, "w", encoding="utf-8") as f:
|
||
json.dump(index_data, f, ensure_ascii=False, indent=2)
|
||
|
||
md_lines = [
|
||
"# 文档目录整理报告\n",
|
||
f"- 来源目录: {dir_p}",
|
||
f"- 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
|
||
f"- 文档总数: {total}",
|
||
f"- 类别数: {len(categories)}\n",
|
||
]
|
||
for cat, items in sorted(categories.items(), key=lambda x: -len(x[1])):
|
||
md_lines.append(f"## {cat} ({len(items)}份)")
|
||
for item in items[:50]:
|
||
meta = item.get("metadata", {})
|
||
title = meta.get("title", item["file"])[:80]
|
||
date_str = meta.get("date", "")
|
||
inst = meta.get("institution", "")
|
||
code = meta.get("stock_code", "")
|
||
extras = " / ".join(filter(None, [date_str, inst, code]))
|
||
line = f"- **{title}** ({item['type']})"
|
||
if extras:
|
||
line += f" — {extras}"
|
||
md_lines.append(line)
|
||
md_lines.append("")
|
||
with open(summary_file, "w", encoding="utf-8") as f:
|
||
f.write("\n".join(md_lines))
|
||
|
||
return {
|
||
"total": total,
|
||
"categories": {cat: len(items) for cat, items in categories.items()},
|
||
"index_file": str(index_file),
|
||
"summary_file": str(summary_file),
|
||
}
|
||
|
||
def compare(self, file_paths: List[str]) -> Dict[str, Any]:
|
||
"""多文档并排对比。"""
|
||
results = []
|
||
for fp in file_paths:
|
||
r = self.analyze(fp)
|
||
results.append(r)
|
||
|
||
if not results:
|
||
return {"compared": 0, "files": [], "dimensions": {}, "differences": []}
|
||
|
||
dimensions = {
|
||
"文档类型": [], "文件类型": [], "日期": [], "机构": [],
|
||
"股票代码": [], "文档规模（字符数）": [], "章节数": [],
|
||
"财务指标数": [], "风险因素数": [], "关键指标": [],
|
||
}
|
||
|
||
for r in results:
|
||
dimensions["文档类型"].append(r.get("doc_category", ""))
|
||
dimensions["文件类型"].append(r.get("file_type", ""))
|
||
dimensions["日期"].append(r.get("metadata", {}).get("date", ""))
|
||
dimensions["机构"].append(r.get("metadata", {}).get("institution", ""))
|
||
dimensions["股票代码"].append(r.get("metadata", {}).get("stock_code", ""))
|
||
dimensions["文档规模（字符数）"].append(r.get("stats", {}).get("char_count", 0))
|
||
dimensions["章节数"].append(r.get("stats", {}).get("section_count", 0))
|
||
dimensions["财务指标数"].append(r.get("stats", {}).get("indicator_count", 0))
|
||
dimensions["风险因素数"].append(r.get("stats", {}).get("risk_count", 0))
|
||
dimensions["关键指标"].append(r.get("financial_indicators", {}))
|
||
|
||
differences = []
|
||
for dim, values in dimensions.items():
|
||
if dim == "关键指标":
|
||
continue
|
||
unique_vals = set(str(v) for v in values)
|
||
if len(unique_vals) > 1:
|
||
file_names = [Path(fp).name for fp in file_paths]
|
||
labeled = [f"{file_names[i]}: {values[i]}" for i in range(len(values))]
|
||
differences.append(f"[{dim}] {', '.join(labeled)}")
|
||
|
||
return {
|
||
"compared": len(results),
|
||
"files": [
|
||
{"name": Path(fp).name, "path": fp,
|
||
"category": r.get("doc_category", ""),
|
||
"summary": r.get("summary", "")[:200]}
|
||
for fp, r in zip(file_paths, results)
|
||
],
|
||
"dimensions": dimensions,
|
||
"differences": differences,
|
||
}
|
||
|
||
def to_markdown(self, analysis: Dict[str, Any]) -> str:
|
||
"""将 analyze() 结果转换为结构化 Markdown 报告。"""
|
||
if analysis.get("error"):
|
||
return f"## 文档分析失败\n\n错误: {analysis['error']}"
|
||
|
||
lines = [
|
||
"# 文档分析报告\n",
|
||
f"- **文件**: {Path(analysis['file_path']).name}",
|
||
f"- **格式**: {analysis.get('file_type', '').upper()}",
|
||
f"- **分类**: {analysis.get('doc_category', '未知')} (置信度: {analysis.get('category_confidence', 0)}%)",
|
||
f"- **分析时间**: {analysis.get('stats', {}).get('analyzed_at', '')}\n",
|
||
]
|
||
|
||
meta = analysis.get("metadata", {})
|
||
if any(meta.values()):
|
||
lines.append("## 元数据\n")
|
||
if meta.get("title"):
|
||
lines.append(f"- **标题**: {meta['title']}")
|
||
if meta.get("institution"):
|
||
lines.append(f"- **机构**: {meta['institution']}")
|
||
if meta.get("date"):
|
||
lines.append(f"- **日期**: {meta['date']}")
|
||
if meta.get("stock_code"):
|
||
lines.append(f"- **股票代码**: {meta['stock_code']}")
|
||
lines.append("")
|
||
|
||
lines.append("## 内容摘要\n")
|
||
lines.append(f"> {analysis.get('summary', '无法提取摘要')[:500]}\n")
|
||
|
||
sections = analysis.get("sections", [])
|
||
if sections:
|
||
lines.append(f"## 章节结构 ({len(sections)} 个章节)\n")
|
||
for sec in sections[:30]:
|
||
indent = " " * (sec.get("level", 1) - 1)
|
||
preview = sec.get('preview', '')
|
||
lines.append(f"{indent}- **{sec['title']}** _{preview[:80].replace(chr(10), ' ')}_\n")
|
||
lines.append("")
|
||
|
||
indicators = analysis.get("financial_indicators", {})
|
||
if indicators:
|
||
lines.append(f"## 财务指标 ({len(indicators)} 项)\n")
|
||
for key, value in indicators.items():
|
||
if key not in ("amounts_yi", "total_yi", "percentages"):
|
||
lines.append(f"- **{key}**: {value}")
|
||
lines.append("")
|
||
|
||
risks = analysis.get("risk_factors", [])
|
||
if risks:
|
||
lines.append(f"## 风险因素 ({len(risks)} 条)\n")
|
||
for i, risk in enumerate(risks, 1):
|
||
lines.append(f"{i}. {risk}")
|
||
lines.append("")
|
||
|
||
glossary = analysis.get("glossary", {})
|
||
if glossary:
|
||
lines.append(f"## 关键术语 ({len(glossary)} 条)\n")
|
||
for term, defn in list(glossary.items())[:15]:
|
||
lines.append(f"- **{term}**: {defn}")
|
||
|
||
stats = analysis.get("stats", {})
|
||
lines.append(f"\n---\n*文档字符数: {stats.get('char_count', 0):,} | "
|
||
f"章节: {stats.get('section_count', 0)} | "
|
||
f"指标: {stats.get('indicator_count', 0)} | "
|
||
f"风险: {stats.get('risk_count', 0)}*")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ==================== CLI 入口 ====================
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
|
||
if len(sys.argv) < 3:
|
||
print("用法: python document_analyzer.py <命令> <参数>")
|
||
print(" analyze <文件路径> -- 深度分析单份文档")
|
||
print(" organize <目录路径> -- 批量整理文档目录")
|
||
print(" compare <文件1,文件2,...> -- 对比多份文档")
|
||
print(" markdown <文件路径> -- 深度分析并以 Markdown 输出")
|
||
sys.exit(1)
|
||
|
||
cmd = sys.argv[1]
|
||
target = sys.argv[2]
|
||
analyzer = DocumentAnalyzer()
|
||
|
||
if cmd == "analyze":
|
||
result = analyzer.analyze(target)
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
elif cmd == "organize":
|
||
result = analyzer.organize_directory(target)
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
elif cmd == "compare":
|
||
paths = [p.strip() for p in target.split(",") if p.strip()]
|
||
result = analyzer.compare(paths)
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
elif cmd == "markdown":
|
||
result = analyzer.analyze(target)
|
||
print(analyzer.to_markdown(result))
|
||
else:
|
||
print(f"未知命令: {cmd}")
|
||
sys.exit(1)
|