mirror of
https://github.com/spalencsar/bearhub.git
synced 2026-07-10 11:04:15 +02:00
[snap] improvement -> full support refactored to use the Snapd socket instead of the Ubuntu's old Snap API
This commit is contained in:
Vinicius Moreira
@@ -1,174 +1,15 @@
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
from io import StringIO
|
||||
from typing import List, Tuple, Set
|
||||
from typing import Tuple
|
||||
|
||||
from bauh.commons.system import run_cmd, new_subprocess, SimpleProcess
|
||||
from bauh.gems.snap.model import SnapApplication
|
||||
from bauh.commons.system import run_cmd, SimpleProcess
|
||||
|
||||
BASE_CMD = 'snap'
|
||||
RE_SNAPD_STATUS = re.compile('\s+')
|
||||
RE_SNAPD_SERVICES = re.compile(r'snapd\.\w+.+')
|
||||
|
||||
|
||||
def is_installed():
|
||||
res = run_cmd('which snap', print_error=False)
|
||||
return res and not res.strip().startswith('which ')
|
||||
|
||||
|
||||
def is_snapd_running() -> bool:
|
||||
output = run_cmd('systemctl list-units')
|
||||
|
||||
snapd_services = RE_SNAPD_SERVICES.findall(output)
|
||||
|
||||
socket, socket_running, service, service_running = False, False, False, False
|
||||
if snapd_services:
|
||||
for service_line in snapd_services:
|
||||
line_split = RE_SNAPD_STATUS.split(service_line)
|
||||
|
||||
running = line_split[3] in {'listening', 'running'}
|
||||
|
||||
if line_split[0] == 'snapd.service':
|
||||
service = True
|
||||
service_running = running
|
||||
elif line_split[0] == 'snapd.socket':
|
||||
socket = True
|
||||
socket_running = running
|
||||
|
||||
return socket and socket_running and (not service or service_running)
|
||||
|
||||
|
||||
def app_str_to_json(app: str) -> dict:
|
||||
app_data = [word for word in app.split(' ') if word]
|
||||
app_json = {
|
||||
'name': app_data[0],
|
||||
'version': app_data[1],
|
||||
'rev': app_data[2],
|
||||
'tracking': app_data[3],
|
||||
'publisher': app_data[4] if len(app_data) >= 5 else None,
|
||||
'notes': app_data[5] if len(app_data) >= 6 else None
|
||||
}
|
||||
|
||||
return app_json
|
||||
|
||||
|
||||
def get_info(app_name: str, attrs: tuple = None):
|
||||
full_info_lines = run_cmd('{} info {}'.format(BASE_CMD, app_name))
|
||||
|
||||
data = {}
|
||||
|
||||
if full_info_lines:
|
||||
re_attrs = r'\w+' if not attrs else '|'.join(attrs)
|
||||
info_map = re.findall(r'({}):\s+(.+)'.format(re_attrs), full_info_lines)
|
||||
|
||||
for info in info_map:
|
||||
val = info[1].strip()
|
||||
|
||||
if info[0] == 'installed':
|
||||
val_split = [s for s in val.split(' ') if s]
|
||||
data['version'] = val_split[0]
|
||||
|
||||
if len(val_split) > 2:
|
||||
data['size'] = val_split[2]
|
||||
else:
|
||||
data[info[0]] = val
|
||||
|
||||
if not attrs or 'description' in attrs:
|
||||
desc = re.findall(r'\|\n+((\s+.+\n+)+)', full_info_lines)
|
||||
data['description'] = ''.join([w.strip() for w in desc[0][0].strip().split('\n')]).replace('.', '.\n') if desc else None
|
||||
|
||||
if not attrs or 'commands' in attrs:
|
||||
commands = re.findall(r'commands:\s*\n*((\s+-\s.+\s*\n)+)', full_info_lines)
|
||||
data['commands'] = commands[0][0].strip().replace('- ', '').split('\n') if commands else None
|
||||
|
||||
return data
|
||||
|
||||
|
||||
def read_installed(info_path: str) -> List[dict]:
|
||||
res = run_cmd('{} list'.format(BASE_CMD), print_error=False)
|
||||
|
||||
apps = []
|
||||
|
||||
if res and len(res) > 0:
|
||||
lines = res.split('\n')
|
||||
|
||||
if not lines[0].startswith('error'):
|
||||
for idx, app_str in enumerate(lines):
|
||||
if idx > 0 and app_str:
|
||||
apps.append(app_str_to_json(app_str))
|
||||
|
||||
info_out = new_subprocess(['cat', *[info_path.format(a['name']) for a in apps]]).stdout
|
||||
|
||||
idx = -1
|
||||
for o in new_subprocess(['grep', '-E', '(summary|apps)', '--colour=never'], stdin=info_out).stdout:
|
||||
if o:
|
||||
line = o.decode()
|
||||
|
||||
if line.startswith('summary:'):
|
||||
idx += 1
|
||||
apps[idx]['summary'] = line.split(':')[1].strip()
|
||||
else:
|
||||
apps[idx]['apps_field'] = True
|
||||
|
||||
return apps
|
||||
|
||||
|
||||
def get_app_info_path() -> str:
|
||||
if os.path.exists('/snap'):
|
||||
return '/snap/{}/current/meta/snap.yaml'
|
||||
elif os.path.exists('/var/lib/snapd/snap'):
|
||||
return '/var/lib/snapd/snap/{}/current/meta/snap.yaml'
|
||||
else:
|
||||
return None
|
||||
|
||||
|
||||
def has_apps_field(name: str, info_path: str) -> bool:
|
||||
info_out = new_subprocess(['cat', info_path.format(name)]).stdout
|
||||
|
||||
res = False
|
||||
for o in new_subprocess(['grep', '-E', 'apps', '--colour=never'], stdin=info_out).stdout:
|
||||
if o:
|
||||
line = o.decode()
|
||||
|
||||
if line.startswith('apps:'):
|
||||
res = True
|
||||
|
||||
return res
|
||||
|
||||
|
||||
def search(word: str, exact_name: bool = False) -> List[dict]:
|
||||
apps = []
|
||||
|
||||
res = run_cmd('{} find "{}"'.format(BASE_CMD, word), print_error=False)
|
||||
|
||||
if res:
|
||||
res = res.split('\n')
|
||||
|
||||
if not res[0].startswith('No matching'):
|
||||
for idx, app_str in enumerate(res):
|
||||
if idx > 0 and app_str:
|
||||
app_data = [word for word in app_str.split(' ') if word]
|
||||
|
||||
if exact_name and app_data[0] != word:
|
||||
continue
|
||||
|
||||
apps.append({
|
||||
'name': app_data[0],
|
||||
'version': app_data[1],
|
||||
'publisher': app_data[2],
|
||||
'notes': app_data[3] if app_data[3] != '-' else None,
|
||||
'summary': app_data[4] if len(app_data) == 5 else '',
|
||||
'rev': None,
|
||||
'tracking': None,
|
||||
'type': None
|
||||
})
|
||||
|
||||
if exact_name and len(apps) > 0:
|
||||
break
|
||||
|
||||
return apps
|
||||
def is_installed() -> bool:
|
||||
return bool(run_cmd('which {}'.format(BASE_CMD), print_error=False))
|
||||
|
||||
|
||||
def uninstall_and_stream(app_name: str, root_password: str) -> SimpleProcess:
|
||||
@@ -199,56 +40,16 @@ def refresh_and_stream(app_name: str, root_password: str) -> SimpleProcess:
|
||||
shell=True)
|
||||
|
||||
|
||||
def run(app: SnapApplication, logger: logging.Logger):
|
||||
info = get_info(app.name, 'commands')
|
||||
app_name = app.name.lower()
|
||||
|
||||
if info.get('commands'):
|
||||
|
||||
logger.info('Available commands found for {}: {}'.format(app_name, info['commands']))
|
||||
|
||||
commands = [c.strip() for c in info['commands']]
|
||||
|
||||
# trying to find an exact match command:
|
||||
command = None
|
||||
|
||||
for c in commands:
|
||||
if c.lower() == app_name:
|
||||
command = c
|
||||
logger.info("Found exact match command for '{}'".format(app_name))
|
||||
break
|
||||
|
||||
if not command:
|
||||
for c in commands:
|
||||
if not c.endswith('.apm'):
|
||||
command = c
|
||||
|
||||
if command:
|
||||
logger.info("Running '{}'".format(command))
|
||||
subprocess.Popen(['{} run {}'.format(BASE_CMD, command)], shell=True, env={**os.environ})
|
||||
return
|
||||
|
||||
logger.error("No valid command found for '{}'".format(app_name))
|
||||
else:
|
||||
logger.error("No command found for '{}'".format(app_name))
|
||||
def run(cmd: str):
|
||||
subprocess.Popen(['snap run {}'.format(cmd)], shell=True, env={**os.environ})
|
||||
|
||||
|
||||
def is_api_available() -> Tuple[bool, str]:
|
||||
output = StringIO()
|
||||
for o in SimpleProcess(['snap', 'search']).instance.stdout:
|
||||
for o in SimpleProcess([BASE_CMD, 'search']).instance.stdout:
|
||||
if o:
|
||||
output.write(o.decode())
|
||||
|
||||
output.seek(0)
|
||||
output = output.read()
|
||||
return 'error:' not in output, output
|
||||
|
||||
|
||||
def list_installed_names() -> Set[str]:
|
||||
res = run_cmd('{} list'.format(BASE_CMD), print_error=False)
|
||||
|
||||
if res:
|
||||
lines = res.split('\n')
|
||||
|
||||
if not lines[0].startswith('error'):
|
||||
return {l.split(' ')[0].strip() for l in lines[1:] if l}
|
||||
|
||||
Reference in New Issue
Block a user