[feat] negodata/backend: 카드 도메인 CRUD 추가 + 견적·auth 보강

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

negodata/backend/crud/card_crud.py

@@ -0,0 +1,104 @@
+from abc import ABC, abstractmethod
+from typing import Optional, Tuple
+
+from sqlalchemy import select, func, and_, or_, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.enums import ErrorType
+from common.logger import LOG
+from common.utils.gtime import GTime
+
+
+# 협상카드 CRUD. nego_cards/wild_cards 두 테이블에 공통으로 쓰는 제네릭 구현.
+class ICardCRUD(ABC):
+    @abstractmethod
+    async def search(self, cdb: AsyncSession, model, user_id, search, skip, limit) -> Tuple[ErrorType, list, int]:
+        pass
+
+    @abstractmethod
+    async def get_by_id(self, cdb: AsyncSession, model, pk_col, card_id) -> Tuple[ErrorType, object]:
+        pass
+
+    @abstractmethod
+    async def add(self, cdb: AsyncSession, card) -> ErrorType:
+        pass
+
+    @abstractmethod
+    async def update(self, cdb: AsyncSession, model, pk_col, card_id, data: dict) -> ErrorType:
+        pass
+
+    @abstractmethod
+    async def soft_delete(self, cdb: AsyncSession, model, pk_col, card_id) -> ErrorType:
+        pass
+
+
+class CardCRUD(ICardCRUD):
+    async def search(
+        self, cdb: AsyncSession, model, user_id, search: Optional[str], skip: int, limit: int
+    ) -> Tuple[ErrorType, list, int]:
+        try:
+            conditions = [model.deleted == False, model.user_id == user_id]  # noqa: E712
+            if search:
+                conditions.append(
+                    or_(
+                        model.name.ilike(f"%{search}%"),
+                        model.number.ilike(f"%{search}%"),
+                        model.script.ilike(f"%{search}%"),
+                    )
+                )
+            where = and_(*conditions)
+
+            cnt_err, cnt_rows = await DB_SESSION_MNG.execute(cdb, select(func.count()).select_from(model).where(where))
+            if cnt_err != ErrorType.SUCCESS:
+                return cnt_err, [], 0
+            total = int(cnt_rows[0] or 0) if cnt_rows else 0
+
+            list_err, rows = await DB_SESSION_MNG.execute(
+                cdb,
+                select(model).where(where).order_by(model.created_at.desc()).offset(skip).limit(limit),
+            )
+            if list_err != ErrorType.SUCCESS:
+                return list_err, [], 0
+            return ErrorType.SUCCESS, list(rows), total
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, [], 0
+
+    async def get_by_id(self, cdb: AsyncSession, model, pk_col, card_id) -> Tuple[ErrorType, object]:
+        try:
+            query = select(model).where(pk_col == card_id, model.deleted == False).limit(1)  # noqa: E712
+            err_type, row_list = await DB_SESSION_MNG.execute(cdb, query)
+            if err_type != ErrorType.SUCCESS:
+                return err_type, None
+            if len(row_list) != 1:
+                return ErrorType.DB_INVALID_KEY, None
+            return ErrorType.SUCCESS, row_list[0]
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, None
+
+    async def add(self, cdb: AsyncSession, card) -> ErrorType:
+        try:
+            return await DB_SESSION_MNG.insert(cdb, card)
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED
+
+    async def update(self, cdb: AsyncSession, model, pk_col, card_id, data: dict) -> ErrorType:
+        try:
+            if not data:
+                return ErrorType.SUCCESS
+            query = update(model).where(pk_col == card_id).values(**data)
+            return await DB_SESSION_MNG.add(cdb, query)
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED
+
+    async def soft_delete(self, cdb: AsyncSession, model, pk_col, card_id) -> ErrorType:
+        try:
+            query = update(model).where(pk_col == card_id).values(deleted=True, updated_at=GTime.UTC())
+            return await DB_SESSION_MNG.add(cdb, query)
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED

negodata/backend/crud/quotation_crud.py

