113 lines
3.2 KiB
Python
113 lines
3.2 KiB
Python
"""Risk prediction request and response contracts."""
|
|
|
|
from pydantic import Field
|
|
|
|
from app.domain.risk_engine import RiskLevel
|
|
from app.schemas.common import ApiSchema
|
|
|
|
|
|
class RiskIndicator(ApiSchema):
|
|
name: str = Field(min_length=1, max_length=64)
|
|
value: float = Field(ge=0, le=100)
|
|
weight: float = Field(default=1, gt=0, le=100)
|
|
|
|
|
|
class RiskAssessmentRequest(ApiSchema):
|
|
subject_id: str = Field(min_length=1, max_length=128)
|
|
indicators: list[RiskIndicator] = Field(default_factory=list, max_length=50)
|
|
product_category: str | None = Field(default=None, max_length=100)
|
|
material_type: str | None = Field(default=None, max_length=200)
|
|
supplier: str | None = Field(default=None, max_length=300)
|
|
factory: str | None = Field(default=None, max_length=300)
|
|
stage_new: str | None = Field(default=None, max_length=100)
|
|
defect_level: str | None = Field(default=None, max_length=10)
|
|
|
|
|
|
class RiskFactorResult(ApiSchema):
|
|
name: str
|
|
value: float
|
|
weight: float
|
|
contribution: float
|
|
|
|
|
|
class CategoryWarning(ApiSchema):
|
|
category: str
|
|
anomaly_count: int
|
|
risk_level: str
|
|
similar_category: str | None = None
|
|
similar_count: int | None = None
|
|
suggestion: str | None = None
|
|
|
|
|
|
class SupplierFrequency(ApiSchema):
|
|
supplier_name: str | None = None
|
|
cooperation_count: int | None = None
|
|
anomaly_count: int | None = None
|
|
alpha: float | None = None
|
|
tier: str | None = None
|
|
suggestion: str | None = None
|
|
frequency_basis: str = "暂无匹配采购订单"
|
|
note: str | None = None
|
|
data_available: bool = False
|
|
|
|
|
|
class StandardJudgment(ApiSchema):
|
|
category: str
|
|
risk_point: str
|
|
rule: str
|
|
source: str | None = None
|
|
status: str = "待检测"
|
|
|
|
|
|
class RiskDimensionScore(ApiSchema):
|
|
dimension: str
|
|
score: float
|
|
weight: float
|
|
status: str = "database"
|
|
reason: str | None = None
|
|
sample_size: int | None = None
|
|
raw_value: float | int | None = None
|
|
|
|
|
|
class SimilarCase(ApiSchema):
|
|
case_id: int | None = None
|
|
date: str | None = None
|
|
product: str | None = None
|
|
category: str | None = None
|
|
material: str | None = None
|
|
anomaly_type: str | None = None
|
|
description: str | None = None
|
|
root_cause: str | None = None
|
|
defect_level: str | None = None
|
|
problem: str | None = None
|
|
solution: str | None = None
|
|
similarity: float | None = None
|
|
|
|
|
|
class RiskAssessmentResult(ApiSchema):
|
|
subject_id: str
|
|
score: float
|
|
level: RiskLevel
|
|
model_version: str
|
|
factors: list[RiskFactorResult]
|
|
risk_badge: str | None = None
|
|
alert_text: str | None = None
|
|
category_warning: CategoryWarning | None = None
|
|
supplier_freq: SupplierFrequency | None = None
|
|
standard_judgment: list[StandardJudgment] = Field(default_factory=list)
|
|
dimension_scores: list[RiskDimensionScore] = Field(default_factory=list)
|
|
similar_cases: list[SimilarCase] = Field(default_factory=list)
|
|
ai_suggestions: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class ModelInfo(ApiSchema):
|
|
model_version: str
|
|
score_range: str = "0-100"
|
|
levels: tuple[RiskLevel, ...] = (
|
|
RiskLevel.LOW,
|
|
RiskLevel.MODERATELY_LOW,
|
|
RiskLevel.MEDIUM,
|
|
RiskLevel.HIGH,
|
|
RiskLevel.VERY_HIGH,
|
|
)
|