o2o-negosium-original/negodata/backend/services/quotation_service.py

import re
import uuid
from datetime import timezone, timedelta
from typing import Optional

from fastapi import Depends

from common.anchoring import (
    SAMPLEABLE_SUPPLIER_TYPES,
    calc_anchoring_price,
    calc_price_range_index,
    fetch_current_values,
    get_base_anchoring_value,
)
from common.authz import is_owner_or_admin
from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import quotations, sessions, chats, versions, version_nego_cards, version_wild_cards
from common.enums import CloseOutcome, CloseReason, DBWRType, ErrorType, NotificationType, PriceGateAction, QuotationStatus, QuotationType, SessionStatus
from common.logger import LOG
from common.models.gmodel import PageParams
from common.utils.gtime import GTime
from config.server_configs import web_server_config
from crud.quotation_crud import IQuotationCRUD, QuotationCRUD
from router.v1.quotation.protocol import (
    ChatMessageData,
    QuotationCardData,
    QuotationData,
    SessionData,
    Req_CreateQuotation,
    Res_CreateQuotation,
    Res_DeleteQuotation,
    Res_NotifySessions,
    Res_Quotation,
    Res_QuotationCards,
    Res_QuotationList,
    Res_QuotationResult,
    Res_QuotationSessions,
    Res_QuotationStatus,
    Res_SessionChat,
    Res_TargetBreakdown,
    TargetCandidate,
)
from services.email import EmailUnavailable, build_invite_email, send_email
from services.notification import create_notification


