awg-gus-bot/awg_gus/main.py

from __future__ import annotations

import argparse
import asyncio
import contextlib
import getpass
import hmac
import json
import logging
import os
from pathlib import Path
import secrets
import signal
import sys
import time
from urllib.parse import urlsplit
import uuid

from gusvoice_bot import Client, GusVoiceApiError

from . import __version__
from .bot import Bot
from .core import Awg, Delivery, State, client_for
from .profiles import assert_unique_bot

DEFAULT_CONFIG = "/etc/awg-gus-bot.json"
DEFAULT_STATE = "/var/lib/awg-gus-bot"


def validate_url(url):
    u = urlsplit(url.strip())
    if u.scheme != "https" or not u.hostname or u.username or u.password or u.query or u.fragment:
        raise ValueError("Укажите HTTPS-адрес Гуся без логина, пароля и параметров.")
    if u.path.rstrip("/") not in {"", "/api"}:
        raise ValueError("Нужен адрес сайта/сервера API, без /api/bot/v1.")
    return url.strip().rstrip("/").removesuffix("/api")


def validate_config(d):
    validate_url(d["url"])
    if not isinstance(d.get("token"), str) or not d["token"].startswith("gvb_"):
        raise ValueError("Требуется токен бота Гуся gvb_…")
    if not isinstance(d.get("owners"), list) or not d["owners"]:
        raise ValueError("Не задан владелец")
    for owner in d["owners"]:
        uuid.UUID(owner)
    uuid.UUID(d["bot_id"])
    for key in ("awg2", "state_dir"):
        if not Path(d[key]).is_absolute():
            raise ValueError(f"{key}: нужен абсолютный путь")
    if "label" in d and (not isinstance(d["label"], str) or len(d["label"]) > 80
                         or any(ord(c) < 32 for c in d["label"])):
        raise ValueError("Название подключения должно быть одной строкой до 80 символов")
    return d


def load_config(path):
    with open(path, encoding="utf-8") as f:
        return validate_config(json.load(f))


def save_config(path, data):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(path.name + ".new")
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
            f.write("\n")
            f.flush()
            os.fsync(f.fileno())
        if os.name == "posix":
            tmp.chmod(0o600)
        os.replace(tmp, path)
    finally:
        with contextlib.suppress(FileNotFoundError):
            tmp.unlink()


def check_features(me):
    uuid.UUID(me["id"])
    # HTML rendering was introduced with API 3; do not silently send HTML to an older server.
    if int(me.get("api_version") or 0) < 3:
        raise ValueError("Нужен GusVoice Bot API 3 или новее (кнопки, команды и HTML). Обновите Гуся.")
    features = set(me.get("features") or [])
    if not {"keyboards", "commands", "parse_mode"}.issubset(features):
        raise ValueError("Инстанс не объявляет необходимые возможности: keyboards, commands, parse_mode.")


def pair(client, code, *, duration=600, now=time.monotonic):
    deadline = now() + duration
    offset = 0
    while now() < deadline:
        updates = client.get_updates(offset=offset, timeout=10, allowed_updates=["message"])
        for u in updates:
            offset = max(offset, int(u["update_id"]) + 1)
            msg = u.get("message") or {}
            chat, author = msg.get("chat") or {}, msg.get("from") or {}
            text = (msg.get("text") or "").strip()
            if chat.get("type") != "dm" or author.get("bot"):
                continue
            if not hmac.compare_digest(text.encode("utf-8"), ("/start " + code).encode("utf-8")):
                continue
            try:
                uuid.UUID(author["id"])
                uuid.UUID(chat["id"])
            except (KeyError, ValueError, TypeError, AttributeError):
                continue
            client.get_updates(offset=offset, timeout=0, allowed_updates=[])
            return author["id"], chat["id"], offset
    raise ValueError("Время привязки истекло. Запустите настройку ещё раз.")


