28b4a9e797
- POST /v1/item/{id}/lowest-price: 상품 1건 LPS 즉시 검색요청(manual
우선순위). queued/duplicated/unavailable 상태 반환, 실패는
ErrorType.LPS_UNAVAILABLE(2000 블록 신설)
- GET /v1/item/{id}/lowest-price: 대표 최저가 + 최근 수집 이력(최신순,
LowestPriceEntry 타입화). 조회 전 lps_db 증분 동기화 1회 수행 —
5분 크론을 기다리지 않는 실시간 폴링 UX(멱등·저비용)
- 동시성 방어 2겹: 프로세스 내 asyncio.Lock(크론·온디맨드 직렬화) +
uq_iilp_item_crawl_time 유니크 인덱스(alters/2026-07-10, init.sql
멱등 반영·dev DB 적용) — 경합 진 쪽 tx 실패 후 다음 tick 흡수
- LpsSyncService 무인자 생성자(FastAPI Depends 호환)
- compose: negodata-backend 에 LPS_DB_HOST·LPS_BASE_URL env(미설정 시
연동 비활성으로 조용히 동작)
검증(도커 컨테이너 e2e): 인증→POST queued→LPS 크롤→GET 온디맨드
동기화로 이력 즉시 노출(not_found 정책: 대표값 미변경 확인),
company 스코프 차단(타사 상품 ITEM_NOT_FOUND) 확인
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
225 lines
12 KiB
Python
225 lines
12 KiB
Python
"""LPS(인터넷 최저가 검색) 동기화 서비스 — 요청·수집·반영의 도메인 로직.
|
||
|
||
흐름 (요청은 API, 결과는 DB — 2026-07-10 설계 결정)
|
||
① 요청: 갱신이 오래된 상품을 골라 LPS API(POST /v1/lps/search)로 enqueue.
|
||
product_code = items.item_id(uuid 문자열) — 이걸로 결과가 자동 매핑된다.
|
||
enqueue 를 DB insert 로 하지 않는 이유: LPS 의 활성중복 dedupe·pg_notify 워커 깨움을 우회하게 됨.
|
||
② 수집: lps_db.price_history(읽기전용)를 워터마크(max crawl_end_time) 증분으로 읽는다.
|
||
③ 반영: 이력은 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
|
||
|
||
# 재검색 주기·요청 배치 크기 기본값. 상품당 실측 비용 ~$0.004 이므로
|
||
# (상품 수 × 24h 주기)로 월 비용이 바로 계산된다. 필요 시 여기만 조정.
|
||
REFRESH_HOURS = 24 # 마지막 수집이 이보다 오래된 상품만 재요청
|
||
REQUEST_BATCH_LIMIT = 500 # 요청 잡 1회가 enqueue 하는 최대 상품 수(비용 상한)
|
||
ENQUEUE_CHUNK = 100 # LPS API 1콜에 담는 상품 수
|
||
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)
|
||
|
||
# ---- ① 요청: 오래된 상품을 LPS 에 검색 enqueue ----------------------
|
||
async def request_stale_searches(self) -> Counter:
|
||
results = Counter()
|
||
if not self.available():
|
||
return results
|
||
|
||
cutoff = _utc_now_aware() - timedelta(hours=REFRESH_HOURS) # timestamptz 비교 — aware 필수(위 sync_results 주석)
|
||
|
||
err, rows = await DB_SESSION_MNG.execute_lambda(
|
||
DBType.MAIN.value, DBWRType.DB_READ.value,
|
||
lambda s: self.crud.stale_items(s, cutoff, REQUEST_BATCH_LIMIT),
|
||
)
|
||
if err != ErrorType.SUCCESS or not rows:
|
||
return results
|
||
|
||
payload_items = []
|
||
for item_id, name, model_name, spec, manufacturer, price in rows:
|
||
if not (name or "").strip():
|
||
results["skipped_no_name"] += 1
|
||
continue
|
||
payload_items.append({
|
||
"product_code": str(item_id),
|
||
"product_name": name,
|
||
"job_type": "batch",
|
||
"model": model_name or "",
|
||
"specification": spec or "",
|
||
"company": manufacturer or "",
|
||
"price": str(price) if price else "",
|
||
})
|
||
|
||
base = web_server_config.lps_base_url.rstrip("/")
|
||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||
for i in range(0, len(payload_items), ENQUEUE_CHUNK):
|
||
chunk = payload_items[i:i + ENQUEUE_CHUNK]
|
||
try:
|
||
r = await client.post(f"{base}/v1/lps/search", json={"data": chunk})
|
||
r.raise_for_status()
|
||
body = r.json()
|
||
results["accepted"] += int(body.get("accepted", 0))
|
||
results["duplicated"] += sum(1 for it in body.get("items", []) if it.get("duplicated"))
|
||
except Exception as ex:
|
||
# LPS 다운은 이 배치의 실패일 뿐 — 다음 tick 이 같은 대상(여전히 stale)을 다시 요청한다.
|
||
results["http_error"] += 1
|
||
LOG.w(f"[lps-sync] 검색요청 실패(청크 {i // ENQUEUE_CHUNK}): {type(ex).__name__}: {ex}")
|
||
return results
|
||
|
||
# ---- 단건 즉시 요청 (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
|