Merge branch 'feature/negodata-lps' — negodata↔LPS 인터넷 최저가 연동(수동 트리거 전용)

docker-compose.yml

@@ -38,6 +38,9 @@ services:
       RELOAD: "1"                      # uvicorn --reload 활성 → 소스 저장 시 자동 재기동(재빌드 불필요)
       SCHEDULER_ENABLED: "1"           # 마감 크론 활성(단일 워커라 중복 없음). 운영 다중 워커면 1개 프로세스에서만 1
       PYTHONUNBUFFERED: "1"            # 컨테이너 로그 실시간 출력(stdout 버퍼링 끔)
+      # ── LPS(인터넷 최저가) 연동 — 미설정이면 연동 비활성으로 조용히 동작 ──
+      LPS_DB_HOST: host.docker.internal          # lps_db 읽기전용(수집 배치·조회 API)
+      LPS_BASE_URL: http://host.docker.internal:9600   # 검색요청 enqueue. lps-api 컨테이너 사용 시 http://lps-api:9600
     volumes:
       - ./negodata/backend:/app       # 호스트 소스 = 컨테이너 코드. 이게 있어야 수정이 즉시 반영됨
     ports:

negodata/backend/common/database/db_session_manager.py

@@ -9,7 +9,7 @@ from common.database.model.models import MAIN_BASE
 from common.enums import DBType, DBWRType, ErrorType
 from common.logger import LOG
 from common.singleton import Singleton
-from config.server_configs import main_db_config
+from config.server_configs import main_db_config, lps_db_config
 
 
 class DBSessionManager(Singleton):
@@ -47,6 +47,12 @@ class DBSessionManager(Singleton):
             DBType.MAIN.value: self.create_engine(DBType.MAIN.value, DBWRType.DB_READ.value),
         }
 
+        # LPS 결과 DB(lps_db) — **읽기전용**: read 엔진만 등록한다(write 로 열면 KeyError = 앱 차원 가드).
+        # config 의 name 이 비어 있으면 미사용(스키마 없는 환경에서도 부팅 가능, 배치는 is_registered 로 스킵).
+        if lps_db_config.name:
+            self.__db_type_map[DBType.LPS.value] = lps_db_config
+            self.__read_session[DBType.LPS.value] = self.create_engine(DBType.LPS.value, DBWRType.DB_READ.value)
+
     def create_engine(self, db_type: int, db_wr_type: int):
         db_config = self.__db_type_map.get(db_type)
         if not db_config:
@@ -83,6 +89,10 @@ class DBSessionManager(Singleton):
         )
         return scoped_session
 
+    def is_registered(self, db_type: int) -> bool:
+        """해당 논리 DB 가 등록돼 있는지 — 선택 연동(LPS 등)의 배치가 실행 전 확인하는 용도."""
+        return db_type in self.__db_type_map
+
     async def dispose_all(self):
         """모든 엔진의 커넥션 풀을 정리한다. 앱 종료/테스트 종료 시 호출한다.
         호출하지 않으면 풀 커넥션이 이벤트 루프 종료 후 GC 되며 경고를 남긴다.

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"}

negodata/backend/common/enums.py

@@ -82,6 +82,10 @@ class ErrorType(Enum):
     EMAIL_NOT_CONFIGURED = 1900  # ACS/SMTP 둘 다 미설정 — 발송 불가(설정 필요)
     EMAIL_SEND_FAILED = auto()   # 발송 시도했으나 전부 실패(수신자 0 성공)
 
+    # LPS(인터넷 최저가 검색) 연동 관련 에러
+    LPS_UNAVAILABLE = 2000       # lps_db 미등록(연동 비활성 환경) — 기능 사용 불가
+    LPS_REQUEST_FAILED = auto()  # LPS 검색요청 API 호출 실패(LPS 다운/네트워크)
+
 
 # ErrorType 의 HTTP_* 값과 status_code 를 맞춰 router 단에서 raise 한다.
 EXCEPTION_FORBIDDEN = HTTPException(status_code=ErrorType.HTTP_FORBIDDEN.value, detail=ErrorType.HTTP_FORBIDDEN.name)
@@ -99,6 +103,7 @@ class DBType(Enum):
     """
 
     MAIN = 1
+    LPS = 2   # 인터넷 최저가 검색(lps_db) — 읽기전용(price_history 동기화 배치용, write 엔진 미등록)
 
 
 class DBWRType(Enum):
