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.enums import DBWRType, ErrorType from crud.quotation_crud import IQuotationCRUD, QuotationCRUD from router.v1.quotation.protocol import ( AsyncJob, QuotationData, Res_CreateQuotation, Res_DeleteQuotation, Res_Quotation, Res_QuotationCards, Res_QuotationList, Res_QuotationResult, Res_QuotationSessions, Res_QuotationStatus, Res_SessionChat, ) class QuotationService: """견적 비즈니스 로직. quotations 테이블에는 company_id 가 없어 회사 스코핑은 하지 않는다(토큰 검증만). user_id 는 생성 시 소유자로만 기록한다(조회/변경 시 소유권 필터 없음). """ def __init__(self, quotation_crud: IQuotationCRUD = Depends(QuotationCRUD)): self.quotation_crud = quotation_crud async def _fetch(self, qt_id: uuid.UUID): """견적 단건 조회. (ErrorType, quotation|None) 반환. (회사 스코프 없음)""" err_type, quotation = await DB_SESSION_MNG.execute_lambda( quotations.DBType(), DBWRType.DB_READ.value, lambda s: self.quotation_crud.get_by_id(s, qt_id), ) if err_type != ErrorType.SUCCESS or quotation is None: 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: res = Res_QuotationList(page=page, size=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), ) if err_type != ErrorType.SUCCESS: res.result.SetResult(err_type) return res res.quotations = [QuotationData.model_validate(r) for r in rows] res.total = total return res async def get_quotation(self, qt_id: str) -> Res_Quotation: res = Res_Quotation() err_type, quotation = await self._fetch(uuid.UUID(qt_id)) if err_type != ErrorType.SUCCESS: res.result.SetResult(err_type) return res res.quotation = QuotationData.model_validate(quotation) return res async def create_quotation(self, user_id: str, data: dict) -> Res_CreateQuotation: res = Res_CreateQuotation() quotation = quotations(**data, user_id=uuid.UUID(user_id)) err_type = await DB_SESSION_MNG.execute_lambda_run( [quotations.DBType()], [lambda s: self.quotation_crud.add_quotation(s, quotation)], ) if err_type != ErrorType.SUCCESS: res.result.SetResult(err_type) return res # 서버 기본값(created_at/updated_at) 로드 위해 재조회 (응답 shape 은 Res_CreateQuotation 유지). f_err, fresh = await DB_SESSION_MNG.execute_lambda( quotations.DBType(), DBWRType.DB_READ.value, lambda s: self.quotation_crud.get_by_id(s, quotation.qt_id), ) res.quotation = QuotationData.model_validate(fresh if f_err == ErrorType.SUCCESS and fresh is not None else quotation) # 네고시움 백엔드 비동기 요청은 스텁이므로 row 만 생성한다. res.async_job = AsyncJob(status="submitted", message="견적 생성 작업 요청됨(스텁)") return res async def stop_quotation(self, qt_id: str) -> Res_Quotation: res = Res_Quotation() qt_uuid = uuid.UUID(qt_id) # 존재 확인 err_type, _ = await self._fetch(qt_uuid) if err_type != ErrorType.SUCCESS: res.result.SetResult(err_type) return res # 상태를 '견적마감'으로 변경(실제 DB 업데이트) err_type = await DB_SESSION_MNG.execute_lambda_run( [quotations.DBType()], [lambda s: self.quotation_crud.update_quotation(s, qt_uuid, {"status": "견적마감"})], ) if err_type != ErrorType.SUCCESS: res.result.SetResult(err_type) return res # 갱신 후 재조회 return await self.get_quotation(qt_id) async def delete_quotation(self, qt_id: str) -> Res_DeleteQuotation: res = Res_DeleteQuotation() qt_uuid = uuid.UUID(qt_id) err_type, _ = await self._fetch(qt_uuid) if err_type != ErrorType.SUCCESS: res.result.SetResult(err_type) return res err_type = await DB_SESSION_MNG.execute_lambda_run( [quotations.DBType()], [lambda s: self.quotation_crud.soft_delete(s, qt_uuid)], ) if err_type != ErrorType.SUCCESS: res.result.SetResult(err_type) return res async def get_status(self, qt_id: str) -> Res_QuotationStatus: res = Res_QuotationStatus() err_type, quotation = await self._fetch(uuid.UUID(qt_id)) if err_type != ErrorType.SUCCESS: res.result.SetResult(err_type) return res res.qt_id = quotation.qt_id res.job_status = quotation.status res.message = "ok" return res async def get_result(self, qt_id: str) -> Res_QuotationResult: res = Res_QuotationResult() err_type, quotation = await self._fetch(uuid.UUID(qt_id)) if err_type != ErrorType.SUCCESS: res.result.SetResult(err_type) return res # 낙찰 결과는 quotations 컬럼에서 직접 노출. results 테이블 미존재로 result_count 는 0. res.qt_id = quotation.qt_id res.winner_supplier_id = quotation.preferred_sp_id res.winner_supplier_name = quotation.preferred_sp_name res.is_equal_bid = quotation.equal_bid_yn res.equal_bid_data = quotation.equal_bid_data res.result_count = 0 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)) if err_type != ErrorType.SUCCESS: res.result.SetResult(err_type) return res res.qt_id = quotation.qt_id res.sessions = [] res.total = 0 return res async def list_chats(self, session_id: str) -> Res_SessionChat: # chats 모델 미존재 — 빈 목록 반환(스텁). res = Res_SessionChat() res.session_id = uuid.UUID(session_id) res.messages = [] return res async def list_cards(self, qt_id: str) -> Res_QuotationCards: # quotation_cards 모델 미존재 — 존재 검증 후 빈 목록 반환(스텁). res = Res_QuotationCards() err_type, quotation = await self._fetch(uuid.UUID(qt_id)) if err_type != ErrorType.SUCCESS: res.result.SetResult(err_type) return res res.qt_id = quotation.qt_id res.cards = [] return res