diff --git a/backend/src/api/admin_brands.py b/backend/src/api/admin_brands.py new file mode 100644 index 00000000..28f92e55 --- /dev/null +++ b/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() + ] + } diff --git a/backend/src/main.py b/backend/src/main.py index 2da51076..8662b013 100644 --- a/backend/src/main.py +++ b/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 (배포 서버 분리 케이스 보호). diff --git a/docs/issues/2026-05-06-superadmin-brand-picker.md b/docs/issues/2026-05-06-superadmin-brand-picker.md new file mode 100644 index 00000000..538c378d --- /dev/null +++ b/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` diff --git a/docs/issues/README.md b/docs/issues/README.md index cd9a87b7..e26d739a 100644 --- a/docs/issues/README.md +++ b/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) | ## 관련 디렉토리 diff --git a/tests/test_admin_brands.py b/tests/test_admin_brands.py new file mode 100644 index 00000000..0cc3b8cd --- /dev/null +++ b/tests/test_admin_brands.py @@ -0,0 +1,211 @@ +"""슈퍼어드민 brand picker 엔드포인트 검증. + +- require_superadmin: master/manager 403 +- list_admin_brands: 시뮬 가능 업종 매핑·industry 필터·페이징 +- list_supported_industries: 10종 모두 반환 +- _resolve_business_type: indutyMlsfcNm → canonical key 매핑 +- _industry_match_clause: 잘못된 key 400 +""" + +from __future__ import annotations + +from unittest.mock import MagicMock +from uuid import uuid4 + +import pytest +from fastapi import HTTPException +from src.api import admin_brands +from src.api.admin_brands import _industry_match_clause, _resolve_business_type, require_superadmin +from src.services.jwt_auth import UserContext + +# --------------------------------------------------------------------------- +# require_superadmin +# --------------------------------------------------------------------------- + + +def _ctx(role: str) -> UserContext: + return UserContext(user_id=str(uuid4()), role=role, email="x@y.z") + + +def test_require_superadmin_master_blocked(): + with pytest.raises(HTTPException) as exc: + require_superadmin(_ctx("master")) + assert exc.value.status_code == 403 + + +def test_require_superadmin_manager_blocked(): + with pytest.raises(HTTPException) as exc: + require_superadmin(_ctx("manager")) + assert exc.value.status_code == 403 + + +def test_require_superadmin_pass(): + out = require_superadmin(_ctx("superadmin")) + assert out.role == "superadmin" + + +# --------------------------------------------------------------------------- +# _resolve_business_type +# --------------------------------------------------------------------------- + + +def test_resolve_korean(): + out = _resolve_business_type("한식") + assert out == ("한식", "CS100001") + + +def test_resolve_coffee_variants(): + # FTC 표기 "음료 (커피 외)" 도 커피로 매핑 + assert _resolve_business_type("커피")[1] == "CS100010" + assert _resolve_business_type("음료 (커피 외)")[1] == "CS100010" + + +def test_resolve_western_servery(): + # FTC 표기 "서양식" → 양식 + assert _resolve_business_type("서양식")[0] == "양식" + + +def test_resolve_pizza_to_fastfood(): + # 정책: 피자는 패스트푸드로 흡수 + assert _resolve_business_type("피자")[0] == "패스트푸드" + + +def test_resolve_unknown_returns_none(): + assert _resolve_business_type("기타외식") is None + assert _resolve_business_type(None) is None + assert _resolve_business_type("") is None + + +# --------------------------------------------------------------------------- +# _industry_match_clause +# --------------------------------------------------------------------------- + + +def test_industry_clause_specific(): + clause, params = _industry_match_clause("커피") + # 커피 ftc_keywords 5개 (커피·카페·음료 (커피 외)·음료·디저트) + assert clause.count("ILIKE") == 5 + assert any("커피" in v for v in params.values()) + + +def test_industry_clause_all(): + clause, params = _industry_match_clause(None) + # 10종 합집합 — 키 개수가 5 이상은 되어야 (대략 50+) + assert clause.count("ILIKE") >= 30 + assert len(params) >= 30 + + +def test_industry_clause_invalid_key(): + with pytest.raises(HTTPException) as exc: + _industry_match_clause("기타외식") + assert exc.value.status_code == 400 + + +# --------------------------------------------------------------------------- +# list_admin_brands — DB 가로채 SQL 흐름 검증 +# --------------------------------------------------------------------------- + + +class _SQLCapture: + def __init__(self, total: int, rows: list): + self.total = total + self.rows = rows + self.executed: list[tuple[str, dict]] = [] + + def make_engine(self): + cap = self + + class _Conn: + def __enter__(self_inner): + return self_inner + + def __exit__(self_inner, *exc): + return False + + def execute(self_inner, stmt, params=None): + cap.executed.append((str(stmt.text), dict(params or {}))) + result = MagicMock() + if "COUNT(*)" in str(stmt.text): + result.scalar_one.return_value = cap.total + else: + result.fetchall.return_value = cap.rows + return result + + class _Engine: + def connect(self_inner): + return _Conn() + + return _Engine() + + +class _Row: + def __init__(self, mapping): + self._mapping = mapping + + +def _row(brand_name, industry_medium, source="ftc", **kw): + return _Row( + { + "brand_name": brand_name, + "corp_name": kw.get("corp_name"), + "biz_number": kw.get("biz_number"), + "industry_medium": industry_medium, + "franchise_count": kw.get("franchise_count"), + "avg_sales": kw.get("avg_sales"), + "source": source, + } + ) + + +def test_list_admin_brands_filters_unsupported_industries(monkeypatch): + cap = _SQLCapture( + total=3, + rows=[ + _row("스타벅스", "커피", franchise_count=1500), + _row("이상한브랜드", "기타외식"), # 매핑 안 되는 업종 → skip + _row("BBQ", "치킨", franchise_count=2000), + ], + ) + monkeypatch.setattr(admin_brands, "get_sync_engine", lambda *_a, **_k: cap.make_engine()) + + out = admin_brands.list_admin_brands(q=None, industry=None, page=1, size=50, _user=_ctx("superadmin")) + # 매핑 안 되는 brand 는 응답에서 제외 + brand_names = [b["brand_name"] for b in out["items"]] + assert "이상한브랜드" not in brand_names + assert {"스타벅스", "BBQ"}.issubset(set(brand_names)) + + starbucks = next(b for b in out["items"] if b["brand_name"] == "스타벅스") + assert starbucks["business_type"] == "커피" + assert starbucks["cs_code"] == "CS100010" + + +def test_list_admin_brands_search_param_propagates(monkeypatch): + cap = _SQLCapture(total=0, rows=[]) + monkeypatch.setattr(admin_brands, "get_sync_engine", lambda *_a, **_k: cap.make_engine()) + + admin_brands.list_admin_brands(q="스타", industry="커피", page=2, size=10, _user=_ctx("superadmin")) + # 첫 SQL = COUNT, 두 번째 = SELECT + count_sql, count_params = cap.executed[0] + list_sql, list_params = cap.executed[1] + + assert "ILIKE :q_pat" in count_sql + assert count_params["q_pat"] == "%스타%" + assert list_params["limit"] == 10 + assert list_params["offset"] == 10 # (page-1) * size = 1 * 10 + + +def test_list_admin_brands_supported_industries_metadata(monkeypatch): + cap = _SQLCapture(total=0, rows=[]) + monkeypatch.setattr(admin_brands, "get_sync_engine", lambda *_a, **_k: cap.make_engine()) + + out = admin_brands.list_admin_brands(q=None, industry=None, page=1, size=50, _user=_ctx("superadmin")) + assert len(out["supported_industries"]) == 10 + keys = {it["key"] for it in out["supported_industries"]} + assert {"한식", "커피", "치킨", "패스트푸드"}.issubset(keys) + + +def test_list_supported_industries_returns_10(): + out = admin_brands.list_supported_industries(_user=_ctx("superadmin")) + assert len(out["industries"]) == 10 + cs_codes = {it["cs_code"] for it in out["industries"]} + assert cs_codes == {f"CS10000{i}" for i in range(1, 10)} | {"CS100010"}