from __future__ import annotations

import asyncio
import json
import os
import re
import sys
from typing import Any

import aiohttp
from loguru import logger

from ..config import settings
from ..constants import PUBLICATION_STATUS_PUBLISHED, WORKER_TG_REACTOR
from ..db import fetch_bool_setting, fetch_int_setting, fetch_setting, get_pool
from ..heartbeat import HeartbeatReporter
from ..jobs import is_worker_enabled

PLATFORM_TG = "tg"
REACTION_STATUS_PENDING = "pending"
REACTION_STATUS_IN_PROGRESS = "in_progress"
REACTION_STATUS_REACTED = "reacted"
REACTION_STATUS_FAILED = "failed"


def split_tokens(value: Any) -> list[str]:
    if isinstance(value, list):
        candidates = [str(item or "") for item in value]
    else:
        candidates = re.split(r"[\s,;]+", str(value or ""))
    return [token.strip() for token in candidates if token.strip()]


def normalize_reactions(value: Any) -> list[str]:
    if not isinstance(value, list):
        value = []
    reactions = []
    for item in value:
        emoji = str(item or "").strip()
        if emoji:
            reactions.append(emoji)
    return reactions or ["👍"]


def parse_message_ids(row: dict[str, Any]) -> list[str]:
    raw = row.get("attachments_json")
    values: list[Any] = []
    if isinstance(raw, list):
        values = raw
    elif isinstance(raw, str):
        try:
            parsed = json.loads(raw)
            if isinstance(parsed, list):
                values = parsed
        except json.JSONDecodeError:
            pass
    if not values and row.get("external_id"):
        values = str(row["external_id"]).split(",")
    message_ids = []
    for value in values:
        text = str(value or "").strip()
        if text:
            message_ids.append(text)
    return message_ids


