Spaces:

IamSamkk
/

mindfull

Configuration error

mindfull / analyze_codebase.py
IamSamk
Mindfull Gradio Space deploy
27caffe 4 months ago
41.6 kB
"""
Codebase Call Graph Analyzer
Traces function calls, assets (model checkpoints, audio, images, configs),
URLs, environment variables, and subprocess commands across all Python files.
Usage: python analyze_codebase.py
"""

import ast
import os
import re
import sys
from pathlib import Path
from collections import defaultdict
from typing import Dict, Set, List, Tuple, Optional

# ─── Asset extension groups ───────────────────────────────────────────────────
ASSET_EXTS = {
    "model":   {".pth", ".pt", ".ckpt", ".safetensors", ".bin", ".onnx", ".mar"},
    "audio":   {".wav", ".mp3", ".flac", ".ogg", ".m4a"},
    "video":   {".mp4", ".avi", ".mov", ".mkv"},
    "image":   {".png", ".jpg", ".jpeg", ".bmp", ".gif", ".webp"},
    "config":  {".json", ".yaml", ".yml", ".cfg", ".ini", ".toml", ".txt"},
    "data":    {".csv", ".npy", ".npz", ".pkl", ".h5", ".hdf5"},
}

# ─── Configuration ────────────────────────────────────────────────────────────

ROOT = Path(__file__).parent

# Directories to skip entirely
SKIP_DIRS = {"venv", "__pycache__", ".git", "node_modules", "gfpgan"}

# Starting entry points (traced in order)
ENTRY_POINTS = [
    "mindfull_web_api.py",
    "mindfull_pipeline.py",
    "mindfull_config.py",
    "simple_video_gen.py",
    "simple_audio_gen.py",
    "sadtalker+wav2lip/enhanced_pipeline.py",
    "sadtalker+wav2lip/simple_pipeline.py",
    "sadtalker+wav2lip/wav2lip/inference.py",
    "sadtalker+wav2lip/sadtalker/inference.py",
    "python313_compat_patch.py",
    "model_manager.py",
    "ollama_to_f5tts.py",
    "infer_cli.py",
]

# ─── Data Structures ──────────────────────────────────────────────────────────

class FunctionInfo:
    def __init__(self, name: str, file: str, lineno: int, is_method: bool = False, class_name: str = None):
        self.name = name
        self.file = file
        self.lineno = lineno
        self.is_method = is_method
        self.class_name = class_name
        self.calls: List[str] = []          # raw call names seen in body
        self.docstring: Optional[str] = None

    def full_name(self):
        if self.class_name:
            return f"{self.class_name}.{self.name}"
        return self.name

    def __repr__(self):
        return f"<Func {self.full_name()} @ {self.file}:{self.lineno}>"


class AssetRef:
    """A non-Python file referenced as a string literal in source code."""
    def __init__(self, path_str: str, asset_type: str, lineno: int, in_func: Optional[str]):
        self.path_str = path_str          # raw string from source
        self.asset_type = asset_type      # model / audio / image / config / …
        self.lineno = lineno
        self.in_func = in_func            # function name, or None for module-level
        self.exists = Path(path_str).exists() or Path(ROOT / path_str).exists()

    def __repr__(self):
        return f"<Asset {self.asset_type}:{self.path_str}>"


class FileInfo:
    def __init__(self, path: str):
        self.path = path                        # relative path from ROOT
        self.imports: Dict[str, str] = {}       # alias -> module string
        self.from_imports: Dict[str, Tuple[str, str]] = {}  # name -> (module, original_name)
        self.functions: List[FunctionInfo] = []
        self.classes: List[str] = []
        self.top_level_calls: List[str] = []    # calls outside any function
        self.third_party: Set[str] = set()      # modules clearly not local
        self.assets: List[AssetRef] = []        # file-path string literals
        self.urls: List[Tuple[int, str, Optional[str]]] = []      # (line, url, func)
        self.env_vars: List[Tuple[int, str, Optional[str]]] = []  # (line, var, func)
        self.subproc_cmds: List[Tuple[int, str, Optional[str]]] = [] # (line, cmd, func)


# ─── Helpers ──────────────────────────────────────────────────────────────────

def collect_py_files(root: Path) -> List[Path]:
    files = []
    for p in root.rglob("*.py"):
        if any(skip in p.parts for skip in SKIP_DIRS):
            continue
        files.append(p)
    return sorted(files)


def rel(path: Path) -> str:
    try:
        return str(path.relative_to(ROOT)).replace("\\", "/")
    except ValueError:
        return str(path).replace("\\", "/")


