Blame for plus/serve_muscriptor.py

"""MuScriptor-сервер и публичная ссылка, одной командой.

    python3 plus/serve_muscriptor.py                 # наши веса с HF, порт 8222
    python3 plus/serve_muscriptor.py --device cuda   # на T4
    python3 plus/serve_muscriptor.py --no-tunnel     # только локально

Поднимает ШТАТНЫЙ сервер MuScriptor (python -m muscriptor serve), тот самый,
что отдаёт веб-приложение «MuScriptor: Music to MIDI and Sheet Music», и
открывает туннель, печатая ссылку. Своего интерфейса здесь нет.
"""

from __future__ import annotations

import argparse
import collections
import json
import os
import pathlib
import re
import shutil
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.request

LINKS_FILE = "muscriptor_links.txt"
DEFAULT_MODEL = "hf://Nokia123543/muscriptor-large-fork/model.safetensors"
PACKAGE = "muscriptor"
TUNNEL_ORDER = ("serveo", "localhostrun", "pinggy", "cloudflared", "tuns", "tunnl", "tunnl-stable")   # serveo первым: постоянный поддомен, без заглушки


def say(text: str) -> None:
    print(text, flush=True)


def ensure_package() -> None:
    import importlib.util

    if importlib.util.find_spec(PACKAGE) is not None:
        return
    say("ставлю пакет %s" % PACKAGE)
    subprocess.run([sys.executable, "-m", "pip", "install", "-q", PACKAGE], check=False)


def ensure_ssh() -> str | None:
    """ssh есть в большинстве образов; если нет, пробуем доставить."""
    found = shutil.which("ssh")
    if found:
        return found
    say("ssh не найден, пробую доставить openssh-client")
    for command in (["apt-get", "install", "-y", "-q", "openssh-client"],
                    [sys.executable, "-m", "pip", "install", "-q", "sshtunnel"]):
        try:
            subprocess.run(command, check=False, timeout=300,
                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        except Exception:
            pass
        found = shutil.which("ssh")
        if found:
            say("ssh установлен: " + found)
            return found
    say("ssh поставить не удалось: останутся http-туннели")
    return None


def ensure_key() -> bool:
    """Ключ нужен для стабильной ссылки (один адрес при переподключении)."""
    home = pathlib.Path.home() / ".ssh"
    if any((home / name).exists() for name in
           ("id_ed25519", "id_rsa", "id_ecdsa")):
        return True
    home.mkdir(parents=True, exist_ok=True)
    try:
        subprocess.run(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f",
                        str(home / "id_ed25519")], check=False,
                       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=60)
    except Exception:
        return False
    if (home / "id_ed25519").exists():
        say("сгенерирован ssh-ключ: " + str(home / "id_ed25519"))
        return True
    return False


def healthy(port: int, timeout: float = 3.0) -> bool:
    try:
        with urllib.request.urlopen("http://127.0.0.1:%d/health" % port, timeout=timeout) as answer:
            return answer.status == 200
    except Exception:
        return False


