SSP 控制項 / 檢查項目（AO）現況說明功能 — Implementation Plan

-- SSP Control Implementation 欄位異動
-- 1) catalog_control_id 改為 nullable
ALTER TABLE oscal.system_security_plan_control_implementations
    ALTER COLUMN catalog_control_id DROP NOT NULL;

-- 2) 新增欄位
ALTER TABLE oscal.system_security_plan_control_implementations
    ADD COLUMN IF NOT EXISTS control_identifier VARCHAR(100),
    ADD COLUMN IF NOT EXISTS control_origination VARCHAR(50) DEFAULT 'organization',
    ADD COLUMN IF NOT EXISTS remarks TEXT;

-- 3) 補回填 control_identifier（從 catalog_controls 反查）
UPDATE oscal.system_security_plan_control_implementations ci
SET control_identifier = cc.control_id
FROM oscal.catalog_controls cc
WHERE ci.catalog_control_id = cc.id
  AND ci.control_identifier IS NULL;

-- 4) control_identifier 設為 NOT NULL（回填完成後）
ALTER TABLE oscal.system_security_plan_control_implementations
    ALTER COLUMN control_identifier SET NOT NULL;

-- 5) 新增 index
CREATE INDEX IF NOT EXISTS ix_ssp_ctrl_impl_control_identifier
    ON oscal.system_security_plan_control_implementations (system_security_plan_id, control_identifier);

-- 6) 更新 unique constraint（改用 control_identifier 取代 catalog_control_id）
ALTER TABLE oscal.system_security_plan_control_implementations
    DROP CONSTRAINT IF EXISTS uq_ssp_ctrl_impl_ssp_catalog_control;
ALTER TABLE oscal.system_security_plan_control_implementations
    ADD CONSTRAINT uq_ssp_ctrl_impl_ssp_control_identifier
    UNIQUE (system_security_plan_id, control_identifier);

-- ============================================================
-- SSP Control Implementation Objectives 新表
-- ============================================================
CREATE TABLE IF NOT EXISTS oscal.ssp_control_implementation_objectives (
    id              SERIAL PRIMARY KEY,
    uid             UUID NOT NULL DEFAULT gen_random_uuid(),

    system_security_plan_id  INTEGER NOT NULL
        REFERENCES oscal.system_security_plans(id) ON DELETE CASCADE,
    control_implementation_id INTEGER NOT NULL
        REFERENCES oscal.system_security_plan_control_implementations(id)
        ON DELETE CASCADE,

    control_identifier    VARCHAR(100) NOT NULL,
    statement_identifier  VARCHAR(100) NOT NULL,

    implementation_status      VARCHAR(50),
    implementation_description TEXT,
    remarks                    TEXT,
    reference_documents        JSONB DEFAULT '[]'::jsonb,

    created_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at     TIMESTAMPTZ NOT NULL DEFAULT now(),
    created_user   VARCHAR(100),
    updated_user   VARCHAR(100),

    CONSTRAINT uq_ssp_obj_ctrl_impl_stmt
        UNIQUE (control_implementation_id, statement_identifier)
);

CREATE INDEX IF NOT EXISTS ix_ssp_obj_uid
    ON oscal.ssp_control_implementation_objectives (uid);
CREATE INDEX IF NOT EXISTS ix_ssp_obj_ctrl_impl_id
    ON oscal.ssp_control_implementation_objectives (control_implementation_id);
CREATE INDEX IF NOT EXISTS ix_ssp_obj_ssp_ctrl_id
    ON oscal.ssp_control_implementation_objectives (system_security_plan_id, control_identifier);

control_identifier: Mapped[str] = mapped_column(
    String(100),
    nullable=False,
    comment="控制項編號，如 AC.L1-3.1.1",
)

control_origination: Mapped[Optional[str]] = mapped_column(
    String(50),
    default="organization",
    comment="organization | system-specific | customer-configured | inherited | shared",
)

remarks: Mapped[Optional[str]] = mapped_column(
    Text,
    comment="補充備註",
)

objectives: Mapped[list["OscalSspControlImplementationObjective"]] = relationship(
    "OscalSspControlImplementationObjective",
    back_populates="control_implementation",
    cascade="all, delete-orphan",
    lazy="select",
)

import uuid
from typing import Optional, TYPE_CHECKING
from datetime import datetime

from sqlalchemy import (
    String, Text, UUID, ForeignKey, DateTime, func,
    Index, UniqueConstraint, Integer,
)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship

from jedi_common.session.database.declarative_base import Base

if TYPE_CHECKING:
    from jedi_oscal.infra.model.ssp.ssp import OscalSystemSecurityPlan
    from jedi_oscal.infra.model.ssp.ssp_control_implementation import (
        OscalSystemSecurityPlanControlImplementation,
    )


