from __future__ import annotations

import asyncio
import html
import os
import re
import sys
import time
from pathlib import Path

import aiohttp
from aiogram import Bot
from aiogram.client.session.aiohttp import AiohttpSession
from aiogram.client.telegram import TelegramAPIServer
from aiogram.exceptions import TelegramRetryAfter
from aiogram.types import BufferedInputFile, FSInputFile, InputMediaPhoto, InputMediaVideo
from loguru import logger

from ..config import settings
from ..constants import (
    JOB_TYPE_VK_STORAGE_COPY,
    MEDIA_STATUS_FAILED,
    MEDIA_STATUS_LINK_ONLY,
    MEDIA_STATUS_UPLOADED,
    POST_STATUS_FAILED,
    POST_STATUS_STORAGE_READY,
    WORKER_STORAGE_UPLOADER,
)
from ..db import fetch_float_setting, fetch_int_setting, fetch_setting, get_pool
from ..heartbeat import HeartbeatReporter
from ..jobs import ack_done, ack_retry, claim_job, is_worker_enabled, recover_stale_jobs

TMP_DIR = Path("/tmp")
TMP_PREFIX = "vkparser_tg_media_"
MAX_MEDIA_GROUP = 10


def parse_topic(value: str) -> tuple[int, int | None]:
    raw = str(value or "").strip()
    if not raw:
        raise RuntimeError("TG_MEDIA_CHANNEL_ID is empty")
    if ":" in raw:
        chat_id, thread_id = raw.split(":", 1)
        return int(chat_id), int(thread_id)
    return int(raw), None


def parse_vk_video_url(vk_url: str) -> tuple[int, int] | None:
    match = re.search(r"video(-?\d+)_(\d+)", vk_url or "")
    if not match:
        return None
    return int(match.group(1)), int(match.group(2))


def split_message_chunks(text: str, limit: int) -> list[str]:
    text = str(text or "").strip()
    if not text:
        return []
    chunks: list[str] = []
    while len(text) > limit:
        split_at = text.rfind("\n", 0, limit)
        if split_at < limit // 2:
            split_at = text.rfind(" ", 0, limit)
        if split_at < limit // 2:
            split_at = limit
        chunks.append(text[:split_at].strip())
        text = text[split_at:].strip()
    if text:
        chunks.append(text)
    return chunks


def original_link(post: dict) -> str:
    url = str(post.get("original_url") or "").strip()
    return f'<a href="{url}">#{int(post["id"])}</a>' if url else f'#{int(post["id"])}'


