import json
import asyncio
import logging
import os
import re
import shutil
import sys
import tempfile
import time
import zipfile
from importlib.metadata import PackageNotFoundError, version
from pathlib import Path
from typing import Any

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

from app.config import settings as app_settings
from app.core.crypto import decrypt_secret, encrypt_secret
from app.models.system_setting import SystemSetting
from app.services import system_setting_service


log = logging.getLogger("qycw.notification.plugins")

BACKEND_ROOT = Path(__file__).resolve().parents[2]
MODE_ROOT = BACKEND_ROOT / "mode"
MAX_PLUGIN_ZIP_BYTES = 100 * 1024 * 1024
PLUGIN_TYPES = {"email", "sms", "verification", "payment"}
_MASK = "********"
PACKAGE_PATTERN = re.compile(r"^[A-Za-z0-9_.-]+([<>=!~]=?[A-Za-z0-9_.*+-]+)?(,[<>=!~]=?[A-Za-z0-9_.*+-]+)*$")
PACKAGE_NAME_PATTERN = re.compile(r"^[A-Za-z0-9_.-]+")
VERSION_SPEC_PATTERN = re.compile(r"[<>=!~]")
VERIFICATION_MODE_PRIORITY = ("two_factor", "three_factor", "face_redirect")


def _python_executable() -> str:
    if os.name == "nt":
        local_python = BACKEND_ROOT / "venv" / "Scripts" / "python.exe"
    else:
        local_python = BACKEND_ROOT / "venv" / "bin" / "python"
    if local_python.exists():
        return str(local_python)
    return sys.executable


def _safe_read_json(path: Path) -> dict[str, Any] | None:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return None
    return data if isinstance(data, dict) else None


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


def _join_plugin_ids(items: list[str]) -> str:
    seen: set[str] = set()
    result: list[str] = []
    for item in items:
        text = str(item or "").strip()
        if text and text not in seen:
            seen.add(text)
            result.append(text)
    return ",".join(result)


def _normalize_plugin_manifest(data: dict[str, Any]) -> dict[str, Any]:
    item = dict(data)
    schema = item.get("config_schema")
    if "config_fields" not in item and isinstance(schema, dict) and isinstance(schema.get("fields"), list):
        item["config_fields"] = schema["fields"]
    requirements = item.get("requirements")
    if not isinstance(requirements, dict):
        requirements = {}
    requirements.setdefault("python_packages", [])
    requirements.setdefault("network", [])
    item["requirements"] = requirements
    runtime = str(item.get("runtime") or "").strip()
    if runtime in {"jsonrpc-stdio", "process"} and not item.get("provider"):
        item["provider"] = f"plugin:{item.get('id')}"
    item.setdefault("entry_module", "")
    item.setdefault("factory", "")
    item.setdefault("configurable", True)
    item.setdefault("config_prefix", str(item.get("type") or ""))
    item.setdefault("test_supported", False)
    item.setdefault("test", {})
    item.setdefault("docs", "")
    item.setdefault("author", {"name": "QingYun Cloud", "url": ""})
    return item


def _normalize_manifest(data: dict[str, Any], path: Path) -> dict[str, Any] | None:
    data = _normalize_plugin_manifest(data)
    plugin_type = data.get("type")
    plugin_id = data.get("id")
    provider = data.get("provider")
    if plugin_type not in PLUGIN_TYPES or not plugin_id or not provider:
        return None
    item = dict(data)
    item["manifest_path"] = str(path.relative_to(BACKEND_ROOT)).replace("\\", "/")
    return item


