import http.server import socketserver import subprocess import os import signal import urllib.parse import re import urllib.request import zipfile import stat PORT = 8001 BASE_DIR = os.path.dirname(__file__) HTML_TEMPLATE_PATH = os.path.join(BASE_DIR, 'taskmgr.html') AUTO_CONFIG_TEMPLATE_PATH = os.path.join(BASE_DIR, 'auto-config.html') # New template paths ABOUT_TEMPLATE_PATH = os.path.join(BASE_DIR, 'about.html') ANDROID_TEMPLATE_PATH = os.path.join(BASE_DIR, 'android.html') CLI_APPS_DIR = os.path.expanduser('~/usr/local/bin') # CLI apps directory WEB_APPS_DIR = os.path.expanduser('~/apps') # Web apps directory AVAILABLE_APPS_URL = 'http://berrystore.sw7ft.com/bins/' WEB_APPS_URL = 'http://berrystore.sw7ft.com/apps/' # New: APKs directory APKS_URL = 'http://berrystore.sw7ft.com/apks/' PROFILE_FILE = os.path.expanduser('~/.profile') app_ports = {} # Store mapping of PID to port class TaskManagerHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): parsed_path = urllib.parse.urlparse(self.path) # Handle /action?... if self.path.startswith('/action'): params = urllib.parse.parse_qs(parsed_path.query) action = params.get('action', [''])[0] app_name = params.get('app_name', [''])[0] pid = params.get('pid', [''])[0] app_type = params.get('app_type', ['cli'])[0] if action == 'start' and app_name: self.start_app(app_name, app_type) elif action == 'stop' and pid: self.stop_app(pid) elif action == 'install' and app_name: self.install_app(app_name, app_type) elif action == 'delete' and app_name: self.delete_app(app_name, app_type) elif action == 'enable_auto_start' and app_name: self.enable_auto_start_app(app_name, app_type) elif action == 'disable_auto_start' and app_name: self.disable_auto_start_app(app_name) referer = self.headers.get('Referer', '/') self.send_response(303) self.send_header('Location', referer) self.end_headers() # Handle /auto-config page elif self.path.startswith('/auto-config'): self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() html = self.generate_auto_config_html() self.wfile.write(html.encode()) # New: handle /about page elif self.path == '/about': self.serve_about_page() # New: handle /android page elif self.path == '/android': self.serve_android_page() # Otherwise serve the main page else: self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() html = self.generate_html() self.wfile.write(html.encode()) # ------------------------------------------------------------------------- # New methods for /about and /android # ------------------------------------------------------------------------- def serve_about_page(self): """Serve the about.html page.""" try: with open(ABOUT_TEMPLATE_PATH, 'r') as file: html = file.read() self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() self.wfile.write(html.encode()) except Exception as e: print(f"Error loading about.html: {e}") self.send_error(500, 'Error loading about.html') def serve_android_page(self): """Serve the android.html page, after populating it with the list of .apk files.""" try: # Fetch the list of .apk files apks = self.fetch_android_apks() with open(ANDROID_TEMPLATE_PATH, 'r') as file: html = file.read() # Build a simple list or table of the APK files # For each APK, we can provide a download link or just list them if not apks: apks_html = '

No APKs found.

