Add v2 support for independent GusVoice connections to one AWG server

awg_gus/profiles.py

@@ -0,0 +1,60 @@
+"""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 отключит его работу.")