def module_to_file(module: str, from_file: Path) -> Optional[str]:
    """Try to resolve a module name to a file in the project."""
    parts = module.replace(".", "/")

    # Relative to the file's own directory
    candidates = [
        from_file.parent / (parts + ".py"),
        from_file.parent / parts / "__init__.py",
        ROOT / (parts + ".py"),
        ROOT / parts / "__init__.py",
    ]
    for c in candidates:
        if c.exists():
            return rel(c)
    return None


def get_call_name(node) -> Optional[str]:
    """Extract a simple string name from a Call node."""
    if isinstance(node.func, ast.Name):
        return node.func.id
    if isinstance(node.func, ast.Attribute):
        # e.g. self.foo() -> foo,  obj.method() -> method
        return node.func.attr
    return None


# ─── AST Parser ───────────────────────────────────────────────────────────────

def parse_file(path: Path) -> Optional[FileInfo]:
    info = FileInfo(rel(path))
    try:
        src = path.read_text(encoding="utf-8", errors="ignore")
        tree = ast.parse(src, filename=str(path))
    except SyntaxError as e:
        print(f"  [SKIP] Syntax error in {rel(path)}: {e}")
        return None

    # ── Collect imports ──────────────────────────────────────────

    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                name = alias.asname or alias.name
                info.imports[name] = alias.name
        elif isinstance(node, ast.ImportFrom):
            module = node.module or ""
            for alias in node.names:
                imported_name = alias.asname or alias.name
                info.from_imports[imported_name] = (module, alias.name)

    # ── Collect functions and their calls ────────────────────────

    current_class = [None]

    class Visitor(ast.NodeVisitor):
        def __init__(self):
            self._func_stack: List[FunctionInfo] = []

        def visit_ClassDef(self, node):
            prev = current_class[0]
            current_class[0] = node.name
            info.classes.append(node.name)
            self.generic_visit(node)
            current_class[0] = prev

        def visit_FunctionDef(self, node):
            self._visit_func(node)

        def visit_AsyncFunctionDef(self, node):
            self._visit_func(node)

        def _visit_func(self, node):
            fi = FunctionInfo(
                name=node.name,
                file=info.path,
                lineno=node.lineno,
                is_method=(current_class[0] is not None),
                class_name=current_class[0],
            )
            # Docstring
            if (node.body and isinstance(node.body[0], ast.Expr)
                    and isinstance(node.body[0].value, ast.Constant)
                    and isinstance(node.body[0].value.value, str)):
                fi.docstring = node.body[0].value.value

            # Calls inside this function's body
            for child in ast.walk(node):
                if isinstance(child, ast.Call):
                    cname = get_call_name(child)
                    if cname and cname not in fi.calls:
                        fi.calls.append(cname)

            self._func_stack.append(fi)
            info.functions.append(fi)
            self.generic_visit(node)
            self._func_stack.pop()

    Visitor().visit(tree)

    # Top-level calls (outside functions/classes)
    for node in tree.body:
        if isinstance(node, ast.Expr) and isinstance(node.value, ast.Call):
            cname = get_call_name(node.value)
            if cname:
                info.top_level_calls.append(cname)

    # ── Asset / URL / ENV / subprocess scan ─────────────────────
    _extract_assets(tree, info)

    return info


def _current_func_at_line(info: FileInfo, lineno: int) -> Optional[str]:
    """Return the innermost function name that contains `lineno`."""
    # Functions are ordered by source; find last one whose lineno <= target
    best = None
    for fn in info.functions:
        if fn.lineno <= lineno:
            best = fn.full_name()
    return best


_URL_RE = re.compile(r'https?://[^\s\'"\\>]+')
_ENV_GETENV = {"getenv", "environ"}


_EXEC_NAMES = {"python", "python3", "python.exe", "ffmpeg", "ffmpeg.exe",
               "ffprobe", "git", "pip", "conda", "bash", "sh", "cmd", "powershell"}

