import http.server import socketserver import subprocess import os import signal import urllib.parse import re import urllib.request import zipfile import stat import time import threading import socket import json 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 # Performance optimization: Caching CACHE_DURATION = 300 # 5 minutes cache REQUEST_TIMEOUT = 10 # 10 seconds timeout cache = {} def get_cached_or_fetch(url, cache_key): """Get data from cache or fetch from URL with timeout""" current_time = time.time() # Check if we have cached data that's still valid if cache_key in cache: cached_time, cached_data = cache[cache_key] if current_time - cached_time < CACHE_DURATION: return cached_data # Fetch new data with timeout try: request = urllib.request.Request(url) with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT) as response: if response.status != 200: print(f"Failed to fetch {url}: Status {response.status}") return None data = response.read().decode() # Cache the result cache[cache_key] = (current_time, data) return data except (urllib.error.URLError, socket.timeout) as e: print(f"Error fetching {url}: {e}") # Return stale cache if available if cache_key in cache: cached_time, cached_data = cache[cache_key] print(f"Using stale cache for {cache_key}") return cached_data return None 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() # Handle lazy loading endpoints elif self.path == '/api/available-cli': self.serve_available_cli_json() elif self.path == '/api/available-web': self.serve_available_web_json() elif self.path.startswith('/api/app-details'): self.serve_app_details() elif self.path == '/news': self.serve_news_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()) def serve_available_cli_json(self): """Serve available CLI apps as JSON for lazy loading""" try: html = self.generate_available_cli_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 available CLI apps: {e}") self.send_error(500, 'Error loading CLI apps') def serve_available_web_json(self): """Serve available web apps as JSON for lazy loading""" try: html = self.generate_available_web_apps_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 available web apps: {e}") self.send_error(500, 'Error loading web apps') def serve_app_details(self): """Serve app details including overview.md content""" try: parsed_path = urllib.parse.urlparse(self.path) params = urllib.parse.parse_qs(parsed_path.query) app_name = params.get('app_name', [''])[0] app_type = params.get('app_type', ['cli'])[0] if not app_name: self.send_error(400, 'App name required') return # Get app description from overview.md description = self.get_app_description(app_name, app_type) # Create install URL app_encoded = urllib.parse.quote(app_name + '.zip') install_url = f'/action?action=install&app_name={app_encoded}&app_type={app_type}' # Return JSON response response = { 'description': description, 'install_url': install_url } self.send_response(200) self.send_header('Content-type', 'application/json') self.end_headers() self.wfile.write(json.dumps(response).encode()) except Exception as e: print(f"Error serving app details: {e}") self.send_error(500, 'Error loading app details') def get_app_description(self, app_name, app_type): """ Fetch app description from overview.md file on the server. Returns formatted HTML description or fallback text. """ try: # Construct URL to fetch overview.md from the appropriate server if app_type == 'cli': overview_url = urllib.parse.urljoin(AVAILABLE_APPS_URL, f'{app_name}/overview.md') else: # web overview_url = urllib.parse.urljoin(WEB_APPS_URL, f'{app_name}/overview.md') # Fetch the overview.md file request = urllib.request.Request(overview_url) with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT) as response: if response.status == 200: content = response.read().decode('utf-8') # Convert markdown to simple HTML (basic conversion) html_content = self.markdown_to_html(content) return html_content else: return f'

No description available for {app_name}.

' except (urllib.error.URLError, urllib.error.HTTPError, socket.timeout) as e: print(f"Error fetching overview.md for {app_name}: {e}") return f'

No description available for {app_name}.

' except Exception as e: print(f"Error processing overview.md for {app_name}: {e}") return f'

No description available for {app_name}.

' def markdown_to_html(self, markdown_text): """ Convert basic markdown to HTML for display in modal. Handles common markdown elements like headers, paragraphs, lists, and links. """ try: lines = markdown_text.split('\n') html_lines = [] in_list = False for line in lines: line = line.strip() if not line: if in_list: html_lines.append('') in_list = False html_lines.append('
') continue # Headers if line.startswith('# '): html_lines.append(f'

{line[2:]}

