← 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()