61 lines
2.7 KiB
Python
61 lines
2.7 KiB
Python
"""One local connection per GusVoice bot; legacy v1 paths remain valid."""
|
|
from dataclasses import dataclass
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import uuid
|
|
|
|
|
|
def profile_name(value):
|
|
if not isinstance(value, str) or not re.fullmatch(r"[a-z][a-z0-9_-]{0,31}", value):
|
|
raise ValueError("Имя подключения: строчная латиница, цифры, _ и -, до 32 символов; первая — буква.")
|
|
return value
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Layout:
|
|
root: Path = Path("/")
|
|
|
|
def config(self, name):
|
|
profile_name(name)
|
|
return self.root / ("etc/awg-gus-bot.json" if name == "default"
|
|
else f"etc/awg-gus-bot/instances/{name}.json")
|
|
|
|
def state(self, name):
|
|
profile_name(name)
|
|
base = self.root / "var/lib/awg-gus-bot"
|
|
return base if name == "default" else base / "instances" / name
|
|
|
|
def names(self):
|
|
names = ["default"] if self.config("default").is_file() else []
|
|
for path in sorted((self.root / "etc/awg-gus-bot/instances").glob("*.json")):
|
|
name = profile_name(path.stem)
|
|
if name == "default":
|
|
raise ValueError("Имя default зарезервировано для первого подключения.")
|
|
names.append(name)
|
|
return names
|
|
|
|
@staticmethod
|
|
def unit(name):
|
|
profile_name(name)
|
|
return "awg-gus-bot.service" if name == "default" else f"awg-gus-bot@{name}.service"
|
|
|
|
|
|
def assert_unique_bot(target, candidate, layout=None):
|
|
"""Before polling: aliases/rotated tokens must not steal another bot's updates."""
|
|
layout = layout or Layout()
|
|
target = Path(target).resolve()
|
|
for name in layout.names():
|
|
path = layout.config(name)
|
|
try:
|
|
existing = json.loads(path.read_text(encoding="utf-8"))
|
|
same_bot = uuid.UUID(existing["bot_id"]) == uuid.UUID(candidate["bot_id"])
|
|
same_token = existing["token"] == candidate["token"]
|
|
except (ValueError, KeyError, TypeError, AttributeError) as exc:
|
|
raise ValueError(f"Проверьте настройки подключения {name}: неверный формат.") from exc
|
|
if path.resolve() == target:
|
|
if not same_bot or existing["url"] != candidate["url"]:
|
|
raise ValueError("Это другой бот/сервер. Добавьте новое подключение вместо перенастройки старого.")
|
|
elif same_bot or same_token:
|
|
raise ValueError(f"Этот бот уже подключён как {name}. Повторный polling отключит его работу.")
|