From 5906acc48adefa23ca2fecb428f3bb030e38afaa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=AF=BC=ED=97=8C?= Date: Thu, 9 Jul 2026 11:13:28 +0900 Subject: [PATCH] =?UTF-8?q?feat(lps):=20=EC=B5=9C=EC=A0=80=EA=B0=80=20?= =?UTF-8?q?=EC=9D=B4=EB=A0=A5(price=5Fhistory)=20=E2=80=94=20=ED=8A=B8?= =?UTF-8?q?=EB=A6=AC=EA=B1=B0=20=EA=B8=B0=EB=B0=98=20=EC=8B=9C=EA=B3=84?= =?UTF-8?q?=EC=97=B4=20=EA=B7=B8=EB=9E=98=ED=94=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 같은 상품 반복 검색 시 최저가를 스냅샷으로 적재 → 네이버/쿠팡/최종 3개 선 그래프. 배치 아님(조회된 상품만, 실제 검색 시각에 기록) — 트래픽/리소스 절약. - price_history 테이블: product_code·triggered_at(X축)·naver/coupang/final 최저가+상세·outcome - crud/price_history: record() + list_by_product(시각 오름차순) - handler: AI 매칭 후 소스별 min + 전체 min 스냅샷 기록(_price_snapshot). found/not_found 기록, 네거티브 캐시 히트·기술실패는 미기록 - API: GET /v1/lps/products/{product_code}/history → 그래프 데이터(시각 오름차순) - tests: 스냅샷 계산/기록·조회/핸들러 기록규칙/API → 전체 54/54 Co-Authored-By: Claude Opus 4.8 (1M context) --- lps/common/database/model/models.py | 34 +++++++++++ lps/crud/price_history.py | 63 +++++++++++++++++++ lps/router/v1/lps/protocol.py | 19 ++++++ lps/router/v1/lps/search.py | 14 ++++- lps/services/lps_service.py | 23 ++++++- lps/tests/test_lps_api.py | 18 ++++++ lps/tests/test_price_history.py | 93 +++++++++++++++++++++++++++++ lps/worker/handlers.py | 34 ++++++++++- lps/worker_main.py | 6 +- 9 files changed, 299 insertions(+), 5 deletions(-) create mode 100644 lps/crud/price_history.py create mode 100644 lps/tests/test_price_history.py diff --git a/lps/common/database/model/models.py b/lps/common/database/model/models.py index b2bcbcc..44b26cf 100644 --- a/lps/common/database/model/models.py +++ b/lps/common/database/model/models.py @@ -68,6 +68,40 @@ class search_negative(MAIN_BASE): created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("now()")) +class price_history(MAIN_BASE): + """상품별 최저가 스냅샷(트리거 기반). 네이버/쿠팡/최종 최저가를 검색 시점마다 적재해 + 시계열 그래프(X=triggered_at, Y=가격, 3개 선)로 본다. 배치 아님 — 조회된 상품만 기록.""" + + @staticmethod + def DBType(): + return DBType.MAIN.value + + __tablename__ = "price_history" + + id = Column(UUID(as_uuid=True), primary_key=True, server_default=text("gen_random_uuid()")) + product_code = Column(String(100), nullable=False) # 상품 식별(조회 키) + job_id = Column(UUID(as_uuid=True), nullable=True) # 검색 잡 연결(추적) + triggered_at = Column(DateTime(timezone=True), nullable=False, server_default=text("now()")) # X축(검색 실행 시각) + outcome = Column(String(20), nullable=False) # found / not_found + matched_count = Column(Integer, nullable=True) # AI 매칭 건수 + + naver_lowest = Column(Integer, nullable=True) # 네이버 최저가(같은 상품) + naver_name = Column(String(300), nullable=True) + naver_url = Column(Text, nullable=True) + coupang_lowest = Column(Integer, nullable=True) # 쿠팡 최저가(같은 상품) + coupang_name = Column(String(300), nullable=True) + coupang_url = Column(Text, nullable=True) + final_lowest = Column(Integer, nullable=True) # 전체 최저가(Y축 핵심) + final_source = Column(String(20), nullable=True) # 최종 최저가 소스 + + created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("now()")) + + __table_args__ = ( + # 특정 상품 시계열 조회 최적화 + Index("ix_price_history_product", "product_code", "triggered_at"), + ) + + class bot_detection(MAIN_BASE): """봇 감지 이력 — '이 IP로 몇 번째 요청에서, 어떤 방식으로 차단됐나'를 축적해 패턴 분석. (예: SELECT avg(ip_request_no) → IP당 평균 몇 요청 만에 감지되는지)""" diff --git a/lps/crud/price_history.py b/lps/crud/price_history.py new file mode 100644 index 0000000..904d147 --- /dev/null +++ b/lps/crud/price_history.py @@ -0,0 +1,63 @@ +"""최저가 스냅샷 CRUD — 트리거 시점마다 기록하고, 상품별 시계열로 조회(그래프).""" + +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", +) + + +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) + VALUES + (:product_code, :job_id, :outcome, :matched_count, + :naver_lowest, :naver_name, :naver_url, + :coupang_lowest, :coupang_name, :coupang_url, + :final_lowest, :final_source) + """) + params = {k: event.get(k) for k in _FIELDS} + 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 + 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) diff --git a/lps/router/v1/lps/protocol.py b/lps/router/v1/lps/protocol.py index cfa0c28..3618e57 100644 --- a/lps/router/v1/lps/protocol.py +++ b/lps/router/v1/lps/protocol.py @@ -45,3 +45,22 @@ class Res_JobStatus(Res_WebPacketProtocol): class Res_QueueStats(Res_WebPacketProtocol): counts: dict[str, int] = Field(default_factory=dict, description="상태별 잡 개수") + + +class PricePoint(BaseModel): + triggered_at: str = Field(description="관측 시각(X축)") + outcome: str + matched_count: Optional[int] = None + naver: Optional[int] = Field(None, description="네이버 최저가") + coupang: Optional[int] = Field(None, description="쿠팡 최저가") + final: Optional[int] = Field(None, description="전체 최저가(Y축)") + final_source: Optional[str] = None + naver_name: Optional[str] = None + naver_url: Optional[str] = None + coupang_name: Optional[str] = None + coupang_url: Optional[str] = None + + +class Res_PriceHistory(Res_WebPacketProtocol): + product_code: Optional[str] = None + points: list[PricePoint] = Field(default_factory=list, description="시각 오름차순 스냅샷(그래프용)") diff --git a/lps/router/v1/lps/search.py b/lps/router/v1/lps/search.py index 22dc1c1..52a6e78 100644 --- a/lps/router/v1/lps/search.py +++ b/lps/router/v1/lps/search.py @@ -2,9 +2,11 @@ from fastapi import APIRouter, Depends +from fastapi import Query + from router.v1.validator.dependencies import RemoveNoneResponse from services.lps_service import LpsService -from router.v1.lps.protocol import Res_JobStatus, Res_QueueStats, Res_Search, Req_Search +from router.v1.lps.protocol import Res_JobStatus, Res_PriceHistory, Res_QueueStats, Res_Search, Req_Search router = APIRouter(prefix="/v1/lps", tags=["LPS"], responses={404: {"description": "Not found"}}) @@ -37,3 +39,13 @@ async def job_status(job_id: str, service: LpsService = Depends()): ) async def queue_stats(service: LpsService = Depends()): return RemoveNoneResponse(await service.stats()) + + +@router.get( + path="/products/{product_code}/history", + response_model=Res_PriceHistory, + summary="최저가 이력(그래프)", + description="상품의 트리거별 최저가 스냅샷(네이버/쿠팡/최종)을 시각 오름차순으로 반환. 가격 시계열 그래프용.", +) +async def price_history(product_code: str, limit: int = Query(100, ge=1, le=1000), service: LpsService = Depends()): + return RemoveNoneResponse(await service.price_history(product_code, limit)) diff --git a/lps/services/lps_service.py b/lps/services/lps_service.py index 7620f66..385b607 100644 --- a/lps/services/lps_service.py +++ b/lps/services/lps_service.py @@ -10,9 +10,12 @@ from fastapi import Depends from common.enums import ErrorType, JobStatus, JobType from crud.job_crud import JobQueue +from crud.price_history import PriceHistory from router.v1.lps.protocol import ( EnqueuedItem, + PricePoint, Res_JobStatus, + Res_PriceHistory, Res_QueueStats, Res_Search, SearchItem, @@ -23,8 +26,9 @@ _PRIORITY = {"new": 1, "single": 2, "negowiz": 3, "batch": 4} class LpsService: - def __init__(self, queue: JobQueue = Depends(JobQueue)): + def __init__(self, queue: JobQueue = Depends(JobQueue), history: PriceHistory = Depends(PriceHistory)): self.queue = queue + self.history = history async def submit_search(self, items: list[SearchItem]) -> Res_Search: res = Res_Search() @@ -65,3 +69,20 @@ class LpsService: res = Res_QueueStats() res.counts = await self.queue.counts() return res + + async def price_history(self, product_code: str, limit: int = 100) -> Res_PriceHistory: + res = Res_PriceHistory(product_code=product_code) + rows = await self.history.list_by_product(product_code, limit) + res.points = [ + PricePoint( + triggered_at=r["triggered_at"].isoformat(timespec="seconds"), + outcome=r["outcome"], + matched_count=r["matched_count"], + naver=r["naver_lowest"], coupang=r["coupang_lowest"], final=r["final_lowest"], + final_source=r["final_source"], + naver_name=r["naver_name"], naver_url=r["naver_url"], + coupang_name=r["coupang_name"], coupang_url=r["coupang_url"], + ) + for r in rows + ] + return res diff --git a/lps/tests/test_lps_api.py b/lps/tests/test_lps_api.py index 0b52cc1..a5e5925 100644 --- a/lps/tests/test_lps_api.py +++ b/lps/tests/test_lps_api.py @@ -51,6 +51,24 @@ async def test_job_status_not_found(client, clean_jobs): assert r2.json()["result"]["desc"] == "LPS_JOB_NOT_FOUND" +async def test_price_history_endpoint(client, db_engine): + from crud.price_history import PriceHistory + async with db_engine.begin() as conn: + await conn.execute(text("TRUNCATE price_history")) + ph = PriceHistory() + await ph.record({"product_code": "GRAPH1", "outcome": "found", "final_lowest": 2500, + "final_source": "naver", "naver_lowest": 2500, "coupang_lowest": 2700, "matched_count": 2}) + + r = await client.get("/v1/lps/products/GRAPH1/history") + assert r.status_code == 200 + body = r.json() + assert body["product_code"] == "GRAPH1" + assert len(body["points"]) == 1 + pt = body["points"][0] + assert pt["final"] == 2500 and pt["naver"] == 2500 and pt["coupang"] == 2700 and pt["final_source"] == "naver" + assert "triggered_at" in pt + + async def test_queue_stats(client, clean_jobs): await client.post("/v1/lps/search", json={"data": [ {"product_code": "A", "product_name": "x"}, diff --git a/lps/tests/test_price_history.py b/lps/tests/test_price_history.py new file mode 100644 index 0000000..00def63 --- /dev/null +++ b/lps/tests/test_price_history.py @@ -0,0 +1,93 @@ +"""최저가 이력 — 소스별 min 스냅샷 계산 + 기록/조회 CRUD + 핸들러 기록 규칙.""" + +import pytest_asyncio +from sqlalchemy import text + +from common.enums import JobType +from crud.price_history import PriceHistory +from services.search.contract import NormalizedProduct +from worker.handlers import _price_snapshot, build_search_handler + + +def _np(source, price, name=None): + return NormalizedProduct(source=source, name=name or f"{source}-{price}", price=price, detail_url=f"http://{source}/{price}") + + +# ── 스냅샷 계산 (순수) ───────────────────────────────────────────── +def test_snapshot_source_lowest_and_final(): + matched = [_np("naver", 3000), _np("naver", 2000), _np("coupang", 2500)] + s = _price_snapshot(matched) + assert s["naver_lowest"] == 2000 and s["coupang_lowest"] == 2500 + assert s["final_lowest"] == 2000 and s["final_source"] == "naver" + assert s["matched_count"] == 3 + + +def test_snapshot_single_source_only(): + s = _price_snapshot([_np("coupang", 1500)]) + assert s["naver_lowest"] is None and s["coupang_lowest"] == 1500 + assert s["final_lowest"] == 1500 and s["final_source"] == "coupang" + + +def test_snapshot_empty_is_all_null(): + s = _price_snapshot([]) + assert s["final_lowest"] is None and s["naver_lowest"] is None and s["matched_count"] == 0 + + +# ── CRUD (실 DB) ─────────────────────────────────────────────────── +@pytest_asyncio.fixture +async def ph(db_engine): + async with db_engine.begin() as conn: + await conn.execute(text("TRUNCATE price_history")) + return PriceHistory() + + +async def test_record_and_list_time_ordered(ph): + await ph.record({"product_code": "P1", "outcome": "found", "final_lowest": 2000, "final_source": "naver", + "naver_lowest": 2000, "coupang_lowest": 2500, "matched_count": 3}) + await ph.record({"product_code": "P1", "outcome": "found", "final_lowest": 1900, "final_source": "coupang", + "naver_lowest": 2100, "coupang_lowest": 1900, "matched_count": 2}) + await ph.record({"product_code": "P2", "outcome": "found", "final_lowest": 999}) # 다른 상품 + + points = await ph.list_by_product("P1") + assert len(points) == 2 # P2 제외 + assert [p["final_lowest"] for p in points] == [2000, 1900] # 시각 오름차순 + assert points[0]["triggered_at"] <= points[1]["triggered_at"] + + +# ── 핸들러 기록 규칙 ─────────────────────────────────────────────── +class _Rec: + def __init__(self): self.events = [] + async def record(self, e): self.events.append(e) + + +class _FakeAdapter: + def __init__(self, source, products): self.source = source; self._p = products + async def search(self, q, limit=40): return self._p + + +class _Neg: + def __init__(self, neg): self._neg = neg + async def is_negative(self, k): return self._neg + async def put(self, *a, **k): pass + + +def _job(**p): + p.setdefault("product_name", "x"); p.setdefault("product_code", "PC1") + return {"job_type": JobType.SEARCH.value, "attempts": 1, "job_id": "j1", "payload": p} + + +async def test_handler_records_found_snapshot(): + rec = _Rec() + adapters = {"naver": _FakeAdapter("naver", [_np("naver", 2000)]), "coupang": _FakeAdapter("coupang", [_np("coupang", 1800)])} + await build_search_handler(adapters, history=rec)(_job()) + assert len(rec.events) == 1 + e = rec.events[0] + assert e["product_code"] == "PC1" and e["outcome"] == "found" + assert e["final_lowest"] == 1800 and e["final_source"] == "coupang" + + +async def test_handler_skips_record_on_negative_cache_hit(): + rec = _Rec() + adapters = {"naver": _FakeAdapter("naver", [_np("naver", 100)])} + await build_search_handler(adapters, neg_cache=_Neg(True), history=rec)(_job()) + assert rec.events == [] # 캐시 히트 → 새 관측 없음 → 미기록 diff --git a/lps/worker/handlers.py b/lps/worker/handlers.py index bc8ccb3..3152b72 100644 --- a/lps/worker/handlers.py +++ b/lps/worker/handlers.py @@ -17,11 +17,28 @@ import asyncio from common.enums import JobType from common.logger import LOG -from services.search.contract import SearchAdapter +from services.search.contract import SearchAdapter, NormalizedProduct from services.search.util import parse_price from services.pipeline.core import apply_filters, rank_result +def _price_snapshot(matched: list[NormalizedProduct]) -> dict: + """매칭 목록에서 소스별 최저가 + 전체 최저가 스냅샷을 만든다(price_history 기록용).""" + def lowest(src): + items = [p for p in matched if p.source == src] + return min(items, key=lambda p: p.price) if items else None + + n, c = lowest("naver"), lowest("coupang") + finals = [x for x in (n, c) if x] + f = min(finals, key=lambda p: p.price) if finals else None + return { + "matched_count": len(matched), + "naver_lowest": n.price if n else None, "naver_name": n.name if n else None, "naver_url": n.detail_url if n else None, + "coupang_lowest": c.price if c else None, "coupang_name": c.name if c else None, "coupang_url": c.detail_url if c else None, + "final_lowest": f.price if f else None, "final_source": f.source if f else None, + } + + def build_search_handler( adapters: dict[str, SearchAdapter], sources: list[str] | None = None, @@ -31,12 +48,23 @@ def build_search_handler( keyword_gen=None, max_rounds: int = 3, neg_cache=None, + history=None, ): """검색 핸들러 생성. judge: SimilarityJudge(같은 상품 판정) / keyword_gen: KeywordGenerator(정밀·광역 재검색어) / - neg_cache: NegativeCache(TTL not_found 캐시). 모두 선택 — 없으면 해당 단계 생략.""" + neg_cache: NegativeCache(TTL not_found 캐시) / history: PriceHistory(최저가 스냅샷). + 모두 선택 — 없으면 해당 단계 생략.""" use = list(sources) if sources else list(adapters.keys()) + async def _record_history(product_code: str, job_id, outcome: str, matched: list): + if history is None: + return + event = {"product_code": product_code, "job_id": job_id, "outcome": outcome, **_price_snapshot(matched)} + try: + await history.record(event) + except Exception as ex: + LOG.e_no_callstack(f"[history] 스냅샷 기록 실패(무시): {ex}") + async def _search_round(query: str): """한 라운드: 모든 소스 동시 검색 → (products, per_source, tech_failed).""" results = await asyncio.gather(*[adapters[s].search(query, limit=limit) for s in use], return_exceptions=True) @@ -99,6 +127,7 @@ def build_search_handler( if candidates: # 찾음 → 조기 종료 result = rank_result(candidates, len(products), stages, top_n) result.update(outcome="found", query=query, round=label, rounds_tried=rounds_done, sources=per_source) + await _record_history(cache_key, job.get("job_id"), "found", candidates) return result if tech_failed: # 0매칭인데 소스가 죽어 있었음 → '없음'이라 단정 불가 → 기술 재시도 @@ -109,6 +138,7 @@ def build_search_handler( await neg_cache.put(cache_key, reason=f"not_found after {rounds_done} rounds") result = rank_result([], 0, last_stages, top_n) result.update(outcome="not_found", query=base_query, rounds_tried=rounds_done, sources=last_sources) + await _record_history(cache_key, job.get("job_id"), "not_found", []) return result return handler diff --git a/lps/worker_main.py b/lps/worker_main.py index 816a15c..16162b2 100644 --- a/lps/worker_main.py +++ b/lps/worker_main.py @@ -13,6 +13,7 @@ from config.server_configs import web_server_config, openai_config from crud.job_crud import JobQueue from crud.negative_cache import NegativeCache from crud.bot_detection import BotDetectionLog +from crud.price_history import PriceHistory from services.search.proxy import DecodoProxy from services.search.coupang.adapter import CoupangAdapter from services.search.naver.adapter import NaverAdapter @@ -42,7 +43,10 @@ async def main(concurrency: int = 1): judge = SimilarityJudge() if has_openai else None keyword_gen = KeywordGenerator() if has_openai else None LOG.i(f"AI(판정+검색어생성): {'ON' if has_openai else 'OFF(키 없음)'}") - handler = build_search_handler(adapters, judge=judge, keyword_gen=keyword_gen, neg_cache=NegativeCache()) + handler = build_search_handler( + adapters, judge=judge, keyword_gen=keyword_gen, + neg_cache=NegativeCache(), history=PriceHistory(), + ) stop = asyncio.Event() listeners: list[JobListener] = []