class TelegramReactor:
    def __init__(self) -> None:
        self.pool = None
        self.heartbeat = HeartbeatReporter(WORKER_TG_REACTOR, 30)
        self.bot_tokens: list[str] = []
        self.reactions: list[str] = ["👍"]
        self.delay_sec = 60
        self.interval_sec = 60
        self.reaction_pause_sec = 2
        self.react_all_messages = False
        self.max_attempts = 3
        self.retry_backoff_max_sec = 30
        self.since = ""

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

    async def reload_settings(self) -> None:
        tokens = split_tokens(await fetch_setting("tg_reactor_bot_tokens", ""))
        if not tokens:
            fallback = str(await fetch_setting("tg_poster_bot_token", "") or "").strip() or settings.tg_bot_token
            tokens = split_tokens(fallback)
        if not tokens:
            raise RuntimeError("tg_reactor_bot_tokens, tg_poster_bot_token and TG_BOT_TOKEN are empty")
        self.bot_tokens = tokens
        self.reactions = normalize_reactions(await fetch_setting("tg_reactor_reactions_json", ["👍"]))
        self.since = str(await fetch_setting("tg_reactor_since", "") or "").strip()
        self.delay_sec = max(0, await fetch_int_setting("tg_reactor_delay_sec", 60))
        self.interval_sec = max(10, await fetch_int_setting("tg_reactor_interval_sec", 60))
        self.reaction_pause_sec = max(0, await fetch_int_setting("tg_reactor_reaction_pause_sec", 2))
        self.react_all_messages = await fetch_bool_setting("tg_reactor_react_all_messages", False)
        self.max_attempts = max(1, await fetch_int_setting("tg_reactor_max_attempts", 3))
        self.retry_backoff_max_sec = max(1, await fetch_int_setting("tg_reactor_retry_backoff_max_sec", 30))
        logger.info(
            "TG reactor config: bots={} reactions={} since={} delay={}s pause={}s all_messages={}",
            len(self.bot_tokens),
            self.reactions,
            self.since or "-",
            self.delay_sec,
            self.reaction_pause_sec,
            self.react_all_messages,
        )

    async def load_publications(self, limit: int = 50) -> list[dict[str, Any]]:
        rows = await self.pool.fetch(
            """
            SELECT pp.*
            FROM post_publications pp
            WHERE pp.platform=$1
              AND pp.status=$2
              AND pp.published_at IS NOT NULL
              AND pp.published_at <= NOW() - ($3::int * INTERVAL '1 second')
              AND ($4::text = '' OR pp.published_at >= $4::timestamptz)
            ORDER BY pp.published_at DESC, pp.id DESC
            LIMIT $5::int
            """,
            PLATFORM_TG,
            PUBLICATION_STATUS_PUBLISHED,
            self.delay_sec,
            self.since,
            limit,
        )
        return [dict(row) for row in rows]

    async def ensure_reaction_row(
        self,
        publication: dict[str, Any],
        reactor_key: str,
        reaction: str,
        message_id: str,
    ) -> dict[str, Any] | None:
        row = await self.pool.fetchrow(
            """
            INSERT INTO post_reactions(
              publication_id, raw_post_id, platform, reactor_key, reaction, target_id, message_id, status, updated_at
            )
            VALUES($1, $2, $3, $4, $5, $6, $7, $8, NOW())
            ON CONFLICT (publication_id, reactor_key, message_id) DO NOTHING
            RETURNING *
            """,
            int(publication["id"]),
            int(publication["raw_post_id"]),
            PLATFORM_TG,
            reactor_key,
            reaction,
            str(publication["target_id"]),
            message_id,
            REACTION_STATUS_PENDING,
        )
        if row:
            return dict(row)
        row = await self.pool.fetchrow(
            """
            SELECT *
            FROM post_reactions
            WHERE publication_id=$1 AND reactor_key=$2 AND message_id=$3
            """,
            int(publication["id"]),
            reactor_key,
            message_id,
        )
        if not row or row["status"] == REACTION_STATUS_REACTED or int(row["attempts"] or 0) >= self.max_attempts:
            return None
        return dict(row)

    async def claim_reaction(self, reaction_id: int) -> dict[str, Any] | None:
        row = await self.pool.fetchrow(
            """
            UPDATE post_reactions
            SET status=$2,
                attempts=attempts + 1,
                error=NULL,
                updated_at=NOW()
            WHERE id=$1
              AND status IN ($3, $4)
              AND attempts < $5
            RETURNING *
            """,
            reaction_id,
            REACTION_STATUS_IN_PROGRESS,
            REACTION_STATUS_PENDING,
            REACTION_STATUS_FAILED,
            self.max_attempts,
        )
        return dict(row) if row else None

    async def set_reaction(self, token: str, chat_id: str, message_id: str, reaction: str) -> None:
        url = f"https://api.telegram.org/bot{token}/setMessageReaction"
        payload = {
            "chat_id": int(chat_id),
            "message_id": int(message_id),
            "reaction": [{"type": "emoji", "emoji": reaction}],
            "is_big": False,
        }
        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session:
            async with session.post(url, json=payload) as resp:
                data = await resp.json(content_type=None)
        if not data.get("ok"):
            description = data.get("description") or data
            retry_after = ((data.get("parameters") or {}).get("retry_after")) if isinstance(data, dict) else None
            if retry_after:
                raise RuntimeError(f"Telegram retry_after={retry_after}: {description}")
            raise RuntimeError(str(description))

    async def mark_reacted(self, reaction_id: int) -> None:
        await self.pool.execute(
            """
            UPDATE post_reactions
            SET status=$2,
                error=NULL,
                reacted_at=NOW(),
                updated_at=NOW()
            WHERE id=$1
            """,
            reaction_id,
            REACTION_STATUS_REACTED,
        )

    async def mark_failed(self, reaction_id: int, error: str) -> None:
        await self.pool.execute(
            """
            UPDATE post_reactions
            SET status=$2,
                error=$3,
                updated_at=NOW()
            WHERE id=$1
            """,
            reaction_id,
            REACTION_STATUS_FAILED,
            error[:1000],
        )

    async def process_publication(self, publication: dict[str, Any]) -> int:
        message_ids = parse_message_ids(publication)
        if not message_ids:
            return 0
        if not self.react_all_messages:
            message_ids = message_ids[:1]

        done = 0
        for message_id in message_ids:
            for idx, token in enumerate(self.bot_tokens):
                reaction = self.reactions[idx % len(self.reactions)]
                reactor_key = f"bot-{idx + 1}:{reaction}"
                row = await self.ensure_reaction_row(publication, reactor_key, reaction, message_id)
                if not row:
                    continue
                claimed = await self.claim_reaction(int(row["id"]))
                if not claimed:
                    continue
                try:
                    await self.set_reaction(token, str(publication["target_id"]), message_id, reaction)
                    await self.mark_reacted(int(claimed["id"]))
                    done += 1
                    logger.info(
                        "TG reacted raw_post={} publication={} message={} reactor={}",
                        publication["raw_post_id"],
                        publication["id"],
                        message_id,
                        reactor_key,
                    )
                    if self.reaction_pause_sec:
                        await asyncio.sleep(self.reaction_pause_sec)
                except Exception as exc:
                    await self.mark_failed(int(claimed["id"]), str(exc))
                    logger.warning(
                        "TG reaction failed raw_post={} publication={} message={} reactor={}: {}",
                        publication["raw_post_id"],
                        publication["id"],
                        message_id,
                        reactor_key,
                        exc,
                    )
                    await asyncio.sleep(min(2 ** int(claimed["attempts"]), self.retry_backoff_max_sec))
        return done

    async def run_once(self) -> bool:
        await self.reload_settings()
        enabled = await is_worker_enabled(self.pool, WORKER_TG_REACTOR)
        app_enabled = await fetch_bool_setting("tg_reactor_enabled", False)
        if not enabled or not app_enabled:
            await self.heartbeat.beat(self.pool, status="disabled", force=True)
            return False
        publications = await self.load_publications()
        if not publications:
            await self.heartbeat.beat(self.pool, status="idle")
            return False
        total = 0
        for publication in publications:
            await self.heartbeat.beat(
                self.pool,
                status="reacting",
                meta={"publication_id": int(publication["id"]), "raw_post_id": int(publication["raw_post_id"])},
                force=True,
            )
            total += await self.process_publication(publication)
        await self.heartbeat.beat(self.pool, status="idle", meta={"reactions": total}, force=True)
        return bool(total)

    async def run_loop(self) -> None:
        await self.init()
        logger.info("{} started pid={}", WORKER_TG_REACTOR, os.getpid())
        while True:
            try:
                await self.run_once()
            except Exception as exc:
                logger.exception("TG reactor loop error: {}", exc)
            await asyncio.sleep(self.interval_sec)


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


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