You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
79 lines
2.4 KiB
79 lines
2.4 KiB
"""Хранилище редактируемых промптов для AI-функций. |
|
|
|
Промпты хранятся в app/data/prompts.json и загружаются при старте. |
|
Если файл не существует или повреждён — используются встроенные defaults. |
|
""" |
|
from __future__ import annotations |
|
|
|
import json |
|
import logging |
|
import os |
|
import threading |
|
from pathlib import Path |
|
from typing import Any |
|
|
|
_log = logging.getLogger(__name__) |
|
_lock = threading.Lock() |
|
|
|
_DATA_FILE = Path(__file__).parent.parent / 'data' / 'prompts.json' |
|
|
|
_store: dict[str, dict] | None = None |
|
|
|
|
|
def _load() -> dict[str, dict]: |
|
if _DATA_FILE.exists(): |
|
try: |
|
with _DATA_FILE.open('r', encoding='utf-8') as f: |
|
data = json.load(f) |
|
if isinstance(data, dict): |
|
return data |
|
except Exception as e: |
|
_log.warning('prompt_store: не удалось загрузить %s: %s', _DATA_FILE, e) |
|
return {} |
|
|
|
|
|
def _get_store() -> dict[str, dict]: |
|
global _store |
|
if _store is None: |
|
with _lock: |
|
if _store is None: |
|
_store = _load() |
|
return _store |
|
|
|
|
|
def get_all() -> dict[str, dict]: |
|
"""Вернуть все промпты (id → {label, description, system, user, vars}).""" |
|
return dict(_get_store()) |
|
|
|
|
|
def get_prompt(prompt_id: str) -> dict[str, Any] | None: |
|
return _get_store().get(prompt_id) |
|
|
|
|
|
def get_system(prompt_id: str, default: str = '') -> str: |
|
p = get_prompt(prompt_id) |
|
return p.get('system', default) if p else default |
|
|
|
|
|
def get_user(prompt_id: str, default: str = '') -> str: |
|
p = get_prompt(prompt_id) |
|
return p.get('user', default) if p else default |
|
|
|
|
|
def save_all(data: dict[str, dict]) -> None: |
|
"""Сохранить все промпты в JSON-файл и обновить кеш.""" |
|
global _store |
|
_DATA_FILE.parent.mkdir(parents=True, exist_ok=True) |
|
with _lock: |
|
with _DATA_FILE.open('w', encoding='utf-8') as f: |
|
json.dump(data, f, ensure_ascii=False, indent=2) |
|
_store = data |
|
|
|
|
|
def save_prompt(prompt_id: str, system: str, user: str) -> None: |
|
"""Обновить system/user для одного промпта.""" |
|
store = dict(_get_store()) |
|
if prompt_id not in store: |
|
raise KeyError(f'Промпт {prompt_id!r} не найден.') |
|
store[prompt_id] = {**store[prompt_id], 'system': system, 'user': user} |
|
save_all(store)
|
|
|