class OscalSspControlImplementationObjective(Base):
    """SSP Control Implementation Objective (AO 層級)"""

    __tablename__ = "ssp_control_implementation_objectives"
    __table_args__ = (
        UniqueConstraint(
            "control_implementation_id", "statement_identifier",
            name="uq_ssp_obj_ctrl_impl_stmt",
        ),
        Index("ix_ssp_obj_uid", "uid"),
        Index("ix_ssp_obj_ctrl_impl_id", "control_implementation_id"),
        Index("ix_ssp_obj_ssp_ctrl_id", "system_security_plan_id", "control_identifier"),
        {"schema": "oscal"},
    )

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    uid: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True), default=uuid.uuid4, unique=True, nullable=False,
    )

    system_security_plan_id: Mapped[int] = mapped_column(
        ForeignKey("oscal.system_security_plans.id", ondelete="CASCADE"),
        nullable=False,
    )
    control_implementation_id: Mapped[int] = mapped_column(
        ForeignKey("oscal.system_security_plan_control_implementations.id", ondelete="CASCADE"),
        nullable=False,
    )

    control_identifier: Mapped[str] = mapped_column(String(100), nullable=False)
    statement_identifier: Mapped[str] = mapped_column(String(100), nullable=False)

    implementation_status: Mapped[Optional[str]] = mapped_column(String(50))
    implementation_description: Mapped[Optional[str]] = mapped_column(Text)
    remarks: Mapped[Optional[str]] = mapped_column(Text)
    reference_documents: Mapped[Optional[list]] = mapped_column(
        JSONB, default=list, server_default="'[]'::jsonb",
    )

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), default=func.now(), nullable=False,
    )
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), default=func.now(), onupdate=func.now(), nullable=False,
    )
    created_user: Mapped[Optional[str]] = mapped_column(String(100))
    updated_user: Mapped[Optional[str]] = mapped_column(String(100))

    # Relationships
    control_implementation: Mapped["OscalSystemSecurityPlanControlImplementation"] = relationship(
        "OscalSystemSecurityPlanControlImplementation",
        back_populates="objectives",
    )
    system_security_plan: Mapped["OscalSystemSecurityPlan"] = relationship(
        "OscalSystemSecurityPlan",
    )

from .ssp_control_implementation_objective import OscalSspControlImplementationObjective

control_identifier: Optional[str] = None,
control_origination: Optional[str] = None,
remarks: Optional[str] = None,
objectives: Optional[list] = None,  # list["ControlImplementationObjectiveEntity"]

self.control_identifier = control_identifier
self.control_origination = control_origination
self.remarks = remarks
self.objectives = objectives or []

from datetime import datetime
from typing import Optional
import uuid


class ControlImplementationObjectiveEntity:
    """SSP Control Implementation Objective domain entity."""

    def __init__(
        self,
        id: Optional[int] = None,
        uid: Optional[uuid.UUID] = None,
        system_security_plan_id: Optional[int] = None,
        control_implementation_id: Optional[int] = None,
        control_identifier: Optional[str] = None,
        statement_identifier: Optional[str] = None,
        implementation_status: Optional[str] = None,
        implementation_description: Optional[str] = None,
        remarks: Optional[str] = None,
        reference_documents: Optional[list] = None,
        created_user: Optional[str] = None,
        updated_user: Optional[str] = None,
        created_at: Optional[datetime] = None,
        updated_at: Optional[datetime] = None,
    ):
        self.id = id
        self.uid = uid
        self.system_security_plan_id = system_security_plan_id
        self.control_implementation_id = control_implementation_id
        self.control_identifier = control_identifier
        self.statement_identifier = statement_identifier
        self.implementation_status = implementation_status
        self.implementation_description = implementation_description
        self.remarks = remarks
        self.reference_documents = reference_documents or []
        self.created_user = created_user
        self.updated_user = updated_user
        self.created_at = created_at
        self.updated_at = updated_at

from datetime import datetime
from typing import Optional
import uuid


class SspControlImplObjectiveQueryEntity:
    """Query container for filtering ControlImplementationObjective."""

    def __init__(
        self,
        id: Optional[int] = None,
        uid: Optional[uuid.UUID] = None,
        system_security_plan_id: Optional[int] = None,
        control_implementation_id: Optional[int] = None,
        control_identifier: Optional[str] = None,
        statement_identifier: Optional[str] = None,
        implementation_status: Optional[str] = None,
    ):
        self.id = id
        self.uid = uid
        self.system_security_plan_id = system_security_plan_id
        self.control_implementation_id = control_implementation_id
        self.control_identifier = control_identifier
        self.statement_identifier = statement_identifier
        self.implementation_status = implementation_status

control_identifier=getattr(model, "control_identifier", None),
control_origination=getattr(model, "control_origination", None),
remarks=getattr(model, "remarks", None),
objectives=ControlImplementationObjectiveMapper.to_list_entity(
    getattr(model, "objectives", []) or []
),

m.control_identifier = getattr(entity, "control_identifier", None)
m.control_origination = getattr(entity, "control_origination", None)
m.remarks = getattr(entity, "remarks", None)

from jedi_oscal.infra.mapper.ssp.control_implementation_objective_mapper import ControlImplementationObjectiveMapper

from typing import List, Optional

from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_entity import ControlImplementationObjectiveEntity
from jedi_oscal.infra.model.ssp.ssp_control_implementation_objective import OscalSspControlImplementationObjective


