settled-reach/tooling/planet-gen/atlas_cohesion_audit.py

#!/usr/bin/env python3
"""
atlas_cohesion_audit.py — Quality analysis script for atlas atlas_* tables.

Runs a set of SQL queries against server/data/systems.db atlas_* tables to
identify names that need hand-refining:

  1. Stem duplicates within a system (same root token, different bodies)
  2. Cross-feature same-body near-duplicates (e.g. Aureus River + Aureus Range)
  3. Cardinal direction density per body (Western X / Northern X patterns)
  4. Generic / lazy outputs (Great %, Hilly %, Sector %, The %)
  5. Empty or single-character names
  6. Earth-echo concentration in a given system

Usage:
    python3 tooling/planet-gen/atlas_cohesion_audit.py
    python3 tooling/planet-gen/atlas_cohesion_audit.py --system GJ 144
    python3 tooling/planet-gen/atlas_cohesion_audit.py --body GJ144d

Decisions: D-191 (atlas pipeline, corridor palettes, markers.json format)
"""

import argparse
import sqlite3
import sys
from pathlib import Path

REPO_ROOT = (Path(__file__).resolve().parent / ".." / "..").resolve()
DB_PATH = REPO_ROOT / "server" / "data" / "systems.db"

CARDINALS = ("North", "South", "East", "West", "Northern", "Southern",
             "Eastern", "Western", "Upper", "Lower", "Far", "Kita", "Minami",
             "Higashi", "Nishi")

LAZY_PATTERNS = (
    "Great %", "Hilly %", "Sector %", "Flat %", "High %",
    "Big %", "Little %", "Low %", "Deep %", "Old %",
    "% Bluff", "% View", "% Gulch",
    "Flatland %", "Prairie %", "Desert %", "Canyon %",
    "% Crossing", "% Bend", "% Fork", "% Run", "% Creek",
    "% Point", "% Pass", "% Peak", "% Ridge", "% Stream",
    "% Ground", "% Slope", "% Mesa",
)

EARTH_ECHOES = (
    "Manchester", "London", "Paris", "Berlin", "Madrid",
    "Sierra Nevada", "Blue Ridge", "Appalachian",
    "Route %", "Highway %", "Interstate %",
    "New York", "Los Angeles", "Chicago",
    "Timberline", "Riverbend", "Greenfield",
    "Stony Creek", "Iron Creek", "Pine Gulch", "Valley View",
)


def get_conn(db_path: Path) -> sqlite3.Connection:
    conn = sqlite3.connect(str(db_path))
    conn.row_factory = sqlite3.Row
    return conn


def section(title: str) -> None:
    print(f"\n{'='*70}")
    print(f"  {title}")
    print('='*70)


def run_all_features_for_body(conn, body_id: str) -> list[tuple[str, str]]:
    """Return (table_label, name) for all named features on a body."""
    features = []
    for table, label in [
        ("atlas_cities", "city"),
        ("atlas_rivers", "river"),
        ("atlas_oceans", "ocean"),
        ("atlas_mountain_ranges", "mountain"),
        ("atlas_pois", "poi"),
    ]:
        rows = conn.execute(
            f"SELECT name FROM {table} WHERE body_id=? AND name != ''", (body_id,)
        ).fetchall()
        for r in rows:
            features.append((label, r["name"]))
    return features


def stem_of(name: str) -> str:
    """Extract first significant word as a rough stem."""
    parts = name.replace("The ", "").replace("the ", "").strip().split()
    return parts[0].lower() if parts else ""


def report_empty_names(conn, system_id: str | None, body_id: str | None) -> None:
    section("EMPTY OR SINGLE-CHARACTER NAMES")
    where_clauses = []
    params = []
    if body_id:
        where_clauses.append("body_id = ?")
        params.append(body_id)
    elif system_id:
        where_clauses.append("body_id IN (SELECT body_id FROM bodies WHERE system_id = ?)")
        params.append(system_id)
    w = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
    found = 0
    for table in ("atlas_cities", "atlas_rivers", "atlas_oceans",
                  "atlas_mountain_ranges", "atlas_pois"):
        rows = conn.execute(
            f"SELECT body_id, local_id, name FROM {table} {w} "
            f"AND (name = '' OR LENGTH(name) < 2)", params
        ).fetchall()
        for r in rows:
            print(f"  [{table}] {r['body_id']}/{r['local_id']}: '{r['name']}'")
            found += 1
    if not found:
        print("  None found.")


