282 lines
8.4 KiB
Python
282 lines
8.4 KiB
Python
"""Реквизиты организации для учётных документов (СП-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()
|