Coverage for src\genesis\writer\models.py: 100%

1"""Writer 子系统数据模型（Phase 5）。""" 

2from __future__ import annotations 

3 

4from dataclasses import dataclass, field 

5from typing import Literal 

6 

7from genesis.data_models import ElementType, SheetType 

8 

9# 单表渲染行数上限（防 token 爆炸；样本量小，通常不触发） 

10MAX_ROWS_PER_TABLE = 200 

11 

12# design.md §6.8 ①「DataGate.load(structured_source, selector=该章数据)」的章节级选择器： 

13# 章节占位符 id → 本章对应的 Excel Sheet 类型 

14# - introduction 概览章注入全部类型表 

15# - 未登记的章节 id 缺省为空列表 → 仅 GENERIC 自由記述作背景 

16_ALL_SHEET_TYPES = list(SheetType) 

17CHAPTER_SHEET_TYPES: dict[str, list[SheetType]] = { 

18 "introduction": _ALL_SHEET_TYPES, 

19 "function_list": [SheetType.FUNCTION], 

20 "screen_list": [SheetType.SCREEN], 

21 "report_list": [SheetType.REPORT], 

22 "db_design": [SheetType.DATABASE], 

23 "if_definition": [SheetType.INTERFACE], 

24 "batch_list": [SheetType.BATCH], 

25} 

26 

27# 章节 id → 影响调查要素类型（None = 展示全部，用于概览章） 

28CHAPTER_IMPACT_ELEMENT: dict[str, ElementType | None] = { 

29 "function_list": ElementType.FUNCTION, 

30 "screen_list": ElementType.SCREEN, 

31 "report_list": ElementType.REPORT, 

32 "db_design": ElementType.DB, 

33 "if_definition": ElementType.IF, 

34 "batch_list": ElementType.BATCH, 

35 "introduction": None, 

36} 

37 

38 

39@dataclass 

40class ContentBlock: 

41 """LLM 生成的内容块。注意：table.headers/caption、list.items/style 在渲染至 

42 DocxInjector.Block 时显式丢弃（renderer 中声明并测试）。""" 

43 

44 block_id: str 

45 type: Literal["paragraph", "heading", "table", "list", "note"] 

46 level: int | None = None 

47 text: str | None = None 

48 caption: str | None = None 

49 headers: list[str] | None = None 

50 rows: list[list[str]] | None = None 

51 items: list[str] | None = None 

52 style: str | None = None 

53 source_uris: list[str] = field(default_factory=list) 

54 

55 @classmethod 

56 def from_dict(cls, block_id: str, data: dict) -> "ContentBlock": 

57 """从 LLM 输出的 block dict 安全构造内容块。""" 

58 return cls( 

59 block_id=str(block_id), 

60 type=data.get("type", "paragraph"), 

61 level=data.get("level"), 

62 text=data.get("text"), 

63 caption=data.get("caption"), 

64 headers=data.get("headers"), 

65 rows=data.get("rows"), 

66 items=data.get("items"), 

67 style=data.get("style"), 

68 source_uris=data.get("source_uris", []), 

69 ) 

70 

71 

72@dataclass 

73class ChapterContent: 

74 chapter_id: str 

75 version: int 

76 title: str 

77 blocks: list[ContentBlock] 

78 

79 @classmethod 

80 def from_llm(cls, chapter_id: str, title: str, data: dict) -> "ChapterContent": 

81 """从 LLM 结构化输出（含 title、blocks 列表）构造章节内容。 

82 

83 对每个 block dict 用 ContentBlock.from_dict 安全取值；block_id 缺省为序号字符串。 

84 """ 

85 blocks: list[ContentBlock] = [] 

86 for i, b in enumerate(data.get("blocks", [])): 

87 blocks.append(ContentBlock.from_dict(str(b.get("block_id", i)), b)) 

88 return cls(chapter_id=chapter_id, version=1, title=title, blocks=blocks) 