def low_memory_env(threads: int | None = None) -> dict:
    """Убрать раздувание памяти аллокаторами, не трогая точность модели.

    Веса остаются float32, никакого квантования: экономия идёт за счёт числа
    потоков (у каждого своя арена malloc) и возврата освобождённых страниц ОС.
    """
    cores = os.cpu_count() or 4
    count = threads or max(1, min(4, cores // 2))
    env = dict(os.environ)
    env.update({
        "OMP_NUM_THREADS": str(count),
        "MKL_NUM_THREADS": str(count),
        "OPENBLAS_NUM_THREADS": str(count),
        "NUMEXPR_NUM_THREADS": str(count),
        "MALLOC_ARENA_MAX": "2",
        "MALLOC_TRIM_THRESHOLD_": "1000000",
        "MALLOC_MMAP_THRESHOLD_": "131072",
        "TOKENIZERS_PARALLELISM": "false",
        "PYTORCH_CPU_ALLOC_CONF": "garbage_collection_threshold:0.6",
    })
    return env


def start_server(port: int, model: str, device: str, dtype: str | None,
                 env: dict | None = None) -> subprocess.Popen:
    command = [sys.executable, "-m", "muscriptor", "serve",
               "--model", model, "--device", device, "--host", "0.0.0.0", "--port", str(port)]
    if dtype:
        command += ["--dtype", dtype]
    log = open("muscriptor_server.log", "a")
    say("запускаю: " + " ".join(command[1:]))
    process = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT,
                               stdin=open(os.devnull, "rb"),
                               env=env or dict(os.environ))
    deadline = time.time() + 600
    started = time.time()
    while time.time() < deadline:
        if process.poll() is not None:
            raise SystemExit("сервер умер при старте, смотри muscriptor_server.log")
        if healthy(port):
            return process
        waited = int(time.time() - started)
        if waited and waited % 15 == 0:
            say("      жду загрузки модели, %d с (первый раз качаются веса)"
                % waited)
            time.sleep(1)
        time.sleep(2)
    raise SystemExit("сервер не ответил за 10 минут, смотри muscriptor_server.log")


RESERVED_HOSTS = ("console.", "www.", "admin.", "docs.", "blog.", "status.")


def pick_url(pattern: "re.Pattern[str]", line: str, subdomain: str = "") -> str | None:
    """Выбрать настоящую ссылку туннеля, пропустив служебные адреса провайдера."""
    for found in pattern.finditer(line):
        url = found.group(0)
        host = url.split("://", 1)[-1].lower()
        if host.startswith(RESERVED_HOSTS):
            continue
        return url
    return None


class Tunnel:
    def __init__(self, name: str, argv: list[str], pattern: str):
        self.name, self.pattern = name, re.compile(pattern)
        self.url: str | None = None
        self.tail: collections.deque[str] = collections.deque(maxlen=40)
        self.failed = False
        self.proc = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                                     stdin=open(os.devnull, "rb"), text=True, bufsize=1)
        threading.Thread(target=self._drain, daemon=True).start()

    def _drain(self) -> None:
        assert self.proc.stdout is not None
        for line in self.proc.stdout:
            self.tail.append(line.rstrip())
            low = line.lower()
            if "forwarding failed" in low or "administratively prohibited" in low:
                self.failed = True
                continue
            candidate = pick_url(self.pattern, line, SUBDOMAIN)
            if not candidate:
                continue
            if self.url is None:
                self.url = candidate
            elif SUBDOMAIN and SUBDOMAIN not in self.url and SUBDOMAIN in candidate:
                self.url = candidate      # свой поддомен важнее случайного

    def alive(self) -> bool:
        return self.proc.poll() is None

    def kill(self) -> None:
        try:
            self.proc.kill()
        except Exception:
            pass


DEFAULT_SUBDOMAIN = "nokia1709"
SUBDOMAIN = os.environ.get("MUSCRIPTOR_SUBDOMAIN",
                           DEFAULT_SUBDOMAIN).strip()


def forward_spec(port: int, wants_name: bool) -> str:
    """Строка проброса: с постоянным поддоменом, если он задан."""
    if wants_name and SUBDOMAIN:
        return "%s:80:127.0.0.1:%d" % (SUBDOMAIN, port)
    return "80:127.0.0.1:%d" % port


SSH_OPTS = ["-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null",
            "-o", "ServerAliveInterval=15", "-o", "ServerAliveCountMax=3",
            "-o", "TCPKeepAlive=yes", "-o", "ExitOnForwardFailure=yes"]


SSH_USER = os.environ.get("MUSCRIPTOR_SSH_USER", "nokey")
CUSTOM_SSH = os.environ.get("MUSCRIPTOR_SSH", "")


def custom_spec(port: int) -> tuple[str, list[str], str] | None:
    """Свой ssh-адрес: пользователь@хост или пользователь@хост:порт."""
    if not CUSTOM_SSH:
        return None
    target, ssh_port = CUSTOM_SSH, None
    if ":" in target.rsplit("@", 1)[-1]:
        target, ssh_port = target.rsplit(":", 1)
    argv = ["ssh", "-T", *SSH_OPTS]
    if ssh_port:
        argv += ["-p", ssh_port]
    argv += ["-R", "80:127.0.0.1:%d" % port, target]
    return ("custom", argv, r"https?://[a-zA-Z0-9][a-zA-Z0-9.-]*\.[a-zA-Z]{2,}")


CLOUDFLARED_URL = ("https://github.com/cloudflare/cloudflared/releases/latest/"
                   "download/cloudflared-linux-amd64")


