Compare commits

..

2 Commits

3 changed files with 273 additions and 249 deletions

View File

@@ -1,249 +0,0 @@
#!/usr/bin/env python3
"""
Cross-Repo Dependency Graph Builder
Scans repos for import/require/reference patterns and builds a directed
dependency graph. Detects circular dependencies. Outputs DOT and Mermaid.
Usage:
python3 scripts/dependency_graph.py /path/to/repos/
python3 scripts/dependency_graph.py --repos repo1,repo2,repo3 --format mermaid
python3 scripts/dependency_graph.py --repos-dir /path/to/ --format dot --output deps.dot
Patterns detected:
- Python: import X, from X import Y
- JavaScript: require("X"), import ... from "X"
- Go: import "X"
- Ansible: include_role, import_role
- Docker/Compose: image: X, depends_on
- Config references: repo-name in YAML/TOML/JSON
"""
import argparse
import json
import os
import re
import sys
from collections import defaultdict
from pathlib import Path
# Known repo names for matching
KNOWN_REPOS = [
"hermes-agent", "timmy-config", "timmy-home", "the-nexus", "the-door",
"the-beacon", "fleet-ops", "burn-fleet", "timmy-dispatch", "turboquant",
"compounding-intelligence", "the-playground", "second-son-of-timmy",
"ai-safety-review", "the-echo-pattern", "timmy-academy", "wolf",
"the-testament",
]
def normalize_repo_name(name: str) -> str:
"""Normalize a repo name for comparison."""
return name.lower().replace("_", "-").replace(".git", "").strip()
def scan_file_for_deps(filepath: str, content: str, own_repo: str) -> set:
"""Scan a file's content for references to other repos."""
deps = set()
own_norm = normalize_repo_name(own_repo)
for repo in KNOWN_REPOS:
repo_norm = normalize_repo_name(repo)
if repo_norm == own_norm:
continue
# Direct name references
patterns = [
repo, # exact name
repo.replace("-", "_"), # underscore variant
repo.replace("-", ""), # no separator
f"/{repo}/", # path reference
f'"{repo}"', # quoted
f"'{repo}'", # single quoted
f"Timmy_Foundation/{repo}", # full Gitea path
f"Timmy_Foundation.{repo}", # Python module path
]
for pattern in patterns:
if pattern in content:
deps.add(repo)
break
return deps
def scan_repo(repo_path: str, repo_name: str = None) -> dict:
"""Scan a repo directory for dependencies."""
path = Path(repo_path)
if not path.is_dir():
return {"error": f"Not a directory: {repo_path}"}
if not repo_name:
repo_name = path.name
deps = set()
files_scanned = 0
exts = {".py", ".js", ".ts", ".go", ".yaml", ".yml", ".toml", ".json",
".md", ".sh", ".bash", ".Dockerfile", ".tf", ".hcl"}
for fpath in path.rglob("*"):
if not fpath.is_file():
continue
if fpath.suffix not in exts:
continue
# Skip common non-source dirs
parts = fpath.parts
if any(p in (".git", "node_modules", "__pycache__", ".venv", "venv",
"vendor", "dist", "build", ".tox") for p in parts):
continue
try:
content = fpath.read_text(errors="ignore")
except:
continue
file_deps = scan_file_for_deps(str(fpath), content, repo_name)
deps.update(file_deps)
files_scanned += 1
return {
"repo": repo_name,
"dependencies": sorted(deps),
"files_scanned": files_scanned,
}
def detect_cycles(graph: dict) -> list:
"""Detect circular dependencies using DFS."""
cycles = []
visited = set()
rec_stack = set()
def dfs(node, path):
visited.add(node)
rec_stack.add(node)
for neighbor in graph.get(node, {}).get("dependencies", []):
if neighbor not in visited:
result = dfs(neighbor, path + [neighbor])
if result:
return result
elif neighbor in rec_stack:
cycle_start = path.index(neighbor)
return path[cycle_start:] + [neighbor]
rec_stack.remove(node)
return None
for node in graph:
if node not in visited:
cycle = dfs(node, [node])
if cycle:
cycles.append(cycle)
return cycles
def to_dot(graph: dict) -> str:
"""Generate DOT format output."""
lines = ["digraph dependencies {"]
lines.append(" rankdir=LR;")
lines.append(" node [shape=box, style=filled, fillcolor="#1a1a2e", fontcolor="#e6edf3"];")
lines.append(" edge [color="#4a4a6a"];")
lines.append("")
for repo, data in sorted(graph.items()):
dep_count = len(data.get("dependencies", []))
fill = "#2d1b69" if dep_count > 2 else "#16213e"
lines.append(f' "{repo}" [fillcolor="{fill}"];')
for dep in data.get("dependencies", []):
lines.append(f' "{repo}" -> "{dep}";')
lines.append("}")
return "\n".join(lines)
def to_mermaid(graph: dict) -> str:
"""Generate Mermaid format output."""
lines = ["graph LR"]
for repo, data in sorted(graph.items()):
for dep in data.get("dependencies", []):
lines.append(f" {repo.replace('-','_')} --> {dep.replace('-','_')}")
# Add node labels
lines.append("")
for repo in sorted(graph.keys()):
lines.append(f" {repo.replace('-','_')}[{repo}]")
return "\n".join(lines)
def main():
parser = argparse.ArgumentParser(description="Build cross-repo dependency graph")
parser.add_argument("repos_dir", nargs="?", help="Directory containing repos")
parser.add_argument("--repos", help="Comma-separated list of repo paths")
parser.add_argument("--format", choices=["dot", "mermaid", "json"], default="json")
parser.add_argument("--output", "-o", help="Output file (default: stdout)")
parser.add_argument("--cycles-only", action="store_true", help="Only report cycles")
args = parser.parse_args()
results = {}
repo_paths = []
if args.repos:
repo_paths = [p.strip() for p in args.repos.split(",")]
elif args.repos_dir:
base = Path(args.repos_dir)
repo_paths = [str(p) for p in base.iterdir() if p.is_dir() and not p.name.startswith(".")]
else:
parser.print_help()
sys.exit(1)
for rpath in repo_paths:
name = Path(rpath).name
print(f"Scanning {name}...", file=sys.stderr)
result = scan_repo(rpath, name)
if "error" not in result:
results[name] = result
# Detect cycles
cycles = detect_cycles(results)
if args.cycles_only:
if cycles:
print("CIRCULAR DEPENDENCIES DETECTED:")
for cycle in cycles:
print(f" {' -> '.join(cycle)}")
sys.exit(1)
else:
print("No circular dependencies found.")
sys.exit(0)
# Output
output = {}
if args.format == "dot":
output = to_dot(results)
elif args.format == "mermaid":
output = to_mermaid(results)
else:
output = json.dumps({
"repos": results,
"cycles": cycles,
"summary": {
"total_repos": len(results),
"total_deps": sum(len(r["dependencies"]) for r in results.values()),
"cycles_found": len(cycles),
}
}, indent=2)
if args.output:
Path(args.output).write_text(output)
print(f"Written to {args.output}", file=sys.stderr)
else:
print(output)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,162 @@
#!/usr/bin/env python3
"""
Gitea Issue Body Parser
Extracts structured data from Gitea issue markdown bodies:
- Title
- Context section
- Acceptance criteria (checkboxes)
- Labels
- Epic/parent references
Usage:
python3 scripts/gitea_issue_parser.py <issue_body.txt
python3 scripts/gitea_issue_parser.py --url https://forge.../api/v1/repos/.../issues/123
echo "issue body" | python3 scripts/gitea_issue_parser.py --stdin
Output: JSON with {title, context, criteria[], labels[], epic_ref}
"""
import argparse
import json
import re
import sys
from typing import Optional
def parse_issue_body(body: str, title: str = "", labels: list = None) -> dict:
"""Parse a Gitea issue body into structured JSON."""
result = {
"title": title,
"context": "",
"criteria": [],
"labels": labels or [],
"epic_ref": None,
"sections": {},
}
if not body:
return result
# Extract epic/parent reference from title or body
epic_pattern = r"#(\d+)"
title_refs = re.findall(epic_pattern, title)
body_refs = re.findall(epic_pattern, body[:200]) # Check early body refs
# Look for "Closes #N" or "Part of #N" or "Epic: #N"
close_match = re.search(r"(?:Closes?|Fixes?|Resolves?)\s+#(\d+)", body, re.IGNORECASE)
part_match = re.search(r"(?:Part of|Epic|Parent|Blocks?)\s+#(\d+)", body, re.IGNORECASE)
if close_match:
result["epic_ref"] = f"#{close_match.group(1)}"
elif part_match:
result["epic_ref"] = f"#{part_match.group(1)}"
elif title_refs:
result["epic_ref"] = f"#{title_refs[0]}"
elif body_refs:
result["epic_ref"] = f"#{body_refs[0]}"
# Split into sections by ## headers
section_pattern = r"^##\s+(.+)$"
lines = body.split("\n")
current_section = None
current_content = []
for line in lines:
header_match = re.match(section_pattern, line)
if header_match:
# Save previous section
if current_section:
result["sections"][current_section] = "\n".join(current_content).strip()
current_section = header_match.group(1).strip().lower()
current_content = []
else:
current_content.append(line)
# Save last section
if current_section:
result["sections"][current_section] = "\n".join(current_content).strip()
# Extract context
for key in ["context", "background", "description", "problem"]:
if key in result["sections"]:
result["context"] = result["sections"][key]
break
# Extract acceptance criteria (checkboxes)
criteria_section = None
for key in ["acceptance criteria", "acceptance_criteria", "criteria", "requirements", "definition of done"]:
if key in result["sections"]:
criteria_section = result["sections"][key]
break
if criteria_section:
checkbox_pattern = r"-\s*\[[ xX]?\]\s*(.+)"
for match in re.finditer(checkbox_pattern, criteria_section):
result["criteria"].append(match.group(1).strip())
# Also try plain numbered/bulleted lists if no checkboxes found
if not result["criteria"]:
list_pattern = r"^\s*(?:\d+\.|-|\*)\s+(.+)"
for match in re.finditer(list_pattern, criteria_section, re.MULTILINE):
result["criteria"].append(match.group(1).strip())
# If no sectioned criteria found, scan whole body for checkboxes
if not result["criteria"]:
for match in re.finditer(r"-\s*\[[ xX]?\]\s*(.+)", body):
result["criteria"].append(match.group(1).strip())
return result
def parse_from_url(api_url: str, token: str = None) -> dict:
"""Parse an issue from a Gitea API URL."""
import urllib.request
headers = {}
if token:
headers["Authorization"] = f"token {token}"
req = urllib.request.Request(api_url, headers=headers)
resp = json.loads(urllib.request.urlopen(req, timeout=30).read())
title = resp.get("title", "")
body = resp.get("body", "")
labels = [l["name"] for l in resp.get("labels", [])]
return parse_issue_body(body, title, labels)
def main():
parser = argparse.ArgumentParser(description="Parse Gitea issue body into structured JSON")
parser.add_argument("input", nargs="?", help="Issue body file (or - for stdin)")
parser.add_argument("--url", help="Gitea API URL for the issue")
parser.add_argument("--stdin", action="store_true", help="Read from stdin")
parser.add_argument("--token", help="Gitea API token (or set GITEA_TOKEN env var)")
parser.add_argument("--title", default="", help="Issue title (for epic ref extraction)")
parser.add_argument("--labels", nargs="*", default=[], help="Issue labels")
parser.add_argument("--pretty", action="store_true", help="Pretty-print JSON output")
args = parser.parse_args()
import os
token = args.token or os.environ.get("GITEA_TOKEN")
if args.url:
result = parse_from_url(args.url, token)
elif args.stdin or (args.input and args.input == "-"):
body = sys.stdin.read()
result = parse_issue_body(body, args.title, args.labels)
elif args.input:
with open(args.input) as f:
body = f.read()
result = parse_issue_body(body, args.title, args.labels)
else:
parser.print_help()
sys.exit(1)
indent = 2 if args.pretty else None
print(json.dumps(result, indent=indent))
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,111 @@
#!/usr/bin/env python3
"""Tests for gitea_issue_parser."""
import json
import sys
import os
sys.path.insert(0, os.path.dirname(__file__))
from gitea_issue_parser import parse_issue_body
def test_basic_structure():
body = """## Context
This is the background.
## Acceptance Criteria
- [ ] First criterion
- [x] Second criterion (already done)
- [ ] Third criterion
## Labels
`pipeline`, `extraction`
"""
result = parse_issue_body(body, "Test Issue", ["pipeline", "extraction"])
assert result["title"] == "Test Issue"
assert "background" in result["context"].lower()
assert len(result["criteria"]) == 3
assert "First criterion" in result["criteria"]
assert result["labels"] == ["pipeline", "extraction"]
print("PASS: test_basic_structure")
def test_epic_ref():
body = "Closes #645\n\nSome description."
result = parse_issue_body(body, "feat: thing (#688)")
assert result["epic_ref"] == "#645"
print("PASS: test_epic_ref")
def test_epic_ref_from_title():
body = "Some description without close ref."
result = parse_issue_body(body, "feat: scene descriptions (#645)")
assert result["epic_ref"] == "#645"
print("PASS: test_epic_ref_from_title")
def test_no_checkboxes():
body = """## Requirements
1. First thing
2. Second thing
3. Third thing
"""
result = parse_issue_body(body)
assert len(result["criteria"]) == 3
print("PASS: test_no_checkboxes")
def test_empty_body():
result = parse_issue_body("", "Empty Issue")
assert result["title"] == "Empty Issue"
assert result["criteria"] == []
assert result["context"] == ""
print("PASS: test_empty_body")
def test_real_issue_format():
body = """Closes #681
## Changes
Add `#!/usr/bin/env python3` shebang to 6 Python scripts.
## Verification
All 6 files confirmed missing shebangs before fix.
## Impact
Scripts can now be executed directly.
"""
result = parse_issue_body(body, "fix: add python3 shebangs (#685)")
assert result["epic_ref"] == "#681"
assert "shebang" in result["context"].lower()
print("PASS: test_real_issue_format")
def test_all_sections_captured():
body = """## Context
Background info.
## Acceptance Criteria
- [ ] Do thing
## Labels
`test`
"""
result = parse_issue_body(body)
assert "context" in result["sections"]
assert "acceptance criteria" in result["sections"]
print("PASS: test_all_sections_captured")
if __name__ == "__main__":
test_basic_structure()
test_epic_ref()
test_epic_ref_from_title()
test_no_checkboxes()
test_empty_body()
test_real_issue_format()
test_all_sections_captured()
print("\nAll tests passed.")