source: lliurex-store/trunk/fuentes/python3-lliurex-store.install/usr/share/lliurexstore/plugins/appImageManager.py @ 7721

Last change on this file since 7721 was 7721, checked in by Juanma, 21 months ago

Availabe snaps are loaded from available sections

File size: 20.0 KB
Line 
1#The name of the main class must match the file name in lowercase
2import re
3import urllib
4from urllib.request import Request
5from urllib.request import urlretrieve
6import shutil
7import json
8import os
9import sys
10import threading
11import queue
12import time
13import random
14import gi
15from gi.repository import Gio
16gi.require_version('AppStreamGlib', '1.0')
17from gi.repository import AppStreamGlib as appstream
18from bs4 import BeautifulSoup
19#from subprocess import call
20
21class appimagemanager:
22        def __init__(self):
23                self.dbg=False
24                self.progress=0
25                self.partial_progress=0
26                self.plugin_actions={'install':'appimage','remove':'appimage','pkginfo':'appimage','load':'appimage'}
27                self.result={}
28                self.result['data']={}
29                self.result['status']={}
30                self.cache_dir=os.getenv("HOME")+"/.cache/lliurex-store"
31                self.icons_dir=self.cache_dir+"/icons"
32                self.bundles_dir=self.cache_dir+"/xmls/appimage"
33                self.bundle_types=['appimg']
34                self.appimage_dir=os.getenv("HOME")+"/.local/bin"
35                self.repos={'appimagehub':{'type':'json','url':'https://appimage.github.io/feed.json','url_info':''}}
36                #Appimages not stored in a repo must be listed in this file, providing at least the download url
37                self.external_appimages="/usr/share/lliurex-store/files/external_appimages.json"
38                self.locale=['ca_ES@valencia','ca@valencia','qcv','ca','ca_ES','es_ES','es','en_US','en_GB','en','C']
39                self.disabled=False
40                self.icon_cache_enabled=True
41                self.image_cache_enabled=True
42                self.apps_for_store=queue.Queue()
43        #def __init__
44
45        def set_debug(self,dbg=True):
46                self.dbg=dbg
47                self._debug ("Debug enabled")
48        #def set_debug
49
50        def _debug(self,msg=''):
51                if self.dbg:
52                        print ('DEBUG appimage: %s'%msg)
53        #def debug
54
55        def register(self):
56                return(self.plugin_actions)
57
58        def enable(self,state=False):
59                self.disable=state
60
61        def execute_action(self,action,applist=None,store=None):
62                if store:
63                        self.store=store
64                else:
65                        self.store=appstream.Store()
66                self.appimage_store=appstream.Store()
67                self.progress=0
68                self.result['status']={'status':-1,'msg':''}
69                self.result['data']=[]
70                self.threads=[]
71                dataList=[]
72                if self.disabled:
73                        self._set_status(9)
74                        self.result['data']=self.store
75                else:
76                        self._chk_installDir()
77                        if action=='load':
78                                self._load_appimage_store()
79                                #wait till threads end (if any)
80                                self._debug("Ending threads...")
81                                for th in threading.enumerate():
82                                        if th.is_alive():
83                                                try:
84                                                        th.join(5)
85                                                except:
86                                                        pass
87                                while not self.apps_for_store.empty():
88                                        app=self.apps_for_store.get()
89                                        self.store.add_app(app)
90                                self.result['data']=self.store
91                        else:
92                                for app_info in applist:
93                                        self.partial_progress=0
94                                        if action=='install':
95                                                dataList.append(self._install_appimage(app_info))
96                                        if action=='remove':
97                                                dataList.append(self._remove_appimage(app_info))
98                                        if action=='pkginfo':
99                                                dataList.append(self._get_info(app_info))
100                                        self.progress+=int(self.partial_progress/len(applist))-1
101                                self.result['data']=list(dataList)
102                self.progress=100
103                return(self.result)
104
105        def _set_status(self,status,msg=''):
106                self.result['status']={'status':status,'msg':msg}
107        #def _set_status
108
109        def _callback(self,partial_size=0,total_size=0):
110                limit=99
111                if partial_size!=0 and total_size!=0:
112                        inc=round(partial_size/total_size,2)*100
113                        self.progress=inc
114                else:
115                        inc=1
116                        margin=limit-self.progress
117                        inc=round(margin/limit,3)
118                        self.progress=(self.progress+inc)
119                if (self.progress>limit):
120                        self.progress=limit
121
122        def _chk_installDir(self):
123                msg_status=True
124                if not os.path.isdir(self.appimage_dir):
125                        try:
126                                os.makedirs(self.appimage_dir)
127                        except:
128                                msg_status=False
129                return msg_status                               
130
131        def _install_appimage(self,app_info):
132                app_info=self._get_info(app_info)
133                self._debug("Installing %s"%app_info)
134                if app_info['state']=='installed':
135                        self._set_status(4)
136                else:
137                        if 'appimage' in app_info['channel_releases'].keys():
138                                appimage_url=app_info['channel_releases']['appimage'][0]
139                        self._debug("Downloading "+appimage_url)
140                        dest_path=self.appimage_dir+'/'+app_info['package']
141                        if appimage_url:
142                                try:
143                                        req=Request(appimage_url, headers={'User-Agent':'Mozilla/5.0'})
144                                        with urllib.request.urlopen(req) as response, open(dest_path, 'wb') as out_file:
145                                                bf=16*1024
146                                                acumbf=0
147                                                app_size=int(response.info()['Content-Length'])
148                                                while True:
149                                                        if acumbf>=app_size:
150                                                            break
151                                                        shutil.copyfileobj(response, out_file,bf)
152                                                        acumbf=acumbf+bf
153                                                        self._callback(acumbf,app_size)
154                                        st = os.stat(dest_path)
155                                        os.chmod(dest_path, st.st_mode | 0o111)
156                                        self._set_status(0)
157                                except Exception as e:
158                                        print(e)
159                                        self._set_status(5)
160                        else:
161                                self._set_status(12)
162                return app_info
163        #def _install_appimage
164
165        def _remove_appimage(self,app_info):
166                self._debug("Removing "+app_info['package'])
167                if os.path.isfile(self.appimage_dir+'/'+app_info['package']):
168                        try:
169                                call([self.appimage_dir+"/"+app_info['package'], "--remove-appimage-desktop-integration"])
170                        except:
171                                pass
172                        try:
173                                os.remove(self.appimage_dir+"/"+app_info['package'])
174                                self._set_status(0)
175                        except:
176                                self._set_status(6)
177                return(app_info)
178        #def _remove_appimage
179
180        def _load_appimage_store(self,store=None):
181                self._get_bundles_catalogue()
182                self._get_external_catalogue()
183                if os.path.exists(self.bundles_dir):
184                        for bundle_type in self.bundle_types:
185                                self._debug("Loading %s catalog"%bundle_type)
186                                store=self._generic_file_load(self.bundles_dir+'/'+bundle_type,store)
187                return(store)
188        #def load_bundles_catalog(self)
189       
190        def _generic_file_load(self,target_path,store):
191                icon_path='/usr/share/icons/hicolor/128x128'
192                if not os.path.isdir(target_path):
193                        os.makedirs(target_path)
194                files=os.listdir(target_path)
195                for target_file in os.listdir(target_path):
196                        if target_file.endswith('appdata.xml'):
197                                store_path=Gio.File.new_for_path(target_path+'/'+target_file)
198                                self._debug("Adding file "+target_path+'/'+target_file)
199                                try:
200                                        store.from_file(store_path,icon_path,None)
201                                except Exception as e:
202                                        self._debug("Couldn't add file "+target_file+" to store")
203                                        self._debug("Reason: "+str(e))
204                return(store)
205        #def _generic_file_load
206
207        def _get_bundles_catalogue(self):
208                applist=[]
209                appdict={}
210                all_apps=[]
211                outdir=self.bundles_dir+'/appimg/'
212                #Load repos
213                for repo_name,repo_info in self.repos.items():
214                        if not os.path.isdir(self.bundles_dir):
215                                try:
216                                        os.makedirs(self.bundles_dir)
217                                except:
218                                        self._debug("appImage catalogue could not be fetched: Permission denied")
219                        self._debug("Fetching repo %s"%repo_info['url'])
220                        if repo_info['type']=='json':
221                                applist=self._process_appimage_json(self._fetch_repo(repo_info['url']),repo_name)
222
223                        self._debug("Fetched repo "+repo_info['url'])
224                        self._th_generate_xml_catalog(applist,outdir,repo_info['url_info'],repo_info['url'],repo_name)
225                        all_apps.extend(applist)
226                return True
227        #def _get_bundles_catalogue
228
229        def _get_external_catalogue(self):
230                applist=[]
231                all_apps=[]
232                outdir=self.bundles_dir+'/appimg/'
233                #Load external apps
234                for app_name,app_info in self._get_external_appimages().items():
235                        if os.path.isdir(self.bundles_dir):
236                                appinfo=self._init_appinfo()
237                                if 'name' in app_info.keys():
238                                        appinfo['name']=app_info['name']
239                                else:
240                                        appinfo['name']=app_info['url'].split('/')[-1]
241                                appinfo['package']=app_info['url'].split('/')[-1]
242                                if 'homepage' in app_info.keys():
243                                        appinfo['homepage']=app_info['homepage']
244                                else:
245                                        appinfo['homepage']='/'.join(app_info['url'].split('/')[0:-1])
246                                appinfo['installerUrl']=app_info['url']
247                                if 'description' in app_info.keys():
248                                        if type(app_info['description'])==type({}):
249                                                for lang in app_info['description']:
250                                                        appinfo['description'].update({lang:app_info['description'][lang]})
251                                        else:
252                                                appinfo['description'].update({"C":appimage['description']})
253                                if 'categories' in app_info.keys():
254                                        appinfo['categories']=app_info['categories']
255                                if 'keywords' in app_info.keys():
256                                        appinfo['keywords']=app_info['keywords']
257                                if 'version' in app_info.keys():
258                                        appinfo['version']=app_info['version']
259                                self._debug("Fetching external appimage %s"%app_info['url'])
260                                appinfo['bundle']='appimage'
261                                self._debug("External:\n%s\n-------"%appinfo)
262                                applist.append(appinfo)
263                        else:
264                                self._debug("External appImage could not be fetched: Permission denied")
265                self._th_generate_xml_catalog(applist,outdir,app_info['url_info'],app_info['url'],app_name)
266                self._debug("Fetched appimage "+app_info['url'])
267                all_apps.extend(applist)
268                self._debug("Removing old entries...")
269#               self._clean_bundle_catalogue(all_apps,outdir)
270                return(True)
271        #def _get_external_catalogue
272       
273        def _fetch_repo(self,repo):
274                req=Request(repo, headers={'User-Agent':'Mozilla/5.0'})
275                with urllib.request.urlopen(req) as f:
276                        content=(f.read().decode('utf-8'))
277               
278                return(content)
279        #def _fetch_repo
280       
281        def _get_external_appimages(self):
282                external_appimages={}
283                if os.path.isfile(self.external_appimages):
284                        try:
285                                with open(self.external_appimages) as appimages:
286                                        external_appimages=json.load(appimages)
287                        except:
288                                self._debug("Can't load %s"%self.external_appimages)
289                return external_appimages
290        #def _get_external_appimages
291       
292        def _process_appimage_json(self,data,repo_name):
293                applist=[]
294                json_data=json.loads(data)
295                if 'items' in json_data.keys():
296                        for appimage in json_data['items']:
297                                appinfo=self._th_process_appimage(appimage)
298                                if appinfo:
299                                        applist.append(appinfo)
300                return (applist)
301        #_process_appimage_json
302
303        def _th_process_appimage(self,appimage):
304                appinfo=None
305                releases=[]
306                if 'links' in appimage.keys():
307                        if appimage['links']:
308                                appinfo=self.load_json_appinfo(appimage)
309                return(appinfo)
310        #def _th_process_appimage
311
312        def load_json_appinfo(self,appimage):
313                appinfo=self._init_appinfo()
314                appinfo['name']=appimage['name']
315                appinfo['package']=appimage['name']
316                if 'license' in appimage.keys():
317                        appinfo['license']=appimage['license']
318                appinfo['summary']=''
319                if 'description' in appimage.keys():
320                        if type(appinfo['description'])==type({}):
321                                for lang in appinfo['description'].keys():
322                                        appinfo['description'].update({lang:appimage['description'][lang]})
323                        else:
324                                appinfo['description']={"C":appimage['description']}
325                if 'categories' in appimage.keys():
326                        appinfo['categories']=appimage['categories']
327                if 'icon' in appimage.keys():
328                        appinfo['icon']=appimage['icon']
329                if 'icons' in appimage.keys():
330                        self._debug("Loading icon %s"%appimage['icons'])
331                        if appimage['icons']:
332                                self._debug("Loading icon %s"%appimage['icons'][0])
333                                appinfo['icon']=appimage['icons'][0]
334                if 'screenshots' in appimage.keys():
335                        appinfo['thumbnails']=appimage['screenshots']
336                if 'links' in appimage.keys():
337                        if appimage['links']:
338                                for link in appimage['links']:
339                                        if 'url' in link.keys() and link['type']=='Download':
340                                                appinfo['installerUrl']=link['url']
341                if 'authors' in appimage.keys():
342                        if appimage['authors']:
343                                for author in appimage['authors']:
344                                        if 'url' in author.keys():
345                                                self._debug("Author: %s"%author['url'])
346                                                appinfo['homepage']=author['url']
347                else:
348                        appinfo['homepage']='/'.join(appinfo['installerUrl'].split('/')[0:-1])
349                appinfo['bundle']=['appimage']
350                return appinfo
351        #def load_json_appinfo
352
353        def _th_generate_xml_catalog(self,applist,outdir,info_url,repo,repo_name):
354                maxconnections = 2
355                threads=[]
356                semaphore = threading.BoundedSemaphore(value=maxconnections)
357                random_applist = list(applist)
358                random.shuffle(random_applist)
359                for app in applist:
360                        th=threading.Thread(target=self._th_write_xml, args = (app,outdir,info_url,repo,repo_name,semaphore))
361                        threads.append(th)
362                        th.start()
363        #def _th_generate_xml_catalog
364
365        def     _th_write_xml(self,appinfo,outdir,info_url,repo,repo_name,semaphore):
366                semaphore.acquire()
367                self._add_appimage(appinfo)
368                semaphore.release()
369        #def _th_write_xml
370
371        def _add_appimage(self,appinfo):
372                #Search in local store for the app
373                sw_new=False
374                app=appstream.App()
375                app_orig=self.store.get_app_by_pkgname(appinfo['package'].lower())
376                if not app_orig:
377                        app_orig=self.store.get_app_by_id(appinfo['package'].lower()+".desktop")
378                if app_orig:
379                        self._debug("Extending app %s"%appinfo['package'])
380                        app=self._copy_app_from_appstream(app_orig,app)
381                else:
382                        self._debug("Generating new %s"%appinfo['package'])
383                if appinfo['name'].endswith('.appimage'):
384                        app.set_id("appimagehub.%s"%appinfo['name'].lower())
385                        app.set_name("C",appinfo['name'])
386                else:
387                        app.set_id("appimagehub.%s"%appinfo['name'].lower()+'.appimage')
388                        app.set_name("C",appinfo['name']+".appimage")
389                if appinfo['package'].endswith('.appimage'):
390                        app.add_pkgname(appinfo['package'].lower())
391                else:
392                        app.add_pkgname(appinfo['package'].lower()+".appimage")
393                app.set_id_kind=appstream.IdKind.DESKTOP
394                sw_new=True
395
396                icon=appstream.Icon()
397                screenshot=appstream.Screenshot()
398                if appinfo['license']:
399                        app.set_project_license(appinfo['license'])
400                bundle=appstream.Bundle()
401                bundle.set_kind(bundle.kind_from_string('APPIMAGE'))
402                if appinfo['package'].endswith('.appimage'):
403                        bundle.set_id(appinfo['package'])
404                else:
405                        bundle.set_id(appinfo['package']+'.appimage')
406                app.add_bundle(bundle)
407                if 'keywords' in appinfo.keys():
408                        for keyword in appinfo['keywords']:
409                                app.add_keyword("C",keyword)
410                        if 'appimage' not in appinfo['keywords']:
411                                app.add_keyword("C","appimage")
412                else:
413                        app.add_keyword("C","appimage")
414                app.add_url(appstream.UrlKind.UNKNOWN,appinfo['installerUrl'])
415                app.add_url(appstream.UrlKind.HOMEPAGE,appinfo['homepage'])
416                if sw_new:
417                        app.add_keyword("C",appinfo['package'])
418                        if not appinfo['name'].endswith('.appimage'):
419                                app.set_name("C",appinfo['name']+".appimage")
420                        for lang,desc in appinfo['description'].items():
421                                description="This is an AppImage bundle of app %s. It hasn't been tested by our developers and comes from a 3rd party dev team. Please use it carefully.\n%s"%(appinfo['name'],desc)
422                                summary=' '.join(list(description.split(' ')[:8]))
423                                if len(description.split(' '))>8:
424                                        summary+="... "
425                                app.set_description(lang,description)
426                                app.set_comment(lang,summary)
427                        if 'categories' in appinfo.keys():
428                                for category in appinfo['categories']:
429                                        app.add_category(category)
430                                if 'appimage' not in appinfo['categories']:
431                                        app.add_category("appimage")
432                        else:
433                                app.add_category("appimage")
434                if appinfo['icon']:
435                        if self.icon_cache_enabled:
436                                icon.set_kind(appstream.IconKind.LOCAL)
437                                icon.set_name(self._download_file(appinfo['icon'],appinfo['name'],self.icons_dir))
438                        else:
439                                icon.set_kind(appstream.IconKind.REMOTE)
440                                icon.set_name(pkg.get_icon())
441                                icon.set_url(pkg.get_icon())
442                        app.add_icon(icon)
443                if appinfo['thumbnails']:
444                        img=appstream.Image()
445                        if not appinfo['thumbnails'][0].startswith('http'):
446                                        appinfo['screenshot']=appinfo['thumbnails'][0]
447                                        appinfo['screenshot']="https://appimage.github.io/database/%s"%appinfo['screenshot']
448                        img.set_kind(appstream.ImageKind.SOURCE)
449                        img.set_url(appinfo['screenshot'])
450                        screenshot.add_image(img)
451                        app.add_screenshot(screenshot)
452                #Adds the app to the store
453                self.apps_for_store.put(app)
454                if not os.path.isfile(self.bundles_dir+'/'+app.get_id_filename()):
455                        gioFile=Gio.File.new_for_path('%s/%s.xml'%(self.bundles_dir,app.get_id_filename()))
456                        app.to_file(gioFile)
457        #def _add_appimage
458
459        def _copy_app_from_appstream(self,app_orig,app):
460                app.set_id("appimage."+app_orig.get_id())
461                for category in app_orig.get_categories():
462                        app.add_category(category)
463                for screenshot in app_orig.get_screenshots():
464                        app.add_screenshot(screenshot)
465                for icon in app_orig.get_icons():
466                        app.add_icon(icon)
467                for localeItem in self.locale:
468                        if app_orig.get_name(localeItem):
469                                app.set_name(localeItem,app_orig.get_name(localeItem)+".appimage")
470                        if app_orig.get_description(localeItem):
471                                app.set_description(localeItem,app_orig.get_description(localeItem))
472                        if app_orig.get_comment(localeItem):
473                                app.set_comment(localeItem,app_orig.get_comment(localeItem))
474                app.set_origin(app_orig.get_origin())
475                return app
476
477        def _clean_bundle_catalogue(self,applist,outdir):
478                xml_files_list=[]
479                applist=[item.lower() for item in applist]
480                for xml_file in os.listdir(outdir):
481                        if xml_file.endswith('appdata.xml'):
482                                xml_files_list.append(xml_file.lower().replace('appdata.xml','appimage'))
483       
484                if xml_files_list:
485                        xml_discard_list=list(set(xml_files_list).difference(applist))
486                        for discarded_file in xml_discard_list:
487                                os.remove(outdir+'/'+discarded_file.replace('appimage','appdata.xml'))
488        #def _clean_bunlde_catalogue
489
490        def _download_file(self,url,app_name,dest_dir):
491                target_file=dest_dir+'/'+app_name+".png"
492                if not url.startswith('http'):
493                        url="https://appimage.github.io/database/%s"%url
494                if not os.path.isfile(target_file):
495                        if not os.path.isfile(target_file):
496                                self._debug("Downloading %s to %s"%(url,target_file))
497                                try:
498                                        with urllib.request.urlopen(url) as response, open(target_file, 'wb') as out_file:
499                                                bf=16*1024
500                                                acumbf=0
501                                                file_size=int(response.info()['Content-Length'])
502                                                while True:
503                                                        if acumbf>=file_size:
504                                                            break
505                                                        shutil.copyfileobj(response, out_file,bf)
506                                                        acumbf=acumbf+bf
507                                        st = os.stat(target_file)
508                                except Exception as e:
509                                        self._debug("Unable to download %s"%url)
510                                        self._debug("Reason: %s"%e)
511                                        target_file=''
512                return(target_file)
513        #def _download_file
514       
515        def _chk_bundle_dir(self,outdir):
516                msg_status=True
517                if not os.path.isdir(outdir):
518                        try:
519                                os.makedirs(outdir)
520                        except Exception as e:
521                                msg_status=False
522                                print(e)
523                return(os.access(outdir,os.W_OK|os.R_OK|os.X_OK|os.F_OK))
524        #def _chk_bundle_dir
525       
526        def _init_appinfo(self):
527                appInfo={'appstream_id':'',\
528                'id':'',\
529                'name':'',\
530                'version':'',\
531                'channel_releases':{},\
532                'component':'',\
533                'package':'',\
534                'license':'',\
535                'summary':'',\
536                'description':{},\
537                'categories':[],\
538                'icon':'',\
539                'screenshot':'',\
540                'thumbnails':[],\
541                'video':'',\
542                'homepage':'',\
543                'installerUrl':'',\
544                'state':'',\
545                'depends':'',\
546                'kudos':'',\
547                'suggests':'',\
548                'extraInfo':'',\
549                'size':'',\
550                'bundle':'',\
551                'updatable':'',\
552                }
553                return(appInfo)
554        #def _init_appinfo
555       
556        def _get_info(self,app_info):
557                if app_info['installerUrl']:
558                        self._debug("installer: %s"%app_info['installerUrl'])
559                        app_info['channel_releases']={'appimage':[]}
560                        app_info['channel_releases']['appimage']=self._get_releases(app_info)
561                app_info['state']='available'
562                if os.path.isfile(self.appimage_dir+'/'+app_info['package']):
563                        app_info['state']='installed'
564                #Get size
565                if 'appimage' in app_info['channel_releases'].keys():
566                        if app_info['channel_releases']['appimage'][0]:
567                                appimage_url=app_info['channel_releases']['appimage'][0]
568                                dest_path=self.appimage_dir+'/'+app_info['package']
569                                if appimage_url:
570                                        try:
571                                                with urllib.request.urlopen(appimage_url) as response:
572                                                        app_info['size']=(response.info()['Content-Length'])
573                                        except:
574                                                app_info['size']=0
575                        else:
576                                app_info['size']=0
577                        #Version (unaccurate aprox)
578                        app_info['version']=app_info['channel_releases']['appimage'][0].split('/')[-2]
579
580                self._set_status(0)
581                self.partial_progress=100
582                return(app_info)
583        #def _get_info
584
585        def _get_releases(self,app_info):
586                releases=[]
587                releases_page=''
588                self._debug("Info url: %s"%app_info['installerUrl'])
589                url_source=""
590                try:
591                        if 'github' in app_info['installerUrl']:
592                                releases_page="https://github.com"
593                        if 'gitlab' in app_info['installerUrl']:
594                                releases_page="https://gitlab.com"
595                        if 'opensuse' in app_info['installerUrl'].lower():
596                                releases_page=""
597                                url_source="opensuse"
598
599                        if url_source or releases_page:
600                                with urllib.request.urlopen(app_info['installerUrl']) as f:
601                                        content=(f.read().decode('utf-8'))
602                                        soup=BeautifulSoup(content,"html.parser")
603                                        package_a=soup.findAll('a', attrs={ "href" : re.compile(r'.*\.[aA]pp[iI]mage$')})
604
605                                        for package_data in package_a:
606                                                if url_source=="opensuse":
607                                                        package_name=package_data.findAll('a', attrs={"class" : "mirrorbrain-btn"})
608                                                else:
609                                                        package_name=package_data.findAll('strong', attrs={ "class" : "pl-1"})
610                                                package_link=package_data['href']
611                                                package_link=releases_page+package_link
612                                                releases.append(package_link)
613                                                self._debug("Link: %s"%package_link)
614                        else:
615                                releases=[app_info['installerUrl']]
616                except Exception as e:
617                        print(e)
618                self._debug(releases)
619                return releases
620        #def _get_releases
621       
Note: See TracBrowser for help on using the repository browser.