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=2, ge=1, le=5) customer_count: int = Field(default=1, ge=0, le=3) 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", "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}"] supplier_name = f"内存模拟供应商 {token[-3:].upper()}{index + 1}" product_name = ( f"{request.origin}-{request.destination}动态Mock方案{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 * (0.76 + index * 0.025), -2) success_rate = min(0.98, 0.87 + index * 0.035) product_class = ( GESLI.AirFreightProduct if transport_mode == "空运" else GESLI.ContainerOceanProduct ) route_type = "直达" if transport_mode == "空运" and index == 0 else ( "直航" if index % 2 == 0 else "中转" ) 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, GESLI.FreightForwarderSupplier), (supplier, RDFS.label, Literal(supplier_name, lang="zh")), (supplier, GESLI.platformSupplierType, Literal("内存Mock供应商")), (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("Mock标准服务")), (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("本地动态Mock")), (offering, GESLI.agencyLevel, Literal("一级模拟代理")), (offering, GESLI.capacityCommitment, Literal(f"模拟可用运力{10 + index * 5}单位")), (offering, GESLI.capacityAvailable, Literal(True)), (offering, GESLI.supportsDangerousGoods, Literal(index == 0)), (offering, GESLI.bookingSuccessRate, Literal(success_rate, datatype=XSD.decimal)), (offering, GESLI.spaceReleaseSuccessRate, Literal(success_rate - 0.01, 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 (14 + index * 2) 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)