def _extract_assets(tree: ast.AST, info: FileInfo) -> None:
    """Walk the AST and pull out asset paths, URLs, env vars, subprocess calls."""

    # ── Pass 1: capture cmd-variable assignments  (cmd = ["ffmpeg", ...]) ──
    # Maps variable name -> (lineno, first-token string)
    _cmd_vars: Dict[str, Tuple[int, str]] = {}
    for node in ast.walk(tree):
        if not isinstance(node, ast.Assign):
            continue
        val = node.value
        if not (isinstance(val, ast.List) and val.elts):
            continue
        first_elt = val.elts[0]
        first_str: Optional[str] = None
        if isinstance(first_elt, ast.Constant) and isinstance(first_elt.value, str):
            first_str = first_elt.value
        elif isinstance(first_elt, ast.Attribute):
            # sys.executable  → keep as placeholder
            first_str = "<python>"
        if first_str is None:
            continue
        lower = first_str.lower().replace("\\\\", "/")
        base = lower.split("/")[-1].split(".")[0]
        if base in _EXEC_NAMES or lower == "<python>":
            # collect all string parts for a readable label
            parts = []
            for el in val.elts[:8]:
                if isinstance(el, ast.Constant) and isinstance(el.value, str):
                    parts.append(el.value)
                elif isinstance(el, ast.Attribute) and el.attr == "executable":
                    parts.append("<python>")
                else:
                    parts.append("...")
            cmd_preview = " ".join(parts)[:120]
            for tgt in node.targets:
                if isinstance(tgt, ast.Name):
                    _cmd_vars[tgt.id] = (node.lineno, cmd_preview)

    for node in ast.walk(tree):
        lineno = getattr(node, "lineno", 0)
        func_ctx = None  # resolved lazily below

        # ── String literals ─────────────────────────────────────
        if isinstance(node, ast.Constant) and isinstance(node.value, str):
            s = node.value.strip()

            # File-path heuristic: contains a known extension
            suffix = Path(s).suffix.lower()
            for atype, exts in ASSET_EXTS.items():
                if suffix in exts and len(s) > 3:
                    if func_ctx is None:
                        func_ctx = _current_func_at_line(info, lineno)
                    info.assets.append(AssetRef(s, atype, lineno, func_ctx))
                    break

            # URL
            if _URL_RE.match(s):
                if func_ctx is None:
                    func_ctx = _current_func_at_line(info, lineno)
                info.urls.append((lineno, s, func_ctx))

        # ── os.environ / os.getenv ───────────────────────────────
        elif isinstance(node, ast.Call):
            fname = get_call_name(node)
            if fname in {"getenv"}:
                if node.args and isinstance(node.args[0], ast.Constant):
                    if func_ctx is None:
                        func_ctx = _current_func_at_line(info, lineno)
                    info.env_vars.append((lineno, node.args[0].value, func_ctx))

            # subprocess.run / call / Popen  with a plain string command
            if fname in {"run", "call", "Popen", "check_output", "check_call", "system"}:
                if node.args:
                    first = node.args[0]
                    cmd_str = None
                    if isinstance(first, ast.Constant) and isinstance(first.value, str):
                        cmd_str = first.value[:120]
                    elif isinstance(first, ast.List) and first.elts:
                        parts = []
                        for el in first.elts[:6]:
                            if isinstance(el, ast.Constant) and isinstance(el.value, str):
                                parts.append(el.value)
                            elif isinstance(el, ast.Attribute) and el.attr == "executable":
                                parts.append("<python>")
                            else:
                                parts.append("...")
                        if parts:
                            cmd_str = " ".join(parts)[:120]
                    elif isinstance(first, ast.Name):
                        # variable — look up in our cmd-variable map
                        if first.id in _cmd_vars:
                            cmd_str = _cmd_vars[first.id][1]
                    if cmd_str:
                        if func_ctx is None:
                            func_ctx = _current_func_at_line(info, lineno)
                        info.subproc_cmds.append((lineno, cmd_str, func_ctx))

        # ── os.environ["VAR"] subscript ──────────────────────────
        elif isinstance(node, ast.Subscript):
            if (isinstance(node.value, ast.Attribute)
                    and node.value.attr == "environ"
                    and isinstance(node.slice, ast.Constant)
                    and isinstance(node.slice.value, str)):
                if func_ctx is None:
                    func_ctx = _current_func_at_line(info, lineno)
                info.env_vars.append((lineno, node.slice.value, func_ctx))


# ─── Resolver ─────────────────────────────────────────────────────────────────

STDLIB_ROOTS = {
    "os", "sys", "re", "json", "time", "subprocess", "pathlib", "shutil",
    "tempfile", "typing", "collections", "functools", "itertools", "logging",
    "argparse", "hashlib", "math", "random", "copy", "io", "abc", "enum",
    "threading", "asyncio", "dataclasses", "datetime", "inspect", "gc",
    "traceback", "warnings", "platform", "struct", "socket", "http",
    "urllib", "uuid", "base64", "csv", "glob", "fnmatch",
}

THIRD_PARTY_ROOTS = {
    "torch", "cv2", "numpy", "np", "PIL", "flask", "fastapi", "requests",
    "scipy", "librosa", "soundfile", "sf", "tqdm", "yaml", "omegaconf",
    "imageio", "skimage", "einops", "transformers", "diffusers", "gradio",
    "pydantic", "uvicorn", "starlette", "aiofiles", "f5_tts", "cached_path",
    "vocos", "safetensors",
}

def is_local(module: str) -> bool:
    root = module.split(".")[0]
    return root not in STDLIB_ROOTS and root not in THIRD_PARTY_ROOTS


