"""Instance realtime metrics collection and Redis cache buffering."""
from __future__ import annotations

import json
import logging
from datetime import datetime, timedelta, timezone
from typing import Any

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

from app.core.crypto import decrypt_secret
from app.models.cloud_server import CloudServer
from app.models.instance import Instance
from app.models.instance_real_data_snapshot import InstanceRealDataSnapshot
from app.redis_client import k, redis_client
from app.services.dcim_cloud_client import DcimCloudClient, DcimCloudCredentials, DcimCloudError
from app.services import zjmf_instance_adapter

log = logging.getLogger("qycw.instance_real_data")

REAL_DATA_TTL_SECONDS = 300
REAL_DATA_HISTORY_LIMIT = 600
REAL_DATA_HISTORY_TTL_SECONDS = 24 * 60 * 60
REAL_DATA_HISTORY_QUERY_HOURS = 24
REAL_DATA_SNAPSHOT_RETENTION_DAYS = 3
REAL_DATA_DB_BUFFER_KEY = k("instances", "real_data", "db_buffer")
REAL_DATA_DB_LAST_FLUSH_KEY = k("instances", "real_data", "last_flush")
REAL_DATA_DB_FLUSH_INTERVAL_SECONDS = 15 * 60
REAL_DATA_DB_FLUSH_MIN_ITEMS = 50


def real_data_cache_key(instance_id: int) -> str:
    return k("instances", "real_data", str(instance_id))


def real_data_history_key(instance_id: int) -> str:
    return k("instances", "real_data", "history", str(instance_id))


async def get_cached_real_data(instance_id: int) -> dict[str, Any] | None:
    raw = await redis_client.get(real_data_cache_key(instance_id))
    if not raw:
        return None
    try:
        data = json.loads(raw)
    except ValueError:
        return None
    return data if isinstance(data, dict) else None


async def get_cached_real_data_history(instance_id: int) -> list[dict[str, Any]]:
    rows = await redis_client.lrange(real_data_history_key(instance_id), 0, -1)
    items: list[dict[str, Any]] = []
    for raw in rows:
        try:
            data = json.loads(raw)
        except ValueError:
            continue
        if isinstance(data, dict):
            items.append(data)
    return items


def _snapshot_to_payload(snapshot: InstanceRealDataSnapshot) -> dict[str, Any]:
    return {
        "id": snapshot.upstream_id,
        "instance_id": snapshot.instance_id,
        "kvm": snapshot.kvm,
        "cpu_usage": snapshot.cpu_usage,
        "memory_usage": snapshot.memory_usage,
        "memory_total": snapshot.memory_total,
        "memory_usable": snapshot.memory_usable,
        "current_in_bw": snapshot.current_in_bw,
        "current_out_bw": snapshot.current_out_bw,
        "current_read_byte": snapshot.current_read_byte,
        "current_write_byte": snapshot.current_write_byte,
        "collected_at": snapshot.collected_at.isoformat(),
    }


async def get_real_data_history(instance_id: int, db: AsyncSession) -> list[dict[str, Any]]:
    now = datetime.now(timezone.utc)
    since = now - timedelta(hours=REAL_DATA_HISTORY_QUERY_HOURS)
    res = await db.execute(
        select(InstanceRealDataSnapshot)
        .where(
            InstanceRealDataSnapshot.instance_id == instance_id,
            InstanceRealDataSnapshot.collected_at >= since,
        )
        .order_by(InstanceRealDataSnapshot.collected_at.asc())
    )
    merged: dict[str, dict[str, Any]] = {}
    for snapshot in res.scalars().all():
        item = _snapshot_to_payload(snapshot)
        merged[str(item["collected_at"])] = item

    for item in await get_cached_real_data_history(instance_id):
        collected_at = item.get("collected_at")
        if isinstance(collected_at, str):
            merged[collected_at] = item

    return [
        item
        for _, item in sorted(
            merged.items(),
            key=lambda pair: pair[0],
        )
    ]


async def cleanup_old_real_data_snapshots(db: AsyncSession) -> int:
    cutoff = datetime.now(timezone.utc) - timedelta(days=REAL_DATA_SNAPSHOT_RETENTION_DAYS)
    res = await db.execute(
        delete(InstanceRealDataSnapshot).where(
            InstanceRealDataSnapshot.collected_at < cutoff
        )
    )
    await db.commit()
    return int(res.rowcount or 0)