89 

90 

91@dataclass 

92class ChapterSpec: 

93 """template_mapper 产出：驱动 WriterAgent 串行顺序。""" 

94 

95 chapter_id: str 

96 title: str 

97 section_placeholder: str | None = None # 如 "{{section:db_design}}"，无则 None 

98 sub_headings: list[str] = field(default_factory=list) # 本章 H2/H3 子节标题（§6.5 归并） 

99 

100 

101@dataclass 

102class GenerationContext: 

103 chapter_id: str 

104 title: str 

105 template_marker: ChapterSpec 

106 structured_source: object | None 

107 write_rules: list[str] 

108 design_rules: list[str] 

109 template_styles: set[str] 

110 prior_state: object | None = None # WriterState，避免循环 import 用 object 

111 impact_report: object | None = None # ImpactReport 影响调查书（生成主上下文） 

112 output_language: str = "auto" # "auto" | "zh" | "ja"（步骤 1：用户可选输出语言） 

113 

114 def _language_instruction(self) -> str: 

115 """根据 output_language 生成【语言约束】段的具体指令（步骤 1）。""" 

116 if self.output_language == "zh": 

117 return "必须使用简体中文撰写（标题、正文与所有说明一律中文）。" 

118 if self.output_language == "ja": 

119 return "必ず日本語で記述すること（タイトル・本文・すべての説明は日本語）。" 

120 # auto：沿用与标题语言一致的旧语义（向后兼容既有日文文档） 

121 return ( 

122 f"必须与章节标题「{self.title}」所用自然语言保持一致：" 

123 "标题为日文则用日文撰写，为中文则用中文撰写，依此类推。" 

124 ) 

125 

126 def to_vars(self) -> dict: 

127 """返回供 prompt 渲染的变量字典。""" 

128 tm = self.template_marker 

129 template_marker = f"{tm.chapter_id}:{tm.title}" if tm is not None else "" 

130 return { 

131 "chapter_id": self.chapter_id, 

132 "title": self.title, 

133 "template_marker": template_marker, 

134 "write_rules": "\n".join(self.write_rules), 

135 "design_rules": "\n".join(self.design_rules), 

136 "template_styles": ", ".join(sorted(self.template_styles)), 

137 "sub_headings": "\n".join( 

138 f"- {h}" for h in (getattr(tm, "sub_headings", None) or []) 

139 ), 

140 "prior_state": str(self.prior_state) if self.prior_state is not None else "", 

141 "language_instruction": self._language_instruction(), 

142 "data": _format_chapter_data( 

143 self.structured_source, 

144 CHAPTER_SHEET_TYPES.get(self.chapter_id, []), 

145 ), 

146 "impact": _format_impact( 

147 self.impact_report, 

148 CHAPTER_IMPACT_ELEMENT.get(self.chapter_id, None), 

149 self.output_language, 

150 ), 

151 } 

152 

153 

154def _render_table(tb) -> list[str]: 

155 """将单张 ExcelTable 渲染为可读 Markdown 行（表名行 + 表头 + 分隔 + 数据行）。""" 

156 lines = [f"### 表: {tb.name}（{tb.detected_type.value}）"] 

157 headers = list(tb.headers) 

158 lines.append("| " + " | ".join(headers) + " |") 

159 lines.append("|" + "|".join([" --- "] * len(headers)) + "|") 

160 for row in tb.rows[:MAX_ROWS_PER_TABLE]: 

161 cells = [] 

162 for h in headers: 

163 v = row.get(h) 

164 value = getattr(v, "value", v) 

165 cells.append("" if value is None else str(value)) 

166 lines.append("| " + " | ".join(cells) + " |") 

167 return lines 

168 

169 

170def _format_chapter_data(structured_source: object | None, sheet_types: list[SheetType]) -> str: 

