cada96e874
- math.isfinite() guard в out_of_range check (NaN bypass: NaN<MIN/MAX → False) - Same guard в weight_profiles._validate_weights_dict (pre-existing dup bug) - _POI_WEIGHTS теперь импорт из weight_profiles._SYSTEM_POI_WEIGHTS (SSOT) - Test test_inline_weights_rejects_nan (raw NaN payload → 422)
339 lines
11 KiB
Python
339 lines
11 KiB
Python
"""User weight profiles для Site Finder POI scoring.
|
||
|
||
Per #114 (Макс feedback): кастомизируемые веса для каждой POI категории.
|
||
Table schema: data/sql/90_user_weight_profiles.sql.
|
||
|
||
API surface:
|
||
- list_profiles(db, user_id) → list[WeightProfile]
|
||
- get_profile(db, user_id, profile_id) → WeightProfile | None
|
||
- get_default_profile(db, user_id) → WeightProfile | None
|
||
- create_profile(db, payload) → WeightProfile
|
||
- update_profile(db, user_id, profile_id, payload) → WeightProfile | None
|
||
- delete_profile(db, user_id, profile_id) → bool
|
||
- resolve_weights(db, user_id, profile_id) → dict[str, float]
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import math
|
||
from datetime import datetime
|
||
from typing import Any
|
||
|
||
from pydantic import BaseModel, Field, field_validator
|
||
from sqlalchemy import text
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Allowed POI categories — single source of truth; imported by api/v1/parcels.py
|
||
ALLOWED_CATEGORIES: set[str] = {
|
||
"school",
|
||
"kindergarten",
|
||
"pharmacy",
|
||
"hospital",
|
||
"shop_mall",
|
||
"shop_supermarket",
|
||
"shop_small",
|
||
"park",
|
||
"bus_stop",
|
||
"metro_stop",
|
||
"tram_stop",
|
||
}
|
||
|
||
# Weight value bounds (per #114 spec)
|
||
MIN_WEIGHT: float = -2.0
|
||
MAX_WEIGHT: float = 3.0
|
||
|
||
# System defaults — single source of truth; imported as _POI_WEIGHTS by api/v1/parcels.py
|
||
_SYSTEM_POI_WEIGHTS: dict[str, float] = {
|
||
"school": 1.5,
|
||
"kindergarten": 1.5,
|
||
"pharmacy": 0.8,
|
||
"hospital": 0.6,
|
||
"shop_mall": 1.2,
|
||
"shop_supermarket": 1.0,
|
||
"shop_small": 0.5,
|
||
"park": 1.8,
|
||
"bus_stop": 0.3,
|
||
"metro_stop": 1.5,
|
||
"tram_stop": -0.5,
|
||
}
|
||
|
||
# ── SQL helpers ────────────────────────────────────────────────────────────────
|
||
|
||
_SELECT_COLS = """
|
||
id, user_id, profile_name, weights, is_default, description,
|
||
created_at, updated_at
|
||
"""
|
||
|
||
_SELECT_BY_USER = f"""
|
||
SELECT {_SELECT_COLS}
|
||
FROM user_weight_profiles
|
||
WHERE user_id = :user_id
|
||
ORDER BY is_default DESC, id ASC
|
||
"""
|
||
|
||
_SELECT_BY_ID = f"""
|
||
SELECT {_SELECT_COLS}
|
||
FROM user_weight_profiles
|
||
WHERE user_id = :user_id
|
||
AND id = :profile_id
|
||
"""
|
||
|
||
_SELECT_DEFAULT = f"""
|
||
SELECT {_SELECT_COLS}
|
||
FROM user_weight_profiles
|
||
WHERE user_id = :user_id
|
||
AND is_default = TRUE
|
||
LIMIT 1
|
||
"""
|
||
|
||
_UNSET_DEFAULT = """
|
||
UPDATE user_weight_profiles
|
||
SET is_default = FALSE
|
||
WHERE user_id = :user_id
|
||
AND is_default = TRUE
|
||
"""
|
||
|
||
_INSERT = """
|
||
INSERT INTO user_weight_profiles
|
||
(user_id, profile_name, weights, is_default, description)
|
||
VALUES
|
||
(:user_id, :profile_name, CAST(:weights AS jsonb), :is_default, :description)
|
||
RETURNING id, created_at, updated_at
|
||
"""
|
||
|
||
|
||
# ── Pydantic models ────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _validate_weights_dict(v: dict[str, float]) -> dict[str, float]:
|
||
"""Shared validation logic for weights dict."""
|
||
bad_keys = set(v.keys()) - ALLOWED_CATEGORIES
|
||
if bad_keys:
|
||
raise ValueError(f"Unknown POI categories: {sorted(bad_keys)}")
|
||
for k, w in v.items():
|
||
if not isinstance(w, int | float):
|
||
raise ValueError(f"Weight for '{k}' must be number, got {type(w).__name__}")
|
||
if not math.isfinite(w) or w < MIN_WEIGHT or w > MAX_WEIGHT:
|
||
raise ValueError(f"Weight for '{k}' = {w} out of bounds [{MIN_WEIGHT}, {MAX_WEIGHT}]")
|
||
return v
|
||
|
||
|
||
class WeightProfileBase(BaseModel):
|
||
profile_name: str = Field(..., min_length=1, max_length=64)
|
||
weights: dict[str, float] = Field(default_factory=dict)
|
||
is_default: bool = False
|
||
description: str | None = None
|
||
|
||
@field_validator("weights")
|
||
@classmethod
|
||
def _validate_weights(cls, v: dict[str, float]) -> dict[str, float]:
|
||
return _validate_weights_dict(v)
|
||
|
||
|
||
class WeightProfileCreate(WeightProfileBase):
|
||
user_id: str = Field(..., min_length=1, max_length=128)
|
||
|
||
|
||
class WeightProfileUpdate(BaseModel):
|
||
profile_name: str | None = Field(None, min_length=1, max_length=64)
|
||
weights: dict[str, float] | None = None
|
||
is_default: bool | None = None
|
||
description: str | None = None
|
||
|
||
@field_validator("weights")
|
||
@classmethod
|
||
def _validate_weights(cls, v: dict[str, float] | None) -> dict[str, float] | None:
|
||
if v is None:
|
||
return v
|
||
return _validate_weights_dict(v)
|
||
|
||
|
||
class WeightProfile(WeightProfileBase):
|
||
id: int
|
||
user_id: str
|
||
created_at: datetime
|
||
updated_at: datetime
|
||
|
||
model_config = {"from_attributes": True}
|
||
|
||
|
||
# ── Row mapper ─────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _row_to_profile(r: Any) -> WeightProfile:
|
||
"""Map SQLAlchemy mapping row → WeightProfile."""
|
||
weights_raw = r["weights"]
|
||
if isinstance(weights_raw, str):
|
||
weights_raw = json.loads(weights_raw)
|
||
weights: dict[str, float] = weights_raw or {}
|
||
|
||
return WeightProfile(
|
||
id=r["id"],
|
||
user_id=r["user_id"],
|
||
profile_name=r["profile_name"],
|
||
weights=weights,
|
||
is_default=bool(r["is_default"]),
|
||
description=r["description"],
|
||
created_at=r["created_at"],
|
||
updated_at=r["updated_at"],
|
||
)
|
||
|
||
|
||
# ── CRUD service ───────────────────────────────────────────────────────────────
|
||
|
||
|
||
def list_profiles(db: Any, user_id: str) -> list[WeightProfile]:
|
||
"""Вернуть все профили пользователя, default первым."""
|
||
rows = db.execute(text(_SELECT_BY_USER), {"user_id": user_id}).mappings().all()
|
||
return [_row_to_profile(r) for r in rows]
|
||
|
||
|
||
def get_profile(db: Any, user_id: str, profile_id: int) -> WeightProfile | None:
|
||
"""Вернуть профиль по id (scoped к пользователю)."""
|
||
row = (
|
||
db.execute(text(_SELECT_BY_ID), {"user_id": user_id, "profile_id": profile_id})
|
||
.mappings()
|
||
.first()
|
||
)
|
||
if row is None:
|
||
return None
|
||
return _row_to_profile(row)
|
||
|
||
|
||
def get_default_profile(db: Any, user_id: str) -> WeightProfile | None:
|
||
"""Вернуть профиль-default пользователя или None."""
|
||
row = db.execute(text(_SELECT_DEFAULT), {"user_id": user_id}).mappings().first()
|
||
if row is None:
|
||
return None
|
||
return _row_to_profile(row)
|
||
|
||
|
||
def create_profile(db: Any, payload: WeightProfileCreate) -> WeightProfile:
|
||
"""Создать новый профиль.
|
||
|
||
Если is_default=True — снять is_default у всех существующих профилей
|
||
пользователя в одной транзакции.
|
||
"""
|
||
if payload.is_default:
|
||
db.execute(text(_UNSET_DEFAULT), {"user_id": payload.user_id})
|
||
|
||
result = (
|
||
db.execute(
|
||
text(_INSERT),
|
||
{
|
||
"user_id": payload.user_id,
|
||
"profile_name": payload.profile_name,
|
||
"weights": json.dumps(payload.weights, ensure_ascii=False),
|
||
"is_default": payload.is_default,
|
||
"description": payload.description,
|
||
},
|
||
)
|
||
.mappings()
|
||
.first()
|
||
)
|
||
|
||
db.commit()
|
||
|
||
assert result is not None, "INSERT RETURNING вернул пустой результат"
|
||
|
||
return WeightProfile(
|
||
id=result["id"],
|
||
user_id=payload.user_id,
|
||
profile_name=payload.profile_name,
|
||
weights=payload.weights,
|
||
is_default=payload.is_default,
|
||
description=payload.description,
|
||
created_at=result["created_at"],
|
||
updated_at=result["updated_at"],
|
||
)
|
||
|
||
|
||
def update_profile(
|
||
db: Any, user_id: str, profile_id: int, payload: WeightProfileUpdate
|
||
) -> WeightProfile | None:
|
||
"""Обновить поля профиля (PATCH-style). Вернуть None если не найден.
|
||
|
||
При установке is_default=True — снять is_default у остальных профилей
|
||
пользователя в одной транзакции.
|
||
"""
|
||
existing = get_profile(db, user_id, profile_id)
|
||
if existing is None:
|
||
return None
|
||
|
||
sets: list[str] = ["updated_at = NOW()"]
|
||
params: dict[str, Any] = {"user_id": user_id, "profile_id": profile_id}
|
||
|
||
if payload.profile_name is not None:
|
||
sets.append("profile_name = :profile_name")
|
||
params["profile_name"] = payload.profile_name
|
||
|
||
if payload.weights is not None:
|
||
sets.append("weights = CAST(:weights AS jsonb)")
|
||
params["weights"] = json.dumps(payload.weights, ensure_ascii=False)
|
||
|
||
if payload.description is not None:
|
||
sets.append("description = :description")
|
||
params["description"] = payload.description
|
||
|
||
if payload.is_default is not None:
|
||
sets.append("is_default = :is_default")
|
||
params["is_default"] = payload.is_default
|
||
if payload.is_default:
|
||
db.execute(text(_UNSET_DEFAULT), {"user_id": user_id})
|
||
|
||
if len(sets) > 1: # есть что обновлять кроме updated_at
|
||
db.execute(
|
||
text(
|
||
f"UPDATE user_weight_profiles SET {', '.join(sets)}"
|
||
" WHERE user_id = :user_id AND id = :profile_id"
|
||
),
|
||
params,
|
||
)
|
||
db.commit()
|
||
|
||
return get_profile(db, user_id, profile_id)
|
||
|
||
|
||
def delete_profile(db: Any, user_id: str, profile_id: int) -> bool:
|
||
"""Удалить профиль. Вернуть True если удалён, False если не найден."""
|
||
result = db.execute(
|
||
text(
|
||
"DELETE FROM user_weight_profiles"
|
||
" WHERE user_id = :user_id AND id = :profile_id"
|
||
" RETURNING id"
|
||
),
|
||
{"user_id": user_id, "profile_id": profile_id},
|
||
).first()
|
||
if result is None:
|
||
return False
|
||
db.commit()
|
||
return True
|
||
|
||
|
||
def resolve_weights(db: Any, user_id: str | None, profile_id: int | None) -> dict[str, float]:
|
||
"""Вернуть эффективные веса для analyze_parcel.
|
||
|
||
Порядок приоритетов:
|
||
1. profile_id задан → загрузить именно этот профиль
|
||
2. user_id задан → загрузить default-профиль пользователя
|
||
3. Иначе → вернуть системные значения _SYSTEM_POI_WEIGHTS
|
||
"""
|
||
if profile_id is not None and user_id is not None:
|
||
profile = get_profile(db, user_id, profile_id)
|
||
if profile is not None and profile.weights:
|
||
logger.debug(
|
||
"resolve_weights: user=%s profile_id=%s → custom weights", user_id, profile_id
|
||
)
|
||
return dict(profile.weights)
|
||
|
||
if user_id is not None:
|
||
profile = get_default_profile(db, user_id)
|
||
if profile is not None and profile.weights:
|
||
logger.debug("resolve_weights: user=%s → default profile weights", user_id)
|
||
return dict(profile.weights)
|
||
|
||
logger.debug("resolve_weights: returning system defaults")
|
||
return dict(_SYSTEM_POI_WEIGHTS)
|