from dataclasses import dataclass from pathlib import Path from typing import Awaitable, Callable, Iterable, Sequence from watchfiles import Change, awatch import asyncio import uuid from src.common.logger import get_logger logger = get_logger("file_watcher") @dataclass(frozen=True) class FileChange: change_type: Change path: Path ChangeCallback = Callable[[Sequence[FileChange]], Awaitable[None] | None] @dataclass(frozen=True) class FileWatchSubscription: subscription_id: str callback: ChangeCallback paths: tuple[Path, ...] change_types: frozenset[Change] | None @dataclass class SubscriptionState: consecutive_failures: int = 0 cooldown_until_monotonic: float = 0.0 @dataclass class FileWatcherStats: batches_seen: int = 0 changes_seen: int = 0 callbacks_succeeded: int = 0 callbacks_failed: int = 0 callbacks_timed_out: int = 0 callbacks_skipped_cooldown: int = 0 restart_count: int = 0 class FileWatcher: def __init__( self, paths: Iterable[Path], debounce_ms: int = 600, callback_timeout_s: float = 10.0, callback_failure_threshold: int = 3, callback_cooldown_s: float = 30.0, force_polling: bool = False, ) -> None: self._paths = [path.resolve() for path in paths] self._debounce_ms = debounce_ms self._force_polling = force_polling self._callback_timeout_s = callback_timeout_s self._callback_failure_threshold = callback_failure_threshold self._callback_cooldown_s = callback_cooldown_s self._running = False self._task: asyncio.Task[None] | None = None self._subscriptions: dict[str, FileWatchSubscription] = {} self._subscription_states: dict[str, SubscriptionState] = {} self._stats = FileWatcherStats() @property def running(self) -> bool: return self._running @property def stats(self) -> FileWatcherStats: return FileWatcherStats( batches_seen=self._stats.batches_seen, changes_seen=self._stats.changes_seen, callbacks_succeeded=self._stats.callbacks_succeeded, callbacks_failed=self._stats.callbacks_failed, callbacks_timed_out=self._stats.callbacks_timed_out, callbacks_skipped_cooldown=self._stats.callbacks_skipped_cooldown, restart_count=self._stats.restart_count, ) def subscribe( self, callback: ChangeCallback, *, paths: Iterable[Path] | None = None, change_types: Iterable[Change] | None = None, ) -> str: if not callable(callback): raise TypeError("callback 必须是可调用对象") normalized_paths = tuple(path.resolve() for path in paths) if paths is not None else () normalized_change_types = frozenset(change_types) if change_types is not None else None subscription_id = str(uuid.uuid4()) self._subscriptions[subscription_id] = FileWatchSubscription( subscription_id=subscription_id, callback=callback, paths=normalized_paths, change_types=normalized_change_types, ) self._subscription_states[subscription_id] = SubscriptionState() return subscription_id def unsubscribe(self, subscription_id: str) -> bool: removed = self._subscriptions.pop(subscription_id, None) is not None self._subscription_states.pop(subscription_id, None) return removed async def start(self) -> None: if self._running: return if not self._subscriptions: raise RuntimeError("启动文件监视器前必须至少注册一个订阅") self._running = True self._ready_event = asyncio.Event() self._task = asyncio.create_task(self._run()) await self._ready_event.wait() async def stop(self) -> None: if not self._running: return self._running = False if self._task is None: return self._task.cancel() try: await self._task except asyncio.CancelledError: return finally: self._task = None async def _run(self) -> None: while self._running: try: async for changes in awatch( *self._paths, debounce=self._debounce_ms, force_polling=self._force_polling, yield_on_timeout=True, ): if not self._ready_event.is_set(): self._ready_event.set() if not self._running: break if not changes: continue normalized_changes = self._normalize_changes(changes) if not normalized_changes: continue self._stats.batches_seen += 1 self._stats.changes_seen += len(normalized_changes) try: await self._dispatch_changes(normalized_changes) except Exception as exc: logger.warning(f"文件变更分发失败: {exc}") except asyncio.CancelledError: return except Exception as exc: self._stats.restart_count += 1 logger.error(f"文件监视器运行异常，将在1秒后重试: {exc}") if self._running: await asyncio.sleep(1.0) async def _dispatch_changes(self, changes: Sequence[FileChange]) -> None: for subscription in list(self._subscriptions.values()): matched_changes = self._match_changes(changes, subscription) if not matched_changes: continue state = self._subscription_states.get(subscription.subscription_id) if state is None: continue now_monotonic = asyncio.get_running_loop().time() if state.cooldown_until_monotonic > now_monotonic: self._stats.callbacks_skipped_cooldown += 1 continue try: await asyncio.wait_for(self._invoke_callback(subscription.callback, matched_changes), timeout=self._callback_timeout_s) state.consecutive_failures = 0 self._stats.callbacks_succeeded += 1 except asyncio.TimeoutError: self._stats.callbacks_timed_out += 1 self._stats.callbacks_failed += 1 self._mark_callback_failure(subscription.subscription_id) logger.warning( f"文件变更回调执行超时（subscription_id={subscription.subscription_id}, timeout={self._callback_timeout_s}s）" ) except Exception as exc: self._stats.callbacks_failed += 1 self._mark_callback_failure(subscription.subscription_id) logger.warning(f"文件变更回调执行失败（subscription_id={subscription.subscription_id}）: {exc}") async def _invoke_callback(self, callback: ChangeCallback, changes: Sequence[FileChange]) -> None: if asyncio.iscoroutinefunction(callback): await callback(changes) return await asyncio.to_thread(callback, changes) def _mark_callback_failure(self, subscription_id: str) -> None: state = self._subscription_states.get(subscription_id) if state is None: return state.consecutive_failures += 1 if state.consecutive_failures >= self._callback_failure_threshold: now_monotonic = asyncio.get_running_loop().time() state.cooldown_until_monotonic = now_monotonic + self._callback_cooldown_s state.consecutive_failures = 0 logger.warning( f"文件变更回调进入冷却（subscription_id={subscription_id}, cooldown={self._callback_cooldown_s}s）" ) def _match_changes(self, changes: Sequence[FileChange], subscription: FileWatchSubscription) -> list[FileChange]: matched: list[FileChange] = [] for change in changes: if subscription.change_types is not None and change.change_type not in subscription.change_types: continue if subscription.paths and not any(self._path_matches(change.path, path) for path in subscription.paths): continue matched.append(change) return matched def _path_matches(self, changed_path: Path, subscribed_path: Path) -> bool: if subscribed_path.is_dir(): return changed_path == subscribed_path or changed_path.is_relative_to(subscribed_path) return changed_path == subscribed_path def _normalize_changes(self, changes: set[tuple[Change, str]]) -> list[FileChange]: return [FileChange(change_type=change, path=Path(path).resolve()) for change, path in changes]