Phase A2 Implementation Plan — Excel Parser + 解析 API

# 看 docx parse_jobs migration 內 RLS policy 寫法
grep -n "POLICY\|ALTER TABLE.*ssp_docx_parse_jobs\|ROW LEVEL" \
  scripts/sql/2026-05-01-ssp-docx-parser-consolidated.sql \
  scripts/sql/ssp_docx_parse_jobs_migration.sql 2>/dev/null

grep -n "def add_profile\|catalog_uid\|include_controls\|metadata" \
  ~/Projects/Jedicogy/module/jedi-python-package/jedi-oscal/jedi_oscal/app/services/profile/profile_service.py \
  | head -30

sed -n '120,190p' app/module_frame/service/module_frame_service.py

sed -n '155,200p' app/oscal/service/ssp_docx_import_app_service.py

grep -rn "party_reconciliation_service\|PartyReconciliation" di_containers/ 2>/dev/null | head -10
grep -n "def reconcile" domain/oscal/ app/oscal/ 2>/dev/null | head -10

# openpyxl 內 Data Validation list 跟 Form Control checkbox 哪個跨 Excel 版本穩定
poetry run python3 -c "
from openpyxl import Workbook
from openpyxl.worksheet.datavalidation import DataValidation
wb = Workbook()
ws = wb.active
dv = DataValidation(type='list', formula1='\"TRUE,FALSE\"', allow_blank=True)
ws.add_data_validation(dv)
dv.add('B2:B10')
ws['B2'] = True
wb.save('/tmp/checkbox_test.xlsx')
print('TRUE/FALSE dropdown OK')
"

# app/module_frame/excel_template/sheet_definitions.py
# SHEET_CONTROLS columns 在 control_id 後加：
SHEET_CONTROLS = SheetDef(
    sheet_name="07_控制項與AO",
    columns=(
        ColumnDef("statement_id", ..., hidden=True),          # A 欄
        ColumnDef("include_in_profile", "field.include_in_profile",
                  required=False, enum_values=("TRUE", "FALSE")),   # B 欄（A2 新增）
        ColumnDef("control_id", ..., required=True),                # C 欄（原 B → C）
        ColumnDef("control_name", ...),                              # D 欄
        ColumnDef("objective_id", ...),
        ColumnDef("objective_name", ...),
        ColumnDef("impl_status", ..., enum_values=(...), required=True),
        ColumnDef("statement", ..., required=True),
        ColumnDef("reference_doc", ...),
    ),
)

# app/module_frame/excel_template/sheet_definitions.py（或 generator 主檔）
TEMPLATE_VERSION = "v1.2.0"   # 原 v1.1.0

grep -rn "ws\\['[A-Z]2'\\]\\|cell(row=2,\\s*column=[0-9]" \
  tests/test_excel_template_*.py | head -20

@dataclass
class ValidationError:
    sheet: str
    row: int
    field: str
    code: str
    message: str
    blocking: bool = False

@dataclass
class ParsedExcel:
    template_version: str
    metadata: dict | None                  # 01 sheet
    parties_org: list[dict]                # 02 sheet
    parties_person: list[dict]             # 03 sheet
    devices: list[dict]                    # 04 sheet
    info_systems: list[dict]               # 05 sheet
    leveraged: list[dict]                  # 06 sheet
    controls_with_aos: list[dict]          # 07 sheet（含 objectives 巢狀）
    ref_docs: list[dict]                   # 08 sheet
    validation_errors: list[ValidationError]

    def to_dict(self) -> dict:
        """寫入 parse_job.parsed_result (JSONB) 用"""

class ExcelParser:
    def parse(self, file_path: str, source_type: str, source_uid: str) -> ParsedExcel:
        wb = load_workbook(file_path, data_only=True)
        template_version = self._read_template_version(wb)
        self._check_template_version(template_version)
        
        validation_errors: list[ValidationError] = []
        result = ParsedExcel(
            template_version=template_version,
            metadata=self._parse_metadata_sheet(wb, validation_errors),
            parties_org=self._parse_parties_org_sheet(wb, validation_errors),
            parties_person=self._parse_parties_person_sheet(wb, validation_errors),
            devices=self._parse_devices_sheet(wb, validation_errors),
            info_systems=self._parse_info_systems_sheet(wb, validation_errors),
            leveraged=self._parse_leveraged_sheet(wb, validation_errors),
            controls_with_aos=self._parse_controls_sheet(wb, validation_errors),
            ref_docs=self._parse_ref_docs_sheet(wb, validation_errors),
            validation_errors=validation_errors,
        )
        return result

