| |
| import json, os, re, hashlib, subprocess |
| from pathlib import Path |
| from collections import defaultdict, Counter |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| OUT = ROOT / 'inventory' / 'evidence' |
| TRIAGE = ROOT / 'triage' |
| GH = ROOT / 'github' / 'issue_drafts' |
| JOBS = ROOT / 'jobs' / 'pending' |
| OUT.mkdir(parents=True, exist_ok=True) |
|
|
| SCAN_ROOTS = [Path('/home/smlflg/Projekte'), Path('/home/smlflg/Documents'), Path('/home/smlflg/Desktop'), Path('/home/smlflg/.hermes'), Path('/home/smlflg/.codex'), Path('/home/smlflg/.claude'), Path('/home/smlflg/.openclaude')] |
| SKIP = {'node_modules','.venv','venv','__pycache__','.cache','dist','build','.next','target','vendor','.git/objects'} |
| SECRET_PAT = re.compile(r'(\.env|pem$|id_rsa|id_ed25519|credential|secret|token|apikey)', re.I) |
| TODO_PAT = re.compile(r'\b(TODO|FIXME)\b|^- \[ \]', re.I) |
| PLAN_PAT = re.compile(r'(plan|roadmap|backlog|idea|brainstorm|handoff|NEXT_ISSUES|AGENT_RESULT)', re.I) |
| THEMES = { |
| 'agent_control_tower_near': re.compile(r'(agent.?control|command.?tower|komm?ando|zentrale|cockpit|dashboard|arena|swarm|orchestr|intake|project.?tower)', re.I), |
| 'hermes_near': re.compile(r'(hermes|skill|profile|gateway|delegate|cronjob|tui|mcp|agentprojectminenraeumer)', re.I), |
| 'uni_near': re.compile(r'(uni|study|lernen|lern|semester|klausur|tutor|course|math|dsa|algorithm|agenttestunisupport)', re.I), |
| } |
| TEST_FILES = re.compile(r'(^test_|_test\.|\.test\.|\.spec\.|tests?$)', re.I) |
| README = re.compile(r'^readme(\..*)?$', re.I) |
| TEXT_EXT = {'.md','.txt','.yaml','.yml','.json','.jsonl','.toml','.ini','.py','.js','.ts','.tsx','.jsx','.rs','.go','.sh'} |
|
|
| def sh(args, cwd=None, timeout=8): |
| try: |
| p = subprocess.run(args, cwd=cwd, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout) |
| return p.returncode, p.stdout.strip(), p.stderr.strip() |
| except Exception as e: |
| return 124, '', str(e) |
|
|
| def skip_path(p): |
| parts = set(p.parts) |
| return bool(parts & SKIP) or any('/'+s in str(p) for s in ['/node_modules/','/.venv/','/venv/','/__pycache__/','/.cache/','/dist/','/build/','/.next/','/target/','/vendor/']) |
|
|
| def walk(root, max_files=6000): |
| n=0 |
| for dp, dns, fns in os.walk(root): |
| d=Path(dp) |
| dns[:] = [x for x in dns if not skip_path(d/x) and x != 'objects'] |
| if skip_path(d): |
| continue |
| for fn in fns: |
| n += 1 |
| if n > max_files: |
| return |
| yield d/fn |
|
|
| def git_repos(): |
| repos=set() |
| for root in SCAN_ROOTS: |
| if not root.exists(): continue |
| if (root/'.git').exists(): repos.add(root) |
| for dp,dns,fns in os.walk(root): |
| d=Path(dp) |
| if skip_path(d): |
| dns[:] = [] |
| continue |
| if '.git' in dns: |
| repos.add(d) |
| dns[:] = [x for x in dns if x != '.git'] |
| dns[:] = [x for x in dns if not skip_path(d/x)] |
| return sorted(repos, key=lambda p: str(p).lower()) |
|
|
| def owner_for(path, repos): |
| best=None |
| for r in repos: |
| try: |
| path.relative_to(r) |
| if best is None or len(str(r)) > len(str(best)): best=r |
| except ValueError: pass |
| return best |
|
|
| def relslug(p): |
| return re.sub(r'[^A-Za-z0-9_.-]+','_', str(p).replace('/home/smlflg/',''))[:120] |
|
|
| def safe_read(path, max_chars=120000): |
| if SECRET_PAT.search(path.name): return None |
| if path.suffix.lower() not in TEXT_EXT and not README.match(path.name): return None |
| try: |
| return path.read_text(encoding='utf-8', errors='ignore')[:max_chars] |
| except Exception: |
| return None |
|
|
| def main(): |
| repos = git_repos() |
| repo_rows=[]; readmes=[]; todos=[]; agent_results=[]; next_issues=[]; plans=[]; tests=[]; broken=[] |
| branch_rows=[]; uncommitted=[]; theme_rows=[]; files_by_hash=defaultdict(list); duplicate_projects=[] |
| repo_set=set(map(str,repos)) |
|
|
| for repo in repos: |
| code,out,err=sh(['git','status','--porcelain=v1','-b'], cwd=repo) |
| status_lines=out.splitlines() if out else [] |
| dirty=[l for l in status_lines if not l.startswith('##')] |
| if dirty: uncommitted.append({'repo':str(repo),'count':len(dirty),'sample':dirty[:20]}) |
| codeb, branches, errb=sh(['git','branch','--format=%(refname:short)|%(committerdate:iso8601)|%(objectname:short)|%(subject)'], cwd=repo) |
| br=[] |
| for line in branches.splitlines(): |
| parts=line.split('|',3) |
| if parts and parts[0]: br.append({'name':parts[0], 'date':parts[1] if len(parts)>1 else '', 'sha':parts[2] if len(parts)>2 else '', 'subject':parts[3] if len(parts)>3 else ''}) |
| if len(br)>1: branch_rows.append({'repo':str(repo),'branches':br}) |
| remote=sh(['git','remote','get-url','origin'], cwd=repo)[1] |
| last=sh(['git','log','-1','--format=%cI|%s'], cwd=repo)[1] |
| repo_rows.append({'path':str(repo),'remote':remote,'last_commit':last,'dirty_count':len(dirty),'branch_count':len(br) or 1}) |
| name_blob=(str(repo)+' '+remote).lower() |
| matched=[k for k,pat in THEMES.items() if pat.search(name_blob)] |
| if matched: theme_rows.append({'repo':str(repo),'themes':matched,'evidence':'path/remote/name match'}) |
|
|
| for root in SCAN_ROOTS: |
| if not root.exists(): continue |
| for p in walk(root): |
| if SECRET_PAT.search(p.name): continue |
| filename_relevant = README.match(p.name) or p.name in {'AGENT_RESULT.md','NEXT_ISSUES.md'} or PLAN_PAT.search(p.name) or TEST_FILES.search(p.name) or '/tests/' in str(p) |
| text=safe_read(p) |
| text_relevant = bool(text and TODO_PAT.search(text[:120000])) |
| if not filename_relevant and not text_relevant: |
| continue |
| own=owner_for(p,repos) |
| if README.match(p.name): readmes.append({'path':str(p),'repo':str(own) if own else None}) |
| if p.name == 'AGENT_RESULT.md': |
| agent_results.append({'path':str(p),'repo':str(own) if own else None, 'preview':(text or '')[:1000]}) |
| if p.name == 'NEXT_ISSUES.md': |
| next_issues.append({'path':str(p),'repo':str(own) if own else None, 'preview':(text or '')[:1000]}) |
| if PLAN_PAT.search(p.name) or any(part in {'.hermes','plans'} for part in p.parts): |
| plans.append({'path':str(p),'repo':str(own) if own else None}) |
| if TEST_FILES.search(p.name) or '/tests/' in str(p): |
| tests.append({'path':str(p),'repo':str(own) if own else None}) |
| if text: |
| for i,line in enumerate(text.splitlines()[:3000],1): |
| if TODO_PAT.search(line): |
| todos.append({'path':str(p),'repo':str(own) if own else None,'line':i,'text':line.strip()[:240]}) |
| |
| |
|
|
| |
| names=defaultdict(list) |
| for r in repos: |
| norm=re.sub(r'[-_ ]?(v\d+|v\d+-v\d+|copy|backup|old|lite|ng|dashboard|ui)$','',r.name.lower()) |
| norm=re.sub(r'[-_ ]+','',norm) |
| names[norm].append(str(r)) |
| for norm,paths in names.items(): |
| if len(paths)>1: |
| duplicate_projects.append({'key':norm,'repos':paths}) |
| duplicate_files=[{'sha1':h,'paths':ps[:20],'count':len(ps)} for h,ps in files_by_hash.items() if len(ps)>1][:500] |
|
|
| |
| test_repos=Counter(x['repo'] for x in tests if x.get('repo')) |
| readme_repos=Counter(x['repo'] for x in readmes if x.get('repo')) |
| for r in repo_rows: |
| reasons=[] |
| if r['dirty_count']>0: reasons.append(f"uncommitted_changes={r['dirty_count']}") |
| if test_repos.get(r['path'],0)==0: reasons.append('no_tests_detected') |
| if readme_repos.get(r['path'],0)==0: reasons.append('no_readme_detected') |
| if reasons and ('no_tests_detected' in reasons or 'no_readme_detected' in reasons): |
| broken.append({'repo':r['path'],'reasons':reasons}) |
|
|
| artifacts = { |
| 'git_repos.json': repo_rows, |
| 'readmes.json': readmes, |
| 'todos_fixmes.json': todos, |
| 'agent_results.json': agent_results, |
| 'next_issues.json': next_issues, |
| 'old_plans.json': plans, |
| 'uncommitted_changes.json': uncommitted, |
| 'open_branches.json': branch_rows, |
| 'tests.json': tests, |
| 'broken_projects.json': broken, |
| 'duplicate_projects.json': duplicate_projects, |
| 'duplicate_files.json': duplicate_files, |
| 'theme_clusters.json': theme_rows, |
| } |
| for name,data in artifacts.items(): (OUT/name).write_text(json.dumps(data,indent=2,ensure_ascii=False),encoding='utf-8') |
|
|
| |
| lines=['# Category Index',''] |
| for name,data in artifacts.items(): |
| lines.append(f'- {name}: {len(data)}') |
| lines += ['', '## Theme counts'] |
| tc=Counter(t for row in theme_rows for t in row['themes']) |
| for k,v in tc.most_common(): lines.append(f'- {k}: {v}') |
| (TRIAGE/'CATEGORY_INDEX.md').write_text('\n'.join(lines)+'\n',encoding='utf-8') |
|
|
| |
| digest = ROOT/'OWNER_DIGEST.md' |
| prev = digest.read_text(encoding='utf-8') if digest.exists() else '# Owner Digest\n' |
| digest.write_text(prev.rstrip()+"\n\n## Focused category scan\n"+'\n'.join(lines[2:])+'\n',encoding='utf-8') |
|
|
| print(json.dumps({k:len(v) for k,v in artifacts.items()},indent=2)) |
|
|
| if __name__=='__main__': main() |
|
|
