import base64
import datetime
import json
import os
import re
import subprocess
import sys
import time


def required(name):
    value = os.environ.get(name, "").strip()
    if not value:
        raise RuntimeError(f"missing required environment variable {name}")
    return value


def run(command, timeout_seconds):
    return subprocess.run(
        command,
        check=False,
        capture_output=True,
        text=True,
        timeout=timeout_seconds,
    )


def stderr_tail(result, limit=1200):
    return (result.stderr or "")[-limit:]


def list_group_states(kafka_command, timeout_seconds):
    result = run(kafka_command + ["--list", "--state"], timeout_seconds)
    if result.returncode != 0:
        raise RuntimeError(f"Kafka group state listing failed: {stderr_tail(result)}")
    states = {}
    header_seen = False
    for raw_line in result.stdout.splitlines():
        fields = raw_line.split()
        if len(fields) >= 2 and fields[0] == "GROUP" and fields[1] == "STATE":
            header_seen = True
            continue
        if header_seen and len(fields) >= 2:
            states[fields[0]] = fields[1]
    if not header_seen:
        raise RuntimeError("Kafka group state listing did not include the GROUP STATE header")
    return states, stderr_tail(result)


def iso_time(timestamp_ms):
    return datetime.datetime.fromtimestamp(timestamp_ms / 1000, datetime.timezone.utc).isoformat().replace("+00:00", "Z")


def classify(states, policy, now_ms):
    pattern = re.compile(policy["timestamp"]["pattern"])
    capture_group = int(policy["timestamp"]["captureGroup"])
    minimum_age_ms = int(policy["minimumGroupAgeHours"]) * 60 * 60 * 1000
    inactive_states = set(policy["inactiveStates"])
    prefix_matched = []
    pattern_matched = []
    inactive_matched = []
    age_eligible = []
    candidates = []
    invalid_timestamp_count = 0
    for name, state in states.items():
        if not name.startswith(policy["prefix"]):
            continue
        prefix_matched.append(name)
        match = pattern.fullmatch(name)
        if match is None:
            continue
        pattern_matched.append(name)
        try:
            created_ms = int(match.group(capture_group))
        except (IndexError, TypeError, ValueError):
            invalid_timestamp_count += 1
            continue
        age_ms = now_ms - created_ms
        if age_ms < 0:
            invalid_timestamp_count += 1
            continue
        if state in inactive_states:
            inactive_matched.append(name)
        if age_ms >= minimum_age_ms:
            age_eligible.append(name)
        if state in inactive_states and age_ms >= minimum_age_ms:
            candidates.append({
                "name": name,
                "state": state,
                "createdAt": iso_time(created_ms),
                "ageHours": round(age_ms / 3600000, 3),
            })
    candidates.sort(key=lambda item: (-item["ageHours"], item["name"]))
    return {
        "prefixMatchedCount": len(prefix_matched),
        "patternMatchedCount": len(pattern_matched),
        "inactiveMatchedCount": len(inactive_matched),
        "ageEligibleCount": len(age_eligible),
        "invalidTimestampCount": invalid_timestamp_count,
        "candidates": candidates,
    }


def delete_candidates(kafka_command, candidates, batch_size, timeout_seconds):
    batch_evidence = []
    for start in range(0, len(candidates), batch_size):
        batch = candidates[start:start + batch_size]
        command = kafka_command + ["--delete"]
        for candidate in batch:
            command.extend(["--group", candidate["name"]])
        result = run(command, timeout_seconds)
        batch_evidence.append({
            "size": len(batch),
            "exitCode": result.returncode,
            "stderrTail": stderr_tail(result),
        })
    return batch_evidence


def main():
    target = required("KAFKA_TARGET")
    namespace = required("KAFKA_NAMESPACE")
    cluster = required("KAFKA_CLUSTER")
    bootstrap = required("KAFKA_BOOTSTRAP")
    confirm = required("KAFKA_GROUP_CONFIRM") == "true"
    full = required("KAFKA_FULL") == "true"
    candidate_limit = int(required("KAFKA_CANDIDATE_LIMIT"))
    timeout_seconds = int(required("KAFKA_COMMAND_TIMEOUT_SECONDS"))
    batch_size = int(required("KAFKA_DELETE_BATCH_SIZE"))
    policy = json.loads(base64.b64decode(required("KAFKA_POLICY_B64")).decode("utf-8"))
    pod_result = run([
        "kubectl", "-n", namespace, "get", "pod",
        "-l", f"strimzi.io/cluster={cluster},strimzi.io/kind=Kafka",
        "-o", "jsonpath={.items[0].metadata.name}",
    ], timeout_seconds)
    pod = pod_result.stdout.strip()
    if pod_result.returncode != 0 or not pod:
        raise RuntimeError(f"Kafka broker pod not found: {stderr_tail(pod_result)}")
    kafka_command = [
        "kubectl", "-n", namespace, "exec", pod, "--",
        "bin/kafka-consumer-groups.sh", "--bootstrap-server", bootstrap,
    ]
    observed_at_ms = int(time.time() * 1000)
    before_states, list_stderr = list_group_states(kafka_command, timeout_seconds)
    selection = classify(before_states, policy, observed_at_ms)
    eligible_candidates = selection.pop("candidates")
    selected_candidates = eligible_candidates[:candidate_limit]
    batch_evidence = []
    deletions = []
    after_states = before_states
    if confirm and selected_candidates:
        batch_evidence = delete_candidates(kafka_command, selected_candidates, batch_size, timeout_seconds)
        after_states, _ = list_group_states(kafka_command, timeout_seconds)
        inactive_states = set(policy["inactiveStates"])
        for candidate in selected_candidates:
            name = candidate["name"]
            state_after = after_states.get(name)
            if state_after is None:
                status = "deleted"
            elif state_after not in inactive_states:
                status = "state-changed"
            else:
                status = "failed"
            deletions.append({"name": name, "status": status, "stateAfter": state_after})
    display_deletions = deletions if full else deletions[:candidate_limit]
    deleted_count = sum(1 for item in deletions if item["status"] == "deleted")
    state_changed_count = sum(1 for item in deletions if item["status"] == "state-changed")
    failed_count = sum(1 for item in deletions if item["status"] == "failed")
    payload = {
        "ok": failed_count == 0,
        "target": target,
        "namespace": namespace,
        "cluster": cluster,
        "pod": pod,
        "mode": "confirmed" if confirm else "plan",
        "mutation": confirm,
        "observedAt": iso_time(observed_at_ms),
        "policy": policy,
        "summary": {
            "totalGroupCount": len(before_states),
            **selection,
            "eligibleCount": len(eligible_candidates),
            "selectedCount": len(selected_candidates),
            "omittedCount": max(0, len(eligible_candidates) - len(selected_candidates)),
            "deleteAttemptedCount": len(deletions),
            "deletedCount": deleted_count,
            "stateChangedCount": state_changed_count,
            "failedCount": failed_count,
        },
        "candidates": selected_candidates,
        "eligibleCandidates": eligible_candidates if full else [],
        "deletions": display_deletions,
        "batchEvidence": batch_evidence if full else [],
        "stderrTail": list_stderr,
        "valuesPrinted": False,
    }
    print(json.dumps(payload, ensure_ascii=False, indent=2))
    return 0 if payload["ok"] else 1


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as error:
        confirmed = os.environ.get("KAFKA_GROUP_CONFIRM", "").strip() == "true"
        print(json.dumps({
            "ok": False,
            "error": str(error),
            "mutation": confirmed,
            "mutationOutcome": "unknown" if confirmed else "not-attempted",
        }, ensure_ascii=False))
        sys.exit(1)