refactor(infra): replace AppFloyo REST bridge with report generator

- gremlin_appflowy_bridge.py: deprecated shim → delegates to new module
- gremlin_lean_report.py: generates JSON + Markdown reports from Gremlin
  with module theorem counts, sorries, import/dependency metrics
- AppFloyo Cloud v0.16.5 lacks programmatic database REST API;
  reports can be imported via AppFloyo UI instead
- GoTrue auth (port 9999) is working for future API use
This commit is contained in:
allaun 2026-06-30 03:14:05 -05:00
parent d7c60cf60c
commit 6468138112
2 changed files with 283 additions and 0 deletions

View file

@ -0,0 +1,30 @@
#!/usr/bin/env -S uv run
# /// script
# requires-python = ">=3.11"
# dependencies = [
# "gremlinpython",
# "python-dotenv",
# ]
# ///
"""gremlin_appflowy_bridge.py — DEPRECATED. Use gremlin_lean_report.py instead.
The old AppFloyo Cloud REST API (/api/databases) does not exist in
self-hosted v0.16.5. The replacement generates Markdown/JSON reports
that can be imported into AppFloyo via the UI, or consumed by other
automation.
See: gremlin_lean_report.py
signatures/appflowy_deployment_status.json
"""
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "shim"))
from gremlin_lean_report import main
if __name__ == "__main__":
main()

View file

