74 lines
2.1 KiB
Python
74 lines
2.1 KiB
Python
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import aiofiles
|
|
|
|
CONFIG_PATH = Path(os.getenv("CONFIG_PATH", "/app/data/config.json"))
|
|
|
|
|
|
async def load_config() -> dict:
|
|
if CONFIG_PATH.is_dir():
|
|
return {}
|
|
if not CONFIG_PATH.exists():
|
|
return {}
|
|
async with aiofiles.open(CONFIG_PATH, "r") as f:
|
|
data = await f.read()
|
|
return json.loads(data)
|
|
|
|
|
|
async def save_config(config: dict) -> None:
|
|
CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
async with aiofiles.open(CONFIG_PATH, "w") as f:
|
|
await f.write(json.dumps(config, indent=2))
|
|
|
|
|
|
async def _get_guild_config(guild_id: int) -> dict:
|
|
config = await load_config()
|
|
return config.get(str(guild_id), {})
|
|
|
|
|
|
async def get_output_channel(guild_id: int) -> Optional[int]:
|
|
guild_config = await _get_guild_config(guild_id)
|
|
return guild_config.get("output_channel_id")
|
|
|
|
|
|
async def set_output_channel(guild_id: int, channel_id: int) -> None:
|
|
config = await load_config()
|
|
guild_key = str(guild_id)
|
|
guild_config = config.get(guild_key, {})
|
|
guild_config["output_channel_id"] = channel_id
|
|
config[guild_key] = guild_config
|
|
await save_config(config)
|
|
|
|
|
|
async def get_retry_state(guild_id: int) -> Optional[dict]:
|
|
guild_config = await _get_guild_config(guild_id)
|
|
retry_state = guild_config.get("retry_state")
|
|
return retry_state if isinstance(retry_state, dict) else None
|
|
|
|
|
|
async def set_retry_state(guild_id: int, file_path: str, source_channel_id: int | None, reason: str) -> None:
|
|
config = await load_config()
|
|
guild_key = str(guild_id)
|
|
guild_config = config.get(guild_key, {})
|
|
guild_config["retry_state"] = {
|
|
"file_path": file_path,
|
|
"source_channel_id": source_channel_id,
|
|
"reason": reason,
|
|
}
|
|
config[guild_key] = guild_config
|
|
await save_config(config)
|
|
|
|
|
|
async def clear_retry_state(guild_id: int) -> None:
|
|
config = await load_config()
|
|
guild_key = str(guild_id)
|
|
guild_config = config.get(guild_key)
|
|
if not guild_config:
|
|
return
|
|
guild_config.pop("retry_state", None)
|
|
config[guild_key] = guild_config
|
|
await save_config(config)
|