def _parse_real_data_payload(text: str, body: Any) -> Any:
    if body is not None:
        return body
    stripped = text.strip()
    if not stripped:
        return None
    try:
        return json.loads(stripped)
    except ValueError:
        return None


def _extract_real_data_items(payload: Any) -> list[dict[str, Any]]:
    if isinstance(payload, dict):
        payload = payload.get("data") or payload.get("list") or payload
        if isinstance(payload, dict):
            payload = payload.get("list") or payload.get("data") or []
    if not isinstance(payload, list):
        return []
    return [item for item in payload if isinstance(item, dict)]


def _float_or_none(value: Any) -> float | None:
    if value is None or value == "":
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


async def _mark_buffer_seen() -> None:
    exists = await redis_client.exists(REAL_DATA_DB_LAST_FLUSH_KEY)
    if not exists:
        await redis_client.set(
            REAL_DATA_DB_LAST_FLUSH_KEY,
            datetime.now(timezone.utc).isoformat(),
        )


async def _write_real_data_sample(
    instance_id: int,
    item: dict[str, Any],
    collected_at: str,
) -> None:
    cache_payload = {**item, "instance_id": instance_id, "collected_at": collected_at}
    encoded = json.dumps(cache_payload, ensure_ascii=False)
    pipe = redis_client.pipeline()
    pipe.set(real_data_cache_key(instance_id), encoded, ex=REAL_DATA_TTL_SECONDS)
    pipe.rpush(real_data_history_key(instance_id), encoded)
    pipe.ltrim(real_data_history_key(instance_id), -REAL_DATA_HISTORY_LIMIT, -1)
    pipe.expire(real_data_history_key(instance_id), REAL_DATA_HISTORY_TTL_SECONDS)
    pipe.hset(REAL_DATA_DB_BUFFER_KEY, str(instance_id), encoded)
    await pipe.execute()


async def _collect_zjmf_group(
    db: AsyncSession,
    group_rows: list[tuple[Instance, CloudServer]],
    collected_at: str,
) -> int:
    """魔方财务实例逐台采集监控图表（charts 接口按主机维度，不支持批量）。"""
    written = 0
    for instance, _ in group_rows:
        if not instance.upstream_id:
            continue
        try:
            item = await zjmf_instance_adapter.collect_real_data_for_instance(db, instance)
        except Exception as e:  # noqa: BLE001
            log.warning("real_data zjmf collect instance=%s failed: %s", instance.id, e)
            continue
        if not item:
            continue
        await _write_real_data_sample(instance.id, item, collected_at)
        written += 1
    return written


async def collect_instance_real_data(db: AsyncSession) -> int:
    res = await db.execute(
        select(Instance, CloudServer)
        .join(CloudServer, CloudServer.id == Instance.cloud_server_id)
        .where(
            Instance.upstream_id.is_not(None),
            Instance.cloud_server_id.is_not(None),
            CloudServer.disabled.is_(False),
        )
        .order_by(CloudServer.id.asc(), Instance.id.asc())
    )
    rows = res.all()
    if not rows:
        return 0

    grouped: dict[int, list[tuple[Instance, CloudServer]]] = {}
    for instance, cloud_server in rows:
        grouped.setdefault(cloud_server.id, []).append((instance, cloud_server))

    written = 0
    collected_at = datetime.now(timezone.utc).isoformat()

    for _, group_rows in grouped.items():
        cloud_server = group_rows[0][1]
        if zjmf_instance_adapter.is_zjmf_server(cloud_server):
            written += await _collect_zjmf_group(db, group_rows, collected_at)
            continue
        upstream_ids = [instance.upstream_id for instance, _ in group_rows if instance.upstream_id]
        if not upstream_ids:
            continue

        creds = DcimCloudCredentials(
            base_url=cloud_server.base_url,
            username=cloud_server.username,
            password=decrypt_secret(cloud_server.password_encrypted),
            account_type=cloud_server.account_type,
        )
        client = DcimCloudClient(creds, server_id=cloud_server.id)

        try:
            code, text, body = await client.request(
                "/clouds/real_data",
                data={"id": upstream_ids},
                method="POST",
                timeout=12.0,
            )
        except DcimCloudError as e:
            log.warning("real_data collect cloud_server=%s failed: %s", cloud_server.id, e)
            continue

        payload = _parse_real_data_payload(text, body)
        if not (200 <= code < 300) or payload is None:
            log.warning("real_data collect cloud_server=%s http=%s body=%s", cloud_server.id, code, text[:200])
            continue

        items = _extract_real_data_items(payload)
        if not items:
            continue

        by_upstream_id = {str(item.get("id")): item for item in items if item.get("id") is not None}
        pipe = redis_client.pipeline()
        batch_count = 0
        for instance, _ in group_rows:
            item = by_upstream_id.get(str(instance.upstream_id))
            if not item:
                continue
            cache_payload = {**item, "instance_id": instance.id, "collected_at": collected_at}
            encoded = json.dumps(cache_payload, ensure_ascii=False)
            pipe.set(
                real_data_cache_key(instance.id),
                encoded,
                ex=REAL_DATA_TTL_SECONDS,
            )
            pipe.rpush(real_data_history_key(instance.id), encoded)
            pipe.ltrim(real_data_history_key(instance.id), -REAL_DATA_HISTORY_LIMIT, -1)
            pipe.expire(real_data_history_key(instance.id), REAL_DATA_HISTORY_TTL_SECONDS)
            pipe.hset(REAL_DATA_DB_BUFFER_KEY, str(instance.id), encoded)
            batch_count += 1
        if batch_count:
            await pipe.execute()
            written += batch_count

    if written:
        await _mark_buffer_seen()
    return written