171 """按章节定向格式化要件定义数据（design.md §6.8 ① selector=该章数据）。 

172 

173 - 命中 sheet_types 的表全部注入（章节主题数据） 

174 - GENERIC（自由記述）作为通用背景始终注入 

175 - structured_source 为 None 或无任何可注入表时返回空串 

176 """ 

177 if structured_source is None: 

178 return "" 

179 tables = getattr(structured_source, "tables", None) or [] 

180 matched = [ 

181 t for t in tables 

182 if t.detected_type in sheet_types and t.detected_type != SheetType.GENERIC 

183 ] 

184 generic = [t for t in tables if t.detected_type == SheetType.GENERIC] 

185 selected = matched + generic 

186 if not selected: 

187 return "" 

188 lines: list[str] = [] 

189 for tb in selected: 

190 lines.extend(_render_table(tb)) 

191 lines.append("") 

192 return "\n".join(lines).rstrip() 

193 

194 

195# 影响调查标签本地化（步骤 B）：auto/ja 默认日文，zh 中文 

196# 注：方括号标记 [..] 保留（中日通用），仅标签词本地化 

197_IMPACT_LABELS: dict[str, dict[str, str]] = { 

198 "zh": { 

199 "new": "新建", "modified": "变更", "deleted": "删除", "warning": "警告", 

200 "affected": "受影响", 

201 }, 

202 "ja": { 

203 "new": "新規", "modified": "変更", "deleted": "削除", "warning": "警告", 

204 "affected": "受影响", 

205 }, 

206} 

207 

208 

209def _format_impact( 

210 report: object | None, 

211 element_type: ElementType | None = None, 

212 output_language: str = "auto", 

213) -> str: 

214 """将影响调查书格式化为 prompt 可读文本（无报告/无分析时为空串）。 

215 

216 element_type 非 None 时仅保留该类型要素（章节级定向，design.md §6.8 ①）； 

217 警告始终保留（不依赖要素类型）。 

218 output_language 控制标签语言（步骤 B）；auto 回落到 ja 标签。 

219 """ 

220 if report is None: 

221 return "" 

222 ca = getattr(report, "change_analysis", None) 

223 if ca is None: 

224 return "" 

225 

226 def keep(el) -> bool: 

227 return element_type is None or el.element_type == element_type.value 

228 

229 lab = _IMPACT_LABELS.get(output_language, _IMPACT_LABELS["ja"]) 

230 summary = getattr(report, "summary", {}) or {} 

231 lines = [f"project_type={getattr(ca, 'project_type', '')}"] 

232 lines.append( 

233 "summary: new={new} modified={modified} deleted={deleted} " 

234 "unchanged={unchanged} warnings={warnings}".format( 

235 new=summary.get("new", 0), modified=summary.get("modified", 0), 

236 deleted=summary.get("deleted", 0), unchanged=summary.get("unchanged", 0), 

237 warnings=summary.get("warnings", 0), 

238 ) 

239 ) 

240 for el in getattr(ca, "new_elements", []) or []: 

241 if keep(el): 

242 lines.append(f"[{lab['new']}] {el.element_id} {el.element_type} {el.name}") 

243 for el in getattr(ca, "modified_elements", []) or []: 

244 if keep(el): 

245 impacted = ", ".join(el.impacted_existing) or "-" 

246 lines.append(f"[{lab['modified']}] {el.element_id} {el.element_type} {el.name} → {lab['affected']}: {impacted}") 

247 for el in getattr(ca, "deleted_elements", []) or []: 

248 if keep(el): 

249 impacted = ", ".join(el.impacted_existing) or "-" 

250 lines.append(f"[{lab['deleted']}] {el.element_id} {el.element_type} {el.name} → {lab['affected']}: {impacted}") 

251 for w in getattr(ca, "warnings", []) or []: 

252 lines.append(f"[{lab['warning']}] {w.element_id}: {w.issue}") 

253 return "\n".join(lines)