SSP 控制項現況說明 — 批次匯入匯出 Implementation Plan

{
  "status": true,
  "data": {
    "valid_count": 45,
    "error_count": 3,
    "items": [
      {
        "row": 2,
        "control_identifier": "AC.L1-3.1.1",
        "objective_id": null,
        "implementation_status": "implemented",
        "implementation_description": "已建立存取控制機制...",
        "remarks": null,
        "errors": []
      },
      {
        "row": 5,
        "control_identifier": "IA.L1-3.5.1",
        "objective_id": "[a]",
        "implementation_status": "invalid_value",
        "implementation_description": "...",
        "remarks": null,
        "errors": [
          {"field": "implementation_status", "message": "無效的實作狀態值，允許值：implemented, partial, not_applicable, not_implemented, inherited, unknown"}
        ]
      }
    ]
  }
}

{
  "status": true,
  "data": {
    "updated_controls": 15,
    "updated_objectives": 30,
    "total": 45
  }
}

"""SSP 控制項現況 — 批次匯入匯出 Serializers"""
from marshmallow import Schema, fields


class ImportValidationErrorSchema(Schema):
    """單一欄位的驗證錯誤"""
    field = fields.String()
    message = fields.String()


class ImportRowSchema(Schema):
    """匯入驗證結果的單筆資料"""
    row = fields.Integer()
    control_identifier = fields.String()
    objective_id = fields.String(allow_none=True, load_default=None)
    implementation_status = fields.String(allow_none=True, load_default=None)
    implementation_description = fields.String(allow_none=True, load_default=None)
    remarks = fields.String(allow_none=True, load_default=None)
    obj_implementation_status = fields.String(allow_none=True, load_default=None)
    obj_implementation_description = fields.String(allow_none=True, load_default=None)
    obj_remarks = fields.String(allow_none=True, load_default=None)
    errors = fields.List(fields.Nested(ImportValidationErrorSchema), load_default=[])


class ImportValidationResponseSchema(Schema):
    """匯入驗證結果（Step 2 回傳）"""
    valid_count = fields.Integer()
    error_count = fields.Integer()
    items = fields.List(fields.Nested(ImportRowSchema))


class ImportConfirmItemSchema(Schema):
    """確認匯入的單筆資料（不含 errors）"""
    row = fields.Integer()
    control_identifier = fields.String(required=True)
    objective_id = fields.String(allow_none=True, load_default=None)
    implementation_status = fields.String(allow_none=True, load_default=None)
    implementation_description = fields.String(allow_none=True, load_default=None)
    remarks = fields.String(allow_none=True, load_default=None)
    obj_implementation_status = fields.String(allow_none=True, load_default=None)
    obj_implementation_description = fields.String(allow_none=True, load_default=None)
    obj_remarks = fields.String(allow_none=True, load_default=None)


class ImportConfirmRequestSchema(Schema):
    """確認匯入的 Request"""
    items = fields.List(fields.Nested(ImportConfirmItemSchema), required=True)


class ImportConfirmResponseSchema(Schema):
    """確認匯入的 Response"""
    updated_controls = fields.Integer()
    updated_objectives = fields.Integer()
    total = fields.Integer()

git add api/oscal/serializers/ssp/ssp_control_impl_import.py
git commit -m "feat(ssp-import): add marshmallow schemas for batch import/export"

"""SSP 控制項現況說明 — 批次匯入匯出 Service

負責：
- 匯出：查詢 SSP 所有控制項 + objectives，產生 Excel BytesIO
- 匯入：解析 Excel、驗證資料、批次 upsert

注意：讀取操作直接使用 domain services（避免穿透 SspControlImplementationService 的私有成員），
寫入操作複用 SspControlImplementationService 的 upsert 方法。
"""
import os
from io import BytesIO

from openpyxl import load_workbook
from openpyxl.styles import Font, PatternFill, Alignment
from openpyxl.workbook import Workbook
from openpyxl.worksheet.datavalidation import DataValidation

from jedi_common.handler.exception import BadRequestError, NotFound
from jedi_common.session.database.db import transaction
from jedi_common.session.database.session_context import get_session

