o2o-negosium-original/lps/crud/price_history.py

"""최저가 스냅샷 CRUD — 트리거 시점마다 기록하고, 상품별 시계열로 조회(그래프)."""

import json

from sqlalchemy import text

from common.database.db_session_manager import DB_SESSION_MNG
from common.enums import DBType, DBWRType

_FIELDS = (
    "product_code", "job_id", "outcome", "matched_count",
    "naver_lowest", "naver_name", "naver_url",
    "coupang_lowest", "coupang_name", "coupang_url",
    "final_lowest", "final_source", "final_rating", "final_review_count",
    "final_shipping_fee", "final_shipping_type", "final_shipping_label",
    "partial",
)


class PriceHistory:
    DB = DBType.MAIN.value

    async def record(self, event: dict):
        """스냅샷 1건 저장. triggered_at 은 now()(관측 시각). 로깅 실패가 검색을 막지 않도록 호출부에서 예외 처리."""
        sql = text("""
            INSERT INTO price_history
                (product_code, job_id, outcome, matched_count,
                 naver_lowest, naver_name, naver_url,
                 coupang_lowest, coupang_name, coupang_url,
                 final_lowest, final_source, final_rating, final_review_count,
                 final_shipping_fee, final_shipping_type, final_shipping_label,
                 by_mall, sources, partial)
            VALUES
                (:product_code, :job_id, :outcome, :matched_count,
                 :naver_lowest, :naver_name, :naver_url,
                 :coupang_lowest, :coupang_name, :coupang_url,
                 :final_lowest, :final_source, :final_rating, :final_review_count,
                 :final_shipping_fee, :final_shipping_type, :final_shipping_label,
                 CAST(:by_mall AS jsonb), CAST(:sources AS jsonb), COALESCE(:partial, FALSE))
        """)
        params = {k: event.get(k) for k in _FIELDS}
        for col in ("by_mall", "sources"):   # JSONB 는 문자열로 넘겨 CAST 한다
            v = event.get(col)
            params[col] = json.dumps(v, ensure_ascii=False) if v is not None else None
        s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_WRITE.value)
        try:
            await s.execute(sql, params)
            await s.commit()
        except Exception:
            await s.rollback()
            raise
        finally:
            await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_WRITE.value)

    async def list_by_product(self, product_code: str, limit: int = 100) -> list[dict]:
        """상품의 최근 스냅샷을 시각 오름차순(그래프 플롯용)으로 반환. 최근 limit 건."""
        sql = text("""
            SELECT * FROM (
                SELECT triggered_at, outcome, matched_count,
                       naver_lowest, naver_name, naver_url,
                       coupang_lowest, coupang_name, coupang_url,
                       final_lowest, final_source, by_mall
                FROM price_history
                WHERE product_code = :pc
                ORDER BY triggered_at DESC
                LIMIT :lim
            ) t ORDER BY triggered_at ASC
        """)
        s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value)
        try:
            rows = (await s.execute(sql, {"pc": product_code, "lim": limit})).mappings().all()
            return [dict(r) for r in rows]
        finally:
            await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)

    async def list_products(self, q: str | None = None, limit: int = 50) -> list[dict]:
        """이력이 있는 상품 목록(관리자 FE) — 상품별 최신 스냅샷 + 검색 횟수, 최근 검색순.
        상품명은 price_history 에 없어 최신 스냅샷의 매칭 상품명(네이버 우선)으로 대신한다."""
        where = "WHERE product_code ILIKE :q" if q else ""
        sql = text(f"""
            SELECT * FROM (
                SELECT DISTINCT ON (product_code)
                       product_code, triggered_at, outcome,
                       naver_lowest, coupang_lowest, final_lowest, final_source,
                       COALESCE(naver_name, coupang_name) AS display_name,
                       count(*) OVER (PARTITION BY product_code) AS searches
                FROM price_history {where}
                ORDER BY product_code, triggered_at DESC
            ) t ORDER BY triggered_at DESC LIMIT :lim
        """)
        params: dict = {"lim": limit}
        if q:
            params["q"] = f"%{q}%"
        s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value)
        try:
            return [dict(r) for r in (await s.execute(sql, params)).mappings().all()]
        finally:
            await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value)