ProjectIntakeTower/verifier/focused_category_scan.py

#!/usr/bin/env python3
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]})
                # duplicate file hashing intentionally skipped in focused scan;
                # project-level duplicate heuristic is cheaper and more useful here.

    # duplicated project heuristic: same basename variants and nested related repos
    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]

    # broken projects: dirty + no readme/test, or prior verifier failed if available
    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')

    # Markdown category index
    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')

    # update digest append summary
    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()