Stage 1 Implementation Plan — Docx Import Parity (Parser/Adapter only)

def extract_metadata_table(doc: Document) -> dict:
    """Find Table #0 (first table before Introduction H1). Returns dict
    with keys: document_serial_no, version, published, title, organization_name."""

def extract_party_tables(doc: Document) -> list[dict]:
    """Find all K/V tables under H1 Introduction (before next H1/H2).
    Returns list of {name, title, address, telephone, email, party_type}."""

def extract_leveraged_csp_table(doc: Document) -> list[dict]:
    """Find table under H2 Leveraged External Systems with col0 header
    'CSP/CSO Name'. Returns list of {provider, service_name, fedramp_package_id,
    nature_of_agreement, impact_level, data_types, authorized_users}."""

def extract_leveraged_category_table(doc: Document) -> list[dict]:
    """Find table under H2 Leveraged External Systems with col0 header
    '類別 Category' (or English 'Category'). Returns list of {category,
    name_description, purpose, protocol, security_auth}."""

def extract_revision_history_table(doc: Document) -> list[dict]:
    """Find table under H1 APPENDIX 2 Revision History. Returns list of
    {version, date, amendment, description}."""

def extract_system_characteristic_from_metadata(doc: Document) -> dict:
    """Heuristic — Stage 1 只抽 organization_name 從 Table #0 cell
    含 'Limited' / 'Inc.' / '公司' / 'Co.' pattern。Returns dict with
    keys: organization_name (str | None)."""

def _normalize_label(text: str) -> str:
    """Normalize label cell text for fuzzy matching: strip whitespace,
    remove punctuation (：:.,), lowercase, NFKC normalize."""

def _iter_body_blocks(doc):
    """Walk document body in order, yielding ('p', paragraph) or ('t', table)."""

def _find_section_blocks(doc, heading_match: str, heading_level: int = 1):
    """Find all body blocks (tables only) between a matching heading and the next
    heading of same-or-higher level. heading_match is a substring match on H1/H2 text."""

LABEL_MAP_METADATA = {
    'serno': 'document_serial_no',
    'serialno': 'document_serial_no',
    '序號': 'document_serial_no',
    'version': 'version',
    '版本': 'version',
    'issuedate': 'published',
    'issue date': 'published',
    '發布日期': 'published',
}
# 整 cell 含 'Limited' / 'Co.' / 'Inc.' / '公司' → organization_name
# 整 cell 含 'System Security Plan' or '系統安全計畫' → title

LABEL_MAP_PARTY = {
    'name': 'name',
    '姓名': 'name',
    'title': 'title',
    '職稱': 'title',
    'address': 'address',
    'officeaddress': 'address',
    '地址': 'address',
    'phone': 'telephone',
    'workphone': 'telephone',
    '電話': 'telephone',
    'email': 'email',
    'e-mailaddress': 'email',
    'emailaddress': 'email',
    '信箱': 'email',
}
# party_type: 有 'title' label 出現 → person；只有 name+address+phone → organization

LEVERAGED_CSP_HEADERS = {
    'cspcsoname': 'provider',
    'csoservice': 'service_name',
    'fedramppackageid': 'fedramp_package_id',
    'natureofagreement': 'nature_of_agreement',
    'impactlevel': 'impact_level',
    'datatypes': 'data_types',
    'authorizedusers': 'authorized_users',
    'authorizedusersauthentication': 'authorized_users',
}
# placeholder skip: row 全是 '[...]' bracket 文字 → skip

LEVERAGED_CAT_HEADERS = {
    '類別': 'category',
    'category': 'category',
    '名稱描述': 'name_description',
    'namedescription': 'name_description',
    '功能目的': 'purpose',
    'purposefunction': 'purpose',
    '傳輸方式與協定': 'protocol',
    'protocolmethod': 'protocol',
    '安全驗證機制': 'security_auth',
    'securityauth': 'security_auth',
}

REVISION_HEADERS = {
    'version': 'version',
    'date': 'date',
    'amendment': 'amendment',
    'description': 'description',
}

# tests/oscal/test_docx_section_extractors.py
import pytest
from docx import Document
from domain.oscal.parser.docx_section_extractors import (
    extract_metadata_table,
    extract_party_tables,
    extract_leveraged_csp_table,
    extract_leveraged_category_table,
    extract_revision_history_table,
    extract_system_characteristic_from_metadata,
)