VALID_PARTICIPANT_ROLES = {"manager", "reviewer", "auditor", "viewer"}

def validate_person_role(value: str, sheet: str, row: int) -> tuple[str, ValidationError | None]:
    role = (value or "").strip().lower()
    if not role:
        return "", None
    if role not in VALID_PARTICIPANT_ROLES:
        return role, ValidationError(
            sheet=sheet, row=row, field="role", code="INVALID_ROLE",
            message=f"role '{role}' 不在 manager/reviewer/auditor/viewer 範圍",
            blocking=True,
        )
    return role, None

_EXCEL_MAX_SIZE = 10 * 1024 * 1024  # 10MB
_PARSE_JOB_TTL_HOURS = 24

class SspExcelImportAppService:
    def __init__(self, parse_job_domain_service, parser, file_upload_service,
                 module_frame_domain_service, profile_service, module_frame_service,
                 party_domain_service, responsible_party_domain_service,
                 oscal_framework_version_domain_service, catalog_control_domain_service,
                 party_reconciliation_service=None, ...):
        self._parse_job = parse_job_domain_service
        self._parser = parser
        # ... 其他 dep
    
    @transaction
    def upload_and_parse(self, file, source_type, source_uid, metadata, user_context) -> dict:
        # 1. file validation（副檔名 / size）
        # 2. source 存在性 + 權限檢查
        # 3. 建 parse_job(status=pending)
        # 4. 走兩軌 storage：tempfile + file_upload_service
        # 5. self._parser.parse(tmp_path, source_type, source_uid) → ParsedExcel
        # 6. write_parsed_result(parsed.to_dict()) + status=awaiting_review
        # 7. unlink tempfile
        # 8. return {parse_uid, status, file_uid, summary}
    
    @transaction
    def get_parse_result(self, parse_uid, user_context) -> dict:
        # 1. get_one + tenant check + ttl check + is_active
        # 2. return job.to_dict() including parsed_result
    
    @transaction
    def discard_parse(self, parse_uid, user_context) -> dict:
        # 1. permission check
        # 2. domain_service.discard
        # 3. return {parse_uid, status: 'discarded'}
    
    @transaction
    def confirm_import(self, parse_uid, payload, user_context) -> dict:
        # 1. get + status / ttl / blocking-validation 守門
        # 2. dispatch by source_type:
        #    - 'framework_version' → _confirm_superset_flow
        #    - 'module_frame'      → _confirm_update_flow
        # 3. write_import_summary + status=completed
    
    def _confirm_superset_flow(self, job, parsed_result, decisions, overrides, user_context):
        # 1. 取 framework_version + catalog
        # 2. 從 controls_with_aos 篩納入 profile 的 control_ids → catalog_control_uid list
        # 3. ProfileService.add_profile({catalog_uid, include_controls, title, ...})
        # 4. ModuleFrameService.add_module_frame({oscal_profile_uid=新, oscal_framework_version_uid, name, ...})
        # 5. _write_parties / _write_items / _write_control_defaults / _write_ref_docs (context_type='module_frame', context_id=新 MF.id)
        # 6. return import_summary
    
    def _confirm_update_flow(self, job, parsed_result, decisions, overrides, user_context):
        # 1. 取既有 MF（已驗存在）
        # 2. 不動 profile
        # 3. _write_parties / items / control_defaults / ref_docs upsert
        # 4. return import_summary
    
    # _write_* helper：A2 直接寫，不靠 decisions 拆 append/update/skip（A5 補完）

class SspExcelImportParseRequestSchema(Schema):
    """POST /ssp-excel-imports/parse — form fields"""
    source_type = fields.Str(required=True, validate=validate.OneOf(['framework_version', 'module_frame']))
    source_uid = fields.Str(required=True)
    metadata = fields.Dict(required=False)   # JSON-encoded form field

class SspExcelImportParseResponseSchema(Schema):
    """POST /parse response"""
    parse_uid = fields.Str()
    status = fields.Str()
    file_uid = fields.Str()
    summary = fields.Dict(allow_none=True)
    error_code = fields.Str(allow_none=True)
    error_message = fields.Str(allow_none=True)

