spalencsar/bearhub
1
0
Fork 0
mirror of https://github.com/spalencsar/bearhub.git synced 2026-07-09 16:44:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

[refactoring] CategoriesDownloader as a common class

Browse Source
This commit is contained in:
Vinicius Moreira
2019-10-22 17:10:24 -03:00
parent c358d077be
commit 64e4822b2d
7 changed files with 118 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
bauh/gems/snap/__init__.py
View File

@@ -4,3 +4,5 @@ from bauh.api.constants import CACHE_PATH
ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
SNAP_CACHE_PATH = CACHE_PATH + '/snap'
CATEGORIES_FILE_PATH = SNAP_CACHE_PATH + '/categories.txt'
URL_CATEGORIES_FILE = 'https://raw.githubusercontent.com/vinifmor/bauh-files/master/snap/categories.txt'

8
bauh/gems/snap/controller.py
View File

@@ -10,12 +10,13 @@ from bauh.api.abstract.handler import ProcessWatcher
from bauh.api.abstract.model import SoftwarePackage, PackageHistory, PackageUpdate, PackageSuggestion, \
SuggestionPriority
from bauh.api.abstract.view import SingleSelectComponent, SelectViewType, InputOption
from bauh.commons.category import CategoriesDownloader
from bauh.commons.html import bold
from bauh.commons.system import SystemProcess, ProcessHandler, new_root_subprocess
from bauh.gems.snap import snap, suggestions
from bauh.gems.snap import snap, suggestions, URL_CATEGORIES_FILE, SNAP_CACHE_PATH, CATEGORIES_FILE_PATH
from bauh.gems.snap.constants import SNAP_API_URL
from bauh.gems.snap.model import SnapApplication
from bauh.gems.snap.worker import SnapAsyncDataLoader, CategoriesDownloader
from bauh.gems.snap.worker import SnapAsyncDataLoader
RE_AVAILABLE_CHANNELS = re.compile(re.compile(r'(\w+)\s+(snap install.+)'))
@@ -32,7 +33,8 @@ class SnapManager(SoftwareManager):
self.logger = context.logger
self.ubuntu_distro = context.distro == 'ubuntu'
self.categories = {}
self.categories_downloader = CategoriesDownloader(self.http_client, self.logger, self, context.disk_cache)
self.categories_downloader = CategoriesDownloader('snap', self.http_client, self.logger, self, context.disk_cache,
URL_CATEGORIES_FILE, SNAP_CACHE_PATH, CATEGORIES_FILE_PATH)
def map_json(self, app_json: dict, installed: bool, disk_loader: DiskCacheLoader, internet: bool = True) -> SnapApplication:
app = SnapApplication(publisher=app_json.get('publisher'),

91
bauh/gems/snap/worker.py
View File

@@ -1,18 +1,11 @@
import logging
import os
import traceback
from pathlib import Path
from threading import Thread
from typing import Dict, List
import requests
from bauh.api.abstract.cache import MemoryCache
from bauh.api.abstract.context import ApplicationContext
from bauh.api.abstract.controller import SoftwareManager
from bauh.api.abstract.model import PackageStatus
from bauh.api.http import HttpClient
from bauh.gems.snap import snap, SNAP_CACHE_PATH
from bauh.gems.snap import snap
from bauh.gems.snap.constants import SNAP_API_URL
from bauh.gems.snap.model import SnapApplication
@@ -75,85 +68,3 @@ class SnapAsyncDataLoader(Thread):
if self.persist:
self.manager.cache_to_disk(pkg=self.app, icon_bytes=None, only_icon=False)
class CategoriesDownloader(Thread):
URL_CATEGORIES_FILE = 'https://raw.githubusercontent.com/vinifmor/bauh-files/master/snap/categories.txt'
CATEGORIES_FILE_PATH = SNAP_CACHE_PATH + '/categories.txt'
def __init__(self, http_client: HttpClient, logger: logging.Logger, manager: SoftwareManager, disk_cache: bool):
super(CategoriesDownloader, self).__init__(daemon=True)
self.http_client = http_client
self.logger = logger
self.manager = manager
self.disk_cache = disk_cache
def _read_categories_from_disk(self) -> dict:
if self.disk_cache and os.path.exists(self.CATEGORIES_FILE_PATH):
self.logger.info("Reading cached categories from the disk")
with open(self.CATEGORIES_FILE_PATH) as f:
categories = f.read()
return self._map_categories(categories)
return {}
def _map_categories(self, categories: str) -> dict:
categories_map = {}
for l in categories.split('\n'):
if l:
data = l.split('=')
categories_map[data[0]] = [c.strip() for c in data[1].split(',') if c]
return categories_map
def _cache_categories_to_disk(self, categories: str):
self.logger.info('Caching Snap categories to the disk')
try:
Path(SNAP_CACHE_PATH).mkdir(parents=True, exist_ok=True)
with open(self.CATEGORIES_FILE_PATH, 'w+') as f:
f.write(categories)
self.logger.info("Snap categories cached to the disk as '{}'".format(self.CATEGORIES_FILE_PATH))
except:
self.logger.error("Could not cache Snap categories to the disk as '{}'".format(self.CATEGORIES_FILE_PATH))
traceback.print_exc()
def get_categories(self) -> Dict[str, List[str]]:
self.logger.info('Downloading Snap category definitions from {}'.format(self.URL_CATEGORIES_FILE))
try:
res = self.http_client.get(self.URL_CATEGORIES_FILE)
if res:
try:
categories = self._map_categories(res.text)
self.logger.info('Loaded categories for {} Snap applications'.format(len(categories)))
if self.disk_cache and categories:
Thread(target=self._cache_categories_to_disk, args=(res.text,), daemon=True).start()
return categories
except:
self.logger.error("Could not parse categories definitions")
traceback.print_exc()
else:
self.logger.info('Could not download {}'.format(self.URL_CATEGORIES_FILE))
except requests.exceptions.ConnectionError:
self.logger.warning('The internet connection seems to be off.')
return self._read_categories_from_disk()
def run(self):
categories = self.get_categories()
if categories:
self.logger.info("Settings categories to {}".format(self.manager.__class__.__name__))
self.manager.categories = categories
self.logger.info('Finished')