"""
상품명(iname) 기반 후보 group_id 조회 유틸.

목표
- UI(`SearchGroupPositionUI_FullIntegrated.py`)의 iname 검색 후보축소 로직을 재사용 가능하게 분리
- `dome_group_match.py`의 폴백(iname으로 그룹 후보 찾기)에 바로 활용

주의
- 외부 패키지(rapidfuzz 등) 없이도 동작하도록, DB(pg_trgm) 사용 가능 시 DB similarity 우선
- pg_trgm이 없으면 ILIKE 토큰 필터로만 후보를 좁힌다(최종 스코어링은 호출자에서 수행 가능)
"""

from __future__ import annotations

from dataclasses import dataclass
import math
import re
from typing import Iterable, List, Optional, Sequence, Tuple


def norm_iname_db_exact(s: str | None) -> str:
    """
    DB에서 정확일치 비교에 쓰는 정규화.
    - SQL의: lower(regexp_replace(trim(m.iname), '\\s+', ' ', 'g')) 와 동일한 규칙(최대한)
    """
    s2 = str(s or "").strip().lower()
    s2 = re.sub(r"\s+", " ", s2)
    return s2


def normalize_iname_for_tokens(text: str | None) -> str:
    """
    토큰 기반 검색용 정규화(너무 공격적이지 않게).
    - 소문자화
    - 특수문자 제거/공백 정리
    - 한글/영문/숫자/공백 위주로 남김
    """
    if not text:
        return ""
    t = str(text).lower()
    t = re.sub(r"[^0-9a-z가-힣\s]+", " ", t)
    t = re.sub(r"\s+", " ", t).strip()
    return t


def tokenize_iname(norm: str, *, min_token_len: int = 2, max_tokens: int = 5) -> List[str]:
    toks = [t for t in str(norm or "").split(" ") if t]
    mt = int(min_token_len)
    toks = [t for t in toks if len(t) >= mt]
    if max_tokens and max_tokens > 0:
        toks = toks[: int(max_tokens)]
    return toks


_DIM_TOKEN_RE = re.compile(r"^\d+(?:x\d+){1,}$", re.IGNORECASE)


def _ilike_pattern_for_token(t: str, *, is_first: bool) -> str:
    """
    ILIKE 패턴 생성.
    - 치수 토큰(예: 28x21x7)은 DB에서 '×', 공백, '*', 대문자 X 등으로 저장될 수 있어
      '%28%21%7%' 형태로 완화 매칭(순서 보장)한다.
    - 그 외는 기존 정책: 첫 토큰 길이>=3이면 prefix, 아니면 contains.
    """
    s = str(t or "")
    if _DIM_TOKEN_RE.match(s):
        parts = [p for p in s.lower().split("x") if p]
        return "%" + "%".join(parts) + "%"
    if is_first and len(s) >= 3:
        return f"{s}%"
    return f"%{s}%"


def _token_specificity_score(t: str) -> tuple:
    """
    토큰 "구체성" 점수(정렬용).
    - 숫자 포함/치수 토큰은 강한 식별자일 가능성이 높으므로 우선
    - 길이가 길수록 우선
    """
    s = str(t or "")
    is_dim = bool(_DIM_TOKEN_RE.match(s))
    has_digit = any(ch.isdigit() for ch in s)
    return (int(is_dim), int(has_digit), len(s))


