from __future__ import annotations

from collections import Counter, defaultdict
from datetime import datetime, timedelta, timezone
from typing import Any

from sqlalchemy import desc, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.crypto import decrypt_secret
from app.models.instance import Instance, InstanceStatus
from app.models.order import Order
from app.models.product import Product, ProductApiType, ProductStatus
from app.models.provision_interface import ProvisionInterface
from app.models.system_setting import SystemSetting
from app.models.user import User
from app.models.user_verification_record import UserVerificationRecord
from app.services import (
    notification_plugin_service,
    provision_interface_service,
    provision_plugin_service,
    system_setting_service,
)


NOTIFICATION_TYPES = ("email", "sms", "verification", "payment")
TYPE_LABELS = {
    "email": "邮件",
    "sms": "短信",
    "verification": "实名",
    "payment": "支付",
}
INSTANCE_STATUS_LABELS = {
    InstanceStatus.PROVISIONING.value: "开通中",
    InstanceStatus.ACTIVE.value: "运行中",
    InstanceStatus.SUSPENDED.value: "已暂停",
    InstanceStatus.TERMINATED.value: "已销毁",
    InstanceStatus.FAILED.value: "开通失败",
    InstanceStatus.PROVISION_FAILED.value: "待人工处理",
}
_SECRET_MASK = "********"


async def build_plugin_diagnostics(db: AsyncSession) -> dict[str, Any]:
    settings = await system_setting_service.ensure_system_settings(db)
    notification_channels = await _notification_channels(db, settings)
    provision = await _provision_diagnostics(db)
    verification = await _verification_diagnostics(db)
    security = await _security_diagnostics(db, settings)

    warnings = _collect_warnings(notification_channels, provision, verification, security)
    error_count = sum(1 for item in warnings if item["level"] == "error")
    warning_count = sum(1 for item in warnings if item["level"] == "warning")

    notification_plugin_count = sum(int(item["total_plugins"]) for item in notification_channels)
    provision_plugin_count = len(provision["plugins"])
    enabled_plugins = sum(int(item["enabled_count"]) for item in notification_channels) + int(
        provision["summary"]["enabled_interfaces"]
    )

    return {
        "summary": {
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "total_plugins": notification_plugin_count + provision_plugin_count,
            "notification_plugins": notification_plugin_count,
            "provision_plugins": provision_plugin_count,
            "enabled_plugins": enabled_plugins,
            "issue_count": error_count + warning_count,
            "error_count": error_count,
            "warning_count": warning_count,
            "recent_failed_instances": provision["summary"]["failed_instances"],
            "pending_redirect_verifications": verification["pending_redirect"],
            "product_binding_issues": provision["summary"]["product_binding_issues"],
            "plaintext_secret_count": security["plaintext_secret_count"],
        },
        "warnings": warnings,
        "channels": notification_channels,
        "provision": provision,
        "verification": verification,
        "security": security,
    }


async def _notification_channels(
    db: AsyncSession,
    settings: dict[str, SystemSetting],
) -> list[dict[str, Any]]:
    channels: list[dict[str, Any]] = []
    for plugin_type in NOTIFICATION_TYPES:
        plugins = await notification_plugin_service.list_plugins_with_state(db, plugin_type)
        enabled = [item for item in plugins if item.get("enabled")]
        current = [item for item in plugins if item.get("current")]
        missing_required: list[dict[str, str]] = []
        warnings: list[str] = []

        expected_missing = _missing_expected_plugin(plugin_type, settings, plugins)
        if expected_missing:
            warnings.append(expected_missing)

        for plugin in enabled or current:
            missing_required.extend(_missing_required_plugin_fields(plugin, settings))

        if not plugins:
            status = "error"
            warnings.append(f"{TYPE_LABELS[plugin_type]}通道没有安装可用插件")
        elif expected_missing:
            status = "error"
        elif missing_required:
            status = "error"
        elif not enabled:
            status = "disabled"
        else:
            status = "ok"

        channels.append(
            {
                "type": plugin_type,
                "label": TYPE_LABELS[plugin_type],
                "status": status,
                "total_plugins": len(plugins),
                "enabled_count": len(enabled),
                "current_count": len(current),
                "current_plugins": [_plugin_summary(item) for item in current],
                "enabled_plugins": [_plugin_summary(item) for item in enabled],
                "missing_required": missing_required,
                "warnings": warnings,
                "actions": [
                    {"label": "配置功能插件", "path": "/admin/notification-plugins"},
                ],
            }
        )
    return channels


