from __future__ import annotations import asyncio import json import os import tempfile from contextlib import asynccontextmanager from pathlib import Path from typing import Any from config import settings DEFAULT_FILES: dict[str, Any] = { "guilds.json": {}, "guild_settings.json": {}, "guild_resources.json": {}, "trusted_users.json": {}, "security_events.json": [], "backups.json": {}, "commands.json": [], "lockdown_states.json": {}, "licenses.json": {}, "customers.json": [], "meta.json": {"next_event_id": 1, "next_backup_id": 1, "next_command_id": 1}, } class JsonStorage: def __init__(self) -> None: self.base = Path(settings.data_dir) self.lock = asyncio.Lock() async def initialize(self) -> None: self.base.mkdir(parents=True, exist_ok=True) async with self.lock: for name, default in DEFAULT_FILES.items(): path = self.base / name if not path.exists(): self._atomic_write(path, default) def _read(self, name: str, default: Any) -> Any: path = self.base / name if not path.exists(): return default try: with path.open("r", encoding="utf-8") as f: return json.load(f) except (json.JSONDecodeError, OSError): return default def _atomic_write(self, path: Path, data: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) fd, tmp_name = tempfile.mkstemp(prefix=path.name + ".", suffix=".tmp", dir=str(path.parent)) try: with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) f.flush() os.fsync(f.fileno()) os.replace(tmp_name, path) finally: if os.path.exists(tmp_name): try: os.unlink(tmp_name) except OSError: pass async def read(self, name: str, default: Any) -> Any: await self.initialize() async with self.lock: return self._read(name, default) async def write(self, name: str, data: Any) -> None: await self.initialize() async with self.lock: self._atomic_write(self.base / name, data) async def _next_id_locked(self, key: str) -> int: meta = self._read("meta.json", DEFAULT_FILES["meta.json"].copy()) current = int(meta.get(key, 1)) meta[key] = current + 1 self._atomic_write(self.base / "meta.json", meta) return current async def upsert_guild(self, guild) -> None: await self.initialize() async with self.lock: guilds = self._read("guilds.json", {}) gid = str(guild.id) guilds[gid] = { "guild_id": gid, "name": guild.name, "icon_url": str(guild.icon.url) if guild.icon else "", "owner_id": str(guild.owner_id or ""), } self._atomic_write(self.base / "guilds.json", guilds) settings_data = self._read("guild_settings.json", {}) settings_data.setdefault(gid, { "enabled": True, "aggressive_mode": True, "automatic_backups": True, "backup_interval_hours": 6, "keep_backups": 30, "log_channel_id": "", "web_log_channel_id": "", "backup_keep": 30, }) self._atomic_write(self.base / "guild_settings.json", settings_data) async def sync_guild_resources(self, guild) -> None: """ Exporta usuarios y canales del Discord para que la web pueda mostrarlos en el Lockdown Selectivo sin pedir IDs manualmente. """ await self.initialize() async with self.lock: resources = self._read("guild_resources.json", {}) gid = str(guild.id) members = [] for member in guild.members: if member.bot: continue avatar = "" try: avatar = str(member.display_avatar.url) except Exception: avatar = "" members.append({ "id": str(member.id), "username": str(member), "display_name": member.display_name, "avatar": avatar, "owner": member.id == guild.owner_id, }) members.sort(key=lambda x: (not x["owner"], x["display_name"].lower())) channels = [] me = guild.me for channel in guild.text_channels: can_send = True if me: try: can_send = channel.permissions_for(me).send_messages except Exception: can_send = False channels.append({ "id": str(channel.id), "name": channel.name, "position": channel.position, "category": channel.category.name if channel.category else "", "can_send": bool(can_send), }) channels.sort(key=lambda x: (x["category"].lower(), x["position"], x["name"].lower())) resources[gid] = { "members": members, "channels": channels, } self._atomic_write(self.base / "guild_resources.json", resources) async def settings_for(self, guild_id: int) -> dict[str, Any]: base = await self.read("guild_settings.json", {}) panel = await self.read("web_settings.json", {}) gid = str(guild_id) merged = dict(base.get(gid, {})) merged.update(dict(panel.get(gid, {}))) return merged async def update_settings(self, guild_id: int, updates: dict[str, Any]) -> None: await self.initialize() async with self.lock: data = self._read("guild_settings.json", {}) gid = str(guild_id) current = dict(data.get(gid, {})) current.update(updates) data[gid] = current self._atomic_write(self.base / "guild_settings.json", data) async def update_panel_settings(self, guild_id: int, updates: dict[str, Any]) -> None: await self.initialize() async with self.lock: data = self._read("web_settings.json", {}) gid = str(guild_id) current = dict(data.get(gid, {})) current.update(updates) data[gid] = current self._atomic_write(self.base / "web_settings.json", data) async def is_licensed(self, guild_id: int) -> bool: return True async def trusted_ids(self, guild_id: int) -> set[int]: data = await self.read("trusted_users.json", {}) rows = data.get(str(guild_id), []) result = set() for row in rows: try: result.add(int(row["user_id"])) except Exception: pass return result async def add_trusted(self, guild_id: int, user_id: int, label: str = "") -> None: await self.initialize() async with self.lock: data = self._read("trusted_users.json", {}) gid = str(guild_id) rows = list(data.get(gid, [])) rows = [r for r in rows if str(r.get("user_id")) != str(user_id)] rows.append({"user_id": str(user_id), "label": label}) data[gid] = rows self._atomic_write(self.base / "trusted_users.json", data) async def remove_trusted(self, guild_id: int, user_id: int) -> None: await self.initialize() async with self.lock: data = self._read("trusted_users.json", {}) gid = str(guild_id) data[gid] = [r for r in data.get(gid, []) if str(r.get("user_id")) != str(user_id)] self._atomic_write(self.base / "trusted_users.json", data) async def add_event( self, guild_id: int, event_type: str, severity: str, actor_id: int | None, actor_name: str, target: str, details: dict[str, Any] | None = None, blocked: bool = False, ) -> None: from datetime import datetime, timezone await self.initialize() async with self.lock: events = self._read("security_events.json", []) event_id = await self._next_id_locked("next_event_id") events.append({ "id": event_id, "guild_id": str(guild_id), "event_type": event_type, "severity": severity, "actor_id": str(actor_id) if actor_id else None, "actor_name": actor_name, "target": target, "details": details or {}, "blocked": bool(blocked), "created_at": datetime.now(timezone.utc).isoformat(), }) events = events[-5000:] self._atomic_write(self.base / "security_events.json", events) async def events_for(self, guild_id: int, limit: int = 100) -> list[dict[str, Any]]: events = await self.read("security_events.json", []) rows = [e for e in events if str(e.get("guild_id")) == str(guild_id)] return list(reversed(rows[-limit:])) async def create_backup_record(self, guild_id: int, reason: str, data: dict[str, Any], created_by: str) -> int: from datetime import datetime, timezone await self.initialize() async with self.lock: backups = self._read("backups.json", {}) backup_id = await self._next_id_locked("next_backup_id") gid = str(guild_id) rows = list(backups.get(gid, [])) rows.append({ "id": backup_id, "guild_id": gid, "reason": reason, "backup": data, "created_by": created_by, "created_at": datetime.now(timezone.utc).isoformat(), }) backups[gid] = rows self._atomic_write(self.base / "backups.json", backups) return backup_id async def get_backup(self, guild_id: int, backup_id: int) -> dict[str, Any] | None: backups = await self.read("backups.json", {}) for row in backups.get(str(guild_id), []): if int(row.get("id", 0)) == int(backup_id): return row return None async def backups_for(self, guild_id: int, limit: int = 30) -> list[dict[str, Any]]: backups = await self.read("backups.json", {}) rows = list(backups.get(str(guild_id), [])) return list(reversed(rows[-limit:])) async def trim_backups(self, guild_id: int, keep: int) -> None: await self.initialize() async with self.lock: backups = self._read("backups.json", {}) gid = str(guild_id) keep = max(1, min(int(keep or 30), 200)) backups[gid] = list(backups.get(gid, []))[-keep:] self._atomic_write(self.base / "backups.json", backups) async def save_lockdown_state(self, guild_id: int, mode: str, data: dict[str, Any]) -> None: from datetime import datetime, timezone await self.initialize() async with self.lock: states = self._read("lockdown_states.json", {}) gid = str(guild_id) states.setdefault(gid, {}) states[gid][mode] = { "active": True, "state": data, "created_at": datetime.now(timezone.utc).isoformat(), } self._atomic_write(self.base / "lockdown_states.json", states) async def get_lockdown_state(self, guild_id: int, mode: str) -> dict[str, Any] | None: states = await self.read("lockdown_states.json", {}) row = states.get(str(guild_id), {}).get(mode) if not row or not row.get("active"): return None return row async def clear_lockdown_state(self, guild_id: int, mode: str) -> None: await self.initialize() async with self.lock: states = self._read("lockdown_states.json", {}) gid = str(guild_id) if gid in states and mode in states[gid]: states[gid][mode]["active"] = False self._atomic_write(self.base / "lockdown_states.json", states) async def enqueue_command(self, command: dict[str, Any]) -> int: from datetime import datetime, timezone await self.initialize() async with self.lock: rows = self._read("commands.json", []) cid = await self._next_id_locked("next_command_id") command = dict(command) command["id"] = cid command.setdefault("status", "pending") command.setdefault("created_at", datetime.now(timezone.utc).isoformat()) rows.append(command) rows = rows[-2000:] self._atomic_write(self.base / "commands.json", rows) return cid async def pending_commands(self, limit: int = 10) -> list[dict[str, Any]]: rows = await self.read("commands.json", []) return [r for r in rows if r.get("status") == "pending"][:limit] async def update_command(self, command_id: int, updates: dict[str, Any]) -> None: await self.initialize() async with self.lock: rows = self._read("commands.json", []) for row in rows: if int(row.get("id", 0)) == int(command_id): row.update(updates) break self._atomic_write(self.base / "commands.json", rows) async def commands_for(self, guild_id: int, limit: int = 15) -> list[dict[str, Any]]: rows = await self.read("commands.json", []) filtered = [r for r in rows if str(r.get("guild_id")) == str(guild_id)] return list(reversed(filtered[-limit:])) storage = JsonStorage()