Eric-ontology/app/market/service.py

from __future__ import annotations

from datetime import date, datetime
import json
from pathlib import Path
from threading import RLock
from uuid import uuid4
from zoneinfo import ZoneInfo

from rdflib import Graph, Literal, Namespace, RDF, URIRef, XSD

from app.market.filters import rejection_reasons
from app.market.models import CandidateRecord, InquiryRequest
from app.market.pricing import build_prices
from app.market.scoring import CustomerSignals, score_candidates
from app.repository import RDFRepository


GESLI = Namespace("https://gesli.example/ontology#")
RUN = Namespace("https://gesli.example/recommendation/")


class MarketService:
    def __init__(self, root: Path, repository: RDFRepository):
        self.root = root
        self.repository = repository
        self.config = json.loads(
            (root / "config" / "market-strategies.json").read_text(encoding="utf-8")
        )
        self.runs: dict[str, dict] = {}
        self.audit_graph = Graph()
        self._lock = RLock()

    def customers(self) -> list[dict]:
        rows = self.repository.query_file(
            self.root / "queries" / "market-customers.rq"
        )["rows"]
        return [
            {
                "id": row["customer"],
                "name": row["customerName"],
                "industry": row.get("industry", ""),
                "preferred_route": row.get("routeName", ""),
                "repeat_purchase_count": _as_int(row.get("repeatPurchaseCount")),
                "price_sensitivity": _as_float(row.get("customerPriceSensitivity"), 0.5),
                "time_sensitivity": _as_float(row.get("customerTimeSensitivity"), 0.5),
                "reliability_sensitivity": _as_float(
                    row.get("customerReliabilitySensitivity"), 0.5
                ),
            }
            for row in rows
        ]

    def form_options(self) -> dict:
        candidates = self._candidate_records()
        context_rows = self.repository.query_file(
            self.root / "queries" / "market-context.rq"
        )["rows"]
        configured_routes = {
            (row.get("origin", ""), row.get("destination", ""))
            for row in context_rows
            if row.get("kind") == "benchmark" and row.get("origin") and row.get("destination")
        }
        origins = sorted({origin for origin, _ in configured_routes})
        destinations = sorted({destination for _, destination in configured_routes})
        route_types = sorted({
            item.route_type
            for item in candidates
            if item.route_type and (item.origin, item.destination) in configured_routes
        })
        benchmark_by_route = {
            (row.get("origin", ""), row.get("destination", "")): _as_float(
                row.get("amount")
            )
            for row in context_rows
            if row.get("kind") == "benchmark"
        }
        routes = []
        for origin, destination in sorted(configured_routes):
            route_candidates = [
                item
                for item in candidates
                if item.origin == origin
                and item.destination == destination
                and item.departure_date
                and item.valid_to
                and item.valid_to >= date(2026, 7, 20)
            ]
            departures = [item.departure_date for item in route_candidates]
            transit_days = [item.transit_days for item in route_candidates]
            routes.append({
                "origin": origin,
                "destination": destination,
                "route_types": sorted({
                    item.route_type for item in route_candidates if item.route_type
                }),
                "target_amount": benchmark_by_route.get((origin, destination), 0),
                "requested_departure_date": (
                    min(departures).isoformat() if departures else "2026-07-27"
                ),
                "max_transit_days": max(transit_days) if transit_days else 30,
            })
        return {
            "customers": self.customers(),
            "origins": origins,
            "destinations": destinations,
            "route_types": route_types,
            "routes": routes,
            "defaults": InquiryRequest(customer="https://gesli.example/sample#Haier").model_dump(
                mode="json"
            ),
        }

    def recommend(self, inquiry: InquiryRequest) -> dict:
        customer_row = self._find_customer(inquiry.customer)
        customer = self._customer_signals(customer_row)
        component_cost, component_names = self._component_costs(
            inquiry.evaluation_date, inquiry.origin
        )
        component_product_codes = {
            code
            for codes in self.config["component_product_codes_by_origin"].values()
            for code in codes
        }

        route_candidates = [
            candidate
            for candidate in self._candidate_records()
            if candidate.origin == inquiry.origin
            and candidate.destination == inquiry.destination
            and candidate.currency == "CNY"
            and candidate.product_code not in component_product_codes
        ]

        feasible: list[CandidateRecord] = []
        rejected: list[dict] = []
        for candidate in route_candidates:
            reasons = rejection_reasons(candidate, inquiry)
            if reasons:
                rejected.append(self._rejected_payload(candidate, reasons))
            else:
                feasible.append(candidate)

        if not feasible:
            raise ValueError("没有满足当前硬条件的供应商产品和班期")

        benchmark, strategy = self._market_context(inquiry)
        scored, weights = score_candidates(feasible, inquiry, customer)

        candidates: list[dict] = []
        for item in scored:
            candidate: CandidateRecord = item.pop("candidate")
            total_supplier_cost = candidate.supplier_cost + component_cost
            prices = build_prices(
                benchmark_amount=benchmark["amount"],
                strategy_rate=strategy["adjustment_rate"],
                solution_type=item["solution_type"],
                customer=customer,
                salesperson_adjustment=inquiry.salesperson_adjustment,
                supplier_cost=total_supplier_cost,
                config=self.config,
            )
            payload = self._candidate_payload(candidate, item, prices, component_names)
            candidates.append(payload)

        top = candidates[0]
        score_gap = top["total_score"] - candidates[1]["total_score"] if len(candidates) > 1 else 10
        confidence = round(min(0.98, 0.78 + max(0, score_gap) / 100), 2)
        top["confidence_score"] = confidence

        run_id = f"rec-{uuid4().hex[:12]}"
        generated_at = datetime.now(ZoneInfo("Asia/Shanghai")).isoformat(timespec="seconds")
        result = {
            "run_id": run_id,
            "generated_at": generated_at,
            "normalized_inquiry": {
                **inquiry.model_dump(mode="json"),
                "customer_name": customer_row["customerName"],
                "route": f"{inquiry.origin} → {inquiry.destination}",
                "dangerous_goods_label": "危险品" if inquiry.dangerous_goods else "非危险品",
            },
            "pipeline": {
                "discovered": len(route_candidates),
                "rejected": len(rejected),
                "feasible": len(feasible),
                "generated_solutions": len(candidates),
            },
            "recommendation": {
                "candidate_id": top["id"],
                "solution_type": top["solution_type"],
                "product_name": top["product_name"],
                "supplier_name": top["supplier_name"],
                "final_quote_amount": top["pricing"]["final_quote_amount"],
                "currency": top["pricing"]["currency"],
                "confidence_score": confidence,
                "summary": self._recommendation_summary(top, strategy, customer_row),
            },
            "market_context": {
                "benchmark_amount": benchmark["amount"],
                "currency": benchmark["currency"],
                "strategy_name": strategy["name"],
                "strategy_objective": strategy["objective"],
                "strategy_adjustment_rate": strategy["adjustment_rate"],
            },
            "weights": weights,
            "candidates": candidates,
            "rejected_candidates": rejected,
        }
        self._store_run(run_id, result, inquiry, strategy)
        return result

    def get_run(self, run_id: str) -> dict | None:
        with self._lock:
            return self.runs.get(run_id)

    def reset_runtime(self) -> None:
        with self._lock:
            self.runs.clear()
            self.audit_graph = Graph()

    def get_audit(self, run_id: str) -> dict | None:
        run = self.get_run(run_id)
        if not run:
            return None
        run_uri = RUN[run_id]
        with self._lock:
            triples = list(self.audit_graph.triples((run_uri, None, None)))
            evaluations = [
                evaluation
                for _, _, evaluation in self.audit_graph.triples(
                    (run_uri, GESLI.hasCandidateEvaluation, None)
                )
            ]
            for evaluation in evaluations:
                triples.extend(self.audit_graph.triples((evaluation, None, None)))
        return {
            "run_id": run_id,
            "generated_at": run["generated_at"],
            "input": run["normalized_inquiry"],
            "weights": run["weights"],
            "market_context": run["market_context"],
            "candidate_count": len(run["candidates"]),
            "rejected_count": len(run["rejected_candidates"]),
            "rdf_assertions": len(triples),
            "sources": self.repository.loaded_file_names,
            "business_explanation": self._business_audit(run),
        }

    def _candidate_records(self) -> list[CandidateRecord]:
        rows = self.repository.query_file(
            self.root / "queries" / "market-candidates.rq"
        )["rows"]
        return [
            CandidateRecord(
                quote_uri=row["quote"],
                quote_no=row["quoteNo"],
                product_uri=row["product"],
                product_code=row.get("productCode", ""),
                product_name=row.get("productName", ""),
                brand_name=row.get("brandName", ""),
                transport_mode=row.get("transportMode", ""),
                offering_uri=row["offering"],
                offering_code=row.get("offeringCode", ""),
                supplier_uri=row["supplier"],
                supplier_name=row["supplierName"],
                schedule_uri=row.get("schedule", ""),
                schedule_code=row.get("scheduleCode", ""),
                route_type=row.get("routeType", ""),
                service_form=row.get("serviceForm", ""),
                origin=row.get("originLocation", ""),
                destination=row.get("destinationLocation", ""),
                currency=row["currency"],
                supplier_cost=_as_float(row.get("supplierCostAmount")),
                valid_from=_as_date(row.get("validFrom")),
                valid_to=_as_date(row.get("validTo")),
                departure_date=_as_date(row.get("estimatedDepartureTime")),
                arrival_date=_as_date(row.get("estimatedArrivalTime")),
                transit_days=_as_float(row.get("estimatedTransitDays")),
                booking_success_rate=_as_float(row.get("bookingSuccessRate"), 0.8),
                space_release_success_rate=_as_float(
                    row.get("spaceReleaseSuccessRate"), 0.8
                ),
                etd_on_time_rate=_as_float(row.get("etdAtdOnTimeRate"), 0.85),
                eta_on_time_rate=_as_float(row.get("etaAtaOnTimeRate"), 0.85),
                supplier_active=_as_bool(row.get("supplierActive"), True),
                capacity_available=_as_bool(row.get("capacityAvailable"), True),
                supports_dangerous_goods=_as_bool(
                    row.get("supportsDangerousGoods"), False
                ),
                dangerous_goods_only=_as_bool(row.get("dangerousGoodsOnly"), False),
                agency_level=row.get("agencyLevel", ""),
                offering_channel=row.get("offeringChannel", ""),
                capacity_commitment=row.get("capacityCommitment", ""),
                spot_space_status=row.get("spotSpaceStatus") or "需确认",
                spot_space_teu=_as_float(row.get("spotSpaceTeu")),
                solution_assurance=_as_float(row.get("solutionAssuranceScore"), 0.72),
                service_support_score=_as_float(row.get("serviceSupportScore"), 0.72),
                relationship_tier=row.get("relationshipTier") or "基础合作",
                relationship_strength=_as_float(row.get("relationshipStrength"), 0.60),
                low_price_access_probability=_as_float(
                    row.get("lowPriceAccessProbability"), 0.50
                ),
                priority_space_probability=_as_float(
                    row.get("prioritySpaceProbability"), 0.60
                ),
                relationship_evidence=(
                    row.get("relationshipEvidence") or "按标准合作机制执行"
                ),
            )
            for row in rows
        ]

    def _find_customer(self, customer_id: str) -> dict:
        rows = self.repository.query_file(
            self.root / "queries" / "market-customers.rq"
        )["rows"]
        for row in rows:
            if row["customer"] == customer_id or row["customerName"] == customer_id:
                return row
        raise ValueError("客户不存在或没有可用画像")

    @staticmethod
    def _customer_signals(row: dict) -> CustomerSignals:
        service_requirement = "；".join(
            value
            for value in (
                row.get("notificationRequirement", ""),
                row.get("documentRequirement", ""),
                row.get("exceptionResponseRequirement", ""),
            )
            if value
        )
        return CustomerSignals(
            price_sensitivity=_as_float(row.get("customerPriceSensitivity"), 0.5),
            time_sensitivity=_as_float(row.get("customerTimeSensitivity"), 0.5),
            reliability_sensitivity=_as_float(
                row.get("customerReliabilitySensitivity"), 0.5
            ),
            repeat_purchase_count=_as_int(row.get("repeatPurchaseCount")),
            overdue_days=_as_int(row.get("overdueDays")),
            service_requirement=service_requirement,
        )

    def _component_costs(
        self, evaluation_date: date, origin: str
    ) -> tuple[float, list[str]]:
        candidates = self._candidate_records()
        total = 0.0
        names: list[str] = []
        product_codes = self.config["component_product_codes_by_origin"].get(
            origin,
            self.config["component_product_codes_by_origin"].get("__default__", []),
        )
        for product_code in product_codes:
            valid = [
                item
                for item in candidates
                if item.product_code == product_code
                and (not item.valid_from or item.valid_from <= evaluation_date)
                and (not item.valid_to or evaluation_date <= item.valid_to)
            ]
            if valid:
                selected = min(valid, key=lambda item: item.supplier_cost)
                total += selected.supplier_cost
                names.append(selected.product_name)
        return total, names

    def _market_context(self, inquiry: InquiryRequest) -> tuple[dict, dict]:
        rows = self.repository.query_file(
            self.root / "queries" / "market-context.rq"
        )["rows"]
        applicable = [
            row
            for row in rows
            if row.get("origin") == inquiry.origin
            and row.get("destination") == inquiry.destination
            and _within(inquiry.evaluation_date, row.get("validFrom"), row.get("validTo"))
        ]
        benchmark_row = next((row for row in applicable if row["kind"] == "benchmark"), None)
        strategy_row = next((row for row in applicable if row["kind"] == "strategy"), None)
        if not benchmark_row or not strategy_row:
            raise ValueError("当前路线缺少有效的市场基准价格或启用策略")
        return (
            {
                "uri": benchmark_row["resource"],
                "amount": _as_float(benchmark_row["amount"]),
                "currency": benchmark_row["currency"],
            },
            {
                "uri": strategy_row["resource"],
                "name": strategy_row["strategyName"],
                "objective": strategy_row["strategyObjective"],
                "adjustment_rate": _as_float(strategy_row["strategyAdjustmentRate"]),
            },
        )

    @staticmethod
    def _candidate_payload(
        candidate: CandidateRecord,
        scores: dict,
        prices: dict,
        component_names: list[str],
    ) -> dict:
        reliability = round(
            (candidate.booking_success_rate + candidate.space_release_success_rate) / 2,
            2,
        )
        reasons = [
            f"我方与{candidate.supplier_name}为{candidate.relationship_tier}：{candidate.relationship_evidence}",
            _spot_space_summary(candidate.spot_space_status, candidate.spot_space_teu),
            f"低价获取概率{candidate.low_price_access_probability:.0%}，优先舱位概率{candidate.priority_space_probability:.0%}",
            f"供应商订舱与放舱综合成功率{reliability:.0%}",
            f"预计运输时间{candidate.transit_days:g}天",
            f"{candidate.route_type or '当前'}路由，{candidate.capacity_commitment or '运力已确认'}",
        ]
        focus_reasons = {
            "方案优选": "方案保障、供应商合作与履约表现综合最优",
            "关系优选": "船公司或订舱代理的合作优势最突出",
            "价格参考": "当前可执行候选中采购价格最具参考性",
            "时效参考": "当前可执行候选中预计运输时间最短",
        }
        reasons.insert(0, focus_reasons.get(scores["solution_type"], "可执行的备选运输方案"))

        return {
            "id": candidate.quote_uri,
            "quote_no": candidate.quote_no,
            "solution_type": scores["solution_type"],
            "rank": scores["rank"],
            "brand_name": candidate.brand_name,
            "product_name": candidate.product_name,
            "supplier_name": candidate.supplier_name,
            "offering_channel": candidate.offering_channel,
            "schedule_code": candidate.schedule_code,
            "departure_date": candidate.departure_date.isoformat() if candidate.departure_date else "",
            "arrival_date": candidate.arrival_date.isoformat() if candidate.arrival_date else "",
            "transit_days": candidate.transit_days,
            "route_type": candidate.route_type,
            "capacity_commitment": candidate.capacity_commitment,
            "components": [*component_names, candidate.product_name],
            "reliability_rate": reliability,
            "price_score": scores["price_score"],
            "transit_score": scores["transit_score"],
            "supplier_execution_score": scores["supplier_execution_score"],
            "customer_fit_score": scores["customer_fit_score"],
            "solution_assurance_score": scores["solution_assurance_score"],
            "relationship_advantage_score": scores["relationship_advantage_score"],
            "spot_space_score": scores["spot_space_score"],
            "total_score": scores["total_score"],
            "confidence_score": None,
            "relationship": {
                "supplier_name": candidate.supplier_name,
                "supplier_channel": candidate.offering_channel,
                "tier": candidate.relationship_tier,
                "strength": candidate.relationship_strength,
                "low_price_access_probability": candidate.low_price_access_probability,
                "priority_space_probability": candidate.priority_space_probability,
                "evidence": candidate.relationship_evidence,
            },
            "spot_space": {
                "status": candidate.spot_space_status,
                "teu": candidate.spot_space_teu,
                "summary": _spot_space_summary(
                    candidate.spot_space_status, candidate.spot_space_teu
                ),
            },
            "reasons": reasons,
            "pricing": prices,
        }

    @staticmethod
    def _rejected_payload(candidate: CandidateRecord, reasons: list[str]) -> dict:
        return {
            "id": candidate.quote_uri,
            "quote_no": candidate.quote_no,
            "product_name": candidate.product_name,
            "supplier_name": candidate.supplier_name,
            "schedule_code": candidate.schedule_code,
            "reasons": reasons,
        }

    @staticmethod
    def _recommendation_summary(
        candidate: dict, strategy: dict, customer_row: dict
    ) -> str:
        return (
            f"为{customer_row['customerName']}推荐{candidate['solution_type']}："
            f"{candidate['supplier_name']}的{candidate['product_name']}。"
            f"综合得分{candidate['total_score']:.1f}，采用“{strategy['name']}”，"
            f"当前{candidate['spot_space']['summary']}，"
            f"在方案保障、供应商合作、现舱与履约可靠性之间取得当前最优平衡。"
        )

    @staticmethod
    def _business_audit(run: dict) -> dict:
        top = run["candidates"][0]
        alternatives = []
        for candidate in run["candidates"][1:]:
            alternatives.append({
                "title": f"{candidate['solution_type']}：{candidate['supplier_name']}",
                "detail": (
                    f"{candidate['spot_space']['summary']}；"
                    f"我方与该供应商为{candidate['relationship']['tier']}；"
                    f"预计{candidate['transit_days']:g}天到达。"
                ),
            })
        rejected = [
            f"{candidate['supplier_name']}：{'；'.join(candidate['reasons'])}"
            for candidate in run["rejected_candidates"][:3]
        ]
        return {
            "decision": (
                f"本次为{run['normalized_inquiry']['customer_name']}的"
                f"{run['normalized_inquiry']['route']}询价，在"
                f"{len(run['candidates'])}个可执行方案中选择了{top['supplier_name']}。"
            ),
            "why_recommended": [
                f"{top['spot_space']['summary']}，可直接支持本次订舱安排。",
                (
                    f"我方与{top['supplier_name']}为{top['relationship']['tier']}，"
                    f"{top['relationship']['evidence']}。"
                ),
                (
                    f"方案为{top['route_type']}，预计{top['transit_days']:g}天，"
                    f"方案保障得分{top['solution_assurance_score']}分。"
                ),
                f"建议向客户报价{top['pricing']['final_quote_amount']:,.0f} {top['pricing']['currency']}。",
            ],
            "alternatives": alternatives,
            "rejected": rejected,
            "selection_basis": "优先看方案保障、与船公司或订舱代理的合作深度和现舱；再比较履约、客户需求、时效与价格。价格用于利润校验和横向比较，但不单独决定首选。",
        }

    def _store_run(
        self, run_id: str, result: dict, inquiry: InquiryRequest, strategy: dict
    ) -> None:
        run_uri = RUN[run_id]
        generated_at = Literal(result["generated_at"], datatype=XSD.dateTime)
        with self._lock:
            self.runs[run_id] = result
            self.audit_graph.add((run_uri, RDF.type, GESLI.RecommendationRun))
            self.audit_graph.add((run_uri, GESLI.generatedAt, generated_at))
            self.audit_graph.add((run_uri, GESLI.usesMarketStrategy, URIRef(strategy["uri"])))
            self.audit_graph.add((run_uri, GESLI.recommendationReason, Literal(result["recommendation"]["summary"])))
            customer_uri = URIRef(inquiry.customer)
            self.audit_graph.add((run_uri, GESLI.quoteForCustomer, customer_uri))
            for candidate in result["candidates"]:
                evaluation_uri = RUN[f"{run_id}/candidate/{candidate['rank']}"]
                self.audit_graph.add((evaluation_uri, RDF.type, GESLI.CandidateEvaluation))
                self.audit_graph.add((run_uri, GESLI.hasCandidateEvaluation, evaluation_uri))
                self.audit_graph.add((evaluation_uri, GESLI.evaluatedQuote, URIRef(candidate["id"])))
                self.audit_graph.add((evaluation_uri, GESLI.recommendationRank, Literal(candidate["rank"])))
                self.audit_graph.add((evaluation_uri, GESLI.totalScore, Literal(candidate["total_score"])))
                self.audit_graph.add((
                    evaluation_uri,
                    GESLI.solutionAssuranceEvaluationScore,
                    Literal(candidate["solution_assurance_score"]),
                ))
                self.audit_graph.add((
                    evaluation_uri,
                    GESLI.relationshipAdvantageScore,
                    Literal(candidate["relationship_advantage_score"]),
                ))
                self.audit_graph.add((
                    evaluation_uri,
                    GESLI.spotSpaceScore,
                    Literal(candidate["spot_space_score"]),
                ))
            top_evaluation = RUN[f"{run_id}/candidate/1"]
            self.audit_graph.add((run_uri, GESLI.recommendedEvaluation, top_evaluation))


def _as_float(value: str | None, default: float = 0.0) -> float:
    try:
        return float(value) if value not in (None, "") else default
    except (TypeError, ValueError):
        return default


def _spot_space_summary(status: str, teu: float) -> str:
    status_label = status if status.startswith("现舱") else f"现舱{status}"
    if teu > 0:
        return f"{status_label}，当前可订{teu:g} TEU"
    return f"{status_label}，需向供应商确认"


def _as_int(value: str | None, default: int = 0) -> int:
    try:
        return int(value) if value not in (None, "") else default
    except (TypeError, ValueError):
        return default


def _as_bool(value: str | None, default: bool = False) -> bool:
    if value in (None, ""):
        return default
    return str(value).lower() in {"true", "1", "yes"}


def _as_date(value: str | None) -> date | None:
    if not value:
        return None
    try:
        return date.fromisoformat(value[:10])
    except ValueError:
        return None


def _within(current: date, start: str | None, end: str | None) -> bool:
    start_date = _as_date(start)
    end_date = _as_date(end)
    return (not start_date or start_date <= current) and (not end_date or current <= end_date)