"""카페24 예약 실행 worker. python -m app.modules.cafe24.worker --loop 60 # 60초마다 확인 (운영) python -m app.modules.cafe24.worker --once # 한 번만 처리하고 종료 왜 별도 프로세스인가 예약은 브라우저를 닫아도, 아무도 화면을 보고 있지 않아도 그 시각에 실행돼야 한다. 웹 요청 안에서 기다리는 방식은 프록시 타임아웃·재기동에 그대로 무너진다. compose 서비스 `dbx-cafe24-worker` 가 web 과 같은 이미지로 이 모듈을 돌린다. 한 번에 한 건씩 처리한다 `claim_due_schedule` 이 `FOR UPDATE SKIP LOCKED` 로 한 건을 잠그고 PROCESSING 으로 바꾼다. worker 가 실수로 두 개 떠도 같은 예약이 두 번 적용되지 않는다. 적용 순서는 화면 편집과 같다 카페24 현재값 재조회 → BACKUP revision → PUT → SUCCESS + 감사로그 실패하면 재시도 예산(store.MAX_RETRY) 안에서 간격을 두고 다시 시도하고, 소진되면 FAILED 로 확정한다. 되돌리기는 쓰지 않는다. 토큰 갱신은 TokenService 가 행 잠금 안에서 하므로 web 과 동시에 떠 있어도 안전하다. """ from __future__ import annotations import argparse import logging import os import sys import time from datetime import timedelta from typing import Any from app.integrations.cafe24 import Cafe24Error, build_cafe24_api, products from app.timezone import now_kst from . import store logger = logging.getLogger("cafe24.worker") ACTOR = "SCHEDULER" def _apply(store_db: Any, api: Any, row: dict[str, Any]) -> str: """예약 1건 적용. 성공 시 사람이 읽을 요약 문자열.""" schedule_id = int(row["id"]) product_no = int(row["product_no"]) revision_id = row.get("revision_id") set_display = row.get("set_display") set_selling = row.get("set_selling") html: str | None = None if revision_id: revision = store_db.get_revision(int(revision_id)) if not revision: raise Cafe24Error(f"예약이 가리키는 버전 {revision_id} 을 찾을 수 없습니다.") html = revision.get("html_content") or "" if not html.strip(): raise Cafe24Error(f"버전 {revision_id} 의 내용이 비어 있습니다.") # 상세설명을 바꿀 때는 쓰기 직전 현재값을 읽어 백업한다(로컬 값을 믿지 않는다). backup_id = 0 if html is not None: current = products.fetch_descriptions(api.client, product_no) backup_id = store_db.add_revision( product_no=product_no, html_content=current.description, revision_type=store.REVISION_BACKUP, memo=f"예약 #{schedule_id} 적용 직전 자동 백업", created_by=ACTOR, ) if current.mobile_description and current.mobile_description != current.description: store_db.add_revision( product_no=product_no, html_content=current.mobile_description, revision_type=store.REVISION_BACKUP, memo=f"예약 #{schedule_id} 적용 직전 자동 백업 (모바일)", created_by=ACTOR, ) # PC/모바일은 구분하지 않는다 — mobile_description 은 보내지 않고 # separated_mobile_description="F" 로 "PC 상세설명과 동일"을 강제한다. # (mobile_description 을 직접 보내면 카페24가 그 설정을 "직접 등록"으로 바꿔버린다.) products.update_product( api.client, product_no, description=html, separated_mobile_description="F" if html is not None else None, display=set_display, selling=set_selling, ) if html is not None: # 화면 편집(apply)과 동일 — 카페24 관리자 API 의 쓰기 직후 읽기 지연을 # 여기서 짧게 흡수한다(쇼핑몰에는 바로 반영되지만 관리자 조회만 뒤쳐질 때가 있다). products.wait_for_description(api.client, product_no, html) summary = store.describe_schedule_action( has_html=html is not None, set_display=set_display, set_selling=set_selling ) store_db.log_audit( actor=ACTOR, action="schedule_apply", product_no=product_no, revision_id=int(revision_id) if revision_id else None, schedule_id=schedule_id, result="SUCCESS", detail=summary + (f" (백업 {backup_id})" if backup_id else ""), ) return summary def process_once(store_db: Any, api: Any) -> int: """실행할 예약을 모두 처리한다. 처리한 건수를 돌려준다.""" handled = 0 while True: with store_db.claim_due_schedule(now=now_kst()) as row: if row is None: return handled # 잠금은 여기서 이미 풀렸다. 상태가 PROCESSING 이라 다른 worker 가 집지 않는다. schedule_id = int(row["id"]) product_no = int(row["product_no"]) try: summary = _apply(store_db, api, row) except Cafe24Error as exc: retry_count = int(row.get("retry_count") or 0) if store.can_retry(retry_count): wait = store.retry_backoff_seconds(retry_count) store_db.finish_schedule( schedule_id, status=store.STATUS_PENDING, error=str(exc), next_retry_at=now_kst() + timedelta(seconds=wait), retry_count=retry_count + 1, ) logger.warning( "예약 #%s 상품 %s 실패 — %s초 후 재시도 (%s/%s): %s", schedule_id, product_no, wait, retry_count + 1, store.MAX_RETRY, exc, ) else: store_db.finish_schedule( schedule_id, status=store.STATUS_FAILED, error=str(exc) ) store_db.log_audit( actor=ACTOR, action="schedule_apply", product_no=product_no, schedule_id=schedule_id, result="FAIL", detail=str(exc), ) logger.error("예약 #%s 상품 %s 최종 실패: %s", schedule_id, product_no, exc) except Exception as exc: # noqa: BLE001 — 한 건의 사고가 worker 를 죽이면 안 된다. store_db.finish_schedule( schedule_id, status=store.STATUS_FAILED, error=f"{type(exc).__name__}: {exc}" ) logger.exception("예약 #%s 처리 중 예상치 못한 오류", schedule_id) else: store_db.finish_schedule(schedule_id, status=store.STATUS_SUCCESS) logger.info("예약 #%s 상품 %s 적용 완료 — %s", schedule_id, product_no, summary) handled += 1 def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="카페24 예약 실행 worker") parser.add_argument("--loop", type=int, default=0, help="확인 간격(초). 0 이면 한 번만") parser.add_argument("--once", action="store_true", help="한 번만 처리하고 종료") args = parser.parse_args(argv) logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s", stream=sys.stdout, ) dsn = (os.getenv("CAFE24_DB_URL") or "").strip() if not dsn: logger.error("CAFE24_DB_URL 이 설정되지 않았습니다. worker 를 시작할 수 없습니다.") return 1 # 지연 import — psycopg 가 없는 개발 환경에서도 이 모듈을 열어볼 수 있게. from .db import Cafe24Store # noqa: WPS433 store_db = Cafe24Store(dsn) api = build_cafe24_api(store_db) interval = 0 if args.once else max(0, int(args.loop)) logger.info("카페24 예약 worker 시작 (간격 %s초)", interval or "단발") try: while True: try: count = process_once(store_db, api) if count: logger.info("예약 %s건 처리", count) except Exception: # noqa: BLE001 — DB 순간 장애로 죽지 않게 logger.exception("예약 처리 루프에서 오류 — 다음 주기에 다시 시도") if not interval: return 0 time.sleep(interval) except KeyboardInterrupt: logger.info("종료 요청 — worker 를 멈춥니다.") return 0 finally: store_db.close() if __name__ == "__main__": raise SystemExit(main())