sitesync

class SiteSyncThread(threading.Thread):
    """
        Separate thread running synchronization server with asyncio loop.
        Stopped when tray is closed.
    """
    def __init__(self, addon):
        self.log = Logger.get_logger(self.__class__.__name__)
        super(SiteSyncThread, self).__init__()
        self.addon = addon
        self.loop = None
        self.is_running = False
        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=3)
        self.timer = None

    def run(self):
        self.is_running = True

        try:
            self.log.info("Starting SiteSync")
            self.loop = asyncio.new_event_loop()  # create new loop for thread
            asyncio.set_event_loop(self.loop)
            self.loop.set_default_executor(self.executor)

            asyncio.ensure_future(self.check_shutdown(), loop=self.loop)
            asyncio.ensure_future(self.sync_loop(), loop=self.loop)
            self.log.info("SiteSync Started")
            self.loop.run_forever()
        except Exception:
            self.log.warning(
                "SiteSync service has failed", exc_info=True
            )
        finally:
            self.loop.close()  # optional

    async def sync_loop(self):
        """
            Runs permanently, each time:
                - gets list of collections in DB
                - gets list of active remote providers (has configuration,
                    credentials)
                - for each project_name it looks for representations that
                  should be synced
                - synchronize found collections
                - update representations - fills error messages for exceptions
                - waits X seconds and repeat
        Returns:

        """
        while self.is_running and not self.addon.is_paused():
            try:
                start_time = time.time()
                self.addon.set_sync_project_settings()  # clean cache
                project_name = None
                enabled_projects = self.addon.get_enabled_projects()
                for project_name in enabled_projects:
                    await self._sync_project(project_name)

                duration = time.time() - start_time
                self.log.debug("One loop took {:.2f}s".format(duration))

                delay = self.addon.get_loop_delay(project_name)
                self.log.debug(
                    "Waiting for {} seconds to new loop".format(delay)
                )
                self.timer = asyncio.create_task(self.run_timer(delay))
                await asyncio.gather(self.timer)

            except ConnectionResetError:
                self.log.warning(
                    "ConnectionResetError in sync loop, trying next loop",
                    exc_info=True)
            except asyncio.exceptions.CancelledError:
                # cancelling timer
                pass
            except ResumableError:
                self.log.warning(
                    "ResumableError in sync loop, trying next loop",
                    exc_info=True)
            except Exception:
                self.stop()
                self.log.warning(
                    "Unhandled except. in sync loop, stopping server",
                    exc_info=True)

    def stop(self):
        """Sets is_running flag to false, 'check_shutdown' shuts server down"""
        self.is_running = False

    async def check_shutdown(self):
        """ Future that is running and checks if server should be running
            periodically.
        """
        while self.is_running:
            if self.addon.long_running_tasks:
                task = self.addon.long_running_tasks.pop()
                self.log.info("starting long running")
                await self.loop.run_in_executor(None, task["func"])
                self.log.info("finished long running")
                self.addon.projects_processed.remove(task["project_name"])
            await asyncio.sleep(0.5)

        tasks = [
            task
            for task in asyncio.all_tasks()
            if task is not asyncio.current_task()
        ]
        list(map(lambda task: task.cancel(), tasks))  # cancel all the tasks
        results = await asyncio.gather(*tasks, return_exceptions=True)
        self.log.debug(
            f"Finished awaiting cancelled tasks, results: {results}...")
        await self.loop.shutdown_asyncgens()
        # to really make sure everything else has time to stop
        self.executor.shutdown(wait=True)
        await asyncio.sleep(0.07)
        self.loop.stop()

    async def run_timer(self, delay):
        """Wait for 'delay' seconds to start next loop"""
        await asyncio.sleep(delay)

    def reset_timer(self):
        """Called when waiting for next loop should be skipped"""
        self.log.debug("Resetting timer")
        if self.timer:
            self.timer.cancel()
            self.timer = None

    def _working_sites(self, project_name, sync_config):
        if self.addon.is_project_paused(project_name):
            self.log.debug("Both sites same, skipping")
            return None, None

        local_site = self.addon.get_active_site(project_name)
        remote_site = self.addon.get_remote_site(project_name)
        if local_site == remote_site:
            self.log.debug("{}-{} sites same, skipping".format(
                local_site, remote_site))
            return None, None

        local_site_config = sync_config.get("sites")[local_site]
        remote_site_config = sync_config.get("sites")[remote_site]
        if not all([
            _site_is_working(
                self.addon, project_name, local_site, local_site_config
            ),
            _site_is_working(
                self.addon, project_name, remote_site, remote_site_config
            )
        ]):
            self.log.debug((
                "Some of the sites {} - {} in {} is not working properly"
            ).format(local_site, remote_site, project_name))

            return None, None

        return local_site, remote_site

    def _get_remote_provider_info(
        self, project_name, remote_site, site_preset
    ):
        remote_provider = self.addon.get_provider_for_site(site=remote_site)
        handler = lib.factory.get_provider(
            remote_provider,
            project_name,
            remote_site,
            presets=site_preset
        )
        limit = lib.factory.get_provider_batch_limit(remote_provider)

        return handler, remote_provider, limit

    async def _sync_project(self, project_name):
        self.log.info(f"Processing '{project_name}'")
        preset = self.addon.sync_project_settings[project_name]

        local_site, remote_site = self._working_sites(
            project_name, preset
        )
        if not local_site or not remote_site:
            return

        remote_site_preset = preset.get("sites")[remote_site]

        handler, remote_provider, limit = self._get_remote_provider_info(
            project_name,
            remote_site,
            remote_site_preset
        )

        repre_states = self.addon.get_sync_representations(
            project_name,
            local_site,
            remote_site,
            limit
        )

        task_files_to_process = []
        files_processed_info = []
        # process only unique file paths in one batch
        # multiple representation could have same file path
        # (textures),
        # upload process can find already uploaded file and
        # reuse same id
        processed_file_path = set()

        # first call to get_provider could be expensive, its
        # building folder tree structure in memory
        # call only if needed, eg. DO_UPLOAD or DO_DOWNLOAD
        for repre_state in repre_states:
            repre_id = repre_state["representationId"]
            # QUESTION Why is not project passed in?
            # QUESTION Why there is not option to check all representations
            #    in one batch?
            if self.addon.is_representation_paused(repre_id):
                continue
            file_states = repre_state.get("files") or []
            for file_state in file_states:
                # skip already processed files
                # WARNING Using empty string for path is dangerous!!!
                file_path = file_state.get("path", "")
                if file_path in processed_file_path:
                    continue
                status = self.addon.check_status(
                    file_state,
                    local_site,
                    remote_site,
                    preset.get("config")
                )
                if (status == SyncStatus.DO_UPLOAD and
                        len(task_files_to_process) < limit):
                    tree = handler.get_tree()
                    task = asyncio.create_task(
                        upload(
                            self.addon,
                            project_name,
                            file_state,
                            repre_state,
                            remote_provider,
                            remote_site,
                            tree,
                            remote_site_preset
                        )
                    )
                    task_files_to_process.append(task)
                    # store info for exception handlingy
                    files_processed_info.append((
                        file_state,
                        repre_state,
                        remote_site,
                        "remote",
                        project_name
                    ))
                    processed_file_path.add(file_path)

                if (status == SyncStatus.DO_DOWNLOAD and
                        len(task_files_to_process) < limit):
                    tree = handler.get_tree()
                    task = asyncio.create_task(
                        download(
                            self.addon,
                            project_name,
                            file_state,
                            repre_state,
                            remote_provider,
                            remote_site,
                            tree,
                            remote_site_preset
                        )
                    )
                    task_files_to_process.append(task)

                    files_processed_info.append((
                        file_state,
                        repre_state,
                        local_site,
                        "local",
                        project_name
                    ))
                    processed_file_path.add(file_path)

        self.log.debug("Sync tasks count {}".format(
            len(task_files_to_process)
        ))
        files_created = await asyncio.gather(
            *task_files_to_process,
            return_exceptions=True
        )

        for file_result, info in zip(files_created, files_processed_info):
            file_state, repre_status, site_name, side, project_name = info
            error = None
            if isinstance(file_result, BaseException):
                error = str(file_result)
                self.log.warning(error, exc_info=True)
                file_result = None  # it is exception >> no id >> reset

            self.addon.update_db(
                project_name=project_name,
                new_file_id=file_result,
                file=file_state,
                repre_status=repre_status,
                site_name=site_name,
                side=side,
                error=error
            )

            repre_id = repre_status["representationId"]
            self.addon.handle_alternate_site(
                project_name,
                repre_id,
                site_name,
                file_state["fileHash"]
            )