"""Simbiot Harness — транскрипция голосовых сообщений Telegram.

Пайплайн: file_id (в MessageReceived) → getFile → download → ASR →
MessageTranscribed (транскрипт в проекцию inbox) → минутный уровень
видит голосовое как обычный текст и отвечает по смыслу `tg-reply`.

ASR — два яруса:
  1) whisper-asr-webservice (docker 127.0.0.1:8084, openai_whisper base):
     быстрый, но короткие русские фразы иногда глуховатые;
  2) fallback: faster-whisper large-v3-turbo (CPU int8, модель уже в
     HF-кэше сервера, ~5-15с на короткое голосовое) — точнее на русском.
     Живёт в системном python3 (в venv харнеса пакета нет) — вызываем
     через subprocess-обёртку `_fw_asr_worker.py`.

Вызывает инбокс-поллер сразу после MessageReceived (транскрипт в сторе
ещё до первого минутного тика — уровню не нужно качать файлы, его
таймаут мал). Аудио кэшируется по file_unique_id: повторная расшифровка
не качает из Telegram и не перегоняет через ASR.
"""
import base64
import json
import os
import subprocess
import urllib.error
import urllib.parse
import urllib.request

from . import es

# ─── конфигурация ───────────────────────────────────────────────────────────
_HARNESS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
from . import paths as _paths  # t-261004113351-2858c1
_ENV_PATH = _paths.env_path()
_FW_WORKER = os.path.join(_HARNESS_DIR, "harness", "_fw_asr_worker.py")
_FW_PYTHON = "/usr/bin/python3"  # faster-whisper живёт в системном python

ASR_URL = os.environ.get("SIMBIOT_ASR_URL", "http://127.0.0.1:8084")

VOICE_DIR = _paths.voice_dir()  # SIMBIOT_HOME → <home>/voice; legacy без хома
# что расшифровываем: voice — главное; audio/video_note/video — та же речь
# (t-261004152714: видео-вложения Alex получали not_transcribable — звук
# из контейнера вообще не извлекался; теперь тянем дорожку ffmpeg'ом)
TRANSCRIBABLE_KINDS = {"voice", "audio", "video_note", "video"}
# длина, сек: длиннее не тянем (поллер не должен вязнуть в 10-мин озвучках)
MAX_DURATION = 300
# аудио-кэш: file_unique_id → ogg-файл (путь задаёт paths.voice_dir выше)

TG_TOKEN = None


def _tg_init():
    """Токен бота из .env (кэш в глобале)."""
    global TG_TOKEN
    if TG_TOKEN is None:
        if os.path.exists(_ENV_PATH):
            with open(_ENV_PATH) as f:
                for line in f:
                    line = line.strip()
                    if line.startswith("SIMBIOT_TG_TOKEN="):
                        TG_TOKEN = line.split("=", 1)[1].strip()
    return TG_TOKEN


# ───────────────────────────── TG download ──────────────────────────────────

def _download_voice(file_id: str, file_unique_id: str = ""):
    """getFile → download. Аудио кэшируется на диске по file_unique_id.

    Возвращает (bytes, cache_path) или (None, None) при неудаче.
    Старые голосовые (файл истёк у Telegram) дают 400 — это не ошибка
    поллера, просто расшифровка невозможна.
    """
    if not _tg_init():
        return None, None
    cache = os.path.join(VOICE_DIR, (file_unique_id or file_id) + ".ogg")
    if os.path.exists(cache):  # кэш: повтор не качает из Telegram
        with open(cache, "rb") as f:
            return f.read(), cache
    try:
        url = f"https://api.telegram.org/bot{TG_TOKEN}/getFile"
        data = urllib.parse.urlencode({"file_id": file_id}).encode()
        with urllib.request.urlopen(
                urllib.request.Request(url, data=data), timeout=30) as r:
            resp = json.loads(r.read().decode())
        if not resp.get("ok"):
            return None, None
        path = resp["result"].get("file_path")
        if not path:
            return None, None
        durl = f"https://api.telegram.org/file/bot{TG_TOKEN}/{path}"
        with urllib.request.urlopen(durl, timeout=60) as r:
            audio = r.read()
    except urllib.error.HTTPError as e:
        # 400 = file_id истёк (TG хранит файлы ограниченно) — тихо отдаём отказ
        print(f"[voice] getFile HTTP {e.code}: файл недоступен", flush=True)
        return None, None
    except Exception as e:
        print(f"[voice] download error: {e}", flush=True)
        return None, None
    try:  # кэш — best effort
        os.makedirs(VOICE_DIR, exist_ok=True)
        with open(cache, "wb") as f:
            f.write(audio)
    except Exception:
        pass
    return audio, cache