async def _provision_diagnostics(db: AsyncSession) -> dict[str, Any]:
    plugin_manifests = [provision_plugin_service.admin_manifest(item) for item in provision_plugin_service.list_plugins()]
    plugin_map = {str(item.get("id")): item for item in plugin_manifests}
    interfaces = await provision_interface_service.list_interfaces(db)
    products = await _plugin_products(db)

    interface_rows = [_interface_diagnostic(item, plugin_map) for item in interfaces]
    product_issues = _product_binding_issues(products, interfaces, plugin_map)
    recent_failures, failed_count, status_counts = await _recent_provision_failures(db)
    stuck_count = await _stuck_provisioning_count(db)

    interface_counts: dict[str, int] = defaultdict(int)
    enabled_interface_counts: dict[str, int] = defaultdict(int)
    for item in interfaces:
        interface_counts[item.plugin_id] += 1
        if item.enabled:
            enabled_interface_counts[item.plugin_id] += 1

    product_counts: dict[str, int] = defaultdict(int)
    active_product_counts: dict[str, int] = defaultdict(int)
    for product in products:
        plugin_id = str(product.api_identifier or "").strip()
        if not plugin_id:
            continue
        product_counts[plugin_id] += 1
        if _enum_value(product.status) == ProductStatus.ACTIVE.value:
            active_product_counts[plugin_id] += 1

    plugin_rows = []
    for manifest in plugin_manifests:
        plugin_id = str(manifest.get("id") or "")
        plugin_rows.append(
            {
                "id": plugin_id,
                "name": manifest.get("name") or plugin_id,
                "version": manifest.get("version"),
                "status": manifest.get("status"),
                "capabilities": manifest.get("capabilities") or [],
                "interface_count": interface_counts[plugin_id],
                "enabled_interface_count": enabled_interface_counts[plugin_id],
                "product_count": product_counts[plugin_id],
                "active_product_count": active_product_counts[plugin_id],
                "config_field_count": len(manifest.get("config_fields") or []),
                "purchase_field_count": len(
                    ((manifest.get("product_page") or {}).get("purchase_fields") or [])
                    if isinstance(manifest.get("product_page"), dict)
                    else []
                ),
                "manage_action_count": len(manifest.get("manage_actions") or []),
            }
        )

    return {
        "summary": {
            "plugin_count": len(plugin_rows),
            "interface_count": len(interface_rows),
            "enabled_interfaces": sum(1 for item in interfaces if item.enabled),
            "product_count": len(products),
            "active_product_count": sum(
                1 for item in products if _enum_value(item.status) == ProductStatus.ACTIVE.value
            ),
            "product_binding_issues": len(product_issues),
            "failed_instances": failed_count,
            "stuck_provisioning": stuck_count,
        },
        "plugins": plugin_rows,
        "interfaces": interface_rows,
        "product_issues": product_issues,
        "recent_failures": recent_failures,
        "instance_status_counts": status_counts,
        "actions": [
            {"label": "开通接口", "path": "/admin/provision-plugins"},
            {"label": "失败实例", "path": "/admin/instances"},
            {"label": "任务中心", "path": "/admin/tasks"},
        ],
    }


