diff --git a/negodata/backend/common/database/model/models.py b/negodata/backend/common/database/model/models.py index 35d3822..583b3d6 100644 --- a/negodata/backend/common/database/model/models.py +++ b/negodata/backend/common/database/model/models.py @@ -111,6 +111,26 @@ class items(MainTableMixin, MAIN_BASE): delivery_fee_yn = Column(Boolean, nullable=True) +class item_internet_lowest_prices(MainTableMixin, MAIN_BASE): + """인터넷 최저가 수집 이력(시도 단위, 성공/실패 모두 기록). + 확정 대표값은 items.internet_lowest_price 에 박제된다(N건 이력 → 1건 대표 패턴). + LPS 동기화 배치(services/lps_sync_service)가 lps_db.price_history 를 읽어 채운다.""" + + __tablename__ = "item_internet_lowest_prices" + __table_args__ = {"schema": "partner"} + + lp_id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + item_id = Column(UUID(as_uuid=True), nullable=False, index=True) + + lp_price = Column(BigInteger, nullable=True) # 수집한 최저가(원). 실패면 NULL + website = Column(SmallInteger, nullable=False) # LowestPriceWebsite 코드 + success_yn = Column(Boolean, nullable=False) + fail_reason = Column(String(100), nullable=True) # 실패 사유(예: not_found) + ai_model = Column(SmallInteger, nullable=True) # (예약) AI 모델 코드 — LPS 계약엔 미포함 + crawl_duration_ms = Column(Integer, nullable=True) # (예약) 수집 소요 — LPS 계약엔 미포함 + crawl_end_time = Column(DateTime(timezone=True), nullable=False) # 수집 완료 시각(=price_history.created_at, 워터마크 기준) + + class suppliers(MainTableMixin, MAIN_BASE): __tablename__ = "suppliers" __table_args__ = {"schema": "partner"} diff --git a/negodata/backend/common/enums.py b/negodata/backend/common/enums.py index 37d3972..ea7852f 100644 --- a/negodata/backend/common/enums.py +++ b/negodata/backend/common/enums.py @@ -265,6 +265,25 @@ class SupplierType(CodeEnum): SOLE_AGENCY = 3 # 총판 +class LowestPriceWebsite(CodeEnum): + """partner.item_internet_lowest_prices.website — 최저가 수집 사이트 코드. + LPS(lps_db.price_history.final_source)의 소스 문자열을 코드값으로 매핑한다.""" + + NAVER = 1 + COUPANG = 2 + GMARKET = 3 + AUCTION = 4 + ST11 = 5 + ETC = 99 + + @classmethod + def from_source(cls, source) -> "LowestPriceWebsite": + return { + "naver": cls.NAVER, "coupang": cls.COUPANG, + "gmarket": cls.GMARKET, "auction": cls.AUCTION, "st11": cls.ST11, + }.get((source or "").lower(), cls.ETC) + + class CardUsageType(CodeEnum): """nego_cards/wild_cards.usage_type 코드값. 협상카드 사용 범위(신규/재 견적·협상 양쪽 적용). 공통=모두 적용(기본), 신규견적전용, 재견적전용.""" diff --git a/negodata/backend/config/config_models.py b/negodata/backend/config/config_models.py index 4992542..51eee43 100644 --- a/negodata/backend/config/config_models.py +++ b/negodata/backend/config/config_models.py @@ -10,6 +10,7 @@ class WebServerConfig(ConfigModel): client_url: str = "" nego_chat_url: str = "http://localhost:3300" agent_base_url: str = "http://localhost:9500" # 협상 agent(9500). 공용 카탈로그 변경 알림용. + lps_base_url: str = "http://localhost:9600" # 인터넷 최저가 검색 LPS(9600). 검색요청 enqueue 용. class LogConfig(ConfigModel): diff --git a/negodata/backend/config/server_configs.py b/negodata/backend/config/server_configs.py index e96cf7a..4bcf500 100644 --- a/negodata/backend/config/server_configs.py +++ b/negodata/backend/config/server_configs.py @@ -59,3 +59,7 @@ def _apply_lps_db_env_override(cfg: LpsDBConfig): _apply_lps_db_env_override(lps_db_config) + +# LPS API 주소 env override (도커: http://lps-api:9600 또는 host.docker.internal:9600) +if os.environ.get("LPS_BASE_URL"): + web_server_config.lps_base_url = os.environ["LPS_BASE_URL"] diff --git a/negodata/backend/crud/lps_sync_crud.py b/negodata/backend/crud/lps_sync_crud.py new file mode 100644 index 0000000..27fd77c --- /dev/null +++ b/negodata/backend/crud/lps_sync_crud.py @@ -0,0 +1,145 @@ +"""LPS 동기화 CRUD — negosium(main) 쪽 조회/반영 + lps_db(읽기전용) 결과 조회. + +경계 규칙 +- lps_db 는 LPS 서비스 소유라 ORM 모델을 만들지 않는다 — anchoring reader 선례대로 + raw table() 로 **price_history 만** 읽는다(그 외 테이블은 LPS 내부 구현으로 간주). +- 워터마크는 별도 상태 테이블 없이 partner.item_internet_lowest_prices 의 + max(crawl_end_time) 을 쓴다(crawl_end_time = price_history.created_at 을 그대로 보존). + 반영이 중간에 실패해도 다음 tick 이 같은 지점부터 다시 읽는다(자기 치유). +""" +from abc import ABC, abstractmethod +from typing import Tuple + +from sqlalchemy import column, func, or_, select, table, update +from sqlalchemy.ext.asyncio import AsyncSession + +from common.database.db_session_manager import DB_SESSION_MNG +from common.database.model.models import item_internet_lowest_prices, items +from common.enums import ErrorType +from common.logger import LOG + +# lps_db.price_history — LPS 가 검색 1건마다 남기는 append-only 스냅샷(읽기 계약). +_price_history = table( + "price_history", + column("product_code"), # = negosium items.item_id (요청 시 그대로 넣는다) + column("outcome"), # found | not_found + column("final_lowest"), # 전체 최저가(원) + column("final_source"), # naver | coupang | (폴백몰) + column("created_at"), +) + + +class ILpsSyncCRUD(ABC): + @abstractmethod + async def watermark(self, cdb: AsyncSession) -> Tuple[ErrorType, object]: + pass + + @abstractmethod + async def stale_items(self, cdb: AsyncSession, cutoff, limit) -> Tuple[ErrorType, list]: + pass + + @abstractmethod + async def existing_item_ids(self, cdb: AsyncSession, item_ids: list) -> Tuple[ErrorType, set]: + pass + + @abstractmethod + async def fetch_results_since(self, ldb: AsyncSession, since) -> Tuple[ErrorType, list]: + pass + + @abstractmethod + async def add_history_rows(self, cdb: AsyncSession, rows: list) -> ErrorType: + pass + + @abstractmethod + async def update_item_lowest(self, cdb: AsyncSession, item_id, price) -> ErrorType: + pass + + +class LpsSyncCRUD(ILpsSyncCRUD): + # ---- negosium(main) 조회 ------------------------------------------- + async def watermark(self, cdb: AsyncSession): + """수집 이력의 최신 crawl_end_time — 증분 동기화 기준점. 이력이 없으면 None(전체 수집).""" + try: + q = select(func.max(item_internet_lowest_prices.crawl_end_time)).where( + item_internet_lowest_prices.deleted == False # noqa: E712 + ) + return ErrorType.SUCCESS, (await cdb.execute(q)).scalar() + except Exception as ex: + LOG.e_no_callstack(f"[lps-sync] watermark 조회 실패: {ex}") + return ErrorType.DB_RUN_FAILED, None + + async def stale_items(self, cdb: AsyncSession, cutoff, limit: int): + """재검색 대상 상품 — 수집 이력이 없거나 마지막 수집이 cutoff 이전인 활성 상품. + 반환 행: (item_id, name, model_name, spec, manufacturer, price)""" + try: + latest = ( + select( + item_internet_lowest_prices.item_id.label("item_id"), + func.max(item_internet_lowest_prices.crawl_end_time).label("last_ts"), + ) + .where(item_internet_lowest_prices.deleted == False) # noqa: E712 + .group_by(item_internet_lowest_prices.item_id) + .subquery() + ) + q = ( + select(items.item_id, items.name, items.model_name, items.spec, items.manufacturer, items.price) + .join(latest, items.item_id == latest.c.item_id, isouter=True) + .where( + items.deleted == False, # noqa: E712 + or_(latest.c.last_ts.is_(None), latest.c.last_ts < cutoff), + ) + .order_by(latest.c.last_ts.asc().nullsfirst()) # 오래된 것부터(이력 없는 신규 최우선) + .limit(limit) + ) + return ErrorType.SUCCESS, (await cdb.execute(q)).all() + except Exception as ex: + LOG.e_no_callstack(f"[lps-sync] stale_items 조회 실패: {ex}") + return ErrorType.DB_RUN_FAILED, [] + + async def existing_item_ids(self, cdb: AsyncSession, item_ids: list): + """전달된 id 중 실재하는 활성 상품 id 집합 — 결과 반영 전 매핑 검증용.""" + if not item_ids: + return ErrorType.SUCCESS, set() + try: + q = select(items.item_id).where( + items.item_id.in_(item_ids), + items.deleted == False, # noqa: E712 + ) + return ErrorType.SUCCESS, {r[0] for r in (await cdb.execute(q)).all()} + except Exception as ex: + LOG.e_no_callstack(f"[lps-sync] existing_item_ids 조회 실패: {ex}") + return ErrorType.DB_RUN_FAILED, set() + + # ---- lps_db(읽기전용) 조회 ----------------------------------------- + async def fetch_results_since(self, ldb: AsyncSession, since): + """워터마크 이후의 price_history 증분. since 가 None 이면 전체(첫 동기화).""" + try: + q = select( + _price_history.c.product_code, + _price_history.c.outcome, + _price_history.c.final_lowest, + _price_history.c.final_source, + _price_history.c.created_at, + ).order_by(_price_history.c.created_at.asc()) + if since is not None: + q = q.where(_price_history.c.created_at > since) + return ErrorType.SUCCESS, (await ldb.execute(q)).all() + except Exception as ex: + LOG.e_no_callstack(f"[lps-sync] price_history 조회 실패: {ex}") + return ErrorType.DB_RUN_FAILED, [] + + # ---- negosium(main) 반영 (execute_lambda_run 안에서 호출) ----------- + async def add_history_rows(self, cdb: AsyncSession, rows: list) -> ErrorType: + """수집 이력 일괄 insert. rows = item_internet_lowest_prices ORM 객체 리스트.""" + if not rows: + return ErrorType.SUCCESS + return await DB_SESSION_MNG.insert(cdb, rows, err_msg="lps-sync history insert 실패") + + async def update_item_lowest(self, cdb: AsyncSession, item_id, price) -> ErrorType: + """상품 대표 최저가 박제 — items.internet_lowest_price + yn 플래그.""" + q = ( + update(items) + .where(items.item_id == item_id, items.deleted == False) # noqa: E712 + .values(internet_lowest_price=price, internet_lowest_price_yn=True) + ) + return await DB_SESSION_MNG.add(cdb, q, err_msg=f"lps-sync 최저가 반영 실패 item={item_id}") diff --git a/negodata/backend/scheduler/__init__.py b/negodata/backend/scheduler/__init__.py index 3031f0b..ab5e0a0 100644 --- a/negodata/backend/scheduler/__init__.py +++ b/negodata/backend/scheduler/__init__.py @@ -4,6 +4,8 @@ 잡 ① close_expired_quotations : 5분마다(KST) — 마감시각 지난 견적 마감 잡 ② close_negotiated_quotations: 5분마다(KST) — 모든 세션 협상 끝난 견적 즉시 마감(타입 무관) +잡 ③ request_lps_searches : 매일 04:00(KST) — 갱신 오래된 상품을 LPS 에 검색 요청(비용 발생 잡) +잡 ④ sync_lps_results : 5분마다(KST) — lps_db 결과 증분 수집 → 이력 append + 상품 최저가 박제 """ import os @@ -51,8 +53,26 @@ def start_scheduler(): misfire_grace_time=600, max_instances=1, ) + # 잡 ③ LPS 검색 요청(비용 발생) — 새벽 1회. LPS 미설정 환경이면 잡 내부에서 스킵. + _scheduler.add_job( + jobs.request_lps_searches, + CronTrigger(hour=4, minute=0), + id="request_lps_searches", + coalesce=True, + misfire_grace_time=3600, # 재기동 등으로 놓쳐도 1시간 내면 실행 + max_instances=1, + ) + # 잡 ④ LPS 결과 수집·반영 — 증분·멱등이라 잦아도 안전 + _scheduler.add_job( + jobs.sync_lps_results, + CronTrigger(minute="*/5"), + id="sync_lps_results", + coalesce=True, + misfire_grace_time=600, + max_instances=1, + ) _scheduler.start() - LOG.i("[scheduler] started (KST, both every 5min)") + LOG.i("[scheduler] started (KST: 견적마감 2잡 5분 · LPS 요청 04:00 · LPS 수집 5분)") def shutdown_scheduler(): diff --git a/negodata/backend/scheduler/jobs.py b/negodata/backend/scheduler/jobs.py index 2b5d94c..09258f3 100644 --- a/negodata/backend/scheduler/jobs.py +++ b/negodata/backend/scheduler/jobs.py @@ -79,3 +79,41 @@ async def close_negotiated_quotations() -> int: if results: LOG.i(f"[scheduler] close_negotiated: {_format_results(results)}") return sum(results.values()) + + +# ---- LPS(인터넷 최저가) 동기화 ------------------------------------------ + +async def request_lps_searches() -> int: + """[잡③] 갱신이 오래된 상품을 LPS 에 검색 요청(enqueue). 매일 새벽 1회. + 비용이 발생하는 잡(상품당 ~$0.004) — 주기·상한은 lps_sync_service 상수로 관리. + LPS 미설정 환경이면 조용히 스킵(available=False).""" + from services.lps_sync_service import LpsSyncService + + service = LpsSyncService() + if not service.available(): + return 0 + results = await service.request_stale_searches() + if results: + LOG.i( + f"[scheduler] lps_request: 접수 {results['accepted']} / 활성중복 {results['duplicated']} / " + f"이름없음 {results['skipped_no_name']} / HTTP오류 {results['http_error']}" + ) + return results["accepted"] + + +async def sync_lps_results() -> int: + """[잡④] lps_db.price_history 증분을 읽어 수집 이력 append + 상품 대표 최저가 박제. 5분마다. + 워터마크(=이력의 max crawl_end_time) 기준 증분이라 멱등 — 실패 tick 은 다음 tick 이 흡수.""" + from services.lps_sync_service import LpsSyncService + + service = LpsSyncService() + if not service.available(): + return 0 + results = await service.sync_results() + if results: + LOG.i( + f"[scheduler] lps_sync: found {results['found']} / not_found {results['not_found']} / " + f"상품반영 {results['items_updated']} / 비uuid스킵 {results['skipped_not_uuid']} / " + f"미존재상품 {results['skipped_unknown_item']} / 트랜잭션오류 {results['tx_error']}" + ) + return results["items_updated"] diff --git a/negodata/backend/services/lps_sync_service.py b/negodata/backend/services/lps_sync_service.py new file mode 100644 index 0000000..ca61877 --- /dev/null +++ b/negodata/backend/services/lps_sync_service.py @@ -0,0 +1,168 @@ +"""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 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) + + +class LpsSyncService: + def __init__(self, crud: ILpsSyncCRUD = None): + self.crud = crud or 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 + + # ---- ②+③ 수집·반영: price_history 증분 → 이력 append + 대표값 박제 -- + async def sync_results(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