from __future__ import annotations

import asyncio
import aiohttp
import json
import os
import re
import sys
from datetime import datetime, time, timedelta
from pathlib import Path
from typing import Any
from zoneinfo import ZoneInfo

from aiogram import Bot
from aiogram.exceptions import TelegramRetryAfter
from aiogram.types import FSInputFile, InputMediaPhoto, InputMediaVideo
from loguru import logger

from ..config import settings
from ..constants import (
    PUBLICATION_STATUS_FAILED,
    PUBLICATION_STATUS_PENDING,
    PUBLICATION_STATUS_PUBLISHED,
    WORKER_TG_POSTER,
)
from ..db import fetch_bool_setting, fetch_float_setting, fetch_int_setting, fetch_setting, get_pool
from ..heartbeat import HeartbeatReporter
from ..jobs import is_worker_enabled
from ..text_utils import build_publication_text

LOCAL_TZ = ZoneInfo("Asia/Yekaterinburg")
MAX_MEDIA_GROUP = 10
MAX_RICH_MEDIA = 50
MAX_RICH_TEXT = 32768
PROJECT_ROOT = Path(__file__).resolve().parents[3]
UPLOAD_ROOT = PROJECT_ROOT / "uploads"


def parse_topic(value: str) -> tuple[int, int | None]:
    raw = str(value or "").strip()
    if not raw:
        raise RuntimeError("tg_poster_chat_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 valid_time(value: str) -> bool:
    return bool(re.fullmatch(r"(?:[01]\d|2[0-3]):[0-5]\d", str(value or "").strip()))


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 tg_message_url(chat_id: int, message_id: int) -> str:
    chat = str(abs(int(chat_id)))
    if chat.startswith("100"):
        chat = chat[3:]
    return f"https://t.me/c/{chat}/{message_id}"


def local_upload_path(url: str) -> Path | None:
    raw = str(url or "").strip()
    prefix = "/uploads/"
    if not raw.startswith(prefix):
        return None
    path = (UPLOAD_ROOT / raw.removeprefix(prefix)).resolve()
    try:
        path.relative_to(UPLOAD_ROOT.resolve())
    except ValueError:
        return None
    return path if path.is_file() else None


class RichMessageUnavailable(RuntimeError):
    pass


class TelegramPoster:
    def __init__(self) -> None:
        self.pool = None
        self.bot: Bot | None = None
        self.bot_token = ""
        self.chat_id: int | None = None
        self.thread_id: int | None = None
        self.heartbeat = HeartbeatReporter(WORKER_TG_POSTER, 30)
        self.schedule: list[dict[str, Any]] = []
        self.media_group_max_items = MAX_MEDIA_GROUP
        self.caption_limit = 1024
        self.message_limit = 4096
        self.max_attempts = 3
        self.retry_backoff_max_sec = 30
        self.send_delay_sec = 1.0
        self.overflow_caption = "⬇️ Описание"
        self.recent_window = 20
        self.category_repeat_penalty = 3.0
        self.source_repeat_penalty = 4.0

    async def init(self) -> None:
        self.pool = await get_pool()
        await self.reload_settings()

    async def reload_settings(self) -> None:
        token = str(await fetch_setting("tg_poster_bot_token", "") or "").strip() or settings.tg_bot_token
        if not token:
            raise RuntimeError("tg_poster_bot_token and TG_BOT_TOKEN are empty")
        if token != self.bot_token:
            if self.bot:
                await self.bot.session.close()
            self.bot = Bot(token=token)
            self.bot_token = token
        self.chat_id, self.thread_id = parse_topic(str(await fetch_setting("tg_poster_chat_id", "-1003712724327") or ""))
        self.schedule = self.normalize_schedule(await fetch_setting("tg_poster_schedule_json", []))
        self.media_group_max_items = max(1, min(MAX_MEDIA_GROUP, await fetch_int_setting("tg_poster_media_group_max_items", 10)))
        self.caption_limit = max(128, min(1024, await fetch_int_setting("tg_poster_caption_limit", 1024)))
        self.message_limit = max(512, min(4096, await fetch_int_setting("tg_poster_message_limit", 4096)))
        self.max_attempts = max(1, await fetch_int_setting("tg_poster_max_attempts", 3))
        self.retry_backoff_max_sec = max(1, await fetch_int_setting("tg_poster_retry_backoff_max_sec", 30))
        self.send_delay_sec = max(0.0, await fetch_float_setting("tg_poster_send_delay_sec", 1.0))
        self.overflow_caption = str(await fetch_setting("tg_poster_text_overflow_caption", self.overflow_caption) or self.overflow_caption)
        self.recent_window = max(1, await fetch_int_setting("tg_poster_recent_window", 20))
        self.category_repeat_penalty = max(0.0, await fetch_float_setting("tg_poster_category_repeat_penalty", 3.0))
        self.source_repeat_penalty = max(0.0, await fetch_float_setting("tg_poster_source_repeat_penalty", 4.0))
        logger.info("TG poster config: chat={} schedule={} caption_limit={}", self.chat_id, self.schedule, self.caption_limit)

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

    def normalize_schedule(self, value: Any) -> list[dict[str, Any]]:
        if not isinstance(value, list):
            value = []
        rows: list[dict[str, Any]] = []
        for idx, item in enumerate(value):
            if not isinstance(item, dict):
                continue
            time_value = str(item.get("time") or "").strip()
            if not valid_time(time_value):
                continue
            count = max(0, min(20, int(item.get("count") or 0)))
            if count < 1:
                continue
            rows.append(
                {
                    "id": str(item.get("id") or f"slot-{idx + 1}").strip()[:80],
                    "time": time_value,
                    "count": count,
                    "enabled": bool(item.get("enabled", True)),
                }
            )
        return sorted(rows, key=lambda row: row["time"])

    def chat_kwargs(self) -> dict[str, Any]:
        kwargs: dict[str, Any] = {"chat_id": int(self.chat_id), "parse_mode": "HTML"}
        if self.thread_id:
            kwargs["message_thread_id"] = int(self.thread_id)
        return kwargs

    async def tg_retry(self, fn):
        for attempt in range(1, self.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:
                logger.warning("Telegram post error without retry to avoid duplicate send: {}", exc)
                raise
        raise RuntimeError("telegram retry exhausted")

    async def due_slots(self) -> list[dict[str, Any]]:
        now = datetime.now(LOCAL_TZ)
        today = now.date()
        due: list[dict[str, Any]] = []
        for row in self.schedule:
            if not row.get("enabled"):
                continue
            hour, minute = [int(part) for part in str(row["time"]).split(":", 1)]
            slot_start = datetime.combine(today, time(hour=hour, minute=minute), tzinfo=LOCAL_TZ)
            slot_end = slot_start + timedelta(minutes=10)
            if not (slot_start <= now < slot_end):
                continue
            inserted = await self.pool.fetchrow(
                """
                INSERT INTO publication_runs(poster, schedule_id, scheduled_for, scheduled_time, planned_count, status)
                VALUES($1, $2, $3, $4, $5, 'started')
                ON CONFLICT (poster, schedule_id, scheduled_for) DO NOTHING
                RETURNING id
                """,
                "tg",
                row["id"],
                today,
                row["time"],
                int(row["count"]),
            )
            if inserted:
                due.append({**row, "run_id": int(inserted["id"]), "date": today})
        return due

    async def load_recent(self) -> list[dict[str, Any]]:
        rows = await self.pool.fetch(
            """
            SELECT final_category_tag, rewrite_category_tag, final_source_tag, rewrite_source_tag
            FROM raw_posts
            WHERE publication_status=$1
              AND tg_publication_chat_id=$2
            ORDER BY published_at DESC NULLS LAST, id DESC
            LIMIT $3::int
            """,
            PUBLICATION_STATUS_PUBLISHED,
            self.chat_id,
            self.recent_window,
        )
        return [dict(row) for row in rows]

    async def load_candidates(self, limit: int) -> list[dict[str, Any]]:
        rows = await self.pool.fetch(
            """
            SELECT rp.*, s.name AS source_name, s.tag AS source_tag
            FROM raw_posts rp
            JOIN sources s ON s.id=rp.source_id
            WHERE rp.status='storage_ready'
              AND rp.rewrite_status='ready'
              AND rp.editorial_status='accepted'
              AND COALESCE(rp.publication_status, 'pending') = $1
              AND EXISTS (
                SELECT 1
                FROM raw_post_media rpm
                WHERE rpm.raw_post_id=rp.id
                  AND COALESCE(rpm.editor_hidden, FALSE)=FALSE
                  AND rpm.media_type IN ('photo', 'video')
                  AND COALESCE(rpm.tg_file_id, rpm.storage_attachment_id, '') <> ''
              )
              AND NOT EXISTS (
                SELECT 1
                FROM raw_post_media rpm
                WHERE rpm.raw_post_id=rp.id
                  AND COALESCE(rpm.editor_hidden, FALSE)=FALSE
                  AND rpm.media_type IN ('photo', 'video')
                  AND COALESCE(rpm.tg_file_id, rpm.storage_attachment_id, '') = ''
              )
            ORDER BY COALESCE(rp.reviewed_at, rp.edited_at, rp.rewritten_at, rp.created_at) ASC,
                     rp.id ASC
            LIMIT $2::int
            """,
            PUBLICATION_STATUS_PENDING,
            max(limit, 50),
        )
        return [dict(row) for row in rows]

    def choose_posts(self, candidates: list[dict[str, Any]], recent: list[dict[str, Any]], count: int) -> list[dict[str, Any]]:
        selected: list[dict[str, Any]] = []
        recent_categories = [
            str(row.get("final_category_tag") or row.get("rewrite_category_tag") or "")
            for row in recent
        ]
        recent_sources = [
            str(row.get("final_source_tag") or row.get("rewrite_source_tag") or "")
            for row in recent
        ]
        pool = list(candidates)
        while pool and len(selected) < count:
            best_idx = 0
            selected_categories = [str(row.get("final_category_tag") or row.get("rewrite_category_tag") or "") for row in selected]
            selected_sources = [str(row.get("final_source_tag") or row.get("rewrite_source_tag") or row.get("source_tag") or "") for row in selected]
            previous_category = selected_categories[-1] if selected_categories else (recent_categories[0] if recent_categories else "")
            previous_source = selected_sources[-1] if selected_sources else (recent_sources[0] if recent_sources else "")
            best_rank: tuple[int, float, int] | None = None
            for idx, post in enumerate(pool):
                category = str(post.get("final_category_tag") or post.get("rewrite_category_tag") or "")
                source = str(post.get("final_source_tag") or post.get("rewrite_source_tag") or post.get("source_tag") or "")
                immediate_repeats = int(bool(category) and category == previous_category) + int(bool(source) and source == previous_source)
                repeat_penalty = (
                    (recent_categories.count(category) + selected_categories.count(category)) * self.category_repeat_penalty
                    + (recent_sources.count(source) + selected_sources.count(source)) * self.source_repeat_penalty
                )
                rank = (immediate_repeats, repeat_penalty, idx)
                if best_rank is None or rank < best_rank:
                    best_idx = idx
                    best_rank = rank
            selected.append(pool.pop(best_idx))
        return selected

    async def load_media(self, raw_post_id: int) -> list[dict[str, Any]]:
        rows = await self.pool.fetch(
            """
            SELECT *
            FROM raw_post_media
            WHERE raw_post_id=$1
              AND COALESCE(editor_hidden, FALSE)=FALSE
              AND media_type IN ('photo', 'video')
            ORDER BY sort_order ASC, id ASC
            """,
            raw_post_id,
        )
        return [dict(row) for row in rows]

    def media_value(self, item: dict[str, Any]) -> Any | None:
        stored = item.get("tg_file_id") or item.get("storage_attachment_id")
        if stored:
            return str(stored)
        url = str(item.get("original_url") or "").strip()
        path = local_upload_path(url)
        if path:
            return FSInputFile(path)
        return None

    def build_text(self, post: dict[str, Any]) -> str:
        category_tag = str(post.get("final_category_tag") or post.get("rewrite_category_tag") or post.get("final_category") or post.get("rewrite_category") or "")
        source_tag = str(post.get("final_source_tag") or post.get("rewrite_source_tag") or post.get("source_tag") or "")
        return build_publication_text(post.get("final_text") or post.get("rewritten_text") or "", category_tag, source_tag, format_title=True, parse_mode="html")

    async def send_text(self, text: str) -> list[int]:
        message_ids: list[int] = []
        for chunk in split_message_chunks(text, self.message_limit):
            message = await self.tg_retry(lambda text=chunk: self.bot.send_message(text=text, **self.chat_kwargs()))
            message_ids.append(int(message.message_id))
            if self.send_delay_sec:
                await asyncio.sleep(self.send_delay_sec)
        return message_ids

    async def send_media_items(self, items: list[dict[str, Any]], caption: str | None = None) -> list[int]:
        if not items:
            return []
        if len(items) == 1:
            item = items[0]
            if item["type"] == "photo":
                message = await self.tg_retry(
                    lambda: self.bot.send_photo(photo=item["media"], caption=caption, **self.chat_kwargs())
                )
            else:
                message = await self.tg_retry(
                    lambda: self.bot.send_video(video=item["media"], caption=caption, **self.chat_kwargs())
                )
            return [int(message.message_id)]
        group = []
        for idx, item in enumerate(items):
            media_caption = caption if idx == 0 else None
            if item["type"] == "photo":
                group.append(InputMediaPhoto(media=item["media"], caption=media_caption))
            else:
                group.append(InputMediaVideo(media=item["media"], caption=media_caption))
        messages = await self.tg_retry(lambda: self.bot.send_media_group(media=group, **self.chat_kwargs()))
        return [int(message.message_id) for message in messages]

    def rich_media_value(self, item: dict[str, Any]) -> str | None:
        stored = str(item.get("tg_file_id") or item.get("storage_attachment_id") or "").strip()
        return stored or None

    def rich_text_html(self, text: str) -> str:
        paragraphs = []
        for paragraph in str(text or "").strip().split("\n\n"):
            body = "<br/>".join(line for line in paragraph.splitlines() if line.strip())
            if body:
                paragraphs.append(f"<p>{body}</p>")
        return "\n".join(paragraphs)

    def build_rich_message(self, text: str, media_rows: list[dict[str, Any]]) -> dict[str, Any] | None:
        if len(media_rows) > MAX_RICH_MEDIA:
            return None

        rich_media = []
        media_tags = []
        for idx, item in enumerate(media_rows[:MAX_RICH_MEDIA]):
            media_type = str(item.get("media_type") or "")
            if media_type not in ("photo", "video"):
                continue
            media = self.rich_media_value(item)
            if not media:
                return None
            media_id = f"m{idx}"
            input_type = "photo" if media_type == "photo" else "video"
            rich_media.append({"id": media_id, "media": {"type": input_type, "media": media}})
            if media_type == "photo":
                media_tags.append(f'<img src="tg://photo?id={media_id}"/>')
            else:
                media_tags.append(f'<video src="tg://video?id={media_id}"></video>')
        if not rich_media:
            return None

        rich_text = self.rich_text_html(text)
        if len(rich_text) > MAX_RICH_TEXT:
            return None

        media_html = media_tags[0] if len(media_tags) == 1 else f"<tg-collage>{''.join(media_tags)}</tg-collage>"
        return {
            "html": f"{media_html}\n{rich_text}" if rich_text else media_html,
            "media": rich_media,
        }

    async def send_rich_message(self, rich_message: dict[str, Any]) -> list[int]:
        data: dict[str, Any] = {
            "chat_id": int(self.chat_id),
            "rich_message": rich_message,
        }
        if self.thread_id:
            data["message_thread_id"] = int(self.thread_id)

        url = f"https://api.telegram.org/bot{self.bot_token}/sendRichMessage"
        timeout = aiohttp.ClientTimeout(total=90)
        async with aiohttp.ClientSession(timeout=timeout) as session:
            async with session.post(url, json=data) as response:
                payload = await response.json(content_type=None)

        if payload.get("ok"):
            message_id = payload.get("result", {}).get("message_id")
            if message_id:
                return [int(message_id)]
            raise RichMessageUnavailable("sendRichMessage returned no message_id")

        description = str(payload.get("description") or f"HTTP {response.status}")
        if "Too Many Requests" in description and isinstance(payload.get("parameters"), dict):
            retry_after = float(payload["parameters"].get("retry_after") or 0)
            if retry_after > 0:
                logger.warning("Telegram rich message flood control, sleep {}s", retry_after)
                await asyncio.sleep(retry_after + 0.5)
        raise RichMessageUnavailable(description)

    async def send_legacy_media_post(self, text: str, media_rows: list[dict[str, Any]]) -> list[int]:
        media_items = []
        for item in media_rows:
            media = self.media_value(item)
            if media is None:
                continue
            media_type = str(item.get("media_type") or "")
            if media_type == "photo":
                media_items.append({"type": "photo", "media": media})
            elif media_type == "video":
                media_items.append({"type": "video", "media": media})

        if not media_items:
            return await self.send_text(text)

        if len(text) <= self.caption_limit:
            first_caption = text
            text_chunks: list[str] = []
        else:
            first_caption = self.overflow_caption[: self.caption_limit]
            text_chunks = split_message_chunks(text, self.message_limit)

        first = media_items[: self.media_group_max_items]
        message_ids = []
        message_ids.extend(await self.send_media_items(first, first_caption))
        if self.send_delay_sec:
            await asyncio.sleep(self.send_delay_sec)

        rest = media_items[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]
            message_ids.extend(await self.send_media_items(chunk))
            if self.send_delay_sec:
                await asyncio.sleep(self.send_delay_sec)

        for chunk in text_chunks:
            message_ids.extend(await self.send_text(chunk))

        return message_ids

    async def send_post(self, post: dict[str, Any]) -> list[int]:
        text = self.build_text(post)
        media_rows = await self.load_media(int(post["id"]))

        if not media_rows:
            return await self.send_text(text)

        rich_message = self.build_rich_message(text, media_rows)
        if rich_message:
            try:
                return await self.send_rich_message(rich_message)
            except RichMessageUnavailable as exc:
                logger.warning("Telegram rich message failed, fallback to legacy media post: {}", exc)

        return await self.send_legacy_media_post(text, media_rows)

    async def mark_published(self, post_id: int, message_ids: list[int]) -> None:
        async with self.pool.acquire() as conn:
            async with conn.transaction():
                await conn.execute(
                    """
                    UPDATE raw_posts
                    SET publication_status=$2,
                        editorial_status='published',
                        publication_error=NULL,
                        published_at=NOW(),
                        tg_publication_chat_id=$3,
                        tg_publication_thread_id=$4,
                        tg_publication_message_ids=$5,
                        tg_publication_url=$6,
                        updated_at=NOW()
                    WHERE id=$1
                    """,
                    post_id,
                    PUBLICATION_STATUS_PUBLISHED,
                    self.chat_id,
                    self.thread_id,
                    message_ids,
                    tg_message_url(int(self.chat_id), message_ids[0]) if message_ids else None,
                )
                await conn.execute(
                    """
                    INSERT INTO post_publications(raw_post_id, platform, status, target_id, external_id, url, attachments_json, published_at, updated_at)
                    VALUES($1, 'tg', $2, $3, $4, $5, $6::jsonb, NOW(), NOW())
                    ON CONFLICT (raw_post_id, platform) DO UPDATE
                    SET status=EXCLUDED.status,
                        target_id=EXCLUDED.target_id,
                        external_id=EXCLUDED.external_id,
                        url=EXCLUDED.url,
                        error=NULL,
                        attachments_json=EXCLUDED.attachments_json,
                        published_at=NOW(),
                        updated_at=NOW()
                    """,
                    post_id,
                    PUBLICATION_STATUS_PUBLISHED,
                    str(self.chat_id),
                    ",".join(str(mid) for mid in message_ids),
                    tg_message_url(int(self.chat_id), message_ids[0]) if message_ids else None,
                    json.dumps(message_ids, ensure_ascii=False),
                )

    async def mark_failed(self, post_id: int, error: str) -> None:
        async with self.pool.acquire() as conn:
            async with conn.transaction():
                await conn.execute(
                    """
                    UPDATE raw_posts
                    SET publication_status=$2,
                        editorial_status='publish_failed',
                        publication_error=$3,
                        updated_at=NOW()
                    WHERE id=$1
                    """,
                    post_id,
                    PUBLICATION_STATUS_FAILED,
                    error[:1000],
                )
                await conn.execute(
                    """
                    INSERT INTO post_publications(raw_post_id, platform, status, target_id, error, updated_at)
                    VALUES($1, 'tg', $2, $3, $4, NOW())
                    ON CONFLICT (raw_post_id, platform) DO UPDATE
                    SET status=EXCLUDED.status,
                        target_id=EXCLUDED.target_id,
                        error=EXCLUDED.error,
                        updated_at=NOW()
                    """,
                    post_id,
                    PUBLICATION_STATUS_FAILED,
                    str(self.chat_id),
                    error[:1000],
                )
    async def process_slot(self, slot: dict[str, Any]) -> int:
        count = int(slot["count"])
        recent = await self.load_recent()
        candidates = await self.load_candidates(max(50, count * 10))
        posts = self.choose_posts(candidates, recent, count)
        published = 0
        for post in posts:
            await self.heartbeat.beat(self.pool, status="publishing", meta={"post_id": int(post["id"]), "slot": slot["id"]}, force=True)
            try:
                message_ids = await self.send_post(post)
                await self.mark_published(int(post["id"]), message_ids)
                published += 1
                logger.info("TG published raw_post={} messages={}", post["id"], message_ids)
            except Exception as exc:
                await self.mark_failed(int(post["id"]), str(exc))
                logger.exception("TG publish failed raw_post={}: {}", post["id"], exc)
        await self.pool.execute(
            """
            UPDATE publication_runs
            SET published_count=$2,
                status=$3,
                error=$4,
                updated_at=NOW()
            WHERE id=$1
            """,
            int(slot["run_id"]),
            published,
            "done" if published == count else "partial",
            None if published == count else f"published {published} of {count}",
        )
        return published

    async def run_once(self) -> bool:
        await self.reload_settings()
        enabled = await is_worker_enabled(self.pool, WORKER_TG_POSTER)
        app_enabled = await fetch_bool_setting("tg_poster_enabled", False)
        if not enabled or not app_enabled:
            await self.heartbeat.beat(self.pool, status="disabled", force=True)
            return False
        slots = await self.due_slots()
        if not slots:
            await self.heartbeat.beat(self.pool, status="idle")
            return False
        total = 0
        for slot in slots:
            total += await self.process_slot(slot)
        await self.heartbeat.beat(self.pool, status="idle", meta={"published": total}, force=True)
        return bool(total)

    async def run_loop(self) -> None:
        await self.init()
        logger.info("{} started pid={}", WORKER_TG_POSTER, os.getpid())
        try:
            while True:
                await self.run_once()
                await asyncio.sleep(max(10, await fetch_int_setting("tg_poster_interval_sec", 60)))
        finally:
            await self.close()


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


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