class ControlImplementationObjectiveMapper:
    @staticmethod
    def to_entity(model: OscalSspControlImplementationObjective) -> ControlImplementationObjectiveEntity:
        return ControlImplementationObjectiveEntity(
            id=getattr(model, "id", None),
            uid=getattr(model, "uid", None),
            system_security_plan_id=getattr(model, "system_security_plan_id", None),
            control_implementation_id=getattr(model, "control_implementation_id", None),
            control_identifier=getattr(model, "control_identifier", None),
            statement_identifier=getattr(model, "statement_identifier", None),
            implementation_status=getattr(model, "implementation_status", None),
            implementation_description=getattr(model, "implementation_description", None),
            remarks=getattr(model, "remarks", None),
            reference_documents=getattr(model, "reference_documents", None) or [],
            created_user=getattr(model, "created_user", None),
            updated_user=getattr(model, "updated_user", None),
            created_at=getattr(model, "created_at", None),
            updated_at=getattr(model, "updated_at", None),
        )

    @staticmethod
    def to_list_entity(models: list) -> list[ControlImplementationObjectiveEntity]:
        return [ControlImplementationObjectiveMapper.to_entity(m) for m in models]

    @staticmethod
    def to_model(
        entity: ControlImplementationObjectiveEntity,
        model: Optional[OscalSspControlImplementationObjective] = None,
    ) -> OscalSspControlImplementationObjective:
        m = model or OscalSspControlImplementationObjective()
        if getattr(entity, "uid", None) is not None:
            m.uid = entity.uid
        m.system_security_plan_id = entity.system_security_plan_id
        m.control_implementation_id = entity.control_implementation_id
        m.control_identifier = entity.control_identifier
        m.statement_identifier = entity.statement_identifier
        m.implementation_status = getattr(entity, "implementation_status", None)
        m.implementation_description = getattr(entity, "implementation_description", None)
        m.remarks = getattr(entity, "remarks", None)
        m.reference_documents = getattr(entity, "reference_documents", None) or []
        if hasattr(m, "created_user"):
            m.created_user = getattr(entity, "created_user", None)
        if hasattr(m, "updated_user"):
            m.updated_user = getattr(entity, "updated_user", None)
        return m

    @staticmethod
    def to_list_model(entities: list) -> list[OscalSspControlImplementationObjective]:
        return [ControlImplementationObjectiveMapper.to_model(e) for e in entities]

from jedi_common.session.database.repository.base_repository import IBaseRepo


class IControlImplementationObjectiveRepo(IBaseRepo):
    pass

from jedi_common.session.database.repository.base_repository_impl import BaseRepositoryImpl
from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_entity import ControlImplementationObjectiveEntity
from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_query_entity import SspControlImplObjectiveQueryEntity
from jedi_oscal.domain.repository.ssp.control_implementation_objective_repo import IControlImplementationObjectiveRepo
from jedi_oscal.infra.mapper.ssp.control_implementation_objective_mapper import ControlImplementationObjectiveMapper
from jedi_oscal.infra.model.ssp.ssp_control_implementation_objective import OscalSspControlImplementationObjective


class ControlImplementationObjectiveRepoImpl(
    IControlImplementationObjectiveRepo[ControlImplementationObjectiveEntity, SspControlImplObjectiveQueryEntity],
    BaseRepositoryImpl[
        ControlImplementationObjectiveEntity,
        SspControlImplObjectiveQueryEntity,
        OscalSspControlImplementationObjective,
        ControlImplementationObjectiveMapper,
    ],
):
    def __init__(self):
        super().__init__(
            mapper=ControlImplementationObjectiveMapper,
            model=OscalSspControlImplementationObjective,
        )

from jedi_oscal.infra.model.ssp.ssp_control_implementation import OscalSystemSecurityPlanControlImplementation

def get_by_ssp_and_identifier(self, ssp_id: int, control_identifier: str):
    """依 ssp_id + control_identifier 取得單筆，含 objectives"""
    model = self.session.query(OscalSystemSecurityPlanControlImplementation).filter(
        OscalSystemSecurityPlanControlImplementation.system_security_plan_id == ssp_id,
        OscalSystemSecurityPlanControlImplementation.control_identifier == control_identifier,
    ).first()
    if not model:
        return None
    return self.mapper.to_entity(model)

def get_all_by_ssp_id(self, ssp_id: int):
    """取得某 SSP 下所有 control implementations"""
    models = self.session.query(OscalSystemSecurityPlanControlImplementation).filter(
        OscalSystemSecurityPlanControlImplementation.system_security_plan_id == ssp_id,
    ).all()
    return self.mapper.to_list_entity(models)

from abc import abstractmethod

@abstractmethod
def get_by_ssp_and_identifier(self, ssp_id: int, control_identifier: str):
    pass

@abstractmethod
def get_all_by_ssp_id(self, ssp_id: int):
    pass

from .ssp_control_impl_objective_repo_impl import ControlImplementationObjectiveRepoImpl

from .control_implementation_objective_repo import IControlImplementationObjectiveRepo

def get_by_ssp_and_identifier(self, ssp_id: int, control_identifier: str):
    return self.control_implementation_repo.get_by_ssp_and_identifier(ssp_id, control_identifier)

def get_all_by_ssp_id(self, ssp_id: int):
    return self.control_implementation_repo.get_all_by_ssp_id(ssp_id)

from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_entity import ControlImplementationObjectiveEntity
from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_query_entity import SspControlImplObjectiveQueryEntity
from jedi_oscal.domain.repository.ssp import IControlImplementationObjectiveRepo


class ControlImplementationObjectiveDomainService:
    def __init__(self, repo: IControlImplementationObjectiveRepo):
        self.repo = repo

    def get_all(self, _filter: SspControlImplObjectiveQueryEntity = None):
        return self.repo.get_all_by_fields(_filter)

    def get_one(self, _filter: SspControlImplObjectiveQueryEntity):
        return self.repo.get_one_by_fields(_filter)

    def get_by_uid(self, uid: str):
        return self.repo.get_by_uid(uid)

    def get_by_id(self, id: int):
        return self.repo.get_by_id(id)

    def add(self, entity: ControlImplementationObjectiveEntity):
        return self.repo.add(entity)

    def update(self, entity: ControlImplementationObjectiveEntity):
        return self.repo.update(entity)

    def delete_by_id(self, id: int):
        return self.repo.delete_by_id(id)

    def delete_by_uid(self, uid: str):
        return self.repo.delete_by_uid(uid)

