o2o-negosium-original/negodata/backend/services/lps_sync_service.py

"""LPS(인터넷 최저가 검색) 동기화 서비스 — 요청·수집·반영의 도메인 로직.

흐름 (요청은 API, 결과는 DB — 2026-07-10 설계 결정)
  ① 요청: **수동 트리거 전용**(상품 화면 → POST /v1/item/{id}/lowest-price → 여기의
     request_search_for_item). 자동 요청 배치는 두지 않는다(2026-07-10 협의 — 크롤 비용이
     사용자 행동에만 비례하게). product_code = items.item_id(uuid 문자열) — 이걸로 결과가
     자동 매핑된다. enqueue 를 DB insert 로 하지 않는 이유: LPS 의 활성중복 dedupe·pg_notify
     워커 깨움을 우회하게 됨.
  ② 수집: lps_db.price_history(읽기전용)를 워터마크(max crawl_end_time) 증분으로 읽는다.
     (5분 크론 + 조회 API 의 온디맨드 — 트리거된 검색의 반영 백스톱, 크롤 비용 0)
  ③ 반영: 이력은 partner.item_internet_lowest_prices 에 append(성공/실패 모두),
     성공분의 상품별 최신값을 items.internet_lowest_price 에 박제(+yn).

방어
- LPS 미설정 환경(lps_db 미등록)이면 조용히 스킵 — 연동은 선택 기능, 본 서비스를 못 멈춘다.
- product_code 가 uuid 가 아니거나(item_id 아님 — 예: LPS 자체 부하테스트 잡) 상품이 없으면 스킵.
- 반영은 한 트랜잭션(execute_lambda_run) — 부분 반영으로 워터마크가 오염되지 않는다.
"""
import asyncio
import uuid
from collections import Counter
from datetime import timedelta, timezone

import httpx

from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import item_internet_lowest_prices
from common.enums import DBType, DBWRType, ErrorType, LowestPriceWebsite
from common.logger import LOG
from config.server_configs import web_server_config
from crud.lps_sync_crud import ILpsSyncCRUD, LpsSyncCRUD

SCAN_FLOOR_DAYS = 3         # price_history 스캔 하한(일) — 스킵행(비uuid 등)은 워터마크를
                            # 못 올리므로, 바닥 없이는 같은 행을 영원히 재스캔하게 된다


def _utc_now_aware():
    """aware UTC now — 이 서비스는 timestamptz 비교/워터마크에 쓰므로 GTime(naive)이 아닌 aware 를 쓴다."""
    from datetime import datetime
    return datetime.now(timezone.utc)


# 온디맨드 동기화(조회 API)와 크론 잡의 동시 실행 직렬화(프로세스 내).
# 프로세스 간 경합은 uq_iilp_item_crawl_time 유니크 인덱스가 최종 방어(진 쪽 tx 실패 → 다음 tick 흡수).
_sync_lock = asyncio.Lock()


