o2o-negosium-original/negodata/backend/scheduler/__init__.py
민헌 523661e3d2 feat(negodata): LPS 최저가 동기화 배치 — 요청(API)·수집(lps_db)·반영(items)
설계(2026-07-10 결정): 요청은 LPS API(dedupe·워커알림 보존), 결과는
lps_db.price_history 직접 읽기(읽기전용 엔진). product_code=item_id
로 결과가 자동 매핑된다.

- 잡③ request_lps_searches(매일 04:00 KST): 24h 이상 미수집 상품을
  최대 500건 enqueue(job_type=batch, 100건/콜 청크) — 비용 발생 잡
- 잡④ sync_lps_results(5분): 워터마크(max crawl_end_time) 증분 수집
  → item_internet_lowest_prices append(성공/실패 모두) + 성공분 최신값
  items.internet_lowest_price 박제(+yn). 한 트랜잭션(부분반영 방지)
- 스캔 바닥 3일 — 비uuid·미존재 상품 행이 워터마크를 못 올려도
  재스캔 범위 유한
- ⚠️ 시각은 aware UTC 통일 — naive 를 timestamptz 파라미터로 넘기면
  PG 세션 타임존(KST) 해석으로 9시간 어긋남(중복 수집 실측 버그 수정)
- LowestPriceWebsite 코드 enum(naver=1 coupang=2 …), iilp ORM 모델,
  lps_base_url config(+LPS_BASE_URL env)

검증: 실상품 1건 e2e(요청→크롤 found 15,000원→이력+박제 반영),
멱등성(재실행 빈 카운터), 비uuid 12건 스킵, 대상선정 쿼리 500건 상한

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-10 15:38:57 +09:00

85 lines
3.1 KiB
Python

"""백그라운드 스케줄러(크론) 패키지 — '언제'(when) 담당.
다중 워커(운영)에서 잡이 워커마다 중복 실행되면 안 되므로 SCHEDULER_ENABLED=1 인 프로세스에서만 등록한다.
잡 ① close_expired_quotations : 5분마다(KST) — 마감시각 지난 견적 마감
잡 ② close_negotiated_quotations: 5분마다(KST) — 모든 세션 협상 끝난 견적 즉시 마감(타입 무관)
잡 ③ request_lps_searches : 매일 04:00(KST) — 갱신 오래된 상품을 LPS 에 검색 요청(비용 발생 잡)
잡 ④ sync_lps_results : 5분마다(KST) — lps_db 결과 증분 수집 → 이력 append + 상품 최저가 박제
"""
import os
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
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
# 한국시간 기준, 두 잡 모두 5분마다 실행.
_scheduler = AsyncIOScheduler(timezone="Asia/Seoul")
# 잡 ① 마감시각 지난 견적 처리
_scheduler.add_job(
jobs.close_expired_quotations,
CronTrigger(minute="*/5"),
id="close_expired_quotations",
coalesce=True, # 밀린 실행이 쌓여도 1번만
misfire_grace_time=600, # 정시보다 늦게 깨어나도 10분 내면 실행
max_instances=1,
)
# 잡 ② 모든 세션 협상 끝난 견적 즉시 마감(타입 무관)
_scheduler.add_job(
jobs.close_negotiated_quotations,
CronTrigger(minute="*/5"),
id="close_negotiated_quotations",
coalesce=True,
misfire_grace_time=600,
max_instances=1,
)
# 잡 ③ LPS 검색 요청(비용 발생) — 새벽 1회. LPS 미설정 환경이면 잡 내부에서 스킵.
_scheduler.add_job(
jobs.request_lps_searches,
CronTrigger(hour=4, minute=0),
id="request_lps_searches",
coalesce=True,
misfire_grace_time=3600, # 재기동 등으로 놓쳐도 1시간 내면 실행
max_instances=1,
)
# 잡 ④ LPS 결과 수집·반영 — 증분·멱등이라 잦아도 안전
_scheduler.add_job(
jobs.sync_lps_results,
CronTrigger(minute="*/5"),
id="sync_lps_results",
coalesce=True,
misfire_grace_time=600,
max_instances=1,
)
_scheduler.start()
LOG.i("[scheduler] started (KST: 견적마감 2잡 5분 · LPS 요청 04:00 · LPS 수집 5분)")
def shutdown_scheduler():
"""lifespan shutdown 에서 호출."""
global _scheduler
if _scheduler is not None:
_scheduler.shutdown(wait=False)
_scheduler = None
LOG.i("[scheduler] stopped")