def configure(args):
    if not sys.stdin.isatty():
        raise ValueError("Настройка требует терминала. Скачайте install.sh в файл и запустите sudo bash install.sh.")
    print("AWG для Гуся: настройка отдельного бота. Mini App не используется.")
    url = validate_url(input("HTTPS-адрес Гуся: "))
    token = getpass.getpass("Токен нового бота gvb_… (ввод скрыт): ").strip()
    if not token.startswith("gvb_"):
        raise ValueError("Токен должен начинаться с gvb_")
    client = Client(token, url)
    me = client.get_me()
    check_features(me)
    if client.get_webhook_info().get("url"):
        raise ValueError("У этого бота настроен webhook. Создайте отдельного бота для AWG.")
    resolved = validate_url(client.api_url())
    candidate = {"url": resolved, "token": token, "bot_id": me["id"]}
    assert_unique_bot(args.config, candidate, getattr(args, "layout", None))
    code = secrets.token_urlsafe(18)
    print(f"\nОткройте личную беседу с @{me.get('username', '')} в Гусе.")
    print("Отправьте следующее сообщение (код действует 10 минут):\n")
    print("/start " + code)
    print("\nЖду сообщение…", flush=True)
    owner, chat, offset = pair(client, code)
    d = {"url": resolved, "token": token, "owners": [owner], "bot_id": me["id"],
         "awg2": args.awg2, "state_dir": args.state_dir}
    if getattr(args, "label", None):
        d["label"] = args.label
    validate_config(d)
    state = State(Path(args.state_dir) / "state.sqlite3", resolved + "|" + me["id"])
    state.set("offset", offset)
    with state.db:
        state.db.execute("INSERT OR REPLACE INTO chats VALUES (?,?)", (owner, chat))
    state.db.close()
    save_config(args.config, d)
    # Pairing is complete at this point; a delivery failure must not discard the valid config.
    with contextlib.suppress(Exception):
        client.send_message(chat, "✅ Аккаунт привязан. После завершения установки отправьте /menu.")
    print("Владелец привязан. Конфигурация сохранена.")


async def check(args):
    d = load_config(args.config)
    me = await client_for(d).get_me()
    check_features(me)
    if me["id"] != d["bot_id"]:
        raise ValueError("Токен принадлежит другому боту. Повторите настройку с отдельным state_dir.")
    if (await client_for(d).get_webhook_info()).get("url"):
        raise ValueError("Задан webhook: нужен отдельный бот для polling.")
    await Awg(d["awg2"]).call("version", timeout=30)
    print(f"OK: AWG для Гуся {__version__}; GusVoice API {me['api_version']}; awg2 api отвечает.")


async def run(args):
    d = load_config(args.config)
    client = client_for(d)
    me = await client.get_me()
    check_features(me)
    if me["id"] != d["bot_id"]:
        raise ValueError("Неверная привязка бота. Выполните setup.")
    state = State(Path(d["state_dir"]) / "state.sqlite3", d["url"] + "|" + d["bot_id"])
    # flock prevents accidental second local process; the API prevents remote duplicate consumers.
    import fcntl
    lock = open(Path(d["state_dir"]) / "process.lock", "a")
    try:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError as e:
        raise ValueError("Этот бот уже запущен") from e
    bot = Bot(d, Awg(d["awg2"]), Delivery(client), state)
    task = asyncio.create_task(bot.run())
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, task.cancel)
    logging.info("AWG Gus bot %s started", __version__)
    try:
        await task
    except asyncio.CancelledError:
        pass
    finally:
        state.db.close()
        lock.close()


def main():
    if len(sys.argv) > 1 and sys.argv[1] == "manage":
        from .manage import main as manage_main
        return manage_main(sys.argv[2:])
    parser = argparse.ArgumentParser(description="AWG Toolza bot for GusVoice")
    parser.add_argument("command", choices=["run", "setup", "check"])
    parser.add_argument("--config", default=DEFAULT_CONFIG)
    parser.add_argument("--awg2", default="/usr/local/bin/awg2")
    parser.add_argument("--state-dir", default=DEFAULT_STATE)
    args = parser.parse_args()
    os.umask(0o077)
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
    try:
        if args.command == "setup":
            configure(args)
        else:
            asyncio.run(check(args) if args.command == "check" else run(args))
    except GusVoiceApiError as e:
        print(f"Ошибка GusVoice API: HTTP {e.code}. Проверьте адрес, токен и включение ботов.", file=sys.stderr)
        sys.exit(1)
    except (ValueError, FileNotFoundError, KeyError) as e:
        # No raw request URLs or token-bearing exception repr in logs.
        print(str(e) if isinstance(e, ValueError) else "Не найден или неверно заполнен конфигурационный файл.", file=sys.stderr)
        sys.exit(1)
    except KeyboardInterrupt:
        sys.exit(130)
    except Exception as e:
        print(f"Ошибка: {type(e).__name__}. Проверьте сеть, сертификаты и доступность awg2.", file=sys.stderr)
        sys.exit(1)