class LpsSyncService:
    # FastAPI Depends() 로도 쓰이므로 무인자 생성자(파라미터가 있으면 DI 가 의존성으로 해석해 부팅 실패).
    # 테스트는 인스턴스 생성 후 .crud 교체로 주입한다.
    def __init__(self):
        self.crud: ILpsSyncCRUD = LpsSyncCRUD()

    @staticmethod
    def available() -> bool:
        """LPS 연동 활성 여부 — lps_db 가 등록된 환경에서만 배치가 돈다."""
        return DB_SESSION_MNG.is_registered(DBType.LPS.value)

    # ---- 단건 즉시 요청 (lowest-price 트리거 API 용) ---------------------
    async def request_search_for_item(self, item) -> tuple:
        """상품 1건을 즉시 LPS 에 검색 요청(수동 트리거 — job_type=manual, 배치보다 높은 우선순위).
        반환: (status, message) — queued | duplicated | unavailable."""
        if not self.available():
            return "unavailable", "LPS 연동이 비활성 상태입니다(설정 없음)"
        payload = {
            "product_code": str(item.item_id),
            "product_name": item.name,
            "job_type": "manual",
            "model": item.model_name or "",
            "specification": item.spec or "",
            "company": item.manufacturer or "",
            "price": str(item.price) if item.price else "",
        }
        base = web_server_config.lps_base_url.rstrip("/")
        try:
            async with httpx.AsyncClient(timeout=10.0) as client:
                r = await client.post(f"{base}/v1/lps/search", json={"data": [payload]})
                r.raise_for_status()
                body = r.json()
            entry = (body.get("items") or [{}])[0]
            if entry.get("duplicated"):
                return "duplicated", "이미 검색이 진행 중입니다(활성 중복)"
            return "queued", "최저가 검색이 접수됐습니다"
        except Exception as ex:
            LOG.w(f"[lps-sync] 단건 검색요청 실패 item={item.item_id}: {type(ex).__name__}: {ex}")
            return "unavailable", "검색 서비스에 연결할 수 없습니다"

    # ---- 조회 뷰 (lowest-price 조회 API 용) ------------------------------
    async def lowest_price_view(self, item_id, limit: int = 10) -> tuple:
        """대표 최저가 + 최근 수집 이력. 조회 전에 증분 동기화를 한 번 돌려
        5분 크론을 기다리지 않고 최신 결과를 반영한다(멱등·증분이라 저비용).
        반환: (ErrorType, history_rows)."""
        if self.available():
            try:
                await self.sync_results()
            except Exception as ex:  # 조회는 동기화 실패에도 계속(마지막 반영분이라도 보여준다)
                LOG.w(f"[lps-sync] 조회 전 동기화 실패(무시): {type(ex).__name__}: {ex}")
        return await DB_SESSION_MNG.execute_lambda(
            DBType.MAIN.value, DBWRType.DB_READ.value,
            lambda s: self.crud.recent_history(s, item_id, limit),
        )

    # ---- ②+③ 수집·반영: price_history 증분 → 이력 append + 대표값 박제 --
    async def sync_results(self) -> Counter:
        async with _sync_lock:   # 크론 잡과 온디맨드 조회의 프로세스 내 동시 실행 직렬화
            return await self._sync_results_locked()

    async def _sync_results_locked(self) -> Counter:
        results = Counter()
        if not self.available():
            return results

        err, wm = await DB_SESSION_MNG.execute_lambda(
            DBType.MAIN.value, DBWRType.DB_READ.value, lambda s: self.crud.watermark(s))
        if err != ErrorType.SUCCESS:
            return results

        # 스캔 시작점 = max(워터마크, 최근 N일 바닥) — 반영 안 되는 행(비uuid·미존재 상품)이
        # 섞여 있어도 재스캔 범위가 유한하게 유지된다(N일 지나면 자연 소멸).
        # ⚠️ 이 경로의 시각은 전부 aware UTC 로 다룬다. naive 를 timestamptz 파라미터로 넘기면
        # PG 가 세션 타임존(KST)으로 해석해 9시간 어긋난 증분을 읽는다(중복 수집 실측 버그).
        floor = _utc_now_aware() - timedelta(days=SCAN_FLOOR_DAYS)
        since = max(wm, floor) if wm is not None else floor

        err, rows = await DB_SESSION_MNG.execute_lambda(
            DBType.LPS.value, DBWRType.DB_READ.value, lambda s: self.crud.fetch_results_since(s, since))
        if err != ErrorType.SUCCESS or not rows:
            return results

        # product_code(uuid=item_id) 검증 — LPS 부하테스트 등 비상품 코드는 조용히 스킵
        parsed = []
        for code, outcome, final_lowest, final_source, created_at in rows:
            try:
                parsed.append((uuid.UUID(code), outcome, final_lowest, final_source, created_at))
            except (ValueError, AttributeError, TypeError):
                results["skipped_not_uuid"] += 1

        err, existing = await DB_SESSION_MNG.execute_lambda(
            DBType.MAIN.value, DBWRType.DB_READ.value,
            lambda s: self.crud.existing_item_ids(s, list({p[0] for p in parsed})),
        )
        if err != ErrorType.SUCCESS:
            return results

        history_rows, latest_found = [], {}   # latest_found: item_id → (created_at, price)
        for item_id, outcome, final_lowest, final_source, created_at in parsed:
            if item_id not in existing:
                results["skipped_unknown_item"] += 1
                continue
            found = outcome == "found" and final_lowest is not None
            history_rows.append(item_internet_lowest_prices(
                item_id=item_id,
                lp_price=final_lowest if found else None,
                website=LowestPriceWebsite.from_source(final_source).value,
                success_yn=found,
                fail_reason=None if found else (outcome or "unknown")[:100],
                crawl_end_time=created_at,   # 워터마크 기준값 — price_history.created_at 그대로 보존
            ))
            results["found" if found else "not_found"] += 1
            if found and (item_id not in latest_found or created_at > latest_found[item_id][0]):
                latest_found[item_id] = (created_at, final_lowest)

        if not history_rows:
            return results

        funcs = [lambda s, r=history_rows: self.crud.add_history_rows(s, r)]
        funcs += [
            (lambda s, i=item_id, p=price: self.crud.update_item_lowest(s, i, p))
            for item_id, (_, price) in latest_found.items()
        ]
        err = await DB_SESSION_MNG.execute_lambda_run([DBType.MAIN.value] * len(funcs), funcs)
        if err != ErrorType.SUCCESS:
            LOG.e_no_callstack(f"[lps-sync] 반영 트랜잭션 실패: {err.name} (다음 tick 재시도)")
            return Counter({"tx_error": 1})
        results["items_updated"] = len(latest_found)
        return results