def ensure_cloudflared() -> bool:
    """Резерв без ssh: бинарник тянем один раз, регистрация не нужна."""
    if shutil.which("cloudflared"):
        return True
    target = pathlib.Path("/usr/local/bin/cloudflared")
    say("резервный бинарник не найден, качаю (около 40 МБ)")
    try:
        urllib.request.urlretrieve(CLOUDFLARED_URL, str(target))
    except Exception as problem:
        say("скачать не удалось: %s" % problem)
        return False
    try:
        target.chmod(0o755)
    except Exception:
        pass
    return target.exists()


def specs(port: int) -> list[tuple[str, list[str], str]]:
    head = [custom_spec(port)] if custom_spec(port) else []
    return head + [
        ("tunnl",
         ["ssh", "-T", *SSH_OPTS, "-R", "80:127.0.0.1:%d" % port, "proxy.tunnl.gg"],
         r"https://[a-z0-9-]+\.tunnl\.gg"),
        ("tunnl-stable",
         ["ssh", "-T", *SSH_OPTS, "-R", "80:127.0.0.1:%d" % port, "stable@proxy.tunnl.gg"],
         r"https://[a-z0-9-]+\.tunnl\.gg"),
        ("tuns",
         ["ssh", "-T", *SSH_OPTS, "-R", "80:127.0.0.1:%d" % port, "tuns.sh"],
         r"https://[a-z0-9-]+\.tuns\.sh"),
        ("localhostrun",
         ["ssh", "-T", *SSH_OPTS, "-R", "80:127.0.0.1:%d" % port, SSH_USER + "@localhost.run"],
         r"https://[a-z0-9]+\.lhr\.life"),
        ("serveo",
         ["ssh", "-T", *SSH_OPTS, "-R", forward_spec(port, True), "serveo.net"],
         r"https://(?!console\.|www\.)[a-z0-9-]+(?:\.[a-z0-9-]+)*\."
         r"(?:serveousercontent|serveo)\.[a-z]+"),
        ("pinggy",
         ["ssh", "-T", *SSH_OPTS, "-p", "443", "-R0:127.0.0.1:%d" % port, "a.pinggy.io"],
         r"https://[a-z0-9-]+\.(?:a\.)?pinggy\.(?:link|online)"),
        ("cloudflared",
         ["cloudflared", "tunnel", "--url", "http://127.0.0.1:%d" % port, "--no-autoupdate",
          "--protocol", "http2"],
         r"https://[a-z0-9-]+\.trycloudflare\.com"),
    ]


BROWSER_UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
              "(KHTML, like Gecko) Chrome/128.0 Safari/537.36")


BYPASS_HEADERS = {"serveo-skip-browser-warning": "true"}
PAGE_MARKS = ("continue to", "continue to site", "proceed", "phishing",
              "browser warning", "serveo")


def probe_api(url: str, timeout: float = 10.0) -> tuple[bool, str]:
    """API через туннель: JSON со списком инструментов, заглушка — не поломка."""
    request = urllib.request.Request(url + "/instruments", headers=BYPASS_HEADERS)
    try:
        with urllib.request.urlopen(request, timeout=timeout) as answer:
            status = answer.status
            body = answer.read().decode("utf-8", "ignore")
    except urllib.error.HTTPError as problem:
        return False, "API код %d" % problem.code
    except Exception as problem:
        return False, "API не ответил (%s)" % type(problem).__name__
    if status != 200:
        return False, "API код %d" % status
    try:
        data = json.loads(body)
    except Exception:
        low = body.lower()
        if any(mark in low for mark in PAGE_MARKS):
            return True, "API за страницей-предупреждением провайдера"
        return False, "API отдал не JSON"
    names = data.get("instruments") if isinstance(data, dict) else None
    if not names:
        return False, "API отдал пустой список инструментов"
    return True, "API: %d инструментов" % len(names)


def probe_page(url: str, timeout: float = 15.0) -> tuple[int, str]:
    """Что реально увидит браузер: код и начало страницы."""
    request = urllib.request.Request(url + "/", headers={
        "User-Agent": BROWSER_UA,
        "Accept": "text/html,application/xhtml+xml",
        "Accept-Language": "ru,en;q=0.8",
    })
    try:
        with urllib.request.urlopen(request, timeout=timeout) as answer:
            body = answer.read(4000).decode("utf-8", "ignore")
            return answer.status, body
    except urllib.error.HTTPError as problem:
        try:
            body = problem.read(800).decode("utf-8", "ignore")
        except Exception:
            body = ""
        return problem.code, body
    except Exception as problem:
        return 0, str(problem)