async def _verification_diagnostics(db: AsyncSession) -> dict[str, Any]:
    pending_redirect = await db.scalar(
        select(func.count())
        .select_from(UserVerificationRecord)
        .where(
            UserVerificationRecord.mode == "face_redirect",
            UserVerificationRecord.status == "processing",
            UserVerificationRecord.provider_task_id.is_not(None),
        )
    )
    processing_rows = await db.execute(
        select(UserVerificationRecord, User)
        .join(User, User.id == UserVerificationRecord.user_id)
        .where(UserVerificationRecord.status.in_(["pending", "processing"]))
        .order_by(desc(UserVerificationRecord.updated_at))
        .limit(10)
    )
    status_rows = await db.execute(
        select(UserVerificationRecord.status, func.count())
        .select_from(UserVerificationRecord)
        .group_by(UserVerificationRecord.status)
    )
    status_counts = {str(status): int(count) for status, count in status_rows.all()}
    records = []
    for record, user in processing_rows.all():
        records.append(
            {
                "id": record.id,
                "user_id": record.user_id,
                "username": user.username,
                "provider": record.provider,
                "mode": record.mode,
                "status": record.status,
                "external_id": record.provider_task_id,
                "created_at": _iso(record.created_at),
                "updated_at": _iso(record.updated_at),
            }
        )
    return {
        "pending_redirect": int(pending_redirect or 0),
        "processing_records": records,
        "status_counts": status_counts,
        "actions": [
            {"label": "实名插件", "path": "/admin/notification-plugins"},
            {"label": "用户实名", "path": "/admin/users"},
        ],
    }


async def _security_diagnostics(
    db: AsyncSession,
    settings: dict[str, SystemSetting],
) -> dict[str, Any]:
    secret_items: list[dict[str, Any]] = []
    encrypted_count = 0
    plaintext_count = 0

    notification_plugins = notification_plugin_service.list_plugins()
    plugin_secret_fields: dict[str, dict[str, str]] = {}
    for plugin in notification_plugins:
        for field in plugin.get("config_fields") or []:
            if not isinstance(field, dict):
                continue
            key = str(field.get("key") or "").strip()
            if key and _field_is_secret(field):
                plugin_secret_fields[key] = {
                    "plugin": str(plugin.get("name") or plugin.get("id") or key),
                    "label": str(field.get("label") or key),
                }

    for key, item in settings.items():
        field = plugin_secret_fields.get(key)
        spec = system_setting_service.SETTING_SPECS.get(key)
        if not field and not system_setting_service.is_secret_setting_key(key, spec):
            continue
        value = item.value
        if value in (None, ""):
            secret_items.append(
                {
                    "scope": "system_setting",
                    "key": key,
                    "label": field["label"] if field else item.label,
                    "plugin": field["plugin"] if field else None,
                    "status": "empty",
                }
            )
            continue
        if _is_encrypted(value):
            encrypted_count += 1
            status = "encrypted"
        else:
            plaintext_count += 1
            status = "plaintext"
        secret_items.append(
            {
                "scope": "system_setting",
                "key": key,
                "label": field["label"] if field else item.label,
                "plugin": field["plugin"] if field else None,
                "status": status,
            }
        )

    interface_rows = await db.execute(select(ProvisionInterface).order_by(ProvisionInterface.id.desc()))
    for interface in interface_rows.scalars().all():
        stored = interface.config if isinstance(interface.config, dict) else {}
        values = stored.get("values") if isinstance(stored.get("values"), dict) else {}
        secret_keys = stored.get("secret_keys") if isinstance(stored.get("secret_keys"), list) else []
        for key in secret_keys:
            value = values.get(key)
            if value in (None, "", _SECRET_MASK):
                status = "empty"
            elif _is_encrypted(value):
                encrypted_count += 1
                status = "encrypted"
            else:
                plaintext_count += 1
                status = "plaintext"
            secret_items.append(
                {
                    "scope": "provision_interface",
                    "key": str(key),
                    "label": str(key),
                    "plugin": interface.plugin_id,
                    "interface_id": interface.id,
                    "interface_name": interface.name,
                    "status": status,
                }
            )

    return {
        "secret_count": len(secret_items),
        "encrypted_secret_count": encrypted_count,
        "plaintext_secret_count": plaintext_count,
        "items": secret_items,
    }


