Compare commits
2 Commits
feat/93-de
...
feat/90-is
| Author | SHA1 | Date | |
|---|---|---|---|
| 2f38c52691 | |||
| b614220eff |
@@ -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()
|
||||
162
scripts/gitea_issue_parser.py
Normal file
162
scripts/gitea_issue_parser.py
Normal 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()
|
||||
111
scripts/test_gitea_issue_parser.py
Normal file
111
scripts/test_gitea_issue_parser.py
Normal 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.")
|
||||
Reference in New Issue
Block a user