from __future__ import annotations

import asyncio
import hashlib
import json
from datetime import datetime, timedelta, timezone

from loguru import logger

from ..config import settings
from ..constants import (
    JOB_TYPE_VK_STORAGE_COPY,
    PLATFORM_VK,
    POST_STATUS_SKIPPED,
    POST_STATUS_STORAGE_PENDING,
    SOURCE_STATUS_ERROR,
    SOURCE_STATUS_OK,
    WORKER_PARSER,
)
from ..db import fetch_bool_setting, fetch_float_setting, fetch_int_setting, get_pool
from ..heartbeat import HeartbeatReporter
from ..jobs import is_worker_enabled
from ..vk_api import (
    VKAPIClient,
    VKAPIError,
    extract_media,
    is_deleted_or_invalid,
    is_fatal_source_error,
    is_repost,
    post_vk_url,
)


def utc_from_ts(value: int) -> datetime:
    return datetime.fromtimestamp(int(value), tz=timezone.utc)


def make_hash(*parts: str) -> str:
    h = hashlib.sha256()
    for part in parts:
        h.update((part or "").strip().lower().encode("utf-8"))
        h.update(b"\0")
    return h.hexdigest()


class VKParserWorker:
    def __init__(self) -> None:
        self.pool = None
        self.heartbeat = HeartbeatReporter(WORKER_PARSER, 30)

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

    async def active_sources(self) -> list[dict]:
        rows = await self.pool.fetch(
            """
            SELECT *
            FROM sources
            WHERE platform=$1
              AND active=TRUE
              AND archived_at IS NULL
            ORDER BY last_checked_at NULLS FIRST, priority ASC, id ASC
            """,
            PLATFORM_VK,
        )
        return [dict(r) for r in rows]

    async def mark_source_error(self, source_id: int, message: str) -> None:
        await self.pool.execute(
            """
            UPDATE sources
            SET status=$2,
                status_msg=$3,
                last_checked_at=NOW(),
                updated_at=NOW()
            WHERE id=$1
            """,
            source_id,
            SOURCE_STATUS_ERROR,
            message[:1000],
        )

    async def deactivate_source(self, source_id: int, message: str) -> None:
        await self.pool.execute(
            """
            UPDATE sources
            SET active=FALSE,
                status=$2,
                status_msg=$3,
                last_checked_at=NOW(),
                updated_at=NOW()
            WHERE id=$1
            """,
            source_id,
            SOURCE_STATUS_ERROR,
            message[:1000],
        )

    async def mark_source_ok(self, source_id: int, last_parsed_at: datetime | None) -> None:
        await self.pool.execute(
            """
            UPDATE sources
            SET status=$2,
                status_msg=NULL,
                last_checked_at=NOW(),
                last_parsed_at=COALESCE($3, last_parsed_at),
                updated_at=NOW()
            WHERE id=$1
            """,
            source_id,
            SOURCE_STATUS_OK,
            last_parsed_at,
        )

    async def resolve_source_if_needed(self, client: VKAPIClient, source: dict) -> dict:
        if source.get("external_owner_id"):
            return source
        external_id, owner_id, resolved_name = await client.resolve_group(source["url"] or source.get("external_id") or "")
        await self.pool.execute(
            """
            UPDATE sources
            SET external_id=$2,
                external_owner_id=$3,
                name=CASE WHEN name='' OR name=url THEN $4 ELSE name END,
                updated_at=NOW()
            WHERE id=$1
            """,
            int(source["id"]),
            external_id,
            owner_id,
            resolved_name,
        )
        source["external_id"] = external_id
        source["external_owner_id"] = owner_id
        if not source.get("name") or source.get("name") == source.get("url"):
            source["name"] = resolved_name
        return source

    async def known_post_ids(self, source_id: int, external_post_ids: list[str]) -> set[str]:
        if not external_post_ids:
            return set()
        rows = await self.pool.fetch(
            """
            SELECT external_post_id
            FROM raw_posts
            WHERE source_id=$1 AND external_post_id=ANY($2::text[])
            """,
            source_id,
            external_post_ids,
        )
        return {str(r["external_post_id"]) for r in rows}

    async def known_content_hashes(self, hashes: list[str]) -> set[str]:
        if not hashes:
            return set()
        rows = await self.pool.fetch(
            """
            SELECT content_hash
            FROM raw_posts
            WHERE content_hash=ANY($1::text[])
            """,
            hashes,
        )
        return {str(r["content_hash"]) for r in rows}

    async def save_post(
        self,
        source: dict,
        post: dict,
        *,
        status: str,
        skip_reason: str | None = None,
        create_storage_job: bool = True,
    ) -> int | None:
        source_id = int(source["id"])
        owner_id = int(post.get("owner_id") or source["external_owner_id"])
        external_post_id = str(post["id"])
        raw_text = str(post.get("text") or "").strip()
        posted_at = utc_from_ts(int(post["date"]))
        media = extract_media(post)
        media_ids = ",".join(m.attachment_id for m in media)
        text_hash = make_hash(raw_text)
        content_hash = make_hash(raw_text, media_ids)
        original_url = post_vk_url(owner_id, external_post_id)

        async with self.pool.acquire() as conn:
            async with conn.transaction():
                raw_post_id = await conn.fetchval(
                    """
                    INSERT INTO raw_posts(
                      source_id, platform, external_post_id, external_owner_id,
                      original_url, raw_text, raw_json, text_hash, content_hash,
                      posted_at, status, skip_reason
                    )
                    VALUES($1,$2,$3,$4,$5,$6,$7::jsonb,$8,$9,$10,$11,$12)
                    ON CONFLICT (source_id, external_post_id) DO NOTHING
                    RETURNING id
                    """,
                    source_id,
                    PLATFORM_VK,
                    external_post_id,
                    owner_id,
                    original_url,
                    raw_text,
                    json.dumps(post, ensure_ascii=False),
                    text_hash,
                    content_hash,
                    posted_at,
                    status,
                    skip_reason,
                )
                if raw_post_id is None:
                    return None

                for m in media:
                    await conn.execute(
                        """
                        INSERT INTO raw_post_media(
                          raw_post_id, platform, media_type, original_url, original_attachment_id,
                          width, height, duration_sec, sort_order
                        )
                        VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)
                        """,
                        raw_post_id,
                        PLATFORM_VK,
                        m.media_type,
                        m.original_url,
                        m.attachment_id,
                        m.width,
                        m.height,
                        m.duration_sec,
                        m.sort_order,
                    )

                if create_storage_job:
                    await conn.execute(
                        """
                        INSERT INTO jobs(type, entity_type, entity_id, payload_json, status)
                        VALUES($1, 'raw_post', $2, '{}'::jsonb, 'pending')
                        ON CONFLICT DO NOTHING
                        """,
                        JOB_TYPE_VK_STORAGE_COPY,
                        raw_post_id,
                    )
        return int(raw_post_id)

    async def parse_source(self, client: VKAPIClient, source: dict) -> int:
        source = await self.resolve_source_if_needed(client, source)
        source_id = int(source["id"])
        owner_id = int(source["external_owner_id"])
        page_size = max(1, min(100, await fetch_int_setting("vk_wall_page_size", 50)))
        lookback_days = max(1, await fetch_int_setting("parser_new_source_lookback_days", 14))
        overlap_minutes = max(0, await fetch_int_setting("parser_reparse_overlap_minutes", 120))

        last_parsed_at = source.get("last_parsed_at")
        parse_from = source.get("parse_from")
        if last_parsed_at:
            since_dt = last_parsed_at - timedelta(minutes=overlap_minutes)
        elif parse_from:
            since_dt = parse_from
        else:
            since_dt = datetime.now(timezone.utc) - timedelta(days=lookback_days)
        if since_dt.tzinfo is None:
            since_dt = since_dt.replace(tzinfo=timezone.utc)
        since_ts = int(since_dt.timestamp())

        fetched: list[dict] = []
        offset = 0
        stop = False
        while not stop:
            response = await client.get_wall_posts(owner_id=owner_id, count=page_size, offset=offset)
            items = response.get("items", []) or []
            if not items:
                break
            for post in items:
                if is_deleted_or_invalid(post):
                    continue
                if int(post["date"]) <= since_ts:
                    stop = True
                    break
                fetched.append(post)
            if len(items) < page_size:
                break
            offset += page_size

        known = await self.known_post_ids(source_id, [str(p["id"]) for p in fetched if p.get("id")])
        candidates = [p for p in fetched if str(p.get("id")) not in known]
        saved = 0
        max_seen: datetime | None = None
        min_text_length = max(0, await fetch_int_setting("parser_min_text_length", 0))
        skip_reposts = await fetch_bool_setting("parser_skip_reposts", True)
        skip_empty_text = await fetch_bool_setting("parser_skip_empty_text", True)
        skip_no_media = await fetch_bool_setting("parser_skip_no_media", True)
        skip_short_text = await fetch_bool_setting("parser_skip_text_too_short", True)
        store_skipped = await fetch_bool_setting("parser_store_skipped_posts", False)
        dedupe_content_hash = await fetch_bool_setting("parser_dedupe_content_hash", True)

        hash_by_post_id: dict[str, str] = {}
        if dedupe_content_hash:
            candidate_hashes: list[str] = []
            for post in candidates:
                if is_deleted_or_invalid(post) or (skip_reposts and is_repost(post)):
                    continue
                media = extract_media(post)
                media_ids = ",".join(m.attachment_id for m in media)
                content_hash = make_hash(str(post.get("text") or "").strip(), media_ids)
                hash_by_post_id[str(post["id"])] = content_hash
                candidate_hashes.append(content_hash)
            known_hashes = await self.known_content_hashes(candidate_hashes)
        else:
            known_hashes = set()

        for post in candidates:
            if is_deleted_or_invalid(post):
                continue
            text = str(post.get("text") or "").strip()
            posted_at = utc_from_ts(int(post["date"]))
            if max_seen is None or posted_at > max_seen:
                max_seen = posted_at

            if skip_reposts and is_repost(post):
                continue

            media = extract_media(post)
            content_hash = hash_by_post_id.get(str(post["id"]))
            if dedupe_content_hash and content_hash in known_hashes:
                continue

            skip_reason: str | None = None
            if skip_empty_text and not text:
                skip_reason = "empty_text"
            elif skip_no_media and not media:
                skip_reason = "no_media"
            elif skip_short_text and len(text) < min_text_length:
                skip_reason = "text_too_short"

            if skip_reason:
                if not store_skipped:
                    continue
                raw_id = await self.save_post(
                    source,
                    post,
                    status=POST_STATUS_SKIPPED,
                    skip_reason=skip_reason,
                    create_storage_job=False,
                )
            else:
                raw_id = await self.save_post(
                    source,
                    post,
                    status=POST_STATUS_STORAGE_PENDING,
                    skip_reason=None,
                    create_storage_job=True,
                )
            if raw_id:
                saved += 1
                if content_hash:
                    known_hashes.add(content_hash)

        await self.mark_source_ok(source_id, max_seen or last_parsed_at)
        logger.info(
            "Parsed source {}: fetched={} known={} candidates={} saved={}",
            source.get("name"),
            len(fetched),
            len(known),
            len(candidates),
            saved,
        )
        return saved

    async def run_once(self) -> None:
        enabled = await is_worker_enabled(self.pool, WORKER_PARSER)
        if not enabled:
            await self.heartbeat.beat(self.pool, status="disabled", force=True)
            return

        rps = await fetch_int_setting("vk_requests_per_second", 3)
        timeout_total = await fetch_int_setting("vk_api_timeout_total_sec", 15)
        timeout_connect = await fetch_int_setting("vk_api_timeout_connect_sec", 5)
        rate_limit_sleep = await fetch_float_setting("vk_rate_limit_sleep_sec", 1.0)
        retry_attempts = await fetch_int_setting("vk_api_retry_attempts", 3)
        retry_min_delay = await fetch_float_setting("vk_api_retry_min_delay_sec", 2.0)
        retry_max_delay = await fetch_float_setting("vk_api_retry_max_delay_sec", 10.0)
        source_pause = max(0.0, await fetch_float_setting("parser_source_pause_sec", 0.0))
        sources = await self.active_sources()
        await self.heartbeat.beat(self.pool, meta={"sources": len(sources)})
        if not sources:
            logger.info("No active VK sources")
            return

        logger.info(
            "Parser cycle: sources={} rps={} timeout={}/{} retry={}/{}..{} rate_limit_sleep={} source_pause={}",
            len(sources),
            rps,
            timeout_total,
            timeout_connect,
            retry_attempts,
            retry_min_delay,
            retry_max_delay,
            rate_limit_sleep,
            source_pause,
        )

        async with VKAPIClient(
            rps=rps,
            timeout_total_sec=timeout_total,
            timeout_connect_sec=timeout_connect,
            rate_limit_sleep_sec=rate_limit_sleep,
            retry_attempts=retry_attempts,
            retry_min_delay_sec=retry_min_delay,
            retry_max_delay_sec=retry_max_delay,
        ) as client:
            for source in sources:
                try:
                    await self.parse_source(client, source)
                except VKAPIError as e:
                    if is_fatal_source_error(e):
                        await self.deactivate_source(int(source["id"]), str(e))
                        logger.warning("VK source deactivated {}: {}", source.get("name"), e)
                    else:
                        await self.mark_source_error(int(source["id"]), str(e))
                        logger.warning("VK source temporary error {}: {}", source.get("name"), e)
                except Exception as e:
                    await self.mark_source_error(int(source["id"]), str(e))
                    logger.exception("Unexpected source error {}: {}", source.get("name"), e)
                if source_pause:
                    await asyncio.sleep(source_pause)

    async def run_loop(self) -> None:
        await self.init()
        logger.info("{} started", WORKER_PARSER)
        while True:
            try:
                await self.run_once()
            except Exception as e:
                logger.exception("Parser loop error: {}", e)
            interval = max(10, await fetch_int_setting("parser_interval_sec", 300))
            await asyncio.sleep(interval)


async def main() -> None:
    logger.remove()
    logger.add(lambda msg: print(msg, end=""), level=settings.log_level)
    worker = VKParserWorker()
    await worker.run_loop()


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