| |
| """Shared utilities for ProjectIntakeTower. |
| |
| Safety rules: |
| - never write outside this repository |
| - never read known secret files |
| - cap text reads aggressively |
| - skip dependency/build/binary trees |
| """ |
| from __future__ import annotations |
|
|
| import fnmatch |
| import hashlib |
| import json |
| import os |
| import re |
| import subprocess |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Any, Dict, Iterable, List, Optional |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| CONFIG = ROOT / "config" |
| INV = ROOT / "inventory" |
| EVIDENCE = INV / "evidence" |
| CARDS = INV / "project_cards" |
| TRIAGE = ROOT / "triage" |
| JOBS = ROOT / "jobs" |
| PENDING = JOBS / "pending" |
| GH_DRAFTS = ROOT / "github" / "issue_drafts" |
|
|
| SCAN_ROOTS_YAML = CONFIG / "scan_roots.yaml" |
|
|
| PROJECT_MARKERS = { |
| ".git", "README", "README.md", "README.txt", "pyproject.toml", "requirements.txt", "setup.py", |
| "package.json", "pnpm-lock.yaml", "yarn.lock", "Cargo.toml", "go.mod", "pom.xml", "build.gradle", |
| "Makefile", "Dockerfile", "docker-compose.yml", "AGENTS.md", "CLAUDE.md", ".cursorrules", ".hermes_async", |
| ".hermes", "next.config.js", "vite.config.ts", "tsconfig.json", "Gemfile", "composer.json" |
| } |
| TECH_MARKERS = { |
| "pyproject.toml": "python", "requirements.txt": "python", "setup.py": "python", "uv.lock": "python", |
| "package.json": "node", "pnpm-lock.yaml": "node", "yarn.lock": "node", "tsconfig.json": "typescript", |
| "Cargo.toml": "rust", "go.mod": "go", "pom.xml": "java", "build.gradle": "java/gradle", |
| "Dockerfile": "docker", "docker-compose.yml": "docker", "AGENTS.md": "agent", "CLAUDE.md": "agent", |
| ".hermes_async": "hermes-async", ".cursorrules": "agent", "Gemfile": "ruby", "composer.json": "php", |
| } |
| IGNORE_DIR_NAMES = { |
| ".git", "node_modules", ".venv", "venv", "env", "__pycache__", ".cache", "dist", "build", ".next", |
| "target", "vendor", ".tox", ".mypy_cache", ".pytest_cache", "coverage", ".turbo", ".pnpm-store" |
| } |
| SENSITIVE_PATTERNS = [ |
| ".env", ".env.*", "*.pem", "*.key", "id_rsa", "id_ed25519", "credentials*", "secrets*", |
| "*secret*", "*token*", "*apikey*", "*api_key*", "*.kdbx", "known_hosts", "authorized_keys" |
| ] |
| BINARY_EXT = { |
| ".png", ".jpg", ".jpeg", ".gif", ".webp", ".ico", ".svgz", ".mp4", ".mov", ".avi", ".mkv", |
| ".zip", ".tar", ".gz", ".tgz", ".xz", ".7z", ".rar", ".pdf", ".sqlite", ".db", ".bin", |
| ".so", ".dylib", ".dll", ".exe", ".class", ".jar", ".pyc", ".pyo", ".woff", ".woff2", ".ttf" |
| } |
| TEXT_EXT = { |
| "", ".md", ".txt", ".rst", ".yaml", ".yml", ".json", ".jsonl", ".toml", ".ini", ".cfg", |
| ".py", ".js", ".ts", ".tsx", ".jsx", ".rs", ".go", ".sh", ".rb", ".php", ".java", ".kt", |
| ".c", ".cpp", ".h", ".hpp", ".css", ".html", ".xml", ".sql" |
| } |
| TODO_RE = re.compile(r"\b(TODO|FIXME|HACK|XXX|BUG|WIP|REVIEW|OPEN|NEXT)\b|^- \[ \]", re.I) |
| PLAN_RE = re.compile(r"(plan|roadmap|backlog|todo|idea|brainstorm|handoff|unfinished|wip)", re.I) |
| SECRET_VALUE_RE = re.compile(r"(BEGIN (RSA |OPENSSH |EC |DSA )?PRIVATE KEY|AWS_SECRET_ACCESS_KEY|ghp_[A-Za-z0-9_]{20,}|sk-[A-Za-z0-9]{20,}|xox[baprs]-)") |
|
|
|
|
| def ensure_dirs() -> None: |
| for d in [CONFIG, INV, EVIDENCE, CARDS, TRIAGE, PENDING, JOBS / "running", JOBS / "done", JOBS / "failed", GH_DRAFTS, ROOT / "scripts", ROOT / "verifier"]: |
| d.mkdir(parents=True, exist_ok=True) |
|
|
|
|
| def clear_generated_files(directory: Path, pattern: str) -> int: |
| removed = 0 |
| if not directory.exists(): |
| return removed |
| for path in directory.glob(pattern): |
| if path.is_file(): |
| path.unlink() |
| removed += 1 |
| return removed |
|
|
|
|
| def now_iso() -> str: |
| return datetime.now(timezone.utc).replace(microsecond=0).isoformat() |
|
|
|
|
| def safe_slug(text: str, fallback: str = "item") -> str: |
| s = re.sub(r"[^A-Za-z0-9_.-]+", "-", text).strip("-._").lower()[:90] |
| return s or fallback |
|
|
|
|
| def short_id(path: str) -> str: |
| return hashlib.sha1(path.encode("utf-8", "ignore")).hexdigest()[:10] |
|
|
|
|
| def parse_roots() -> List[Dict[str, Any]]: |
| if not SCAN_ROOTS_YAML.exists(): |
| return [ |
| {"path": "~/Projekte", "enabled": True, "purpose": "primary_projects"}, |
| {"path": "~/Documents", "enabled": True, "purpose": "documents"}, |
| {"path": "~/Desktop", "enabled": True, "purpose": "desktop"}, |
| {"path": "~/.hermes", "enabled": True, "purpose": "hermes_agent_state"}, |
| {"path": "~/.codex", "enabled": True, "purpose": "codex_agent_state"}, |
| {"path": "~/.claude", "enabled": True, "purpose": "claude_agent_state"}, |
| {"path": "~/.openclaude", "enabled": True, "purpose": "openclaude_agent_state"}, |
| ] |
| roots: List[Dict[str, Any]] = [] |
| cur: Optional[Dict[str, Any]] = None |
| for line in SCAN_ROOTS_YAML.read_text(encoding="utf-8", errors="ignore").splitlines(): |
| s = line.strip() |
| if s.startswith("- path:"): |
| cur = {"path": s.split(":", 1)[1].strip().strip('"\''), "enabled": True, "purpose": ""} |
| roots.append(cur) |
| elif cur and s.startswith("enabled:"): |
| cur["enabled"] = s.split(":", 1)[1].strip().lower() == "true" |
| elif cur and s.startswith("purpose:"): |
| cur["purpose"] = s.split(":", 1)[1].strip().strip('"\'') |
| return roots |
|
|
|
|
| def expand_path(p: str) -> Path: |
| return Path(os.path.expandvars(p)).expanduser().resolve() |
|
|
|
|
| def is_sensitive_name(name: str) -> bool: |
| lower = name.lower() |
| return any(fnmatch.fnmatch(lower, pat.lower()) for pat in SENSITIVE_PATTERNS) |
|
|
|
|
| def is_binary_or_unsafe(path: Path) -> bool: |
| return path.suffix.lower() in BINARY_EXT or is_sensitive_name(path.name) |
|
|
|
|
| def should_skip_dir(path: Path) -> bool: |
| return path.name in IGNORE_DIR_NAMES or is_sensitive_name(path.name) |
|
|
|
|
| def is_under_repo(path: Path) -> bool: |
| try: |
| path.resolve().relative_to(ROOT.resolve()) |
| return True |
| except ValueError: |
| return False |
|
|
|
|
| def text_snippet(path: Path, max_bytes: int = 120_000, max_chars: int = 1200) -> str: |
| if is_binary_or_unsafe(path): |
| return "" |
| try: |
| if path.stat().st_size > max_bytes: |
| return "" |
| data = path.read_text(encoding="utf-8", errors="ignore")[:max_chars] |
| except Exception: |
| return "" |
| if SECRET_VALUE_RE.search(data): |
| return "[redacted: secret-like content detected]" |
| return data.strip() |
|
|
|
|
| def iter_tree(root: Path, max_dirs: int = 20000, max_files: int = 120000) -> Iterable[Path]: |
| dirs_seen = 0 |
| files_seen = 0 |
| for dirpath, dirnames, filenames in os.walk(root): |
| d = Path(dirpath) |
| dirs_seen += 1 |
| if dirs_seen > max_dirs: |
| return |
| dirnames[:] = [x for x in dirnames if not should_skip_dir(d / x)] |
| for fn in filenames: |
| files_seen += 1 |
| if files_seen > max_files: |
| return |
| yield d / fn |
|
|
|
|
| def dir_mtime(path: Path) -> str: |
| latest = 0.0 |
| try: |
| latest = path.stat().st_mtime |
| except Exception: |
| pass |
| try: |
| for p in list(path.iterdir())[:300]: |
| if is_sensitive_name(p.name): |
| continue |
| try: |
| latest = max(latest, p.stat().st_mtime) |
| except Exception: |
| pass |
| except Exception: |
| pass |
| if not latest: |
| return "" |
| return datetime.fromtimestamp(latest, timezone.utc).replace(microsecond=0).isoformat() |
|
|
|
|
| def git_status(repo: Path) -> Dict[str, Any]: |
| env = os.environ.copy() |
| env["GIT_OPTIONAL_LOCKS"] = "0" |
| def run(args: List[str], timeout: int = 5) -> str: |
| try: |
| p = subprocess.run(args, cwd=repo, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, timeout=timeout) |
| return p.stdout.strip() if p.returncode == 0 else "" |
| except Exception: |
| return "" |
| status = run(["git", "status", "--porcelain", "--untracked-files=no"]) |
| branch = run(["git", "rev-parse", "--abbrev-ref", "HEAD"]) |
| remote = run(["git", "remote", "get-url", "origin"]) |
| last = run(["git", "log", "-1", "--format=%cI|%s"], timeout=4) |
| lines = [x for x in status.splitlines() if x.strip()] |
| return { |
| "branch": branch or "unknown", |
| "remote": remote[:300], |
| "dirty_tracked_count": len(lines), |
| "dirty": bool(lines), |
| "last_commit": last[:500], |
| "status_sample": lines[:10], |
| } |
|
|
|
|
| def load_jsonl(path: Path) -> List[Dict[str, Any]]: |
| rows: List[Dict[str, Any]] = [] |
| if not path.exists(): |
| return rows |
| for line in path.read_text(encoding="utf-8", errors="ignore").splitlines(): |
| if not line.strip(): |
| continue |
| rows.append(json.loads(line)) |
| return rows |
|
|
|
|
| def write_jsonl(path: Path, rows: List[Dict[str, Any]]) -> None: |
| path.write_text("\n".join(json.dumps(r, ensure_ascii=False, sort_keys=True) for r in rows) + ("\n" if rows else ""), encoding="utf-8") |
|
|
|
|
| def md_escape(s: Any) -> str: |
| return str(s).replace("|", "\\|").replace("\n", " ") |
|
|
|
|
| def project_sort_key(p: Dict[str, Any]): |
| order = {"needs_human": 0, "active": 1, "quick_win": 2, "review": 3, "merge_candidate": 4, "archive_candidate": 5} |
| return (order.get(p.get("triage_status", "review"), 9), -int(p.get("score", 0)), p.get("name", "")) |
|
|