from jedi_oscal.domain.services.ssp.ssp_domain_service import SystemSecurityPlanDomainService
from jedi_oscal.domain.services.ssp.control_implementation_domain_service import ControlImplementationDomainService
from jedi_oscal.domain.services.ssp.control_implementation_objective_domain_service import ControlImplementationObjectiveDomainService
from app.oscal.service.ssp_control_implementation_service import SspControlImplementationService
from common.code.grc_error_code import GrcErrorCode

VALID_STATUSES = [
    "implemented", "partial", "not_applicable",
    "not_implemented", "inherited", "unknown",
]

HEADER_LABELS = [
    "控制項識別碼",       # A
    "控制項名稱",         # B
    "實作狀態",           # C
    "現況描述",           # D
    "備註",               # E
    "檢查項目識別碼",     # F
    "檢查項目名稱",       # G
    "檢查項目實作狀態",   # H
    "檢查項目現況描述",   # I
    "檢查項目備註",       # J
]

HEADER_FONT_WHITE = Font(bold=True, size=11, color="FFFFFF")
HEADER_FILL = PatternFill("solid", start_color="4472C4")
READONLY_FILL = PatternFill("solid", start_color="F2F2F2")
WRAP_ALIGNMENT = Alignment(wrap_text=True, vertical="top")

MAX_FILE_SIZE = 10 * 1024 * 1024  # 10 MB