from .control_implementation_objective_domain_service import ControlImplementationObjectiveDomainService

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Optional

from jedi_oscal.app.dto.base.oscal_base_dto import OscalBaseDTO
from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_entity import ControlImplementationObjectiveEntity


@dataclass
class ControlImplementationObjectiveDTO(OscalBaseDTO):
    uid: Optional[str] = None
    system_security_plan_id: Optional[int] = None
    control_implementation_id: Optional[int] = None
    control_identifier: Optional[str] = None
    statement_identifier: Optional[str] = None
    implementation_status: Optional[str] = None
    implementation_description: Optional[str] = None
    remarks: Optional[str] = None
    reference_documents: list = field(default_factory=list)

    @staticmethod
    def from_entity(entity: ControlImplementationObjectiveEntity) -> "ControlImplementationObjectiveDTO":
        return ControlImplementationObjectiveDTO(
            id=getattr(entity, "id", None),
            uid=str(entity.uid) if getattr(entity, "uid", None) else None,
            system_security_plan_id=getattr(entity, "system_security_plan_id", None),
            control_implementation_id=getattr(entity, "control_implementation_id", None),
            control_identifier=getattr(entity, "control_identifier", None),
            statement_identifier=getattr(entity, "statement_identifier", None),
            implementation_status=getattr(entity, "implementation_status", None),
            implementation_description=getattr(entity, "implementation_description", None),
            remarks=getattr(entity, "remarks", None),
            reference_documents=getattr(entity, "reference_documents", None) or [],
            created_at=getattr(entity, "created_at", None),
            updated_at=getattr(entity, "updated_at", None),
            created_user=getattr(entity, "created_user", None),
            updated_user=getattr(entity, "updated_user", None),
        )

    @staticmethod
    def from_entity_list(entities: list) -> list["ControlImplementationObjectiveDTO"]:
        return [ControlImplementationObjectiveDTO.from_entity(e) for e in entities]

control_identifier: Optional[str] = None
control_origination: Optional[str] = None
remarks: Optional[str] = None
objectives: list = field(default_factory=list)  # list[ControlImplementationObjectiveDTO]

from jedi_oscal.app.dto.ssp.ssp_control_impl_objective_dto import ControlImplementationObjectiveDTO

control_identifier=getattr(entity, "control_identifier", None),
control_origination=getattr(entity, "control_origination", None),
remarks=getattr(entity, "remarks", None),
objectives=ControlImplementationObjectiveDTO.from_entity_list(
    getattr(entity, "objectives", []) or []
),

# ── SSP Control Implementation ──────────────────────────────────────
GRC_SSP_NOT_FOUND = ("SSP 系統安全計畫不存在", "GRC_404012")
GRC_SSP_CONTROL_IMPL_NOT_FOUND = ("SSP 控制項實作記錄不存在", "GRC_404013")
GRC_SSP_OBJECTIVE_NOT_FOUND = ("SSP 檢查項目記錄不存在", "GRC_404014")

from marshmallow import Schema, fields


class ObjectiveResponseSchema(Schema):
    uid = fields.Str()
    statement_identifier = fields.Str()
    implementation_status = fields.Str(allow_none=True)
    implementation_description = fields.Str(allow_none=True)
    remarks = fields.Str(allow_none=True)
    reference_documents = fields.List(fields.Dict(), load_default=[])
    created_at = fields.DateTime()
    updated_at = fields.DateTime()


class ObjectiveUpdateRequestSchema(Schema):
    implementation_status = fields.Str(load_default=None, allow_none=True)
    implementation_description = fields.Str(load_default=None, allow_none=True)
    remarks = fields.Str(load_default=None, allow_none=True)
    reference_documents = fields.List(fields.Dict(), load_default=None, allow_none=True)

control_identifier = fields.Str()
control_origination = fields.Str(allow_none=True)
remarks = fields.Str(allow_none=True)
from api.oscal.serializers.ssp.ssp_control_impl_objective import ObjectiveResponseSchema
# ...（放在檔案頂部 import 區）
objectives = fields.List(fields.Nested(ObjectiveResponseSchema), dump_default=[])

class ControlImplementationUpdateRequestSchema(Schema):
    implementation_status = fields.Str(load_default=None, allow_none=True)
    implementation_description = fields.Str(load_default=None, allow_none=True)
    remarks = fields.Str(load_default=None, allow_none=True)


class ControlImplementationListItemSchema(Schema):
    control_identifier = fields.Str()
    implementation_status = fields.Str(allow_none=True)
    implementation_description = fields.Str(allow_none=True)
    objective_count = fields.Int()
    objective_described_count = fields.Int()
    has_description = fields.Bool()


class ControlImplementationListSummarySchema(Schema):
    total = fields.Int()
    described = fields.Int()
    not_described = fields.Int()
    completion_rate = fields.Float()


class ControlImplementationListResponseSchema(Schema):
    items = fields.List(fields.Nested(ControlImplementationListItemSchema))
    summary = fields.Nested(ControlImplementationListSummarySchema)