def _split_required_optional_tokens(
    tokens: Sequence[str],
    req_n: int,
    *,
    mode: str = "head",
    hybrid_head_n: int = 2,
    hybrid_specificity_n: int = 1,
) -> tuple[list[str], list[str]]:
    """
    required/optional 토큰 분리.

    - mode="head"(기본): **기존 정책**처럼 앞에서부터 req_n개를 required로 사용
      - 실무에서 '브랜드/제품군' 토큰이 앞에 오는 경우가 많아, dimension(28x21x7) 같은 토큰을
        required로 강제하지 않는 것이 더 안정적일 수 있음.
    - mode="specificity": 치수/숫자/긴 토큰을 우선으로 required 선택(필요 시 사용)
    - mode="hybrid": "앞 토큰 일부 + 구체 토큰 일부"를 required로 선택(성능 최적화용)
      - 예: head 2개 + specificity 1개 => (브랜드/제품군) + (치수/숫자/긴 토큰 1개)
    """
    toks = [str(t) for t in (tokens or []) if str(t)]
    if not toks:
        return [], []
    # req_n <= 0 이면 required를 0개로 허용한다.
    # (기존 구현은 max(1, ...)로 최소 1개를 강제해서 "head 조건이 없는 것처럼" 보일 수 있음)
    try:
        req_n_i = int(req_n)
    except Exception:
        req_n_i = 0
    if req_n_i <= 0:
        return [], toks
    n = min(req_n_i, len(toks))
    mode_l = str(mode or "").lower()
    if mode_l in ("head", ""):
        return toks[:n], toks[n:]
    if mode_l == "specificity":
        scored = [(i, toks[i], _token_specificity_score(toks[i])) for i in range(len(toks))]
        top = sorted(scored, key=lambda x: x[2], reverse=True)[:n]
        req_idx = {i for (i, _t, _s) in top}
        req_tokens = [toks[i] for i in sorted(req_idx)]
        opt_tokens = [toks[i] for i in range(len(toks)) if i not in req_idx]
        return req_tokens, opt_tokens
    if mode_l == "hybrid":
        # head에서 먼저 일부 고정 + 나머지는 구체성 우선으로 채운다.
        hn = max(0, int(hybrid_head_n))
        sn = max(0, int(hybrid_specificity_n))
        head_take = min(len(toks), hn)
        req_idx: set[int] = set(range(head_take))

        # 나머지에서 구체성 top-K
        remaining = [i for i in range(len(toks)) if i not in req_idx]
        if remaining and (sn > 0) and (len(req_idx) < n):
            scored = [(i, toks[i], _token_specificity_score(toks[i])) for i in remaining]
            top = sorted(scored, key=lambda x: x[2], reverse=True)[: min(sn, n - len(req_idx))]
            req_idx.update({i for (i, _t, _s) in top})

        # 아직 모자라면 head 순서대로 채운다(원래 정책에 가깝게 유지)
        if len(req_idx) < n:
            for i in range(len(toks)):
                if i in req_idx:
                    continue
                req_idx.add(i)
                if len(req_idx) >= n:
                    break

        req_tokens = [toks[i] for i in sorted(req_idx)]
        opt_tokens = [toks[i] for i in range(len(toks)) if i not in req_idx]
        return req_tokens, opt_tokens

    # default: head
    return toks[:n], toks[n:]


@dataclass(frozen=True)
class WhereClause:
    sql: str
    params: Tuple


def build_ilike_where_clause(
    tokens: Sequence[str],
    *,
    placeholder_style: str = "psycopg2",
    start_index: int = 1,
) -> WhereClause:
    """
    UI와 동일한 토큰 ILIKE 전략:
    - 첫 토큰은 prefix 매칭(길이>=3)으로 인덱스 활용 가능성↑
    - 나머지는 contains
    """
    where_parts: List[str] = []
    params: List[str] = []

    def _ph(i: int) -> str:
        # i는 1-based
        if placeholder_style == "asyncpg":
            return f"${i}"
        return "%s"

    for i, t in enumerate(tokens):
        where_parts.append(f"m.iname ILIKE {_ph(int(start_index) + len(params))}")
        params.append(_ilike_pattern_for_token(str(t), is_first=(i == 0)))

    return WhereClause(sql=" AND ".join(where_parts), params=tuple(params))


def _reorder_tokens_for_best_prefix(tokens: Sequence[str]) -> list[str]:
    """
    build_ilike_where_clause는 첫 토큰만 prefix 매칭이 가능하므로,
    AND 토큰들의 "매칭 집합"은 그대로 두고(순서만 변경),
    prefix로 두었을 때 가장 선별력이 큰 토큰을 1번으로 오게 재정렬한다.

    - dimension 토큰은 prefix 효과가 없으므로 뒤로
    - 길이>=3 우선 (prefix 가능)
    - 구체성 점수(치수/숫자/길이) + 길이로 tie-break
    """
    toks = [str(t) for t in (tokens or []) if str(t)]
    if len(toks) <= 1:
        return toks

    def _score(t: str) -> tuple:
        s = str(t or "")
        is_dim = bool(_DIM_TOKEN_RE.match(s))
        can_prefix = int(len(s) >= 3 and not is_dim)
        spec = _token_specificity_score(s)
        return (can_prefix, spec[0], spec[1], spec[2])

    best_i = 0
    best_s = _score(toks[0])
    for i in range(1, len(toks)):
        sc = _score(toks[i])
        if sc > best_s:
            best_s = sc
            best_i = i
    if best_i == 0:
        return toks
    return [toks[best_i]] + toks[:best_i] + toks[best_i + 1 :]