def _validate_manifest(data: dict[str, Any]) -> dict[str, Any]:
    data = _normalize_plugin_manifest(data)
    required = [
        "schema_version",
        "id",
        "type",
        "name",
        "version",
        "description",
        "provider",
        "entry_module",
        "factory",
        "configurable",
        "config_prefix",
        "test_supported",
        "capabilities",
        "requirements",
        "config_fields",
        "test",
        "docs",
        "author",
    ]
    missing = [key for key in required if key not in data]
    if missing:
        raise HTTPException(status_code=400, detail=f"plugin.json 缺少字段：{', '.join(missing)}")

    plugin_type = data.get("type")
    plugin_id = str(data.get("id") or "").strip()
    provider = str(data.get("provider") or "").strip()
    runtime = str(data.get("runtime") or "").strip()
    if plugin_type not in PLUGIN_TYPES:
        raise HTTPException(status_code=400, detail="插件 type 只能是 email、sms、verification 或 payment")
    if not plugin_id.replace("_", "").replace("-", "").isalnum():
        raise HTTPException(status_code=400, detail="插件 id 只能包含字母、数字、下划线和短横线")
    expected_prefix = f"mode.{plugin_type}.{plugin_id}."
    if runtime not in {"jsonrpc-stdio", "process"} and not provider.startswith(expected_prefix):
        raise HTTPException(status_code=400, detail=f"插件 provider 必须以 {expected_prefix} 开头")
    if runtime in {"jsonrpc-stdio", "process"}:
        entry = data.get("entry")
        if not isinstance(entry, dict) and not isinstance(data.get("endpoint"), str):
            raise HTTPException(status_code=400, detail="Go 插件必须提供 entry")
    fields = data.get("config_fields")
    if not isinstance(fields, list):
        raise HTTPException(status_code=400, detail="config_fields 必须是数组")
    for index, field in enumerate(fields):
        if not isinstance(field, dict):
            raise HTTPException(status_code=400, detail=f"config_fields[{index}] 必须是对象")
        for key in ["key", "name", "label", "type", "required", "secret", "default", "placeholder", "help", "validation"]:
            if key not in field:
                raise HTTPException(status_code=400, detail=f"config_fields[{index}] 缺少 {key}")
    return data


def _inspect_plugin_zip(zip_path: Path) -> tuple[dict[str, Any], str]:
    if zip_path.stat().st_size > MAX_PLUGIN_ZIP_BYTES:
        raise HTTPException(status_code=400, detail="插件包不能超过 5MB")
    try:
        with zipfile.ZipFile(zip_path) as archive:
            names = [name for name in archive.namelist() if name and not name.endswith("/")]
            if not names:
                raise HTTPException(status_code=400, detail="插件包为空")
            for name in names:
                pure = Path(name)
                if pure.is_absolute() or ".." in pure.parts:
                    raise HTTPException(status_code=400, detail="插件包包含非法路径")
            if "plugin.json" in archive.namelist():
                root = ""
                manifest_name = "plugin.json"
            else:
                roots = {Path(name).parts[0] for name in names if Path(name).parts}
                if len(roots) != 1:
                    raise HTTPException(status_code=400, detail="插件包必须包含唯一根目录")
                root = roots.pop()
                manifest_name = f"{root}/plugin.json"
            if manifest_name not in archive.namelist():
                raise HTTPException(status_code=400, detail="插件包根目录缺少 plugin.json")
            try:
                manifest = json.loads(archive.read(manifest_name).decode("utf-8"))
            except Exception as exc:
                raise HTTPException(status_code=400, detail="plugin.json 不是有效 JSON") from exc
    except zipfile.BadZipFile as exc:
        raise HTTPException(status_code=400, detail="插件包不是有效 zip 文件") from exc

    if not isinstance(manifest, dict):
        raise HTTPException(status_code=400, detail="plugin.json 必须是对象")
    manifest = _validate_manifest(manifest)
    if root and str(manifest["id"]) != root:
        raise HTTPException(status_code=400, detail="插件包根目录名必须等于 plugin.json 的 id")
    return manifest, root


def list_plugins(plugin_type: str | None = None) -> list[dict[str, Any]]:
    roots: list[Path]
    if plugin_type in PLUGIN_TYPES:
        roots = [MODE_ROOT / plugin_type]
    else:
        roots = [MODE_ROOT / "email", MODE_ROOT / "sms", MODE_ROOT / "verification", MODE_ROOT / "payment"]

    items: list[dict[str, Any]] = []
    for root in roots:
        if not root.exists():
            continue
        for path in root.glob("*/plugin.json"):
            data = _safe_read_json(path)
            if not data:
                continue
            item = _normalize_manifest(data, path)
            if item:
                items.append(item)
    return sorted(items, key=lambda item: (str(item.get("type")), str(item.get("name"))))