' else: apks_html = '' # Replace a placeholder in android.html, e.g. , with our content html = html.replace('', apks_html) self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() self.wfile.write(html.encode()) except Exception as e: print(f"Error serving android.html: {e}") self.send_error(500, 'Error serving android.html') # ------------------------------------------------------------------------- # Existing methods (plus new helper fetch_android_apks) # ------------------------------------------------------------------------- def install_app(self, app_name, app_type='cli'): try: if app_type == 'cli': download_url = urllib.parse.urljoin(AVAILABLE_APPS_URL, app_name) install_dir = CLI_APPS_DIR elif app_type == 'web': download_url = urllib.parse.urljoin(WEB_APPS_URL, app_name) install_dir = WEB_APPS_DIR else: print(f"Unknown app type:{app_type}" ) return print(f"Downloading {download_url}") with urllib.request.urlopen(download_url) as response: if response.status != 200: print(f"Failed to download {download_url}: Status {response.status}") return zip_data = response.read() if not os.path.exists(install_dir): os.makedirs(install_dir) print(f"Created installation directory at: {install_dir}") zip_path = os.path.join(install_dir, app_name) with open(zip_path, 'wb') as f: f.write(zip_data) print(f"Downloaded {app_name} to {zip_path}") with zipfile.ZipFile(zip_path, 'r') as zip_ref: if app_type == 'web': app_folder = os.path.splitext(app_name)[0] extract_path = os.path.join(install_dir, app_folder) os.makedirs(extract_path, exist_ok=True) zip_ref.extractall(extract_path) print(f"Extracted {app_name} to {extract_path}") else: # For CLI apps, handle lib/ directories specially lib_dir = os.path.expanduser('~/usr/local/lib') bin_dir = os.path.expanduser('~/usr/local/bin') for member in zip_ref.namelist(): if member.startswith('lib/'): # Extract lib files to ~/usr/local/lib/ # Remove 'lib/' prefix and extract to lib directory lib_file = member[4:] # Remove 'lib/' prefix if lib_file: # Skip empty directory entries target_path = os.path.join(lib_dir, lib_file) os.makedirs(os.path.dirname(target_path), exist_ok=True) with zip_ref.open(member) as source, open(target_path, 'wb') as target: target.write(source.read()) print(f"Extracted library: {lib_file} to {lib_dir}") elif member.startswith('bin/'): # Extract bin files to ~/usr/local/bin/ bin_file = member[4:] # Remove 'bin/' prefix if bin_file: # Skip empty directory entries target_path = os.path.join(bin_dir, bin_file) os.makedirs(os.path.dirname(target_path), exist_ok=True) with zip_ref.open(member) as source, open(target_path, 'wb') as target: target.write(source.read()) print(f"Extracted binary: {bin_file} to {bin_dir}") else: # For other files, extract to bin directory if not member.endswith('/'): # Skip directories target_path = os.path.join(bin_dir, member) os.makedirs(os.path.dirname(target_path), exist_ok=True) with zip_ref.open(member) as source, open(target_path, 'wb') as target: target.write(source.read()) print(f"Extracted: {member} to {bin_dir}") print(f"Extracted {app_name} with proper lib/bin separation") os.remove(zip_path) print(f"Removed zip file: {zip_path}") # Set executable permissions for installed files if app_type == 'web': # For web apps, set permissions in the install directory for root, dirs, files in os.walk(install_dir): for file in files: file_path = os.path.join(root, file) if os.path.isfile(file_path): st = os.stat(file_path) os.chmod(file_path, st.st_mode | stat.S_IEXEC) else: # For CLI apps, set permissions in both bin and lib directories for directory in [os.path.expanduser('~/usr/local/bin'), os.path.expanduser('~/usr/local/lib')]: if os.path.exists(directory): for root, dirs, files in os.walk(directory): for file in files: file_path = os.path.join(root, file) if os.path.isfile(file_path): st = os.stat(file_path) os.chmod(file_path, st.st_mode | stat.S_IEXEC) print(f"Successfully installed {app_name}") # Note: PATH setup is now handled automatically by ensure_cli_paths() except Exception as e: print(f"Error installing {app_name}: {e}") # Ensure CLI paths are set up in .profile after installation if app_type == 'cli': self.ensure_cli_paths() def ensure_cli_paths(self): """Ensure ~/usr/local/bin and ~/usr/local/lib are in PATH and LD_LIBRARY_PATH""" try: # Create directories if they don't exist os.makedirs(os.path.expanduser('~/usr/local/bin'), exist_ok=True) os.makedirs(os.path.expanduser('~/usr/local/lib'), exist_ok=True) # Check if .profile exists if not os.path.exists(PROFILE_FILE): # Create .profile if it doesn't exist with open(PROFILE_FILE, 'w') as f: f.write('umask 002\n') # Read current profile with open(PROFILE_FILE, 'r') as f: profile_content = f.read() # Check if PATH export already exists path_export = 'export PATH="$HOME/usr/local/bin:$PATH"' lib_export = 'export LD_LIBRARY_PATH="$HOME/usr/local/lib:$LD_LIBRARY_PATH"' lines_to_add = [] if path_export not in profile_content: lines_to_add.append(path_export) print("Added ~/usr/local/bin to PATH in .profile") if lib_export not in profile_content: lines_to_add.append(lib_export) print("Added ~/usr/local/lib to LD_LIBRARY_PATH in .profile") if lines_to_add: with open(PROFILE_FILE, 'a') as f: f.write('\n# CLI Apps PATH and Library Path Setup\n') for line in lines_to_add: f.write(f'{line}\n') print("CLI paths have been automatically configured in .profile") else: print("CLI paths already configured in .profile") except Exception as e: print(f"Error setting up CLI paths: {e}") def delete_app(self, app_name, app_type='cli'): try: if app_type == 'web': app_folder = os.path.splitext(app_name)[0] app_path = os.path.join(WEB_APPS_DIR, app_folder) else: app_path = os.path.join(CLI_APPS_DIR, app_name) if not os.path.exists(app_path): print(f"App {app_name} does not exist.") return if os.path.isfile(app_path) or os.path.islink(app_path): os.remove(app_path) print(f"Deleted app: {app_path}") elif os.path.isdir(app_path): subprocess.run(['rm', '-rf', app_path]) print(f"Deleted directory app: {app_path}") except Exception as e: print(f"Error deleting {app_name}: {e}") def start_app(self, app_name, app_type='cli'): try: if app_type == 'web': app_folder = os.path.splitext(app_name)[0] app_path = os.path.join(WEB_APPS_DIR, app_folder, 'app.py') else: app_path = os.path.join(CLI_APPS_DIR, app_name) if not os.path.exists(app_path): print(f"App {app_name} does not exist.") return print(f"Starting app: {app_path}") python_path = os.path.expanduser('~/python3_ntoarmv7-qnx-static/tools/python3/python3') if not os.path.exists(python_path): python_path = 'python3' command = [python_path, '-u', app_path] process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) while True: output_line = process.stdout.readline() if not output_line: break output_line = output_line.strip() print(f"App output: {output_line}") # Try to detect port in output match = re.search(r'(\d{4,5})', output_line) if match: port = match.group(1) app_ports[process.pid] = port print(f"App started on port {port}") break print(f'Successfully started app {app_name} with PID {process.pid}') except Exception as e: print(f'Error starting {app_name}: {e}') def stop_app(self, pid): try: os.kill(int(pid), signal.SIGTERM) print(f'Stopped process {pid}') if int(pid) in app_ports: del app_ports[int(pid)] except Exception as e: print(f'Error stopping process {pid}: {e}') def generate_html(self): try: with open(HTML_TEMPLATE_PATH, 'r') as file: html = file.read() processes_html = self.generate_processes_html() installed_apps_html = self.generate_installed_apps_html() available_cli_html = self.generate_available_cli_html() available_web_html = self.generate_available_web_apps_html() processes_html = processes_html if processes_html is not None else '' installed_apps_html = installed_apps_html if installed_apps_html is not None else '' available_cli_html = available_cli_html if available_cli_html is not None else '' available_web_html = available_web_html if available_web_html is not None else '' html = html.replace('', processes_html) html = html.replace('', installed_apps_html) html = html.replace('', available_cli_html) html = html.replace('', available_web_html) return html except Exception as e: print(f'Error loading HTML template: {e}') return '