class SspControlImplImportService:
    """SSP 控制項現況批次匯入匯出"""

    def __init__(
        self,
        ssp_domain_service: SystemSecurityPlanDomainService,
        ctrl_impl_domain_service: ControlImplementationDomainService,
        objective_domain_service: ControlImplementationObjectiveDomainService,
        ssp_ctrl_impl_service: SspControlImplementationService,
    ):
        self._ssp_ds = ssp_domain_service
        self._ctrl_ds = ctrl_impl_domain_service
        self._obj_ds = objective_domain_service
        self._ssp_svc = ssp_ctrl_impl_service

    def _get_ssp_or_404(self, ssp_uid: str):
        ssp = self._ssp_ds.get_by_uid(ssp_uid)
        if not ssp:
            raise NotFound(GrcErrorCode.GRC_SSP_NOT_FOUND)
        return ssp

    @transaction
    def export_excel(self, ssp_uid: str) -> BytesIO:
        """匯出 SSP 控制項現況為 Excel"""
        session = get_session()
        ssp = self._get_ssp_or_404(ssp_uid)
        all_ctrls = self._ctrl_ds.get_all_by_ssp_id(ssp.id)

        # 查 catalog control titles
        ctrl_titles = self._load_catalog_titles(session, ssp.id)

        wb = Workbook()
        ws = wb.active
        ws.title = "SSP控制項現況"

        # Header row
        ws.append(HEADER_LABELS)
        for col_idx in range(1, len(HEADER_LABELS) + 1):
            cell = ws.cell(row=1, column=col_idx)
            cell.font = HEADER_FONT_WHITE
            cell.fill = HEADER_FILL
            cell.alignment = WRAP_ALIGNMENT

        # Column widths
        col_widths = [18, 25, 18, 40, 25, 15, 30, 18, 40, 25]
        for i, w in enumerate(col_widths, 1):
            ws.column_dimensions[chr(64 + i)].width = w

        # Status dropdown
        status_dv = DataValidation(
            type="list",
            formula1='"' + ",".join(VALID_STATUSES) + '"',
            allow_blank=True,
        )
        status_dv.error = "請選擇有效的實作狀態"
        status_dv.errorTitle = "無效值"
        ws.add_data_validation(status_dv)

        row_num = 2
        for ctrl in sorted(all_ctrls, key=lambda c: c.control_identifier):
            ws.cell(row=row_num, column=1, value=ctrl.control_identifier)
            ws.cell(row=row_num, column=2, value=ctrl_titles.get(ctrl.control_identifier, ""))
            ws.cell(row=row_num, column=3, value=ctrl.implementation_status if ctrl.implementation_status != "unknown" else "")
            ws.cell(row=row_num, column=4, value=ctrl.implementation_description or "")
            ws.cell(row=row_num, column=5, value=ctrl.remarks or "")
            for col in [1, 2]:
                ws.cell(row=row_num, column=col).fill = READONLY_FILL
            ws.cell(row=row_num, column=4).alignment = WRAP_ALIGNMENT
            status_dv.add(ws.cell(row=row_num, column=3))
            row_num += 1

            for obj in sorted(ctrl.objectives or [], key=lambda o: o.statement_identifier):
                ws.cell(row=row_num, column=1, value=ctrl.control_identifier)
                ws.cell(row=row_num, column=6, value=obj.statement_identifier)
                ws.cell(row=row_num, column=8, value=obj.implementation_status or "")
                ws.cell(row=row_num, column=9, value=obj.implementation_description or "")
                ws.cell(row=row_num, column=10, value=obj.remarks or "")
                for col in [1, 2, 6, 7]:
                    ws.cell(row=row_num, column=col).fill = READONLY_FILL
                ws.cell(row=row_num, column=9).alignment = WRAP_ALIGNMENT
                status_dv.add(ws.cell(row=row_num, column=8))
                row_num += 1

        ws.freeze_panes = "A2"
        excel_bytes = BytesIO()
        wb.save(excel_bytes)
        excel_bytes.seek(0)
        return excel_bytes

    def _load_catalog_titles(self, session, ssp_id: int) -> dict:
        """從 catalog 載入控制項標題（control_identifier → title）"""
        from jedi_oscal.infra.model.ssp.ssp_control_implementation import SspControlImplementation
        from jedi_oscal.infra.model.catalog.catalog_control import OscalCatalogControl

        rows = (
            session.query(
                SspControlImplementation.control_identifier,
                OscalCatalogControl.title,
            )
            .outerjoin(
                OscalCatalogControl,
                SspControlImplementation.catalog_control_id == OscalCatalogControl.id,
            )
            .filter(SspControlImplementation.system_security_plan_id == ssp_id)
            .all()
        )
        return {identifier: title for identifier, title in rows if title}

    @staticmethod
    def _validate_file(file):
        """驗證上傳檔案：副檔名 .xlsx、大小 <= 10MB"""
        filename = getattr(file, "filename", "") or ""
        if not filename.lower().endswith(".xlsx"):
            raise BadRequestError(GrcErrorCode.GRC_IMPORT_INVALID_FILE)
        file.seek(0, os.SEEK_END)
        size = file.tell()
        file.seek(0)
        if size > MAX_FILE_SIZE:
            raise BadRequestError(GrcErrorCode.GRC_IMPORT_FILE_TOO_LARGE)

    @transaction
    def validate_import(self, ssp_uid: str, file) -> dict:
        """解析 Excel 並驗證，回傳驗證結果預覽"""
        self._validate_file(file)
        ssp = self._get_ssp_or_404(ssp_uid)

        try:
            wb = load_workbook(file, data_only=True)
            ws = wb.active
        except Exception:
            raise BadRequestError(GrcErrorCode.GRC_IMPORT_INVALID_FILE)

        # 驗證 header 完全匹配
        headers = [cell.value for cell in ws[1]]
        if headers[:len(HEADER_LABELS)] != HEADER_LABELS:
            raise BadRequestError(GrcErrorCode.GRC_IMPORT_INVALID_FILE)

        # 載入該 SSP 現有資料
        all_ctrls = self._ctrl_ds.get_all_by_ssp_id(ssp.id)
        valid_identifiers = {c.control_identifier for c in all_ctrls}
        valid_objectives = {}
        for c in all_ctrls:
            for obj in (c.objectives or []):
                valid_objectives.setdefault(c.control_identifier, set()).add(
                    obj.statement_identifier
                )

        items = []
        current_ctrl_id = None

        for row_idx, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
            if not row or all(v is None or str(v).strip() == "" for v in row):
                continue

            ctrl_id = str(row[0]).strip() if row[0] else None
            obj_id = str(row[5]).strip() if row[5] else None
            errors = []

            if ctrl_id:
                current_ctrl_id = ctrl_id
            else:
                ctrl_id = current_ctrl_id

            if not ctrl_id:
                errors.append({"field": "control_identifier", "message": "控制項識別碼不可為空"})
            elif ctrl_id not in valid_identifiers:
                errors.append({"field": "control_identifier", "message": f"控制項 {ctrl_id} 不存在於此 SSP"})

            is_objective_row = bool(obj_id)

            if is_objective_row:
                if ctrl_id and obj_id and ctrl_id in valid_objectives:
                    if obj_id not in valid_objectives.get(ctrl_id, set()):
                        errors.append({"field": "objective_id", "message": f"檢查項目 {obj_id} 不存在於控制項 {ctrl_id}"})
                obj_status = str(row[7]).strip() if row[7] else None
                if obj_status and obj_status not in VALID_STATUSES:
                    errors.append({
                        "field": "obj_implementation_status",
                        "message": f"無效的實作狀態值，允許值：{', '.join(VALID_STATUSES)}",
                    })
                item = {
                    "row": row_idx,
                    "control_identifier": ctrl_id,
                    "objective_id": obj_id,
                    "obj_implementation_status": obj_status,
                    "obj_implementation_description": str(row[8]).strip() if row[8] else None,
                    "obj_remarks": str(row[9]).strip() if row[9] else None,
                    "errors": errors,
                }
            else:
                impl_status = str(row[2]).strip() if row[2] else None
                if impl_status and impl_status not in VALID_STATUSES:
                    errors.append({
                        "field": "implementation_status",
                        "message": f"無效的實作狀態值，允許值：{', '.join(VALID_STATUSES)}",
                    })
                item = {
                    "row": row_idx,
                    "control_identifier": ctrl_id,
                    "objective_id": None,
                    "implementation_status": impl_status,
                    "implementation_description": str(row[3]).strip() if row[3] else None,
                    "remarks": str(row[4]).strip() if row[4] else None,
                    "errors": errors,
                }
            items.append(item)

        valid_count = sum(1 for i in items if not i["errors"])
        error_count = sum(1 for i in items if i["errors"])
        return {"valid_count": valid_count, "error_count": error_count, "items": items}

    @transaction
    def confirm_import(self, ssp_uid: str, items: list, curr_user: str) -> dict:
        """確認匯入 — 批次 upsert（複用 SspControlImplementationService）"""
        updated_controls = 0
        updated_objectives = 0

        for item in items:
            ctrl_id = item.get("control_identifier")
            obj_id = item.get("objective_id")
            if not ctrl_id:
                continue

            if obj_id:
                kwargs = {}
                if item.get("obj_implementation_status"):
                    kwargs["implementation_status"] = item["obj_implementation_status"]
                if item.get("obj_implementation_description") is not None:
                    kwargs["implementation_description"] = item["obj_implementation_description"]
                if item.get("obj_remarks") is not None:
                    kwargs["remarks"] = item["obj_remarks"]
                if kwargs:
                    self._ssp_svc.update_objective(
                        ssp_uid, ctrl_id, obj_id, curr_user, **kwargs
                    )
                    updated_objectives += 1
            else:
                kwargs = {}
                if item.get("implementation_status"):
                    kwargs["implementation_status"] = item["implementation_status"]
                if item.get("implementation_description") is not None:
                    kwargs["implementation_description"] = item["implementation_description"]
                if item.get("remarks") is not None:
                    kwargs["remarks"] = item["remarks"]
                if kwargs:
                    self._ssp_svc.update_control_implementation(
                        ssp_uid, ctrl_id, curr_user, **kwargs
                    )
                    updated_controls += 1

        return {
            "updated_controls": updated_controls,
            "updated_objectives": updated_objectives,
            "total": updated_controls + updated_objectives,
        }