def preview_plugin_zip(zip_path: Path) -> dict[str, Any]:
    manifest, root = _inspect_plugin_zip(zip_path)
    existing = any(item.get("id") == manifest["id"] and item.get("type") == manifest["type"] for item in list_plugins(str(manifest["type"])))
    data = dict(manifest)
    data["root_dir"] = root
    data["exists"] = existing
    data["install_path"] = str((MODE_ROOT / str(manifest["type"]) / str(manifest["id"])).relative_to(BACKEND_ROOT)).replace("\\", "/")
    return data


def install_plugin_zip(zip_path: Path, *, overwrite: bool = False, operator_id: int | None = None) -> dict[str, Any]:
    manifest, root = _inspect_plugin_zip(zip_path)
    plugin_type = str(manifest["type"])
    plugin_id = str(manifest["id"])
    target = (MODE_ROOT / plugin_type / plugin_id).resolve()
    allowed_root = (MODE_ROOT / plugin_type).resolve()
    if allowed_root not in target.parents:
        raise HTTPException(status_code=400, detail="插件安装路径无效")

    existing = next((item for item in list_plugins(plugin_type) if item.get("id") == plugin_id), None)
    if existing:
        if existing.get("protected"):
            raise HTTPException(status_code=400, detail="不能覆盖内置插件")
        if not overwrite:
            raise HTTPException(status_code=409, detail="插件已存在，如需覆盖请开启 overwrite")

    with tempfile.TemporaryDirectory(prefix="qycw_plugin_") as tmp:
        tmp_path = Path(tmp)
        with zipfile.ZipFile(zip_path) as archive:
            archive.extractall(tmp_path)
        source = (tmp_path / root).resolve() if root else tmp_path.resolve()
        if not source.is_dir():
            raise HTTPException(status_code=400, detail="插件目录无效")
        if target.exists():
            shutil.rmtree(target)
        shutil.copytree(source, target)

    log.warning(
        "notification plugin installed: type=%s id=%s overwrite=%s operator_id=%s",
        plugin_type,
        plugin_id,
        overwrite,
        operator_id,
    )
    return get_plugin(plugin_type, plugin_id)


def get_plugin(plugin_type: str, plugin_id: str) -> dict[str, Any]:
    if plugin_type not in PLUGIN_TYPES:
        raise HTTPException(status_code=400, detail="插件类型无效")
    for item in list_plugins(plugin_type):
        if item.get("id") == plugin_id:
            return item
    raise HTTPException(status_code=404, detail="插件不存在")


async def list_plugins_with_state(db: AsyncSession, plugin_type: str | None = None) -> list[dict[str, Any]]:
    settings = await system_setting_service.ensure_system_settings(db)
    email_provider = str(settings["notification.email.provider"].value)
    email_enabled = bool(settings["notification.email.enabled"].value)
    sms_provider = str(settings["notification.sms.provider"].value)
    sms_enabled = bool(settings["notification.sms.enabled"].value)
    verification_provider = str(settings["verification.provider"].value)
    verification_enabled = bool(settings["verification.enabled"].value)
    payment_enabled_ids = set(_split_plugin_ids(settings["payment.enabled_plugins"].value))

    rows = []
    for item in list_plugins(plugin_type):
        provider = str(item.get("provider"))
        current = False
        enabled = False
        if item.get("type") == "email":
            current = provider == email_provider
            enabled = current and email_enabled
        if item.get("type") == "sms":
            current = provider == sms_provider
            enabled = current and sms_enabled
        if item.get("type") == "verification":
            current = provider == verification_provider
            enabled = current and verification_enabled
        if item.get("type") == "payment":
            current = str(item.get("id")) in payment_enabled_ids
            enabled = current
        row = dict(item)
        row["current"] = current
        row["enabled"] = enabled
        rows.append(row)
    return rows