') elif line.startswith('## '): html_lines.append(f'

{line[3:]}

') elif line.startswith('### '): html_lines.append(f'

{line[4:]}

') # Lists elif line.startswith('- ') or line.startswith('* '): if not in_list: html_lines.append('') in_list = False html_lines.append(f'

{line}

') # Close any open list if in_list: html_lines.append('') return '\n'.join(html_lines) except Exception as e: print(f"Error converting markdown to HTML: {e}") return f'

{markdown_text}

' def serve_news_page(self): """Serve the news page with content from news.json""" try: # Load news from JSON file news_data = self.load_news_data() # Generate HTML content news_html = self.generate_news_html(news_data) self.send_response(200) self.send_header('Content-type', 'text/html') self.end_headers() self.wfile.write(news_html.encode()) except Exception as e: print(f"Error serving news page: {e}") self.send_error(500, 'Error loading news') def load_news_data(self): """Load news data from news.json file""" try: news_file_path = os.path.join(BASE_DIR, 'news.json') if not os.path.exists(news_file_path): return {"news": [], "last_updated": ""} with open(news_file_path, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: print(f"Error loading news data: {e}") return {"news": [], "last_updated": ""} def generate_news_html(self, news_data): """Generate HTML for the news page""" try: # Start with the base HTML structure html = f''' BB10 News

BB10 News

Back to Task Manager Auto Start Apps Android Apps About Sw7ft

Latest Updates

''' # Add news items if news_data.get('news'): for news_item in news_data['news']: # Convert markdown content to HTML content_html = self.markdown_to_html(news_item['content']) # Format date date_str = news_item.get('date', 'Unknown') author_str = news_item.get('author', 'Unknown') category_str = news_item.get('category', 'general') html += f'''
{news_item['title']}
{date_str} by {author_str} {category_str}
{content_html}
''' else: html += '''

No news available at the moment.

Check back soon for updates!

''' # Add last updated footer last_updated = news_data.get('last_updated', '') if last_updated: html += f'''
Last updated: {last_updated}
''' html += '''
''' return html except Exception as e: print(f"Error generating news HTML: {e}") return '''

Error loading news

