from __future__ import annotations

import json
import os
import re
import shutil
import subprocess
from datetime import UTC, datetime
from pathlib import Path

from .config import SourceConfig
from .source_db import export_filter


def ensure_profile(config: SourceConfig) -> dict[str, object]:
    """Materialize the `ticloud` profile this source uses from its own secrets, so auth is
    portable instead of bound to a hand-created ~/.ticloud/config.toml on one machine.

    `ticloud` (unlike most TiDB Cloud SDKs) does NOT read API keys from the environment — it
    only reads a named profile from its local config file. For a cloud runner that file does
    not exist, so we bootstrap it here from the per-source API key pair (injected via
    secrets.env / a secret manager). Idempotent and back-compatible:

    - If the named profile already exists locally (e.g. a developer ran `config create`, or a
      prior run bootstrapped it), this is a no-op — we do not clobber it.
    - Else, if the source provides an API key pair, create the profile from it.
    - Else, leave it: the engine will pass `--profile` and fail loudly if it is truly absent,
      which is the same behavior as before this function existed.
    """
    profile = config.tidb_cli_profile
    if not profile:
        return {"profile": None, "bootstrapped": False, "reason": "no profile configured"}
    if profile in _existing_profiles():
        return {"profile": profile, "bootstrapped": False, "reason": "already present"}
    public_key = config.tidb_cloud_public_key
    private_key = config.tidb_cloud_private_key
    if not (public_key and private_key):
        return {"profile": profile, "bootstrapped": False, "reason": "no api keys in secrets"}
    subprocess.run(
        [
            find_ticloud_bin(), "config", "create",
            "--profile-name", profile,
            "--public-key", public_key,
            "--private-key", private_key,
        ],
        check=True, capture_output=True, text=True,
    )
    return {"profile": profile, "bootstrapped": True, "reason": "created from source secrets"}


def _existing_profiles() -> set[str]:
    """Profile names known to the local `ticloud` config. First line is a header; each
    remaining line starts with the profile name (the active one is tab-suffixed `(active)`)."""
    result = subprocess.run(
        [find_ticloud_bin(), "config", "list"], capture_output=True, text=True
    )
    lines = result.stdout.splitlines()[1:]
    return {line.split()[0] for line in lines if line.strip()}


def start_export(config: SourceConfig, run_date: str | None = None, run_id: str | None = None) -> dict[str, str]:
    ensure_profile(config)
    run_date = run_date or datetime.now(UTC).strftime("%Y-%m-%d")
    run_id = run_id or datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
    s3_uri = config.s3_run_uri(run_date, run_id)
    command = [
        find_ticloud_bin(),
        "serverless",
        "export",
        "create",
        "--profile",
        config.tidb_cli_profile,
        "--cluster-id",
        config.tidb_cluster_id,
        "--target-type",
        "S3",
        "--s3.uri",
        s3_uri,
        "--s3.access-key-id",
        config.export_access_key_id,
        "--s3.secret-access-key",
        config.export_secret_access_key,
        "--filter",
        export_filter(config),
        "--file-type",
        "PARQUET",
        "--parquet.compression",
        config.parquet_compression,
    ]
    result = subprocess.run(command, check=True, capture_output=True, text=True)
    payload = {"run_date": run_date, "run_id": run_id, "s3_uri": s3_uri}
    export_id = parse_export_id(result.stdout)
    if export_id:
        # Bridges start-export -> write-manifest without a manual `ticloud export list`.
        payload["export_id"] = export_id
    return payload


def parse_export_id(stdout: str) -> str | None:
    """Best-effort extraction of the created export's id from `export create` output.

    `ticloud serverless export create` does not accept `--output` and prints a plain line
    like ``export exp-xxxx is running now`` (older CLIs emitted JSON). Try JSON first, then
    fall back to the ``exp-...`` token in the text. Returns None if neither matches (the
    export still ran, so we degrade to "no id" rather than failing).
    """
    try:
        data = json.loads(stdout)
    except (json.JSONDecodeError, TypeError):
        data = None
    if isinstance(data, dict):
        # Structured form is authoritative — don't regex-scan JSON text for a stray id.
        nested = data.get("export")
        nested_id = nested.get("exportId") if isinstance(nested, dict) else None
        return data.get("exportId") or data.get("id") or nested_id
    # Plain-text form: `export exp-xxxx is running now`.
    match = re.search(r"\bexp-[a-z0-9]+\b", stdout or "")
    return match.group(0) if match else None


def find_ticloud_bin() -> str:
    if os.environ.get("TICLOUD_BIN"):
        return os.environ["TICLOUD_BIN"]
    found = shutil.which("ticloud")
    if found:
        return found
    home_bin = Path.home() / ".ticloud" / "bin" / "ticloud"
    if home_bin.exists():
        return str(home_bin)
    raise SystemExit("ticloud CLI is required")