Add previous code

android_manager/android_manager.py

@@ -9,10 +9,15 @@ import sys
 import os
 import json
 import time
+import re
 import logging
 from logging.handlers import SysLogHandler
 import click
 import click_config_file
+from ppadb.client import Client as AdbClient
+import ppadb
+import requests
+import yaml
 
 
 HOME_FOLDER = os.environ.get('HOME', os.environ.get('USERPROFILE', '/'))
@@ -46,6 +51,210 @@ class AndroidManager:
             "last_update": 0,
         }
         self.data = self._read_cached_data()
+        self.lines = None
+        self.client = AdbClient(host=self.config['host'], port=self.config['port'])
+        self.devices = []
+        for device_object in self.client.devices():
+            dev = {}
+            dev['object'] = device_object
+            try:
+                dev['properties'] = device_object.get_properties()
+                dev['name'] = dev['properties']['ro.semc.product.name']
+                device_object.shell('dumpsys bluetooth_manager', handler=self._handle_output)
+                for line in self.lines:
+                    match = re.search(r'^  Name: (.*)$', line)
+                    if match:
+                        dev['name'] = match.group(1)
+                    match = re.search(r'^  address: (.*)$', line)
+                    if match:
+                        dev['bt_address'] = match.group(1)
+                self.devices.append(dev)
+                self._log.debug(
+                    "Found Android device: %s",
+                    dev['name']
+                )
+            except Exception as error:
+                self._log.error(f"Error reading device properties. Maybe the device is off-line? {error}")
+
+        self.get_installed_apps()
+
+    def get_installed_apps(self, device_name=None):
+        '''Get a list of installed app in a device or all'''
+        for device in self.devices:
+            if device_name == device['name'] or device_name is None:
+                device['object'].shell("pm list packages -3 -f", handler=self._handle_output)
+                device['apps'] = []
+                for line in self.lines:
+                    app = {}
+                    match = re.search(r'^package:(.*\.apk)=(.*)$', line)
+                    if match:
+                        app['path'] = os.path.dirname(match.group(1))
+                        app['id'] = match.group(2)
+                    else:
+                        self._log.error(
+                            "RegExp failed to identify app in line: '%s'",
+                            line
+                        )
+                    device['apps'].append(app)
+
+    def restore_apps(self, backup_text_file, device_product_name):
+        '''Restore apps from a text file using F-Droid repository'''
+        if not os.path.exists(backup_text_file):
+            self._log.error(
+                "The backup file '%s' doesn't exist.",
+                backup_text_file
+            )
+            return False
+        session = requests.Session()
+        with open(backup_text_file, 'r', encoding='utf-8') as backup_file:
+            content = backup_file.read()
+            for app_code in content.split():
+                self._log.info(
+                    "Processing app '%s' for restoration...",
+                    app_code
+                )
+                app_url = f"https://gitlab.com/fdroid/fdroiddata/-/raw/master/metadata/{app_code}.yml"
+                result = session.get(app_url)
+                if result.status_code > 399:
+                    self._log.error(
+                        "Error %s fetching app data file from '%s'.",
+                        result.status_code,
+                        app_url
+                    )
+                    continue
+                app_data = yaml.load(result.content, Loader=yaml.Loader)
+                latest_build = 0
+                if not app_data:
+                    self._log.error(
+                        "Error loading YAML data from remote application file '%s'. Result: %s",
+                        app_url,
+                        result.content
+                    )
+                    continue
+                if 'Builds' not in app_data:
+                    self._log.error(
+                        "'Builds' section not found in application data. %s",
+                        app_data
+                    )
+                    continue
+                for build in app_data['Builds']:
+                    if build['versionCode'] > latest_build:
+                        latest_build = build['versionCode']
+                if latest_build == 0:
+                    self._log.error(
+                        "Unable to find the latest version code in the app data file '%s'",
+                        app_url
+                    )
+                    continue
+                self._log.debug(
+                    "Latest build version code name seems to be '%s'",
+                    latest_build
+                )
+                build_url = f"https://verification.f-droid.org/{app_code}_{latest_build}.apk.json"
+                result = session.get(
+                    build_url
+                )
+                if result.status_code > 399:
+                    self._log.error(
+                        "Error %s fetching build information from file '%s'.",
+                        result.status_code,
+                        build_url
+                    )
+                    continue
+                build_data = json.loads(result.content)
+                apk_file_url = None
+                for build in build_data.keys():
+                    apk_file_url = build_data[build]['url']
+                    self._log.debug(
+                        "Found file: %s",
+                        apk_file_url
+                    )
+                if not apk_file_url:
+                    self._log.error(
+                        "Error finding APK file in '%s'. %s",
+                        build_url,
+                        result.content
+                    )
+                    continue
+                result = session.get(
+                    apk_file_url
+                )
+                apk_file = os.path.join(os.path.dirname(backup_text_file), f"{app_code}.apk")
+                target_device = None
+                with open(apk_file, 'w+b') as apk_object_file:
+                    apk_object_file.write(result.content)
+                    if device_product_name:
+                        for device in self.devices:
+                            if device['name'] == device_product_name:
+                                target_device = device
+                    else:
+                        target_device=self.devices[0]
+                    if target_device:
+                        self._log.debug(
+                            "Installing app '%s' with version code '%s' to device '%s'...",
+                            app_code,
+                            latest_build,
+                            target_device['name']
+                        )
+                        try:
+                            target_device['object'].install(apk_file)
+                            self._log.debug(
+                                "Installed without errors."
+                            )
+                        except ppadb.InstallError as error:
+                            if 'INSTALL_FAILED_ALREADY_EXISTS' in error:
+                                self._log.debug(
+                                    'App already installed, not installing again. %s',
+                                    error
+                                )
+                            else:
+                                self._log.error(
+                                    "Error installing application '%s' with version code '%s' to device '%s'. %s",
+                                    app_code,
+                                    latest_build,
+                                    target_device['name'],
+                                    error
+                                )
+        return True
+
+    def backup_apps(self, backup_path, just_names):
+        '''Backup apps either to a text file or to a APK files'''
+        for device in self.devices:
+            local_directory = os.path.join(backup_path, device.get('name', 'unknown_device'))
+            if not os.path.exists(local_directory):
+                os.mkdir(local_directory)
+            if just_names and os.path.exists(os.path.join(local_directory, 'apks_list.txt')):
+                os.remove(os.path.join(local_directory, 'apks_list.txt'))
+            for app in device['apps']:
+                remote_filename = f"{app['path']}/base.apk"
+                if not just_names:
+                    local_filename = os.path.join(local_directory, f"{app['id']}.apk")
+                    self._log.info(
+                        "Saving '%s' into '%s'...",
+                        remote_filename,
+                        local_filename
+                    )
+                    device['object'].pull(remote_filename, local_filename)
+                else:
+                    file_name = os.path.join(local_directory, 'apks_list.txt')
+                    self._log.info(
+                        "Saving list of apps into file '%s'...",
+                        file_name
+                    )
+                    with open(file_name, 'a', encoding='UTF-8') as list_file:
+                        list_file.write(f"{app['id']}\n")
+
+    def _handle_output(self, connection):
+        buffer = ""
+        while True:
+            data = connection.read(1024)
+            if not data:
+                break
+            # print(data)
+            buffer += data.decode('utf-8', errors='ignore')
+
+        connection.close()
+        self.lines = buffer.strip().split('\n')
 
     def close(self):
         '''Close class and save data'''
