o2o-negosium-original/negodata/backend/services/quotation/pricing.py

"""목표가·앵커링가 계산 + 산정내역 응답 (QuotationService 파사드의 가격 부분)."""
import uuid
from typing import Optional

from common.anchoring import (
    SAMPLEABLE_SUPPLIER_TYPES,
    calc_anchoring_price,
    calc_price_range_index,
    fetch_current_values,
    get_base_anchoring_value,
)
from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import quotations, sessions
from common.enums import DBWRType, ErrorType, QuotationType
from common.logger import LOG
from router.v1.quotation.protocol import Res_TargetBreakdown, TargetCandidate


class PricingMixin:
    # 인터넷 평균 수수료율(상수). 시장 평균값이라 견적/세팅별로 두지 않고 고정. 목표가=인터넷최저가×(1−값).
    INTERNET_AVERAGE_FEE = 0.078

    # TODO 목표마진율(quotation_settings.target_margin_rate)을 세팅에서 상수로 강등 검토 (2026-07-27)
    #      KTC 원본은 하드코딩 상수 0.065 (task_get_rq_price.py). 쓰임새도 판매가 후보 하나뿐.
    #      강등 시 딸려가는 것: quotation_settings 컬럼(날짜 SQL 파일)·세팅 화면·산정내역 표기·테스트.

    # 목표가 후보 basis 코드 ↔ 표시 라벨(산정내역 응답에서 프론트가 그대로 표기).
    _CANDIDATE_LABELS = {"md": "구매담당자 제시가", "internet": "인터넷 최저가", "purchase": "매입가", "selling": "판매가"}

    # 가격 소스별로, 회사 설정 hidden_fields 에서 쓰는 필드 이름. 숨긴 가격은 목표가 후보에서도 뺀다.
    _SOURCE_HIDDEN_FIELD = {"internet": "internet_lowest_price", "purchase": "purchase_price", "selling": "selling_price"}

    @staticmethod
    def _candidates(md_price=None, internet_lowest=None, purchase=None, selling=None, fee=0.0, margin=0.0, is_new=False, hidden=None) -> list[tuple[str, float]]:
        """목표가 후보 [(소스, 후보가)] 목록. 생성(_calc_target_price)과 산정내역 화면이 공용.

        md_price 있으면 그 값 하나만. 없으면:
          신규(is_new): 인터넷최저가 × (1−fee)
          재:           인터넷최저가 × (1−fee) · 매입가 · 판매가 × (1−margin)
        값이 없거나 hidden(회사가 숨긴 필드)에 든 소스는 제외."""
        if md_price:
            return [("md", float(int(md_price)))]
        # (소스, 가격, 차감율) 표 — 조건에 맞는 소스만 남겨 후보가 = 가격 × (1−차감율)
        if is_new:
            table = [("internet", internet_lowest, fee)]
        else:
            table = [("internet", internet_lowest, fee), ("purchase", purchase, margin), ("selling", selling, margin)]
        hidden = hidden or set()
        # 후보가는 10원 단위 반올림(IMK #11) — 목표가·산정내역·자동채움이 다 이 값으로 일치.
        return [
            (basis, round(int(price) * (1 - (rate or 0.0)) / 10) * 10)
            for basis, price, rate in table
            if price and PricingMixin._SOURCE_HIDDEN_FIELD[basis] not in hidden
        ]

    @staticmethod
    def _calc_target_price(md_price=None, internet_lowest=None, purchase=None, selling=None, fee=0.0, margin=0.0, is_new=False, hidden=None) -> int:
        """세션에 박을 목표가를 확정한다: 구매담당자 제시가가 있으면 그 값 그대로, 없으면 후보 중 가장 싼 값.

        차감율(fee/margin)이 1 이상이면 목표가가 0이나 음수가 되므로 설정 오류로 막는다.
        후보를 하나도 못 만들면 ValueError — 이 견적은 생성 자체가 불가능하다."""
        if not md_price:
            if not 0.0 <= (fee or 0.0) < 1.0:
                raise ValueError(f"인터넷 수수료율은 0 이상 1 미만이어야 합니다: fee={fee}")
            if not is_new and not 0.0 <= (margin or 0.0) < 1.0:
                raise ValueError(f"목표 마진율은 0 이상 1 미만이어야 합니다: margin={margin}")
        cands = PricingMixin._candidates(md_price, internet_lowest, purchase, selling, fee, margin, is_new, hidden)
        if not cands:
            raise ValueError("목표가 계산 불가: 구매담당자 제시가·인터넷최저가" + ("" if is_new else "·매입가·판매가") + " 모두 없음")
        return int(min(v for _, v in cands))

    async def _load_target_inputs(
        self, item_ids: list[uuid.UUID], qt_setting_id, user_id
    ) -> tuple[dict, float, float, set]:
        """목표가 계산에 필요한 값들을 한 번에 모아온다.

        - prices: 상품마다 (인터넷최저가, 매입가, 판매가) — DB 조회
        - margin: 판매가에서 깎을 목표마진율 — 견적 세팅(quotation_settings) 조회
        - fee: 인터넷최저가에서 깎을 수수료율 — DB 아님, 고정 상수 INTERNET_AVERAGE_FEE
        - hidden: 회사 설정(companies.settings)의 숨김 가격 필드 — 숨긴 가격은 목표가 후보에서도 뺀다

        견적 생성(_build_quotation)과 산정내역 화면(get_target_breakdown)이 똑같이 이 함수를 쓴다 —
        그래야 만들 때 계산한 목표가와 화면에 보여주는 근거가 어긋나지 않는다."""
        prices = {}
        if item_ids:
            _err, prices = await DB_SESSION_MNG.execute_lambda(
                quotations.DBType(),
                DBWRType.DB_READ.value,
                lambda s: self.quotation_crud.get_item_prices(s, item_ids),
            )
            prices = prices if _err == ErrorType.SUCCESS else {}
        _err, rates = await DB_SESSION_MNG.execute_lambda(
            quotations.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.get_setting_rates(s, qt_setting_id),
        )
        rates = rates if _err == ErrorType.SUCCESS else {}
        fee = self.INTERNET_AVERAGE_FEE        # 인터넷가 차감 수수료율(상수)
        margin = rates.get("margin") or 0.0    # 판매가 차감 목표마진율
        user_uuid = uuid.UUID(user_id) if isinstance(user_id, str) else user_id
        settings = await DB_SESSION_MNG.execute_lambda(
            quotations.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.get_company_settings(s, user_uuid),
        )
        hidden = set(settings.get("hidden_fields") or [])
        return prices, fee, margin, hidden

    def _resolve_target_prices(
        self, *, qt_id, item_ids: list[uuid.UUID], prices: dict,
        md_price, fee, margin, is_new: bool, hidden, inherited_target_prices: Optional[dict],
    ) -> dict[uuid.UUID, int]:
        """상품마다 목표가를 정한다.

        재생성 라운드는 직전 라운드의 목표가를 그대로 물려받는다(KTC 규칙).
        그 외에는 인터넷최저가·매입가·판매가로 만든 후보 중 가장 싼 값을 목표가로 쓴다.
        후보를 하나도 못 만드는 상품이 있으면 ValueError — 호출한 쪽이 견적 생성 실패로 처리한다."""
        target_prices = {}
        for iid in item_ids:
            if inherited_target_prices and iid in inherited_target_prices:
                target_prices[iid] = inherited_target_prices[iid]
                continue
            internet, purchase, selling = prices.get(iid) or (None, None, None)
            try:
                target_prices[iid] = self._calc_target_price(md_price, internet, purchase, selling, fee, margin, is_new=is_new, hidden=hidden)
            except ValueError as ex:
                LOG.w(
                    f"[목표가 산정불가] qt_id={qt_id} item={iid} is_new={is_new} "
                    f"md={md_price} internet={internet} purchase={purchase} selling={selling} :: {ex}"
                )
                raise
        return target_prices

    async def _resolve_anchors(
        self, item_ids: list[uuid.UUID], supplier_ids: list[uuid.UUID], target_prices: dict[uuid.UUID, int]
    ) -> dict[tuple[uuid.UUID, uuid.UUID], tuple[int, int]]:
        """(상품×공급사) 조합마다 앵커링가를 계산한다(앵커링 v1.2, 인수인계.md §1.3).
        앵커링 값은 사용자 입력이 아니라, 칸(상품의 회사 × 공급유형 × 가격구간)마다
        배치(schedules/anchoring)가 조정해 둔 현재값을 읽어 쓴다.
        칸 값을 못 찾으면(공급유형 미지정·조정 이력 없음·조회 실패) 기본 시작값 테이블로 폴백 —
        반환: {(item_id, supplier_id): (anchoring_value ‰, anchoring_price 원)}"""
        _err, item_companies = await DB_SESSION_MNG.execute_lambda(
            quotations.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.get_item_companies(s, item_ids),
        )
        item_companies = item_companies if _err == ErrorType.SUCCESS else {}
        _err, supply_types = await DB_SESSION_MNG.execute_lambda(
            quotations.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.get_supply_types(s, item_ids, supplier_ids),
        )
        supply_types = supply_types if _err == ErrorType.SUCCESS else {}
        value_map = {}
        sampleable_supply_types = sorted({t for t in supply_types.values() if t in SAMPLEABLE_SUPPLIER_TYPES})
        if sampleable_supply_types and item_companies:
            value_map = await DB_SESSION_MNG.execute_lambda(
                quotations.DBType(),
                DBWRType.DB_READ.value,
                lambda s: fetch_current_values(s, list(set(item_companies.values())), sampleable_supply_types),
            )

        anchors = {}
        for iid in item_ids:
            tp = target_prices[iid]
            price_range = calc_price_range_index(tp)
            company = item_companies.get(iid)
            for sid in supplier_ids:
                supply_type = supply_types.get((iid, sid))
                value = value_map.get((company, supply_type, price_range)) if company is not None else None
                if value is None:
                    value = get_base_anchoring_value(price_range)
                ap = calc_anchoring_price(tp, value)  # 목표가×(1−value), 10원 반올림(정수 연산 — schedules 앵커와 동일)
                anchors[(iid, sid)] = (value, ap)
        return anchors

    async def get_target_breakdown(self, session_id: str, company_id=None) -> Res_TargetBreakdown:
        """목표가 모달의 산정내역 응답.

        저장된 목표가·앵커링가는 그대로 내려주고, 후보 목록은 _candidates 로 다시 계산한다.
        채택 체크 = 저장 목표가와 값이 같은 후보. 없으면(재생성 상속 등) is_inherited=True."""
        res = Res_TargetBreakdown()
        err_type, got = await DB_SESSION_MNG.execute_lambda(
            sessions.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.get_session_with_supplier(s, uuid.UUID(session_id)),
        )
        if err_type != ErrorType.SUCCESS or got is None:
            res.result.SetResult(err_type if err_type != ErrorType.SUCCESS else ErrorType.QUOTATION_NOT_FOUND)
            return res
        sess = got[0]
        err_type, quotation = await self._fetch(sess.quotation_id, company_id)
        if err_type != ErrorType.SUCCESS or quotation is None:
            res.result.SetResult(err_type if err_type != ErrorType.SUCCESS else ErrorType.QUOTATION_NOT_FOUND)
            return res

        # 산정 입력(재료)은 생성과 같은 로더를 공유 — 생성값과 표시값이 어긋나지 않는다.
        prices, fee, margin, hidden = await self._load_target_inputs(
            [sess.item_id], quotation.qt_setting_id, quotation.user_id
        )
        internet, purchase, selling = (prices or {}).get(sess.item_id) or (None, None, None)
        is_new = QuotationType.is_new(quotation.type)
        md = quotation.md_price

        # 산정내역 화면은 구매담당자 제시가(md)가 있어도 후보를 다 보여준다(생성 로직과 달리 '표시용').
        # 생성 시엔 md 가 있으면 md 하나만 쓰지만(_calc_target_price), 여기선 md + 상품후보를 함께 나열해 근거를 드러낸다.
        base_cands = self._candidates(None, internet, purchase, selling, fee, margin, is_new, hidden)
        cands = ([("md", float(int(md)))] if md else []) + base_cands
        chosen_basis = next((b for b, v in cands if int(v) == sess.target_price), None)
        is_inherited = chosen_basis is None

        res.is_new = is_new
        res.is_inherited = is_inherited
        res.md_price = int(md) if md else None
        res.internet_lowest = int(internet) if internet is not None else None
        res.purchase = int(purchase) if purchase is not None else None
        res.selling = int(selling) if selling is not None else None
        res.fee = fee
        res.margin = margin
        res.candidates = [TargetCandidate(basis=b, label=self._CANDIDATE_LABELS.get(b, b), value=int(v)) for b, v in cands]
        res.hidden_price_fields = sorted(hidden & {"internet_lowest_price", "purchase_price", "selling_price"})
        res.chosen_basis = chosen_basis
        res.target_price = sess.target_price
        res.anchoring_price = sess.anchoring_price
        res.anchoring_value = (sess.anchoring_value or 0) / 1000  # 세션 ‰ → 비율(main 프론트 '목표가×(1−값)' 표시용)
        return res