REFERENCE_DOCX = Path(__file__).parent.parent / 'docs/features/FR-011.2-2605-ssp-import-export-phase2/reference/ASIA-CMMC-SSP-DRAFT-with-user-info-202604.docx'

@pytest.fixture
def ref_doc():
    return Document(str(REFERENCE_DOCX))

def test_extract_metadata_returns_title(ref_doc):
    result = extract_metadata_table(ref_doc)
    assert 'title' in result
    assert 'System Security Plan' in result['title'] or 'SSP' in result['title']

def test_extract_parties_includes_billows_persons(ref_doc):
    parties = extract_party_tables(ref_doc)
    # reference docx with-user-info 樣板有填 Billows 4 個 person
    assert len(parties) >= 1
    person_names = [p['name'] for p in parties if p.get('name')]
    assert any('Billows' in n or 'Johnson' in n or 'Bob' in n for n in person_names)

def test_extract_leveraged_csp_returns_at_least_placeholder(ref_doc):
    rows = extract_leveraged_csp_table(ref_doc)
    # reference docx 樣板 row1 是 placeholder ([...]) 應 skip
    # row2 應 有 data（除非 reference 樣板沒填）
    assert isinstance(rows, list)
    # 至少 fields 對 (即使空 list)

def test_extract_revision_history_returns_v10(ref_doc):
    rows = extract_revision_history_table(ref_doc)
    assert any(r.get('version') == 'V1.0' for r in rows)

# scripts/scrub_customer_docx.py
"""Scrub customer-identifying strings out of the docx, preserving structure.

Usage:
    python scripts/scrub_customer_docx.py SOURCE DEST
"""

SCRUB_RULES = {
    'AIR ASIA Company Limited.': 'Acme Demo Co., Ltd.',
    '亞航': 'Acme Demo',
    'Microsoft Windows Update': 'Demo External Service A',
    '病毒碼與威脅情資同步更新': '[Demo service description]',
    '印表機': 'Demo Interconnect B',
    # Add more as observed
}

def scrub(src_path: str, dst_path: str) -> None:
    from docx import Document
    doc = Document(src_path)
    # Run replace at the run level to preserve formatting
    for paragraph in doc.paragraphs:
        for run in paragraph.runs:
            for orig, repl in SCRUB_RULES.items():
                if orig in run.text:
                    run.text = run.text.replace(orig, repl)
    for tbl in doc.tables:
        for row in tbl.rows:
            for cell in row.cells:
                for paragraph in cell.paragraphs:
                    for run in paragraph.runs:
                        for orig, repl in SCRUB_RULES.items():
                            if orig in run.text:
                                run.text = run.text.replace(orig, repl)
    doc.save(dst_path)