from typing import Optional
from jedi_common.session.database.db import transaction
from jedi_common.handler.exception import NotFound
from common.code.grc_error_code import GrcErrorCode

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 jedi_oscal.domain.services.ssp.ssp_domain_service import SystemSecurityPlanDomainService
from jedi_oscal.domain.entity.ssp.ssp_control_implementation_entity import ControlImplementationEntity
from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_entity import ControlImplementationObjectiveEntity
from jedi_oscal.domain.entity.ssp.ssp_control_impl_objective_query_entity import SspControlImplObjectiveQueryEntity
from jedi_oscal.app.dto.ssp.ssp_control_implementation_dto import ControlImplementationDTO
from jedi_oscal.app.dto.ssp.ssp_control_impl_objective_dto import ControlImplementationObjectiveDTO


class SspControlImplementationService:
    def __init__(
        self,
        ssp_domain_service: SystemSecurityPlanDomainService,
        control_impl_domain_service: ControlImplementationDomainService,
        objective_domain_service: ControlImplementationObjectiveDomainService,
    ):
        self._ssp_ds = ssp_domain_service
        self._ctrl_ds = control_impl_domain_service
        self._obj_ds = objective_domain_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

    def _get_or_create_control_impl(self, ssp_id: int, control_identifier: str, curr_user: str):
        """取得或自動建立 control implementation"""
        entity = self._ctrl_ds.get_by_ssp_and_identifier(ssp_id, control_identifier)
        if entity:
            return entity
        new_entity = ControlImplementationEntity(
            system_security_plan_id=ssp_id,
            control_identifier=control_identifier,
            implementation_status="unknown",  # 不可為 None，ORM 欄位 nullable=False
            created_user=curr_user,
            updated_user=curr_user,
        )
        return self._ctrl_ds.add(new_entity)

    @transaction
    def get_control_implementation(self, ssp_uid: str, control_identifier: str):
        """API 2.1: 取得控制項的 SSP 現況說明（含 objectives）"""
        ssp = self._get_ssp_or_404(ssp_uid)
        entity = self._ctrl_ds.get_by_ssp_and_identifier(ssp.id, control_identifier)
        if not entity:
            raise NotFound(GrcErrorCode.GRC_SSP_CONTROL_IMPL_NOT_FOUND)
        return ControlImplementationDTO.from_entity(entity)

    @transaction
    def update_control_implementation(self, ssp_uid: str, control_identifier: str, curr_user: str, **kwargs):
        """API 2.2: 更新控制項層級的 SSP 現況說明（upsert）"""
        ssp = self._get_ssp_or_404(ssp_uid)
        entity = self._get_or_create_control_impl(ssp.id, control_identifier, curr_user)

        if "implementation_status" in kwargs:
            entity.implementation_status = kwargs["implementation_status"]
        if "implementation_description" in kwargs:
            entity.implementation_description = kwargs["implementation_description"]
        if "remarks" in kwargs:
            entity.remarks = kwargs["remarks"]
        entity.updated_user = curr_user

        updated = self._ctrl_ds.update(entity)
        return ControlImplementationDTO.from_entity(updated)

    @transaction
    def update_objective(self, ssp_uid: str, control_identifier: str, statement_identifier: str, curr_user: str, **kwargs):
        """API 2.3: 更新 AO 層級的 SSP 現況說明（upsert）"""
        ssp = self._get_ssp_or_404(ssp_uid)
        ctrl = self._get_or_create_control_impl(ssp.id, control_identifier, curr_user)

        # 查詢是否已存在
        query = SspControlImplObjectiveQueryEntity(
            control_implementation_id=ctrl.id,
            statement_identifier=statement_identifier,
        )
        obj_entity = self._obj_ds.get_one(query)

        if obj_entity:
            if "implementation_status" in kwargs:
                obj_entity.implementation_status = kwargs["implementation_status"]
            if "implementation_description" in kwargs:
                obj_entity.implementation_description = kwargs["implementation_description"]
            if "remarks" in kwargs:
                obj_entity.remarks = kwargs["remarks"]
            if "reference_documents" in kwargs and kwargs["reference_documents"] is not None:
                obj_entity.reference_documents = kwargs["reference_documents"]
            obj_entity.updated_user = curr_user
            updated = self._obj_ds.update(obj_entity)
        else:
            new_obj = ControlImplementationObjectiveEntity(
                system_security_plan_id=ssp.id,
                control_implementation_id=ctrl.id,
                control_identifier=control_identifier,
                statement_identifier=statement_identifier,
                implementation_status=kwargs.get("implementation_status"),
                implementation_description=kwargs.get("implementation_description"),
                remarks=kwargs.get("remarks"),
                reference_documents=kwargs.get("reference_documents") or [],
                created_user=curr_user,
                updated_user=curr_user,
            )
            updated = self._obj_ds.add(new_obj)

        return ControlImplementationObjectiveDTO.from_entity(updated)

    @transaction
    def add_objective_document(self, ssp_uid: str, control_identifier: str, statement_identifier: str, file_ref: dict, curr_user: str):
        """API 2.4: 新增程序書關聯到 objective 的 reference_documents"""
        ssp = self._get_ssp_or_404(ssp_uid)
        ctrl = self._get_or_create_control_impl(ssp.id, control_identifier, curr_user)

        query = SspControlImplObjectiveQueryEntity(
            control_implementation_id=ctrl.id,
            statement_identifier=statement_identifier,
        )
        obj_entity = self._obj_ds.get_one(query)
        if not obj_entity:
            raise NotFound(GrcErrorCode.GRC_SSP_OBJECTIVE_NOT_FOUND)

        docs = obj_entity.reference_documents or []
        docs.append(file_ref)
        obj_entity.reference_documents = docs
        obj_entity.updated_user = curr_user
        updated = self._obj_ds.update(obj_entity)
        return ControlImplementationObjectiveDTO.from_entity(updated)

    @transaction
    def remove_objective_document(self, ssp_uid: str, control_identifier: str, statement_identifier: str, file_id: str, curr_user: str):
        """API 2.5: 移除程序書關聯"""
        ssp = self._get_ssp_or_404(ssp_uid)
        ctrl = self._ctrl_ds.get_by_ssp_and_identifier(ssp.id, control_identifier)
        if not ctrl:
            raise NotFound(GrcErrorCode.GRC_SSP_CONTROL_IMPL_NOT_FOUND)

        query = SspControlImplObjectiveQueryEntity(
            control_implementation_id=ctrl.id,
            statement_identifier=statement_identifier,
        )
        obj_entity = self._obj_ds.get_one(query)
        if not obj_entity:
            raise NotFound(GrcErrorCode.GRC_SSP_OBJECTIVE_NOT_FOUND)

        docs = obj_entity.reference_documents or []
        obj_entity.reference_documents = [d for d in docs if d.get("file_id") != file_id]
        obj_entity.updated_user = curr_user
        self._obj_ds.update(obj_entity)

    @transaction
    def list_control_implementations(self, ssp_uid: str, control_identifiers: Optional[list] = None, status: Optional[str] = None, has_description: Optional[bool] = None):
        """API 2.6: 批次取得多個控制項的 SSP 現況"""
        ssp = self._get_ssp_or_404(ssp_uid)
        all_ctrls = self._ctrl_ds.get_all_by_ssp_id(ssp.id)

        # 篩選
        if control_identifiers:
            all_ctrls = [c for c in all_ctrls if c.control_identifier in control_identifiers]
        if status:
            all_ctrls = [c for c in all_ctrls if c.implementation_status == status]

        items = []
        described_count = 0
        for ctrl in all_ctrls:
            obj_count = len(ctrl.objectives) if ctrl.objectives else 0
            obj_described = sum(
                1 for o in (ctrl.objectives or [])
                if o.implementation_description
            )
            has_desc = bool(ctrl.implementation_description)

            if has_description is not None and has_desc != has_description:
                continue

            items.append({
                "control_identifier": ctrl.control_identifier,
                "implementation_status": ctrl.implementation_status,
                "implementation_description": ctrl.implementation_description,
                "objective_count": obj_count,
                "objective_described_count": obj_described,
                "has_description": has_desc,
            })
            if has_desc:
                described_count += 1

        total = len(items)
        return {
            "items": items,
            "summary": {
                "total": total,
                "described": described_count,
                "not_described": total - described_count,
                "completion_rate": round(described_count / total, 2) if total > 0 else 0,
            },
        }