@@ -264,6 +269,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 코드값. 협상카드 사용 범위(신규/재 견적·협상 양쪽 적용).
     공통=모두 적용(기본), 신규견적전용, 재견적전용."""

negodata/backend/config/config.local.toml.example

@@ -32,6 +32,20 @@ pool_size = 10
 max_overflow = 20
 sslmode = ""              # 로컬: "" / 관리형 DB: "require"|"verify-ca"|"verify-full"
 
+# 인터넷 최저가 검색(LPS) 결과 DB — 읽기전용(price_history 동기화 배치).
+# name 을 비우면 LPS 연동 비활성(엔진 미등록·배치 스킵). 도커는 LPS_DB_HOST 등 env 로 override.
+[LpsDBConfig]
+db_type = "postgresql"
+name = "lps_db"
+read_host = "127.0.0.1"
+read_port = 5432
+read_id = "postgres"
+read_pw = "postgres"
+show_log = false
+pool_size = 2
+max_overflow = 2
+sslmode = ""
+
 [JwtToken]
 access_key = "<JWT_ACCESS_SECRET>"
 refresh_key = "<JWT_REFRESH_SECRET>"

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):
@@ -39,6 +40,23 @@ class MainDBConfig(ConfigModel):
     sslmode: str = ""
 
 
+# 인터넷 최저가 검색(LPS)의 결과 DB(lps_db) — **읽기전용** 접속.
+# LPS 는 별도 서비스(자체 스키마 소유)이고, negodata 는 price_history 를 주기 배치로 읽어만 온다.
+# 그래서 read_* 만 둔다(write 엔진 미등록 = 앱 차원 읽기전용 강제). name 이 비면 미사용(등록 스킵).
+class LpsDBConfig(ConfigModel):
+    db_type: str = "postgresql"
+    name: str = ""            # 비우면 LPS 연동 비활성(엔진 미등록, 배치 스킵)
+    read_host: str = ""
+    read_port: int = 5432
+    read_id: str = ""
+    read_pw: str = ""
+    show_log: bool = False
+    # 배치 전용이라 작은 풀이면 충분 (동시 사용처 = 스케줄러 잡 1개)
+    pool_size: int = 2
+    max_overflow: int = 2
+    sslmode: str = ""
+
+
 class JwtToken(ConfigModel):
     access_key: str = ""
     refresh_key: str = ""

negodata/backend/config/server_configs.py

@@ -1,7 +1,7 @@
 import os
 
 from config.config_loader import Configs
-from config.config_models import WebServerConfig, LogConfig, MainDBConfig, JwtToken, StorageConfig, MailConfig
+from config.config_models import WebServerConfig, LogConfig, MainDBConfig, LpsDBConfig, JwtToken, StorageConfig, MailConfig
 
 # 실행 환경 결정 (기본 local). 환경변수 APP_ENV 로 변경.
 APP_ENV = os.environ.get("APP_ENV", "local")
@@ -18,6 +18,8 @@ configs = Configs(_config_file)
 web_server_config: WebServerConfig = configs.get(WebServerConfig)
 log_config: LogConfig = configs.get(LogConfig)
 main_db_config: MainDBConfig = configs.get(MainDBConfig)
+# [LpsDBConfig] 섹션이 없는 toml(구버전)에서도 죽지 않게 기본값 폴백(name 빈 값 → LPS 연동 비활성).
+lps_db_config: LpsDBConfig = configs.get(LpsDBConfig) or LpsDBConfig()
 jwt_token_config: JwtToken = configs.get(JwtToken)
 storage_config: StorageConfig = configs.get(StorageConfig)
 # [MailConfig] 섹션이 없는 toml(구버전)에서도 죽지 않도록 기본값으로 폴백(전 필드 빈 값 → 발송 시 EmailUnavailable).
@@ -40,3 +42,24 @@ def _apply_db_env_override(cfg: MainDBConfig):
 
 
 _apply_db_env_override(main_db_config)
+
+
+# LPS 결과 DB(읽기전용) env override — 도커에서 host 등만 교체. 로컬은 toml 그대로.
+def _apply_lps_db_env_override(cfg: LpsDBConfig):
+    if os.environ.get("LPS_DB_HOST"):
+        cfg.read_host = os.environ["LPS_DB_HOST"]
+    if os.environ.get("LPS_DB_PORT"):
+        cfg.read_port = int(os.environ["LPS_DB_PORT"])
+    if os.environ.get("LPS_DB_USER"):
+        cfg.read_id = os.environ["LPS_DB_USER"]
+    if os.environ.get("LPS_DB_PASSWORD"):
+        cfg.read_pw = os.environ["LPS_DB_PASSWORD"]
+    if os.environ.get("LPS_DB_NAME"):
+        cfg.name = os.environ["LPS_DB_NAME"]
+
+
+_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"]

negodata/backend/crud/lps_sync_crud.py

@@ -0,0 +1,134 @@
+"""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, 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 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 recent_history(self, cdb: AsyncSession, item_id, limit) -> 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 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()
+
+    async def recent_history(self, cdb: AsyncSession, item_id, limit: int):
+        """상품의 최근 수집 이력(최신순) — lowest-price 조회 API 용. idx_iilp_item_crawl_time 사용."""
+        try:
+            q = (
+                select(item_internet_lowest_prices)
+                .where(
+                    item_internet_lowest_prices.item_id == item_id,
+                    item_internet_lowest_prices.deleted == False,  # noqa: E712
+                )
+                .order_by(item_internet_lowest_prices.crawl_end_time.desc())
+                .limit(limit)
+            )
+            return ErrorType.SUCCESS, (await cdb.execute(q)).scalars().all()
+        except Exception as ex:
+            LOG.e_no_callstack(f"[lps-sync] recent_history 조회 실패: {ex}")
+            return ErrorType.DB_RUN_FAILED, []
+
+    # ---- 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}")

