"""Server-side connection checks for external integrations.

Credential files are referenced by absolute path. Secret values never cross the
admin HTTP API and are never included in errors or successful responses.
"""

from __future__ import annotations

import hashlib
import hmac
import json
import os
import re
import socket
import time
import uuid
from pathlib import Path
from typing import Callable
from urllib.error import HTTPError, URLError
from urllib.parse import quote, unquote, urlparse
from urllib.request import Request, urlopen


POSTGRES_SCHEMES = {"postgres", "postgresql"}
POSTGRES_SSL_MODES = {"disable", "allow", "prefer", "require", "verify-ca", "verify-full"}
COS_BUCKET_PATTERN = re.compile(r"^[a-z0-9][a-z0-9-]{1,62}-[0-9]+$")
COS_REGION_PATTERN = re.compile(r"^[a-z][a-z0-9-]{1,30}$")
COS_API_HOST_PATTERN = re.compile(r"\.cos\.(?P<region>[a-z][a-z0-9-]{1,30})\.myqcloud\.com$")


class IntegrationAdapterError(RuntimeError):
    def __init__(self, code: str, message: str, http_status: int = 503) -> None:
        super().__init__(message)
        self.code = code
        self.message = message
        self.http_status = http_status


def _read_json_credentials(request_ref: str, environment_key: str, label: str) -> dict[str, object]:
    ref = os.environ.get(environment_key, "").strip() or request_ref.strip()
    if not ref:
        raise IntegrationAdapterError("CONFIG_MISSING", f"未配置{label}凭据文件", 400)
    credential_path = Path(ref)
    if not credential_path.is_absolute():
        raise IntegrationAdapterError("INVALID_CREDENTIAL_REF", f"{label}凭据引用必须是绝对路径", 400)
    try:
        payload = json.loads(credential_path.read_text(encoding="utf-8"))
    except FileNotFoundError as exc:
        raise IntegrationAdapterError("CREDENTIAL_FILE_NOT_FOUND", f"{label}凭据文件不存在", 400) from exc
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise IntegrationAdapterError("CREDENTIAL_FILE_INVALID", f"{label}凭据文件不可读取或格式错误", 400) from exc
    if not isinstance(payload, dict):
        raise IntegrationAdapterError("CREDENTIAL_FILE_INVALID", f"{label}凭据文件格式错误", 400)
    return payload


def _postgres_connection_settings(endpoint: str, identifier: str, credentials_ref: str) -> dict[str, object]:
    parsed = urlparse(endpoint.strip())
    if parsed.scheme not in POSTGRES_SCHEMES or not parsed.hostname:
        raise IntegrationAdapterError(
            "INVALID_DATABASE_ENDPOINT",
            "数据库地址应为 postgresql://主机:端口/数据库名",
            400,
        )
    if parsed.username is not None or parsed.password is not None:
        raise IntegrationAdapterError("DATABASE_SECRET_IN_ENDPOINT", "数据库地址中不得填写账号或密码", 400)
    database = unquote(parsed.path.lstrip("/")).strip()
    if not database:
        raise IntegrationAdapterError("INVALID_DATABASE_ENDPOINT", "数据库地址缺少数据库名", 400)
    if identifier.strip() and identifier.strip() != database:
        raise IntegrationAdapterError("DATABASE_IDENTIFIER_MISMATCH", "数据库标识与地址中的数据库名不一致", 400)
    try:
        port = parsed.port or 5432
    except ValueError as exc:
        raise IntegrationAdapterError("INVALID_DATABASE_ENDPOINT", "数据库端口无效", 400) from exc

    credentials = _read_json_credentials(
        credentials_ref,
        "AITASKER_DATABASE_CREDENTIALS_FILE",
        "PostgreSQL",
    )
    user = credentials.get("user")
    password = credentials.get("password")
    sslmode = str(credentials.get("sslmode") or "require")
    if not isinstance(user, str) or not user.strip() or not isinstance(password, str) or not password:
        raise IntegrationAdapterError(
            "DATABASE_CREDENTIALS_INVALID",
            "PostgreSQL 凭据文件缺少 user 或 password",
            400,
        )
    if sslmode not in POSTGRES_SSL_MODES:
        raise IntegrationAdapterError("DATABASE_CREDENTIALS_INVALID", "PostgreSQL sslmode 配置无效", 400)
    return {
        "host": parsed.hostname,
        "port": port,
        "dbname": database,
        "user": user.strip(),
        "password": password,
        "sslmode": sslmode,
        "connect_timeout": 5,
    }