INTERSTITIAL_MARKS = ("continue", "warning", "proceed", "restricted",
                      "phishing", "confirm you are human", "продолж")


def is_interstitial(body: str) -> bool:
    low = body.lower()
    provider = any(name in low for name in ("tunnl", "pinggy", "localhost.run",
                                            "serveo", "lhr.life"))
    return provider and any(mark in low for mark in INTERSTITIAL_MARKS)


def looks_like_app(status: int, body: str) -> tuple[bool, str]:
    if status != 200:
        return False, "код %s" % (status or "нет ответа")
    low = body.lower()
    if "muscriptor" in low:
        return True, "страница MuScriptor"
    if is_interstitial(body):
        return True, "страница-предупреждение провайдера, приложение за ней"
    if "<html" in low:
        return False, "открылась чужая страница (%d байт)" % len(body)
    return False, "ответ без страницы (%d байт)" % len(body)


def reachable(url: str, attempts: int = 4) -> bool:
    for _ in range(attempts):
        try:
            with urllib.request.urlopen(url + "/health", timeout=8) as answer:
                if answer.status == 200:
                    return True
        except Exception:
            pass
        time.sleep(3)
    return False


def open_tunnel(port: int, order: tuple[str, ...]) -> Tunnel:
    table = {name: (argv, pattern) for name, argv, pattern in specs(port)}
    fails = []
    for name in order:
        if name not in table:
            continue
        argv, pattern = table[name]
        if not shutil.which(argv[0]) and name == "cloudflared":
            ensure_cloudflared()
        if not shutil.which(argv[0]):
            say("  [%s] нет команды %s, пропускаю" % (name, argv[0]))
            continue
        say("  [%s] открываю" % name)
        tunnel = Tunnel(name, argv, pattern)
        deadline, started = time.time() + 60, time.time()
        while (time.time() < deadline and tunnel.url is None
               and tunnel.alive() and not tunnel.failed):
            time.sleep(1)
            waited = int(time.time() - started)
            if waited and waited % 10 == 0:
                say("      [%s] ждём адрес, %d с" % (name, waited))
        if tunnel.url:
            status, body = probe_page(tunnel.url)
            fine, note = looks_like_app(status, body)
            if not fine and reachable(tunnel.url):
                fine, note = True, "жив, страницу проверить не вышло"
            if fine:
                say("  [%s] работает: %s (%s)" % (name, tunnel.url, note))
                return tunnel
            say("  [%s] адрес есть, приложение не отдаёт: %s" % (name, note))
        fails.append((name, tunnel.url, list(tunnel.tail)[-2:]))
        tunnel.kill()
    for name, url, tail in fails:
        say("  [%s] не вышло: %s | %s" % (name, url or "нет адреса", " / ".join(tail)))
    raise SystemExit("ни один туннель не поднялся")


BAND_INSTRUMENTS = ("drums,electric_bass,acoustic_bass,chromatic_percussion,"
                    "acoustic_piano,electric_piano,clean_electric_guitar,synth_pad")


def ensure_audio_tools() -> bool:
    """fluidsynth рендерит MIDI в WAV на сервере; pip его не ставит."""
    if shutil.which("fluidsynth"):
        return True
    say("      fluidsynth не найден, ставлю (нужен для рендера аудио)")
    env = dict(os.environ, DEBIAN_FRONTEND="noninteractive")
    steps = (["apt-get", "update", "-qq"],
             ["apt-get", "install", "-y", "--no-install-recommends", "fluidsynth"])
    for argv in steps:
        try:
            run = subprocess.run(argv, env=env, stdout=subprocess.PIPE,
                                 stderr=subprocess.STDOUT, timeout=600)
            if run.returncode != 0:
                tail = run.stdout.decode("utf-8", "replace").strip().splitlines()[-2:]
                say("      %s: %s" % (argv[0], " / ".join(tail) or "код %d" % run.returncode))
        except Exception as problem:
            say("      %s не сработал: %s" % (argv[0], problem))
    if shutil.which("fluidsynth"):
        say("      fluidsynth готов: %s" % shutil.which("fluidsynth"))
        return True
    say("      без fluidsynth транскрипция работает, рендер в WAV нет")
    return False


WIDE_INSTRUMENTS = ("acoustic_piano,electric_piano,chromatic_percussion,organ,acoustic_guitar,clean_electric_guitar,acoustic_bass,electric_bass,string_ensemble,voice,synth_pad,drums")