negodata/backend/router/v1/item/item.py

@@ -2,10 +2,13 @@ from uuid import UUID
 
 from fastapi import APIRouter, Depends, File, Query, UploadFile
 
+from common.enums import ErrorType
 from common.models.gmodel import PageParams, UserInfo
 from router.v1.validator.dependencies import IsValidAccessToken, RemoveNoneResponse
 from services.item_service import ItemService
+from services.lps_sync_service import LpsSyncService
 from .protocol import (
+    LowestPriceEntry,
     Req_CheckCodes,
     Req_CreateItem,
     Req_UpdateItem,
@@ -77,18 +80,43 @@ async def delete_item(item_id: UUID, service: ItemService = Depends(), user_info
     return RemoveNoneResponse(await service.delete_item(user_info.company_id, str(item_id), user_info.user_id, user_info.role))
 
 
-@router.post(path="/{item_id}/lowest-price", response_model=Res_LowestPriceTrigger, summary="최저가 수집 요청(스텁)")
+@router.post(path="/{item_id}/lowest-price", response_model=Res_LowestPriceTrigger, summary="최저가 수집 요청")
-async def trigger_lowest_price(item_id: UUID, service: ItemService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
+async def trigger_lowest_price(
-    # 존재/소유 확인만 (없으면 result 에 ITEM_NOT_FOUND)
+    item_id: UUID,
+    service: ItemService = Depends(),
+    lps: LpsSyncService = Depends(),
+    user_info: UserInfo = Depends(IsValidAccessToken),
+):
+    """상품 1건을 LPS 에 즉시 검색 요청(수동 트리거, manual 우선순위). 결과는 GET lowest-price 폴링."""
+    got = await service.get_item(user_info.company_id, str(item_id))   # 존재/소유(company 스코프) 확인
+    if got.item is None:
+        return RemoveNoneResponse(got)
+    status, message = await lps.request_search_for_item(got.item)
+    res = Res_LowestPriceTrigger(item_id=str(item_id), status=status, message=message)
+    if status == "unavailable":
+        res.result.SetResult(ErrorType.LPS_UNAVAILABLE)
+    return RemoveNoneResponse(res)
+
+
+@router.get(path="/{item_id}/lowest-price", response_model=Res_LowestPriceResult, summary="최저가 수집 결과")
+async def get_lowest_price(
+    item_id: UUID,
+    service: ItemService = Depends(),
+    lps: LpsSyncService = Depends(),
+    user_info: UserInfo = Depends(IsValidAccessToken),
+):
+    """대표 최저가(items.internet_lowest_price) + 최근 수집 이력(최신순).
+    조회 전에 lps_db 증분 동기화를 한 번 수행해 5분 크론을 기다리지 않는다(멱등·저비용)."""
     got = await service.get_item(user_info.company_id, str(item_id))
     if got.item is None:
         return RemoveNoneResponse(got)
-    return RemoveNoneResponse(Res_LowestPriceTrigger(item_id=str(item_id), status="queued", message="최저가 수집 요청됨(스텁)"))
+    err, history = await lps.lowest_price_view(item_id)
-
+    res = Res_LowestPriceResult(
-
+        item_id=str(item_id),
-@router.get(path="/{item_id}/lowest-price", response_model=Res_LowestPriceResult, summary="최저가 수집 결과(스텁)")
+        lowest_price=got.item.internet_lowest_price,
-async def get_lowest_price(item_id: UUID, service: ItemService = Depends(), user_info: UserInfo = Depends(IsValidAccessToken)):
+        results=[LowestPriceEntry.model_validate(h) for h in history],
-    got = await service.get_item(user_info.company_id, str(item_id))
+        message="" if history else "수집 이력이 없습니다",
-    if got.item is None:
+    )
-        return RemoveNoneResponse(got)
+    if err != ErrorType.SUCCESS:
-    return RemoveNoneResponse(Res_LowestPriceResult(item_id=str(item_id), results=[], message="최저가 수집 결과 없음(스텁)"))
+        res.result.SetResult(err)
+    return RemoveNoneResponse(res)

negodata/backend/router/v1/item/protocol.py

@@ -126,11 +126,24 @@ class Res_ItemImage(Res_WebPacketProtocol):
 
 class Res_LowestPriceTrigger(Res_WebPacketProtocol):
     item_id: str = ""
-    status: str = ""
+    status: str = ""      # queued(접수) | duplicated(이미 진행 중) | unavailable(연동 비활성/LPS 다운)
     message: str = ""
 
 
+class LowestPriceEntry(WebPacketProtocol):
+    """최저가 수집 이력 1건(partner.item_internet_lowest_prices)."""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    lp_price: Optional[int] = None      # 수집한 최저가(원). 실패면 None
+    website: int = 0                    # LowestPriceWebsite 코드(1=naver 2=coupang …)
+    success_yn: bool = False
+    fail_reason: Optional[str] = None
+    crawl_end_time: Optional[datetime] = None
+
+
 class Res_LowestPriceResult(Res_WebPacketProtocol):
     item_id: str = ""
-    results: list = []
+    lowest_price: Optional[int] = None    # 상품에 박제된 대표 최저가(items.internet_lowest_price)
+    results: list[LowestPriceEntry] = []  # 최근 수집 이력(최신순)
     message: str = ""

negodata/backend/scheduler/__init__.py

@@ -4,6 +4,11 @@
 
 잡 ① close_expired_quotations  : 5분마다(KST) — 마감시각 지난 견적 마감
 잡 ② close_negotiated_quotations: 5분마다(KST) — 모든 세션 협상 끝난 견적 즉시 마감(타입 무관)
+잡 ③ sync_lps_results          : 5분마다(KST) — lps_db 결과 증분 수집 → 이력 append + 상품 최저가 박제
+                                  (수동 트리거된 검색의 반영 백스톱 — 크롤을 일으키지 않음, 비용 0)
+
+LPS 검색 **요청**은 배치로 돌리지 않는다(2026-07-10 협의) — 상품 화면의 수동 트리거
+(POST /v1/item/{id}/lowest-price)로만 검색한다. 크롤 비용이 사용자 행동에만 비례하게.
 """
 import os
 
@@ -51,8 +56,17 @@ def start_scheduler():
         misfire_grace_time=600,
         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 수집 5분 — LPS 요청 배치 없음, 수동 트리거 전용)")
 
 
 def shutdown_scheduler():

negodata/backend/scheduler/jobs.py

@@ -79,3 +79,24 @@ async def close_negotiated_quotations() -> int:
     if results:
         LOG.i(f"[scheduler] close_negotiated: {_format_results(results)}")
     return sum(results.values())
+
+
+# ---- LPS(인터넷 최저가) 동기화 ------------------------------------------
+
+async def sync_lps_results() -> int:
+    """[잡③] lps_db.price_history 증분을 읽어 수집 이력 append + 상품 대표 최저가 박제. 5분마다.
+    검색 **요청**은 하지 않는다(수동 트리거 전용, 2026-07-10 협의) — 이 잡은 반영 백스톱.
+    워터마크(=이력의 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"]

negodata/backend/services/lps_sync_service.py

@@ -0,0 +1,176 @@
+"""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
+
+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
+
+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)
+
+    # ---- 단건 즉시 요청 (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

negodata/front/src/api/generated/item/item.ts

@@ -644,7 +644,8 @@ export const useDeleteItem = <TError = void | HTTPValidationError,
       return useMutation(mutationOptions, queryClient);
     }
     /**
- * @summary 최저가 수집 요청(스텁)
+ * 상품 1건을 LPS 에 즉시 검색 요청(수동 트리거, manual 우선순위). 결과는 GET lowest-price 폴링.
+ * @summary 최저가 수집 요청
  */
 export const triggerLowestPrice = (
     itemId: string,
@@ -690,7 +691,7 @@ const {mutation: mutationOptions, request: requestOptions} = options ?
     export type TriggerLowestPriceMutationError = void | HTTPValidationError
 
     /**
- * @summary 최저가 수집 요청(스텁)
+ * @summary 최저가 수집 요청
  */
 export const useTriggerLowestPrice = <TError = void | HTTPValidationError,
     TContext = unknown>(options?: { mutation?:UseMutationOptions<Awaited<ReturnType<typeof triggerLowestPrice>>, TError,{itemId: string}, TContext>, request?: SecondParameter<typeof customFetch>}
@@ -706,7 +707,9 @@ export const useTriggerLowestPrice = <TError = void | HTTPValidationError,
       return useMutation(mutationOptions, queryClient);
     }
     /**
- * @summary 최저가 수집 결과(스텁)
+ * 대표 최저가(items.internet_lowest_price) + 최근 수집 이력(최신순).
+조회 전에 lps_db 증분 동기화를 한 번 수행해 5분 크론을 기다리지 않는다(멱등·저비용).
+ * @summary 최저가 수집 결과
  */
 export const getLowestPrice = (
     itemId: string,
@@ -777,7 +780,7 @@ export function useGetLowestPrice<TData = Awaited<ReturnType<typeof getLowestPri
  , queryClient?: QueryClient
   ):  UseQueryResult<TData, TError> & { queryKey: DataTag<QueryKey, TData, TError> }
 /**
- * @summary 최저가 수집 결과(스텁)
+ * @summary 최저가 수집 결과
  */
 
 export function useGetLowestPrice<TData = Awaited<ReturnType<typeof getLowestPrice>>, TError = void | HTTPValidationError>(

negodata/front/src/api/generated/model/index.ts

@@ -78,6 +78,10 @@ export * from './listNotificationsParams';
 export * from './listQuotationsParams';
 export * from './listSuppliersParams';
 export * from './listUsersParams';
+export * from './lowestPriceEntry';
+export * from './lowestPriceEntryCrawlEndTime';
+export * from './lowestPriceEntryFailReason';
+export * from './lowestPriceEntryLpPrice';
 export * from './notificationData';
 export * from './notificationDataCreatedAt';
 export * from './notificationDataData';
@@ -278,6 +282,7 @@ export * from './resItemSupplyTypeListMsg';
 export * from './resLogin';
 export * from './resLoginMsg';
 export * from './resLowestPriceResult';
+export * from './resLowestPriceResultLowestPrice';
 export * from './resLowestPriceResultMsg';
 export * from './resLowestPriceTrigger';
 export * from './resLowestPriceTriggerMsg';

negodata/front/src/api/generated/model/lowestPriceEntry.ts

@@ -0,0 +1,20 @@
+/**
+ * Generated by orval v7.21.0 🍺
+ * Do not edit manually.
+ * Negodata Api Server
+ * OpenAPI spec version: 0.1.0
+ */
+import type { LowestPriceEntryLpPrice } from './lowestPriceEntryLpPrice';
+import type { LowestPriceEntryFailReason } from './lowestPriceEntryFailReason';
+import type { LowestPriceEntryCrawlEndTime } from './lowestPriceEntryCrawlEndTime';
+
+/**
+ * 최저가 수집 이력 1건(partner.item_internet_lowest_prices).
+ */
+export interface LowestPriceEntry {
+  lp_price?: LowestPriceEntryLpPrice;
+  website?: number;
+  success_yn?: boolean;
+  fail_reason?: LowestPriceEntryFailReason;
+  crawl_end_time?: LowestPriceEntryCrawlEndTime;
+}

negodata/front/src/api/generated/model/lowestPriceEntryCrawlEndTime.ts

@@ -0,0 +1,8 @@
+/**
+ * Generated by orval v7.21.0 🍺
+ * Do not edit manually.
+ * Negodata Api Server
+ * OpenAPI spec version: 0.1.0
+ */
+
+export type LowestPriceEntryCrawlEndTime = string | null;

negodata/front/src/api/generated/model/lowestPriceEntryFailReason.ts

@@ -0,0 +1,8 @@
+/**
+ * Generated by orval v7.21.0 🍺
+ * Do not edit manually.
+ * Negodata Api Server
+ * OpenAPI spec version: 0.1.0
+ */
+
+export type LowestPriceEntryFailReason = string | null;

negodata/front/src/api/generated/model/lowestPriceEntryLpPrice.ts

@@ -0,0 +1,8 @@
+/**
+ * Generated by orval v7.21.0 🍺
+ * Do not edit manually.
+ * Negodata Api Server
+ * OpenAPI spec version: 0.1.0
+ */
+
+export type LowestPriceEntryLpPrice = number | null;

negodata/front/src/api/generated/model/resLowestPriceResult.ts

@@ -6,11 +6,14 @@
  */
 import type { ErrorInfo } from './errorInfo';
 import type { ResLowestPriceResultMsg } from './resLowestPriceResultMsg';
+import type { ResLowestPriceResultLowestPrice } from './resLowestPriceResultLowestPrice';
+import type { LowestPriceEntry } from './lowestPriceEntry';
 
 export interface ResLowestPriceResult {
   result?: ErrorInfo;
   msg?: ResLowestPriceResultMsg;
   item_id?: string;
-  results?: unknown[];
+  lowest_price?: ResLowestPriceResultLowestPrice;
+  results?: LowestPriceEntry[];
   message?: string;
 }

negodata/front/src/api/generated/model/resLowestPriceResultLowestPrice.ts

@@ -0,0 +1,8 @@
+/**
+ * Generated by orval v7.21.0 🍺
+ * Do not edit manually.
+ * Negodata Api Server
+ * OpenAPI spec version: 0.1.0
+ */
+
+export type ResLowestPriceResultLowestPrice = number | null;

negodata/front/src/features/products/components/PriceUpdateModal.tsx

@@ -1,10 +1,12 @@
-import { useState } from 'react';
+import { useRef, useState } from 'react';
+import { useQueryClient } from '@tanstack/react-query';
 import { toast } from 'sonner';
 import { Globe, X, AlertCircle, Loader2, Cpu, RefreshCw } from 'lucide-react';
 import { showToast } from '@/lib/notify';
 import { Typography } from '@/components/ui/typography';
 import { Button } from '@/components/ui/button';
 import { useScrollLock } from '@/lib/useScrollLock';
+import { triggerLowestPrice, getLowestPrice } from '@/api/generated/item/item';
 import type { Product } from '../types';
 
 type PriceUpdateModalProps = {
@@ -15,62 +17,113 @@ type PriceUpdateModalProps = {
   onClose: () => void;
 };
 
-// 인터넷 최저가 실시간 수집 데모 모달. 크롤링 진행 state는 이 컴포넌트가 소유한다.
+const POLL_INTERVAL_MS = 5_000; // GET lowest-price 폴링 간격(서버가 조회 시 lps_db 증분 동기화를 겸함)
-// NOTE: 서버 미연동 — 진행 애니메이션/로그만 데모.
+const POLL_TIMEOUT_MS = 300_000; // 상품당 수십 초 × 순차 처리 감안한 전체 상한(5분)
+
+// 인터넷 최저가 실시간 수집 모달 — LPS 연동.
+// 흐름: 선택 상품마다 POST(수집 요청, 큐 접수) → GET 폴링(요청 시각 이후의 수집 이력이 생기면 완료).
+// 폴링이 시간을 초과해도 서버 검색은 계속되고, 5분 주기 동기화 배치가 결과를 자동 반영한다.
 export function PriceUpdateModal({ open, products, selectedIds, onDone, onClose }: PriceUpdateModalProps) {
   useScrollLock(open); // 모달 열린 동안 배경(부모) 스크롤 잠금
+  const queryClient = useQueryClient();
   const [isCrawling, setIsCrawling] = useState(false);
   const [crawlingProgress, setCrawlingProgress] = useState(0);
   const [crawlerLogs, setCrawlerLogs] = useState<string[]>([]);
+  const cancelledRef = useRef(false); // 닫기 시 폴링 루프 중단(서버 검색은 계속)
 
   if (!open) return null;
 
-  const handleStartCrawling = () => {
+  const pushLog = (line: string) => setCrawlerLogs((prev) => [...prev, line]);
+  const nameOf = (id: string) => products.find((p) => p.item_id === id)?.name || id;
+
+  const handleStartCrawling = async () => {
     if (selectedIds.length === 0) {
       toast.error('업데이트할 상품을 1개 이상 선택해 주십시오.');
       return;
     }
 
+    cancelledRef.current = false;
     setIsCrawling(true);
-    setCrawlingProgress(10);
+    setCrawlingProgress(5);
     setCrawlerLogs([
-      '[System] 실시간 최저가 수집용 웹 크롤러 엔진 가동...',
+      '[System] 인터넷 최저가 검색(LPS) 요청 접수 중...',
-      '[System] API Endpoint: https://api.commerce-crawler.co.kr/v2/itemsync',
+      `[Target] 선택된 ${selectedIds.length}개 상품`,
-      `[Target] 선택된 ${selectedIds.length}개 상품의 고유 코드 및 품목 매핑 중...`,
     ]);
+    const startedAt = new Date().toISOString(); // 이 시각 이후의 수집 이력만 "이번 요청 결과"로 인정
 
-    // Fast simulation timers
+    // 1) 상품별 수집 요청(POST) — 실패/중복은 로그로 구분하고 계속 진행
-    setTimeout(() => {
+    const pending = new Set<string>();
-      setCrawlingProgress(35);
+    for (const id of selectedIds) {
-      setCrawlerLogs((prev) => [
+      try {
-        ...prev,
+        const r = await triggerLowestPrice(id);
-        ...selectedIds.map((id) => {
+        if (r.status === 'queued') {
-          const p = products.find((prod) => prod.item_id === id);
+          pending.add(id);
-          return `[크롤링] '${p?.name || id}' 인터넷 최저가 비교 검색 수집 진행`;
+          pushLog(`[접수] '${nameOf(id)}' 검색 큐 등록`);
-        }),
+        } else if (r.status === 'duplicated') {
-        `[Search] 외부 커머스 유통 플랫폼(Coupang, Gmarket, Danawa) 지표 추출 시작...`,
+          pending.add(id); // 이미 진행 중 → 결과는 폴링으로 같이 받는다
-      ]);
+          pushLog(`[진행중] '${nameOf(id)}' 이미 검색이 진행 중 — 결과 대기에 합류`);
-    }, 500);
+        } else {
+          pushLog(`[불가] '${nameOf(id)}' ${r.message || '검색 서비스 연결 불가'}`);
+        }
+      } catch {
+        pushLog(`[오류] '${nameOf(id)}' 요청 실패`);
+      }
+    }
 
-    setTimeout(() => {
+    if (pending.size === 0) {
-      setCrawlingProgress(72);
+      showToast('접수된 상품이 없습니다. 검색 서비스 상태를 확인해 주세요.', 'error');
-      setCrawlerLogs((prev) => [
-        ...prev,
-        `[OCR] 수집 완료된 실시간 HTML/DOM 가격 노드 데이터 분석 중...`,
-        `[Sync] 정제 단가 적용 (부가세 보정 및 할인 쿠폰 혜택가 산정)`,
-      ]);
-    }, 1100);
-
-    setTimeout(() => {
-      setCrawlingProgress(100);
-      // NOTE: 인터넷 최저가 동기화는 전용 엔드포인트 연동 예정. 현재는 진행 애니메이션만 데모(서버 미반영).
-      showToast(`선택한 ${selectedIds.length}개 상품의 인터넷 최저가 동기화는 준비 중입니다(데모).`, 'info');
-      onDone();
       setIsCrawling(false);
-      onClose();
       setCrawlingProgress(0);
-      setCrawlerLogs([]);
+      return;
-    }, 2000);
+    }
+    setCrawlingProgress(15);
+    pushLog(`[Search] ${pending.size}개 상품 크롤링 진행 — 네이버·쿠팡 수집 및 AI 동일상품 판정...`);
+
+    // 2) 폴링 — GET 이 서버측 증분 동기화를 겸함. 요청 시각 이후 이력이 생긴 상품부터 완료 처리.
+    const total = pending.size;
+    let found = 0;
+    let notFound = 0;
+    const deadline = Date.now() + POLL_TIMEOUT_MS;
+    while (pending.size > 0 && Date.now() < deadline && !cancelledRef.current) {
+      await new Promise((r) => setTimeout(r, POLL_INTERVAL_MS));
+      for (const id of [...pending]) {
+        try {
+          const r = await getLowestPrice(id);
+          const fresh = (r.results ?? []).find((e) => (e.crawl_end_time ?? '') >= startedAt);
+          if (!fresh) continue;
+          pending.delete(id);
+          if (fresh.success_yn && fresh.lp_price != null) {
+            found += 1;
+            pushLog(`[완료] '${nameOf(id)}' 최저가 ${fresh.lp_price.toLocaleString()}원 반영`);
+          } else {
+            notFound += 1;
+            pushLog(`[미발견] '${nameOf(id)}' 동일 상품을 찾지 못함(기존 값 유지)`);
+          }
+        } catch {
+          /* 일시 오류는 다음 tick 재시도 */
+        }
+        setCrawlingProgress(15 + Math.round(((total - pending.size) / total) * 85));
+      }
+    }
+
+    // 3) 마무리 — 목록 갱신(테이블 인터넷 최저가 컬럼 반영) 후 종료
+    const timedOut = pending.size > 0 && !cancelledRef.current;
+    if (timedOut) {
+      pushLog(`[대기초과] ${pending.size}개 상품은 아직 검색 중 — 완료되면 주기 동기화로 자동 반영됩니다.`);
+    }
+    await queryClient.invalidateQueries({ queryKey: ['/v1/item/list'] });
+    setCrawlingProgress(100);
+    if (!cancelledRef.current) {
+      showToast(
+        `최저가 수집 완료: 반영 ${found} · 미발견 ${notFound}${timedOut ? ` · 검색중 ${pending.size}(자동 반영 예정)` : ''}`,
+        found > 0 ? 'success' : 'info',
+      );
+    }
+    onDone();
+    setIsCrawling(false);
+    onClose();
+    setCrawlingProgress(0);
+    setCrawlerLogs([]);
   };
 
   return (
@@ -81,7 +134,7 @@ export function PriceUpdateModal({ open, products, selectedIds, onDone, onClose
         <div className="flex items-center justify-between pb-4 border-b border-border">
           <div className="flex items-center gap-2 text-foreground">
             <Globe className="text-rose-500 animate-pulse" size={18} />
-            <Typography variant="h3">인터넷 최저가 API 실시간 수집 및 동기화</Typography>
+            <Typography variant="h3">인터넷 최저가 실시간 수집 및 동기화</Typography>
           </div>
           {!isCrawling && (
             <button
@@ -99,7 +152,7 @@ export function PriceUpdateModal({ open, products, selectedIds, onDone, onClose
             <Typography variant="small" className="text-[11.5px] leading-relaxed text-left flex items-start gap-2">
               <AlertCircle size={15} className="text-rose-500 shrink-0 mt-0.5" />
               <span>
-                총 <span className="font-bold text-rose-600 dark:text-rose-400 underline decoration-rose-500/50 decoration-2">{selectedIds.length}개</span> 품목에 대하여 네이버 오픈마켓, 다나와, 쿠팡 및 B2B 공공 유통망의 최저가 데이터를 수집 및 비교 분석하여 최신 최저가(minPrice) 필드로 다이렉트 동기화합니다.
+                총 <span className="font-bold text-rose-600 dark:text-rose-400 underline decoration-rose-500/50 decoration-2">{selectedIds.length}개</span> 품목에 대하여 네이버 쇼핑·쿠팡의 최저가를 수집하고 AI 가 동일 상품을 판정하여 인터넷 최저가 필드로 동기화합니다. 상품당 수십 초가 소요될 수 있습니다.
               </span>
             </Typography>
           </div>
@@ -116,7 +169,9 @@ export function PriceUpdateModal({ open, products, selectedIds, onDone, onClose
                     <div className="flex items-center gap-1.5 font-mono">
                       <span className="text-muted-foreground">{(prod.price ?? 0).toLocaleString()}원</span>
                       <span className="text-muted-foreground">→</span>
-                      <span className="text-rose-500 font-bold">자동 가공 수집</span>
+                      <span className="text-rose-500 font-bold">
+                        {prod.internet_lowest_price != null ? `${Number(prod.internet_lowest_price).toLocaleString()}원 갱신` : '신규 수집'}
+                      </span>
                     </div>
                   </div>
                 ) : null;
@@ -130,7 +185,7 @@ export function PriceUpdateModal({ open, products, selectedIds, onDone, onClose
               <div className="flex justify-between text-[11px] text-muted-foreground">
                 <span className="flex items-center gap-1.5 font-bold text-foreground">
                   <Loader2 size={13} className="animate-spin text-rose-500" />
-                  실시간 크롤링 엔진 데이터 동기화 동작 중...
+                  최저가 검색·수집 진행 중... (닫아도 검색은 계속됩니다)
                 </span>
                 <span className="font-bold text-rose-500">{crawlingProgress}%</span>
               </div>
@@ -148,7 +203,7 @@ export function PriceUpdateModal({ open, products, selectedIds, onDone, onClose
                 {crawlerLogs.map((log, idx) => (
                   <div key={idx} className="flex items-start gap-1">
                     <span className="text-zinc-600 select-none shrink-0">&gt;</span>
-                    <span className={`text-left break-all ${log.includes('[System]') ? 'text-zinc-400 font-bold' : log.includes('[OCR]') ? 'text-blue-400' : 'text-emerald-400'}`}>
+                    <span className={`text-left break-all ${log.includes('[System]') || log.includes('[Search]') ? 'text-zinc-400 font-bold' : log.includes('[완료]') ? 'text-emerald-400' : log.includes('[미발견]') || log.includes('[불가]') || log.includes('[오류]') || log.includes('[대기초과]') ? 'text-amber-400' : 'text-rose-300'}`}>
                       {log}
                     </span>
                   </div>
@@ -159,7 +214,7 @@ export function PriceUpdateModal({ open, products, selectedIds, onDone, onClose
             <div className="text-center py-4 border border-dashed border-border rounded bg-muted/10 space-y-2">
               <Cpu size={24} className="mx-auto text-muted-foreground/60" />
               <Typography variant="muted" className="text-[11px]">
-                "최저가 업데이트 시작" 버튼을 누르시면 실시간 스크래핑 엔진이 시작됩니다.
+                "인터넷 최저가 가동" 버튼을 누르시면 실시간 수집이 시작됩니다.
               </Typography>
             </div>
           )}
@@ -167,7 +222,15 @@ export function PriceUpdateModal({ open, products, selectedIds, onDone, onClose
 
         {/* Modal Footer */}
         <div className="flex justify-end gap-2 pt-4 border-t border-border">
-          <Button type="button" variant="outline" size="sm" disabled={isCrawling} onClick={onClose}>
+          <Button
+            type="button"
+            variant="outline"
+            size="sm"
+            onClick={() => {
+              cancelledRef.current = true; // 진행 중이면 폴링만 중단(서버 검색은 계속 → 주기 동기화로 반영)
+              onClose();
+            }}
+          >
             닫기
           </Button>
           <Button type="button" variant="destructive" size="sm" disabled={isCrawling} onClick={handleStartCrawling}>

postgres-init/alters/2026-07-10-iilp-unique-crawl.sql

@@ -0,0 +1,8 @@
+-- 2026-07-10 · LPS 최저가 수집 이력 중복 방지
+-- partner.item_internet_lowest_prices 는 LPS 동기화(negodata 배치 + lowest-price API 의
+-- 온디맨드 동기화)가 채운다. 같은 수집분(item_id, crawl_end_time)이 동시 호출 경합으로
+-- 두 번 들어가지 않도록 유니크 인덱스를 건다 — 경합의 진 쪽 트랜잭션은 실패하고,
+-- 다음 tick 워터마크가 흡수한다(중복 0 보장).
+CREATE UNIQUE INDEX IF NOT EXISTS uq_iilp_item_crawl_time
+    ON partner.item_internet_lowest_prices (item_id, crawl_end_time)
+    WHERE deleted = FALSE;

postgres-init/init-data/init.sql

@@ -446,6 +446,8 @@ CREATE INDEX IF NOT EXISTS idx_sessions_end_time      ON negotiation.sessions (e
 
 -- 상품별 최신 크롤링 최저가 조회
 CREATE INDEX IF NOT EXISTS idx_iilp_item_crawl_time   ON partner.item_internet_lowest_prices (item_id, crawl_end_time DESC) WHERE deleted = FALSE;
+-- [2026-07-10] LPS 동기화 중복 방지 — 같은 수집분(item_id, crawl_end_time)은 1행만(경합 시 진 쪽 tx 실패)
+CREATE UNIQUE INDEX IF NOT EXISTS uq_iilp_item_crawl_time ON partner.item_internet_lowest_prices (item_id, crawl_end_time) WHERE deleted = FALSE;
 
 
 -- ============================================================