def resolve_calls(all_files: Dict[str, FileInfo]) -> Dict[str, List[Tuple[str, str]]]:
    """
    Returns call_graph[file_rel][func_name] = [(target_file, target_func), ...]
    as a flat dict: key = "file::func", value = list of "file::func"
    """
    # Build lookup: function_name -> [(file, FunctionInfo)]
    name_index: Dict[str, List[Tuple[str, FunctionInfo]]] = defaultdict(list)
    for frel, finfo in all_files.items():
        for fn in finfo.functions:
            name_index[fn.name].append((frel, fn))

    call_graph: Dict[str, List[str]] = defaultdict(list)

    for frel, finfo in all_files.items():
        file_path = ROOT / frel.replace("/", os.sep)

        for fn in finfo.functions:
            key = f"{frel}::{fn.full_name()}"
            for raw_call in fn.calls:
                # Check from_imports first
                if raw_call in finfo.from_imports:
                    mod, orig = finfo.from_imports[raw_call]
                    target_file = module_to_file(mod, file_path)
                    if target_file and target_file in all_files:
                        call_graph[key].append(f"{target_file}::{orig}")
                        continue
                    elif not is_local(mod):
                        call_graph[key].append(f"[{mod}]::{orig}")
                        continue

                # Check regular imports (module.func pattern won't hit here, but alias might)
                if raw_call in finfo.imports:
                    mod = finfo.imports[raw_call]
                    target_file = module_to_file(mod, file_path)
                    if target_file:
                        call_graph[key].append(f"{target_file}::<module>")
                        continue

                # Global function name match across project files
                if raw_call in name_index:
                    for tfile, tfn in name_index[raw_call]:
                        entry = f"{tfile}::{tfn.full_name()}"
                        if entry not in call_graph[key]:
                            call_graph[key].append(entry)
                    continue

                # Unknown — could be builtin, stdlib, or third-party
                root_mod = raw_call.split(".")[0]
                if root_mod in THIRD_PARTY_ROOTS:
                    call_graph[key].append(f"[lib:{root_mod}]::{raw_call}")

    return call_graph


# ─── README Generator ─────────────────────────────────────────────────────────