ALL_INSTRUMENTS = ("acoustic_piano,electric_piano,chromatic_percussion,organ,acoustic_guitar,clean_electric_guitar,distorted_electric_guitar,acoustic_bass,electric_bass,violin,viola,cello,contrabass,orchestral_harp,timpani,string_ensemble,synth_strings,voice,orchestra_hit,trumpet,trombone,tuba,french_horn,brass_section,soprano_and_alto_sax,tenor_sax,baritone_sax,oboe,english_horn,bassoon,clarinet,flutes,synth_lead,synth_pad,drums")


def expand_instruments(text: str) -> list[str]:
    """Разбор набора: all → все группы, auto → пусто, иначе список через запятую."""
    cleaned = (text or "").strip()
    if not cleaned or cleaned.lower() == "auto":
        return []
    if cleaned.lower() in ("all", "все"):
        return ALL_INSTRUMENTS.split(",")
    if cleaned.lower() in ("wide", "широкий"):
        return WIDE_INSTRUMENTS.split(",")
    return [name.strip() for name in cleaned.split(",") if name.strip()]

MAX_GROUPS = 10                      # шире модель размазывается: замерено
BAND_LIST = "drums,electric_bass,acoustic_bass,chromatic_percussion,acoustic_piano,electric_piano,clean_electric_guitar,synth_pad".split(",")


def guard_instruments(names: list[str], label: str = "") -> tuple[list[str], str]:
    """Свести опасный набор к проверенному и объяснить почему."""
    if not names:
        return names, ""
    band_alias = [n for n in names if n.lower() in ("band", "набор")]
    if band_alias:
        return list(BAND_LIST), "%s: это слово-псевдоним, беру проверенный набор" % label
    known = set(ALL_INSTRUMENTS.split(","))
    unknown = [n for n in names if n not in known]
    if unknown:
        names = [n for n in names if n in known]
        if not names:
            return list(BAND_LIST), ("%s: неизвестные группы %s, беру проверенный набор"
                                     % (label, ", ".join(unknown[:4])))
        return names, "%s: выкинул неизвестные группы %s" % (label, ", ".join(unknown[:4]))
    if len(names) > MAX_GROUPS:
        return list(BAND_LIST), (
            "%s: %d групп из %d — модель размазывается (замер: 35 групп дали 0 нот,"
            " 12 групп хуже на 9 попаданий). Беру проверенный набор из %d."
            % (label or "набор", len(names), len(names), len(BAND_LIST)))
    return names, ""

def start_proxy(server_port: int, listen_port: int,
                instruments: str = "band",
                multi_pass: str = "",
                no_articulation: bool = False,
                local_font: bool = False) -> subprocess.Popen:
    """SSE-прокси: держит поток /transcribe живым через медленный туннель."""
    script = pathlib.Path(__file__).with_name("sse_proxy.py")
    if not script.exists():
        script = pathlib.Path(__file__).parent / "sse_proxy.py"
    proc = subprocess.Popen(
        [sys.executable, "-u", str(script),
         "--listen", str(listen_port), "--target", str(server_port),
         "--instruments", instruments]
        + (["--multi-pass", multi_pass] if multi_pass else [])
        + (["--font-cdn", "none"] if local_font else []),
        stdout=open("muscriptor_proxy.log", "a"), stderr=subprocess.STDOUT)
    for _ in range(20):
        time.sleep(0.5)
        if proc.poll() is not None:
            say("прокси не поднялся, работаю напрямую (поток может рваться)")
            return proc
        try:
            with urllib.request.urlopen("http://127.0.0.1:%d/health" % listen_port,
                                        timeout=3) as answer:
                if answer.status == 200:
                    return proc
        except Exception:
            continue
    return proc


def speed_probe(url: str, timeout: float = 25.0) -> tuple[float, float, int]:
    """Вернуть (ttfb, всего, байт) для страницы по ссылке."""
    import time as _time
    started = _time.time()
    try:
        with urllib.request.urlopen(url.rstrip("/") + "/", timeout=timeout) as answer:
            first = answer.read(1)
            first_at = _time.time()
            rest = answer.read()
            return first_at - started, _time.time() - started, len(first) + len(rest)
    except Exception:
        return -1.0, _time.time() - started, 0