def _coerce_field_value(field: dict[str, Any], value: Any) -> Any:
    field_type = field.get("type")
    if field_type == "boolean":
        if isinstance(value, bool):
            return value
        if isinstance(value, str):
            return value.strip().lower() in {"1", "true", "yes", "on", "enabled"}
        return bool(value)
    if field_type == "integer":
        try:
            number = int(value)
        except (TypeError, ValueError):
            number = int(field.get("default") or 0)
        validation = field.get("validation") if isinstance(field.get("validation"), dict) else {}
        if "min" in validation:
            number = max(int(validation["min"]), number)
        if "max" in validation:
            number = min(int(validation["max"]), number)
        return number
    if value is None:
        return field.get("default", "")
    text = str(value).strip()
    validation = field.get("validation") if isinstance(field.get("validation"), dict) else {}
    max_length = validation.get("max_length")
    if max_length is not None:
        text = text[: int(max_length)]
    return text


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 _decrypt_field_value(field: dict[str, Any], value: Any) -> Any:
    if not _field_is_secret(field) or value in (None, ""):
        return value
    try:
        return decrypt_secret(str(value))
    except Exception:
        return value


def _stored_field_value(field: dict[str, Any], value: Any, current_value: Any = None) -> Any:
    if system_setting_service.is_masked_secret_value(value) and current_value not in (None, ""):
        return current_value
    coerced = _coerce_field_value(field, value)
    if not _field_is_secret(field) or coerced in (None, ""):
        return coerced
    try:
        decrypt_secret(str(coerced))
        return coerced
    except Exception:
        return encrypt_secret(str(coerced))


def _public_field_value(field: dict[str, Any], value: Any) -> Any:
    plain = _decrypt_field_value(field, value)
    if _field_is_secret(field) and plain not in (None, ""):
        return _MASK
    return plain


async def get_plugin_config(db: AsyncSession, plugin_type: str, plugin_id: str) -> dict[str, Any]:
    plugin = get_plugin(plugin_type, plugin_id)
    await system_setting_service.ensure_system_settings(db)
    keys = [field.get("key") for field in plugin.get("config_fields", []) if field.get("key")]
    existing = {}
    if keys:
        result = await db.execute(select(SystemSetting).where(SystemSetting.key.in_(keys)))
        existing = {item.key: item for item in result.scalars().all()}
    values = {}
    for field in plugin.get("config_fields", []):
        key = field.get("key")
        if not key:
            continue
        if key in existing:
            raw_value = existing[key].value
            stored_value = _stored_field_value(field, raw_value)
            if _field_is_secret(field) and stored_value != raw_value:
                existing[key].value = stored_value
                raw_value = stored_value
            values[key] = _public_field_value(field, raw_value)
        else:
            values[key] = _public_field_value(field, field.get("default"))
    await db.flush()
    return {"plugin": plugin, "values": values}


async def save_plugin_config(
    db: AsyncSession,
    plugin_type: str,
    plugin_id: str,
    values: dict[str, Any],
) -> dict[str, Any]:
    plugin = get_plugin(plugin_type, plugin_id)
    fields = {field.get("key"): field for field in plugin.get("config_fields", []) if field.get("key")}
    unknown = sorted(set(values) - set(fields))
    if unknown:
        raise HTTPException(status_code=400, detail=f"未知插件配置项：{', '.join(unknown)}")

    await system_setting_service.ensure_system_settings(db)
    existing = {}
    if fields:
        result = await db.execute(select(SystemSetting).where(SystemSetting.key.in_(fields)))
        existing = {item.key: item for item in result.scalars().all()}

    for key, value in values.items():
        field = fields[key]
        current_value = existing[key].value if key in existing else None
        coerced = _stored_field_value(field, value, current_value)
        item = existing.get(key)
        if not item:
            item = SystemSetting(
                key=key,
                value=coerced,
                label=str(field.get("label") or key),
                description=str(field.get("help") or ""),
            )
            db.add(item)
        else:
            item.value = coerced
            item.label = str(field.get("label") or key)
            item.description = str(field.get("help") or "")

    await db.flush()
    return await get_plugin_config(db, plugin_type, plugin_id)