''' # ------------------------------------------------------------------------- # 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') 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}") request = urllib.request.Request(download_url) with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT) 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 file: {member} to {bin_dir}") # Make binaries executable if app_type == 'cli': self.ensure_cli_paths() extracted_files = zip_ref.namelist() for file in extracted_files: if not file.endswith('/'): if file.startswith('bin/'): file_path = os.path.join(bin_dir, file[4:]) else: file_path = os.path.join(bin_dir, file) if os.path.exists(file_path): os.chmod(file_path, stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH) print(f"Made {file_path} executable") os.remove(zip_path) print(f"Removed temporary zip file: {zip_path}") except Exception as e: print(f"Error installing app {app_name}: {e}") def ensure_cli_paths(self): """Ensure CLI paths exist""" for path in [CLI_APPS_DIR, os.path.expanduser('~/usr/local/lib')]: if not os.path.exists(path): os.makedirs(path) print(f"Created directory: {path}") def delete_app(self, app_name, app_type='cli'): try: if app_type == 'cli': app_path = os.path.join(CLI_APPS_DIR, app_name) if os.path.exists(app_path): os.remove(app_path) print(f"Deleted CLI app: {app_path}") elif app_type == 'web': app_path = os.path.join(WEB_APPS_DIR, app_name) if os.path.exists(app_path): import shutil shutil.rmtree(app_path) print(f"Deleted web app: {app_path}") except Exception as e: print(f"Error deleting app {app_name}: {e}") def start_app(self, app_name, app_type='cli'): try: if app_type == 'web': app_path = os.path.join(WEB_APPS_DIR, app_name, 'app.py') if os.path.exists(app_path): process = subprocess.Popen(['python3', app_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE) print(f"Started web app: {app_name} with PID: {process.pid}") # Give the app a moment to start up and bind to a port time.sleep(2) # Try to detect the actual port the app is using port = self.detect_app_port(process.pid, app_name) if port: app_ports[process.pid] = port print(f"Detected port {port} for {app_name}") else: print(f"Could not detect port for {app_name}") else: print(f"Web app not found: {app_path}") else: # Handle CLI apps app_path = os.path.join(CLI_APPS_DIR, app_name) if os.path.exists(app_path): # Make sure the file is executable os.chmod(app_path, os.stat(app_path).st_mode | stat.S_IXUSR) # Start the CLI app in background process = subprocess.Popen([app_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE) print(f"Started CLI app: {app_name} with PID: {process.pid}") else: print(f"CLI app not found: {app_path}") except Exception as e: print(f"Error starting app {app_name}: {e}") def detect_app_port(self, pid, app_name): """ Detect the actual port a web app is running on by checking netstat output """ try: print(f"DEBUG: Detecting port for PID {pid}, app {app_name}") # Method 1: Try netstat (Linux/QNX style) try: output = subprocess.check_output(['netstat', '-tlnp'], timeout=5, stderr=subprocess.DEVNULL).decode() print(f"DEBUG: netstat output: {output}") lines = output.strip().split('\n') for line in lines: print(f"DEBUG: Checking line: {line}") if str(pid) in line and ('LISTEN' in line or 'tcp' in line): print(f"DEBUG: Found matching line: {line}") # Parse the port from netstat output # Example: tcp 0 0 0.0.0.0:8080 0.0.0.0:* LISTEN 1234/python3 parts = line.split() for part in parts: if ':' in part and part.count(':') == 1: try: port = int(part.split(':')[1]) if 8000 <= port <= 9000: # Common web app port range print(f"DEBUG: Found port {port} in netstat") return port except (ValueError, IndexError): continue except (subprocess.CalledProcessError, FileNotFoundError) as e: print(f"DEBUG: netstat not available: {e}") # Method 2: Try ss command (alternative to netstat) try: output = subprocess.check_output(['ss', '-tlnp'], timeout=5, stderr=subprocess.DEVNULL).decode() print(f"DEBUG: ss output: {output}") lines = output.strip().split('\n') for line in lines: if str(pid) in line and 'LISTEN' in line: print(f"DEBUG: Found matching ss line: {line}") # Parse ss output format parts = line.split() for part in parts: if ':' in part and part.count(':') == 1: try: port = int(part.split(':')[1]) if 8000 <= port <= 9000: print(f"DEBUG: Found port {port} in ss") return port except (ValueError, IndexError): continue except (subprocess.CalledProcessError, FileNotFoundError) as e: print(f"DEBUG: ss not available: {e}") # Method 3: Parse the app.py file for port configuration app_path = os.path.join(WEB_APPS_DIR, app_name, 'app.py') if os.path.exists(app_path): try: with open(app_path, 'r') as f: content = f.read() # Look for common port patterns import re # Look for server_address = ('0.0.0.0', 8004) pattern port_match = re.search(r'server_address\s*=\s*\([^,]+,\s*(\d+)\)', content) if port_match: port = int(port_match.group(1)) print(f"DEBUG: Found port {port} in app.py (server_address tuple)") return port # Look for HTTPServer(('0.0.0.0', 8004), Handler) pattern port_match = re.search(r'HTTPServer\(\s*\([^,]+,\s*(\d+)\)', content) if port_match: port = int(port_match.group(1)) print(f"DEBUG: Found port {port} in app.py (HTTPServer tuple)") return port # Look for port = 8080 or similar port_match = re.search(r'port\s*=\s*(\d+)', content) if port_match: port = int(port_match.group(1)) print(f"DEBUG: Found port {port} in app.py (port =)") return port # Look for app.run(port=8080) or similar port_match = re.search(r'app\.run\([^)]*port\s*=\s*(\d+)', content) if port_match: port = int(port_match.group(1)) print(f"DEBUG: Found port {port} in app.py (app.run)") return port # Look for http.server.HTTPServer(('', 8080)) port_match = re.search(r'HTTPServer\([^)]*(\d+)\)', content) if port_match: port = int(port_match.group(1)) print(f"DEBUG: Found port {port} in app.py (HTTPServer)") return port # Look for socketserver.TCPServer(("", 8080)) port_match = re.search(r'TCPServer\([^)]*(\d+)\)', content) if port_match: port = int(port_match.group(1)) print(f"DEBUG: Found port {port} in app.py (TCPServer)") return port except Exception as e: print(f"Error parsing app.py for port: {e}") # Method 4: Try common ports for this app (as last resort) # Note: This method can be unreliable as it finds the first open port, # which might not be the correct app. Only use if other methods fail. common_ports = [8000, 8080, 8001, 8002, 8003, 8004, 8005, 8006, 8007, 8008, 8009, 8010] for port in common_ports: try: # Try to connect to the port to see if it's listening sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(0.1) result = sock.connect_ex(('127.0.0.1', port)) sock.close() if result == 0: # Port is open print(f"DEBUG: Found open port {port} via socket test (fallback method)") return port except Exception as e: print(f"DEBUG: Socket test failed for port {port}: {e}") continue print(f"DEBUG: No port found for PID {pid}") return None except Exception as e: print(f"Error detecting port for PID {pid}: {e}") return None def stop_app(self, pid): try: os.kill(int(pid), signal.SIGTERM) print(f"Stopped app with PID: {pid}") # Remove from port mapping if int(pid) in app_ports: del app_ports[int(pid)] except Exception as e: print(f"Error stopping app {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() # For performance, load available apps with lazy loading placeholders available_cli_html = '
Loading CLI apps...
' available_web_html = '
Loading web apps...
' 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 '' html = html.replace('', processes_html) html = html.replace('', installed_apps_html) html = html.replace('', available_cli_html) html = html.replace('', available_web_html) # Add lazy loading JavaScript (ES5 compatible) lazy_script = ''' ''' html = html.replace('', lazy_script + '') return html except Exception as e: print(f'Error loading HTML template: {e}') return '

Error loading HTML template

' def extract_app_name_from_command(self, cmd): """ Extract a user-friendly app name from a command string. Examples: - 'python3 /data/apps/myapp/app.py' -> 'myapp' - '/usr/local/bin/tool' -> 'tool' - 'python3 /data/apps/webserver/app.py' -> 'webserver' - 'myapp' -> 'myapp' """ try: # Remove common prefixes and extract the app name if '/data/apps/' in cmd: # Web app pattern: /data/apps/appname/app.py parts = cmd.split('/data/apps/') if len(parts) > 1: app_part = parts[1].split('/')[0] return app_part elif '/usr/local/bin/' in cmd: # CLI app pattern: /usr/local/bin/toolname parts = cmd.split('/usr/local/bin/') if len(parts) > 1: app_part = parts[1].split()[0] # Take first part before any args return app_part elif 'app.py' in cmd: # Generic app.py pattern if '/apps/' in cmd: parts = cmd.split('/apps/') if len(parts) > 1: app_part = parts[1].split('/')[0] return app_part elif '/data/' in cmd: parts = cmd.split('/data/') if len(parts) > 1: app_part = parts[1].split('/')[0] return app_part # Handle direct executable names (CLI apps) if not '/' in cmd and not 'python' in cmd.lower(): # This might be a direct executable name app_name = cmd.split()[0] # Take first part before any args return app_name # Fallback: try to extract from the end of the path if cmd.endswith('app.py'): # Remove 'app.py' and get the directory name clean_cmd = cmd.replace('app.py', '').rstrip('/') app_name = os.path.basename(clean_cmd) if app_name: return app_name # Last resort: return a cleaned version of the command return os.path.basename(cmd.split()[0]) if cmd.split() else 'Unknown App' except Exception as e: print(f"Error extracting app name from command '{cmd}': {e}") return 'Unknown App' def generate_processes_html(self): try: processes = self.get_python_processes() print(f"DEBUG: Found {len(processes)} processes") for pid, cmd in processes: print(f"DEBUG: Process {pid}: {cmd}") rows = '' for pid, cmd in processes: # Extract user-friendly app name instead of showing full command app_name = self.extract_app_name_from_command(cmd) # Check if we already have a port for this PID launch_link = '' pid_int = int(pid) if pid_int in app_ports: port = app_ports[pid_int] launch_link = f' - Launch UI' else: # Try to detect port for this process if it's a web app if 'app.py' in cmd or '/data/apps/' in cmd: detected_port = self.detect_app_port(pid_int, app_name) if detected_port: app_ports[pid_int] = detected_port launch_link = f' - Launch UI' print(f"Detected port {detected_port} for existing process {pid} ({app_name})") rows += f'{pid}{app_name}Stop{launch_link}' if not processes: rows = 'No running apps detected. Try refreshing the page.' return rows except Exception as e: print(f'Error generating processes HTML: {e}') return 'Error loading processes' 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() if available_zips is None: return '

Error loading Command Line Utilities.

' 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] apps_html += f'''
{app_name}
Click for details
''' 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() if available_zips is None: return '

Error loading 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] apps_html += f'''
{app_name}
Click for details
''' 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', ...]). """ html_content = get_cached_or_fetch(APKS_URL, 'android_apks') if html_content is None: return [] try: # 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 parsing Android APKs: {e}") return [] def fetch_available_zips(self): html_content = get_cached_or_fetch(AVAILABLE_APPS_URL, 'available_cli_zips') if html_content is None: return None try: 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 parsing available CLI zips: {e}") return None def fetch_web_apps(self): html_content = get_cached_or_fetch(WEB_APPS_URL, 'available_web_apps') if html_content is None: return None try: 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 parsing web apps: {e}") return None def get_python_processes(self): """ IMPROVED: Better process detection for QNX - looks for all relevant processes """ processes = [] try: # Use pidin ar to get all processes, then filter for relevant ones output = subprocess.check_output(['pidin', 'ar'], timeout=5, stderr=subprocess.DEVNULL).decode() lines = output.strip().split('\n') for line in lines: if line.strip(): parts = line.split() if len(parts) >= 2: pid = parts[0] cmd = ' '.join(parts[1:]) # Look for various types of running apps: # 1. Python processes (web apps, CLI tools) # 2. Processes in /data/apps/ directory # 3. Processes in /usr/local/bin/ directory # 4. Any process that looks like an app is_relevant = ( 'python' in cmd.lower() or '/data/apps/' in cmd or '/usr/local/bin/' in cmd or 'app.py' in cmd or (cmd.endswith('.py') and not 'taskapp.py' in cmd) # Exclude taskapp itself ) if is_relevant: # Clean up command path for display if '/data/' in cmd: index = cmd.find('/data/') + len('/data/') cmd = cmd[index:] elif '/usr/local/bin/' in cmd: index = cmd.find('/usr/local/bin/') + len('/usr/local/bin/') cmd = cmd[index:] processes.append((pid, cmd)) except subprocess.CalledProcessError: # Fallback: try with different pidin options try: output = subprocess.check_output(['pidin', '-f', '%a %A %n %p'], timeout=5, stderr=subprocess.DEVNULL).decode() lines = output.strip().split('\n') for line in lines: if line.strip(): parts = line.split() if len(parts) >= 2: pid = parts[0] cmd = ' '.join(parts[1:]) is_relevant = ( 'python' in cmd.lower() or '/data/apps/' in cmd or '/usr/local/bin/' in cmd or 'app.py' in cmd or (cmd.endswith('.py') and not 'taskapp.py' in cmd) ) if is_relevant: if '/data/' in cmd: index = cmd.find('/data/') + len('/data/') cmd = cmd[index:] elif '/usr/local/bin/' in cmd: index = cmd.find('/usr/local/bin/') + len('/usr/local/bin/') cmd = cmd[index:] processes.append((pid, cmd)) except Exception as e: print(f'Error getting processes (fallback): {e}') 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() print("Starting optimized Task Manager...") print("Performance improvements:") print("- Caching network requests (5min TTL)") print("- Timeout handling (10s timeout)") print("- Optimized pidin usage") print("- Lazy loading for available apps") with socketserver.TCPServer(("", PORT), TaskManagerHandler) as httpd: print(f"Serving on port {PORT}") httpd.serve_forever()