from dependency_injector.wiring import inject, Provide
from flask import request
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_implementation import (
    ControlImplementationResponseSchema,
    ControlImplementationUpdateRequestSchema,
    ControlImplementationListResponseSchema,
)
from api.oscal.serializers.ssp.ssp_control_impl_objective import (
    ObjectiveResponseSchema,
    ObjectiveUpdateRequestSchema,
)
from app.oscal.service.ssp_control_implementation_service import SspControlImplementationService
from common.util.response_util import return_response
from di_containers.containers import Containers

import logging

logger = logging.getLogger(__name__)


class SspControlImplementationDetailRoute(MethodResource):
    """GET / PUT /ssp/<ssp_uid>/control-implementation/<control_identifier>"""

    @doc(description="取得控制項的 SSP 現況說明", tags=["SSP"])
    @marshal_with(ControlImplementationResponseSchema, apply=False)
    @jwt_required()
    @inject
    def get(
        self,
        ssp_uid: str,
        control_identifier: str,
        ssp_ctrl_service: SspControlImplementationService = Provide[
            Containers.oscal_container.ssp_control_implementation_service
        ],
    ):
        result = ssp_ctrl_service.get_control_implementation(ssp_uid, control_identifier)
        return return_response(True, ControlImplementationResponseSchema().dump(result))

    @doc(description="更新控制項層級的 SSP 現況說明", tags=["SSP"])
    @use_kwargs(ControlImplementationUpdateRequestSchema, location="json", apply=False)
    @jwt_required()
    @inject
    def put(
        self,
        ssp_uid: str,
        control_identifier: str,
        ssp_ctrl_service: SspControlImplementationService = Provide[
            Containers.oscal_container.ssp_control_implementation_service
        ],
    ):
        user = get_user_context()
        payload = request.get_json(silent=True) or {}
        data = ControlImplementationUpdateRequestSchema().load(payload)

        result = ssp_ctrl_service.update_control_implementation(
            ssp_uid, control_identifier, user.login_name, **data
        )
        return return_response(True, ControlImplementationResponseSchema().dump(result))


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

    @doc(description="批次取得多個控制項的 SSP 現況", tags=["SSP"])
    @marshal_with(ControlImplementationListResponseSchema, apply=False)
    @jwt_required()
    @inject
    def get(
        self,
        ssp_uid: str,
        ssp_ctrl_service: SspControlImplementationService = Provide[
            Containers.oscal_container.ssp_control_implementation_service
        ],
    ):
        control_ids_str = request.args.get("control_identifiers")
        control_ids = control_ids_str.split(",") if control_ids_str else None
        status = request.args.get("status")
        has_desc_str = request.args.get("has_description")
        has_description = None
        if has_desc_str is not None:
            has_description = has_desc_str.lower() == "true"

        result = ssp_ctrl_service.list_control_implementations(
            ssp_uid,
            control_identifiers=control_ids,
            status=status,
            has_description=has_description,
        )
        return return_response(True, ControlImplementationListResponseSchema().dump(result))