class SspExcelImportPreviewResponseSchema(Schema):
    """GET /<uid>"""
    uid = fields.Str()
    source_type = fields.Str()
    source_uid = fields.Str()
    status = fields.Str()
    file_uid = fields.Str()
    file_name = fields.Str()
    parsed_result = fields.Dict()
    created_at = fields.DateTime()
    ttl_expires_at = fields.DateTime()

class SspExcelImportConfirmRequestSchema(Schema):
    """POST /<uid>/confirm"""
    decisions = fields.List(fields.Dict(), required=False)
    overrides = fields.Dict(required=False)

class SspExcelImportConfirmResponseSchema(Schema):
    parse_uid = fields.Str()
    status = fields.Str()
    module_frame_uid = fields.Str()
    profile_uid = fields.Str(allow_none=True)
    import_summary = fields.Dict()

GRC_EXCEL_INVALID_FILE                = ("Excel 檔格式無效", "GRC_400065")
GRC_EXCEL_FILE_TOO_LARGE              = ("Excel 檔超過大小限制（10MB）", "GRC_400066")
GRC_EXCEL_SOURCE_TYPE_INVALID         = ("Excel 匯入 source_type 無效（需 framework_version 或 module_frame）", "GRC_400067")
GRC_EXCEL_SOURCE_UID_REQUIRED         = ("Excel 匯入 source_uid 必填", "GRC_400068")
GRC_EXCEL_TEMPLATE_VERSION_UNSUPPORTED = ("Excel 樣板版本不支援，請重新下載最新樣板", "GRC_400069")
GRC_EXCEL_VALIDATION_ERRORS_BLOCKING  = ("Excel 內容含阻斷級錯誤，請修正後重新上傳", "GRC_400070")
GRC_EXCEL_PARSE_JOB_NOT_FOUND         = ("Excel 解析任務不存在", "GRC_404033")
GRC_EXCEL_PARSE_JOB_NOT_AWAITING      = ("Excel 解析任務狀態不是 awaiting_review，無法確認匯入", "GRC_412013")
GRC_EXCEL_PARSE_JOB_TTL_EXPIRED       = ("Excel 解析任務已過期（24h）", "GRC_412014")

TOKEN=$(./scripts/dev_login.sh)
FW_VERSION_UID="..."   # dev DB 既有 framework_version
MF_UID="..."           # dev DB 既有 MF

# Superset flow — 拿 A1 superset 樣板填好後上傳
curl -X POST \
  -H "Authorization: Bearer $TOKEN" \
  -H "X-Tenant-ID: 102" \
  -F "file=@/tmp/test_superset.xlsx" \
  -F "source_type=framework_version" \
  -F "source_uid=$FW_VERSION_UID" \
  -F 'metadata={"target_mf_name":"A2 Smoke MF","target_group":"smoke","target_version":"0.1","target_frequency":"annual"}' \
  http://localhost:8000/api/1.0/ssp-excel-imports/parse

# 拿 parse_uid → GET → confirm
PARSE_UID="..."
curl -H "Authorization: Bearer $TOKEN" \
  http://localhost:8000/api/1.0/ssp-excel-import/$PARSE_UID

curl -X POST \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"decisions":[],"overrides":{}}' \
  http://localhost:8000/api/1.0/ssp-excel-import/$PARSE_UID/confirm

# MF update flow — 拿 A1 filled 樣板填好上傳
curl -X POST \
  -H "Authorization: Bearer $TOKEN" \
  -F "file=@/tmp/test_update.xlsx" \
  -F "source_type=module_frame" \
  -F "source_uid=$MF_UID" \
  http://localhost:8000/api/1.0/ssp-excel-imports/parse

# 重複 GET + confirm

-- 確認新 MF / profile 寫入
SELECT name, oscal_profile_uid, oscal_framework_version_uid FROM compliance.module_frames WHERE name = 'A2 Smoke MF';
SELECT uid, status FROM oscal.oscal_profiles WHERE uid = '<新 profile uid>';
SELECT COUNT(*) FROM oscal.oscal_parties WHERE uid IN (...);
SELECT COUNT(*) FROM oscal.ssp_system_implementation_items WHERE scope_type='module_frame' AND scope_id=<新 MF.id>;