xst
2122 lines
77 KiB
Python
2122 lines
77 KiB
Python
import argparse
|
||
import filecmp
|
||
import json
|
||
from pathlib import Path
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from typing import Any
|
||
import xml.etree.ElementTree as ET
|
||
from zipfile import BadZipFile, ZipFile
|
||
import zlib
|
||
|
||
try:
|
||
from docx import Document as DocxDocument
|
||
except ImportError:
|
||
DocxDocument = None
|
||
|
||
try:
|
||
from pyantiword.antiword_wrapper import extract_text as extract_doc_text
|
||
except ImportError:
|
||
try:
|
||
from pyantiword import extract_text as extract_doc_text
|
||
except ImportError:
|
||
extract_doc_text = None
|
||
|
||
from export_excel import (
|
||
LEGACY_COLUMNS,
|
||
REQUIRED_COLUMNS,
|
||
TYPE_ENUMS,
|
||
export_markdown_to_excel,
|
||
load_markdown_table,
|
||
)
|
||
from openpyxl import load_workbook
|
||
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||
REQUIREMENTS_DIR = REPO_ROOT / "requirements"
|
||
RAW_REQUIREMENTS_DIR = REPO_ROOT / "source_docs" / "requirements_raw"
|
||
TECHNICAL_SOLUTIONS_DIR = REPO_ROOT / "source_docs" / "technical_solutions"
|
||
DECISIONS_DIR = REPO_ROOT / "decisions"
|
||
DECISIONS_APPLIED_DIR = DECISIONS_DIR / "applied"
|
||
VERSIONS_DIR = REPO_ROOT / "output" / "versions"
|
||
MAX_SNAPSHOT_VERSIONS = 3
|
||
PROJECT_PROFILE_FILE = REPO_ROOT / "knowledge_base" / "00_project" / "project_profile.md"
|
||
CORE_TERMINOLOGY_FILE = REPO_ROOT / "knowledge_base" / "01_standards" / "terminology.md"
|
||
SUPPORTED_REQUIREMENT_SUFFIXES = {".md", ".doc", ".docx", ".pdf", ".txt"}
|
||
DOCX_NAMESPACE = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"}
|
||
PDF_STREAM_RE = re.compile(rb"stream\r?\n(.*?)\r?\nendstream", re.DOTALL)
|
||
PDF_TEXT_BLOCK_RE = re.compile(rb"BT(.*?)ET", re.DOTALL)
|
||
PDF_ARRAY_TEXT_RE = re.compile(rb"\[(.*?)\]\s*TJ", re.DOTALL)
|
||
PDF_SIMPLE_TEXT_RE = re.compile(rb"(?<!\[)\((?:\\.|[^\\)])*\)\s*Tj", re.DOTALL)
|
||
PDF_INLINE_STRING_RE = re.compile(rb"\((?:\\.|[^\\)])*\)")
|
||
TOPIC_NOISE_TOKENS = (
|
||
"产品需求文档",
|
||
"需求文档",
|
||
"技术方案",
|
||
"技术设计",
|
||
"设计文档",
|
||
"技术文档",
|
||
"产品需求",
|
||
"需求说明",
|
||
"技术说明",
|
||
"需求",
|
||
"方案",
|
||
"设计",
|
||
"prd",
|
||
)
|
||
KNOWLEDGE_BASE_FILES = [
|
||
CORE_TERMINOLOGY_FILE,
|
||
REPO_ROOT / "knowledge_base" / "01_standards" / "test_case_template.md",
|
||
REPO_ROOT / "knowledge_base" / "01_standards" / "definition_of_done.md",
|
||
REPO_ROOT / "knowledge_base" / "01_standards" / "review_checklist.md",
|
||
REPO_ROOT / "knowledge_base" / "02_history" / "common_missed_scenes.md",
|
||
REPO_ROOT / "knowledge_base" / "02_history" / "historical_defects.md",
|
||
REPO_ROOT / "knowledge_base" / "02_history" / "marketing_rules.md",
|
||
REPO_ROOT / "knowledge_base" / "03_best_practices" / "payment_flow_cases.md",
|
||
REPO_ROOT / "knowledge_base" / "03_best_practices" / "marketing_activity_cases.md",
|
||
REPO_ROOT / "knowledge_base" / "03_best_practices" / "order_manage_cases.md",
|
||
REPO_ROOT / "knowledge_base" / "03_best_practices" / "data_reporting_cases.md",
|
||
]
|
||
OPTIONAL_TERMINOLOGY_RULES = [
|
||
{
|
||
"name": "saas",
|
||
"path": REPO_ROOT / "knowledge_base" / "01_standards" / "terminology_optional_saas.md",
|
||
"keywords": [
|
||
"导购",
|
||
"导购关系",
|
||
"分销",
|
||
"分销员",
|
||
"推广员",
|
||
"佣金",
|
||
"佣金结算",
|
||
"渠道活码",
|
||
"渠道码",
|
||
"企微",
|
||
"企业微信",
|
||
"企微客户",
|
||
"社群",
|
||
"客户标签",
|
||
"客户分群",
|
||
"客户归属",
|
||
"私域",
|
||
"召回活动",
|
||
"生命周期",
|
||
"储值",
|
||
"储值余额",
|
||
"会员储值",
|
||
"礼品卡",
|
||
"赠送金额",
|
||
"冻结余额",
|
||
"可用余额",
|
||
],
|
||
}
|
||
]
|
||
NORMATIVE_KEYWORDS = (
|
||
"必须",
|
||
"应",
|
||
"需要",
|
||
"支持",
|
||
"允许",
|
||
"可",
|
||
"禁止",
|
||
"不得",
|
||
"不能",
|
||
"不可",
|
||
"仅",
|
||
"只",
|
||
)
|
||
NEGATIVE_POLARITY_KEYWORDS = (
|
||
"不支持",
|
||
"禁止",
|
||
"不得",
|
||
"不能",
|
||
"不可",
|
||
"仅支持",
|
||
"只支持",
|
||
"仅限",
|
||
"只可",
|
||
"不可与",
|
||
)
|
||
POSITIVE_POLARITY_KEYWORDS = (
|
||
"支持",
|
||
"允许",
|
||
"可以",
|
||
"可",
|
||
"必须",
|
||
"应",
|
||
"需要",
|
||
)
|
||
RELATED_SIMILARITY_THRESHOLD = 0.015
|
||
CONFLICT_SIMILARITY_THRESHOLD = 0.22
|
||
MAX_RELATED_REQUIREMENTS = 5
|
||
NUMERIC_TOKEN_RE = re.compile(r"\d+(?:\.\d+)?(?:%|分钟|秒|元|天|次)?")
|
||
VERSION_DIR_RE = re.compile(r"^v(\d+)$")
|
||
LEGACY_EXCEL_EXPORT_RE = re.compile(r"^(?P<stem>.+)_\d{8}_\d{6}\.xlsx$")
|
||
DECISION_FILE_RE = re.compile(r"确认状态[:：]\s*`?(已确认|待确认|已驳回|已拒绝)`?")
|
||
DECISION_EXPORT_RE = re.compile(r"是否允许在确认前继续导出[:：]\s*`?(允许|不允许)`?")
|
||
DECISION_RELATION_TYPE_RE = re.compile(r"关系类型[:：]\s*`?(补充|替代|并行)`?")
|
||
OUT_OF_SCOPE_HEADING_KEYWORDS = (
|
||
"不在本期范围",
|
||
"非本期范围",
|
||
"不支持",
|
||
"不包含",
|
||
"out of scope",
|
||
)
|
||
|
||
|
||
def resolve_requirement_path(requirement: str) -> Path:
|
||
path = Path(requirement)
|
||
return path if path.is_absolute() else REPO_ROOT / path
|
||
|
||
|
||
def build_paths(base_name: str) -> dict[str, Path]:
|
||
return {
|
||
"analysis": REPO_ROOT / "output" / "analysis" / f"{base_name}_分析.md",
|
||
"relation_report": REPO_ROOT / "output" / "analysis" / f"{base_name}_关联与冲突.md",
|
||
"test_points": REPO_ROOT / "output" / "test_points" / f"{base_name}_测试点.md",
|
||
"test_cases": REPO_ROOT / "output" / "test_cases" / f"{base_name}_测试用例.md",
|
||
"manifest": REPO_ROOT / "output" / "manifests" / f"{base_name}.json",
|
||
"normalized_input_dir": REPO_ROOT / "output" / "normalized_inputs" / base_name,
|
||
"excel_dir": REPO_ROOT / "output" / "excel_reports",
|
||
"version_root": VERSIONS_DIR / base_name,
|
||
"decision_file": DECISIONS_DIR / f"{base_name}_确认结论.md",
|
||
"maintained_requirement": REQUIREMENTS_DIR / f"{base_name}.md",
|
||
}
|
||
|
||
|
||
def ensure_output_dirs(paths: dict[str, Path]) -> None:
|
||
for key in ("analysis", "relation_report", "test_points", "test_cases", "manifest", "excel_dir"):
|
||
paths[key].parent.mkdir(parents=True, exist_ok=True)
|
||
paths["normalized_input_dir"].mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
def validate_requirement_file(requirement_path: Path) -> None:
|
||
if not requirement_path.exists():
|
||
raise FileNotFoundError(f"需求文档不存在：{requirement_path}")
|
||
if requirement_path.stat().st_size == 0:
|
||
raise ValueError(f"需求文档为空：{requirement_path}")
|
||
if requirement_path.suffix.lower() not in SUPPORTED_REQUIREMENT_SUFFIXES:
|
||
raise ValueError(
|
||
"当前流水线仅支持以下需求文档格式："
|
||
f"{', '.join(sorted(SUPPORTED_REQUIREMENT_SUFFIXES))}。"
|
||
f"当前文件：{requirement_path}"
|
||
)
|
||
|
||
|
||
def validate_knowledge_base() -> None:
|
||
missing = [str(file_path) for file_path in KNOWLEDGE_BASE_FILES if not file_path.exists()]
|
||
if missing:
|
||
raise FileNotFoundError("缺少知识库或规范文件：\n" + "\n".join(missing))
|
||
|
||
optional_missing = [
|
||
str(rule["path"])
|
||
for rule in OPTIONAL_TERMINOLOGY_RULES
|
||
if not Path(rule["path"]).exists()
|
||
]
|
||
if optional_missing:
|
||
raise FileNotFoundError("缺少可选术语文件：\n" + "\n".join(optional_missing))
|
||
|
||
|
||
def validate_project_profile() -> None:
|
||
if not PROJECT_PROFILE_FILE.exists():
|
||
raise FileNotFoundError(f"缺少项目画像文件：{PROJECT_PROFILE_FILE}")
|
||
if PROJECT_PROFILE_FILE.stat().st_size == 0:
|
||
raise ValueError(f"项目画像文件为空：{PROJECT_PROFILE_FILE}")
|
||
|
||
|
||
def safe_read_text(path: Path) -> str:
|
||
suffix = path.suffix.lower()
|
||
if suffix == ".doc":
|
||
return read_doc_text_file(path)
|
||
if suffix == ".docx":
|
||
return read_docx_text(path)
|
||
if suffix == ".pdf":
|
||
return read_pdf_text(path)
|
||
return path.read_text(encoding="utf-8", errors="ignore")
|
||
|
||
|
||
def read_doc_text_file(path: Path) -> str:
|
||
if extract_doc_text is None:
|
||
raise RuntimeError(
|
||
"当前环境缺少 .doc 解析依赖 `pyantiword`，请先执行 `pip install -r requirements.txt`。"
|
||
)
|
||
|
||
try:
|
||
return (extract_doc_text(str(path)) or "").strip()
|
||
except Exception as exc:
|
||
raise ValueError(f"无法解析 DOC 文件：{path}\n{exc}") from exc
|
||
|
||
|
||
def read_docx_text(path: Path) -> str:
|
||
library_text = read_docx_text_via_library(path)
|
||
if library_text.strip():
|
||
return library_text.strip()
|
||
|
||
try:
|
||
with ZipFile(path) as archive:
|
||
document_xml = archive.read("word/document.xml")
|
||
except KeyError as exc:
|
||
raise ValueError(f"DOCX 文件缺少 word/document.xml：{path}") from exc
|
||
except BadZipFile as exc:
|
||
raise ValueError(f"无法解析 DOCX 文件：{path}") from exc
|
||
|
||
root = ET.fromstring(document_xml)
|
||
body = root.find("w:body", DOCX_NAMESPACE)
|
||
if body is None:
|
||
return ""
|
||
|
||
lines: list[str] = []
|
||
for child in body:
|
||
tag = child.tag.rsplit("}", 1)[-1]
|
||
if tag == "p":
|
||
text = extract_docx_paragraph_text(child)
|
||
if text:
|
||
lines.append(text)
|
||
elif tag == "tbl":
|
||
lines.extend(extract_docx_table_lines(child))
|
||
return "\n".join(lines).strip()
|
||
|
||
|
||
def read_docx_text_via_library(path: Path) -> str:
|
||
if DocxDocument is None:
|
||
return ""
|
||
|
||
try:
|
||
document = DocxDocument(str(path))
|
||
except Exception:
|
||
return ""
|
||
|
||
lines: list[str] = []
|
||
iter_inner_content = getattr(document, "iter_inner_content", None)
|
||
if callable(iter_inner_content):
|
||
for block in iter_inner_content():
|
||
if hasattr(block, "rows"):
|
||
lines.extend(extract_python_docx_table_lines(block))
|
||
continue
|
||
text = normalize_extracted_text(getattr(block, "text", ""))
|
||
if text:
|
||
lines.append(text)
|
||
else:
|
||
lines.extend(
|
||
text
|
||
for text in (
|
||
normalize_extracted_text(paragraph.text)
|
||
for paragraph in getattr(document, "paragraphs", [])
|
||
)
|
||
if text
|
||
)
|
||
for table in getattr(document, "tables", []):
|
||
lines.extend(extract_python_docx_table_lines(table))
|
||
return "\n".join(lines).strip()
|
||
|
||
|
||
def extract_python_docx_table_lines(table: object) -> list[str]:
|
||
lines: list[str] = []
|
||
for row in getattr(table, "rows", []):
|
||
cells: list[str] = []
|
||
for cell in row.cells:
|
||
content_lines: list[str] = []
|
||
iter_inner_content = getattr(cell, "iter_inner_content", None)
|
||
if callable(iter_inner_content):
|
||
for item in iter_inner_content():
|
||
if hasattr(item, "rows"):
|
||
content_lines.extend(extract_python_docx_table_lines(item))
|
||
continue
|
||
text = normalize_extracted_text(getattr(item, "text", ""))
|
||
if text:
|
||
content_lines.append(text)
|
||
else:
|
||
fallback_text = normalize_extracted_text(getattr(cell, "text", ""))
|
||
if fallback_text:
|
||
content_lines.append(fallback_text)
|
||
|
||
cells.append(" / ".join(content_lines).strip())
|
||
if any(cells):
|
||
lines.append("| " + " | ".join(cells) + " |")
|
||
return lines
|
||
|
||
|
||
def extract_docx_paragraph_text(node: ET.Element) -> str:
|
||
parts = [item.text or "" for item in node.findall(".//w:t", DOCX_NAMESPACE)]
|
||
return "".join(parts).strip()
|
||
|
||
|
||
def extract_docx_table_lines(node: ET.Element) -> list[str]:
|
||
lines: list[str] = []
|
||
for row in node.findall("./w:tr", DOCX_NAMESPACE):
|
||
cells: list[str] = []
|
||
for cell in row.findall("./w:tc", DOCX_NAMESPACE):
|
||
paragraphs = [
|
||
extract_docx_paragraph_text(paragraph)
|
||
for paragraph in cell.findall(".//w:p", DOCX_NAMESPACE)
|
||
]
|
||
text = " ".join(part for part in paragraphs if part).strip()
|
||
cells.append(text)
|
||
if any(cells):
|
||
lines.append("| " + " | ".join(cells) + " |")
|
||
return lines
|
||
|
||
|
||
def read_pdf_text(path: Path) -> str:
|
||
raw = path.read_bytes()
|
||
chunks: list[str] = []
|
||
|
||
for stream in PDF_STREAM_RE.findall(raw):
|
||
decoded = decode_pdf_stream(stream)
|
||
if not decoded:
|
||
continue
|
||
text = extract_text_from_pdf_stream(decoded)
|
||
if text:
|
||
chunks.append(text)
|
||
|
||
result = "\n".join(part.strip() for part in chunks if part.strip()).strip()
|
||
if result:
|
||
return result
|
||
|
||
fallback = extract_printable_pdf_text(raw)
|
||
if fallback:
|
||
return fallback
|
||
|
||
return "> ⚠️ 待确认：PDF 未提取到可用文本，可能是扫描件、图片型 PDF 或使用了不受当前解析器支持的编码。"
|
||
|
||
|
||
def decode_pdf_stream(stream: bytes) -> bytes:
|
||
candidate = stream.strip(b"\r\n")
|
||
if not candidate:
|
||
return b""
|
||
for decoder in (zlib.decompress, lambda data: data):
|
||
try:
|
||
return decoder(candidate)
|
||
except Exception:
|
||
continue
|
||
return b""
|
||
|
||
|
||
def extract_text_from_pdf_stream(stream: bytes) -> str:
|
||
pieces: list[str] = []
|
||
for block in PDF_TEXT_BLOCK_RE.findall(stream):
|
||
array_matches = PDF_ARRAY_TEXT_RE.findall(block)
|
||
for array_content in array_matches:
|
||
pieces.extend(decode_pdf_string(token) for token in PDF_INLINE_STRING_RE.findall(array_content))
|
||
|
||
simple_matches = PDF_SIMPLE_TEXT_RE.findall(block)
|
||
for token in simple_matches:
|
||
inline_match = PDF_INLINE_STRING_RE.search(token)
|
||
if inline_match:
|
||
pieces.append(decode_pdf_string(inline_match.group(0)))
|
||
|
||
cleaned = [normalize_extracted_text(piece) for piece in pieces]
|
||
return "\n".join(item for item in cleaned if item)
|
||
|
||
|
||
def extract_printable_pdf_text(raw: bytes) -> str:
|
||
printable_chunks = re.findall(rb"[\x20-\x7e]{6,}", raw)
|
||
text = "\n".join(
|
||
normalize_extracted_text(chunk.decode("latin-1", errors="ignore"))
|
||
for chunk in printable_chunks
|
||
)
|
||
lines = [line for line in text.splitlines() if len(line.strip()) >= 6]
|
||
return "\n".join(lines[:400]).strip()
|
||
|
||
|
||
def decode_pdf_string(token: bytes) -> str:
|
||
if token.startswith(b"(") and token.endswith(b")"):
|
||
token = token[1:-1]
|
||
token = re.sub(rb"\\([()\\])", rb"\1", token)
|
||
token = token.replace(b"\\n", b"\n").replace(b"\\r", b"\r").replace(b"\\t", b"\t")
|
||
token = re.sub(
|
||
rb"\\([0-7]{1,3})",
|
||
lambda match: bytes([int(match.group(1), 8)]),
|
||
token,
|
||
)
|
||
return token.decode("utf-8", errors="ignore") or token.decode("latin-1", errors="ignore")
|
||
|
||
|
||
def normalize_extracted_text(text: str) -> str:
|
||
text = text.replace("\x00", " ")
|
||
text = re.sub(r"[ \t]+", " ", text)
|
||
return text.strip()
|
||
|
||
|
||
def normalize_topic_key(name: str) -> str:
|
||
normalized = name.lower()
|
||
for token in TOPIC_NOISE_TOKENS:
|
||
normalized = normalized.replace(token.lower(), "")
|
||
normalized = re.sub(r"[_\-\s]+", "", normalized)
|
||
normalized = re.sub(r"v\d+(?:\.\d+)*", "", normalized)
|
||
normalized = re.sub(r"版本\d+(?:\.\d+)*", "", normalized)
|
||
normalized = re.sub(r"[()（）【】\[\]]", "", normalized)
|
||
return normalized.strip()
|
||
|
||
|
||
def is_relative_to(path: Path, parent: Path) -> bool:
|
||
try:
|
||
path.resolve().relative_to(parent.resolve())
|
||
return True
|
||
except ValueError:
|
||
return False
|
||
|
||
|
||
def is_requirement_markdown(path: Path) -> bool:
|
||
return is_relative_to(path, REQUIREMENTS_DIR) and path.suffix.lower() == ".md"
|
||
|
||
|
||
def iter_requirement_documents() -> list[Path]:
|
||
result: list[Path] = []
|
||
for directory in (REQUIREMENTS_DIR, RAW_REQUIREMENTS_DIR):
|
||
if not directory.exists():
|
||
continue
|
||
for item in sorted(directory.rglob("*")):
|
||
if not item.is_file():
|
||
continue
|
||
if item.suffix.lower() not in SUPPORTED_REQUIREMENT_SUFFIXES:
|
||
continue
|
||
if item.stat().st_size == 0:
|
||
continue
|
||
if item.name.startswith("~$"): # 跳过 Office 临时锁定文件
|
||
continue
|
||
result.append(item)
|
||
return result
|
||
|
||
|
||
def prefer_maintained_requirement_documents(paths: list[Path]) -> list[Path]:
|
||
maintained_keys = {
|
||
normalize_topic_key(path.stem)
|
||
for path in paths
|
||
if is_requirement_markdown(path)
|
||
}
|
||
result: list[Path] = []
|
||
for path in paths:
|
||
if (
|
||
is_relative_to(path, RAW_REQUIREMENTS_DIR)
|
||
and normalize_topic_key(path.stem) in maintained_keys
|
||
):
|
||
continue
|
||
result.append(path)
|
||
return result
|
||
|
||
|
||
def select_technical_solution_files(requirement_path: Path) -> list[Path]:
|
||
if not TECHNICAL_SOLUTIONS_DIR.exists():
|
||
return []
|
||
|
||
requirement_key = normalize_topic_key(requirement_path.stem)
|
||
requirement_text = safe_read_text(requirement_path)
|
||
requirement_tokens = to_ngrams(f"{requirement_path.stem}\n{requirement_text}")
|
||
matches: list[tuple[float, Path]] = []
|
||
|
||
for candidate in sorted(TECHNICAL_SOLUTIONS_DIR.rglob("*")):
|
||
if not candidate.is_file():
|
||
continue
|
||
if candidate.suffix.lower() not in SUPPORTED_REQUIREMENT_SUFFIXES:
|
||
continue
|
||
candidate_key = normalize_topic_key(candidate.stem)
|
||
filename_score = 1.0 if requirement_key and candidate_key == requirement_key else 0.0
|
||
candidate_text = safe_read_text(candidate)
|
||
similarity_score = jaccard_similarity(
|
||
requirement_tokens,
|
||
to_ngrams(f"{candidate.stem}\n{candidate_text}"),
|
||
)
|
||
score = max(filename_score, similarity_score)
|
||
if score < 0.08:
|
||
continue
|
||
matches.append((score, candidate))
|
||
|
||
matches.sort(key=lambda item: item[0], reverse=True)
|
||
return [path for _, path in matches[:3]]
|
||
|
||
|
||
def write_normalized_document(source_path: Path, target_path: Path, document_role: str) -> Path:
|
||
body = safe_read_text(source_path).strip()
|
||
lines = [
|
||
f"# {source_path.stem}",
|
||
"",
|
||
f"> 文档角色：{document_role}",
|
||
f"> 原始来源：`{to_repo_relative(source_path)}`",
|
||
"",
|
||
]
|
||
if body:
|
||
lines.append(body)
|
||
else:
|
||
lines.append("> ⚠️ 待确认：文档解析结果为空，请人工检查原始文件内容或格式。")
|
||
target_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
|
||
return target_path
|
||
|
||
|
||
def extract_in_scope_requirement_text(text: str) -> str:
|
||
lines = text.splitlines()
|
||
result: list[str] = []
|
||
skip_level: int | None = None
|
||
|
||
for raw_line in lines:
|
||
line = raw_line.rstrip()
|
||
stripped = line.strip()
|
||
if stripped.startswith("#"):
|
||
level = len(stripped) - len(stripped.lstrip("#"))
|
||
heading_text = stripped[level:].strip().lower()
|
||
|
||
if skip_level is not None and level <= skip_level:
|
||
skip_level = None
|
||
|
||
if any(keyword in heading_text for keyword in OUT_OF_SCOPE_HEADING_KEYWORDS):
|
||
skip_level = level
|
||
continue
|
||
|
||
if skip_level is not None:
|
||
continue
|
||
result.append(line)
|
||
|
||
return "\n".join(result)
|
||
|
||
|
||
def to_repo_relative(path: Path) -> str:
|
||
try:
|
||
return str(path.resolve().relative_to(REPO_ROOT))
|
||
except ValueError:
|
||
return str(path.resolve())
|
||
|
||
|
||
def normalize_text(text: str) -> str:
|
||
return re.sub(r"\s+", "", text).lower()
|
||
|
||
|
||
def to_ngrams(text: str, n: int = 2) -> set[str]:
|
||
normalized = normalize_text(text)
|
||
if not normalized:
|
||
return set()
|
||
if len(normalized) < n:
|
||
return {normalized}
|
||
return {normalized[i : i + n] for i in range(len(normalized) - n + 1)}
|
||
|
||
|
||
def jaccard_similarity(left: set[str], right: set[str]) -> float:
|
||
if not left or not right:
|
||
return 0.0
|
||
union = left | right
|
||
if not union:
|
||
return 0.0
|
||
return len(left & right) / len(union)
|
||
|
||
|
||
def discover_other_requirements(requirement_path: Path) -> list[Path]:
|
||
target = requirement_path.resolve()
|
||
result: list[Path] = []
|
||
for item in iter_requirement_documents():
|
||
if item.resolve() == target:
|
||
continue
|
||
if (
|
||
is_requirement_markdown(item)
|
||
and is_relative_to(requirement_path, RAW_REQUIREMENTS_DIR)
|
||
and normalize_topic_key(item.stem) == normalize_topic_key(requirement_path.stem)
|
||
):
|
||
continue
|
||
result.append(item)
|
||
return prefer_maintained_requirement_documents(result)
|
||
|
||
|
||
def find_related_requirements(requirement_path: Path) -> list[dict[str, Any]]:
|
||
target_content = safe_read_text(requirement_path)
|
||
target_tokens = to_ngrams(f"{requirement_path.name}\n{target_content}")
|
||
related: list[dict[str, Any]] = []
|
||
|
||
for candidate in discover_other_requirements(requirement_path):
|
||
candidate_content = safe_read_text(candidate)
|
||
score = jaccard_similarity(
|
||
target_tokens,
|
||
to_ngrams(f"{candidate.name}\n{candidate_content}"),
|
||
)
|
||
if score < RELATED_SIMILARITY_THRESHOLD:
|
||
continue
|
||
related.append(
|
||
{
|
||
"path": candidate,
|
||
"score": score,
|
||
"content": candidate_content,
|
||
}
|
||
)
|
||
|
||
related.sort(key=lambda item: item["score"], reverse=True)
|
||
return related[:MAX_RELATED_REQUIREMENTS]
|
||
|
||
|
||
def select_optional_terminology_files(
|
||
requirement_path: Path,
|
||
related_requirements: list[dict[str, Any]],
|
||
technical_solution_files: list[Path],
|
||
) -> list[dict[str, Any]]:
|
||
texts = [
|
||
extract_in_scope_requirement_text(safe_read_text(requirement_path)),
|
||
safe_read_text(PROJECT_PROFILE_FILE),
|
||
requirement_path.name,
|
||
]
|
||
texts.extend(safe_read_text(path) for path in technical_solution_files)
|
||
combined_text = "\n".join(texts)
|
||
|
||
selected: list[dict[str, Any]] = []
|
||
for rule in OPTIONAL_TERMINOLOGY_RULES:
|
||
matched_keywords = [
|
||
keyword for keyword in rule["keywords"] if keyword in combined_text
|
||
]
|
||
if not matched_keywords:
|
||
continue
|
||
selected.append(
|
||
{
|
||
"name": rule["name"],
|
||
"path": Path(rule["path"]),
|
||
"matched_keywords": matched_keywords[:10],
|
||
}
|
||
)
|
||
return selected
|
||
|
||
|
||
def extract_rule_lines(text: str) -> list[str]:
|
||
result: list[str] = []
|
||
seen: set[str] = set()
|
||
|
||
for raw_line in text.splitlines():
|
||
line = raw_line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
|
||
line = re.sub(r"^\s*[-*]\s*", "", line)
|
||
line = re.sub(r"^\s*\d+[.)]\s*", "", line)
|
||
line = line.strip()
|
||
if len(line) < 8:
|
||
continue
|
||
if not any(keyword in line for keyword in NORMATIVE_KEYWORDS):
|
||
continue
|
||
if line in seen:
|
||
continue
|
||
|
||
seen.add(line)
|
||
result.append(line)
|
||
return result
|
||
|
||
|
||
def classify_polarity(rule_line: str) -> str:
|
||
if any(keyword in rule_line for keyword in NEGATIVE_POLARITY_KEYWORDS):
|
||
return "negative"
|
||
if any(keyword in rule_line for keyword in POSITIVE_POLARITY_KEYWORDS):
|
||
return "positive"
|
||
return "neutral"
|
||
|
||
|
||
def extract_numeric_tokens(rule_line: str) -> set[str]:
|
||
return set(NUMERIC_TOKEN_RE.findall(rule_line))
|
||
|
||
|
||
def build_conflict_candidates(
|
||
requirement_path: Path,
|
||
related_requirements: list[dict[str, Any]],
|
||
) -> list[dict[str, Any]]:
|
||
target_rules = extract_rule_lines(safe_read_text(requirement_path))
|
||
target_rule_items = [
|
||
{
|
||
"line": line,
|
||
"polarity": classify_polarity(line),
|
||
"numbers": extract_numeric_tokens(line),
|
||
"tokens": to_ngrams(line),
|
||
}
|
||
for line in target_rules
|
||
]
|
||
|
||
conflicts: list[dict[str, Any]] = []
|
||
dedup: set[tuple[str, str, str]] = set()
|
||
|
||
for related in related_requirements:
|
||
related_rules = extract_rule_lines(related["content"])
|
||
related_rule_items = [
|
||
{
|
||
"line": line,
|
||
"polarity": classify_polarity(line),
|
||
"numbers": extract_numeric_tokens(line),
|
||
"tokens": to_ngrams(line),
|
||
}
|
||
for line in related_rules
|
||
]
|
||
|
||
for target_item in target_rule_items:
|
||
for history_item in related_rule_items:
|
||
similarity = jaccard_similarity(target_item["tokens"], history_item["tokens"])
|
||
if similarity < CONFLICT_SIMILARITY_THRESHOLD:
|
||
continue
|
||
|
||
conflict_type = ""
|
||
suggestion = ""
|
||
opposite_polarity = {
|
||
target_item["polarity"],
|
||
history_item["polarity"],
|
||
} == {"positive", "negative"}
|
||
|
||
if opposite_polarity:
|
||
conflict_type = "规则方向冲突"
|
||
suggestion = "建议补充版本范围、生效条件和优先级，明确保留哪条规则。"
|
||
elif (
|
||
target_item["numbers"]
|
||
and history_item["numbers"]
|
||
and target_item["numbers"] != history_item["numbers"]
|
||
):
|
||
conflict_type = "数值口径冲突"
|
||
suggestion = "建议统一阈值口径，并在需求中明确变更原因与兼容策略。"
|
||
else:
|
||
continue
|
||
|
||
key = (target_item["line"], history_item["line"], conflict_type)
|
||
if key in dedup:
|
||
continue
|
||
dedup.add(key)
|
||
|
||
conflicts.append(
|
||
{
|
||
"target_rule": target_item["line"],
|
||
"history_rule": history_item["line"],
|
||
"history_path": related["path"],
|
||
"type": conflict_type,
|
||
"suggestion": suggestion,
|
||
"similarity": similarity,
|
||
}
|
||
)
|
||
|
||
conflicts.sort(key=lambda item: item["similarity"], reverse=True)
|
||
return conflicts[:20]
|
||
|
||
|
||
def escape_table_cell(text: str) -> str:
|
||
return text.replace("|", "\\|").replace("\n", " ").strip()
|
||
|
||
|
||
def write_relation_report(
|
||
requirement_path: Path,
|
||
related_requirements: list[dict[str, Any]],
|
||
technical_solution_files: list[Path],
|
||
conflicts: list[dict[str, Any]],
|
||
report_path: Path,
|
||
) -> None:
|
||
lines: list[str] = [
|
||
f"# {requirement_path.stem} 关联需求与冲突检查",
|
||
"",
|
||
"## 目标需求",
|
||
f"- `{to_repo_relative(requirement_path)}`",
|
||
"",
|
||
"## 项目画像",
|
||
f"- `{to_repo_relative(PROJECT_PROFILE_FILE)}`",
|
||
"",
|
||
"## 关联技术方案",
|
||
]
|
||
|
||
if technical_solution_files:
|
||
for path in technical_solution_files:
|
||
lines.append(f"- `{to_repo_relative(path)}`")
|
||
else:
|
||
lines.append("- 未识别到同主题技术方案文档。")
|
||
|
||
lines.extend([
|
||
"",
|
||
"## 关联需求识别",
|
||
])
|
||
|
||
if related_requirements:
|
||
lines.extend(
|
||
[
|
||
"| 序号 | 关联需求 | 相似度 |",
|
||
"| :--- | :--- | :--- |",
|
||
]
|
||
)
|
||
for index, item in enumerate(related_requirements, start=1):
|
||
lines.append(
|
||
f"| {index} | `{to_repo_relative(item['path'])}` | {item['score']:.4f} |"
|
||
)
|
||
else:
|
||
lines.append("- 未识别到相似度达到阈值的历史需求文档。")
|
||
|
||
lines.extend(["", "## 潜在冲突与修改建议"])
|
||
|
||
if conflicts:
|
||
lines.extend(
|
||
[
|
||
"| 序号 | 当前需求条目 | 历史需求条目 | 历史来源 | 冲突类型 | 修改建议 |",
|
||
"| :--- | :--- | :--- | :--- | :--- | :--- |",
|
||
]
|
||
)
|
||
for index, item in enumerate(conflicts, start=1):
|
||
lines.append(
|
||
"| "
|
||
+ " | ".join(
|
||
[
|
||
str(index),
|
||
escape_table_cell(item["target_rule"]),
|
||
escape_table_cell(item["history_rule"]),
|
||
f"`{to_repo_relative(item['history_path'])}`",
|
||
item["type"],
|
||
escape_table_cell(item["suggestion"]),
|
||
]
|
||
)
|
||
+ " |"
|
||
)
|
||
lines.extend(
|
||
[
|
||
"",
|
||
"> ⚠️ 待确认：以上条目为机器识别候选冲突，请产品/业务确认最终规则口径，并回写需求文档。",
|
||
]
|
||
)
|
||
else:
|
||
lines.append("- 暂未识别到明显冲突条目。建议在需求评审时继续人工确认。")
|
||
|
||
report_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
|
||
|
||
|
||
def write_manifest(
|
||
requirement_path: Path,
|
||
paths: dict[str, Path],
|
||
related_requirements: list[dict[str, Any]],
|
||
technical_solution_files: list[Path],
|
||
normalized_requirement_file: Path,
|
||
normalized_technical_solution_files: list[Path],
|
||
optional_terminology_files: list[dict[str, Any]],
|
||
conflict_count: int,
|
||
confirmation_gate: dict[str, Any],
|
||
) -> Path:
|
||
effective_terminology_files = [CORE_TERMINOLOGY_FILE] + [
|
||
item["path"] for item in optional_terminology_files
|
||
]
|
||
manifest = {
|
||
"requirement": str(requirement_path),
|
||
"requirement_source_file": str(requirement_path),
|
||
"requirement_input_type": requirement_path.suffix.lower().lstrip("."),
|
||
"base_name": requirement_path.stem,
|
||
"maintained_requirement_file": str(paths["maintained_requirement"]),
|
||
"normalized_requirement_file": str(normalized_requirement_file),
|
||
"technical_solution_files": [str(path) for path in technical_solution_files],
|
||
"normalized_technical_solution_files": [str(path) for path in normalized_technical_solution_files],
|
||
"analysis_file": str(paths["analysis"]),
|
||
"relation_report_file": str(paths["relation_report"]),
|
||
"test_points_file": str(paths["test_points"]),
|
||
"test_cases_file": str(paths["test_cases"]),
|
||
"excel_output_dir": str(paths["excel_dir"]),
|
||
"project_profile_file": str(PROJECT_PROFILE_FILE),
|
||
"knowledge_base_files": [str(file_path) for file_path in KNOWLEDGE_BASE_FILES],
|
||
"effective_terminology_files": [str(file_path) for file_path in effective_terminology_files],
|
||
"optional_terminology_files": [
|
||
{
|
||
"name": item["name"],
|
||
"path": str(item["path"]),
|
||
"matched_keywords": item["matched_keywords"],
|
||
}
|
||
for item in optional_terminology_files
|
||
],
|
||
"related_requirements": [
|
||
{"path": str(item["path"]), "similarity": round(item["score"], 6)}
|
||
for item in related_requirements
|
||
],
|
||
"conflict_candidates_count": conflict_count,
|
||
"confirmation_gate": confirmation_gate,
|
||
}
|
||
write_manifest_data(paths["manifest"], manifest)
|
||
return paths["manifest"]
|
||
|
||
|
||
def write_manifest_data(manifest_path: Path, manifest: dict[str, Any]) -> None:
|
||
manifest_path.write_text(
|
||
json.dumps(manifest, ensure_ascii=False, indent=2) + "\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
|
||
def count_pending_confirmation_markers(paths: list[Path]) -> int:
|
||
count = 0
|
||
for path in paths:
|
||
if not path.exists():
|
||
continue
|
||
count += safe_read_text(path).count("> ⚠️ 待确认：")
|
||
return count
|
||
|
||
|
||
def list_decision_files(base_name: str) -> list[Path]:
|
||
if not DECISIONS_DIR.exists():
|
||
return []
|
||
|
||
return sorted(
|
||
[
|
||
item
|
||
for item in DECISIONS_DIR.glob(f"{base_name}*_确认结论.md")
|
||
if item.is_file()
|
||
],
|
||
key=lambda item: (item.stat().st_mtime, item.name),
|
||
reverse=True,
|
||
)
|
||
|
||
|
||
def extract_single_line_field(content: str, label: str) -> str:
|
||
pattern = re.compile(rf"^- {re.escape(label)}:\s*(.*)$", re.MULTILINE)
|
||
match = pattern.search(content)
|
||
if not match:
|
||
return ""
|
||
return match.group(1).strip()
|
||
|
||
|
||
def clean_field_value(value: str) -> str:
|
||
value = value.strip()
|
||
value = value.strip("`")
|
||
return value.strip()
|
||
|
||
|
||
def split_field_values(value: str) -> list[str]:
|
||
if not value:
|
||
return []
|
||
normalized = (
|
||
value.replace("；", "\n")
|
||
.replace("，", "\n")
|
||
.replace(",", "\n")
|
||
.replace("、", "\n")
|
||
)
|
||
return [clean_field_value(item) for item in normalized.splitlines() if clean_field_value(item)]
|
||
|
||
|
||
def requirement_reference_aliases(requirement_path: Path) -> set[str]:
|
||
aliases = {
|
||
clean_field_value(str(requirement_path.resolve())).lower(),
|
||
clean_field_value(to_repo_relative(requirement_path)).lower(),
|
||
clean_field_value(requirement_path.name).lower(),
|
||
clean_field_value(requirement_path.stem).lower(),
|
||
normalize_topic_key(requirement_path.stem),
|
||
}
|
||
return {item for item in aliases if item}
|
||
|
||
|
||
def requirement_reference_matches(reference: str, requirement_path: Path) -> bool:
|
||
normalized_reference = clean_field_value(reference).lower()
|
||
if not normalized_reference:
|
||
return False
|
||
|
||
stem = Path(normalized_reference).stem if "/" in normalized_reference or "." in normalized_reference else normalized_reference
|
||
aliases = requirement_reference_aliases(requirement_path)
|
||
return normalized_reference in aliases or normalize_topic_key(stem) in aliases
|
||
|
||
|
||
def resolve_requirement_reference(reference: str) -> Path:
|
||
value = clean_field_value(reference)
|
||
if not value:
|
||
raise ValueError("需求引用为空。")
|
||
|
||
candidate_path = resolve_requirement_path(value)
|
||
if candidate_path.exists():
|
||
validate_requirement_file(candidate_path)
|
||
return candidate_path
|
||
|
||
matches = [
|
||
path
|
||
for path in iter_requirement_documents()
|
||
if requirement_reference_matches(value, path)
|
||
]
|
||
if not matches:
|
||
raise FileNotFoundError(f"未找到需求引用：{value}")
|
||
if len(matches) > 1:
|
||
options = "\n".join(f"- {to_repo_relative(path)}" for path in matches)
|
||
raise ValueError(f"需求引用存在歧义：{value}\n候选项：\n{options}")
|
||
return matches[0]
|
||
|
||
|
||
def parse_decision_file(decision_file: Path) -> dict[str, Any]:
|
||
content = safe_read_text(decision_file)
|
||
status_match = DECISION_FILE_RE.search(content)
|
||
raw_status = status_match.group(1) if status_match else ""
|
||
status_map = {
|
||
"已确认": "confirmed",
|
||
"待确认": "pending",
|
||
"已驳回": "rejected",
|
||
"已拒绝": "rejected",
|
||
}
|
||
export_match = DECISION_EXPORT_RE.search(content)
|
||
export_policy = export_match.group(1) if export_match else None
|
||
relation_match = DECISION_RELATION_TYPE_RE.search(content)
|
||
relation_type = relation_match.group(1) if relation_match else None
|
||
|
||
return {
|
||
"file": decision_file,
|
||
"content": content,
|
||
"status": status_map.get(raw_status, "unknown"),
|
||
"raw_status": raw_status or None,
|
||
"allow_export_before_confirmation": export_policy == "允许" if export_policy else None,
|
||
"current_requirement_refs": split_field_values(extract_single_line_field(content, "当前需求")),
|
||
"related_requirement_refs": split_field_values(extract_single_line_field(content, "关联历史需求")),
|
||
"rerun_requirement_refs": split_field_values(extract_single_line_field(content, "需要重跑的需求列表")),
|
||
"relation_type": relation_type,
|
||
"rewrite_current": clean_field_value(extract_single_line_field(content, "是否需要回写当前需求")) == "是",
|
||
"rewrite_history": clean_field_value(extract_single_line_field(content, "是否需要回写历史需求")) == "是",
|
||
"effective_scope": clean_field_value(extract_single_line_field(content, "生效范围")),
|
||
"invalid_scope": clean_field_value(extract_single_line_field(content, "失效范围")),
|
||
"impact_modules": clean_field_value(extract_single_line_field(content, "影响模块")),
|
||
"impact_roles": clean_field_value(extract_single_line_field(content, "影响角色")),
|
||
"impact_data_scope": clean_field_value(extract_single_line_field(content, "影响接口或数据口径")),
|
||
"version_boundary": clean_field_value(extract_single_line_field(content, "需要补充的版本边界")),
|
||
"inheritance_note": clean_field_value(extract_single_line_field(content, "需要补充的替代/继承说明")),
|
||
"compatibility_strategy": clean_field_value(extract_single_line_field(content, "需要补充的兼容策略")),
|
||
"rerun_order": clean_field_value(extract_single_line_field(content, "重跑顺序建议")),
|
||
}
|
||
|
||
|
||
def inspect_decision_file(decision_file: Path) -> dict[str, Any]:
|
||
parsed = parse_decision_file(decision_file)
|
||
return {
|
||
"file": parsed["file"],
|
||
"status": parsed["status"],
|
||
"raw_status": parsed["raw_status"],
|
||
"allow_export_before_confirmation": parsed["allow_export_before_confirmation"],
|
||
}
|
||
|
||
|
||
def find_relevant_decision_files(requirement_path: Path) -> list[Path]:
|
||
explicit = list_decision_files(requirement_path.stem)
|
||
if not DECISIONS_DIR.exists():
|
||
return explicit
|
||
|
||
candidates: list[Path] = []
|
||
seen: set[Path] = set()
|
||
for item in explicit + sorted(DECISIONS_DIR.glob("*确认结论.md"), key=lambda path: path.name):
|
||
if not item.is_file() or item in seen:
|
||
continue
|
||
seen.add(item)
|
||
parsed = parse_decision_file(item)
|
||
references = parsed["current_requirement_refs"] + parsed["related_requirement_refs"]
|
||
if item.stem.startswith(requirement_path.stem) or any(
|
||
requirement_reference_matches(reference, requirement_path)
|
||
for reference in references
|
||
):
|
||
candidates.append(item)
|
||
|
||
return sorted(
|
||
candidates,
|
||
key=lambda item: (item.stat().st_mtime, item.name),
|
||
reverse=True,
|
||
)
|
||
|
||
|
||
def build_confirmation_gate(
|
||
requirement_path: Path,
|
||
base_name: str,
|
||
related_requirements: list[dict[str, Any]],
|
||
conflicts: list[dict[str, Any]],
|
||
normalized_requirement_file: Path,
|
||
normalized_technical_solution_files: list[Path],
|
||
) -> dict[str, Any]:
|
||
pending_markers_count = count_pending_confirmation_markers(
|
||
[normalized_requirement_file, *normalized_technical_solution_files]
|
||
)
|
||
reasons: list[str] = []
|
||
if related_requirements:
|
||
reasons.append(
|
||
f"识别到 {len(related_requirements)} 个历史相似需求，需确认与旧需求的关系类型、生效范围和是否需要回写历史需求。"
|
||
)
|
||
if conflicts:
|
||
reasons.append(
|
||
f"识别到 {len(conflicts)} 个潜在冲突候选，需确认最终规则口径后再继续校验或导出。"
|
||
)
|
||
if pending_markers_count:
|
||
reasons.append(
|
||
f"标准化输入中包含 {pending_markers_count} 处待确认项，需先补充或确认关键前提。"
|
||
)
|
||
|
||
decision_files = find_relevant_decision_files(requirement_path)
|
||
decision_info = inspect_decision_file(decision_files[0]) if decision_files else None
|
||
required = bool(reasons)
|
||
|
||
if not required:
|
||
status = "not_required"
|
||
elif decision_info is None:
|
||
status = "missing"
|
||
else:
|
||
status = decision_info["status"]
|
||
|
||
gate = {
|
||
"required": required,
|
||
"reasons": reasons,
|
||
"pending_markers_count": pending_markers_count,
|
||
"decision_file": str(decision_info["file"]) if decision_info else None,
|
||
"decision_status": status,
|
||
"decision_status_label": decision_info["raw_status"] if decision_info else None,
|
||
"allow_export_before_confirmation": (
|
||
decision_info["allow_export_before_confirmation"] if decision_info else None
|
||
),
|
||
"candidate_decision_files": [str(path) for path in decision_files],
|
||
"suggested_decision_file": str(DECISIONS_DIR / f"{base_name}_确认结论.md"),
|
||
}
|
||
return gate
|
||
|
||
|
||
def load_manifest_data(manifest_path: Path) -> dict[str, Any]:
|
||
if not manifest_path.exists():
|
||
raise FileNotFoundError(f"缺少清单文件：{manifest_path}。请先执行 prepare。")
|
||
return json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def refresh_manifest_confirmation_gate(paths: dict[str, Path]) -> dict[str, Any]:
|
||
manifest = load_manifest_data(paths["manifest"])
|
||
base_name = manifest.get("base_name", paths["test_cases"].stem.replace("_测试用例", ""))
|
||
requirement_path = Path(manifest.get("requirement_source_file", manifest.get("requirement", "")))
|
||
related_requirements = manifest.get("related_requirements", [])
|
||
conflicts = [
|
||
{"placeholder": True}
|
||
for _ in range(int(manifest.get("conflict_candidates_count", 0) or 0))
|
||
]
|
||
normalized_requirement_file = Path(
|
||
manifest.get("normalized_requirement_file", paths["normalized_input_dir"] / "requirement.md")
|
||
)
|
||
normalized_technical_solution_files = [
|
||
Path(item) for item in manifest.get("normalized_technical_solution_files", [])
|
||
]
|
||
confirmation_gate = build_confirmation_gate(
|
||
requirement_path=requirement_path,
|
||
base_name=base_name,
|
||
related_requirements=related_requirements,
|
||
conflicts=conflicts,
|
||
normalized_requirement_file=normalized_requirement_file,
|
||
normalized_technical_solution_files=normalized_technical_solution_files,
|
||
)
|
||
manifest["confirmation_gate"] = confirmation_gate
|
||
write_manifest_data(paths["manifest"], manifest)
|
||
return manifest
|
||
|
||
|
||
def ensure_confirmation_resolved(paths: dict[str, Path], command_name: str) -> dict[str, Any]:
|
||
manifest = refresh_manifest_confirmation_gate(paths)
|
||
gate = manifest.get("confirmation_gate", {})
|
||
if not gate.get("required"):
|
||
return manifest
|
||
|
||
if gate.get("decision_status") == "confirmed":
|
||
return manifest
|
||
|
||
reasons = gate.get("reasons", [])
|
||
reason_block = "\n".join(f"- {item}" for item in reasons) if reasons else "- 待补充确认原因"
|
||
decision_file = gate.get("decision_file") or gate.get("suggested_decision_file")
|
||
raise ValueError(
|
||
f"{command_name} 已阻断：当前需求仍需人工确认。\n"
|
||
f"建议确认单：{decision_file}\n"
|
||
"阻断原因：\n"
|
||
f"{reason_block}\n"
|
||
"请在确认单中明确写出 `确认状态：已确认` 后再继续。"
|
||
)
|
||
|
||
|
||
def build_maintenance_note_path(decision_file: Path) -> Path:
|
||
return DECISIONS_APPLIED_DIR / f"{decision_file.stem}_维护说明.md"
|
||
|
||
|
||
def render_maintenance_note(
|
||
decision: dict[str, Any],
|
||
current_requirement: Path,
|
||
related_requirements: list[Path],
|
||
rerun_targets: list[Path],
|
||
) -> str:
|
||
lines = [
|
||
f"# {current_requirement.stem} 确认结论维护说明",
|
||
"",
|
||
f"- 来源确认单：`{to_repo_relative(Path(decision['file']))}`",
|
||
f"- 当前需求：`{to_repo_relative(current_requirement)}`",
|
||
f"- 关系类型：{decision.get('relation_type') or '未填写'}",
|
||
f"- 是否回写当前需求：{'是' if decision.get('rewrite_current') else '否'}",
|
||
f"- 是否回写历史需求：{'是' if decision.get('rewrite_history') else '否'}",
|
||
"",
|
||
"## 关联历史需求",
|
||
]
|
||
if related_requirements:
|
||
for path in related_requirements:
|
||
lines.append(f"- `{to_repo_relative(path)}`")
|
||
else:
|
||
lines.append("- 未填写")
|
||
|
||
lines.extend(
|
||
[
|
||
"",
|
||
"## 生效边界",
|
||
f"- 生效范围：{decision.get('effective_scope') or '未填写'}",
|
||
f"- 失效范围：{decision.get('invalid_scope') or '未填写'}",
|
||
f"- 影响模块：{decision.get('impact_modules') or '未填写'}",
|
||
f"- 影响角色：{decision.get('impact_roles') or '未填写'}",
|
||
f"- 影响接口或数据口径：{decision.get('impact_data_scope') or '未填写'}",
|
||
"",
|
||
"## 维护说明",
|
||
f"- 版本边界补充：{decision.get('version_boundary') or '未填写'}",
|
||
f"- 替代/继承说明：{decision.get('inheritance_note') or '未填写'}",
|
||
f"- 兼容策略：{decision.get('compatibility_strategy') or '未填写'}",
|
||
"",
|
||
"## 重跑计划",
|
||
]
|
||
)
|
||
for path in rerun_targets:
|
||
lines.append(f"- `{to_repo_relative(path)}`")
|
||
if not rerun_targets:
|
||
lines.append("- 未填写")
|
||
if decision.get("rerun_order"):
|
||
lines.append(f"- 重跑顺序建议：{decision['rerun_order']}")
|
||
|
||
return "\n".join(lines).rstrip() + "\n"
|
||
|
||
|
||
def extract_normalized_requirement_body(text: str) -> str:
|
||
lines = text.splitlines()
|
||
result: list[str] = []
|
||
started = False
|
||
|
||
for index, line in enumerate(lines):
|
||
if index == 0 and line.startswith("# "):
|
||
continue
|
||
if not started:
|
||
if line.startswith("> "):
|
||
continue
|
||
if not line.strip():
|
||
continue
|
||
started = True
|
||
result.append(line)
|
||
|
||
return "\n".join(result).strip()
|
||
|
||
|
||
def find_applied_maintenance_notes(requirement_path: Path) -> list[Path]:
|
||
if not DECISIONS_APPLIED_DIR.exists():
|
||
return []
|
||
|
||
matches: list[Path] = []
|
||
for item in sorted(DECISIONS_APPLIED_DIR.glob("*_维护说明.md")):
|
||
if not item.is_file():
|
||
continue
|
||
content = safe_read_text(item)
|
||
references = [
|
||
to_repo_relative(requirement_path),
|
||
str(requirement_path.resolve()),
|
||
requirement_path.name,
|
||
requirement_path.stem,
|
||
]
|
||
if any(reference and reference in content for reference in references):
|
||
matches.append(item)
|
||
return sorted(matches, key=lambda item: (item.stat().st_mtime, item.name), reverse=True)
|
||
|
||
|
||
def render_maintained_requirement(
|
||
requirement_path: Path,
|
||
manifest: dict[str, Any],
|
||
normalized_requirement_file: Path,
|
||
) -> str:
|
||
maintained_requirement_path = Path(
|
||
manifest.get(
|
||
"maintained_requirement_file",
|
||
build_paths(manifest.get("base_name", requirement_path.stem))["maintained_requirement"],
|
||
)
|
||
)
|
||
normalized_text = safe_read_text(normalized_requirement_file)
|
||
body = extract_normalized_requirement_body(normalized_text)
|
||
maintenance_notes = find_applied_maintenance_notes(requirement_path)
|
||
technical_solutions = [Path(item) for item in manifest.get("technical_solution_files", [])]
|
||
|
||
lines = [
|
||
f"# {manifest.get('base_name', requirement_path.stem)}",
|
||
"",
|
||
"> 文档角色：正式维护版需求",
|
||
f"> 原始来源：`{to_repo_relative(requirement_path)}`",
|
||
f"> 标准化来源：`{to_repo_relative(normalized_requirement_file)}`",
|
||
f"> 输入类型：`{manifest.get('requirement_input_type', requirement_path.suffix.lstrip('.'))}`",
|
||
"",
|
||
"## 维护信息",
|
||
f"- 当前维护文件：`{to_repo_relative(maintained_requirement_path)}`",
|
||
]
|
||
if maintenance_notes:
|
||
for index, note in enumerate(maintenance_notes, start=1):
|
||
label = "最近维护说明" if index == 1 else "历史维护说明"
|
||
lines.append(f"- {label}：`{to_repo_relative(note)}`")
|
||
else:
|
||
lines.append("- 最近维护说明：暂无")
|
||
|
||
lines.extend(["", "## 技术方案来源"])
|
||
if technical_solutions:
|
||
for path in technical_solutions:
|
||
lines.append(f"- `{to_repo_relative(path)}`")
|
||
else:
|
||
lines.append("- 未关联技术方案")
|
||
|
||
lines.extend(["", "## 正式维护内容", ""])
|
||
if body:
|
||
lines.append(body)
|
||
else:
|
||
lines.append("> ⚠️ 待确认：标准化需求正文为空，请先检查原始需求文档解析结果。")
|
||
|
||
return "\n".join(lines).rstrip() + "\n"
|
||
|
||
|
||
def sync_maintained_requirement(requirement_path: Path) -> Path:
|
||
paths = build_paths(requirement_path.stem)
|
||
manifest = load_manifest_data(paths["manifest"])
|
||
maintained_requirement_path = Path(
|
||
manifest.get("maintained_requirement_file", paths["maintained_requirement"])
|
||
)
|
||
maintained_requirement_path.parent.mkdir(parents=True, exist_ok=True)
|
||
normalized_requirement_file = Path(manifest["normalized_requirement_file"])
|
||
content = render_maintained_requirement(
|
||
requirement_path=requirement_path,
|
||
manifest=manifest,
|
||
normalized_requirement_file=normalized_requirement_file,
|
||
)
|
||
maintained_requirement_path.write_text(content, encoding="utf-8")
|
||
manifest["maintained_requirement_file"] = str(maintained_requirement_path)
|
||
write_manifest_data(paths["manifest"], manifest)
|
||
return maintained_requirement_path
|
||
|
||
|
||
def write_maintenance_note(
|
||
decision: dict[str, Any],
|
||
current_requirement: Path,
|
||
related_requirements: list[Path],
|
||
rerun_targets: list[Path],
|
||
) -> Path:
|
||
DECISIONS_APPLIED_DIR.mkdir(parents=True, exist_ok=True)
|
||
note_path = build_maintenance_note_path(Path(decision["file"]))
|
||
note_path.write_text(
|
||
render_maintenance_note(
|
||
decision=decision,
|
||
current_requirement=current_requirement,
|
||
related_requirements=related_requirements,
|
||
rerun_targets=rerun_targets,
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
return note_path
|
||
|
||
|
||
def build_rerun_targets(decision: dict[str, Any], requirement_path: Path) -> list[Path]:
|
||
targets: list[Path] = [requirement_path]
|
||
for reference in decision.get("rerun_requirement_refs", []):
|
||
resolved = resolve_requirement_reference(reference)
|
||
if resolved not in targets:
|
||
targets.append(resolved)
|
||
return targets
|
||
|
||
|
||
def run_pipeline_for_requirement(requirement_path: Path) -> None:
|
||
relative_requirement = to_repo_relative(requirement_path)
|
||
commands = [
|
||
["prepare", "--requirement", relative_requirement],
|
||
["verify", "--requirement", relative_requirement],
|
||
["export", "--requirement", relative_requirement],
|
||
]
|
||
for command in commands:
|
||
subprocess.run(
|
||
[sys.executable, str(REPO_ROOT / "scripts" / "case_pipeline.py"), *command],
|
||
check=True,
|
||
)
|
||
sync_maintained_requirement(requirement_path)
|
||
|
||
|
||
def collect_snapshot_sources(
|
||
paths: dict[str, Path],
|
||
excel_path: Path,
|
||
include_manifest: bool = True,
|
||
) -> list[tuple[Path, Path]]:
|
||
sources = [
|
||
(paths["analysis"], Path(paths["analysis"].name)),
|
||
(paths["relation_report"], Path(paths["relation_report"].name)),
|
||
(paths["test_points"], Path(paths["test_points"].name)),
|
||
(paths["test_cases"], Path(paths["test_cases"].name)),
|
||
(excel_path, Path(excel_path.name)),
|
||
]
|
||
if include_manifest:
|
||
sources.append((paths["manifest"], Path(paths["manifest"].name)))
|
||
|
||
normalized_root = paths["normalized_input_dir"]
|
||
if normalized_root.exists():
|
||
for item in sorted(normalized_root.rglob("*")):
|
||
if item.is_file():
|
||
relative = item.relative_to(normalized_root)
|
||
sources.append((item, Path("normalized_inputs") / relative))
|
||
|
||
return sources
|
||
|
||
|
||
def find_latest_snapshot(version_root: Path) -> tuple[int, Path] | None:
|
||
if not version_root.exists():
|
||
return None
|
||
|
||
versions: list[tuple[int, Path]] = []
|
||
for child in version_root.iterdir():
|
||
if not child.is_dir():
|
||
continue
|
||
match = VERSION_DIR_RE.match(child.name)
|
||
if not match:
|
||
continue
|
||
versions.append((int(match.group(1)), child))
|
||
|
||
if not versions:
|
||
return None
|
||
return max(versions, key=lambda item: item[0])
|
||
|
||
|
||
def list_snapshot_versions(version_root: Path) -> list[tuple[int, Path]]:
|
||
if not version_root.exists():
|
||
return []
|
||
|
||
versions: list[tuple[int, Path]] = []
|
||
for child in version_root.iterdir():
|
||
if not child.is_dir():
|
||
continue
|
||
match = VERSION_DIR_RE.match(child.name)
|
||
if not match:
|
||
continue
|
||
versions.append((int(match.group(1)), child))
|
||
|
||
return sorted(versions, key=lambda item: item[0])
|
||
|
||
|
||
def snapshot_matches(snapshot_dir: Path, sources: list[tuple[Path, Path]]) -> bool:
|
||
for source, relative in sources:
|
||
target = snapshot_dir / relative
|
||
if not target.exists() or not target.is_file():
|
||
return False
|
||
if not files_match(source, target):
|
||
return False
|
||
return True
|
||
|
||
|
||
def files_match(source: Path, target: Path) -> bool:
|
||
if source.suffix.lower() == ".xlsx" and target.suffix.lower() == ".xlsx":
|
||
return excel_files_match(source, target)
|
||
return filecmp.cmp(source, target, shallow=False)
|
||
|
||
|
||
def excel_files_match(source: Path, target: Path) -> bool:
|
||
source_workbook = load_workbook(source, read_only=True, data_only=True)
|
||
target_workbook = load_workbook(target, read_only=True, data_only=True)
|
||
|
||
try:
|
||
if source_workbook.sheetnames != target_workbook.sheetnames:
|
||
return False
|
||
|
||
for sheet_name in source_workbook.sheetnames:
|
||
source_sheet = source_workbook[sheet_name]
|
||
target_sheet = target_workbook[sheet_name]
|
||
if source_sheet.max_row != target_sheet.max_row or source_sheet.max_column != target_sheet.max_column:
|
||
return False
|
||
|
||
for source_row, target_row in zip(
|
||
source_sheet.iter_rows(values_only=True),
|
||
target_sheet.iter_rows(values_only=True),
|
||
):
|
||
if tuple(source_row) != tuple(target_row):
|
||
return False
|
||
finally:
|
||
source_workbook.close()
|
||
target_workbook.close()
|
||
|
||
return True
|
||
|
||
|
||
def is_legacy_excel_export(file_path: Path, case_stem: str) -> bool:
|
||
match = LEGACY_EXCEL_EXPORT_RE.match(file_path.name)
|
||
return bool(match and match.group("stem") == case_stem)
|
||
|
||
|
||
def find_legacy_excel_exports(paths: dict[str, Path]) -> list[Path]:
|
||
case_stem = paths["test_cases"].stem
|
||
excel_dir = paths["excel_dir"]
|
||
if not excel_dir.exists():
|
||
return []
|
||
|
||
return sorted(
|
||
[
|
||
item
|
||
for item in excel_dir.iterdir()
|
||
if item.is_file() and is_legacy_excel_export(item, case_stem)
|
||
],
|
||
key=lambda item: item.name,
|
||
)
|
||
|
||
|
||
def existing_snapshot_excel_files(version_root: Path) -> list[Path]:
|
||
if not version_root.exists():
|
||
return []
|
||
|
||
files: list[Path] = []
|
||
for child in sorted(version_root.iterdir()):
|
||
if not child.is_dir() or not VERSION_DIR_RE.match(child.name):
|
||
continue
|
||
for item in child.glob("*.xlsx"):
|
||
files.append(item)
|
||
return files
|
||
|
||
|
||
def extract_base_name_from_paths(paths: dict[str, Path]) -> str:
|
||
stem = paths["test_cases"].stem
|
||
if stem.endswith("_测试用例"):
|
||
return stem[: -len("_测试用例")]
|
||
return stem
|
||
|
||
|
||
def next_snapshot_version(version_root: Path) -> int:
|
||
latest = find_latest_snapshot(version_root)
|
||
if latest is None:
|
||
return 1
|
||
return latest[0] + 1
|
||
|
||
|
||
def find_matching_snapshot(version_root: Path, sources: list[tuple[Path, Path]]) -> tuple[int, Path] | None:
|
||
if not version_root.exists():
|
||
return None
|
||
|
||
versions: list[tuple[int, Path]] = []
|
||
for child in version_root.iterdir():
|
||
if not child.is_dir():
|
||
continue
|
||
match = VERSION_DIR_RE.match(child.name)
|
||
if not match:
|
||
continue
|
||
versions.append((int(match.group(1)), child))
|
||
|
||
for version_number, snapshot_dir in sorted(versions, key=lambda item: item[0], reverse=True):
|
||
if snapshot_matches(snapshot_dir, sources):
|
||
return version_number, snapshot_dir
|
||
|
||
return None
|
||
|
||
|
||
def decide_snapshot_version(paths: dict[str, Path], excel_path: Path) -> tuple[int, Path, bool]:
|
||
version_root = paths["version_root"]
|
||
sources = collect_snapshot_sources(paths, excel_path, include_manifest=False)
|
||
matched = find_matching_snapshot(version_root, sources)
|
||
if matched is not None:
|
||
return matched[0], matched[1], False
|
||
|
||
version_root.mkdir(parents=True, exist_ok=True)
|
||
version_number = next_snapshot_version(version_root)
|
||
snapshot_dir = version_root / f"v{version_number}"
|
||
snapshot_dir.mkdir(parents=True, exist_ok=False)
|
||
|
||
return version_number, snapshot_dir, True
|
||
|
||
|
||
def write_snapshot_files(paths: dict[str, Path], excel_path: Path, snapshot_dir: Path) -> None:
|
||
sources = collect_snapshot_sources(paths, excel_path, include_manifest=True)
|
||
for source, relative in sources:
|
||
target = snapshot_dir / relative
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
shutil.copy2(source, target)
|
||
|
||
|
||
def write_legacy_excel_snapshot(snapshot_dir: Path, base_name: str, legacy_file: Path) -> Path:
|
||
snapshot_dir.mkdir(parents=True, exist_ok=False)
|
||
target_excel = snapshot_dir / f"{base_name}_测试用例.xlsx"
|
||
shutil.copy2(legacy_file, target_excel)
|
||
note = "\n".join(
|
||
[
|
||
f"# {base_name} 历史 Excel 导入说明",
|
||
"",
|
||
f"- 导入来源：`{legacy_file.name}`",
|
||
"- 导入类型：旧时间戳命名 Excel",
|
||
"- 说明：该快照仅迁移了历史 Excel，本次未重建当时对应的分析、测试点、测试用例 Markdown 与 manifest。",
|
||
"- 目的：统一历史文件归档路径，并清理 `output/excel_reports/` 中的旧时间戳文件。",
|
||
"",
|
||
]
|
||
)
|
||
(snapshot_dir / "snapshot_note.md").write_text(note, encoding="utf-8")
|
||
return target_excel
|
||
|
||
|
||
def infer_snapshot_type(snapshot_dir: Path) -> str:
|
||
if (snapshot_dir / "snapshot_note.md").exists():
|
||
return "legacy_excel_import"
|
||
|
||
has_manifest = any(item.name != "snapshot_meta.json" for item in snapshot_dir.glob("*.json"))
|
||
has_analysis = any(snapshot_dir.glob("*_分析.md"))
|
||
has_relation = any(snapshot_dir.glob("*_关联与冲突.md"))
|
||
has_test_points = any(snapshot_dir.glob("*_测试点.md"))
|
||
has_test_cases = any(snapshot_dir.glob("*_测试用例.md"))
|
||
has_excel = any(snapshot_dir.glob("*.xlsx"))
|
||
|
||
if has_manifest and has_analysis and has_relation and has_test_points and has_test_cases and has_excel:
|
||
return "full_pipeline"
|
||
return "partial_snapshot"
|
||
|
||
|
||
def build_snapshot_summary(snapshot_dir: Path) -> list[str]:
|
||
summary: list[str] = []
|
||
if any(item.name != "snapshot_meta.json" for item in snapshot_dir.glob("*.json")):
|
||
summary.append("manifest")
|
||
if any(snapshot_dir.glob("*_分析.md")):
|
||
summary.append("analysis")
|
||
if any(snapshot_dir.glob("*_关联与冲突.md")):
|
||
summary.append("relation_report")
|
||
if any(snapshot_dir.glob("*_测试点.md")):
|
||
summary.append("test_points")
|
||
if any(snapshot_dir.glob("*_测试用例.md")):
|
||
summary.append("test_cases_markdown")
|
||
if any(snapshot_dir.glob("*.xlsx")):
|
||
summary.append("excel")
|
||
if (snapshot_dir / "normalized_inputs").exists():
|
||
summary.append("normalized_inputs")
|
||
if (snapshot_dir / "snapshot_note.md").exists():
|
||
summary.append("migration_note")
|
||
return summary
|
||
|
||
|
||
def write_snapshot_metadata(snapshot_dir: Path, base_name: str) -> Path:
|
||
version_name = snapshot_dir.name
|
||
snapshot_type = infer_snapshot_type(snapshot_dir)
|
||
meta = {
|
||
"base_name": base_name,
|
||
"version": version_name,
|
||
"snapshot_type": snapshot_type,
|
||
"summary": build_snapshot_summary(snapshot_dir),
|
||
}
|
||
note_file = snapshot_dir / "snapshot_note.md"
|
||
if note_file.exists():
|
||
meta["note_file"] = str(note_file)
|
||
|
||
meta_path = snapshot_dir / "snapshot_meta.json"
|
||
meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
return meta_path
|
||
|
||
|
||
def refresh_version_index(version_root: Path, base_name: str) -> Path:
|
||
version_root.mkdir(parents=True, exist_ok=True)
|
||
version_entries: list[tuple[int, Path, dict[str, Any]]] = []
|
||
|
||
for version_number, child in list_snapshot_versions(version_root):
|
||
meta_path = write_snapshot_metadata(child, base_name)
|
||
meta = json.loads(meta_path.read_text(encoding="utf-8"))
|
||
version_entries.append((version_number, child, meta))
|
||
|
||
lines = [
|
||
f"# {base_name} 版本索引",
|
||
"",
|
||
"| 版本 | 类型 | 内容摘要 | 目录 |",
|
||
"| --- | --- | --- | --- |",
|
||
]
|
||
|
||
type_labels = {
|
||
"full_pipeline": "完整流水线快照",
|
||
"legacy_excel_import": "历史 Excel 导入",
|
||
"partial_snapshot": "部分产物快照",
|
||
}
|
||
|
||
for version_number, snapshot_dir, meta in sorted(version_entries, key=lambda item: item[0]):
|
||
summary = "、".join(meta.get("summary", [])) or "-"
|
||
lines.append(
|
||
"| "
|
||
+ " | ".join(
|
||
[
|
||
f"v{version_number}",
|
||
type_labels.get(meta.get("snapshot_type", ""), meta.get("snapshot_type", "未知")),
|
||
summary,
|
||
f"`{to_repo_relative(snapshot_dir)}`",
|
||
]
|
||
)
|
||
+ " |"
|
||
)
|
||
|
||
if len(lines) == 4:
|
||
lines.append("| - | - | - | - |")
|
||
|
||
index_path = version_root / "VERSION_INDEX.md"
|
||
index_path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8")
|
||
return index_path
|
||
|
||
|
||
def prune_old_snapshots(version_root: Path, keep: int = MAX_SNAPSHOT_VERSIONS) -> list[Path]:
|
||
versions = list_snapshot_versions(version_root)
|
||
if len(versions) <= keep:
|
||
return []
|
||
|
||
removable = versions[: len(versions) - keep]
|
||
removed_dirs: list[Path] = []
|
||
for _, snapshot_dir in removable:
|
||
shutil.rmtree(snapshot_dir)
|
||
removed_dirs.append(snapshot_dir)
|
||
return removed_dirs
|
||
|
||
|
||
def migrate_legacy_excel_exports(paths: dict[str, Path]) -> dict[str, int]:
|
||
legacy_files = find_legacy_excel_exports(paths)
|
||
if not legacy_files:
|
||
return {"found": 0, "imported": 0, "deduplicated": 0, "removed": 0}
|
||
|
||
version_root = paths["version_root"]
|
||
version_root.mkdir(parents=True, exist_ok=True)
|
||
snapshot_excels = existing_snapshot_excel_files(version_root)
|
||
imported = 0
|
||
deduplicated = 0
|
||
removed = 0
|
||
|
||
for legacy_file in legacy_files:
|
||
duplicated = any(files_match(legacy_file, snapshot_excel) for snapshot_excel in snapshot_excels)
|
||
if duplicated:
|
||
deduplicated += 1
|
||
else:
|
||
version_number = next_snapshot_version(version_root)
|
||
snapshot_dir = version_root / f"v{version_number}"
|
||
target_excel = write_legacy_excel_snapshot(snapshot_dir, extract_base_name_from_paths(paths), legacy_file)
|
||
snapshot_excels.append(target_excel)
|
||
imported += 1
|
||
|
||
legacy_file.unlink()
|
||
removed += 1
|
||
|
||
return {
|
||
"found": len(legacy_files),
|
||
"imported": imported,
|
||
"deduplicated": deduplicated,
|
||
"removed": removed,
|
||
}
|
||
|
||
|
||
def update_manifest_version_info(
|
||
manifest_path: Path,
|
||
excel_path: Path,
|
||
version_number: int,
|
||
snapshot_dir: Path,
|
||
) -> None:
|
||
if not manifest_path.exists():
|
||
return
|
||
|
||
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
||
manifest["current_excel_file"] = str(excel_path)
|
||
manifest["versioning_scheme"] = {
|
||
"current_files": "固定文件名，始终表示当前最新版",
|
||
"snapshot_rule": "仅在 export 成功且产物内容发生变化时递增版本",
|
||
"snapshot_dir_pattern": "output/versions/{BASE_NAME}/vN/",
|
||
}
|
||
manifest["latest_snapshot_version"] = f"v{version_number}"
|
||
manifest["latest_snapshot_dir"] = str(snapshot_dir)
|
||
manifest["latest_snapshot_type"] = infer_snapshot_type(snapshot_dir)
|
||
write_manifest_data(manifest_path, manifest)
|
||
|
||
|
||
def print_legacy_migration_result(result: dict[str, int]) -> None:
|
||
if result["found"] == 0:
|
||
print("🧹 历史时间戳 Excel: 0")
|
||
return
|
||
|
||
print(
|
||
"🧹 历史时间戳 Excel: "
|
||
f"发现 {result['found']} 个，导入版本 {result['imported']} 个，"
|
||
f"重复去重 {result['deduplicated']} 个，清理原文件 {result['removed']} 个"
|
||
)
|
||
|
||
|
||
def print_pruned_snapshots(removed_dirs: list[Path]) -> None:
|
||
if not removed_dirs:
|
||
print(f"🗃️ 版本保留上限: {MAX_SNAPSHOT_VERSIONS}，本次无需淘汰旧版本")
|
||
return
|
||
|
||
removed_names = ", ".join(snapshot_dir.name for snapshot_dir in removed_dirs)
|
||
print(
|
||
f"🗃️ 版本保留上限: {MAX_SNAPSHOT_VERSIONS}，"
|
||
f"已淘汰旧版本 {len(removed_dirs)} 个: {removed_names}"
|
||
)
|
||
|
||
|
||
def command_prepare(requirement: str) -> None:
|
||
requirement_path = resolve_requirement_path(requirement)
|
||
validate_requirement_file(requirement_path)
|
||
validate_knowledge_base()
|
||
validate_project_profile()
|
||
paths = build_paths(requirement_path.stem)
|
||
ensure_output_dirs(paths)
|
||
|
||
technical_solution_files = select_technical_solution_files(requirement_path)
|
||
normalized_requirement_file = write_normalized_document(
|
||
source_path=requirement_path,
|
||
target_path=paths["normalized_input_dir"] / "requirement.md",
|
||
document_role="需求文档",
|
||
)
|
||
normalized_technical_solution_files = [
|
||
write_normalized_document(
|
||
source_path=path,
|
||
target_path=paths["normalized_input_dir"] / f"technical_solution_{index:02d}.md",
|
||
document_role="技术方案",
|
||
)
|
||
for index, path in enumerate(technical_solution_files, start=1)
|
||
]
|
||
related_requirements = find_related_requirements(requirement_path)
|
||
optional_terminology_files = select_optional_terminology_files(
|
||
requirement_path,
|
||
related_requirements,
|
||
technical_solution_files,
|
||
)
|
||
conflicts = build_conflict_candidates(requirement_path, related_requirements)
|
||
confirmation_gate = build_confirmation_gate(
|
||
requirement_path=requirement_path,
|
||
base_name=requirement_path.stem,
|
||
related_requirements=related_requirements,
|
||
conflicts=conflicts,
|
||
normalized_requirement_file=normalized_requirement_file,
|
||
normalized_technical_solution_files=normalized_technical_solution_files,
|
||
)
|
||
write_relation_report(
|
||
requirement_path=requirement_path,
|
||
related_requirements=related_requirements,
|
||
technical_solution_files=technical_solution_files,
|
||
conflicts=conflicts,
|
||
report_path=paths["relation_report"],
|
||
)
|
||
manifest_path = write_manifest(
|
||
requirement_path=requirement_path,
|
||
paths=paths,
|
||
related_requirements=related_requirements,
|
||
technical_solution_files=technical_solution_files,
|
||
normalized_requirement_file=normalized_requirement_file,
|
||
normalized_technical_solution_files=normalized_technical_solution_files,
|
||
optional_terminology_files=optional_terminology_files,
|
||
conflict_count=len(conflicts),
|
||
confirmation_gate=confirmation_gate,
|
||
)
|
||
|
||
print(f"✅ 准备完成: {requirement_path}")
|
||
print(f"📦 清单文件: {manifest_path}")
|
||
print(f"📄 标准化需求输入: {normalized_requirement_file}")
|
||
print(f"🧩 关联技术方案数量: {len(technical_solution_files)}")
|
||
print(f"📝 分析输出: {paths['analysis']}")
|
||
print(f"🔎 关联与冲突输出: {paths['relation_report']}")
|
||
print(f"🧪 测试点输出: {paths['test_points']}")
|
||
print(f"📋 用例输出: {paths['test_cases']}")
|
||
print(f"📚 关联需求数量: {len(related_requirements)}")
|
||
print(f"📖 可选术语文件: {len(optional_terminology_files)}")
|
||
print(f"⚠️ 潜在冲突候选: {len(conflicts)}")
|
||
if confirmation_gate["required"]:
|
||
if confirmation_gate["decision_status"] == "confirmed":
|
||
print(f"✅ 人工确认: 已确认 -> {confirmation_gate['decision_file']}")
|
||
else:
|
||
print("🛑 人工确认: 必需")
|
||
print(f"📝 确认单建议路径: {confirmation_gate['suggested_decision_file']}")
|
||
else:
|
||
print("🟢 人工确认: 当前无需额外确认")
|
||
|
||
|
||
def verify_case_columns(case_file: Path) -> list[list[str]]:
|
||
headers, rows = load_markdown_table(case_file)
|
||
if headers not in (REQUIRED_COLUMNS, LEGACY_COLUMNS):
|
||
raise ValueError(
|
||
"测试用例表头不符合标准。\n"
|
||
f"期望: {REQUIRED_COLUMNS} 或 {LEGACY_COLUMNS}\n"
|
||
f"实际: {headers}"
|
||
)
|
||
if headers == REQUIRED_COLUMNS:
|
||
type_index = headers.index("类型")
|
||
invalid_types = sorted({row[type_index] for row in rows if row[type_index] and row[type_index] not in TYPE_ENUMS})
|
||
if invalid_types:
|
||
raise ValueError(f"测试用例存在非法类型枚举：{', '.join(invalid_types)}。")
|
||
return rows
|
||
|
||
|
||
def command_verify(requirement: str) -> None:
|
||
requirement_path = resolve_requirement_path(requirement)
|
||
validate_requirement_file(requirement_path)
|
||
paths = build_paths(requirement_path.stem)
|
||
|
||
for name in ("analysis", "relation_report", "test_points", "test_cases"):
|
||
target = paths[name]
|
||
if not target.exists():
|
||
raise FileNotFoundError(f"缺少输出文件：{target}")
|
||
if target.stat().st_size == 0:
|
||
raise ValueError(f"输出文件为空：{target}")
|
||
|
||
manifest = ensure_confirmation_resolved(paths, "verify")
|
||
rows = verify_case_columns(paths["test_cases"])
|
||
print(f"✅ 校验通过: {requirement_path.name}")
|
||
print(f"📄 分析文件: {paths['analysis']}")
|
||
print(f"🔎 关联与冲突文件: {paths['relation_report']}")
|
||
print(f"🧪 测试点文件: {paths['test_points']}")
|
||
print(f"📋 测试用例文件: {paths['test_cases']}")
|
||
print(f"📊 用例数量: {len(rows)}")
|
||
gate = manifest.get("confirmation_gate", {})
|
||
if gate.get("required"):
|
||
print(f"📝 确认单: {gate.get('decision_file')}")
|
||
|
||
|
||
def command_export(requirement: str) -> None:
|
||
requirement_path = resolve_requirement_path(requirement)
|
||
validate_requirement_file(requirement_path)
|
||
paths = build_paths(requirement_path.stem)
|
||
if not paths["test_cases"].exists():
|
||
raise FileNotFoundError(f"测试用例文件不存在：{paths['test_cases']}")
|
||
|
||
ensure_confirmation_resolved(paths, "export")
|
||
verify_case_columns(paths["test_cases"])
|
||
output_path = export_markdown_to_excel(paths["test_cases"], paths["excel_dir"])
|
||
legacy_result = migrate_legacy_excel_exports(paths)
|
||
version_number, snapshot_dir, created = decide_snapshot_version(paths, output_path)
|
||
if created:
|
||
write_snapshot_files(paths, output_path, snapshot_dir)
|
||
pruned_dirs = prune_old_snapshots(paths["version_root"])
|
||
index_path = refresh_version_index(paths["version_root"], extract_base_name_from_paths(paths))
|
||
latest = find_latest_snapshot(paths["version_root"])
|
||
if latest is None:
|
||
raise FileNotFoundError(f"版本目录为空：{paths['version_root']}")
|
||
version_number, snapshot_dir = latest
|
||
update_manifest_version_info(paths["manifest"], output_path, version_number, snapshot_dir)
|
||
maintained_requirement_path = sync_maintained_requirement(requirement_path)
|
||
|
||
print(f"✅ Excel 已生成: {output_path}")
|
||
print(f"🧾 正式维护版: {maintained_requirement_path}")
|
||
if created:
|
||
print(f"📦 版本快照: v{version_number} -> {snapshot_dir}")
|
||
else:
|
||
print(f"♻️ 版本未变: v{version_number} -> {snapshot_dir}")
|
||
print_legacy_migration_result(legacy_result)
|
||
print_pruned_snapshots(pruned_dirs)
|
||
print(f"🗂️ 版本索引: {index_path}")
|
||
|
||
|
||
def command_sync_maintained(requirement: str) -> None:
|
||
requirement_path = resolve_requirement_path(requirement)
|
||
validate_requirement_file(requirement_path)
|
||
paths = build_paths(requirement_path.stem)
|
||
if not paths["manifest"].exists():
|
||
raise FileNotFoundError(f"缺少清单文件：{paths['manifest']}。请先执行 prepare。")
|
||
|
||
maintained_requirement_path = sync_maintained_requirement(requirement_path)
|
||
print(f"✅ 正式维护版已同步: {maintained_requirement_path}")
|
||
print(f"📄 原始来源: {requirement_path}")
|
||
print(f"🧾 manifest: {paths['manifest']}")
|
||
|
||
|
||
def command_apply_confirmation(requirement: str, dry_run: bool) -> None:
|
||
requirement_path = resolve_requirement_path(requirement)
|
||
validate_requirement_file(requirement_path)
|
||
paths = build_paths(requirement_path.stem)
|
||
manifest = refresh_manifest_confirmation_gate(paths)
|
||
gate = manifest.get("confirmation_gate", {})
|
||
decision_file = gate.get("decision_file")
|
||
if not decision_file:
|
||
fallback_files = find_relevant_decision_files(requirement_path)
|
||
if fallback_files:
|
||
decision_file = str(fallback_files[0])
|
||
if not decision_file:
|
||
raise FileNotFoundError("未找到可用确认单。")
|
||
|
||
decision = parse_decision_file(Path(decision_file))
|
||
if decision["status"] != "confirmed":
|
||
raise ValueError(f"确认单未处于已确认状态：{decision_file}")
|
||
|
||
related_requirements = [
|
||
resolve_requirement_reference(reference)
|
||
for reference in decision.get("related_requirement_refs", [])
|
||
]
|
||
rerun_targets = build_rerun_targets(decision, requirement_path)
|
||
note_path = build_maintenance_note_path(Path(decision["file"]))
|
||
|
||
print(f"✅ 确认单已读取: {decision_file}")
|
||
print(f"🧭 关系类型: {decision.get('relation_type') or '未填写'}")
|
||
print(f"📝 维护说明路径: {note_path}")
|
||
print(f"🔁 重跑需求数量: {len(rerun_targets)}")
|
||
for target in rerun_targets:
|
||
print(f" - {to_repo_relative(target)}")
|
||
|
||
if dry_run:
|
||
print("🧪 dry-run: 未写入维护说明，未执行重跑。")
|
||
return
|
||
|
||
note_path = write_maintenance_note(
|
||
decision=decision,
|
||
current_requirement=requirement_path,
|
||
related_requirements=related_requirements,
|
||
rerun_targets=rerun_targets,
|
||
)
|
||
for target in rerun_targets:
|
||
run_pipeline_for_requirement(target)
|
||
|
||
print(f"📌 维护说明已写入: {note_path}")
|
||
print("🚀 确认后续跑已完成")
|
||
|
||
|
||
def command_migrate_history(requirement: str | None, all_requirements: bool) -> None:
|
||
if not requirement and not all_requirements:
|
||
raise ValueError("请指定 --requirement，或使用 --all 扫描全部历史 Excel。")
|
||
|
||
base_names: list[str] = []
|
||
if all_requirements:
|
||
for excel_file in sorted((REPO_ROOT / "output" / "excel_reports").glob("*.xlsx")):
|
||
if LEGACY_EXCEL_EXPORT_RE.match(excel_file.name):
|
||
stem = excel_file.stem
|
||
base_name = re.sub(r"_\d{8}_\d{6}$", "", stem)
|
||
if base_name.endswith("_测试用例"):
|
||
base_name = base_name[: -len("_测试用例")]
|
||
if base_name not in base_names:
|
||
base_names.append(base_name)
|
||
else:
|
||
requirement_path = resolve_requirement_path(requirement or "")
|
||
validate_requirement_file(requirement_path)
|
||
base_names.append(requirement_path.stem)
|
||
|
||
total_found = 0
|
||
total_imported = 0
|
||
total_deduplicated = 0
|
||
total_removed = 0
|
||
|
||
for base_name in base_names:
|
||
paths = build_paths(base_name)
|
||
result = migrate_legacy_excel_exports(paths)
|
||
pruned_dirs = prune_old_snapshots(paths["version_root"])
|
||
index_path = refresh_version_index(paths["version_root"], extract_base_name_from_paths(paths))
|
||
total_found += result["found"]
|
||
total_imported += result["imported"]
|
||
total_deduplicated += result["deduplicated"]
|
||
total_removed += result["removed"]
|
||
print(f"📁 {base_name}")
|
||
print_legacy_migration_result(result)
|
||
print_pruned_snapshots(pruned_dirs)
|
||
print(f"🗂️ 版本索引: {index_path}")
|
||
|
||
print(
|
||
"✅ 历史 Excel 迁移完成: "
|
||
f"发现 {total_found} 个，导入版本 {total_imported} 个，"
|
||
f"重复去重 {total_deduplicated} 个，清理原文件 {total_removed} 个"
|
||
)
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="测试用例生成流水线辅助脚本。")
|
||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||
|
||
prepare_parser = subparsers.add_parser("prepare", help="校验输入并生成本次任务清单。")
|
||
prepare_parser.add_argument("--requirement", required=True, help="需求文档路径。")
|
||
|
||
verify_parser = subparsers.add_parser("verify", help="校验本次任务产物。")
|
||
verify_parser.add_argument("--requirement", required=True, help="需求文档路径。")
|
||
|
||
export_parser = subparsers.add_parser("export", help="导出本次任务的 Excel。")
|
||
export_parser.add_argument("--requirement", required=True, help="需求文档路径。")
|
||
|
||
apply_confirmation_parser = subparsers.add_parser(
|
||
"apply-confirmation",
|
||
help="读取已确认的确认单，写出维护说明并重跑受影响需求。",
|
||
)
|
||
apply_confirmation_parser.add_argument("--requirement", required=True, help="需求文档路径。")
|
||
apply_confirmation_parser.add_argument("--dry-run", action="store_true", help="仅校验确认单和重跑目标，不执行写入或重跑。")
|
||
|
||
sync_maintained_parser = subparsers.add_parser(
|
||
"sync-maintained",
|
||
help="同步 requirements/ 下的正式维护版 Markdown。",
|
||
)
|
||
sync_maintained_parser.add_argument("--requirement", required=True, help="需求文档路径。")
|
||
|
||
migrate_parser = subparsers.add_parser("migrate-history", help="迁移旧时间戳 Excel 到版本目录。")
|
||
migrate_parser.add_argument("--requirement", help="按需求文档迁移对应 BASE_NAME 的历史 Excel。")
|
||
migrate_parser.add_argument("--all", action="store_true", help="扫描并迁移全部历史时间戳 Excel。")
|
||
|
||
args = parser.parse_args()
|
||
|
||
if args.command == "prepare":
|
||
command_prepare(args.requirement)
|
||
elif args.command == "verify":
|
||
command_verify(args.requirement)
|
||
elif args.command == "export":
|
||
command_export(args.requirement)
|
||
elif args.command == "sync-maintained":
|
||
command_sync_maintained(args.requirement)
|
||
elif args.command == "apply-confirmation":
|
||
command_apply_confirmation(args.requirement, args.dry_run)
|
||
elif args.command == "migrate-history":
|
||
command_migrate_history(args.requirement, args.all)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|