git add app/oscal/service/ssp_control_impl_import_service.py
git commit -m "feat(ssp-import): add import/export service with Excel generation, validation, batch upsert"

    # ── SSP Import ─────────────────────────────────────────────────────
    GRC_IMPORT_INVALID_FILE          = ("上傳檔案格式無效，請使用匯出的 Excel 範本",  "GRC_400001")
    GRC_IMPORT_FILE_TOO_LARGE        = ("上傳檔案超過大小限制（10MB）",              "GRC_400002")

git add common/code/grc_error_code.py
git commit -m "feat(ssp-import): add GRC_IMPORT_INVALID_FILE error code"

"""SSP 控制項現況 — 批次匯入匯出 Routes

端點：
- GET  /ssp/<ssp_uid>/control-implementations/export         → 匯出 Excel
- POST /ssp/<ssp_uid>/control-implementations/import         → 上傳 Excel + 驗證
- POST /ssp/<ssp_uid>/control-implementations/import/confirm → 確認匯入
"""
from dependency_injector.wiring import inject, Provide
from flask import request, send_file
from flask_apispec import MethodResource, doc, use_kwargs, marshal_with
from flask_jwt_extended import jwt_required
from jedi_common.session.auth.auth_context import get_user_context

from api.oscal.serializers.ssp.ssp_control_impl_import import (
    ImportValidationResponseSchema,
    ImportConfirmRequestSchema,
    ImportConfirmResponseSchema,
)
from app.oscal.service.ssp_control_impl_import_service import SspControlImplImportService
from common.util.response_util import return_response
from di_containers.containers import Containers


