src/config.py - master - kuzminaa/PDM

src/config.py
92 строки3 KB
kuzminaa
refactor: migrate modules to src/, add logging system, add assembly workshop
29 июл 2026, 23:52
29 июл 2026, 23:52a5323aa
import os
import json
import threading
import logging

logger = logging.getLogger(__name__)

SETTINGS_NAME = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "settings.json")

LAST_INPUT_DOC_DIR = "last_input_doc_directory"

DEFAULT_ROLES_STAFF = {
    "dev": ["---", "Разработчик А.А.", "Иванов И.И."],
    "chk": ["---", "Проверяющий Б.Б.", "Петров П.П."],
    "ctl": ["---", "Нормоконтролер В.В.", "Сидоров С.С."],
    "app": ["---", "Утверждающий Г.Г.", "Кузнецов А.В."],
    "agr": ["---", "Согласующий Д.Д.", "Петров П.П.", "Сидоров С.С."]
}

_SETTINGS_CACHE = {}
_write_delay = 0.5
_write_timer = None
_lock = threading.Lock()


def init_cache():
    """Загружает настройки с диска в память один раз при старте."""
    global _SETTINGS_CACHE
    with _lock:
        _SETTINGS_CACHE = DEFAULT_ROLES_STAFF.copy()
        _SETTINGS_CACHE[LAST_INPUT_DOC_DIR] = os.path.expanduser('~')
        if not os.path.exists(SETTINGS_NAME):
            with open(SETTINGS_NAME, "w", encoding="utf-8") as f:
                json.dump(_SETTINGS_CACHE, f, ensure_ascii=False, indent=4)
        else:
            try:
                with open(SETTINGS_NAME, "r", encoding="utf-8") as f:
                    _SETTINGS_CACHE = json.load(f)
            except Exception:
                _SETTINGS_CACHE = DEFAULT_ROLES_STAFF.copy()
        if LAST_INPUT_DOC_DIR not in _SETTINGS_CACHE:
            _SETTINGS_CACHE[LAST_INPUT_DOC_DIR] = os.path.expanduser("~")


def get_role_staff(role_key):
    """Мгновенно возвращает список из кэша в ОЗУ без обращения к диску."""
    with _lock:
        return _SETTINGS_CACHE.get(role_key, ["---"])


def add_fio_to_cache(role_key, fio):
    """Добавляет ФИО в кэш ОЗУ и синхронизирует с диском с задержкой."""
    if not fio or fio == "---": return
    with _lock:
        if role_key not in _SETTINGS_CACHE:
            _SETTINGS_CACHE[role_key] = ["---"]
        
        if fio not in _SETTINGS_CACHE[role_key]:
            _SETTINGS_CACHE[role_key].append(fio)
            _schedule_save_to_disk()


def get_last_doc_directory():
    with _lock:
        return _SETTINGS_CACHE.get(LAST_INPUT_DOC_DIR, os.path.expanduser("~"))


def save_last_doc_directory(path):
    with _lock:
        if not path: return
        if _SETTINGS_CACHE.get(LAST_INPUT_DOC_DIR) != path:
            _SETTINGS_CACHE[LAST_INPUT_DOC_DIR] = path 
            _schedule_save_to_disk()


def _schedule_save_to_disk():
    global _write_timer
    if _write_timer:
        _write_timer.cancel()
    _write_timer = threading.Timer(_write_delay, _save_cache_to_disk)
    _write_timer.start()


def _save_cache_to_disk():
    """Служебная функция синхронизации кэша ОЗУ с файлом JSON."""
    with _lock:
        cache_snapshot = _SETTINGS_CACHE.copy()
    try:
        with open(SETTINGS_NAME, "w", encoding="utf-8") as f:
            json.dump(cache_snapshot, f, ensure_ascii=False, indent=4)
    except Exception as e:
        logger.error(f"Ошибка сохранения настроек: {e}")