"""Discover homemade user directories in a generated app checkout.""" from __future__ import annotations import json import re from collections.abc import Iterable from dataclasses import asdict, dataclass from pathlib import Path from typing import Any from src.domain.codegen.auth_users_adapter import ADAPTER_MARKER, local_admin_user_routes from src.domain.codegen.legacy_store_retire import RETIRED_MARKER _SKIP_DIRS = { ".git", "node_modules", ".nuxt", ".output", "dist", "coverage", ".ruff_cache", "__pycache__", ".venv", "venv", } _CODE_SUFFIXES = {".ts", ".js", ".mjs", ".cjs", ".vue", ".json"} _EMAIL_RE = re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}") _ROLE_RE = re.compile( r"(?:role|user_role|userRole)\s*[\"']?\s*[:=]\s*[\"']" r"(super_admin|superadmin|admin|staff|owner|member|customer|user|student)[\"']", re.IGNORECASE, ) _NAME_RE = re.compile( r"(?:name|full_name|fullName)\s*[\"']?\s*[:=]\s*[\"']([^\"']{1,80})[\"']" ) _PASSWORD_HINT = re.compile(r"password|bcrypt|scrypt|argon2|pbkdf2", re.IGNORECASE) _SEED_HINT = re.compile( r"usr_seed_|users\.json|useStorage\(['\"](?:users|data|auth)|" r"auth:users|SEED_USERS|user-db|records-db|data-bridge|local-auth|app-storage", re.IGNORECASE, ) _ROLE_ALIASES = { "super_admin": "super_admin", "superadmin": "super_admin", "owner": "super_admin", "admin": "admin", "staff": "admin", "administrator": "admin", "counselor": "admin", "admissions": "admin", "member": "member", "customer": "member", "user": "member", "student": "member", } @dataclass(frozen=True) class DiscoveredUser: email: str role: str name: str source: str legacy_data: dict[str, Any] | None = None @dataclass class RepoScan: users: list[DiscoveredUser] local_route_files: list[str] seed_markers: list[str] password_store_files: list[str] record_adapter_present: bool auth_adapter_present: bool def as_dict(self) -> dict[str, Any]: data = asdict(self) data["users"] = [asdict(u) for u in self.users] return data def normalize_role(raw: str | None) -> str: key = (raw or "").strip().lower().replace("-", "_").replace(" ", "_") return _ROLE_ALIASES.get(key, "member") def _iter_code_files(root: Path, *, include_backups: bool = False) -> Iterable[Path]: for path in root.rglob("*"): if not path.is_file(): continue if any(part in _SKIP_DIRS for part in path.parts): continue name = path.name is_backup = name.endswith(".p2hs-local.bak") if is_backup and not include_backups: continue suffix = path.suffix.lower() if is_backup: inner = name[: -len(".p2hs-local.bak")] suffix = Path(inner).suffix.lower() if suffix not in _CODE_SUFFIXES: continue yield path def _rel(root: Path, path: Path) -> str: return str(path.relative_to(root)).replace("\\", "/") def _users_from_json(value: Any, source: str) -> list[DiscoveredUser]: found: list[DiscoveredUser] = [] def walk(node: Any) -> None: if isinstance(node, dict): email = node.get("email") if isinstance(email, str) and _EMAIL_RE.fullmatch(email.strip()): found.append(DiscoveredUser( email=email.strip().lower(), role=normalize_role(str(node.get("role") or "")), name=str(node.get("name") or node.get("full_name") or "").strip(), source=source, )) for child in node.values(): walk(child) elif isinstance(node, list): for child in node: walk(child) walk(value) return found def _users_from_text(text: str, source: str) -> list[DiscoveredUser]: found: list[DiscoveredUser] = [] for match in _EMAIL_RE.finditer(text): email = match.group(0).lower() start = max(0, match.start() - 240) end = min(len(text), match.end() + 240) window = text[start:end] email_at = match.start() - start nearest = None nearest_dist = None for role_match in _ROLE_RE.finditer(window): dist = abs(role_match.start() - email_at) if nearest_dist is None or dist < nearest_dist: nearest = role_match nearest_dist = dist if nearest is None or (nearest_dist or 0) > 160: if "usr_seed_" not in window: continue nearest_dist = 0 name_match = _NAME_RE.search(window) found.append(DiscoveredUser( email=email, role=normalize_role(nearest.group(1) if nearest else "member"), name=(name_match.group(1).strip() if name_match else ""), source=source, )) return found def scan_generated_repo(repo_path: str) -> RepoScan: """Find local admin-user routes and email/role pairs in a generated checkout.""" root = Path(repo_path) users: list[DiscoveredUser] = [] seed_markers: list[str] = [] password_files: list[str] = [] local_routes: list[str] = [] for path in local_admin_user_routes(str(root)): local_routes.append(_rel(root, path)) for path in _iter_code_files(root, include_backups=True): rel = _rel(root, path) try: text = path.read_text(encoding="utf-8") except OSError: continue if ADAPTER_MARKER in text or RETIRED_MARKER in text: continue if _SEED_HINT.search(text): seed_markers.append(rel) password_hint = _PASSWORD_HINT.search(text) admin_route = "admin/users" in rel.replace("\\", "/") if password_hint and (admin_route or "usr_seed_" in text): password_files.append(rel) extracted = [] if path.suffix.lower() == ".json": try: extracted = _users_from_json(json.loads(text), rel) except json.JSONDecodeError: extracted = _users_from_text(text, rel) else: extracted = _users_from_text(text, rel) users.extend(extracted) deduped: dict[str, DiscoveredUser] = {} for user in users: current = deduped.get(user.email) if current is None or (current.role == "member" and user.role != "member"): deduped[user.email] = user record_present = (root / "functions/_lib/p2hs-records.mjs").is_file() or ( root / "server/utils/p2hs-records.mjs" ).is_file() auth_present = (root / "functions/_lib/p2hs-auth-users.mjs").is_file() or ( root / "server/utils/p2hs-auth-users.mjs" ).is_file() return RepoScan( users=list(deduped.values()), local_route_files=local_routes, seed_markers=seed_markers, password_store_files=password_files, record_adapter_present=record_present, auth_adapter_present=auth_present, )