async def flush_real_data_buffer_to_db(db: AsyncSession, *, force: bool = False) -> int:
    buffered_count = await redis_client.hlen(REAL_DATA_DB_BUFFER_KEY)
    if buffered_count <= 0:
        return 0

    now = datetime.now(timezone.utc)
    should_flush = force or buffered_count >= REAL_DATA_DB_FLUSH_MIN_ITEMS
    last_flush_raw = await redis_client.get(REAL_DATA_DB_LAST_FLUSH_KEY)
    if last_flush_raw:
        try:
            last_flush = datetime.fromisoformat(last_flush_raw)
        except ValueError:
            last_flush = None
        if last_flush and (now - last_flush).total_seconds() >= REAL_DATA_DB_FLUSH_INTERVAL_SECONDS:
            should_flush = True
    elif not should_flush:
        await redis_client.set(REAL_DATA_DB_LAST_FLUSH_KEY, now.isoformat())

    if not should_flush:
        return 0

    raw_items = await redis_client.hgetall(REAL_DATA_DB_BUFFER_KEY)
    snapshots: list[InstanceRealDataSnapshot] = []
    flushed_fields: list[str] = []
    for field, raw in raw_items.items():
        try:
            item = json.loads(raw)
        except ValueError:
            flushed_fields.append(field)
            continue
        if not isinstance(item, dict):
            flushed_fields.append(field)
            continue

        collected_at = now
        raw_collected_at = item.get("collected_at")
        if isinstance(raw_collected_at, str):
            try:
                collected_at = datetime.fromisoformat(raw_collected_at)
            except ValueError:
                collected_at = now

        snapshots.append(
            InstanceRealDataSnapshot(
                instance_id=int(item.get("instance_id") or field),
                upstream_id=int(item["id"]) if item.get("id") is not None else None,
                kvm=str(item["kvm"]) if item.get("kvm") is not None else None,
                cpu_usage=_float_or_none(item.get("cpu_usage")),
                memory_usage=_float_or_none(item.get("memory_usage")),
                memory_total=str(item["memory_total"]) if item.get("memory_total") is not None else None,
                memory_usable=str(item["memory_usable"]) if item.get("memory_usable") is not None else None,
                current_in_bw=str(item["current_in_bw"]) if item.get("current_in_bw") is not None else None,
                current_out_bw=str(item["current_out_bw"]) if item.get("current_out_bw") is not None else None,
                current_read_byte=_float_or_none(item.get("current_read_byte")),
                current_write_byte=_float_or_none(item.get("current_write_byte")),
                collected_at=collected_at,
                raw_json=json.dumps(item, ensure_ascii=False),
            )
        )
        flushed_fields.append(field)

    if snapshots:
        db.add_all(snapshots)
        await db.commit()
    if flushed_fields:
        await redis_client.hdel(REAL_DATA_DB_BUFFER_KEY, *flushed_fields)
    await redis_client.set(REAL_DATA_DB_LAST_FLUSH_KEY, now.isoformat())
    return len(snapshots)
