diff --git a/.gitignore b/.gitignore index d7747ed..41a5963 100644 --- a/.gitignore +++ b/.gitignore @@ -37,6 +37,8 @@ venv/ __pycache__/ *.pyc *.mmdb +pythonlib-dev/ +run-pw-dev.py # Closed source patches private @@ -47,3 +49,5 @@ closedsrc bundle *.pem .DS_store + +settings/ \ No newline at end of file diff --git a/pythonlib/camoufox/__main__.py b/pythonlib/camoufox/__main__.py index c014e6b..caff3ee 100644 --- a/pythonlib/camoufox/__main__.py +++ b/pythonlib/camoufox/__main__.py @@ -1,25 +1,168 @@ """ -CLI package manager for Camoufox. - -Adapted from https://github.com/daijro/hrequests/blob/main/hrequests/__main__.py +CLI package manager for Camoufox """ from importlib.metadata import PackageNotFoundError from importlib.metadata import version as pkg_version from os import environ -from typing import Optional +from typing import Any, List, Optional, Tuple -import click +import rich_click as click from .addons import DefaultAddons, maybe_download_addons -from .locale import ALLOW_GEOIP, download_mmdb, remove_mmdb -from .pkgman import INSTALL_DIR, CamoufoxFetcher, installed_verstr, rprint +from .geolocation import ( + ALLOW_GEOIP, + GEOIP_DIR, + _load_geoip_repos, + download_mmdb, + get_mmdb_path, + load_geoip_config, + remove_mmdb, + save_geoip_config, +) +from .multiversion import ( + REPO_CACHE_FILE, + InstalledVersion, + list_installed, + load_config, + load_repo_cache, + print_tree, + remove_version, + save_config, + save_repo_cache, + set_active, +) +from .pkgman import ( + INSTALL_DIR, + AvailableVersion, + CamoufoxFetcher, + RepoConfig, + installed_verstr, + list_available_versions, + rprint, +) -try: - from browserforge.download import download as update_browserforge -except ImportError: - # Account for other Browserforge versions - from browserforge.download import Download as update_browserforge + +def _inquirer_select( + choices: List[Tuple[str, Any]], + message: str, +) -> Optional[Any]: + """ + Generic inquirer selection. Returns selected value or None + """ + import inquirer + from inquirer.themes import GreenPassion + + try: + result = inquirer.prompt( + [inquirer.List('item', message=message, choices=choices, carousel=True)], + theme=GreenPassion(), + ) + return result['item'] if result else None + except KeyboardInterrupt: + return None + + +def _find_installed(specifier: str) -> Optional[InstalledVersion]: + """ + Find installed version by channel path, build, or full version string + """ + spec = specifier.lower() + installed = list_installed() + + for v in installed: + if any( + [ + v.channel_path.lower() == spec, + v.relative_path.lower() == spec, + v.version.build.lower() == spec, + v.version.full_string.lower() == spec, + ] + ): + return v + + parts = spec.split('/') + if len(parts) == 2: + repo, ctype = parts + is_pre = ctype == "prerelease" + for v in installed: + if v.repo_name == repo and v.is_prerelease == is_pre: + return v + + return None + + +def _get_geoip_source_name() -> str: + """ + Get the name of the active GeoIP source + """ + try: + return load_geoip_config().get('name', 'Default') + except Exception: + return "Default" + + +def _do_sync(spoof_os=None, spoof_arch=None) -> bool: + """ + Sync repos and browserforge data. Returns True on success + """ + rprint("Syncing repositories...", fg="yellow") + + cache = {'repos': [], 'spoof_os': spoof_os, 'spoof_arch': spoof_arch} + + for repo_config in RepoConfig.load_repos(): + rprint(f" {repo_config.name}...", fg="cyan", nl=False) + try: + versions = list_available_versions( + repo_config=repo_config, + include_prerelease=True, + spoof_os=spoof_os, + spoof_arch=spoof_arch, + ) + repo_data = { + 'name': repo_config.name, + 'repo': repo_config.repo, + 'versions': [ + { + 'version': v.version.version, + 'build': v.version.build, + 'url': v.url, + 'is_prerelease': v.is_prerelease, + 'asset_id': v.asset_id, + 'asset_size': v.asset_size, + 'asset_updated_at': v.asset_updated_at, + } + for v in versions + ], + } + cache['repos'].append(repo_data) + rprint(f" {len(versions)} versions", fg="green") + except Exception as e: + rprint(f" Error: {e}", fg="red") + + save_repo_cache(cache) + total = sum(len(r['versions']) for r in cache['repos']) + platform_str = f" ({spoof_os}/{spoof_arch})" if spoof_os else "" + rprint(f"\nSynced {total} versions from {len(cache['repos'])} repos{platform_str}.", fg="green") + + rprint("Updating browserforge...", fg="yellow") + try: + from browserforge.download import download as update_browserforge + except ImportError: + from browserforge.download import Download as update_browserforge + update_browserforge(headers=True, fingerprints=True) + + return True + + +def _ensure_synced() -> bool: + """ + Ensure repo cache exists. Returns True if synced, False if not + """ + if not REPO_CACHE_FILE.exists(): + rprint("No repo cache found. Run 'camoufox sync' first.", fg="red") + return False + return True class CamoufoxUpdate(CamoufoxFetcher): @@ -27,47 +170,35 @@ class CamoufoxUpdate(CamoufoxFetcher): Checks & updates Camoufox """ - def __init__(self) -> None: - """ - Initializes the CamoufoxUpdate class - """ - super().__init__() - self.current_verstr: Optional[str] + def __init__( + self, + repo_config: Optional[RepoConfig] = None, + selected_version: Optional[AvailableVersion] = None, + ) -> None: + super().__init__(repo_config=repo_config, selected_version=selected_version) try: self.current_verstr = installed_verstr() except FileNotFoundError: self.current_verstr = None def is_updated_needed(self) -> bool: - # Camoufox is not installed - if self.current_verstr is None: - return True - # If the installed version is not the latest version - if self.current_verstr != self.verstr: - return True - return False + return self.current_verstr is None or self.current_verstr != self.verstr - def update(self) -> None: - """ - Updates Camoufox if needed - """ - # Check if the version is the same as the latest available version - if not self.is_updated_needed(): + def update(self, replace: bool = False, i_know_what_im_doing: bool = False) -> None: + if not self.is_updated_needed() and not replace: rprint("Camoufox binaries up to date!", fg="green") rprint(f"Current version: v{self.current_verstr}", fg="green") return - # Download updated file - if self.current_verstr is not None: - # Display an updating message - rprint( - f"Updating Camoufox binaries from v{self.current_verstr} => v{self.verstr}", - fg="yellow", - ) - else: - rprint(f"Fetching Camoufox binaries v{self.verstr}...", fg="yellow") - # Install the new version - self.install() + if self.is_prerelease and not i_know_what_im_doing: + rprint(f"Warning: v{self.verstr} is a prerelease version!", fg="yellow") + if not click.confirm("Continue with prerelease installation?"): + rprint("Installation cancelled.", fg="red") + return + + action = "Installing" if self.current_verstr else "Fetching" + rprint(f"{action} Camoufox v{self.verstr}...", fg="yellow") + self.install(replace=replace) @click.group() @@ -75,35 +206,508 @@ def cli() -> None: pass -@cli.command(name='fetch') +@cli.command(name='sync') +@click.option('--spoof-os', type=click.Choice(['mac', 'win', 'lin']), help='Spoof OS') @click.option( - '--browserforge', is_flag=True, help='Update browserforge\'s header and fingerprint definitions' + '--spoof-arch', type=click.Choice(['x86_64', 'i686', 'arm64']), help='Spoof architecture' ) -def fetch(browserforge=False) -> None: +def sync(spoof_os, spoof_arch): """ - Fetch the latest version of Camoufox and optionally update Browserforge's database + Sync available versions from remote repositories. """ - CamoufoxUpdate().update() - # Fetch the GeoIP database - if ALLOW_GEOIP: - download_mmdb() + _do_sync(spoof_os=spoof_os, spoof_arch=spoof_arch) - # Download default addons - maybe_download_addons(list(DefaultAddons)) - if browserforge: - update_browserforge(headers=True, fingerprints=True) +@cli.command(name='fetch') +@click.argument('version', default=None, required=False) +def fetch(version): + """ + Install the active version, or a specific version. + + \b + Examples: + camoufox fetch # install active version + camoufox fetch official/135.0-beta.25 # install specific version + """ + _do_sync() + + cache = load_repo_cache() + config = load_config() + + if version: + if '/' not in version: + rprint("Format: /- (e.g., official/135.0-beta.25)", fg="red") + return + repo_name, ver_str = version.split('/', 1) + ver_str = ver_str.lstrip('v') + elif config.get('pinned'): + channel = config.get('channel', '') + repo_name = channel.split('/')[0] if '/' in channel else channel + ver_str = config['pinned'] + elif config.get('channel'): + channel = config['channel'] + if '/' in channel: + repo_name, ctype = channel.split('/', 1) + else: + repo_name, ctype = channel, 'stable' + for repo_data in cache.get('repos', []): + if repo_data['name'].lower() != repo_name.lower(): + continue + versions = repo_data.get('versions', []) + if ctype == 'prerelease': + candidates = [v for v in versions if v.get('is_prerelease')] + else: + candidates = [v for v in versions if not v.get('is_prerelease')] + if candidates: + ver_str = f"{candidates[0]['version']}-{candidates[0]['build']}" + break + else: + rprint(f"No versions found for channel '{channel}'.", fg="red") + return + else: + rprint("No channel set. Run 'camoufox select' first.", fg="red") + return + + for repo_data in cache.get('repos', []): + if repo_data['name'].lower() != repo_name.lower(): + continue + for v in repo_data['versions']: + if f"{v['version']}-{v['build']}" == ver_str: + from .pkgman import Version + + selected = AvailableVersion( + version=Version(v['build'], v['version']), + url=v['url'], + is_prerelease=v.get('is_prerelease', False), + ) + repo_config = RepoConfig.find_by_name(repo_data['name']) + try: + CamoufoxUpdate(repo_config=repo_config, selected_version=selected).update() + except Exception as e: + msg = str(e) + if '404' in msg or 'Not Found' in msg: + rprint("Release not found (404). Asset may have been removed.", fg="red") + rprint("Run 'camoufox sync' to refresh available versions.", fg="yellow") + else: + rprint(f"Error: {msg}", fg="red") + return + if ALLOW_GEOIP: + download_mmdb() + maybe_download_addons(list(DefaultAddons)) + return + + rprint(f"Version '{version or ver_str}' not found in cache.", fg="red") + + +def _set_channel(repo_name: str, channel_type: str): + """ + Set to track a channel (fetches latest on fetch) + """ + config = load_config() + config['channel'] = f"{repo_name}/{channel_type}" + config.pop('pinned', None) + save_config(config) + click.secho(f"Channel: {repo_name.lower()}/{channel_type}", fg="cyan", bold=True) + + # Check if latest for this channel is already installed + is_pre = channel_type == "prerelease" + cache = load_repo_cache() + for repo_data in cache.get('repos', []): + if repo_data['name'].lower() != repo_name.lower(): + continue + versions = repo_data.get('versions', []) + candidates = [v for v in versions if v.get('is_prerelease', False) == is_pre] + if candidates: + latest_build = candidates[0]['build'] + for inst in list_installed(): + if inst.version.build == latest_build and inst.repo_name == repo_name.lower(): + set_active(inst.relative_path) + click.secho(f"Latest is installed: {inst.channel_path}", fg="green") + return + break + + click.secho("Run 'camoufox fetch' to install latest.", fg="yellow") + + +def _set_pinned(repo_name: str, channel_type: str, ver_data: dict, inst): + """ + Pin to a specific version + """ + config = load_config() + config['channel'] = f"{repo_name}/{channel_type}" + config['pinned'] = f"{ver_data['version']}-{ver_data['build']}" + save_config(config) + ver_str = f"{ver_data['version']}-{ver_data['build']}" + display = f"{repo_name.lower()}/{channel_type}/{ver_str}" + if inst: + set_active(inst.relative_path) + click.secho(f"Pinned: {display} (installed)", fg="green") + else: + click.secho(f"Pinned: {display}", fg="cyan", bold=True) + click.secho("Run 'camoufox fetch' to install.", fg="yellow") + + +@cli.command(name='set') +@click.argument('specifier', required=False) +@click.option('--geoip', is_flag=True, help='Select GeoIP source instead') +def set_cmd(specifier, geoip): + """ + \b + Interactive selector for versions and settings + Or, pass a specifier to activate directly: + Pin version: camoufox set official/stable/134.0.2-beta.20 + Auto-update channel: camoufox set official/stable + """ + if geoip: + _select_geoip_source() + return + + if specifier: + parts = specifier.lower().split('/') + + # 2-part: set channel (e.g. official/stable) + if len(parts) == 2: + repo_name, ctype = parts + if ctype not in ('stable', 'prerelease'): + rprint(f"Unknown channel type '{ctype}'. Use 'stable' or 'prerelease'.", fg="red") + return + _set_channel(repo_name, ctype) + return + + # 3-part: pin version (e.g. official/stable/146.0.1-beta.25) + if len(parts) == 3: + repo_name, ctype, ver_str = parts + if ctype not in ('stable', 'prerelease'): + rprint(f"Unknown channel type '{ctype}'. Use 'stable' or 'prerelease'.", fg="red") + return + # Activate if already installed + target = _find_installed(specifier) + if target: + set_active(target.relative_path) + rprint(f"Pinned: {target.channel_path} (installed)", fg="green") + else: + click.secho(f"Pinned: {repo_name}/{ctype}/{ver_str}", fg="cyan", bold=True) + rprint("Run 'camoufox fetch' to install.", fg="yellow") + # Save pin config either way + config = load_config() + config['channel'] = f"{repo_name}/{ctype}" + config['pinned'] = ver_str + save_config(config) + return + + rprint(f"Invalid specifier '{specifier}'.", fg="red") + rprint("Use: repo/channel or repo/channel/version", fg="yellow") + return + + if not _ensure_synced(): + return + + import inquirer + from inquirer.themes import GreenPassion + + cache = load_repo_cache() + installed = {v.version.build: v for v in list_installed()} + + if not cache.get('repos'): + rprint("No versions in cache. Run 'camoufox sync' first.", fg="red") + return + + channels = [] + for repo_data in cache['repos']: + name = repo_data['name'] + versions = repo_data.get('versions', []) + stable = [v for v in versions if not v.get('is_prerelease')] + prereleases = [v for v in versions if v.get('is_prerelease')] + if stable: + channels.append((name, 'stable', stable[0])) + if prereleases: + channels.append((name, 'prerelease', prereleases[0])) + + config = load_config() + channel = config.get('channel', '') + pinned = config.get('pinned') + + if pinned: + click.secho(f"Pinned: {channel.lower()}/{pinned}", fg="cyan") + elif channel: + click.secho(f"Channel: {channel.lower()}", fg="cyan") + else: + click.secho("(no channel set)", fg="yellow") + click.echo() + + channel_versions = {} + for repo_data in cache['repos']: + name = repo_data['name'] + versions = repo_data.get('versions', []) + stable = [v for v in versions if not v.get('is_prerelease')] + prereleases = [v for v in versions if v.get('is_prerelease')] + if stable: + channel_versions[(name, 'stable')] = stable + if prereleases: + channel_versions[(name, 'prerelease')] = prereleases + + while True: + choices = [("Set channel", 'channel')] + for (name, ctype), versions in channel_versions.items(): + label = f"Pin version: {click.style(f'{name.lower()}/{ctype}', fg='cyan', bold=True)}" + choices.append((label, ('pin', name, ctype, versions))) + choices.append((click.style("Exit", fg="bright_black"), 'exit')) + + answer = inquirer.prompt( + [inquirer.List('action', message="Select", choices=choices, carousel=True)], + theme=GreenPassion(), + ) + if not answer: + return + + action = answer['action'] + + if action == 'exit': + return + + elif action == 'channel': + ch_choices = [] + for name, ctype, latest in channels: + ver_str = f"v{latest['version']}-{latest['build']}" + is_current = channel == f"{name}/{ctype}" + label = f"{name.lower()}/{ctype} (latest: {ver_str})" + if is_current: + label = click.style(label, fg="green", bold=True) + " (current)" + ch_choices.append((label, (name, ctype, latest))) + ch_choices.append((click.style("Back", fg="bright_black"), None)) + + ch_answer = inquirer.prompt( + [ + inquirer.List( + 'channel', message="Set channel", choices=ch_choices, carousel=True + ) + ], + theme=GreenPassion(), + ) + if not ch_answer or ch_answer['channel'] is None: + continue + + repo_name, ctype, _ = ch_answer['channel'] + _set_channel(repo_name, ctype) + return + + elif isinstance(action, tuple) and action[0] == 'pin': + _, rname, ctype, versions = action + + v_choices = [] + for i, v in enumerate(versions): + build = v['build'] + full_ver = f"{v['version']}-{build}" + inst = installed.get(build) + is_last = i == len(versions) - 1 + + prefix = "└── " if is_last else "├── " + + is_pinned = pinned == full_ver + if is_pinned and inst: + color = "green" + bold = True + suffix = " (pinned)" + elif is_pinned: + color = "cyan" + bold = True + suffix = " (pinned, not installed)" + elif inst: + color = None # white + bold = False + suffix = " (installed)" + else: + color = "bright_black" # grayed out + bold = False + suffix = "" + + ver_str = click.style(f"v{full_ver}", fg=color, bold=bold) + v_choices.append((f"{prefix}{ver_str}{suffix}", v)) + + v_choices.append((click.style("Back", fg="bright_black"), None)) + + default_val = versions[0] if versions else None + v_answer = inquirer.prompt( + [ + inquirer.List( + 'version', + message=f"Pin version ({rname.lower()}/{ctype})", + choices=v_choices, + default=default_val, + ) + ], + theme=GreenPassion(), + ) + if not v_answer or v_answer['version'] is None: + continue + + ver_data = v_answer['version'] + inst = installed.get(ver_data['build']) + _set_pinned(rname, ctype, ver_data, inst) + return + + +def _select_geoip_source(): + """ + Interactive selection of GeoIP source + """ + repos, _ = _load_geoip_repos() + if not repos: + rprint("No GeoIP sources configured.", fg="red") + return + + current = load_geoip_config().get('name', '') + choices = [(r['name'] + (" [active]" if r.get('name') == current else ""), r) for r in repos] + + selected = _inquirer_select(choices, "Select GeoIP source") + if not selected: + return + + save_geoip_config(selected) + rprint(f"GeoIP source: {selected['name']}", fg="green") + + +@cli.command(name='list') +@click.argument('mode', default='installed', type=click.Choice(['installed', 'all'])) +@click.option('--path', 'show_paths', is_flag=True, help='Show full paths') +def list_cmd(mode, show_paths): + """ + List Camoufox versions. + + \b + MODES: + installed Show installed versions (default) + all Show all available versions from synced repos + """ + if mode == 'all': + _list_all(show_paths) + else: + _list_installed(show_paths) + + +def _list_installed(show_paths: bool): + """ + List installed versions + """ + print_tree(show_paths=show_paths) + + click.echo() + click.secho("geoip/", fg="cyan", bold=True, nl=False) + if show_paths and GEOIP_DIR.exists(): + click.secho(f" -> {GEOIP_DIR}", fg="bright_black") + else: + click.echo() + + if GEOIP_DIR.exists(): + mmdb = get_mmdb_path() + if mmdb.exists(): + click.echo(f" └── {mmdb.name} ", nl=False) + click.secho(f"({_get_geoip_source_name()})", fg="green") + else: + rprint(" └── Not downloaded", fg="yellow") + else: + rprint(" └── Not configured", fg="yellow") + + +def _list_all(_show_paths: bool): + """ + List all available versions from synced repos + """ + if not _ensure_synced(): + return + + cache = load_repo_cache() + installed = {v.version.build: v for v in list_installed()} + + rprint("Available versions:\n", fg="yellow") + + for repo_data in cache.get('repos', []): + rname = repo_data['name'] + versions = repo_data.get('versions', []) + + click.secho(f"{rname}/", fg="cyan", bold=True) + + for i, v in enumerate(versions): + build = v['build'] + full_ver = f"{v['version']}-{build}" + inst = installed.get(build) + is_last = i == len(versions) - 1 + + prefix = "└── " if is_last else "├── " + color = "green" if inst and inst.is_active else None + + click.echo(f" {prefix}", nl=False) + click.secho(f"v{full_ver}", fg=color, bold=inst and inst.is_active, nl=False) + + if v.get('is_prerelease'): + click.secho(" (prerelease)", fg="yellow", nl=False) + else: + click.secho(" (stable)", fg="blue", nl=False) + + if inst: + if inst.is_active: + click.secho(" (installed, active)", fg="green", bold=True, nl=False) + else: + click.secho(" (installed)", fg="green", nl=False) + + click.echo() + + click.echo() @cli.command(name='remove') -def remove() -> None: +@click.argument('version_path', required=False) +@click.option('--all', 'remove_all', is_flag=True, help='Remove everything') +@click.option('--yes', '-y', is_flag=True, help='Skip confirmation prompts') +def remove(version_path, remove_all, yes): """ - Remove all downloaded files + \b + Remove installed version(s) + Or, pass a specifier to remove directly: + camoufox remove official/stable/134.0.2-beta.20 """ - if not CamoufoxUpdate().cleanup(): - rprint("Camoufox binaries not found!", fg="red") - # Remove the GeoIP database - remove_mmdb() + installed = list_installed() + has_geoip = GEOIP_DIR.exists() + + if remove_all or version_path == 'all': + if not installed and not has_geoip: + rprint("Nothing to remove.", fg="yellow") + return + if installed and (yes or click.confirm(f"Remove all {len(installed)} browser version(s)?")): + for v in installed: + remove_version(v.path) + rprint(f"Removed {len(installed)} version(s).", fg="green") + if has_geoip and (yes or click.confirm("Remove GeoIP database?")): + remove_mmdb() + return + + if not installed: + rprint("No browser versions installed.", fg="yellow") + if has_geoip and (yes or click.confirm("Remove GeoIP database?")): + remove_mmdb() + return + + if version_path: + target = _find_installed(version_path) + if not target: + rprint(f"Version '{version_path}' not found.", fg="red") + return + else: + choices = [ + (v.channel_path + (" [active]" if v.is_active else ""), v) + for v in installed + ] + target = _inquirer_select(choices, "Select version to remove") + if not target: + rprint("Cancelled.", fg="yellow") + return + + if yes or click.confirm(f"Remove {target.channel_path}?"): + remove_version(target.path) + rprint(f"Removed {target.channel_path}", fg="green") + + if has_geoip and (yes or click.confirm("Also remove GeoIP?", default=False)): + remove_mmdb() @cli.command(name='test') @@ -119,11 +723,11 @@ def test(url: Optional[str] = None, executable_path: Optional[str] = None) -> No page = browser.new_page() if url: page.goto(url) - page.pause() # Open the Playwright inspector + page.pause() @cli.command(name='server') -def server() -> None: +def server(): """ Launch a Playwright server """ @@ -132,41 +736,104 @@ def server() -> None: launch_server() -@cli.command(name='path') -def path() -> None: +@cli.command(name='gui') +@click.option('--debug', is_flag=True, help="Enable debug options in the GUI.") +def gui(debug): """ - Display the path to the Camoufox executable + Launch the Camouman GUI (requires PySide6) """ - rprint(INSTALL_DIR, fg="green") + try: + from .gui import main + + main(debug=debug) + except ImportError: + rprint("GUI requires PySide6. Install with: pip install 'camoufox\\[gui]'", fg="red") @cli.command(name='version') -def version() -> None: +def version(): """ - Display the current version + Display version info """ - # python package version try: - rprint(f"Pip package:\tv{pkg_version('camoufox')}", fg="green") + rprint(f"Pip package:\t\tv{pkg_version('camoufox')}", fg="green") except PackageNotFoundError: - rprint("Pip package:\tNot installed!", fg="red") + rprint("Pip package:\t\tNot installed!", fg="red") - updater = CamoufoxUpdate() - bin_ver = updater.current_verstr + active_v = None + for v in list_installed(): + if v.is_active: + active_v = v + break - # If binaries are not downloaded - if not bin_ver: - rprint("Camoufox:\tNot downloaded!", fg="red") + if not active_v: + rprint("Active:\t\t\tNot installed!", fg="red") return - # Print the base version - rprint(f"Camoufox:\tv{bin_ver} ", fg="green", nl=False) - # Check for Camoufox updates - if updater.is_updated_needed(): - rprint(f"(Latest supported: v{updater.verstr})", fg="red") + config = load_config() + pinned = config.get('pinned') + channel = config.get('channel', '') + + # Channel + if pinned: + rprint(f"Channel:\t\t{channel.lower()} (Version pinned)", fg="cyan") + elif channel: + rprint(f"Channel:\t\t{channel.lower()} (Following updates)", fg="cyan") + + # Version with update status + rprint(f"Version:\t\tv{active_v.version.full_string} ", fg="green", nl=False) + if pinned: + click.echo() + elif channel: + repo_name, ctype = channel.split('/', 1) if '/' in channel else ('', '') + is_pre = ctype == "prerelease" + latest_build = None + cache = load_repo_cache() + for repo_data in cache.get('repos', []): + if repo_data['name'].lower() != repo_name.lower(): + continue + candidates = [v for v in repo_data.get('versions', []) if v.get('is_prerelease', False) == is_pre] + if candidates: + latest_build = candidates[0]['build'] + break + if latest_build and active_v.version.build == latest_build: + rprint("(Up to date!)", fg="yellow") + elif latest_build: + rprint(f"(Latest: {latest_build})", fg="red") + else: + click.echo() else: - rprint("(Up to date!)", fg="yellow") + click.echo() + + if REPO_CACHE_FILE.exists(): + from datetime import datetime, timezone + + mtime = REPO_CACHE_FILE.stat().st_mtime + dt = datetime.fromtimestamp(mtime, tz=timezone.utc).astimezone() + rprint(f"Last repos sync:\t{dt.strftime('%Y-%m-%d %H:%M')}", fg="bright_black") + + +@cli.command(name='active') +def active_cmd(): + """ + Print the current active version + """ + installed = list_installed() + for v in installed: + if v.is_active: + click.echo(v.channel_path) + return + rprint("No active version.", fg="yellow") + + +@cli.command(name='path') +def path_cmd(): + """ + Print the install directory path + """ + click.echo(INSTALL_DIR) if __name__ == '__main__': cli() + cli() diff --git a/pythonlib/camoufox/addons.py b/pythonlib/camoufox/addons.py index dfff0b6..191707d 100644 --- a/pythonlib/camoufox/addons.py +++ b/pythonlib/camoufox/addons.py @@ -4,7 +4,10 @@ from multiprocessing import Lock from typing import List, Optional from .exceptions import InvalidAddonPath -from .pkgman import get_path, unzip, webdl +from .pkgman import INSTALL_DIR, unzip, webdl + +# Addons are stored in a shared folder, not per-browser version +ADDONS_DIR = INSTALL_DIR / "addons" class DefaultAddons(Enum): @@ -55,9 +58,9 @@ def download_and_extract(url: str, extract_path: str, name: str) -> None: def get_addon_path(addon_name: str) -> str: """ - Returns a path to the addon + Returns a path to the addon in the shared addons folder. """ - return get_path(os.path.join("addons", addon_name)) + return str(ADDONS_DIR / addon_name) def maybe_download_addons( diff --git a/pythonlib/camoufox/geolocation.py b/pythonlib/camoufox/geolocation.py new file mode 100644 index 0000000..e7ca481 --- /dev/null +++ b/pythonlib/camoufox/geolocation.py @@ -0,0 +1,261 @@ +""" +Helpers to fetch geolocation, timezone, and locale data given an IP +""" + +import shutil +import tempfile +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple, cast + +from platformdirs import user_cache_dir +from yaml import CDumper, CLoader +from yaml import dump as yaml_dump +from yaml import load as yaml_load + +from .exceptions import NotInstalledGeoIPExtra, UnknownIPLocation +from .ip import validate_ip +from .locale import SELECTOR, Geolocation +from .pkgman import LOCAL_DATA, rprint, unzip, webdl + +try: + import maxminddb # type: ignore +except ImportError: + ALLOW_GEOIP = False +else: + ALLOW_GEOIP = True + + +GEOIP_DIR = Path(user_cache_dir("camoufox")) / "geoip" +MMDB_DIR = GEOIP_DIR / "mmdb" +GEOIP_CONFIG = GEOIP_DIR / "config.yml" + + +def _find_in(data: Dict, key: str) -> Any: + """ + Resolve a dotted path in a nested dict + """ + for part in key.split('.'): + if not isinstance(data, dict): + return None + data = data.get(part) + if data is None: + return None + return data + + +def _load_geoip_repos() -> Tuple[List[Dict], str]: + """ + Load GeoIP repos and default name from repos.yml + """ + with open(LOCAL_DATA / 'repos.yml', 'r') as f: + data = yaml_load(f, Loader=CLoader) + geoip_repos = data.get('geoip', []) + default_name = data.get('default', {}).get('geoip', 'GeoLite2') + return geoip_repos, default_name + + +def _get_geoip_config_by_name(name: Optional[str] = None) -> Dict: + """ + Get GeoIP config by name from repos.yml. If None, uses default + """ + repos, default_name = _load_geoip_repos() + target_name = name or default_name + + def _validate_repo(repo: Dict) -> Dict: + if 'urls' not in repo: + raise ValueError(f"GeoIP repo '{repo.get('name')}' missing required urls") + if 'paths' not in repo: + raise ValueError(f"GeoIP repo '{repo.get('name')}' missing required paths") + return repo + + for repo in repos: + if repo.get('name', '').lower() == target_name.lower(): + return _validate_repo(repo) + + if name: + available = [r.get('name', 'Unknown') for r in repos] + raise ValueError(f"GeoIP database '{name}' not found. Available: {available}") + + if repos: + return _validate_repo(repos[0]) + raise ValueError("No GeoIP repos configured in repos.yml") + + +def load_geoip_config() -> Dict: + """ + Load active GeoIP config from disk, falling back to repos.yml default + """ + if GEOIP_CONFIG.exists(): + with open(GEOIP_CONFIG, 'r') as f: + saved = yaml_load(f, Loader=CLoader) + try: + return _get_geoip_config_by_name(saved.get('name')) + except (ValueError, KeyError): + return saved + return _get_geoip_config_by_name(None) + + +def save_geoip_config(config: Dict) -> None: + """ + Save active GeoIP source name to disk + """ + GEOIP_DIR.mkdir(parents=True, exist_ok=True) + with open(GEOIP_CONFIG, 'w') as f: + yaml_dump({'name': config['name']}, f, Dumper=CDumper, default_flow_style=False) + + +def get_mmdb_path(ip_version: str = 'ipv4', config: Optional[Dict] = None) -> Path: + """ + Get path to the mmdb file for the specified IP version + """ + if config is None: + config = load_geoip_config() + name = config.get('name', 'geolite2').lower() + urls = config.get('urls', {}) + if 'combined' in urls: + return MMDB_DIR / f"{name}-combined.mmdb" + return MMDB_DIR / f"{name}-{ip_version}.mmdb" + + +def geoip_allowed() -> None: + """ + Checks if the geoip2 module is available + """ + if not ALLOW_GEOIP: + raise NotInstalledGeoIPExtra( + 'Please install the geoip extra to use this feature: pip install camoufox[geoip]' + ) + + +def download_mmdb( + source: Optional[str] = None, + progress_callback: Optional[callable] = None, +) -> None: + """ + Downloads the GeoIP database(s) to geoip/mmdb/ + """ + geoip_allowed() + + config = _get_geoip_config_by_name(source) if source else load_geoip_config() + urls = config['urls'] + name = config['name'].lower() + + MMDB_DIR.mkdir(parents=True, exist_ok=True) + + extract = config.get('extract', False) + dl_desc = f'Downloading {config["name"]}' + ex_desc = f'Extracting {config["name"]}' + max_len = max(len(dl_desc), len(ex_desc)) + dl_desc = dl_desc.ljust(max_len) + ex_desc = ex_desc.ljust(max_len) + + for ip_ver, url_list in urls.items(): + mmdb_path = MMDB_DIR / f"{name}-{ip_ver}.mmdb" + + if isinstance(url_list, str): + url_list = [url_list] + + last_error = None + for url in url_list: + try: + with tempfile.NamedTemporaryFile(suffix='.zip' if extract else '.mmdb') as tmp: + webdl( + url, + desc=dl_desc, + buffer=tmp, + bar=progress_callback is None, + progress_callback=progress_callback, + ) + if extract: + with tempfile.TemporaryDirectory() as tmpdir: + unzip(tmp, tmpdir, desc=ex_desc, bar=progress_callback is None) + mmdb_files = list(Path(tmpdir).rglob('*.mmdb')) + if not mmdb_files: + raise ValueError("No .mmdb file found in archive") + shutil.move(str(mmdb_files[0]), str(mmdb_path)) + else: + tmp.seek(0) + with open(mmdb_path, 'wb') as dst: + shutil.copyfileobj(tmp, dst) + break + except Exception as e: + last_error = e + continue + else: + raise last_error or Exception(f"Failed to download {ip_ver}") + + save_geoip_config(config) + + +def remove_mmdb() -> None: + """ + Removes the GeoIP database and config + """ + if not GEOIP_DIR.exists(): + rprint("GeoIP database not found.") + return + + shutil.rmtree(GEOIP_DIR) + rprint("GeoIP database removed.") + + +def needs_update(config: Optional[Dict] = None) -> bool: + """ + Check if the GeoIP database needs an update (older than 30 days) + """ + from datetime import datetime, timedelta + + if config is None: + config = load_geoip_config() + + update_days = 30 + + ipv4_path = get_mmdb_path('ipv4', config) + if not ipv4_path.exists(): + return True + + mtime = datetime.fromtimestamp(ipv4_path.stat().st_mtime) + age = datetime.now() - mtime + return age > timedelta(days=update_days) + + +def get_geolocation(ip: str, geoip_db: Optional[str] = None) -> Geolocation: + """ + Gets the geolocation for an IP address + """ + import maxminddb + + validate_ip(ip) + ip_version = 'ipv6' if ':' in ip else 'ipv4' + mmdb_path = get_mmdb_path(ip_version) + + if not mmdb_path.exists() or needs_update(): + download_mmdb() + mmdb_path = get_mmdb_path(ip_version) + + if geoip_db: + config = _get_geoip_config_by_name(geoip_db) + else: + config = load_geoip_config() + paths = config['paths'] + + with maxminddb.open_database(str(mmdb_path)) as reader: + resp = cast(Dict[str, Any], reader.get(ip)) + + if not resp: + raise UnknownIPLocation(f"IP not found in database: {ip}") + + iso_code = _find_in(resp, paths['iso_code']) + longitude = _find_in(resp, paths['longitude']) + latitude = _find_in(resp, paths['latitude']) + timezone = _find_in(resp, paths['timezone']) + + iso_code = str(iso_code).upper() + locale = SELECTOR.from_region(iso_code) + + return Geolocation( + locale=locale, + longitude=float(longitude), + latitude=float(latitude), + timezone=str(timezone), + ) diff --git a/pythonlib/camoufox/gui/__init__.py b/pythonlib/camoufox/gui/__init__.py new file mode 100644 index 0000000..4fa621d --- /dev/null +++ b/pythonlib/camoufox/gui/__init__.py @@ -0,0 +1,16 @@ +""" +Camouman - QML-based GUI for managing Camoufox versions +""" + +import os + +# Force Basic style before PySide6 import to avoid Breeze conflicts +os.environ["QT_QUICK_CONTROLS_STYLE"] = "Basic" +os.environ["QT_QPA_PLATFORMTHEME"] = "" +os.environ["QT_STYLE_OVERRIDE"] = "" +os.environ["KDE_FULL_SESSION"] = "" +os.environ["DESKTOP_SESSION"] = "" + +from .backend import main + +__all__ = ['main'] diff --git a/pythonlib/camoufox/gui/assets/SegUIVar.ttf b/pythonlib/camoufox/gui/assets/SegUIVar.ttf new file mode 100644 index 0000000..dd8c264 Binary files /dev/null and b/pythonlib/camoufox/gui/assets/SegUIVar.ttf differ diff --git a/pythonlib/camoufox/gui/assets/icon.ico b/pythonlib/camoufox/gui/assets/icon.ico new file mode 100644 index 0000000..ffdac08 Binary files /dev/null and b/pythonlib/camoufox/gui/assets/icon.ico differ diff --git a/pythonlib/camoufox/gui/assets/segmdl2.ttf b/pythonlib/camoufox/gui/assets/segmdl2.ttf new file mode 100755 index 0000000..066f108 Binary files /dev/null and b/pythonlib/camoufox/gui/assets/segmdl2.ttf differ diff --git a/pythonlib/camoufox/gui/backend.py b/pythonlib/camoufox/gui/backend.py new file mode 100644 index 0000000..4017232 --- /dev/null +++ b/pythonlib/camoufox/gui/backend.py @@ -0,0 +1,968 @@ +import os +import sys +import tempfile +from enum import IntEnum +from pathlib import Path + +from PySide6.QtCore import ( + Property, + QAbstractListModel, + QModelIndex, + QObject, + Qt, + QThread, + QUrl, + Signal, + Slot, +) +from PySide6.QtGui import QGuiApplication, QIcon +from PySide6.QtQml import QQmlApplicationEngine +from PySide6.QtQuickControls2 import QQuickStyle + +from ..multiversion import ( + BROWSERS_DIR, + get_cached_versions, + get_repo_name, + list_installed, + load_config, + load_repo_cache, + remove_version, + save_config, + save_repo_cache, + set_active, +) +from ..pkgman import RepoConfig, unzip, webdl + +# Workers + +class Worker(QThread): + progress = Signal(float) + status = Signal(str) + done = Signal(bool, str) + + def _progress(self, downloaded, total): + if total > 0: + self.progress.emit(downloaded / total) + + +class DownloadWorker(Worker): + def __init__(self, repo_config, version): + super().__init__() + self.repo_config = repo_config + self.version = version + + def run(self): + try: + import shlex + + import orjson + + self.status.emit("Downloading...") + repo_name = get_repo_name(self.repo_config.repo) + folder = f"{self.version.version.version}-{self.version.version.build}" + path = BROWSERS_DIR / repo_name / folder + path.mkdir(parents=True, exist_ok=True) + + with tempfile.NamedTemporaryFile() as f: + webdl(self.version.url, buffer=f, bar=False, progress_callback=self._progress) + self.status.emit("Extracting...") + self.progress.emit(-1) + unzip(f, str(path), bar=False) + + (path / 'version.json').write_bytes(orjson.dumps(self.version.to_metadata())) + + if sys.platform != 'win32': + os.system(f'chmod -R 755 {shlex.quote(str(path))}') + + set_active(f"browsers/{repo_name}/{folder}") + self.done.emit(True, f"Installed v{self.version.version.full_string}") + except Exception as e: + msg = str(e) + if '404' in msg or 'Not Found' in msg: + self.done.emit(False, "Release not found (404). Please resync.") + else: + self.done.emit(False, msg) + + +class SyncWorker(Worker): + def __init__(self, spoof_os=None, spoof_arch=None): + super().__init__() + self.spoof_os = spoof_os + self.spoof_arch = spoof_arch + + def run(self): + try: + from datetime import datetime + + from ..pkgman import list_available_versions + + self.status.emit("Syncing...") + cache = {'repos': []} + + for rc in RepoConfig.load_repos(): + self.status.emit(f"Syncing {rc.name}...") + versions = list_available_versions( + rc, + include_prerelease=True, + spoof_os=self.spoof_os, + spoof_arch=self.spoof_arch, + ) + cache['repos'].append({ + 'name': rc.name, + 'repo': rc.repo, + 'versions': [ + { + 'version': v.version.version, + 'build': v.version.build, + 'url': v.url, + 'is_prerelease': v.is_prerelease, + 'asset_id': v.asset_id, + 'asset_size': v.asset_size, + 'asset_updated_at': v.asset_updated_at, + } + for v in versions + ], + }) + + cache['spoof_os'] = self.spoof_os + cache['spoof_arch'] = self.spoof_arch + cache['sync_time'] = datetime.now().strftime('%-m/%-d/%Y %-I:%M %p') + save_repo_cache(cache) + total = sum(len(r['versions']) for r in cache['repos']) + self.done.emit(True, f"Synced {total} versions") + except Exception as e: + self.done.emit(False, str(e)) + + +class GeoIPWorker(Worker): + def __init__(self, source): + super().__init__() + self.source = source + + def run(self): + try: + from ..geolocation import download_mmdb + + download_mmdb(source=self.source, progress_callback=self._progress) + self.done.emit(True, f"Installed: {self.source}") + except Exception as e: + self.done.emit(False, str(e)) + + +# Models + +class Roles(IntEnum): + Display = Qt.ItemDataRole.UserRole + 1 + Build = Qt.ItemDataRole.UserRole + 2 + IsHeader = Qt.ItemDataRole.UserRole + 3 + IsPrerelease = Qt.ItemDataRole.UserRole + 4 + IsActive = Qt.ItemDataRole.UserRole + 5 + IsInstalled = Qt.ItemDataRole.UserRole + 6 + Section = Qt.ItemDataRole.UserRole + 7 + Expanded = Qt.ItemDataRole.UserRole + 8 + IsPinned = Qt.ItemDataRole.UserRole + 9 + + +_ROLE_ATTRS = { + Roles.Display: 'display', Roles.Build: 'build', + Roles.IsHeader: 'is_header', Roles.IsPrerelease: 'is_prerelease', + Roles.IsActive: 'is_active', Roles.IsInstalled: 'is_installed', + Roles.Section: 'section', Roles.Expanded: 'expanded', + Roles.IsPinned: 'is_pinned', +} + +_BOOL_ROLES = { + Roles.IsHeader, Roles.IsPrerelease, Roles.IsActive, + Roles.IsInstalled, Roles.Expanded, Roles.IsPinned, +} + +_ROLE_NAMES = { + Roles.Display: b"display", Roles.Build: b"build", + Roles.IsHeader: b"isHeader", Roles.IsPrerelease: b"isPrerelease", + Roles.IsActive: b"isActive", Roles.IsInstalled: b"isInstalled", + Roles.Section: b"section", Roles.Expanded: b"expanded", + Roles.IsPinned: b"isPinned", +} + + +class VersionItem: + __slots__ = ( + 'display', 'build', 'is_header', 'is_prerelease', 'is_active', + 'is_pinned', 'is_installed', 'section', 'expanded', + 'version_data', 'installed_data', + ) + + def __init__( + self, display, build="", is_header=False, is_prerelease=False, + is_active=False, is_pinned=False, is_installed=False, + section="", expanded=True, version_data=None, installed_data=None, + ): + self.display = display + self.build = build + self.is_header = is_header + self.is_prerelease = is_prerelease + self.is_active = is_active + self.is_pinned = is_pinned + self.is_installed = is_installed + self.section = section + self.expanded = expanded + self.version_data = version_data + self.installed_data = installed_data + + +class VersionModel(QAbstractListModel): + def __init__(self, parent=None): + super().__init__(parent) + self._items = [] + + def rowCount(self, parent=QModelIndex()): + return len(self._items) + + def data(self, index, role=Qt.ItemDataRole.DisplayRole): + if not index.isValid() or index.row() >= len(self._items): + return False if role in _BOOL_ROLES else "" + attr = _ROLE_ATTRS.get(role) + return getattr(self._items[index.row()], attr, "") if attr else "" + + def roleNames(self): + return _ROLE_NAMES + + def set_items(self, items): + self.beginResetModel() + self._items = items + self.endResetModel() + + def get(self, index): + return self._items[index] if 0 <= index < len(self._items) else None + + +OS_OPTIONS = ["(auto)", "mac", "win", "lin"] +ARCH_OPTIONS = ["(auto)", "x86_64", "i686", "arm64"] + + +# Backend + +class Backend(QObject): + reposChanged = Signal() + busyChanged = Signal() + progressChanged = Signal() + statusChanged = Signal() + selectionChanged = Signal() + geoipChanged = Signal() + infoChanged = Signal() + debugChanged = Signal() + currentRepoChanged = Signal() + installPrompt = Signal(int, str, str) + + def __init__(self): + super().__init__() + self._repo_configs = list(RepoConfig.load_repos()) + self._repos = [r.name for r in self._repo_configs] + self._version_model = VersionModel(self) + self._current_repo = self._repo_configs[0] if self._repo_configs else None + + self._selected = -1 + self._busy = False + self._progress = -1.0 + self._status_text = "" + self._status_color = "#888" + self._installed_only = False + self._sections = {"stable": True, "prerelease": True} + self._worker = None + self._channel_data = None + + self._geoip_sources = [] + self._geoip_names = [] + self._geoip_installed = "" + self._geoip_downloaded = [] + self._geoip_path = "" + self._geoip_size = "" + self._geoip_mtime = "" + self._geoip_busy = False + self._geoip_progress = -1.0 + self._lookup_result = "" + self._lookup_ok = True + + self._spoof_os_idx = 0 + self._spoof_arch_idx = 0 + self._load_spoof_from_cache() + self._load_geoip() + + # Properties + + @Property(list, notify=reposChanged) + def repos(self): + return self._repos + + @Property(QObject, constant=True) + def versionModel(self): + return self._version_model + + @Property(bool, notify=busyChanged) + def busy(self): + return self._busy + + @Property(float, notify=progressChanged) + def progress(self): + return self._progress + + @Property(str, notify=statusChanged) + def statusText(self): + return self._status_text + + @Property(str, notify=statusChanged) + def statusColor(self): + return self._status_color + + @Property(bool, notify=selectionChanged) + def canInstall(self): + item = self._version_model.get(self._selected) + return item and not item.is_header and not item.is_installed + + @Property(bool, notify=selectionChanged) + def canUninstall(self): + item = self._version_model.get(self._selected) + return item and not item.is_header and item.is_installed + + @Property(str, notify=selectionChanged) + def selectedVersion(self): + item = self._version_model.get(self._selected) + return item.display if item else "" + + @Property(bool, notify=selectionChanged) + def selectedIsPrerelease(self): + item = self._version_model.get(self._selected) + return item.is_prerelease if item else False + + @Property(list, notify=geoipChanged) + def geoipSources(self): + return self._geoip_sources + + @Property(str, notify=geoipChanged) + def geoipInstalled(self): + return self._geoip_installed + + @Property(list, notify=geoipChanged) + def geoipDownloaded(self): + return self._geoip_downloaded + + @Property(str, notify=geoipChanged) + def geoipPath(self): + return self._geoip_path + + @Property(str, notify=geoipChanged) + def geoipSize(self): + return self._geoip_size + + @Property(str, notify=geoipChanged) + def geoipMtime(self): + return self._geoip_mtime + + @Property(bool, notify=geoipChanged) + def geoipBusy(self): + return self._geoip_busy + + @Property(float, notify=geoipChanged) + def geoipProgress(self): + return self._geoip_progress + + @Property(str, notify=geoipChanged) + def lookupResult(self): + return self._lookup_result + + @Property(bool, notify=geoipChanged) + def lookupSuccess(self): + return self._lookup_ok + + @Property(str, notify=infoChanged) + def activeBrowserText(self): + cfg = load_config() + if cfg.get('pinned'): + return f"v{cfg['pinned']}" + channel = cfg.get('channel', '') + if channel: + _, keys, latest = self._build_channels() + try: + return latest[keys.index(channel)] or channel + except ValueError: + return channel + return "(no channel set)" + + @Property(str, notify=infoChanged) + def activeBrowserColor(self): + cfg = load_config() + return "#26a69a" if cfg.get('channel') or cfg.get('pinned') else "#888888" + + @Property(str, notify=infoChanged) + def followedChannel(self): + return load_config().get('channel', '') + + @Property(list, notify=infoChanged) + def channels(self): + return self._build_channels()[0] + + @Property(list, notify=infoChanged) + def channelKeys(self): + return self._build_channels()[1] + + @Property(list, notify=infoChanged) + def channelLatest(self): + return self._build_channels()[2] + + @Property(str, notify=infoChanged) + def activeLabel(self): + cfg = load_config() + channel = cfg.get('channel', '') + pinned = cfg.get('pinned') + if channel: + channels, keys, _ = self._build_channels() + try: + return f"Active: following {channels[keys.index(channel)]}" + except ValueError: + return f"Active: following {channel}" + if pinned: + return f"Active: pinned to v{pinned}" + return "" + + @Property(str, notify=infoChanged) + def libraryVersion(self): + return self._pkg_version('camoufox') + + @Property(str, notify=infoChanged) + def playwrightVersion(self): + return self._pkg_version('playwright') + + @Property(str, notify=infoChanged) + def browserforgeVersion(self): + return self._pkg_version('browserforge') + + @Property(str, notify=infoChanged) + def lastSyncTime(self): + cache = load_repo_cache() + raw = cache.get('sync_time', 'Never') if cache else 'Never' + if raw == 'Never': + return raw + try: + from datetime import datetime + + dt = datetime.strptime(raw, '%Y-%m-%d %H:%M') + return dt.strftime('%-m/%-d/%Y %-I:%M %p') + except ValueError: + return raw + + @Property(str, notify=infoChanged) + def reposInfo(self): + cache = load_repo_cache() + if not cache: + return "Run sync" + repos = cache.get('repos', []) + total = sum(len(r.get('versions', [])) for r in repos) + return f"{len(repos)} repos, {total} versions" + + @Property(list, constant=True) + def spoofOsOptions(self): + return OS_OPTIONS + + @Property(list, constant=True) + def spoofArchOptions(self): + return ARCH_OPTIONS + + @Property(int, notify=debugChanged) + def spoofOsIndex(self): + return self._spoof_os_idx + + @Property(int, notify=debugChanged) + def spoofArchIndex(self): + return self._spoof_arch_idx + + @Property(int, notify=currentRepoChanged) + def currentRepoIndex(self): + if self._current_repo: + for i, rc in enumerate(self._repo_configs): + if rc.name == self._current_repo.name: + return i + return 0 + + # Internal + + @staticmethod + def _pkg_version(pkg): + try: + from importlib.metadata import version + + return version(pkg) + except Exception: + return "?" + + def _build_channels(self): + """ + Build channel names, keys, and latest version strings (cached) + """ + if self._channel_data is not None: + return self._channel_data + + cache = load_repo_cache() + channels, keys, latest = [], [], [] + + for rc in self._repo_configs: + repo_versions = [] + if cache: + for repo in cache.get('repos', []): + if repo['name'].lower() == rc.name.lower(): + repo_versions = repo.get('versions', []) + break + + stable = [v for v in repo_versions if not v.get('is_prerelease')] + prereleases = [v for v in repo_versions if v.get('is_prerelease')] + + channels.append(rc.name) + keys.append(rc.name) + latest.append(f"v{stable[0]['version']}-{stable[0]['build']}" if stable else "") + + if prereleases: + channels.append(f"{rc.name} (Prerelease)") + keys.append(f"{rc.name}/prerelease") + latest.append(f"v{prereleases[0]['version']}-{prereleases[0]['build']}") + + self._channel_data = (channels, keys, latest) + return self._channel_data + + # Slots + + @Slot(int) + def selectRepo(self, index): + if 0 <= index < len(self._repo_configs): + self._current_repo = self._repo_configs[index] + self._refresh() + + @Slot(bool) + def setInstalledOnly(self, value): + self._installed_only = value + self._refresh() + + @Slot(str) + def toggleSection(self, section): + self._sections[section] = not self._sections.get(section, True) + self._refresh() + + @Slot(int) + def selectVersion(self, index): + self._selected = index + self.selectionChanged.emit() + + @Slot(int) + def setActive(self, index): + item = self._version_model.get(index) + if not item or item.is_header: + return + + cfg = load_config() + cfg.pop('channel', None) + cfg['pinned'] = f"{item.version_data.version.version}-{item.version_data.version.build}" + cfg.update({ + 'active_repo': self._current_repo.name, + 'active_build': item.version_data.version.build, + 'active_version': item.version_data.version.version, + }) + save_config(cfg) + + if item.installed_data: + set_active(item.installed_data.relative_path) + + self._refresh() + self.infoChanged.emit() + + @Slot(int) + def setFollowedChannel(self, index): + _, keys, _ = self._build_channels() + if not (0 <= index < len(keys)): + return + + key = keys[index] + cfg = load_config() + is_follow = cfg.get('channel', '') != key + + if not is_follow: + cfg.pop('channel', None) + else: + cfg['channel'] = key + cfg.pop('pinned', None) + + repo_name, ctype = (key.split('/', 1) + ['stable'])[:2] + cache = load_repo_cache() + if cache: + for repo in cache.get('repos', []): + if repo['name'].lower() == repo_name.lower(): + is_pre = ctype == 'prerelease' + candidates = [ + v for v in repo.get('versions', []) + if v.get('is_prerelease') == is_pre + ] + if candidates: + cfg['active_build'] = candidates[0]['build'] + cfg['active_version'] = candidates[0]['version'] + cfg['active_repo'] = repo_name + break + + for rc in self._repo_configs: + if rc.name.lower() == repo_name.lower(): + self._current_repo = rc + self.currentRepoChanged.emit() + break + + save_config(cfg) + self._refresh() + self.infoChanged.emit() + + if is_follow: + is_pre = '/' in key and key.split('/')[1] == 'prerelease' + for idx, item in enumerate(self._version_model._items): + if item.is_header: + continue + if item.is_prerelease == is_pre: + if not item.is_installed: + self._selected = idx + self.selectionChanged.emit() + self.installPrompt.emit(idx, item.display, item.build) + break + + @Slot() + def installSelected(self): + item = self._version_model.get(self._selected) + if item and not item.is_header and not item.is_installed: + self._run_worker(DownloadWorker(self._current_repo, item.version_data), self._on_done) + + @Slot() + def uninstallSelected(self): + item = self._version_model.get(self._selected) + if not item or not item.is_installed or not item.installed_data: + return + try: + remove_version(item.installed_data.path) + self._set_status(f"Uninstalled {item.display}", "#2ecc71") + self._refresh() + self.infoChanged.emit() + except Exception as e: + self._set_status(str(e), "#e74c3c") + + @Slot() + def refresh(self): + self._refresh() + self._set_status("Refreshed", "#888") + + @Slot() + def sync(self): + spoof_os = OS_OPTIONS[self._spoof_os_idx] if self._spoof_os_idx > 0 else None + spoof_arch = ARCH_OPTIONS[self._spoof_arch_idx] if self._spoof_arch_idx > 0 else None + self._run_worker(SyncWorker(spoof_os, spoof_arch), self._on_done) + + @Slot() + def cancelOperation(self): + if self._worker and self._worker.isRunning(): + self._worker.terminate() + self._worker.wait() + self._busy = False + self._progress = -1 + self.busyChanged.emit() + self._set_status("Cancelled", "#f39c12") + + @Slot(str) + def downloadGeoip(self, source): + if not self._geoip_names or source not in self._geoip_names: + return + if source == self._geoip_installed: + from ..geolocation import needs_update + + if not needs_update(): + return + self._start_geoip_download(source) + + @Slot() + def refreshGeoip(self): + if self._geoip_installed: + self._start_geoip_download(self._geoip_installed) + + @Slot() + def deleteGeoipData(self): + from ..geolocation import remove_mmdb + + remove_mmdb() + self._load_geoip() + + @Slot(str) + def deleteGeoipSource(self, source): + from ..geolocation import MMDB_DIR, _get_geoip_config_by_name + + try: + name = _get_geoip_config_by_name(source)['name'].lower() + for f in MMDB_DIR.glob(f"{name}-*.mmdb"): + f.unlink() + except Exception: + pass + self._load_geoip() + + @Slot(str) + def setActiveGeoip(self, source): + from ..geolocation import _get_geoip_config_by_name, save_geoip_config + + if source not in self._geoip_downloaded: + return + save_geoip_config(_get_geoip_config_by_name(source)) + self._load_geoip() + + @Slot(int) + def setSpoofOs(self, index): + self._spoof_os_idx = index + self.debugChanged.emit() + + @Slot(int) + def setSpoofArch(self, index): + self._spoof_arch_idx = index + self.debugChanged.emit() + + @Slot() + def openGeoipFolder(self): + from ..geolocation import MMDB_DIR + + if not MMDB_DIR.exists(): + return + + import subprocess + + path = str(MMDB_DIR) + if sys.platform == 'win32': + subprocess.Popen(['explorer', path]) + elif sys.platform == 'darwin': + subprocess.Popen(['open', path]) + else: + subprocess.Popen(['xdg-open', path]) + + @Slot(str) + def lookupIp(self, ip): + if not ip.strip(): + self._lookup_result = "Enter IP" + self._lookup_ok = False + self.geoipChanged.emit() + return + + try: + from ..geolocation import get_geolocation, get_mmdb_path + + if not get_mmdb_path().exists(): + self._lookup_result = "Database not downloaded" + self._lookup_ok = False + self.geoipChanged.emit() + return + + geo = get_geolocation(ip.strip()) + self._lookup_result = "
".join([ + f"Country: {geo.locale.region}", + f"TZ: {geo.timezone}", + f"Lat/Lon: {geo.latitude:.4f}, {geo.longitude:.4f}", + ]) + self._lookup_ok = True + except Exception as e: + self._lookup_result = str(e) + self._lookup_ok = False + + self.geoipChanged.emit() + + # Internal helpers + + def _load_geoip(self): + self._geoip_installed = "" + self._geoip_downloaded = [] + self._geoip_path = "" + self._geoip_size = "" + self._geoip_mtime = "" + + try: + from ..geolocation import ( + ALLOW_GEOIP, + MMDB_DIR, + _load_geoip_repos, + get_mmdb_path, + load_geoip_config, + ) + + if not ALLOW_GEOIP: + self.geoipChanged.emit() + return + + repos, _ = _load_geoip_repos() + self._geoip_names = self._geoip_sources = [r.get('name', '?') for r in repos] + + if MMDB_DIR.exists(): + for repo in repos: + nm = repo.get('name', '').lower() + if (MMDB_DIR / f"{nm}-combined.mmdb").exists() or \ + (MMDB_DIR / f"{nm}-ipv4.mmdb").exists(): + self._geoip_downloaded.append(repo.get('name', '')) + + config = load_geoip_config() + path = get_mmdb_path('ipv4', config) + if path.exists(): + from datetime import datetime + + self._geoip_installed = config.get('name', '') + self._geoip_path = str(path.parent) + stat = path.stat() + size = stat.st_size + if 'combined' not in config.get('urls', {}): + ipv6 = get_mmdb_path('ipv6', config) + if ipv6.exists(): + size += ipv6.stat().st_size + self._geoip_size = f"{size / (1024 * 1024):.1f} MB" + self._geoip_mtime = datetime.fromtimestamp(stat.st_mtime).strftime( + '%Y-%m-%d %H:%M' + ) + except Exception: + pass + + self.geoipChanged.emit() + + def _load_spoof_from_cache(self): + cache = load_repo_cache() + if not cache: + return + spoof_os = cache.get('spoof_os') + spoof_arch = cache.get('spoof_arch') + if spoof_os and spoof_os in OS_OPTIONS: + self._spoof_os_idx = OS_OPTIONS.index(spoof_os) + if spoof_arch and spoof_arch in ARCH_OPTIONS: + self._spoof_arch_idx = ARCH_OPTIONS.index(spoof_arch) + + def _refresh(self): + items = [] + self._selected = -1 + + if not self._current_repo: + self._version_model.set_items(items) + return + + installed = {v.version.build: v for v in list_installed()} + cfg = load_config() + active = cfg.get('active_build') + pinned = cfg.get('pinned') + versions = get_cached_versions(self._current_repo.name) + + if not versions: + self._set_status("No cache. Run Sync.", "#f39c12") + self._version_model.set_items(items) + return + + for section, is_prerelease in [("stable", False), ("prerelease", True)]: + version_list = [v for v in versions if v.is_prerelease == is_prerelease] + if not version_list: + continue + + expanded = self._sections.get(section, True) + items.append(VersionItem( + section.capitalize(), is_header=True, + is_prerelease=is_prerelease, section=section, expanded=expanded, + )) + + if expanded: + for v in version_list: + inst = installed.get(v.version.build) + if self._installed_only and not inst: + continue + items.append(VersionItem( + f"v{v.version.version}", v.version.build, + is_prerelease=is_prerelease, + is_active=(active == v.version.build), + is_pinned=(pinned == v.version.full_string if pinned else False), + is_installed=bool(inst), + version_data=v, installed_data=inst, + )) + + self._version_model.set_items(items) + self.selectionChanged.emit() + + def _set_status(self, text, color): + self._status_text = text + self._status_color = color + self.statusChanged.emit() + + def _on_worker_progress(self, value): + self._progress = value + self.progressChanged.emit() + + def _on_worker_status(self, status): + self._set_status(status, "#3498db") + + def _on_geoip_progress(self, value): + self._geoip_progress = value + self.geoipChanged.emit() + + def _on_geoip_status(self, status): + self.geoipChanged.emit() + + def _run_worker(self, worker, done_callback): + self._busy = True + self._progress = -1 + self.busyChanged.emit() + self._worker = worker + worker.progress.connect(self._on_worker_progress) + worker.status.connect(self._on_worker_status) + worker.done.connect(done_callback) + worker.start() + + def _on_done(self, ok, msg): + self._busy = False + self._progress = -1 + self.busyChanged.emit() + self._set_status(msg if ok else f"Error: {msg}", "#2ecc71" if ok else "#e74c3c") + if ok: + self._channel_data = None + self._refresh() + self.infoChanged.emit() + self._load_spoof_from_cache() + self.debugChanged.emit() + + def _on_geoip_done(self, ok, msg): + self._geoip_busy = False + self._geoip_progress = -1 + self._load_geoip() + + def _start_geoip_download(self, source): + self._geoip_busy = True + self._geoip_progress = -1 + self.geoipChanged.emit() + self._worker = GeoIPWorker(source) + self._worker.progress.connect(self._on_geoip_progress) + self._worker.status.connect(self._on_geoip_status) + self._worker.done.connect(self._on_geoip_done) + self._worker.start() + + +def main(debug=False): + QQuickStyle.setStyle("Basic") + + # Suppress Breeze style warnings on KDE + from PySide6.QtCore import QtMsgType, qInstallMessageHandler + + def _msg_filter(mode, ctx, msg): + if 'org/kde/breeze' in msg: + return + if mode in (QtMsgType.QtCriticalMsg, QtMsgType.QtFatalMsg): + print(msg, file=sys.stderr) + + qInstallMessageHandler(_msg_filter) + + app = QGuiApplication(sys.argv) + app.setWindowIcon(QIcon(str(Path(__file__).parent / "assets/icon.ico"))) + engine = QQmlApplicationEngine() + engine.rootContext().setContextProperty("debugMode", debug) + + backend = Backend() + backend.setParent(engine) + engine.rootContext().setContextProperty("backend", backend) + engine.load(QUrl.fromLocalFile(str(Path(__file__).parent / "qml/main.qml"))) + + if not engine.rootObjects(): + sys.exit(-1) + + ret = app.exec() + del engine + sys.exit(ret) diff --git a/pythonlib/camoufox/gui/qml/main.qml b/pythonlib/camoufox/gui/qml/main.qml new file mode 100644 index 0000000..a296e82 --- /dev/null +++ b/pythonlib/camoufox/gui/qml/main.qml @@ -0,0 +1,1307 @@ +import QtQuick +import QtQuick.Controls +import QtQuick.Layouts + +ApplicationWindow { + id: root + visible: true + width: 800 + height: 520 + minimumWidth: 640 + minimumHeight: 400 + title: "Camoufox Manager" + color: c.bg + + FontLoader { id: segoe; source: "../assets/SegUIVar.ttf" } + FontLoader { id: mdl2; source: "../assets/segmdl2.ttf" } + + // 4pt spacing scale + property real scale: 1.0 + readonly property int row: Math.round(32 * scale) + readonly property int s1: Math.round(4 * scale) + readonly property int s2: Math.round(8 * scale) + readonly property int s3: Math.round(12 * scale) + readonly property int s4: Math.round(16 * scale) + readonly property int textSm: Math.round(12 * scale) + readonly property int textMd: Math.round(14 * scale) + readonly property int colW: Math.round(100 * scale) + readonly property string fontMain: segoe.name + readonly property string fontIcon: mdl2.name + + property string geoipDlgSource: "" + + QtObject { + id: c + readonly property color bg: "#181818" + readonly property color fg: "#1f1f1f" + readonly property color raised: "#282828" + readonly property color border: "#383838" + readonly property color text: "#ffffff" + readonly property color muted: "#a0a0a0" + readonly property color dim: "#606060" + readonly property color accent: "#569cd6" + readonly property color ok: "#6b9e7e" + readonly property color err: "#f14c4c" + } + + // Primitives + + component T: Text { + color: c.text + font.family: fontMain + font.pixelSize: textSm + } + + component Muted: Text { + color: c.muted + font.family: fontMain + font.pixelSize: textSm + } + + component Bold: Text { + color: c.text + font.family: fontMain + font.pixelSize: textSm + font.weight: Font.DemiBold + } + + component Header: Text { + color: c.muted + font.family: fontMain + font.pixelSize: Math.round(10 * scale) + font.weight: Font.DemiBold + font.letterSpacing: 0.5 + } + + component Icon: Text { + property string icon: "" + text: icon + color: c.text + font.family: fontIcon + font.pixelSize: textSm + } + + component Rule: Rectangle { + width: parent ? parent.width : 0 + height: 1 + color: c.border + } + + component Btn: Rectangle { + property string text: "" + property string icon: "" + property bool on: true + property color accent: c.text + signal clicked + + width: _row.implicitWidth + s3 + height: row - s2 + radius: s1 + color: on && _ma.containsMouse ? c.raised : "transparent" + border.color: c.border + border.width: 1 + opacity: on ? 1 : 0.4 + + Row { + id: _row + anchors.centerIn: parent + spacing: (icon && text) ? s1 : 0 + + Icon { + visible: !!icon + anchors.verticalCenter: parent.verticalCenter + icon: _row.parent.icon + color: _row.parent.on ? _row.parent.accent : c.muted + } + + T { + visible: !!text + anchors.verticalCenter: parent.verticalCenter + text: _row.parent.text + color: _row.parent.on ? _row.parent.accent : c.muted + } + } + + MouseArea { + id: _ma + anchors.fill: parent + hoverEnabled: true + cursorShape: on ? Qt.PointingHandCursor : Qt.ArrowCursor + onClicked: if (parent.on) parent.clicked() + } + } + + component Tag: Rectangle { + property alias text: lbl.text + property color accent: c.ok + + width: lbl.width + s3 + height: row - s3 + radius: s1 + color: Qt.rgba(accent.r, accent.g, accent.b, 0.15) + border.color: accent + border.width: 1 + + T { + id: lbl + anchors.centerIn: parent + color: parent.accent + font.capitalization: Font.Capitalize + } + } + + component Check: Rectangle { + property bool on: false + + width: s3 + height: s3 + radius: s1 + color: on ? c.accent : "transparent" + border.color: on ? c.accent : c.muted + border.width: 1 + + Text { + anchors.centerIn: parent + text: "\uE73E" + color: c.bg + font.family: fontIcon + font.pixelSize: Math.round(8 * scale) + visible: parent.on + } + } + + component Input: Rectangle { + property alias text: inp.text + property alias input: inp + property string placeholder: "" + + width: Math.round(160 * scale) + height: row - s2 + radius: s1 + color: c.raised + border.color: inp.activeFocus ? c.accent : c.border + border.width: 1 + + TextInput { + id: inp + anchors.fill: parent + leftPadding: s2 + rightPadding: s2 + color: c.text + font.family: fontMain + font.pixelSize: textSm + verticalAlignment: Text.AlignVCenter + } + + Muted { + anchors.left: parent.left + anchors.leftMargin: s2 + anchors.verticalCenter: parent.verticalCenter + text: placeholder + visible: !inp.text && !inp.activeFocus + } + } + + component ProgressBar: Rectangle { + id: _bar + property real value: -1 + property bool active: false + + height: s1 + radius: s1 / 2 + color: c.raised + + Rectangle { + id: _fill + x: 0 + width: _bar.value >= 0 ? _bar.width * _bar.value : _bar.width * 0.3 + height: parent.height + radius: parent.radius + color: c.accent + } + + Timer { + interval: 16; repeat: true + running: _bar.active && _bar.value < 0 + property real pos: 0 + property bool fwd: true + onTriggered: { + if (fwd) { pos += 0.02; if (pos >= 0.7) fwd = false } + else { pos -= 0.02; if (pos <= 0) fwd = true } + _fill.x = _bar.width * pos + } + onRunningChanged: if (!running) { pos = 0; _fill.x = 0 } + } + } + + component Combo: ComboBox { + id: cb + implicitWidth: colW + implicitHeight: row - s2 + + background: Rectangle { + color: c.raised + radius: s1 + border.color: c.border + border.width: 1 + } + + contentItem: T { + text: cb.displayText + leftPadding: s2 + verticalAlignment: Text.AlignVCenter + } + + delegate: ItemDelegate { + required property int index + required property string modelData + width: cb.width + height: row - s2 + contentItem: T { + text: modelData + verticalAlignment: Text.AlignVCenter + } + background: Rectangle { + color: highlighted ? c.raised : c.fg + } + } + } + + component SideRow: Rectangle { + property bool sel: false + property bool bar: false + signal clicked + + width: parent ? parent.width : 0 + height: row + color: sel ? c.raised : ma.containsMouse ? c.raised : "transparent" + + Rectangle { + visible: bar && sel + width: s1 + height: parent.height + color: c.accent + } + + MouseArea { + id: ma + anchors.fill: parent + hoverEnabled: true + cursorShape: Qt.PointingHandCursor + onClicked: parent.clicked() + } + } + + component Panel: Rectangle { + default property alias content: col.children + property string title: "" + + color: c.bg + + ColumnLayout { + anchors.fill: parent + spacing: 0 + + Rectangle { + Layout.fillWidth: true + height: row + color: c.fg + + Rule { anchors.bottom: parent.bottom } + + Header { + anchors.left: parent.left + anchors.leftMargin: s3 + anchors.verticalCenter: parent.verticalCenter + text: title + } + } + + ColumnLayout { + id: col + Layout.fillWidth: true + Layout.margins: s4 + spacing: s3 + } + + Item { Layout.fillHeight: true } + } + } + + // Dialog + + property string dlgAct: "" + property string dlgVer: "" + property string dlgBuild: "" + property int dlgIdx: -1 + + function showDlg(act, idx, ver, build) { + backend.selectVersion(idx) + dlgAct = act + dlgIdx = idx + dlgVer = ver + dlgBuild = build + } + + Rectangle { + anchors.fill: parent + color: "#80000000" + visible: dlgAct !== "" + z: 100 + MouseArea { + anchors.fill: parent + hoverEnabled: true + onClicked: dlgAct = "" + } + } + + Rectangle { + visible: dlgAct !== "" + anchors.centerIn: parent + width: Math.round(340 * scale) + height: dlgContent.height + s4 * 2 + color: c.fg + border.color: c.border + radius: s2 + z: 101 + + Column { + id: dlgContent + anchors.left: parent.left + anchors.right: parent.right + anchors.top: parent.top + anchors.margins: s4 + spacing: s3 + + Bold { + text: dlgAct === "install" || dlgAct === "promptInstall" ? "Install" + : dlgAct === "uninstall" ? "Uninstall" : "Set Active" + font.pixelSize: textMd + } + + T { + width: parent.width + wrapMode: Text.Wrap + text: dlgAct === "promptInstall" + ? "Camoufox " + dlgVer + "-" + dlgBuild + " is now active but not installed. Install it now?" + : (dlgAct === "install" ? "Install" : dlgAct === "uninstall" ? "Remove" : "Set") + + " Camoufox " + dlgVer + "-" + dlgBuild + + (dlgAct === "uninstall" ? " from disk?" : dlgAct === "setActive" ? " as active?" : "?") + } + + T { + visible: (dlgAct === "install" || dlgAct === "promptInstall") && backend.selectedIsPrerelease + width: parent.width + color: c.err + wrapMode: Text.Wrap + text: "Warning: Prerelease versions may be unstable." + } + + Item { width: 1; height: s1 } + + Row { + anchors.right: parent.right + spacing: s2 + + Btn { + text: dlgAct === "promptInstall" ? "Skip" : "Cancel" + onClicked: dlgAct = "" + } + + Btn { + text: dlgAct === "install" || dlgAct === "promptInstall" ? "Install" + : dlgAct === "uninstall" ? "Uninstall" : "Set Active" + accent: dlgAct === "uninstall" ? c.err : dlgAct === "setActive" ? c.accent : c.ok + onClicked: { + if (dlgAct === "install" || dlgAct === "promptInstall") { + backend.installSelected() + } else if (dlgAct === "uninstall") { + backend.uninstallSelected() + } else { + backend.setActive(dlgIdx) + backend.selectVersion(dlgIdx) + if (backend.canInstall) { + dlgAct = "promptInstall" + return + } + } + dlgAct = "" + } + } + } + } + } + + Connections { + target: backend + function onCurrentRepoChanged() { + repoList.currentIndex = backend.currentRepoIndex + } + function onInstallPrompt(idx, display, build) { + showDlg("promptInstall", idx, display, build) + } + } + + // Layout + + ColumnLayout { + anchors.fill: parent + spacing: 0 + + // Tab bar + Rectangle { + Layout.fillWidth: true + height: row + color: c.fg + + Rule { anchors.bottom: parent.bottom } + + Row { + anchors.left: parent.left + anchors.leftMargin: s3 + anchors.verticalCenter: parent.verticalCenter + + Repeater { + model: ["Browsers", "GeoIP", "Info"] + + Rectangle { + width: tabLbl.width + s4 * 2 + height: row + color: "transparent" + + T { + id: tabLbl + anchors.centerIn: parent + text: modelData + color: tabs.currentIndex === index ? c.accent : c.muted + } + + Rectangle { + anchors.bottom: parent.bottom + width: parent.width + height: s1 / 2 + color: c.accent + visible: tabs.currentIndex === index + } + + MouseArea { + anchors.fill: parent + cursorShape: Qt.PointingHandCursor + onClicked: tabs.currentIndex = index + } + } + } + } + } + + RowLayout { + Layout.fillWidth: true + Layout.fillHeight: true + spacing: 0 + + // Sidebar + Rectangle { + Layout.preferredWidth: Math.round(210 * scale) + Layout.fillHeight: true + color: c.fg + visible: tabs.currentIndex === 0 + + Rectangle { + anchors.right: parent.right + width: 1 + height: parent.height + color: c.border + } + + Column { + anchors.top: parent.top + anchors.left: parent.left + anchors.right: parent.right + anchors.rightMargin: 1 + + Rectangle { + width: parent.width + height: row + color: "transparent" + + Header { + anchors.left: parent.left + anchors.leftMargin: s3 + anchors.verticalCenter: parent.verticalCenter + text: "CHANNEL" + } + } + + Repeater { + model: backend.channels + + Rectangle { + width: parent ? parent.width : 0 + height: row + s2 + color: chMa.containsMouse ? c.raised : "transparent" + + Column { + anchors.left: parent.left + anchors.leftMargin: s4 + anchors.right: parent.right + anchors.rightMargin: s3 + anchors.verticalCenter: parent.verticalCenter + spacing: 1 + + Row { + spacing: s2 + Check { on: backend.followedChannel === backend.channelKeys[index]; anchors.verticalCenter: parent.verticalCenter } + T { text: (backend.followedChannel === backend.channelKeys[index] ? "Following " : "Follow ") + modelData; anchors.verticalCenter: parent.verticalCenter } + } + + Muted { + leftPadding: s3 + s2 + text: backend.channelLatest[index] ? ("(latest: " + backend.channelLatest[index] + ")") : "(sync first)" + font.pixelSize: Math.round(10 * scale) + } + } + + MouseArea { + id: chMa + anchors.fill: parent + hoverEnabled: true + cursorShape: Qt.PointingHandCursor + onClicked: backend.setFollowedChannel(index) + } + } + } + + Rule {} + + Rectangle { + width: parent.width + height: row + color: "transparent" + + Header { + anchors.left: parent.left + anchors.leftMargin: s3 + anchors.verticalCenter: parent.verticalCenter + text: "REPOSITORIES" + } + } + + ListView { + id: repoList + width: parent.width + height: contentHeight + interactive: false + model: backend.repos + + delegate: SideRow { + sel: ListView.isCurrentItem + bar: true + + T { + anchors.left: parent.left + anchors.leftMargin: s4 + anchors.verticalCenter: parent.verticalCenter + text: modelData + } + + onClicked: { + repoList.currentIndex = index + backend.selectRepo(index) + } + } + + Component.onCompleted: backend.selectRepo(0) + } + } + + Column { + anchors.bottom: parent.bottom + anchors.left: parent.left + anchors.right: parent.right + anchors.rightMargin: 1 + + Rule {} + + Rectangle { + width: parent.width + height: row + color: "transparent" + + Header { + anchors.left: parent.left + anchors.leftMargin: s3 + anchors.verticalCenter: parent.verticalCenter + text: "FILTERS" + } + } + + SideRow { + Row { + anchors.left: parent.left + anchors.leftMargin: s4 + anchors.verticalCenter: parent.verticalCenter + spacing: s2 + + Check { + id: filterChk + anchors.verticalCenter: parent.verticalCenter + } + + T { + text: "Installed only" + anchors.verticalCenter: parent.verticalCenter + } + } + + onClicked: { + filterChk.on = !filterChk.on + backend.setInstalledOnly(filterChk.on) + } + } + } + } + + StackLayout { + id: tabs + Layout.fillWidth: true + Layout.fillHeight: true + + // Browsers + Rectangle { + color: c.bg + + ColumnLayout { + anchors.fill: parent + spacing: 0 + + Rectangle { + Layout.fillWidth: true + height: row + color: c.fg + + Rule { anchors.bottom: parent.bottom } + + Row { + anchors.left: parent.left + anchors.leftMargin: s3 + anchors.verticalCenter: parent.verticalCenter + spacing: s3 + + Rectangle { + width: s3 + height: s3 + radius: s1 + color: "transparent" + border.color: c.dim + border.width: 1 + anchors.verticalCenter: parent.verticalCenter + } + + Header { + text: "VERSION" + width: colW + anchors.verticalCenter: parent.verticalCenter + } + + Header { + text: "BUILD" + width: colW + anchors.verticalCenter: parent.verticalCenter + } + + Header { + text: "STATUS" + anchors.verticalCenter: parent.verticalCenter + } + } + } + + ListView { + id: vList + Layout.fillWidth: true + Layout.fillHeight: true + clip: true + model: backend.versionModel + + ScrollBar.vertical: ScrollBar { + contentItem: Rectangle { + implicitWidth: s2 + radius: s1 + color: c.border + } + } + + delegate: Rectangle { + id: vrow + property bool hov: hover.hovered + + width: vList.width + height: row + color: model.isHeader ? c.fg : + model.isActive ? Qt.rgba(c.accent.r, c.accent.g, c.accent.b, 0.06) : + hov ? Qt.rgba(1, 1, 1, 0.02) : "transparent" + + Rectangle { + anchors.top: parent.top + width: parent.width + height: 1 + color: model.isHeader && index > 0 ? c.border : "transparent" + } + + Rectangle { + anchors.bottom: parent.bottom + width: parent.width + height: 1 + color: model.isHeader ? c.border : "transparent" + } + + Rectangle { + width: s1 + height: parent.height + color: model.isActive && !model.isHeader ? c.accent : "transparent" + } + + HoverHandler { id: hover } + + Bold { + anchors.left: parent.left + anchors.leftMargin: s3 + anchors.verticalCenter: parent.verticalCenter + visible: model.isHeader + text: model.display + } + + RowLayout { + anchors.fill: parent + anchors.leftMargin: s3 + anchors.rightMargin: s2 + spacing: s3 + visible: !model.isHeader + + Check { + on: model.isPinned + MouseArea { + anchors.fill: parent + cursorShape: Qt.PointingHandCursor + onClicked: showDlg("setActive", index, model.display, model.build) + } + } + + T { + text: model.display + color: model.isInstalled ? c.text : c.dim + Layout.preferredWidth: colW + } + + Muted { + text: model.build + Layout.preferredWidth: colW + } + + Row { + spacing: s2 + Layout.fillWidth: true + + Tag { + visible: model.isInstalled + text: "installed" + } + + Tag { + visible: model.isActive + text: "active" + accent: c.accent + } + } + + Btn { + visible: vrow.hov && !model.isInstalled + icon: "\uE896" + text: "Download" + accent: c.ok + on: !backend.busy + onClicked: showDlg("install", index, model.display, model.build) + } + + Btn { + visible: vrow.hov && model.isInstalled + icon: "\uE74D" + text: "Delete" + accent: c.err + on: !backend.busy + onClicked: showDlg("uninstall", index, model.display, model.build) + } + } + } + } + + } + } + + // GeoIP + Rectangle { + color: c.bg + + ColumnLayout { + anchors.fill: parent + spacing: 0 + + Rectangle { + Layout.fillWidth: true + height: row + color: c.fg + + Rule { anchors.bottom: parent.bottom } + + Header { + anchors.left: parent.left + anchors.leftMargin: s3 + anchors.verticalCenter: parent.verticalCenter + text: "GEOIP DATABASE" + } + } + + Column { + Layout.fillWidth: true + spacing: 0 + + Repeater { + model: backend.geoipSources + + Rectangle { + id: geoRow + property bool hov: geoHover.hovered + property bool active: backend.geoipInstalled === modelData + property bool downloaded: backend.geoipDownloaded.indexOf(modelData) >= 0 + + width: parent.width + height: row + color: active ? Qt.rgba(c.accent.r, c.accent.g, c.accent.b, 0.06) : + hov ? Qt.rgba(1, 1, 1, 0.02) : "transparent" + + Rectangle { + width: s1 + height: parent.height + color: geoRow.active ? c.accent : "transparent" + } + + HoverHandler { id: geoHover } + + MouseArea { + anchors.fill: parent + cursorShape: Qt.PointingHandCursor + onClicked: { + if (backend.geoipBusy) return + if (!geoRow.downloaded) { + root.geoipDlgSource = modelData + } else if (!geoRow.active) { + backend.setActiveGeoip(modelData) + } + } + } + + RowLayout { + anchors.fill: parent + anchors.leftMargin: s3 + anchors.rightMargin: s3 + spacing: s3 + + T { + text: modelData + color: geoRow.downloaded ? c.text : c.dim + Layout.fillWidth: true + } + + Tag { + visible: geoRow.active + text: "active" + accent: c.accent + } + + Btn { + visible: geoRow.hov && !geoRow.downloaded + icon: "\uE896" + text: "Download" + accent: c.ok + on: !backend.geoipBusy + onClicked: root.geoipDlgSource = modelData + } + + Btn { + visible: geoRow.hov && geoRow.downloaded && !geoRow.active + icon: "\uE74D" + text: "Delete" + accent: c.err + on: !backend.geoipBusy + onClicked: backend.deleteGeoipSource(modelData) + } + } + } + } + } + + Rule { Layout.fillWidth: true; Layout.topMargin: s3 } + + // IP Lookup + ColumnLayout { + Layout.fillWidth: true + Layout.margins: s4 + spacing: s2 + + Bold { text: "IP Lookup" } + + Row { + spacing: s2 + + Input { + id: ipIn + placeholder: "Enter IP..." + input.onAccepted: backend.lookupIp(text) + } + + Btn { + text: "Lookup" + onClicked: backend.lookupIp(ipIn.text) + } + } + + T { + visible: backend.lookupResult.length > 0 + text: backend.lookupResult + textFormat: Text.RichText + color: backend.lookupSuccess ? c.ok : c.err + } + } + + Item { Layout.fillHeight: true } + + Rectangle { + Layout.fillWidth: true + visible: backend.geoipInstalled.length > 0 || backend.geoipBusy + height: visible ? metaCol.height + s4 * 2 : 0 + color: c.fg + + Rule { anchors.top: parent.top } + + RowLayout { + anchors.fill: parent + anchors.margins: s4 + spacing: s4 + + Column { + id: metaCol + Layout.fillWidth: true + spacing: s1 + + Row { + spacing: s2 + Muted { text: "Path:" } + T { text: backend.geoipPath } + } + + Row { + spacing: s2 + Muted { text: "Size:" } + T { text: backend.geoipSize } + } + + Row { + spacing: s2 + Muted { text: "Downloaded:" } + T { text: backend.geoipMtime } + } + } + + Btn { + icon: "\uE72C" + accent: c.accent + on: !backend.geoipBusy && backend.geoipInstalled.length > 0 + onClicked: backend.refreshGeoip() + } + + Btn { + icon: "\uE838" + accent: c.text + on: backend.geoipInstalled.length > 0 + onClicked: backend.openGeoipFolder() + } + + Btn { + icon: "\uE74D" + accent: c.err + on: !backend.geoipBusy && backend.geoipInstalled.length > 0 + onClicked: backend.deleteGeoipData() + } + } + } + + ProgressBar { + Layout.fillWidth: true + visible: backend.geoipBusy + height: s2 + radius: 0 + value: backend.geoipProgress + active: backend.geoipBusy + } + } + + // GeoIP download dialog + Rectangle { + anchors.fill: parent + color: "#80000000" + visible: root.geoipDlgSource !== "" + z: 100 + + MouseArea { + anchors.fill: parent + hoverEnabled: true + onClicked: root.geoipDlgSource = "" + } + } + + Rectangle { + visible: root.geoipDlgSource !== "" + anchors.centerIn: parent + width: Math.round(320 * scale) + height: Math.round(140 * scale) + color: c.fg + border.color: c.border + radius: s2 + z: 101 + + ColumnLayout { + anchors.fill: parent + anchors.margins: s4 + spacing: s3 + + Bold { + text: "Download GeoIP Database" + font.pixelSize: textMd + } + + T { + Layout.fillWidth: true + wrapMode: Text.Wrap + text: "Download " + root.geoipDlgSource + " database? This will download both IPv4 and IPv6 files." + } + + Item { Layout.fillHeight: true } + + Row { + Layout.alignment: Qt.AlignRight + spacing: s2 + + Btn { + text: "Cancel" + onClicked: root.geoipDlgSource = "" + } + + Btn { + text: "Download" + accent: c.ok + onClicked: { + backend.downloadGeoip(root.geoipDlgSource) + root.geoipDlgSource = "" + } + } + } + } + } + } + + // Info + Panel { + title: "SYSTEM INFO" + + GridLayout { + columns: 2 + columnSpacing: s4 * 2 + rowSpacing: s2 + + Muted { text: "Active Browser" } + Bold { text: backend.activeBrowserText; color: backend.activeBrowserColor } + + Muted { text: "Python Library" } + T { text: backend.libraryVersion } + + Muted { text: "Playwright" } + T { text: backend.playwrightVersion } + + Muted { text: "Browserforge" } + T { text: backend.browserforgeVersion } + + Muted { text: "Last Sync" } + T { text: backend.lastSyncTime } + + Muted { text: "Repos" } + T { text: backend.reposInfo } + + Muted { text: "Website" } + T { + text: "camoufox.com" + textFormat: Text.RichText + onLinkActivated: link => Qt.openUrlExternally(link) + } + } + + Rule { visible: debugMode } + + Bold { text: "Debug Options"; visible: debugMode } + + Row { + visible: debugMode + spacing: s4 + + Row { + spacing: s2 + Muted { text: "Spoof OS"; anchors.verticalCenter: parent.verticalCenter } + Combo { + model: backend.spoofOsOptions + currentIndex: backend.spoofOsIndex + onActivated: backend.setSpoofOs(currentIndex) + implicitWidth: Math.round(90 * scale) + } + } + + Row { + spacing: s2 + Muted { text: "Spoof Arch"; anchors.verticalCenter: parent.verticalCenter } + Combo { + model: backend.spoofArchOptions + currentIndex: backend.spoofArchIndex + onActivated: backend.setSpoofArch(currentIndex) + implicitWidth: Math.round(90 * scale) + } + } + } + + Muted { + visible: debugMode + text: "Set spoof options then Sync on Browsers tab" + font.pixelSize: Math.round(10 * scale) + } + + Rule { visible: debugMode } + + Row { + visible: debugMode + spacing: s3 + + Muted { text: "UI Scale"; anchors.verticalCenter: parent.verticalCenter } + + Slider { + id: scaleSlider + from: 0.8 + to: 1.5 + stepSize: 0.1 + value: root.scale + onPressedChanged: if (!pressed) root.scale = value + implicitWidth: Math.round(140 * scale) + implicitHeight: row - s2 + + background: Rectangle { + x: scaleSlider.leftPadding + y: scaleSlider.topPadding + scaleSlider.availableHeight / 2 - s1 / 2 + width: scaleSlider.availableWidth + height: s1 + radius: s1 / 2 + color: c.raised + + Rectangle { + width: scaleSlider.visualPosition * parent.width + height: s1 + radius: s1 / 2 + color: c.accent + } + } + + handle: Rectangle { + x: scaleSlider.leftPadding + scaleSlider.visualPosition * (scaleSlider.availableWidth - width) + y: scaleSlider.topPadding + scaleSlider.availableHeight / 2 - height / 2 + width: s4 + height: s4 + radius: s2 + color: scaleSlider.pressed ? c.accent : c.text + } + } + + T { text: Math.round(scaleSlider.value * 100) + "%" } + } + } + } + } + + // Status bar + Rectangle { + visible: tabs.currentIndex === 0 + Layout.fillWidth: true + height: row + color: c.fg + + Rule { anchors.top: parent.top } + + RowLayout { + anchors.fill: parent + anchors.leftMargin: s3 + anchors.rightMargin: s3 + spacing: s2 + + Btn { + visible: debugMode + text: "Refresh" + on: !backend.busy + onClicked: backend.refresh() + } + + Btn { + icon: "\uE895" + text: "Sync Upstream" + accent: c.accent + on: !backend.busy + onClicked: backend.sync() + } + + Muted { + text: "Synced: " + backend.lastSyncTime + visible: !backend.busy + } + + Item { Layout.fillWidth: true } + + T { + visible: backend.statusText.length > 0 + text: backend.statusText + color: backend.statusColor + Layout.maximumWidth: Math.round(200 * scale) + elide: Text.ElideRight + } + + Row { + visible: backend.busy + spacing: s2 + + ProgressBar { + width: Math.round(80 * scale) + value: backend.progress + active: backend.busy + anchors.verticalCenter: parent.verticalCenter + } + + Btn { + text: "Cancel" + accent: c.err + onClicked: backend.cancelOperation() + } + } + + T { + visible: backend.activeLabel.length > 0 + text: backend.activeLabel + color: c.accent + Layout.maximumWidth: Math.round(200 * scale) + elide: Text.ElideRight + } + } + } + } +} diff --git a/pythonlib/camoufox/locale.py b/pythonlib/camoufox/locale.py index 7a0662a..63ecc94 100644 --- a/pythonlib/camoufox/locale.py +++ b/pythonlib/camoufox/locale.py @@ -1,29 +1,18 @@ import xml.etree.ElementTree as ET # nosec from dataclasses import dataclass -from typing import Any, Dict, Iterable, List, Optional, Set, Tuple, Union, cast +from typing import Any, Dict, Iterable, List, Optional, Set, Tuple, Union import numpy as np from language_tags import tags -from camoufox.pkgman import LOCAL_DATA, GitHubDownloader, rprint, webdl +from camoufox.pkgman import LOCAL_DATA from camoufox.warnings import LeakWarning from .exceptions import ( InvalidLocale, - MissingRelease, - NotInstalledGeoIPExtra, - UnknownIPLocation, UnknownLanguage, UnknownTerritory, ) -from .ip import validate_ip - -try: - import geoip2.database # type: ignore -except ImportError: - ALLOW_GEOIP = False -else: - ALLOW_GEOIP = True """ @@ -185,98 +174,6 @@ def _join_unique(seq: Iterable[str]) -> str: return ', '.join(x for x in seq if not (x in seen or seen.add(x))) -""" -Helpers to fetch geolocation, timezone, and locale data given an IP. -""" - -MMDB_FILE = LOCAL_DATA / 'GeoLite2-City.mmdb' -MMDB_REPO = "P3TERX/GeoLite.mmdb" - - -class MaxMindDownloader(GitHubDownloader): - """ - MaxMind database downloader from a GitHub repository. - """ - - def check_asset(self, asset: Dict) -> Optional[str]: - # Check for the first -City.mmdb file - if asset['name'].endswith('-City.mmdb'): - return asset['browser_download_url'] - return None - - def missing_asset_error(self) -> None: - raise MissingRelease('Failed to find GeoIP database release asset') - - -def geoip_allowed() -> None: - """ - Checks if the geoip2 module is available. - """ - if not ALLOW_GEOIP: - raise NotInstalledGeoIPExtra( - 'Please install the geoip extra to use this feature: pip install camoufox[geoip]' - ) - - -def download_mmdb() -> None: - """ - Downloads the MaxMind GeoIP2 database. - """ - geoip_allowed() - - asset_url = MaxMindDownloader(MMDB_REPO).get_asset() - - with open(MMDB_FILE, 'wb') as f: - webdl( - asset_url, - desc='Downloading GeoIP database', - buffer=f, - ) - - -def remove_mmdb() -> None: - """ - Removes the MaxMind GeoIP2 database. - """ - if not MMDB_FILE.exists(): - rprint("GeoIP database not found.") - return - - MMDB_FILE.unlink() - rprint("GeoIP database removed.") - - -def get_geolocation(ip: str) -> Geolocation: - """ - Gets the geolocation for an IP address. - """ - # Check if the database is downloaded - if not MMDB_FILE.exists(): - download_mmdb() - - # Validate the IP address - validate_ip(ip) - - with geoip2.database.Reader(str(MMDB_FILE)) as reader: - resp = reader.city(ip) - iso_code = cast(str, resp.registered_country.iso_code).upper() - location = resp.location - - # Check if any required attributes are missing - if any(not getattr(location, attr) for attr in ('longitude', 'latitude', 'time_zone')): - raise UnknownIPLocation(f"Unknown IP location: {ip}") - - # Get a statistically correct locale based on the country code - locale = SELECTOR.from_region(iso_code) - - return Geolocation( - locale=locale, - longitude=cast(float, resp.location.longitude), - latitude=cast(float, resp.location.latitude), - timezone=cast(str, resp.location.time_zone), - ) - - """ Gets a random language based on the territory code. """ diff --git a/pythonlib/camoufox/multiversion.py b/pythonlib/camoufox/multiversion.py new file mode 100644 index 0000000..d056476 --- /dev/null +++ b/pythonlib/camoufox/multiversion.py @@ -0,0 +1,412 @@ +""" +Manager for handling multiple Camoufox versions side by side +""" + +import os +import shlex +import shutil +import tempfile +from dataclasses import dataclass +from pathlib import Path +from typing import TYPE_CHECKING, Dict, List, Optional + +if TYPE_CHECKING: + from .pkgman import AvailableVersion + +import orjson +import rich_click as click + +from .pkgman import INSTALL_DIR, OS_NAME, Version, rprint, unzip + +BROWSERS_DIR: Path = INSTALL_DIR / "browsers" +CONFIG_FILE: Path = INSTALL_DIR / "config.json" +REPO_CACHE_FILE: Path = INSTALL_DIR / "repo_cache.json" + + +def load_config() -> Dict: + """ + Load user config from disk, or return empty dict + """ + if CONFIG_FILE.exists(): + try: + return orjson.loads(CONFIG_FILE.read_bytes()) + except orjson.JSONDecodeError: + pass + return {} + + +def save_config(config: Dict) -> None: + """ + Save user config to disk + """ + INSTALL_DIR.mkdir(parents=True, exist_ok=True) + CONFIG_FILE.write_bytes(orjson.dumps(config, option=orjson.OPT_INDENT_2)) + + +def load_repo_cache() -> Dict: + """ + Load cached repo data from disk + """ + if REPO_CACHE_FILE.exists(): + try: + return orjson.loads(REPO_CACHE_FILE.read_bytes()) + except orjson.JSONDecodeError: + pass + return {} + + +def save_repo_cache(cache: Dict) -> None: + """ + Save repo cache to disk + """ + INSTALL_DIR.mkdir(parents=True, exist_ok=True) + REPO_CACHE_FILE.write_bytes(orjson.dumps(cache, option=orjson.OPT_INDENT_2)) + + +def get_cached_versions(repo_name: Optional[str] = None) -> List['AvailableVersion']: + """ + Get cached available versions, optionally filtered by repo + """ + from .pkgman import AvailableVersion, Version + + cache = load_repo_cache() + if not cache.get('repos'): + return [] + + versions = [] + for repo_data in cache['repos']: + if repo_name and repo_data['name'].lower() != repo_name.lower(): + continue + for v in repo_data.get('versions', []): + versions.append( + AvailableVersion( + version=Version(build=v['build'], version=v['version']), + url=v['url'], + is_prerelease=v.get('is_prerelease', False), + asset_id=v.get('asset_id'), + asset_size=v.get('asset_size'), + asset_updated_at=v.get('asset_updated_at'), + ) + ) + + versions.sort(key=lambda x: x.version, reverse=True) + return versions + + +def get_cached_repo_names() -> List[str]: + """ + Get list of repo names in cache + """ + cache = load_repo_cache() + return [r['name'] for r in cache.get('repos', [])] + + +def get_repo_name(github_repo: str) -> str: + """ + Get display name for a repo from repos.yml, lowercased + """ + from .pkgman import RepoConfig + + for repo in RepoConfig.load_repos(): + if repo.repo == github_repo: + return repo.name.lower() + return github_repo.split('/')[0].lower() + + +@dataclass +class InstalledVersion: + """ + Information about an installed Camoufox version + """ + + repo_name: str + version: Version + path: Path + is_active: bool = False + is_prerelease: bool = False + asset_id: Optional[int] = None + asset_size: Optional[int] = None + asset_updated_at: Optional[str] = None + + @property + def relative_path(self) -> str: + """ + Path relative to INSTALL_DIR (like browsers/official/134.0.2-beta.20) + """ + return f"browsers/{self.repo_name}/{self.version.full_string}" + + @property + def channel_path(self) -> str: + """ + Channel display string (like official/stable/134.0.2-beta.20) + """ + ctype = "prerelease" if self.is_prerelease else "stable" + return f"{self.repo_name}/{ctype}/{self.version.full_string}" + + def get_changes(self, available: 'AvailableVersion') -> List[str]: + """ + Compare with an available version and return change indicators + """ + changes: List[str] = [] + if self.is_prerelease and not available.is_prerelease: + changes.append("prerelease -> stable") + elif not self.is_prerelease and available.is_prerelease: + changes.append("stable -> prerelease") + + if self.asset_updated_at and available.asset_updated_at: + if self.asset_updated_at != available.asset_updated_at: + changes.append("asset updated") + elif self.asset_size and available.asset_size: + if self.asset_size != available.asset_size: + changes.append("asset updated") + + return changes + + +def find_installed_by_build( + build: str, repo_name: Optional[str] = None +) -> Optional[InstalledVersion]: + """ + Find an installed version by its build string + """ + for v in list_installed(): + if v.version.build == build: + if repo_name is None or v.repo_name == repo_name: + return v + return None + + +def list_installed() -> List[InstalledVersion]: + """ + Scan browsers/ for installed versions, sorted by repo then version descending + """ + installed: List[InstalledVersion] = [] + config = load_config() + active = config.get('active_version') + + if not BROWSERS_DIR.exists(): + return installed + + for repo_dir in BROWSERS_DIR.iterdir(): + if not repo_dir.is_dir() or repo_dir.name.startswith('.'): + continue + + for version_dir in repo_dir.iterdir(): + if not version_dir.is_dir(): + continue + + version_json = version_dir / 'version.json' + if not version_json.exists(): + continue + + try: + ver = Version.from_path(version_dir) + with open(version_json, 'rb') as f: + version_data = orjson.loads(f.read()) + rel_path = f"browsers/{repo_dir.name}/{ver.full_string}" + installed.append( + InstalledVersion( + repo_name=repo_dir.name, + version=ver, + path=version_dir, + is_active=(rel_path == active), + is_prerelease=version_data.get('prerelease', False), + asset_id=version_data.get('asset_id'), + asset_size=version_data.get('asset_size'), + asset_updated_at=version_data.get('asset_updated_at'), + ) + ) + except (FileNotFoundError, orjson.JSONDecodeError): + continue + + installed.sort(key=lambda x: (x.repo_name, x.version), reverse=True) + return installed + + +def get_active_path() -> Optional[Path]: + """ + Get path to active version. Auto-selects newest if none set + """ + config = load_config() + active = config.get('active_version') + + if active: + path = INSTALL_DIR / active + if path.exists() and (path / 'version.json').exists(): + return path + + installed = list_installed() + if installed: + config['active_version'] = installed[0].relative_path + save_config(config) + return installed[0].path + + return None + + +def set_active(relative_path: str) -> None: + """ + Set the active version by its relative path + """ + config = load_config() + config['active_version'] = relative_path + save_config(config) + + +def find_installed_version(specifier: str) -> Optional[Path]: + """ + Find an installed version by path, build, full version, or repo/build + """ + installed = list_installed() + if not installed: + return None + + specifier_lower = specifier.lower() + + for v in installed: + if v.relative_path == specifier or v.relative_path == f"browsers/{specifier}": + return v.path + if f"browsers/{v.repo_name}/{v.version.full_string}".endswith(specifier): + return v.path + if f"{v.repo_name}/{v.version.build}".lower() == specifier_lower: + return v.path + if v.version.build.lower() == specifier_lower: + return v.path + if v.version.full_string.lower() == specifier_lower: + return v.path + if v.version.version and v.version.version.lower() == specifier_lower: + return v.path + + return None + + +def install_versioned(fetcher, replace: bool = False) -> bool: + """ + Install to browsers/{repo_name}/{version}-{build}/ + """ + repo_name = get_repo_name(fetcher.github_repo) + version_folder = f"{fetcher.version}-{fetcher.build}" + install_path = BROWSERS_DIR / repo_name / version_folder + + if install_path.exists() and (install_path / 'version.json').exists(): + if not replace: + installed_v = find_installed_by_build(fetcher.build, repo_name) + change_msg = "" + if installed_v and fetcher._selected_version: + changes = installed_v.get_changes(fetcher._selected_version) + if changes: + change_msg = f" ({', '.join(changes)})" + + rprint(f"Version v{fetcher.verstr} already installed{change_msg}.", fg="yellow") + if change_msg: + rprint("Use --replace to update with the new release.", fg="yellow") + else: + rprint("Use --replace to reinstall.", fg="yellow") + if not load_config().get('active_version'): + set_active(f"browsers/{repo_name}/{version_folder}") + return False + rprint(f"Replacing: {install_path}", fg="yellow") + shutil.rmtree(install_path) + + try: + install_path.mkdir(parents=True, exist_ok=True) + + with tempfile.NamedTemporaryFile() as temp_file: + fetcher.download_file(temp_file, fetcher.url) + rprint(f'Extracting Camoufox: {install_path}') + unzip(temp_file, str(install_path)) + + if fetcher._selected_version: + metadata = fetcher._selected_version.to_metadata() + else: + metadata = { + 'version': fetcher.version, + 'build': fetcher.build, + 'prerelease': fetcher.is_prerelease, + } + with open(install_path / 'version.json', 'wb') as f: + f.write(orjson.dumps(metadata)) + + if OS_NAME != 'win': + os.system(f'chmod -R 755 {shlex.quote(str(install_path))}') # nosec + + set_active(f"browsers/{repo_name}/{version_folder}") + + rprint(f'\nCamoufox v{fetcher.verstr} installed.', fg="green") + rprint(f'Path: {install_path}', fg="green") + return True + + except Exception as e: + rprint(f"Error: {e}", fg="red") + if install_path.exists(): + shutil.rmtree(install_path) + raise + + +def remove_version(path: Path) -> bool: + """ + Remove a specific version installation + """ + if not path.exists(): + return False + + rprint(f'Removing: {path}') + shutil.rmtree(path) + + parent = path.parent + if parent.exists() and parent != BROWSERS_DIR and not any(parent.iterdir()): + parent.rmdir() + if BROWSERS_DIR.exists() and not any(BROWSERS_DIR.iterdir()): + BROWSERS_DIR.rmdir() + + config = load_config() + try: + rel_path = str(path.relative_to(INSTALL_DIR)) + if config.get('active_version') == rel_path: + remaining = list_installed() + config['active_version'] = remaining[0].relative_path if remaining else None + save_config(config) + except ValueError: + pass # Path not relative to INSTALL_DIR + + return True + + +def print_tree(show_header: bool = True, show_paths: bool = False) -> None: + """ + Print installed versions as a tree + """ + installed = list_installed() + + if not installed: + rprint("No versions installed.", fg="yellow") + rprint("Run `camoufox fetch` to install.", fg="yellow") + return + + if show_header: + rprint("Installed versions:\n", fg="yellow") + + current_repo = None + for i, v in enumerate(installed): + is_last = (i == len(installed) - 1) or (installed[i + 1].repo_name != v.repo_name) + + if v.repo_name != current_repo: + current_repo = v.repo_name + click.secho(f"{current_repo}/", fg="cyan", bold=True, nl=False) + if show_paths: + click.secho(f" -> {BROWSERS_DIR / current_repo}", fg="bright_black") + else: + click.echo() + + branch = "└── " if is_last else "├── " + color = "green" if v.is_active else None + + click.echo(f" {branch}", nl=False) + click.secho(f"v{v.version.full_string}", fg=color, bold=v.is_active, nl=False) + if v.is_prerelease: + click.secho(" (prerelease)", fg="yellow", nl=False) + else: + click.secho(" (stable)", fg="blue", nl=False) + if v.is_active: + click.secho(" (active)", fg="green", bold=True, nl=False) + click.echo() diff --git a/pythonlib/camoufox/pkgman.py b/pythonlib/camoufox/pkgman.py index b6f0c3f..46600a9 100644 --- a/pythonlib/camoufox/pkgman.py +++ b/pythonlib/camoufox/pkgman.py @@ -1,7 +1,6 @@ import os import platform import re -import shlex import shutil import sys import tempfile @@ -9,14 +8,23 @@ from dataclasses import dataclass from functools import total_ordering from io import BufferedWriter, BytesIO from pathlib import Path -from typing import Any, Dict, List, Literal, Optional, Tuple, Union +from typing import Any, Callable, Dict, List, Literal, Optional, Tuple, Union from zipfile import ZipFile -import click import orjson import requests from platformdirs import user_cache_dir -from tqdm import tqdm +from rich.console import Console +from rich.progress import ( + BarColumn, + DownloadColumn, + Progress, + SpinnerColumn, + TaskProgressColumn, + TextColumn, + TimeRemainingColumn, + TransferSpeedColumn, +) from typing_extensions import TypeAlias from yaml import CLoader, load @@ -31,7 +39,6 @@ from .exceptions import ( DownloadBuffer: TypeAlias = Union[BytesIO, tempfile._TemporaryFileWrapper, BufferedWriter] -# Map machine architecture to Camoufox binary name ARCH_MAP: Dict[str, str] = { 'amd64': 'x86_64', 'x86_64': 'x86_64', @@ -54,14 +61,12 @@ OS_NAME: Literal['mac', 'win', 'lin'] = OS_MAP[sys.platform] INSTALL_DIR: Path = Path(user_cache_dir("camoufox")) LOCAL_DATA: Path = Path(os.path.abspath(__file__)).parent -# The supported architectures for each OS OS_ARCH_MATRIX: Dict[str, List[str]] = { 'win': ['x86_64', 'i686'], 'mac': ['x86_64', 'arm64'], 'lin': ['x86_64', 'arm64', 'i686'], } -# The relative path to the camoufox executable LAUNCH_FILE = { 'win': 'camoufox.exe', 'mac': '../MacOS/camoufox', @@ -69,33 +74,215 @@ LAUNCH_FILE = { } -def rprint(*a, **k): - click.secho(*a, **k, bold=True) +console = Console() + + +def rprint(msg: str, fg: Optional[str] = None, nl: bool = True) -> None: + """ + Print a styled message + """ + style = f"bold {fg}" if fg else "bold" + console.print(msg, style=style, end="\n" if nl else "", highlight=False) + + +def _parse_semver(version: str) -> Tuple[int, ...]: + """ + Parse a semver string into a comparable tuple + """ + version = version.lstrip('^~') + parts = [] + for part in version.split('.'): + try: + parts.append(int(part)) + except ValueError: + parts.append(0) + while len(parts) < 3: + parts.append(0) + return tuple(parts) + + +def _get_library_version() -> str: + """ + Get the current library version + """ + from importlib.metadata import version + + try: + return version('camoufox') + except Exception: + return '0.0.0' + + +def _find_version_constraints(versions: List[Dict], library_version: str) -> Optional[Dict]: + """ + Find browser build constraints for the current library version. + Each entry has python_library {min, max} and browser {min, max}. + """ + lib_parts = _parse_semver(library_version) + for entry in versions: + py_lib = entry.get('python_library', {}) + lib_min = _parse_semver(py_lib.get('min', '0')) + lib_max = _parse_semver(py_lib.get('max', '999')) + if lib_min <= lib_parts < lib_max: + return entry.get('browser') + return None + + +@dataclass +class RepoConfig: + """ + Configuration for a Camoufox repository + """ + + repo: str + name: str + pattern: str + os_map: Dict[str, str] + arch_map: Dict[str, str] + build_min: Optional[str] = None + build_max: Optional[str] = None + + @staticmethod + def load_repos() -> List['RepoConfig']: + """ + Load repository configurations from repos.yml + """ + repos_path = LOCAL_DATA / 'repos.yml' + with open(repos_path, 'r') as f: + data = load(f, Loader=CLoader) + return [RepoConfig.from_dict(r) for r in data.get('browsers', [])] + + @staticmethod + def get_default_name() -> str: + """ + Get the default repo name from repos.yml + """ + repos_path = LOCAL_DATA / 'repos.yml' + with open(repos_path, 'r') as f: + data = load(f, Loader=CLoader) + return data.get('default', {}).get('browser', 'Official') + + @staticmethod + def from_dict(d: Dict) -> 'RepoConfig': + """ + Create RepoConfig from dictionary + """ + if 'pattern' not in d: + raise ValueError(f"Repo '{d.get('name', 'unknown')}' missing required pattern") + + build_min: Optional[str] = None + build_max: Optional[str] = None + if d.get('versions'): + library_version = _get_library_version() + browser = _find_version_constraints(d['versions'], library_version) + if browser: + build_min = browser.get('min') + build_max = browser.get('max') + + return RepoConfig( + repo=d['repo'], + name=d['name'], + pattern=d['pattern'], + os_map=OS_MAP, + arch_map=ARCH_MAP, + build_min=build_min, + build_max=build_max, + ) + + @staticmethod + def get_default() -> 'RepoConfig': + """ + Get the default repository config + """ + default_name = RepoConfig.get_default_name() + repo = RepoConfig.find_by_name(default_name) + if repo: + return repo + return RepoConfig.load_repos()[0] + + @staticmethod + def find_by_name(name: str) -> Optional['RepoConfig']: + """ + Find a repo config by name (case-insensitive) + """ + name_lower = name.lower() + for repo in RepoConfig.load_repos(): + if repo.name.lower() == name_lower: + return repo + return None + + def get_os_name(self, spoof_os: Optional[str] = None) -> str: + """ + Get the mapped OS name + """ + if spoof_os: + return spoof_os + os_name = self.os_map.get(sys.platform) + if not os_name: + raise UnsupportedOS(f"OS {sys.platform} is not supported") + return os_name + + def get_arch(self, spoof_arch: Optional[str] = None) -> str: + """ + Get the mapped architecture + """ + if spoof_arch: + return spoof_arch + plat_arch = platform.machine().lower() + arch = self.arch_map.get(plat_arch) + if not arch: + raise UnsupportedArchitecture(f"Architecture {plat_arch} is not supported") + return arch + + def build_pattern( + self, spoof_os: Optional[str] = None, spoof_arch: Optional[str] = None + ) -> re.Pattern: + """ + Build asset regex from the config pattern string + """ + replacements = { + 'name': r'(?P\w+)', + 'version': r'(?P[^-]+)', + 'build': r'(?P[^-]+)', + 'os': re.escape(self.get_os_name(spoof_os)), + 'arch': re.escape(self.get_arch(spoof_arch)), + } + pattern = self.pattern.replace('.', r'\.') + regex = re.sub(r'\{(\w+)\}', lambda m: replacements.get(m[1], m[0]), pattern) + return re.compile(regex) + + def is_version_supported(self, version: 'Version') -> bool: + """ + Check if a version is within the repo's supported build range + """ + if self.build_min is None or self.build_max is None: + return True + build_min = Version(build=self.build_min) + build_max = Version(build=self.build_max) + return build_min <= version <= build_max @total_ordering @dataclass class Version: """ - A version string that can be compared to other version strings. - Stores versions up to 5 parts. + A comparable version string (up to 5 parts) """ - release: str + build: str version: Optional[str] = None def __post_init__(self) -> None: - # Build an internal sortable structure self.sorted_rel = tuple( [ - *(int(x) if x.isdigit() else ord(x[0]) - 1024 for x in self.release.split('.')), - *(0 for _ in range(5 - self.release.count('.'))), + *(int(x) if x.isdigit() else ord(x[0]) - 1024 for x in self.build.split('.')), + *(0 for _ in range(5 - self.build.count('.'))), ] ) @property def full_string(self) -> str: - return f"{self.version}-{self.release}" + return f"{self.version}-{self.build}" def __eq__(self, other) -> bool: return self.sorted_rel == other.sorted_rel @@ -109,7 +296,7 @@ class Version: @staticmethod def from_path(path: Optional[Path] = None) -> 'Version': """ - Get the version from the given path. + Get the version from version.json at the given path """ version_path = (path or INSTALL_DIR) / 'version.json' if not os.path.exists(version_path): @@ -119,55 +306,55 @@ class Version: ) with open(version_path, 'rb') as f: version_data = orjson.loads(f.read()) - return Version(**version_data) + if 'release' in version_data: + version_data['build'] = version_data.pop('release') + elif 'tag' in version_data: + version_data['build'] = version_data.pop('tag') + return Version( + build=version_data['build'], + version=version_data.get('version'), + ) @staticmethod def is_supported_path(path: Path) -> bool: """ - Check if the version at the given path is supported. + Check if the version at the given path is supported """ return Version.from_path(path) >= VERSION_MIN @staticmethod def build_minmax() -> Tuple['Version', 'Version']: - return Version(release=CONSTRAINTS.MIN_VERSION), Version(release=CONSTRAINTS.MAX_VERSION) + return Version(build=CONSTRAINTS.MIN_VERSION), Version(build=CONSTRAINTS.MAX_VERSION) -# The minimum and maximum supported versions VERSION_MIN, VERSION_MAX = Version.build_minmax() class GitHubDownloader: """ - Manages fetching and installing GitHub releases. + Manages fetching GitHub releases """ def __init__(self, github_repo: str) -> None: self.github_repo = github_repo self.api_url = f"https://api.github.com/repos/{github_repo}/releases" + self.is_prerelease: bool = False - def check_asset(self, asset: Dict) -> Any: + def check_asset(self, asset: Dict, release: Optional[Dict] = None) -> Any: """ - Compare the asset to determine if it's the desired asset. - - Args: - asset: Asset information from GitHub API - - Returns: - Any: Data to be returned if this is the desired asset, or None/False if not + Return truthy data if this is the desired asset, else None """ return asset.get('browser_download_url') def missing_asset_error(self) -> None: """ - Raise a MissingRelease exception if no release is found. + Raise a MissingRelease exception """ raise MissingRelease(f"Could not find a release asset in {self.github_repo}.") def get_asset(self) -> Any: """ - Fetch the latest release from the GitHub API. - Gets the first asset that returns a truthy value from check_asset. + Fetch the first matching release asset from GitHub """ resp = requests.get(self.api_url, timeout=20) resp.raise_for_status() @@ -176,79 +363,102 @@ class GitHubDownloader: for release in releases: for asset in release['assets']: - if data := self.check_asset(asset): + if data := self.check_asset(asset, release): + self.is_prerelease = release.get('prerelease', False) return data self.missing_asset_error() -class CamoufoxFetcher(GitHubDownloader): +@dataclass +class AvailableVersion: """ - Handles fetching and installing the latest version of Camoufox. + Information about an available Camoufox version from GitHub """ - def __init__(self) -> None: - super().__init__("daijro/camoufox") + version: Version + url: str + is_prerelease: bool + # GitHub metadata for tracking changes + asset_id: Optional[int] = None + asset_size: Optional[int] = None + asset_updated_at: Optional[str] = None + + @property + def display(self) -> str: + """ + Display string for the version + """ + pre = " (prerelease)" if self.is_prerelease else "" + return f"v{self.version.full_string}{pre}" + + def to_metadata(self) -> Dict[str, Any]: + """ + Return metadata dict for storing in version.json + """ + return { + 'version': self.version.version, + 'build': self.version.build, + 'prerelease': self.is_prerelease, + 'asset_id': self.asset_id, + 'asset_size': self.asset_size, + 'asset_updated_at': self.asset_updated_at, + } + + +class CamoufoxFetcher(GitHubDownloader): + """ + Handles fetching and installing Camoufox + """ + + def __init__( + self, + repo_config: Optional[RepoConfig] = None, + selected_version: Optional[AvailableVersion] = None, + ) -> None: + self.repo_config = repo_config or RepoConfig.get_default() + super().__init__(self.repo_config.repo) self.arch = self.get_platform_arch() self._version_obj: Optional[Version] = None - self.pattern: re.Pattern = re.compile( - rf'camoufox-(?P.+)-(?P.+)-{OS_NAME}\.{self.arch}\.zip' - ) + self._selected_version: Optional[AvailableVersion] = None + self.pattern: re.Pattern = self.repo_config.build_pattern() - self.fetch_latest() + if selected_version: + self._selected_version = selected_version + self._version_obj = selected_version.version + self._url = selected_version.url + self.is_prerelease = selected_version.is_prerelease + else: + self.fetch_latest() - def check_asset(self, asset: Dict) -> Optional[Tuple[Version, str]]: + def check_asset( + self, asset: Dict, release: Optional[Dict] = None + ) -> Optional[Tuple[Version, str]]: """ - Finds the latest release from a GitHub releases API response that - supports the Camoufox version constraints, the OS, and architecture. - - Returns: - Optional[Tuple[Version, str]]: The version and URL of a release + Match a release asset against version constraints, OS, and arch """ - # Search through releases for the first supported version match = self.pattern.match(asset['name']) if not match: return None - # Check if the version is supported - version = Version(release=match['release'], version=match['version']) - if not version.is_supported(): + version = Version(build=match['build'], version=match['version']) + if not self.repo_config.is_version_supported(version): return None - # Asset was found. Return data return version, asset['browser_download_url'] def missing_asset_error(self) -> None: - """ - Raise a MissingRelease exception if no release is found. - """ raise MissingRelease( f"No matching release found for {OS_NAME} {self.arch} in the " - f"supported range: ({CONSTRAINTS.as_range()}). " - "Please update the Python library." + f"supported range. Please update the Python library." ) - @staticmethod - def get_platform_arch() -> str: + def get_platform_arch(self) -> str: """ - Get the current platform and architecture information. - - Returns: - str: The architecture of the current platform - - Raises: - UnsupportedArchitecture: If the current architecture is not supported + Get the current platform architecture """ - - # Check if the architecture is supported for the OS - plat_arch = platform.machine().lower() - if plat_arch not in ARCH_MAP: - raise UnsupportedArchitecture(f"Architecture {plat_arch} is not supported") - - arch = ARCH_MAP[plat_arch] - - # Check if the architecture is supported for the OS + arch = self.repo_config.get_arch() if arch not in OS_ARCH_MATRIX[OS_NAME]: raise UnsupportedArchitecture(f"Architecture {arch} is not supported for {OS_NAME}") @@ -256,39 +466,21 @@ class CamoufoxFetcher(GitHubDownloader): def fetch_latest(self) -> None: """ - Fetch the URL of the latest camoufox release for the current platform. - Sets the version, release, and url properties. - - Raises: - requests.RequestException: If there's an error fetching release data - ValueError: If no matching release is found for the current platform + Fetch the latest camoufox release for the current platform """ - release_data = self.get_asset() - - # Set the version and URL - self._version_obj, self._url = release_data + self._version_obj, self._url = self.get_asset() @staticmethod def download_file(file: DownloadBuffer, url: str) -> DownloadBuffer: """ - Download a file from the given URL and return it as BytesIO. - - Args: - file (DownloadBuffer): The buffer to download to - url (str): The URL to download the file from - - Returns: - DownloadBuffer: The downloaded file content as a BytesIO object + Download a file from the given URL """ rprint(f'Downloading package: {url}') return webdl(url, buffer=file) def extract_zip(self, zip_file: DownloadBuffer) -> None: """ - Extract the contents of a zip file to the installation directory. - - Args: - zip_file (DownloadBuffer): The zip file content as a BytesIO object + Extract a zip file to the installation directory """ rprint(f'Extracting Camoufox: {INSTALL_DIR}') unzip(zip_file, str(INSTALL_DIR)) @@ -296,7 +488,7 @@ class CamoufoxFetcher(GitHubDownloader): @staticmethod def cleanup() -> bool: """ - Clean up the old installation. + Clean up the old installation """ if INSTALL_DIR.exists(): rprint(f'Cleaning up cache: {INSTALL_DIR}') @@ -306,153 +498,173 @@ class CamoufoxFetcher(GitHubDownloader): def set_version(self) -> None: """ - Set the version in the INSTALL_DIR/version.json file + Write version.json to INSTALL_DIR """ with open(INSTALL_DIR / 'version.json', 'wb') as f: - f.write(orjson.dumps({'version': self.version, 'release': self.release})) + f.write(orjson.dumps({'version': self.version, 'build': self.build})) - def install(self) -> None: + def install(self, replace: bool = False) -> None: """ - Download and install the latest version of camoufox. - - Raises: - Exception: If any error occurs during the installation process + Download and install camoufox to a versioned subdirectory """ - # Clean up old installation - self.cleanup() - try: - # Install to directory - INSTALL_DIR.mkdir(parents=True, exist_ok=True) + from .multiversion import install_versioned - # Fetch the latest zip - with tempfile.NamedTemporaryFile() as temp_file: - self.download_file(temp_file, self.url) - self.extract_zip(temp_file) - self.set_version() - - # Set permissions on INSTALL_DIR - if OS_NAME != 'win': - os.system(f'chmod -R 755 {shlex.quote(str(INSTALL_DIR))}') # nosec - - rprint('\nCamoufox successfully installed.', fg="yellow") - except Exception as e: - rprint(f"Error installing Camoufox: {str(e)}") - self.cleanup() - raise + install_versioned(self, replace=replace) @property def url(self) -> str: - """ - Url of the fetched latest version of camoufox. - - Returns: - str: The version of the installed camoufox - - Raises: - ValueError: If the version is not available (fetch_latest not ran) - """ if self._url is None: raise ValueError("Url is not available. Make sure to run fetch_latest first.") return self._url @property def version(self) -> str: - """ - Version of the fetched latest version of camoufox. - - Returns: - str: The version of the installed camoufox - - Raises: - ValueError: If the version is not available (fetch_latest not ran) - """ if self._version_obj is None or not self._version_obj.version: raise ValueError("Version is not available. Make sure to run the fetch_latest first.") return self._version_obj.version @property - def release(self) -> str: - """ - Release of the fetched latest version of camoufox. - - Returns: - str: The release of the installed camoufox - - Raises: - ValueError: If the release information is not available (fetch_latest not ran) - """ + def build(self) -> str: if self._version_obj is None: raise ValueError( - "Release information is not available. Make sure to run the installation first." + "Build information is not available. Make sure to run the installation first." ) - return self._version_obj.release + return self._version_obj.build @property def verstr(self) -> str: - """ - Fetches the version and release in version-release format - - Returns: - str: The version of the installed camoufox - """ if self._version_obj is None: raise ValueError("Version is not available. Make sure to run the installation first.") return self._version_obj.full_string +def list_available_versions( + repo_config: Optional[RepoConfig] = None, + include_prerelease: bool = True, + spoof_os: Optional[str] = None, + spoof_arch: Optional[str] = None, +) -> List[AvailableVersion]: + """ + Fetch all supported versions from GitHub for the current platform + """ + config = repo_config or RepoConfig.get_default() + api_url = f"https://api.github.com/repos/{config.repo}/releases" + pattern = config.build_pattern(spoof_os=spoof_os, spoof_arch=spoof_arch) + + os_name = spoof_os or OS_NAME + arch = config.get_arch(spoof_arch) + if arch not in OS_ARCH_MATRIX.get(os_name, []): + raise UnsupportedArchitecture(f"Architecture {arch} is not supported for {os_name}") + + resp = requests.get(api_url, timeout=20) + resp.raise_for_status() + releases = resp.json() + + versions: List[AvailableVersion] = [] + seen_builds: set = set() + + for release in releases: + is_prerelease = release.get('prerelease', False) + + if is_prerelease and not include_prerelease: + continue + + for asset in release['assets']: + match = pattern.match(asset['name']) + if not match: + continue + + version = Version(build=match['build'], version=match['version']) + if not config.is_version_supported(version): + continue + + if version.build in seen_builds: + continue + seen_builds.add(version.build) + + versions.append( + AvailableVersion( + version=version, + url=asset['browser_download_url'], + is_prerelease=is_prerelease, + asset_id=asset.get('id'), + asset_size=asset.get('size'), + asset_updated_at=asset.get('updated_at'), + ) + ) + + versions.sort(key=lambda x: x.version, reverse=True) + return versions + + def installed_verstr() -> str: """ - Get the full version string of the installed camoufox. + Get the full version string of the active install """ - return Version.from_path().full_string + from .multiversion import get_active_path + + active = get_active_path() + return Version.from_path(active).full_string def camoufox_path(download_if_missing: bool = True) -> Path: """ - Full path to the camoufox folder. + Full path to the active camoufox folder """ + from .multiversion import get_active_path + + active = get_active_path() + if active and Version.from_path(active).is_supported(): + return active - # Ensure the directory exists and is not empty if not os.path.exists(INSTALL_DIR) or not os.listdir(INSTALL_DIR): if not download_if_missing: raise FileNotFoundError(f"Camoufox executable not found at {INSTALL_DIR}") - # Camoufox exists and the the version is supported elif os.path.exists(INSTALL_DIR) and Version.from_path().is_supported(): return INSTALL_DIR - # Ensure the version is supported else: if not download_if_missing: raise UnsupportedVersion("Camoufox executable is outdated.") - # Install and recheck CamoufoxFetcher().install() return camoufox_path() def get_path(file: str) -> str: """ - Get the path to the camoufox executable. + Get the path to a file in the camoufox directory """ if OS_NAME == 'mac': return os.path.abspath(camoufox_path() / 'Camoufox.app' / 'Contents' / 'Resources' / file) return str(camoufox_path() / file) -def launch_path() -> str: +def launch_path(browser_path: Optional[Path] = None) -> str: """ - Get the path to the camoufox executable. + Get the path to the camoufox executable """ - launch_path = get_path(LAUNCH_FILE[OS_NAME]) - if not os.path.exists(launch_path): - # Not installed error + if browser_path: + if OS_NAME == 'mac': + exec_path = os.path.abspath( + browser_path / 'Camoufox.app' / 'Contents' / 'Resources' / LAUNCH_FILE[OS_NAME] + ) + else: + exec_path = str(browser_path / LAUNCH_FILE[OS_NAME]) + else: + exec_path = get_path(LAUNCH_FILE[OS_NAME]) + + if not os.path.exists(exec_path): raise CamoufoxNotInstalled( - f"Camoufox is not installed at {camoufox_path()}. Please run `camoufox fetch` to install." + f"Camoufox is not installed at {browser_path or camoufox_path()}. Please run `camoufox fetch` to install." ) - return launch_path + return exec_path + + +ProgressCallback: TypeAlias = 'Callable[[int, int], None]' def webdl( @@ -460,20 +672,10 @@ def webdl( desc: Optional[str] = None, buffer: Optional[DownloadBuffer] = None, bar: bool = True, + progress_callback: Optional[ProgressCallback] = None, ) -> DownloadBuffer: """ - Download a file from the given URL and return it as BytesIO. - - Args: - url (str): The URL to download the file from - buffer (Optional[BytesIO]): A BytesIO object to store the downloaded file - bar (bool): Whether to show the progress bar - - Returns: - DownloadBuffer: The downloaded file content as a BytesIO object - - Raises: - requests.RequestException: If there's an error downloading the file + Download a file from the given URL """ response = requests.get(url, stream=True) response.raise_for_status() @@ -483,16 +685,38 @@ def webdl( if buffer is None: buffer = BytesIO() - with tqdm( - total=total_size, - unit='iB', - bar_format=None if bar else '{desc}: {percentage:3.0f}%', - unit_scale=True, - desc=desc, - ) as progress_bar: + if progress_callback: + downloaded = 0 + last_update = 0 + for data in response.iter_content(block_size * 4): + size = buffer.write(data) + downloaded += size + if downloaded - last_update >= 65536 or downloaded == total_size: + progress_callback(downloaded, total_size) + last_update = downloaded + elif bar: + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + BarColumn(), + DownloadColumn(), + TransferSpeedColumn(), + TimeRemainingColumn(), + console=console, + ) as progress: + task = progress.add_task(desc or "Downloading", total=total_size) + for data in response.iter_content(block_size): + size = buffer.write(data) + progress.update(task, advance=size) + else: + downloaded = 0 for data in response.iter_content(block_size): size = buffer.write(data) - progress_bar.update(size) + downloaded += size + if total_size: + pct = (downloaded / total_size) * 100 + print(f"\r{desc}: {pct:.0f}%", end="", flush=True) + print(f"\r{desc}: Complete" if desc else "") buffer.seek(0) return buffer @@ -505,25 +729,35 @@ def unzip( bar: bool = True, ) -> None: """ - Extract the contents of a zip file to the installation directory. - - Args: - zip_file (BytesIO): The zip file content as a BytesIO object - - Raises: - zipfile.BadZipFile: If the zip file is invalid or corrupted - OSError: If there's an error creating directories or writing files + Extract a zip file to the given path """ with ZipFile(zip_file) as zf: - for member in tqdm( - zf.infolist(), desc=desc, bar_format=None if bar else '{desc}: {percentage:3.0f}%' - ): - zf.extract(member, extract_path) + members = zf.infolist() + if bar: + with Progress( + SpinnerColumn(), + TextColumn("[progress.description]{task.description}"), + BarColumn(), + TaskProgressColumn(), + console=console, + ) as progress: + task = progress.add_task(desc or "Extracting", total=len(members)) + for member in members: + zf.extract(member, extract_path) + progress.update(task, advance=1) + else: + for i, member in enumerate(members): + zf.extract(member, extract_path) + if desc: + pct = ((i + 1) / len(members)) * 100 + print(f"\r{desc}: {pct:.0f}%", end="", flush=True) + if desc: + print(f"\r{desc}: Complete") def load_yaml(file: str) -> Dict[str, Any]: """ - Loads a local YAML file and returns it as a dictionary. + Load a local YAML file as a dictionary """ with open(LOCAL_DATA / file, 'r') as f: return load(f, Loader=CLoader) diff --git a/pythonlib/camoufox/repos.yml b/pythonlib/camoufox/repos.yml new file mode 100644 index 0000000..eb41ce4 --- /dev/null +++ b/pythonlib/camoufox/repos.yml @@ -0,0 +1,55 @@ +# Default configurations +default: + browser: Official + geoip: MaxMind GeoLite2 + +# Browser repositories +browsers: + - repo: daijro/camoufox + name: Official + pattern: "{name}-{version}-{build}-{os}.{arch}.zip" + versions: + - python_library: + min: "0.5.0" + max: "1" + browser: + min: "beta.19" + max: "1" + + - repo: coryking/camoufox + name: CoryKing + pattern: "{name}-{version}-{build}-{os}.{arch}.zip" + versions: + - python_library: + min: "0.5.0" + max: "1" + # assume all browsers + +# GeoIP database repositories +geoip: + # GeoLite2 City - Full city-level data with timezone + - name: MaxMind GeoLite2 + urls: + ipv4: + - https://cdn.jsdelivr.net/npm/@ip-location-db/geolite2-city-mmdb/geolite2-city-ipv4.mmdb + - https://raw.githubusercontent.com/sapics/ip-location-db/refs/heads/main/geolite2-city-mmdb/geolite2-city-ipv4.mmdb + ipv6: + - https://cdn.jsdelivr.net/npm/@ip-location-db/geolite2-city-mmdb/geolite2-city-ipv6.mmdb + - https://raw.githubusercontent.com/sapics/ip-location-db/refs/heads/main/geolite2-city-mmdb/geolite2-city-ipv6.mmdb + paths: + iso_code: country_code + longitude: longitude + latitude: latitude + timezone: timezone + + # GeoIP All-in-One - Combined IPv4/IPv6 with all fields + - name: GeoIP AIO by daijro (experimental) + extract: true + urls: + combined: + - https://github.com/daijro/geoip-all-in-one/releases/latest/download/geoip-aio-all.mmdb.zip + paths: + iso_code: country.iso_code + longitude: location.longitude + latitude: location.latitude + timezone: location.time_zone diff --git a/pythonlib/camoufox/utils.py b/pythonlib/camoufox/utils.py index 7158ba8..fb47955 100644 --- a/pythonlib/camoufox/utils.py +++ b/pythonlib/camoufox/utils.py @@ -23,7 +23,8 @@ from .exceptions import ( ) from .fingerprints import from_browserforge, generate_fingerprint from .ip import Proxy, public_ip, valid_ipv4, valid_ipv6 -from .locale import geoip_allowed, get_geolocation, handle_locales +from .geolocation import geoip_allowed, get_geolocation +from .locale import handle_locales from .pkgman import OS_NAME, get_path, installed_verstr, launch_path from .virtdisplay import VirtualDisplay from .warnings import LeakWarning @@ -344,6 +345,7 @@ def launch_options( disable_coop: Optional[bool] = None, webgl_config: Optional[Tuple[str, str]] = None, geoip: Optional[Union[str, bool]] = None, + geoip_db: Optional[str] = None, humanize: Optional[Union[bool, float]] = None, locale: Optional[Union[str, List[str]]] = None, addons: Optional[List[str]] = None, @@ -357,6 +359,7 @@ def launch_options( headless: Optional[bool] = None, main_world_eval: Optional[bool] = None, executable_path: Optional[Union[str, Path]] = None, + browser: Optional[str] = None, firefox_user_prefs: Optional[Dict[str, Any]] = None, proxy: Optional[Dict[str, str]] = None, enable_cache: Optional[bool] = None, @@ -390,6 +393,9 @@ def launch_options( geoip (Optional[Union[str, bool]]): Calculate longitude, latitude, timezone, country, & locale based on the IP address. Pass the target IP address to use, or `True` to find the IP address automatically. + geoip_db (Optional[str]): + Name of the GeoIP database to use (e.g., "MaxMind"). + If not specified, uses the configured default. humanize (Optional[Union[bool, float]]): Humanize the cursor movement. Takes either `True`, or the MAX duration in seconds of the cursor movement. @@ -426,6 +432,12 @@ def launch_options( To use this, prepend "mw:" to the script: page.evaluate("mw:" + script). executable_path (Optional[Union[str, Path]]): Custom Camoufox browser executable path. + browser (Optional[str]): + Select a specific installed browser version. Can be: + - Repo/build like "official/beta.20" + - Build alone like "beta.20" + - Full version like "134.0.2-beta.20" + If not specified, uses the active version. firefox_user_prefs (Optional[Dict[str, Any]]): Firefox user preferences to set. proxy (Optional[Dict[str, str]]): @@ -560,7 +572,7 @@ def launch_options( elif valid_ipv6(geoip): set_into(config, 'webrtc:ipv6', geoip) - geolocation = get_geolocation(geoip) + geolocation = get_geolocation(geoip, geoip_db=geoip_db) config.update(geolocation.as_config()) # Raise a warning when a proxy is being used without spoofing geolocation. @@ -648,15 +660,27 @@ def launch_options( # Prepare the executable path if executable_path: executable_path = str(executable_path) + elif browser: + # Select a specific installed browser version + from .multiversion import find_installed_version + + browser_path = find_installed_version(browser) + if not browser_path: + raise ValueError( + f"Browser version '{browser}' not found. Run `camoufox list` to see installed versions." + ) + executable_path = launch_path(browser_path) else: executable_path = launch_path() - return { + resp = { "executable_path": executable_path, "args": args, "env": env_vars, "firefox_user_prefs": firefox_user_prefs, - "proxy": proxy, "headless": headless, **(launch_options if launch_options is not None else {}), } + if proxy: + resp["proxy"] = proxy + return resp diff --git a/pythonlib/pyproject.toml b/pythonlib/pyproject.toml index ed74351..6d27d52 100644 --- a/pythonlib/pyproject.toml +++ b/pythonlib/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api" [tool.poetry] name = "camoufox" -version = "0.4.11" +version = "0.5.0" description = "Wrapper around Playwright to help launch Camoufox" authors = ["daijro "] license = "MIT" @@ -28,14 +28,14 @@ classifiers = [ [tool.poetry.dependencies] python = "^3.10" -click = "*" +rich-click = "*" +rich = "*" requests = "*" orjson = "*" browserforge = "^1.2.1" playwright = "*" pyyaml = "*" platformdirs = "*" -tqdm = "*" numpy = "*" ua_parser = "*" typing_extensions = "*" @@ -43,10 +43,13 @@ screeninfo = "*" lxml = "*" language-tags = "*" pysocks = "*" +inquirer = "*" geoip2 = {version = "*", optional = true} +PySide6 = {version = "*", optional = true} [tool.poetry.extras] geoip = ["geoip2"] +gui = ["PySide6"] [tool.poetry.scripts] camoufox = "camoufox.__main__:cli"