[wip] negodata/backend: 견적 마감 스케줄러

- scheduler/(__init__·jobs): 마감 처리·낙찰자 선정 잡
- quotation_crud·service: 스케줄러 연동 조회/마감 로직
- web_main·requirements·docker-compose·config: 스케줄러 기동 설정

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Mina Choi 2026-06-22 17:10:26 +09:00
parent ae103fd0f1
commit 5406094ea1
9 changed files with 383 additions and 22 deletions

View File

@ -37,3 +37,9 @@ access_key = "<JWT_ACCESS_SECRET>"
refresh_key = "<JWT_REFRESH_SECRET>"
access_expire_min = 30
refresh_expire_day = 7
# 협상 agent(포트 9500) 접속. use_mock=true 면 agent 미연동 — 내장 mock 응답 사용(통합 테스트/로컬 기본).
[AgentConfig]
base_url = "http://127.0.0.1:9500"
timeout_sec = 10.0
use_mock = true

View File

@ -35,6 +35,10 @@ services:
environment:
APP_ENV: local
DB_HOST: host.docker.internal # 컨테이너→호스트 DB (config.local.toml의 127.0.0.1 override)
RELOAD: "1" # uvicorn --reload 활성 → 소스 저장 시 자동 재기동(재빌드 불필요)
SCHEDULER_ENABLED: "1" # 마감 크론 활성(단일 워커라 중복 없음). 운영 다중 워커면 1개 프로세스에서만 1
volumes:
- ./negodata/backend:/app # 호스트 소스 = 컨테이너 코드. 이게 있어야 수정이 즉시 반영됨
ports:
- "9400:9400"
extra_hosts:

View File