def _database_error(exc: Exception) -> IntegrationAdapterError:
    detail = str(exc).lower()
    if "password authentication failed" in detail or "authentication failed" in detail:
        return IntegrationAdapterError("DATABASE_AUTH_FAILED", "PostgreSQL 认证失败，请检查账号和密码", 502)
    if "timeout" in detail or "timed out" in detail:
        return IntegrationAdapterError("DATABASE_TIMEOUT", "PostgreSQL 连接超时，请检查 SSH 隧道", 504)
    unreachable_markers = (
        "connection refused",
        "could not connect",
        "failed to connect",
        "network is unreachable",
        "no connection could be made",
        "server closed the connection",
    )
    if any(marker in detail for marker in unreachable_markers):
        return IntegrationAdapterError("DATABASE_UNREACHABLE", "PostgreSQL 无法连接，请检查 SSH 隧道和端口", 502)
    return IntegrationAdapterError("DATABASE_TEST_FAILED", "PostgreSQL 连接或查询失败", 502)


def test_postgresql_connection(
    endpoint: str,
    identifier: str,
    credentials_ref: str,
    connector: Callable[..., object] | None = None,
) -> dict[str, object]:
    settings = _postgres_connection_settings(endpoint, identifier, credentials_ref)
    if connector is None:
        try:
            import psycopg  # type: ignore[import-not-found]
        except ImportError as exc:
            raise IntegrationAdapterError(
                "DATABASE_DRIVER_MISSING",
                "缺少 PostgreSQL 驱动，请安装 psycopg[binary]",
                503,
            ) from exc
        connector = psycopg.connect

    connection = None
    try:
        connection = connector(**settings)
        with connection.cursor() as cursor:
            cursor.execute("SELECT 1")
            row = cursor.fetchone()
        if not row or row[0] != 1:
            raise IntegrationAdapterError("DATABASE_TEST_FAILED", "PostgreSQL 查询校验失败", 502)
    except IntegrationAdapterError:
        raise
    except Exception as exc:
        raise _database_error(exc) from exc
    finally:
        if connection is not None:
            try:
                connection.close()
            except Exception:
                pass
    return {
        "ok": True,
        "service": "database",
        "status": "CONNECTED",
        "message": "PostgreSQL 真实连接通过（SELECT 1）",
    }


def _cos_credentials(credentials_ref: str, endpoint: str) -> tuple[str, str, str]:
    credentials = _read_json_credentials(
        credentials_ref,
        "TENCENT_COS_CREDENTIALS_FILE",
        "腾讯云 COS",
    )
    secret_id = credentials.get("secretId")
    secret_key = credentials.get("secretKey")
    if not isinstance(secret_id, str) or not secret_id.strip() or not isinstance(secret_key, str) or not secret_key:
        raise IntegrationAdapterError(
            "COS_CREDENTIALS_INVALID",
            "腾讯云 COS 凭据文件缺少 secretId 或 secretKey",
            400,
        )

    region = str(os.environ.get("TENCENT_COS_REGION") or credentials.get("region") or "").strip()
    parsed = urlparse(endpoint.strip())
    if endpoint.strip() and (parsed.scheme != "https" or not parsed.hostname):
        raise IntegrationAdapterError("INVALID_COS_ENDPOINT", "COS 访问域名必须使用 HTTPS", 400)
    if not region and parsed.hostname:
        match = COS_API_HOST_PATTERN.search(parsed.hostname)
        region = match.group("region") if match else ""
    if not COS_REGION_PATTERN.fullmatch(region):
        raise IntegrationAdapterError(
            "COS_REGION_MISSING",
            "COS 凭据文件缺少有效 region，例如 ap-beijing",
            400,
        )
    return secret_id.strip(), secret_key, region


def _cos_authorization(method: str, path: str, host: str, secret_id: str, secret_key: str) -> str:
    start = int(time.time()) - 30
    key_time = f"{start};{start + 600}"
    header_list = "host"
    http_string = f"{method.lower()}\n{path}\n\nhost={host}\n"
    string_to_sign = f"sha1\n{key_time}\n{hashlib.sha1(http_string.encode('utf-8')).hexdigest()}\n"
    sign_key = hmac.new(secret_key.encode("utf-8"), key_time.encode("utf-8"), hashlib.sha1).hexdigest()
    signature = hmac.new(sign_key.encode("utf-8"), string_to_sign.encode("utf-8"), hashlib.sha1).hexdigest()
    values = {
        "q-sign-algorithm": "sha1",
        "q-ak": secret_id,
        "q-sign-time": key_time,
        "q-key-time": key_time,
        "q-header-list": header_list,
        "q-url-param-list": "",
        "q-signature": signature,
    }
    return "&".join(f"{key}={quote(value, safe=';-_.~')}" for key, value in values.items())


