from __future__ import annotations from datetime import date, datetime, time, timedelta, timezone from pathlib import Path from uuid import uuid4 from pydantic import BaseModel, Field from rdflib import Literal, Namespace, RDF, RDFS, XSD from app.market import MarketService from app.repository import RDFRepository GESLI = Namespace("https://gesli.example/ontology#") MOCK = Namespace("https://gesli.example/runtime-mock/") class MockBatchRequest(BaseModel): origin: str destination: str supplier_count: int = Field(default=4, ge=1, le=8) customer_count: int = Field(default=2, ge=0, le=5) evaluation_date: date = date(2026, 7, 20) class LocalMockDataService: def __init__( self, root: Path, repository: RDFRepository, market_service: MarketService, ): self.root = root self.repository = repository self.market_service = market_service def catalog(self) -> dict: candidate_rows = self.repository.query_file( self.root / "queries" / "market-candidates.rq" )["rows"] context_rows = self.repository.query_file( self.root / "queries" / "market-context.rq" )["rows"] customers = self.market_service.customers() evaluation_date = date(2026, 7, 20) routes: dict[tuple[str, str], dict] = {} for row in candidate_rows: origin = row.get("originLocation", "") destination = row.get("destinationLocation", "") if not origin or not destination: continue key = (origin, destination) route = routes.setdefault(key, { "origin": origin, "destination": destination, "products": set(), "suppliers": set(), "quotes": set(), "current_quotes": 0, "transport_modes": set(), }) route["products"].add(row.get("product", "")) route["suppliers"].add(row.get("supplier", "")) route["quotes"].add(row.get("quote", "")) if row.get("transportMode"): route["transport_modes"].add(row["transportMode"]) if _within(evaluation_date, row.get("validFrom"), row.get("validTo")): route["current_quotes"] += 1 route_rows = [ { "origin": route["origin"], "destination": route["destination"], "transport_modes": " / ".join(sorted(route["transport_modes"])), "product_count": len(route["products"]), "supplier_count": len(route["suppliers"]), "quote_count": len(route["quotes"]), "current_quote_count": route["current_quotes"], } for route in routes.values() ] route_rows.sort(key=lambda item: (item["origin"], item["destination"])) quote_rows = [] for row in candidate_rows: valid_from = row.get("validFrom", "") valid_to = row.get("validTo", "") if _within(evaluation_date, valid_from, valid_to): status = "当前有效" elif valid_to and valid_to < evaluation_date.isoformat(): status = "已过期" else: status = "尚未生效" quote_rows.append({ "quote_no": row.get("quoteNo", ""), "route": f"{row.get('originLocation', '')} → {row.get('destinationLocation', '')}", "transport_mode": row.get("transportMode", ""), "product_name": row.get("productName", ""), "supplier_name": row.get("supplierName", ""), "supplier_cost_amount": _as_float(row.get("supplierCostAmount")), "currency": row.get("currency", ""), "valid_to": valid_to, "status": status, "dangerous_goods": row.get("supportsDangerousGoods", "false") == "true", "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 "按标准合作机制执行" ), "spot_space_status": row.get("spotSpaceStatus") or "需确认", "spot_space_teu": _as_float(row.get("spotSpaceTeu")), "runtime_mock": row.get("quote", "").startswith(str(MOCK)), }) quote_rows.sort( key=lambda item: ( item["route"], item["status"], item["supplier_cost_amount"] ) ) strategies = [ { "name": row.get("strategyName", ""), "objective": row.get("strategyObjective", ""), "route": f"{row.get('origin', '')} → {row.get('destination', '')}", "adjustment_rate": _as_float(row.get("strategyAdjustmentRate")), "valid_to": row.get("validTo", ""), } for row in context_rows if row.get("kind") == "strategy" ] return { "summary": { "triple_count": self.repository.triple_count, "in_memory_triple_count": self.repository.in_memory_triple_count, "customer_count": len(customers), "supplier_count": len({row.get("supplier") for row in candidate_rows}), "product_count": len({row.get("product") for row in candidate_rows}), "quote_count": len({row.get("quote") for row in candidate_rows}), "route_count": len(route_rows), "strategy_count": len(strategies), }, "routes": route_rows, "customers": customers, "quotes": quote_rows, "strategies": strategies, "loaded_files": self.repository.loaded_file_names, } def generate(self, request: MockBatchRequest) -> dict: route_context = next( ( route for route in self.market_service.form_options()["routes"] if route["origin"] == request.origin and route["destination"] == request.destination ), None, ) if not route_context: raise ValueError("只能为已经配置市场基准和策略的路线生成Mock数据") token = uuid4().hex[:8] benchmark = float(route_context["target_amount"]) transport_mode = self._route_transport_mode( request.origin, request.destination ) triples = [] generated_suppliers = [] generated_customers = [] for index in range(request.supplier_count): suffix = f"{token}-{index + 1}" brand = MOCK[f"brand-{suffix}"] supplier = MOCK[f"supplier-{suffix}"] product = MOCK[f"product-{suffix}"] offering = MOCK[f"offering-{suffix}"] schedule = MOCK[f"schedule-{suffix}"] quote = MOCK[f"quote-{suffix}"] performance = MOCK[f"performance-{suffix}"] profile = self._supplier_profile(index) supplier_name = ( f"印度航线Mock{profile['supplier_label']} {token[-3:].upper()}{index + 1}" ) product_name = ( f"{request.origin}-{request.destination}{profile['product_label']}方案{index + 1}" ) departure = request.evaluation_date + timedelta(days=8 + index) transit_days = self._transit_days(transport_mode, index) arrival = departure + timedelta(days=transit_days) cost = round(benchmark * profile["cost_ratio"], -2) success_rate = profile["booking_success_rate"] product_class = ( GESLI.AirFreightProduct if transport_mode == "空运" else GESLI.ContainerOceanProduct ) route_type = "直达" if transport_mode == "空运" and index < 2 else profile["route_type"] triples.extend([ (brand, RDF.type, GESLI.ServiceBrand), (brand, GESLI.brandCode, Literal(f"MOCK-{token}-{index + 1}")), (brand, GESLI.brandName, Literal(f"Mock品牌 {index + 1}")), (supplier, RDF.type, profile["supplier_class"]), (supplier, RDFS.label, Literal(supplier_name, lang="zh")), (supplier, GESLI.platformSupplierType, Literal(profile["supplier_type"])), (supplier, GESLI.isActivePartner, Literal(True)), (product, RDF.type, product_class), (product, GESLI.productCode, Literal(f"MOCK-PRODUCT-{suffix}")), (product, GESLI.productName, Literal(product_name)), (product, GESLI.hasServiceBrand, brand), (product, GESLI.hasProductPerformanceProfile, performance), (product, GESLI.transportMode, Literal(transport_mode)), (product, GESLI.serviceForm, Literal(profile["service_form"])), (product, GESLI.routeType, Literal(route_type)), (product, GESLI.originLocation, Literal(request.origin)), (product, GESLI.destinationLocation, Literal(request.destination)), (product, GESLI.estimatedTransitDays, Literal(transit_days, datatype=XSD.decimal)), (performance, RDF.type, GESLI.ProductPerformanceProfile), (performance, GESLI.etdAtdOnTimeRate, Literal(success_rate, datatype=XSD.decimal)), (performance, GESLI.etaAtaOnTimeRate, Literal(success_rate - 0.02, datatype=XSD.decimal)), (offering, RDF.type, GESLI.SupplierProductOffering), (offering, GESLI.offeringCode, Literal(f"MOCK-OFFER-{suffix}")), (offering, GESLI.offeredBySupplier, supplier), (offering, GESLI.offeringProduct, product), (offering, GESLI.offeringChannel, Literal(profile["offering_channel"])), (offering, GESLI.agencyLevel, Literal(profile["agency_level"])), (offering, GESLI.capacityCommitment, Literal(profile["capacity_commitment"])), (offering, GESLI.capacityAvailable, Literal(True)), (offering, GESLI.spotSpaceStatus, Literal(profile["spot_space_status"])), (offering, GESLI.spotSpaceTeu, Literal(profile["spot_space_teu"], datatype=XSD.decimal)), (offering, GESLI.supportsDangerousGoods, Literal(index == 0)), (offering, GESLI.solutionAssuranceScore, Literal(profile["solution_assurance"], datatype=XSD.decimal)), (offering, GESLI.serviceSupportScore, Literal(profile["service_support_score"], datatype=XSD.decimal)), (offering, GESLI.relationshipTier, Literal(profile["relationship_tier"])), (offering, GESLI.relationshipStrength, Literal(profile["relationship_strength"], datatype=XSD.decimal)), (offering, GESLI.lowPriceAccessProbability, Literal(profile["low_price_access_probability"], datatype=XSD.decimal)), (offering, GESLI.prioritySpaceProbability, Literal(profile["priority_space_probability"], datatype=XSD.decimal)), (offering, GESLI.relationshipEvidence, Literal(profile["relationship_evidence"])), (offering, GESLI.bookingSuccessRate, Literal(success_rate, datatype=XSD.decimal)), (offering, GESLI.spaceReleaseSuccessRate, Literal(profile["space_release_success_rate"], datatype=XSD.decimal)), (schedule, RDF.type, GESLI.ScheduledServiceInstance), (schedule, GESLI.scheduleCode, Literal(f"MOCK-SCHEDULE-{suffix}")), (schedule, GESLI.scheduleOfProduct, product), (schedule, GESLI.documentCutoffTime, _date_time_literal(departure - timedelta(days=2))), (schedule, GESLI.estimatedDepartureTime, _date_time_literal(departure)), (schedule, GESLI.estimatedArrivalTime, _date_time_literal(arrival)), (schedule, GESLI.estimatedTransitDays, Literal(transit_days, datatype=XSD.decimal)), (quote, RDF.type, GESLI.SupplierQuote), (quote, GESLI.quoteNo, Literal(f"MOCK-QUOTE-{suffix}")), (quote, GESLI.quotedBySupplier, supplier), (quote, GESLI.quotedForOffering, offering), (quote, GESLI.quoteAppliesToSchedule, schedule), (quote, GESLI.supplierCostAmount, Literal(cost, datatype=XSD.decimal)), (quote, GESLI.currency, Literal("CNY")), (quote, GESLI.validFrom, Literal(request.evaluation_date, datatype=XSD.date)), (quote, GESLI.validTo, Literal(request.evaluation_date + timedelta(days=30), datatype=XSD.date)), ]) generated_suppliers.append(supplier_name) for index in range(request.customer_count): suffix = f"{token}-{index + 1}" customer = MOCK[f"customer-{suffix}"] trade = MOCK[f"customer-trade-{suffix}"] credit = MOCK[f"customer-credit-{suffix}"] customer_name = f"印度目的港模拟客户 {token[-3:].upper()}{index + 1}" triples.extend([ (customer, RDF.type, GESLI.Customer), (customer, RDFS.label, Literal(customer_name, lang="zh")), (customer, GESLI.hasTradeProfile, trade), (customer, GESLI.hasCreditProfile, credit), (trade, RDF.type, GESLI.CustomerTradeProfile), (trade, GESLI.industry, Literal("印度市场Mock行业")), (trade, GESLI.routeName, Literal(f"{request.origin}-{request.destination}")), (trade, GESLI.repeatPurchaseCount, Literal(4 + index * 6)), (trade, GESLI.quoteWinRate, Literal(0.5 + index * 0.08, datatype=XSD.decimal)), (trade, GESLI.customerPriceSensitivity, Literal(0.45 + index * 0.15, datatype=XSD.decimal)), (trade, GESLI.customerTimeSensitivity, Literal(0.75 - index * 0.1, datatype=XSD.decimal)), (trade, GESLI.customerReliabilitySensitivity, Literal(0.7 + index * 0.1, datatype=XSD.decimal)), (credit, RDF.type, GESLI.CustomerCreditProfile), (credit, GESLI.overdueDays, Literal(index)), ]) generated_customers.append(customer_name) added_triples = self.repository.add_triples(triples) self.market_service.reset_runtime() return { "batch_id": token, "added_triples": added_triples, "suppliers": generated_suppliers, "customers": generated_customers, "catalog": self.catalog(), } def reset(self) -> dict: self.repository.reload() self.market_service.reset_runtime() return self.catalog() def _route_transport_mode(self, origin: str, destination: str) -> str: rows = self.repository.query_file( self.root / "queries" / "market-candidates.rq" )["rows"] return next( ( row.get("transportMode", "海运") for row in rows if row.get("originLocation") == origin and row.get("destinationLocation") == destination ), "海运", ) @staticmethod def _transit_days(transport_mode: str, index: int) -> int: return (3 + index) if transport_mode == "空运" else (16 + index * 2) @staticmethod def _supplier_profile(index: int) -> dict: profiles = [ { "supplier_label": "船公司直营", "supplier_class": GESLI.Carrier, "supplier_type": "船公司", "product_label": "优先直航", "route_type": "直航", "service_form": "优先舱位包柜", "offering_channel": "船公司直营", "agency_level": "产品品牌本地分支机构", "capacity_commitment": "年度协议优先保舱20TEU", "spot_space_status": "可立即订舱", "spot_space_teu": 18, "relationship_tier": "战略合作", "relationship_strength": 0.96, "low_price_access_probability": 0.90, "priority_space_probability": 0.95, "relationship_evidence": "年度协议、价格窗口直连与优先放舱额度", "solution_assurance": 0.98, "service_support_score": 0.94, "booking_success_rate": 0.97, "space_release_success_rate": 0.96, "cost_ratio": 0.91, }, { "supplier_label": "一级订舱代理", "supplier_class": GESLI.ShippingAgencyCompany, "supplier_type": "订舱代理", "product_label": "稳定直航", "route_type": "直航", "service_form": "稳定舱位包柜", "offering_channel": "一级订舱代理", "agency_level": "一级代理", "capacity_commitment": "月度保舱12TEU", "spot_space_status": "现舱有限", "spot_space_teu": 8, "relationship_tier": "核心合作", "relationship_strength": 0.87, "low_price_access_probability": 0.82, "priority_space_probability": 0.86, "relationship_evidence": "核心代理授权与固定订舱窗口", "solution_assurance": 0.90, "service_support_score": 0.90, "booking_success_rate": 0.94, "space_release_success_rate": 0.92, "cost_ratio": 0.86, }, { "supplier_label": "稳定合作代理", "supplier_class": GESLI.FreightForwarderSupplier, "supplier_type": "货代公司", "product_label": "常规直航", "route_type": "直航", "service_form": "标准包柜", "offering_channel": "长期合作货代", "agency_level": "稳定合作代理", "capacity_commitment": "周度滚动确认8TEU", "spot_space_status": "现舱有限", "spot_space_teu": 4, "relationship_tier": "稳定合作", "relationship_strength": 0.74, "low_price_access_probability": 0.70, "priority_space_probability": 0.72, "relationship_evidence": "长期出货记录与周度保舱协同", "solution_assurance": 0.83, "service_support_score": 0.82, "booking_success_rate": 0.90, "space_release_success_rate": 0.88, "cost_ratio": 0.79, }, { "supplier_label": "市场订舱代理", "supplier_class": GESLI.ShippingAgencyCompany, "supplier_type": "订舱代理", "product_label": "中转经济", "route_type": "中转", "service_form": "经济包柜", "offering_channel": "市场订舱代理", "agency_level": "二级代理", "capacity_commitment": "按单票确认舱位", "spot_space_status": "需确认", "spot_space_teu": 0, "relationship_tier": "常规合作", "relationship_strength": 0.52, "low_price_access_probability": 0.46, "priority_space_probability": 0.48, "relationship_evidence": "按市场舱位和临时报价执行", "solution_assurance": 0.67, "service_support_score": 0.68, "booking_success_rate": 0.84, "space_release_success_rate": 0.82, "cost_ratio": 0.72, }, { "supplier_label": "区域货代", "supplier_class": GESLI.FreightForwarderSupplier, "supplier_type": "货代公司", "product_label": "加班船直航", "route_type": "直航", "service_form": "加班船包柜", "offering_channel": "区域货代", "agency_level": "合作货代", "capacity_commitment": "加班船窗口优先确认6TEU", "spot_space_status": "现舱有限", "spot_space_teu": 3, "relationship_tier": "稳定合作", "relationship_strength": 0.69, "low_price_access_probability": 0.63, "priority_space_probability": 0.70, "relationship_evidence": "区域市场协同与加班船资源预留", "solution_assurance": 0.80, "service_support_score": 0.79, "booking_success_rate": 0.89, "space_release_success_rate": 0.88, "cost_ratio": 0.82, }, ] return profiles[index % len(profiles)] 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_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 ) def _date_time_literal(value: date) -> Literal: dt = datetime.combine( value, time(hour=12), tzinfo=timezone(timedelta(hours=8)), ) return Literal(dt.isoformat(), datatype=XSD.dateTime)