ProjectIntakeTower/scripts/intake_common.py

#!/usr/bin/env python3
"""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", ""))