"""스케줄 잡 로직(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 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 import QuotationService async def _close_each(service: QuotationService, qt_ids) -> Counter: """대상 견적마다 close_and_decide 를 호출하되, 한 건의 예외가 배치 전체를 멈추지 않도록 격리한다. (예전 per-item try/continue 보존 — 한 견적의 DB 오류 등으로 나머지 견적이 이번 tick 에서 누락되면 안 됨.) 반환: 결과(CloseOutcome) 카운트 + 예외 발생 건수('error').""" results = Counter() for qt_id in qt_ids: try: results[await service.close_and_decide(qt_id)] += 1 except Exception as ex: results["error"] += 1 LOG.e_no_callstack(f"[scheduler] close_and_decide 실패 qt={qt_id}: {ex}") return results def _format_results(results: Counter) -> str: return ( f"낙찰 {results[CloseOutcome.AWARDED]} / 개찰 {results[CloseOutcome.OPENED]} / " f"마감 {results[CloseOutcome.CLOSED]} / 오류 {results['error']}" ) async def close_expired_quotations() -> int: """[잡①] 마감일이 지난 견적을 자동 마감 처리한다. 5분마다 실행(scheduler/__init__.py). 대상: 마감 시각이 이미 지났는데 아직 마감되지 않은(삭제되지도 않은) 견적. 처리: 견적마다 close_and_decide 로 결과 판정(낙찰 확정 / 개찰=낙찰자 미정 마감). 반환: 처리한 견적 수.""" crud = QuotationCRUD() service = QuotationService(crud) 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 results = await _close_each(service, qt_ids) if results: LOG.i(f"[scheduler] close_expired: {_format_results(results)}") return sum(results.values()) async def close_negotiated_quotations() -> int: """[잡②] 모든 세션의 협상이 끝난 견적은 마감일을 기다리지 않고 바로 마감한다(견적 타입 무관). 대상: 아직 마감되지 않았고, 진행중·미시작 세션이 하나도 없는(= 모두 종결된) 견적. 5분마다 실행. 처리: 견적마다 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_all_sessions_ended(s), ) if err_type != ErrorType.SUCCESS: LOG.e_no_callstack(f"[scheduler] close_negotiated 대상 조회 실패: {err_type.name}") return 0 results = await _close_each(service, qt_ids) if results: LOG.i(f"[scheduler] close_negotiated: {_format_results(results)}") return sum(results.values()) # ---- LPS(인터넷 최저가) 동기화 ------------------------------------------ async def sync_lps_results() -> int: """[잡③] lps_db.price_history 증분을 읽어 수집 이력 append + 상품 대표 최저가 박제. 5분마다. 검색 **요청**은 하지 않는다(수동 트리거 전용, 2026-07-10 협의) — 이 잡은 반영 백스톱. 워터마크(=이력의 max crawl_end_time) 기준 증분이라 멱등 — 실패 tick 은 다음 tick 이 흡수.""" from services.lps_sync_service import LpsSyncService service = LpsSyncService() if not service.available(): return 0 results = await service.sync_results() if results: LOG.i( f"[scheduler] lps_sync: found {results['found']} / not_found {results['not_found']} / " f"상품반영 {results['items_updated']} / 비uuid스킵 {results['skipped_not_uuid']} / " f"미존재상품 {results['skipped_unknown_item']} / 트랜잭션오류 {results['tx_error']}" ) return results["items_updated"]