# domain/oscal/adapter/cmmc_ssp_adapter.py
class CmmcSspAdapter:
    def adapt(self, parsed_docx, structure, candidates) -> ParsedSsp:
        # ... existing logic builds parsed_ssp ...

        # NEW (Stage 1): enrich with section extractor output
        from docx import Document
        from domain.oscal.parser.docx_section_extractors import (
            extract_metadata_table,
            extract_party_tables,
            extract_leveraged_csp_table,
            extract_leveraged_category_table,
            extract_revision_history_table,
            extract_system_characteristic_from_metadata,
        )

        # The adapter receives `structure` (DocxStructure) but for section
        # extraction we need the raw Document. The caller (app service) needs
        # to provide it — adjust signature OR re-open the docx in adapter.
        # Simpler path: adapter gets `doc` injected as additional kw arg.

        meta_dict = extract_metadata_table(self._doc)
        if meta_dict.get('title') and not parsed_ssp.metadata.title:
            parsed_ssp.metadata.title = meta_dict['title']
        if meta_dict.get('version'):
            parsed_ssp.metadata.version = meta_dict['version']
        if meta_dict.get('document_serial_no'):
            parsed_ssp.metadata.document_serial_no = meta_dict['document_serial_no']
        if meta_dict.get('published'):
            parsed_ssp.metadata.published = meta_dict['published']

        # Revision history
        revisions = extract_revision_history_table(self._doc)
        if revisions:
            parsed_ssp.metadata.revision_history = revisions

        # Parties — merge with existing (dedup by name + email)
        party_dicts = extract_party_tables(self._doc)
        existing_names = {p.name for p in parsed_ssp.parties}
        for pd in party_dicts:
            if pd.get('name') and pd['name'] not in existing_names:
                parsed_ssp.parties.append(ParsedParty(
                    name=pd['name'],
                    party_type=pd.get('party_type', 'person'),
                    title=pd.get('title'),
                    address=pd.get('address'),
                    telephone_number=pd.get('telephone'),
                    email_address=pd.get('email'),
                ))

        # Leveraged (T6 CSP)
        for row in extract_leveraged_csp_table(self._doc):
            ls = ParsedLeveragedService(
                title=row.get('service_name') or row.get('provider') or '',
                provider=row.get('provider'),
                fedramp_package_id=row.get('fedramp_package_id'),
                props={
                    'nature_of_agreement': row.get('nature_of_agreement'),
                    'impact_level': row.get('impact_level'),
                    'data_types': row.get('data_types'),
                    'authorized_users': row.get('authorized_users'),
                } if any(row.get(k) for k in ('nature_of_agreement','impact_level','data_types','authorized_users')) else None,
            )
            parsed_ssp.leveraged_services.append(ls)

        # Leveraged (T7 Category)
        for row in extract_leveraged_category_table(self._doc):
            ls = ParsedLeveragedService(
                title=row.get('name_description') or '',
                purpose=row.get('purpose'),
                protocol=row.get('protocol'),
                props={
                    'category': row.get('category'),
                    'security_auth': row.get('security_auth'),
                } if row.get('category') or row.get('security_auth') else None,
            )
            parsed_ssp.leveraged_services.append(ls)

        # System characteristics — organization name from Table #0
        sc_dict = extract_system_characteristic_from_metadata(self._doc)
        if sc_dict.get('organization_name') and not getattr(parsed_ssp.system_characteristics, 'organization_name', None):
            # ParsedSystemCharacteristics 沒這欄位 — 先塞 description 後綴
            existing = parsed_ssp.system_characteristics.description or ''
            parsed_ssp.system_characteristics.description = (
                f"{existing}\n組織名稱: {sc_dict['organization_name']}".strip()
            )

        return parsed_ssp

# app/oscal/service/ssp_docx_import_app_service.py
except Exception as e:
    logger.error(
        f"SSP docx parse failed (parse_uid={job.uid}, framework={framework}): {e}",
        exc_info=True,  # ← 補這行
    )
    error_code_str = GrcErrorCode.GRC_DOCX_PARSE_FAILED.value[1]
    error_msg = str(e)
    self._parse_job.write_error(job.uid, error_code_str, error_msg, user_context.login_name)
    # ... (rest unchanged)

parsed_ssp = adapter.adapt(parsed, structure, candidates)

from docx import Document as _Document
doc = _Document(file_path)
parsed_ssp = adapter.adapt(parsed, structure, candidates, doc=doc)

# scripts/smoke_docx_import.py
"""Standalone smoke test — bypass FE/HTTP, directly call upload_and_parse on
the customer docx and print parsed_result.

Run:
    set -a; source .env; set +a
    poetry run python scripts/smoke_docx_import.py
"""
import json, sys
from pathlib import Path
from werkzeug.datastructures import FileStorage
from main_app import create_app

CUSTOMER_DOCX = Path('docs/reference/亞航-CMMC-SSP-20260520-1會議討論版.docx')

def main():
    app = create_app()
    with app.app_context():
        # ... set up user context (blsadmin tenant 102) ...
        # ... call app_service.upload_and_parse(...) ...
        # ... print result['parsed_result'] keys ...
        pass

if __name__ == '__main__':
    main()

# 取 blsadmin token 從現有開發環境 .env / 重 login API
TOKEN="..."  # 從 BE log 抓或 重打 login

# Upload + parse
curl -sS -X POST http://localhost:8000/api/1.0/ssp-docx-imports/parse \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Tenant-Id: 102" \
  -H "X-Org-Unit-Id: 98" \
  -F "file=@docs/reference/亞航-CMMC-SSP-20260520-1會議討論版.docx" \
  -F "framework=cmmc-l1" \
  -F "source_type=module_frame" \
  -F "mode=full" | tee /tmp/parse_response.json

# Extract parse_uid
UID=$(python -c "import json; print(json.load(open('/tmp/parse_response.json'))['data']['parse_uid'])")

# Get preview
curl -sS -X GET http://localhost:8000/api/1.0/ssp-docx-import/$UID \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Tenant-Id: 102" \
  -H "X-Org-Unit-Id: 98" | jq '.data | keys'

---
type: feat
modules: [oscal, ssp_docx_import]
commit: <multiple — list 各 step commit hash>
---