Eric-ontology/app/mock_data.py

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)