def _build_optional_token_expr(
    tokens: Sequence[str],
    *,
    placeholder_style: str,
    start_index: int,
) -> tuple[str, str, tuple]:
    """
    optional 토큰용 표현식 생성.

    - **OR 조건**: (m.iname ILIKE ... OR ...)
    - **카운트(매칭 개수)**: (CASE WHEN ... THEN 1 ELSE 0 END + ...)

    주의:
    - asyncpg는 `$n`을 재사용할 수 있어 OR/COUNT에서 같은 placeholder를 공유 가능
    - psycopg2는 `%s` 재사용이 불가능(등장 횟수만큼 파라미터 필요)하므로
      sync 경로에서는 COUNT 표현식을 사용하지 않도록 `cnt_sql="0"`으로 둔다.

    반환: (or_sql, cnt_sql, params)
    """
    toks = [str(t) for t in tokens if str(t)]
    if not toks:
        return "", "0", tuple()

    def _ph(i: int) -> str:
        if placeholder_style == "asyncpg":
            return f"${i}"
        return "%s"

    conds: list[str] = []
    cnt_terms: list[str] = []
    params: list[str] = []

    for idx, t in enumerate(toks):
        ph = _ph(int(start_index) + idx)
        conds.append(f"m.iname ILIKE {ph}")
        if placeholder_style == "asyncpg":
            cnt_terms.append(f"(CASE WHEN m.iname ILIKE {ph} THEN 1 ELSE 0 END)")
        params.append(f"%{t}%")

    or_sql = "(" + " OR ".join(conds) + ")"
    cnt_sql = "(" + " + ".join(cnt_terms) + ")" if (placeholder_style == "asyncpg" and cnt_terms) else "0"
    return or_sql, cnt_sql, tuple(params)


def _compute_req_opt_counts(
    n_tokens: int,
    *,
    required_token_count: int,
    optional_min_hits: int,
    min_hit_ratio: float | None,
) -> tuple[int, int]:
    """
    required/optional 토큰 개수 결정 로직.

    - min_hit_ratio가 주어지면: **전체 토큰 N개 중 ceil(N * ratio) 개를 반드시 AND로 매칭(required)**
      - 요청 정책: "전체가 12개면 최소 6개는 반드시 일치(and 조건)"
      - 따라서 required_token_count/optional_min_hits(고정값)은 무시(=비율 정책이 우선)
    - min_hit_ratio가 None이면: 기존처럼 required_token_count + optional_min_hits(고정) 정책을 사용
    """
    n = max(0, int(n_tokens))
    if n <= 0:
        return 0, 0

    if min_hit_ratio is not None:
        r = float(min_hit_ratio)
        # 0~1 범위로 클램프(과도한 값 방지)
        r = 0.0 if r < 0.0 else (1.0 if r > 1.0 else r)
        min_hits = max(1, int(math.ceil(n * r)))
        # ✅ 비율 정책은 "그 개수만큼 AND"가 목적이므로 optional은 쓰지 않는다.
        req_n = min(n, min_hits)
        return int(req_n), 0

    req_n = max(1, min(n, int(required_token_count)))
    opt_min = max(0, int(optional_min_hits))
    return req_n, opt_min


_PG_TRGM_SYNC_CACHE: Optional[bool] = None
_PG_TRGM_ASYNC_CACHE: Optional[bool] = None


def db_has_pg_trgm_sync(conn) -> bool:
    global _PG_TRGM_SYNC_CACHE
    if _PG_TRGM_SYNC_CACHE is not None:
        return bool(_PG_TRGM_SYNC_CACHE)
    try:
        cur = conn.cursor()
        # ⚠️ psycopg2는 SQL 에러가 나면 커넥션이 "failed transaction" 상태가 되어
        # 이후 모든 쿼리가 실패한다. outer transaction을 깨지 않도록 SAVEPOINT로 격리한다.
        sp = "sp_iname_pg_trgm_check"
        try:
            cur.execute(f"SAVEPOINT {sp}")
        except Exception:
            # autocommit 등으로 SAVEPOINT가 안 되면(=트랜잭션 밖) 그냥 시도
            sp = ""
        try:
            cur.execute("SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'pg_trgm')")
            _PG_TRGM_SYNC_CACHE = bool(cur.fetchone()[0])
        finally:
            if sp:
                try:
                    cur.execute(f"RELEASE SAVEPOINT {sp}")
                except Exception:
                    # RELEASE 실패 시에도 상태 정리는 시도
                    try:
                        cur.execute(f"ROLLBACK TO SAVEPOINT {sp}")
                    except Exception:
                        pass
    except Exception:
        # SAVEPOINT 격리가 실패한 경우에도 커넥션 상태가 깨지지 않게 정리
        try:
            conn.rollback()
        except Exception:
            pass
        _PG_TRGM_SYNC_CACHE = False
    return bool(_PG_TRGM_SYNC_CACHE)