class SspControlImplExportRoute(MethodResource):
    """GET /ssp/<ssp_uid>/control-implementations/export"""

    @doc(description="匯出 SSP 控制項現況為 Excel", tags=["SSP Import/Export"])
    @jwt_required()
    @inject
    def get(
        self,
        ssp_uid: str,
        ssp_import_service: SspControlImplImportService = Provide[
            Containers.oscal_container.ssp_control_impl_import_service
        ],
    ):
        excel_bytes = ssp_import_service.export_excel(ssp_uid)
        return send_file(
            excel_bytes,
            mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            as_attachment=True,
            download_name=f"ssp-control-implementations-{ssp_uid[:8]}.xlsx",
        )


class SspControlImplImportRoute(MethodResource):
    """POST /ssp/<ssp_uid>/control-implementations/import"""

    @doc(description="上傳 Excel 並驗證匯入資料", tags=["SSP Import/Export"])
    @marshal_with(ImportValidationResponseSchema, apply=False)
    @jwt_required()
    @inject
    def post(
        self,
        ssp_uid: str,
        ssp_import_service: SspControlImplImportService = Provide[
            Containers.oscal_container.ssp_control_impl_import_service
        ],
    ):
        file = request.files.get("file")
        if not file:
            from jedi_common.handler.exception import BadRequestError
            from common.code.grc_error_code import GrcErrorCode
            raise BadRequestError(GrcErrorCode.GRC_IMPORT_INVALID_FILE)

        result = ssp_import_service.validate_import(ssp_uid, file)
        return return_response(True, ImportValidationResponseSchema().dump(result))


class SspControlImplImportConfirmRoute(MethodResource):
    """POST /ssp/<ssp_uid>/control-implementations/import/confirm"""

    @doc(description="確認匯入 — 批次寫入 SSP 控制項現況", tags=["SSP Import/Export"])
    @use_kwargs(ImportConfirmRequestSchema, location="json", apply=False)
    @marshal_with(ImportConfirmResponseSchema, apply=False)
    @jwt_required()
    @inject
    def post(
        self,
        ssp_uid: str,
        ssp_import_service: SspControlImplImportService = Provide[
            Containers.oscal_container.ssp_control_impl_import_service
        ],
    ):
        payload = request.get_json(silent=True) or {}
        data = ImportConfirmRequestSchema().load(payload)
        user = get_user_context()
        result = ssp_import_service.confirm_import(
            ssp_uid, data["items"], user.login_name
        )
        return return_response(True, ImportConfirmResponseSchema().dump(result))

git add api/oscal/routes/ssp/ssp_control_impl_import_route.py
git commit -m "feat(ssp-import): add export/import/confirm API routes"

from api.oscal.routes.ssp.ssp_control_impl_import_route import (
    SspControlImplExportRoute,
    SspControlImplImportRoute,
    SspControlImplImportConfirmRoute,
)

