Kyle
0bbcaa8ee5
- History panel defaults to "This diagram" / "All" toggle - "This diagram" filters git log to commits that touched only that diagram's JSON file (git log -- diagrams/0001_*.json) - Automatically switches to "All" when no diagram is open - Restore flow still works per-commit, per-file Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
292 lines
10 KiB
Python
292 lines
10 KiB
Python
import json
|
|
import re
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .. import models
|
|
from ..database import get_db
|
|
|
|
router = APIRouter(tags=["git"])
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────────────
|
|
|
|
def _run(*args, cwd: Path) -> subprocess.CompletedProcess:
|
|
return subprocess.run(
|
|
["git", *args],
|
|
capture_output=True, text=True, cwd=str(cwd), timeout=30,
|
|
)
|
|
|
|
|
|
def _repo_root() -> Path:
|
|
backend_dir = Path(__file__).parent.parent.parent
|
|
r = _run("rev-parse", "--show-toplevel", cwd=backend_dir)
|
|
if r.returncode != 0:
|
|
raise HTTPException(500, "Not in a git repository")
|
|
return Path(r.stdout.strip())
|
|
|
|
|
|
def _diagrams_dir(repo_root: Path) -> Path:
|
|
d = repo_root / "diagrams"
|
|
d.mkdir(exist_ok=True)
|
|
return d
|
|
|
|
|
|
def _safe_name(diagram_id: int, name: str) -> str:
|
|
safe = re.sub(r"[^\w\-]", "_", name)
|
|
return f"{diagram_id:04d}_{safe}.json"
|
|
|
|
|
|
def _diagram_to_dict(diagram, db: Session) -> dict:
|
|
bundle_label = {b.id: b.label for b in diagram.wire_bundles}
|
|
return {
|
|
"id": diagram.id,
|
|
"name": diagram.name,
|
|
"description": diagram.description or "",
|
|
"sheets": [
|
|
{"id": s.id, "name": s.name, "position": s.position}
|
|
for s in diagram.sheets
|
|
],
|
|
"wire_bundles": [
|
|
{"id": b.id, "label": b.label, "jacket_color": b.jacket_color}
|
|
for b in diagram.wire_bundles
|
|
],
|
|
"devices": [
|
|
{
|
|
"id": d.id, "sheet_id": d.sheet_id,
|
|
"device_type": d.device_type, "label": d.label,
|
|
"reference": d.reference, "x": d.x, "y": d.y,
|
|
"width": d.width, "height": d.height,
|
|
"properties": d.properties, "pins": d.pins,
|
|
}
|
|
for d in diagram.devices
|
|
],
|
|
"wires": [
|
|
{
|
|
"id": w.id, "sheet_id": w.sheet_id, "label": w.label,
|
|
"from_device_id": w.from_device_id, "from_pin": w.from_pin,
|
|
"to_device_id": w.to_device_id, "to_pin": w.to_pin,
|
|
"color_primary": w.color_primary, "color_stripe": w.color_stripe,
|
|
"gauge": w.gauge, "length": w.length, "length_unit": w.length_unit,
|
|
"notes": w.notes, "waypoints": w.waypoints or [],
|
|
"twisted_pair": bool(w.twisted_pair), "twist_pitch": w.twist_pitch,
|
|
"shielded": bool(w.shielded), "show_size_label": bool(w.show_size_label),
|
|
"bundle_id": w.bundle_id, "bundle_label": bundle_label.get(w.bundle_id),
|
|
}
|
|
for w in diagram.wires
|
|
],
|
|
}
|
|
|
|
|
|
def _import_data(data: dict, db: Session, name_override: Optional[str] = None) -> models.Diagram:
|
|
new_diag = models.Diagram(
|
|
name=name_override or data.get("name", "Restored Diagram"),
|
|
description=data.get("description", ""),
|
|
)
|
|
db.add(new_diag)
|
|
db.flush()
|
|
|
|
sheet_map = {}
|
|
for s in data.get("sheets", []):
|
|
ns = models.Sheet(
|
|
diagram_id=new_diag.id,
|
|
name=s.get("name", "Sheet 1"),
|
|
position=s.get("position", 0),
|
|
)
|
|
db.add(ns)
|
|
db.flush()
|
|
if "id" in s:
|
|
sheet_map[s["id"]] = ns.id
|
|
|
|
bundle_map = {}
|
|
for b in data.get("wire_bundles", []):
|
|
nb = models.WireBundle(
|
|
diagram_id=new_diag.id,
|
|
label=b.get("label", ""),
|
|
jacket_color=b.get("jacket_color", "#2a2a2a"),
|
|
)
|
|
db.add(nb)
|
|
db.flush()
|
|
if "id" in b:
|
|
bundle_map[b["id"]] = nb.id
|
|
|
|
dev_map = {}
|
|
for dev in data.get("devices", []):
|
|
nd = models.Device(
|
|
diagram_id=new_diag.id,
|
|
sheet_id=sheet_map.get(dev.get("sheet_id")),
|
|
device_type=dev.get("device_type", "generic"),
|
|
label=dev.get("label", ""),
|
|
reference=dev.get("reference", ""),
|
|
x=dev.get("x", 100), y=dev.get("y", 100),
|
|
width=dev.get("width", 120), height=dev.get("height", 60),
|
|
properties=dev.get("properties", {}),
|
|
pins=dev.get("pins", []),
|
|
)
|
|
db.add(nd)
|
|
db.flush()
|
|
if "id" in dev:
|
|
dev_map[dev["id"]] = nd.id
|
|
|
|
for w in data.get("wires", []):
|
|
nw = models.Wire(
|
|
diagram_id=new_diag.id,
|
|
sheet_id=sheet_map.get(w.get("sheet_id")),
|
|
label=w.get("label", ""),
|
|
from_device_id=dev_map.get(w.get("from_device_id")),
|
|
from_pin=w.get("from_pin", ""),
|
|
to_device_id=dev_map.get(w.get("to_device_id")),
|
|
to_pin=w.get("to_pin", ""),
|
|
color_primary=w.get("color_primary", "#FF0000"),
|
|
color_stripe=w.get("color_stripe"),
|
|
gauge=w.get("gauge", "18 AWG"),
|
|
length=w.get("length"),
|
|
length_unit=w.get("length_unit", "in"),
|
|
notes=w.get("notes", ""),
|
|
waypoints=w.get("waypoints", []),
|
|
twisted_pair=w.get("twisted_pair", False),
|
|
twist_pitch=w.get("twist_pitch", 16.0),
|
|
shielded=w.get("shielded", False),
|
|
show_size_label=w.get("show_size_label", False),
|
|
bundle_id=bundle_map.get(w.get("bundle_id")),
|
|
)
|
|
db.add(nw)
|
|
|
|
db.commit()
|
|
db.refresh(new_diag)
|
|
return new_diag
|
|
|
|
|
|
# ── Endpoints ─────────────────────────────────────────────────────────────────
|
|
|
|
@router.get("/status")
|
|
def git_status():
|
|
repo_root = _repo_root()
|
|
|
|
branch_r = _run("rev-parse", "--abbrev-ref", "HEAD", cwd=repo_root)
|
|
branch = branch_r.stdout.strip() if branch_r.returncode == 0 else "unknown"
|
|
|
|
status_r = _run("status", "--porcelain", cwd=repo_root)
|
|
changed = [l for l in status_r.stdout.splitlines() if l.strip()]
|
|
|
|
ahead_r = _run("rev-list", "--count", "@{u}..HEAD", cwd=repo_root)
|
|
ahead = int(ahead_r.stdout.strip()) if ahead_r.returncode == 0 and ahead_r.stdout.strip().isdigit() else 0
|
|
|
|
return {
|
|
"branch": branch,
|
|
"is_clean": len(changed) == 0,
|
|
"changed_count": len(changed),
|
|
"commits_ahead": ahead,
|
|
}
|
|
|
|
|
|
@router.get("/log")
|
|
def git_log(diagram_id: Optional[int] = None):
|
|
repo_root = _repo_root()
|
|
cmd = ["log", "--pretty=format:%H|%h|%s|%an|%ai", "--follow", "-50"]
|
|
if diagram_id is not None:
|
|
# Filter to commits that touched this diagram's file (glob by id prefix)
|
|
cmd += ["--", f"diagrams/{diagram_id:04d}_*.json"]
|
|
r = _run(*cmd, cwd=repo_root)
|
|
if r.returncode != 0:
|
|
return []
|
|
entries = []
|
|
for line in r.stdout.splitlines():
|
|
parts = line.split("|", 4)
|
|
if len(parts) == 5:
|
|
entries.append({
|
|
"hash": parts[0], "short": parts[1],
|
|
"message": parts[2], "author": parts[3], "date": parts[4],
|
|
})
|
|
return entries
|
|
|
|
|
|
class CommitBody(BaseModel):
|
|
message: str
|
|
diagram_id: Optional[int] = None # None = all diagrams
|
|
|
|
|
|
@router.post("/commit")
|
|
def git_commit(body: CommitBody, db: Session = Depends(get_db)):
|
|
if not body.message.strip():
|
|
raise HTTPException(400, "Commit message required")
|
|
repo_root = _repo_root()
|
|
diagrams_dir = _diagrams_dir(repo_root)
|
|
|
|
if body.diagram_id is not None:
|
|
# Export only the specified diagram
|
|
diagram = db.query(models.Diagram).filter(models.Diagram.id == body.diagram_id).first()
|
|
if not diagram:
|
|
raise HTTPException(404, "Diagram not found")
|
|
to_export = [diagram]
|
|
else:
|
|
to_export = db.query(models.Diagram).all()
|
|
|
|
exported_files = []
|
|
for diagram in to_export:
|
|
data = _diagram_to_dict(diagram, db)
|
|
fname = _safe_name(diagram.id, diagram.name)
|
|
(diagrams_dir / fname).write_text(json.dumps(data, indent=2), encoding="utf-8")
|
|
exported_files.append(f"diagrams/{fname}")
|
|
|
|
# Stage only the exported files (not the whole diagrams/ dir)
|
|
add_r = _run("add", *exported_files, cwd=repo_root)
|
|
if add_r.returncode != 0:
|
|
raise HTTPException(500, f"git add failed: {add_r.stderr}")
|
|
|
|
commit_r = _run("commit", "-m", body.message.strip(), cwd=repo_root)
|
|
if commit_r.returncode != 0:
|
|
out = commit_r.stdout + commit_r.stderr
|
|
if "nothing to commit" in out:
|
|
return {"status": "nothing_to_commit", "output": "Nothing to commit — diagram unchanged since last commit"}
|
|
raise HTTPException(500, f"git commit failed: {commit_r.stderr}")
|
|
|
|
return {"status": "committed", "output": commit_r.stdout.strip()}
|
|
|
|
|
|
@router.post("/push")
|
|
def git_push():
|
|
repo_root = _repo_root()
|
|
r = _run("push", cwd=repo_root)
|
|
if r.returncode != 0:
|
|
raise HTTPException(500, f"git push failed: {r.stderr.strip()}")
|
|
return {"status": "pushed", "output": (r.stdout + r.stderr).strip()}
|
|
|
|
|
|
@router.get("/history/{commit_hash}")
|
|
def git_history_files(commit_hash: str):
|
|
repo_root = _repo_root()
|
|
r = _run("ls-tree", "--name-only", commit_hash, "diagrams/", cwd=repo_root)
|
|
if r.returncode != 0:
|
|
return {"files": []}
|
|
files = [f for f in r.stdout.splitlines() if f.endswith(".json")]
|
|
return {"files": files}
|
|
|
|
|
|
class RestoreBody(BaseModel):
|
|
commit_hash: str
|
|
filepath: str # e.g. "diagrams/0001_My_Diagram.json"
|
|
name: Optional[str] = None
|
|
|
|
|
|
@router.post("/restore")
|
|
def git_restore(body: RestoreBody, db: Session = Depends(get_db)):
|
|
repo_root = _repo_root()
|
|
r = _run("show", f"{body.commit_hash}:{body.filepath}", cwd=repo_root)
|
|
if r.returncode != 0:
|
|
raise HTTPException(404, f"File not found in that commit")
|
|
|
|
try:
|
|
data = json.loads(r.stdout)
|
|
except json.JSONDecodeError:
|
|
raise HTTPException(500, "Invalid JSON in git history")
|
|
|
|
short_hash = body.commit_hash[:7]
|
|
name = body.name or f"{data.get('name', 'Diagram')} (restored {short_hash})"
|
|
new_diag = _import_data(data, db, name_override=name)
|
|
return {"id": new_diag.id, "name": new_diag.name}
|