mirror of
https://github.com/daijro/camoufox.git
synced 2026-09-09 00:00:39 +00:00
Add version manager in python library
This commit is contained in:
daijro
+749
-82
@@ -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: <repo>/<version>-<build> (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()
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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),
|
||||
)
|
||||
@@ -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']
|
||||
Binary file not shown.
Binary file not shown.
|
After Width: | Height: | Size: 27 KiB |
Executable
BIN
Binary file not shown.
@@ -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 = "<br>".join([
|
||||
f"<b>Country:</b> {geo.locale.region}",
|
||||
f"<b>TZ:</b> {geo.timezone}",
|
||||
f"<b>Lat/Lon:</b> {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)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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.
|
||||
"""
|
||||
|
||||
@@ -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()
|
||||
+444
-210
@@ -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<name>\w+)',
|
||||
'version': r'(?P<version>[^-]+)',
|
||||
'build': r'(?P<build>[^-]+)',
|
||||
'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<version>.+)-(?P<release>.+)-{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)
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
@@ -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 <daijro.dev@gmail.com>"]
|
||||
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"
|
||||
|
||||
Reference in New Issue
Block a user