1import os
 2import os.path
 3import re
 4import sys
 5import threading
 6from collections.abc import Callable
 7
 8import watchfiles
 9
10COMPILED_EXT_RE = re.compile(r"py[co]$")
11
12
13class Reloader(threading.Thread):
14    """File change reloader using watchfiles for cross-platform native file watching."""
15
16    def __init__(self, callback: Callable[[str], None], watch_html: bool) -> None:
17        super().__init__()
18        self.daemon = True
19        self._callback = callback
20        self._watch_html = watch_html
21
22    def get_watch_paths(self) -> set[str]:
23        """Get all directories to watch for changes."""
24        paths = set()
25
26        # Get directories from loaded Python modules
27        for module in tuple(sys.modules.values()):
28            if not hasattr(module, "__file__") or not module.__file__:
29                continue
30            # Convert .pyc/.pyo to .py and get directory
31            file_path = COMPILED_EXT_RE.sub("py", module.__file__)
32            dir_path = os.path.dirname(os.path.abspath(file_path))
33            if os.path.isdir(dir_path):
34                paths.add(dir_path)
35
36        # Add current working directory for .env files
37        cwd = os.getcwd()
38        if os.path.isdir(cwd):
39            paths.add(cwd)
40
41        return paths
42
43    def run(self) -> None:
44        """Watch for file changes and trigger callback."""
45        watch_paths = self.get_watch_paths()
46
47        for changes in watchfiles.watch(*watch_paths, rust_timeout=1000):
48            for change_type, file_path in changes:
49                should_reload = False
50                filename = os.path.basename(file_path)
51
52                # Python files: reload on modify/add
53                if change_type in (
54                    watchfiles.Change.modified,
55                    watchfiles.Change.added,
56                ) and file_path.endswith(".py"):
57                    should_reload = True
58
59                # .env files: reload on modify/add/delete
60                if change_type in (
61                    watchfiles.Change.modified,
62                    watchfiles.Change.added,
63                    watchfiles.Change.deleted,
64                ) and filename.startswith(".env"):
65                    should_reload = True
66
67                # HTML files: only reload on add/delete (Jinja auto-reloads modifications)
68                if (
69                    self._watch_html
70                    and change_type
71                    in (
72                        watchfiles.Change.added,
73                        watchfiles.Change.deleted,
74                    )
75                    and file_path.endswith(".html")
76                ):
77                    should_reload = True
78
79                if should_reload:
80                    self._callback(file_path)