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)))
        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,
        }

    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", ""),
            )
            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"供应商订舱与放舱综合成功率{reliability:.0%}",
            f"预计运输时间{candidate.transit_days:g}天",
            f"{candidate.route_type or '当前'}路由，{candidate.capacity_commitment or '运力已确认'}",
        ]
        if scores["solution_type"] == "经济方案":
            reasons.insert(0, "当前可执行候选中采购价格最有竞争力")
        elif scores["solution_type"] == "时效方案":
            reasons.insert(0, "当前可执行候选中预计运输时间最短")
        else:
            reasons.insert(0, "客户偏好、供应商可靠性与价格综合表现最均衡")

        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"],
            "product_strategy_score": scores["product_strategy_score"],
            "total_score": scores["total_score"],
            "confidence_score": None,
            "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"在时效、履约可靠性和客户价格接受度之间取得当前最优平衡。"
        )

    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"])))
            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 _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)