def generate_readme(
    all_files: Dict[str, FileInfo],
    call_graph: Dict[str, List[str]],
    entry_points: List[str],
) -> str:
    lines = []

    lines += [
        "# Codebase Call Graph & Function Map",
        "",
        "Auto-generated by `analyze_codebase.py`.",
        "",
        "**Legend**",
        "- `→` = calls",
        "- `[lib:X]` = third-party library",
        "- `[stdlib:X]` = Python standard library",
        "- Line numbers link to source definitions",
        "",
        "---",
        "",
    ]

    # ── Section 1: File summary ──────────────────────────────────
    lines += ["## 1. Files in Project", ""]
    lines += ["| File | Classes | Functions | Imports |",
              "|------|---------|-----------|---------|"]
    for frel in sorted(all_files):
        fi = all_files[frel]
        classes = ", ".join(fi.classes) if fi.classes else "—"
        func_count = len(fi.functions)
        imp_count = len(fi.imports) + len(fi.from_imports)
        lines.append(f"| `{frel}` | {classes} | {func_count} | {imp_count} |")
    lines += ["", "---", ""]

    # ── Section 2: Entry point trace ────────────────────────────
    lines += ["## 2. Entry Points", ""]
    for ep in entry_points:
        ep_norm = ep.replace("\\", "/")
        if ep_norm in all_files:
            lines.append(f"- **`{ep_norm}`** ✓")
        else:
            lines.append(f"- ~~`{ep_norm}`~~ *(not found)*")
    lines += ["", "---", ""]

    # ── Section 3: Per-file function breakdown ───────────────────
    lines += ["## 3. Function Map  (`filename → function → calls`)", ""]

    for frel in sorted(all_files):
        fi = all_files[frel]
        if not fi.functions:
            continue

        lines += [f"### `{frel}`", ""]

        # Imports summary for this file
        if fi.from_imports or fi.imports:
            lines.append("**Imports:**")
            seen_mods: Set[str] = set()
            for name, (mod, orig) in fi.from_imports.items():
                if mod not in seen_mods:
                    seen_mods.add(mod)
                    resolved = module_to_file(mod, ROOT / frel.replace("/", os.sep))
                    tag = f"→ `{resolved}`" if resolved else ("*(stdlib)*" if not is_local(mod) else "*(external)*")
                    lines.append(f"- `from {mod} import ...`  {tag}")
            for alias, mod in fi.imports.items():
                if mod not in seen_mods:
                    seen_mods.add(mod)
                    resolved = module_to_file(mod, ROOT / frel.replace("/", os.sep))
                    tag = f"→ `{resolved}`" if resolved else ("*(stdlib)*" if not is_local(mod) else "*(external)*")
                    lines.append(f"- `import {mod}`  {tag}")
            lines.append("")

        # Functions
        lines.append("**Functions:**")
        lines.append("")
        lines.append("| # | Function | Line | Calls |")
        lines.append("|---|----------|------|-------|")

        for i, fn in enumerate(fi.functions, 1):
            key = f"{frel}::{fn.full_name()}"
            raw_calls = fn.calls
            resolved_calls = call_graph.get(key, [])

            # Build call cell
            if not raw_calls:
                call_cell = "*(none)*"
            else:
                # Show resolved targets when available, fallback to raw name
                resolved_map = {}
                for rc in resolved_calls:
                    parts = rc.split("::")
                    func_part = parts[-1]
                    resolved_map[func_part] = rc

                call_parts = []
                for c in raw_calls:
                    if c in resolved_map:
                        target = resolved_map[c]
                        if target.startswith("["):
                            call_parts.append(f"`{target}`")
                        else:
                            tfile, tfunc = target.rsplit("::", 1)
                            call_parts.append(f"`{tfunc}` *({tfile})*")
                    else:
                        call_parts.append(f"`{c}`")

                call_cell = ", ".join(call_parts[:8])
                if len(call_parts) > 8:
                    call_cell += f" *+{len(call_parts)-8} more*"

            prefix = f"{fn.class_name}." if fn.class_name else ""
            lines.append(f"| {i} | `{prefix}{fn.name}` | L{fn.lineno} | {call_cell} |")

        lines += ["", ""]

    # ── Section 4: Full call graph (deduplicated) ────────────────
    lines += ["---", "", "## 4. Full Call Graph (all edges)", ""]
    lines += ["```"]
    for key in sorted(call_graph):
        frel, fname = key.split("::", 1)
        short_f = frel.split("/")[-1]
        targets = call_graph[key]
        if targets:
            for t in targets:
                if "::" in t:
                    tfile, tfunc = t.rsplit("::", 1)
                    tshort = tfile.split("/")[-1] if not tfile.startswith("[") else tfile
                    lines.append(f"{short_f}::{fname}  →  {tshort}::{tfunc}")
    lines += ["```", ""]

    # ── Section 5: Assets, URLs, Env vars, Subprocess ──────────
    lines += ["---", "", "## 5. Non-Python Assets Referenced", ""]
    lines += ["> All string literals that look like file paths, grouped by type.", ""]

    # Collect all assets across files
    asset_rows: Dict[str, List[Tuple[str, int, str, bool]]] = defaultdict(list)
    for frel, fi in all_files.items():
        for a in fi.assets:
            asset_rows[a.asset_type].append((frel, a.lineno, a.path_str, a.exists))

    for atype in ["model", "audio", "video", "image", "config", "data"]:
        rows = asset_rows.get(atype, [])
        if not rows:
            continue
        lines.append(f"### {atype.capitalize()} Files")
        lines.append("")
        lines.append("| File | Line | Path | On Disk |")
        lines.append("|------|------|------|---------|")
        seen = set()
        for frel, ln, path_str, exists in sorted(rows, key=lambda x: (x[0], x[1])):
            key = (frel, path_str)
            if key in seen:
                continue
            seen.add(key)
            short_f = frel.split("/")[-1]
            disk = "✓" if exists else "✗ missing"
            safe = path_str.replace("|", "&#124;")[:80]
            lines.append(f"| `{short_f}` | L{ln} | `{safe}` | {disk} |")
        lines.append("")

    # URLs
    lines += ["### URLs Referenced", ""]
    lines += ["| File | Line | URL | In Function |"]
    lines += ["|------|------|-----|-------------|"]
    url_seen = set()
    for frel, fi in sorted(all_files.items()):
        for ln, url, func in fi.urls:
            key = (frel, url)
            if key in url_seen:
                continue
            url_seen.add(key)
            short_f = frel.split("/")[-1]
            func_str = f"`{func}`" if func else "*(module level)*"
            lines.append(f"| `{short_f}` | L{ln} | `{url[:80]}` | {func_str} |")
    if not url_seen:
        lines.append("| — | — | *(none found)* | — |")
    lines.append("")

    # Env vars
    lines += ["### Environment Variables Read", ""]
    lines += ["| File | Line | Variable | In Function |"]
    lines += ["|------|------|----------|-------------|"]
    env_seen = set()
    for frel, fi in sorted(all_files.items()):
        for ln, var, func in fi.env_vars:
            key = (frel, var)
            if key in env_seen:
                continue
            env_seen.add(key)
            short_f = frel.split("/")[-1]
            func_str = f"`{func}`" if func else "*(module level)*"
            lines.append(f"| `{short_f}` | L{ln} | `{var}` | {func_str} |")
    if not env_seen:
        lines.append("| — | — | *(none found)* | — |")
    lines.append("")

    # Subprocess commands
    lines += ["### Subprocess Commands Launched", ""]
    lines += ["| File | Line | Command (truncated) | In Function |"]
    lines += ["|------|------|---------------------|-------------|"]
    cmd_seen = set()
    for frel, fi in sorted(all_files.items()):
        for ln, cmd, func in fi.subproc_cmds:
            key = (frel, cmd[:60])
            if key in cmd_seen:
                continue
            cmd_seen.add(key)
            short_f = frel.split("/")[-1]
            func_str = f"`{func}`" if func else "*(module level)*"
            safe_cmd = cmd.replace("|", "&#124;")[:80]
            lines.append(f"| `{short_f}` | L{ln} | `{safe_cmd}` | {func_str} |")
    if not cmd_seen:
        lines.append("| — | — | *(none found)* | — |")
    lines += ["", "---", ""]

    # ── Section 6: Third-party libraries used ────────────────────
    lines += ["## 6. Third-Party & External Libraries Referenced", ""]
    lib_usages: Dict[str, Set[str]] = defaultdict(set)
    for key, targets in call_graph.items():
        frel = key.split("::")[0]
        for t in targets:
            if t.startswith("[lib:"):
                lib_name = t.split("[lib:")[1].split("]")[0]
                lib_usages[lib_name].add(frel)

    # Also pull from imports directly
    for frel, fi in all_files.items():
        for name, (mod, orig) in fi.from_imports.items():
            root = mod.split(".")[0]
            if root in THIRD_PARTY_ROOTS:
                lib_usages[root].add(frel)
        for alias, mod in fi.imports.items():
            root = mod.split(".")[0]
            if root in THIRD_PARTY_ROOTS:
                lib_usages[root].add(frel)

    lines.append("| Library | Used in |")
    lines.append("|---------|---------|")
    for lib in sorted(lib_usages):
        files_str = ", ".join(f"`{f.split('/')[-1]}`" for f in sorted(lib_usages[lib]))
        lines.append(f"| `{lib}` | {files_str} |")

    lines += ["", "---", "", "*Generated by `analyze_codebase.py` — re-run any time to refresh.*", ""]
    return "\n".join(lines)