async def db_has_pg_trgm_async(conn) -> bool:
    global _PG_TRGM_ASYNC_CACHE
    if _PG_TRGM_ASYNC_CACHE is not None:
        return bool(_PG_TRGM_ASYNC_CACHE)
    try:
        # nested transaction으로 격리(실패 시 outer abort 방지)
        async with conn.transaction():
            v = await conn.fetchval("SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'pg_trgm')")
        _PG_TRGM_ASYNC_CACHE = bool(v)
    except Exception:
        _PG_TRGM_ASYNC_CACHE = False
    return bool(_PG_TRGM_ASYNC_CACHE)


def candidate_group_ids_by_iname_sync(
    conn,
    *,
    iname: str,
    exclude_icode: str | int | None = None,
    min_ratio: int = 70,
    limit: int = 5000,
    min_token_len: int = 2,
    # 0 또는 음수면 제한 없음(토큰 잘림 방지). 50% 규칙은 "전체 토큰 개수" 기준이므로 기본은 무제한이 안전.
    max_tokens: int = 0,
    required_token_count: int = 2,
    optional_min_hits: int = 1,
    min_hit_ratio: float | None = 0.5,
    max_rows_no_trgm: int = 10,
    # 성능: group_map 전체가 아니라 "그룹 winner 1행"만 대상으로 후보를 만든다.
    # - group_id당 1행이므로 GROUP BY/MAX 집계가 불필요해지고, 읽는 row/page 수가 크게 줄어든다.
    winner_only: bool = True,
    # 성능: required 토큰 선택 모드
    required_token_mode: str = "head",
    hybrid_head_n: int = 2,
    hybrid_specificity_n: int = 1,
    # 성능: 토큰이 많은 상품명에서 ILIKE AND가 너무 길어지면 인덱스 스캔 비용이 커질 수 있어,
    # SQL WHERE에 들어가는 required 토큰 수를 제한할 수 있다(후보 생성용이므로 false positive는 허용).
    # - 0 또는 음수면 제한 없음(기본: 기존 동작 유지)
    max_req_tokens_sql: int = 0,
    prefer_best_prefix_token: bool = True,
    # 성능: pg_trgm 결과(out)가 비었을 때 ILIKE(req-only) 폴백을 얼마나 허용할지 제어
    ilike_fallback_when_trgm_empty: bool = True,
    ilike_fallback_requires_digit_or_dim: bool = False,
) -> List[int]:
    """
    group_map 전체 상품명 대상으로 iname 후보 group_id를 조회.
    - pg_trgm 있으면 similarity 기반으로 상위 group_id 반환(정렬됨)
    - 없으면 ILIKE 토큰 필터로 distinct group_id만 반환(정렬 보장 없음)

    토큰 매칭 정책
    - min_hit_ratio가 설정된 경우(기본 0.5):
      - 전체 토큰 N개 중 ceil(N*min_hit_ratio)개를 **반드시 AND(required)로 매칭**
      - 이 모드에서는 required_token_count / optional_min_hits 값은 정책상 사용되지 않는다(무시)
    - min_hit_ratio=None 인 경우에만:
      - required_token_count + optional_min_hits(기존 방식)로 동작
    """
    iname_s = str(iname or "")
    excl_icode = str(exclude_icode).strip() if (exclude_icode is not None) else ""
    norm = normalize_iname_for_tokens(iname_s)
    tokens = tokenize_iname(norm, min_token_len=min_token_len, max_tokens=max_tokens)
    if not tokens:
        return []

    has_trgm = db_has_pg_trgm_sync(conn)

    # 토큰 정책:
    # - 기본: 전체 토큰 N개 중 50% 이상(ceil(N*0.5)) 매칭되면 통과하도록 required/optional을 자동 계산
    # - pg_trgm 없음 fallback: 성능 이슈 방지를 위해 "카운트/랭킹"은 하지 않고,
    #   **50% 정책을 required AND로 근사 적용**한다.
    if (not has_trgm) and (min_hit_ratio is not None):
        # 50% 기준을 만족시키기 위해 최소 매칭 개수만큼을 required로 올리고(optional=0)
        r = float(min_hit_ratio)
        r = 0.0 if r < 0.0 else (1.0 if r > 1.0 else r)
        min_hits = max(1, int(math.ceil(len(tokens) * r)))
        req_n, opt_min = min(len(tokens), min_hits), 0
    else:
        req_n, opt_min = _compute_req_opt_counts(
            len(tokens),
            required_token_count=required_token_count,
            optional_min_hits=optional_min_hits,
            min_hit_ratio=min_hit_ratio,
        )
    req_tokens, opt_tokens = _split_required_optional_tokens(
        tokens,
        req_n,
        mode=required_token_mode,
        hybrid_head_n=hybrid_head_n,
        hybrid_specificity_n=hybrid_specificity_n,
    )
    if opt_min <= 0:
        opt_tokens = []
    if not opt_tokens:
        opt_min = 0

    req_tokens_sql = list(req_tokens)
    if int(max_req_tokens_sql) > 0 and len(req_tokens_sql) > int(max_req_tokens_sql):
        req_tokens_sql = req_tokens_sql[: int(max_req_tokens_sql)]
    # head 모드에서는 토큰 "순서" 자체가 정책이므로 재정렬하지 않는다.
    # (재정렬을 켜면 첫 토큰(prefix)이 바뀌어 '조립식%, %회전형%, %3면꽃이%' 같은 기대와 달라질 수 있음)
    if bool(prefer_best_prefix_token) and str(required_token_mode or "").lower() not in ("head", ""):
        req_tokens_sql = _reorder_tokens_for_best_prefix(req_tokens_sql)
    where_req = build_ilike_where_clause(req_tokens_sql, placeholder_style="psycopg2", start_index=1)
    opt_or_sql, _opt_cnt_sql, opt_params = _build_optional_token_expr(
        opt_tokens,
        placeholder_style="psycopg2",
        start_index=1 + len(where_req.params),
    )
    where_sql_req_only = where_req.sql
    where_sql_trgm = where_req.sql
    if opt_or_sql and opt_min > 0:
        # sync 경로에서는 optional_min_hits > 1을 엄밀히 강제하지 않는다(OR로 "최소 1개"만 보장).
        where_sql_trgm = f"({where_sql_trgm}) AND {opt_or_sql}"
    cur = conn.cursor()

    if has_trgm:
        # similarity()는 0~1.0 → threshold
        sim_threshold = max(0.0, min(1.0, float(min_ratio) / 100.0))
        try:
            # ⚠️ pg_trgm/similarity 함수 미존재/권한 문제 등이 발생할 수 있으므로 SAVEPOINT로 격리
            sp = "sp_iname_trgm_query"
            try:
                cur.execute(f"SAVEPOINT {sp}")
            except Exception:
                sp = ""
            if winner_only:
                from_clause = """
                  FROM mlinkdw.shopprod_group2 g
                  JOIN mlinkdw.shopprod_group_map2 m ON m.group_id = g.group_id AND m.vender_code = g.winner_vender_code AND m.icode = g.winner_icode
                """
            else:
                from_clause = "FROM mlinkdw.shopprod_group_map2 m"

            cur.execute(
                f"""
                WITH cand AS (
                  SELECT
                    m.group_id,
                    mlinkdw.similarity(lower(m.iname), lower(%s)) AS sim
                  {from_clause}
                  WHERE {where_sql_trgm} {("AND m.icode <> %s") if excl_icode else ""}
                )
                SELECT group_id
                FROM cand
                WHERE sim >= %s
                ORDER BY sim DESC
                LIMIT %s
                """,
                tuple(
                    [iname_s]
                    + list(where_req.params)
                    + list(opt_params)
                    + ([excl_icode] if excl_icode else [])
                    + [sim_threshold, int(limit)]
                ),
            )
            rows = cur.fetchall()
            out: List[int] = []
            for r in rows:
                try:
                    out.append(int(r[0]))
                except Exception:
                    continue
            if sp:
                try:
                    cur.execute(f"RELEASE SAVEPOINT {sp}")
                except Exception:
                    pass
            return out
        except Exception:
            # 권한/함수 미존재 등 → 아래 ILIKE fallback로 진행
            try:
                # outer transaction을 깨지 않도록 savepoint로 되돌린다.
                try:
                    if "sp" in locals() and sp:
                        cur.execute(f"ROLLBACK TO SAVEPOINT {sp}")
                        cur.execute(f"RELEASE SAVEPOINT {sp}")
                    else:
                        conn.rollback()
                except Exception:
                    try:
                        conn.rollback()
                    except Exception:
                        pass
            except Exception:
                pass

    # pg_trgm이 있었던 케이스에서 out이 비었을 때, ILIKE 폴백 자체를 최소화하고 싶으면 여기서 차단한다.
    if has_trgm:
        if not bool(ilike_fallback_when_trgm_empty):
            return []
        if bool(ilike_fallback_requires_digit_or_dim):
            has_specific = False
            for t in req_tokens_sql:
                s = str(t or "")
                if _DIM_TOKEN_RE.match(s) or any(ch.isdigit() for ch in s):
                    has_specific = True
                    break
            if not has_specific:
                return []

    # fallback(pg_trgm 없음): ILIKE "필수 토큰 AND"만으로 후보 group_id 축소(최종 similarity 판단은 호출자)
    # - 성능 이슈 방지를 위해 optional OR/카운트/랭킹은 사용하지 않는다.
    if winner_only:
        from_clause2 = """
          FROM mlinkdw.shopprod_group2 g
          JOIN mlinkdw.shopprod_group_map2 m
            ON m.group_id = g.group_id
           AND m.vender_code = g.winner_vender_code
           AND m.icode = g.winner_icode
        """
    else:
        from_clause2 = "FROM mlinkdw.shopprod_group_map2 m"
    cur.execute(
        f"""
        SELECT DISTINCT m.group_id
        {from_clause2}
        WHERE {where_sql_req_only} {("AND m.icode <> %s") if excl_icode else ""}
        LIMIT %s
        """,
        tuple(list(where_req.params) + ([excl_icode] if excl_icode else []) + [int(max_rows_no_trgm)]),
    )
    rows = cur.fetchall()
    out2: List[int] = []
    for r in rows:
        try:
            out2.append(int(r[0]))
        except Exception:
            continue
    return out2[: int(limit)]


