lookyloo/lookyloo.py
def _set_capture_cache(self, capture_dir: Path, force: bool=False, redis_pipeline: Optional[Redis]=None) -> None:
        '''Populate the redis cache for a capture. Mostly used on the index page.'''
        if force or not self.redis.exists(str(capture_dir)):
            # (re)build cache
            pass
        else:
            return

        with (capture_dir / 'uuid').open() as f:
            uuid = f.read().strip()

        har_files = sorted(capture_dir.glob('*.har'))

        error_cache: Dict[str, str] = {}
        if (capture_dir / 'error.txt').exists():
            # Something went wrong
            with (capture_dir / 'error.txt').open() as _error:
                content = _error.read()
                try:
                    error_to_cache = json.loads(content)
                    if isinstance(error_to_cache, dict) and error_to_cache.get('details'):
                        error_to_cache = error_to_cache.get('details')
                except json.decoder.JSONDecodeError:
                    # old format
                    error_to_cache = content
                error_cache['error'] = f'The capture {capture_dir.name} has an error: {error_to_cache}'

        fatal_error = False
        if har_files:
            try:
                har = HarFile(har_files[0], uuid)
            except Har2TreeError as e:
                error_cache['error'] = str(e)
                fatal_error = True
        else:
            error_cache['error'] = f'No har files in {capture_dir.name}'
            fatal_error = True

        if (capture_dir / 'categories').exists():
            with (capture_dir / 'categories').open() as _categories:
                categories = [c.strip() for c in _categories.readlines()]
        else:
            categories = []

        if not redis_pipeline:
            p = self.redis.pipeline()
        else:
            p = redis_pipeline  # type: ignore
        p.hset('lookup_dirs', uuid, str(capture_dir))
        if error_cache:
            if 'HTTP Error' not in error_cache['error']:
                self.logger.warning(error_cache['error'])
            p.hmset(str(capture_dir), error_cache)  # type: ignore

        if not fatal_error:
            redirects = har.initial_redirects
            incomplete_redirects = False
            if redirects and har.need_tree_redirects:
                # load tree from disk, get redirects
                ct = load_pickle_tree(capture_dir)
                if ct:
                    redirects = ct.redirects
                else:
                    # Pickle not available
                    incomplete_redirects = True

            cache: Dict[str, Union[str, int]] = {'uuid': uuid,
                                                 'title': har.initial_title,
                                                 'timestamp': har.initial_start_time,
                                                 'url': har.root_url,
                                                 'redirects': json.dumps(redirects),
                                                 'categories': json.dumps(categories),
                                                 'capture_dir': str(capture_dir),
                                                 'incomplete_redirects': 1 if incomplete_redirects else 0}
            if (capture_dir / 'no_index').exists():  # If the folders claims anonymity
                cache['no_index'] = 1

            if (capture_dir / 'parent').exists():  # The capture was initiated from an other one
                with (capture_dir / 'parent').open() as f:
                    cache['parent'] = f.read().strip()

            p.hmset(str(capture_dir), cache)  # type: ignore
        if not redis_pipeline:
            p.execute()
        # If the cache is re-created for some reason, pop from the local cache.
        self._captures_index.pop(uuid, None)
Populate the redis cache for a capture. Mostly used on the index page.
Populate the redis cache for a capture. Mostly used on the index page.
Populate the redis cache for a capture . Mostly used on the index page .
def _set_capture_cache(self, capture_dir: Path, force: bool=False, redis_pipeline: Optional[Redis]=None) -> None:
        if force or not self.redis.exists(str(capture_dir)):
            pass
        else:
            return
        with (capture_dir / 'uuid').open() as f:
            uuid = f.read().strip()
        har_files = sorted(capture_dir.glob('*.har'))
        error_cache: Dict[str, str] = {}
        if (capture_dir / 'error.txt').exists():
            with (capture_dir / 'error.txt').open() as _error:
                content = _error.read()
                try:
                    error_to_cache = json.loads(content)
                    if isinstance(error_to_cache, dict) and error_to_cache.get('details'):
                        error_to_cache = error_to_cache.get('details')
                except json.decoder.JSONDecodeError:
                    error_to_cache = content
                error_cache['error'] = f'The capture {capture_dir.name} has an error: {error_to_cache}'
        fatal_error = False
        if har_files:
            try:
                har = HarFile(har_files[0], uuid)
            except Har2TreeError as e:
                error_cache['error'] = str(e)
                fatal_error = True
        else:
            error_cache['error'] = f'No har files in {capture_dir.name}'
            fatal_error = True
        if (capture_dir / 'categories').exists():
            with (capture_dir / 'categories').open() as _categories:
                categories = [c.strip() for c in _categories.readlines()]
        else:
            categories = []
        if not redis_pipeline:
            p = self.redis.pipeline()
        else:
            p = redis_pipeline  
        p.hset('lookup_dirs', uuid, str(capture_dir))
        if error_cache:
            if 'HTTP Error' not in error_cache['error']:
                self.logger.warning(error_cache['error'])
            p.hmset(str(capture_dir), error_cache)  
        if not fatal_error:
            redirects = har.initial_redirects
            incomplete_redirects = False
            if redirects and har.need_tree_redirects:
                ct = load_pickle_tree(capture_dir)
                if ct:
                    redirects = ct.redirects
                else:
                    incomplete_redirects = True
            cache: Dict[str, Union[str, int]] = {'uuid': uuid,
                                                 'title': har.initial_title,
                                                 'timestamp': har.initial_start_time,
                                                 'url': har.root_url,
                                                 'redirects': json.dumps(redirects),
                                                 'categories': json.dumps(categories),
                                                 'capture_dir': str(capture_dir),
                                                 'incomplete_redirects': 1 if incomplete_redirects else 0}
            if (capture_dir / 'no_index').exists():  
                cache['no_index'] = 1
            if (capture_dir / 'parent').exists():  
                with (capture_dir / 'parent').open() as f:
                    cache['parent'] = f.read().strip()
            p.hmset(str(capture_dir), cache)  
        if not redis_pipeline:
            p.execute()
        self._captures_index.pop(uuid, None)
