from abc import ABC, abstractmethod from datetime import datetime, timezone from typing import Optional, Tuple from sqlalchemy import asc, desc, select, update from sqlalchemy.ext.asyncio import AsyncSession from common.database.db_session_manager import DB_SESSION_MNG from common.database.model.models import chats, items, sessions from common.enums import ErrorType, SessionStatus from common.logger import LOG # 협상 채팅 CRUD. 메시지 로그(negotiation.chats)와 종료 시 세션 입찰 확정(negotiation.sessions)을 다룬다. # chats / sessions 모두 NEGOTIATION 논리 DB 라 한 트랜잭션(execute_lambda_run)으로 묶을 수 있다. class IChatCRUD(ABC): @abstractmethod async def list_by_session(self, cdb: AsyncSession, session_id) -> Tuple[ErrorType, list]: pass @abstractmethod async def get_last(self, cdb: AsyncSession, session_id) -> Tuple[ErrorType, Tuple[int, Optional[int], Optional[dict]]]: """마지막 메시지의 (seq, sender, meta). 없으면 (0, None, None). 동시전송 가드 + seq 채번 + 입력-모드 검증(직전 봇 meta.input_mode)에 사용.""" pass @abstractmethod async def insert_message(self, cdb: AsyncSession, message: chats) -> ErrorType: pass @abstractmethod async def soft_delete_message(self, cdb: AsyncSession, chat_id) -> ErrorType: pass @abstractmethod async def get_item_by_id(self, cdb: AsyncSession, item_id) -> Tuple[ErrorType, items]: pass @abstractmethod async def finalize_session( self, cdb: AsyncSession, session_id, status: int, bid_price: Optional[int] = None, reject_reason: Optional[str] = None, reject_price: Optional[int] = None, ) -> ErrorType: pass @abstractmethod async def update_last_offer_price(self, cdb: AsyncSession, session_id, price: int) -> ErrorType: pass class ChatCRUD(IChatCRUD): async def list_by_session(self, cdb: AsyncSession, session_id) -> Tuple[ErrorType, list]: try: # (session_id, seq) 유니크 인덱스가 정렬 스캔을 커버한다. query = ( select(chats) .where(chats.session_id == session_id, chats.deleted == False) # noqa: E712 .order_by(asc(chats.seq)) ) err_type, rows = await DB_SESSION_MNG.execute(cdb, query, "list_by_session failed.") if err_type != ErrorType.SUCCESS: return err_type, [] return ErrorType.SUCCESS, rows except Exception as ex: LOG.e_no_callstack(ex) return ErrorType.DB_RUN_FAILED, [] async def get_last(self, cdb: AsyncSession, session_id) -> Tuple[ErrorType, Tuple[int, Optional[int], Optional[dict]]]: try: query = ( select(chats.seq, chats.sender, chats.meta) .where(chats.session_id == session_id, chats.deleted == False) # noqa: E712 .order_by(desc(chats.seq)) .limit(1) ) err_type, rows = await DB_SESSION_MNG.execute(cdb, query, "get_last failed.") if err_type != ErrorType.SUCCESS: return err_type, (0, None, None) if not rows: return ErrorType.SUCCESS, (0, None, None) return ErrorType.SUCCESS, (rows[0][0], rows[0][1], rows[0][2]) except Exception as ex: LOG.e_no_callstack(ex) return ErrorType.DB_RUN_FAILED, (0, None, None) async def insert_message(self, cdb: AsyncSession, message: chats) -> ErrorType: try: return await DB_SESSION_MNG.insert(cdb, message, raise_error=False) except Exception as ex: LOG.e_no_callstack(ex) return ErrorType.DB_RUN_FAILED async def soft_delete_message(self, cdb: AsyncSession, chat_id) -> ErrorType: # agent 실패 시 선점(pre-claim)한 유저 메시지를 되돌린다. 부분 유니크(WHERE deleted=FALSE)라 seq 가 다시 비워진다. try: query = update(chats).where(chats.chat_id == chat_id).values(deleted=True) return await DB_SESSION_MNG.add(cdb, query) except Exception as ex: LOG.e_no_callstack(ex) return ErrorType.DB_RUN_FAILED async def get_item_by_id(self, cdb: AsyncSession, item_id) -> Tuple[ErrorType, items]: try: query = select(items).where(items.item_id == item_id, items.deleted == False).limit(1) # noqa: E712 err_type, row_list = await DB_SESSION_MNG.execute(cdb, query, f"get_item_by_id({item_id}) failed.") 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 finalize_session( self, cdb: AsyncSession, session_id, status: int, bid_price: Optional[int] = None, reject_reason: Optional[str] = None, reject_price: Optional[int] = None, ) -> ErrorType: try: values = {"status": status} if bid_price is not None: values["bid_price"] = bid_price values["bid_at"] = datetime.now(timezone.utc) if reject_reason is not None: values["reject_reason"] = reject_reason[:255] if reject_price is not None: values["reject_price"] = reject_price # 진행중일 때만 전이 — negodata 일괄마감/중복 전송 경합이 종료된 세션을 되살리지 못하게 가드 query = ( update(sessions) .where(sessions.session_id == session_id, sessions.status == SessionStatus.IN_PROGRESS.value) .values(**values) ) return await DB_SESSION_MNG.add(cdb, query) except Exception as ex: LOG.e_no_callstack(ex) return ErrorType.DB_RUN_FAILED async def update_last_offer_price(self, cdb: AsyncSession, session_id, price: int) -> ErrorType: """협력사 마지막 제시가 갱신 — 가격 입력 턴의 봇 메시지 저장과 같은 트랜잭션에서 호출. 진행 중엔 매 가격 입력마다 덮어쓰고 종료 후엔 불변. 앵커링 표본 판정에서 "가격을 한 번이라도 써낸 협상"을 가르는 기준값(NULL=가격 흔적 없음 → 집계 제외). status 가드: 일괄마감 등으로 이미 종료된 세션의 흔적을 사후에 바꾸지 못하게 한다(판정 결정성 보호). """ try: query = ( update(sessions) .where(sessions.session_id == session_id, sessions.status == SessionStatus.IN_PROGRESS.value) .values(last_offer_price=price) ) return await DB_SESSION_MNG.add(cdb, query) except Exception as ex: LOG.e_no_callstack(ex) return ErrorType.DB_RUN_FAILED