class SspObjectiveDetailRoute(MethodResource):
    """PUT /ssp/<ssp_uid>/control-implementation/<control_identifier>/objective/<statement_identifier>"""

    @doc(description="更新 AO 層級的 SSP 現況說明", tags=["SSP"])
    @use_kwargs(ObjectiveUpdateRequestSchema, location="json", apply=False)
    @jwt_required()
    @inject
    def put(
        self,
        ssp_uid: str,
        control_identifier: str,
        statement_identifier: str,
        ssp_ctrl_service: SspControlImplementationService = Provide[
            Containers.oscal_container.ssp_control_implementation_service
        ],
    ):
        user = get_user_context()
        payload = request.get_json(silent=True) or {}
        data = ObjectiveUpdateRequestSchema().load(payload)

        result = ssp_ctrl_service.update_objective(
            ssp_uid, control_identifier, statement_identifier, user.login_name, **data
        )
        return return_response(True, ObjectiveResponseSchema().dump(result))


class SspObjectiveDocumentRoute(MethodResource):
    """DELETE /ssp/<ssp_uid>/control-implementation/<ci>/objective/<si>/document/<file_id>"""

    @doc(description="移除程序書關聯", tags=["SSP"])
    @jwt_required()
    @inject
    def delete(
        self,
        ssp_uid: str,
        control_identifier: str,
        statement_identifier: str,
        file_id: str,
        ssp_ctrl_service: SspControlImplementationService = Provide[
            Containers.oscal_container.ssp_control_implementation_service
        ],
    ):
        user = get_user_context()
        ssp_ctrl_service.remove_objective_document(
            ssp_uid, control_identifier, statement_identifier, file_id, user.login_name
        )
        return return_response(True, {})

from jedi_oscal.infra.repository.ssp import ControlImplementationRepoImpl, ControlImplementationObjectiveRepoImpl
from jedi_oscal.domain.services.ssp import ControlImplementationDomainService, ControlImplementationObjectiveDomainService
from app.oscal.service.ssp_control_implementation_service import SspControlImplementationService

# SSP Control Implementation
control_implementation_repo = providers.Singleton(ControlImplementationRepoImpl)
control_implementation_objective_repo = providers.Singleton(ControlImplementationObjectiveRepoImpl)

control_implementation_domain_service = providers.Factory(
    ControlImplementationDomainService,
    control_implementation_repo=control_implementation_repo,
)

control_implementation_objective_domain_service = providers.Factory(
    ControlImplementationObjectiveDomainService,
    repo=control_implementation_objective_repo,
)

ssp_control_implementation_service = providers.Factory(
    SspControlImplementationService,
    ssp_domain_service=ssp_domain_service,
    control_impl_domain_service=control_implementation_domain_service,
    objective_domain_service=control_implementation_objective_domain_service,
)

from api.oscal.routes.ssp.ssp_control_implementation_route import (
    SspControlImplementationDetailRoute,
    SspControlImplementationListRoute,
    SspObjectiveDetailRoute,
    SspObjectiveDocumentRoute,
)

# SSP Control Implementation
api.add_resource(
    SspControlImplementationListRoute,
    '/ssp/<string:ssp_uid>/control-implementations',
)
api.add_resource(
    SspControlImplementationDetailRoute,
    '/ssp/<string:ssp_uid>/control-implementation/<string:control_identifier>',
)
api.add_resource(
    SspObjectiveDetailRoute,
    '/ssp/<string:ssp_uid>/control-implementation/<string:control_identifier>/objective/<string:statement_identifier>',
)
api.add_resource(
    SspObjectiveDocumentRoute,
    '/ssp/<string:ssp_uid>/control-implementation/<string:control_identifier>/objective/<string:statement_identifier>/document/<string:file_id>',
)

control_impl_domain_service: ControlImplementationDomainService,
objective_domain_service: ControlImplementationObjectiveDomainService,

control_impl_domain_service=oscal_container.control_implementation_domain_service,
objective_domain_service=oscal_container.control_implementation_objective_domain_service,

