feat(admin): superadmin brand picker — /admin/brands 라우터

backend/src/api/admin_brands.py

@@ -0,0 +1,247 @@
+"""슈퍼어드민 전용 brand picker.
+
+엔드포인트:
+- GET /admin/brands — 시뮬 가능 업종 (CS100001~CS100010) 의 brand 통합 목록.
+  소스: ftc_brand_franchise + biz_brand_mapping (회원가입 본부 매핑) UNION.
+  검색: brand_name ILIKE :q OR corp_name ILIKE :q.
+  필터: industry (canonical key, 예: "한식")
+  페이징: page (1+), size (1~200, 기본 50)
+
+권한: role == "superadmin" 만 허용. 다른 역할은 403.
+
+응답:
+{
+  "total": int,
+  "page": int,
+  "size": int,
+  "supported_industries": list[{key, label, cs_code}],   # 시뮬 가능 업종 10종
+  "items": list[BrandItem]
+}
+"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from pydantic import BaseModel
+from sqlalchemy import text
+
+from src.config.business_type_mapping import BUSINESS_TYPE_MAPPING
+from src.database.sync_engine import get_sync_engine
+from src.services.jwt_auth import UserContext, get_current_user
+
+router = APIRouter(prefix="/admin", tags=["admin-brands"])
+
+
+def _db_url() -> str:
+    from src.config.settings import settings
+
+    return settings.postgres_url
+
+
+def require_superadmin(user: UserContext = Depends(get_current_user)) -> UserContext:
+    """role == 'superadmin' 강제. master/manager 모두 403."""
+    if user.role != "superadmin":
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="superadmin 전용 엔드포인트입니다.",
+        )
+    return user
+
+
+class BrandItem(BaseModel):
+    brand_name: str
+    corp_name: Optional[str] = None
+    biz_number: Optional[str] = None
+    business_type: str  # canonical key (예: "한식")
+    cs_code: str  # CS100001 ~ CS100010
+    industry_medium: Optional[str] = None  # FTC 원본 indutyMlsfcNm
+    franchise_count: Optional[int] = None
+    avg_sales: Optional[int] = None
+    source: str  # "ftc" | "biz_brand_mapping"
+
+
+def _industry_match_clause(industry_key: str | None) -> tuple[str, dict[str, Any]]:
+    """canonical key (예: '한식') → ftc_brand_franchise.indutyMlsfcNm ILIKE 절.
+
+    None 이면 모든 시뮬 가능 업종 (10종 ftc_keywords 합집합).
+    """
+    if industry_key:
+        entry = BUSINESS_TYPE_MAPPING.get(industry_key)
+        if not entry:
+            raise HTTPException(
+                status_code=400,
+                detail=f"지원하지 않는 업종 key: {industry_key}",
+            )
+        keywords = entry["ftc_keywords"]
+    else:
+        keywords = []
+        for entry in BUSINESS_TYPE_MAPPING.values():
+            keywords.extend(entry["ftc_keywords"])
+
+    placeholders = []
+    params: dict[str, Any] = {}
+    for i, kw in enumerate(keywords):
+        ph = f"ftc_kw_{i}"
+        placeholders.append(f"COALESCE(\"indutyMlsfcNm\", '') ILIKE :{ph}")
+        params[ph] = f"%{kw}%"
+    return "(" + " OR ".join(placeholders) + ")", params
+
+
+def _resolve_business_type(industry_medium: str | None) -> tuple[str, str] | None:
+    """FTC indutyMlsfcNm → canonical key + cs_code 매핑.
+
+    여러 업종 키워드가 같은 indutyMlsfcNm 에 매칭될 수 있으므로
+    가장 먼저 매칭되는 entry 를 사용.
+    """
+    if not industry_medium:
+        return None
+    haystack = industry_medium.lower()
+    for key, entry in BUSINESS_TYPE_MAPPING.items():
+        for kw in entry["ftc_keywords"]:
+            if kw.lower() in haystack:
+                return key, entry["cs_code"]
+    return None
+
+
+@router.get("/brands")
+def list_admin_brands(
+    q: Optional[str] = Query(default=None, description="brand_name 또는 corp_name 부분 일치"),
+    industry: Optional[str] = Query(
+        default=None,
+        description="canonical 업종 key (예: 한식, 커피). 미지정 시 시뮬 가능 10종 전체",
+    ),
+    page: int = Query(default=1, ge=1),
+    size: int = Query(default=50, ge=1, le=200),
+    _user: UserContext = Depends(require_superadmin),
+) -> dict[str, Any]:
+    """시뮬 가능 업종의 brand 통합 목록 (FTC + biz_brand_mapping)."""
+
+    industry_clause, industry_params = _industry_match_clause(industry)
+
+    where_search = ""
+    search_params: dict[str, Any] = {}
+    if q and q.strip():
+        where_search = " AND (b.brand_name ILIKE :q_pat OR COALESCE(b.corp_name, '') ILIKE :q_pat)"
+        search_params["q_pat"] = f"%{q.strip()}%"
+
+    # FTC + biz_brand_mapping UNION:
+    # - ftc_brand_franchise: 정보공개서 16K+ brand 본문
+    # - biz_brand_mapping: 회원가입 본부의 가맹본부 매핑 (SPOTTER 사용 본부)
+    # 같은 brand_name 이 양쪽에 있을 수 있어 MAX/COALESCE 로 우선순위:
+    # franchise_count·avg_sales 는 ftc 우선, biz_number 는 biz_brand_mapping 만 보유
+    base_sql = f"""
+        WITH ftc AS (
+            SELECT
+                "brandNm" AS brand_name,
+                "corpNm" AS corp_name,
+                NULL::text AS biz_number,
+                "indutyMlsfcNm" AS industry_medium,
+                "frcsCnt" AS franchise_count,
+                "avrgSlsAmt" AS avg_sales,
+                'ftc' AS source
+            FROM ftc_brand_franchise
+            WHERE {industry_clause}
+              AND "brandNm" IS NOT NULL
+        ),
+        biz AS (
+            SELECT
+                brand_name,
+                company_name AS corp_name,
+                biz_number,
+                industry_medium,
+                franchise_count,
+                avg_sales,
+                'biz_brand_mapping' AS source
+            FROM biz_brand_mapping
+            WHERE brand_name IS NOT NULL
+        ),
+        combined AS (
+            SELECT * FROM ftc
+            UNION ALL
+            SELECT * FROM biz
+        ),
+        deduped AS (
+            SELECT DISTINCT ON (brand_name, COALESCE(corp_name, ''))
+                brand_name, corp_name, biz_number, industry_medium,
+                franchise_count, avg_sales, source
+            FROM combined
+            ORDER BY brand_name, COALESCE(corp_name, ''),
+                     CASE WHEN source = 'biz_brand_mapping' THEN 0 ELSE 1 END,
+                     franchise_count DESC NULLS LAST
+        )
+        SELECT * FROM deduped b
+        WHERE 1=1{where_search}
+    """
+
+    params: dict[str, Any] = {**industry_params, **search_params}
+    offset = (page - 1) * size
+
+    engine = get_sync_engine(_db_url())
+    with engine.connect() as conn:
+        total = conn.execute(
+            text(f"SELECT COUNT(*) FROM ({base_sql}) t"),
+            params,
+        ).scalar_one()
+
+        rows = conn.execute(
+            text(
+                f"""
+                {base_sql}
+                ORDER BY franchise_count DESC NULLS LAST, brand_name
+                LIMIT :limit OFFSET :offset
+                """
+            ),
+            {**params, "limit": size, "offset": offset},
+        ).fetchall()
+
+    items: list[dict[str, Any]] = []
+    for r in rows:
+        m = dict(r._mapping)
+        bt = _resolve_business_type(m.get("industry_medium"))
+        # industry_medium 이 시뮬 가능 10종에 매핑 안 되면 skip — UNION 이후에도 잡종 brand 가 들어올 수 있음
+        if bt is None:
+            continue
+        bt_key, cs_code = bt
+        items.append(
+            {
+                "brand_name": m["brand_name"],
+                "corp_name": m.get("corp_name"),
+                "biz_number": m.get("biz_number"),
+                "business_type": bt_key,
+                "cs_code": cs_code,
+                "industry_medium": m.get("industry_medium"),
+                "franchise_count": m.get("franchise_count"),
+                "avg_sales": m.get("avg_sales"),
+                "source": m["source"],
+            }
+        )
+
+    supported = [{"key": k, "label": v["label_kr"], "cs_code": v["cs_code"]} for k, v in BUSINESS_TYPE_MAPPING.items()]
+
+    return {
+        "total": int(total or 0),
+        "page": page,
+        "size": size,
+        "supported_industries": supported,
+        "items": items,
+    }
+
+
+@router.get("/brands/industries")
+def list_supported_industries(
+    _user: UserContext = Depends(require_superadmin),
+) -> dict[str, Any]:
+    """시뮬 가능 업종 메타정보만 가볍게 반환 (drop-down 초기 로딩)."""
+    return {
+        "industries": [
+            {
+                "key": k,
+                "label": v["label_kr"],
+                "cs_code": v["cs_code"],
+                "kakao_category": v["kakao_category"],
+            }
+            for k, v in BUSINESS_TYPE_MAPPING.items()
+        ]
+    }

backend/src/main.py

@@ -165,6 +165,11 @@ async def _check_rate_limit(ip: str) -> tuple[bool, int]:
 
 app.include_router(_sensitivity_router)
 
+# --- admin_brands REST (슈퍼어드민 brand picker — 시뮬 가능 brand 통합 목록) ---
+from src.api.admin_brands import router as _admin_brands_router  # noqa: E402
+
+app.include_router(_admin_brands_router)
+
 
 # customer_revenue MLP 모델 startup 시 워밍업 — 첫 미리보기 호출 latency 0.5~1초 → ~100ms.
 # 가중치 부재 환경에선 silent skip (배포 서버 분리 케이스 보호).

docs/issues/2026-05-06-superadmin-brand-picker.md

@@ -0,0 +1,131 @@
+# 2026-05-06 — 슈퍼어드민 brand picker (P1, 진행중)
+
+## 증상 / 요구
+
+슈퍼어드민이 모든 가맹본부의 brand 를 자유롭게 선택해 시뮬 가능해야 함.
+일반 master/manager 는 회원가입 시 매핑된 corp 의 brand 만 시뮬 가능.
+
+현재 슈퍼어드민 (`role=superadmin`) 은 `simulation_ai`/`simulation_foresee` 의 **저장된 이력만** 조회 가능.
+시뮬 신규 실행 시 brand picker 가 없어 자기 corp(`SPOTTER Admin`)의 brand 만 사용 가능 → 사실상 불가.
+
+## 진단 환경
+
+- 브랜치: `IM3-superadmin-brand-picker` (worktree, origin/dev base)
+- 기준 commit: `c8e730dc` (origin/dev HEAD)
+- DB: ftc_brand_franchise (~16K brand) + biz_brand_mapping (회원가입 본부 brand)
+
+## 해결 — Backend `/admin/brands`
+
+### 신규 라우트 (이번 PR)
+
+| 메서드 | 경로 | 설명 | 권한 |
+|--------|------|------|------|
+| GET | `/admin/brands/industries` | 시뮬 가능 10종 메타 (label, cs_code) | superadmin |
+| GET | `/admin/brands` | brand picker 목록 | superadmin |
+
+`/admin/brands` 쿼리 파라미터:
+- `q`: brand_name / corp_name 부분 일치 (ILIKE)
+- `industry`: canonical key (한식·커피 등) — `BUSINESS_TYPE_MAPPING.keys()` 만 허용
+- `page` / `size`: 페이징 (기본 1 / 50, 최대 200)
+
+응답:
+```json
+{
+  "total": 14571,
+  "page": 1,
+  "size": 50,
+  "supported_industries": [{"key": "한식", "label": "한식음식점", "cs_code": "CS100001"}, ...],
+  "items": [
+    {
+      "brand_name": "메가커피",
+      "corp_name": "(주)앤하우스",
+      "biz_number": null,
+      "business_type": "커피",
+      "cs_code": "CS100010",
+      "industry_medium": "커피",
+      "franchise_count": 3325,
+      "avg_sales": 30000,
+      "source": "ftc"
+    }
+  ]
+}
+```
+
+### 데이터 소스 통합
+
+`ftc_brand_franchise` (FTC 정보공개서) + `biz_brand_mapping` (회원가입 매핑) UNION:
+- FTC: `brandNm`, `corpNm`, `indutyMlsfcNm`, `frcsCnt`, `avrgSlsAmt` — biz_number 없음
+- biz_brand_mapping: `brand_name`, `company_name`, `biz_number`, `industry_medium` — biz_number 있음
+- DISTINCT ON (brand_name, corp_name) 으로 중복 제거 (biz_brand_mapping 우선)
+
+### 시뮬 가능 업종 필터링
+
+`backend/src/config/business_type_mapping.py` 의 `BUSINESS_TYPE_MAPPING` 단일 source:
+- 10종: 한식·중식·일식·양식·제과·패스트푸드·치킨·분식·호프·커피
+- CS100001 ~ CS100010
+- `ftc_keywords` 로 FTC `indutyMlsfcNm` 매칭 (예: 양식 → "서양식", 패스트푸드 → "피자" 흡수)
+
+기타외식·편의점 등은 시뮬 흐름 미지원 → 응답에서 제외.
+
+## 미해결 (다음 단계)
+
+### 1. WIP 머지 후 superadmin bypass 필요
+
+`IM3-263-ai-summary-layout` 브랜치 commit `66b874e7` 의 `_validate_and_resolve_brand` 함수가 dev 머지되면:
+- master/manager: 운영 외 업종 차단됨 (정상)
+- **superadmin: 차단되면 안 됨** — 모든 업종 자유
+
+**필요 패치** (해당 PR 머지 후 follow-up):
+
+```python
+# backend/src/main.py
+def _validate_and_resolve_brand(input_data, current_user=None):
+    # superadmin: corp 검증·brand override 우회
+    if current_user and current_user.role == "superadmin":
+        return
+    biz_number = input_data.biz_number or _resolve_user_biz_number(current_user)
+    ...
+```
+
+3줄 추가. `corp_brand_resolver.get_corp_industries` 도 superadmin 시 `industries=None` 반환.
+
+### 2. Frontend brand picker UI (다음 PR)
+
+- `AuthContext.role === "superadmin"` 감지
+- 시뮬 입력 폼에 brand picker 모달 (typeahead 검색)
+- 선택 → `biz_number` + `brand_name` + `business_type` 자동 채움
+- 매출 프리뷰: `franchise_count`, `avg_sales` 표시
+
+### 3. 알려진 한계
+
+- 동일 brand 가 FTC 여러 yr 또는 다른 source 에서 corp_name 미세하게 다르면 중복 노출 가능 (예: "(주)앤하우스" vs "앤하우스(주)")
+- 응답 items 의 Python post-filter 가 SQL filter 와 불일치 → 페이지당 items 수 < size 가능. total 도 SQL 기준이라 페이지 수 계산 시 오차.
+- 후속 개선: `_resolve_business_type` 로직을 SQL CASE 식 또는 캐시 컬럼으로 이전.
+
+## 영향 매트릭스
+
+| 영역 | 변경 |
+|------|------|
+| backend/src/api/admin_brands.py | 신규 라우터 |
+| backend/src/main.py | router 등록 4줄 |
+| tests/test_admin_brands.py | 15 케이스 신규 |
+| frontend | 미변경 (다음 PR) |
+
+## 검증
+
+- ruff check / format: clean
+- pytest: 15/15 PASS
+- E2E (real DB): 14,571 brand 노출, 커피 6,850, 검색 정상
+
+## 책임 영역
+
+- A1 (찬영): `backend/src/api/`, services/ — 본 PR 범위
+- 다음 단계 superadmin bypass: 본인 영역 내 (services + main.py)
+- 프론트 brand picker: B1·B2 영역 (별도 협의)
+
+## 참고
+
+- 슈퍼어드민 role 도입: `33afb1aa feat(auth): superadmin role`
+- corp_brand_resolver WIP: `66b874e7 feat(corp): 사업자번호 기반 운영 업종 dropdown 자동 차단` (IM3-263)
+- 단일 source mapping: `backend/src/config/business_type_mapping.py`
+- 이전 ultrareview: `docs/issues/2026-05-05-codebase-ultrareview.md`

docs/issues/README.md

@@ -29,6 +29,7 @@ YYYY-MM-DD-<짧은-슬러그>.md
 | [`2026-04-28-summary-tab-empty-cards.md`](./2026-04-28-summary-tab-empty-cards.md) | 🔴 High | 미해결 | B1·B2·C1 (A1 영역 외) |
 | [`2026-04-28-end-to-end-data-flow-gaps.md`](./2026-04-28-end-to-end-data-flow-gaps.md) | 🔴 High (24건 drift) | 미해결 | B1·B2·C1 (P0 4건은 main.py + state.py + synthesis_node + SummaryTab) |
 | [`2026-05-05-codebase-ultrareview.md`](./2026-05-05-codebase-ultrareview.md) | 🔴 Critical (P0 2건 + P1 24건 + P2 다수) | 미해결 | A1 일부 (services/SQL, DB 네이밍) + 타 팀원 (agents/simulation/frontend/infra) |
+| [`2026-05-06-superadmin-brand-picker.md`](./2026-05-06-superadmin-brand-picker.md) | 🟡 P1 (1단계 완료) | Backend `/admin/brands` 완료, FE picker + WIP 머지 후 bypass 잔존 | A1 (backend) + B1·B2 (frontend) |
 
 ## 관련 디렉토리