#!/usr/bin/env python3 from __future__ import annotations import json import os import re from pathlib import Path from intake_common import * MAX_PROJECTS = 500 MAX_TODO_PER_PROJECT = 40 def markers_for_dir(d: Path): names = set() try: for p in d.iterdir(): if is_sensitive_name(p.name): continue if p.name in PROJECT_MARKERS or p.name.upper().startswith("README"): names.add("README" if p.name.upper().startswith("README") else p.name) except Exception: pass return sorted(names) def tech_for_markers(markers): tech = sorted({TECH_MARKERS[m] for m in markers if m in TECH_MARKERS}) if "README" in markers and not tech: tech.append("docs/notes") return tech or ["unknown"] def find_candidates(roots): candidates = {} raw = [] ts = now_iso() for r in roots: root = expand_path(r["path"]) raw.append({"type": "root", "path": str(root), "exists": root.exists(), "purpose": r.get("purpose", ""), "ts": ts}) if not root.exists(): raw.append({"type": "warning", "path": str(root), "reason": "missing_root", "ts": ts}) continue dirs_seen = 0 for dirpath, dirnames, filenames in os.walk(root): d = Path(dirpath) dirs_seen += 1 if dirs_seen > 20000: raw.append({"type": "warning", "path": str(root), "reason": "max_dirs_reached", "ts": ts}) break dirnames[:] = [x for x in dirnames if not should_skip_dir(d / x)] names = set(filenames) | set(dirnames) markers = sorted({"README" if x.upper().startswith("README") else x for x in names if x in PROJECT_MARKERS or x.upper().startswith("README")}) if markers: # Avoid making every nested subfolder a project if parent already has strong marker, except nested git repos. is_git = ".git" in names key = str(d.resolve()) candidates[key] = {"path": d.resolve(), "markers": markers, "reason": "git_repo" if is_git else "project_markers"} raw.append({"type": "candidate", "path": key, "markers": markers, "reason": candidates[key]["reason"], "ts": ts}) if is_git: dirnames[:] = [x for x in dirnames if x != ".git"] # record plan/agent files as raw metadata only for fn in filenames[:300]: if is_sensitive_name(fn): raw.append({"type": "sensitive_skipped", "path": str(d / fn), "ts": ts}) elif PLAN_RE.search(fn) or fn in {"AGENTS.md", "CLAUDE.md", ".cursorrules"}: raw.append({"type": "agent_or_plan_artifact", "path": str(d / fn), "ts": ts}) return list(candidates.values())[:MAX_PROJECTS], raw def readme_summary(d: Path): readmes = [] try: readmes = [p for p in d.iterdir() if p.is_file() and p.name.upper().startswith("README") and not is_sensitive_name(p.name)] except Exception: return "", None if not readmes: return "", None p = sorted(readmes, key=lambda x: len(x.name))[0] txt = text_snippet(p, max_bytes=150_000, max_chars=1800) if not txt: return "", str(p) lines = [x.strip() for x in txt.splitlines() if x.strip() and not x.strip().startswith(("!", "[!"))] summary = " ".join(lines[:8])[:900] return summary, str(p) def collect_todos(d: Path): todos = [] plan_hits = [] agent_hits = [] files_seen = 0 for path in iter_tree(d, max_dirs=2500, max_files=16000): if files_seen > 16000: break files_seen += 1 if is_sensitive_name(path.name): continue if path.name in {"AGENTS.md", "CLAUDE.md", ".cursorrules"} or any(part in {".hermes", ".codex", ".claude", ".openclaude"} for part in path.parts): agent_hits.append(str(path)) if PLAN_RE.search(path.name): plan_hits.append(str(path)) if path.suffix.lower() not in TEXT_EXT or path.suffix.lower() in BINARY_EXT: continue try: if path.stat().st_size > 160_000: continue text = path.read_text(encoding="utf-8", errors="ignore") except Exception: continue if SECRET_VALUE_RE.search(text[:2000]): continue for i, line in enumerate(text.splitlines()[:1600], 1): if TODO_RE.search(line): todos.append({"file": str(path), "line": i, "text": line.strip()[:260]}) if len(todos) >= MAX_TODO_PER_PROJECT: break if len(todos) >= MAX_TODO_PER_PROJECT: break return todos, plan_hits[:50], agent_hits[:50] def score_project(project): score = 0 reasons = [] if project["is_git_repo"]: score += 8; reasons.append("git repo") if project["git_status"].get("dirty"): score += 20; reasons.append("tracked dirty work") tc = project["todo_fixme_count"] if tc: score += min(25, tc * 2); reasons.append(f"{tc} TODO/FIXME/WIP hints") if project["readme_summary"]: score += 6; reasons.append("README present") if "agent" in project["technologies"] or "hermes-async" in project["technologies"]: score += 10; reasons.append("agentized") if project["plan_hit_count"]: score += min(15, project["plan_hit_count"]); reasons.append("plans/backlogs found") if not project["readme_summary"] and project["technologies"] == ["unknown"]: score -= 8; reasons.append("unclear purpose") if project["path"].startswith(str(Path.home() / ".")): score += 6; reasons.append("agent/config artifact root") status = "review" if project["risk_level"] == "high" or (project["git_status"].get("dirty") and not project["readme_summary"]): status = "needs_human" elif score >= 35: status = "active" elif score >= 18: status = "quick_win" elif score <= 4: status = "archive_candidate" return max(score, 0), reasons, status def main(): ensure_dirs() clear_generated_files(EVIDENCE, "*.json") roots = [r for r in parse_roots() if r.get("enabled", True)] candidates, raw = find_candidates(roots) projects = [] evidence_index = [] for c in candidates: d = c["path"] markers = markers_for_dir(d) or c["markers"] tech = tech_for_markers(markers) is_git = (d / ".git").exists() gs = git_status(d) if is_git else {"branch": "", "remote": "", "dirty": False, "dirty_tracked_count": 0, "last_commit": "", "status_sample": []} rs, readme_path = readme_summary(d) todos, plan_hits, agent_hits = collect_todos(d) risk = "high" if str(d).startswith(str(Path.home() / ".")) else ("medium" if is_git and gs.get("dirty") else "low") pid = f"{safe_slug(d.name)}-{short_id(str(d))}" evidence = { "project_id": pid, "path": str(d), "readme_path": readme_path, "readme_summary": rs, "todo_fixme_snippets": todos, "plan_hits": plan_hits, "agent_artifacts": agent_hits, "secret_policy": "secret-like filenames/content skipped; snippets capped", } ev_path = EVIDENCE / f"{pid}.json" ev_path.write_text(json.dumps(evidence, ensure_ascii=False, indent=2), encoding="utf-8") project = { "project_id": pid, "name": d.name, "path": str(d), "last_modified": dir_mtime(d), "is_git_repo": is_git, "git_status": gs, "markers": markers, "technologies": tech, "readme_path": readme_path, "readme_summary": rs, "todo_fixme_count": len(todos), "todo_fixme_snippets": todos[:8], "plan_hit_count": len(plan_hits), "agent_artifact_count": len(agent_hits), "risk_level": risk, "evidence_file": str(ev_path.relative_to(ROOT)), "detected_reason": c["reason"], } score, reasons, status = score_project(project) project["score"] = score project["score_reasons"] = reasons project["triage_status"] = status projects.append(project) evidence_index.append({"project_id": pid, "evidence_file": project["evidence_file"]}) projects.sort(key=project_sort_key) raw.append({"type": "summary", "projects_found": len(projects), "ts": now_iso()}) write_jsonl(INV / "raw_scan.jsonl", raw) write_jsonl(INV / "projects.jsonl", projects) (EVIDENCE / "evidence_index.json").write_text(json.dumps(evidence_index, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps({"status": "succeeded", "projects": len(projects), "raw_events": len(raw)}, indent=2)) if __name__ == "__main__": main()