# ─────────── тяжёлые воркеры ВНЕ cgroup поллера (t-261004160531-780666) ─────
# faster-whisper large-v3-turbo (int8) живёт в ~2-3G RSS, ffmpeg тоже кусается:
# заспавненные поллером как обычные дети они падали в ЕГО cgroup — systemd
# Consumption показывал 3.7G peak у simbiot-harness-tg. Теперь тяжёлое уходит
# в собственный transient-scope (systemd-run --scope) со своим MemoryMax;
# поллер остаётся лёгким (<100M), его юнит держит MemoryMax=1G.
# Нет systemd (локалка/тесты) — прозрачный fallback на прямой запуск.
_ASR_SCOPE_MEM = os.environ.get("SIMBIOT_ASR_SCOPE_MEM", "3G")
_HAS_SYSTEMD = os.path.isdir("/run/systemd/system") and os.path.exists("/usr/bin/systemd-run")


def _heavy_argv(argv):
    """argv тяжёлого воркера → запуск в отдельном systemd-scope с MemoryMax."""
    if not _HAS_SYSTEMD:
        return list(argv)
    import uuid
    unit = f"simbiot-asr-{uuid.uuid4().hex[:10]}"
    return ["systemd-run", "--scope", "--quiet", "--collect",
            f"--unit={unit}", f"-p", f"MemoryMax={_ASR_SCOPE_MEM}",
            "--"] + list(argv)


def _heavy_run(argv, input_bytes=None, timeout=600):
    """Тяжёлый subprocess: scope-изоляция памяти + killpg на таймауте.

    systemd-run --scope выполняет команду синхронно, поэтому таймаут
    работает как раньше; start_new_session + killpg (паттерн
    isolated_runner, e-261003004801) не оставляют осиротевший воркер,
    если тик умер раньше него.
    """
    import signal
    proc = subprocess.Popen(
        _heavy_argv(argv), stdin=subprocess.PIPE, stdout=subprocess.PIPE,
        stderr=subprocess.PIPE, start_new_session=True)
    try:
        out, err = proc.communicate(input=input_bytes, timeout=timeout)
        return proc.returncode, out or b"", err or b""
    except subprocess.TimeoutExpired:
        try:
            os.killpg(proc.pid, signal.SIGKILL)
        except (ProcessLookupError, PermissionError, OSError):
            pass
        try:  # дождаться смерти группы, чтобы зомби не сидели в scope
            proc.communicate(timeout=10)
        except Exception:
            pass
        return None, b"", b""


# ────────────── извлечение аудио из видеоконтейнера ─────────────────────────


def extract_audio_track(container: bytes):
    """Видеоконтейнер (mp4 и т.п.) → ogg/opus аудиодорожка через ffmpeg.

    Возвращает bytes дорожки или None (нет дорожки / ffmpeg отказал).
    ffmpeg есть на хосте (6.1.1) — тянуть в venv нечего. t-261004152714
    """
    import tempfile
    src = dst = None
    try:
        with tempfile.NamedTemporaryFile(prefix="simbiot-v_", delete=False) as tf:
            tf.write(container)
            src = tf.name
        # -vn: только звук; моно 16к — формат, который любят оба ASR-яруса.
        # ffmpeg тяжёлый → в свой scope (память поллера не растёт)
        rc, out, err = _heavy_run(
            ["ffmpeg", "-v", "error", "-i", src, "-vn", "-ac", "1",
             "-ar", "16000", "-f", "ogg", "-c:a", "libopus", "-"],
            timeout=180)
        if rc != 0 or not out:
            err = err.decode()[:200]
            print(f"[voice] ffmpeg: дорожка не извлечена ({err})", flush=True)
            return None
        return out
    except Exception as e:
        print(f"[voice] ffmpeg error: {e}", flush=True)
        return None
    finally:
        for p in (src, dst):
            if p and os.path.exists(p):
                try:
                    os.unlink(p)
                except OSError:
                    pass


# ─────────────────────────────── ASR ярусы ──────────────────────────────────

def transcribe_bytes(audio: bytes, filename: str = "audio.ogg",
                     long: bool = False):
    """Байты аудио → текст. '' = тишина/пусто, None = не смогли.

    Ярус 1 — локальный ASR-контейнер whisper base (быстро, ~1с; короткие
    русские фразы распознаёт с ошибками). Ярус 2 — faster-whisper
    large-v3-turbo (точнее на русском, ~7-15с на короткое голосовое).
    Второй ярус включается: при пустом результате первого или при
    long=True (короткие фразы и длинные записи — где base слабее всего).
    """
    text = _asr_webservice(audio, filename)
    if text:
        return text, "whisper-asr-base"
    # короткие фразы (≤30с) — зона, где base ошибается чаще всего;
    # длинные записи base тянет хорошо, но large-v3-turbo точнее
    fw = _asr_faster_whisper(audio, filename)
    if fw:  # непустой результат второго яруса
        return fw, "faster-whisper-large-v3-turbo"
    return text, None  # '' = тишина, None = оба яруса недоступны