# ─── Unused File Detection ────────────────────────────────────────────────────

# Dirs to skip when walking disk for non-Python assets
ASSET_SKIP_DIRS = {"venv", "__pycache__", ".git", "node_modules", "outputs", "temp",
                   "temp_enhanced_pipeline", "datasets-1", "gfpgan"}

# These Python file patterns are entry points by convention — never "unused"
ALWAYS_KEEP_PATTERNS = {
    "analyze_codebase.py",   # this script
    "__init__.py",
    "__main__.py",
    "setup.py",
    "conftest.py",
}


def build_import_graph(all_files: Dict[str, FileInfo]) -> Dict[str, Set[str]]:
    """
    Returns imported_by[file] = set of files that import it.
    Also returns the full set of reachable files when starting from entry points.
    """
    # imported_by: file -> who imports it
    imported_by: Dict[str, Set[str]] = defaultdict(set)

    for frel, fi in all_files.items():
        file_path = ROOT / frel.replace("/", os.sep)

        for alias, mod in fi.imports.items():
            target = module_to_file(mod, file_path)
            if target and target in all_files:
                imported_by[target].add(frel)

        for name, (mod, orig) in fi.from_imports.items():
            target = module_to_file(mod, file_path)
            if target and target in all_files:
                imported_by[target].add(frel)

    return imported_by


def reachable_from_entries(
    entry_points: List[str],
    imported_by: Dict[str, Set[str]],
    all_files: Dict[str, FileInfo],
) -> Set[str]:
    """
    BFS from entry points following imports to find all reachable files.
    We need forward edges (what does a file import), so invert imported_by.
    """
    # Build forward: importer -> set of files it imports
    imports_map: Dict[str, Set[str]] = defaultdict(set)
    for target, importers in imported_by.items():
        for importer in importers:
            imports_map[importer].add(target)

    visited: Set[str] = set()
    queue: List[str] = []

    for ep in entry_points:
        ep_norm = ep.replace("\\", "/")
        if ep_norm in all_files:
            queue.append(ep_norm)

    while queue:
        current = queue.pop()
        if current in visited:
            continue
        visited.add(current)
        for dep in imports_map.get(current, set()):
            if dep not in visited:
                queue.append(dep)

    return visited


def categorise_python_file(frel: str, fi: FileInfo) -> str:
    """Return a human label for why a Python file exists."""
    name = frel.split("/")[-1]
    if name in ALWAYS_KEEP_PATTERNS:
        return "framework/convention"
    if name.startswith("test_"):
        return "test file"
    # Has a __main__ guard
    src_path = ROOT / frel.replace("/", os.sep)
    try:
        src = src_path.read_text(encoding="utf-8", errors="ignore")
        if '__name__ == "__main__"' in src or "__name__ == '__main__'" in src:
            return "standalone script (has __main__)"
    except Exception:
        pass
    if not fi.functions and not fi.classes:
        return "config/constants only"
    return "unreachable module"


