from __future__ import annotations

import argparse
import asyncio
import json
import os
import re
import sys
from datetime import date, datetime, time, timedelta, timezone
from typing import Any
from zoneinfo import ZoneInfo

from aiogram import Bot
from aiogram.exceptions import TelegramRetryAfter
from loguru import logger

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

LOCAL_TZ = ZoneInfo("Asia/Yekaterinburg")
MESSAGE_LIMIT = 4096


def valid_time(value: str) -> bool:
    return bool(re.fullmatch(r"(?:[01]\d|2[0-3]):[0-5]\d", str(value or "").strip()))


def parse_recipients(value: Any) -> list[int]:
    raw_items: list[Any]
    if isinstance(value, list):
        raw_items = value
    elif isinstance(value, str):
        raw = value.strip()
        if not raw:
            return []
        try:
            parsed = json.loads(raw)
            raw_items = parsed if isinstance(parsed, list) else [parsed]
        except json.JSONDecodeError:
            raw_items = re.split(r"[\s,;]+", raw)
    else:
        raw_items = [value]
    out: list[int] = []
    for item in raw_items:
        try:
            recipient_id = int(str(item).strip())
        except Exception:
            continue
        if recipient_id not in out:
            out.append(recipient_id)
    return out


def day_bounds(report_date: date) -> tuple[datetime, datetime]:
    start_local = datetime.combine(report_date, time.min, tzinfo=LOCAL_TZ)
    end_local = start_local + timedelta(days=1)
    return start_local.astimezone(timezone.utc), end_local.astimezone(timezone.utc)


def fmt_int(value: Any) -> str:
    try:
        return f"{int(value):,}".replace(",", " ")
    except Exception:
        return "0"


def split_chunks(text: str, limit: int = MESSAGE_LIMIT) -> 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 = limit
        chunks.append(text[:split_at].strip())
        text = text[split_at:].strip()
    if text:
        chunks.append(text)
    return chunks


