diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..3e732dd --- /dev/null +++ b/app/__init__.py @@ -0,0 +1 @@ +"""DOCX compliance analysis application.""" diff --git a/app/analyzer.py b/app/analyzer.py new file mode 100644 index 0000000..01ad3e8 --- /dev/null +++ b/app/analyzer.py @@ -0,0 +1,222 @@ +from __future__ import annotations + +import json +import re +from collections import Counter +from typing import Protocol + +import requests + +from app.config import ProviderConfig +from app.docx_parser import ParsedDocument +from app.report_generator import AnalysisReport +from app.skill_loader import Skill + + +class SupportsPost(Protocol): + def post(self, url: str, **kwargs): ... + + +IMPORTANT_TERMS = { + "需求", + "接口", + "测试", + "合格性", + "追踪", + "追溯", + "配置", + "质量", + "部署", + "安装", + "验收", + "设计", + "资源", + "风险", + "计划", + "说明", + "文档", + "CSCI", + "GJB", +} + + +def _tokens(text: str) -> list[str]: + ascii_tokens = re.findall(r"[A-Za-z][A-Za-z0-9_-]{2,}", text) + chinese_terms = [term for term in IMPORTANT_TERMS if term in text] + return [token.lower() for token in ascii_tokens] + chinese_terms + + +def select_relevant_skills(parsed: ParsedDocument, skills: list[Skill], max_skills: int = 6) -> list[Skill]: + document_tokens = Counter(_tokens(parsed.text + "\n" + "\n".join(h.text for h in parsed.headings))) + scored: list[tuple[int, Skill]] = [] + + for skill in skills: + skill_text = f"{skill.slug}\n{skill.name}\n{skill.description}\n{skill.use_when}\n{skill.content[:3000]}" + score = 0 + skill_tokens = set(_tokens(skill_text)) + for token, count in document_tokens.items(): + if token in skill_tokens: + score += count + if parsed.filename.lower().endswith(".docx"): + score += 1 + if score > 0: + scored.append((score, skill)) + + scored.sort(key=lambda item: (-item[0], item[1].slug)) + if not scored: + return skills[:max_skills] + return [skill for _, skill in scored[:max_skills]] + + +def build_analysis_prompt(parsed: ParsedDocument, skills: list[Skill]) -> str: + skill_sections = [] + for skill in skills: + skill_sections.append( + f"## {skill.slug}\n名称: {skill.name}\n描述: {skill.description}\n适用条件: {skill.use_when}\n规范内容:\n{skill.content[:6000]}" + ) + + document_outline = "\n".join(f"- H{heading.level} {heading.text}" for heading in parsed.headings[:80]) or "未识别到标题。" + document_text = parsed.text[:18000] + + return f"""你是军用软件文档符合性审查助手。请依据给定 GJB438C/GJB2786 技能规范，分析上传 DOCX 是否符合规范。 + +请输出中文 Markdown，必须包含以下小节： +1. 总体结论 +2. 符合项 +3. 不符合项 +4. 缺失章节或缺失证据 +5. 整改建议 +6. 需人工复核事项 + +要求： +- 每个问题尽量引用文档中的标题、关键词或证据摘要。 +- 不要编造未在文档中出现的证据。 +- 如果无法判断，标记为“需人工复核”。 + +# 文件 +{parsed.filename} + +# 文档目录 +{document_outline} + +# 待检查技能 +{chr(10).join(skill_sections)} + +# 文档正文摘录 +{document_text} +""" + + +class LLMClient: + def __init__(self, provider: ProviderConfig, session: SupportsPost | None = None, timeout: int = 120) -> None: + self.provider = provider + self.session = session or requests.Session() + self.timeout = timeout + + def complete(self, prompt: str) -> str: + headers = {"Content-Type": "application/json"} + if self.provider.api_key and self.provider.api_key != "EMPTY": + headers["Authorization"] = f"Bearer {self.provider.api_key}" + + payload = { + "model": self.provider.model, + "messages": [{"role": "user", "content": prompt}], + "temperature": self.provider.temperature, + "max_tokens": self.provider.max_tokens, + } + response = self.session.post( + self.provider.chat_completions_url, + headers=headers, + json=payload, + timeout=self.timeout, + ) + response.raise_for_status() + data = response.json() + return data["choices"][0]["message"]["content"] + + +def heuristic_analysis(parsed: ParsedDocument, skills: list[Skill]) -> AnalysisReport: + headings_text = "\n".join(h.text for h in parsed.headings) + full_text = parsed.text + findings: list[dict[str, str]] = [] + recommendations: list[str] = [] + + required_terms = ["范围", "引用文档", "需求", "合格性", "追踪", "接口"] + for term in required_terms: + present = term in full_text or term in headings_text + findings.append( + { + "status": "符合" if present else "需整改", + "item": f"检查关键内容：{term}", + "evidence": "文档中已发现相关表述" if present else "未在解析文本中发现明确表述", + } + ) + if not present: + recommendations.append(f"补充或明确“{term}”相关章节与证据。") + + if not parsed.headings: + findings.append({"status": "需整改", "item": "章节结构", "evidence": "未识别到 Word 标题样式"}) + recommendations.append("使用 Word 标题样式组织章节，便于目录和符合性审查。") + + issue_count = sum(1 for item in findings if item["status"] != "符合") + summary = "通过" if issue_count == 0 else "部分通过，需人工复核" + raw_output = "未调用模型，已使用本地启发式规则生成初步分析。" + + return AnalysisReport( + source_filename=parsed.filename, + provider_name="local", + model_name="heuristic", + matched_skills=[skill.slug for skill in skills], + summary=summary, + findings=findings, + recommendations=recommendations or ["保持现有章节结构，并由人工进行最终符合性确认。"], + raw_model_output=raw_output, + ) + + +def report_from_model_output( + parsed: ParsedDocument, + skills: list[Skill], + provider_name: str, + model_name: str, + output: str, +) -> AnalysisReport: + findings = [{"status": "模型分析", "item": "完整分析结果", "evidence": output[:1200]}] + recommendations = _extract_recommendations(output) + return AnalysisReport( + source_filename=parsed.filename, + provider_name=provider_name, + model_name=model_name, + matched_skills=[skill.slug for skill in skills], + summary=_extract_summary(output), + findings=findings, + recommendations=recommendations, + raw_model_output=output, + ) + + +def _extract_summary(output: str) -> str: + for line in output.splitlines(): + normalized = line.strip(" #：:") + if "总体结论" in normalized and len(normalized) > 4: + return normalized + return "模型已生成分析结果，需人工复核" + + +def _extract_recommendations(output: str) -> list[str]: + recommendations: list[str] = [] + in_section = False + for line in output.splitlines(): + stripped = line.strip() + if "整改建议" in stripped or "修改建议" in stripped: + in_section = True + continue + if in_section and stripped.startswith("#"): + break + if in_section and stripped.lstrip("-0123456789.、 "): + recommendations.append(stripped.lstrip("-0123456789.、 ")) + return recommendations[:10] or ["按模型分析结果逐项整改，并进行人工复核。"] + + +def serialize_prompt_debug(prompt: str) -> str: + return json.dumps({"prompt_preview": prompt[:2000]}, ensure_ascii=False, indent=2) diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..f43b370 --- /dev/null +++ b/app/config.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +import yaml + + +@dataclass(frozen=True) +class ProviderConfig: + api_key: str + base_url: str + max_tokens: int + model: str + temperature: float + + @property + def chat_completions_url(self) -> str: + base_url = self.base_url.rstrip("/") + if base_url.endswith("/chat/completions"): + return base_url + return f"{base_url}/chat/completions" + + +@dataclass(frozen=True) +class ApiSettings: + provider_name: str + provider: ProviderConfig + + +def load_api_config(path: Path | str = Path("configs/api_config.yaml"), provider_name: str | None = None) -> ApiSettings: + config_path = Path(path) + data = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {} + selected_name = provider_name or data.get("default_provider") + providers = data.get("providers", {}) + + if not selected_name: + raise ValueError("api_config.yaml missing default_provider") + if selected_name not in providers: + raise ValueError(f"provider not found in api_config.yaml: {selected_name}") + + provider_data = providers[selected_name] + provider = ProviderConfig( + api_key=str(provider_data.get("api_key", "")), + base_url=str(provider_data["base_url"]), + max_tokens=int(provider_data.get("max_tokens", 4096)), + model=str(provider_data["model"]), + temperature=float(provider_data.get("temperature", 0.7)), + ) + return ApiSettings(provider_name=selected_name, provider=provider) diff --git a/app/docx_parser.py b/app/docx_parser.py new file mode 100644 index 0000000..2fed0c7 --- /dev/null +++ b/app/docx_parser.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +from docx import Document + + +@dataclass(frozen=True) +class Heading: + level: int + text: str + + +@dataclass(frozen=True) +class ParsedDocument: + filename: str + text: str + paragraphs: list[str] + headings: list[Heading] + tables: list[list[list[str]]] + + +def _heading_level(style_name: str) -> int | None: + if not style_name.lower().startswith("heading"): + return None + parts = style_name.split() + if parts and parts[-1].isdigit(): + return int(parts[-1]) + return 1 + + +def parse_docx(path: Path | str, display_filename: str | None = None) -> ParsedDocument: + docx_path = Path(path) + document = Document(docx_path) + paragraphs: list[str] = [] + headings: list[Heading] = [] + tables: list[list[list[str]]] = [] + text_parts: list[str] = [] + + for paragraph in document.paragraphs: + text = paragraph.text.strip() + if not text: + continue + paragraphs.append(text) + text_parts.append(text) + level = _heading_level(paragraph.style.name if paragraph.style else "") + if level is not None: + headings.append(Heading(level=level, text=text)) + + for table in document.tables: + rows: list[list[str]] = [] + for row in table.rows: + values = [cell.text.strip() for cell in row.cells] + if any(values): + rows.append(values) + text_parts.append(" | ".join(values)) + if rows: + tables.append(rows) + + return ParsedDocument( + filename=display_filename or docx_path.name, + text="\n".join(text_parts), + paragraphs=paragraphs, + headings=headings, + tables=tables, + ) diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..a38c85e --- /dev/null +++ b/app/main.py @@ -0,0 +1,300 @@ +from __future__ import annotations + +import shutil +import threading +import time +from dataclasses import dataclass, field +from pathlib import Path +from uuid import uuid4 +from typing import Callable + +from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile +from fastapi.responses import FileResponse, HTMLResponse +from fastapi.staticfiles import StaticFiles +from fastapi.templating import Jinja2Templates + +from app.analyzer import ( + LLMClient, + build_analysis_prompt, + heuristic_analysis, + report_from_model_output, + select_relevant_skills, +) +from app.config import load_api_config +from app.docx_parser import parse_docx +from app.report_generator import generate_docx_report, generate_markdown_report +from app.skill_loader import load_skill_catalog + + +ROOT_DIR = Path(__file__).resolve().parent.parent +UPLOAD_DIR = ROOT_DIR / "uploads" +OUTPUT_DIR = ROOT_DIR / "outputs" +SKILL_DIR = ROOT_DIR / "GJB438C-2021_prd_skills" +CONFIG_PATH = ROOT_DIR / "configs" / "api_config.yaml" +MAX_UPLOAD_BYTES = 30 * 1024 * 1024 +ProgressCallback = Callable[[int, str], None] + + +@dataclass +class AnalysisTask: + task_id: str + source_filename: str + status: str = "queued" + progress: int = 0 + message: str = "任务已创建" + summary: str = "" + matched_skills: list[str] = field(default_factory=list) + downloads: dict[str, str] = field(default_factory=dict) + error: str = "" + created_at: float = field(default_factory=time.time) + updated_at: float = field(default_factory=time.time) + + def to_dict(self) -> dict[str, object]: + return { + "task_id": self.task_id, + "source_filename": self.source_filename, + "status": self.status, + "progress": self.progress, + "message": self.message, + "summary": self.summary, + "matched_skills": self.matched_skills, + "downloads": self.downloads, + "error": self.error, + "created_at": self.created_at, + "updated_at": self.updated_at, + } + + +class AnalysisTaskStore: + def __init__(self) -> None: + self._tasks: dict[str, AnalysisTask] = {} + self._lock = threading.Lock() + + def create(self, source_filename: str) -> AnalysisTask: + task = AnalysisTask(task_id=uuid4().hex, source_filename=source_filename) + with self._lock: + self._tasks[task.task_id] = task + return task + + def update( + self, + task_id: str, + *, + status: str | None = None, + progress: int | None = None, + message: str | None = None, + summary: str | None = None, + matched_skills: list[str] | None = None, + downloads: dict[str, str] | None = None, + error: str | None = None, + ) -> AnalysisTask: + with self._lock: + task = self._tasks[task_id] + if status is not None: + task.status = status + if progress is not None: + task.progress = progress + if message is not None: + task.message = message + if summary is not None: + task.summary = summary + if matched_skills is not None: + task.matched_skills = matched_skills + if downloads is not None: + task.downloads = downloads + if error is not None: + task.error = error + task.updated_at = time.time() + return task + + def get(self, task_id: str) -> AnalysisTask | None: + with self._lock: + return self._tasks.get(task_id) + + +TASK_STORE = AnalysisTaskStore() + +app = FastAPI(title="GJB438C DOCX 规范分析") +templates = Jinja2Templates(directory=str(ROOT_DIR / "app" / "templates")) +app.mount("/static", StaticFiles(directory=str(ROOT_DIR / "app" / "static")), name="static") + + +def analyze_saved_docx( + upload_path: Path, + provider: str | None = None, + use_model: bool = True, + display_filename: str | None = None, + progress_callback: ProgressCallback | None = None, +) -> dict[str, object]: + def progress(percent: int, message: str) -> None: + if progress_callback is not None: + progress_callback(percent, message) + + progress(5, "正在解析 DOCX 文档") + parsed = parse_docx(upload_path, display_filename=display_filename) + progress(20, "DOCX 解析完成，正在加载技能规范") + skills = load_skill_catalog(SKILL_DIR) + progress(35, "技能规范已加载，正在匹配候选技能") + selected_skills = select_relevant_skills(parsed, skills) + progress(50, f"已匹配 {len(selected_skills)} 项技能，正在读取模型配置") + settings = load_api_config(CONFIG_PATH, provider_name=provider or None) + + if use_model: + progress(65, f"正在调用 {settings.provider.model} 进行分析") + prompt = build_analysis_prompt(parsed, selected_skills) + try: + output = LLMClient(settings.provider).complete(prompt) + report = report_from_model_output( + parsed, + selected_skills, + settings.provider_name, + settings.provider.model, + output, + ) + except Exception as exc: + report = heuristic_analysis(parsed, selected_skills) + report = report.__class__( + source_filename=report.source_filename, + provider_name=settings.provider_name, + model_name=f"{settings.provider.model} (调用失败，已降级)", + matched_skills=report.matched_skills, + summary=f"{report.summary}；模型调用失败：{exc}", + findings=report.findings, + recommendations=report.recommendations, + raw_model_output=f"模型调用失败：{exc}\n\n{report.raw_model_output}", + ) + else: + progress(70, "已关闭模型分析，正在使用本地规则生成结果") + report = heuristic_analysis(parsed, selected_skills) + + progress(85, "正在生成 Markdown 分析文档") + markdown_path = generate_markdown_report(report, OUTPUT_DIR) + progress(100, "分析完成") + + return { + "source_filename": parsed.filename, + "summary": report.summary, + "matched_skills": report.matched_skills, + "downloads": {"markdown": f"/download/{markdown_path.name}"}, + "markdown_filename": markdown_path.name, + } + + +def _run_analysis_task( + task_id: str, + upload_path: Path, + provider: str | None, + use_model: bool, + display_filename: str, +) -> None: + def on_progress(progress: int, message: str) -> None: + TASK_STORE.update(task_id, status="running", progress=progress, message=message) + + try: + TASK_STORE.update(task_id, status="running", progress=1, message="任务已启动") + result = analyze_saved_docx( + upload_path, + provider=provider, + use_model=use_model, + display_filename=display_filename, + progress_callback=on_progress, + ) + TASK_STORE.update( + task_id, + status="completed", + progress=100, + message="分析完成", + summary=str(result["summary"]), + matched_skills=list(result["matched_skills"]), + downloads=dict(result["downloads"]), + ) + except Exception as exc: + TASK_STORE.update(task_id, status="error", progress=100, message="分析失败", error=str(exc)) + + +@app.get("/", response_class=HTMLResponse) +def index(request: Request) -> HTMLResponse: + settings = load_api_config(CONFIG_PATH) + skills = load_skill_catalog(SKILL_DIR) + return templates.TemplateResponse( + request, + "index.html", + { + "default_provider": settings.provider_name, + "skill_count": len(skills), + }, + ) + + +@app.post("/analyze") +async def analyze_docx( + file: UploadFile = File(...), + provider: str | None = Form(None), + use_model: str = Form("true"), +): + if not file.filename or not file.filename.lower().endswith(".docx"): + raise HTTPException(status_code=400, detail="仅支持上传 .docx 文件") + + UPLOAD_DIR.mkdir(parents=True, exist_ok=True) + OUTPUT_DIR.mkdir(parents=True, exist_ok=True) + upload_path = UPLOAD_DIR / f"{uuid4().hex}_{Path(file.filename).name}" + + content = await file.read() + if len(content) > MAX_UPLOAD_BYTES: + raise HTTPException(status_code=413, detail="文件超过 30MB 限制") + upload_path.write_bytes(content) + + should_use_model = use_model.lower() in {"1", "true", "yes", "on"} + task = TASK_STORE.create(Path(file.filename).name) + threading.Thread( + target=_run_analysis_task, + args=(task.task_id, upload_path, provider, should_use_model, Path(file.filename).name), + daemon=True, + ).start() + return { + "task_id": task.task_id, + "status_url": f"/tasks/{task.task_id}", + "status": task.status, + "progress": task.progress, + "message": "任务已提交", + } + + +@app.get("/tasks/{task_id}") +def get_task(task_id: str) -> dict[str, object]: + task = TASK_STORE.get(task_id) + if task is None: + raise HTTPException(status_code=404, detail="任务不存在") + return task.to_dict() + + +@app.get("/download/{filename}") +def download_report(filename: str): + safe_name = Path(filename).name + path = OUTPUT_DIR / safe_name + if not path.exists() or not path.is_file(): + raise HTTPException(status_code=404, detail="报告不存在") + + media_type = "application/octet-stream" + if path.suffix == ".docx": + media_type = "application/vnd.openxmlformats-officedocument.wordprocessingml.document" + elif path.suffix == ".md": + media_type = "text/markdown; charset=utf-8" + + return FileResponse(path, filename=path.name, media_type=media_type) + + +@app.post("/cleanup") +def cleanup_runtime_files() -> dict[str, int]: + removed = 0 + for directory in (UPLOAD_DIR, OUTPUT_DIR): + if not directory.exists(): + continue + for path in directory.iterdir(): + if path.is_file(): + path.unlink() + removed += 1 + elif path.is_dir(): + shutil.rmtree(path) + removed += 1 + return {"removed": removed} diff --git a/app/report_generator.py b/app/report_generator.py new file mode 100644 index 0000000..1b7db24 --- /dev/null +++ b/app/report_generator.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from uuid import uuid4 + +from docx import Document + + +@dataclass(frozen=True) +class AnalysisReport: + source_filename: str + provider_name: str + model_name: str + matched_skills: list[str] + summary: str + findings: list[dict[str, str]] + recommendations: list[str] + raw_model_output: str + + +def _safe_stem(filename: str) -> str: + stem = Path(filename).stem or "analysis" + safe = "".join(ch if ch.isalnum() or ch in ("-", "_") else "_" for ch in stem) + return safe[:60] or "analysis" + + +def _report_base_path(report: AnalysisReport, output_dir: Path, suffix: str) -> Path: + output_dir.mkdir(parents=True, exist_ok=True) + return output_dir / f"{_safe_stem(report.source_filename)}_{uuid4().hex[:8]}_analysis.{suffix}" + + +def generate_markdown_report(report: AnalysisReport, output_dir: Path | str) -> Path: + path = _report_base_path(report, Path(output_dir), "md") + lines = [ + "# DOCX 规范分析报告", + "", + "## 基本信息", + "", + f"- 源文件：{report.source_filename}", + f"- 分析时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", + f"- 模型供应商：{report.provider_name}", + f"- 模型名称：{report.model_name}", + f"- 命中技能：{', '.join(report.matched_skills) or '无'}", + "", + "## 总体结论", + "", + report.summary, + "", + "## 技能符合性矩阵", + "", + "| 状态 | 检查项 | 证据或说明 |", + "| --- | --- | --- |", + ] + for finding in report.findings: + lines.append( + f"| {finding.get('status', '')} | {finding.get('item', '')} | {finding.get('evidence', '').replace('|', '/')} |" + ) + lines.extend(["", "## 修改建议", ""]) + for item in report.recommendations: + lines.append(f"- {item}") + lines.extend(["", "## 模型分析原文", "", report.raw_model_output]) + path.write_text("\n".join(lines), encoding="utf-8") + return path + + +def generate_docx_report(report: AnalysisReport, output_dir: Path | str) -> Path: + path = _report_base_path(report, Path(output_dir), "docx") + document = Document() + document.add_heading("DOCX 规范分析报告", level=0) + document.add_heading("基本信息", level=1) + for label, value in [ + ("源文件", report.source_filename), + ("分析时间", datetime.now().strftime("%Y-%m-%d %H:%M:%S")), + ("模型供应商", report.provider_name), + ("模型名称", report.model_name), + ("命中技能", ", ".join(report.matched_skills) or "无"), + ]: + document.add_paragraph(f"{label}：{value}") + + document.add_heading("总体结论", level=1) + document.add_paragraph(report.summary) + + document.add_heading("技能符合性矩阵", level=1) + table = document.add_table(rows=1, cols=3) + table.style = "Table Grid" + headers = table.rows[0].cells + headers[0].text = "状态" + headers[1].text = "检查项" + headers[2].text = "证据或说明" + for finding in report.findings: + row = table.add_row().cells + row[0].text = finding.get("status", "") + row[1].text = finding.get("item", "") + row[2].text = finding.get("evidence", "") + + document.add_heading("修改建议", level=1) + for item in report.recommendations: + document.add_paragraph(item, style="List Bullet") + + document.add_heading("模型分析原文", level=1) + for line in report.raw_model_output.splitlines() or ["无"]: + document.add_paragraph(line) + + document.add_paragraph("说明：模型分析结果需人工复核，不应直接作为正式审查结论。") + document.save(path) + return path diff --git a/app/skill_loader.py b/app/skill_loader.py new file mode 100644 index 0000000..70ca73e --- /dev/null +++ b/app/skill_loader.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass +from pathlib import Path + + +@dataclass(frozen=True) +class Skill: + slug: str + name: str + description: str + use_when: str + content: str + path: Path | None + + +INDEX_ROW_RE = re.compile( + r"^\|\s*\[([^\]]+)\]\(([^)]+)\)\s*\|\s*([^|]+?)\s*\|\s*([^|]+?)\s*\|" +) +FRONT_MATTER_RE = re.compile(r"^---\n(.*?)\n---\n", re.DOTALL) + + +def _front_matter_value(content: str, key: str) -> str | None: + match = FRONT_MATTER_RE.match(content) + if not match: + return None + for line in match.group(1).splitlines(): + if line.startswith(f"{key}:"): + return line.split(":", 1)[1].strip() + return None + + +def load_skill_catalog(root: Path | str = Path("GJB438C-2021_prd_skills")) -> list[Skill]: + root_path = Path(root) + index_path = root_path / "index.md" + skills: list[Skill] = [] + seen: set[str] = set() + + if index_path.exists(): + for line in index_path.read_text(encoding="utf-8").splitlines(): + match = INDEX_ROW_RE.match(line) + if not match: + continue + slug, relative_path, description, use_when = [part.strip() for part in match.groups()] + skill_path = root_path / relative_path + content = skill_path.read_text(encoding="utf-8") if skill_path.exists() else "" + name = _front_matter_value(content, "name") or slug + skills.append( + Skill( + slug=slug, + name=name, + description=description, + use_when=use_when, + content=content, + path=skill_path, + ) + ) + seen.add(slug) + + for skill_path in sorted(root_path.glob("*/SKILL.md")): + slug = skill_path.parent.name + if slug in seen: + continue + content = skill_path.read_text(encoding="utf-8") + skills.append( + Skill( + slug=slug, + name=_front_matter_value(content, "name") or slug, + description=_front_matter_value(content, "description") or "", + use_when="", + content=content, + path=skill_path, + ) + ) + + return skills diff --git a/app/static/app.js b/app/static/app.js new file mode 100644 index 0000000..58c4ee5 --- /dev/null +++ b/app/static/app.js @@ -0,0 +1,72 @@ +const form = document.querySelector("#upload-form"); +const result = document.querySelector("#result"); +const summary = document.querySelector("#summary"); +const skills = document.querySelector("#skills"); +const mdLink = document.querySelector("#download-md"); +const progressBar = document.querySelector("#analysis-progress"); +const statusText = document.querySelector("#analysis-status"); +const button = form.querySelector("button"); + +async function pollTask(statusUrl) { + while (true) { + const response = await fetch(statusUrl); + const payload = await response.json(); + + progressBar.style.width = `${payload.progress || 0}%`; + statusText.textContent = payload.message || "分析中"; + + if (payload.status === "completed") { + return payload; + } + + if (payload.status === "error") { + throw new Error(payload.error || "分析失败"); + } + + await new Promise((resolve) => setTimeout(resolve, 1000)); + } +} + +form.addEventListener("submit", async (event) => { + event.preventDefault(); + button.disabled = true; + button.textContent = "分析中..."; + result.hidden = true; + progressBar.style.width = "0%"; + statusText.textContent = "任务提交中..."; + + const data = new FormData(form); + if (!data.has("use_model")) { + data.set("use_model", "false"); + } + + try { + const response = await fetch("/analyze", { + method: "POST", + body: data, + }); + const payload = await response.json(); + if (!response.ok) { + throw new Error(payload.detail || "分析失败"); + } + + result.hidden = false; + const task = await pollTask(payload.status_url); + summary.textContent = task.summary; + skills.innerHTML = ""; + task.matched_skills.forEach((name) => { + const item = document.createElement("span"); + item.textContent = name; + skills.appendChild(item); + }); + mdLink.href = task.downloads.markdown; + } catch (error) { + summary.textContent = error.message; + skills.innerHTML = ""; + result.hidden = false; + statusText.textContent = "分析失败"; + } finally { + button.disabled = false; + button.textContent = "开始分析"; + } +}); diff --git a/app/static/styles.css b/app/static/styles.css new file mode 100644 index 0000000..31601e4 --- /dev/null +++ b/app/static/styles.css @@ -0,0 +1,218 @@ +:root { + color-scheme: light; + font-family: "Inter", "Segoe UI", Arial, sans-serif; + background: #f5f7fb; + color: #172033; +} + +* { + box-sizing: border-box; +} + +body { + margin: 0; +} + +.shell { + min-height: 100vh; + display: grid; + place-items: center; + padding: 32px; +} + +.panel { + width: min(880px, 100%); + background: #ffffff; + border: 1px solid #dbe2ee; + border-radius: 8px; + box-shadow: 0 20px 60px rgba(23, 32, 51, 0.08); + padding: 32px; +} + +.header { + display: flex; + justify-content: space-between; + gap: 24px; + align-items: flex-start; + margin-bottom: 28px; +} + +.eyebrow { + margin: 0 0 8px; + color: #5d6d83; + font-size: 13px; + letter-spacing: 0; +} + +h1, +h2 { + margin: 0; +} + +h1 { + font-size: 32px; + font-weight: 700; +} + +h2 { + font-size: 20px; +} + +.meta { + display: flex; + gap: 8px; + flex-wrap: wrap; + justify-content: flex-end; +} + +.meta span, +.skills span { + border: 1px solid #cfd8e6; + border-radius: 6px; + padding: 6px 9px; + color: #44546a; + background: #f8fafc; + font-size: 13px; +} + +.form { + display: grid; + gap: 20px; +} + +.drop-zone { + border: 1px dashed #8aa1bd; + border-radius: 8px; + padding: 34px; + background: #fbfcfe; + display: grid; + gap: 8px; + cursor: pointer; +} + +.drop-zone input { + width: 100%; +} + +.drop-title { + font-size: 18px; + font-weight: 700; +} + +.drop-subtitle { + color: #5d6d83; +} + +.controls { + display: grid; + grid-template-columns: 1fr auto; + gap: 16px; + align-items: end; +} + +label { + display: grid; + gap: 8px; + color: #344054; + font-size: 14px; +} + +select { + height: 42px; + border: 1px solid #cfd8e6; + border-radius: 6px; + padding: 0 12px; + background: #ffffff; +} + +.checkbox { + display: flex; + align-items: center; + gap: 8px; + height: 42px; +} + +button, +.downloads a { + border: 0; + border-radius: 6px; + background: #1f6feb; + color: #ffffff; + min-height: 44px; + padding: 0 18px; + font-weight: 700; + cursor: pointer; + text-decoration: none; + display: inline-flex; + align-items: center; + justify-content: center; +} + +button:disabled { + background: #8aa1bd; + cursor: wait; +} + +.result { + margin-top: 28px; + border-top: 1px solid #dbe2ee; + padding-top: 24px; +} + +.progress-wrap { + display: grid; + gap: 8px; + margin: 14px 0 18px; +} + +.progress-track { + height: 10px; + border-radius: 999px; + background: #e8edf4; + overflow: hidden; +} + +.progress-bar { + height: 100%; + border-radius: inherit; + background: linear-gradient(90deg, #1f6feb, #4f8df5); + transition: width 0.2s ease; +} + +.status-text { + margin: 0; + color: #5d6d83; + font-size: 14px; +} + +.skills, +.downloads { + display: flex; + flex-wrap: wrap; + gap: 8px; + margin-top: 16px; +} + +.downloads a:last-child { + background: #334155; +} + +@media (max-width: 680px) { + .shell { + padding: 16px; + } + + .panel { + padding: 22px; + } + + .header, + .controls { + grid-template-columns: 1fr; + display: grid; + } + + .meta { + justify-content: flex-start; + } +} diff --git a/app/templates/index.html b/app/templates/index.html new file mode 100644 index 0000000..d1e1249 --- /dev/null +++ b/app/templates/index.html @@ -0,0 +1,67 @@ + + + + + + GJB438C DOCX 规范分析 + + + +
+
+
+
+

GJB438C / GJB2786

+

DOCX 规范分析

+
+
+ {{ skill_count }} 项技能 + 默认 {{ default_provider }} +
+
+ +
+ + +
+ + +
+ + +
+ + +
+
+ + + diff --git a/plant.md b/plant.md new file mode 100644 index 0000000..2fef089 --- /dev/null +++ b/plant.md @@ -0,0 +1,75 @@ +# 内网离线移植步骤 + +## 1. 迁移原则 + +不要直接拷贝现有 `.venv` 到内网服务器。Ubuntu 20.04 默认 Python 版本偏低，而本项目要求 `python >= 3.10`，且虚拟环境对路径和系统 ABI 敏感。推荐迁移“项目源代码 + `uv.lock` + 离线 wheel 包 + Python 运行时”。 + +## 2. 在线准备机打包 + +在一台可联网机器上完成一次完整解析与下载： + +```bash +uv sync +uv export --frozen --no-dev -o requirements.txt +uv pip download -r requirements.txt -d wheelhouse +``` + +同时打包以下内容： + +- `app/` +- `scripts/` +- `tests/` +- `GJB438C-2021_prd_skills/` +- `configs/` +- `pyproject.toml` +- `uv.lock` +- `requirements.txt` +- `wheelhouse/` +- `uv` 可执行文件（Linux x86_64） + +如果目标机没有 Python 3.10+，还要额外准备一个可移植的 Python 3.10/3.12 运行时目录。 + +## 3. 目标机落地 + +将打包内容拷贝到 Ubuntu 20.04 内网服务器，例如 `/opt/docx-analyzer`。 + +```bash +cd /opt/docx-analyzer +chmod +x uv +``` + +如果系统已有 Python 3.10+： + +```bash +UV_CACHE_DIR=/opt/docx-analyzer/.uv-cache ./uv sync --frozen --offline +``` + +如果系统没有满足版本的 Python： + +```bash +./uv venv .venv --python /opt/python/bin/python3.12 +UV_CACHE_DIR=/opt/docx-analyzer/.uv-cache ./uv sync --frozen --offline +``` + +## 4. 启动与验证 + +启动服务： + +```bash +UV_CACHE_DIR=/opt/docx-analyzer/.uv-cache ./uv run uvicorn app.main:app --host 0.0.0.0 --port 8010 +``` + +验证命令： + +```bash +UV_CACHE_DIR=/opt/docx-analyzer/.uv-cache ./uv run pytest +curl http://127.0.0.1:8010/ +``` + +分析样例文件时，确认 `configs/api_config.yaml` 中的 `intranet` 配置指向内网模型地址，且网络可访问。 + +## 5. 运维建议 + +- 用 `systemd` 把 `uvicorn` 注册成服务。 +- 把 `uploads/`、`outputs/`、`.uv-cache/` 放在独立目录并定期清理。 +- 首次上线先用 `use_model=false` 验证本地流程，再切到 `intranet` 模型。 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..97d580d --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,25 @@ +[project] +name = "linux-format-docs-check" +version = "0.1.0" +description = "Offline-friendly DOCX compliance analysis web app for GJB438C skill documents." +readme = "plan.md" +requires-python = ">=3.10" +dependencies = [ + "fastapi>=0.116.0", + "jinja2>=3.1.0", + "python-docx>=1.1.0", + "python-multipart>=0.0.9", + "pyyaml>=6.0.0", + "requests>=2.32.0", + "uvicorn>=0.35.0", +] + +[dependency-groups] +dev = [ + "httpx>=0.28.0", + "pytest>=8.3.0", +] + +[tool.pytest.ini_options] +testpaths = ["tests"] +pythonpath = ["."] diff --git a/scripts/analyze_docx.py b/scripts/analyze_docx.py new file mode 100644 index 0000000..4420ceb --- /dev/null +++ b/scripts/analyze_docx.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +ROOT_DIR = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT_DIR)) + +from app.main import analyze_saved_docx + + +def main() -> None: + parser = argparse.ArgumentParser(description="Analyze a DOCX file and generate downloadable reports.") + parser.add_argument("docx", type=Path, help="Path to the DOCX file.") + parser.add_argument("--provider", default=None, help="Provider name from configs/api_config.yaml.") + parser.add_argument("--use-model", action="store_true", help="Call the configured LLM API.") + args = parser.parse_args() + + result = analyze_saved_docx(args.docx, provider=args.provider, use_model=args.use_model) + print(f"source: {result['source_filename']}") + print(f"summary: {result['summary']}") + print("matched_skills:") + for skill in result["matched_skills"]: + print(f"- {skill}") + print("downloads:") + print(f"- docx: {result['downloads']['docx']}") + print(f"- markdown: {result['downloads']['markdown']}") + + +if __name__ == "__main__": + main() diff --git a/tests/test_analyzer.py b/tests/test_analyzer.py new file mode 100644 index 0000000..bcb6b90 --- /dev/null +++ b/tests/test_analyzer.py @@ -0,0 +1,71 @@ +from app.analyzer import LLMClient, build_analysis_prompt, select_relevant_skills +from app.config import ProviderConfig +from app.docx_parser import ParsedDocument +from app.skill_loader import Skill + + +class DummyResponse: + def raise_for_status(self) -> None: + return None + + def json(self) -> dict: + return {"choices": [{"message": {"content": "分析结果"}}]} + + +class DummySession: + def __init__(self) -> None: + self.calls = [] + + def post(self, url: str, **kwargs): + self.calls.append((url, kwargs)) + return DummyResponse() + + +def test_select_relevant_skills_prefers_matching_text() -> None: + parsed = ParsedDocument( + filename="srs.docx", + text="软件需求规格说明 CSCI 能力需求 接口需求 合格性规定", + paragraphs=[], + headings=[], + tables=[], + ) + skills = [ + Skill("requirements", "软件需求规格说明", "需求文档", "编写软件需求", "需求能力接口", None), + Skill("deployment", "部署准备", "部署文档", "软件部署交付", "安装部署", None), + ] + + selected = select_relevant_skills(parsed, skills, max_skills=1) + + assert [skill.slug for skill in selected] == ["requirements"] + + +def test_llm_client_posts_openai_compatible_payload() -> None: + provider = ProviderConfig( + api_key="secret", + base_url="https://api.deepseek.com/v1/chat/completions", + max_tokens=99, + model="deepseek-chat", + temperature=0.1, + ) + session = DummySession() + client = LLMClient(provider, session=session) + + content = client.complete("检查文档") + + assert content == "分析结果" + url, kwargs = session.calls[0] + assert url == "https://api.deepseek.com/v1/chat/completions" + assert kwargs["headers"]["Authorization"] == "Bearer secret" + assert kwargs["json"]["model"] == "deepseek-chat" + assert kwargs["json"]["messages"][0]["content"] == "检查文档" + + +def test_build_analysis_prompt_contains_document_and_skill() -> None: + parsed = ParsedDocument("a.docx", "正文", [], [], []) + skill = Skill("s1", "技能一", "描述", "使用条件", "规范内容", None) + + prompt = build_analysis_prompt(parsed, [skill]) + + assert "a.docx" in prompt + assert "技能一" in prompt + assert "符合项" in prompt diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..9d353f1 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,47 @@ +from pathlib import Path + +from app.config import load_api_config + + +def test_load_api_config_selects_default_provider(tmp_path: Path) -> None: + config_path = tmp_path / "api_config.yaml" + config_path.write_text( + """ +default_provider: intranet +providers: + intranet: + api_key: EMPTY + base_url: http://model.local/v1 + max_tokens: 1024 + model: qwen3-coder + temperature: 0.2 +""".strip(), + encoding="utf-8", + ) + + settings = load_api_config(config_path) + + assert settings.provider_name == "intranet" + assert settings.provider.model == "qwen3-coder" + assert settings.provider.chat_completions_url == "http://model.local/v1/chat/completions" + + +def test_load_api_config_accepts_full_chat_completions_url(tmp_path: Path) -> None: + config_path = tmp_path / "api_config.yaml" + config_path.write_text( + """ +default_provider: deepseek +providers: + deepseek: + api_key: key + base_url: https://api.deepseek.com/v1/chat/completions + max_tokens: 4000 + model: deepseek-chat + temperature: 0.7 +""".strip(), + encoding="utf-8", + ) + + settings = load_api_config(config_path, provider_name="deepseek") + + assert settings.provider.chat_completions_url == "https://api.deepseek.com/v1/chat/completions" diff --git a/tests/test_docx_parser.py b/tests/test_docx_parser.py new file mode 100644 index 0000000..01988af --- /dev/null +++ b/tests/test_docx_parser.py @@ -0,0 +1,24 @@ +from pathlib import Path + +from docx import Document + +from app.docx_parser import parse_docx + + +def test_parse_docx_extracts_headings_paragraphs_and_tables(tmp_path: Path) -> None: + docx_path = tmp_path / "sample.docx" + document = Document() + document.add_heading("软件需求规格说明", level=1) + document.add_paragraph("本文档描述 CSCI 的能力需求和接口需求。") + table = document.add_table(rows=1, cols=2) + table.cell(0, 0).text = "需求编号" + table.cell(0, 1).text = "REQ-001" + document.save(docx_path) + + parsed = parse_docx(docx_path) + + assert parsed.filename == "sample.docx" + assert "软件需求规格说明" in parsed.text + assert "REQ-001" in parsed.text + assert parsed.headings[0].text == "软件需求规格说明" + assert parsed.tables[0][0] == ["需求编号", "REQ-001"] diff --git a/tests/test_report_generator.py b/tests/test_report_generator.py new file mode 100644 index 0000000..34eb86b --- /dev/null +++ b/tests/test_report_generator.py @@ -0,0 +1,29 @@ +from pathlib import Path + +from docx import Document + +from app.report_generator import AnalysisReport, generate_docx_report, generate_markdown_report + + +def test_generate_reports_create_markdown_and_docx(tmp_path: Path) -> None: + report = AnalysisReport( + source_filename="sample.docx", + provider_name="deepseek", + model_name="deepseek-chat", + matched_skills=["gjb438c-software-requirements-spec-structure"], + summary="部分通过", + findings=[{"status": "需整改", "item": "缺少追溯性矩阵", "evidence": "未发现相关章节"}], + recommendations=["补充需求双向追溯矩阵。"], + raw_model_output="模型输出", + ) + + md_path = generate_markdown_report(report, tmp_path) + docx_path = generate_docx_report(report, tmp_path) + + assert md_path.exists() + assert "缺少追溯性矩阵" in md_path.read_text(encoding="utf-8") + assert docx_path.exists() + parsed = Document(docx_path) + text = "\n".join(p.text for p in parsed.paragraphs) + assert "DOCX 规范分析报告" in text + assert "部分通过" in text diff --git a/tests/test_skill_loader.py b/tests/test_skill_loader.py new file mode 100644 index 0000000..a9036fd --- /dev/null +++ b/tests/test_skill_loader.py @@ -0,0 +1,14 @@ +from pathlib import Path + +from app.skill_loader import load_skill_catalog + + +def test_load_skill_catalog_reads_index_and_skill_files() -> None: + skills = load_skill_catalog(Path("GJB438C-2021_prd_skills")) + + assert len(skills) >= 30 + skill_names = {skill.slug for skill in skills} + assert "gjb438c-software-requirements-spec-structure" in skill_names + target = next(skill for skill in skills if skill.slug == "gjb438c-software-requirements-spec-structure") + assert "软件需求规格说明" in target.content + assert target.path.name == "SKILL.md" diff --git a/tests/test_web.py b/tests/test_web.py new file mode 100644 index 0000000..7b17ed6 --- /dev/null +++ b/tests/test_web.py @@ -0,0 +1,55 @@ +from pathlib import Path + +from docx import Document + +from app.main import OUTPUT_DIR, ROOT_DIR, analyze_saved_docx + + +def test_index_template_contains_upload_ui() -> None: + html = (ROOT_DIR / "app" / "templates" / "index.html").read_text(encoding="utf-8") + js = (ROOT_DIR / "app" / "static" / "app.js").read_text(encoding="utf-8") + + assert "DOCX 规范分析" in html + assert 'type="file"' in html + assert "analysis-progress" in html + assert "analysis-status" in html + assert "下载 Markdown 报告" in html + assert "