"""Реквизиты организации для учётных документов (СП-20, журнал). Хранение: recipes.db.""" from __future__ import annotations import json import os from pathlib import Path from typing import Any, Dict, Tuple from flask import Flask from config import Config ORG_SETTINGS_ROW_ID = 1 ORG_SETTINGS_FIELDS = ( "organization_name", "okpo", "farm_name", "department", "brigade", "animal_type", "responsible_person", "warehouse_keeper", "zootechnician", "document_number_prefix", ) DEFAULT_ORG_SETTINGS: Dict[str, str] = { "organization_name": "", "okpo": "", "farm_name": "", "department": "", "brigade": "", "animal_type": "КРС", "responsible_person": "", "warehouse_keeper": "", "zootechnician": "", "document_number_prefix": "СП20", } SIGNATURE_ROLES = ("zootechnician", "warehouse_keeper", "recipient") def get_org_settings_path() -> Path: custom = os.getenv("WESP_ORG_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_org_settings.json" def get_signatures_dir() -> Path: return Path(Config.DATA_DIR) / "signatures" def _read_legacy_json_raw() -> Dict[str, Any]: path = get_org_settings_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: return {} def _normalize_settings(raw: Dict[str, Any]) -> Dict[str, str]: merged = dict(DEFAULT_ORG_SETTINGS) for field in ORG_SETTINGS_FIELDS: if field in raw: merged[field] = str(raw[field] or "").strip() return merged def _row_to_dict(row: Any) -> Dict[str, str]: return {field: str(getattr(row, field, "") or "") for field in ORG_SETTINGS_FIELDS} def _apply_settings_to_row(row: Any, settings: Dict[str, str]) -> None: merged = _normalize_settings(settings) for field in ORG_SETTINGS_FIELDS: setattr(row, field, merged[field]) def _persist_settings(settings: Dict[str, str]) -> Dict[str, str]: from app import db from app.models.org_settings import OrgSettings from app.timeutil import utc_now_naive merged = _normalize_settings(settings) row = db.session.get(OrgSettings, ORG_SETTINGS_ROW_ID) if row is None: row = OrgSettings(id=ORG_SETTINGS_ROW_ID, updated_at=utc_now_naive()) db.session.add(row) _apply_settings_to_row(row, merged) row.updated_at = utc_now_naive() db.session.commit() return merged def create_org_settings_table(bind) -> None: from app.models.org_settings import OrgSettings OrgSettings.__table__.create(bind, checkfirst=True) def import_org_settings_legacy(bind) -> bool: """Импорт legacy JSON-файла в typed-колонки (миграция / first-run ETL).""" 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("org_settings"): return False settings = _normalize_settings(_read_legacy_json_raw()) cols = {c["name"] for c in insp.get_columns("org_settings")} has_payload = "payload" in cols typed_cols = [f for f in ORG_SETTINGS_FIELDS if f in cols] if not typed_cols and not has_payload: return False def _import(conn: Connection) -> bool: existing = conn.execute(text("SELECT id FROM org_settings WHERE id = 1")).fetchone() if existing is not None: return False if typed_cols: assignments = ", ".join(f"{field} = :{field}" for field in typed_cols) params = {field: settings[field] for field in typed_cols} params["updated_at"] = utc_now_naive() conn.execute( text(f"INSERT INTO org_settings (id, {', '.join(typed_cols)}, updated_at) VALUES (1, {', '.join(':' + f for f in typed_cols)}, :updated_at)"), params, ) else: conn.execute( text( "INSERT INTO org_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 import_org_settings_json(bind) -> bool: """Alias для совместимости с migration 0007.""" return import_org_settings_legacy(bind) def ensure_org_settings_table(bind, *, import_json: bool = False) -> None: create_org_settings_table(bind) if import_json: import_org_settings_legacy(bind) def import_legacy_json_to_db(*, force: bool = False) -> bool: from app import db from app.models.org_settings import OrgSettings row = db.session.get(OrgSettings, ORG_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 _persist_settings(raw if raw else DEFAULT_ORG_SETTINGS) return True def read_org_settings_raw() -> Dict[str, Any]: from app import db from app.models.org_settings import OrgSettings row = db.session.get(OrgSettings, ORG_SETTINGS_ROW_ID) if row is None: import_legacy_json_to_db() row = db.session.get(OrgSettings, ORG_SETTINGS_ROW_ID) if row is None: return {} return _row_to_dict(row) def write_org_settings(partial: Dict[str, Any]) -> None: current = read_org_settings_raw() for k, v in partial.items(): if k in ORG_SETTINGS_FIELDS: current[k] = v if v is None else str(v).strip() _persist_settings(current) def _farm_fallback() -> str: farms = list_farms_from_db() if not farms: return "" return farms[0] def list_farms_from_db() -> list[str]: try: from app.models import FeedDispenser rows = ( FeedDispenser.query.filter(FeedDispenser.is_deleted.is_(False)) .filter(FeedDispenser.is_active.is_(True)) .order_by(FeedDispenser.farm.asc(), FeedDispenser.name.asc()) .all() ) seen: set[str] = set() out: list[str] = [] for row in rows: farm = str(row.farm or "").strip() if not farm or farm in seen: continue seen.add(farm) out.append(farm) return out except Exception: return [] def get_org_settings(app: Flask | None = None) -> Dict[str, str]: merged = dict(DEFAULT_ORG_SETTINGS) merged.update({k: str(v) for k, v in read_org_settings_raw().items() if k in ORG_SETTINGS_FIELDS}) if not merged.get("farm_name"): merged["farm_name"] = _farm_fallback() return merged def validate_org_settings_payload(data: Dict[str, Any]) -> Tuple[bool, str, Dict[str, str]]: if not isinstance(data, dict): return False, "Ожидается JSON-объект", {} out: Dict[str, str] = {} for field in ORG_SETTINGS_FIELDS: if field in data: out[field] = str(data[field] or "").strip() return True, "", out def signature_path(role: str) -> Path: if role not in SIGNATURE_ROLES: raise ValueError(f"Unknown signature role: {role}") return get_signatures_dir() / f"{role}.png" def list_signature_status() -> Dict[str, bool]: d = get_signatures_dir() return {role: (d / f"{role}.png").is_file() for role in SIGNATURE_ROLES} def save_signature_png(role: str, png_bytes: bytes) -> None: if role not in SIGNATURE_ROLES: raise ValueError(f"Unknown signature role: {role}") if not png_bytes or len(png_bytes) < 32: raise ValueError("Пустой файл подписи") if len(png_bytes) > 512_000: raise ValueError("Файл подписи слишком большой") d = get_signatures_dir() d.mkdir(parents=True, exist_ok=True) signature_path(role).write_bytes(png_bytes) def delete_signature(role: str) -> None: p = signature_path(role) if p.is_file(): p.unlink()