# Step 7.5：初始化 SSP Control Implementations + Objectives
if ap and ssp:
    for group in (ap.groups or []):
        for control in (group.controls or []):
            ctrl_entity = ControlImplementationEntity(
                system_security_plan_id=ssp.id,
                control_identifier=control.control_id,
                catalog_control_id=getattr(control, "catalog_control_id", None),
                implementation_status="unknown",  # 不可為 None，ORM 欄位 nullable=False
                created_user=curr_user,
                updated_user=curr_user,
            )
            ctrl_entity = self._ctrl_impl_ds.add(ctrl_entity)

    # 重新取得帶 objectives 用的 ctrl map
    all_ctrls = self._ctrl_impl_ds.get_all_by_ssp_id(ssp.id)
    ctrl_map = {c.control_identifier: c for c in all_ctrls}

    for task in (ap.tasks or []):
        # 找到 task 對應的 control_identifier
        # task 透過 assessment_task_controls 關聯到 ap_control
        for task_ctrl in (getattr(task, "controls", []) or []):
            ctrl_id_str = getattr(task_ctrl, "control_id", None)
            if not ctrl_id_str:
                continue
            parent_ctrl = ctrl_map.get(ctrl_id_str)
            if not parent_ctrl:
                continue

            obj_entity = ControlImplementationObjectiveEntity(
                system_security_plan_id=ssp.id,
                control_implementation_id=parent_ctrl.id,
                control_identifier=ctrl_id_str,
                statement_identifier=task.task_code or task.title or str(task.id),
                implementation_status=None,
                created_user=curr_user,
                updated_user=curr_user,
            )
            self._obj_ds.add(obj_entity)

cd ~/Projects/Jedicogy/module/jedi-python-package/jedi-oscal
pip install -e .

psql -h $DB_HOST -U $DB_USER -d compliance_manager -f scripts/sql/ssp_control_implementation_migration.sql

python main_app.py

# 取得 control implementations 列表
curl -H "Authorization: Bearer $TOKEN" \
  http://localhost:8000/api/1.0/ssp/{ssp_uid}/control-implementations

# 更新控制項描述
curl -X PUT -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
  -d '{"implementation_status":"implemented","implementation_description":"test"}' \
  http://localhost:8000/api/1.0/ssp/{ssp_uid}/control-implementation/AC.L1-3.1.1

# 更新 AO 描述
curl -X PUT -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
  -d '{"implementation_description":"AO test"}' \
  http://localhost:8000/api/1.0/ssp/{ssp_uid}/control-implementation/AC.L1-3.1.1/objective/[a]

# SSP 現況
ssp_implementation_status: str = None,
ssp_implementation_description: str = None,
ssp_has_description: bool = False,

self.ssp_implementation_status = ssp_implementation_status
self.ssp_implementation_description = ssp_implementation_description
self.ssp_has_description = ssp_has_description

ssp_implementation_status: str = None,
ssp_implementation_description: str = None,
ssp_has_description: bool = False,

from jedi_oscal.infra.model.ssp.ssp_control_implementation import OscalSystemSecurityPlanControlImplementation
from jedi_oscal.infra.model.ssp.ssp_control_implementation_objective import OscalSspControlImplementationObjective

from jedi_oscal.infra.model.ap.assessment_plan import OscalAssessmentPlan

ap_row = session.query(OscalAssessmentPlan.ssp_id).filter(
    OscalAssessmentPlan.id == assessment_plan_id
).first()
ssp_id = ap_row.ssp_id if ap_row else None

# SSP 現況 subquery
ssp_impl_status = session.query(
    OscalSystemSecurityPlanControlImplementation.control_identifier,
    OscalSystemSecurityPlanControlImplementation.implementation_status,
    OscalSystemSecurityPlanControlImplementation.implementation_description,
).filter(
    OscalSystemSecurityPlanControlImplementation.system_security_plan_id == ssp_id,
).subquery("ssp_impl")

# 在 main query 加 outerjoin
query = query.outerjoin(
    ssp_impl_status,
    OscalAssessmentPlanControl.control_id == ssp_impl_status.c.control_identifier,
).add_columns(
    ssp_impl_status.c.implementation_status.label("ssp_implementation_status"),
    ssp_impl_status.c.implementation_description.label("ssp_implementation_description"),
)

ssp_implementation_status=getattr(row, "ssp_implementation_status", None),
ssp_implementation_description=getattr(row, "ssp_implementation_description", None),
ssp_has_description=bool(getattr(row, "ssp_implementation_description", None)),

# 批次取得 SSP objectives
if ssp_id:
    ssp_objectives = session.query(
        OscalSspControlImplementationObjective.statement_identifier,
        OscalSspControlImplementationObjective.control_identifier,
        OscalSspControlImplementationObjective.implementation_status,
        OscalSspControlImplementationObjective.implementation_description,
    ).filter(
        OscalSspControlImplementationObjective.system_security_plan_id == ssp_id,
    ).all()

    # 建立 lookup: (control_identifier, statement_identifier) → obj
    ssp_obj_map = {}
    for obj in ssp_objectives:
        key = (obj.control_identifier, obj.statement_identifier)
        ssp_obj_map[key] = obj

ssp_obj = ssp_obj_map.get((control_identifier, task.task_code))
ao_entity.ssp_implementation_status = ssp_obj.implementation_status if ssp_obj else None
ao_entity.ssp_implementation_description = ssp_obj.implementation_description if ssp_obj else None
ao_entity.ssp_has_description = bool(ssp_obj.implementation_description) if ssp_obj else False

ssp_implementation_status: Optional[str] = None
ssp_implementation_description: Optional[str] = None
ssp_has_description: bool = False

ssp_implementation_status: Optional[str] = None
ssp_implementation_description: Optional[str] = None
ssp_has_description: bool = False

ssp_implementation_status = fields.Str(allow_none=True)
ssp_implementation_description = fields.Str(allow_none=True)
ssp_has_description = fields.Bool(dump_default=False)

ssp_implementation_status = fields.Str(allow_none=True)
ssp_implementation_description = fields.Str(allow_none=True)
ssp_has_description = fields.Bool(dump_default=False)