feat: SourceParser 门面全量输入聚合 StructuredSource

This commit is contained in:
lhl
2026-08-10 14:57:04 +08:00
parent 9ab11da87d
commit 7f29cc9cf3
3 changed files with 172 additions and 0 deletions
+73
View File
@@ -0,0 +1,73 @@
from __future__ import annotations
from pathlib import Path
from genesis.data_models import StructuredSource
from genesis.parsers.excel_parser import ExcelParser
from genesis.parsers.rule_doc_parser import RuleDocParser
from genesis.parsers.word_template_parser import WordTemplateParser
XLSX_EXTS = (".xlsx", ".xls")
DOCX_EXT = ".docx"
class SourceParser:
"""全量输入门面:Excel 要件定义 + Word 模板 + Word 规则 → StructuredSource。
角色由调用方按 api-design file_type 语义显式传入(requirements/template/
write_instruction/rules),不做基于文件名的隐式猜测(spec §3.4)。
"""
def __init__(self) -> None:
self._excel = ExcelParser()
def parse(
self,
requirement_paths: list[str | Path] | None = None,
template_path: str | Path | None = None,
write_instruction_paths: list[str | Path] | None = None,
rule_paths: list[str | Path] | None = None,
) -> StructuredSource:
requirement_paths = requirement_paths or []
write_instruction_paths = write_instruction_paths or []
rule_paths = rule_paths or []
tables = []
comments = []
for p in requirement_paths:
path = Path(p)
if path.suffix.lower() not in XLSX_EXTS:
raise ValueError(f"不支持的文件类型: {path.suffix or '(无扩展名)'}")
if not path.exists():
raise FileNotFoundError(str(p))
result = self._excel.parse(path)
tables.extend(result.tables)
comments.extend(result.comments)
template = None
if template_path is not None:
tpath = Path(template_path)
if tpath.suffix.lower() != DOCX_EXT:
raise ValueError(f"不支持的文件类型: {tpath.suffix or '(无扩展名)'}")
if not tpath.exists():
raise FileNotFoundError(str(template_path))
template = WordTemplateParser().parse(tpath)
rule_docs = []
for p in [*write_instruction_paths, *rule_paths]:
path = Path(p)
if path.suffix.lower() != DOCX_EXT:
raise ValueError(f"不支持的文件类型: {path.suffix or '(无扩展名)'}")
if not path.exists():
raise FileNotFoundError(str(p))
# 做成说明书与记入规则均为 Type A 写入规则 → writeapi-design §2.2
rule_docs.append(RuleDocParser().parse(path, category="write"))
return StructuredSource(
tables=tables,
template=template,
rule_docs=rule_docs,
image_analyses=[],
existing_system=None,
comments=comments,
)