async def enable_plugin(db: AsyncSession, plugin_type: str, plugin_id: str) -> dict[str, Any]:
    plugin = get_plugin(plugin_type, plugin_id)
    if plugin_type == "email":
        await system_setting_service.update_feature_settings(
            db,
            {
                "notification.email.provider": str(plugin["provider"]),
                "notification.email.enabled": True,
            },
        )
    elif plugin_type == "sms":
        await system_setting_service.update_feature_settings(
            db,
            {
                "notification.sms.provider": str(plugin["provider"]),
                "notification.sms.enabled": True,
            },
        )
    elif plugin_type == "verification":
        capabilities = set(plugin.get("capabilities") or [])
        default_mode = next((mode for mode in VERIFICATION_MODE_PRIORITY if mode in capabilities), "two_factor")
        await system_setting_service.update_feature_settings(
            db,
            {
                "verification.provider": str(plugin["provider"]),
                "verification.enabled": True,
                "verification.default_mode": default_mode,
            },
        )
    else:
        settings = await system_setting_service.ensure_system_settings(db)
        enabled = _split_plugin_ids(settings["payment.enabled_plugins"].value)
        await system_setting_service.update_feature_settings(
            db,
            {"payment.enabled_plugins": _join_plugin_ids([*enabled, str(plugin["id"])])},
        )
    await db.flush()
    return {"enabled": True, "plugin": plugin}


async def disable_plugin(db: AsyncSession, plugin_type: str, plugin_id: str) -> dict[str, Any]:
    plugin = get_plugin(plugin_type, plugin_id)
    settings = await system_setting_service.ensure_system_settings(db)
    if plugin_type == "email" and str(settings["notification.email.provider"].value) == str(plugin["provider"]):
        settings["notification.email.enabled"].value = False
    if plugin_type == "sms" and str(settings["notification.sms.provider"].value) == str(plugin["provider"]):
        settings["notification.sms.enabled"].value = False
    if plugin_type == "verification" and str(settings["verification.provider"].value) == str(plugin["provider"]):
        settings["verification.enabled"].value = False
    if plugin_type == "payment":
        enabled = [item for item in _split_plugin_ids(settings["payment.enabled_plugins"].value) if item != str(plugin["id"])]
        settings["payment.enabled_plugins"].value = _join_plugin_ids(enabled)
    await db.flush()
    return {"enabled": False, "plugin": plugin}


async def delete_plugin(db: AsyncSession, plugin_type: str, plugin_id: str) -> dict[str, Any]:
    plugin = get_plugin(plugin_type, plugin_id)
    if plugin.get("protected"):
        raise HTTPException(status_code=400, detail="内置插件不允许删除")

    settings = await system_setting_service.ensure_system_settings(db)
    if plugin_type == "verification":
        provider_key = "verification.provider"
        enabled_key = "verification.enabled"
    elif plugin_type == "payment":
        if str(plugin["id"]) in set(_split_plugin_ids(settings["payment.enabled_plugins"].value)):
            raise HTTPException(status_code=400, detail="请先禁用该插件再删除")
        provider_key = ""
        enabled_key = ""
    else:
        provider_key = f"notification.{plugin_type}.provider"
        enabled_key = f"notification.{plugin_type}.enabled"
    if provider_key and str(settings[provider_key].value) == str(plugin["provider"]) and bool(settings[enabled_key].value):
        raise HTTPException(status_code=400, detail="请先禁用该插件再删除")

    target = (MODE_ROOT / plugin_type / plugin_id).resolve()
    allowed_root = (MODE_ROOT / plugin_type).resolve()
    if allowed_root not in target.parents or not target.is_dir():
        raise HTTPException(status_code=400, detail="插件目录无效")

    shutil.rmtree(target)
    return {"deleted": True, "plugin_id": plugin_id, "type": plugin_type}


