328 lines
10 KiB
Python
328 lines
10 KiB
Python
"""Глобальные настройки контроля отклонений (recipes.db, singleton)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
from copy import deepcopy
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Tuple
|
|
|
|
from config import Config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SETTINGS_VERSION = 1
|
|
SETTINGS_ROW_ID = 1
|
|
|
|
DEFAULT_SETTINGS: Dict[str, Any] = {
|
|
"version": SETTINGS_VERSION,
|
|
"loading": {
|
|
"enabled": True,
|
|
"warning_pct": 10.0,
|
|
"critical_pct": 15.0,
|
|
"warning_min_sec": 3,
|
|
"critical_min_sec": 1,
|
|
"warning_max_sec": 180,
|
|
"critical_max_sec": 600,
|
|
"notify_warning": True,
|
|
"notify_critical": True,
|
|
},
|
|
"unloading": {
|
|
"enabled": True,
|
|
"warning_pct": 5.0,
|
|
"critical_pct": 10.0,
|
|
"notify_warning": True,
|
|
"notify_critical": True,
|
|
},
|
|
"mix_time": {
|
|
"enabled": True,
|
|
"warning_delta_sec": 30,
|
|
"critical_delta_sec": 60,
|
|
"notify_warning": True,
|
|
"notify_critical": True,
|
|
},
|
|
"left_in_mixer": {
|
|
"enabled": True,
|
|
"warning_min_kg": 5.0,
|
|
"warning_min_pct": 2.0,
|
|
"critical_min_kg": 15.0,
|
|
"notify_warning": True,
|
|
"notify_critical": True,
|
|
},
|
|
}
|
|
|
|
|
|
def get_legacy_json_path() -> Path:
|
|
custom = os.getenv("WESP_FEED_QUALITY_SETTINGS_PATH", "").strip()
|
|
if custom:
|
|
p = Path(custom)
|
|
if not p.is_absolute():
|
|
p = Path(Config.BASE_DIR) / p
|
|
return p
|
|
return Path(Config.DATA_DIR) / "wesp_feed_quality_settings.json"
|
|
|
|
|
|
def default_settings() -> Dict[str, Any]:
|
|
return deepcopy(DEFAULT_SETTINGS)
|
|
|
|
|
|
def _coerce_bool(value: Any, default: bool) -> bool:
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, str):
|
|
return value.strip().lower() in ("1", "true", "yes", "on")
|
|
if isinstance(value, (int, float)):
|
|
return bool(value)
|
|
return default
|
|
|
|
|
|
def _coerce_float(value: Any, default: float, *, min_v: float, max_v: float) -> float:
|
|
try:
|
|
num = float(value)
|
|
except (TypeError, ValueError):
|
|
num = default
|
|
return max(min_v, min(max_v, num))
|
|
|
|
|
|
def _coerce_int(value: Any, default: int, *, min_v: int, max_v: int) -> int:
|
|
try:
|
|
num = int(round(float(value)))
|
|
except (TypeError, ValueError):
|
|
num = default
|
|
return max(min_v, min(max_v, num))
|
|
|
|
|
|
def _merge_section(
|
|
raw: Dict[str, Any],
|
|
defaults: Dict[str, Any],
|
|
*,
|
|
float_fields: Tuple[Tuple[str, float, float, float], ...] = (),
|
|
int_fields: Tuple[Tuple[str, int, int, int], ...] = (),
|
|
) -> Dict[str, Any]:
|
|
out = deepcopy(defaults)
|
|
if not isinstance(raw, dict):
|
|
return out
|
|
if "enabled" in raw:
|
|
out["enabled"] = _coerce_bool(raw.get("enabled"), bool(defaults.get("enabled", True)))
|
|
for key, default, min_v, max_v in float_fields:
|
|
if key in raw:
|
|
out[key] = _coerce_float(raw.get(key), float(defaults[key]), min_v=min_v, max_v=max_v)
|
|
for key, default, min_v, max_v in int_fields:
|
|
if key in raw:
|
|
out[key] = _coerce_int(raw.get(key), int(defaults[key]), min_v=min_v, max_v=max_v)
|
|
for notify_key in ("notify_warning", "notify_critical"):
|
|
if notify_key in raw:
|
|
out[notify_key] = _coerce_bool(raw.get(notify_key), bool(defaults.get(notify_key, True)))
|
|
return out
|
|
|
|
|
|
def normalize_settings(partial: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Валидирует и нормализует настройки; critical >= warning где применимо."""
|
|
base = default_settings()
|
|
raw = partial if isinstance(partial, dict) else {}
|
|
|
|
loading = _merge_section(
|
|
raw.get("loading") or {},
|
|
base["loading"],
|
|
float_fields=(
|
|
("warning_pct", 10.0, 0.1, 100.0),
|
|
("critical_pct", 15.0, 0.1, 100.0),
|
|
),
|
|
int_fields=(
|
|
("warning_min_sec", 3, 0, 3600),
|
|
("critical_min_sec", 1, 0, 3600),
|
|
("warning_max_sec", 180, 0, 86400),
|
|
("critical_max_sec", 600, 0, 86400),
|
|
),
|
|
)
|
|
if loading["critical_pct"] < loading["warning_pct"]:
|
|
loading["critical_pct"] = loading["warning_pct"]
|
|
if loading["critical_min_sec"] > loading["warning_min_sec"]:
|
|
loading["critical_min_sec"] = loading["warning_min_sec"]
|
|
if loading["critical_max_sec"] < loading["warning_max_sec"]:
|
|
loading["critical_max_sec"] = loading["warning_max_sec"]
|
|
|
|
unloading = _merge_section(
|
|
raw.get("unloading") or {},
|
|
base["unloading"],
|
|
float_fields=(
|
|
("warning_pct", 5.0, 0.1, 100.0),
|
|
("critical_pct", 10.0, 0.1, 100.0),
|
|
),
|
|
)
|
|
if unloading["critical_pct"] < unloading["warning_pct"]:
|
|
unloading["critical_pct"] = unloading["warning_pct"]
|
|
|
|
mix_time = _merge_section(
|
|
raw.get("mix_time") or {},
|
|
base["mix_time"],
|
|
int_fields=(
|
|
("warning_delta_sec", 30, 1, 3600),
|
|
("critical_delta_sec", 60, 1, 3600),
|
|
),
|
|
)
|
|
if mix_time["critical_delta_sec"] < mix_time["warning_delta_sec"]:
|
|
mix_time["critical_delta_sec"] = mix_time["warning_delta_sec"]
|
|
|
|
left_in_mixer = _merge_section(
|
|
raw.get("left_in_mixer") or {},
|
|
base["left_in_mixer"],
|
|
float_fields=(
|
|
("warning_min_kg", 5.0, 0.0, 10000.0),
|
|
("warning_min_pct", 2.0, 0.0, 100.0),
|
|
("critical_min_kg", 15.0, 0.0, 10000.0),
|
|
),
|
|
)
|
|
if left_in_mixer["critical_min_kg"] < left_in_mixer["warning_min_kg"]:
|
|
left_in_mixer["critical_min_kg"] = left_in_mixer["warning_min_kg"]
|
|
|
|
return {
|
|
"version": SETTINGS_VERSION,
|
|
"loading": loading,
|
|
"unloading": unloading,
|
|
"mix_time": mix_time,
|
|
"left_in_mixer": left_in_mixer,
|
|
}
|
|
|
|
|
|
def _read_legacy_json_raw() -> Dict[str, Any]:
|
|
path = get_legacy_json_path()
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
return data if isinstance(data, dict) else {}
|
|
except Exception:
|
|
logger.warning("Не удалось прочитать %s", path, exc_info=True)
|
|
return {}
|
|
|
|
|
|
def _payload_to_settings(payload: str | None) -> Dict[str, Any]:
|
|
if not payload:
|
|
return default_settings()
|
|
try:
|
|
raw = json.loads(payload)
|
|
if not isinstance(raw, dict):
|
|
return default_settings()
|
|
return normalize_settings(raw)
|
|
except Exception:
|
|
return default_settings()
|
|
|
|
|
|
def _persist_settings(settings: Dict[str, Any]) -> Dict[str, Any]:
|
|
from app import db
|
|
from app.models.feed_quality_settings import FeedQualitySettings
|
|
from app.timeutil import utc_now_naive
|
|
|
|
merged = normalize_settings(settings)
|
|
row = db.session.get(FeedQualitySettings, SETTINGS_ROW_ID)
|
|
if row is None:
|
|
row = FeedQualitySettings(
|
|
id=SETTINGS_ROW_ID,
|
|
payload=json.dumps(merged, ensure_ascii=False),
|
|
updated_at=utc_now_naive(),
|
|
)
|
|
db.session.add(row)
|
|
else:
|
|
row.payload = json.dumps(merged, ensure_ascii=False)
|
|
row.updated_at = utc_now_naive()
|
|
db.session.commit()
|
|
return merged
|
|
|
|
|
|
def import_legacy_json_to_db(*, force: bool = False) -> bool:
|
|
"""Импортирует JSON в БД, если строки нет или force=True."""
|
|
from app import db
|
|
from app.models.feed_quality_settings import FeedQualitySettings
|
|
|
|
row = db.session.get(FeedQualitySettings, SETTINGS_ROW_ID)
|
|
if row is not None and not force:
|
|
return False
|
|
raw = _read_legacy_json_raw()
|
|
if not raw and row is not None:
|
|
return False
|
|
settings = normalize_settings(raw) if raw else default_settings()
|
|
_persist_settings(settings)
|
|
return True
|
|
|
|
|
|
def create_feed_quality_settings_table(bind) -> None:
|
|
from app.models.feed_quality_settings import FeedQualitySettings
|
|
|
|
FeedQualitySettings.__table__.create(bind, checkfirst=True)
|
|
|
|
|
|
def import_feed_quality_settings_json(bind) -> bool:
|
|
"""Импорт legacy JSON в пустую таблицу (миграция, без Flask app context)."""
|
|
from sqlalchemy import inspect, text
|
|
from sqlalchemy.engine import Connection, Engine
|
|
|
|
from app.timeutil import utc_now_naive
|
|
|
|
insp = inspect(bind)
|
|
if not insp.has_table("feed_quality_settings"):
|
|
return False
|
|
|
|
def _import(conn: Connection) -> bool:
|
|
existing = conn.execute(
|
|
text("SELECT id FROM feed_quality_settings WHERE id = 1")
|
|
).fetchone()
|
|
if existing is not None:
|
|
return False
|
|
raw = _read_legacy_json_raw()
|
|
settings = normalize_settings(raw) if raw else default_settings()
|
|
conn.execute(
|
|
text(
|
|
"INSERT INTO feed_quality_settings (id, payload, updated_at) "
|
|
"VALUES (1, :payload, :updated_at)"
|
|
),
|
|
{
|
|
"payload": json.dumps(settings, ensure_ascii=False),
|
|
"updated_at": utc_now_naive(),
|
|
},
|
|
)
|
|
if conn.in_transaction():
|
|
conn.commit()
|
|
return True
|
|
|
|
if isinstance(bind, Engine):
|
|
with bind.connect() as conn:
|
|
return _import(conn)
|
|
return _import(bind)
|
|
|
|
|
|
def ensure_feed_quality_settings_table(bind, *, import_json: bool = False) -> None:
|
|
create_feed_quality_settings_table(bind)
|
|
if import_json:
|
|
import_feed_quality_settings_json(bind)
|
|
|
|
|
|
def get_feed_quality_settings() -> Dict[str, Any]:
|
|
from app import db
|
|
from app.models.feed_quality_settings import FeedQualitySettings
|
|
|
|
row = db.session.get(FeedQualitySettings, SETTINGS_ROW_ID)
|
|
if row is None:
|
|
import_legacy_json_to_db()
|
|
row = db.session.get(FeedQualitySettings, SETTINGS_ROW_ID)
|
|
if row is None:
|
|
return default_settings()
|
|
return _payload_to_settings(row.payload)
|
|
|
|
|
|
def save_feed_quality_settings(partial: Dict[str, Any]) -> Dict[str, Any]:
|
|
current = get_feed_quality_settings()
|
|
merged_input = deepcopy(current)
|
|
for key, val in (partial or {}).items():
|
|
if key in ("loading", "unloading", "mix_time", "left_in_mixer") and isinstance(val, dict):
|
|
merged_input[key] = {**current.get(key, {}), **val}
|
|
else:
|
|
merged_input[key] = val
|
|
return _persist_settings(normalize_settings(merged_input))
|
|
|
|
|
|
def settings_for_api() -> Dict[str, Any]:
|
|
return {"settings": get_feed_quality_settings(), "defaults": default_settings()}
|