class DailyReportWorker:
    def __init__(self) -> None:
        self.pool = None
        self.bot: Bot | None = None
        self.bot_token = ""
        self.recipients: list[int] = []
        self.report_time = "00:04"
        self.interval_sec = 60
        self.heartbeat = HeartbeatReporter(WORKER_DAILY_REPORT, 30)

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

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

    async def reload_settings(self) -> None:
        token = (
            str(await fetch_setting("daily_report_bot_token", "") or "").strip()
            or str(await fetch_setting("tg_poster_bot_token", "") or "").strip()
            or settings.tg_bot_token
        )
        if not token:
            raise RuntimeError("daily_report_bot_token, 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.recipients = parse_recipients(await fetch_setting("daily_report_recipient_ids", [442509142]))
        if not self.recipients:
            raise RuntimeError("daily_report_recipient_ids is empty")
        report_time = str(await fetch_setting("daily_report_time", "00:04") or "00:04").strip()
        self.report_time = report_time if valid_time(report_time) else "00:04"
        self.interval_sec = max(10, await fetch_int_setting("daily_report_interval_sec", 60))
        logger.info("Daily report config: time={} recipients={}", self.report_time, self.recipients)

    async def save_last_sent_date(self, report_date: date) -> None:
        await self.pool.execute(
            """
            UPDATE app_settings
            SET value_json=$2::jsonb,
                updated_at=NOW()
            WHERE key=$1
            """,
            "daily_report_last_sent_date",
            json.dumps(report_date.isoformat()),
        )

    async def countval(self, sql: str, *args: Any) -> int:
        return int(await self.pool.fetchval(sql, *args) or 0)

    async def collect(self, report_date: date) -> dict[str, Any]:
        start_utc, end_utc = day_bounds(report_date)
        now_utc = datetime.now(timezone.utc)
        async with self.pool.acquire() as conn:
            raw_collected = await conn.fetchval(
                "SELECT COUNT(*) FROM raw_posts WHERE created_at >= $1 AND created_at < $2",
                start_utc,
                end_utc,
            )
            storage_ready = await conn.fetchval(
                "SELECT COUNT(*) FROM raw_posts WHERE status='storage_ready' AND updated_at >= $1 AND updated_at < $2",
                start_utc,
                end_utc,
            )
            media_failed = await conn.fetchval(
                "SELECT COUNT(*) FROM raw_post_media WHERE status IN ('failed','link_only') AND updated_at >= $1 AND updated_at < $2",
                start_utc,
                end_utc,
            )
            qualification_rows = await conn.fetch(
                """
                SELECT COALESCE(qualification_status, 'pending') AS status, COUNT(*) AS count
                FROM raw_posts
                WHERE qualified_at >= $1 AND qualified_at < $2
                GROUP BY COALESCE(qualification_status, 'pending')
                """,
                start_utc,
                end_utc,
            )
            writer_ready = await conn.fetchval(
                "SELECT COUNT(*) FROM raw_posts WHERE rewrite_status='ready' AND rewritten_at >= $1 AND rewritten_at < $2",
                start_utc,
                end_utc,
            )
            review_now = await conn.fetchval(
                """
                SELECT COUNT(*)
                FROM raw_posts
                WHERE rewrite_status='ready'
                  AND COALESCE(editorial_status, 'review')='review'
                """
            )
            accepted_today = await conn.fetchval(
                """
                SELECT COUNT(*)
                FROM raw_posts
                WHERE reviewed_at >= $1 AND reviewed_at < $2
                  AND editorial_status='accepted'
                """,
                start_utc,
                end_utc,
            )
            rejected_today = await conn.fetchval(
                """
                SELECT COUNT(*)
                FROM raw_posts
                WHERE reviewed_at >= $1 AND reviewed_at < $2
                  AND editorial_status='rejected'
                """,
                start_utc,
                end_utc,
            )
            publication_rows = await conn.fetch(
                """
                SELECT platform, status, COUNT(*) AS count
                FROM post_publications
                WHERE updated_at >= $1 AND updated_at < $2
                GROUP BY platform, status
                ORDER BY platform, status
                """,
                start_utc,
                end_utc,
            )
            tg_queue = await conn.fetchval(
                """
                SELECT COUNT(*)
                FROM raw_posts rp
                WHERE rp.status='storage_ready'
                  AND rp.rewrite_status='ready'
                  AND rp.editorial_status='accepted'
                  AND COALESCE(rp.publication_status, 'pending')='pending'
                  AND EXISTS (
                    SELECT 1 FROM raw_post_media m
                    WHERE m.raw_post_id=rp.id
                      AND COALESCE(m.editor_hidden,FALSE)=FALSE
                      AND m.media_type IN ('photo','video')
                      AND COALESCE(m.tg_file_id, m.storage_attachment_id, '') <> ''
                  )
                  AND NOT EXISTS (
                    SELECT 1 FROM raw_post_media m
                    WHERE m.raw_post_id=rp.id
                      AND COALESCE(m.editor_hidden,FALSE)=FALSE
                      AND m.media_type IN ('photo','video')
                      AND COALESCE(m.tg_file_id, m.storage_attachment_id, '') = ''
                  )
                """
            )
            vk_queue = await conn.fetchval(
                """
                SELECT COUNT(*)
                FROM raw_posts rp
                LEFT JOIN post_publications pp ON pp.raw_post_id=rp.id AND pp.platform='vk'
                WHERE rp.status='storage_ready'
                  AND rp.rewrite_status='ready'
                  AND COALESCE(rp.editorial_status, 'review') IN ('accepted', 'published', 'publish_failed')
                  AND COALESCE(pp.status, 'pending') IN ('pending', 'publish_failed')
                  AND EXISTS (
                    SELECT 1 FROM raw_post_media rpm
                    WHERE rpm.raw_post_id=rp.id
                      AND COALESCE(rpm.editor_hidden, FALSE)=FALSE
                      AND COALESCE(rpm.editor_added, FALSE)=FALSE
                      AND (
                        (rpm.media_type='photo' AND rpm.original_attachment_id LIKE 'photo%')
                        OR (rpm.media_type='video' AND rpm.original_attachment_id LIKE 'video%')
                      )
                  )
                """
            )
            publish_failed_now = await conn.fetchval(
                """
                SELECT COUNT(*)
                FROM raw_posts
                WHERE rewrite_status='ready'
                  AND COALESCE(editorial_status, '')='publish_failed'
                """
            )
            workers = await conn.fetch(
                """
                SELECT wc.name, wc.enabled, wh.heartbeat_at, wh.status
                FROM worker_controls wc
                LEFT JOIN worker_heartbeats wh ON wh.name=wc.name
                ORDER BY wc.name
                """
            )
            top_errors = await conn.fetch(
                """
                SELECT platform, error, COUNT(*) AS count
                FROM post_publications
                WHERE status='publish_failed'
                  AND updated_at >= $1 AND updated_at < $2
                  AND COALESCE(error, '') <> ''
                GROUP BY platform, error
                ORDER BY COUNT(*) DESC
                LIMIT 5
                """,
                start_utc,
                end_utc,
            )
        return {
            "date": report_date,
            "raw_collected": raw_collected,
            "storage_ready": storage_ready,
            "media_failed": media_failed,
            "qualification": {row["status"]: int(row["count"]) for row in qualification_rows},
            "writer_ready": writer_ready,
            "review_now": review_now,
            "accepted_today": accepted_today,
            "rejected_today": rejected_today,
            "publications": {(row["platform"], row["status"]): int(row["count"]) for row in publication_rows},
            "tg_queue": tg_queue,
            "vk_queue": vk_queue,
            "publish_failed_now": publish_failed_now,
            "workers": [dict(row) for row in workers],
            "top_errors": [dict(row) for row in top_errors],
            "now_utc": now_utc,
        }

    def render(self, data: dict[str, Any]) -> str:
        q = data["qualification"]
        pubs = data["publications"]
        workers_lines: list[str] = []
        for row in data["workers"]:
            heartbeat_at = row.get("heartbeat_at")
            is_stale = True
            if heartbeat_at:
                is_stale = (data["now_utc"] - heartbeat_at).total_seconds() > 600
            marker = "OK" if row.get("enabled") and heartbeat_at and not is_stale else ("OFF" if not row.get("enabled") else "STALE")
            workers_lines.append(f"{marker} {row['name']}: {row.get('status') or '-'}")
        errors_lines = [
            f"- {row['platform']}: {row['count']} x {str(row['error'])[:120]}"
            for row in data["top_errors"]
        ]
        if not errors_lines:
            errors_lines = ["- нет"]
        report_date = data["date"].strftime("%d.%m.%Y")
        return "\n".join(
            [
                f"Ежедневный отчет N8 Parser за {report_date}",
                "",
                "Сбор:",
                f"raw собрано: {fmt_int(data['raw_collected'])}",
                f"storage ready за день: {fmt_int(data['storage_ready'])}",
                f"media failed/link_only: {fmt_int(data['media_failed'])}",
                "",
                "AI:",
                f"принято категоризатором: {fmt_int(q.get('accepted', 0))}",
                f"отклонено категоризатором: {fmt_int(q.get('rejected', 0))}",
                f"ошибки категоризатора: {fmt_int(q.get('failed', 0))}",
                f"рерайтов готово: {fmt_int(data['writer_ready'])}",
                "",
                "Редакторская:",
                f"принято редактором за день: {fmt_int(data['accepted_today'])}",
                f"отклонено редактором за день: {fmt_int(data['rejected_today'])}",
                f"на проверке сейчас: {fmt_int(data['review_now'])}",
                f"ошибка публикации сейчас: {fmt_int(data['publish_failed_now'])}",
                "",
                "Публикации за день:",
                f"TG опубликовано: {fmt_int(pubs.get(('tg', 'published'), 0))}",
                f"VK опубликовано: {fmt_int(pubs.get(('vk', 'published'), 0))}",
                f"TG ошибки: {fmt_int(pubs.get(('tg', 'publish_failed'), 0))}",
                f"VK ошибки: {fmt_int(pubs.get(('vk', 'publish_failed'), 0))}",
                "",
                "Очередь сейчас:",
                f"TG кандидатов: {fmt_int(data['tg_queue'])}",
                f"VK кандидатов: {fmt_int(data['vk_queue'])}",
                "",
                "Ошибки дня:",
                *errors_lines,
                "",
                "Воркеры:",
                *workers_lines,
            ]
        )

    async def send_text(self, text: str) -> None:
        if not self.bot:
            raise RuntimeError("bot is not initialized")
        for recipient_id in self.recipients:
            for chunk in split_chunks(text):
                while True:
                    try:
                        await self.bot.send_message(recipient_id, chunk, disable_web_page_preview=True)
                        break
                    except TelegramRetryAfter as exc:
                        await asyncio.sleep(float(exc.retry_after) + 1)

    async def send_report(self, report_date: date, save_last_sent: bool = False) -> str:
        await self.reload_settings()
        data = await self.collect(report_date)
        text = self.render(data)
        await self.send_text(text)
        if save_last_sent:
            await self.save_last_sent_date(report_date)
        await self.heartbeat.beat(
            self.pool,
            status="sent",
            meta={"date": report_date.isoformat(), "recipients": self.recipients},
            force=True,
        )
        return text

    async def due_report_date(self) -> date | None:
        now = datetime.now(LOCAL_TZ)
        hour, minute = [int(part) for part in self.report_time.split(":", 1)]
        slot_start = datetime.combine(now.date(), time(hour=hour, minute=minute), tzinfo=LOCAL_TZ)
        slot_end = slot_start + timedelta(minutes=10)
        if not (slot_start <= now < slot_end):
            return None
        report_date = now.date() - timedelta(days=1)
        last_sent = str(await fetch_setting("daily_report_last_sent_date", "") or "").strip()
        if last_sent == report_date.isoformat():
            return None
        return report_date

    async def run_once(self) -> bool:
        await self.reload_settings()
        enabled = await is_worker_enabled(self.pool, WORKER_DAILY_REPORT)
        app_enabled = await fetch_bool_setting("daily_report_enabled", True)
        if not enabled or not app_enabled:
            await self.heartbeat.beat(self.pool, status="disabled", force=True)
            return False
        report_date = await self.due_report_date()
        if not report_date:
            await self.heartbeat.beat(self.pool, status="idle")
            return False
        await self.send_report(report_date, save_last_sent=True)
        return True

    async def run_loop(self) -> None:
        await self.init()
        logger.info("{} started pid={}", WORKER_DAILY_REPORT, os.getpid())
        while True:
            try:
                await self.run_once()
            except Exception as exc:
                logger.exception("Daily report loop error: {}", exc)
                if self.pool:
                    await self.heartbeat.beat(self.pool, status="error", meta={"error": str(exc)}, force=True)
            await asyncio.sleep(self.interval_sec)


async def run_send_now(report_date: date | None = None) -> str:
    worker = DailyReportWorker()
    await worker.init()
    try:
        return await worker.send_report(report_date or datetime.now(LOCAL_TZ).date(), save_last_sent=False)
    finally:
        await worker.close()


async def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--send-now", action="store_true")
    parser.add_argument("--date", help="Report date YYYY-MM-DD, default today in EKB for --send-now")
    args = parser.parse_args()
    logger.remove()
    logger.add(sys.stdout, level=settings.log_level)
    if args.send_now:
        report_date = date.fromisoformat(args.date) if args.date else None
        text = await run_send_now(report_date)
        print(text)
        return
    worker = DailyReportWorker()
    await worker.run_loop()


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