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]]

def test_xxx(client, headers, tenant_id):
    res = client.post("/api/1.0/ssp-docx-imports/parse", ...)

---
type: feat
modules: [ssp-doc-parser, oscal]
issue: docs/features/FR-022-2604-ssp-doc-parser/  # link to feature folder
---

psql -h localhost -p 25432 -U cmmgr -d compliance_manager -f scripts/sql/ssp_docx_parse_jobs_migration.sql

psql -h localhost -p 25432 -U cmmgr -d compliance_manager -c "\d oscal.ssp_docx_parse_jobs"

git add scripts/sql/ssp_docx_parse_jobs_migration.sql
git commit -m "feat(ssp-docx): add ssp_docx_parse_jobs migration"

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

git add domain/oscal/entity/ssp_docx_parse_job_entity.py
git commit -m "feat(ssp-docx): add SspDocxParseJobEntity and QueryEntity"

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: ...

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)

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,
        )

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()

# 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"

pytest tests/test_ssp_docx_parse_job_repo.py -v

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"

# 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"

pytest tests/test_control_id_matcher.py -v
# Expected: ImportError / ModuleNotFoundError

# 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

pytest tests/test_control_id_matcher.py -v
# Expected: 5 passed

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)"

# 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

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)

# 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）

# 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)

pytest tests/test_docx_parser_core_structure.py -v
# Expected: 2 passed

# 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

# 接續 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

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"

# 在 _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，仍給使用者參考
    ))

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)

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

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

# 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

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

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=...)

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()

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)

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: ...

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()

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）

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): ...

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)

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

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)

api.add_resource(SspDocxImportParseRoute, '/ssp-docx-imports/parse')
api.add_resource(SspDocxImportRoute, '/ssp-docx-import/<string:parse_uid>')
api.add_resource(SspDocxImportConfirmRoute, '/ssp-docx-import/<string:parse_uid>/confirm')

# 接續 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,
)

# 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=...,  # 新增
)

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

@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（可選）

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 已被更新

pytest tests/test_ssp_docx_import_e2e.py -v -s

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"

cd compliance-manager-fe
npm install vuedraggable@next

# 在主對話中：
# Agent({
#   subagent_type: "feature-test-planner",
#   prompt: "請為 docs/features/FR-022-2604-ssp-doc-parser/ 產出 test-plan.md..."
# })

---
type: feat
modules: [ssp-doc-parser, oscal]
issue: docs/features/FR-022-2604-ssp-doc-parser/
commit: <hash backfill 後>
---

# 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

cd compliance-manager-be
pytest tests/ -q -x  # 所有既有測試不能 break