| |
| 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: |
| |
| 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"] |
| |
| 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() |
|
|