async def candidate_group_ids_by_iname_async(
    conn,
    *,
    iname: str,
    exclude_icode: str | int | None = None,
    min_ratio: int = 70,
    limit: int = 5000,
    min_token_len: int = 2,
    # 0 또는 음수면 제한 없음(토큰 잘림 방지). 50% 규칙은 "전체 토큰 개수" 기준이므로 기본은 무제한이 안전.
    max_tokens: int = 0,
    required_token_count: int = 2,
    optional_min_hits: int = 1,
    min_hit_ratio: float | None = 0.5,
    max_rows_no_trgm: int = 10,
    merge_ilike_when_trgm_empty_only: bool = True,
    debug_sql: bool = False,
    # 성능: group_map 전체가 아니라 "그룹 winner 1행"만 대상으로 후보를 만든다.
    winner_only: bool = True,
    # 성능: required 토큰 선택 모드
    required_token_mode: str = "head",
    hybrid_head_n: int = 2,
    hybrid_specificity_n: int = 1,
    # 성능: SQL WHERE에 들어가는 required 토큰 수 제한(0이면 제한 없음)
    max_req_tokens_sql: int = 0,
    prefer_best_prefix_token: bool = True,
    # 성능: pg_trgm 결과(out)가 비었을 때 ILIKE(req-only) 폴백을 얼마나 허용할지 제어
    # - 기본(True): 기존 동작 유지(단, merge_ilike_when_trgm_empty_only=True이면 out이 비었을 때만 수행)
    # - requires_digit_or_dim=True면: 숫자/치수 토큰이 하나라도 있을 때만 폴백 수행(흔한 토큰 폭발 방지)
    ilike_fallback_when_trgm_empty: bool = True,
    ilike_fallback_requires_digit_or_dim: bool = False,
) -> List[int]:
    # asyncpg connection 전용 가드 (psycopg2 연결이 들어오면 명확히 안내)
    if not hasattr(conn, "fetch"):
        raise TypeError(
            "candidate_group_ids_by_iname_async()는 asyncpg Connection이 필요합니다. "
            "psycopg2 connection을 쓰는 코드라면 candidate_group_ids_by_iname_sync()를 호출하세요."
        )

    iname_s = str(iname or "")
    excl_icode = str(exclude_icode).strip() if (exclude_icode is not None) else ""
    norm = normalize_iname_for_tokens(iname_s)
    tokens = tokenize_iname(norm, min_token_len=min_token_len, max_tokens=max_tokens)
    if not tokens:
        return []

    has_trgm = await db_has_pg_trgm_async(conn)

    # pg_trgm이 없을 때도 50% 정책을 적용할 수는 있으나,
    # "어떤 토큰이든 50% 이상"을 정확 구현하려면 ILIKE 카운트가 필요하여 성능 부담이 큼.
    # 따라서 pg_trgm 없음 환경에서는 50% 기준을 required AND로 근사 적용한다(optional=0).
    if (not has_trgm) and (min_hit_ratio is not None):
        r = float(min_hit_ratio)
        r = 0.0 if r < 0.0 else (1.0 if r > 1.0 else r)
        min_hits = max(1, int(math.ceil(len(tokens) * r)))
        req_n, opt_min = min(len(tokens), min_hits), 0
    else:
        req_n, opt_min = _compute_req_opt_counts(
            len(tokens),
            required_token_count=required_token_count,
            optional_min_hits=optional_min_hits,
            min_hit_ratio=min_hit_ratio,
        )
        if debug_sql and (min_hit_ratio is not None):
            try:
                print(
                    "[DEBUG][iname_search] NOTE: min_hit_ratio 모드에서는 "
                    "required_token_count/optional_min_hits가 무시됩니다."
                )
            except Exception:
                pass
        # debug/가시성: 50% 규칙의 "목표 총 매칭 수" 계산
        if min_hit_ratio is not None:
            try:
                r_dbg = float(min_hit_ratio)
                r_dbg = 0.0 if r_dbg < 0.0 else (1.0 if r_dbg > 1.0 else r_dbg)
                min_hits = max(1, int(math.ceil(len(tokens) * r_dbg)))
            except Exception:
                min_hits = None
        else:
            min_hits = None
    req_tokens, opt_tokens = _split_required_optional_tokens(
        tokens,
        req_n,
        mode=required_token_mode,
        hybrid_head_n=hybrid_head_n,
        hybrid_specificity_n=hybrid_specificity_n,
    )
    if opt_min <= 0:
        opt_tokens = []
    if not opt_tokens:
        opt_min = 0

    req_tokens_sql = list(req_tokens)
    if int(max_req_tokens_sql) > 0 and len(req_tokens_sql) > int(max_req_tokens_sql):
        req_tokens_sql = req_tokens_sql[: int(max_req_tokens_sql)]
    # head 모드에서는 토큰 "순서" 자체가 정책이므로 재정렬하지 않는다.
    if bool(prefer_best_prefix_token) and str(required_token_mode or "").lower() not in ("head", ""):
        req_tokens_sql = _reorder_tokens_for_best_prefix(req_tokens_sql)

    # pg_trgm query에서는 $1을 "검색어(iname_s)"로 사용하므로 required token placeholder는 $2부터 시작해야 한다.
    where_pg_req = build_ilike_where_clause(req_tokens_sql, placeholder_style="asyncpg", start_index=2)
    opt_or_pg, opt_cnt_pg, opt_params_pg = _build_optional_token_expr(
        opt_tokens,
        placeholder_style="asyncpg",
        start_index=2 + len(where_pg_req.params),
    )

    # pg_trgm 없이 ILIKE만 쓸 때는 token placeholder를 $1부터 사용
    where_ilike_req = build_ilike_where_clause(req_tokens_sql, placeholder_style="asyncpg", start_index=1)
    opt_or_ilike, opt_cnt_ilike, opt_params_ilike = _build_optional_token_expr(
        opt_tokens,
        placeholder_style="asyncpg",
        start_index=1 + len(where_ilike_req.params),
    )

    out: List[int] = []
    if has_trgm:
        sim_threshold = max(0.0, min(1.0, float(min_ratio) / 100.0))
        try:
            # nested transaction으로 격리(실패 시 outer abort 방지)
            async with conn.transaction():
                # placeholder index 계산
                n_req = len(where_pg_req.params)
                n_opt = len(opt_params_pg)
                excl_pos = 2 + n_req + n_opt
                sim_pos = excl_pos + (1 if excl_icode else 0)
                optmin_pos = sim_pos + 1
                limit_pos = sim_pos + (2 if opt_min > 0 else 1)
                if winner_only:
                    from_clause = """
                      FROM mlinkdw.shopprod_group2 g
                      JOIN mlinkdw.shopprod_group_map2 m
                        ON m.group_id = g.group_id
                       AND m.vender_code = g.winner_vender_code
                       AND m.icode = g.winner_icode
                    """
                else:
                    from_clause = "FROM mlinkdw.shopprod_group_map2 m"

                q = f"""
                    WITH cand AS (
                      SELECT
                        m.group_id,
                        mlinkdw.similarity(lower(m.iname), lower($1)) AS sim,
                        {opt_cnt_pg} AS opt_cnt
                      {from_clause}
                      WHERE ({where_pg_req.sql})
                        {("AND " + opt_or_pg) if (opt_or_pg and opt_min > 0) else ""}
                        {f"AND m.icode <> ${excl_pos}" if excl_icode else ""}
                    )
                    SELECT group_id
                    FROM cand
                    WHERE sim >= ${sim_pos}
                      {f"AND opt_cnt >= ${optmin_pos}" if opt_min > 0 else ""}
                    ORDER BY sim DESC, opt_cnt DESC
                    LIMIT ${limit_pos}
                    """
                args = (
                    tuple([iname_s])
                    + where_pg_req.params
                    + opt_params_pg
                    + ((excl_icode,) if excl_icode else tuple())
                    + (
                    (sim_threshold, opt_min, int(limit)) if opt_min > 0 else (sim_threshold, int(limit))
                    )
                )
                if debug_sql:
                    try:
                        print("[DEBUG][iname_search] candidate_group_ids_by_iname_async(pg_trgm) SQL:\n" + q.strip())
                        print("[DEBUG][iname_search] args=" + repr(args))
                    except Exception:
                        pass
                rows = await conn.fetch(q, *args)
            for r in rows:
                try:
                    out.append(int(r["group_id"]))
                except Exception:
                    try:
                        out.append(int(r[0]))
                    except Exception:
                        continue
        except Exception as e:
            # 함수 미존재 등 → 아래 ILIKE fallback
            if debug_sql:
                try:
                    print(f"[DEBUG][iname_search] pg_trgm query failed: {type(e).__name__}: {e}")
                except Exception:
                    pass
            pass

    # pg_trgm 결과가 있고, "out이 비었을 때만 ILIKE 추가검색" 모드면 즉시 반환(성능 우선)
    if has_trgm and out and bool(merge_ilike_when_trgm_empty_only):
        if debug_sql:
            try:
                print(f"[DEBUG][iname_search] trgm_out_only mode: returning trgm out (len={len(out)})")
            except Exception:
                pass
        return out[: int(limit)]

    # pg_trgm 결과가 비었을 때도 ILIKE 폴백 자체를 최소화하고 싶으면 여기서 차단한다.
    if has_trgm and (not out) and bool(merge_ilike_when_trgm_empty_only):
        if not bool(ilike_fallback_when_trgm_empty):
            if debug_sql:
                try:
                    print("[DEBUG][iname_search] ILIKE fallback disabled when trgm out is empty")
                except Exception:
                    pass
            return []
        if bool(ilike_fallback_requires_digit_or_dim):
            has_specific = False
            for t in req_tokens_sql:
                s = str(t or "")
                if _DIM_TOKEN_RE.match(s) or any(ch.isdigit() for ch in s):
                    has_specific = True
                    break
            if not has_specific:
                if debug_sql:
                    try:
                        print("[DEBUG][iname_search] ILIKE fallback skipped (no digit/dim token in required tokens)")
                    except Exception:
                        pass
                return []

    # ✅ req-only ILIKE 후보를 소량 merge
    # - pg_trgm이 있어도 similarity threshold/정렬 때문에 특정 group_id가 누락될 수 있으므로
    #   (특히 "이우 훼미리박스 ..." 같은 일반 토큰 케이스) ILIKE 후보를 max_rows_no_trgm 만큼 추가한다.
    # - 성능 이슈 방지를 위해 optional OR/카운트/랭킹은 사용하지 않는다.
    try:
        async with conn.transaction():
            n_req2 = len(where_ilike_req.params)
            excl_pos2 = n_req2 + 1
            limit_pos2 = n_req2 + (2 if excl_icode else 1)
            if winner_only:
                from_clause2 = """
                    FROM mlinkdw.shopprod_group2 g
                    JOIN mlinkdw.shopprod_group_map2 m
                      ON m.group_id = g.group_id
                     AND m.vender_code = g.winner_vender_code
                     AND m.icode = g.winner_icode
                """
            else:
                from_clause2 = "FROM mlinkdw.shopprod_group_map2 m"
            q2 = f"""
                SELECT DISTINCT m.group_id
                {from_clause2}
                WHERE ({where_ilike_req.sql}) {f"AND m.icode <> ${excl_pos2}" if excl_icode else ""}
                LIMIT ${limit_pos2}
                """
            args2 = tuple(where_ilike_req.params) + ((excl_icode,) if excl_icode else tuple()) + (int(max_rows_no_trgm),)
            if debug_sql:
                try:
                    print(
                        "[DEBUG][iname_search] "
                        f"n_tokens={len(tokens)} min_hit_ratio={min_hit_ratio!r} min_hits={locals().get('min_hits', None)!r} "
                        f"required_token_count={required_token_count} -> req_n={req_n}, opt_min={opt_min}, has_trgm={has_trgm}"
                    )
                    print(f"[DEBUG][iname_search] tokens={tokens!r} req_tokens={req_tokens!r}")
                    print("[DEBUG][iname_search] candidate_group_ids_by_iname_async(ILIKE req-only) SQL:\n" + q2.strip())
                    print("[DEBUG][iname_search] args=" + repr(args2))
                except Exception:
                    pass
            rows2 = await conn.fetch(q2, *args2)
    except Exception as e:
        if debug_sql:
            try:
                print(f"[DEBUG][iname_search] ILIKE query failed: {type(e).__name__}: {e}")
            except Exception:
                pass
        rows2 = []

    # merge(dedup, order-preserving)
    seen = set()
    merged: List[int] = []
    for gid in out:
        if gid not in seen:
            seen.add(gid)
            merged.append(gid)
    for r in rows2:
        try:
            gid2 = int(r["group_id"])
        except Exception:
            try:
                gid2 = int(r[0])
            except Exception:
                continue
        if gid2 not in seen:
            seen.add(gid2)
            merged.append(gid2)

    if debug_sql:
        try:
            print(f"[DEBUG][iname_search] merged_gids(head)={merged[:20]!r} total={len(merged)}")
        except Exception:
            pass

    return merged[: int(limit)]