import logging import concurrent from pathlib import Path from concurrent.futures import as_completed from tqdm import tqdm from colorama import Fore, Back, Style from inotify_simple import flags as iflags from co3.resources import DiskResource from co3 import Differ, Syncer, Database from execlog.event import Event from execlog.router import CancelledFrameError from execlog.routers import PathRouter from execlog.util.generic import color_text logger = logging.getLogger(__name__) class PathDiffer(Differ[Path]): def __init__( self, database: Database, ): super().__init__(DiskResource(), database) def l_transform(self, item): ''' Transform ``(path, head)`` tuple from ``DiskResource``. ''' return Path(*item) class PathRouterSyncer(Syncer[Path]): def __init__( self, differ: PathDiffer, router: PathRouter, ): super().__init__(differ) self.router = router def _construct_event( self, fpath: str | Path, endpoint: str | Path, action: bytes ): return Event( endpoint=str(endpoint), name=str(Path(fpath).relative_to(endpoint)), action=[action], # synthetic action to match any flag filters ) def handle_l_excl(self, path: Path, disk_pairs: list): ''' Handle disk exclusive paths (i.e., those added to disk since last sync). ''' return [ self._construct_event(str(path), endpoint, iflags.CREATE) for endpoint, _ in disk_pairs ] def handle_r_excl(self, path: Path, db_vals: list): ''' Handle database exclusive paths (i.e., those deleted from disk since last sync). Searches for matching endpoints under the attached router and creates corresponding events. .. admonition:: On lack of endpoints This method handles database exclusive items, i.e., paths no longer on disk but still in the database. For typical Router designs, it is not important to preserve possible endpoints of origin for this kind of event; what matters is the absolute path of the file to be removed. In general, file events are associated solely with a path, but we in some cases may be sensitive to the base path seen to be "triggering" that file event, as router methods can hook in to specific endpoints. This has somewhat dubious effects, as multiple events (with the same action) are dispatched for the same file, purely to observe the Router convention of endpoints and allowing independent trajectories through the execution sequence. One concern here is that you might, in theory, want to respond to the same file deletion event in different ways under different endpoints. This will be accessible when picking up such an event live, as endpoints are grouped by watch descriptor and can be all be triggered from the single file event. This is the one case where we can't *really* simulate the event taking place with the available data, and instead have to peer into the router to see what root paths the file could theoretically trigger. Most of the time, this won't be too problematic, since we'll be watching the same paths and can tell where a deleted file would've been. But there are cases where a watch path endpoint may be abandoned, and thus no callback will be there to receive the DELETE event. *Routers should heavily consider implementing a global DELETE handler to prevent these cases if it's critical to respond to deletions.* Otherwise, we still make an attempt to propagate under appropriate endpoints, allowing for possible "deconstructor-like" behavior of specific filetypes (e.g., cleaning up auxiliary elements, writing to a log, creating a backup, etc). ''' return [ self._construct_event(str(path), str(endpoint), iflags.DELETE) for endpoint in self.router.routemap if Path(path).is_relative_to(Path(endpoint)) ] def handle_lr_int(self, path: Path, path_tuples: tuple[list, list]): ''' Handle paths reflected both in the database and on disk. Paths only reach this method if still present after being passed through ``filter_diff_sets``, which will filter out those files that are up-to-date in the database. ''' return [ self._construct_event(str(path), endpoint, iflags.MODIFY) for endpoint, _ in path_tuples[0] ] def filter_diff_sets(self, l_excl, r_excl, lr_int): total_disk_files = len(l_excl) + len(lr_int) total_joint_files = len(lr_int) def file_out_of_sync(p): _, db_el = lr_int[p] db_mtime = float(db_el[0].get('mtime','0')) disk_mtime = Path(p).stat().st_mtime return disk_mtime > db_mtime lr_int = {p:v for p,v in lr_int.items() if file_out_of_sync(p)} # compute out-of-sync details oos_count = len(l_excl) + len(lr_int) oos_prcnt = oos_count / max(total_disk_files, 1) * 100 logger.info(color_text(f'{len(l_excl)} new files to add', Fore.GREEN)), logger.info(color_text(f'{len(lr_int)} modified files [{total_joint_files} up-to-date]', Fore.YELLOW)), logger.info(color_text(f'{len(r_excl)} files to remove', Fore.RED)), logger.info(color_text(f'({oos_prcnt:.2f}%) of disk files out-of-sync', Style.DIM)), return l_excl, r_excl, lr_int def process_chunk(self, event_sets): chunk_events = [e for event_set in event_sets for e in event_set] # 1) flush synthetic events for the batch through the chained router # 2) block until completed and sweep up the collected inserts event_futures = self.router.submit(chunk_events) # note: we structure this future waiting like this for the TQDM view results = [] success = 0 cancelled = 0 errored = 0 submitted = len(event_futures) progress_bar = tqdm( total=len(event_futures), desc=f'Awaiting chunk futures [submitted {len(event_futures)}]' ) for future in as_completed(event_futures): try: callback_future_list = future.result() results.append(callback_future_list) success += 1 except concurrent.futures.CancelledError as e: cancelled += 1 #logger.error(f'Event future cancelled; "{e}"') except CancelledFrameError as e: errored += 1 pass #logger.error(f'Event frame cancelled; "{e}"') except Exception as e: errored += 1 logger.warning(f'Sync job failed with unknown exception "{e}"') suc_txt = color_text(f'{success}', Fore.GREEN) can_txt = color_text(f'{cancelled}', Fore.YELLOW) err_txt = color_text(f'{errored}', Fore.RED) tot_txt = color_text(f'{success+cancelled+errored}', Style.BRIGHT) progress_bar.set_description( f'Awaiting chunk futures [{tot_txt} / {submitted} | {suc_txt} {can_txt} {err_txt}]' ) progress_bar.update(n=1) progress_bar.close() return results def shutdown(self): super().shutdown() self.router.shutdown()