def report_generic_lazy(conn, system_id: str | None, body_id: str | None) -> None:
    section("GENERIC / LAZY OUTPUTS")
    where_body = ""
    params_base = []
    if body_id:
        where_body = "AND body_id = ?"
        params_base = [body_id]
    elif system_id:
        where_body = "AND body_id IN (SELECT body_id FROM bodies WHERE system_id = ?)"
        params_base = [system_id]

    found = 0
    for table, label in [
        ("atlas_cities", "city"), ("atlas_rivers", "river"),
        ("atlas_oceans", "ocean"), ("atlas_mountain_ranges", "mountain"),
        ("atlas_pois", "poi"),
    ]:
        hits = {}
        for pattern in LAZY_PATTERNS:
            rows = conn.execute(
                f"SELECT body_id, local_id, name FROM {table} "
                f"WHERE name LIKE ? {where_body}",
                [pattern] + params_base,
            ).fetchall()
            for r in rows:
                key = f"{r['body_id']}/{r['local_id']}"
                if key not in hits:
                    hits[key] = (r['body_id'], r['local_id'], r['name'], label)

        for key, (bid, lid, name, lbl) in sorted(hits.items()):
            print(f"  [{lbl}] {bid}/{lid}: '{name}'")
            found += 1

    if not found:
        print("  None found.")


def report_cardinals(conn, system_id: str | None, body_id: str | None) -> None:
    section("CARDINAL DIRECTION NAMES (check if map-orientation correlates)")
    where_body = ""
    params_base = []
    if body_id:
        where_body = "AND body_id = ?"
        params_base = [body_id]
    elif system_id:
        where_body = "AND body_id IN (SELECT body_id FROM bodies WHERE system_id = ?)"
        params_base = [system_id]

    by_body: dict[str, list] = {}
    for table, label in [
        ("atlas_cities", "city"), ("atlas_rivers", "river"),
        ("atlas_oceans", "ocean"), ("atlas_mountain_ranges", "mountain"),
    ]:
        for cardinal in CARDINALS:
            rows = conn.execute(
                f"SELECT body_id, local_id, name FROM {table} "
                f"WHERE (name LIKE ? OR name LIKE ?) {where_body}",
                [f"{cardinal} %", f"% {cardinal} %"] + params_base,
            ).fetchall()
            for r in rows:
                by_body.setdefault(r["body_id"], []).append(
                    (label, r["local_id"], r["name"])
                )

    if not by_body:
        print("  None found.")
        return
    for bid, entries in sorted(by_body.items()):
        print(f"  {bid}: {len(entries)} cardinal name(s)")
        for lbl, lid, name in entries:
            print(f"    [{lbl}] {lid}: '{name}'")


def report_earth_echoes(conn, system_id: str | None, body_id: str | None) -> None:
    section("EARTH-ECHO CONCENTRATION (allowed but check intentionality)")
    where_body = ""
    params_base = []
    if body_id:
        where_body = "AND body_id = ?"
        params_base = [body_id]
    elif system_id:
        where_body = "AND body_id IN (SELECT body_id FROM bodies WHERE system_id = ?)"
        params_base = [system_id]

    by_body: dict[str, list] = {}
    for table, label in [
        ("atlas_cities", "city"), ("atlas_rivers", "river"),
        ("atlas_oceans", "ocean"), ("atlas_mountain_ranges", "mountain"),
    ]:
        for pattern in EARTH_ECHOES:
            rows = conn.execute(
                f"SELECT body_id, local_id, name FROM {table} "
                f"WHERE name LIKE ? {where_body}",
                [f"%{pattern.replace('%', '')}%"] + params_base,
            ).fetchall()
            for r in rows:
                key = (r["body_id"], label, r["local_id"])
                by_body.setdefault(r["body_id"], []).append(
                    (label, r["local_id"], r["name"])
                )

    if not by_body:
        print("  None found.")
        return
    for bid, entries in sorted(by_body.items()):
        print(f"  {bid}: {len(entries)} earth-echo(s)")
        for lbl, lid, name in entries:
            print(f"    [{lbl}] {lid}: '{name}'")


def report_same_body_stem_dupes(conn, system_id: str | None, body_id: str | None) -> None:
    section("SAME-BODY CROSS-FEATURE STEM DUPLICATES")
    where = ""
    params = []
    if body_id:
        where = "WHERE body_id = ?"
        params = [body_id]
    elif system_id:
        where = "WHERE system_id = ?"
        params = [system_id]

    bodies_q = conn.execute(
        f"SELECT body_id FROM bodies {where}", params
    ).fetchall()

    found = 0
    for row in bodies_q:
        bid = row["body_id"]
        features = run_all_features_for_body(conn, bid)
        stem_to_features: dict[str, list] = {}
        for label, name in features:
            s = stem_of(name)
            if s and len(s) > 3:
                stem_to_features.setdefault(s, []).append((label, name))
        for stem, items in sorted(stem_to_features.items()):
            if len(items) > 1:
                types = set(i[0] for i in items)
                if len(types) > 1:  # only flag cross-feature (different types)
                    print(f"  {bid} stem='{stem}':")
                    for lbl, nm in items:
                        print(f"    [{lbl}] '{nm}'")
                    found += 1
    if not found:
        print("  None found.")


