From 7f948075e5d956883b5038255d31a8e092e520e4 Mon Sep 17 00:00:00 2001 From: Mina Choi Date: Tue, 23 Jun 2026 16:36:22 +0900 Subject: [PATCH] =?UTF-8?q?[feat]=20negodata/backend:=20=EA=B2=AC=EC=A0=81?= =?UTF-8?q?=20=EC=9E=90=EB=8F=99=EB=A7=88=EA=B0=90=20=EC=8A=A4=EC=BC=80?= =?UTF-8?q?=EC=A4=84=EB=9F=AC=20+=20=EB=8B=A4=EC=9D=8C=20=EB=9D=BC?= =?UTF-8?q?=EC=9A=B4=EB=93=9C=20=EC=9E=AC=EC=83=9D=EC=84=B1(=EC=9E=90?= =?UTF-8?q?=EB=8F=99/=EC=88=98=EB=8F=99)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - APScheduler 크론(KST 5분): 마감일 지난 견적 / 모든 세션 종결 견적 → close_and_decide - 마감 판정: 단독 최저가 낙찰 확정 / 동가·전원미참여 다음 라운드 재생성 / 거부·한도 마감 - 재생성: 같은 견적번호 체인(round+1, 이름 '(N차)'), 공급사 수로 재협상·재견적, 사유별 한도(미참여1+동가1) - 수동 재생성 POST /regenerate/{qt_id} (마지막 차수만 허용) - CloseOutcome enum + 잡 결과별 로그 Co-Authored-By: Claude Opus 4.8 --- docker-compose.yml | 1 + negodata/backend/common/enums.py | 9 + negodata/backend/conftest.py | 29 +- negodata/backend/crud/quotation_crud.py | 90 ++++- .../backend/router/v1/quotation/protocol.py | 15 +- .../backend/router/v1/quotation/quotation.py | 8 + negodata/backend/scheduler/__init__.py | 44 +-- negodata/backend/scheduler/jobs.py | 128 ++----- .../backend/services/quotation_service.py | 317 +++++++++++++++--- negodata/backend/tests/test_scheduler.py | 189 +++++++++++ 10 files changed, 641 insertions(+), 189 deletions(-) create mode 100644 negodata/backend/tests/test_scheduler.py diff --git a/docker-compose.yml b/docker-compose.yml index b5d45c5..927f11f 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -37,6 +37,7 @@ services: DB_HOST: host.docker.internal # 컨테이너→호스트 DB (config.local.toml의 127.0.0.1 override) RELOAD: "1" # uvicorn --reload 활성 → 소스 저장 시 자동 재기동(재빌드 불필요) SCHEDULER_ENABLED: "1" # 마감 크론 활성(단일 워커라 중복 없음). 운영 다중 워커면 1개 프로세스에서만 1 + PYTHONUNBUFFERED: "1" # 컨테이너 로그 실시간 출력(stdout 버퍼링 끔) volumes: - ./negodata/backend:/app # 호스트 소스 = 컨테이너 코드. 이게 있어야 수정이 즉시 반영됨 ports: diff --git a/negodata/backend/common/enums.py b/negodata/backend/common/enums.py index 7c4d97c..9c1b0f7 100644 --- a/negodata/backend/common/enums.py +++ b/negodata/backend/common/enums.py @@ -57,6 +57,7 @@ class ErrorType(Enum): # 견적 관련 에러 QUOTATION_NOT_FOUND = 1500 + QUOTATION_NOT_LATEST_ROUND = auto() # 마지막 차수가 아닌 견적을 재생성하려 함 # 견적 설정 관련 에러 QUOTATION_SETTING_NOT_FOUND = 1600 @@ -142,6 +143,14 @@ class SessionStatus(CodeEnum): REJECTED = 5 +class CloseOutcome(Enum): + """견적 마감 판정 결과(close_and_decide 반환값). 내부 제어·로그용 — DB 저장/프론트 노출 안 함.""" + + AWARDED = "awarded" # 단독 낙찰 확정 + REGENERATED = "regenerated" # 다음 라운드 재생성 + CLOSED = "closed" # 그냥 마감 + + class ChatSender(CodeEnum): """negotiation.chats.sender 코드값. 채팅 발신 주체.""" diff --git a/negodata/backend/conftest.py b/negodata/backend/conftest.py index 8a8b2aa..da559b8 100644 --- a/negodata/backend/conftest.py +++ b/negodata/backend/conftest.py @@ -1,8 +1,9 @@ -# 테스트도 APP_ENV=local 로 실행한다 (config.local.toml 사용). +# 테스트는 APP_ENV=test 로 실행한다 (config.test.toml → negosium_test_db, dev DB 와 분리). +# 이 픽스처들은 TRUNCATE 를 하므로 dev DB(negosium_db)와 절대 공유하면 안 된다(아래 db_engine 안전가드 참고). # config.server_configs 가 import 되는 순간 config..toml 을 읽으므로 가장 먼저 설정. import os -os.environ.setdefault("APP_ENV", "local") +os.environ.setdefault("APP_ENV", "test") import uuid @@ -15,6 +16,11 @@ from common.database.model.models import MAIN_BASE from config.server_configs import main_db_config +# 모델이 쓰는 스키마. test DB 는 비어 있을 수 있어 create_all 전에 직접 만든다. +# 또 TRUNCATE 가 unqualified 테이블명을 쓰므로 이 스키마들을 search_path 에 얹어 해석시킨다. +_SCHEMAS = ("company", "quotation", "card", "negotiation", "partner") + + def _write_url(cfg) -> str: pw = f":{cfg.write_pw}" if cfg.write_pw else "" return f"postgresql+asyncpg://{cfg.write_id}{pw}@{cfg.write_host}:{cfg.write_port}/{cfg.name}" @@ -24,17 +30,28 @@ def _write_url(cfg) -> str: async def db_engine(): """테스트용 스키마를 보장하고, 매 테스트 시작 시 테이블을 비워 격리한다. - 앱(DB_SESSION_MNG)은 자체 엔진으로 같은 DB(config.test.toml)에 접속하므로, - 여기서 만든 스키마를 그대로 공유한다. + ⚠ 이 픽스처는 TRUNCATE 한다 → dev DB(negosium_db)를 가리키면 실데이터가 날아간다. + 그래서 test 전용 DB(이름에 'test')가 아니면 즉시 중단한다(config.test.toml / APP_ENV=test). + 앱(DB_SESSION_MNG)도 APP_ENV=test 면 같은 test DB 에 접속하므로 여기서 만든 스키마를 공유한다. """ - engine = create_async_engine(_write_url(main_db_config)) + # 안전가드: dev DB 오염 방지. negosium_test_db 이외엔 절대 실행하지 않는다. + assert "test" in main_db_config.name, ( + f"테스트가 비-test DB('{main_db_config.name}')를 가리킵니다. " + "APP_ENV=test(config.test.toml)로 실행하세요. dev DB 보호를 위해 중단합니다." + ) + engine = create_async_engine( + _write_url(main_db_config), + connect_args={"server_settings": {"search_path": ",".join(_SCHEMAS) + ",public"}}, + ) async with engine.begin() as conn: + for sch in _SCHEMAS: + await conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {sch}")) await conn.run_sync(MAIN_BASE.metadata.create_all) # 이미 있으면 skip # negodata 도메인 테이블 전부 비워 격리 (CASCADE: FK 미설정이라 안전망) await conn.execute( text( "TRUNCATE TABLE tbl_account, users, companies, items, suppliers, " - "quotation_settings, quotations RESTART IDENTITY CASCADE" + "quotation_settings, quotations, sessions RESTART IDENTITY CASCADE" ) ) yield engine diff --git a/negodata/backend/crud/quotation_crud.py b/negodata/backend/crud/quotation_crud.py index a22f69a..9a1dddb 100644 --- a/negodata/backend/crud/quotation_crud.py +++ b/negodata/backend/crud/quotation_crud.py @@ -10,7 +10,7 @@ from common.database.model.models import ( quotations, sessions, chats, nego_cards, wild_cards, items, suppliers, quotation_settings, version_nego_cards, version_wild_cards, ) -from common.enums import ErrorType, QuotationStatus, QuotationType, SessionStatus +from common.enums import ErrorType, QuotationStatus, SessionStatus from common.logger import LOG from common.utils.gtime import GTime @@ -93,13 +93,25 @@ class IQuotationCRUD(ABC): pass @abstractmethod - async def list_requote_done(self, cdb: AsyncSession) -> Tuple[ErrorType, list]: + async def list_all_sessions_ended(self, cdb: AsyncSession) -> Tuple[ErrorType, list]: + pass + + @abstractmethod + async def list_chain_equal_flags(self, cdb: AsyncSession, number, current_round) -> Tuple[ErrorType, list]: + pass + + @abstractmethod + async def chain_max_round(self, cdb: AsyncSession, number) -> Tuple[ErrorType, int]: pass @abstractmethod async def list_done_sessions(self, cdb: AsyncSession, qt_id) -> Tuple[ErrorType, list]: pass + @abstractmethod + async def list_sessions_status(self, cdb: AsyncSession, qt_id) -> Tuple[ErrorType, list]: + pass + @abstractmethod async def bulk_update_quotation_status(self, cdb: AsyncSession, qt_ids, status: int) -> ErrorType: pass @@ -379,24 +391,32 @@ class QuotationCRUD(IQuotationCRUD): LOG.e_no_callstack(ex) return ErrorType.DB_RUN_FAILED, [] - async def list_requote_done(self, cdb: AsyncSession) -> Tuple[ErrorType, list]: - """[잡②] 재견적(REQUOTE) 중 협상완료(DONE) 세션이 1건 이상이고 아직 안 닫힌 견적 qt_id 목록. - 재견적은 세션이 독립적이라 하나라도 완료되면 나머지를 기다리지 않고 마감 대상.""" + async def list_all_sessions_ended(self, cdb: AsyncSession) -> Tuple[ErrorType, list]: + """[잡②] 아직 안 닫혔는데 모든 세션이 종결된 견적 qt_id 목록(견적 타입 무관 — KTC 와 동일). + 종결 = 협상완료/협상거부/미참여 → 즉 진행중(IN_PROGRESS)·미시작(CREATED) 세션이 하나도 없음. + 세션이 1개 이상 있어야 하며, 마감일과 무관하게 협상이 다 끝났으면 즉시 마감 대상.""" try: - done_exists = ( + # 아직 안 끝난(진행중·미시작) 세션이 하나라도 있으면 제외 + pending_exists = ( select(sessions.session_id) .where( sessions.quotation_id == quotations.qt_id, - sessions.status == SessionStatus.DONE.value, + sessions.status.in_([SessionStatus.CREATED.value, SessionStatus.IN_PROGRESS.value]), sessions.deleted == False, # noqa: E712 ) .exists() ) + # 세션이 최소 1개는 있어야(세션 없는 견적은 대상 아님) + any_session = ( + select(sessions.session_id) + .where(sessions.quotation_id == quotations.qt_id, sessions.deleted == False) # noqa: E712 + .exists() + ) query = select(quotations.qt_id).where( - quotations.type == QuotationType.REQUOTE.value, quotations.status != QuotationStatus.CLOSED.value, quotations.deleted == False, # noqa: E712 - done_exists, + any_session, + ~pending_exists, ) err_type, rows = await DB_SESSION_MNG.execute(cdb, query) if err_type != ErrorType.SUCCESS: @@ -427,6 +447,58 @@ class QuotationCRUD(IQuotationCRUD): LOG.e_no_callstack(ex) return ErrorType.DB_RUN_FAILED, [] + async def list_chain_equal_flags(self, cdb: AsyncSession, number, current_round) -> Tuple[ErrorType, list]: + """[재생성 한도] 같은 견적번호(체인)의 이전 라운드(round < current_round)들의 equal_bid_yn 목록. 삭제 제외. + True=동가로 닫힌 라운드 / 그 외(False·NULL)=미참여로 닫힌 라운드.""" + try: + query = select(quotations.equal_bid_yn).where( + quotations.number == number, + quotations.round < current_round, + quotations.deleted == False, # noqa: E712 + ) + 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 chain_max_round(self, cdb: AsyncSession, number) -> Tuple[ErrorType, int]: + """같은 견적번호(체인)의 최대 round. 다음 라운드 = 이 값 + 1. + uq_quotations_number(number, round) 충돌 방지 — 원본 round+1이 아니라 체인 최신 기준으로 매긴다.""" + try: + query = select(func.max(quotations.round)).where( + quotations.number == number, + quotations.deleted == False, # noqa: E712 + ) + err_type, rows = await DB_SESSION_MNG.execute(cdb, query) + if err_type != ErrorType.SUCCESS: + return err_type, 0 + # 단일 컬럼 select → rows[0] 이 스칼라(max) 값. 행 없거나 전부 NULL이면 None. + mx = rows[0] if rows else None + return ErrorType.SUCCESS, int(mx or 0) + except Exception as ex: + LOG.e_no_callstack(ex) + return ErrorType.DB_RUN_FAILED, 0 + + async def list_sessions_status(self, cdb: AsyncSession, qt_id) -> Tuple[ErrorType, list]: + """[마감 판정] 견적의 모든 세션 → (status, supplier_id, bid_price, supplier_name). 삭제 제외. + 공급사가 지워졌어도 세션 집계엔 포함되도록 outerjoin(이때 name 은 None).""" + try: + query = ( + select(sessions.status, sessions.supplier_id, sessions.bid_price, suppliers.name) + .outerjoin(suppliers, suppliers.supplier_id == sessions.supplier_id) + .where(sessions.quotation_id == qt_id, sessions.deleted == False) # noqa: E712 + ) + 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 bulk_update_quotation_status(self, cdb: AsyncSession, qt_ids, status: int) -> ErrorType: """[잡①] 여러 견적의 status 를 한 번에 전이.""" try: diff --git a/negodata/backend/router/v1/quotation/protocol.py b/negodata/backend/router/v1/quotation/protocol.py index d72172f..f302cf5 100644 --- a/negodata/backend/router/v1/quotation/protocol.py +++ b/negodata/backend/router/v1/quotation/protocol.py @@ -16,7 +16,6 @@ class Req_CreateQuotation(QuotationProtocol): qt_setting_id: uuid.UUID version_id: Optional[uuid.UUID] = None # 미지정 시 기본 전략 버전(card.versions 시드)으로 채움 name: str = "" - number: str = "" type: int = 0 status: int = 0 start_time: Optional[datetime] = None # 미지정 시 생성 시각(UTC) @@ -31,6 +30,10 @@ class Req_CreateQuotation(QuotationProtocol): card_ids: list[uuid.UUID] = [] # 선택 협상카드. 버전을 만들어 묶고 quotation.version_id 로 연결 +class Req_RegenerateQuotation(QuotationProtocol): + supplier_ids: list[uuid.UUID] = [] # 다음 라운드에 부를 공급사(프론트 선택). 상품·기간·번호는 원 견적에서 이어받음 + + class QuotationData(WebPacketProtocol): model_config = ConfigDict(from_attributes=True) @@ -95,15 +98,9 @@ class SessionData(WebPacketProtocol): url: str = "" # 세션 chat 실행 URL(공급사 협상 프론트). DB 미저장 — session_id 로 구성 -class AsyncJob(WebPacketProtocol): - status: str = "" - message: str = "" - - class Res_CreateQuotation(Res_WebPacketProtocol): - quotation: Optional[QuotationData] = None - sessions: list[SessionData] = [] # 견적 생성과 함께 만들어진 협상 세션(각 url 포함) - async_job: Optional[AsyncJob] = None + qt_id: Optional[uuid.UUID] = None # 생성된 견적 id(프론트가 상세 오버레이 오픈에 사용) + session_count: int = 0 # 함께 생성된 협상 세션 수(토스트 표시용). 세션 풀바디는 별도 GET 으로 조회 class Res_QuotationStatus(Res_WebPacketProtocol): diff --git a/negodata/backend/router/v1/quotation/quotation.py b/negodata/backend/router/v1/quotation/quotation.py index 8c18431..2c60663 100644 --- a/negodata/backend/router/v1/quotation/quotation.py +++ b/negodata/backend/router/v1/quotation/quotation.py @@ -8,6 +8,7 @@ from router.v1.validator.dependencies import IsValidAccessToken, RemoveNoneRespo from services.quotation_service import QuotationService from .protocol import ( Req_CreateQuotation, + Req_RegenerateQuotation, Res_CreateQuotation, Res_DeleteQuotation, Res_Quotation, @@ -53,6 +54,13 @@ async def stop_quotation(qt_id: UUID, service: QuotationService = Depends(), use return RemoveNoneResponse(await service.stop_quotation(str(qt_id))) +@router.post(path="/regenerate/{qt_id}", response_model=Res_CreateQuotation, summary="견적 재생성(다음 라운드)") +async def regenerate_quotation( + qt_id: UUID, req: Req_RegenerateQuotation, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken) +): + return RemoveNoneResponse(await service.regenerate_quotation(str(qt_id), req.supplier_ids)) + + # ----- 견적 상세 (FK로 연결된 하위 데이터 / 일부는 모델 미존재로 스텁) ----- @router.get(path="/{qt_id}/status", response_model=Res_QuotationStatus, summary="견적 상태 조회") async def get_quotation_status(qt_id: UUID, service: QuotationService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)): diff --git a/negodata/backend/scheduler/__init__.py b/negodata/backend/scheduler/__init__.py index 9955173..3031f0b 100644 --- a/negodata/backend/scheduler/__init__.py +++ b/negodata/backend/scheduler/__init__.py @@ -1,17 +1,15 @@ """백그라운드 스케줄러(크론) 패키지 — '언제'(when) 담당. -router/(HTTP 진입점)와 동급의 '시간 진입점' 계층. APScheduler 수명주기와 잡 등록(타이밍)만 책임지고, -실제로 하는 일(what)은 scheduler/jobs.py 에 있다. +다중 워커(운영)에서 잡이 워커마다 중복 실행되면 안 되므로 SCHEDULER_ENABLED=1 인 프로세스에서만 등록한다. -- 다중 워커(운영)에서 잡이 워커마다 중복 실행되면 안 되므로 SCHEDULER_ENABLED=1 인 프로세스에서만 등록한다. - (개발은 RELOAD=1 단일 워커라 docker-compose 에서 SCHEDULER_ENABLED=1 로 켠다.) -- apscheduler import 는 start_scheduler() 안에서 한다 → 미설치(이미지 미재빌드) 상태라도 API 는 부팅된다. - -잡 ① close_expired_quotations : 매일 UTC 00:10(KST 09:10) — 마감시각 지난 견적 마감 -잡 ② complete_requote_quotations: 1시간마다 — 재견적 중 DONE 세션 있으면 즉시 마감 +잡 ① close_expired_quotations : 5분마다(KST) — 마감시각 지난 견적 마감 +잡 ② close_negotiated_quotations: 5분마다(KST) — 모든 세션 협상 끝난 견적 즉시 마감(타입 무관) """ import os +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.cron import CronTrigger + from common.logger import LOG from scheduler import jobs @@ -33,36 +31,28 @@ def start_scheduler(): if _scheduler is not None: return - try: - from apscheduler.schedulers.asyncio import AsyncIOScheduler - from apscheduler.triggers.cron import CronTrigger - from apscheduler.triggers.interval import IntervalTrigger - except ImportError: - # 의존성 미설치(이미지 미재빌드) → API 는 살리고 스케줄러만 끈다. - LOG.e_no_callstack("[scheduler] apscheduler 미설치 → 스케줄러 비활성. requirements 재설치(이미지 재빌드) 필요") - return - - _scheduler = AsyncIOScheduler(timezone="UTC") - # 잡 ① 마감시간 처리: 매일 UTC 00:10 + # 한국시간 기준, 두 잡 모두 5분마다 실행. + _scheduler = AsyncIOScheduler(timezone="Asia/Seoul") + # 잡 ① 마감시각 지난 견적 처리 _scheduler.add_job( jobs.close_expired_quotations, - CronTrigger(hour=0, minute=10), + CronTrigger(minute="*/5"), id="close_expired_quotations", - coalesce=True, # 밀린 실행이 여러 번 쌓여도 1번만 - misfire_grace_time=3600, # 정시보다 늦게 깨어나도 1시간 내면 실행 + coalesce=True, # 밀린 실행이 쌓여도 1번만 + misfire_grace_time=600, # 정시보다 늦게 깨어나도 10분 내면 실행 max_instances=1, ) - # 잡 ② 재견적 협상완료 처리: 1시간마다 + # 잡 ② 모든 세션 협상 끝난 견적 즉시 마감(타입 무관) _scheduler.add_job( - jobs.complete_requote_quotations, - IntervalTrigger(hours=1), - id="complete_requote_quotations", + jobs.close_negotiated_quotations, + CronTrigger(minute="*/5"), + id="close_negotiated_quotations", coalesce=True, misfire_grace_time=600, max_instances=1, ) _scheduler.start() - LOG.i("[scheduler] started (close_expired=daily 00:10 UTC, complete_requote=hourly)") + LOG.i("[scheduler] started (KST, both every 5min)") def shutdown_scheduler(): diff --git a/negodata/backend/scheduler/jobs.py b/negodata/backend/scheduler/jobs.py index c320a11..bf104c7 100644 --- a/negodata/backend/scheduler/jobs.py +++ b/negodata/backend/scheduler/jobs.py @@ -1,17 +1,27 @@ +"""스케줄 잡 로직(what). '언제 도느냐'(scheduler/__init__.py)와 분리된, 잡이 실제로 하는 일. + +두 잡 모두 '대상 견적을 골라' → 견적마다 QuotationService.close_and_decide 를 호출한다. +마감 + 결과 판정(낙찰 확정 / 다음 라운드 재생성 / 그냥 마감)은 전부 도메인(close_and_decide)이 책임지고, +여기 잡은 '어떤 견적을 고르냐(대상 선정)'와 '언제 도느냐'만 담당한다. +""" +from collections import Counter + from common.database.db_session_manager import DB_SESSION_MNG -from common.database.model.models import quotations, sessions -from common.enums import DBWRType, ErrorType, QuotationStatus, SessionStatus +from common.database.model.models import quotations +from common.enums import CloseOutcome, DBWRType, ErrorType from common.logger import LOG from common.utils.gtime import GTime from crud.quotation_crud import QuotationCRUD +from services.quotation_service import QuotationService async def close_expired_quotations() -> int: - """[잡①] 마감일이 지난 견적을 자동으로 견적마감 처리한다. 하루 한 번 실행. + """[잡①] 마감일이 지난 견적을 자동 마감 처리한다. 하루 한 번 실행. 대상: 마감 시각이 이미 지났는데 아직 마감되지 않은(삭제되지도 않은) 견적. - 처리: 그 견적들을 견적마감 상태로 바꾸고, 아직 시작 전인 세션은 미참여로 정리한다. - 반환: 마감 처리한 견적 수.""" + 처리: 견적마다 close_and_decide 로 결과 판정(낙찰 확정 / 다음 라운드 재생성 / 그냥 마감). + 반환: 처리한 견적 수.""" crud = QuotationCRUD() + service = QuotationService(crud) now = GTime.UTC() err_type, qt_ids = await DB_SESSION_MNG.execute_lambda( quotations.DBType(), @@ -21,104 +31,34 @@ async def close_expired_quotations() -> int: if err_type != ErrorType.SUCCESS: LOG.e_no_callstack(f"[scheduler] close_expired 대상 조회 실패: {err_type.name}") return 0 - if not qt_ids: - return 0 - err_type = await DB_SESSION_MNG.execute_lambda_run( - [quotations.DBType()], - [ - lambda s: crud.bulk_update_quotation_status(s, qt_ids, QuotationStatus.CLOSED.value), - lambda s: crud.bulk_update_sessions_status( - s, qt_ids, [SessionStatus.CREATED.value, SessionStatus.IN_PROGRESS.value], SessionStatus.NOT_PARTICIPATED.value - ), - ], - ) - if err_type != ErrorType.SUCCESS: - LOG.e_no_callstack(f"[scheduler] close_expired 마감 실패: {err_type.name}") - return 0 - LOG.i(f"[scheduler] close_expired: {len(qt_ids)}건 견적마감") - return len(qt_ids) + results = Counter() + for qt_id in qt_ids: + results[await service.close_and_decide(qt_id)] += 1 + if results: + LOG.i(f"[scheduler] close_expired: 낙찰 {results[CloseOutcome.AWARDED]} / 재생성 {results[CloseOutcome.REGENERATED]} / 마감 {results[CloseOutcome.CLOSED]}") + return sum(results.values()) -async def complete_requote_quotations() -> int: - """[잡②] 재견적은 협상완료된 세션이 생기면 나머지를 기다리지 않고 바로 마감한다. 한 시간마다 실행. - 대상: 아직 마감되지 않은 재견적 견적 중, 협상완료된 세션이 있는 것. - 낙찰: 협상완료된 세션 중 입찰가가 가장 낮은 공급사를 낙찰자로 정한다. 같은 최저가가 둘 이상이면(동가) 낙찰자를 비우고 동가 정보만 남긴다. - (현재 재견적은 견적당 세션이 하나라 실제로는 단독 낙찰만 일어나지만, 모델상 1:N이라 일반 규칙을 그대로 둔다.) - 처리: 낙찰 정보를 기록하고 견적을 견적마감 상태로 바꾸며, 아직 시작 전인 세션은 미참여로 정리한다. - 반환: 마감 처리한 견적 수.""" +async def close_negotiated_quotations() -> int: + """[잡②] 모든 세션의 협상이 끝난 견적은 마감일을 기다리지 않고 바로 마감한다(견적 타입 무관). + 대상: 아직 마감되지 않았고, 진행중·미시작 세션이 하나도 없는(= 모두 종결된) 견적. 한 시간마다 실행. + 처리: 견적마다 close_and_decide 로 결과 판정(낙찰 확정 / 다음 라운드 재생성 / 그냥 마감). + 반환: 처리한 견적 수.""" crud = QuotationCRUD() + service = QuotationService(crud) err_type, qt_ids = await DB_SESSION_MNG.execute_lambda( quotations.DBType(), DBWRType.DB_READ.value, - lambda s: crud.list_requote_done(s), + lambda s: crud.list_all_sessions_ended(s), ) if err_type != ErrorType.SUCCESS: - LOG.e_no_callstack(f"[scheduler] complete_requote 대상 조회 실패: {err_type.name}") - return 0 - if not qt_ids: + LOG.e_no_callstack(f"[scheduler] close_negotiated 대상 조회 실패: {err_type.name}") return 0 - closed = 0 + results = Counter() for qt_id in qt_ids: - e2, done_rows = await DB_SESSION_MNG.execute_lambda( - sessions.DBType(), - DBWRType.DB_READ.value, - lambda s, q=qt_id: crud.list_done_sessions(s, q), - ) - if e2 != ErrorType.SUCCESS: - LOG.e_no_callstack(f"[scheduler] complete_requote DONE세션 조회 실패 qt_id={qt_id}: {e2.name}") - continue - - # 현재 재견적은 세션이 하나라 사실상 단독 낙찰만 타지만, 모델상 1:N이라 일반 규칙(_pick_winner)을 그대로 쓴다. - winner, equal = _pick_winner(done_rows) - # 단독 낙찰과 동가는 상호배타(KTC 정본). 플래그를 명시적으로 박는다. - data = { - "status": QuotationStatus.CLOSED.value, - "preferred_sp_yn": winner is not None, - "equal_bid_yn": equal is not None, - } - if winner is not None: - data["preferred_sp_id"] = winner["supplier_id"] - data["preferred_sp_name"] = (winner["name"] or "")[:20] - if equal is not None: - data["equal_bid_data"] = equal - - e3 = await DB_SESSION_MNG.execute_lambda_run( - [quotations.DBType()], - [ - lambda s, d=data, q=qt_id: crud.update_quotation(s, q, d), - lambda s, q=qt_id: crud.update_sessions_status( - s, q, [SessionStatus.CREATED.value, SessionStatus.IN_PROGRESS.value], SessionStatus.NOT_PARTICIPATED.value - ), - ], - ) - if e3 == ErrorType.SUCCESS: - closed += 1 - else: - LOG.e_no_callstack(f"[scheduler] complete_requote 마감 실패 qt_id={qt_id}: {e3.name}") - - if closed: - LOG.i(f"[scheduler] complete_requote: {closed}건 견적마감") - return closed - - -def _pick_winner(done_rows): - """협상완료된 세션들 중에서 낙찰자를 정한다(KTC 정본 규칙). complete_requote_quotations 전용 헬퍼. - 입찰가가 매겨진 세션들 가운데 가장 낮은 가격을 부른 공급사를 낙찰자로 본다. - - 최저가를 부른 곳이 한 곳뿐이면: 그 공급사를 낙찰자로 정하고, 동가는 없다. - - 최저가가 둘 이상으로 같으면(동가): 낙찰자는 비우고 동가 정보(최저가와 그 공급사들)만 남긴다. - - 입찰가가 매겨진 세션이 하나도 없으면: 낙찰자도 동가 정보도 없다. - 낙찰자와 동가 정보를 한 쌍으로 돌려주며, 둘은 동시에 채워지지 않는다(단독 낙찰 또는 동가, 둘 중 하나).""" - cands = [(sid, int(bp), name) for sid, bp, name in done_rows if bp is not None] - if not cands: - return None, None - min_price = min(c[1] for c in cands) - tied = [c for c in cands if c[1] == min_price] - if len(tied) > 1: # 동가입찰: 최저가가 여럿 → 낙찰 미지정, 동가만 기록 - equal = { - "price": min_price, - "suppliers": [{"supplier_id": str(sid), "name": name} for sid, _, name in tied], - } - return None, equal - return {"supplier_id": tied[0][0], "name": tied[0][2]}, None + results[await service.close_and_decide(qt_id)] += 1 + if results: + LOG.i(f"[scheduler] close_negotiated: 낙찰 {results[CloseOutcome.AWARDED]} / 재생성 {results[CloseOutcome.REGENERATED]} / 마감 {results[CloseOutcome.CLOSED]}") + return sum(results.values()) diff --git a/negodata/backend/services/quotation_service.py b/negodata/backend/services/quotation_service.py index 146a834..4cf7690 100644 --- a/negodata/backend/services/quotation_service.py +++ b/negodata/backend/services/quotation_service.py @@ -1,11 +1,13 @@ +import re import uuid from datetime import timezone +from typing import Optional from fastapi import Depends from common.database.db_session_manager import DB_SESSION_MNG from common.database.model.models import quotations, sessions, chats, versions, version_nego_cards, version_wild_cards -from common.enums import DBWRType, ErrorType, QuotationStatus, SessionStatus +from common.enums import CloseOutcome, DBWRType, ErrorType, QuotationStatus, QuotationType, SessionStatus from common.models.gmodel import PageParams from common.utils.gtime import GTime from config.server_configs import web_server_config @@ -38,6 +40,9 @@ class QuotationService: # 기본 전략 버전(card.versions 시드). 견적 생성 시 version_id 미지정이면 이 값으로 채운다. DEFAULT_VERSION_ID = uuid.UUID("00000000-0000-0000-0000-000000000030") + # 재생성 한도: 한 체인(같은 견적번호)에서 사유(미참여/동가)별 최대 1번까지 재생성(순서 무관, 같은 사유 2번 불가). + MAX_REGEN_PER_CAUSE = 1 + def __init__(self, quotation_crud: IQuotationCRUD = Depends(QuotationCRUD)): self.quotation_crud = quotation_crud @@ -134,21 +139,92 @@ class QuotationService: return res async def create_quotation(self, user_id: str, req: Req_CreateQuotation) -> Res_CreateQuotation: + """[프론트] 신규 견적 생성. 요청값을 보정한 뒤 공통 빌더(_build_quotation)에 위임한다.""" + return await self._build_quotation( + user_id=user_id, + qt_setting_id=req.qt_setting_id, + version_id=req.version_id or self.DEFAULT_VERSION_ID, + name=req.name, + number=self._gen_number(), # 견적번호는 항상 서버 생성(프론트 입력란 없음) + type_=req.type, + status=req.status or QuotationStatus.CREATED.value, + round_=req.round or 1, + start_time=self._naive_utc(req.start_time or GTime.UTC()), + end_time=self._naive_utc(req.end_time), + manager_name=req.manager_name, + manager_email=req.manager_email, + manager_contact_number=req.manager_contact_number, + memo=req.memo, + item_ids=req.item_ids, + supplier_ids=req.supplier_ids, + card_ids=req.card_ids, + ) + + async def regenerate_next_round(self, original_qt_id: uuid.UUID, supplier_ids: list) -> Res_CreateQuotation: + """[마감 후속] 결판 안 난 견적의 '다음 라운드'를 새로 만든다. + + 플로우: + 1) 원 견적 + 세션을 조회해 대상 상품(item)을 복원 + 2) 타입 결정 — 다음 라운드 공급사가 1곳이면 재협상(RENEGO), 여러 곳이면 재견적(REQUOTE) + 3) 같은 견적번호 + round+1 로 다음 라운드 생성 (협상기간은 원 견적과 같은 길이) + + 견적번호(number)를 원본 그대로 이어받아 '같은 번호 = 한 체인'으로 묶는다(parent_id 대체). + supplier_ids: 다음 라운드에 부를 공급사(동가면 동가 업체만, 그 외엔 원 견적 공급사 전체). + """ res = Res_CreateQuotation() - # 세션 생성용 입력은 quotations 컬럼이 아니므로 분리한다(상품 × 공급사 조합마다 세션 1개). - item_ids = req.item_ids - supplier_ids = req.supplier_ids - card_ids = req.card_ids + # 1) 원 견적 + 세션 조회 → 대상 상품 복원 + err_type, original = await self._fetch(original_qt_id) + if err_type != ErrorType.SUCCESS or original is None: + 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, original_qt_id), + ) + item_ids = list({r.item_id for r in rows}) if err_type == ErrorType.SUCCESS else [] - # NOT NULL 컬럼 보정(프론트 미전송 시 서버 디폴트). - version_id = req.version_id or self.DEFAULT_VERSION_ID - number = req.number or self._gen_number() - status = req.status or QuotationStatus.CREATED.value - round_ = req.round or 1 # ORM 기본값은 flush 시점이라, 세션 스냅샷용으로 미리 확정한다 - # DB 컬럼이 naive 라, tz-aware 로 들어온 시각(프론트 toISOString)을 UTC naive 로 맞춘다. - start_time = self._naive_utc(req.start_time or GTime.UTC()) - end_time = self._naive_utc(req.end_time) + # 2) 타입 결정: 공급사 1곳 → 재협상 / 여러 곳 → 재견적 + next_type = QuotationType.RENEGO.value if len(supplier_ids) <= 1 else QuotationType.REQUOTE.value + + # 3) 다음 라운드의 견적 생성 + now = GTime.UTC() + duration = original.end_time - original.start_time + # 진입 경로(크론 마감 / 수동 regenerate_quotation) 모두 '마지막 차수'만 넘기므로 +1 이 곧 체인 다음 차수. + next_round = original.round + 1 + # 이름에 '(N차)' 표기. 원래 이름 기준(기존 '(M차)' 표기는 떼고 새로) + name 컬럼 50자 제한 보호. + suffix = f" ({next_round}차)" + base_name = re.sub(r"\s*\(\d+차\)\s*$", "", original.name or "")[: 50 - len(suffix)] + return await self._build_quotation( + user_id=str(original.user_id), + qt_setting_id=original.qt_setting_id, + version_id=original.version_id, # 카드 버전은 원본 그대로 이어씀 + name=f"{base_name}{suffix}", # 예: "삼성 견적 (2차)" + number=original.number, # ← 원본 번호 따라감(새 번호 생성 X) + type_=next_type, + status=QuotationStatus.CREATED.value, + round_=next_round, + start_time=now, + end_time=now + duration, + manager_name=original.manager_name, + manager_email=original.manager_email, + manager_contact_number=original.manager_contact_number, + memo=original.memo, + item_ids=item_ids, + supplier_ids=list(supplier_ids), + card_ids=[], # 새 버전 안 만듦(원본 version_id 재사용) + ) + + async def _build_quotation( + self, *, + user_id: str, qt_setting_id, version_id, name: str, number: str, + type_: int, status: int, round_: int, start_time, end_time, + manager_name, manager_email, manager_contact_number, memo, + item_ids: list, supplier_ids: list, card_ids: list, + ) -> Res_CreateQuotation: + """견적 1건 + (상품×공급사) 세션들을 한 트랜잭션으로 생성하는 공통 빌더.""" + res = Res_CreateQuotation() # 세션 목표가 입력(상품 단가 + 견적 세팅 목표 마진율). 읽기 트랜잭션에서 먼저 조회. prices = {} @@ -162,12 +238,12 @@ class QuotationService: _err, margin = await DB_SESSION_MNG.execute_lambda( quotations.DBType(), DBWRType.DB_READ.value, - lambda s: self.quotation_crud.get_target_margin(s, req.qt_setting_id), + lambda s: self.quotation_crud.get_target_margin(s, qt_setting_id), ) margin = margin if _err == ErrorType.SUCCESS else None # 선택 협상카드가 있으면 새 버전을 만들어 카드들을 묶고, quotation.version_id 로 연결한다. - # (quotation↔card 는 chats 가 아니라 version → version_nego_cards/version_wild_cards 로 연결.) + # (quotation↔card 는 version → version_nego_cards/version_wild_cards 로 연결.) version_obj = None link_rows = [] if card_ids: @@ -197,21 +273,22 @@ class QuotationService: quotation = quotations( qt_id=qt_id, user_id=uuid.UUID(user_id), - qt_setting_id=req.qt_setting_id, + qt_setting_id=qt_setting_id, version_id=version_id, - name=req.name, + name=name, number=number, - type=req.type, + type=type_, status=status, round=round_, start_time=start_time, end_time=end_time, - manager_name=req.manager_name, - manager_email=req.manager_email, - manager_contact_number=req.manager_contact_number, - memo=req.memo, + manager_name=manager_name, + manager_email=manager_email, + manager_contact_number=manager_contact_number, + memo=memo, ) + # 상품 × 공급사 조합마다 세션 1개. session_objs = [] for iid in item_ids: tp = self._calc_target_price(prices.get(iid), margin) @@ -238,21 +315,188 @@ class QuotationService: ops.append(lambda s: self.quotation_crud.add_rows(s, link_rows)) ops.append(lambda s: self.quotation_crud.add_quotation(s, quotation)) ops.append(lambda s: self.quotation_crud.add_sessions(s, session_objs)) - err_type = await DB_SESSION_MNG.execute_lambda_run( - [quotations.DBType()], - ops, - ) + err_type = await DB_SESSION_MNG.execute_lambda_run([quotations.DBType()], ops) if err_type != ErrorType.SUCCESS: res.result.SetResult(err_type) return res - # 프론트는 생성 응답 본문을 화면에 쓰지 않고 qt_id 로 목록/상세를 재조회한다. - # 그래서 새 견적 id 와 세션 수만 돌려준다(재조회 get_by_id·세션 풀바디·url 생략). + # 프론트는 생성 응답 본문을 화면에 안 쓰고 qt_id 로 재조회 → 새 id 와 세션 수만 반환. res.qt_id = qt_id res.session_count = len(session_objs) return res + # ----- 마감 판정 + @staticmethod + def _pick_winner(done_rows) -> tuple[Optional[dict], Optional[dict]]: + """협상완료 세션들 중 낙찰자 판정. done_rows: [(supplier_id, bid_price, supplier_name), ...]. + 입찰가가 매겨진 세션 중 최저가가 단독이면 그 공급사를 낙찰로, 동가(둘+)면 낙찰은 비우고 동가 정보만 남긴다. + 단독/동가는 상호배타. 반환: (winner|None, equal|None).""" + cands = [(sid, int(bp), name) for sid, bp, name in done_rows if bp is not None] + if not cands: + return None, None + min_price = min(c[1] for c in cands) + tied = [c for c in cands if c[1] == min_price] + if len(tied) > 1: + equal = {"price": min_price, "suppliers": [{"supplier_id": str(sid), "name": name} for sid, _, name in tied]} + return None, equal + return {"supplier_id": tied[0][0], "name": tied[0][2]}, None + + async def _award_and_close(self, qt_uuid, winner) -> None: + """단독 낙찰 확정 + 마감 + 미완료(미시작·진행중) 세션 미참여.""" + data = { + "status": QuotationStatus.CLOSED.value, + "preferred_sp_yn": True, + "preferred_sp_id": winner["supplier_id"], + "preferred_sp_name": (winner["name"] or "")[:20], + "equal_bid_yn": False, + } + await DB_SESSION_MNG.execute_lambda_run( + [quotations.DBType()], + [ + lambda s: self.quotation_crud.update_quotation(s, qt_uuid, data), + lambda s: self.quotation_crud.update_sessions_status( + s, qt_uuid, [SessionStatus.CREATED.value, SessionStatus.IN_PROGRESS.value], SessionStatus.NOT_PARTICIPATED.value + ), + ], + ) + + async def _just_close(self, qt_uuid) -> None: + """그냥 마감 + 미완료 세션 미참여.""" + await DB_SESSION_MNG.execute_lambda_run( + [quotations.DBType()], + [ + lambda s: self.quotation_crud.update_quotation(s, qt_uuid, {"status": QuotationStatus.CLOSED.value}), + lambda s: self.quotation_crud.update_sessions_status( + s, qt_uuid, [SessionStatus.CREATED.value, SessionStatus.IN_PROGRESS.value], SessionStatus.NOT_PARTICIPATED.value + ), + ], + ) + + async def _close_as_equal(self, qt_uuid, equal) -> None: + """동가로 마감 + 미완료 세션 미참여. equal_bid_yn/data 를 기록해 둔다 + (재생성 한도 계산이 이 플래그로 동가 라운드를 식별하고, 프론트도 동가 정보를 그대로 쓴다).""" + data = { + "status": QuotationStatus.CLOSED.value, + "preferred_sp_yn": False, + "equal_bid_yn": True, + "equal_bid_data": equal, + } + await DB_SESSION_MNG.execute_lambda_run( + [quotations.DBType()], + [ + lambda s: self.quotation_crud.update_quotation(s, qt_uuid, data), + lambda s: self.quotation_crud.update_sessions_status( + s, qt_uuid, [SessionStatus.CREATED.value, SessionStatus.IN_PROGRESS.value], SessionStatus.NOT_PARTICIPATED.value + ), + ], + ) + + async def close_and_decide(self, qt_id) -> CloseOutcome: + """[마감] 견적을 마감하면서 결과를 판정한다. + 1) 협상완료 중 최저가 단독 → 그 공급사 낙찰 확정 + 2) 협상완료 중 최저가 동가 → 다음 라운드 재생성(동가 업체끼리) [체인에 동가 재생성 이력 없을 때만] + 3) 협상거부 세션이 하나라도 있음 → 그냥 마감 (재생성 안 함) + 4) 전원 미참여(완료·거부 0) → 다음 라운드 재생성(원 견적 공급사 전체) [체인에 미참여 재생성 이력 없을 때만] + 5) 그 외 / 한도 도달 → 그냥 마감 + 동가를 거부보다 먼저 본다: 동률은 '협상완료'한 업체들 간 경쟁이라 무관한 다른 업체의 거부로 막지 않는다. + 재생성 한도: 한 체인(같은 견적번호)에서 '미참여' 1번 + '동가' 1번(순서 무관, 같은 사유 2번은 불가). + 공통: status→CLOSED, 미시작·진행중 세션→미참여.""" + qt_uuid = qt_id if isinstance(qt_id, uuid.UUID) else uuid.UUID(str(qt_id)) + + err_type, original = await self._fetch(qt_uuid) + if err_type != ErrorType.SUCCESS or original is None: + return CloseOutcome.CLOSED + + err_type, rows = await DB_SESSION_MNG.execute_lambda( + sessions.DBType(), + DBWRType.DB_READ.value, + lambda s: self.quotation_crud.list_sessions_status(s, qt_uuid), + ) + rows = rows if err_type == ErrorType.SUCCESS else [] + + done = [(r.supplier_id, r.bid_price, r.name) for r in rows if r.status == SessionStatus.DONE.value] + has_rejected = any(r.status == SessionStatus.REJECTED.value for r in rows) + winner, equal = self._pick_winner(done) + + # 1) 단독 낙찰 → 확정 + if winner is not None: + await self._award_and_close(qt_uuid, winner) + return CloseOutcome.AWARDED + + # 동가/미참여 재생성은 사유별 한도(각 1번, 순서 무관) 확인 후 + no_part_used, equal_used = await self._chain_regen_counts(original.number, original.round) + + # 2) 동가 → 동가 업체끼리 다음 라운드 (거부보다 먼저: tie 해소 우선, 체인에 동가 이력 없을 때만) + if equal is not None and equal_used < self.MAX_REGEN_PER_CAUSE: + tied_ids = [uuid.UUID(sp["supplier_id"]) for sp in equal["suppliers"]] + await self._close_as_equal(qt_uuid, equal) # 동가 기록(equal_bid_yn) 후 마감 + await self.regenerate_next_round(qt_uuid, tied_ids) + return CloseOutcome.REGENERATED + # 3) 협상거부 있음 → 마감만 (재생성 안 함) + if has_rejected: + await self._just_close(qt_uuid) + return CloseOutcome.CLOSED + # 4) 전원 미참여 → 공급사 전체로 다음 라운드 (체인에 미참여 재생성 이력 없을 때만) + if not done and rows and no_part_used < self.MAX_REGEN_PER_CAUSE: + supplier_ids = list({r.supplier_id for r in rows}) + await self._just_close(qt_uuid) + await self.regenerate_next_round(qt_uuid, supplier_ids) + return CloseOutcome.REGENERATED + # 5) 그 외 / 한도 도달 → 마감만 + await self._just_close(qt_uuid) + return CloseOutcome.CLOSED + + async def _chain_regen_counts(self, number: str, current_round: int) -> tuple[int, int]: + """체인(같은 견적번호) 이전 라운드들의 재생성 사유 횟수. 반환: (미참여 횟수, 동가 횟수). + 동가로 닫힌 라운드는 equal_bid_yn=True 로 기록되므로 그 플래그로 센다. + (이전 라운드는 동가 아니면 미참여 둘뿐 — 단독낙찰·거부는 체인을 끝냄 → equal_bid_yn 이 True 아니면 미참여).""" + err_type, flags = await DB_SESSION_MNG.execute_lambda( + quotations.DBType(), + DBWRType.DB_READ.value, + lambda s: self.quotation_crud.list_chain_equal_flags(s, number, current_round), + ) + if err_type != ErrorType.SUCCESS: + return 0, 0 + equal = sum(1 for f in flags if f is True) + no_part = len(flags) - equal + return no_part, equal + + async def regenerate_quotation(self, qt_id: str, supplier_ids: list) -> Res_CreateQuotation: + """[프론트] 마감된 견적을 골라 수동으로 다음 라운드를 생성한다. + 크론/수동마감의 자동 재생성과 달리 사유·체인 한도 판정 없이, 프론트가 고른 공급사로 바로 만든다. + 상품·기간·견적번호·카드버전은 원 견적에서 이어받는다(regenerate_next_round).""" + res = Res_CreateQuotation() + qt_uuid = uuid.UUID(qt_id) + + err_type, original = await self._fetch(qt_uuid) + if err_type != ErrorType.SUCCESS or original is None: + res.result.SetResult(err_type) + return res + # 마감된 견적만 재생성(진행 중인 라운드를 또 찍어 같은 번호가 동시에 살아있는 걸 막는다). + if original.status != QuotationStatus.CLOSED.value: + res.result.SetResult(ErrorType.INVALID_REQUEST_DATA) + return res + # 공급사 미선택이면 세션이 0건이라 의미 없음. + if not supplier_ids: + res.result.SetResult(ErrorType.INVALID_REQUEST_DATA) + return res + # 마지막 차수에서만 재생성 — 옛 라운드/뒤 라운드 살아있는데 또 생성하는 걸 막고(uq(number,round) 충돌도 예방), + # 마지막이 아니면 명시적 에러를 던진다. + _e, max_round = await DB_SESSION_MNG.execute_lambda( + quotations.DBType(), + DBWRType.DB_READ.value, + lambda s: self.quotation_crud.chain_max_round(s, original.number), + ) + if _e == ErrorType.SUCCESS and max_round and original.round < max_round: + res.result.SetResult(ErrorType.QUOTATION_NOT_LATEST_ROUND) + res.msg = "마지막 차수의 견적에서만 다음 라운드를 생성할 수 있습니다." + return res + + return await self.regenerate_next_round(qt_uuid, supplier_ids) + async def stop_quotation(self, qt_id: str) -> Res_Quotation: + """[프론트] 수동 견적마감. 크론과 똑같은 마감 판정(close_and_decide)을 탄다 + (단독낙찰 확정 / 동가·미참여면 다음 라운드 재생성 / 거부·한도면 그냥 마감).""" res = Res_Quotation() qt_uuid = uuid.UUID(qt_id) @@ -262,22 +506,7 @@ class QuotationService: res.result.SetResult(err_type) return res - # 견적 '견적마감'(CLOSED) + 딸린 세션 정리를 한 트랜잭션으로. - # 세션은 아직 시작 전(협상생성)인 것만 미참여로 떨군다. 협상중/완료/거부/미참여는 그대로 둔다. - err_type = await DB_SESSION_MNG.execute_lambda_run( - [quotations.DBType()], - [ - lambda s: self.quotation_crud.update_quotation(s, qt_uuid, {"status": QuotationStatus.CLOSED.value}), - lambda s: self.quotation_crud.update_sessions_status( - s, qt_uuid, [SessionStatus.CREATED.value, SessionStatus.IN_PROGRESS.value], SessionStatus.NOT_PARTICIPATED.value - ), - ], - ) - if err_type != ErrorType.SUCCESS: - res.result.SetResult(err_type) - return res - - # 갱신 후 재조회 + await self.close_and_decide(qt_uuid) return await self.get_quotation(qt_id) async def delete_quotation(self, qt_id: str) -> Res_DeleteQuotation: diff --git a/negodata/backend/tests/test_scheduler.py b/negodata/backend/tests/test_scheduler.py new file mode 100644 index 0000000..73abe7d --- /dev/null +++ b/negodata/backend/tests/test_scheduler.py @@ -0,0 +1,189 @@ +"""scheduler 잡 e2e — '대상 선정'(어떤 견적을 고르나) + close_and_decide 위임 결과 검증. + +실행 전제: PostgreSQL(negodata_db). docker compose up -d 후 python -m pytest tests/test_scheduler.py. +잡은 HTTP 엔드포인트가 없어 scheduler.jobs 함수를 직접 호출한다(앱과 같은 DB_SESSION_MNG 사용 → mock 불필요). +세션 상태(DONE/REJECTED/bid_price 등)는 협상 프론트가 만드는 값이라 API 로 못 만든다 → SQL 로 직접 시드. +""" +import asyncio +import uuid +from datetime import datetime + +import pytest_asyncio +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.interval import IntervalTrigger +from sqlalchemy import text + +from common.enums import QuotationStatus, QuotationType, SessionStatus +from scheduler import jobs + +PAST = datetime(2020, 1, 1) # 마감시각 지남(잡① 대상) +FUTURE = datetime(2999, 1, 1) # 마감시각 미래(잡① 제외) + + +@pytest_asyncio.fixture +async def clean(db_engine): + """conftest 의 db_engine 은 quotations 만 비우고 sessions 는 안 비운다(FK 미설정 → CASCADE 대상 아님). + 잡②(close_negotiated)는 전체 견적을 스캔하므로 다른 테스트가 남긴 세션이 결과를 흔든다 → sessions 도 비워 격리.""" + async with db_engine.begin() as conn: + await conn.execute(text("TRUNCATE TABLE sessions, quotations RESTART IDENTITY CASCADE")) + return db_engine + + +# ----- 시드 헬퍼 (FK 미설정이라 user/item/supplier 없이 임의 uuid 로 충분) ----- +async def _add_quotation(engine, *, status=QuotationStatus.ACTIVE.value, end_time=PAST, deleted=False): + qt_id = uuid.uuid4() + async with engine.begin() as conn: + await conn.execute( + text( + "INSERT INTO quotations " + "(qt_id, user_id, qt_setting_id, version_id, name, number, type, status, " + " round, iteration, start_time, end_time, deleted) VALUES " + "(:qt_id, :user_id, :qt_setting_id, :version_id, :name, :number, :type, :status, " + " :round, :iteration, :start_time, :end_time, :deleted)" + ), + { + "qt_id": qt_id, "user_id": uuid.uuid4(), "qt_setting_id": uuid.uuid4(), + "version_id": uuid.uuid4(), "name": "견적", "number": f"Q-{qt_id.hex[:8]}", + "type": QuotationType.REQUOTE.value, "status": status, "round": 1, "iteration": 0, + "start_time": PAST, "end_time": end_time, "deleted": deleted, + }, + ) + return qt_id + + +async def _add_session(engine, qt_id, *, status, bid_price=None, supplier_id=None): + async with engine.begin() as conn: + await conn.execute( + text( + "INSERT INTO sessions " + "(session_id, quotation_id, item_id, supplier_id, qt_number, qt_round, qt_type, " + " target_price, status, bid_price, end_time) VALUES " + "(:session_id, :quotation_id, :item_id, :supplier_id, :qt_number, :qt_round, :qt_type, " + " :target_price, :status, :bid_price, :end_time)" + ), + { + "session_id": uuid.uuid4(), "quotation_id": qt_id, "item_id": uuid.uuid4(), + "supplier_id": supplier_id or uuid.uuid4(), "qt_number": "Q", "qt_round": 1, + "qt_type": QuotationType.REQUOTE.value, "target_price": 0, + "status": status, "bid_price": bid_price, "end_time": PAST, + }, + ) + + +async def _quotation_row(engine, qt_id): + async with engine.begin() as conn: + return (await conn.execute( + text("SELECT status, preferred_sp_yn, preferred_sp_id FROM quotations WHERE qt_id = :id"), + {"id": qt_id}, + )).first() + + +# ----- 잡① close_expired_quotations : 대상 선정(마감시각 지난 미마감만) ----- +async def test_close_expired_picks_only_due_and_open(clean): + engine = clean + due = await _add_quotation(engine, status=QuotationStatus.ACTIVE.value, end_time=PAST) + future = await _add_quotation(engine, status=QuotationStatus.ACTIVE.value, end_time=FUTURE) + already = await _add_quotation(engine, status=QuotationStatus.CLOSED.value, end_time=PAST) + deleted = await _add_quotation(engine, status=QuotationStatus.ACTIVE.value, end_time=PAST, deleted=True) + + n = await jobs.close_expired_quotations() + + assert n == 1 # 마감 대상은 due 1건뿐 + assert (await _quotation_row(engine, due)).status == QuotationStatus.CLOSED.value + assert (await _quotation_row(engine, future)).status == QuotationStatus.ACTIVE.value # 미래 → 안 건드림 + assert (await _quotation_row(engine, already)).status == QuotationStatus.CLOSED.value # 원래부터 CLOSED + assert (await _quotation_row(engine, deleted)).status == QuotationStatus.ACTIVE.value # 삭제분 → 제외 + + +# ----- 잡② close_negotiated_quotations : 대상 선정(전 세션 종결 + 세션 1개+) ----- +async def test_close_negotiated_picks_when_all_sessions_ended(clean): + engine = clean + # 전 세션 종결(거부) → 대상 + ended = await _add_quotation(engine, status=QuotationStatus.ACTIVE.value, end_time=FUTURE) + await _add_session(engine, ended, status=SessionStatus.REJECTED.value) + # 진행중 세션 하나라도 있으면 → 제외 + pending = await _add_quotation(engine, status=QuotationStatus.ACTIVE.value, end_time=FUTURE) + await _add_session(engine, pending, status=SessionStatus.DONE.value, bid_price=100) + await _add_session(engine, pending, status=SessionStatus.IN_PROGRESS.value) + # 세션 0개 → 제외 + no_session = await _add_quotation(engine, status=QuotationStatus.ACTIVE.value, end_time=FUTURE) + + await jobs.close_negotiated_quotations() + + assert (await _quotation_row(engine, ended)).status == QuotationStatus.CLOSED.value + assert (await _quotation_row(engine, pending)).status == QuotationStatus.ACTIVE.value + assert (await _quotation_row(engine, no_session)).status == QuotationStatus.ACTIVE.value + + +# ----- close_and_decide 위임 결과 스모크(잡①을 통해) ----- +async def test_award_single_lowest(clean): + engine = clean + qt = await _add_quotation(engine, end_time=PAST) + winner = uuid.uuid4() + await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=100, supplier_id=winner) + await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=200) + + await jobs.close_expired_quotations() + + row = await _quotation_row(engine, qt) + assert row.status == QuotationStatus.CLOSED.value + assert row.preferred_sp_yn is True + assert str(row.preferred_sp_id) == str(winner) # 최저가 단독 → 낙찰 확정 + + +async def test_rejected_just_closes(clean): + engine = clean + qt = await _add_quotation(engine, end_time=PAST) + await _add_session(engine, qt, status=SessionStatus.REJECTED.value) # 입찰 없는 거부만 + + await jobs.close_expired_quotations() + + row = await _quotation_row(engine, qt) + assert row.status == QuotationStatus.CLOSED.value + assert not row.preferred_sp_yn # 거부 → 낙찰 없이 그냥 마감 + + +# ----- 스케줄러 와이어링(start_scheduler) : DB 불필요 ----- +async def test_scheduler_disabled_without_env(monkeypatch): + import scheduler + monkeypatch.delenv("SCHEDULER_ENABLED", raising=False) + scheduler._scheduler = None + scheduler.start_scheduler() + assert scheduler._scheduler is None # SCHEDULER_ENABLED != 1 → 미기동 + + +async def test_scheduler_registers_both_jobs(monkeypatch): + import scheduler + monkeypatch.setenv("SCHEDULER_ENABLED", "1") + scheduler._scheduler = None + scheduler.start_scheduler() + try: + ids = {j.id for j in scheduler._scheduler.get_jobs()} + assert ids == {"close_expired_quotations", "close_negotiated_quotations"} + finally: + scheduler.shutdown_scheduler() + assert scheduler._scheduler is None + + +# ----- 스케줄러가 실제로 잡을 호출해 마감까지 가는지(라이브) ----- +async def test_scheduler_actually_runs_job_and_closes(clean): + """스케줄러에 잡을 걸면 정말 호출돼 견적이 마감되는지 확인. + 운영 트리거는 CronTrigger(minute='*/5')라 분 경계까지 기다려야 하므로, 여기선 + 1초 IntervalTrigger 로 같은 잡을 걸어 '스케줄러 → 잡 호출 → 마감' 경로만 몇 초 안에 검증한다.""" + engine = clean + qt = await _add_quotation(engine, end_time=PAST) + await _add_session(engine, qt, status=SessionStatus.DONE.value, bid_price=100) + + sched = AsyncIOScheduler(timezone="Asia/Seoul") + sched.add_job(jobs.close_expired_quotations, IntervalTrigger(seconds=1), max_instances=1) + sched.start() + try: + row = None + for _ in range(25): # 최대 ~5초 폴링(잡은 1초 뒤 첫 발화) + await asyncio.sleep(0.2) + row = await _quotation_row(engine, qt) + if row.status == QuotationStatus.CLOSED.value: + break + assert row is not None and row.status == QuotationStatus.CLOSED.value # 크론이 잡을 호출해 마감 + finally: + sched.shutdown(wait=False)