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)