87 lines
2.5 KiB
Python
87 lines
2.5 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import threading
|
||
|
|
|
||
|
|
from config import DATA_DIR
|
||
|
|
from tickets.data import GuildTicketConfig, TicketData
|
||
|
|
|
||
|
|
logger = logging.getLogger("vision")
|
||
|
|
|
||
|
|
TICKETS_DIR = DATA_DIR / "tickets"
|
||
|
|
_CONFIG_FILE = TICKETS_DIR / "config.json"
|
||
|
|
|
||
|
|
_lock = threading.Lock()
|
||
|
|
_config_cache: dict[int, GuildTicketConfig] = {}
|
||
|
|
_tickets_cache: dict[int, TicketData] = {}
|
||
|
|
|
||
|
|
|
||
|
|
def _ensure_dirs():
|
||
|
|
TICKETS_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
|
|
||
|
|
|
||
|
|
def _load_all_configs() -> dict[int, GuildTicketConfig]:
|
||
|
|
global _config_cache
|
||
|
|
_ensure_dirs()
|
||
|
|
if _config_cache:
|
||
|
|
return _config_cache
|
||
|
|
if _CONFIG_FILE.exists():
|
||
|
|
with open(_CONFIG_FILE, "r", encoding="utf-8") as f:
|
||
|
|
raw = json.load(f)
|
||
|
|
_config_cache = {int(k): GuildTicketConfig.from_dict(v) for k, v in raw.items()}
|
||
|
|
else:
|
||
|
|
_config_cache = {}
|
||
|
|
return _config_cache
|
||
|
|
|
||
|
|
|
||
|
|
def _save_all_configs():
|
||
|
|
_ensure_dirs()
|
||
|
|
data = {str(k): v.to_dict() for k, v in _config_cache.items()}
|
||
|
|
with open(_CONFIG_FILE, "w", encoding="utf-8") as f:
|
||
|
|
json.dump(data, f, indent=2, ensure_ascii=False)
|
||
|
|
|
||
|
|
|
||
|
|
def get_config(guild_id: int) -> GuildTicketConfig:
|
||
|
|
with _lock:
|
||
|
|
configs = _load_all_configs()
|
||
|
|
if guild_id not in configs:
|
||
|
|
configs[guild_id] = GuildTicketConfig(guild_id=guild_id)
|
||
|
|
return configs[guild_id]
|
||
|
|
|
||
|
|
|
||
|
|
def save_config(config: GuildTicketConfig):
|
||
|
|
with _lock:
|
||
|
|
_config_cache[config.guild_id] = config
|
||
|
|
_save_all_configs()
|
||
|
|
|
||
|
|
|
||
|
|
def get_ticket(channel_id: int) -> TicketData | None:
|
||
|
|
with _lock:
|
||
|
|
if channel_id in _tickets_cache:
|
||
|
|
return _tickets_cache[channel_id]
|
||
|
|
ticket_file = TICKETS_DIR / f"{channel_id}.json"
|
||
|
|
if ticket_file.exists():
|
||
|
|
with open(ticket_file, "r", encoding="utf-8") as f:
|
||
|
|
ticket = TicketData.from_dict(json.load(f))
|
||
|
|
_tickets_cache[channel_id] = ticket
|
||
|
|
return ticket
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def save_ticket(ticket: TicketData):
|
||
|
|
with _lock:
|
||
|
|
_tickets_cache[ticket.channel_id] = ticket
|
||
|
|
_ensure_dirs()
|
||
|
|
ticket_file = TICKETS_DIR / f"{ticket.channel_id}.json"
|
||
|
|
with open(ticket_file, "w", encoding="utf-8") as f:
|
||
|
|
json.dump(ticket.to_dict(), f, indent=2, ensure_ascii=False)
|
||
|
|
|
||
|
|
|
||
|
|
def delete_ticket(channel_id: int):
|
||
|
|
with _lock:
|
||
|
|
_tickets_cache.pop(channel_id, None)
|
||
|
|
ticket_file = TICKETS_DIR / f"{channel_id}.json"
|
||
|
|
if ticket_file.exists():
|
||
|
|
ticket_file.unlink()
|