def write_link(url: str, port: int, public_port: int | None = None) -> None:
    lines = ["MuScriptor " + time.strftime("%H:%M:%S"), "ссылка: " + url,
             "локально: http://127.0.0.1:%d" % port]
    if public_port and public_port != port:
        lines.append("локально через прокси: http://127.0.0.1:%d" % public_port)
    with open(LINKS_FILE, "w", encoding="utf-8") as handle:
        handle.write("\n".join(lines) + "\n")
    say("\n" + "\n".join(lines))


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="MuScriptor: сервер и ссылка")
    parser.add_argument("--port", type=int, default=8222)
    parser.add_argument("--model", default=DEFAULT_MODEL,
                        help="путь к safetensors, имя размера или hf://org/repo/file")
    parser.add_argument("--device", default="auto", help="auto | cpu | cuda | cuda:0")
    parser.add_argument("--dtype", default=None, help="float32 | float16 | bfloat16")
    parser.add_argument("--subdomain", default=DEFAULT_SUBDOMAIN,
                        help="постоянный поддомен serveo, например nokia1709")
    parser.add_argument("--tunnel", default=None, help="только этот туннель")
    parser.add_argument("--ssh", default=None,
                        help="свой ssh-туннель: пользователь@хост или пользователь@хост:порт")
    parser.add_argument("--ssh-user", default=None,
                        help="имя пользователя для локального туннеля (по умолчанию nokey)")
    parser.add_argument("--no-tunnel", action="store_true", help="без публичной ссылки")
    parser.add_argument("--keep", type=int, default=0, help="держать N секунд и выйти")
    parser.add_argument("--low-memory", action="store_true",
                        help="экономный режим для слабого ПК: меньше потоков и арен,"
                             " веса всё так же float32")
    parser.add_argument("--threads", type=int, default=None,
                        help="сколько потоков дать модели (по умолчанию половина ядер)")
    parser.add_argument("--skip-audio", action="store_true",
                        help="не ставить fluidsynth (рендер аудио будет недоступен)")
    parser.add_argument("--local-font", action="store_true",
                        help="звуковой банк отдавать с сервера, а не с CDN"
                             " (по умолчанию идёт с CDN: 38 МБ не через туннель)")
    parser.add_argument("--no-articulation", action="store_true",
                        help="не пересчитывать концы нот и не ставить легато/стакато")
    parser.add_argument("--multi-pass", default="",
                        help="второй проход с другим набором инструментов и слияние,"
                             " например: chromatic_percussion,electric_bass,acoustic_bass")
    parser.add_argument("--instruments", default="band",
                        help="инструменты для модели: band (по умолчанию, 8 групп), auto"
                             " или список через запятую")
    args = parser.parse_args(argv)
    global SSH_USER, CUSTOM_SSH, SUBDOMAIN
    if args.subdomain:
        SUBDOMAIN = args.subdomain.strip()
        os.environ["MUSCRIPTOR_SUBDOMAIN"] = SUBDOMAIN
    if args.ssh_user:
        SSH_USER = args.ssh_user
    if args.ssh:
        CUSTOM_SSH = args.ssh

    ensure_package()
    checked, warning = guard_instruments(expand_instruments(args.instruments),
                                         "--instruments")
    if warning:
        say("      " + warning)
        args.instruments = ",".join(checked)
    if args.multi_pass:
        checked_pass, warning = guard_instruments(expand_instruments(args.multi_pass),
                                                  "--multi-pass")
        if warning:
            say("      " + warning)
            args.multi_pass = ",".join(checked_pass)
    if not args.skip_audio:
        ensure_audio_tools()
    ensure_ssh()
    ensure_key()
    env = None
    if args.low_memory or args.threads:
        env = low_memory_env(args.threads)
        say("      экономный режим: потоков %s, арен malloc 2, веса float32"
            % env["OMP_NUM_THREADS"])
    server = start_server(args.port, args.model, args.device, args.dtype, env)
    say("[1/2] сервер отвечает: http://127.0.0.1:%d" % args.port)
    proxy_port = args.port + 1
    proxy = start_proxy(args.port, proxy_port, args.instruments, args.multi_pass,
                        args.no_articulation, args.local_font)
    if proxy.poll() is None:
        say("      поток защищён прокси: http://127.0.0.1:%d" % proxy_port)
    else:
        proxy_port = args.port

    if args.no_tunnel:
        write_link("нет (локальный режим)", args.port, proxy_port)
        say(">>> LOCAL LINK: http://127.0.0.1:%d" % args.port)
        tunnel, order = None, ()
    else:
        say("[2/2] публичная ссылка")
        if SUBDOMAIN and "serveo" in (args.tunnel or "serveo"):
            say("      просим поддомен serveo: %s" % SUBDOMAIN)
        base = TUNNEL_ORDER
        if CUSTOM_SSH and "custom" not in base:
            base = ("custom",) + tuple(base)
        order = (args.tunnel,) + tuple(n for n in base if n != args.tunnel) \
            if args.tunnel else base
        tunnel = open_tunnel(proxy_port, order)
        write_link(tunnel.url or "", args.port, proxy_port)
        say(">>> OPEN THIS LINK: " + str(tunnel.url))
        if getattr(tunnel, "interstitial", False):
            say("      ВНИМАНИЕ: провайдер показывает страницу-предупреждение;")
            say("      из-за неё панель инструментов и транскрипция могут не работать.")
            say("      Перезапусти с --tunnel localhostrun")
        if tunnel.url and SUBDOMAIN and SUBDOMAIN not in tunnel.url:
            say("      провайдер выдал другой адрес (свой поддомен занят): %s" % tunnel.url)
        if tunnel.url:
            api_ok, api_note = probe_api(tunnel.url)
            say("      API снаружи: %s" % api_note)
            ttfb, total, size = speed_probe(tunnel.url)
            if ttfb >= 0:
                say("      скорость ссылки: первый байт %.1f с, страница %.1f с"
                    % (ttfb, total))
                if ttfb > 2.5 or total > 5.0:
                    say("      медленная ссылка (%s): попробуй --tunnel pinggy"
                        " или --tunnel serveo" % tunnel.name)
            else:
                say("      скорость ссылки померить не удалось (страница не открылась)")

    started, beat = time.time(), time.time()
    public_fails, checked = 0, time.time()
    reopens = 0
    try:
        while True:
            time.sleep(10)
            if time.time() - beat > 60:
                beat = time.time()
                say("      жив: %.1f мин | сервер %s | ссылка %s"
                    % ((time.time() - started) / 60,
                       "ок" if server.poll() is None else "упал",
                       tunnel.url if tunnel else "локально"))
            if args.keep and time.time() - started > args.keep:
                break
            if server.poll() is not None:
                say("сервер упал, поднимаю заново")
                server = start_server(args.port, args.model, args.device, args.dtype, env)
            if proxy.poll() is not None:
                say("прокси упал, поднимаю заново")
                proxy = start_proxy(args.port, proxy_port, args.instruments,
                                    args.multi_pass, args.no_articulation,
                                    args.local_font)
            if tunnel is not None and tunnel.url and time.time() - checked > 60:
                checked = time.time()
                status, body = probe_page(tunnel.url)
                fine, note = looks_like_app(status, body)
                api_fine, api_note = probe_api(tunnel.url)
                if fine and not api_fine:
                    fine, note = False, api_note
                if fine:
                    public_fails = 0
                else:
                    public_fails += 1
                    say("      ссылка снаружи: %s (%d подряд)" % (note, public_fails))
                    if public_fails >= 3 and reopens < 5:
                        reopens += 1
                        say("      переоткрываю туннель: снаружи не отвечает"
                            " (раз %d из 5)" % reopens)
                        tunnel.kill()
                        tunnel = open_tunnel(proxy_port, order)
                        write_link(tunnel.url or "", args.port, proxy_port)
                        if tunnel.url:
                            say(">>> OPEN THIS LINK: " + tunnel.url)
                        public_fails = 0
                    elif public_fails >= 3:
                        say("      переоткрытий уже 5, оставляю этот туннель:"
                            " проблема не в нём")
            if tunnel is not None and not tunnel.alive():
                say("туннель умер, открываю новый")
                tunnel.kill()
                tunnel = open_tunnel(proxy_port, order)
                write_link(tunnel.url or "", args.port, proxy_port)
                if tunnel.url:
                    say(">>> OPEN THIS LINK: " + tunnel.url)
    except KeyboardInterrupt:
        pass
    if tunnel is not None:
        tunnel.kill()
    if proxy.poll() is None:
        proxy.terminate()
    if server.poll() is None:
        server.terminate()
    say("остановлено")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\nостановлено вручную", flush=True)
        sys.exit(130)