ss-tools/scripts/cleanup_maintenance_banner_charts.py

#!/usr/bin/env python3
"""
Cleanup of orphaned "Maintenance Banner" charts in Superset.

Context (ADR: maintenance banner is a native MARKDOWN element, not a chart):
Before the fix, maintenance:start created a real markdown CHART resource
(slice_name="Maintenance Banner", viz_type="markdown") that was never rendered
— the visible banner is a native MARKDOWN element in the dashboard position_json.
The fix stops creating these charts, but the ones already created remain in the
Charts menu and in dashboard exports. This script finds and deletes them.

Safety:
- DRY-RUN by default. Pass --yes to actually delete.
- Matches charts by slice_name == "Maintenance Banner" AND viz_type == "markdown"
  AND params containing the known placeholder text, so legitimate charts are
  never matched.
- Re-checks every candidate via GET /chart/{id} before deletion.
- The dashboard layouts are NOT touched: banners live as native MARKDOWN elements
  in position_json and are unaffected by chart deletion. Stale maintenance_banners
  DB rows self-heal on the next maintenance:start (reachability liveness check).

Usage:
    python scripts/cleanup_maintenance_banner_charts.py [--yes] \
        [--url https://superset.example.com] [--username admin] [--password ...]
    # or via env: SUPERSET_URL / SUPERSET_USERNAME / SUPERSET_PASSWORD / SUPERSET_VERIFY_SSL

Output: per-chart decision (DELETE/SKIP) + JSON summary on stdout.
"""

from __future__ import annotations

import argparse
import asyncio
import json
import os
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "backend"))

from src.core.config_models import Environment  # noqa: E402
from src.core.superset_client import SupersetClient  # noqa: E402

BANNER_SLICE_NAME = "Maintenance Banner"
BANNER_PLACEHOLDER = "Maintenance banner placeholder"

GREEN = "\033[32m"
RED = "\033[31m"
YELLOW = "\033[33m"
RESET = "\033[0m"


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--url", default=os.environ.get("SUPERSET_URL", ""))
    parser.add_argument("--username", default=os.environ.get("SUPERSET_USERNAME", ""))
    parser.add_argument("--password", default=os.environ.get("SUPERSET_PASSWORD", ""))
    parser.add_argument(
        "--verify-ssl",
        action="store_true",
        default=os.environ.get("SUPERSET_VERIFY_SSL", "1") not in ("0", "false", "False"),
    )
    parser.add_argument(
        "--yes",
        action="store_true",
        help="Actually delete matching charts. Without it the script only reports.",
    )
    return parser.parse_args()


def _looks_like_bogus_banner(chart: dict) -> bool:
    """A candidate chart is bogus only if it is a markdown chart with the banner name.

    The placeholder check happens later on the full chart payload (GET /chart/{id});
    here we only pre-filter the list response columns.
    """
    return (
        chart.get("slice_name") == BANNER_SLICE_NAME
        and chart.get("viz_type") == "markdown"
    )


def _params_contain_placeholder(chart: dict) -> bool:
    params = chart.get("params")
    if not isinstance(params, str):
        return False
    try:
        parsed = json.loads(params)
    except json.JSONDecodeError:
        return False
    markdown = parsed.get("markdown", "") if isinstance(parsed, dict) else ""
    return BANNER_PLACEHOLDER in str(markdown)


async def _find_candidates(client: SupersetClient) -> list[dict]:
    """Fetch all charts named 'Maintenance Banner' (paginated)."""
    candidates: list[dict] = []
    page = 0
    page_size = 100
    while True:
        query = {
            "columns": ["id", "slice_name", "viz_type"],
            "filters": [{"col": "slice_name", "opr": "eq", "value": BANNER_SLICE_NAME}],
            "page_size": page_size,
            "page": page,
        }
        count, charts = await client.get_charts(query)
        for chart in charts:
            if _looks_like_bogus_banner(chart):
                candidates.append(chart)
        if page * page_size + len(charts) >= count or not charts:
            break
        page += 1
    return candidates


async def _delete_chart(client: SupersetClient, chart_id: int) -> None:
    await client.request(method="DELETE", endpoint=f"/chart/{chart_id}")


async def run(args: argparse.Namespace, client: SupersetClient | None = None) -> int:
    if client is None:
        missing = [
            name
            for name, val in (("url", args.url), ("username", args.username), ("password", args.password))
            if not val
        ]
        if missing:
            print(f"{RED}Missing required config: {', '.join(missing)} (args or env){RESET}", file=sys.stderr)
            return 2

        env = Environment(
            id="cleanup",
            name="Cleanup",
            url=args.url,
            username=args.username,
            password=args.password,
            stage="PROD",
            verify_ssl=args.verify_ssl,
        )
        client = SupersetClient(env)

    print(f"{YELLOW}Scanning {args.url} for '{BANNER_SLICE_NAME}' markdown charts...{RESET}")
    candidates = await _find_candidates(client)
    if not candidates:
        print(f"{GREEN}No 'Maintenance Banner' markdown charts found — nothing to do.{RESET}")
        return 0

    print(f"Found {len(candidates)} candidate chart(s). Verifying each via GET /chart/{{id}}...")
    decisions: dict[str, list[dict]] = {"DELETE": [], "SKIP": []}
    for cand in candidates:
        chart_id = int(cand["id"])
        try:
            full = await client.get_chart(chart_id)
        except Exception as exc:  # noqa: BLE001 - report and continue
            decisions["SKIP"].append({"id": chart_id, "reason": f"get_chart failed: {exc}"})
            print(f"  {YELLOW}SKIP {chart_id}: cannot fetch full payload ({exc}){RESET}")
            continue

        if not _params_contain_placeholder(full):
            decisions["SKIP"].append(
                {"id": chart_id, "reason": "params do not contain the banner placeholder — not our chart"}
            )
            print(f"  {YELLOW}SKIP {chart_id}: params lack placeholder — not a bogus banner chart{RESET}")
            continue

        decisions["DELETE"].append({"id": chart_id})
        print(f"  {RED}DELETE {chart_id}{RESET}  ({BANNER_SLICE_NAME}, markdown, placeholder present)")

    if not args.yes:
        print(
            f"\n{YELLOW}DRY-RUN: {len(decisions['DELETE'])} chart(s) would be deleted. "
            f"Re-run with --yes to apply.{RESET}"
        )
    else:
        for item in decisions["DELETE"]:
            try:
                await _delete_chart(client, item["id"])
                print(f"  {GREEN}deleted {item['id']}{RESET}")
            except Exception as exc:  # noqa: BLE001 - report and continue
                item["error"] = str(exc)
                print(f"  {RED}FAILED to delete {item['id']}: {exc}{RESET}")

    summary = {
        "url": args.url,
        "dry_run": not args.yes,
        "candidates_found": len(candidates),
        "decisions": decisions,
    }
    print(f"\n{json.dumps(summary, ensure_ascii=False, indent=2)}")
    return 0


def main() -> int:
    args = _parse_args()
    try:
        return asyncio.run(run(args))
    except KeyboardInterrupt:
        return 130


if __name__ == "__main__":
    sys.exit(main())