def _collect_warnings(
    channels: list[dict[str, Any]],
    provision: dict[str, Any],
    verification: dict[str, Any],
    security: dict[str, Any],
) -> list[dict[str, Any]]:
    warnings: list[dict[str, Any]] = []
    for channel in channels:
        if channel["status"] == "error":
            text = "、".join(channel["warnings"]) or "存在必填配置缺失"
            if channel["missing_required"]:
                labels = "、".join(item["label"] for item in channel["missing_required"])
                text = f"{text}；缺失：{labels}" if text else f"缺失：{labels}"
            warnings.append(
                {
                    "level": "error",
                    "scope": "notification",
                    "message": f"{channel['label']}插件异常",
                    "suggestion": text,
                    "path": "/admin/notification-plugins",
                }
            )
        elif channel["status"] == "disabled":
            warnings.append(
                {
                    "level": "warning",
                    "scope": "notification",
                    "message": f"{channel['label']}通道未启用",
                    "suggestion": "如该通道参与注册、登录、实名或支付流程，请完成配置并启用。",
                    "path": "/admin/notification-plugins",
                }
            )

    for item in provision["interfaces"]:
        if item["status"] in {"error", "warning"}:
            warnings.append(
                {
                    "level": "error" if item["status"] == "error" else "warning",
                    "scope": "provision_interface",
                    "message": f"开通接口 {item['name']} 状态异常",
                    "suggestion": "、".join(item["warnings"]) or item.get("last_test_message") or "请重新测试接口配置。",
                    "path": "/admin/provision-plugins",
                }
            )
    for item in provision["product_issues"]:
        warnings.append(
            {
                "level": "error",
                "scope": "product_binding",
                "message": f"商品 {item['product_name']} 开通绑定异常",
                "suggestion": "、".join(item["issues"]),
                "path": f"/admin/products/{item['product_id']}",
            }
        )
    if int(provision["summary"]["failed_instances"]) > 0:
        warnings.append(
            {
                "level": "error",
                "scope": "provision_instances",
                "message": "存在开通失败实例",
                "suggestion": "请进入实例管理筛选 failed / provision_failed，确认是否需要重试或人工处理。",
                "path": "/admin/instances",
            }
        )
    if int(provision["summary"]["stuck_provisioning"]) > 0:
        warnings.append(
            {
                "level": "warning",
                "scope": "provision_instances",
                "message": "存在长时间开通中的实例",
                "suggestion": "超过 30 分钟仍处于开通中，建议检查上游任务或任务中心。",
                "path": "/admin/tasks",
            }
        )
    if int(verification["pending_redirect"]) > 0:
        warnings.append(
            {
                "level": "warning",
                "scope": "verification",
                "message": "存在 H5 实名处理中记录",
                "suggestion": "如长时间未回调，请确认实名插件回调地址和前端返回地址配置。",
                "path": "/admin/users",
            }
        )
    if int(security["plaintext_secret_count"]) > 0:
        warnings.append(
            {
                "level": "error",
                "scope": "security",
                "message": "发现明文密钥配置",
                "suggestion": "请重新保存对应插件或接口配置，让系统写入加密值。",
                "path": "/admin/plugin-diagnostics",
            }
        )
    return warnings


def _missing_expected_plugin(
    plugin_type: str,
    settings: dict[str, SystemSetting],
    plugins: list[dict[str, Any]],
) -> str | None:
    if plugin_type == "payment":
        enabled_ids = _split_csv(_setting_value(settings, "payment.enabled_plugins"))
        installed_ids = {str(item.get("id")) for item in plugins}
        missing = [item for item in enabled_ids if item not in installed_ids]
        return f"已启用的支付插件不存在：{', '.join(missing)}" if missing else None

    enabled_key = "verification.enabled" if plugin_type == "verification" else f"notification.{plugin_type}.enabled"
    provider_key = "verification.provider" if plugin_type == "verification" else f"notification.{plugin_type}.provider"
    enabled = bool(_setting_value(settings, enabled_key))
    provider = str(_setting_value(settings, provider_key) or "")
    if not enabled:
        return None
    if not any(str(item.get("provider")) == provider for item in plugins):
        return f"当前 provider 未匹配已安装插件：{provider}"
    return None