def _build_cos_request(
    method: str,
    bucket: str,
    region: str,
    object_key: str,
    secret_id: str,
    secret_key: str,
    body: bytes | None = None,
) -> Request:
    host = f"{bucket}.cos.{region}.myqcloud.com"
    path = "/" + quote(object_key, safe="/-_.~")
    request = Request(f"https://{host}{path}", data=body, method=method)
    request.add_header("Host", host)
    request.add_header("Authorization", _cos_authorization(method, path, host, secret_id, secret_key))
    if body is not None:
        request.add_header("Content-Type", "application/octet-stream")
        request.add_header("Content-Length", str(len(body)))
    return request


def _cos_error(exc: Exception) -> IntegrationAdapterError:
    if isinstance(exc, HTTPError):
        if exc.code in (401, 403):
            return IntegrationAdapterError("COS_ACCESS_DENIED", "COS 认证失败或没有测试对象权限", 502)
        if exc.code == 404:
            return IntegrationAdapterError("COS_RESOURCE_NOT_FOUND", "COS Bucket、地域或测试对象不存在", 502)
        if exc.code in (408, 429) or exc.code >= 500:
            return IntegrationAdapterError("COS_TEMPORARILY_UNAVAILABLE", "COS 暂时不可用，请稍后重试", 502)
        return IntegrationAdapterError("COS_TEST_FAILED", f"COS 返回 HTTP {exc.code}", 502)
    if isinstance(exc, (TimeoutError, socket.timeout)):
        return IntegrationAdapterError("COS_TIMEOUT", "COS 连接超时", 504)
    if isinstance(exc, URLError):
        if isinstance(exc.reason, (TimeoutError, socket.timeout)):
            return IntegrationAdapterError("COS_TIMEOUT", "COS 连接超时", 504)
        return IntegrationAdapterError("COS_UNREACHABLE", "COS 网络连接失败", 502)
    return IntegrationAdapterError("COS_TEST_FAILED", "COS 连接测试失败", 502)


def _open_cos_request(opener: Callable[..., object], request: Request) -> bytes:
    with opener(request, timeout=8) as response:
        return response.read(64 * 1024)


def test_cos_connection(
    bucket: str,
    endpoint: str,
    credentials_ref: str,
    opener: Callable[..., object] = urlopen,
) -> dict[str, object]:
    bucket = bucket.strip()
    if not COS_BUCKET_PATTERN.fullmatch(bucket):
        raise IntegrationAdapterError("INVALID_COS_BUCKET", "COS Bucket 名称格式无效", 400)
    secret_id, secret_key, region = _cos_credentials(credentials_ref, endpoint)
    object_key = f"_aitasker_connection_tests/{uuid.uuid4().hex}.txt"
    content = f"aitasker-cos-connection-test:{uuid.uuid4().hex}".encode("ascii")
    uploaded = False
    primary_error: IntegrationAdapterError | None = None

    try:
        put_request = _build_cos_request("PUT", bucket, region, object_key, secret_id, secret_key, content)
        _open_cos_request(opener, put_request)
        uploaded = True
        get_request = _build_cos_request("GET", bucket, region, object_key, secret_id, secret_key)
        returned = _open_cos_request(opener, get_request)
        if returned != content:
            raise IntegrationAdapterError("COS_CONTENT_MISMATCH", "COS 上传后读取内容校验失败", 502)
    except IntegrationAdapterError as exc:
        primary_error = exc
    except Exception as exc:
        primary_error = _cos_error(exc)
    finally:
        if uploaded:
            try:
                delete_request = _build_cos_request("DELETE", bucket, region, object_key, secret_id, secret_key)
                _open_cos_request(opener, delete_request)
            except Exception as exc:
                if primary_error is None:
                    primary_error = _cos_error(exc)

    if primary_error is not None:
        raise primary_error
    return {
        "ok": True,
        "service": "storage",
        "status": "CONNECTED",
        "message": "腾讯云 COS 真实连接通过（上传、读取、删除）",
    }