def collect_disk_assets(root: Path) -> Dict[str, Path]:
    """
    Walk disk and return all non-Python files with known asset extensions.
    Key = relative path string (forward slashes), Value = absolute Path.
    """
    all_exts: Set[str] = set()
    for exts in ASSET_EXTS.values():
        all_exts.update(exts)

    found: Dict[str, Path] = {}
    for p in root.rglob("*"):
        if not p.is_file():
            continue
        if any(skip in p.parts for skip in ASSET_SKIP_DIRS):
            continue
        if p.suffix.lower() in all_exts:
            found[rel(p)] = p
    return found


def normalise_ref(raw: str) -> str:
    """Strip drive letters, normalise slashes, lowercase for fuzzy matching."""
    # e.g. C:\\Users\\...\\foo.wav  ->  foo.wav (just the basename for fuzzy match)
    p = Path(raw.strip())
    return str(p).replace("\\", "/").lower()


def find_unused(
    all_files: Dict[str, FileInfo],
    entry_points: List[str],
) -> str:
    """Generate the UNUSED_FILES.md content."""

    lines = [
        "# Unused & Unreferenced Files",
        "",
        "Auto-generated by `analyze_codebase.py`.",
        "",
        "> **Unused Python** = never imported by any other file in the project and not a declared entry point.  ",
        "> **Unreferenced Asset** = file exists on disk but no string literal in any `.py` file points to it.  ",
        "> Items marked ⚠️ *might still be used* at runtime via dynamic paths — verify before deleting.",
        "",
        "---",
        "",
    ]

    # ── 1. Python files ──────────────────────────────────────────────────────
    imported_by = build_import_graph(all_files)
    reachable   = reachable_from_entries(entry_points, imported_by, all_files)

    # Normalise entry point list for quick lookup
    ep_set = {ep.replace("\\", "/") for ep in entry_points}

    unused_py: List[Tuple[str, str, Set[str]]] = []  # (frel, reason, imported_by_set)
    for frel, fi in sorted(all_files.items()):
        name = frel.split("/")[-1]
        if frel in ep_set:
            continue
        if name in ALWAYS_KEEP_PATTERNS:
            continue
        if frel in reachable:
            continue
        # Not reachable from any entry point
        reason = categorise_python_file(frel, fi)
        imported_by_set = imported_by.get(frel, set())
        unused_py.append((frel, reason, imported_by_set))

    lines += [
        f"## 1. Python Files Never Imported  ({len(unused_py)} files)",
        "",
        "These files are not reachable from any entry point via import chains.",
        "",
        "| File | Reason | Imported by |",
        "|------|--------|-------------|",
    ]
    for frel, reason, iby in unused_py:
        iby_str = ", ".join(f"`{f.split('/')[-1]}`" for f in sorted(iby)) if iby else "*(nothing)*"
        lines.append(f"| `{frel}` | {reason} | {iby_str} |")

    lines += ["", "---", ""]

    # ── 2. Non-Python assets on disk vs referenced ──────────────────────────
    disk_assets = collect_disk_assets(ROOT)

    # Build a set of all referenced path strings (raw + basename variants)
    ref_strings: Set[str] = set()
    for fi in all_files.values():
        for a in fi.assets:
            s = a.path_str.replace("\\", "/").lower()
            ref_strings.add(s)
            # Also add just the basename so partial paths match
            ref_strings.add(Path(s).name)

    unreferenced: Dict[str, List[Tuple[str, str]]] = defaultdict(list)  # type -> [(frel, disk_path)]
    for disk_rel, disk_abs in sorted(disk_assets.items()):
        # Try full relative path match first
        norm_rel = disk_rel.lower()
        basename  = Path(disk_rel).name.lower()
        if norm_rel in ref_strings or basename in ref_strings:
            continue
        # Try matching any suffix of the disk path against references
        parts = disk_rel.replace("\\", "/").lower().split("/")
        matched = any(
            "/".join(parts[i:]) in ref_strings
            for i in range(len(parts))
        )
        if matched:
            continue

        suffix = disk_abs.suffix.lower()
        atype = "other"
        for t, exts in ASSET_EXTS.items():
            if suffix in exts:
                atype = t
                break
        unreferenced[atype].append((disk_rel, str(disk_abs)))

    total_unref = sum(len(v) for v in unreferenced.values())
    lines += [
        f"## 2. Non-Python Assets on Disk But Never Referenced  ({total_unref} files)",
        "",
        "These files exist in the project folder but no Python source file contains a string that matches their path.",
        "",
    ]

    for atype in ["model", "audio", "video", "image", "config", "data", "other"]:
        items = unreferenced.get(atype, [])
        if not items:
            continue
        lines += [
            f"### {atype.capitalize()} Files  ({len(items)})",
            "",
            "| Path | Size |",
            "|------|------|",
        ]
        for disk_rel, disk_abs in items:
            try:
                size = Path(disk_abs).stat().st_size
                size_str = f"{size:,} B" if size < 1024 else (
                    f"{size//1024:,} KB" if size < 1_048_576 else f"{size//1_048_576:,} MB"
                )
            except Exception:
                size_str = "?"
            lines.append(f"| `{disk_rel}` | {size_str} |")
        lines.append("")

    lines += ["---", ""]

    # ── 3. Referenced assets that are MISSING from disk ─────────────────────
    missing_assets: List[Tuple[str, str, int, str]] = []  # (frel, path_str, line, atype)
    for frel, fi in sorted(all_files.items()):
        for a in fi.assets:
            if not a.exists:
                # Skip obvious format strings and very short tokens
                if "%" in a.path_str or "{" in a.path_str or len(a.path_str) < 5:
                    continue
                # Skip strings that are clearly descriptions, not paths
                if " " in a.path_str and not any(c in a.path_str for c in "\\/"):
                    continue
                missing_assets.append((frel, a.path_str, a.lineno, a.asset_type))

    lines += [
        f"## 3. Assets Referenced in Code But Missing on Disk  ({len(missing_assets)} refs)",
        "",
        "These are paths the code expects to exist but were not found. May indicate missing model downloads.",
        "",
        "| Source File | Line | Expected Path | Type |",
        "|-------------|------|---------------|------|",
    ]
    seen_missing: Set[Tuple[str, str]] = set()
    for frel, path_str, lineno, atype in missing_assets:
        key = (frel, path_str)
        if key in seen_missing:
            continue
        seen_missing.add(key)
        short_f = frel.split("/")[-1]
        safe = path_str.replace("|", "&#124;")[:80]
        lines.append(f"| `{short_f}` | L{lineno} | `{safe}` | {atype} |")

    lines += [
        "",
        "---",
        "",
        "*Generated by `analyze_codebase.py` — re-run any time to refresh.*",
        "",
    ]
    return "\n".join(lines)