def _missing_required_plugin_fields(
    plugin: dict[str, Any],
    settings: dict[str, SystemSetting],
) -> list[dict[str, str]]:
    missing = []
    for field in plugin.get("config_fields") or []:
        if not isinstance(field, dict) or not field.get("required"):
            continue
        key = str(field.get("key") or "").strip()
        if not key:
            continue
        value = settings[key].value if key in settings else field.get("default")
        if not _present(value):
            missing.append(
                {
                    "plugin_id": str(plugin.get("id") or ""),
                    "plugin_name": str(plugin.get("name") or plugin.get("id") or ""),
                    "key": key,
                    "label": str(field.get("label") or key),
                }
            )
    return missing


def _interface_diagnostic(
    item: ProvisionInterface,
    plugin_map: dict[str, dict[str, Any]],
) -> dict[str, Any]:
    manifest = plugin_map.get(item.plugin_id)
    warnings: list[str] = []
    missing_required: list[dict[str, str]] = []
    if not manifest:
        warnings.append(f"接口绑定的插件不存在：{item.plugin_id}")
    else:
        stored = item.config if isinstance(item.config, dict) else {}
        values = stored.get("values") if isinstance(stored.get("values"), dict) else {}
        for field in manifest.get("config_fields") or []:
            if not isinstance(field, dict) or not field.get("required"):
                continue
            key = str(field.get("key") or "").strip()
            if key and not _present(values.get(key, field.get("default"))):
                missing_required.append({"key": key, "label": str(field.get("label") or key)})
        if missing_required:
            warnings.append("必填配置缺失")
    if not item.enabled:
        warnings.append("接口已禁用")
    if item.enabled and item.last_test_ok is False:
        warnings.append(item.last_test_message or "最近一次接口测试失败")
    if item.enabled and item.last_test_ok is None:
        warnings.append("接口尚未完成连通性测试")

    status = "ok"
    if not item.enabled:
        status = "disabled"
    elif not manifest or missing_required or item.last_test_ok is False:
        status = "error"
    elif item.last_test_ok is None:
        status = "warning"

    return {
        "id": item.id,
        "name": item.name,
        "plugin_id": item.plugin_id,
        "plugin_name": str((manifest or {}).get("name") or item.plugin_id),
        "enabled": item.enabled,
        "status": status,
        "missing_required": missing_required,
        "warnings": warnings,
        "last_test_ok": item.last_test_ok,
        "last_test_message": item.last_test_message,
        "updated_at": _iso(item.updated_at),
    }


def _product_binding_issues(
    products: list[Product],
    interfaces: list[ProvisionInterface],
    plugin_map: dict[str, dict[str, Any]],
) -> list[dict[str, Any]]:
    interface_map = {item.id: item for item in interfaces}
    rows: list[dict[str, Any]] = []
    for product in products:
        issues: list[str] = []
        plugin_id = str(product.api_identifier or "").strip()
        interface = interface_map.get(int(product.provision_interface_id or 0))
        if not plugin_id:
            issues.append("未选择开通插件")
        elif plugin_id not in plugin_map:
            issues.append(f"开通插件不存在：{plugin_id}")
        if not product.provision_interface_id:
            issues.append("未绑定开通接口")
        elif not interface:
            issues.append(f"开通接口不存在：{product.provision_interface_id}")
        else:
            if not interface.enabled:
                issues.append(f"开通接口已禁用：{interface.name}")
            if plugin_id and interface.plugin_id != plugin_id:
                issues.append(f"接口插件不一致：{interface.plugin_id}")
        if _enum_value(product.status) == ProductStatus.ACTIVE.value and not bool(product.auto_provision):
            issues.append("在售商品未启用自动开通")
        if issues:
            rows.append(
                {
                    "product_id": product.id,
                    "product_name": product.name,
                    "sku": product.sku,
                    "status": _enum_value(product.status),
                    "auto_provision": bool(product.auto_provision),
                    "plugin_id": plugin_id or None,
                    "interface_id": product.provision_interface_id,
                    "issues": issues,
                }
            )
    return rows


async def _plugin_products(db: AsyncSession) -> list[Product]:
    result = await db.execute(
        select(Product)
        .where(Product.api_type == ProductApiType.PLUGIN.value)
        .order_by(Product.sort_order.asc(), Product.id.desc())
    )
    return list(result.scalars().all())


