import json
import os
import re
from typing import Any, Dict, List

from sqlmodel import Session

from core.settings import (
    DEFAULT_CHAT_PULL_PAGE_SIZE,
    DEFAULT_COMMAND_AUTO_UNLOCK_SECONDS,
    DEFAULT_PAGE_SIZE,
    DEFAULT_STT_AUDIO_FILTER,
    DEFAULT_STT_AUDIO_PREPROCESS,
    DEFAULT_STT_DEFAULT_LANGUAGE,
    DEFAULT_STT_FORCE_SIMPLIFIED,
    DEFAULT_STT_INITIAL_PROMPT,
    DEFAULT_STT_MAX_AUDIO_SECONDS,
    DEFAULT_UPLOAD_MAX_MB,
    DEFAULT_WORKSPACE_DOWNLOAD_EXTENSIONS,
    STT_DEVICE,
    STT_ENABLED_DEFAULT,
    STT_MODEL,
)
from models.platform import PlatformSetting
from schemas.platform import LoadingPageSettings, PlatformSettingsPayload, SystemSettingItem

from services.platform_common import utcnow

DEFAULT_ALLOWED_ATTACHMENT_EXTENSIONS: tuple[str, ...] = ()
DEFAULT_ACTIVITY_EVENT_RETENTION_DAYS = 7
ACTIVITY_EVENT_RETENTION_SETTING_KEY = "activity_event_retention_days"
SYS_AUTH_TOKEN_TTL_DAYS_SETTING_KEY = "sys_auth_token_ttl_days"
SETTING_KEYS = (
    "page_size",
    "chat_pull_page_size",
    "command_auto_unlock_seconds",
    "upload_max_mb",
    "allowed_attachment_extensions",
    "workspace_download_extensions",
    "speech_enabled",
)
PROTECTED_SETTING_KEYS = set(SETTING_KEYS) | {
    ACTIVITY_EVENT_RETENTION_SETTING_KEY,
}
DEPRECATED_SETTING_KEYS = {
    "loading_page",
    "speech_max_audio_seconds",
    "speech_default_language",
    "speech_force_simplified",
    "speech_audio_preprocess",
    "speech_audio_filter",
    "speech_initial_prompt",
    "dashboard_activity_page_size",
}
SYSTEM_SETTING_DEFINITIONS: Dict[str, Dict[str, Any]] = {
    "page_size": {
        "name": "分页大小",
        "category": "ui",
        "description": "平台各类列表默认每页条数。",
        "value_type": "integer",
        "value": DEFAULT_PAGE_SIZE,
        "is_public": True,
        "sort_order": 5,
    },
    "chat_pull_page_size": {
        "name": "对话懒加载条数",
        "category": "chat",
        "description": "Bot 对话区向上懒加载时每次读取的消息条数。",
        "value_type": "integer",
        "value": DEFAULT_CHAT_PULL_PAGE_SIZE,
        "is_public": True,
        "sort_order": 8,
    },
    "command_auto_unlock_seconds": {
        "name": "发送按钮自动恢复秒数",
        "category": "chat",
        "description": "对话发送后按钮保持停止态的最长秒数，超时后自动恢复为可发送状态。",
        "value_type": "integer",
        "value": DEFAULT_COMMAND_AUTO_UNLOCK_SECONDS,
        "is_public": True,
        "sort_order": 9,
    },
    "upload_max_mb": {
        "name": "上传大小限制",
        "category": "upload",
        "description": "单文件上传大小限制，单位 MB。",
        "value_type": "integer",
        "value": DEFAULT_UPLOAD_MAX_MB,
        "is_public": False,
        "sort_order": 10,
    },
    "allowed_attachment_extensions": {
        "name": "允许附件后缀",
        "category": "upload",
        "description": "允许上传的附件后缀列表，留空表示不限制。",
        "value_type": "json",
        "value": list(DEFAULT_ALLOWED_ATTACHMENT_EXTENSIONS),
        "is_public": False,
        "sort_order": 20,
    },
    "workspace_download_extensions": {
        "name": "工作区下载后缀",
        "category": "workspace",
        "description": "命中后缀的工作区文件默认走下载模式。",
        "value_type": "json",
        "value": list(DEFAULT_WORKSPACE_DOWNLOAD_EXTENSIONS),
        "is_public": False,
        "sort_order": 30,
    },
    "speech_enabled": {
        "name": "语音识别开关",
        "category": "speech",
        "description": "控制 Bot 语音转写功能是否启用。",
        "value_type": "boolean",
        "value": STT_ENABLED_DEFAULT,
        "is_public": True,
        "sort_order": 32,
    },
    ACTIVITY_EVENT_RETENTION_SETTING_KEY: {
        "name": "活动事件保留天数",
        "category": "maintenance",
        "description": "bot_activity_event 运维事件的保留天数，超期记录会自动清理。",
        "value_type": "integer",
        "value": DEFAULT_ACTIVITY_EVENT_RETENTION_DAYS,
        "is_public": False,
        "sort_order": 34,
    },
    SYS_AUTH_TOKEN_TTL_DAYS_SETTING_KEY: {
        "name": "登录令牌有效天数",
        "category": "auth",
        "description": "用户登录 JWT 的失效天数，默认 7 天，同时作为 Redis 会话 TTL。",
        "value_type": "integer",
        "value": 7,
        "is_public": False,
        "sort_order": 36,
    },
}


