# SSP 文件解析器 Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 實作「SSP docx parser」v1A — 純規則層 + 同步處理 + 拖拉指派 + 三入口共用 wizard，無 LLM。 **Architecture:** 主專案內獨立寫 `DocxParserCore` 與 `IWriteStrategy`（不動 jedi-oscal 套件）；新增 `ssp_docx_parse_jobs` 表暫存 parse 結果；沿用既有 `validate_import` / `confirm_import` 切點 pattern；FE 用 PrimeVue + Pinia + vuedraggable 實作 3 步驟 wizard 與拖拉 UI。 **Tech Stack:** Python 3.11 / Flask-RESTful / SQLAlchemy / python-docx / rapidfuzz；Vue 3 / TypeScript / PrimeVue / Pinia / vuedraggable **Source specs:** - SA: [`api-spec.md`](./api-spec.md) — 23 條 AC、4 個 endpoint、17 個 error code - SD: [`design.md`](./design.md) — DDD 架構、演算法、DB schema、Strategy pattern - FE: [`frontend-spec.md`](./frontend-spec.md) — Vue 元件樹、Pinia store、wizard UI **Repos:** - BE：`/Users/chouraymond/Projects/Billows/Audit-Manager/compliance-manager-be/` - FE：`/Users/chouraymond/Projects/Billows/Audit-Manager/compliance-manager-fe/` - E2E：`/Users/chouraymond/Projects/Billows/Audit-Manager/compliance-manager-test/` --- ## Phase Overview | Phase | 範圍 | Repo | 輸出 | |-------|------|------|------| | **A** | BE — DB migration + ORM model + Entity + Repo + Mapper | BE | 1 表 + 1 model + 1 entity + 1 repo | | **B** | BE — Parser core + ControlIdMatcher utility | BE | DocxParserCore + framework_patterns + matcher | | **C** | BE — Domain service for parse job + WriteStrategy（Module / SSP） | BE | 1 domain service + 2 strategies | | **D** | BE — App service + Route + Serializer + DI container | BE | App service + 3 routes + serializers + DI wiring | | **E** | BE — Error code + InformationSystemService.upsert_by_name + 整合測試 | BE | error codes + helper + integration test | | **F** | FE — Service + Pinia store + Dialog shell | FE | Service + store + top-level dialog | | **G** | FE — Step 0 / Step 1 / Step 2 元件 + diff preview | FE | 3 個 step 元件 + MatchedControlDiff | | **H** | FE — Drag-Drop + Conflict modal + Batch toolbar | FE | 拖拉 UI + 衝突視窗 + 批次操作 | | **I** | FE — 整合到 ModuleFrame.vue / SSP 控制項頁 + i18n | FE | 三入口按鈕 + i18n | | **J** | 跨 repo smoke test + 移交 Phase 4（test-plan） | BE / FE | 端到端 manual test 通過 | **完成準則**： - BE: 全部 pytest 通過（含 fixture docx 整合測試） - FE: 三個入口可開 dialog → 上傳 → 預覽 → 確認，dev server 顯示正確 - E2E: Phase 4 的 test-plan.md 涵蓋率充足，但實際 E2E 跑由 Phase 5 後段處理 --- ## §0 Pre-implementation Verifications & Cross-cutting Constraints > **必讀**：以下事項在 Phase A 開始前已被 verify，是整份 plan 的隱含前提。實作時若發現與下述不符，停下來回報。 ### V-1：DI 結構（OscalContainer flat pattern） **verified**：本專案所有 oscal 相關 service 都直接掛在 `OscalContainer` 上，**不另建 sub-container**。例如 `ssp_control_impl_import_service` 就是 `OscalContainer` 的 provider，inject path 是 `Containers.oscal_container.ssp_control_impl_import_service`。 > 因此本 plan 不另建 `SspDocxImportContainer`（修正 SD §5 的設計）。 > 新增的 provider 全部直接加在 `di_containers/oscal/oscal_containers.py` 的 `OscalContainer` 內。 新 providers 命名（加在 OscalContainer 既有 providers 後）： - `ssp_docx_parse_job_repo` - `ssp_docx_parse_job_domain_service` - `docx_parser_core` - `ssp_docx_module_frame_write_strategy` - `ssp_docx_ssp_write_strategy` - `ssp_docx_import_app_service` Route 用：`Containers.oscal_container.ssp_docx_import_app_service` ### V-2：jedi-oscal 既有 method 名稱 **verified** via reading `~/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/domain/services/ssp/control_implementation_domain_service.py`： `ControlImplementationDomainService` 提供（已 verify）： - `get_one(_filter)`、`get_by_uid(uid)`、`get_by_id(id)`、`get_by_ssp_and_identifier(ssp_id, control_identifier)`、`get_all_by_ssp_id(ssp_id)` - `add(_entity)`、`update(_entity)` - `delete_by_id(id)`、`delete_by_uid(uid)`、`delete_by_ids(ids)`、`delete_by_uids(uids)` **沒有** `update_or_create` / `upsert` — Strategy 必須**自己組 get + (add or update) 兩段邏輯**。 Objective service 同樣 shape（`add` / `update`，無 upsert）。 不修改 jedi-oscal 套件（CLAUDE.md 規範）。 ### V-3：BaseModel 提供欄位 **verified**：`from jedi_common.session.database.model.base_model import BaseModel` 提供： ```python id: Mapped[int] # primary key, autoincrement created_at: Mapped[datetime] # default now() updated_at: Mapped[datetime] # default now() + onupdate now() created_user: Mapped[Optional[str]] # 50 char updated_user: Mapped[Optional[str]] ``` `tenant_id` 從 `TenantScopedMixinModel` 提供（位於同 jedi-common path）。 新表 `SspDocxParseJob` 不需自己定 id/created_*/updated_*/tenant_id，繼承即可。 ### V-4：測試目錄與 conftest **verified**：測試目錄是 `tests/`（plural，**不是** `test/`），既有： - `tests/conftest.py`（root）— session-scoped app/client/token fixtures + function-scoped headers - `tests/cloud_integration/conftest.py`（module-scoped） 依記憶 `tests/DEPENDENCIES.md`： - 既有 fixture：`app`, `client`, `token`, `headers`, `tenant_id` - 測試帳號：blsit / Billows@123!，tenant_id=102, org_unit_id=99, is_admin=False - Response 格式：`{"status": true/false, "data": {...}}` 整合測試直接用既有 fixture： ```python def test_xxx(client, headers, tenant_id): res = client.post("/api/1.0/ssp-docx-imports/parse", ...) ``` ### V-5：Route registration via create_module **verified**：`api/oscal/__init__.py` 內有 `create_module()` 函式： 1. import 所有 route classes 2. 用 `api.add_resource(RouteClass, '/path')` 逐個註冊 3. 回傳 Blueprint 新 route 加入流程： 1. 新增 `api/oscal/routes/ssp/ssp_docx_import_route.py`（含 3 個 Route class） 2. 在 `api/oscal/__init__.py` 的 `create_module()` 內 import + add_resource 3. **不另外註冊**（既有 `config/di_modules.py` 自動掃 `api/**/routes/*_route.py`） ### V-6：jedi_information_system 套件位置 **verified**：`jedi_information_system/` 目錄在主專案 root（**不是** 外部套件）。依記憶「暫置主專案的模組」原則，可直接修改，不需走 jedi 進版流程。 修改 `jedi_information_system/app/service/information_system_service.py` 加 `upsert_by_name` **不需** user 額外確認（不是真正的外部套件）。 ### V-7：跨層 file 路徑速查 | 概念 | 路徑 | |------|------| | 主專案 BE root | `/Users/chouraymond/Projects/Billows/Audit-Manager/compliance-manager-be/` | | jedi-oscal 套件 source | `~/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/` | | jedi_information_system | 主專案內 `jedi_information_system/`（in-tree） | | 客戶 fixture docx | `docs/features/FR-022-2604-ssp-doc-parser/raw-requirement/reference/ASIA-CMMC-SSP-DRAFT-202604.docx` | | FE root | `~/Projects/Billows/Audit-Manager/compliance-manager-fe/` | | E2E root | `~/Projects/Billows/Audit-Manager/compliance-manager-test/` | ### V-8：Changelog 強制要求 **每完成一個 Phase（A/B/C/D/E/F/G/H/I/J）就建立一份 changelog**： ``` docs/changelog/2026-XX-XX-feat-ssp-docx-phase--.md ``` frontmatter: ```yaml --- type: feat modules: [ssp-doc-parser, oscal] issue: docs/features/FR-022-2604-ssp-doc-parser/ # link to feature folder --- ``` 每個 Phase 最後一個 task 強制加： - [ ] **Step F: 寫 Phase X changelog 並 commit** --- ## Pre-flight 確認本機環境： - [ ] **Pre-1: 確認在 `fix/survey-answer-duplicate-issue` 或新建 feature branch** - 建議：`git checkout -b feat/ssp-doc-parser` 從 main 分出 - [ ] **Pre-2: BE 環境 OK** ```bash cd /Users/chouraymond/Projects/Billows/Audit-Manager/compliance-manager-be poetry install pytest tests/ -q # 既有測試應全 pass ``` - [ ] **Pre-3: FE 環境 OK** ```bash cd /Users/chouraymond/Projects/Billows/Audit-Manager/compliance-manager-fe npm install npm run dev # 確認可啟動 ``` - [ ] **Pre-4: dev DB 可連線** ```bash psql -h localhost -p 25432 -U cmmgr -d compliance_manager -c "SELECT 1;" ``` - [ ] **Pre-5: 確認 fixture docx 可開** ```bash python3 -c "from docx import Document; d = Document('docs/features/FR-022-2604-ssp-doc-parser/raw-requirement/reference/ASIA-CMMC-SSP-DRAFT-202604.docx'); print(len(d.paragraphs))" # Expected: 273 ``` --- # Phase A — BE Foundation (DB + ORM) 依 SD §4.1 建立 `ssp_docx_parse_jobs` 表與對應 ORM。 ### Task A.1: SQL Migration **Files:** - Create: `scripts/sql/ssp_docx_parse_jobs_migration.sql` - [ ] **Step 1: 建立 migration 檔** 完整內容見 SD §4.1。檔案開頭 `-- Date: 2026-04-30`，每條語句加日期註解。包含： - `CREATE TABLE oscal.ssp_docx_parse_jobs (...)` - 3 個 index - RLS policy - `GRANT SELECT, INSERT, UPDATE, DELETE ON oscal.ssp_docx_parse_jobs TO cm_app;` - `GRANT USAGE, SELECT ON SEQUENCE oscal.ssp_docx_parse_jobs_id_seq TO cm_app;` - [ ] **Step 2: 用 cmmgr 跑 migration** ⚠️ 依記憶 `feedback_sql_migration_use_cmmgr.md`，用 cmmgr 執行（cm_app 受 RLS 擋寫不進來）： ```bash psql -h localhost -p 25432 -U cmmgr -d compliance_manager -f scripts/sql/ssp_docx_parse_jobs_migration.sql ``` 預期輸出： ``` CREATE TABLE CREATE INDEX (×3) ALTER TABLE CREATE POLICY GRANT (×2) ``` - [ ] **Step 3: 驗證** ```bash psql -h localhost -p 25432 -U cmmgr -d compliance_manager -c "\d oscal.ssp_docx_parse_jobs" ``` 確認 18 個欄位（id / uid / tenant_id / source_type / source_uid / mode / framework / status / file_path / file_name / file_size / parsed_result / error_code / error_message / import_summary / is_active / created_at / created_user / updated_at / updated_user）皆在。 - [ ] **Step 4: Commit** ```bash git add scripts/sql/ssp_docx_parse_jobs_migration.sql git commit -m "feat(ssp-docx): add ssp_docx_parse_jobs migration" ``` --- ### Task A.2: Entity + Query Entity **Files:** - Create: `domain/oscal/entity/ssp_docx_parse_job_entity.py` - [ ] **Step 1: 撰寫 entity** ```python from dataclasses import dataclass, field from datetime import datetime from typing import Optional, Any from jedi_common.session.base.entity import BaseEntity, BaseQueryEntity @dataclass class SspDocxParseJobEntity(BaseEntity): uid: Optional[str] = None tenant_id: Optional[int] = None source_type: Optional[str] = None # 'module_frame' | 'project_ssp' source_uid: Optional[str] = None mode: Optional[str] = None # 'full' | 'statement_only' framework: Optional[str] = None status: Optional[str] = None file_path: Optional[str] = None file_name: Optional[str] = None file_size: Optional[int] = None parsed_result: Optional[dict] = None error_code: Optional[str] = None error_message: Optional[str] = None import_summary: Optional[dict] = None is_active: Optional[bool] = True created_at: Optional[datetime] = None created_user: Optional[str] = None updated_at: Optional[datetime] = None updated_user: Optional[str] = None @dataclass class SspDocxParseJobQueryEntity(BaseQueryEntity): uid: Optional[str] = None tenant_id: Optional[int] = None source_type: Optional[str] = None source_uid: Optional[str] = None status: Optional[str] = None is_active: Optional[bool] = True ``` - [ ] **Step 2: Commit** ```bash git add domain/oscal/entity/ssp_docx_parse_job_entity.py git commit -m "feat(ssp-docx): add SspDocxParseJobEntity and QueryEntity" ``` --- ### Task A.3: Repo Interface **Files:** - Create: `domain/oscal/repository/i_ssp_docx_parse_job_repo.py` - [ ] **Step 1: 撰寫介面** ```python from abc import ABC, abstractmethod from typing import Optional from domain.oscal.entity.ssp_docx_parse_job_entity import ( SspDocxParseJobEntity, SspDocxParseJobQueryEntity ) class ISspDocxParseJobRepo(ABC): @abstractmethod def add(self, entity: SspDocxParseJobEntity) -> SspDocxParseJobEntity: ... @abstractmethod def update(self, id: int, fields: dict, user: str) -> SspDocxParseJobEntity: ... @abstractmethod def get_one(self, query: SspDocxParseJobQueryEntity) -> Optional[SspDocxParseJobEntity]: ... @abstractmethod def deactivate(self, uid: str, user: str) -> None: ... ``` - [ ] **Step 2: Commit** --- ### Task A.4: ORM Model **Files:** - Create: `infra/oscal/model/ssp_docx_parse_job.py` - [ ] **Step 1: 撰寫 SQLAlchemy model** ```python from datetime import datetime from typing import Optional from sqlalchemy import String, Integer, BigInteger, DateTime, Boolean, Text from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column from jedi_common.session.database.base_model import BaseModel from jedi_common.session.database.tenant_scoped import TenantScopedMixinModel class SspDocxParseJob(BaseModel, TenantScopedMixinModel): __tablename__ = "ssp_docx_parse_jobs" __table_args__ = {"schema": "oscal"} uid: Mapped[str] = mapped_column(String(36), unique=True, nullable=False) source_type: Mapped[str] = mapped_column(String(20), nullable=False) source_uid: Mapped[str] = mapped_column(String(36), nullable=False) mode: Mapped[str] = mapped_column(String(20), nullable=False) framework: Mapped[str] = mapped_column(String(40), nullable=False) status: Mapped[str] = mapped_column(String(20), nullable=False, default="pending") file_path: Mapped[Optional[str]] = mapped_column(String(255)) file_name: Mapped[Optional[str]] = mapped_column(String(255)) file_size: Mapped[Optional[int]] = mapped_column(BigInteger) parsed_result: Mapped[Optional[dict]] = mapped_column(JSONB) error_code: Mapped[Optional[str]] = mapped_column(String(40)) error_message: Mapped[Optional[str]] = mapped_column(Text) import_summary: Mapped[Optional[dict]] = mapped_column(JSONB) is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) ``` > 註：`tenant_id` / `id` / `created_at` / `created_user` / `updated_at` / `updated_user` 由 base classes 提供。 - [ ] **Step 2: Commit** --- ### Task A.5: Mapper (Entity ↔ Model) **Files:** - Create: `infra/oscal/mapper/ssp_docx_parse_job_mapper.py` - [ ] **Step 1: 撰寫雙向 mapper** ```python from domain.oscal.entity.ssp_docx_parse_job_entity import SspDocxParseJobEntity from infra.oscal.model.ssp_docx_parse_job import SspDocxParseJob class SspDocxParseJobMapper: @staticmethod def model_to_entity(m: SspDocxParseJob) -> SspDocxParseJobEntity: return SspDocxParseJobEntity( id=m.id, uid=m.uid, tenant_id=m.tenant_id, source_type=m.source_type, source_uid=m.source_uid, mode=m.mode, framework=m.framework, status=m.status, file_path=m.file_path, file_name=m.file_name, file_size=m.file_size, parsed_result=m.parsed_result, error_code=m.error_code, error_message=m.error_message, import_summary=m.import_summary, is_active=m.is_active, created_at=m.created_at, created_user=m.created_user, updated_at=m.updated_at, updated_user=m.updated_user, ) @staticmethod def entity_to_model(e: SspDocxParseJobEntity) -> SspDocxParseJob: return SspDocxParseJob( uid=e.uid, tenant_id=e.tenant_id, source_type=e.source_type, source_uid=e.source_uid, mode=e.mode, framework=e.framework, status=e.status, file_path=e.file_path, file_name=e.file_name, file_size=e.file_size, parsed_result=e.parsed_result, error_code=e.error_code, error_message=e.error_message, import_summary=e.import_summary, is_active=e.is_active if e.is_active is not None else True, created_user=e.created_user, updated_user=e.updated_user, ) ``` - [ ] **Step 2: Commit** --- ### Task A.6: Repo Implementation **Files:** - Create: `infra/oscal/repository/ssp_docx_parse_job_repo_impl.py` - [ ] **Step 1: 撰寫 repo impl** 繼承 `BaseRepositoryImpl`（依 CLAUDE.md DDD 規範，session 自動 lazy）： ```python from typing import Optional from jedi_common.session.database.repository.base_repository_impl import BaseRepositoryImpl from domain.oscal.entity.ssp_docx_parse_job_entity import ( SspDocxParseJobEntity, SspDocxParseJobQueryEntity ) from domain.oscal.repository.i_ssp_docx_parse_job_repo import ISspDocxParseJobRepo from infra.oscal.mapper.ssp_docx_parse_job_mapper import SspDocxParseJobMapper from infra.oscal.model.ssp_docx_parse_job import SspDocxParseJob class SspDocxParseJobRepoImpl( ISspDocxParseJobRepo, BaseRepositoryImpl[SspDocxParseJobEntity, SspDocxParseJobQueryEntity, SspDocxParseJob, SspDocxParseJobMapper] ): def __init__(self): super().__init__(mapper=SspDocxParseJobMapper, model=SspDocxParseJob) def deactivate(self, uid: str, user: str) -> None: m = self.session.query(SspDocxParseJob).filter_by(uid=uid).first() if m: m.is_active = False m.updated_user = user self.session.flush() ``` (其他 add / update / get_one 從 BaseRepositoryImpl 繼承) - [ ] **Step 2: 寫整合測試（用實際 DB）** ```python # tests/test_ssp_docx_parse_job_repo.py import pytest from datetime import datetime from infra.oscal.repository.ssp_docx_parse_job_repo_impl import SspDocxParseJobRepoImpl from domain.oscal.entity.ssp_docx_parse_job_entity import SspDocxParseJobEntity, SspDocxParseJobQueryEntity from jedi_common.session.database.db import session_scope def test_add_and_get_parse_job(): repo = SspDocxParseJobRepoImpl() with session_scope(): entity = SspDocxParseJobEntity( uid="test-uid-001", tenant_id=102, source_type="project_ssp", source_uid="ssp-test-001", mode="statement_only", framework="cmmc-l1", status="pending", created_user="test", updated_user="test", ) added = repo.add(entity) assert added.id is not None fetched = repo.get_one(SspDocxParseJobQueryEntity(uid="test-uid-001")) assert fetched.framework == "cmmc-l1" ``` ⚠️ 依 V-4：用既有 `tests/conftest.py` 提供的 `app` / `client` / `headers` fixtures。test 內必須在 `session_scope()` 或 `@transaction` 內呼叫 repo（CLAUDE.md 規範）。RLS 設定需要 session 層級的 `app.user_id` / `app.allowed_tenant_paths` — 既有 conftest 應已透過 JWT decode + session_scope 設好。 若直接呼叫 repo 時 INSERT 0 rows 靜默失敗（依記憶 `feedback_sql_migration_use_cmmgr.md` 類似情境），可改用 client 觸發 API 路徑進行整合測試（Task E.3 走此模式）。 - [ ] **Step 3: 跑測試** ```bash pytest tests/test_ssp_docx_parse_job_repo.py -v ``` 預期：1 pass - [ ] **Step 4: Commit** ```bash git add infra/oscal/repository/ssp_docx_parse_job_repo_impl.py \ tests/test_ssp_docx_parse_job_repo.py git commit -m "feat(ssp-docx): SspDocxParseJobRepoImpl + integration test" ``` --- # Phase B — BE Parser Core 實作 SD §2 的 DocxParserCore 與 §2.2 / 2.3 / common utility。 ### Task B.1: ControlIdMatcher utility **Files:** - Create: `common/util/control_id_matcher.py` - Test: `tests/test_control_id_matcher.py` - [ ] **Step 1: 撰寫測試（TDD）** ```python # tests/test_control_id_matcher.py import pytest from common.util.control_id_matcher import ( parse_control_id, fuzzy_match_control, detect_dominant_framework ) from domain.oscal.parser.framework_patterns import FRAMEWORK_PATTERNS def test_parse_cmmc_l1_id(): assert parse_control_id("AC.L1-b.1.i", "cmmc-l1") == "AC.L1-b.1.i" assert parse_control_id("Control AC.L1-b.1.iv [FCI DATA]", "cmmc-l1") == "AC.L1-b.1.iv" def test_parse_cmmc_l1_returns_none_for_invalid(): assert parse_control_id("Hello world", "cmmc-l1") is None def test_parse_nist_800_171_id(): assert parse_control_id("3.1.5", "nist-800-171") == "3.1.5" def test_fuzzy_match_returns_best_candidate(): candidates = [ {"control_id": "AC.L1-b.1.i", "control_name": "Authorized Access Control"}, {"control_id": "AC.L1-b.1.ii", "control_name": "Transaction Function Control"}, ] best_id, score = fuzzy_match_control("authorized access control", candidates) assert best_id == "AC.L1-b.1.i" assert score >= 0.7 def test_detect_dominant_framework_cmmc_l1(): headings = ["AC.L1-b.1.i", "AC.L1-b.1.ii", "SI.L1-b.1.xv", "PE.L1-b.1.viii"] assert detect_dominant_framework(headings) == "cmmc-l1" ``` - [ ] **Step 2: 跑測試確認 fail** ```bash pytest tests/test_control_id_matcher.py -v # Expected: ImportError / ModuleNotFoundError ``` - [ ] **Step 3: 實作** ```python # common/util/control_id_matcher.py import re from typing import Optional from rapidfuzz import fuzz from domain.oscal.parser.framework_patterns import FRAMEWORK_PATTERNS def parse_control_id(text: str, framework: str) -> Optional[str]: """從文字內擷取符合 framework 的 control_id；找不到回 None""" pattern = FRAMEWORK_PATTERNS.get(framework) if pattern is None: return None m = pattern.search(text) return m.group(0) if m else None def fuzzy_match_control(text: str, candidates: list) -> tuple[Optional[str], float]: """ 用 rapidfuzz 找最相似的 control_name；回 (control_id, score [0.0-1.0])。 score 線性映射 fuzz.token_set_ratio 0-100 到 0-1。 """ if not candidates: return None, 0.0 best_id = None best_score = 0.0 for c in candidates: name = c.get("control_name") or "" score = fuzz.token_set_ratio(text.lower(), name.lower()) / 100.0 if score > best_score: best_score = score best_id = c["control_id"] return best_id, best_score def detect_dominant_framework(headings: list[str]) -> Optional[str]: """ 判斷 docx 內主要 framework；找不到 0 個或多 framework 並列時回 None。 回傳的 framework 必須命中比例 ≥ 0.6。 """ if not headings: return None counts = {} for fw, pattern in FRAMEWORK_PATTERNS.items(): n = sum(1 for h in headings if pattern.search(h)) if n > 0: counts[fw] = n / len(headings) if not counts: return None best_fw = max(counts, key=counts.get) if counts[best_fw] >= 0.6: return best_fw return None ``` - [ ] **Step 4: 跑測試確認 pass** ```bash pytest tests/test_control_id_matcher.py -v # Expected: 5 passed ``` - [ ] **Step 5: Commit** ```bash git add common/util/control_id_matcher.py tests/test_control_id_matcher.py git commit -m "feat(ssp-docx): control_id_matcher utility (parse + fuzzy + detect)" ``` --- ### Task B.2: framework_patterns **Files:** - Create: `domain/oscal/parser/framework_patterns.py` - [ ] **Step 1: 實作（無 test，純 const）** ```python # domain/oscal/parser/framework_patterns.py import re FRAMEWORK_PATTERNS = { 'cmmc-l1': re.compile(r'\b([A-Z]{2})\.L1-b\.\d+\.[ivxlcdm]+\b', re.I), 'cmmc-l2': re.compile(r'\b([A-Z]{2})\.L2-\d+\.\d+\.\d+\b', re.I), 'nist-800-171': re.compile(r'\b3\.\d+\.\d+\b'), 'iso27001': re.compile(r'\bA\.\d+(\.\d+)*\b'), } # 觸發 FATAL framework_mismatch 的 threshold FRAMEWORK_DOMINANT_THRESHOLD = 0.6 USER_FRAMEWORK_MIN_THRESHOLD = 0.3 ``` - [ ] **Step 2: Commit** --- ### Task B.3: Docx Intermediate dataclasses **Files:** - Create: `domain/oscal/parser/docx_intermediate.py` - [ ] **Step 1: 撰寫 dataclass** ```python from dataclasses import dataclass, field from typing import Optional @dataclass class CandidateControl: control_id: str control_name: str objective_keys: list[str] = field(default_factory=list) current_implementation_description: Optional[str] = None current_objectives_descriptions: dict[str, str] = field(default_factory=dict) @dataclass class ParsedObjective: objective_key: str parsed_description: str @dataclass class MatchedControl: control_id: str control_name: str score: float score_reason: str parsed_implementation_description: str objectives: list[ParsedObjective] = field(default_factory=list) @dataclass class UnmatchedParagraph: paragraph_idx: int text: str context: dict # { preceding_h1, preceding_h2, preceding_h3 } rule_score: float rule_guess_control_id: Optional[str] @dataclass class ParsedDocx: matched_controls: list[MatchedControl] unmatched_paragraphs: list[UnmatchedParagraph] missing_baseline_controls: list[str] # 控制項 ID predicted_module_frame_controls: list[str] # mode=full only summary: dict warnings: list[str] = field(default_factory=list) ``` - [ ] **Step 2: Commit** --- ### Task B.4: DocxParserCore — `_extract_structure` **Files:** - Create: `domain/oscal/parser/docx_parser_core.py` - Test: `tests/test_docx_parser_core_structure.py` > ⚠️ 為了測試方便，先做純解析（不做 matching），再做 matching 邏輯。 - [ ] **Step 1: 寫測試** ```python # tests/test_docx_parser_core_structure.py import pytest from io import BytesIO from docx import Document from domain.oscal.parser.docx_parser_core import DocxParserCore @pytest.fixture def fixture_docx_path(tmp_path): """建立 minimal docx for testing""" path = tmp_path / "minimal.docx" doc = Document() doc.add_heading("Introduction", level=1) doc.add_paragraph("Some intro text") doc.add_heading("Control Implementation", level=1) doc.add_heading("AC.L1-b.1.i — Authorized Access Control", level=3) doc.add_paragraph("We control user groups via AD.") doc.add_heading("AC.L1-b.1.ii", level=3) doc.add_paragraph("Function control description.") doc.save(str(path)) return str(path) def test_extract_structure_counts(fixture_docx_path): parser = DocxParserCore() structure = parser._extract_structure(Document(fixture_docx_path)) assert len(structure.paragraphs) == 6 # heading + body + heading + body + ... h3_paragraphs = [p for p in structure.paragraphs if p.heading_level == 3] assert len(h3_paragraphs) == 2 def test_extract_real_fixture(): parser = DocxParserCore() fixture_path = "docs/features/FR-022-2604-ssp-doc-parser/raw-requirement/reference/ASIA-CMMC-SSP-DRAFT-202604.docx" structure = parser._extract_structure(Document(fixture_path)) h3_count = sum(1 for p in structure.paragraphs if p.heading_level == 3) assert h3_count >= 14 # 至少 14 個控制項 H3（依先前 dump） ``` - [ ] **Step 2: 確認 test fail** - [ ] **Step 3: 實作 `_extract_structure`** ```python # domain/oscal/parser/docx_parser_core.py import re from dataclasses import dataclass from typing import Optional, Iterator from docx import Document from docx.oxml.ns import qn from docx.oxml.text.paragraph import CT_P from docx.oxml.table import CT_Tbl from docx.text.paragraph import Paragraph from docx.table import Table @dataclass class StructureParagraph: paragraph_idx: int text: str heading_level: Optional[int] context: dict # { preceding_h1, preceding_h2, preceding_h3 } @dataclass class StructureTable: after_paragraph_idx: int rows: list[list[str]] # cell texts @dataclass class DocxStructure: paragraphs: list[StructureParagraph] tables: list[StructureTable] class DocxParserCore: HEADING_RE = re.compile(r'^Heading (\d)$') def _is_heading(self, p: Paragraph) -> Optional[int]: m = self.HEADING_RE.match(p.style.name) if m: level = int(m.group(1)) return level if 1 <= level <= 6 else None if p.style.name == 'Title': return 1 return None def _iter_block_items(self, doc: Document) -> Iterator: """yield Paragraph or Table in document order""" body_elm = doc.element.body for child in body_elm.iterchildren(): if isinstance(child, CT_P): yield Paragraph(child, doc) elif isinstance(child, CT_Tbl): yield Table(child, doc) def _extract_structure(self, doc: Document) -> DocxStructure: paragraphs = [] tables = [] ctx_h1 = ctx_h2 = ctx_h3 = None last_para_idx = -1 for idx, item in enumerate(self._iter_block_items(doc)): if isinstance(item, Paragraph): level = self._is_heading(item) if level == 1: ctx_h1, ctx_h2, ctx_h3 = item.text, None, None elif level == 2: ctx_h2, ctx_h3 = item.text, None elif level == 3: ctx_h3 = item.text paragraphs.append(StructureParagraph( paragraph_idx=len(paragraphs), text=item.text, heading_level=level, context={ 'preceding_h1': ctx_h1, 'preceding_h2': ctx_h2, 'preceding_h3': ctx_h3, } )) last_para_idx = len(paragraphs) - 1 elif isinstance(item, Table): rows = [[cell.text.strip() for cell in row.cells] for row in item.rows] tables.append(StructureTable( after_paragraph_idx=last_para_idx, rows=rows, )) return DocxStructure(paragraphs=paragraphs, tables=tables) ``` - [ ] **Step 4: 跑測試** ```bash pytest tests/test_docx_parser_core_structure.py -v # Expected: 2 passed ``` - [ ] **Step 5: Commit** --- ### Task B.5: DocxParserCore — `_match_controls` - [ ] **Step 1: 寫測試** ```python # tests/test_docx_parser_core_match.py import pytest from domain.oscal.parser.docx_parser_core import DocxParserCore from domain.oscal.parser.docx_intermediate import CandidateControl @pytest.fixture def candidates(): return [ CandidateControl( control_id="AC.L1-b.1.i", control_name="Authorized Access Control", objective_keys=["a", "b", "c"], ), CandidateControl( control_id="AC.L1-b.1.ii", control_name="Transaction & Function Control", objective_keys=["a"], ), ] def test_match_h3_with_complete_id(fixture_docx_path, candidates): parser = DocxParserCore() parsed = parser.parse(fixture_docx_path, "cmmc-l1", candidates) assert len(parsed.matched_controls) == 2 ac1 = next(m for m in parsed.matched_controls if m.control_id == "AC.L1-b.1.i") assert ac1.score >= 0.9 assert "AD" in ac1.parsed_implementation_description def test_unmatched_paragraph(fixture_docx_path): parser = DocxParserCore() parsed = parser.parse(fixture_docx_path, "cmmc-l1", candidates=[]) # 沒 candidates → matched_controls 為空，所有有 control_id 的 H3 仍視為「無對應」 assert len(parsed.unmatched_paragraphs) > 0 def test_missing_baseline_control(fixture_docx_path, candidates): candidates.append(CandidateControl( control_id="SI.L1-b.1.xv", # baseline 中有，docx 沒 control_name="System & File Scanning", )) parser = DocxParserCore() parsed = parser.parse(fixture_docx_path, "cmmc-l1", candidates) assert "SI.L1-b.1.xv" in parsed.missing_baseline_controls ``` - [ ] **Step 2: 確認 fail** - [ ] **Step 3: 實作 `parse()` + `_match_controls()`** 依 SD §2.1 algorithm，code 完整： ```python # 接續 docx_parser_core.py from common.util.control_id_matcher import parse_control_id, fuzzy_match_control from domain.oscal.parser.docx_intermediate import ( ParsedDocx, MatchedControl, UnmatchedParagraph, CandidateControl, ParsedObjective ) class DocxParserCore: # ... (前述方法) def parse( self, file_path: str, framework: str, candidates: list[CandidateControl], ) -> ParsedDocx: doc = Document(file_path) structure = self._extract_structure(doc) candidate_dict = {c.control_id: c for c in candidates} matched = [] unmatched = [] used_paragraph_idxs = set() # Pass 1: H3 標題命中 h3_indices = [i for i, p in enumerate(structure.paragraphs) if p.heading_level == 3] for h3_idx_position, h3_idx in enumerate(h3_indices): h3_para = structure.paragraphs[h3_idx] control_id = parse_control_id(h3_para.text, framework) if not control_id: continue used_paragraph_idxs.add(h3_idx) # 收集 H3 之後到下一個 H1/H2/H3 之前的 normal paragraphs next_h_idx = h3_indices[h3_idx_position + 1] if h3_idx_position + 1 < len(h3_indices) else len(structure.paragraphs) content_paras = [] for i in range(h3_idx + 1, next_h_idx): p = structure.paragraphs[i] if p.heading_level is None and p.text.strip(): content_paras.append(p.text) used_paragraph_idxs.add(i) # 收集這個 H3 區塊內的 tables section_tables = [t for t in structure.tables if h3_idx <= t.after_paragraph_idx < next_h_idx] objectives = self._extract_objectives(section_tables, candidate_dict.get(control_id)) matched.append(MatchedControl( control_id=control_id, control_name=candidate_dict.get(control_id, CandidateControl(control_id=control_id, control_name="")).control_name, score=0.95, score_reason="H3 標題完整 ID 命中", parsed_implementation_description="\n".join(content_paras), objectives=objectives, )) # Pass 2: Normal paragraph 內含 control_id (or fuzzy) for i, p in enumerate(structure.paragraphs): if i in used_paragraph_idxs or p.heading_level is not None: continue if not p.text.strip(): continue # 嘗試 control_id 直接命中 cid = parse_control_id(p.text, framework) if cid and cid not in candidate_dict: cid = None if cid: # 視為某個已 matched 的控制項的補充 (append) target = next((m for m in matched if m.control_id == cid), None) if target: target.parsed_implementation_description += "\n" + p.text used_paragraph_idxs.add(i) continue # Fuzzy match best_id, score = fuzzy_match_control(p.text, [ {"control_id": c.control_id, "control_name": c.control_name} for c in candidates ]) if score < 0.4: rule_guess = best_id if score >= 0.4 else None unmatched.append(UnmatchedParagraph( paragraph_idx=i, text=p.text, context=p.context, rule_score=score, rule_guess_control_id=rule_guess, )) # Missing baseline controls matched_ids = {m.control_id for m in matched} baseline_ids = {c.control_id for c in candidates} missing = list(baseline_ids - matched_ids) # Predicted controls (mode=full) predicted = list(matched_ids) # Summary summary = { "total_paragraphs": len(structure.paragraphs), "matched_controls_count": len(matched), "unmatched_paragraphs_count": len(unmatched), "missing_baseline_controls_count": len(missing), "warning_level": self._compute_warning_level(matched, unmatched, missing, candidates), } return ParsedDocx( matched_controls=matched, unmatched_paragraphs=unmatched, missing_baseline_controls=missing, predicted_module_frame_controls=predicted, summary=summary, ) def _extract_objectives(self, tables, candidate): """從 tables 抽出 (a)(b)(c) 評估目標 description""" out = [] for table in tables: for row in table.rows: if len(row) < 2: continue first_cell = row[0] m = re.match(r'^\s*\(([a-z])\)', first_cell) if m: key = m.group(1) if candidate is None or key in candidate.objective_keys: out.append(ParsedObjective( objective_key=key, parsed_description=row[1], )) return out def _compute_warning_level(self, matched, unmatched, missing, candidates): if not candidates: return None match_rate = len(matched) / max(1, len(matched) + len(unmatched)) missing_rate = len(missing) / len(candidates) if match_rate < 0.5 or missing_rate > 0.3: return "partial" return None ``` - [ ] **Step 4: 跑測試** - [ ] **Step 5: Commit** ```bash git add domain/oscal/parser/docx_parser_core.py \ domain/oscal/parser/docx_intermediate.py \ domain/oscal/parser/framework_patterns.py \ tests/test_docx_parser_core_*.py git commit -m "feat(ssp-docx): DocxParserCore — extract structure + match controls" ``` --- ### Task B.5.1: 處理 out-of-baseline 控制項 ID (SD §2.1 step 2 / AC-13 邊界 case) - [ ] **Step 1: 修正測試與行為** 當 docx 內 H3 命中合法 control_id pattern 但 ID **不在 baseline 候選清單**內時，**不能靜默丟棄**，需歸入 `unmatched_paragraphs`： ```python # 在 _match_controls() Pass 1 內： control_id = parse_control_id(h3_para.text, framework) if control_id and control_id in candidate_dict: # 命中且在 baseline → matched matched.append(...) else: # 命中但不在 baseline OR 完全沒命中 → unmatched，rule_guess 帶上 ID unmatched.append(UnmatchedParagraph( paragraph_idx=h3_idx, text=h3_para.text + "\n" + "\n".join(content_paras), # 含下方段落 context=h3_para.context, rule_score=0.30, # 標 medium rule_guess_control_id=control_id, # 即便不在 baseline，仍給使用者參考 )) ``` 理由：使用者可能 docx 寫了 baseline 之外的控制項（如 CMMC L2 但選 L1），讓他們**看到並決定**怎麼處理。 - [ ] **Step 2: 加 test case** ```python def test_h3_with_id_not_in_baseline_goes_to_unmatched(...): # docx 有 IR.L2-3.1.1（在 cmmc-l2 但不在 cmmc-l1 baseline） # candidates 只給 cmmc-l1 控制項 parsed = parser.parse(...) assert any(u.rule_guess_control_id == "IR.L2-3.1.1" for u in parsed.unmatched_paragraphs) ``` - [ ] **Step 3: Commit** --- ### Task B.6: DocxParserCore — FATAL handling - [ ] **Step 1: 寫測試** ```python def test_parse_raises_no_control_id_when_empty(): """完全空白 docx → FATAL""" from common.code.grc_error_code import GrcErrorCode parser = DocxParserCore() # 用 fixture 建立沒 H3 的 docx # ... with pytest.raises(ParseException) as exc: parser.parse(empty_docx_path, "cmmc-l1", candidates) assert exc.value.error_code == GrcErrorCode.GRC_DOCX_NO_CONTROL_ID_FOUND def test_parse_raises_framework_mismatch(): """framework 不符 → FATAL""" parser = DocxParserCore() # docx 全是 NIST 800-171 ID，user 選 cmmc-l1 with pytest.raises(ParseException) as exc: parser.parse(nist_docx_path, "cmmc-l1", cmmc_candidates) assert exc.value.error_code == GrcErrorCode.GRC_DOCX_FRAMEWORK_MISMATCH ``` - [ ] **Step 2: 實作 ParseException 和 FATAL 檢查** ```python class ParseException(Exception): def __init__(self, error_code, original=None): self.error_code = error_code super().__init__(f"{error_code}: {original or ''}") # 在 parse() 開頭加： def parse(self, file_path, framework, candidates): try: doc = Document(file_path) except Exception as e: raise ParseException(GrcErrorCode.GRC_DOCX_PARSE_FAILED, e) structure = self._extract_structure(doc) h3_texts = [p.text for p in structure.paragraphs if p.heading_level == 3] if not h3_texts: raise ParseException(GrcErrorCode.GRC_DOCX_NO_CONTROL_ID_FOUND) dominant = detect_dominant_framework(h3_texts) if dominant is None: raise ParseException(GrcErrorCode.GRC_DOCX_NO_CONTROL_ID_FOUND) if dominant != framework: # 計算 user_match_rate from domain.oscal.parser.framework_patterns import FRAMEWORK_PATTERNS, USER_FRAMEWORK_MIN_THRESHOLD user_pattern = FRAMEWORK_PATTERNS.get(framework) if user_pattern: user_match_rate = sum(1 for t in h3_texts if user_pattern.search(t)) / len(h3_texts) if user_match_rate < USER_FRAMEWORK_MIN_THRESHOLD: raise ParseException(GrcErrorCode.GRC_DOCX_FRAMEWORK_MISMATCH) # ... 繼續正常 parsing ``` - [ ] **Step 3: 跑測試 + Commit** --- ### Task B.7: SSP Metadata extraction（AC-23） > 補入 SD §6.1 漏標的職責：parser 也要抽 Introduction tables 給 SSP metadata 用。 **Files:** - Modify: `domain/oscal/parser/docx_parser_core.py`（加 `_extract_metadata()`） - Modify: `domain/oscal/parser/docx_intermediate.py`（加 `system_metadata` 欄位） - [ ] **Step 1: 擴充 ParsedDocx** ```python # docx_intermediate.py @dataclass class SspSystemMetadata: name: Optional[str] = None description: Optional[str] = None security_sensitivity_level: Optional[str] = None # 'low' | 'moderate' | 'high' deployment_model: Optional[str] = None # 'on-premise' | 'cloud' | ... authorization_boundary: Optional[str] = None @dataclass class ParsedDocx: # ... 既有欄位 system_metadata: Optional[SspSystemMetadata] = None ``` - [ ] **Step 2: 寫測試**（參考 fixture Introduction tables） ```python def test_extract_metadata_from_introduction(real_fixture_path): parser = DocxParserCore() parsed = parser.parse(real_fixture_path, "cmmc-l1", candidates) assert parsed.system_metadata is not None # fixture 第 1 個 table 有 'Ser. NO：CMMC-SSP-...' # 確認至少能抽出 name 或 description ``` - [ ] **Step 3: 實作 `_extract_metadata`** 採保守策略：用啟發式規則從 Introduction H1 段內的 table 抽取，找不到就設 `None`（不阻擋整個解析）： ```python def _extract_metadata(self, structure): """從 H1='Introduction' 區段內的 tables 嘗試抽 SSP 基本資料""" # 找 'Introduction' / '基本資料' H1 之後到下一個 H1 之前的 tables intro_start = next((i for i, p in enumerate(structure.paragraphs) if p.heading_level == 1 and ('Introduction' in p.text or '基本資料' in p.text)), None) if intro_start is None: return None # ...（依 fixture 真實格式調整 cell 對應） return SspSystemMetadata(name=..., description=...) ``` > ⚠️ 抽取規則需以真實 fixture 為依據實驗。預期會有 ~30% 命中率（key/value table 格式各家寫法不同），無法全自動但好過全人工。 - [ ] **Step 4: Commit** --- # Phase C — BE Domain Service + Strategies ### Task C.1: SspDocxParseJobDomainService **Files:** - Create: `domain/oscal/services/ssp_docx_parse_job_domain_service.py` - Test: `tests/test_ssp_docx_parse_job_domain_service.py` - [ ] **Step 1: 寫測試**（簡化：mock repo） ```python def test_create_job_returns_uid_and_pending_status(): repo_mock = MagicMock() service = SspDocxParseJobDomainService(repo=repo_mock) repo_mock.add.return_value = SspDocxParseJobEntity(uid="abc", status="pending") result = service.create(source_type="project_ssp", source_uid="x", mode="statement_only", framework="cmmc-l1", file_path="/tmp/x.docx", file_name="x.docx", file_size=100, user="raymond") assert result.uid == "abc" repo_mock.add.assert_called_once() ``` - [ ] **Step 2: 實作 service** ```python import uuid from typing import Optional from domain.oscal.entity.ssp_docx_parse_job_entity import ( SspDocxParseJobEntity, SspDocxParseJobQueryEntity ) from domain.oscal.repository.i_ssp_docx_parse_job_repo import ISspDocxParseJobRepo class SspDocxParseJobDomainService: def __init__(self, repo: ISspDocxParseJobRepo): self._repo = repo def create(self, source_type, source_uid, mode, framework, file_path, file_name, file_size, tenant_id, user) -> SspDocxParseJobEntity: entity = SspDocxParseJobEntity( uid=str(uuid.uuid4()), tenant_id=tenant_id, source_type=source_type, source_uid=source_uid, mode=mode, framework=framework, file_path=file_path, file_name=file_name, file_size=file_size, status="pending", created_user=user, updated_user=user, ) return self._repo.add(entity) def get_one(self, uid: str) -> Optional[SspDocxParseJobEntity]: return self._repo.get_one(SspDocxParseJobQueryEntity(uid=uid)) def update_status(self, uid: str, status: str, user: str, **fields) -> SspDocxParseJobEntity: job = self.get_one(uid) if not job: return None return self._repo.update(job.id, {"status": status, **fields, "updated_user": user}, user) def write_parsed_result(self, uid: str, parsed_result: dict, user: str): return self.update_status(uid, "awaiting_review", user, parsed_result=parsed_result) def write_error(self, uid: str, error_code: str, error_message: str, user: str): return self.update_status(uid, "failed", user, error_code=error_code, error_message=error_message) def write_import_summary(self, uid: str, summary: dict, user: str): return self.update_status(uid, "completed", user, import_summary=summary) ``` - [ ] **Step 3: 跑測試 + Commit** --- ### Task C.2: WriteStrategy interface **Files:** - Create: `domain/oscal/strategy/i_ssp_docx_write_strategy.py` - [ ] **Step 1: 撰寫** ```python from abc import ABC, abstractmethod from dataclasses import dataclass, field from domain.oscal.parser.docx_intermediate import ParsedDocx @dataclass class ImportDecisions: decisions: list[dict] # [{control_id, action, objectives:[...]}] manual_assignments: list[dict] skipped_paragraph_idxs: list[int] = field(default_factory=list) predicted_controls_user_selection: list[str] = field(default_factory=list) @dataclass class ImportResult: created: int = 0 updated: int = 0 skipped: int = 0 manual_assigned: int = 0 manual_skipped: int = 0 missing_left_blank: int = 0 class IWriteStrategy(ABC): @abstractmethod def write( self, parsed: ParsedDocx, decisions: ImportDecisions, source_uid: str, user_id: str ) -> ImportResult: ... ``` - [ ] **Step 2: Commit** --- ### Task C.3: SspWriteStrategy **Files:** - Create: `domain/oscal/strategy/ssp_write_strategy.py` - Test: `tests/test_ssp_write_strategy.py` - [ ] **Step 1: 寫測試**（mock 所有依賴） ```python def test_ssp_strategy_writes_use_docx_to_implementation_desc(): impl_service_mock = MagicMock() obj_service_mock = MagicMock() info_sys_service_mock = MagicMock() strategy = SspWriteStrategy( control_impl_service=impl_service_mock, objective_service=obj_service_mock, information_system_service=info_sys_service_mock, project_info_system_repo=MagicMock(), ) parsed = ParsedDocx( matched_controls=[MatchedControl( control_id="AC.L1-b.1.i", control_name="...", score=0.95, score_reason="...", parsed_implementation_description="new desc", objectives=[] )], unmatched_paragraphs=[], missing_baseline_controls=[], predicted_module_frame_controls=[], summary={} ) decisions = ImportDecisions( decisions=[{"control_id": "AC.L1-b.1.i", "action": "use_docx", "objectives": []}], manual_assignments=[], skipped_paragraph_idxs=[], ) result = strategy.write(parsed, decisions, source_uid="ssp-001", user_id="raymond") assert result.updated >= 1 impl_service_mock.update_by_ssp_and_identifier.assert_called() ``` - [ ] **Step 2: 實作 strategy** > ⚠️ 依 V-2：jedi-oscal **沒有** `update_or_create` / `upsert`。Strategy 必須自己組 `get + (add or update)` 兩段。 ```python from jedi_common.session.database.db import transaction from jedi_oscal.domain.entity.ssp.ssp_control_implementation_entity import ( ControlImplementationEntity, SspControlImplementationQueryEntity ) from jedi_oscal.domain.entity.ssp.ssp_control_implementation_objective_entity import ( ControlImplementationObjectiveEntity, SspControlImplObjectiveQueryEntity ) class SspWriteStrategy(IWriteStrategy): def __init__( self, control_impl_service, objective_service, ssp_domain_service, information_system_service, project_info_system_repo, ): self._impl = control_impl_service self._obj = objective_service self._ssp = ssp_domain_service self._info_sys = information_system_service self._project_info_sys_repo = project_info_system_repo def write(self, parsed, decisions, source_uid, user_id, tenant_id): result = ImportResult() # 取得 SSP 內部 ID（service 多用 ssp_id 而非 ssp_uid） ssp = self._ssp.get_by_uid(source_uid) if ssp is None: raise NotFound(GrcErrorCode.GRC_SSP_NOT_FOUND) ssp_id = ssp.id # 1. Process matched controls per decision decision_map = {d["control_id"]: d for d in decisions.decisions} for matched in parsed.matched_controls: d = decision_map.get(matched.control_id) if not d or d["action"] == "skip": result.skipped += 1 continue if d["action"] == "use_docx": self._upsert_control_impl( ssp_id=ssp_id, control_identifier=matched.control_id, description=matched.parsed_implementation_description, user=user_id, result=result, ) else: # keep_current → 不變動 result.skipped += 1 continue # objectives 一併略過 # Process objectives — 注意 V-2 需驗證 objective_key obj_decisions = {o["objective_key"]: o for o in d.get("objectives", [])} valid_keys = self._fetch_valid_objective_keys(ssp_id, matched.control_id) for obj in matched.objectives: # 注意：valid_keys 可能為 None（寬鬆放行）/ set()（控制項無 objective）/ set with keys if valid_keys is not None and obj.objective_key not in valid_keys: raise BadRequestError( GrcErrorCode.GRC_DOCX_OBJECTIVE_KEY_INVALID, f"objective_key='{obj.objective_key}' not valid for {matched.control_id}" ) obj_action = obj_decisions.get(obj.objective_key, {}).get("action", "use_docx") if obj_action == "use_docx": self._upsert_objective( ssp_id=ssp_id, control_identifier=matched.control_id, objective_key=obj.objective_key, description=obj.parsed_description, user=user_id, ) # 2. Process manual_assignments for ma in decisions.manual_assignments: paragraph_text = next( (u.text for u in parsed.unmatched_paragraphs if u.paragraph_idx == ma["paragraph_idx"]), None ) if not paragraph_text: continue for target in ma.get("targets", []): self._apply_manual_assignment(ssp_id, target, paragraph_text, user_id) result.manual_assigned += 1 # 3. Skipped manual result.manual_skipped = len(decisions.skipped_paragraph_idxs) # 4. Missing baseline result.missing_left_blank = len(parsed.missing_baseline_controls) # 5. SSP metadata → information_systems（依 SD §6.1） if parsed.system_metadata: self._upsert_information_system(parsed.system_metadata, source_uid, tenant_id, user_id) return result def _upsert_control_impl(self, ssp_id, control_identifier, description, user, result): """jedi-oscal 沒有 upsert，自己組 get + add/update""" existing = self._impl.get_by_ssp_and_identifier(ssp_id, control_identifier) if existing: existing.description = description existing.updated_user = user self._impl.update(existing) result.updated += 1 else: new_entity = ControlImplementationEntity( system_security_plan_id=ssp_id, control_identifier=control_identifier, description=description, created_user=user, updated_user=user, ) self._impl.add(new_entity) result.created += 1 def _upsert_objective(self, ssp_id, control_identifier, objective_key, description, user): existing = self._obj.get_one(SspControlImplObjectiveQueryEntity( system_security_plan_id=ssp_id, control_identifier=control_identifier, statement_identifier=objective_key, )) if existing: existing.description = description existing.updated_user = user self._obj.update(existing) else: self._obj.add(ControlImplementationObjectiveEntity( system_security_plan_id=ssp_id, control_identifier=control_identifier, statement_identifier=objective_key, description=description, created_user=user, updated_user=user, )) def _fetch_valid_objective_keys(self, ssp_id, control_identifier) -> Optional[set]: """從 baseline 控制項定義拿合法的 objective key set 回 None 表示「無法取得 baseline，略過驗證」（寬鬆放行）； 回 set() 表示「已查到但該控制項無 objective」； 回 set 含 keys 表示「驗證清單」 """ # 這部分需依 jedi-oscal 的 catalog/profile 結構查（先 stub，實作時補） # 預期可從 candidate.objective_keys 拿（在 app service 層 build candidates 時要包含） # v1 簡化：寬鬆放行 return None def _apply_manual_assignment(self, ssp_id, target, text, user_id): if target["level"] == "implementation": existing = self._impl.get_by_ssp_and_identifier(ssp_id, target["control_id"]) if target["merge_action"] == "append" and existing: new_desc = (existing.description or "") + "\n\n" + text else: new_desc = text self._upsert_control_impl( ssp_id, target["control_id"], new_desc, user_id, ImportResult() # result 此處不算 ) elif target["level"] == "objective": # 類似邏輯 ... def _upsert_information_system(self, metadata, source_uid, tenant_id, user_id): """SD §6.1 — InformationSystemService.upsert_by_name""" if not metadata.name: return # 寫入或更新 information_systems table info_sys = self._info_sys.upsert_by_name( name=metadata.name, tenant_id=tenant_id, fields={ "description": metadata.description, "security_sensitivity_level": metadata.security_sensitivity_level, "deployment_model": metadata.deployment_model, "authorization_boundary": metadata.authorization_boundary, }, user_id=user_id, ) # TODO: 寫入 project_information_systems 關聯（從 source_uid 找 project_id） ``` > 上面 code 是 reference 級草稿，實作時參考 jedi-oscal 真實 entity 欄位名（如 `description` vs `implementation_description`）微調。每個 `_upsert_*` method 都應有對應 unit test。 - [ ] **Step 3: 加錯誤碼 `GRC_DOCX_OBJECTIVE_KEY_INVALID`** 修正 SA §5：原列 17 個 error code 不含這個。需新增： - `GRC_DOCX_OBJECTIVE_KEY_INVALID` = ("docx 內 objective_key 不在 baseline 中", "GRC_400012") 回到 Task E.1 把它一併加入。 - [ ] **Step 4: 跑測試 + Commit** --- ### Task C.4: ModuleFrameWriteStrategy **Files:** - Create: `domain/oscal/strategy/module_frame_write_strategy.py` - Test: `tests/test_module_frame_write_strategy.py` 類似 Task C.3，但寫到 `module_frame_control_defaults` / `module_frame_control_objective_defaults` + 處理 `mode='full'` 時更新 `oscal_profiles.include_controls`。 - [ ] **Step 1: 寫測試** - [ ] **Step 2: 實作** - [ ] **Step 3: 跑測試 + Commit** --- # Phase D — BE App Service + Routes + Serializers + DI ### Task D.1: Marshmallow Serializers **Files:** - Create: `api/oscal/serializers/ssp/ssp_docx_import.py` - [ ] **Step 1: 撰寫** 依 SA §4.1 / §4.2 / §4.3 完整 schema： ```python from marshmallow import Schema, fields, validate class SspDocxImportParseRequestSchema(Schema): framework = fields.Str(required=True, validate=validate.OneOf(["cmmc-l1", "cmmc-l2", "nist-800-171", "iso27001"])) source_type = fields.Str(required=True, validate=validate.OneOf(["module_frame", "project_ssp"])) source_uid = fields.Str(required=True) mode = fields.Str(required=True, validate=validate.OneOf(["full", "statement_only"])) class ObjectiveDecisionSchema(Schema): objective_key = fields.Str(required=True) action = fields.Str(required=True, validate=validate.OneOf(["use_docx", "keep_current", "skip"])) class ControlDecisionSchema(Schema): control_id = fields.Str(required=True) action = fields.Str(required=True, validate=validate.OneOf(["use_docx", "keep_current", "skip"])) objectives = fields.List(fields.Nested(ObjectiveDecisionSchema), load_default=list) class ManualAssignmentTargetSchema(Schema): control_id = fields.Str(required=True) level = fields.Str(required=True, validate=validate.OneOf(["implementation", "objective"])) objective_key = fields.Str(load_default=None) merge_action = fields.Str(required=True, validate=validate.OneOf(["append", "replace"])) class ManualAssignmentSchema(Schema): paragraph_idx = fields.Int(required=True) targets = fields.List(fields.Nested(ManualAssignmentTargetSchema), required=True) class SspDocxImportConfirmRequestSchema(Schema): decisions = fields.List(fields.Nested(ControlDecisionSchema), required=True) manual_assignments = fields.List(fields.Nested(ManualAssignmentSchema), load_default=list) skipped_paragraph_idxs = fields.List(fields.Int(), load_default=list) predicted_controls_user_selection = fields.List(fields.Str(), load_default=list) # Response schemas (省略，依 SA §4 實作) class SspDocxImportSummarySchema(Schema): ... class SspDocxImportParseResponseSchema(Schema): ... class SspDocxImportPreviewResponseSchema(Schema): ... class SspDocxImportConfirmResponseSchema(Schema): ... ``` - [ ] **Step 2: Commit** --- ### Task D.2: SspDocxImportAppService **Files:** - Create: `app/oscal/service/ssp_docx_import_app_service.py` - Test: `tests/test_ssp_docx_import_app_service.py` 依 SD §1.2 規範：每個 public method 必須 `@transaction` + 第一行權限檢查。 #### D.2.1 — 權限驗證（concrete chain） ```python def _check_permission(self, source_type, source_uid, user_context): """ AC-22 權限矩陣： - source_type='module_frame' → tenant admin OR manager - source_type='project_ssp' → project manager（auditor 嚴格不可） """ user_id = user_context.user_id is_admin = user_context.is_admin if source_type == 'module_frame': # 確認 module_frame 存在 mf = self._module_frame_domain_service.get_by_uid(source_uid) if mf is None: raise NotFound(GrcErrorCode.GRC_DOCX_SOURCE_NOT_FOUND) # is_admin 直接放行；否則檢查是否 manager if is_admin: return mf # TODO: tenant 內任何 manager 可改？依 module_frame 既有 add/edit 權限對齊 # 既有 ModuleFrame.vue Step 2/3 的 PUT API 是何權限？實作前讀現有 module_frame_route.py 確認 if not self._is_tenant_manager(user_id): raise ForbiddenError(GrcErrorCode.GRC_DOCX_NO_PERMISSION_FOR_SOURCE) return mf elif source_type == 'project_ssp': # 從 SSP uid 反查 project_id ssp = self._ssp_domain_service.get_by_uid(source_uid) if ssp is None: raise NotFound(GrcErrorCode.GRC_DOCX_SOURCE_NOT_FOUND) # SSP → assessment_plan → project chain。確認鏈接結構： # ssp.assessment_plan_id → AP → ProjectAssessmentPlanMapping.project_id project_id = self._resolve_project_id_from_ssp(ssp) # 檢查使用者是 project manager participant = self._project_participant_domain_service.get_one( ProjectParticipantQueryEntity(project_id=project_id, user_id=user_id) ) if participant is None or participant.role != "manager": raise ForbiddenError(GrcErrorCode.GRC_DOCX_NO_PERMISSION_FOR_SOURCE) return ssp else: raise BadRequestError(GrcErrorCode.GRC_DOCX_SOURCE_TYPE_INVALID) ``` > ⚠️ **Step 0 verify before implementation**：實作前 grep / read： > 1. `infra/grc/model/grc_project_extension.py` 與相關 entity 確認 ssp ↔ project 鏈接表 > 2. 既有 `api/oscal/routes/ssp/ssp_control_implementation_route.py` 看是否有同類權限檢查可參考 > 3. `is_tenant_manager` 的判定方式（可能是檢查 `user.is_admin` 或屬於某 role group） > 若鏈接結構與想像不同，停下來回報。 #### D.2.2 — 候選控制項載入 ```python def _load_candidates(self, source_type, source, framework): """ 回 list[CandidateControl] - source_type='project_ssp' → 從 ssp.profile.include_controls 拿 - source_type='module_frame' → 從 module_frame.profile.include_controls 拿 若 mode='full'，candidates 是「整個 framework 的全 catalog」，不是 profile subset """ # 從 catalog_domain_service 拿全 framework 控制項作為 fallback catalog_controls = self._catalog_domain_service.get_all_by_framework(framework) # ... # 對每個控制項，組 CandidateControl 物件，包含： # - control_id, control_name # - objective_keys (從 catalog_control_assessment 拿，每個控制項的所有 (a)(b)(c)) # - 若 source_type='project_ssp': # current_implementation_description = control_implementation_service.get_by_ssp_and_identifier(...) # current_objectives_descriptions = objective_service.get_all_by_ssp_and_control_id(...) return candidates ``` > ⚠️ **Step 0 verify**： > 1. `catalog_domain_service.get_all_by_framework()` 是否存在？若不存在，從 `OscalFrameworkVersion → OscalProfile → ProfileControl` 鏈接查 > 2. `catalog_control_assessment` 表 / service 名稱（既有 `infra/oscal/repository/catalog_control_assessment_repo_impl.py` 應有 reading method） > 3. 既有 `SspControlImplementationListRoute` 已 fetch 該 SSP 的 control_implementation list，可參考其 service 呼叫鏈 - [ ] **Step 1: 寫測試**（mock 所有依賴） - [ ] **Step 2: 實作 4 個 public method**： - `upload_and_parse(file, framework, source_type, source_uid, mode, user_context) -> dict` - `get_parse_result(parse_uid, user_context) -> dict`（含 lazy expiration check） - `confirm_import(parse_uid, payload, user_context) -> dict` - `discard_parse(parse_uid, user_context) -> dict` - [ ] **Step 3: 跑測試 + Commit** --- ### Task D.3: Routes **Files:** - Create: `api/oscal/routes/ssp/ssp_docx_import_route.py` - [ ] **Step 1: 撰寫 routes**（4 個 endpoint） ```python from dependency_injector.wiring import inject, Provide from flask import request from flask_apispec import MethodResource, doc, marshal_with, use_kwargs from flask_jwt_extended import jwt_required from jedi_common.session.auth.auth_context import get_user_context from app.oscal.service.ssp_docx_import_app_service import SspDocxImportAppService from api.oscal.serializers.ssp.ssp_docx_import import ( SspDocxImportConfirmRequestSchema, SspDocxImportParseResponseSchema, SspDocxImportPreviewResponseSchema, SspDocxImportConfirmResponseSchema, ) from common.enum.schema_code import AUTH_PARAMS from common.util.response_util import return_response from di_containers.containers import Containers class SspDocxImportParseRoute(MethodResource): @doc(description="上傳 docx + 解析", tags=["SSP Docx Import"], params=AUTH_PARAMS) @marshal_with(SspDocxImportParseResponseSchema, apply=False) @jwt_required() @inject def post(self, app_service: SspDocxImportAppService = Provide[Containers.oscal_container.ssp_docx_import_app_service]): file = request.files.get("file") framework = request.form.get("framework") source_type = request.form.get("source_type") source_uid = request.form.get("source_uid") mode = request.form.get("mode") result = app_service.upload_and_parse(file, framework, source_type, source_uid, mode, get_user_context()) return return_response(True, result) class SspDocxImportRoute(MethodResource): @doc(description="取得 parse job 狀態與預覽", tags=["SSP Docx Import"], params=AUTH_PARAMS) @marshal_with(SspDocxImportPreviewResponseSchema, apply=False) @jwt_required() @inject def get(self, parse_uid: str, app_service: SspDocxImportAppService = Provide[Containers.oscal_container.ssp_docx_import_app_service]): result = app_service.get_parse_result(parse_uid, get_user_context()) return return_response(True, result) @jwt_required() @inject def delete(self, parse_uid: str, app_service: SspDocxImportAppService = Provide[Containers.oscal_container.ssp_docx_import_app_service]): result = app_service.discard_parse(parse_uid, get_user_context()) return return_response(True, result) class SspDocxImportConfirmRoute(MethodResource): @doc(description="確認匯入", tags=["SSP Docx Import"], params=AUTH_PARAMS) @use_kwargs(SspDocxImportConfirmRequestSchema, location="json", apply=False) @marshal_with(SspDocxImportConfirmResponseSchema, apply=False) @jwt_required() @inject def post(self, parse_uid: str, app_service: SspDocxImportAppService = Provide[Containers.oscal_container.ssp_docx_import_app_service]): payload = SspDocxImportConfirmRequestSchema().load(request.get_json(silent=True) or {}) result = app_service.confirm_import(parse_uid, payload, get_user_context()) return return_response(True, result) ``` - [ ] **Step 2: 註冊到 `api/oscal/__init__.py`** ```python api.add_resource(SspDocxImportParseRoute, '/ssp-docx-imports/parse') api.add_resource(SspDocxImportRoute, '/ssp-docx-import/') api.add_resource(SspDocxImportConfirmRoute, '/ssp-docx-import//confirm') ``` - [ ] **Step 3: Commit** --- ### Task D.4: DI Container（依 V-1 修正：直接加在 OscalContainer，不另建 sub-container） **Files:** - Modify: `di_containers/oscal/oscal_containers.py`（OscalContainer 加 6 個 provider） - [ ] **Step 1: 在 OscalContainer 加 providers** ```python # 接續 oscal_containers.py 既有 providers... # === SSP docx import (新增) === ssp_docx_parse_job_repo = providers.Singleton(SspDocxParseJobRepoImpl) ssp_docx_parse_job_domain_service = providers.Factory( SspDocxParseJobDomainService, repo=ssp_docx_parse_job_repo, ) docx_parser_core = providers.Singleton(DocxParserCore) ssp_docx_module_frame_write_strategy = providers.Factory( ModuleFrameWriteStrategy, module_frame_control_default_service=module_frame_container.module_frame_control_default_service, module_frame_control_objective_default_service=module_frame_container.module_frame_control_objective_default_service, profile_domain_service=profile_domain_service, ) ssp_docx_ssp_write_strategy = providers.Factory( SspWriteStrategy, control_impl_service=control_implementation_domain_service, objective_service=control_implementation_objective_domain_service, ssp_domain_service=ssp_domain_service, information_system_service=information_system_container.information_system_service, project_info_system_repo=information_system_container.project_information_system_repo, ) ssp_docx_import_app_service = providers.Factory( SspDocxImportAppService, parse_job_domain=ssp_docx_parse_job_domain_service, parser=docx_parser_core, module_frame_write_strategy=ssp_docx_module_frame_write_strategy, ssp_write_strategy=ssp_docx_ssp_write_strategy, profile_domain_service=profile_domain_service, catalog_domain_service=catalog_domain_service, ssp_domain_service=ssp_domain_service, project_participant_domain_service=project_participant_container.project_participant_domain_service, module_frame_domain_service=module_frame_container.module_frame_domain_service, ) ``` > ⚠️ **Step 0 verify**： > > 1. **OscalContainer deps slots**：`module_frame_container` / `upload_file_container` 已存在於 `oscal_containers.py:64-100`；`information_system_container` / `project_participant_container` **不存在**，需新增 `providers.DependenciesContainer()` slot。 > 2. **避免 declaration ordering 問題**：實作於 `di_containers/containers.py` 時**優先用 `override_providers` 模式**（參考 `containers.py:141` 的既有 `associations_container.override_providers(oscal_container=oscal_container)` 用法），不要重排 Container 宣告順序，避免汙染其他模組。範例： > ```python > oscal_container.override_providers( > information_system_container=information_system_container, > project_participant_container=project_participant_container, > ) > ``` > 3. **既有 module_frame providers 名稱驗證**：`module_frame_container.module_frame_control_default_service` / `.module_frame_control_objective_default_service` 名稱**未驗證**。實作前先讀 `di_containers/module_frame/module_frame_containers.py` 確認 provider 真實名稱（可能是 `control_default_service` 簡稱或其他），對應修正。 > 4. **InformationSystemContainer 既存 provider**：讀 `di_containers/information_system/information_system_container.py` 確認 `information_system_service` 與 `project_information_system_repo` 兩個 provider 都存在。 - [ ] **Step 2: 確認 wiring 在 `di_containers/containers.py` 注入** OscalContainer 在 root Containers 內以 `providers.Container(OscalContainer, ...)` 註冊時，要 inject 必要的 dependencies container： ```python # di_containers/containers.py oscal_container = providers.Container( OscalContainer, config=config.oscal, module_frame_container=module_frame_container, upload_file_container=upload_file_container, information_system_container=information_system_container, # 新增 project_participant_container=..., # 新增 ) ``` > 「information_system_container 是否已存在」需先讀 containers.py 確認；若沒有，要先 declare 它的 sub-container（jedi_information_system 模組對應的 DI）。 - [ ] **Step 3: 跑 BE startup smoke test** ```bash ENV=DEVELOP_PREMISE python main_app.py & sleep 5 curl http://localhost:8000/swagger-ui/ -o /tmp/swagger.html grep "ssp-docx-imports" /tmp/swagger.html # 確認 endpoint 出現 kill %1 ``` - [ ] **Step 4: Commit** --- # Phase E — BE Error Codes + Helper + Integration Test ### Task E.1: 新增 18 個 error code **Files:** - Modify: `common/code/grc_error_code.py` 依 SA §5 表格 + Task C.3 補的 1 個（共 18 個）： 新增清單（接續 GrcErrorCode 既有序號）： | Constant | Code | |----------|------| | `GRC_DOCX_INVALID_FILE` | GRC_400005 | | `GRC_DOCX_FILE_TOO_LARGE` | GRC_400006 | | `GRC_DOCX_FRAMEWORK_REQUIRED` | GRC_400007 | | `GRC_DOCX_SOURCE_TYPE_INVALID` | GRC_400008 | | `GRC_DOCX_MODE_FULL_REQUIRES_MODULE_FRAME` | GRC_400009 | | `GRC_DOCX_DECISIONS_INCOMPLETE` | GRC_400010 | | `GRC_DOCX_PREDICTED_CONTROLS_REQUIRED` | GRC_400011 | | `GRC_DOCX_OBJECTIVE_KEY_INVALID` | GRC_400012 ⬅ **C.3 補** | | `GRC_DOCX_NO_PERMISSION_FOR_SOURCE` | GRC_403002 | | `GRC_DOCX_PARSE_JOB_NO_PERMISSION` | GRC_403003 | | `GRC_DOCX_SOURCE_NOT_FOUND` | GRC_404021 | | `GRC_DOCX_PARSE_JOB_NOT_FOUND` | GRC_404022 | | `GRC_DOCX_PARSE_JOB_NOT_AWAITING` | GRC_412009 | | `GRC_DOCX_PARSE_JOB_EXPIRED` | GRC_412010 | | `GRC_DOCX_PARSE_JOB_ALREADY_COMPLETED` | GRC_412011 | | `GRC_DOCX_NO_CONTROL_ID_FOUND` | GRC_422001 | | `GRC_DOCX_FRAMEWORK_MISMATCH` | GRC_422002 | | `GRC_DOCX_PARSE_FAILED` | GRC_422003 | > 注意 SA §5 列了 17 個，C.3 strategy 實作補了 `GRC_DOCX_OBJECTIVE_KEY_INVALID`，共 18 個。Sync 回 SA §5 表格（**修改 api-spec.md** 加 Code = 'GRC_400012' 一列）。 - [ ] **Step 1: 加 18 個常數到 GrcErrorCode** - [ ] **Step 2: 修 api-spec.md §5 補上第 18 個** - [ ] **Step 3: Commit** --- ### Task E.2: InformationSystemService.upsert_by_name **Files:** - Modify: `jedi_information_system/app/service/information_system_service.py` - Test: `tests/test_information_system_service_upsert.py` > 依 V-6：`jedi_information_system/` 是主專案 in-tree 模組，不是外部套件，可直接修改。 - [ ] **Step 1: 寫測試** - [ ] **Step 2: 實作 upsert_by_name**（依 SD §6.1） - [ ] **Step 3: Commit** --- ### Task E.3: 端到端 BE 整合測試（real fixture） **Files:** - Test: `tests/test_ssp_docx_import_e2e.py` - [ ] **Step 0: 確認可用 fixture** 依 V-4：既有 `tests/conftest.py` 提供 `app` / `client` / `headers` / `tenant_id` fixtures。 本測試需要： - `client`（既有） - `headers`（既有，含 JWT for blsit / tenant_id=102） - 一個有效的 `ssp_uid`（需新增 fixture seed 一筆 SSP record，或使用既有 dev 資料庫的 SSP） 若無現成 SSP fixture，加在 `tests/oscal/conftest.py`（新檔）： ```python @pytest.fixture def seeded_ssp(client, headers): # 經由 API 建立一個臨時 SSP 並回傳 uid res = client.post("/api/1.0/...", json={...}, headers=headers) yield res.json["data"]["uid"] # cleanup（可選） ``` 或從 dev 資料庫直接 query 既有 SSP（簡單但測試不獨立）。 - [ ] **Step 1: 撰寫整合測試** 用真實 fixture： ```python def test_full_flow_with_real_fixture(client, headers, seeded_ssp): # 1. POST /parse with open("docs/features/FR-022-2604-ssp-doc-parser/raw-requirement/reference/ASIA-CMMC-SSP-DRAFT-202604.docx", "rb") as f: res = client.post( "/api/1.0/ssp-docx-imports/parse", data={"file": (f, "test.docx"), "framework": "cmmc-l1", "source_type": "project_ssp", "source_uid": seeded_ssp, "mode": "statement_only"}, headers=headers, content_type="multipart/form-data" ) assert res.status_code == 200 assert res.json["status"] is True parse_uid = res.json["data"]["parse_uid"] # 2. GET /preview res = client.get(f"/api/1.0/ssp-docx-import/{parse_uid}", headers=headers) assert res.status_code == 200 matched = res.json["data"]["matched_controls"] assert len(matched) >= 14 # fixture 有 14+ 控制項 # 3. POST /confirm（全 use_docx） decisions = [{"control_id": m["control_id"], "action": "use_docx", "objectives": []} for m in matched] res = client.post(f"/api/1.0/ssp-docx-import/{parse_uid}/confirm", json={"decisions": decisions}, headers=headers) assert res.status_code == 200 summary = res.json["data"]["import_summary"] assert summary["updated"] >= 14 # 4. 驗證 DB 寫入 # query ssp_control_implementation 確認 implementation_description 已被更新 ``` - [ ] **Step 2: 跑** ```bash pytest tests/test_ssp_docx_import_e2e.py -v -s ``` - [ ] **Step 3: Commit** --- # Phase F — FE Service + Store + Dialog Shell ### Task F.1: API constants **Files:** - Modify: `compliance-manager-fe/src/config/api/api.js` - [ ] **Step 1: 加 endpoints**（依 frontend-spec §8.2） - [ ] **Step 2: Commit** ```bash cd /Users/chouraymond/Projects/Billows/Audit-Manager/compliance-manager-fe git add src/config/api/api.js git commit -m "feat(ssp-docx): add ssp-docx-import API constants" ``` --- ### Task F.2: SspDocxImportService **Files:** - Create: `compliance-manager-fe/src/service/SspDocxImportService.js` - [ ] **Step 1: 撰寫**（依 frontend-spec §8.1） - [ ] **Step 2: Commit** --- ### Task F.3: Pinia Store **Files:** - Create: `compliance-manager-fe/src/stores/sspDocxImport.js` - [ ] **Step 1: 撰寫 store**（依 frontend-spec §7.1） 完整 state + actions（loadPreview / setControlAction / setObjectiveAction / assignParagraph / batchAction / togglePredictedControl / buildDecisionsPayload / reset） - [ ] **Step 2: Commit** --- ### Task F.4: SspDocxImportDialog (top-level shell) **Files:** - Create: `compliance-manager-fe/src/components/grc/SspDocxImportDialog.vue` - [ ] **Step 1: 撰寫 shell**（依 frontend-spec §3.3） ```vue ``` - [ ] **Step 2: Commit** --- # Phase G — FE Wizard Steps ### Task G.1: StepUpload **Files:** - Create: `compliance-manager-fe/src/components/grc/ssp-docx-import/StepUpload.vue` 依 frontend-spec §4，含 framework dropdown + FileUpload + 點下一步呼叫 service.parse() → 進 Step 1。 - [ ] **Step 1-3: 寫元件 + i18n + Commit** --- ### Task G.2: StepPreview shell + MatchedControlDiff **Files:** - Create: `compliance-manager-fe/src/components/grc/ssp-docx-import/StepPreview.vue` - Create: `compliance-manager-fe/src/components/grc/ssp-docx-import/MatchedControlDiff.vue` 依 frontend-spec §5.2.3。先做 matched controls 區塊與 diff，UnmatchedParagraph 區塊留 Phase H。 > ⚠️ 修正 frontend-spec §11.4：StepPreview onMounted 呼叫 `service.getPreview(parseUid)` 時也可能收到 412（lazy expiration on GET，依 SD §2.4 + V-1）。處理： > > ```ts > try { > await store.loadPreview(parseUid) > } catch (err) { > if (err.response?.status === 412) { > toast.add({severity:'warn', summary:'已過期，請重新上傳', life:5000}) > emit('prev') // 回 Step 0 > return > } > throw err > } > ``` - [ ] **Step 1-3: 寫元件 + Commit** --- ### Task G.3: StepResult **Files:** - Create: `compliance-manager-fe/src/components/grc/ssp-docx-import/StepResult.vue` - [ ] **Step 1-3: 寫元件 + Commit** --- ### Task G.4: ParseWarningBanner + BatchActionsToolbar 依 frontend-spec §5.2.1 / §5.2.2 - [ ] **Step 1-3: 寫元件 + Commit** --- ### Task G.5: PredictedControlsSelector (mode=full only) 依 frontend-spec §5.2.6 - [ ] **Step 1-3: 寫元件 + Commit** --- # Phase H — FE Drag-Drop + Conflict Modal ### Task H.1: 安裝 vuedraggable ```bash cd compliance-manager-fe npm install vuedraggable@next ``` - [ ] **Commit:** `git commit -m "chore(deps): add vuedraggable@next"` --- ### Task H.2: UnmatchedParagraphCard **Files:** - Create: `compliance-manager-fe/src/components/grc/ssp-docx-import/UnmatchedParagraphCard.vue` 依 frontend-spec §5.2.4 上半段 - [ ] **Step 1-3: 寫元件 + Commit** --- ### Task H.3: DropTargetTree **Files:** - Create: `compliance-manager-fe/src/components/grc/ssp-docx-import/DropTargetTree.vue` 依 frontend-spec §5.2.4 下半段，整合 vuedraggable 的 drop zone。 - [ ] **Step 1-3: 寫元件 + Commit** --- ### Task H.4: ConflictModal **Files:** - Create: `compliance-manager-fe/src/components/grc/ssp-docx-import/ConflictModal.vue` 依 frontend-spec §5.2.5 - [ ] **Step 1-3: 寫元件 + Commit** --- ### Task H.5: 整合進 StepPreview 回到 StepPreview，把 UnmatchedParagraphCard / DropTargetTree / ConflictModal 串起來。 - [ ] **Step 1: 整合 + 串 store actions** - [ ] **Step 2: 手動 smoke test**（dev server 跑一次完整流程） - [ ] **Step 3: Commit** --- # Phase I — FE 整合到既有頁面 + i18n ### Task I.1: i18n 字串檔 **Files:** - Create: `compliance-manager-fe/src/config/locales/i18n/zh-tw/ssp_docx_import.json` - Create: `compliance-manager-fe/src/config/locales/i18n/en/ssp_docx_import.json` 依 frontend-spec §9.1 完整字串清單。 - [ ] **Step 1: 撰寫 zh-tw** - [ ] **Step 2: 翻譯 en** - [ ] **Step 3: 確保 i18n 載入**（檢查 `i18n.js` import 路徑） - [ ] **Step 4: Commit** --- ### Task I.2: SSP 控制項現況頁加按鈕 **Files:** - Modify: SSP 控制項現況頁面（依 frontend-spec §2.1） - [ ] **Step 1: 加按鈕 + dialog instance + onImported handler** - [ ] **Step 2: dev server 手動測試** - [ ] **Step 3: Commit** --- ### Task I.3: ModuleFrame.vue Step 2 加按鈕 依 frontend-spec §2.2 - [ ] **Step 1-3: 加按鈕 + 串資料 + Commit** --- ### Task I.4: ModuleFrame.vue Step 3 加按鈕 依 frontend-spec §2.3 - [ ] **Step 1-3: 加按鈕 + 串資料 + Commit** --- # Phase J — 跨 repo Smoke Test ### Task J.1: 端到端 manual test 依以下劇本逐一驗證： | # | 場景 | 預期 | |---|------|------| | 1 | 入口 B SSP 控制項頁，匯入 fixture docx | 14+ 控制項出現在 Step 1，diff 正常 | | 2 | Step 1 拖拉一個 unmatched 段落到 AC.L1-b.1.i 主述 | 出現指派提示，無 conflict | | 3 | 拖拉到已有 docx 內容的控制項 | 跳 ConflictModal，選 append 後成功 | | 4 | 點「全部使用 docx」批次按鈕 | 全部 control 的 action 變 use_docx | | 5 | 確認匯入 | Step 2 顯示 summary，DB 確實寫入 | | 6 | 入口 A1 合規資源庫 Step 2 | 顯示 PredictedControlsSelector，可勾選微調 | | 7 | 入口 A1 完成後跳到合規資源庫 Step 3，控制項清單已套 | ✓ | | 8 | FATAL 場景：上傳壞掉的 docx | 顯示錯誤 toast，dialog 不前進 | - [ ] **Step 1: 全部劇本通過** - [ ] **Step 2: 截圖存 `docs/features/FR-022-2604-ssp-doc-parser/screenshots/`**（可選） - [ ] **Step 3: Final commit on FE: `feat(ssp-docx): smoke test passed for all 8 scenarios`** --- ### Task J.2: 移交 Phase 4 — Test Plan - [ ] **Step 1: 呼叫 feature-test-planner agent** ```bash # 在主對話中： # Agent({ # subagent_type: "feature-test-planner", # prompt: "請為 docs/features/FR-022-2604-ssp-doc-parser/ 產出 test-plan.md..." # }) ``` agent 會讀取所有 spec + plan，產出 `docs/features/FR-022-2604-ssp-doc-parser/test-plan.md`。 - [ ] **Step 2: User review test-plan.md** - [ ] **Step 3: 移到 compliance-manager-test/ 寫 E2E features** --- # Phase Changelog 規範（依 V-8） 每個 Phase 完成最後一個 task 時，**必加一份 changelog**： ``` docs/changelog/2026-XX-XX-feat-ssp-docx-phase--.md ``` 範例（Phase A）： ```yaml --- type: feat modules: [ssp-doc-parser, oscal] issue: docs/features/FR-022-2604-ssp-doc-parser/ commit: --- # feat(ssp-docx): Phase A — DB foundation + ORM ## 為什麼 依 implementation-plan.md Phase A 完成 ssp_docx_parse_jobs 表與相關 ORM/Repo。 ## 變更範圍 - 新增：ssp_docx_parse_jobs migration / Entity / Repo / Mapper / Model - 修改：無 ## API 變更 無（Phase A 僅基建，無 endpoint） ## 測試結果 pytest tests/test_ssp_docx_parse_job_repo.py — 1 pass ``` 10 個 Phase = 10 份 changelog。最後（Phase J 完成時）加一份**整合 changelog** 列出本次完整 deliverable，type=feat。 --- # 驗證清單（Phase 完成檢查） - [ ] **BE 驗證：** - [ ] 所有 pytest 通過：`pytest tests/ -q` - [ ] swagger-ui 顯示 4 個新 endpoint - [ ] DB 內 `oscal.ssp_docx_parse_jobs` 表存在且 RLS / GRANT 正確 - [ ] 18 個新 error code 在 GrcErrorCode 內（含 GRC_400012 GRC_DOCX_OBJECTIVE_KEY_INVALID） - [ ] **FE 驗證：** - [ ] dev server 啟動無 error - [ ] 三個入口都看得到「匯入 docx」按鈕 - [ ] manual smoke test 8 個劇本全通過 - [ ] i18n zh-tw / en 都有定義對應字串 - [ ] **跨 repo 驗證：** - [ ] BE + FE 在同一 branch name（如 `feat/ssp-doc-parser`） - [ ] commit 訊息 prefix 一致（`feat(ssp-docx):`） - [ ] **文件驗證：** - [ ] CLAUDE.md「行為規範」已加 changelog（Phase 5 完成時） - [ ] api-spec.md / design.md / frontend-spec.md 沒被偏離（如有偏離已標 SD-OPEN-X） --- # 附錄：與既有功能的相容性檢查 每個 Phase 完成後跑： ```bash cd compliance-manager-be pytest tests/ -q -x # 所有既有測試不能 break ``` 特別關注： - `test_ssp_control_impl_import_service.py`（既有 Excel SSP 匯入不能被影響） - `test_module_frame_template_import_service.py`（既有 module_frame 匯入不能被影響） - 任何涉及 `ssp_control_implementation` / `module_frame_control_defaults` 寫入的測試