async def _recent_provision_failures(
    db: AsyncSession,
) -> tuple[list[dict[str, Any]], int, dict[str, int]]:
    failed_statuses = [InstanceStatus.FAILED, InstanceStatus.PROVISION_FAILED]
    failed_count = await db.scalar(
        select(func.count())
        .select_from(Instance)
        .join(Product, Product.id == Instance.product_id)
        .where(Product.api_type == ProductApiType.PLUGIN.value, Instance.status.in_(failed_statuses))
    )
    status_rows = await db.execute(
        select(Instance.status, func.count())
        .select_from(Instance)
        .join(Product, Product.id == Instance.product_id)
        .where(Product.api_type == ProductApiType.PLUGIN.value)
        .group_by(Instance.status)
    )
    status_counts = {_enum_value(status): int(count) for status, count in status_rows.all()}
    rows = await db.execute(
        select(Instance, Product, Order, User)
        .join(Product, Product.id == Instance.product_id)
        .join(Order, Order.id == Instance.order_id)
        .join(User, User.id == Instance.user_id)
        .where(Product.api_type == ProductApiType.PLUGIN.value, Instance.status.in_(failed_statuses))
        .order_by(desc(Instance.updated_at))
        .limit(10)
    )
    failures = []
    for instance, product, order, user in rows.all():
        status = _enum_value(instance.status)
        failures.append(
            {
                "instance_id": instance.id,
                "product_id": product.id,
                "product_name": product.name,
                "plugin_id": product.api_identifier,
                "order_id": order.id,
                "out_order_no": order.out_order_no,
                "user_id": user.id,
                "username": user.username,
                "status": status,
                "status_label": INSTANCE_STATUS_LABELS.get(status, status),
                "attempts": instance.provision_attempts,
                "last_provision_at": _iso(instance.last_provision_at),
                "updated_at": _iso(instance.updated_at),
                "provision_log": _truncate(instance.provision_log, 240),
            }
        )
    return failures, int(failed_count or 0), status_counts


async def _stuck_provisioning_count(db: AsyncSession) -> int:
    cutoff = datetime.now(timezone.utc) - timedelta(minutes=30)
    count = await db.scalar(
        select(func.count())
        .select_from(Instance)
        .join(Product, Product.id == Instance.product_id)
        .where(
            Product.api_type == ProductApiType.PLUGIN.value,
            Instance.status == InstanceStatus.PROVISIONING,
            Instance.updated_at <= cutoff,
        )
    )
    return int(count or 0)


def _plugin_summary(item: dict[str, Any]) -> dict[str, Any]:
    return {
        "id": item.get("id"),
        "type": item.get("type"),
        "name": item.get("name"),
        "version": item.get("version"),
        "provider": item.get("provider"),
        "enabled": bool(item.get("enabled")),
        "current": bool(item.get("current")),
        "capabilities": item.get("capabilities") or [],
    }


def _field_is_secret(field: dict[str, Any]) -> bool:
    key = str(field.get("key") or "")
    field_type = str(field.get("type") or "").lower()
    return bool(field.get("secret")) or field_type == "password" or system_setting_service.is_secret_setting_key(key)


def _is_encrypted(value: Any) -> bool:
    try:
        decrypt_secret(str(value))
        return True
    except Exception:
        return False


def _present(value: Any) -> bool:
    if value is None:
        return False
    if isinstance(value, str):
        return value.strip() != ""
    if isinstance(value, (list, dict, set, tuple)):
        return bool(value)
    return True


def _setting_value(settings: dict[str, SystemSetting], key: str) -> Any:
    item = settings.get(key)
    return item.value if item else None


def _split_csv(value: Any) -> list[str]:
    return [item.strip() for item in str(value or "").split(",") if item.strip()]


def _enum_value(value: Any) -> str:
    return str(getattr(value, "value", value) or "")


def _iso(value: Any) -> str | None:
    if value is None:
        return None
    if hasattr(value, "isoformat"):
        return value.isoformat()
    return str(value)


def _truncate(value: Any, max_length: int) -> str | None:
    if value in (None, ""):
        return None
    text = str(value)
    return text if len(text) <= max_length else f"{text[:max_length]}..."