@ -7,10 +7,10 @@ from sqlalchemy.ext.asyncio import AsyncSession
from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import (
quotations, sessions, chats, nego_cards, wild_cards, items, quotation_settings,
quotations, sessions, chats, nego_cards, wild_cards, items, suppliers, quotation_settings,
version_nego_cards, version_wild_cards,
)
from common.enums import ErrorType
from common.enums import ErrorType, QuotationStatus, QuotationType, SessionStatus
from common.logger import LOG
from common.utils.gtime import GTime
@ -59,6 +59,10 @@ class IQuotationCRUD(ABC):
async def update_quotation(self, cdb: AsyncSession, qt_id, data: dict) -> ErrorType:
pass
@abstractmethod
async def update_sessions_status(self, cdb: AsyncSession, qt_id, from_statuses: list[int], to_status: int) -> ErrorType:
pass
@abstractmethod
async def soft_delete(self, cdb: AsyncSession, qt_id) -> ErrorType:
pass
@ -83,6 +87,27 @@ class IQuotationCRUD(ABC):
async def item_map(self, cdb: AsyncSession, qt_ids) -> Tuple[ErrorType, dict]:
pass
# ----- 스케줄러(크론) 전용 -----
@abstractmethod
async def list_due_for_close(self, cdb: AsyncSession, now) -> Tuple[ErrorType, list]:
pass
@abstractmethod
async def list_requote_done(self, cdb: AsyncSession) -> Tuple[ErrorType, list]:
pass
@abstractmethod
async def list_done_sessions(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
@abstractmethod
async def bulk_update_sessions_status(self, cdb: AsyncSession, qt_ids, from_statuses: list[int], to_status: int) -> ErrorType:
pass
class QuotationCRUD(IQuotationCRUD):
async def search(
@ -311,6 +336,23 @@ class QuotationCRUD(IQuotationCRUD):
LOG.e_no_callstack(ex)
return ErrorType.DB_RUN_FAILED
async def update_sessions_status(self, cdb: AsyncSession, qt_id, from_statuses: list[int], to_status: int) -> ErrorType:
# 견적에 딸린 세션 중 from_statuses 에 속한 것만 to_status 로 일괄 전이(삭제 제외). 다른 상태는 건드리지 않는다.
try:
query = (
update(sessions)
.where(
sessions.quotation_id == qt_id,
sessions.status.in_(from_statuses),
sessions.deleted == False, # noqa: E712
)
.values(status=to_status, 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
async def soft_delete(self, cdb: AsyncSession, qt_id) -> ErrorType:
try:
query = update(quotations).where(quotations.qt_id == qt_id).values(deleted=True, updated_at=GTime.UTC())
@ -319,6 +361,106 @@ class QuotationCRUD(IQuotationCRUD):
LOG.e_no_callstack(ex)
return ErrorType.DB_RUN_FAILED
# ----- 스케줄러(크론) 전용 -----
async def list_due_for_close(self, cdb: AsyncSession, now) -> Tuple[ErrorType, list]:
"""[잡①] 마감시각이 지났는데 아직 안 닫힌 견적 qt_id 목록.
조건: end_time < now AND status != 견적마감 AND not deleted."""
try:
query = select(quotations.qt_id).where(
quotations.end_time < now,
quotations.status != QuotationStatus.CLOSED.value,
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 list_requote_done(self, cdb: AsyncSession) -> Tuple[ErrorType, list]:
"""[잡②] 재견적(REQUOTE) 중 협상완료(DONE) 세션이 1건 이상이고 아직 안 닫힌 견적 qt_id 목록.
재견적은 세션이 독립적이라 하나라도 완료되면 나머지를 기다리지 않고 마감 대상."""
try:
done_exists = (
select(sessions.session_id)
.where(
sessions.quotation_id == quotations.qt_id,
sessions.status == SessionStatus.DONE.value,
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,
)
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_done_sessions(self, cdb: AsyncSession, qt_id) -> Tuple[ErrorType, list]:
"""[잡②] 견적의 협상완료(DONE) 세션 → (supplier_id, bid_price, supplier_name) 목록. 낙찰자 판정 입력."""
try:
query = (
select(sessions.supplier_id, sessions.bid_price, suppliers.name)
.join(suppliers, suppliers.supplier_id == sessions.supplier_id)
.where(
sessions.quotation_id == qt_id,
sessions.status == SessionStatus.DONE.value,
sessions.deleted == False, # noqa: E712
suppliers.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:
if not qt_ids:
return ErrorType.SUCCESS
query = (
update(quotations)
.where(quotations.qt_id.in_(qt_ids))
.values(status=status, 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
async def bulk_update_sessions_status(self, cdb: AsyncSession, qt_ids, from_statuses: list[int], to_status: int) -> ErrorType:
"""[잡①] 여러 견적에 딸린 세션 중 from_statuses 에 속한 것만 to_status 로 일괄 전이(삭제 제외)."""
try:
if not qt_ids:
return ErrorType.SUCCESS
query = (
update(sessions)
.where(
sessions.quotation_id.in_(qt_ids),
sessions.status.in_(from_statuses),
sessions.deleted == False, # noqa: E712
)
.values(status=to_status, 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
# ----- 견적 상세: 세션 / 채팅 / 사용카드 (읽기 전용) -----
async def list_sessions(self, cdb: AsyncSession, qt_id) -> Tuple[ErrorType, list]:
try:

View File

@ -10,3 +10,4 @@ pydantic>=2.0
python-multipart
openpyxl
httpx
apscheduler>=3.10

View File

@ -0,0 +1,74 @@
"""백그라운드 스케줄러(크론) 패키지 — '언제'(when) 담당.
router/(HTTP 진입점) 동급의 '시간 진입점' 계층. APScheduler 수명주기와 등록(타이밍) 책임지고,
실제로 하는 (what) scheduler/jobs.py 있다.
- 다중 워커(운영)에서 잡이 워커마다 중복 실행되면 되므로 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 세션 있으면 즉시 마감
"""
import os
from common.logger import LOG
from scheduler import jobs
__all__ = ["start_scheduler", "shutdown_scheduler"]
_scheduler = None # AsyncIOScheduler | None
def _is_enabled() -> bool:
return os.environ.get("SCHEDULER_ENABLED", "0") == "1"
def start_scheduler():
"""lifespan startup 에서 호출. SCHEDULER_ENABLED=1 일 때만 스케줄러를 띄운다."""
global _scheduler
if not _is_enabled():
LOG.i("[scheduler] disabled (SCHEDULER_ENABLED != 1)")
return
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
_scheduler.add_job(
jobs.close_expired_quotations,
CronTrigger(hour=0, minute=10),
id="close_expired_quotations",
coalesce=True, # 밀린 실행이 여러 번 쌓여도 1번만
misfire_grace_time=3600, # 정시보다 늦게 깨어나도 1시간 내면 실행
max_instances=1,
)
# 잡 ② 재견적 협상완료 처리: 1시간마다
_scheduler.add_job(
jobs.complete_requote_quotations,
IntervalTrigger(hours=1),
id="complete_requote_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)")
def shutdown_scheduler():
"""lifespan shutdown 에서 호출."""
global _scheduler
if _scheduler is not None:
_scheduler.shutdown(wait=False)
_scheduler = None
LOG.i("[scheduler] stopped")

View File

@ -0,0 +1,124 @@
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.logger import LOG
from common.utils.gtime import GTime
from crud.quotation_crud import QuotationCRUD
async def close_expired_quotations() -> int:
"""[잡①] 마감일이 지난 견적을 자동으로 견적마감 처리한다. 하루 한 번 실행.
대상: 마감 시각이 이미 지났는데 아직 마감되지 않은(삭제되지도 않은) 견적.
처리: 견적들을 견적마감 상태로 바꾸고, 아직 시작 전인 세션은 미참여로 정리한다.
반환: 마감 처리한 견적 ."""
crud = QuotationCRUD()
now = GTime.UTC()
err_type, qt_ids = await DB_SESSION_MNG.execute_lambda(
quotations.DBType(),
DBWRType.DB_READ.value,
lambda s: crud.list_due_for_close(s, now),
)
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)
async def complete_requote_quotations() -> int:
"""[잡②] 재견적은 협상완료된 세션이 생기면 나머지를 기다리지 않고 바로 마감한다. 한 시간마다 실행.
대상: 아직 마감되지 않은 재견적 견적 , 협상완료된 세션이 있는 .
낙찰: 협상완료된 세션 입찰가가 가장 낮은 공급사를 낙찰자로 정한다. 같은 최저가가 이상이면(동가) 낙찰자를 비우고 동가 정보만 남긴다.
(현재 재견적은 견적당 세션이 하나라 실제로는 단독 낙찰만 일어나지만, 모델상 1:N이라 일반 규칙을 그대로 둔다.)
처리: 낙찰 정보를 기록하고 견적을 견적마감 상태로 바꾸며, 아직 시작 전인 세션은 미참여로 정리한다.
반환: 마감 처리한 견적 ."""
crud = QuotationCRUD()
err_type, qt_ids = await DB_SESSION_MNG.execute_lambda(
quotations.DBType(),
DBWRType.DB_READ.value,
lambda s: crud.list_requote_done(s),
)
if err_type != ErrorType.SUCCESS:
LOG.e_no_callstack(f"[scheduler] complete_requote 대상 조회 실패: {err_type.name}")
return 0
if not qt_ids:
return 0
closed = 0
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

View File

@ -274,10 +274,16 @@ class QuotationService:
res.result.SetResult(err_type)
return res
# 상태를 '견적마감'으로 변경(실제 DB 업데이트)
# 견적 '견적마감'(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_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)
@ -387,6 +393,7 @@ class QuotationService:
return res
# chats.seq → ChatMessageData.index 로 매핑. indicator_value(Decimal) → float.
# 말풍선 텍스트는 chats.meta.script 에 영속화돼 있어 그대로 꺼낸다(프론트 하드코딩 X).
res.messages = [
ChatMessageData(
chat_id=r.chat_id,
@ -398,6 +405,8 @@ class QuotationService:
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,
script=(r.meta or {}).get("script"),
step=(r.meta or {}).get("step"),
)
for r in rows
]

View File

@ -6,6 +6,8 @@
# 또는 uvicorn 직접 실행:
# uvicorn router.router:app --reload --host=0.0.0.0 --port=9400
import os
import uvicorn
from common.logger import LOG
@ -21,21 +23,20 @@ if __name__ == "__main__":
LOG.i(f"Server Port : {web_server_config.port}")
LOG.i(f"API Server start time : {router.router.API_SERVER_START_TIME}")
if web_server_config.is_ssl:
uvicorn.run(
"router.router:app",
host="0.0.0.0",
port=web_server_config.port,
access_log=False,
workers=web_server_config.process_count,
ssl_keyfile="./SSL/key.pem",
ssl_certfile="./SSL/cert.pem",
)
# RELOAD=1 (개발 컨테이너) → 소스 변경 시 자동 재기동. reload 와 workers(다중) 는 함께 못 쓰므로 분기.
reload = os.environ.get("RELOAD") == "1"
run_kwargs = dict(
host="0.0.0.0",
port=web_server_config.port,
access_log=False,
)
if reload:
run_kwargs["reload"] = True
else:
uvicorn.run(
"router.router:app",
host="0.0.0.0",
port=web_server_config.port,
access_log=False,
workers=web_server_config.process_count,
)
run_kwargs["workers"] = web_server_config.process_count
if web_server_config.is_ssl:
run_kwargs["ssl_keyfile"] = "./SSL/key.pem"
run_kwargs["ssl_certfile"] = "./SSL/cert.pem"
uvicorn.run("router.router:app", **run_kwargs)

View File

@ -25,7 +25,7 @@ export function QuotationCardsTab({ quotationCardViews }: { quotationCardViews:
<TableCell className="p-3 text-foreground font-sans font-semibold">
{qc.card_id ? (
<Link
to={`/cards?edit=${qc.card_id}`}
to={`/cards?detail=${qc.card_id}`}
className="hover:text-primary hover:underline"
title={`${qc.card_name} — 협상카드 상세로 이동`}
>