class QuotationService:
    """견적 비즈니스 로직.

    회사 스코프(멀티테넌트)는 작성자(user_id)→users.company_id 조인으로 건다(quotations 에 company_id 컬럼이 없음).
    목록(list_quotations)은 회사 스코프로 제한한다. user_id 는 '내 견적만' 추가 필터로도 쓴다.
    """

    # 기본 전략 버전(card.versions 시드). 견적 생성 시 version_id 미지정이면 이 값으로 채운다.
    DEFAULT_VERSION_ID = uuid.UUID("00000000-0000-0000-0000-000000000030")

    # 재생성 라운드의 최소 협상기간(방어적 하한). 원본 협상기간이 비정상적으로 짧으면(또는 0/음수면)
    # 새 라운드가 생성 즉시 만료돼 다음 크론 tick(*/5분)에 또 마감되는 연쇄를 막는다.
    # 정상 견적(수 시간~수일)은 원본 기간을 그대로 쓰며, 이 하한은 비정상적으로 짧은 경우에만 적용된다.
    # TODO 하한값 변경 해야함 !!! feat. MarineYang
    MIN_REGEN_DURATION = timedelta(hours=1)

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

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

    def __init__(self, quotation_crud: IQuotationCRUD = Depends(QuotationCRUD)):
        self.quotation_crud = quotation_crud

    @staticmethod
    def _session_chat_url(session_id) -> str:
        """세션 chat 실행 URL(공급사 협상 프론트). ChatPage 가 session_id 쿼리로 진입한다."""
        base = (web_server_config.nego_chat_url or "").rstrip("/")
        return f"{base}/chat?session_id={session_id}"

    @staticmethod
    def _candidates(md_price=None, internet_lowest=None, purchase=None, selling=None, fee=0.0, margin=0.0, is_new=False, mode=None, hidden=None):
        """목표가 후보 [(basis, value_float)] 목록(빈 값/0 은 제외). md 있으면 md 단독.
        값은 float(인터넷=가격×(1−수수료), 판매가=가격×(1−마진))이며 채택 시 int() 절삭한다.
        _calc_target_price(생성)와 get_target_breakdown(표시)가 공유하는 단일 산정 로직.

        mode='purchase' (회사 설정 features.target_price_mode) 면 매입가만 후보로 쓴다 —
        인터넷최저가·판매가는 신규/재 구분 없이 제외하고 매입가 × (1 − 네고율) 하나로 잡는다(IMK #10).
        hidden (회사 설정 hidden_fields) 에 든 가격 필드는 후보에서 뺀다 — 화면에서 감춘 값이
        목표가를 결정하면 담당자가 산정 근거를 확인할 수 없기 때문."""
        if md_price:
            return [("md", float(int(md_price)))]
        hidden = hidden or set()
        if "internet_lowest_price" in hidden:
            internet_lowest = None
        if "purchase_price" in hidden:
            purchase = None
        if "selling_price" in hidden:
            selling = None
        if mode == "purchase":
            return [("purchase", int(purchase) * (1 - (margin or 0.0)))] if purchase else []
        out = []
        if internet_lowest:
            out.append(("internet", int(internet_lowest) * (1 - (fee or 0.0))))
        if not is_new:  # 재(협상·견적)만 매입가·판매가를 후보에 추가. 신규는 인터넷최저가만.
            if purchase:
                out.append(("purchase", float(int(purchase))))
            if selling:
                out.append(("selling", int(selling) * (1 - (margin or 0.0))))
        return out

    @staticmethod
    def _calc_target_price(md_price=None, internet_lowest=None, purchase=None, selling=None, fee=0.0, margin=0.0, is_new=False, mode=None, hidden=None) -> int:
        """세션 목표가 (KTC 신규/재 분리 로직, 회사 데이터 풍부도에 graceful 적응)
          ① md_price 있으면 → 그대로
          ② 없으면:
               · 신규(NEW_NEGO/NEW_QUOTE) → 인터넷최저가 × (1 − fee)  [인터넷최저가만]
               · 재(RENEGO/REQUOTE)       → 유효 후보 중 min:
                    - 인터넷최저가 × (1 − fee)   ← fee=quotation_settings.internet_average_fee
                    - 매입가 (그대로)
                    - 판매가 × (1 − margin)      ← margin=quotation_settings.target_margin_rate
          ③ 후보 0개 → 견적 생성 불가(ValueError).
        mode='purchase' 면 ②를 무시하고 매입가 × (1 − 네고율) 하나만 후보로 쓴다(IMK #10)."""
        if not md_price:
            # 율은 비율(0~1 미만)이어야 한다. 1 이상이면 (1−율)≤0 → 목표가가 0/음수가 되므로 설정 오류로 막는다.
            if not 0.0 <= (fee or 0.0) < 1.0:
                raise ValueError(f"인터넷 수수료율은 0 이상 1 미만이어야 합니다: fee={fee}")
            if (mode == "purchase" or not is_new) and not 0.0 <= (margin or 0.0) < 1.0:
                raise ValueError(f"목표 마진율은 0 이상 1 미만이어야 합니다: margin={margin}")
        cands = QuotationService._candidates(md_price, internet_lowest, purchase, selling, fee, margin, is_new, mode, hidden)
        if not cands:
            if mode == "purchase":
                raise ValueError("타겟 가격 계산 불가: md_price·매입가 모두 없음")
            raise ValueError("타겟 가격 계산 불가: md_price·인터넷최저가" + ("" if is_new else "·매입가·판매가") + " 모두 없음")
        return int(min(v for _, v in cands))

    @staticmethod
    def _gen_number() -> str:
        """견적번호 자동 생성(미지정 시). EST-YYYYMM-XXXX."""
        now = GTime.UTC()
        return f"EST-{now:%Y%m}-{uuid.uuid4().hex[:4].upper()}"

    async def _fetch(self, qt_id: uuid.UUID, company_id=None):
        """견적 단건 조회. (ErrorType, quotation|None) 반환.
        company_id 가 주어지면 회사 스코프(작성자 회사) 가드 — 남의 회사 견적은 NOT_FOUND. 내부/스케줄러 호출은 None."""
        err_type, quotation = await DB_SESSION_MNG.execute_lambda(
            quotations.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.get_by_id(s, qt_id, company_id),
        )
        if err_type != ErrorType.SUCCESS or quotation is None:
            return ErrorType.QUOTATION_NOT_FOUND, None
        return ErrorType.SUCCESS, quotation

    async def get_target_breakdown(self, session_id: str, company_id=None) -> Res_TargetBreakdown:
        """세션 목표가 산정내역(후보·채택). 저장된 target_price/anchoring 은 그대로 표기하고,
        후보값은 생성과 동일한 _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:
            res.result.SetResult(err_type)
            return res

        _e, prices = await DB_SESSION_MNG.execute_lambda(
            quotations.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.get_item_prices(s, [sess.item_id]),
        )
        internet, purchase, selling = (prices or {}).get(sess.item_id) or (None, None, None)
        _e, rates = await DB_SESSION_MNG.execute_lambda(
            quotations.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.get_setting_rates(s, quotation.qt_setting_id),
        )
        rates = rates or {}
        fee = self.INTERNET_AVERAGE_FEE
        margin = rates.get("margin") or 0.0
        is_new = QuotationType.is_new(quotation.type)
        md = quotation.md_price
        mode, hidden = await DB_SESSION_MNG.execute_lambda(
            quotations.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.get_target_price_mode(s, quotation.user_id),
        )

        cands = self._candidates(md, internet, purchase, selling, fee, margin, is_new, mode, hidden)
        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.target_price_mode = mode
        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

    async def list_quotations(self, company_id, owner, search, status, type_, start_from, start_to, pg: PageParams) -> Res_QuotationList:
        """견적 목록. 회사(company_id) 스코프로 제한하고, owner(user_id) 가 주어지면 '내 견적만'으로 더 좁힌다."""
        res = Res_QuotationList(page=pg.page, size=pg.size)

        company_uuid = uuid.UUID(company_id)
        owner_uuid = uuid.UUID(owner) if owner else None
        err_type, rows, total = await DB_SESSION_MNG.execute_lambda(
            quotations.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.search(s, company_uuid, owner_uuid, search, status, type_, start_from, start_to, pg.skip, pg.size),
        )
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        # 참여 협력사 수(세션 distinct supplier)·대표 상품(세션 item)·작성자명(user→users.name)을
        # 이 페이지 견적들에 대해 각각 한 방으로 모아 합친다(메인 쿼리 비건드림).
        qt_ids = [r.qt_id for r in rows]
        counts = {}
        item_map = {}
        name_map = {}
        if qt_ids:
            cnt_err, got = await DB_SESSION_MNG.execute_lambda(
                quotations.DBType(),
                DBWRType.DB_READ.value,
                lambda s: self.quotation_crud.session_counts(s, qt_ids),
            )
            if cnt_err == ErrorType.SUCCESS:
                counts = got
            im_err, got_im = await DB_SESSION_MNG.execute_lambda(
                quotations.DBType(),
                DBWRType.DB_READ.value,
                lambda s: self.quotation_crud.item_map(s, qt_ids),
            )
            if im_err == ErrorType.SUCCESS:
                item_map = got_im
            user_ids = list({r.user_id for r in rows})
            nm_err, got_nm = await DB_SESSION_MNG.execute_lambda(
                quotations.DBType(),
                DBWRType.DB_READ.value,
                lambda s: self.quotation_crud.user_name_map(s, user_ids),
            )
            if nm_err == ErrorType.SUCCESS:
                name_map = got_nm
        for r in rows:
            r.participation_count = counts.get(r.qt_id, 0)
            item = item_map.get(r.qt_id)
            if item:
                r.item_id, r.item_name = item
            r.creator_name = name_map.get(r.user_id)

        res.quotations = [QuotationData.model_validate(r) for r in rows]
        res.total = total
        return res

    async def get_quotation(self, qt_id: str, company_id=None) -> Res_Quotation:
        res = Res_Quotation()
        err_type, quotation = await self._fetch(uuid.UUID(qt_id), company_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        res.quotation = QuotationData.model_validate(quotation)
        return res

    async def create_quotation(self, user_id: str, req: Req_CreateQuotation) -> Res_CreateQuotation:
        """[프론트] 신규 견적 생성. 요청값을 보정한 뒤 공통 빌더(_build_quotation)에 위임한다.
        생성 성공 시 작성자에게 CREATED 알림(인박스)."""
        number = self._gen_number()  # 견적번호는 항상 서버 생성(프론트 입력란 없음)
        res = await self._build_quotation(
            user_id=user_id,
            qt_setting_id=req.qt_setting_id,
            version_id=req.version_id or self.DEFAULT_VERSION_ID,
            name=req.name,
            number=number,
            type_=req.type,
            status=req.status or QuotationStatus.CREATED.value,
            round_=req.round or 1,
            start_time=req.start_time or GTime.UTC(),
            end_time=req.end_time,
            manager_name=req.manager_name,
            manager_email=req.manager_email,
            manager_contact_number=req.manager_contact_number,
            memo=req.memo,
            md_price=req.md_price,
            item_ids=req.item_ids,
            supplier_ids=req.supplier_ids,
            card_ids=req.card_ids,
            mid_action=req.mid_action,
            over_action=req.over_action,
        )
        if res.result.success:
            await create_notification(
                user_id, NotificationType.CREATED,
                {"qt_name": req.name, "qt_number": number},
                ref_qt_id=res.qt_id,
            )
        return res

    async def regenerate_next_round(self, original_qt_id: uuid.UUID, supplier_ids: list) -> Res_CreateQuotation:
        """[마감 후속] 결판 안 난 견적의 '다음 라운드'를 새로 만든다.

        플로우:
          1) 원 견적 + 세션을 조회해 대상 상품(item)을 복원
          2) 타입 결정 — 다음 라운드 공급사가 1곳이면 재협상(RENEGO), 여러 곳이면 재견적(REQUOTE)
          3) 같은 견적번호 + round+1 로 다음 라운드 생성 (협상기간은 원 견적과 같은 길이)

        견적번호(number)를 원본 그대로 이어받아 '같은 번호 = 한 체인'으로 묶는다(parent_id 대체).
        supplier_ids: 다음 라운드에 부를 공급사(동가면 동가 업체만, 그 외엔 원 견적 공급사 전체).
        """
        res = Res_CreateQuotation()

        # 1) 원 견적 + 세션 조회 → 대상 상품 복원
        err_type, original = await self._fetch(original_qt_id)
        if err_type != ErrorType.SUCCESS or original is None:
            res.result.SetResult(err_type)
            return res
        err_type, rows = await DB_SESSION_MNG.execute_lambda(
            sessions.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.list_sessions(s, original_qt_id),
        )
        item_ids = list({r.item_id for r in rows}) if err_type == ErrorType.SUCCESS else []
        # 재생성은 목표가를 재계산하지 않고 직전 라운드 세션 값을 그대로 상속(KTC 방식).
        # 앵커링가는 상속하지 않는다 — 생성 시점의 칸 rate 로 항상 재계산·박제(앵커링 v1.2 인수인계 규칙 1).
        inherited = {r.item_id: r.target_price for r in rows} if err_type == ErrorType.SUCCESS else {}

        # 2) 타입 결정: 공급사 1곳 → 재협상 / 여러 곳 → 재견적
        next_type = QuotationType.RENEGO.value if len(supplier_ids) <= 1 else QuotationType.REQUOTE.value

        # 3) 다음 라운드의 견적 생성
        now = GTime.UTC()
        # 원본 협상기간을 이어쓰되, 비정상적으로 짧으면 최소 하한을 적용(즉시 만료→연쇄 재마감 방지).
        duration = max(original.end_time - original.start_time, self.MIN_REGEN_DURATION)
        # 다음 차수는 '원본 round+1' 이 아니라 '체인(같은 번호) 최신 round+1'.
        # 크론 마감과 수동 regenerate_quotation 이 같은 체인을 처리하는 타이밍이 엇갈려도
        # 항상 체인 끝에 이어붙어 uq_quotations_number(number, round) 충돌을 막는다.
        _e, chain_max = await DB_SESSION_MNG.execute_lambda(
            quotations.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.chain_max_round(s, original.number),
        )
        base_round = chain_max if (_e == ErrorType.SUCCESS and chain_max) else original.round
        next_round = base_round + 1
        # 이름에 '(N차)' 표기. 원래 이름 기준(기존 '(M차)' 표기는 떼고 새로) + name 컬럼 50자 제한 보호.
        suffix = f" ({next_round}차)"
        base_name = re.sub(r"\s*\(\d+차\)\s*$", "", original.name or "")[: 50 - len(suffix)]
        return await self._build_quotation(
            user_id=str(original.user_id),
            qt_setting_id=original.qt_setting_id,
            version_id=original.version_id,  # 카드 버전은 원본 그대로 이어씀
            name=f"{base_name}{suffix}",     # 예: "삼성 견적 (2차)"
            number=original.number,          # ← 원본 번호 따라감(새 번호 생성 X)
            type_=next_type,
            status=QuotationStatus.CREATED.value,
            round_=next_round,
            start_time=now,
            end_time=now + duration,
            manager_name=original.manager_name,
            manager_email=original.manager_email,
            manager_contact_number=original.manager_contact_number,
            memo=original.memo,
            md_price=original.md_price,
            item_ids=item_ids,
            supplier_ids=list(supplier_ids),
            card_ids=[],  # 새 버전 안 만듦(원본 version_id 재사용)
            mid_action=original.mid_action,    # 낙찰 기준 상속(타입이 REQUOTE 로 바뀌면 빌더가 AWARD 로 재정규화)
            over_action=original.over_action,
            inherited=inherited,  # 직전 라운드 목표가 상속(앵커링가는 현재 rate 로 재계산)
        )

    async def _build_quotation(
        self, *,
        user_id: str, qt_setting_id, version_id, name: str, number: str,
        type_: int, status: int, round_: int, start_time, end_time,
        manager_name, manager_email, manager_contact_number, memo, md_price,
        item_ids: list, supplier_ids: list, card_ids: list,
        mid_action: Optional[int] = None,   # 낙찰 기준(견적 단위). 앵커링가<투찰가≤목표가 처리(AWARD/OPEN)
        over_action: Optional[int] = None,  # 목표가<투찰가 처리(1:1 협상은 항상 OPEN)
        inherited: Optional[dict] = None,  # 재생성 시 {item_id: target_price} 상속(KTC) — 목표가만. 앵커는 항상 재계산
    ) -> Res_CreateQuotation:
        """견적 1건 + (상품×공급사) 세션들을 한 트랜잭션으로 생성하는 공통 빌더."""
        res = Res_CreateQuotation()

        # 낙찰 기준 정규화 — 1:N 경매는 항상 최저가 낙찰(mid=over=AWARD 강제). 1:1 협상은 요청값(미지정=AWARD).
        # create/regenerate 양 경로가 이 빌더를 타므로 불변식을 여기 한 곳에서 강제한다(재생성 시 타입 전환도 자동 재정규화).
        if QuotationType.is_auction(type_):
            mid_action = over_action = PriceGateAction.AWARD.value
        else:
            mid_action = mid_action or PriceGateAction.AWARD.value
            over_action = over_action or PriceGateAction.AWARD.value

        # 세션 목표가 입력(상품별 인터넷최저가/매입가/판매가 + 세팅 율). 읽기 트랜잭션에서 먼저 조회.
        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    # 판매가 차감 목표마진율
        # 앵커링가는 quotation_settings.anchoring_value(구 float 비율)를 더 이상 쓰지 않는다(앵커링 v1.2) —
        # 칸(회사×상품-협력사 공급유형×가격구간)별 조정 anchoring_value(정수 ‰)로 계산한다. 아래 세션 생성부 ②.

        # 선택 협상카드가 있으면 새 버전을 만들어 카드들을 묶고, quotation.version_id 로 연결한다.
        # (quotation↔card 는 version → version_nego_cards/version_wild_cards 로 연결.)
        version_obj = None
        link_rows = []
        if card_ids:
            _err, card_types = await DB_SESSION_MNG.execute_lambda(
                quotations.DBType(),
                DBWRType.DB_READ.value,
                lambda s: self.quotation_crud.classify_card_ids(s, card_ids),
            )
            card_types = card_types if _err == ErrorType.SUCCESS else {}
            new_version_id = uuid.uuid4()
            version_obj = versions(
                version_id=new_version_id,
                user_id=uuid.UUID(user_id),
                code=0,
                name=(number or "견적버전")[:10],
            )
            for cid in card_ids:
                t = card_types.get(cid)
                if t == 1:
                    link_rows.append(version_nego_cards(version_id=new_version_id, nego_card_id=cid))
                elif t == 2:
                    link_rows.append(version_wild_cards(version_id=new_version_id, wild_card_id=cid))
            version_id = new_version_id

        # qt_id 를 미리 발급해 세션 FK(quotation_id)와 묶고, 한 트랜잭션에 함께 insert 한다.
        qt_id = uuid.uuid4()
        quotation = quotations(
            qt_id=qt_id,
            user_id=uuid.UUID(user_id),
            qt_setting_id=qt_setting_id,
            version_id=version_id,
            name=name,
            number=number,
            type=type_,
            status=status,
            round=round_,
            start_time=start_time,
            end_time=end_time,
            manager_name=manager_name,
            manager_email=manager_email,
            manager_contact_number=manager_contact_number,
            memo=memo,
            md_price=md_price,
            mid_action=mid_action,
            over_action=over_action,
        )

        # 상품 × 공급사 조합마다 세션 1개. md/매입/판매/인터넷 후보가 하나도 없으면 목표가 산정 불가 → 생성 실패.
        # 신규(NEW_NEGO/NEW_QUOTE)는 인터넷최저가만, 재(RENEGO/REQUOTE)는 매입가·판매가까지 후보(KTC 신규/재 분리).
        is_new = QuotationType.is_new(type_)

        # 회사별 목표가 산정 모드(features.target_price_mode). 'purchase' 면 매입가 × 네고율만 후보(IMK #10).
        mode, hidden = await DB_SESSION_MNG.execute_lambda(
            quotations.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.get_target_price_mode(s, uuid.UUID(user_id)),
        )

        # ① 목표가 산정 — 재생성(inherited)은 직전 라운드 값 그대로 상속(KTC), 그 외엔 후보 min.
        target_prices = {}
        try:
            for iid in item_ids:
                if inherited and iid in inherited:
                    target_prices[iid] = inherited[iid]
                else:
                    internet, purchase, selling = prices.get(iid) or (None, None, None)
                    target_prices[iid] = self._calc_target_price(md_price, internet, purchase, selling, fee, margin, is_new=is_new, mode=mode, 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}"
            )
            res.result.SetResult(ErrorType.QUOTATION_TARGET_PRICE_UNAVAILABLE)
            return res

        # ② 앵커가 산출 — 칸(items.company_id × supplier_items.supply_type × 목표가 구간) anchoring_value 조회 후
        #    정수 연산으로 박제(앵커링 v1.2, 인수인계.md §1.3). 매핑 미지정/조정 이력 없음/조회 실패는
        #    정적 테이블 시작값 폴백 — 값 조회 때문에 견적 생성이 실패하지 않는다(규칙 6).
        _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),
            )

        session_objs = []
        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)  # 목표가×(1000−value)//1000 — float 곱셈 금지(1원 내림 정확성)
                session_objs.append(
                    sessions(
                        session_id=uuid.uuid4(),
                        quotation_id=qt_id,
                        item_id=iid,
                        supplier_id=sid,
                        qt_number=quotation.number,
                        qt_round=quotation.round,
                        qt_type=quotation.type,
                        target_price=tp,
                        anchoring_price=ap,   # 박제 — 이후 수정 금지(협상 판정·앵커링 학습 기준값)
                        anchoring_value=value,
                        status=SessionStatus.CREATED.value,
                        end_time=quotation.end_time,
                    )
                )

        # 버전 → (버전-카드 매핑) → 견적 → 세션 순으로 한 트랜잭션에 insert(FK 순서 보장).
        ops = []
        if version_obj is not None:
            ops.append(lambda s: self.quotation_crud.add_rows(s, [version_obj]))
            ops.append(lambda s: self.quotation_crud.add_rows(s, link_rows))
        ops.append(lambda s: self.quotation_crud.add_quotation(s, quotation))
        ops.append(lambda s: self.quotation_crud.add_sessions(s, session_objs))
        err_type = await DB_SESSION_MNG.execute_lambda_run([quotations.DBType()], ops)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        # 프론트는 생성 응답 본문을 화면에 안 쓰고 qt_id 로 재조회 → 새 id 와 세션 수만 반환.
        res.qt_id = qt_id
        res.session_count = len(session_objs)
        return res

    # ----- 마감 판정
    @staticmethod
    def _pick_winner(done_rows) -> tuple[Optional[dict], Optional[dict]]:
        """협상완료 세션들 중 낙찰자 판정. done_rows: [(supplier_id, bid_price, supplier_name), ...].
        입찰가가 매겨진 세션 중 최저가가 단독이면 그 공급사를 낙찰로, 동가(둘+)면 낙찰은 비우고 동가 정보만 남긴다.
        단독/동가는 상호배타. 반환: (winner|None, equal|None)."""
        cands = [(sid, int(bp), name) for sid, bp, name in done_rows if bp is not None]
        if not cands:
            return None, None
        min_price = min(c[1] for c in cands)
        tied = [c for c in cands if c[1] == min_price]
        if len(tied) > 1:
            equal = {"price": min_price, "suppliers": [{"supplier_id": str(sid), "name": name} for sid, _, name in tied]}
            return None, equal
        return {"supplier_id": tied[0][0], "name": tied[0][2], "bid_price": min_price}, None

    @staticmethod
    def _gate_action(bid, target, anchor, mid_action, over_action) -> int:
        """가격게이트 판정 → PriceGateAction 코드(AWARD=낙찰 / OPEN=개찰).
          bid ≤ 앵커링가        → 무조건 낙찰(AWARD)
          앵커링가 < bid ≤ 목표가 → mid_action(견적 낙찰 기준)
          목표가 < bid          → over_action(1:1 협상은 항상 OPEN=개찰)
        target/bid 없으면(산정 불가 등) AWARD 폴백(최저가 그대로 낙찰)."""
        if bid is None or target is None:
            return PriceGateAction.AWARD.value
        bid = int(bid)
        if anchor is not None and bid <= int(anchor):
            return PriceGateAction.AWARD.value
        if bid <= int(target):
            return mid_action or PriceGateAction.AWARD.value
        return over_action or PriceGateAction.AWARD.value

    async def _close(self, qt_uuid, close_reason: int, data: Optional[dict] = None) -> None:
        """마감 공통: status→CLOSED + close_reason 기록 + (있으면)추가데이터 + 미완료(미시작·진행중) 세션→미참여.
        close_reason(CloseReason)이 낙찰/개찰 사유 구분의 단일 근거.
        preferred_sp_*/equal_bid_* 는 프론트 표시용으로 함께 채운다(사유 판별은 close_reason 이 담당)."""
        payload = {"status": QuotationStatus.CLOSED.value, "close_reason": close_reason}
        if data:
            payload.update(data)
        await DB_SESSION_MNG.execute_lambda_run(
            [quotations.DBType()],
            [
                lambda s: self.quotation_crud.update_quotation(s, qt_uuid, payload),
                lambda s: self.quotation_crud.update_sessions_status(
                    s, qt_uuid, [SessionStatus.CREATED.value, SessionStatus.IN_PROGRESS.value], SessionStatus.NOT_PARTICIPATED.value
                ),
            ],
        )

    async def _open(self, qt_uuid, original, close_reason: int, reason: str, data: Optional[dict] = None) -> CloseOutcome:
        """개찰 마감 — 낙찰자 미정으로 CLOSED + close_reason(OPEN_*) 기록 + 작성자 알림. 자동 재협상/재생성 없음(담당자 수동 처리).
        결렬(유찰) 아님. 알림 코드는 유지하되 프론트에서 '개찰'로 표기한다."""
        await self._close(qt_uuid, close_reason, data)
        await create_notification(
            original.user_id, NotificationType.FAILURE,
            {"qt_name": original.name, "qt_number": original.number, "reason": reason},
            ref_qt_id=qt_uuid,
        )
        return CloseOutcome.OPENED

    async def close_and_decide(self, qt_id) -> CloseOutcome:
        """[마감] 견적을 마감하며 결과 판정. 협상완료 단독 최저가가 낙찰 기준(가격게이트)을 통과할 때만 낙찰(AWARDED).
          - 단독 최저가: ≤앵커 항상 낙찰 / 앵커~목표 mid_action / 목표초과 over_action(1:1 협상은 항상 OPEN=개찰).
          - 그 외(기준 미달·동가·협상거부·전원 미응찰)는 결렬(유찰)이 아니라 개찰(OPEN_*) — 낙찰자 미정으로 마감.
            자동 재협상/재생성 없음. 다음 라운드는 담당자가 상세에서 수동 재생성(regenerate_quotation)한다.
        공통: 원자적 status→CLOSED 선점, 미시작·진행중 세션→미참여."""
        qt_uuid = qt_id if isinstance(qt_id, uuid.UUID) else uuid.UUID(str(qt_id))

        err_type, original = await self._fetch(qt_uuid)
        if err_type != ErrorType.SUCCESS or original is None:
            return CloseOutcome.CLOSED

        # [동시 마감 가드] 원자적으로 status→CLOSED 선점. 실제로 전이한 호출자만 통과.
        claim_err, claimed = await DB_SESSION_MNG.execute_lambda_claim(
            quotations.DBType(),
            lambda s: self.quotation_crud.claim_for_close(s, qt_uuid),
        )
        if claim_err != ErrorType.SUCCESS or claimed == 0:
            return CloseOutcome.CLOSED

        err_type, rows = await DB_SESSION_MNG.execute_lambda(
            sessions.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.list_sessions_status(s, qt_uuid),
        )
        rows = rows if err_type == ErrorType.SUCCESS else []

        done = [(r.supplier_id, r.bid_price, r.name) for r in rows if r.status == SessionStatus.DONE.value]
        has_rejected = any(r.status == SessionStatus.REJECTED.value for r in rows)
        winner, equal = self._pick_winner(done)

        # 가격게이트 입력: 견적 단위 낙찰 기준(mid/over, 생성 시점 박제) + 세션 목표가/앵커링가(견적당 상품 1개라 세션 공통값)
        mid_action = original.mid_action or PriceGateAction.AWARD.value
        over_action = original.over_action or PriceGateAction.AWARD.value
        target = next((r.target_price for r in rows if r.target_price is not None), None)
        anchor = next((r.anchoring_price for r in rows if r.anchoring_price is not None), None)

        # 1) 단독 최저가가 낙찰 기준 통과 → 낙찰. 미달 → 개찰(가격).
        if winner is not None:
            action = self._gate_action(winner["bid_price"], target, anchor, mid_action, over_action)
            if action == PriceGateAction.AWARD.value:
                await self._close(qt_uuid, CloseReason.AWARDED.value, {
                    "preferred_sp_yn": True, "preferred_sp_id": winner["supplier_id"],
                    "preferred_sp_name": (winner["name"] or "")[:20], "equal_bid_yn": False,
                })
                await create_notification(
                    original.user_id, NotificationType.SUCCESS,
                    {"qt_name": original.name, "qt_number": original.number,
                     "winner_name": winner["name"], "winner_price": winner["bid_price"]},
                    ref_qt_id=qt_uuid,
                )
                return CloseOutcome.AWARDED
            # 개찰(가격) — 낙찰/동가 플래그는 NULL 로 둔다(대시보드 '개찰' 스코프가 preferred/equal 둘 다 NULL 로 집계).
            return await self._open(qt_uuid, original, CloseReason.OPEN_PRICE.value, "price")

        # 2) 동가(최저가 동점) → 개찰(동가). 낙찰자 미정. equal_bid_yn 으로 표기(대시보드 '동가' 스코프).
        if equal is not None:
            return await self._open(qt_uuid, original, CloseReason.OPEN_EQUAL.value, "equal",
                                    {"equal_bid_yn": True, "equal_bid_data": equal})

        # 3) 협상거부 있음 → 개찰(거부).
        if has_rejected:
            return await self._open(qt_uuid, original, CloseReason.OPEN_REJECT.value, "rejected")

        # 4) 전원 미응찰 → 개찰(미응찰).
        return await self._open(qt_uuid, original, CloseReason.OPEN_NOSHOW.value, "no_show")

    async def regenerate_quotation(self, qt_id: str, company_id, supplier_ids: list, user_id=None, role=None) -> Res_CreateQuotation:
        """[프론트] 마감된 견적을 골라 수동으로 다음 라운드를 생성한다.
        크론/수동마감의 자동 재생성과 달리 사유·체인 한도 판정 없이, 프론트가 고른 공급사로 바로 만든다.
        상품·기간·견적번호·카드버전은 원 견적에서 이어받는다(regenerate_next_round)."""
        res = Res_CreateQuotation()
        qt_uuid = uuid.UUID(qt_id)

        err_type, original = await self._fetch(qt_uuid, company_id)
        if err_type != ErrorType.SUCCESS or original is None:
            res.result.SetResult(err_type)
            return res
        # 소유자 게이팅 — 본인 견적 또는 최고관리자만 재생성(user_id 미지정=내부 호출은 스킵).
        if user_id is not None and not is_owner_or_admin(original.user_id, user_id, role):
            res.result.SetResult(ErrorType.ACCOUNT_FORBIDDEN)
            res.msg = "본인이 생성한 견적만 재생성할 수 있습니다."
            return res
        # 마감된 견적만 재생성(진행 중인 라운드를 또 찍어 같은 번호가 동시에 살아있는 걸 막는다).
        if original.status != QuotationStatus.CLOSED.value:
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            return res
        # 공급사 미선택이면 세션이 0건이라 의미 없음.
        if not supplier_ids:
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            return res
        # 마지막 차수에서만 재생성 — 옛 라운드/뒤 라운드 살아있는데 또 생성하는 걸 막고(uq(number,round) 충돌도 예방),
        # 마지막이 아니면 명시적 에러를 던진다.
        _e, max_round = await DB_SESSION_MNG.execute_lambda(
            quotations.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.chain_max_round(s, original.number),
        )
        if _e == ErrorType.SUCCESS and max_round and original.round < max_round:
            res.result.SetResult(ErrorType.QUOTATION_NOT_LATEST_ROUND)
            res.msg = "마지막 차수의 견적에서만 다음 라운드를 생성할 수 있습니다."
            return res

        return await self.regenerate_next_round(qt_uuid, supplier_ids)

    async def stop_quotation(self, qt_id: str, company_id=None, user_id=None, role=None) -> Res_Quotation:
        """[프론트] 수동 견적마감. 크론과 똑같은 마감 판정(close_and_decide)을 탄다
        (단독낙찰 확정 / 동가·미참여면 다음 라운드 재생성 / 거부·한도면 그냥 마감)."""
        res = Res_Quotation()
        qt_uuid = uuid.UUID(qt_id)

        # 존재 확인(+회사 가드)
        err_type, original = await self._fetch(qt_uuid, company_id)
        if err_type != ErrorType.SUCCESS or original is None:
            res.result.SetResult(err_type)
            return res
        # 소유자 게이팅 — 본인 견적 또는 최고관리자만 마감(user_id 미지정=내부 호출은 스킵).
        if user_id is not None and not is_owner_or_admin(original.user_id, user_id, role):
            res.result.SetResult(ErrorType.ACCOUNT_FORBIDDEN)
            res.msg = "본인이 생성한 견적만 마감할 수 있습니다."
            return res

        await self.close_and_decide(qt_uuid)
        return await self.get_quotation(qt_id, company_id)

    async def award_quotation(self, qt_id: str, company_id, user_id, role, winner_supplier_id) -> Res_Quotation:
        """[프론트] 개찰(낙찰자 미정 마감) 견적을 담당자가 직접 낙찰 처리한다.
        투찰한 협상완료(DONE) 세션 중 고른 협력사를 낙찰자로 박고 close_reason 을 AWARDED 로 바꾼다(직접 낙찰).
        자동 낙찰(close_and_decide)과 결과 컬럼은 같되, 알림에 manual 플래그로 '직접 낙찰'임을 남긴다.
        권한: 본인이 생성한 견적만. 단 최고관리자(OWNER)는 회사 내 남의 견적도 낙찰할 수 있다."""
        res = Res_Quotation()
        qt_uuid = uuid.UUID(qt_id)
        sp_uuid = winner_supplier_id if isinstance(winner_supplier_id, uuid.UUID) else uuid.UUID(str(winner_supplier_id))

        # 존재 확인(+회사 가드) — 남의 회사 견적은 NOT_FOUND.
        err_type, original = await self._fetch(qt_uuid, company_id)
        if err_type != ErrorType.SUCCESS or original is None:
            res.result.SetResult(err_type if err_type != ErrorType.SUCCESS else ErrorType.QUOTATION_NOT_FOUND)
            return res

        # 자기 견적만 직접 낙찰 — 최고관리자(OWNER)만 회사 내 남의 견적도 허용. 되돌릴 수 없는 낙찰이라 백엔드에서 강제한다.
        if not is_owner_or_admin(original.user_id, user_id, role):
            res.result.SetResult(ErrorType.ACCOUNT_FORBIDDEN)
            res.msg = "본인이 생성한 견적만 낙찰할 수 있습니다."
            return res

        # 개찰(마감·낙찰자 미정, close_reason ∈ OPEN_*)만 직접 낙찰 대상. 진행중/이미 낙찰은 거부.
        if original.status != QuotationStatus.CLOSED.value or original.close_reason not in (
            CloseReason.OPEN_PRICE.value, CloseReason.OPEN_EQUAL.value,
            CloseReason.OPEN_NOSHOW.value, CloseReason.OPEN_REJECT.value,
        ):
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            res.msg = "개찰(낙찰자 미정) 상태의 견적만 직접 낙찰할 수 있습니다."
            return res

        # 낙찰 후보 = 투찰한 협상완료(DONE) 세션. close_and_decide 와 같은 조회(list_sessions_status,
        # 공급사 삭제돼도 포함되는 outerjoin)를 써서 자동낙찰과 후보 집합을 일치시킨다.
        err_type, rows = await DB_SESSION_MNG.execute_lambda(
            sessions.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.list_sessions_status(s, qt_uuid),
        )
        rows = rows if err_type == ErrorType.SUCCESS else []
        winner = next(
            (r for r in rows
             if r.status == SessionStatus.DONE.value and r.bid_price is not None and r.supplier_id == sp_uuid),
            None,
        )
        if winner is None:
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            res.msg = "선택한 협력사는 이 견적의 낙찰 후보(투찰한 협상완료 협력사)가 아닙니다."
            return res

        # [동시 직접낙찰 가드] 개찰→낙찰 원자 선점. 실제로 전이한 호출자만 통과(재클릭·경합 방어).
        claim_err, claimed = await DB_SESSION_MNG.execute_lambda_claim(
            quotations.DBType(),
            lambda s: self.quotation_crud.claim_for_award(s, qt_uuid, sp_uuid, (winner.name or "")[:20]),
        )
        if claim_err != ErrorType.SUCCESS or claimed == 0:
            res.result.SetResult(ErrorType.INVALID_REQUEST_DATA)
            res.msg = "이미 낙찰 처리된 견적입니다."
            return res

        # 작성자 알림 — 자동낙찰과 같은 SUCCESS 코드, manual 플래그로 '직접 낙찰' 구분.
        await create_notification(
            original.user_id, NotificationType.SUCCESS,
            {"qt_name": original.name, "qt_number": original.number,
             "winner_name": winner.name, "winner_price": winner.bid_price, "manual": True},
            ref_qt_id=qt_uuid,
        )
        return await self.get_quotation(qt_id, company_id)

    async def delete_quotation(self, qt_id: str, company_id=None, user_id=None, role=None) -> Res_DeleteQuotation:
        res = Res_DeleteQuotation()
        qt_uuid = uuid.UUID(qt_id)

        err_type, original = await self._fetch(qt_uuid, company_id)
        if err_type != ErrorType.SUCCESS or original is None:
            res.result.SetResult(err_type)
            return res
        # 소유자 게이팅 — 본인 견적 또는 최고관리자만 삭제(user_id 미지정=내부 호출은 스킵).
        if user_id is not None and not is_owner_or_admin(original.user_id, user_id, role):
            res.result.SetResult(ErrorType.ACCOUNT_FORBIDDEN)
            res.msg = "본인이 생성한 견적만 삭제할 수 있습니다."
            return res

        err_type = await DB_SESSION_MNG.execute_lambda_run(
            [quotations.DBType()],
            [lambda s: self.quotation_crud.soft_delete(s, qt_uuid)],
        )
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
        return res

    async def get_status(self, qt_id: str, company_id=None) -> Res_QuotationStatus:
        res = Res_QuotationStatus()
        err_type, quotation = await self._fetch(uuid.UUID(qt_id), company_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        res.qt_id = quotation.qt_id
        res.job_status = quotation.status
        res.message = "ok"
        return res

    async def get_result(self, qt_id: str, company_id=None) -> Res_QuotationResult:
        res = Res_QuotationResult()
        err_type, quotation = await self._fetch(uuid.UUID(qt_id), company_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res
        # 낙찰 결과는 quotations 컬럼에서 직접 노출. results 테이블 미존재로 result_count 는 0.
        res.qt_id = quotation.qt_id
        res.winner_supplier_id = quotation.preferred_sp_id
        res.winner_supplier_name = quotation.preferred_sp_name
        res.is_equal_bid = quotation.equal_bid_yn
        res.equal_bid_data = quotation.equal_bid_data
        res.result_count = 0
        return res

    async def list_sessions(self, qt_id: str, company_id=None) -> Res_QuotationSessions:
        res = Res_QuotationSessions()
        qt_uuid = uuid.UUID(qt_id)
        err_type, quotation = await self._fetch(qt_uuid, company_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        err_type, rows = await DB_SESSION_MNG.execute_lambda(
            sessions.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.list_sessions(s, qt_uuid),
        )
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        res.qt_id = quotation.qt_id
        # sessions.quotation_id → SessionData.qt_id 로 명시 매핑(컬럼명 불일치).
        res.sessions = [
            SessionData(
                session_id=r.session_id,
                qt_id=r.quotation_id,
                supplier_id=r.supplier_id,
                item_id=r.item_id,
                qt_number=r.qt_number,
                qt_round=r.qt_round,
                qt_type=r.qt_type,
                target_price=r.target_price,
                anchoring_price=r.anchoring_price,
                status=r.status,
                bid_price=r.bid_price,
                bid_at=r.bid_at,
                end_time=r.end_time,
                reject_reason=r.reject_reason,
                reject_price=r.reject_price,
                reject_delivery_type=r.reject_delivery_type,
                email_sent_at=r.email_sent_at,
                custom=r.custom,
                url=self._session_chat_url(r.session_id),
            )
            for r in rows
        ]
        res.total = len(res.sessions)
        return res

    # ----- 협상 초청 메일 (수동 발송)
    async def notify_sessions(self, qt_id: str, company_id=None, user_id=None, role=None) -> Res_NotifySessions:
        """[수동 발송] 견적의 '미발송' 세션(공급사 담당자)에게 협상 초청 메일을 일괄 발송한다.
        대상 = email_sent_at IS NULL + 담당자 이메일 보유."""
        res = Res_NotifySessions()
        qt_uuid = uuid.UUID(qt_id)
        err_type, quotation = await self._fetch(qt_uuid, company_id)
        if err_type != ErrorType.SUCCESS or quotation is None:
            res.result.SetResult(err_type)
            return res
        # 소유자 게이팅 — 본인 견적 또는 최고관리자만 초청메일 발송(user_id 미지정=내부 호출은 스킵).
        if user_id is not None and not is_owner_or_admin(quotation.user_id, user_id, role):
            res.result.SetResult(ErrorType.ACCOUNT_FORBIDDEN)
            res.msg = "본인이 생성한 견적만 초청 메일을 발송할 수 있습니다."
            return res

        err_type, rows = await DB_SESSION_MNG.execute_lambda(
            sessions.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.list_sessions_with_supplier(s, qt_uuid),
        )
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        # 행 언팩: (session, supplier_name, manager_email).
        targets = []  # [(session, name, email)]
        for r in rows:
            sess, sp_name, email = r[0], r[1], r[2]
            res.total += 1
            if sess.email_sent_at is not None:
                continue  # 이미 발송됨 — 재발송은 행 단위 endpoint 로
            if not email:
                res.skipped += 1
                continue
            targets.append((sess, sp_name, email))

        sent_ids = await self._send_invites(quotation, targets, res)
        if sent_ids:
            await self._mark_emailed(sent_ids)
        return res

    async def notify_session(self, session_id: str, company_id=None, user_id=None, role=None) -> Res_NotifySessions:
        """[수동 재발송] 단일 세션(공급사)에 초청 메일 발송(이미 보냈어도 강제 재발송)."""
        res = Res_NotifySessions()
        sess_uuid = uuid.UUID(session_id)
        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, sess_uuid),
        )
        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, sp_name, email = got[0], got[1], got[2]
        res.total = 1
        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)
            return res
        # 소유자 게이팅 — 본인 견적 또는 최고관리자만 재발송(user_id 미지정=내부 호출은 스킵).
        if user_id is not None and not is_owner_or_admin(quotation.user_id, user_id, role):
            res.result.SetResult(ErrorType.ACCOUNT_FORBIDDEN)
            res.msg = "본인이 생성한 견적만 초청 메일을 발송할 수 있습니다."
            return res
        if not email:
            res.skipped = 1
            return res

        sent_ids = await self._send_invites(quotation, [(sess, sp_name, email)], res)
        if sent_ids:
            await self._mark_emailed(sent_ids)
        return res

    async def _send_invites(self, quotation, targets: list, res: Res_NotifySessions) -> list:
        """targets [(session, supplier_name, email)] 에 초청 메일 발송. res.sent/failed 를 채우고
        성공한 session_id 목록을 반환. ACS/SMTP 미설정이면 첫 발송에서 중단(EMAIL_NOT_CONFIGURED)."""
        # 회사 브랜딩(초청 메일 헤더) 한 번 조회 — 견적당 동일.
        email_header = await DB_SESSION_MNG.execute_lambda(
            quotations.DBType(), DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.get_email_header(s, quotation.user_id),
        )
        sent_ids = []
        for sess, sp_name, email in targets:
            subject, html, text = build_invite_email(
                supplier_name=sp_name or "",
                quotation_name=quotation.name,
                qt_number=quotation.number,
                end_time=quotation.end_time,
                chat_url=self._session_chat_url(sess.session_id),
                email_header=email_header,
            )
            try:
                await send_email(email, subject, html, text)
                sent_ids.append(sess.session_id)
                res.sent += 1
            except EmailUnavailable as e:
                res.result.SetResult(ErrorType.EMAIL_NOT_CONFIGURED)  # 발송 채널 없음 — 더 시도해도 무의미
                res.msg = str(e)
                break
            except Exception as ex:
                LOG.e_no_callstack(ex)
                res.failed += 1
        # 보낼 대상이 있었는데 전부 실패면 명시적 실패 코드(설정은 됐으나 발송 실패).
        if res.sent == 0 and res.failed > 0 and res.result.success:
            res.result.SetResult(ErrorType.EMAIL_SEND_FAILED)
        return sent_ids

    async def _mark_emailed(self, session_ids: list) -> None:
        """발송 성공 세션들의 email_sent_at 갱신(write 트랜잭션)."""
        now = GTime.UTC()
        await DB_SESSION_MNG.execute_lambda_run(
            [sessions.DBType()],
            [lambda s: self.quotation_crud.mark_sessions_emailed(s, session_ids, now)],
        )

    async def list_chats(self, session_id: str, company_id=None) -> Res_SessionChat:
        res = Res_SessionChat()
        sess_uuid = uuid.UUID(session_id)
        res.session_id = sess_uuid

        # 회사 가드: chats 는 session 키라 세션→견적→회사로 확인한다(남의 회사 세션이면 NOT_FOUND).
        if company_id is not None:
            g_err, got = await DB_SESSION_MNG.execute_lambda(
                sessions.DBType(),
                DBWRType.DB_READ.value,
                lambda s: self.quotation_crud.get_session_with_supplier(s, sess_uuid),
            )
            if g_err != ErrorType.SUCCESS or got is None:
                res.result.SetResult(g_err if g_err != ErrorType.SUCCESS else ErrorType.QUOTATION_NOT_FOUND)
                return res
            guard_err, _ = await self._fetch(got[0].quotation_id, company_id)
            if guard_err != ErrorType.SUCCESS:
                res.result.SetResult(guard_err)
                return res

        err_type, rows = await DB_SESSION_MNG.execute_lambda(
            chats.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.list_chats(s, sess_uuid),
        )
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        # chats.seq → ChatMessageData.index 로 매핑. indicator_value(Decimal) → float.
        # 말풍선 텍스트는 chats.meta.script 에 영속화돼 있어 그대로 꺼낸다(프론트 하드코딩 X).
        res.messages = [
            ChatMessageData(
                chat_id=r.chat_id,
                session_id=r.session_id,
                card_id=r.card_id,
                index=r.seq,
                sender=r.sender,
                target_price=r.target_price,
                card_used_yn=r.card_used_yn,
                indicator_value=float(r.indicator_value) if r.indicator_value is not None else None,
                card_type=r.card_type,
                script=(r.meta or {}).get("script"),
                step=(r.meta or {}).get("step"),
            )
            for r in rows
        ]
        return res

    async def list_cards(self, qt_id: str, company_id=None) -> Res_QuotationCards:
        res = Res_QuotationCards()
        qt_uuid = uuid.UUID(qt_id)
        err_type, quotation = await self._fetch(qt_uuid, company_id)
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        # 견적의 버전(quotation.version_id)에 묶인 카드를 조회한다(version_nego_cards/version_wild_cards).
        err_type, rows = await DB_SESSION_MNG.execute_lambda(
            quotations.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.quotation_crud.get_version_cards(s, quotation.version_id),
        )
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        res.qt_id = quotation.qt_id
        # rows = [(card_type, card_pk, number, name, script, edit_script, condition, memo), ...].
        cards = []
        for card_type, card_pk, number, name, script, edit, condition, memo in rows:
            is_wild = card_type == 2
            cards.append(
                QuotationCardData(
                    session_card_id=card_pk,
                    qt_id=quotation.qt_id,
                    nego_card_id=None if is_wild else card_pk,
                    wild_card_id=card_pk if is_wild else None,
                    type=card_type,
                    number=number,
                    name=name,
                    script=script,
                    edit_script=edit,
                    condition=condition,
                    memo=memo,
                )
            )
        res.cards = cards
        return res