import gzip
import json
import os
import sys
import traceback
import xml.etree.ElementTree as ET
class Watcher:
def __init__(self):
self.inotify = Inotify()
self.watches = {}
self.handlers = {}
def __add_watch(self, path, mask, handler):
if path in self.watches:
wd = self.watches[path]
self.handlers[wd] = self.handlers[wd] | frozenset([handler])
else:
wd = self.inotify.add_watch(path, mask)
if wd >= 0:
self.watches[path] = wd
self.handlers[wd] = frozenset([handler])
def __rem_watch(self, path, handler):
wd = self.watches[path]
self.handlers[wd] = self.handlers[wd] - frozenset([handler])
if len(self.handlers[wd]) == 0:
self.inotify.rem_watch(wd)
del self.handlers[wd]
del self.watches[path]
def watch_directory(self, path, callback):
events = (IN_CREATE |
IN_MOVED_TO |
IN_MOVED_FROM |
IN_DELETE_SELF |
IN_CLOSE_WRITE |
IN_DELETE |
IN_MOVE_SELF)
def handler(mask, name):
if ((mask & IN_CREATE or mask & IN_MOVED_TO) and
cur_wait_name and name == cur_wait_name):
reset()
elif mask & (IN_DELETE_SELF | IN_MOVE_SELF):
reset()
elif not cur_wait_name and len(name) > 0:
if mask & (IN_CLOSE_WRITE | IN_MOVED_TO | IN_DELETE | IN_MOVED_FROM):
callback(os.path.join(path, name))
def reset():
self.__rem_watch(cur_path, handler)
self.watch_directory(path, callback)
cur_path = path
cur_wait_path = None
cur_wait_name = None
while not os.path.exists(cur_path):
cur_wait_path = cur_path
cur_wait_name = os.path.basename(cur_wait_path)
cur_path = os.path.dirname(cur_path)
self.__add_watch(cur_path, events, handler)
if cur_wait_path and os.path.exists(cur_wait_path):
reset()
return
if not cur_wait_name:
for f in os.listdir(cur_path):
callback(os.path.join(cur_path, f))
def run(self):
def event(wd, mask, name):
if wd in self.handlers:
for h in self.handlers[wd]:
h(mask, name)
self.inotify.run(event)
lang = os.environ.get('LANGUAGE')
def attr_lang(elt):
return elt.attrib.get('{http://www.w3.org/XML/1998/namespace}lang')
def element(xml, tag):
"""Get an XML element from the DOM tag
Retrieves the element with the matching language or by best effort
- If element has the selected language (or language selection is none), return it
- If the selected language can't be found, use empty language as a fallback
- If no elements are found with empty language attribute, use first element found
"""
empty_lang = None
for elt in xml.iter(tag):
language = attr_lang(elt)
if language == lang:
return elt
elif language is None and empty_lang is None:
empty_lang = elt
if empty_lang is not None:
return empty_lang
return xml.find(tag)
def element_value(xml, tag):
elt = element(xml, tag)
return elt.text if elt is not None else None
def convert_description(xml, *, use_lang=True):
if xml is None:
return None
want_lang = lang if use_lang else None
def text(xml):
return " ".join(xml.itertext())
res = []
for c in xml:
if attr_lang(c) != want_lang:
continue
if c.tag == 'p':
res.append(text(c))
elif c.tag == 'ul' or c.tag == 'ol':
res.append({'tag': c.tag, 'items': list(map(text, c.findall('li')))})
if lang is not None and len(res) == 0:
res = convert_description(xml, use_lang=False)
return res
def convert_cached_icon(directory, origin, xml):
icon = xml.text
def try_size(sz):
path = os.path.join(directory, "..", "icons", origin, sz, icon)
return path if os.path.exists(path) else None
return try_size("64x64") or try_size("128x128")
def convert_remote_icon(xml):
url = xml.text
if url.startswith(('http://', 'https://')):
return url
return None
def convert_local_icon(xml):
path = xml.text
if path.startswith("/"):
return path
return None
def convert_stock_icon(xml):
icon = xml.text
def try_size(size: str, extension: str):
path = os.path.join("/usr/share/icons/hicolor", size, "apps", f"{icon}.{extension}")
return path if os.path.exists(path) else None
return (try_size("64x64", "svg") or try_size("64x64", "png") or
try_size("128x128", "svg") or try_size("128x128", "png"))
def find_and_convert_icon(directory, origin, xml):
if xml is None:
return None
icon = xml.find('icon')
if icon is not None:
if icon.attrib['type'] == 'cached':
return convert_cached_icon(directory, origin, icon)
elif icon.attrib['type'] == 'remote':
return convert_remote_icon(icon)
elif icon.attrib['type'] == 'local':
return convert_local_icon(icon)
elif icon.attrib['type'] == 'stock':
return convert_stock_icon(icon)
return None
def convert_screenshots(xml):
if xml is None:
return []
shots = []
for sh in xml.iter('screenshot'):
for img in sh.iter('image'):
if img.attrib['type'] == 'source':
shots.append({'full': img.text})
return shots
def convert_launchables(xml):
ables = []
for elt in xml.iter('launchable'):
launchable_type = elt.attrib['type']
if launchable_type == "cockpit-manifest":
ables.append({'name': elt.text, 'type': launchable_type})
return ables
def convert_urls(xml):
urls = []
for url in xml.iter('url'):
urls.append({'type': url.attrib['type'], 'link': url.text})
return urls
def convert_collection_component(directory, origin, xml):
component_id = element_value(xml, 'id')
pkgname = element_value(xml, 'pkgname')
launchables = convert_launchables(xml)
urls = convert_urls(xml)
if not component_id or not pkgname or len(launchables) == 0:
return None
return {
'id': component_id,
'pkgname': pkgname,
'name': element_value(xml, 'name'),
'summary': element_value(xml, 'summary'),
'description': convert_description(element(xml, 'description')),
'icon': find_and_convert_icon(directory, origin, xml),
'screenshots': convert_screenshots(element(xml, 'screenshots')),
'launchables': launchables,
'urls': urls
}
def convert_upstream_component(file, xml):
if xml.tag != 'component':
return None
launchables = convert_launchables(xml)
if len(launchables) == 0:
return None
urls = convert_urls(xml)
return {
'id': element_value(xml, 'id'),
'name': element_value(xml, 'name'),
'summary': element_value(xml, 'summary'),
'description': convert_description(element(xml, 'description')),
'icon': find_and_convert_icon(dir, '', xml),
'screenshots': convert_screenshots(element(xml, 'screenshots')),
'launchables': launchables,
'installed': True,
'file': file,
'urls': urls
}
class MetainfoDB:
def __init__(self):
self.dumping = False
self.installed_by_file = {}
self.available_by_file = {}
def notice_installed(self, file, xml_root):
if xml_root is not None:
comp = convert_upstream_component(file, xml_root)
if comp is not None:
self.installed_by_file[file] = comp
elif file in self.installed_by_file:
del self.installed_by_file[file]
if self.dumping:
self.dump()
def notice_available(self, file, xml_root):
if xml_root is not None:
info = {}
origin = xml_root.attrib['origin']
for xml_comp in xml_root.iter('component'):
try:
comp = convert_collection_component(os.path.dirname(file), origin, xml_comp)
if comp is not None:
if comp['id'] in info:
pass
else:
info[comp['id']] = comp
except KeyError:
pass
self.available_by_file[file] = info
elif file in self.available_by_file:
del self.available_by_file[file]
if self.dumping:
self.dump()
def dump(self):
comps = {}
for file in self.installed_by_file:
comp = self.installed_by_file[file]
if comp['id'] in comps:
pass
else:
comps[comp['id']] = comp
for file in self.available_by_file:
for comp_id in self.available_by_file[file]:
comp = self.available_by_file[file][comp_id]
if comp['id'] not in comps:
comps[comp['id']] = comp
else:
z = comp.copy()
z.update(comps[comp['id']])
comps[comp['id']] = z
data = {
'components': comps,
'origin_files': list(self.available_by_file.keys())
}
sys.stdout.write(json.dumps(data) + '\n')
sys.stdout.flush()
def start_dumping(self):
self.dump()
self.dumping = True
def watch_db():
watcher = Watcher()
db = MetainfoDB()
def process_file(path, callback):
try:
if not os.path.exists(path):
callback(path, None)
elif path.endswith('.xml'):
callback(path, ET.parse(path).getroot())
elif path.endswith('.xml.gz'):
callback(path, ET.parse(gzip.open(path)).getroot())
except Exception:
callback(path, None)
sys.stderr.write("%s: " % path)
sys.stderr.write("".join(traceback.format_exception_only(sys.exc_info()[0], sys.exc_info()[1])))
sys.stderr.flush()
def installed_callback(path):
process_file(path, lambda path, xml: db.notice_installed(path, xml))
def available_callback(path):
process_file(path, lambda path, xml: db.notice_available(path, xml))
watcher.watch_directory('/usr/share/swcatalog/xml', available_callback)
watcher.watch_directory('/var/cache/swcatalog/xml', available_callback)
watcher.watch_directory('/var/lib/swcatalog/xml', available_callback)
watcher.watch_directory('/usr/share/app-info/xmls', available_callback)
watcher.watch_directory('/var/cache/app-info/xmls', available_callback)
watcher.watch_directory('/usr/share/metainfo', installed_callback)
db.start_dumping()
watcher.run()
watch_db()
