""" services/template_service.py 复刻自 eval_report：report_template_sections 数据的获取方式。 - DEFAULT_TEMPLATE_SECTIONS：系统默认后评价报告章节目录（key, title） - default_section_prompt / default_section_output_contract / default_section_examples： 按章节标题/编号取对应提示词、输出合同、示例 - build_default_template_catalog：默认目录 + 提示词/合同（供上传模版匹配） 说明：eval_report 会额外从《编制细则》与《模版》Word 文档抽取更细的提示词/示例； 本项目默认不含这两个 .doc 文件与 DocParser，故相关函数在缺文件时优雅降级， 回退到 SECTION_PROMPT_RULES / SECTION_EXAMPLE_RULES。 """ from __future__ import annotations import re import uuid from datetime import datetime from functools import lru_cache from pathlib import Path from sqlalchemy.orm import Session from database.models import ReportTemplate, ReportTemplateSection from prompts.report_generation.section_output_contracts import ( DEFAULT_SECTION_OUTPUT_CONTRACT, SECTION_OUTPUT_CONTRACTS, ) from prompts.report_generation.template_prompt_rules import ( DEFAULT_SECTION_PROMPT, SECTION_EXAMPLE_RULES, SECTION_PROMPT_RULES, ) SYSTEM_DEFAULT_TEMPLATE_NAME = "后评价默认模板" GUIDELINE_BASENAME = "炼油化工建设项目后评价报告编制细则（修订）" PROJECT_EXAMPLE_BASENAME = "模版" MAX_SECTION_EXAMPLE_CHARS = 12000 DEFAULT_TEMPLATE_SECTIONS: list[tuple[str, str]] = [ ("1", "1 项目概况"), ("1-1", "1.1 项目基本情况"), ("1-2", "1.2 项目决策要点"), ("1-3", "1.3 项目实施情况"), ("1-4", "1.4 项目运行情况"), ("2", "2 前期工作评价"), ("2-1", "2.1 项目要素评价"), ("2-1-1", "2.1.1 资源与原料评价"), ("2-1-2", "2.1.2 产品方案及市场评价"), ("2-1-2-1", "2.1.2.1 产品方案评价"), ("2-1-2-2", "2.1.2.2 产品市场评价"), ("2-1-3", "2.1.3 工艺方案评价"), ("2-1-3-1", "2.1.3.1 总加工方案评价"), ("2-1-3-2", "2.1.3.2 建设规模及工艺技术方案评价"), ("2-1-3-3", "2.1.3.3 主要设备方案评价"), ("2-1-4", "2.1.4 厂址选择及外部条件评价"), ("2-1-5", "2.1.5 总图及系统配套工程评价"), ("2-1-6", "2.1.6 主要技术指标评价"), ("2-1-7", "2.1.7 风险分析评价"), ("2-2", "2.2 工作程序评价"), ("2-2-1", "2.2.1 编制单位资质及选择方式评价"), ("2-2-2", "2.2.2 编制进度评价"), ("2-2-3", "2.2.3 与专项评价的结合情况"), ("2-2-4", "2.2.4 可行性研究报告的质量评价"), ("2-3", "2.3 前评估工作评价"), ("2-4", "2.4 初步设计评价"), ("2-4-1", "2.4.1 设计单位资质及选择方式评价"), ("2-4-2", "2.4.2 初步设计进度评价"), ("2-4-3", "2.4.3 初步设计质量评价"), ("2-4-4", "2.4.4 初步设计审查工作评价"), ("2-5", "2.5 前期决策程序评价"), ("2-6", "2.6 前期工作评价结论"), ("3", "3 建设实施评价"), ("3-1", "3.1 工程建设管理模式评价"), ("3-2", "3.2 招投标评价"), ("3-3", "3.3 施工图设计评价"), ("3-3-1", "3.3.1 与批复后初步设计符合性评价"), ("3-3-2", "3.3.2 设计进度评价"), ("3-3-3", "3.3.3 施工图设计水平及质量评价"), ("3-3-4", "3.3.4 施工图设计变更管理评价"), ("3-4", "3.4 工程承包商或施工单位评价"), ("3-4-1", "3.4.1 施工准备评价"), ("3-4-2", "3.4.2 施工计划的执行情况"), ("3-5", "3.5 采购工作评价"), ("3-6", "3.6 工程监理评价"), ("3-7", "3.7 工程质量评价"), ("3-8", "3.8 HSE管理评价"), ("3-9", "3.9 三查四定及中间交接"), ("3-10", "3.10 工程竣工验收评价"), ("3-11", "3.11 建设实施评价结论"), ("4", "4 生产运行评价"), ("4-1", "4.1 生产准备评价"), ("4-2", "4.2 联合试运与试生产情况评价"), ("4-3", "4.3 生产运行评价"), ("4-3-1", "4.3.1 原料供应评价"), ("4-3-2", "4.3.2 生产运行总体情况评价"), ("4-3-3", "4.3.3 达标评价"), ("4-3-4", "4.3.4 生产工艺技术评价"), ("4-3-5", "4.3.5 设备运行评价"), ("4-3-6", "4.3.6 公用工程及辅助设施合理性评价"), ("4-4", "4.4 生产运行评价结论"), ("5", "5 投资与经济效益评价"), ("5-1", "5.1 主要经济指标实现程度评价"), ("5-2", "5.2 投资和执行情况评价"), ("5-2-1", "5.2.1 投资控制及变动原因分析"), ("5-2-2", "5.2.2 投资水平分析"), ("5-2-3", "5.2.3 资金来源及到位评价"), ("5-2-4", "5.2.4 投资控制的经验和教训"), ("5-3", "5.3 经济效益分析"), ("5-3-1", "5.3.1 项目投产以来生产经营及效益状况"), ("5-3-2", "5.3.2 项目经济效益后评价"), ("5-4", "5.4 不确定性分析"), ("5-5", "5.5 投资与经济效益评价结论"), ("6", "6 影响与持续性评价"), ("6-1", "6.1 影响评价"), ("6-1-1", "6.1.1 环境影响评价"), ("6-1-2", "6.1.2 安全影响评价"), ("6-1-3", "6.1.3 科技进步影响"), ("6-1-4", "6.1.4 项目社会影响评价"), ("6-1-5", "6.1.5 项目影响评价结论"), ("6-2", "6.2 持续性评价"), ("6-2-1", "6.2.1 资源分析"), ("6-2-2", "6.2.2 产品分析"), ("6-2-3", "6.2.3 主要技术及经济指标对比"), ("6-2-4", "6.2.4 项目持续性评价结论"), ("7", "7 综合评价结论"), ("7-1", "7.1 综合评价结论"), ("7-1-1", "7.1.1 总体评价结论"), ("7-1-2", "7.1.2 成功度评价"), ("7-2", "7.2 主要经验"), ("7-3", "7.3 问题与建议"), ] def default_section_output_contract(section_title: str, section_key: str | None = None) -> str: section_no = _extract_number_prefix(section_title) or _section_key_to_number(section_key) if section_no and section_no in SECTION_OUTPUT_CONTRACTS: return SECTION_OUTPUT_CONTRACTS[section_no] return DEFAULT_SECTION_OUTPUT_CONTRACT def default_section_prompt(section_title: str, section_key: str | None = None) -> str: guideline_prompt = _guideline_prompt_for(section_title, section_key) if guideline_prompt: return guideline_prompt title = _normalize_section_identity(section_title) key = str(section_key or "").strip().lower() for pattern, prompt in SECTION_PROMPT_RULES: p = pattern.lower() if title.startswith(p): return prompt if p.isdigit() and (title.startswith(f"{p} ") or key.startswith(f"{p}-") or key == p): return prompt return DEFAULT_SECTION_PROMPT def build_default_template_catalog() -> list[dict[str, str]]: """系统默认模板章节目录及对应提示词、输出合同（供上传模版匹配）。""" out: list[dict[str, str]] = [] for key, title in DEFAULT_TEMPLATE_SECTIONS: out.append( { "sectionKey": key, "sectionTitle": title, "sectionNumber": _extract_number_prefix(title) or _section_key_to_number(key), "sectionPrompt": default_section_prompt(title, key), "sectionOutputContract": default_section_output_contract(title, key), } ) return out def default_section_examples(section_title: str, section_key: str | None = None) -> str: project_example = _project_example_for(section_title, section_key) if project_example: return project_example title = _normalize_section_identity(section_title) key = str(section_key or "").strip().lower() num = _extract_number_prefix(section_title) or _section_key_to_number(section_key) chapter_no = "" if num: chapter_no = num.split(".")[0] elif key: chapter_no = key.split("-")[0] for prefix, examples in SECTION_EXAMPLE_RULES: p = str(prefix).strip().lower() if chapter_no == p: return examples if title.startswith(f"{p} "): return examples if key.startswith(f"{p}-") or key == p: return examples return "" def _normalize_section_identity(value: str | None) -> str: text = str(value or "").strip().lower() text = text.replace("．", ".").replace("。", ".") text = re.sub(r"\s+", " ", text) return text def _section_key_to_number(section_key: str | None) -> str: key = str(section_key or "").strip() if not key: return "" if re.fullmatch(r"\d+(?:-\d+)*", key): return key.replace("-", ".") return "" def _extract_number_prefix(title: str) -> str: m = re.match(r"^\s*(\d+(?:\.\d+)*)\s*", str(title or "")) return m.group(1) if m else "" def _normalize_heading_key(value: str) -> str: s = str(value or "").strip().lower() s = s.replace("．", ".").replace("。", ".") s = re.sub(r"\s+", "", s) return s def _tuple_from_number(number_str: str) -> tuple[int, ...]: if not number_str: return tuple() parts = [] for p in number_str.split("."): if p.isdigit(): parts.append(int(p)) else: return tuple() return tuple(parts) def _read_doc_text(path: str) -> str: """读取 .doc/.docx 文本。本项目无 DocParser 时返回空串（优雅降级）。""" try: from function.documents.doc_parser import DocParser # type: ignore except Exception: return "" try: return DocParser(path).read() except Exception: return "" @lru_cache(maxsize=1) def _guideline_section_prompt_map() -> dict[str, str]: guideline_path = _resolve_guideline_path() if not guideline_path: return {} raw_text = _read_doc_text(guideline_path) if not raw_text: return {} return _build_guideline_prompt_map(raw_text) def _resolve_guideline_path() -> str | None: root = Path(__file__).resolve().parents[1] candidates = [ root / f"{GUIDELINE_BASENAME}.doc", root / f"{GUIDELINE_BASENAME}.docx", ] for p in candidates: if p.is_file(): return str(p) return None def _resolve_project_example_path() -> str | None: root = Path(__file__).resolve().parents[1] candidates = [ root / f"{PROJECT_EXAMPLE_BASENAME}.doc", root / f"{PROJECT_EXAMPLE_BASENAME}.docx", ] for p in candidates: if p.is_file(): return str(p) return None @lru_cache(maxsize=1) def _project_example_entries() -> list[tuple[str, str]]: path = _resolve_project_example_path() if not path: return [] raw_text = _read_doc_text(path) if not raw_text: return [] return _build_project_example_entries(raw_text) def _build_project_example_entries(text: str) -> list[tuple[str, str]]: lines = str(text or "").splitlines() headings: list[tuple[int, int, str]] = [] for idx, raw in enumerate(lines): line = str(raw or "").strip() m = re.match(r"^\s*(#{1,6})\s*(.+?)\s*$", line) if not m: continue level = len(m.group(1)) heading_title = m.group(2).strip() if not heading_title: continue headings.append((idx, level, heading_title)) out: list[tuple[str, str]] = [] for i, (start_idx, level, title) in enumerate(headings): end_idx = len(lines) for j in range(i + 1, len(headings)): next_idx, next_level, _ = headings[j] if next_level <= level: end_idx = next_idx break body = "\n".join(lines[start_idx + 1 : end_idx]).strip() body = re.sub(r"\n{3,}", "\n\n", body) if not body: continue out.append((title, body)) return out def _project_example_for(section_title: str, section_key: str | None = None) -> str: entries = _project_example_entries() if not entries: return "" target_title = _clean_section_title(section_title) target_key = _section_key_to_number(section_key) target_core = _core_title(target_title or target_key) if not target_core: return "" best_title = "" best_body = "" best_score = -1 for heading, body in entries: heading_clean = _clean_section_title(heading) heading_core = _core_title(heading_clean) score = _title_match_score(target_core, heading_core) if score > best_score: best_score = score best_title = heading_clean best_body = body if best_score < 4 or not best_body: return "" text = f"### {best_title}\n\n{best_body}".strip() if len(text) > MAX_SECTION_EXAMPLE_CHARS: text = text[:MAX_SECTION_EXAMPLE_CHARS].rstrip() + "\n\n（示例过长，已截断）" return text def _clean_section_title(value: str | None) -> str: s = str(value or "").strip() s = re.sub(r"^\s*\d+(?:[.\-]\d+)*\s*", "", s) return s.strip() def _core_title(value: str | None) -> str: s = str(value or "").strip() s = s.replace("（", "(").replace("）", ")") s = re.sub(r"\([^)]*\)", "", s) s = re.sub(r"[、，。；：:（）()\-\s]", "", s) s = s.replace("项目", "") s = s.replace("情况", "") s = s.replace("工作", "") return s.strip().lower() def _title_match_score(target: str, candidate: str) -> int: if not target or not candidate: return 0 if target == candidate: return 100 score = 0 if target in candidate or candidate in target: score += 40 tks_t = re.findall(r"[\u4e00-\u9fa5]{2,8}|[a-z]{2,12}", target) tks_c = re.findall(r"[\u4e00-\u9fa5]{2,8}|[a-z]{2,12}", candidate) if tks_t and tks_c: overlap = len(set(tks_t) & set(tks_c)) score += overlap * 8 ch_overlap = len(set(target) & set(candidate)) score += min(ch_overlap, 20) return score def _build_guideline_prompt_map(text: str) -> dict[str, str]: lines = str(text or "").splitlines() headings: list[tuple[int, str, str, tuple[int, ...]]] = [] for idx, raw in enumerate(lines): line = str(raw or "").strip() m = re.match(r"^\s*#{1,6}\s*(.+?)\s*$", line) if not m: continue heading_title = m.group(1).strip() number = _extract_number_prefix(heading_title) number_tuple = _tuple_from_number(number) if not number_tuple: continue headings.append((idx, heading_title, number, number_tuple)) prompt_map: dict[str, str] = {} for i, (start_idx, heading_title, number, number_tuple) in enumerate(headings): end_idx = len(lines) for j in range(i + 1, len(headings)): next_start, _, _, next_tuple = headings[j] if len(next_tuple) < len(number_tuple) or next_tuple[: len(number_tuple)] != number_tuple: end_idx = next_start break body = "\n".join(lines[start_idx + 1 : end_idx]).strip() body = re.sub(r"\n{3,}", "\n\n", body) if not body: continue key_title = _normalize_heading_key(heading_title) key_number = _normalize_heading_key(number) prompt_map[key_title] = body prompt_map[key_number] = body return prompt_map def _guideline_prompt_for(section_title: str, section_key: str | None = None) -> str: mapping = _guideline_section_prompt_map() if not mapping: return "" title = str(section_title or "").strip() number = _extract_number_prefix(title) or _section_key_to_number(section_key) candidates = [ _normalize_heading_key(title), _normalize_heading_key(number), ] for key in candidates: if key and key in mapping: return mapping[key] return "" def list_templates(db: Session) -> list[ReportTemplate]: return ( db.query(ReportTemplate) .order_by(ReportTemplate.is_default.desc(), ReportTemplate.updated_at.desc()) .all() ) def ensure_default_template(db: Session) -> None: now = datetime.now() system_default = ( db.query(ReportTemplate) .filter(ReportTemplate.name == SYSTEM_DEFAULT_TEMPLATE_NAME) .first() ) if not system_default: system_default = ReportTemplate( id=uuid.uuid4().hex, name=SYSTEM_DEFAULT_TEMPLATE_NAME, description="系统预置模板（细则完整章节）", is_default=True, is_active=True, created_at=now, updated_at=now, ) db.add(system_default) db.flush() current_rows = ( db.query(ReportTemplateSection) .filter(ReportTemplateSection.template_id == system_default.id) .order_by(ReportTemplateSection.section_order.asc()) .all() ) current_pairs = [(r.section_key, r.section_title) for r in current_rows] expected_pairs = list(DEFAULT_TEMPLATE_SECTIONS) db.query(ReportTemplate).update({ReportTemplate.is_default: False}) system_default.is_default = True system_default.is_active = True system_default.updated_at = now if current_pairs == expected_pairs: has_changed = False for row in current_rows: current_examples = str(row.examples or "").strip() new_examples = default_section_examples(row.section_title, row.section_key).strip() if new_examples and current_examples != new_examples: row.examples = new_examples row.updated_at = now has_changed = True current_out = str(getattr(row, "section_output_contract", None) or "").strip() new_out = default_section_output_contract(row.section_title, row.section_key).strip() if not current_out and new_out: row.section_output_contract = new_out row.updated_at = now has_changed = True if has_changed: system_default.updated_at = now db.commit() return db.query(ReportTemplateSection).filter( ReportTemplateSection.template_id == system_default.id ).delete() for i, (key, title) in enumerate(DEFAULT_TEMPLATE_SECTIONS): db.add( ReportTemplateSection( id=uuid.uuid4().hex, template_id=system_default.id, section_key=key, section_title=title, section_prompt="", section_output_contract=default_section_output_contract(title, key), section_order=i, examples=default_section_examples(title, key), created_at=now, updated_at=now, ) ) db.commit()