Error loading HTML template

' def generate_processes_html(self): try: processes = self.get_python_processes() rows = '' for pid, cmd in processes: launch_link = '' if int(pid) in app_ports: port = app_ports[int(pid)] launch_link = f' - Launch UI' rows += f'{pid}{cmd}Stop{launch_link}' return rows except Exception as e: print(f'Error generating processes HTML: {e}') return '' def generate_installed_apps_html(self): try: cli_apps = self.scan_apps_directory(CLI_APPS_DIR, 'cli') web_apps = self.scan_apps_directory(WEB_APPS_DIR, 'web') all_apps = cli_apps + web_apps if not all_apps: return '

No installed apps found.

' apps_html = '' apps_html += '' for app, app_type in all_apps: app_encoded = urllib.parse.quote(app) if app_type == 'web': apps_html += f'' else: apps_html += f'' apps_html += '
App NameTypeAction
{app}Web AppStart | Delete
{app}CLIDelete
' return apps_html except Exception as e: print(f'Error generating installed apps HTML: {e}') return '' def enable_auto_start_app(self, app_name, app_type='cli'): try: if app_type == 'web': app_command = f'python3 "{os.path.join(WEB_APPS_DIR, app_name, "app.py")}" &\n' else: app_command = f'python3 "{os.path.join(CLI_APPS_DIR, app_name)}" &\n' with open(PROFILE_FILE, 'r') as f: profile_contents = f.readlines() AUTO_START_MARKER_START = f'# <<< Auto-Start for {app_name} >>>' AUTO_START_MARKER_END = f'# <<< End Auto-Start for {app_name} >>>' if any(app_command.strip() in line.strip() for line in profile_contents): print(f"Auto-start for {app_name} is already enabled.") return with open(PROFILE_FILE, 'a') as f: f.write(f'\n{AUTO_START_MARKER_START}\n') f.write(app_command) f.write(f'{AUTO_START_MARKER_END}\n') print(f"Enabled auto-start for {app_name}.") except Exception as e: print(f"Error enabling auto start for {app_name}: {e}") def disable_auto_start_app(self, app_name): try: if not os.path.exists(PROFILE_FILE): print(f"{PROFILE_FILE} does not exist.") return AUTO_START_MARKER_START = f'# <<< Auto-Start for {app_name} >>>' AUTO_START_MARKER_END = f'# <<< End Auto-Start for {app_name} >>>' with open(PROFILE_FILE, 'r') as f: profile_lines = f.readlines() updated_lines = [] skip = False for line in profile_lines: if line.strip() == AUTO_START_MARKER_START: skip = True continue if line.strip() == AUTO_START_MARKER_END: skip = False continue if not skip: updated_lines.append(line) with open(PROFILE_FILE, 'w') as f: f.writelines(updated_lines) print(f"Disabled auto-start for {app_name}.") except Exception as e: print(f"Error disabling auto start for {app_name}: {e}") def generate_available_cli_html(self): try: available_zips = self.fetch_available_zips() installed_cli_apps = set(app for app, _ in self.scan_apps_directory(CLI_APPS_DIR, 'cli')) # Exclude installed apps available_zips = [z for z in available_zips if os.path.splitext(z)[0] not in installed_cli_apps] if not available_zips: return '

