252 lines
15 KiB
Python
252 lines
15 KiB
Python
"""목표가·앵커링가 계산 + 산정내역 응답 (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 코드 ↔ 표시 라벨(산정내역 응답에서 프론트가 그대로 표기).
|
||
# *_raw = 네고율 미적용(원가) 변형 — 견적등록에서 네고를 끈 목표가를 역산해 근거를 가려낼 때 쓴다(적용/미적용은 프론트 sub 로 구분).
|
||
_CANDIDATE_LABELS = {
|
||
"md": "구매담당자 제시가", "internet": "인터넷 최저가",
|
||
"purchase": "매입가", "selling": "판매가",
|
||
"purchase_raw": "매입가", "selling_raw": "판매가",
|
||
}
|
||
|
||
# 가격 소스별로, 회사 설정 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, int | None]:
|
||
"""목표가 계산에 필요한 값들을 한 번에 모아온다.
|
||
|
||
- 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 # 판매가 차감 목표마진율
|
||
ceiling_rate = rates.get("done_ceiling_rate") # 회사 완료 상한율(‰), 미조회면 None
|
||
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, ceiling_rate
|
||
|
||
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, _ceiling_rate = 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
|
||
|
||
# 산정내역 화면은 저장된 목표가를 '가격 × 네고율(적용/미적용)'로 역산해 어느 근거인지 스스로 가려낸다.
|
||
# 생성 로직(_calc_target_price)은 네고 적용 후보의 최저만 쓰지만, 견적등록에서 네고를 끄면 목표가가
|
||
# 매입가/판매가 '원가'가 된다. 그래서 원가(미적용) 변형까지 만들어 저장 목표가와 대조하고,
|
||
# 어느 후보에도 안 맞을 때만 구매담당자 제시가(md)로 간주한다.
|
||
target = int(sess.target_price)
|
||
base_cands = self._candidates(None, internet, purchase, selling, fee, margin, is_new, hidden)
|
||
raw_variants = {} # {소스 basis: (원가 basis, 원가값)} — 재견적의 매입가·판매가 네고 미적용(10원 반올림)
|
||
if not is_new:
|
||
if purchase and self._SOURCE_HIDDEN_FIELD["purchase"] not in hidden:
|
||
raw_variants["purchase"] = ("purchase_raw", round(int(purchase) / 10) * 10)
|
||
if selling and self._SOURCE_HIDDEN_FIELD["selling"] not in hidden:
|
||
raw_variants["selling"] = ("selling_raw", round(int(selling) / 10) * 10)
|
||
# 채택 근거: 네고 적용 후보 → 원가(미적용) 변형 → (없으면) 구매담당자 제시가 순으로 목표가와 대조.
|
||
chosen_basis = next((b for b, v in base_cands if int(v) == target), None)
|
||
if chosen_basis is None:
|
||
chosen_basis = next((rb for _s, (rb, rv) in raw_variants.items() if rv == target), None)
|
||
if chosen_basis is None and md:
|
||
chosen_basis = "md"
|
||
# 화면 후보 — 소스별 1행. 채택이 원가 변형이면 그 소스를 원가로 바꿔 '네고율 미적용'으로 표기.
|
||
# 어느 상품후보에도 안 맞아 md 로 간주된 경우에만 구매담당자 제시가 행을 앞세운다.
|
||
cands = [("md", float(int(md)))] if md and chosen_basis == "md" else []
|
||
for b, v in base_cands:
|
||
raw = raw_variants.get(b)
|
||
cands.append(raw if raw and raw[0] == chosen_basis else (b, int(v)))
|
||
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
|