← back to Tk11438 Postgres Migration

inventory.py

56 lines

"""Read-only candidate reconciliation. Outputs metadata, never source/credentials.

Regex findings are candidates, not proof of a PostgreSQL connection. Default
hosts, remote deployment targets, shell -h flags, inherited env, dynamic config,
ignored files, and other machines require separate operational review.
"""
import collections
import datetime
import hashlib
import json
from pathlib import Path
import re

ROOT = Path('/Users/macstudio3/Projects')
OUT = Path(__file__).parent / 'verification'
URL = re.compile(r'postgres(?:ql)?://(?:[^/\s\"\x27]+@)?(?:localhost|127\.0\.0\.1|\[::1\])(?::\d+)?(?:/[^\s\"\x27`]+)?')
HOST = re.compile(r'''(?:["']?(?:host|PGHOST|DB_HOST)["']?\s*[:=]\s*(?:(?:process\.env\.[A-Z_]+\s*\|\|)\s*)?["']?(?:localhost|127\.0\.0\.1|::1)(?=["'\s,;}\n]|$))''')
SUFFIXES = {'.js', '.mjs', '.cjs', '.ts', '.tsx', '.py', '.sh', '.zsh', '.sql', '.json', '.yaml', '.yml', '.toml', '.plist'}

def category(p):
    parts = [x.lower() for x in p.parts]
    if any(x in {'archive', 'archives', '_archive', '_archived', 'backup', 'backups', '.next', 'dist', 'build', '.claude', 'verification', 'reports'} for x in parts) or '.bak' in p.name:
        return 'archive/generated/evidence-review'
    if p.suffix in {'.md', '.txt', '.html', '.csv'}:
        return 'documentation/data-review'
    if p.name.startswith('.env'):
        return 'environment-review'
    if p.suffix in SUFFIXES or p.name in {'tk', 'Dockerfile'}:
        return 'source/config-review'
    return 'other-review'

def main():
    original = {(ROOT / x.removeprefix('./')).absolute() for x in (OUT/'original-218-paths.txt').read_text().splitlines() if x}
    current = {Path(x) for x in (OUT/'current-candidate-paths.txt').read_text().splitlines() if x}
    rows = []
    for p in sorted(original | current):
        row = {'path': str(p), 'project': p.relative_to(ROOT).parts[0] if p != ROOT else '(invalid-directory-entry)', 'original_218': p in original, 'current_scan': p in current, 'category': category(p)}
        try:
            raw = p.read_bytes()
            content = raw.decode('utf-8', errors='replace')
            row.update(sha256=hashlib.sha256(raw).hexdigest(), findings=[])
            for n, line in enumerate(content.splitlines(), 1):
                kinds = []
                if URL.search(line): kinds.append('local-postgres-url')
                if HOST.search(line): kinds.append('loopback-host-review-not-necessarily-postgres')
                if kinds: row['findings'].append({'line': n, 'kinds': kinds})
        except OSError as e:
            row['error'] = type(e).__name__
        rows.append(row)
    summary = {'timestamp': datetime.datetime.now(datetime.timezone.utc).isoformat(), 'original': len(original), 'current_candidates': len(current), 'union': len(rows), 'original_not_in_current_scan': len(original-current), 'by_category': dict(collections.Counter(r['category'] for r in rows)), 'original_by_project': dict(collections.Counter(r['project'] for r in rows if r['original_218'])), 'limitations': __doc__}
    OUT.mkdir(exist_ok=True)
    (OUT/'inventory.json').write_text(json.dumps({'summary': summary, 'files': rows}, indent=2)+'\n')
    print(json.dumps(summary, indent=2))

if __name__ == '__main__': main()