#!/usr/bin/env -S uv run --script
# /// script
# dependencies = ["typer", "ruamel.yaml"]
# ///

import json
import shutil
import subprocess
import sys
from difflib import unified_diff
from pathlib import Path
from typing import Annotated, Any, Optional

import typer
from ruamel.yaml import YAML

app = typer.Typer(help="Add or update a secret field in an existing SealedSecret")


def resolve_file(file: Path) -> Path:
    if file.is_dir():
        candidate = file / "sealedsecrets.yaml"
        if not candidate.is_file():
            raise typer.BadParameter(f"{file} is a directory and contains no sealedsecrets.yaml")
        return candidate
    if not file.is_file():
        raise typer.BadParameter(f"{file} does not exist")
    return file


def looks_like_json(text: str) -> bool:
    for char in text:
        if char in (" ", "\t", "\n", "\r"):
            continue
        return char == "{"
    return False


def extract_header_comments(text: str) -> tuple[str, str]:
    """Split leading comments/blank lines from the rest of the file."""
    lines = text.splitlines(keepends=True)
    idx = 0
    while idx < len(lines):
        stripped = lines[idx].strip()
        if stripped == "" or stripped.startswith("#"):
            idx += 1
        else:
            break
    return "".join(lines[:idx]), "".join(lines[idx:])


def load_documents(text: str) -> list[Any]:
    if looks_like_json(text):
        # JSON is valid YAML, but parse it explicitly so we can normalize it to YAML output.
        return [json.loads(text)]
    yaml = YAML(typ="rt")
    yaml.preserve_quotes = True
    return list(yaml.load_all(text))


def _represent_none(representer, data):
    return representer.represent_scalar("tag:yaml.org,2002:null", "null")


def dump_documents(docs: list[Any]) -> str:
    yaml = YAML(typ="rt")
    yaml.default_flow_style = False
    yaml.preserve_quotes = True
    yaml.width = 4096
    yaml.explicit_start = True
    yaml.representer.add_representer(type(None), _represent_none)
    from ruamel.yaml.compat import StringIO

    stream = StringIO()
    yaml.dump_all(docs, stream)
    return stream.getvalue()


def detect_scope(doc: Any) -> str:
    annotations = doc.get("metadata", {}).get("annotations", {}) or {}
    for key, value in annotations.items():
        if key == "sealedsecrets.bitnami.com/cluster-wide" and str(value).lower() == "true":
            return "cluster-wide"
        if key == "sealedsecrets.bitnami.com/namespace-wide" and str(value).lower() == "true":
            return "namespace-wide"
    return "strict"


def seal_value(value: str, doc: Any, cert: Optional[Path]) -> str:
    metadata = doc.get("metadata", {})
    spec = doc.get("spec", {})

    name = metadata.get("name")
    if not name:
        raise typer.BadParameter("SealedSecret has no metadata.name")

    namespace = metadata.get("namespace") or spec.get("template", {}).get("metadata", {}).get("namespace")
    if not namespace:
        raise typer.BadParameter("SealedSecret has no namespace in metadata or spec.template.metadata")

    cmd = ["kubeseal", "--raw", "--from-file=/dev/stdin"]
    if cert:
        cmd.extend(["--cert", str(cert)])

    scope = detect_scope(doc)
    if scope == "cluster-wide":
        cmd.extend(["--scope", "cluster-wide"])
    elif scope == "namespace-wide":
        cmd.extend(["--namespace", namespace, "--scope", "namespace-wide"])
    else:
        cmd.extend(["--namespace", namespace, "--name", name])

    result = subprocess.run(
        cmd,
        input=value,
        text=True,
        capture_output=True,
    )
    if result.returncode != 0:
        typer.echo(f"kubeseal failed:\n{result.stderr}", err=True)
        raise typer.Exit(result.returncode)

    return result.stdout.strip()


def find_target_document(docs: list[Any], key: str, secret_name: Optional[str]) -> Any:
    if secret_name:
        for doc in docs:
            if doc.get("metadata", {}).get("name") == secret_name:
                return doc
        raise typer.BadParameter(f"No SealedSecret named '{secret_name}' found in file")

    if len(docs) == 1:
        return docs[0]

    candidates = [
        doc for doc in docs
        if key in (doc.get("spec", {}).get("encryptedData", {}) or {})
    ]
    if len(candidates) == 1:
        return candidates[0]
    if len(candidates) > 1:
        names = [doc.get("metadata", {}).get("name", "<unnamed>") for doc in candidates]
        raise typer.BadParameter(
            f"Key '{key}' exists in multiple SealedSecrets ({', '.join(names)}). "
            "Use --secret-name to choose one."
        )

    names = [doc.get("metadata", {}).get("name", "<unnamed>") for doc in docs]
    raise typer.BadParameter(
        f"File contains multiple SealedSecrets and key '{key}' does not exist yet. "
        f"Available: {', '.join(names)}. Use --secret-name to choose one."
    )


@app.command()
def add(
    name: Annotated[str, typer.Option("--name", help="Key name in SealedSecret encryptedData")],
    secret: Annotated[str, typer.Option("--secret", help="Plaintext secret value")],
    file: Annotated[Path, typer.Option("--file", help="Path to sealedsecrets.yaml or its directory")],
    secret_name: Annotated[
        Optional[str],
        typer.Option("--secret-name", help="Name of the SealedSecret resource to update (required for multi-secret files when the key is new)"),
    ] = None,
    cert: Annotated[
        Optional[Path],
        typer.Option("--cert", help="Path to a kubeseal certificate for offline sealing"),
    ] = None,
    dry_run: Annotated[bool, typer.Option("--dry-run", help="Show diff without writing")] = False,
) -> None:
    target_path = resolve_file(file)
    original_text = target_path.read_text()
    header, body = extract_header_comments(original_text)

    docs = load_documents(body)
    if not docs:
        raise typer.BadParameter(f"{target_path} contains no documents")

    target = find_target_document(docs, name, secret_name)
    sealed = seal_value(secret, target, cert)

    if "spec" not in target or target["spec"] is None:
        target["spec"] = {}
    if "encryptedData" not in target["spec"] or target["spec"]["encryptedData"] is None:
        target["spec"]["encryptedData"] = {}

    old_value = target["spec"]["encryptedData"].get(name)
    action = "updated" if old_value is not None else "added"
    target["spec"]["encryptedData"][name] = sealed

    new_text = header + dump_documents(docs)

    if dry_run:
        diff = unified_diff(
            original_text.splitlines(keepends=True),
            new_text.splitlines(keepends=True),
            fromfile=str(target_path),
            tofile=str(target_path),
        )
        sys.stdout.writelines(diff)
        raise typer.Exit()

    backup_path = target_path.with_suffix(target_path.suffix + ".bak")
    shutil.copy2(target_path, backup_path)
    target_path.write_text(new_text)

    typer.echo(f"{action} '{name}' in SealedSecret '{target.get('metadata', {}).get('name')}' ({target_path})")
    typer.echo(f"backup saved to {backup_path}")


if __name__ == "__main__":
    app()