Nexus/scripts/sync_mysql_mcp_env.py

#!/usr/bin/env python3
"""Sync MYSQL_* variables for @yclenove/mysql-mcp-server from NEXUS_DATABASE_URL in .env.

Usage (from project root):
    python scripts/sync_mysql_mcp_env.py
    python scripts/sync_mysql_mcp_env.py --env path/to/.env

Reads NEXUS_DATABASE_URL (mysql+aiomysql://user:pass@host:port/db) and upserts:
  MYSQL_HOST, MYSQL_PORT, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE
  MYSQL_READONLY (default true; use --writable for local dev)
  MYSQL_DATABASE_ALLOWLIST=<db>
"""
from __future__ import annotations

import argparse
import re
import sys
from pathlib import Path
from urllib.parse import unquote

ROOT = Path(__file__).resolve().parent.parent
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))
URL_RE = re.compile(
    r"^mysql\+(?:aiomysql|pymysql)://([^:]+):([^@]+)@([^:/]+)(?::(\d+))?/([^?\s#]+)",
    re.IGNORECASE,
)

MCP_KEYS = [
    "MYSQL_HOST",
    "MYSQL_PORT",
    "MYSQL_USER",
    "MYSQL_PASSWORD",
    "MYSQL_DATABASE",
    "MYSQL_READONLY",
    "MYSQL_DATABASE_ALLOWLIST",
]


def parse_database_url(url: str, *, readonly: bool = True) -> dict[str, str]:
    m = URL_RE.match(url.strip())
    if not m:
        raise ValueError(
            f"Cannot parse NEXUS_DATABASE_URL: {url!r}\n"
            "Expected: mysql+aiomysql://user:pass@host:port/database"
        )
    user, password, host, port, database = m.groups()
    return {
        "MYSQL_HOST": host,
        "MYSQL_PORT": port or "3306",
        "MYSQL_USER": unquote(user),
        "MYSQL_PASSWORD": unquote(password),
        "MYSQL_DATABASE": unquote(database),
        "MYSQL_READONLY": "true" if readonly else "false",
        "MYSQL_DATABASE_ALLOWLIST": unquote(database),
    }


def read_env_lines(path: Path) -> list[str]:
    from server.utils.text_io import read_utf8_lf

    if not path.exists():
        return []
    text = read_utf8_lf(path)
    if not text:
        return []
    return [f"{line}\n" for line in text.splitlines()]


def find_nexus_database_url(lines: list[str]) -> str | None:
    for line in lines:
        s = line.strip()
        if s.startswith("NEXUS_DATABASE_URL=") and not s.startswith("#"):
            return s.split("=", 1)[1].strip()
    return None


def upsert_mcp_block(lines: list[str], values: dict[str, str]) -> list[str]:
    """Remove existing MYSQL_* / MCP section lines, append fresh block."""
    filtered: list[str] = []
    skip_mcp_header = False
    for line in lines:
        stripped = line.strip()
        if stripped.startswith("# ── MySQL MCP"):
            skip_mcp_header = True
            continue
        if skip_mcp_header and (
            stripped.startswith("#")
            or any(stripped.startswith(f"{k}=") for k in MCP_KEYS)
            or stripped == ""
        ):
            if stripped.startswith("# ──") and "MySQL MCP" not in stripped:
                skip_mcp_header = False
                filtered.append(line)
            continue
        if any(stripped.startswith(f"{k}=") for k in MCP_KEYS):
            continue
        skip_mcp_header = False
        filtered.append(line)

    while filtered and filtered[-1].strip() == "":
        filtered.pop()
    if filtered and not filtered[-1].endswith("\n"):
        filtered[-1] += "\n"

    block = [
        "\n",
        "# ── MySQL MCP (@yclenove/mysql-mcp-server — auto-synced, do not commit secrets) ──\n",
    ]
    for key in MCP_KEYS:
        block.append(f"{key}={values[key]}\n")
    filtered.extend(block)
    return filtered


def main() -> int:
    parser = argparse.ArgumentParser(description="Sync MYSQL_* for mysql-mcp-server from NEXUS_DATABASE_URL")
    parser.add_argument("--env", type=Path, default=ROOT / ".env", help="Path to .env file")
    mode = parser.add_mutually_exclusive_group()
    mode.add_argument("--writable", action="store_true", help="MYSQL_READONLY=false (local dev)")
    mode.add_argument("--readonly", action="store_true", help="MYSQL_READONLY=true (default)")
    args = parser.parse_args()
    readonly = not args.writable
    env_path: Path = args.env.resolve()

    lines = read_env_lines(env_path)
    if not lines:
        print(f"ERROR: {env_path} not found. Copy .env.example to .env or run install wizard first.", file=sys.stderr)
        return 1

    url = find_nexus_database_url(lines)
    if not url:
        print("ERROR: NEXUS_DATABASE_URL not found in .env", file=sys.stderr)
        return 1

    try:
        values = parse_database_url(url, readonly=readonly)
    except ValueError as e:
        print(f"ERROR: {e}", file=sys.stderr)
        return 1

    from server.utils.text_io import write_utf8_lf

    updated = upsert_mcp_block(lines, values)
    write_utf8_lf(env_path, "".join(updated))
    print(f"OK: Updated {env_path}")
    mode_label = "readonly" if values["MYSQL_READONLY"] == "true" else "writable"
    print(
        f"  MySQL MCP → {values['MYSQL_USER']}@{values['MYSQL_HOST']}:"
        f"{values['MYSQL_PORT']}/{values['MYSQL_DATABASE']} ({mode_label})"
    )
    return 0


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