def report_cross_body_stem_dupes(conn, system_id: str | None) -> None:
    section("CROSS-BODY STEM DUPLICATES WITHIN SYSTEM (same corridor)")
    if not system_id:
        print("  (requires --system; skipped)")
        return

    bodies_q = conn.execute(
        "SELECT body_id FROM bodies WHERE system_id = ?", (system_id,)
    ).fetchall()

    # Collect all names per table across all bodies
    all_names: dict[str, list[tuple[str, str]]] = {}  # name -> [(body_id, table)]
    for row in bodies_q:
        bid = row["body_id"]
        for table, label in [
            ("atlas_cities", "city"), ("atlas_rivers", "river"),
            ("atlas_oceans", "ocean"), ("atlas_mountain_ranges", "mountain"),
            ("atlas_pois", "poi"),
        ]:
            rows = conn.execute(
                f"SELECT name FROM {table} WHERE body_id=? AND name != ''", (bid,)
            ).fetchall()
            for r in rows:
                name = r["name"]
                s = stem_of(name)
                if s and len(s) > 3:
                    all_names.setdefault(s, []).append((bid, label, name))

    found = 0
    for stem, hits in sorted(all_names.items()):
        bodies_hit = set(h[0] for h in hits)
        if len(bodies_hit) > 1:
            print(f"  stem='{stem}' appears in {len(bodies_hit)} bodies:")
            for bid, label, name in hits:
                print(f"    {bid} [{label}] '{name}'")
            found += 1

    if not found:
        print("  None found.")


def report_summary_score(conn, system_id: str | None, body_id: str | None) -> None:
    section("BODY SUMMARY SCORES")
    where = ""
    params = []
    if body_id:
        where = "WHERE body_id = ?"
        params = [body_id]
    elif system_id:
        where = "WHERE system_id = ?"
        params = [system_id]
    else:
        where = "WHERE inhabited = 1"

    bodies_q = conn.execute(
        f"SELECT body_id, proper_name, population FROM bodies {where}", params
    ).fetchall()

    print(f"  {'body_id':<20} {'name':<20} {'pop':<12} cities rivers oceans mounts pois")
    for row in bodies_q:
        bid = row["body_id"]
        name = row["proper_name"] or ""
        pop = row["population"] or 0

        counts = {}
        for table, key in [
            ("atlas_cities", "c"), ("atlas_rivers", "r"),
            ("atlas_oceans", "o"), ("atlas_mountain_ranges", "m"), ("atlas_pois", "p"),
        ]:
            n = conn.execute(
                f"SELECT COUNT(*) FROM {table} WHERE body_id=?", (bid,)
            ).fetchone()[0]
            counts[key] = n

        print(
            f"  {bid:<20} {name:<20} {pop:<12,} "
            f"{counts['c']:>5} {counts['r']:>6} {counts['o']:>6} "
            f"{counts['m']:>6} {counts['p']:>4}"
        )


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--system", help="GJ catalog ID (e.g. 'GJ 144')")
    parser.add_argument("--body", help="Body ID (e.g. 'GJ144d')")
    parser.add_argument("--db", default=str(DB_PATH), help="Path to systems.db")
    args = parser.parse_args()

    db = Path(args.db)
    if not db.exists():
        print(f"error: database not found at {db}", file=sys.stderr)
        sys.exit(1)

    conn = get_conn(db)

    system_id = args.system
    body_id = args.body

    if body_id and not system_id:
        row = conn.execute(
            "SELECT system_id FROM bodies WHERE body_id=?", (body_id,)
        ).fetchone()
        if row:
            system_id = row["system_id"]

    print("\nAtlas Cohesion Audit")
    print(f"  DB: {db}")
    if system_id:
        print(f"  System: {system_id}")
    if body_id:
        print(f"  Body: {body_id}")

    report_summary_score(conn, system_id, body_id)
    report_empty_names(conn, system_id, body_id)
    report_generic_lazy(conn, system_id, body_id)
    report_cardinals(conn, system_id, body_id)
    report_earth_echoes(conn, system_id, body_id)
    report_same_body_stem_dupes(conn, system_id, body_id)
    if system_id and not body_id:
        report_cross_body_stem_dupes(conn, system_id)

    conn.close()
    print("\nDone.\n")


if __name__ == "__main__":
    main()