def normalize_activity_event_retention_days(raw: Any) -> int:
    try:
        value = int(raw)
    except Exception:
        value = DEFAULT_ACTIVITY_EVENT_RETENTION_DAYS
    return max(1, min(3650, value))


def normalize_extension(raw: Any) -> str:
    text = str(raw or "").strip().lower()
    if not text:
        return ""
    if text.startswith("*."):
        text = text[1:]
    if not text.startswith("."):
        text = f".{text}"
    if not re.fullmatch(r"\.[a-z0-9][a-z0-9._+-]{0,31}", text):
        return ""
    return text


def normalize_extension_list(rows: Any) -> List[str]:
    if not isinstance(rows, list):
        return []
    normalized: List[str] = []
    for item in rows:
        ext = normalize_extension(item)
        if ext and ext not in normalized:
            normalized.append(ext)
    return normalized


def legacy_env_int(name: str, default: int, min_value: int, max_value: int) -> int:
    raw = os.getenv(name)
    if raw is None:
        return default
    try:
        value = int(str(raw).strip())
    except Exception:
        value = default
    return max(min_value, min(max_value, value))


def legacy_env_bool(name: str, default: bool) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return str(raw).strip().lower() in {"1", "true", "yes", "on"}


def legacy_env_extensions(name: str, default: List[str]) -> List[str]:
    raw = os.getenv(name)
    if raw is None:
        return list(default)
    source = re.split(r"[,;\s]+", str(raw))
    normalized: List[str] = []
    for item in source:
        ext = normalize_extension(item)
        if ext and ext not in normalized:
            normalized.append(ext)
    return normalized


def bootstrap_platform_setting_values() -> Dict[str, Any]:
    return {
        "page_size": legacy_env_int("PAGE_SIZE", DEFAULT_PAGE_SIZE, 1, 100),
        "chat_pull_page_size": legacy_env_int(
            "CHAT_PULL_PAGE_SIZE",
            DEFAULT_CHAT_PULL_PAGE_SIZE,
            10,
            500,
        ),
        "command_auto_unlock_seconds": legacy_env_int(
            "COMMAND_AUTO_UNLOCK_SECONDS",
            DEFAULT_COMMAND_AUTO_UNLOCK_SECONDS,
            1,
            600,
        ),
        "upload_max_mb": legacy_env_int("UPLOAD_MAX_MB", DEFAULT_UPLOAD_MAX_MB, 1, 2048),
        "allowed_attachment_extensions": legacy_env_extensions(
            "ALLOWED_ATTACHMENT_EXTENSIONS",
            list(DEFAULT_ALLOWED_ATTACHMENT_EXTENSIONS),
        ),
        "workspace_download_extensions": legacy_env_extensions(
            "WORKSPACE_DOWNLOAD_EXTENSIONS",
            list(DEFAULT_WORKSPACE_DOWNLOAD_EXTENSIONS),
        ),
        "speech_enabled": legacy_env_bool("STT_ENABLED", STT_ENABLED_DEFAULT),
    }


