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

| 欄位 | 內容 |
|---|---|
| Branch | feature/real_doc_import |
| Stage | 1 of 2 |
| Date | 2026-05-24 (overnight) |
| Estimated | ~6h autonomous execution |
| Risk | Medium — 範圍局限在 parser/adapter，不動 confirm path |

## 執行順序

每 step 完成 + verify pass + commit 才下一步。**不 push**。

| Step | 描述 | Verify | Commit type |
|---|---|---|---|
| S0 | gitignore + unstage 客戶檔；建 feature folder；寫 design + plan | files exist | docs |
| S1 | `domain/oscal/parser/docx_section_extractors.py` — 7 個 extractor + helper | extractor 結構 OK (Python imports) | feat |
| S2 | `tests/oscal/test_docx_section_extractors.py` — unit test 用 reference docx | pytest green | test |
| S3 | `scripts/scrub_customer_docx.py` + `tests/data/oscal/customer_sample_scrubbed.docx` | 結構 identical / 敏感字串無 | tweak |
| S4 | `tests/oscal/test_docx_section_extractors_customer.py` — integration with scrubbed | pytest green | test |
| S5 | 改 `cmmc_ssp_adapter.py` — adapter.adapt() 末段呼叫 section extractors | adapter test green | feat |
| S6 | 改 `ssp_docx_import_app_service.py` — generic Exception 加 logger.error；整合 pipeline 驗 parsed_result 含新欄位 | 既有 docx import test green | feat |
| S7 | `scripts/smoke_docx_import.py` — Python script 直接 call upload_and_parse 用客戶檔，印 parsed_result | 印出 metadata + parties + leveraged + revision_history | — |
| S8 | 重啟 BE，tail log 驗 OK | log "Running on http://" + 無 ERROR | — |
| S9 | curl 模擬 FE 上傳客戶檔，驗 preview API | 200 + parsed_result 含 4 categories | — |
| S10 | changelog + handoff SUMMARY | files exist | docs |

## S1 — Extractor 規格

### File layout

```
domain/oscal/parser/docx_section_extractors.py
```

### Public API

```python
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."""
```

### Internal helpers

```python
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."""
```

### 規則細節

**Metadata Table #0**：

```python
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
```

**Parties Tables**：

```python
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 Table**：

```python
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 Category Table**：

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

**Revision History**：

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

### Detection priority

1. Section anchor first (H1 / H2 substring match on raw heading text — match both 中英文混合 cases)
2. Header row match second (col header normalize → in known set)
3. Cell content match last (free-text scan within section, e.g., organization_name)

### Empty / placeholder rules

- Cell text after strip = empty → skip that field (don't include in output dict)
- Row all-empty → skip row
- Row 全 placeholder（cell 文字符合 `^\[.*\]$` 整體）→ skip row
- 整個 anchor 區段 找不到 expected table → 對應 key 在 output 中為 `None` 或 `[]`（不噴 Exception）

## S2 — Unit test 規格

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

## S3 — Scrub Script

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

Verify after scrub:
- Table count identical
- Paragraph count identical
- Heading text identical (no scrub rules on H1/H2/H3)
- 無 `AIR ASIA` / `亞航` 字串

## S5 — Adapter 接駁規格

```python
# 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
```

**Key design choice**: adapter 需要 raw `Document` object 才能跑 extractor。兩條路：

- (a) adapter 在初始化時保留 doc reference
- (b) adapter.adapt() signature 加 `doc` kw arg

選 (b) — call site 是 `ssp_docx_import_app_service` 已有 doc，直接傳。

**ParsedLeveragedService 加 props 欄位**：在 `ssp_intermediate.py` 加 `props: Optional[Dict[str, Any]] = None`。其他 caller 沒讀就是 None，零影響。

## S6 — App Service 改動

### 6.1 補 logger.error

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

### 6.2 Pipeline 整合 — adapter call 傳 doc

既有 adapter call site (line 252):
```python
parsed_ssp = adapter.adapt(parsed, structure, candidates)
```

改為:
```python
from docx import Document as _Document
doc = _Document(file_path)
parsed_ssp = adapter.adapt(parsed, structure, candidates, doc=doc)
```

或更乾淨 — call site 之前已有 `structure = self._parser.extract_structure_from_file(file_path)` 開了一次 docx；可在 parser 加 method 返回 `(structure, doc)` tuple。但**保持改動最小**，本期就在 adapter call 之前另開 doc。

## S7 — Smoke test script

```python
# 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()
```

Smoke test 細節依執行時 app_context 實作（DI container resolve）。

## S9 — Curl 模擬 FE 上傳

```bash
# 取 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'
```

Verify: response 含 `metadata`、`parties`、`leveraged_services`、`revision_history` keys。

## S10 — Changelog + Handoff

### Changelog 結構

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

內容：本期 stage1 改動範圍、未做的 stage2 todo、明早 user 驗證指引。

### Handoff SUMMARY

`docs/features/FR-027-2605-docx-import-parity/handoff/2026-05-24-stage1-overnight-SUMMARY.md`

固定 section：
1. 上線結果摘要（commits / 改動範圍 / 行為差異）
2. 明早 user 驗證指引（curl / DB query / FE 上傳步驟）
3. 已知限制（哪些欄位仍未解 — info_systems / devices / ref_docs 等）
4. Stage 2 TODO 清單 + 工時估
5. 風險回顧（若有 S6 regression 失敗等）

## 失敗 / 中止策略

| 情境 | 處理 |
|---|---|
| S1~S6 任一 step verify red 連續 3 次 | 該 step revert，handoff 寫明狀況，不繼續下 step |
| S6 regression test 全紅 | revert S6 commit；保留 S1~S5（純新增 file），handoff 說明 stage1 partial ship |
| S8 BE 重啟卡住 | 不再嘗試自動修，handoff 詳列 commands |
| S9 curl 取不到 token | 跳過 curl，用 smoke test script 替代驗證 |

## 不做的事

- 不 push (除 user 明確授權)
- 不切 branch
- 不動 jedi-oscal 套件（props 加在主專案 `ssp_intermediate.py` 內）
- 不動 FE
- 不寫 confirm path 分流 / schema_version column / DB migration
- 不加新 endpoint
- 不動 devices / info_systems / ref_docs 樣板