46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
import json
|
||
import os
|
||
import logging
|
||
from typing import List, Dict
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
HISTORY_DIR = "data/chat_history"
|
||
MAX_MESSAGES = 10
|
||
|
||
os.makedirs(HISTORY_DIR, exist_ok=True)
|
||
|
||
def _get_file_path(user_id: int) -> str:
|
||
return os.path.join(HISTORY_DIR, f"{user_id}.json")
|
||
|
||
def load_history(user_id: int) -> List[Dict[str, str]]:
|
||
path = _get_file_path(user_id)
|
||
if not os.path.exists(path):
|
||
return []
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except (json.JSONDecodeError, IOError):
|
||
return []
|
||
|
||
def save_history(user_id: int, history: List[Dict[str, str]]):
|
||
path = _get_file_path(user_id)
|
||
try:
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
json.dump(history, f, ensure_ascii=False, indent=2)
|
||
except IOError as e:
|
||
logger.error(f"Не удалось сохранить историю для user {user_id}: {e}")
|
||
|
||
def add_message(user_id: int, role: str, content: str):
|
||
history = load_history(user_id)
|
||
history.append({"role": role, "content": content})
|
||
if len(history) > MAX_MESSAGES:
|
||
history = history[-MAX_MESSAGES:]
|
||
save_history(user_id, history)
|
||
|
||
def clear_history(user_id: int):
|
||
path = _get_file_path(user_id)
|
||
if os.path.exists(path):
|
||
os.remove(path)
|
||
logger.info(f"История очищена для user {user_id}")
|