438 lines
20 KiB
Python
Executable File
438 lines
20 KiB
Python
Executable File
# /opt/docker/dev/service_finder/backend/app/schemas/asset.py
|
|
from pydantic import BaseModel, ConfigDict, Field, validator, root_validator, model_validator, field_validator
|
|
from typing import Optional, Dict, Any, List
|
|
from uuid import UUID
|
|
from datetime import datetime
|
|
|
|
class AssetCatalogResponse(BaseModel):
|
|
""" A technikai katalógus (Master Data) teljes adattartalma. """
|
|
id: int
|
|
make: str
|
|
model: str
|
|
generation: Optional[str] = None
|
|
engine_variant: Optional[str] = None
|
|
year_from: Optional[int] = None
|
|
year_to: Optional[int] = None
|
|
vehicle_class: Optional[str] = None
|
|
fuel_type: Optional[str] = None
|
|
|
|
# Technikai paraméterek az automatizáláshoz
|
|
power_kw: Optional[int] = None
|
|
engine_capacity: Optional[int] = None
|
|
max_weight_kg: Optional[int] = None
|
|
axle_count: Optional[int] = None
|
|
euro_class: Optional[str] = None
|
|
body_type: Optional[str] = None
|
|
engine_code: Optional[str] = None
|
|
|
|
factory_data: Dict[str, Any] = Field(default_factory=dict)
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
class AssetResponse(BaseModel):
|
|
""" A konkrét járműpéldány (Asset) teljes válaszmodellje - Thick Digital Twin. """
|
|
# === IDENTIFICATION ===
|
|
id: UUID
|
|
vin: Optional[str] = Field(None, min_length=1, max_length=50)
|
|
license_plate: Optional[str] = None
|
|
name: Optional[str] = None
|
|
catalog_id: Optional[int] = None
|
|
|
|
# === CLASSIFICATION ===
|
|
vehicle_class: Optional[str] = None
|
|
brand: Optional[str] = None
|
|
model: Optional[str] = None
|
|
trim_level: Optional[str] = None
|
|
|
|
# === TECHNICAL SPECS ===
|
|
fuel_type: Optional[str] = None
|
|
engine_capacity: Optional[int] = None
|
|
power_kw: Optional[int] = None
|
|
torque_nm: Optional[int] = None
|
|
cylinder_layout: Optional[str] = None
|
|
transmission_type: Optional[str] = None
|
|
drive_type: Optional[str] = None
|
|
euro_classification: Optional[str] = None
|
|
|
|
# === PHYSICAL DIMENSIONS ===
|
|
curb_weight: Optional[int] = None
|
|
max_weight: Optional[int] = None
|
|
cargo_volume_x: Optional[float] = None
|
|
cargo_volume_y: Optional[float] = None
|
|
door_count: Optional[int] = None
|
|
seat_count: Optional[int] = None
|
|
|
|
# === EQUIPMENT ===
|
|
roof_type: Optional[str] = None
|
|
audio_system_type: Optional[str] = None
|
|
individual_equipment: Dict[str, Any] = Field(default_factory=dict)
|
|
|
|
# === STATUS ===
|
|
current_mileage: int = Field(default=0)
|
|
condition_score: int = Field(default=100)
|
|
status: str
|
|
data_status: Optional[str] = None
|
|
is_verified: bool
|
|
verification_method: Optional[str] = None
|
|
catalog_match_score: Optional[float] = None
|
|
|
|
# === TIMELINE ===
|
|
year_of_manufacture: Optional[int] = None
|
|
first_registration_date: Optional[datetime] = None
|
|
created_at: datetime
|
|
updated_at: Optional[datetime] = None
|
|
|
|
# === WARRANTY ===
|
|
is_under_warranty: bool = Field(default=False, description="Érvényes jótállás")
|
|
warranty_expiry_date: Optional[datetime] = Field(None, description="Jótállás érvényességi dátuma")
|
|
|
|
# === SALES MODULE ===
|
|
is_for_sale: bool = Field(default=False)
|
|
price: Optional[float] = None
|
|
currency: str = Field(default="EUR")
|
|
|
|
# === ORGANIZATION & LOCATION ===
|
|
current_organization_id: Optional[int] = None
|
|
branch_id: Optional[UUID] = None
|
|
relocation_performed: bool = Field(default=False)
|
|
|
|
# === IDENTITY RELATIONSHIPS ===
|
|
owner_organization_id: Optional[int] = None
|
|
operator_person_id: Optional[int] = None
|
|
owner_person_id: Optional[int] = None
|
|
operator_org_id: Optional[int] = None
|
|
|
|
# === CATALOG RELATIONSHIP ===
|
|
catalog: Optional[AssetCatalogResponse] = None
|
|
|
|
# === PROFILE COMPLETION ===
|
|
profile_completion_percentage: int = Field(default=0, ge=0, le=100)
|
|
|
|
# === PRIMARY VEHICLE FLAG ===
|
|
is_primary: bool = Field(default=False, description="Elsődleges jármű (tárolva: individual_equipment JSONB)")
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
|
|
class AssetCreate(BaseModel):
|
|
""" Jármű létrehozásához szükséges adatok - Thick Digital Twin támogatással. """
|
|
# === CORE IDENTIFICATION (Required for status determination) ===
|
|
license_plate: Optional[str] = Field(None, min_length=2, max_length=20, description="Rendszám")
|
|
vin: Optional[str] = Field(None, min_length=1, max_length=50, description="VIN szám (opcionális)")
|
|
|
|
# ── Pre-validator: convert empty strings to None for optional string fields ──
|
|
@validator('license_plate', 'vin', 'trim_level',
|
|
'cylinder_layout', 'transmission_type', 'drive_type',
|
|
'euro_classification', 'roof_type', 'audio_system_type',
|
|
'data_status',
|
|
pre=True, always=True)
|
|
def empty_str_to_none(cls, v):
|
|
"""Convert empty strings to None so Pydantic doesn't reject them with min_length."""
|
|
if v is not None and isinstance(v, str) and v.strip() == '':
|
|
return None
|
|
return v
|
|
|
|
# ── Data normalization validators ──
|
|
@field_validator('brand')
|
|
@classmethod
|
|
def normalize_brand(cls, v: Optional[str]) -> Optional[str]:
|
|
"""Brand: UPPERCASE, strip whitespace."""
|
|
if v is None or not isinstance(v, str):
|
|
return v
|
|
return v.strip().upper()
|
|
|
|
@field_validator('model')
|
|
@classmethod
|
|
def normalize_model(cls, v: Optional[str]) -> Optional[str]:
|
|
"""Model: UPPERCASE, remove ALL spaces."""
|
|
if v is None or not isinstance(v, str):
|
|
return v
|
|
return v.upper().replace(' ', '')
|
|
|
|
# ── Model validator: normalize type_designation inside individual_equipment ──
|
|
@model_validator(mode='after')
|
|
def normalize_individual_equipment(self):
|
|
"""Normalize type_designation inside individual_equipment.car_specs.
|
|
|
|
SAFE None handling: checks every level before calling .upper().
|
|
- individual_equipment may be None or empty dict
|
|
- car_specs may be None or missing key
|
|
- type_designation may be None, empty string, or a valid string
|
|
"""
|
|
if not self.individual_equipment:
|
|
return self
|
|
if not isinstance(self.individual_equipment, dict):
|
|
return self
|
|
|
|
car_specs = self.individual_equipment.get('car_specs')
|
|
if not car_specs or not isinstance(car_specs, dict):
|
|
return self
|
|
|
|
td = car_specs.get('type_designation')
|
|
if td is not None and isinstance(td, str) and td.strip():
|
|
car_specs['type_designation'] = td.strip().upper().replace(' ', '')
|
|
return self
|
|
|
|
# ── Root validator: ensure at least one of vin or license_plate is provided ──
|
|
@root_validator(skip_on_failure=True)
|
|
def validate_vin_or_plate(cls, values):
|
|
"""Legalább egy azonosítót meg kell adni: rendszám VAGY VIN szám."""
|
|
license_plate = values.get('license_plate')
|
|
vin = values.get('vin')
|
|
if not license_plate and not vin:
|
|
raise ValueError('Legalább egy azonosítót meg kell adni: rendszám VAGY VIN szám.')
|
|
return values
|
|
|
|
# === CLASSIFICATION (Optional, but affects status) ===
|
|
brand: Optional[str] = Field(None, max_length=100, description="Márka (ha nincs catalog_id)")
|
|
model: Optional[str] = Field(None, max_length=100, description="Modell (ha nincs catalog_id)")
|
|
vehicle_class: Optional[str] = Field(None, max_length=50, description="Járműosztály")
|
|
fuel_type: Optional[str] = Field(None, max_length=50, description="Üzemanyag típus")
|
|
|
|
# === TECHNICAL SPECS (Optional) ===
|
|
catalog_id: Optional[int] = Field(None, description="Opcionális katalógus ID (ha ismert a modell)")
|
|
engine_capacity: Optional[int] = Field(None, ge=0, description="Hengerűrtartalom (cm³)")
|
|
power_kw: Optional[int] = Field(None, ge=0, description="Teljesítmény (kW)")
|
|
torque_nm: Optional[int] = Field(None, ge=0, description="Nyomaték (Nm)")
|
|
cylinder_layout: Optional[str] = Field(None, max_length=50, description="Hengerelrendezés")
|
|
transmission_type: Optional[str] = Field(None, max_length=50, description="Váltó típus")
|
|
drive_type: Optional[str] = Field(None, max_length=50, description="Hajtás")
|
|
euro_classification: Optional[str] = Field(None, max_length=10, description="EURO besorolás")
|
|
|
|
# === PHYSICAL DIMENSIONS (Optional) ===
|
|
curb_weight: Optional[int] = Field(None, ge=0, description="Saját tömeg (kg)")
|
|
max_weight: Optional[int] = Field(None, ge=0, description="Össztömeg (kg)")
|
|
cargo_volume_x: Optional[float] = Field(None, ge=0, description="Csomagtartó hossz (cm)")
|
|
cargo_volume_y: Optional[float] = Field(None, ge=0, description="Csomagtartó szélesség (cm)")
|
|
door_count: Optional[int] = Field(None, ge=0, description="Ajtók száma")
|
|
seat_count: Optional[int] = Field(None, ge=0, description="Ülések száma")
|
|
|
|
# === EQUIPMENT (Optional) ===
|
|
trim_level: Optional[str] = Field(None, max_length=100, description="Felszereltségi szint")
|
|
roof_type: Optional[str] = Field(None, max_length=50, description="Tető típus")
|
|
audio_system_type: Optional[str] = Field(None, max_length=100, description="Hangrendszer")
|
|
individual_equipment: Optional[Dict[str, Any]] = Field(default_factory=dict, description="Egyedi felszerelések")
|
|
|
|
# === MILEAGE (Optional) ===
|
|
current_mileage: Optional[int] = Field(None, ge=0, description="Aktuális km óra állás")
|
|
|
|
# === TIMELINE (Optional) ===
|
|
year_of_manufacture: Optional[int] = Field(None, ge=1900, le=2100, description="Gyártási év")
|
|
first_registration_date: Optional[datetime] = Field(None, description="Első forgalomba helyezés dátuma")
|
|
|
|
# === ORGANIZATION (Optional) ===
|
|
organization_id: Optional[int] = Field(None, description="Szervezet ID (alapértelmezett a felhasználó szervezete)")
|
|
branch_id: Optional[UUID] = Field(None, description="Garázs (Branch) ID, ahova a járművet rendeljük. Ha nincs megadva, a szervezet központi garázsába kerül.")
|
|
|
|
# === WARRANTY ===
|
|
is_under_warranty: bool = Field(default=False, description="Érvényes jótállás")
|
|
warranty_expiry_date: Optional[datetime] = Field(None, description="Jótállás érvényességi dátuma")
|
|
|
|
# === PRIMARY VEHICLE FLAG ===
|
|
is_primary: bool = Field(default=False, description="Elsődleges jármű (tárolva: individual_equipment JSONB)")
|
|
|
|
# === DATA STATUS (for transfer/duplicate scenarios) ===
|
|
data_status: Optional[str] = Field(None, description="Adat státusz (pl. 'draft' tulajdonosváltáskor)")
|
|
|
|
# === STATUS VALIDATION ===
|
|
@validator('status', pre=True, always=True)
|
|
def determine_status(cls, v, values):
|
|
"""Automatikus státusz meghatározás az adatkomplettség alapján."""
|
|
if v is not None:
|
|
return v
|
|
|
|
# Ellenőrizzük az 5 alapvető mezőt
|
|
required_fields = ['license_plate', 'brand', 'model', 'vehicle_class', 'fuel_type']
|
|
has_all_required = all(
|
|
values.get(field) is not None and str(values.get(field)).strip() != ''
|
|
for field in required_fields
|
|
)
|
|
|
|
return "active" if has_all_required else "draft"
|
|
|
|
# === COMPUTED FIELD: status ===
|
|
status: Optional[str] = Field(None, description="Automatikusan számított státusz (draft/active)")
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
|
|
class AssetUpdate(BaseModel):
|
|
""" Jármű adatainak frissítése - minden mező opcionális. """
|
|
# === CORE IDENTIFICATION ===
|
|
license_plate: Optional[str] = Field(None, min_length=2, max_length=20, description="Rendszám")
|
|
vin: Optional[str] = Field(None, min_length=1, max_length=50, description="VIN szám")
|
|
|
|
# ── Pre-validator: convert empty strings to None for optional string fields ──
|
|
@validator('license_plate', 'vin', 'trim_level',
|
|
'cylinder_layout', 'transmission_type', 'drive_type',
|
|
'euro_classification', 'roof_type', 'audio_system_type',
|
|
pre=True, always=True)
|
|
def empty_str_to_none(cls, v):
|
|
"""Convert empty strings to None so Pydantic doesn't reject them with min_length."""
|
|
if v is not None and isinstance(v, str) and v.strip() == '':
|
|
return None
|
|
return v
|
|
|
|
# ── Data normalization validators ──
|
|
@field_validator('brand')
|
|
@classmethod
|
|
def normalize_brand(cls, v: Optional[str]) -> Optional[str]:
|
|
"""Brand: UPPERCASE, strip whitespace."""
|
|
if v is None or not isinstance(v, str):
|
|
return v
|
|
return v.strip().upper()
|
|
|
|
@field_validator('model')
|
|
@classmethod
|
|
def normalize_model(cls, v: Optional[str]) -> Optional[str]:
|
|
"""Model: UPPERCASE, remove ALL spaces."""
|
|
if v is None or not isinstance(v, str):
|
|
return v
|
|
return v.upper().replace(' ', '')
|
|
|
|
# ── Model validator: normalize type_designation inside individual_equipment ──
|
|
@model_validator(mode='after')
|
|
def normalize_individual_equipment(self):
|
|
"""Normalize type_designation inside individual_equipment.car_specs.
|
|
|
|
SAFE None handling: checks every level before calling .upper().
|
|
- individual_equipment may be None or empty dict
|
|
- car_specs may be None or missing key
|
|
- type_designation may be None, empty string, or a valid string
|
|
"""
|
|
if not self.individual_equipment:
|
|
return self
|
|
if not isinstance(self.individual_equipment, dict):
|
|
return self
|
|
|
|
car_specs = self.individual_equipment.get('car_specs')
|
|
if not car_specs or not isinstance(car_specs, dict):
|
|
return self
|
|
|
|
td = car_specs.get('type_designation')
|
|
if td is not None and isinstance(td, str) and td.strip():
|
|
car_specs['type_designation'] = td.strip().upper().replace(' ', '')
|
|
return self
|
|
|
|
# ── Model validator: ensure at least one of vin or license_plate is provided ──
|
|
@model_validator(mode='after')
|
|
def validate_vin_or_plate(self):
|
|
"""Update esetén: ha mindkét mezőt None-ra állítják explicit, az nem engedélyezett.
|
|
|
|
Működés:
|
|
- Ha egyik mező sincs a payload-ban (üres update) → engedélyezve (nincs változás)
|
|
- Ha mindkettő None explicit → hiba (nem lehet mindkettőt elvenni)
|
|
- Ha legalább az egyik meg van adva → OK
|
|
"""
|
|
# model_fields_set tartalmazza az explicit megadott mezőket
|
|
if 'license_plate' not in self.model_fields_set and 'vin' not in self.model_fields_set:
|
|
return self
|
|
|
|
if not self.license_plate and not self.vin:
|
|
raise ValueError('Legalább egy azonosítót meg kell adni: rendszám VAGY VIN szám.')
|
|
return self
|
|
|
|
# === CLASSIFICATION ===
|
|
brand: Optional[str] = Field(None, max_length=100, description="Márka")
|
|
model: Optional[str] = Field(None, max_length=100, description="Modell")
|
|
vehicle_class: Optional[str] = Field(None, max_length=50, description="Járműosztály")
|
|
fuel_type: Optional[str] = Field(None, max_length=50, description="Üzemanyag típus")
|
|
|
|
# === TECHNICAL SPECS ===
|
|
catalog_id: Optional[int] = Field(None, description="Katalógus ID")
|
|
engine_capacity: Optional[int] = Field(None, ge=0, description="Hengerűrtartalom (cm³)")
|
|
power_kw: Optional[int] = Field(None, ge=0, description="Teljesítmény (kW)")
|
|
torque_nm: Optional[int] = Field(None, ge=0, description="Nyomaték (Nm)")
|
|
cylinder_layout: Optional[str] = Field(None, max_length=50, description="Hengerelrendezés")
|
|
transmission_type: Optional[str] = Field(None, max_length=50, description="Váltó típus")
|
|
drive_type: Optional[str] = Field(None, max_length=50, description="Hajtás")
|
|
euro_classification: Optional[str] = Field(None, max_length=10, description="EURO besorolás")
|
|
|
|
# === PHYSICAL DIMENSIONS ===
|
|
curb_weight: Optional[int] = Field(None, ge=0, description="Saját tömeg (kg)")
|
|
max_weight: Optional[int] = Field(None, ge=0, description="Össztömeg (kg)")
|
|
cargo_volume_x: Optional[float] = Field(None, ge=0, description="Csomagtartó hossz (cm)")
|
|
cargo_volume_y: Optional[float] = Field(None, ge=0, description="Csomagtartó szélesség (cm)")
|
|
door_count: Optional[int] = Field(None, ge=0, description="Ajtók száma")
|
|
seat_count: Optional[int] = Field(None, ge=0, description="Ülések száma")
|
|
|
|
# === EQUIPMENT ===
|
|
trim_level: Optional[str] = Field(None, max_length=100, description="Felszereltségi szint")
|
|
roof_type: Optional[str] = Field(None, max_length=50, description="Tető típus")
|
|
audio_system_type: Optional[str] = Field(None, max_length=100, description="Hangrendszer")
|
|
individual_equipment: Optional[Dict[str, Any]] = Field(None, description="Egyedi felszerelések")
|
|
|
|
# === TIMELINE ===
|
|
year_of_manufacture: Optional[int] = Field(None, ge=1900, le=2100, description="Gyártási év")
|
|
first_registration_date: Optional[datetime] = Field(None, description="Első forgalomba helyezés dátuma")
|
|
|
|
# === WARRANTY ===
|
|
is_under_warranty: Optional[bool] = Field(None, description="Érvényes jótállás")
|
|
warranty_expiry_date: Optional[datetime] = Field(None, description="Jótállás érvényességi dátuma")
|
|
|
|
# === STATUS ===
|
|
current_mileage: Optional[int] = Field(None, ge=0, description="Aktuális km óra állás")
|
|
is_primary: Optional[bool] = Field(None, description="Elsődleges jármű")
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
|
|
class AssetEventCreate(BaseModel):
|
|
"""Schema for creating a new AssetEvent (Service Book entry)."""
|
|
event_type: str = Field(..., description="Event type: SERVICE, REPAIR, ACCIDENT, INSPECTION, TIRE_CHANGE, MAINTENANCE, UPGRADE, RECALL")
|
|
event_date: Optional[datetime] = Field(None, description="Event date (default: now)")
|
|
odometer_reading: Optional[int] = Field(None, ge=0, description="Odometer reading at event time")
|
|
description: Optional[str] = Field(None, max_length=2000, description="Event description")
|
|
cost_id: Optional[UUID] = Field(None, description="Associated cost record ID")
|
|
cost_amount: Optional[float] = Field(None, gt=0, description="Cost amount in the given currency. If > 0, an AssetCost record is auto-created.")
|
|
currency: Optional[str] = Field(None, min_length=3, max_length=3, description="Currency code (e.g. HUF, EUR). Defaults to HUF on backend.")
|
|
status: Optional[str] = Field(None, description="Event status (DRAFT, MISSING_TECH_DATA, COMPLETED) - auto-set by backend")
|
|
linked_expense_id: Optional[UUID] = Field(None, description="Associated AssetCost ID (bidirectional link)")
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
|
|
class AssetEventResponse(BaseModel):
|
|
"""Schema for returning an AssetEvent."""
|
|
id: UUID
|
|
asset_id: UUID
|
|
user_id: Optional[int] = None
|
|
organization_id: Optional[int] = None
|
|
event_type: str
|
|
odometer_reading: Optional[int] = None
|
|
description: Optional[str] = None
|
|
cost_id: Optional[UUID] = None
|
|
cost_amount: Optional[float] = Field(None, description="Auto-created cost amount")
|
|
currency: Optional[str] = Field(None, description="Currency code (e.g. HUF)")
|
|
status: str = "DRAFT"
|
|
linked_expense_id: Optional[UUID] = None
|
|
event_date: datetime
|
|
created_at: datetime
|
|
updated_at: Optional[datetime] = None
|
|
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
@classmethod
|
|
def model_validate(cls, obj, **kwargs):
|
|
"""Override to resolve cost_amount and currency from the related AssetCost.
|
|
|
|
GROSS-FIRST: cost_amount resolves from amount_gross (Bruttó) as primary source.
|
|
"""
|
|
data = {}
|
|
if hasattr(obj, '__dict__'):
|
|
# Copy ORM attributes
|
|
for field in cls.model_fields:
|
|
if hasattr(obj, field):
|
|
data[field] = getattr(obj, field)
|
|
|
|
# Resolve cost_amount and currency from the cost relationship
|
|
cost = getattr(obj, 'cost', None)
|
|
if cost is not None:
|
|
if data.get('cost_amount') is None:
|
|
# GROSS-FIRST: prefer amount_gross as the source of truth
|
|
data['cost_amount'] = float(cost.amount_gross) if cost.amount_gross is not None else (
|
|
float(cost.amount_net) if cost.amount_net is not None else None
|
|
)
|
|
if data.get('currency') is None:
|
|
data['currency'] = cost.currency
|
|
return cls(**data) |