def default_platform_settings() -> PlatformSettingsPayload:
    bootstrap = bootstrap_platform_setting_values()
    return PlatformSettingsPayload(
        page_size=int(bootstrap["page_size"]),
        chat_pull_page_size=int(bootstrap["chat_pull_page_size"]),
        command_auto_unlock_seconds=int(bootstrap["command_auto_unlock_seconds"]),
        upload_max_mb=int(bootstrap["upload_max_mb"]),
        allowed_attachment_extensions=list(bootstrap["allowed_attachment_extensions"]),
        workspace_download_extensions=list(bootstrap["workspace_download_extensions"]),
        speech_enabled=bool(bootstrap["speech_enabled"]),
        speech_max_audio_seconds=DEFAULT_STT_MAX_AUDIO_SECONDS,
        speech_default_language=DEFAULT_STT_DEFAULT_LANGUAGE,
        speech_force_simplified=DEFAULT_STT_FORCE_SIMPLIFIED,
        speech_audio_preprocess=DEFAULT_STT_AUDIO_PREPROCESS,
        speech_audio_filter=DEFAULT_STT_AUDIO_FILTER,
        speech_initial_prompt=DEFAULT_STT_INITIAL_PROMPT,
        loading_page=LoadingPageSettings(),
    )


def normalize_setting_key(raw: Any) -> str:
    text = str(raw or "").strip()
    return re.sub(r"[^a-zA-Z0-9_.-]+", "_", text).strip("._-").lower()


def normalize_setting_value(value: Any, value_type: str) -> Any:
    normalized_type = str(value_type or "json").strip().lower() or "json"
    if normalized_type == "integer":
        return int(value or 0)
    if normalized_type == "float":
        return float(value or 0)
    if normalized_type == "boolean":
        if isinstance(value, bool):
            return value
        return str(value or "").strip().lower() in {"1", "true", "yes", "on"}
    if normalized_type == "string":
        return str(value or "")
    if normalized_type == "json":
        return value
    raise ValueError(f"Unsupported value_type: {normalized_type}")


def read_setting_value(row: PlatformSetting) -> Any:
    try:
        value = json.loads(row.value_json or "null")
    except Exception:
        value = None
    return normalize_setting_value(value, row.value_type)


def setting_item_from_row(row: PlatformSetting) -> Dict[str, Any]:
    return SystemSettingItem(
        key=row.key,
        name=row.name,
        category=row.category,
        description=row.description,
        value_type=row.value_type,
        value=read_setting_value(row),
        is_public=bool(row.is_public),
        sort_order=int(row.sort_order or 100),
        created_at=row.created_at.isoformat() + "Z",
        updated_at=row.updated_at.isoformat() + "Z",
    ).model_dump()


def upsert_setting_row(
    session: Session,
    key: str,
    *,
    name: str,
    category: str,
    description: str,
    value_type: str,
    value: Any,
    is_public: bool,
    sort_order: int,
) -> PlatformSetting:
    normalized_key = normalize_setting_key(key)
    if not normalized_key:
        raise ValueError("Setting key is required")
    normalized_type = str(value_type or "json").strip().lower() or "json"
    normalized_value = normalize_setting_value(value, normalized_type)
    now = utcnow()
    row = session.get(PlatformSetting, normalized_key)
    if row is None:
        row = PlatformSetting(
            key=normalized_key,
            name=str(name or normalized_key),
            category=str(category or "general"),
            description=str(description or ""),
            value_type=normalized_type,
            value_json=json.dumps(normalized_value, ensure_ascii=False),
            is_public=bool(is_public),
            sort_order=int(sort_order or 100),
            created_at=now,
            updated_at=now,
        )
    else:
        row.name = str(name or row.name or normalized_key)
        row.category = str(category or row.category or "general")
        row.description = str(description or row.description or "")
        row.value_type = normalized_type
        row.value_json = json.dumps(normalized_value, ensure_ascii=False)
        row.is_public = bool(is_public)
        row.sort_order = int(sort_order or row.sort_order or 100)
        row.updated_at = now
    session.add(row)
    return row


def build_speech_runtime_settings(settings: PlatformSettingsPayload) -> Dict[str, Any]:
    return {
        "enabled": bool(settings.speech_enabled),
        "max_audio_seconds": int(DEFAULT_STT_MAX_AUDIO_SECONDS),
        "default_language": str(DEFAULT_STT_DEFAULT_LANGUAGE or "zh").strip().lower() or "zh",
        "force_simplified": bool(DEFAULT_STT_FORCE_SIMPLIFIED),
        "audio_preprocess": bool(DEFAULT_STT_AUDIO_PREPROCESS),
        "audio_filter": str(DEFAULT_STT_AUDIO_FILTER or "").strip(),
        "initial_prompt": str(DEFAULT_STT_INITIAL_PROMPT or "").strip(),
        "model": STT_MODEL,
        "device": STT_DEVICE,
    }