No Command Line Utilities available.

' apps_html = '
' for zip_file in available_zips: app_name = os.path.splitext(zip_file)[0] app_encoded = urllib.parse.quote(zip_file) apps_html += f'''
{app_name}
Install
''' apps_html += '
' return apps_html except Exception as e: print(f"Error generating available CLI apps HTML: {e}") return '

Error loading Command Line Utilities.

' def generate_available_web_apps_html(self): try: available_zips = self.fetch_web_apps() installed_web_apps = set(app for app, _ in self.scan_apps_directory(WEB_APPS_DIR, 'web')) # Exclude installed apps available_zips = [z for z in available_zips if os.path.splitext(z)[0] not in installed_web_apps] if not available_zips: return '

No Web Apps available.

' apps_html = '
' for zip_file in available_zips: app_name = os.path.splitext(zip_file)[0] app_encoded = urllib.parse.quote(zip_file) apps_html += f'''
{app_name}
Install
''' apps_html += '
' return apps_html except Exception as e: print(f"Error generating available Web Apps HTML: {e}") return '

Error loading Web Apps.

' # New: Fetch .apk files from the berrystore directory def fetch_android_apks(self): """ Fetch and parse the listing of .apk files from APKS_URL. Returns a list of filenames (e.g. ['myapp.apk', 'anotherapp.apk', ...]). """ try: with urllib.request.urlopen(APKS_URL) as response: if response.status != 200: print(f"Failed to fetch Android APKs: Status {response.status}") return [] html_content = response.read().decode() # Look for all anchors linking to something ending in .apk apk_files = re.findall(r'href=["\']([^"\']+\.apk)["\']', html_content) # Just keep the base filenames apk_files = [os.path.basename(a) for a in apk_files] return apk_files except Exception as e: print(f"Error fetching Android APKs: {e}") return [] def fetch_available_zips(self): try: with urllib.request.urlopen(AVAILABLE_APPS_URL) as response: if response.status != 200: print(f"Failed to fetch CLI Utilities: Status {response.status}") return [] html_content = response.read().decode() zip_files = re.findall(r'href=["\']([^"\']+\.zip)["\']', html_content) zip_files = [os.path.basename(z) for z in zip_files] return zip_files except Exception as e: print(f"Error fetching CLI Utilities zips: {e}") return [] def fetch_web_apps(self): try: with urllib.request.urlopen(WEB_APPS_URL) as response: if response.status != 200: print(f"Failed to fetch Web Apps: Status {response.status}") return [] html_content = response.read().decode() zip_files = re.findall(r'href=["\']([^"\']+\.zip)["\']', html_content) zip_files = [os.path.basename(z) for z in zip_files] return zip_files except Exception as e: print(f"Error fetching Web Apps zips: {e}") return [] def get_python_processes(self): processes = [] try: output = subprocess.check_output(['pidin', 'ar']).decode() lines = output.strip().split('\n') for line in lines: if 'python' in line: parts = line.split() pid = parts[0] cmd = ' '.join(parts[1:]) if '/data/' in cmd: index = cmd.find('/data/') + len('/data/') cmd = cmd[index:] processes.append((pid, cmd)) except Exception as e: print(f'Error getting processes: {e}') return processes def scan_apps_directory(self, directory, app_type): apps = [] try: if not os.path.exists(directory): return apps for root, dirs, files in os.walk(directory): for file in files: if app_type == 'web': if file == 'app.py': app_folder = os.path.relpath(root, directory) apps.append((app_folder, 'web')) else: # CLI apps assumed to be executables without .py extension if not file.endswith('.py'): rel_path = os.path.relpath(os.path.join(root, file), directory) apps.append((rel_path, 'cli')) except Exception as e: print(f'Error scanning apps directory: {e}') return apps def generate_auto_config_html(self): """ Generate the HTML for the auto-config page, showing web apps with options to enable or disable auto start. """ try: with open(AUTO_CONFIG_TEMPLATE_PATH, 'r') as file: html = file.read() all_apps = self.get_all_installed_apps() if not all_apps: apps_html = '

No installed web apps found.

' else: apps_html = '' apps_html += '' for app, app_type in all_apps: if app_type == 'web': # Only include web apps app_encoded = urllib.parse.quote(app) if self.is_auto_start_enabled(app): # Show Disable Auto Start link if enabled apps_html += f'' else: # Show Enable Auto Start link if not enabled apps_html += f'' apps_html += '
App NameTypeAuto Start
{app}{app_type.upper()}Disable Auto Start
{app}{app_type.upper()}Enable Auto Start
' html = html.replace('', apps_html) return html except Exception as e: print(f'Error loading auto-config HTML template: {e}') return '

Error loading auto-config template

' def get_all_installed_apps(self): cli_apps = self.scan_apps_directory(CLI_APPS_DIR, 'cli') web_apps = self.scan_apps_directory(WEB_APPS_DIR, 'web') return cli_apps + web_apps def is_auto_start_enabled(self, app_name): """ Check if auto-start is enabled for the given app by inspecting ~/.profile. """ try: if not os.path.exists(PROFILE_FILE): return False with open(PROFILE_FILE, 'r') as f: profile_lines = f.readlines() for line in profile_lines: if f'python3 "{os.path.join(WEB_APPS_DIR, app_name, "app.py")}" &' in line: return True return False except Exception as e: print(f"Error checking auto-start for {app_name}: {e}") return False def check_path(): current_path = os.environ.get('PATH', '') paths = current_path.split(os.pathsep) if CLI_APPS_DIR not in paths: print(f"Warning: {CLI_APPS_DIR} is not in your PATH.") print("Add to ~/.profile:") print(f'export PATH="{CLI_APPS_DIR}:$PATH"') if WEB_APPS_DIR not in paths: print(f"Note: {WEB_APPS_DIR} is not in your PATH.") print("If you want to run web apps from command line, consider adding:") print(f'export PATH="{WEB_APPS_DIR}:$PATH"') if __name__ == '__main__': check_path() with socketserver.TCPServer(("", PORT), TaskManagerHandler) as httpd: print(f"Serving on port {PORT}") httpd.serve_forever()