Mina Choi
71ef28110b
어느 가격이 "공급사에 지불하는 단가"인지는 회사마다 달라, 컬럼을 합치는 대신 회사가 고르게 했다.
판정은 settings.features.nego_baseline_field 1순위, 미설정 회사는 price 만 숨겼으면 purchase_price 폴백.
판정식 정본은 negodata/backend/common/nego_baseline.py (agent·negosium backend 가 같은 규칙 미러).
- negodata front: 회사 설정에 협상 기준가 라디오(선택지마다 실제 나갈 문장 미리보기 + 학습 경고),
'공급사 포털 안내' 탭 신설(협상 유의사항·헬프데스크 연락처), 용어 카탈로그에 target_price·supplier 추가.
인터넷 최저가는 숨김 대상에서 제외(신규 견적의 유일한 목표가 후보).
상품 등록 기본값에서 개발용 더미 제거(price 1,000,000·PROD-BAT-###·800,000·대한민국·10 EA·14).
- agent: get_item_price → get_item_baseline (기준가·호칭·회사 용어사전을 한 쿼리로),
"기존 공급가 대비" 하드코딩을 회사 용어로 치환 + 받침 기준 조사 자동 보정,
협상 스크립트 4종의 협력사·목표가·배송형태 용어를 {label_*} 토큰화, input_options 도 변수 치환 적용.
- negosium: 협상 화면 기준 단가·배송형태 라벨을 회사 설정 기준으로, 유의사항 본문과 헬프데스크 연락처를
하드코딩에서 회사 설정으로(미등록 시 영역 숨김). 유의사항의 VAT·배송비 문구 삭제(IMK 0803 ⑥).
- negodata backend: LPS 검색 가격 힌트를 items.price 고정에서 기준가 규칙으로.
검증: 기준가 설정 3 × 숨김 4 = 12조합에서 협상 멘트·포털 표시·LPS 힌트가 전부 일치.
agent 테스트 176건 통과. 보고서 negodata/docs/nego-baseline-verification.md
217 lines
12 KiB
Python
217 lines
12 KiB
Python
"""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
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
from sqlalchemy import select
|
|
|
|
from common.database.db_session_manager import DB_SESSION_MNG
|
|
from common.database.model.models import companies, item_internet_lowest_prices
|
|
from common.enums import DBType, DBWRType, ErrorType, LowestPriceWebsite
|
|
from common.logger import LOG
|
|
from common.nego_baseline import resolve_baseline_price
|
|
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)
|
|
|
|
async def _company_settings(self, company_id) -> dict:
|
|
"""상품이 속한 회사의 settings(JSONB). 조회 실패/미설정이면 빈 dict."""
|
|
if not company_id:
|
|
return {}
|
|
|
|
async def _q(s):
|
|
query = select(companies.settings).where(
|
|
companies.company_id == company_id, companies.deleted == False, # noqa: E712
|
|
).limit(1)
|
|
return await DB_SESSION_MNG.execute(s, query, "lps company settings failed.", raise_error=False)
|
|
|
|
err_type, rows = await DB_SESSION_MNG.execute_lambda(companies.DBType(), DBWRType.DB_READ.value, _q)
|
|
if err_type != ErrorType.SUCCESS or not rows:
|
|
return {}
|
|
return rows[0] if isinstance(rows[0], dict) else {}
|
|
|
|
# ---- 단건 즉시 요청 (lowest-price 트리거 API 용) ---------------------
|
|
async def request_search_for_item(self, item, force: bool = False, settings: Optional[dict] = None) -> tuple:
|
|
"""상품 1건을 즉시 LPS 에 검색 요청(수동 트리거 — job_type=manual, 배치보다 높은 우선순위).
|
|
force=True 면 LPS 의 네거티브 캐시(24h not_found)를 무시하고 실제로 재검색한다
|
|
(사용자가 '다시 검색'을 누른 경우. 상품명·모델을 고쳐 재시도하는 흐름에 필요).
|
|
settings 는 회사 설정(companies.settings) — 검색 힌트로 보낼 가격 컬럼을 여기서 정한다.
|
|
반환: (status, message) — queued | duplicated | unavailable."""
|
|
if not self.available():
|
|
return "unavailable", "LPS 연동이 비활성 상태입니다(설정 없음)"
|
|
# 가격 힌트는 이 회사가 관리하는 지불 단가로 보낸다 — 협상 기준가와 같은 규칙.
|
|
# 호출부가 안 넘기면 상품의 소속 회사 설정을 직접 읽는다(실패해도 검색은 진행).
|
|
if settings is None:
|
|
settings = await self._company_settings(item.company_id)
|
|
baseline = resolve_baseline_price(item, settings)
|
|
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(baseline) if baseline else "",
|
|
"force": force,
|
|
}
|
|
base = web_server_config.lps_base_url.rstrip("/")
|
|
# LPS API guard: prod 는 lps_api_key 를 채워 X-API-Key 로 인증(개발은 빈값=개방 모드).
|
|
headers = {"X-API-Key": web_server_config.lps_api_key} if web_server_config.lps_api_key else None
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
r = await client.post(f"{base}/v1/lps/search", json={"data": [payload]}, headers=headers)
|
|
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, nv_name, nv_url, cp_name, cp_url, by_mall, created_at in rows:
|
|
try:
|
|
iid = uuid.UUID(code)
|
|
except (ValueError, AttributeError, TypeError):
|
|
results["skipped_not_uuid"] += 1
|
|
continue
|
|
# 출처(찾은 상품명·링크) — 최종 최저가를 낸 소스의 것을 싣는다(근거 검증용)
|
|
src_name, src_url = {
|
|
"naver": (nv_name, nv_url),
|
|
"coupang": (cp_name, cp_url),
|
|
}.get((final_source or "").lower(), (None, None))
|
|
parsed.append((iid, outcome, final_lowest, final_source, src_name, src_url, by_mall, created_at))
|
|
|
|
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, src_name, src_url, by_mall, 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],
|
|
lp_name=(src_name or None) and src_name[:300],
|
|
lp_url=src_url or None,
|
|
by_mall=by_mall or None, # 몰별 스냅샷 그대로 미러링 — 몰별 성공/실패 표시용
|
|
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
|