def _asr_webservice(audio: bytes, filename: str):
    """POST /asr в whisper-asr-webservice. None при недоступности сервиса."""
    import uuid
    boundary = "----simbiot" + uuid.uuid4().hex
    body = (
        f'--{boundary}\r\nContent-Disposition: form-data; name="audio_file"; '
        f'filename="{filename}"\r\nContent-Type: audio/ogg\r\n\r\n'
    ).encode() + audio + f"\r\n--{boundary}--\r\n".encode()
    req = urllib.request.Request(
        f"{ASR_URL}/asr?output=json", data=body,
        headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
    )
    try:
        with urllib.request.urlopen(req, timeout=120) as r:
            data = json.loads(r.read().decode())
            return (data.get("text") or "").strip()
    except Exception as e:
        print(f"[voice] asr-webservice error: {e}", flush=True)
        return None


def _asr_faster_whisper(audio: bytes, filename: str):
    """faster-whisper large-v3-turbo через системный python3 (subprocess).

    В venv харнеса пакета нет (и не тащим: ctranslate2 тянет за собой
    тяжёлые бинарники) — модель живёт в системном python, зовём воркером.
    None при отказе, '' при пустой расшифровке.
    """
    try:
        req = json.dumps({
            "audio_b64": base64.b64encode(audio).decode(),
            "filename": filename,
        }, ensure_ascii=False)
        # faster-whisper large-v3-turbo ~2-3G RSS: только в свой scope,
        # иначе память ложится в cgroup поллера (3.7G peak, t-261004160531)
        rc, out, err = _heavy_run(
            [_FW_PYTHON, _FW_WORKER], input_bytes=req.encode(), timeout=600)
        if rc is None:
            print("[voice] faster-whisper: таймаут (группа убита)", flush=True)
            return None
        out = out.decode() if out else ""
        try:
            data = json.loads(out.strip().splitlines()[-1]) if out.strip() else {}
        except (ValueError, IndexError):
            data = {"error": err.decode()[:300]}
        if data.get("error") or rc != 0:
            print(f"[voice] faster-whisper error: "
                  f"{data.get('error') or rc}", flush=True)
            return None
        return (data.get("text") or "").strip()
    except Exception as e:
        print(f"[voice] faster-whisper subprocess error: {e}", flush=True)
        return None


# ─────────────────────────── оркестрация ────────────────────────────────────

def transcribe_message(msg_id: str, file_id: str, file_unique_id: str = "",
                       duration=None, kind: str = "voice"):
    """Полный путь одного входящего: download + ASR + MessageTranscribed.

    msg_id — 'm-NNN' (aggregate_id). Возвращает (текст, engine):
      текст '' = тишина/пусто, None = расшифровать не смогли;
    идемпотентно: повторный вызов по уже расшифрованному — транскрипт из
    read model без хождений в Telegram/ASR.
    """
    model = es.run_projections()
    m = (model.get("inbox") or {}).get(msg_id)
    if m is None:
        return None, None  # входящее не найдено (нет MessageReceived)
    if m.get("transcribed"):
        return m.get("transcript") or "", m.get("engine")

    reason = None
    if kind not in TRANSCRIBABLE_KINDS:
        reason = "not_transcribable"
    elif duration and duration > MAX_DURATION:
        reason = "too_long"
    if reason:
        es.append_event("inbox", msg_id, "MessageTranscribed", {
            "transcript": "", "engine": None, "reason": reason,
            "duration": duration,
        })
        return "", None

    src_audio, _path = _download_voice(file_id, file_unique_id)
    if src_audio is None:
        es.append_event("inbox", msg_id, "MessageTranscribed", {
            "transcript": "", "engine": None, "reason": "download_failed",
            "duration": duration,
        })
        return None, None
    # видео-контейнер: речь лежит в дорожке — вынимаем её ffmpeg'ом
    if kind == "video":
        audio = extract_audio_track(src_audio)
        if audio is None:
            es.append_event("inbox", msg_id, "MessageTranscribed", {
                "transcript": "", "engine": None, "reason": "audio_track_failed",
                "duration": duration,
            })
            return None, None
    else:
        audio = src_audio
    # короткие русские фразы base-модель коверкает («Туше, что ты тебя
    # слышишь» вместо «Ты теперь слышишь») — для voice/video_note/video ≤60с
    # сразу идём в точный large-v3-turbo, long-записи — как получится
    short_speech = kind in ("voice", "video_note", "video") and (duration or 0) <= 60
    if short_speech:
        text, engine = _asr_faster_whisper(audio, "audio.ogg"), "faster-whisper-large-v3-turbo"
        if text:  # точный путь сработал
            es.append_event("inbox", msg_id, "MessageTranscribed", {
                "transcript": text[:4000], "engine": engine,
                "bytes": len(audio), "duration": duration,
            })
            return text, engine
    text, engine = transcribe_bytes(audio, "audio.ogg")
    es.append_event("inbox", msg_id, "MessageTranscribed", {
        "transcript": (text or "")[:4000],
        "engine": engine if text else None,
        "reason": None if text is not None else "asr_failed",
        "bytes": len(audio),
        "duration": duration,
    })
    return text, engine