8574df4c16
배송 주체가 다르면(쿠팡 로켓 / 판매자로켓 / 네이버 판매자) 배송비 숫자만으로는 비교가
무의미하다. 그래서 **순위는 지금처럼 상품가**로 두되, 배송 정보는 사후 판단이 가능하도록 남긴다.
실측 조사(섬유유연제·생수 2L, 두 소스):
네이버 배송비무료 / 배송비3,000·3,900·4,500·5,000·8,800·9,000·10,000원 /
내일배송 8.6.(목) 도착 · 오늘출발 · 빠른배송 / **배송비포함 혜택가 N원**(가격비교 카드)
쿠팡 내일(목) 도착 보장 · 와우는 무료배송 ∙ 무료반품 ∙ 새벽도착 / 무료배송 ∙ 오늘출발 /
모레(금) 도착 예정. 뱃지=logo_rocket_filter(로켓)·logo_rocket_merchant(판매자로켓)
- NormalizedProduct.shipping_label: 화면 문구 원문. 같은 '무료배송'이어도 주체·조건
(와우회원·최소금액·새벽도착)이 다른데 숫자·분류로는 그게 사라진다
- 네이버: 배송비 문구 + 도착 정보를 잇는다. **_prices() 안에서 뽑는다** — 가격 노드에서
배송비를 decompose 하기 전에 읽어야 해서(나중에 추가했다가 가격비교 카드에서 라벨이 통째로 빔)
- 쿠팡: 배송 문구가 유틸리티 클래스(fw-text-[14px])에 담겨 셀렉터로 못 집는다 → 텍스트 패턴으로
조각을 모으고, 구분자 없이 붙은 상위 컨테이너("내일(목) 도착무료배송")는 조각 2개 이상을
품은 것으로 판별해 버린다. selectolax 의 node.css("*") 가 자기 자신을 포함해
'자손 매칭' 방식은 못 쓴다(실측)
- price_history.final_shipping_fee/type/label 추가(+마이그레이션). fee 는 0=무료,
NULL=미확인(로켓 조건부) — 둘은 다른 뜻이라 기본값을 두지 않았다
교차 검증(파싱값 vs 카드 원문, 6개 규칙): 네이버 40건·쿠팡 40건 **불일치 0**.
테스트 3건 추가, 전체 231 passed.
95 lines
4.4 KiB
Python
95 lines
4.4 KiB
Python
"""최저가 스냅샷 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",
|
|
)
|
|
|
|
|
|
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)
|
|
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))
|
|
""")
|
|
params = {k: event.get(k) for k in _FIELDS}
|
|
by_mall = event.get("by_mall")
|
|
params["by_mall"] = json.dumps(by_mall) if by_mall 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)
|