dbx-main/app/modules/cupang/itemcode.py

r"""itemcode_db 읽기 전용 상품 검색.

cupang 모듈은 itemcode_db 의 상품(낱개코드/세트코드)을 **읽기만** 한다.
cupang_db 에 상품을 복제 저장하지 않는다. 출고 라인에는 product_code 와
product_name_snapshot 만 보존한다.

⚠️ itemcode_db 의 실제 테이블/컬럼명은 이 저장소(main-app)에 정의되어 있지 않다.
   운영 서버에서 다음으로 먼저 구조를 확인한 뒤 환경변수를 설정해야 한다:

     docker exec -it postgres-db psql -U postgres -d itemcode_db -c "\dt"
     docker exec -it postgres-db psql -U postgres -d itemcode_db -c "\d <테이블명>"

환경변수 (모두 미설정 시 검색 비활성 → 폼에서 수동 입력으로 폴백):

  ITEMCODE_DB_URL      읽기 전용 DSN. 예: postgresql://itemcode_ro:<pwd>@postgres-db:5432/itemcode_db
  ITEMCODE_SEARCH_SQL  (선택) 검색 SQL 직접 지정. 아래 자동 생성 대신 사용.
                       반드시 code, name, type 컬럼을 별칭으로 반환하고,
                       %(q)s 파라미터를 LIKE 패턴으로 받는다.

  자동 생성용 (ITEMCODE_SEARCH_SQL 미설정 시):
  ITEMCODE_TABLE       검색 대상 테이블/뷰 (예: products 또는 item_master)
  ITEMCODE_CODE_COL    코드 컬럼명 (기본: product_code)
  ITEMCODE_NAME_COL    상품명 컬럼명 (기본: product_name)
  ITEMCODE_TYPE_COL    (선택) 단품/세트 구분 컬럼명. 없으면 type 은 빈 문자열.

낱개코드와 세트코드가 별도 테이블이면 ITEMCODE_SEARCH_SQL 에 UNION 으로 직접 작성한다.
"""

from __future__ import annotations

import logging
import os
import re
from typing import Any

logger = logging.getLogger("cupang.itemcode")

# 안전한 SQL 식별자(테이블/컬럼)만 허용 — 인젝션 방지.
_IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_.]*$")


def _ident(value: str, *, what: str) -> str:
    v = (value or "").strip()
    if not _IDENT_RE.match(v):
        raise ValueError(f"안전하지 않은 {what} 식별자: {value!r}")
    return v


class ItemcodeReader:
    """itemcode_db 읽기 전용 커넥션 풀 + 상품 검색.

    설정이 없거나 불완전하면 `enabled=False` 로 두고, search()는 빈 리스트를 반환한다.
    앱 부팅이나 cupang 모듈 진입을 막지 않는다.
    """

    def __init__(self) -> None:
        self._pool: Any = None
        self._sql: str | None = None
        self.enabled = False
        self.reason = ""
        self.last_error = ""  # 마지막 조회 오류(진단용, 비밀값 없음)
        self._configure()

    def _configure(self) -> None:
        dsn = os.getenv("ITEMCODE_DB_URL", "").strip()
        if not dsn:
            self.reason = "ITEMCODE_DB_URL 미설정 — 상품 검색 비활성(수동 입력 사용)."
            return

        custom_sql = os.getenv("ITEMCODE_SEARCH_SQL", "").strip()
        if custom_sql:
            self._sql = custom_sql
        else:
            table = os.getenv("ITEMCODE_TABLE", "").strip()
            if not table:
                self.reason = (
                    "ITEMCODE_TABLE(또는 ITEMCODE_SEARCH_SQL) 미설정 — "
                    "상품 검색 비활성(수동 입력 사용)."
                )
                return
            try:
                table_id = _ident(table, what="테이블")
                code_col = _ident(os.getenv("ITEMCODE_CODE_COL", "product_code"), what="코드 컬럼")
                name_col = _ident(os.getenv("ITEMCODE_NAME_COL", "product_name"), what="상품명 컬럼")
                type_col_raw = os.getenv("ITEMCODE_TYPE_COL", "").strip()
                type_expr = _ident(type_col_raw, what="구분 컬럼") if type_col_raw else "''"
            except ValueError as exc:
                self.reason = f"itemcode 검색 설정 오류: {exc}"
                return
            self._sql = (
                f"SELECT {code_col} AS code, {name_col} AS name, {type_expr} AS type "
                f"FROM {table_id} "
                f"WHERE {code_col} ILIKE %(q)s OR {name_col} ILIKE %(q)s "
                f"ORDER BY {code_col} ASC LIMIT %(limit)s"
            )

        # 풀은 lazy open — 부팅 시 itemcode_db 가 잠시 끊겨도 죽지 않게.
        try:
            from psycopg.rows import dict_row
            from psycopg_pool import ConnectionPool

            self._pool = ConnectionPool(
                conninfo=dsn,
                min_size=1,
                max_size=3,
                kwargs={"row_factory": dict_row, "autocommit": True},
                open=False,
            )
            self._pool.open(wait=False)
            self.enabled = True
            self.reason = ""
        except Exception as exc:  # noqa: BLE001 — 설정/드라이버 문제로 모듈을 죽이지 않음
            self.reason = f"itemcode_db 연결 풀 생성 실패: {type(exc).__name__}"

    def search(self, query: str, *, limit: int = 20) -> list[dict[str, Any]]:
        """code/name 부분 일치 검색. 반환: [{"code","name","type"}].

        비활성 상태이거나 조회 실패 시 빈 리스트(예외 비전파 — UI 는 수동 입력 폴백).
        """
        q = (query or "").strip()
        if not q:
            return []
        return self._run(f"%{q}%", limit)

    def list_all(self, *, limit: int = 2000) -> list[dict[str, Any]]:
        """전체 상품 목록(낱개+세트). 설정 화면 왼쪽 리스트용."""
        return self._run("%", limit)

    def _run(self, like: str, limit: int) -> list[dict[str, Any]]:
        if not self.enabled or not self._pool or not self._sql:
            return []
        try:
            with self._pool.connection() as conn:
                rows = conn.execute(
                    self._sql, {"q": like, "limit": int(limit)}
                ).fetchall()
            self.last_error = ""
        except Exception as exc:  # noqa: BLE001 — 모듈을 죽이지 않음. 원인은 로그 + last_error.
            self.last_error = f"{type(exc).__name__}: {exc}"
            logger.exception("itemcode 조회 실패 (SQL/스키마 확인 필요)")
            return []
        out: list[dict[str, Any]] = []
        for r in rows:
            out.append(
                {
                    "code": str(r.get("code") or "").strip(),
                    "name": str(r.get("name") or "").strip(),
                    "type": str(r.get("type") or "").strip(),
                }
            )
        return out

    def close(self) -> None:
        if self._pool is not None:
            self._pool.close()