@ -0,0 +1,253 @@
#!/usr/bin/env -S uv run
# /// script
# requires-python = ">=3.11"
# dependencies = [
# "gremlinpython",
# "python-dotenv",
# ]
# ///
"""gremlin_lean_report.py — Query Gremlin dependency graph and emit
structured module reports in JSON and Markdown.
Flow:
Gremlin (46K vertices, 30K edges)
Query all module vertices + 'imports' edges
Build module dependency map
JSON report (machine-readable)
Markdown report (human-readable, importable into AppFlowy)
Future: AppFloyo Cloud push via WS collab protocol (blocked on
AppFloyo Cloud v0.16.5 lacking programmatic REST API for databases.
See signatures/appflowy_deployment_status.json.)
"""
from __future__ import annotations
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[2]
# ── Gremlin connection ──────────────────────────────────────────────────
def get_gremlin_client():
"""Create Gremlin client from .env.gremlin."""
env_path = ROOT / ".env.gremlin"
if not env_path.exists():
print("ERROR: .env.gremlin not found", file=sys.stderr)
sys.exit(1)
lines = env_path.read_text().splitlines()
cfg = {}
for line in lines:
if "=" in line and not line.startswith("#"):
k, v = line.split("=", 1)
cfg[k.strip()] = v.strip().strip('"')
from gremlin_python.driver import client as gremlin_client, serializer
return gremlin_client.Client(
cfg.get("GREMLIN_ENDPOINT", "wss://mathblob.gremlin.cosmos.azure.com:443/"),
"g",
username=cfg.get("GREMLIN_USERNAME", "/dbs/research/colls/concepts"),
password=cfg.get("GREMLIN_PASSWORD", ""),
message_serializer=serializer.GraphSONSerializersV2d0(),
)
def query_gremlin(query: str, bindings: dict | None = None) -> list[dict]:
"""Run a Gremlin query and return results."""
c = get_gremlin_client()
try:
result = c.submitAsync(query, bindings or {}).result().all().result()
return list(result)
finally:
c.close()
def get_all_modules() -> list[dict]:
"""Get all module vertices with theorem/sorry counts."""
raw = query_gremlin(
'g.V().hasLabel("module").project("id","ns","tc","sc","fc")'
'.by(id).by(coalesce(values("namespace"), constant("")))'
'.by(coalesce(values("theorem_count"), constant(0)))'
'.by(coalesce(values("sorry_count"), constant(0)))'
'.by(coalesce(values("file_path"), constant("")))'
'.fold()'
)
items = raw[0] if raw and isinstance(raw[0], list) else raw
modules = []
for r in items if isinstance(items, list) else []:
if isinstance(r, dict):
modules.append({
"id": r.get("id", ""),
"name": r.get("ns", r["id"].split(".")[-1] if "." in r.get("id", "") else r.get("id", "")),
"theorem_count": int(r.get("tc", 0)),
"sorry_count": int(r.get("sc", 0)),
"file_path": r.get("fc", ""),
})
return modules
def get_dependency_map() -> dict[str, list[str]]:
"""Get module dependency graph via 'imports' edges."""
raw = query_gremlin(
'g.V().hasLabel("module").project("id","deps")'
'.by(id).by(out("imports").hasLabel("module").id().fold())'
'.fold()'
)
items = raw[0] if raw and isinstance(raw[0], list) else raw
result: dict[str, list[str]] = {}
for r in items if isinstance(items, list) else []:
if isinstance(r, dict):
result[r["id"]] = list(r.get("deps", []))
return result
# ── Report generation ──────────────────────────────────────────────────
def build_report(modules: list[dict], deps: dict[str, list[str]]) -> dict:
"""Build a structured report from Gremlin data."""
rev_deps: dict[str, int] = {}
for mod_id, dep_ids in deps.items():
for dep_id in dep_ids:
rev_deps[dep_id] = rev_deps.get(dep_id, 0) + 1
rows = []
for m in modules:
mid = m["id"]
rows.append({
"id": mid,
"name": m["name"],
"theorems": m["theorem_count"],
"sorries": m["sorry_count"],
"imports": len(deps.get(mid, [])),
"imported_by": rev_deps.get(mid, 0),
})
rows.sort(key=lambda r: -r["theorems"])
total_thm = sum(r["theorems"] for r in rows)
total_sor = sum(r["sorries"] for r in rows)
return {
"schema": "gremlin_lean_report_v1",
"generated_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
"source": "Gremlin (Azure Cosmos DB, mathblob)",
"summary": {
"total_modules": len(rows),
"total_theorems": total_thm,
"total_sorries": total_sor,
"total_dependency_edges": sum(len(v) for v in deps.values()),
},
"by_theorems": rows[:10],
"most_imported": sorted(rows, key=lambda r: -r["imported_by"])[:10],
"all_modules": rows,
}
def emit_json(report: dict, out_path: Path) -> None:
"""Write JSON report."""
out_path.write_text(json.dumps(report, indent=2) + "\n")
print(f" JSON: {out_path}")
def emit_markdown(report: dict, out_path: Path) -> None:
"""Write Markdown report (importable into AppFlowy/Obsidian)."""
lines = [
f"# Lean Module Dependency Report",
f"",
f"Generated: {report['generated_at']}",
f"Source: {report['source']}",
f"",
f"## Summary",
f"",
f"| Metric | Value |",
f"|--------|-------|",
f"| Modules | {report['summary']['total_modules']} |",
f"| Theorems | {report['summary']['total_theorems']} |",
f"| Sorries | {report['summary']['total_sorries']} |",
f"| Dependency edges | {report['summary']['total_dependency_edges']} |",
f"",
f"## Top 10 Modules by Theorem Count",
f"",
f"| # | Module | Theorems | Sorries | Imports | Imported By |",
f"|---|--------|----------|---------|---------|-------------|",
]
for i, r in enumerate(report["by_theorems"], 1):
lines.append(f"| {i} | {r['name']} | {r['theorems']} | {r['sorries']} | {r['imports']} | {r['imported_by']} |")
lines += [
f"",
f"## Top 10 Most-Imported Modules",
f"",
f"| # | Module | Imported By | Theorems | Sorries |",
f"|---|--------|-------------|----------|---------|",
]
for i, r in enumerate(report["most_imported"], 1):
lines.append(f"| {i} | {r['name']} | {r['imported_by']} | {r['theorems']} | {r['sorries']} |")
lines += [
f"",
f"## All Modules ({len(report['all_modules'])})",
f"",
f"| Module | Theorems | Sorries | Imports | Imported By |",
f"|--------|----------|---------|---------|-------------|",
]
for r in report["all_modules"]:
lines.append(f"| {r['name']} | {r['theorems']} | {r['sorries']} | {r['imports']} | {r['imported_by']} |")
out_path.write_text("\n".join(lines) + "\n")
print(f" MD: {out_path}")
# ── Main ─────────────────────────────────────────────────────────────────
def main():
out_dir = ROOT / "reports"
out_dir.mkdir(parents=True, exist_ok=True)
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
print("[gremlin] Querying Gremlin...")
modules = get_all_modules()
print(f" Found {len(modules)} module vertices")
deps = get_dependency_map()
edge_count = sum(len(v) for v in deps.values())
print(f" Loaded {edge_count} import edges")
print("[gremlin] Building report...")
report = build_report(modules, deps)
s = report["summary"]
print(f"\n {s['total_modules']} modules, {s['total_theorems']} theorems, {s['total_sorries']} sorries")
print(f" {s['total_dependency_edges']} dependency edges")
print(f"\n Top 5 by theorem count:")
for r in report["by_theorems"][:5]:
print(f" {r['theorems']:4d} thm {r['sorries']:3d} sor "
f"imp={r['imports']:2d} dep={r['imported_by']:3d} | {r['name'][:50]}")
print(f"\n Top 5 most-imported:")
for r in report["most_imported"][:5]:
print(f" {r['imported_by']:3d} dependents | {r['name'][:55]}")
json_path = out_dir / f"lean_report_{ts}.json"
md_path = out_dir / f"lean_report_{ts}.md"
emit_json(report, json_path)
emit_markdown(report, md_path)
print(f"\n[gremlin] Reports written to {out_dir}/")
print(f" To push to AppFloyo Cloud, import the Markdown file via the AppFloyo UI.")
if __name__ == "__main__":
main()