from __future__ import annotations

import json
from dataclasses import asdict, is_dataclass
from pathlib import Path
from typing import Any

from fastapi import HTTPException

from app.config import settings
from app.services.provider_loader import _load_factory


PLUGIN_ROOT = Path(__file__).resolve().parents[2] / "mode" / "provision"
PUBLIC_PAGE_KEYS = {
    "tabs",
    "layout",
    "summary_html",
    "empty_state_html",
    "purchase_fields",
}
PUBLIC_PLUGIN_KEYS = {
    "schema_version",
    "id",
    "type",
    "name",
    "version",
    "status",
    "description",
    "capabilities",
    "manage_actions",
    "test_links",
    "product_page",
    "manage_page",
    "upstream_param_fields",
}
ADMIN_PLUGIN_KEYS = {
    *PUBLIC_PLUGIN_KEYS,
    "provider",
    "config_fields",
    "docs",
    "requirements",
    "author",
}


def _read_json(path: Path) -> dict[str, Any]:
    try:
        with path.open("r", encoding="utf-8") as fh:
            data = json.load(fh)
    except FileNotFoundError:
        raise HTTPException(status_code=404, detail="开通插件不存在")
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"invalid provision plugin manifest: {path}: {exc}") from exc
    if not isinstance(data, dict):
        raise RuntimeError(f"provision plugin manifest must be an object: {path}")
    return data


def _manifest_path(plugin_id: str) -> Path:
    if not plugin_id or not plugin_id.replace("_", "").replace("-", "").isalnum():
        raise HTTPException(status_code=400, detail="插件 ID 格式不正确")
    path = (PLUGIN_ROOT / plugin_id / "plugin.json").resolve()
    root = PLUGIN_ROOT.resolve()
    if root not in path.parents:
        raise HTTPException(status_code=400, detail="插件 ID 格式不正确")
    return path


def _read_plugin_file(plugin_dir: Path, relative_path: str) -> str:
    if not relative_path or Path(relative_path).is_absolute():
        raise RuntimeError("plugin html_file must be a relative path")
    path = (plugin_dir / relative_path).resolve()
    root = plugin_dir.resolve()
    if root not in path.parents:
        raise RuntimeError("plugin html_file escapes plugin directory")
    try:
        return path.read_text(encoding="utf-8")
    except FileNotFoundError as exc:
        raise RuntimeError(f"plugin html_file not found: {relative_path}") from exc


def _normalize_page(manifest: dict[str, Any], key: str, plugin_dir: Path | None) -> dict[str, Any]:
    page = manifest.get(key)
    if not isinstance(page, dict):
        return {"layout": "tabs", "tabs": [], "purchase_fields": []}

    normalized = dict(page)
    if plugin_dir:
        if normalized.get("summary_html_file") and not normalized.get("summary_html"):
            normalized["summary_html"] = _read_plugin_file(plugin_dir, str(normalized["summary_html_file"]))
        if normalized.get("empty_state_html_file") and not normalized.get("empty_state_html"):
            normalized["empty_state_html"] = _read_plugin_file(plugin_dir, str(normalized["empty_state_html_file"]))

    tabs: list[dict[str, Any]] = []
    for raw_tab in normalized.get("tabs") or []:
        if not isinstance(raw_tab, dict):
            continue
        tab = dict(raw_tab)
        if plugin_dir and tab.get("html_file") and not tab.get("html"):
            tab["html"] = _read_plugin_file(plugin_dir, str(tab["html_file"]))
        tab.setdefault("html", "")
        if tab.get("key") and tab.get("label"):
            tabs.append(tab)
    normalized["tabs"] = tabs

    fields: list[dict[str, Any]] = []
    for raw_field in normalized.get("purchase_fields") or []:
        if not isinstance(raw_field, dict):
            continue
        field = dict(raw_field)
        if field.get("key") and field.get("label"):
            field.setdefault("type", "text")
            field.setdefault("required", False)
            fields.append(field)
    normalized["purchase_fields"] = fields
    normalized.setdefault("layout", "tabs")
    return normalized


def _normalize_manifest(
    data: dict[str, Any],
    fallback_id: str | None = None,
    plugin_dir: Path | None = None,
) -> dict[str, Any]:
    manifest = dict(data)
    plugin_id = str(manifest.get("id") or fallback_id or "").strip()
    if not plugin_id:
        raise RuntimeError("provision plugin missing id")
    if manifest.get("type") != "provision":
        raise RuntimeError(f"plugin {plugin_id} is not a provision plugin")
    manifest["id"] = plugin_id
    manifest.setdefault("name", plugin_id)
    manifest.setdefault("version", "1.0.0")
    manifest.setdefault("capabilities", [])
    manifest.setdefault("manage_actions", [])
    manifest.setdefault("test_links", [])
    manifest["product_page"] = _normalize_page(manifest, "product_page", plugin_dir)
    if "manage_page" in manifest and isinstance(manifest.get("manage_page"), dict):
        manifest["manage_page"] = _normalize_page(manifest, "manage_page", plugin_dir)
    return manifest