# SSP 控制項現況 — 批次匯入匯出
api.add_resource(SspControlImplExportRoute, '/ssp/<string:ssp_uid>/control-implementations/export')
api.add_resource(SspControlImplImportRoute, '/ssp/<string:ssp_uid>/control-implementations/import')
api.add_resource(SspControlImplImportConfirmRoute, '/ssp/<string:ssp_uid>/control-implementations/import/confirm')

from app.oscal.service.ssp_control_impl_import_service import SspControlImplImportService

# SSP 控制項現況 — 批次匯入匯出 Service
ssp_control_impl_import_service = providers.Factory(
    SspControlImplImportService,
    ssp_domain_service=ssp_domain_service,
    ctrl_impl_domain_service=control_implementation_domain_service,
    objective_domain_service=control_implementation_objective_domain_service,
    ssp_ctrl_impl_service=ssp_control_implementation_service,
)

git add api/oscal/__init__.py di_containers/oscal/oscal_containers.py
git commit -m "feat(ssp-import): register import/export routes and DI wiring"

# SSP 控制項現況批次匯入匯出

## 需求說明

新增 SSP 控制項現況說明（Control Implementation）的批次匯入匯出功能，
讓使用者能透過 Excel 檔案一次維護專案內所有控制項與檢查項目的現況描述。

## 變更範圍

### 新增檔案
- `app/oscal/service/ssp_control_impl_import_service.py` — 匯入匯出業務邏輯
- `api/oscal/routes/ssp/ssp_control_impl_import_route.py` — API route（匯出/匯入/確認）
- `api/oscal/serializers/ssp/ssp_control_impl_import.py` — Marshmallow schemas

### 修改檔案
- `api/oscal/__init__.py` — 註冊新 route
- `di_containers/oscal/oscal_containers.py` — 註冊 import service
- `common/code/grc_error_code.py` — 新增 GRC_IMPORT_INVALID_FILE error code

## API 變更

### 新增 API

| Method | URL | 說明 |
|--------|-----|------|
| GET | `/api/1.0/ssp/<ssp_uid>/control-implementations/export` | 匯出 Excel（下載） |
| POST | `/api/1.0/ssp/<ssp_uid>/control-implementations/import` | 上傳 Excel + 驗證預覽 |
| POST | `/api/1.0/ssp/<ssp_uid>/control-implementations/import/confirm` | 確認匯入（批次寫入） |

## 參考資訊

- 匯入匯出參考 `api/auth/routes/user_import_route.py` 三步驟模式
- Excel 產生使用 openpyxl（已安裝）
- 批次寫入複用 `SspControlImplementationService` 的 upsert 方法

git add docs/changelog/2026-03-28-ssp-batch-import-export.md
git commit -m "docs: add changelog for SSP batch import/export feature"

python main_app.py
# 確認 Swagger UI 可看到 SSP Import/Export 的三支 API

curl -H "Authorization: Bearer <token>" \
  http://localhost:8000/api/1.0/ssp/<ssp_uid>/control-implementations/export \
  -o test_export.xlsx
# 開啟 Excel 確認：
# - 有所有控制項和 objectives
# - 已有值的欄位有資料
# - 下拉選單可用
# - 灰底唯讀欄位正確

# 修改匯出的 Excel（填寫一些值、故意放錯誤值）
curl -X POST \
  -H "Authorization: Bearer <token>" \
  -F "file=@test_export.xlsx" \
  http://localhost:8000/api/1.0/ssp/<ssp_uid>/control-implementations/import
# 確認回傳驗證結果：valid_count, error_count, items 含 errors

# 用 Step 3 回傳的 valid items 作為 payload
curl -X POST \
  -H "Authorization: Bearer <token>" \
  -H "Content-Type: application/json" \
  -d '{"items": [...]}' \
  http://localhost:8000/api/1.0/ssp/<ssp_uid>/control-implementations/import/confirm
# 確認回傳 updated_controls, updated_objectives 數量
# 再呼叫 GET list API 確認資料已更新