class TelegramStorageUploader:
    def __init__(self) -> None:
        self.pool = None
        self.bot: Bot | None = None
        self.storage_chat_id: int | None = None
        self.storage_thread_id: int | None = None
        self.heartbeat = HeartbeatReporter(WORKER_STORAGE_UPLOADER, 30)
        self.media_group_max_items = MAX_MEDIA_GROUP
        self.media_upload_delay_sec = 1.0
        self.tg_retry_max_attempts = 4
        self.tg_retry_backoff_max_sec = 15
        self.caption_limit = 1024
        self.message_limit = 4096
        self.text_overflow_marker = "Продолжение следующим сообщением."
        self.download_timeout_sec = 45
        self.video_max_size_mb = 49
        self.video_max_height = 720
        self.video_max_duration_sec = 300
        self.yt_dlp_timeout_sec = 300
        self.max_media_attempts = 3
        self.post_job_pause_sec = 0.2
        self.using_local_bot_api = False

    async def init(self) -> None:
        self.pool = await get_pool()
        recovered = await recover_stale_jobs(self.pool, JOB_TYPE_VK_STORAGE_COPY, stale_minutes=20)
        if recovered:
            logger.warning("Recovered stale storage jobs: {}", recovered)

        if not settings.tg_bot_token:
            raise RuntimeError("TG_BOT_TOKEN is empty")
        media_channel = str(await fetch_setting("tg_media_channel_id", settings.tg_media_channel_id) or "").strip()
        self.storage_chat_id, self.storage_thread_id = parse_topic(media_channel)

        local_bot_api_url = str(await fetch_setting("local_bot_api_url", settings.local_bot_api_url) or "").strip()
        if local_bot_api_url:
            session = AiohttpSession(
                api=TelegramAPIServer.from_base(local_bot_api_url.rstrip("/"), is_local=True)
            )
            self.bot = Bot(token=settings.tg_bot_token, session=session)
            try:
                await self.bot.get_me()
                self.using_local_bot_api = True
                logger.info("Using local Telegram Bot API: {}", local_bot_api_url)
            except Exception as exc:
                logger.warning("Local Telegram Bot API unavailable at {}: {}. Falling back to cloud Bot API.", local_bot_api_url, exc)
                await self.bot.session.close()
                self.bot = Bot(token=settings.tg_bot_token)
        else:
            self.bot = Bot(token=settings.tg_bot_token)

        self.media_group_max_items = max(1, min(MAX_MEDIA_GROUP, await fetch_int_setting("telegram_media_group_max_items", 10)))
        self.media_upload_delay_sec = max(0.0, await fetch_float_setting("telegram_media_upload_delay_sec", 1.0))
        self.tg_retry_max_attempts = max(1, await fetch_int_setting("telegram_retry_max_attempts", 4))
        self.tg_retry_backoff_max_sec = max(1, await fetch_int_setting("telegram_retry_backoff_max_sec", 15))
        self.caption_limit = max(128, await fetch_int_setting("telegram_caption_limit", 1024))
        self.message_limit = max(512, await fetch_int_setting("telegram_message_limit", 4096))
        self.text_overflow_marker = str(await fetch_setting("telegram_text_overflow_marker", self.text_overflow_marker))
        self.download_timeout_sec = max(5, await fetch_int_setting("uploader_download_timeout_sec", 45))
        self.video_max_size_mb = max(1, await fetch_int_setting("video_max_size_mb", 49))
        if not self.using_local_bot_api:
            self.video_max_size_mb = min(self.video_max_size_mb, 49)
        self.video_max_height = max(144, await fetch_int_setting("video_max_height", 720))
        self.video_max_duration_sec = max(1, await fetch_int_setting("video_max_duration_sec", 300))
        self.yt_dlp_timeout_sec = max(30, await fetch_int_setting("uploader_yt_dlp_timeout_sec", 300))
        self.max_media_attempts = max(1, await fetch_int_setting("uploader_max_media_attempts", 3))
        self.post_job_pause_sec = max(0.0, await fetch_float_setting("media_post_job_pause_sec", 0.2))

        logger.info(
            "Telegram storage config: media_delay={}s, media_group_max_items={}, tg_retry={}, tg_backoff_max={}s, dl_timeout={}s, ytdlp_timeout={}s, post_pause={}s",
            self.media_upload_delay_sec,
            self.media_group_max_items,
            self.tg_retry_max_attempts,
            self.tg_retry_backoff_max_sec,
            self.download_timeout_sec,
            self.yt_dlp_timeout_sec,
            self.post_job_pause_sec,
        )

    async def close(self) -> None:
        if self.bot:
            await self.bot.session.close()

    def chat_kwargs(self) -> dict:
        kwargs = {"chat_id": self.storage_chat_id}
        if self.storage_thread_id:
            kwargs["message_thread_id"] = self.storage_thread_id
        return kwargs

    async def tg_retry(self, fn):
        for attempt in range(1, self.tg_retry_max_attempts + 1):
            try:
                return await fn()
            except TelegramRetryAfter as exc:
                delay = float(exc.retry_after) + 0.5
                logger.warning("Telegram flood control, sleep {}s", delay)
                await asyncio.sleep(delay)
            except Exception as exc:
                if attempt >= self.tg_retry_max_attempts:
                    raise
                delay = min(2**attempt, self.tg_retry_backoff_max_sec)
                logger.warning("Telegram send error: {}. retry in {}s", exc, delay)
                await asyncio.sleep(delay)
        raise RuntimeError("telegram retry exhausted")

    async def load_raw_post(self, raw_post_id: int) -> dict | None:
        row = await self.pool.fetchrow(
            """
            SELECT rp.*, s.name AS source_name
            FROM raw_posts rp
            JOIN sources s ON s.id=rp.source_id
            WHERE rp.id=$1
            """,
            raw_post_id,
        )
        return dict(row) if row else None

    async def load_media(self, raw_post_id: int) -> list[dict]:
        rows = await self.pool.fetch(
            """
            SELECT *
            FROM raw_post_media
            WHERE raw_post_id=$1
            ORDER BY sort_order ASC, id ASC
            """,
            raw_post_id,
        )
        return [dict(row) for row in rows]

    def media_blocking_error(self, media: list[dict], prepared_media_ids: set[int] | None = None) -> str | None:
        prepared_media_ids = prepared_media_ids or set()
        visible_media = [
            item
            for item in media
            if str(item.get("media_type") or "") in ("photo", "video")
            and not bool(item.get("editor_hidden"))
        ]
        if not visible_media:
            return "no publishable media"
        pending = [
            item
            for item in visible_media
            if str(item.get("status") or "") == "pending"
            and int(item["id"]) not in prepared_media_ids
        ]
        if pending:
            details = ", ".join(
                f"{item.get('media_type')}#{item.get('id')}: {item.get('error') or 'pending'}"
                for item in pending[:5]
            )
            return f"media still pending: {details}"
        failed = [
            item
            for item in visible_media
            if str(item.get("status") or "") == MEDIA_STATUS_FAILED
        ]
        if failed:
            details = ", ".join(
                f"{item.get('media_type')}#{item.get('id')}: {item.get('error') or 'failed'}"
                for item in failed[:5]
            )
            return f"media failed: {details}"
        link_only = [
            item
            for item in visible_media
            if str(item.get("status") or "") == MEDIA_STATUS_LINK_ONLY
        ]
        if link_only:
            details = ", ".join(
                f"{item.get('media_type')}#{item.get('id')}: {item.get('error') or 'link only'}"
                for item in link_only[:5]
            )
            return f"media link only: {details}"
        missing_upload = [
            item
            for item in visible_media
            if not (item.get("tg_file_id") or item.get("storage_attachment_id"))
            and int(item["id"]) not in prepared_media_ids
        ]
        if missing_upload:
            details = ", ".join(
                f"{item.get('media_type')}#{item.get('id')}: missing tg_file_id"
                for item in missing_upload[:5]
            )
            return f"media not uploaded: {details}"
        return None

    async def download_bytes(self, session: aiohttp.ClientSession, url: str) -> bytes | None:
        try:
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=self.download_timeout_sec)) as response:
                if response.status != 200:
                    return None
                return await response.read()
        except Exception:
            return None

    async def mark_media_uploaded(self, media_id: int, file_id: str, unique_id: str | None) -> None:
        await self.pool.execute(
            """
            UPDATE raw_post_media
            SET status=$2,
                storage_attachment_id=$3,
                storage_url=$3,
                tg_file_id=$3,
                tg_file_unique_id=$4,
                error=NULL,
                updated_at=NOW()
            WHERE id=$1
            """,
            media_id,
            MEDIA_STATUS_UPLOADED,
            file_id,
            unique_id,
        )

    async def mark_media_link_only(self, media_id: int, error: str | None = None) -> None:
        await self.pool.execute(
            """
            UPDATE raw_post_media
            SET status=$2,
                error=COALESCE($3, error),
                updated_at=NOW()
            WHERE id=$1
            """,
            media_id,
            MEDIA_STATUS_LINK_ONLY,
            error,
        )

    async def mark_media_failed_attempt(self, media_id: int, error: str) -> None:
        await self.pool.execute(
            """
            UPDATE raw_post_media
            SET attempts=attempts+1,
                status=CASE WHEN attempts + 1 >= $3 THEN $2 ELSE status END,
                error=$4,
                updated_at=NOW()
            WHERE id=$1
            """,
            media_id,
            MEDIA_STATUS_FAILED,
            self.max_media_attempts,
            error[:1000],
        )

    async def download_video(self, vk_url: str, output_path: str) -> dict | None:
        parsed = parse_vk_video_url(vk_url)
        if not parsed:
            return None
        owner_id, video_id = parsed
        max_size = self.video_max_size_mb * 1024 * 1024
        netrc_path = f"{output_path}.netrc"
        fd = os.open(netrc_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        with os.fdopen(fd, "w", encoding="utf-8") as fh:
            fh.write(f"machine vk.com login vk_token password {settings.vk_access_token}\n")
        cmd = [
            sys.executable,
            "-m",
            "yt_dlp",
            "--netrc-location",
            netrc_path,
            f"https://vk.com/video{owner_id}_{video_id}",
            "-o",
            output_path,
            "--no-playlist",
            "-f",
            (
                f"best[height<={self.video_max_height}][filesize<{max_size}]"
                f"/best[height<={self.video_max_height}]"
                f"/bestvideo[height<={self.video_max_height}][filesize<{max_size}]+bestaudio/best"
                f"/bestvideo[height<={self.video_max_height}]+bestaudio/best"
                f"/best[filesize<{max_size}]"
            ),
            "--quiet",
            "--no-warnings",
        ]
        try:
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            _, stderr = await asyncio.wait_for(proc.communicate(), timeout=self.yt_dlp_timeout_sec)
        except asyncio.TimeoutError:
            proc.kill()
            await proc.communicate()
            return {"error": "video download timeout", "permanent": False}
        finally:
            try:
                os.unlink(netrc_path)
            except FileNotFoundError:
                pass
        if proc.returncode != 0 or not os.path.exists(output_path):
            err = (stderr or b"").decode("utf-8", errors="ignore").lower()
            permanent = any(marker in err for marker in ("removed", "unavailable", "private", "access denied"))
            return {"error": "video unavailable or download failed", "permanent": permanent}
        size = os.path.getsize(output_path)
        if size > max_size:
            return {"error": "video too large", "permanent": True}
        return {"path": output_path, "size_bytes": size}

    async def prepare_media(self, raw_post_id: int, media: list[dict]) -> list[dict]:
        prepared: list[dict] = []
        async with aiohttp.ClientSession() as session:
            for item in media:
                media_id = int(item["id"])
                media_type = str(item["media_type"])
                url = str(item.get("original_url") or "")

                if item.get("tg_file_id"):
                    prepared.append({"media_id": media_id, "media_type": media_type, "media": item["tg_file_id"]})
                    continue
                if int(item.get("attempts") or 0) >= self.max_media_attempts:
                    continue

                if media_type == "photo":
                    data = await self.download_bytes(session, url)
                    if not data:
                        await self.mark_media_failed_attempt(media_id, "photo download failed")
                        continue
                    prepared.append(
                        {
                            "media_id": media_id,
                            "media_type": "photo",
                            "media": BufferedInputFile(data, filename=f"photo_{media_id}.jpg"),
                        }
                    )
                    continue

                if media_type == "video":
                    duration = item.get("duration_sec")
                    if duration and int(duration) > self.video_max_duration_sec:
                        await self.mark_media_link_only(media_id, "video too long")
                        continue
                    temp_path = str(TMP_DIR / f"{TMP_PREFIX}{raw_post_id}_{media_id}.mp4")
                    info = await self.download_video(url, temp_path)
                    if not info:
                        await self.mark_media_failed_attempt(media_id, "video download failed")
                        continue
                    if info.get("error"):
                        if info.get("permanent"):
                            await self.mark_media_link_only(media_id, str(info["error"]))
                        else:
                            await self.mark_media_failed_attempt(media_id, str(info["error"]))
                        continue
                    prepared.append(
                        {
                            "media_id": media_id,
                            "media_type": "video",
                            "media": FSInputFile(temp_path, filename="video.mp4"),
                            "tmp_path": temp_path,
                        }
                    )
                    continue

                await self.mark_media_link_only(media_id, "unsupported media type")
        return prepared

    async def send_text_chunk(self, text: str) -> int:
        message = await self.tg_retry(
            lambda: self.bot.send_message(
                text=text,
                parse_mode="HTML",
                disable_web_page_preview=True,
                **self.chat_kwargs(),
            )
        )
        return int(message.message_id)

    def build_text_parts(self, post: dict, has_media: bool, link_only_media: list[dict] | None = None) -> tuple[str | None, list[str]]:
        link = original_link(post)
        raw_text = html.escape(str(post.get("raw_text") or "").strip())
        link_only_media = link_only_media or []
        media_lines = []
        for item in link_only_media:
            url = str(item.get("original_url") or "").strip()
            if not url:
                continue
            media_type = html.escape(str(item.get("media_type") or "media"))
            reason = html.escape(str(item.get("error") or "link only"))
            media_lines.append(f"- {media_type}: <a href=\"{html.escape(url, quote=True)}\">ссылка</a> ({reason})")
        media_note = "\n\nМедиа по ссылке:\n" + "\n".join(media_lines) if media_lines else ""
        suffix = f"\n\n{link}"
        body = raw_text + media_note
        if has_media:
            if len(body + suffix) <= self.caption_limit:
                return body + suffix if body else link, []
            chunks = split_message_chunks(body + suffix, self.message_limit)
            return self.text_overflow_marker[: self.caption_limit], chunks
        chunks = split_message_chunks(body + suffix if body else link, self.message_limit)
        return None, chunks

    async def save_sent_media_ids(self, prepared: list[dict], messages: list) -> None:
        for item, message in zip(prepared, messages):
            if item["media_type"] == "photo" and message.photo:
                photo = message.photo[-1]
                await self.mark_media_uploaded(int(item["media_id"]), photo.file_id, photo.file_unique_id)
            elif item["media_type"] == "video" and message.video:
                video = message.video
                await self.mark_media_uploaded(int(item["media_id"]), video.file_id, video.file_unique_id)

    async def send_storage_post(self, post: dict, prepared: list[dict], media: list[dict]) -> tuple[list[int], int | None]:
        uploaded_media_ids = {int(item["media_id"]) for item in prepared}
        link_only_media = [
            item
            for item in media
            if int(item["id"]) not in uploaded_media_ids and str(item.get("status") or "") == MEDIA_STATUS_LINK_ONLY
        ]
        caption, text_chunks = self.build_text_parts(post, bool(prepared), link_only_media)
        message_ids: list[int] = []

        if prepared:
            first = prepared[: self.media_group_max_items]
            group = []
            for idx, item in enumerate(first):
                cap = caption if idx == 0 else None
                if item["media_type"] == "photo":
                    group.append(InputMediaPhoto(media=item["media"], caption=cap, parse_mode="HTML"))
                elif item["media_type"] == "video":
                    group.append(InputMediaVideo(media=item["media"], caption=cap, parse_mode="HTML"))
            if group:
                messages = await self.tg_retry(lambda: self.bot.send_media_group(media=group, **self.chat_kwargs()))
                await self.save_sent_media_ids(first, messages)
                message_ids.extend(int(msg.message_id) for msg in messages)
                await asyncio.sleep(self.media_upload_delay_sec * len(first))

            rest = prepared[self.media_group_max_items :]
            for start in range(0, len(rest), self.media_group_max_items):
                chunk = rest[start : start + self.media_group_max_items]
                group = [
                    InputMediaPhoto(media=item["media"])
                    if item["media_type"] == "photo"
                    else InputMediaVideo(media=item["media"])
                    for item in chunk
                    if item["media_type"] in ("photo", "video")
                ]
                if group:
                    messages = await self.tg_retry(lambda g=group: self.bot.send_media_group(media=g, **self.chat_kwargs()))
                    await self.save_sent_media_ids(chunk, messages)
                    message_ids.extend(int(msg.message_id) for msg in messages)
                    await asyncio.sleep(self.media_upload_delay_sec * len(chunk))
        for chunk in text_chunks:
            message_ids.append(await self.send_text_chunk(chunk))
        return message_ids, None

    async def mark_post_ready(self, raw_post_id: int, message_ids: list[int], meta_message_id: int | None) -> None:
        await self.pool.execute(
            """
            UPDATE raw_posts
            SET status=$2,
                tg_storage_chat_id=$3,
                tg_storage_thread_id=$4,
                tg_storage_message_ids=$5,
                tg_storage_meta_message_id=$6,
                storage_post_url=$7,
                copied_at=NOW(),
                error_reason=NULL,
                updated_at=NOW()
            WHERE id=$1
            """,
            raw_post_id,
            POST_STATUS_STORAGE_READY,
            self.storage_chat_id,
            self.storage_thread_id,
            message_ids,
            meta_message_id,
            f"tg://resolve?domain=c/{str(abs(int(self.storage_chat_id or 0))).removeprefix('100')}/{message_ids[0]}" if message_ids else None,
        )

    async def mark_post_failed(self, raw_post_id: int, error: str) -> None:
        await self.pool.execute(
            """
            UPDATE raw_posts
            SET status=$2,
                error_reason=$3,
                updated_at=NOW()
            WHERE id=$1
            """,
            raw_post_id,
            POST_STATUS_FAILED,
            error[:1000],
        )

    async def process_job(self, job: dict, worker_id: str) -> None:
        job_id = int(job["id"])
        raw_post_id = int(job["entity_id"])
        await self.heartbeat.beat(self.pool, current_job_id=job_id, force=True)

        post = await self.load_raw_post(raw_post_id)
        if not post:
            await ack_retry(self.pool, job, f"raw_post {raw_post_id} not found", delay_sec=60)
            return
        if post.get("tg_storage_message_ids"):
            await ack_done(self.pool, job_id)
            return

        media = await self.load_media(raw_post_id)
        prepared = await self.prepare_media(raw_post_id, media)
        media = await self.load_media(raw_post_id)
        prepared_media_ids = {int(item["media_id"]) for item in prepared}
        blocking_error = self.media_blocking_error(media, prepared_media_ids)
        if blocking_error:
            for item in prepared:
                tmp_path = item.get("tmp_path")
                if tmp_path:
                    try:
                        if os.path.exists(tmp_path):
                            os.remove(tmp_path)
                    except Exception:
                        pass
            if not blocking_error.startswith("media still pending:"):
                await self.mark_post_failed(raw_post_id, blocking_error)
                await ack_done(self.pool, job_id)
                logger.warning("Telegram storage blocked raw_post={}: {}", raw_post_id, blocking_error)
                return
            await ack_retry(self.pool, job, blocking_error, delay_sec=120)
            logger.warning("Telegram storage retry raw_post={}: {}", raw_post_id, blocking_error)
            return
        try:
            message_ids, meta_message_id = await self.send_storage_post(post, prepared, media)
        finally:
            for item in prepared:
                tmp_path = item.get("tmp_path")
                if tmp_path:
                    try:
                        if os.path.exists(tmp_path):
                            os.remove(tmp_path)
                    except Exception:
                        pass
        await self.mark_post_ready(raw_post_id, message_ids, meta_message_id)
        await ack_done(self.pool, job_id)
        logger.info("Telegram storage done: raw_post={} messages={}", raw_post_id, message_ids)

    async def run_once(self, worker_id: str) -> bool:
        enabled = await is_worker_enabled(self.pool, WORKER_STORAGE_UPLOADER)
        if not enabled:
            await self.heartbeat.beat(self.pool, status="disabled", force=True)
            return False

        job = await claim_job(self.pool, JOB_TYPE_VK_STORAGE_COPY, worker_id)
        if not job:
            await self.heartbeat.beat(self.pool, status="idle")
            return False
        try:
            await self.process_job(job, worker_id)
        except Exception as exc:
            await self.mark_post_failed(int(job["entity_id"]), str(exc))
            await ack_retry(self.pool, job, str(exc), delay_sec=120)
            logger.exception("Telegram storage job {} failed: {}", job["id"], exc)
        return True

    async def cleanup_tmp_files(self) -> None:
        now = time.time()
        for path in TMP_DIR.glob(f"{TMP_PREFIX}*"):
            try:
                if path.is_file() and path.stat().st_mtime < now - 3600:
                    path.unlink(missing_ok=True)
            except Exception:
                pass

    async def run_loop(self) -> None:
        await self.init()
        worker_id = f"{WORKER_STORAGE_UPLOADER}:{os.getpid()}"
        logger.info("{} started", worker_id)
        try:
            while True:
                await self.cleanup_tmp_files()
                had_job = await self.run_once(worker_id)
                if not had_job:
                    await asyncio.sleep(max(1, await fetch_int_setting("uploader_interval_sec", 5)))
                else:
                    await asyncio.sleep(self.post_job_pause_sec)
        finally:
            await self.close()


async def main() -> None:
    logger.remove()
    logger.add(sys.stdout, level=settings.log_level)
    worker = TelegramStorageUploader()
    await worker.run_loop()


if __name__ == "__main__":
    asyncio.run(main())