def list_plugins() -> list[dict[str, Any]]:
    if not PLUGIN_ROOT.exists():
        return []
    plugins: list[dict[str, Any]] = []
    for path in sorted(PLUGIN_ROOT.glob("*/plugin.json")):
        try:
            plugins.append(_normalize_manifest(_read_json(path), path.parent.name, path.parent))
        except Exception:
            continue
    return plugins


def get_plugin_manifest(plugin_id: str) -> dict[str, Any]:
    path = _manifest_path(plugin_id)
    return _normalize_manifest(_read_json(path), plugin_id, path.parent)


def public_manifest(manifest: dict[str, Any]) -> dict[str, Any]:
    data = {key: manifest.get(key) for key in PUBLIC_PLUGIN_KEYS if key in manifest}
    page = data.get("product_page")
    if isinstance(page, dict):
        data["product_page"] = {key: page.get(key) for key in PUBLIC_PAGE_KEYS if key in page}
    else:
        data["product_page"] = {"tabs": []}
    return data


def admin_manifest(manifest: dict[str, Any]) -> dict[str, Any]:
    data = {key: manifest.get(key) for key in ADMIN_PLUGIN_KEYS if key in manifest}
    page = data.get("product_page")
    if isinstance(page, dict):
        data["product_page"] = {key: page.get(key) for key in PUBLIC_PAGE_KEYS if key in page}
    else:
        data["product_page"] = {"tabs": [], "purchase_fields": []}
    return data


def create_plugin_provider(plugin_id: str):
    manifest = get_plugin_manifest(plugin_id)
    provider_path = str(manifest.get("provider") or "").strip()
    if not provider_path:
        raise RuntimeError(f"provision plugin {plugin_id} missing provider")
    return _load_factory(provider_path)(settings)


async def test_plugin(plugin_id: str, config: dict[str, Any] | None = None) -> dict[str, Any]:
    manifest = get_plugin_manifest(plugin_id)
    provider = create_plugin_provider(plugin_id)
    test = getattr(provider, "test", None)
    if not callable(test):
        return {
            "ok": True,
            "message": "插件未实现 test()，仅完成 manifest 加载测试。",
            "plugin": admin_manifest(manifest),
        }
    result = test(config or {})
    if hasattr(result, "__await__"):
        result = await result
    if is_dataclass(result):
        result = asdict(result)
    if not isinstance(result, dict):
        result = {"ok": bool(result), "message": str(result)}
    result.setdefault("plugin", admin_manifest(manifest))
    return result


def ensure_plugin_exists(plugin_id: str) -> None:
    get_plugin_manifest(plugin_id)


def plugin_supports_action(manifest: dict[str, Any], action: str) -> bool:
    for item in manifest.get("manage_actions") or []:
        if isinstance(item, dict) and item.get("key") == action:
            return True
        if item == action:
            return True
    return False


def _present(value: Any) -> bool:
    return value is not None and str(value).strip() != ""


def validate_purchase_fields(plugin_id: str, slider_values: dict[str, Any] | None) -> dict[str, Any]:
    manifest = get_plugin_manifest(plugin_id)
    page = manifest.get("product_page") if isinstance(manifest.get("product_page"), dict) else {}
    fields = page.get("purchase_fields") if isinstance(page.get("purchase_fields"), list) else []
    if not fields:
        return {}

    values = slider_values if isinstance(slider_values, dict) else {}
    plugin_fields = values.get("__plugin_fields") if isinstance(values.get("__plugin_fields"), dict) else {}
    missing: list[str] = []
    invalid: list[str] = []

    for field in fields:
        if not isinstance(field, dict):
            continue
        key = str(field.get("key") or "").strip()
        if not key:
            continue
        value = plugin_fields.get(key)
        if field.get("required") and not _present(value):
            missing.append(str(field.get("label") or key))
            continue
        if not _present(value):
            continue
        if field.get("type") == "select":
            options = field.get("options") if isinstance(field.get("options"), list) else []
            allowed = {str(option.get("value")) for option in options if isinstance(option, dict) and option.get("value") is not None}
            if allowed and str(value) not in allowed:
                invalid.append(str(field.get("label") or key))

    if missing:
        raise HTTPException(status_code=400, detail=f"插件购买字段缺失：{', '.join(missing)}")
    if invalid:
        raise HTTPException(status_code=400, detail=f"插件购买字段无效：{', '.join(invalid)}")
    return dict(plugin_fields)