# ─── Main ─────────────────────────────────────────────────────────────────────

def main():
    print(f"Scanning: {ROOT}")

    py_files = collect_py_files(ROOT)
    print(f"Found {len(py_files)} Python files\n")

    all_files: Dict[str, FileInfo] = {}
    for p in py_files:
        print(f"  Parsing {rel(p)}")
        fi = parse_file(p)
        if fi:
            all_files[fi.path] = fi

    print(f"\nBuilding call graph...")
    call_graph = resolve_calls(all_files)
    total_edges = sum(len(v) for v in call_graph.values())
    print(f"  {len(call_graph)} functions with outgoing calls, {total_edges} total edges")

    print(f"\nGenerating CODEBASE_MAP.md...")
    readme = generate_readme(all_files, call_graph, ENTRY_POINTS)
    out_path = ROOT / "CODEBASE_MAP.md"
    out_path.write_text(readme, encoding="utf-8")

    total_assets = sum(len(fi.assets) for fi in all_files.values())
    total_urls   = sum(len(fi.urls)   for fi in all_files.values())
    total_envs   = sum(len(fi.env_vars) for fi in all_files.values())
    total_cmds   = sum(len(fi.subproc_cmds) for fi in all_files.values())

    print(f"  Done -> {out_path}")
    print(f"  Files analyzed  : {len(all_files)}")
    print(f"  Total functions : {sum(len(fi.functions) for fi in all_files.values())}")
    print(f"  Call graph edges: {total_edges}")
    print(f"  Asset refs      : {total_assets}")
    print(f"  URLs            : {total_urls}  |  Env vars: {total_envs}  |  Subprocess: {total_cmds}")

    print(f"\nGenerating UNUSED_FILES.md...")
    unused_report = find_unused(all_files, ENTRY_POINTS)
    unused_path = ROOT / "UNUSED_FILES.md"
    unused_path.write_text(unused_report, encoding="utf-8")
    print(f"  Done -> {unused_path}")

    # Quick summary stats for unused report
    imported_by = build_import_graph(all_files)
    reachable   = reachable_from_entries(ENTRY_POINTS, imported_by, all_files)
    ep_set      = {ep.replace("\\", "/") for ep in ENTRY_POINTS}
    unused_count = sum(
        1 for frel, fi in all_files.items()
        if frel not in ep_set
        and frel.split("/")[-1] not in ALWAYS_KEEP_PATTERNS
        and frel not in reachable
    )
    disk_asset_count = len(collect_disk_assets(ROOT))
    print(f"  Unused Python files    : {unused_count}")
    print(f"  Disk assets scanned    : {disk_asset_count}")


if __name__ == "__main__":
    main()