from __future__ import annotations

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

import aiohttp
from loguru import logger

from ..config import settings
from ..constants import (
    PUBLICATION_STATUS_FAILED,
    PUBLICATION_STATUS_PENDING,
    PUBLICATION_STATUS_PUBLISHED,
    WORKER_VK_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
from ..vk_api import VKAPIClient, post_vk_url

LOCAL_TZ = ZoneInfo("Asia/Yekaterinburg")
VK_MESSAGE_LIMIT = 16384
VK_MAX_ATTACHMENTS = 10


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


def normalize_schedule(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"])


class VKPoster:
    def __init__(self) -> None:
        self.pool = None
        self.heartbeat = HeartbeatReporter(WORKER_VK_POSTER, 30)
        self.token = ""
        self.owner_id = 0
        self.from_group = True
        self.schedule: list[dict[str, Any]] = []
        self.interval_sec = 60
        self.send_delay_sec = 1.0
        self.recent_window = 20
        self.category_repeat_penalty = 3.0
        self.source_repeat_penalty = 4.0
        self.dry_run = False

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

    def token_expires_soon(self, value: Any) -> bool:
        raw = str(value or "").strip()
        if not raw:
            return False
        try:
            expires_at = datetime.fromisoformat(raw)
            if expires_at.tzinfo is None:
                expires_at = expires_at.replace(tzinfo=timezone.utc)
        except ValueError:
            return False
        return expires_at <= datetime.now(timezone.utc) + timedelta(minutes=15)

    async def save_setting(self, key: str, value: Any) -> None:
        await self.pool.execute(
            """
            UPDATE app_settings
            SET value_json=$2::jsonb,
                updated_at=NOW()
            WHERE key=$1
            """,
            key,
            json.dumps(value, ensure_ascii=False),
        )

    async def refresh_access_token_if_needed(self) -> None:
        expires_at = await fetch_setting("vk_poster_token_expires_at", "")
        if not self.token_expires_soon(expires_at):
            return
        refresh_token = str(await fetch_setting("vk_poster_refresh_token", "") or "").strip()
        client_id = str(await fetch_setting("vk_poster_app_id", "") or "").strip()
        client_secret = str(await fetch_setting("vk_poster_client_secret", "") or "").strip()
        device_id = str(await fetch_setting("vk_poster_token_device_id", "") or "").strip()
        if not refresh_token or not client_id:
            return
        data = {
            "grant_type": "refresh_token",
            "refresh_token": refresh_token,
            "client_id": client_id,
            "device_id": device_id,
            "state": "vk-poster-refresh",
        }
        if client_secret:
            data["client_secret"] = client_secret
        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session:
            async with session.post("https://id.vk.ru/oauth2/auth", data=data) as resp:
                payload = await resp.json(content_type=None)
        if payload.get("error"):
            raise RuntimeError(f"VK token refresh failed: {payload.get('error')}: {payload.get('error_description')}")
        access_token = str(payload.get("access_token") or "").strip()
        if not access_token:
            raise RuntimeError(f"VK token refresh did not return access_token: {payload}")
        new_refresh = str(payload.get("refresh_token") or "").strip() or refresh_token
        new_device_id = str(payload.get("device_id") or "").strip() or device_id
        expires_in = payload.get("expires_in")
        expires_at_value = ""
        try:
            expires_at_value = (datetime.now(timezone.utc) + timedelta(seconds=int(expires_in))).isoformat()
        except Exception:
            pass
        await self.save_setting("vk_poster_access_token", access_token)
        await self.save_setting("vk_poster_refresh_token", new_refresh)
        await self.save_setting("vk_poster_token_device_id", new_device_id)
        await self.save_setting("vk_poster_token_expires_at", expires_at_value)
        self.token = access_token
        logger.info("VK access token refreshed; expires_at={}", expires_at_value)

    async def reload_settings(self) -> None:
        self.token = (
            str(await fetch_setting("vk_poster_access_token", "") or "").strip()
            or settings.vk_group_access_token
            or settings.vk_access_token
        )
        await self.refresh_access_token_if_needed()
        self.owner_id = int(await fetch_int_setting("vk_poster_owner_id", settings.vk_storage_owner_id if settings.vk_storage_group_id else 0))
        self.from_group = await fetch_bool_setting("vk_poster_from_group", True)
        self.schedule = normalize_schedule(await fetch_setting("vk_poster_schedule_json", []))
        self.interval_sec = max(10, await fetch_int_setting("vk_poster_interval_sec", 60))
        self.send_delay_sec = max(0.0, await fetch_float_setting("vk_poster_send_delay_sec", 1.0))
        self.recent_window = max(1, await fetch_int_setting("vk_poster_recent_window", 20))
        self.category_repeat_penalty = max(0.0, await fetch_float_setting("vk_poster_category_repeat_penalty", 3.0))
        self.source_repeat_penalty = max(0.0, await fetch_float_setting("vk_poster_source_repeat_penalty", 4.0))
        self.dry_run = await fetch_bool_setting("vk_poster_dry_run", False)
        if not self.token:
            raise RuntimeError("vk_poster_access_token, VK_GROUP_ACCESS_TOKEN and VK_ACCESS_TOKEN are empty")
        if not self.owner_id:
            raise RuntimeError("vk_poster_owner_id is empty")
        logger.info("VK poster config: owner={} schedule={} dry_run={}", self.owner_id, self.schedule, self.dry_run)

    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
                """,
                "vk",
                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 rp.final_category_tag, rp.rewrite_category_tag, rp.final_source_tag, rp.rewrite_source_tag
            FROM post_publications pp
            JOIN raw_posts rp ON rp.id=pp.raw_post_id
            WHERE pp.platform='vk'
              AND pp.status=$1
              AND pp.target_id=$2
            ORDER BY pp.published_at DESC NULLS LAST, pp.id DESC
            LIMIT $3::int
            """,
            PUBLICATION_STATUS_PUBLISHED,
            str(self.owner_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
            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, $1) = $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 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%'
                      AND COALESCE(rpm.status, '') <> 'link_only'
                    )
                  )
              )
            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
            LIMIT $2::int
            """,
            raw_post_id,
            VK_MAX_ATTACHMENTS,
        )
        return [dict(row) for row in rows]

    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)[:VK_MESSAGE_LIMIT]

    async def media_attachment(self, client: VKAPIClient, item: dict[str, Any]) -> str | None:
        attachment_id = str(item.get("original_attachment_id") or "").strip()
        media_type = str(item.get("media_type") or "").strip()
        if media_type == "video":
            if str(item.get("status") or "").strip() == "link_only":
                return None
            return attachment_id if attachment_id.startswith("video") else None
        if media_type != "photo":
            return None
        if attachment_id.startswith("photo") and not item.get("editor_added"):
            return attachment_id
        return None

    async def send_post(self, post: dict[str, Any]) -> tuple[int, str, list[str]]:
        text = self.build_text(post)
        media_rows = await self.load_media(int(post["id"]))
        async with VKAPIClient(
            token=self.token,
            rps=2,
            timeout_total_sec=max(30, await fetch_int_setting("vk_poster_timeout_sec", 90)),
            retry_attempts=max(1, await fetch_int_setting("vk_poster_max_attempts", 3)),
            retry_min_delay_sec=2,
            retry_max_delay_sec=max(2, await fetch_int_setting("vk_poster_retry_backoff_max_sec", 30)),
        ) as client:
            attachments: list[str] = []
            for item in media_rows:
                try:
                    attachment = await self.media_attachment(client, item)
                except Exception as exc:
                    logger.warning("VK media skipped raw_post={} media_id={}: {}", post["id"], item.get("id"), exc)
                    continue
                if attachment:
                    attachments.append(attachment)
                if len(attachments) >= VK_MAX_ATTACHMENTS:
                    break
            if self.dry_run:
                return 0, "", attachments
            post_id = await client.create_wall_post(self.owner_id, text, attachments, from_group=self.from_group)
            return post_id, post_vk_url(self.owner_id, post_id), attachments

    async def upsert_publication(
        self,
        post_id: int,
        status: str,
        external_id: str | None = None,
        url: str | None = None,
        error: str | None = None,
        attachments: list[str] | None = None,
    ) -> None:
        await self.pool.execute(
            """
            INSERT INTO post_publications(
              raw_post_id, platform, status, target_id, external_id, url, error, attachments_json, published_at, updated_at
            )
            VALUES($1, 'vk', $2, $3, $4, $5, $6, $7::jsonb, CASE WHEN $2=$8 THEN NOW() ELSE NULL END, 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=EXCLUDED.error,
                attachments_json=EXCLUDED.attachments_json,
                published_at=CASE WHEN EXCLUDED.status=$8 THEN COALESCE(EXCLUDED.published_at, NOW()) ELSE post_publications.published_at END,
                updated_at=NOW()
            """,
            post_id,
            status,
            str(self.owner_id),
            external_id,
            url,
            error[:1000] if error else None,
            json.dumps(attachments or [], ensure_ascii=False),
            PUBLICATION_STATUS_PUBLISHED,
        )

    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:
                vk_post_id, url, attachments = await self.send_post(post)
                await self.upsert_publication(int(post["id"]), PUBLICATION_STATUS_PUBLISHED, str(vk_post_id), url, attachments=attachments)
                published += 1
                logger.info("VK published raw_post={} post_id={} attachments={}", post["id"], vk_post_id, len(attachments))
            except Exception as exc:
                await self.upsert_publication(int(post["id"]), PUBLICATION_STATUS_FAILED, error=str(exc))
                logger.exception("VK publish failed raw_post={}: {}", post["id"], exc)
            if self.send_delay_sec:
                await asyncio.sleep(self.send_delay_sec)
        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_VK_POSTER)
        app_enabled = await fetch_bool_setting("vk_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_VK_POSTER, os.getpid())
        while True:
            try:
                await self.run_once()
            except Exception as exc:
                logger.exception("VK poster loop error: {}", exc)
            await asyncio.sleep(self.interval_sec)


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


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