from __future__ import annotations

import hashlib
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from planagent.config import Settings
from planagent.domain.models import utc_now

if TYPE_CHECKING:
    from planagent.domain.models import SourceCursorState


class SourceStateService:
    """数据源游标状态管理——跟踪每个数据源的增量采集状态。"""

    def __init__(self, settings: Settings) -> None:
        self.settings = settings

    async def get_or_create_state(
        self,
        session: AsyncSession,
        source_type: str,
        source_url_or_query: str,
        watch_rule_id: str | None = None,
        tenant_id: str | None = None,
        preset_id: str | None = None,
    ) -> SourceCursorState:
        """获取或创建数据源状态记录。"""
        SourceCursorStateModel = self._source_cursor_state_model()
        state = (
            await session.scalars(
                select(SourceCursorStateModel)
                .where(
                    SourceCursorStateModel.source_type == source_type,
                    SourceCursorStateModel.source_url_or_query == source_url_or_query,
                    SourceCursorStateModel.watch_rule_id == watch_rule_id,
                )
                .limit(1)
            )
        ).first()
        if state is not None:
            return state

        state = SourceCursorStateModel(
            watch_rule_id=watch_rule_id,
            source_type=source_type,
            source_url_or_query=source_url_or_query,
            tenant_id=tenant_id,
            preset_id=preset_id,
        )
        session.add(state)
        await session.flush()
        return state

    async def seed_watch_rule_sources(
        self,
        session: AsyncSession,
        *,
        watch_rule_id: str,
        query: str,
        source_types: list[str],
        tenant_id: str | None = None,
        preset_id: str | None = None,
    ) -> list[SourceCursorState]:
        """Ensure each configured watch source has a cursor state row."""
        states: list[SourceCursorState] = []
        for source_type in sorted({item for item in source_types if item}):
            states.append(
                await self.get_or_create_state(
                    session,
                    source_type=source_type,
                    source_url_or_query=query,
                    watch_rule_id=watch_rule_id,
                    tenant_id=tenant_id,
                    preset_id=preset_id,
                )
            )
        return states

    async def update_after_fetch(
        self,
        session: AsyncSession,
        state_id: str,
        success: bool,
        cursor: str | None = None,
        etag: str | None = None,
        last_modified: str | None = None,
        content_hash: str | None = None,
        raw_source_item_id: str | None = None,
        preserve_changed: bool = False,
    ) -> None:
        """抓取成功/失败后更新游标状态。"""
        SourceCursorStateModel = self._source_cursor_state_model()
        state = await session.get(SourceCursorStateModel, state_id)
        if state is None:
            raise LookupError(f"Source cursor state {state_id} was not found.")

        now = utc_now()
        state.last_checked_at = now
        if success:
            if cursor is not None:
                state.cursor = cursor
            if etag is not None:
                state.etag = etag
            if last_modified is not None:
                state.last_modified = last_modified
            if content_hash is not None:
                state.last_seen_hash = self._normalize_content_hash(content_hash)
            if raw_source_item_id is not None:
                state.last_seen_raw_source_item_id = raw_source_item_id
            state.last_success_at = now
            state.consecutive_failures = 0
            state.health_status = "changed" if preserve_changed else "healthy"
        else:
            state.consecutive_failures = int(state.consecutive_failures or 0) + 1
            state.last_failure_at = now
            state.health_status = "failed" if state.consecutive_failures >= 3 else "degraded"

        state.updated_at = now
        await session.flush()

    async def should_fetch(
        self,
        session: AsyncSession,
        state: SourceCursorState,
        force_full_refresh_every_minutes: int = 24 * 60,
    ) -> bool:
        """判断是否需要抓取（基于分钟级强制刷新间隔和上次失败状态）。"""
        _ = session
        last_success_at = self._normalize_datetime(state.last_success_at)
        last_failure_at = self._normalize_datetime(state.last_failure_at)

        if last_success_at is None and last_failure_at is None:
            return True

        if last_failure_at is not None and (
            last_success_at is None or last_failure_at > last_success_at
        ):
            failures = int(state.consecutive_failures or 0)
            threshold = max(1, int(self.settings.source_failure_circuit_breaker_threshold))
            if failures < threshold:
                return True
            backoff_minutes = self._failure_backoff_minutes(failures, threshold)
            return utc_now() - last_failure_at >= timedelta(minutes=backoff_minutes)

        if last_success_at is None:
            return False

        return utc_now() - last_success_at >= timedelta(minutes=force_full_refresh_every_minutes)

    async def reset_cursor(
        self,
        session: AsyncSession,
        watch_rule_id: str,
    ) -> int:
        """重置游标（强制下次全量抓取）。"""
        SourceCursorStateModel = self._source_cursor_state_model()
        states = list(
            (
                await session.scalars(
                    select(SourceCursorStateModel).where(
                        SourceCursorStateModel.watch_rule_id == watch_rule_id
                    )
                )
            ).all()
        )
        now = utc_now()
        for state in states:
            state.cursor = None
            state.etag = None
            state.last_modified = None
            state.last_seen_hash = None
            state.last_seen_raw_source_item_id = None
            state.health_status = "pending"
            state.updated_at = now

        from planagent.domain.models import WatchRule

        rule = await session.get(WatchRule, watch_rule_id)
        if rule is not None:
            rule.last_cursor_reset_at = now
            rule.next_poll_at = now
        await session.commit()
        return len(states)

    async def list_states(
        self,
        session: AsyncSession,
        watch_rule_id: str | None = None,
        source_type: str | None = None,
    ) -> list[SourceCursorState]:
        """列出数据源状态。"""
        SourceCursorStateModel = self._source_cursor_state_model()
        query = select(SourceCursorStateModel)
        if watch_rule_id is not None:
            query = query.where(SourceCursorStateModel.watch_rule_id == watch_rule_id)
        if source_type is not None:
            query = query.where(SourceCursorStateModel.source_type == source_type)
        query = query.order_by(SourceCursorStateModel.updated_at.desc())
        return list((await session.scalars(query)).all())

    def _normalize_datetime(self, value: datetime | None) -> datetime | None:
        if value is None:
            return None
        if value.tzinfo is None:
            return value.replace(tzinfo=timezone.utc)
        return value

    def _normalize_content_hash(self, value: str) -> str:
        normalized = value.strip()
        if len(normalized) == 64 and all(char in "0123456789abcdefABCDEF" for char in normalized):
            return normalized.lower()
        return hashlib.sha256(normalized.encode("utf-8")).hexdigest()

    def _failure_backoff_minutes(self, failures: int, threshold: int) -> int:
        base_minutes = max(1, int(self.settings.source_failure_backoff_base_minutes))
        max_minutes = max(base_minutes, int(self.settings.source_failure_backoff_max_minutes))
        exponent = min(max(failures - threshold, 0), 8)
        return min(max_minutes, base_minutes * (2**exponent))

    def _source_cursor_state_model(self) -> type[SourceCursorState]:
        from planagent.domain.models import SourceCursorState

        return SourceCursorState