def _set_capture_cache ( self , capture_dir : Path , force : bool = False , redis_pipeline : Optional [ Redis ] = None ) -> None : if force or not self . redis . exists ( str ( capture_dir ) ) : pass else : return with ( capture_dir / 'uuid' ) . open ( ) as f : uuid = f . read ( ) . strip ( ) har_files = sorted ( capture_dir . glob ( '*.har' ) ) error_cache : Dict [ str , str ] = { } if ( capture_dir / 'error.txt' ) . exists ( ) : with ( capture_dir / 'error.txt' ) . open ( ) as _error : content = _error . read ( ) try : error_to_cache = json . loads ( content ) if isinstance ( error_to_cache , dict ) and error_to_cache . get ( 'details' ) : error_to_cache = error_to_cache . get ( 'details' ) except json . decoder . JSONDecodeError : error_to_cache = content error_cache [ 'error' ] = f'The capture {capture_dir.name} has an error: {error_to_cache}' fatal_error = False if har_files : try : har = HarFile ( har_files [ 0 ] , uuid ) except Har2TreeError as e : error_cache [ 'error' ] = str ( e ) fatal_error = True else : error_cache [ 'error' ] = f'No har files in {capture_dir.name}' fatal_error = True if ( capture_dir / 'categories' ) . exists ( ) : with ( capture_dir / 'categories' ) . open ( ) as _categories : categories = [ c . strip ( ) for c in _categories . readlines ( ) ] else : categories = [ ] if not redis_pipeline : p = self . redis . pipeline ( ) else : p = redis_pipeline p . hset ( 'lookup_dirs' , uuid , str ( capture_dir ) ) if error_cache : if 'HTTP Error' not in error_cache [ 'error' ] : self . logger . warning ( error_cache [ 'error' ] ) p . hmset ( str ( capture_dir ) , error_cache ) if not fatal_error : redirects = har . initial_redirects incomplete_redirects = False if redirects and har . need_tree_redirects : ct = load_pickle_tree ( capture_dir ) if ct : redirects = ct . redirects else : incomplete_redirects = True cache : Dict [ str , Union [ str , int ] ] = { 'uuid' : uuid , 'title' : har . initial_title , 'timestamp' : har . initial_start_time , 'url' : har . root_url , 'redirects' : json . dumps ( redirects ) , 'categories' : json . dumps ( categories ) , 'capture_dir' : str ( capture_dir ) , 'incomplete_redirects' : 1 if incomplete_redirects else 0 } if ( capture_dir / 'no_index' ) . exists ( ) : cache [ 'no_index' ] = 1 if ( capture_dir / 'parent' ) . exists ( ) : with ( capture_dir / 'parent' ) . open ( ) as f : cache [ 'parent' ] = f . read ( ) . strip ( ) p . hmset ( str ( capture_dir ) , cache ) if not redis_pipeline : p . execute ( ) self . _captures_index . pop ( uuid , None )
Populate the redis cache for a capture.
Populate the redis cache for a capture .
'''Populate the redis cache for a capture. Mostly used on the index page.''' # (re)build cache # Something went wrong # old format # type: ignore # type: ignore # load tree from disk, get redirects # Pickle not available # If the folders claims anonymity # The capture was initiated from an other one # type: ignore # If the cache is re-created for some reason, pop from the local cache.
self
capture_dir
Path
force
bool
redis_pipeline
Optional[Redis]