def _collect_python_packages() -> list[str]:
    packages: set[str] = set()
    for plugin in list_plugins():
        requirements = plugin.get("requirements")
        if not isinstance(requirements, dict):
            continue
        python_packages = requirements.get("python_packages")
        if not isinstance(python_packages, list):
            continue
        for item in python_packages:
            package = str(item or "").strip()
            if not package:
                continue
            if not PACKAGE_PATTERN.match(package):
                raise HTTPException(status_code=400, detail=f"插件依赖格式不安全：{package}")
            packages.add(package)
    return sorted(packages)


def _missing_python_packages(packages: list[str]) -> list[str]:
    missing: list[str] = []
    for package in packages:
        match = PACKAGE_NAME_PATTERN.match(package)
        if not match:
            missing.append(package)
            continue
        if VERSION_SPEC_PATTERN.search(package):
            missing.append(package)
            continue
        package_name = match.group(0)
        try:
            version(package_name)
        except PackageNotFoundError:
            missing.append(package)
    return missing


async def _run_command(command: list[str] | str, *, shell: bool = False, timeout: int = 300) -> dict[str, Any]:
    try:
        if shell:
            proc = await asyncio.create_subprocess_shell(
                str(command),
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
        else:
            proc = await asyncio.create_subprocess_exec(
                *command,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
    except OSError as exc:
        raise HTTPException(status_code=500, detail=f"命令启动失败：{exc}") from exc

    try:
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
    except asyncio.TimeoutError as exc:
        proc.kill()
        await proc.communicate()
        raise HTTPException(status_code=500, detail="命令执行超时") from exc

    return {
        "returncode": proc.returncode,
        "stdout": stdout.decode("utf-8", errors="replace")[-4000:],
        "stderr": stderr.decode("utf-8", errors="replace")[-4000:],
    }


async def apply_plugin_changes(
    *,
    install_dependencies: bool = True,
    restart_backend: bool = True,
    operator_id: int | None = None,
) -> dict[str, Any]:
    result: dict[str, Any] = {
        "dependencies": {"installed": False, "packages": [], "pending_packages": [], "python": _python_executable(), "result": None},
        "restart": {"requested": bool(restart_backend), "mode": "none", "result": None},
    }

    if install_dependencies:
        packages = _collect_python_packages()
        result["dependencies"]["packages"] = packages
        pending_packages = _missing_python_packages(packages)
        result["dependencies"]["pending_packages"] = pending_packages
        if pending_packages:
            pip_result = await _run_command([_python_executable(), "-m", "pip", "install", *pending_packages], timeout=600)
            result["dependencies"]["installed"] = pip_result["returncode"] == 0
            result["dependencies"]["result"] = pip_result
            if pip_result["returncode"] != 0:
                message = pip_result["stderr"] or pip_result["stdout"] or "插件依赖安装失败"
                raise HTTPException(status_code=500, detail=f"插件依赖安装失败：{message[-800:]}")
        else:
            result["dependencies"]["installed"] = True

    if restart_backend:
        command = app_settings.PLUGIN_RELOAD_COMMAND.strip()
        if command:
            reload_result = await _run_command(command, shell=True, timeout=120)
            result["restart"]["mode"] = "command"
            result["restart"]["result"] = reload_result
            if reload_result["returncode"] != 0:
                raise HTTPException(status_code=500, detail="后端重载命令执行失败")
        else:
            marker = BACKEND_ROOT / "app" / "plugin_reload_marker.py"
            marker.write_text(f'PLUGIN_RELOAD_TS = {time.time()!r}\n', encoding="utf-8")
            result["restart"]["mode"] = "reload_marker"
            result["restart"]["result"] = {
                "path": str(marker.relative_to(BACKEND_ROOT)).replace("\\", "/"),
                "message": "已更新 reload marker；开发环境 uvicorn --reload 会自动重载，生产环境请配置 PLUGIN_RELOAD_COMMAND。",
            }

    log.warning(
        "notification plugin changes applied: install_dependencies=%s restart_backend=%s operator_id=%s",
        install_dependencies,
        restart_backend,
        operator_id,
    )
    result["operator_id"] = operator_id
    return result

