A1: corp 다업종 자동 brand resolve + 운영 외 업종 차단

backend/src/main.py

@@ -77,6 +77,7 @@
 from src.schemas.simulation_input import SimulationInput
 from src.services.auth import AuthService
 from src.services.biz_mapper import BizMapper
+from src.services.corp_brand_resolver import resolve_brand_for_industry
 
 from models.explainability.shap_analysis import explain_tcn_prediction
 from models.explainability.simulation import (
@@ -225,6 +226,40 @@ def _pipeline_key(input_data: Any) -> str:
     return f"{input_data.target_district}:{input_data.business_type}:{input_data.brand_name}:{rent}:{area}:{radius}:{pop_w}"
 
 
+def _validate_and_resolve_brand(input_data: SimulationInput) -> None:
+    """biz_number 입력 시 corp 검증 + 다업종 corp 의 brand auto-resolve.
+
+    동작 (input_data.biz_number 가 입력됐을 때만):
+    1. business_type 이 사용자 corp 의 운영 업종인지 검증.
+    2. 운영 외 업종 → HTTPException(400) + 운영 가능 업종 list 응답.
+    3. 운영 내 업종 + corp 의 해당 업종 brand 가 다른 brand 면 brand_name override.
+
+    biz_number 미입력 (개인사업자 / 비회원) → 검증 skip, 사용자 brand_name 그대로.
+    FTC 미등록 corp → 검증 skip + 경고 로그.
+    """
+    if not input_data.biz_number:
+        return
+
+    result = resolve_brand_for_industry(input_data.biz_number, input_data.business_type)
+
+    if result.get("error") == "INDUSTRY_NOT_OPERATED":
+        raise HTTPException(status_code=400, detail=result)
+
+    if result.get("error") in {"USER_NOT_FOUND", "CORP_NOT_IN_FTC", "INVALID_COMPANY_NAME"}:
+        # 비회원 / FTC 미등록 → 검증 skip, 사용자 brand_name 그대로
+        logger.warning(f"[brand_resolver] {result['error']} biz={input_data.biz_number} — fallback to input.brand_name")
+        return
+
+    # 성공: brand_name override (사용자가 다른 brand 입력했어도 corp 정합 brand 로 교체)
+    resolved_brand = result["brand_name"]
+    if input_data.brand_name != resolved_brand:
+        logger.info(
+            f"[brand_resolver] auto-resolve: input.brand_name='{input_data.brand_name}' → '{resolved_brand}' "
+            f"(corp={result['company_name']}, industry={input_data.business_type})"
+        )
+        input_data.brand_name = resolved_brand
+
+
 _BIZ_TYPE_NORMALIZE: dict[str, str] = {
     "cafe": "카페",
     "coffee": "카페",
@@ -935,6 +970,7 @@ async def analyze_location(input_data: SimulationInput, response: Response):
     그쪽으로 옮길 것. 이 endpoint는 기존 프론트/테스트 호환을 위해 유지하다가
     충분히 검증되면 제거 예정.
     """
+    _validate_and_resolve_brand(input_data)
     from src.config.constants import MAPO_DISTRICTS
 
     # IM3-259: deprecation 헤더 — 클라이언트가 /predict + /analyze/llm 으로 옮길 것을 알림
@@ -990,6 +1026,7 @@ async def analyze_llm(input_data: SimulationInput):
 
     /predict와 독립 병렬 호출 가능. winner는 ranking 단계에서 자체 결정.
     """
+    _validate_and_resolve_brand(input_data)
     from src.config.constants import MAPO_DISTRICTS
     from src.schemas.simulation_output import AnalysisOutput
 
@@ -1059,6 +1096,7 @@ async def analyze_llm(input_data: SimulationInput):
 @app.post("/analyze/llm/async")
 async def analyze_llm_async(input_data: SimulationInput) -> dict[str, Any]:
     """AI 분석 비동기 시작 — 즉시 job_id 반환. LangGraph 노드별 진행률 추적."""
+    _validate_and_resolve_brand(input_data)
     from src.config.constants import MAPO_DISTRICTS
     from src.schemas.simulation_output import AnalysisOutput
     from src.services.job_progress_store import (
@@ -1179,6 +1217,7 @@ async def analyze_quick(input_data: SimulationInput):
 
     응답: { district_rankings, winner_district, top_3_candidates }
     """
+    _validate_and_resolve_brand(input_data)
     from src.agents.nodes.district_ranking import district_ranking_node
     from src.agents.nodes.market_analyst import db_client
 
@@ -1661,6 +1700,7 @@ async def predict_districts(input_data: SimulationInput):
     - target_districts 전체에 대해 TCN/BEP/폐업률/폐업위험도/SHAP 병렬 실행
     - 응답: 동별 예측 결과 리스트 (프론트 멀티라인 차트용)
     """
+    _validate_and_resolve_brand(input_data)
     from src.config.constants import MAPO_DISTRICTS
 
     target_districts = getattr(input_data, "target_districts", None) or [input_data.target_district]
@@ -1724,6 +1764,7 @@ async def predict_districts(input_data: SimulationInput):
 @app.post("/predict/async")
 async def predict_districts_async(input_data: SimulationInput) -> dict[str, Any]:
     """ML 예측 비동기 시작 — 즉시 job_id 반환. 진행률은 status endpoint 폴링."""
+    _validate_and_resolve_brand(input_data)
     from src.config.constants import MAPO_DISTRICTS
     from src.services.job_progress_store import (
         create_job,
@@ -1839,6 +1880,7 @@ async def predict_job_status(job_id: str) -> dict[str, Any]:
 @app.post("/simulate", deprecated=True)
 async def run_simulation(input_data: SimulationInput, response: Response):
     """기본 시뮬레이션 엔드포인트"""
+    _validate_and_resolve_brand(input_data)
     response.headers["Deprecation"] = "true"
     response.headers["Link"] = '</predict>; rel="successor-version", </analyze/llm>; rel="successor-version"'
 

backend/src/schemas/simulation_input.py

@@ -18,6 +18,9 @@ class SimulationInput(BaseModel):
 
     business_type: str = Field(..., description="업종 코드 (cafe, restaurant, convenience)")
     brand_name: str = Field(..., description="브랜드명")
+    # 사용자 회원가입 사업자번호 — corp_brand_resolver 가 다업종 corp 의 적합 brand 자동 선택용.
+    # 미입력 시 회원 검증 skip + brand_name 그대로 사용 (개인사업자 / 비회원 호환).
+    biz_number: str | None = Field(default=None, description="사업자등록번호 (corp 검증 + auto-brand-resolve)")
     target_district: str = Field(..., description="출점 후보 행정동 (대표 1개)")
     target_districts: list[str] = Field(
         default_factory=list, description="사용자가 선택한 후보 행정동 목록 (복수 선택 지원)"
@@ -43,9 +46,7 @@ class SimulationInput(BaseModel):
     )
 
     # 출점 후보지 좌표 — 학교환경위생정화구역(rule_school_zone) 거리 계산 트리거
-    lat: float | None = Field(
-        default=None, description="출점 후보지 위도 (학교 거리 룰 트리거)"
-    )
+    lat: float | None = Field(default=None, description="출점 후보지 위도 (학교 거리 룰 트리거)")
     lon: float | None = Field(default=None, description="출점 후보지 경도")
 
     # [customer_revenue P1-C] 타겟 고객 프로필 — models/customer_revenue/predict.py 입력

backend/src/services/corp_brand_resolver.py

@@ -0,0 +1,150 @@
+"""사업자번호 + 업종 → 같은 corp 의 해당 업종 자동 brand 매핑.
+
+다업종 법인 (예: (주)더본코리아 = 빽다방·홍콩반점·빽보이피자·새마을식당...) 의 경우
+회원가입 시 ``biz_brand_mapping`` 에 top frcsCnt brand 1개만 저장됨.
+시뮬레이션 시 사용자가 다른 업종 (예: 중식) 선택하면 같은 corp 의 해당 업종
+가장 큰 brand (홍콩반점0410) 로 자동 resolve.
+
+운영 외 업종 선택 시 ``INDUSTRY_NOT_OPERATED`` 에러 + 운영 가능 업종 list 반환.
+
+설계:
+- ``users.company_name`` (회원가입 시 기록) 기준 ``ftc_brand_franchise.corpNm`` 매칭
+- corpNm 표기 변형 흡수 — ILIKE + corp 핵심어 추출 (괄호/특수문자 제거)
+- 매칭 brand 중 ``frcsCnt`` 큰 것 1개 선택
+- 운영 외 업종 → 거부 (사용자에게 운영 업종 list 안내)
+
+사용처: ``main.py`` 시뮬 endpoint 호출 직후, 시뮬 input.brand_name override.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+
+import sqlalchemy as sa
+
+from src.config.settings import settings
+
+logger = logging.getLogger(__name__)
+
+
+_engine: sa.Engine | None = None
+
+
+def _get_engine() -> sa.Engine:
+    global _engine
+    if _engine is None:
+        _engine = sa.create_engine(settings.postgres_url)
+    return _engine
+
+
+# corpNm 핵심어 추출용 — '(주)', '㈜', '주식회사' 등 법인 prefix/suffix 제거
+_CORP_NOISE_RE = re.compile(r"\(주\)|㈜|주식회사|\([^)]*\)|\s+")
+
+
+def _normalize_corp(name: str) -> str:
+    """corpNm 정규화 — 법인 표기 noise 제거 후 핵심어 추출."""
+    if not name:
+        return ""
+    return _CORP_NOISE_RE.sub("", name).strip()
+
+
+def get_corp_industries(biz_number: str) -> dict:
+    """사업자번호 → corp 의 운영 brand+업종 list.
+
+    Args:
+        biz_number: 사업자등록번호 (하이픈 제거).
+
+    Returns:
+        ``{"company_name": ..., "brands": [...], "industries": [...]}`` 또는
+        ``{"error": "USER_NOT_FOUND" | "CORP_NOT_IN_FTC", ...}``.
+    """
+    engine = _get_engine()
+    with engine.connect() as c:
+        user = c.execute(
+            sa.text("SELECT company_name FROM users WHERE biz_number = :biz"),
+            {"biz": biz_number},
+        ).first()
+        if not user:
+            return {"error": "USER_NOT_FOUND", "biz_number": biz_number}
+
+        company_name = user._mapping["company_name"]
+        norm = _normalize_corp(company_name)
+        if not norm:
+            return {"error": "INVALID_COMPANY_NAME", "company_name": company_name}
+
+        # ftc_brand_franchise 에서 corpNm 매칭 (정규화 ILIKE)
+        # frcsCnt 큰 row 부터 정렬 — 같은 brand 의 다년 데이터는 max 사용
+        rows = c.execute(
+            sa.text(
+                """
+                SELECT "brandNm", "indutyMlsfcNm", MAX("frcsCnt") AS stores
+                FROM ftc_brand_franchise
+                WHERE "corpNm" IS NOT NULL
+                  AND REGEXP_REPLACE("corpNm", '\\(주\\)|㈜|주식회사|\\([^)]*\\)|\\s+', '', 'g') ILIKE :norm
+                GROUP BY "brandNm", "indutyMlsfcNm"
+                ORDER BY stores DESC NULLS LAST
+                """
+            ),
+            {"norm": f"%{norm}%"},
+        ).fetchall()
+
+    if not rows:
+        return {
+            "error": "CORP_NOT_IN_FTC",
+            "company_name": company_name,
+            "message": f"{company_name} 은(는) FTC 가맹사업 정보공개서에 등록되지 않은 corp 입니다.",
+        }
+
+    brands = [
+        {"name": r._mapping["brandNm"], "industry": r._mapping["indutyMlsfcNm"], "stores": r._mapping["stores"] or 0}
+        for r in rows
+    ]
+    industries = sorted({b["industry"] for b in brands if b["industry"]})
+
+    return {
+        "company_name": company_name,
+        "brands": brands,
+        "industries": industries,
+    }
+
+
+def resolve_brand_for_industry(biz_number: str, industry: str) -> dict:
+    """사업자번호 + 업종 → 같은 corp 의 해당 업종 가장 큰 brand 자동 선택.
+
+    Args:
+        biz_number: 사업자등록번호.
+        industry: 업종명 (FTC indutyMlsfcNm 표기 — 한식/중식/일식/...).
+
+    Returns:
+        성공: ``{"brand_name": ..., "industry": ..., "stores": int,
+                 "alternatives": [...], "company_name": ...}``.
+        실패: ``{"error": "INDUSTRY_NOT_OPERATED" | "USER_NOT_FOUND" | "CORP_NOT_IN_FTC",
+                 "operated_industries": [...], ...}``.
+    """
+    portfolio = get_corp_industries(biz_number)
+    if "error" in portfolio:
+        return portfolio
+
+    matched = [b for b in portfolio["brands"] if b["industry"] == industry]
+    if not matched:
+        return {
+            "error": "INDUSTRY_NOT_OPERATED",
+            "company_name": portfolio["company_name"],
+            "requested_industry": industry,
+            "operated_industries": portfolio["industries"],
+            "message": (
+                f"'{industry}' 업종은 {portfolio['company_name']} 운영 brand 에 없습니다. "
+                f"운영 가능 업종: {', '.join(portfolio['industries'])}"
+            ),
+        }
+
+    # frcsCnt 내림차순 정렬됨 (get_corp_industries 가 보장) — 첫 항목 = top brand
+    top = matched[0]
+    return {
+        "brand_name": top["name"],
+        "industry": top["industry"],
+        "stores": top["stores"],
+        "alternatives": [b["name"] for b in matched[1:]],
+        "company_name": portfolio["company_name"],
+    }