@@ -116,7 +325,7 @@ class AndroidManager:
         return True
 
 
-@click.command()
+@click.group()
 @click.option(
     "--debug-level",
     "-d",
@@ -145,9 +354,39 @@ class AndroidManager:
     default=60*60*24*7,
     help='Max age in seconds for the cache'
 )
+@click.option('--host', '-H', default='127.0.0.1', help='ADB server host name or IP.')
+@click.option('--port', '-p', default=5037, help='ADB server port to use.')
 # @click.option("--dummy","-n", is_flag=True,
 # help="Don't do anything, just show what would be done.")
 @click_config_file.configuration_option()
+@click.pass_context
+def cli(ctx, **kwargs):
+    ctx.ensure_object(dict)
+    ctx.obj['debug_level'] = kwargs['debug_level']
+    ctx.obj['android_manager'] = AndroidManager(**kwargs)
+
+@cli.command()
+@click.option('--backup-path', '-p', required=True, help='Path to save the backed APKs')
+@click.option('--just-names', '-n', is_flag=True, help='Save only the filename')
+@click_config_file.configuration_option()
+@click.pass_context
+def backup_apps(ctx, backup_path, just_names):
+    '''Backup Android apps'''
+    ctx.obj['android_manager'].backup_apps(backup_path, just_names)
+
+@cli.command()
+@click.option('--backup-text-file', '-p', required=True, help='Path to save the backed APKs')
+@click.option(
+    '--device-product-name',
+    '-n',
+    help='Device product name to install the apps. If not indicated it will install in the first device found.'
+)
+@click_config_file.configuration_option()
+@click.pass_context
+def restore_apps(ctx, backup_text_file, device_product_name):
+    '''Restore Android apps'''
+    ctx.obj['android_manager'].restore_apps(backup_text_file, device_product_name)
+
 def __main__(**kwargs):
     obj = AndroidManager(**kwargs)
     obj.close()

requirements.txt

@@ -1,2 +1,5 @@
 click
 click_config_file
+pure-python-adb
+requests
+pyyaml