@@ -6,14 +6,13 @@ from sqlalchemy import select, func, and_, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from common.database.db_session_manager import DB_SESSION_MNG
-from common.database.model.models import quotations
+from common.database.model.models import quotations, sessions, chats, nego_cards, wild_cards
 from common.enums import ErrorType
 from common.logger import LOG
 from common.utils.gtime import GTime
 
 
-# 견적 CRUD. quotations 테이블에는 company_id 가 없어 회사 스코프는 하지 않는다(토큰 검증만).
+# 견적 CRUD.
-# user_id 는 생성 시 소유자로 기록만 한다.
 class IQuotationCRUD(ABC):
     @abstractmethod
     async def search(
@@ -37,6 +36,18 @@ class IQuotationCRUD(ABC):
     async def soft_delete(self, cdb: AsyncSession, qt_id) -> ErrorType:
         pass
 
+    @abstractmethod
+    async def list_sessions(self, cdb: AsyncSession, qt_id) -> Tuple[ErrorType, list]:
+        pass
+
+    @abstractmethod
+    async def list_chats(self, cdb: AsyncSession, session_id) -> Tuple[ErrorType, list]:
+        pass
+
+    @abstractmethod
+    async def list_used_cards(self, cdb: AsyncSession, qt_id) -> Tuple[ErrorType, list]:
+        pass
+
 
 class QuotationCRUD(IQuotationCRUD):
     async def search(
@@ -114,3 +125,66 @@ class QuotationCRUD(IQuotationCRUD):
         except Exception as ex:
             LOG.e_no_callstack(ex)
             return ErrorType.DB_RUN_FAILED
+
+    # ----- 견적 상세: 세션 / 채팅 / 사용카드 (읽기 전용) -----
+    async def list_sessions(self, cdb: AsyncSession, qt_id) -> Tuple[ErrorType, list]:
+        try:
+            query = (
+                select(sessions)
+                .where(sessions.quotation_id == qt_id, sessions.deleted == False)  # noqa: E712
+                .order_by(sessions.created_at.asc())
+            )
+            err_type, rows = await DB_SESSION_MNG.execute(cdb, query)
+            if err_type != ErrorType.SUCCESS:
+                return err_type, []
+            return ErrorType.SUCCESS, list(rows)
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, []
+
+    async def list_chats(self, cdb: AsyncSession, session_id) -> Tuple[ErrorType, list]:
+        try:
+            query = (
+                select(chats)
+                .where(chats.session_id == session_id, chats.deleted == False)  # noqa: E712
+                .order_by(chats.seq.asc())
+            )
+            err_type, rows = await DB_SESSION_MNG.execute(cdb, query)
+            if err_type != ErrorType.SUCCESS:
+                return err_type, []
+            return ErrorType.SUCCESS, list(rows)
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, []
+
+    async def list_used_cards(self, cdb: AsyncSession, qt_id) -> Tuple[ErrorType, list]:
+        """견적의 세션들에서 실제 사용된 카드(chats.card_used_yn)를 카드 카탈로그와 조인.
+        반환: [(chat_row, card_id, name, script), ...].
+        card_type 1=nego_cards / 2=wild_cards 양쪽을 LEFT JOIN 해서 어느 쪽이든 잡는다.
+        """
+        try:
+            query = (
+                select(
+                    chats,
+                    func.coalesce(nego_cards.nego_card_id, wild_cards.wild_card_id).label("card_pk"),
+                    func.coalesce(nego_cards.name, wild_cards.name).label("card_name"),
+                    func.coalesce(nego_cards.script, wild_cards.script).label("card_script"),
+                )
+                .join(sessions, sessions.session_id == chats.session_id)
+                .outerjoin(nego_cards, and_(nego_cards.nego_card_id == chats.card_id, chats.card_type == 1))
+                .outerjoin(wild_cards, and_(wild_cards.wild_card_id == chats.card_id, chats.card_type == 2))
+                .where(
+                    sessions.quotation_id == qt_id,
+                    chats.card_used_yn == True,  # noqa: E712
+                    chats.deleted == False,  # noqa: E712
+                    sessions.deleted == False,  # noqa: E712
+                )
+                .order_by(chats.created_at.asc())
+            )
+            err_type, rows = await DB_SESSION_MNG.execute(cdb, query)
+            if err_type != ErrorType.SUCCESS:
+                return err_type, []
+            return ErrorType.SUCCESS, list(rows)
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, []

negodata/backend/router/router.py

@@ -12,6 +12,7 @@ from config.server_configs import web_server_config
 import router.v1.auth.account
 import router.v1.item.item
 import router.v1.supplier.supplier
+import router.v1.card.card
 import router.v1.quotation.quotation
 import router.v1.quotation_setting.quotation_setting
 import router.v1.enums.enums
@@ -60,6 +61,7 @@ async def healthz():
 app.include_router(router.v1.auth.account.router)
 app.include_router(router.v1.item.item.router)
 app.include_router(router.v1.supplier.supplier.router)
+app.include_router(router.v1.card.card.router)
 app.include_router(router.v1.quotation.quotation.router)
 app.include_router(router.v1.quotation_setting.quotation_setting.router)
 app.include_router(router.v1.enums.enums.router)

negodata/backend/router/v1/auth/protocol.py

@@ -53,4 +53,5 @@ class Res_Me(Res_WebPacketProtocol):
     email: Optional[str] = None
     contact_number: Optional[str] = None
     role: int = UserRole.USER.value
+    role_label: str = ""
     company: Optional[CompanyData] = Field(default=None)

negodata/backend/router/v1/card/card.py

@@ -0,0 +1,50 @@
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, Query
+
+from common.models.gmodel import PageParams, UserInfo
+from router.v1.validator.dependencies import IsValidAccessToken, RemoveNoneResponse
+from services.card_service import CardService
+from .protocol import (
+    Req_CreateCard,
+    Req_UpdateCard,
+    Res_Card,
+    Res_CardList,
+    Res_DeleteCard,
+)
+
+router = APIRouter(prefix="/v1/card", tags=["Card"], responses={404: {"description": "Not found"}})
+
+
+@router.get(path="/list", response_model=Res_CardList, summary="협상카드 목록")
+async def list_cards(
+    service: CardService = Depends(),
+    user_info: UserInfo = Depends(IsValidAccessToken),
+    search: str | None = Query(None, description="카드명/카드번호/스크립트 검색"),
+    pg: PageParams = Depends(),
+):
+    return RemoveNoneResponse(await service.list_cards(user_info.user_id, search, pg))
+
+
+@router.post(path="/create", response_model=Res_Card, summary="협상카드 등록")
+async def create_card(req: Req_CreateCard, service: CardService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
+    return RemoveNoneResponse(
+        await service.create_card(user_info.user_id, req.model_dump(exclude_unset=True))
+    )
+
+
+@router.get(path="/{card_id}", response_model=Res_Card, summary="협상카드 조회")
+async def get_card(card_id: UUID, service: CardService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
+    return RemoveNoneResponse(await service.get_card(user_info.user_id, str(card_id)))
+
+
+@router.patch(path="/update/{card_id}", response_model=Res_Card, summary="협상카드 수정")
+async def update_card(
+    card_id: UUID, req: Req_UpdateCard, service: CardService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)
+):
+    return RemoveNoneResponse(await service.update_card(user_info.user_id, str(card_id), req.model_dump(exclude_unset=True)))
+
+
+@router.delete(path="/delete/{card_id}", response_model=Res_DeleteCard, summary="협상카드 삭제")
+async def delete_card(card_id: UUID, service: CardService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
+    return RemoveNoneResponse(await service.delete_card(user_info.user_id, str(card_id)))

negodata/backend/router/v1/card/protocol.py

@@ -0,0 +1,63 @@
+import uuid
+from datetime import datetime
+from typing import Any, Optional
+
+from pydantic import ConfigDict
+
+from common.enums import CardStatus
+from common.models.gmodel import Res_PageProtocol, Res_WebPacketProtocol, WebPacketProtocol
+
+
+class CardProtocol(WebPacketProtocol):
+    pass
+
+
+class Req_CreateCard(CardProtocol):
+    is_wildcard: bool = False
+    name: Optional[str] = None
+    number: Optional[str] = None
+    script: Optional[str] = None
+    edit_script: Optional[Any] = None
+    status: int = CardStatus.ACTIVE.value  # 와일드카드 적용 여부(available 매핑). 일반카드는 무시.
+    condition: Optional[str] = None  # 와일드카드 전용
+    memo: Optional[str] = None  # 와일드카드 전용
+
+
+class Req_UpdateCard(CardProtocol):
+    name: Optional[str] = None
+    number: Optional[str] = None
+    script: Optional[str] = None
+    edit_script: Optional[Any] = None
+    status: Optional[int] = None
+    condition: Optional[str] = None
+    memo: Optional[str] = None
+
+
+# 통합 카드 표현(nego_cards + wild_cards 공통). nego_card_id 는 출처 테이블의 PK 를 그대로 담는다.
+class CardData(WebPacketProtocol):
+    model_config = ConfigDict(from_attributes=True)
+
+    nego_card_id: uuid.UUID  # 통합 식별자(일반=nego_card_id / 와일드=wild_card_id)
+    user_id: Optional[uuid.UUID] = None
+    is_wildcard: bool = False
+    name: Optional[str] = None
+    number: Optional[str] = None
+    script: Optional[str] = None
+    edit_script: Optional[Any] = None
+    status: int = CardStatus.ACTIVE.value
+    condition: Optional[str] = None
+    memo: Optional[str] = None
+    created_at: Optional[datetime] = None
+    updated_at: Optional[datetime] = None
+
+
+class Res_Card(Res_WebPacketProtocol):
+    card: Optional[CardData] = None
+
+
+class Res_CardList(Res_PageProtocol):
+    cards: list[CardData] = []
+
+
+class Res_DeleteCard(Res_WebPacketProtocol):
+    pass

negodata/backend/router/v1/quotation/protocol.py

@@ -4,7 +4,7 @@ from typing import Any, Optional
 
 from pydantic import ConfigDict
 
-from common.models.gmodel import Res_WebPacketProtocol, WebPacketProtocol
+from common.models.gmodel import Res_PageProtocol, Res_WebPacketProtocol, WebPacketProtocol
 
 
 class QuotationProtocol(WebPacketProtocol):
@@ -59,11 +59,8 @@ class Res_Quotation(Res_WebPacketProtocol):
     quotation: Optional[QuotationData] = None
 
 
-class Res_QuotationList(Res_WebPacketProtocol):
+class Res_QuotationList(Res_PageProtocol):
     quotations: list[QuotationData] = []
-    total: int = 0
-    page: int = 0
-    size: int = 0
 
 
 class Res_DeleteQuotation(Res_WebPacketProtocol):

negodata/backend/router/v1/quotation/quotation.py

@@ -3,7 +3,7 @@ from uuid import UUID
 
 from fastapi import APIRouter, Depends, Query
 
-from common.models.gmodel import UserInfo
+from common.models.gmodel import PageParams, UserInfo
 from router.v1.validator.dependencies import IsValidAccessToken, RemoveNoneResponse
 from services.quotation_service import QuotationService
 from .protocol import (
@@ -35,10 +35,9 @@ async def list_quotations(
     type: str | None = Query(None, description="유형 필터(정확히 일치)"),
     start_from: datetime | None = Query(None, description="시작일시 이후(ISO)"),
     start_to: datetime | None = Query(None, description="시작일시 이전(ISO)"),
-    page: int = Query(1, ge=1),
+    pg: PageParams = Depends(),
-    size: int = Query(20, ge=1, le=100),
 ):
-    return RemoveNoneResponse(await service.list_quotations(status, type, start_from, start_to, page, size))
+    return RemoveNoneResponse(await service.list_quotations(status, type, start_from, start_to, pg))
 
 
 @router.post(path="/create", response_model=Res_CreateQuotation, summary="견적 생성")

negodata/backend/services/auth_service.py

@@ -4,7 +4,7 @@ from fastapi import Depends
 
 from common.database.db_session_manager import DB_SESSION_MNG
 from common.database.model.models import users
-from common.enums import DBWRType, ErrorType, UserStatus, UserRole
+from common.enums import DBWRType, ErrorType, UserStatus, UserRole, ENUM_LABELS
 from common.logger import LOG
 from common.models.gmodel import UserInfo
 from crud.user_crud import IUserCRUD, UserCRUD
@@ -156,6 +156,7 @@ class AuthService:
         res.email = user.email
         res.contact_number = user.contact_number
         res.role = user.role
+        res.role_label = ENUM_LABELS.get(UserRole(user.role), str(user.role))
         res.company = company
         return res
 

negodata/backend/services/card_service.py

@@ -0,0 +1,204 @@
+import uuid
+
+from fastapi import Depends
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.database.model.models import nego_cards, wild_cards
+from common.enums import CardStatus, DBWRType, ErrorType
+from common.models.gmodel import PageParams
+from crud.card_crud import ICardCRUD, CardCRUD
+from router.v1.card.protocol import CardData, Res_Card, Res_CardList, Res_DeleteCard
+
+
+class CardService:
+    """협상카드 비즈니스 로직. nego_cards/wild_cards 두 테이블을 user_id 로 스코프하고
+    프론트용 단일 모델(CardData, is_wildcard 플래그)로 합친다."""
+
+    def __init__(self, card_crud: ICardCRUD = Depends(CardCRUD)):
+        self.card_crud = card_crud
+
+    # ---- 행 → 통합 CardData --------------------------------------------------
+    @staticmethod
+    def _nego_to_data(row) -> CardData:
+        # 일반 협상카드는 상태 개념이 없다(상시 적용) → ACTIVE 고정.
+        return CardData(
+            nego_card_id=row.nego_card_id,
+            user_id=row.user_id,
+            is_wildcard=False,
+            name=row.name,
+            number=row.number,
+            script=row.script,
+            edit_script=row.edit_script,
+            status=CardStatus.ACTIVE.value,
+            created_at=row.created_at,
+            updated_at=row.updated_at,
+        )
+
+    @staticmethod
+    def _wild_to_data(row) -> CardData:
+        # 와일드카드 적용 여부(available) → status(ACTIVE/INACTIVE) 매핑.
+        return CardData(
+            nego_card_id=row.wild_card_id,
+            user_id=row.user_id,
+            is_wildcard=True,
+            name=row.name,
+            number=row.number,
+            script=row.script,
+            edit_script=row.edit_script,
+            status=CardStatus.ACTIVE.value if row.available else CardStatus.INACTIVE.value,
+            condition=row.condition,
+            memo=row.memo,
+            created_at=row.created_at,
+            updated_at=row.updated_at,
+        )
+
+    # ---- 소유 카드 탐색(어느 테이블인지 모를 때) ------------------------------
+    async def _find_owned(self, user_uuid: uuid.UUID, card_id: uuid.UUID):
+        """card_id 를 nego_cards → wild_cards 순으로 찾고 소유권 확인.
+        (ErrorType, model, pk_col, row, is_wildcard) 반환."""
+        err, row = await DB_SESSION_MNG.execute_lambda(
+            nego_cards.DBType(),
+            DBWRType.DB_READ.value,
+            lambda s: self.card_crud.get_by_id(s, nego_cards, nego_cards.nego_card_id, card_id),
+        )
+        if err == ErrorType.SUCCESS and row is not None:
+            if row.user_id != user_uuid:
+                return ErrorType.CARD_NOT_FOUND, None, None, None, False
+            return ErrorType.SUCCESS, nego_cards, nego_cards.nego_card_id, row, False
+
+        err, row = await DB_SESSION_MNG.execute_lambda(
+            wild_cards.DBType(),
+            DBWRType.DB_READ.value,
+            lambda s: self.card_crud.get_by_id(s, wild_cards, wild_cards.wild_card_id, card_id),
+        )
+        if err == ErrorType.SUCCESS and row is not None:
+            if row.user_id != user_uuid:
+                return ErrorType.CARD_NOT_FOUND, None, None, None, True
+            return ErrorType.SUCCESS, wild_cards, wild_cards.wild_card_id, row, True
+
+        return ErrorType.CARD_NOT_FOUND, None, None, None, False
+
+    # ---- 목록 ----------------------------------------------------------------
+    async def list_cards(self, user_id: str, search, pg: PageParams) -> Res_CardList:
+        res = Res_CardList(page=pg.page, size=pg.size)
+        if not user_id:
+            return res
+        user_uuid = uuid.UUID(user_id)
+        # 합쳐서 정렬/페이징하므로 각 테이블에서 skip+limit 까지 받아온다(카드 수가 적어 충분).
+        fetch = pg.skip + pg.size
+
+        err_n, nego_rows, total_n = await DB_SESSION_MNG.execute_lambda(
+            nego_cards.DBType(),
+            DBWRType.DB_READ.value,
+            lambda s: self.card_crud.search(s, nego_cards, user_uuid, search, 0, fetch),
+        )
+        if err_n != ErrorType.SUCCESS:
+            res.result.SetResult(err_n)
+            return res
+
+        err_w, wild_rows, total_w = await DB_SESSION_MNG.execute_lambda(
+            wild_cards.DBType(),
+            DBWRType.DB_READ.value,
+            lambda s: self.card_crud.search(s, wild_cards, user_uuid, search, 0, fetch),
+        )
+        if err_w != ErrorType.SUCCESS:
+            res.result.SetResult(err_w)
+            return res
+
+        merged = [self._nego_to_data(r) for r in nego_rows] + [self._wild_to_data(r) for r in wild_rows]
+        merged.sort(key=lambda c: c.created_at or "", reverse=True)
+        res.cards = merged[pg.skip : pg.skip + pg.size]
+        res.total = total_n + total_w
+        return res
+
+    # ---- 단건 조회 -----------------------------------------------------------
+    async def get_card(self, user_id: str, card_id: str) -> Res_Card:
+        res = Res_Card()
+        err, _model, _pk, row, is_wild = await self._find_owned(uuid.UUID(user_id), uuid.UUID(card_id))
+        if err != ErrorType.SUCCESS:
+            res.result.SetResult(err)
+            return res
+        res.card = self._wild_to_data(row) if is_wild else self._nego_to_data(row)
+        return res
+
+    # ---- 등록 ----------------------------------------------------------------
+    async def create_card(self, user_id: str, data: dict) -> Res_Card:
+        res = Res_Card()
+        user_uuid = uuid.UUID(user_id)
+        is_wildcard = bool(data.get("is_wildcard", False))
+
+        common = dict(
+            user_id=user_uuid,
+            name=data.get("name"),
+            number=data.get("number"),
+            script=data.get("script"),
+            edit_script=data.get("edit_script"),
+        )
+        if is_wildcard:
+            card = wild_cards(
+                **common,
+                condition=data.get("condition"),
+                available=(data.get("status", CardStatus.ACTIVE.value) == CardStatus.ACTIVE.value),
+                memo=data.get("memo"),
+            )
+            model, pk_attr = wild_cards, "wild_card_id"
+        else:
+            card = nego_cards(**common)
+            model, pk_attr = nego_cards, "nego_card_id"
+
+        err = await DB_SESSION_MNG.execute_lambda_run(
+            [model.DBType()],
+            [lambda s: self.card_crud.add(s, card)],
+        )
+        if err != ErrorType.SUCCESS:
+            res.result.SetResult(err)
+            return res
+        # 서버 기본값(created_at 등)은 insert 후 객체에 실리지 않으므로 재조회.
+        return await self.get_card(user_id, str(getattr(card, pk_attr)))
+
+    # ---- 수정 ----------------------------------------------------------------
+    async def update_card(self, user_id: str, card_id: str, data: dict) -> Res_Card:
+        res = Res_Card()
+        user_uuid = uuid.UUID(user_id)
+        card_uuid = uuid.UUID(card_id)
+
+        err, model, pk_col, _row, is_wild = await self._find_owned(user_uuid, card_uuid)
+        if err != ErrorType.SUCCESS:
+            res.result.SetResult(err)
+            return res
+
+        # 해당 테이블에 있는 컬럼만 추린다(없는 필드는 무시). status → available(와일드 전용).
+        allowed = {"name", "number", "script", "edit_script"}
+        if is_wild:
+            allowed |= {"condition", "memo"}
+        payload = {k: v for k, v in data.items() if k in allowed}
+        if is_wild and "status" in data and data["status"] is not None:
+            payload["available"] = data["status"] == CardStatus.ACTIVE.value
+
+        err = await DB_SESSION_MNG.execute_lambda_run(
+            [model.DBType()],
+            [lambda s: self.card_crud.update(s, model, pk_col, card_uuid, payload)],
+        )
+        if err != ErrorType.SUCCESS:
+            res.result.SetResult(err)
+            return res
+        return await self.get_card(user_id, card_id)
+
+    # ---- 삭제(soft) ----------------------------------------------------------
+    async def delete_card(self, user_id: str, card_id: str) -> Res_DeleteCard:
+        res = Res_DeleteCard()
+        user_uuid = uuid.UUID(user_id)
+        card_uuid = uuid.UUID(card_id)
+
+        err, model, pk_col, _row, _is_wild = await self._find_owned(user_uuid, card_uuid)
+        if err != ErrorType.SUCCESS:
+            res.result.SetResult(err)
+            return res
+
+        err = await DB_SESSION_MNG.execute_lambda_run(
+            [model.DBType()],
+            [lambda s: self.card_crud.soft_delete(s, model, pk_col, card_uuid)],
+        )
+        if err != ErrorType.SUCCESS:
+            res.result.SetResult(err)
+        return res

negodata/backend/services/quotation_service.py

@@ -3,12 +3,16 @@ import uuid
 from fastapi import Depends
 
 from common.database.db_session_manager import DB_SESSION_MNG
-from common.database.model.models import quotations
+from common.database.model.models import quotations, sessions, chats
 from common.enums import DBWRType, ErrorType
+from common.models.gmodel import PageParams
 from crud.quotation_crud import IQuotationCRUD, QuotationCRUD
 from router.v1.quotation.protocol import (
     AsyncJob,
+    ChatMessageData,
+    QuotationCardData,
     QuotationData,
+    SessionData,
     Res_CreateQuotation,
     Res_DeleteQuotation,
     Res_Quotation,
@@ -42,14 +46,13 @@ class QuotationService:
             return ErrorType.QUOTATION_NOT_FOUND, None
         return ErrorType.SUCCESS, quotation
 
-    async def list_quotations(self, status, type_, start_from, start_to, page: int, size: int) -> Res_QuotationList:
+    async def list_quotations(self, status, type_, start_from, start_to, pg: PageParams) -> Res_QuotationList:
-        res = Res_QuotationList(page=page, size=size)
+        res = Res_QuotationList(page=pg.page, size=pg.size)
-        skip = (page - 1) * size
 
         err_type, rows, total = await DB_SESSION_MNG.execute_lambda(
             quotations.DBType(),
             DBWRType.DB_READ.value,
-            lambda s: self.quotation_crud.search(s, status, type_, start_from, start_to, skip, size),
+            lambda s: self.quotation_crud.search(s, status, type_, start_from, start_to, pg.skip, pg.size),
         )
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
@@ -154,31 +157,110 @@ class QuotationService:
         return res
 
     async def list_sessions(self, qt_id: str) -> Res_QuotationSessions:
-        # sessions 모델 미존재 — 존재 검증 후 빈 목록 반환(스텁).
         res = Res_QuotationSessions()
-        err_type, quotation = await self._fetch(uuid.UUID(qt_id))
+        qt_uuid = uuid.UUID(qt_id)
+        err_type, quotation = await self._fetch(qt_uuid)
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
             return res
+
+        err_type, rows = await DB_SESSION_MNG.execute_lambda(
+            sessions.DBType(),
+            DBWRType.DB_READ.value,
+            lambda s: self.quotation_crud.list_sessions(s, qt_uuid),
+        )
+        if err_type != ErrorType.SUCCESS:
+            res.result.SetResult(err_type)
+            return res
+
         res.qt_id = quotation.qt_id
-        res.sessions = []
+        # sessions.quotation_id → SessionData.qt_id 로 명시 매핑(컬럼명 불일치).
-        res.total = 0
+        res.sessions = [
+            SessionData(
+                session_id=r.session_id,
+                qt_id=r.quotation_id,
+                supplier_id=r.supplier_id,
+                item_id=r.item_id,
+                qt_number=r.qt_number,
+                qt_round=r.qt_round,
+                qt_type=r.qt_type,
+                target_price=r.target_price,
+                status=r.status,
+                bid_price=r.bid_price,
+                bid_at=r.bid_at,
+                end_time=r.end_time,
+                reject_reason=r.reject_reason,
+                reject_price=r.reject_price,
+                reject_delivery_type=r.reject_delivery_type,
+            )
+            for r in rows
+        ]
+        res.total = len(res.sessions)
         return res
 
     async def list_chats(self, session_id: str) -> Res_SessionChat:
-        # chats 모델 미존재 — 빈 목록 반환(스텁).
         res = Res_SessionChat()
-        res.session_id = uuid.UUID(session_id)
+        sess_uuid = uuid.UUID(session_id)
-        res.messages = []
+        res.session_id = sess_uuid
-        return res
 
-    async def list_cards(self, qt_id: str) -> Res_QuotationCards:
+        err_type, rows = await DB_SESSION_MNG.execute_lambda(
-        # quotation_cards 모델 미존재 — 존재 검증 후 빈 목록 반환(스텁).
+            chats.DBType(),
-        res = Res_QuotationCards()
+            DBWRType.DB_READ.value,
-        err_type, quotation = await self._fetch(uuid.UUID(qt_id))
+            lambda s: self.quotation_crud.list_chats(s, sess_uuid),
+        )
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
             return res
-        res.qt_id = quotation.qt_id
+
-        res.cards = []
+        # chats.seq → ChatMessageData.index 로 매핑. indicator_value(Decimal) → float.
+        res.messages = [
+            ChatMessageData(
+                chat_id=r.chat_id,
+                session_id=r.session_id,
+                card_id=r.card_id,
+                index=r.seq,
+                sender=r.sender,
+                target_price=r.target_price,
+                card_used_yn=r.card_used_yn,
+                indicator_value=float(r.indicator_value) if r.indicator_value is not None else None,
+                card_type=r.card_type,
+            )
+            for r in rows
+        ]
+        return res
+
+    async def list_cards(self, qt_id: str) -> Res_QuotationCards:
+        res = Res_QuotationCards()
+        qt_uuid = uuid.UUID(qt_id)
+        err_type, quotation = await self._fetch(qt_uuid)
+        if err_type != ErrorType.SUCCESS:
+            res.result.SetResult(err_type)
+            return res
+
+        err_type, rows = await DB_SESSION_MNG.execute_lambda(
+            chats.DBType(),
+            DBWRType.DB_READ.value,
+            lambda s: self.quotation_crud.list_used_cards(s, qt_uuid),
+        )
+        if err_type != ErrorType.SUCCESS:
+            res.result.SetResult(err_type)
+            return res
+
+        res.qt_id = quotation.qt_id
+        # rows = [(chat_row, nego_card_id, name, script), ...]. nego/wild 구분은 chats.card_type.
+        cards = []
+        for chat_row, nc_id, nc_name, nc_script in rows:
+            is_wild = chat_row.card_type == 2
+            cards.append(
+                QuotationCardData(
+                    session_card_id=chat_row.chat_id,
+                    qt_id=quotation.qt_id,
+                    nego_card_id=None if is_wild else nc_id,
+                    wild_card_id=nc_id if is_wild else None,
+                    type=chat_row.card_type if chat_row.card_type is not None else 1,
+                    name=nc_name,
+                    script=nc_script,
+                )
+            )
+        res.cards = cards
         return res