From 46ba7cd4508822ec4de61ba0aeea47916cf1df84 Mon Sep 17 00:00:00 2001 From: Austin Horstman Date: Fri, 6 Jun 2025 08:26:58 -0500 Subject: [PATCH] yaziPlugins.update: cleanup Don't think we need to be so explicit on a lot of this. Also don't want to support env variable for plugin update. Signed-off-by: Austin Horstman --- pkgs/by-name/ya/yazi/plugins/update.py | 56 ++++++-------------------- 1 file changed, 12 insertions(+), 44 deletions(-) diff --git a/pkgs/by-name/ya/yazi/plugins/update.py b/pkgs/by-name/ya/yazi/plugins/update.py index fa9c0e55431a..999b04b4f079 100755 --- a/pkgs/by-name/ya/yazi/plugins/update.py +++ b/pkgs/by-name/ya/yazi/plugins/update.py @@ -8,7 +8,6 @@ import re import subprocess import sys from pathlib import Path -from typing import Dict, List, Optional, Tuple import requests from packaging import version @@ -26,7 +25,7 @@ def run_command(cmd: str, capture_output: bool = True) -> str: return result.stdout.strip() if capture_output else "" -def get_plugin_info(nixpkgs_dir: str, plugin_name: str) -> Dict[str, str]: +def get_plugin_info(nixpkgs_dir: str, plugin_name: str) -> dict[str, str]: """Get plugin repository information from Nix""" owner = run_command(f"nix eval --raw -f {nixpkgs_dir} yaziPlugins.\"{plugin_name}\".src.owner") repo = run_command(f"nix eval --raw -f {nixpkgs_dir} yaziPlugins.\"{plugin_name}\".src.repo") @@ -43,7 +42,7 @@ def get_yazi_version(nixpkgs_dir: str) -> str: -def get_github_headers() -> Dict[str, str]: +def get_github_headers() -> dict[str, str]: """Create headers for GitHub API requests""" headers = {"Accept": "application/vnd.github.v3+json"} github_token = os.environ.get("GITHUB_TOKEN") @@ -52,7 +51,7 @@ def get_github_headers() -> Dict[str, str]: return headers -def get_default_branch(owner: str, repo: str, headers: Dict[str, str]) -> str: +def get_default_branch(owner: str, repo: str, headers: dict[str, str]) -> str: """Get the default branch name for a GitHub repository""" api_url = f"https://api.github.com/repos/{owner}/{repo}" @@ -66,7 +65,7 @@ def get_default_branch(owner: str, repo: str, headers: Dict[str, str]) -> str: print("Falling back to 'main' as default branch") return "main" -def fetch_plugin_content(owner: str, repo: str, plugin_pname: str, headers: Dict[str, str]) -> str: +def fetch_plugin_content(owner: str, repo: str, plugin_pname: str, headers: dict[str, str]) -> str: """Fetch the plugin's main.lua content from GitHub""" default_branch = get_default_branch(owner, repo, headers) plugin_path = f"{plugin_pname}/" if owner == "yazi-rs" else "" @@ -88,7 +87,6 @@ def check_version_compatibility(plugin_content: str, plugin_name: str, yazi_vers if required_version == "0": print(f"No version requirement found for {plugin_name}, assuming compatible with any Yazi version") else: - # Check if the plugin is compatible with current Yazi version if version.parse(required_version) > version.parse(yazi_version): message = f"{plugin_name} plugin requires Yazi {required_version}, but we have {yazi_version}" print(message) @@ -97,7 +95,7 @@ def check_version_compatibility(plugin_content: str, plugin_name: str, yazi_vers return required_version -def get_latest_commit(owner: str, repo: str, plugin_pname: str, headers: Dict[str, str]) -> Tuple[str, str]: +def get_latest_commit(owner: str, repo: str, plugin_pname: str, headers: dict[str, str]) -> tuple[str, str]: """Get the latest commit hash and date for the plugin""" default_branch = get_default_branch(owner, repo, headers) @@ -135,12 +133,9 @@ def calculate_sri_hash(owner: str, repo: str, latest_commit: str) -> str: try: new_hash = run_command(f"nix-prefetch-url --unpack --type sha256 {prefetch_url} 2>/dev/null") - # If the hash is not in SRI format, convert it if not new_hash.startswith("sha256-"): - # Try to convert the hash to SRI format new_hash = run_command(f"nix hash to-sri --type sha256 {new_hash} 2>/dev/null") - # If that fails, try another approach if not new_hash.startswith("sha256-"): print("Warning: Failed to get SRI hash directly, trying alternative method...") raw_hash = run_command(f"nix-prefetch-url --type sha256 {prefetch_url} 2>/dev/null") @@ -148,7 +143,6 @@ def calculate_sri_hash(owner: str, repo: str, latest_commit: str) -> str: except Exception as e: raise RuntimeError(f"Error calculating hash: {e}") - # Verify we got a valid SRI hash if not new_hash.startswith("sha256-"): raise RuntimeError(f"Failed to generate valid SRI hash. Output was: {new_hash}") @@ -177,17 +171,13 @@ def update_nix_file(default_nix_path: str, latest_commit: str, new_version: str, """Update the default.nix file with new version, revision and hash""" default_nix_content = read_nix_file(default_nix_path) - # Update the revision in default.nix default_nix_content = re.sub(r'rev = "[^"]*"', f'rev = "{latest_commit}"', default_nix_content) - # Update the version in default.nix if 'version = "' in default_nix_content: default_nix_content = re.sub(r'version = "[^"]*"', f'version = "{new_version}"', default_nix_content) else: - # Add version attribute after pname if it doesn't exist default_nix_content = re.sub(r'(pname = "[^"]*";)', f'\\1\n version = "{new_version}";', default_nix_content) - # Update hash in default.nix if 'hash = "' in default_nix_content: default_nix_content = re.sub(r'hash = "[^"]*"', f'hash = "{new_hash}"', default_nix_content) elif 'fetchFromGitHub' in default_nix_content: @@ -195,10 +185,8 @@ def update_nix_file(default_nix_path: str, latest_commit: str, new_version: str, else: raise RuntimeError(f"Could not find hash attribute in {default_nix_path}") - # Write the updated content back to the file write_nix_file(default_nix_path, default_nix_content) - # Verify the hash was updated updated_content = read_nix_file(default_nix_path) if f'hash = "{new_hash}"' in updated_content or f'sha256 = "{new_hash}"' in updated_content: print(f"Successfully updated hash to: {new_hash}") @@ -206,22 +194,18 @@ def update_nix_file(default_nix_path: str, latest_commit: str, new_version: str, raise RuntimeError(f"Failed to update hash in {default_nix_path}") -def get_all_plugins(nixpkgs_dir: str) -> List[Dict[str, str]]: +def get_all_plugins(nixpkgs_dir: str) -> list[dict[str, str]]: """Get all available Yazi plugins from the Nix expression""" try: - # Get all plugin names plugin_names_json = run_command(f'nix eval --impure --json --expr "builtins.attrNames (import {nixpkgs_dir} {{}}).yaziPlugins"') plugin_names = json.loads(plugin_names_json) - # Filter out known non-plugin attributes (like functions and special attributes) excluded_attrs = ["mkYaziPlugin", "override", "overrideDerivation", "overrideAttrs", "recurseForDerivations"] plugin_names = [name for name in plugin_names if name not in excluded_attrs] plugins = [] for name in plugin_names: - # Check if the attribute is a derivation by trying to get its type try: - # First check if it's a derivation by looking for the pname attribute pname = run_command(f'nix eval --raw -f {nixpkgs_dir} "yaziPlugins.{name}.pname"') plugins.append({ "name": name, # Attribute name in yaziPlugins set @@ -236,20 +220,13 @@ def get_all_plugins(nixpkgs_dir: str) -> List[Dict[str, str]]: raise RuntimeError(f"Error getting plugin list: {e}") -def validate_environment(plugin_name: Optional[str] = None, plugin_pname: Optional[str] = None) -> Tuple[str, Optional[str], Optional[str]]: +def validate_environment(plugin_name: str | None = None, plugin_pname: str | None = None) -> tuple[str, str | None, str | None]: """Validate environment variables and paths""" nixpkgs_dir = os.getcwd() - # If plugin name and pname are not provided, check environment variables - if not plugin_name or not plugin_pname: - plugin_name = os.environ.get("PLUGIN_NAME") - plugin_pname = os.environ.get("PLUGIN_PNAME") - - # For single plugin update, we need both name and pname if plugin_name and not plugin_pname: raise RuntimeError(f"pname not provided for plugin {plugin_name}") - # Validate plugin directory if a specific plugin is specified if plugin_name: plugin_dir = f"{nixpkgs_dir}/pkgs/by-name/ya/yazi/plugins/{plugin_name}" if not Path(f"{plugin_dir}/default.nix").exists(): @@ -258,16 +235,15 @@ def validate_environment(plugin_name: Optional[str] = None, plugin_pname: Option return nixpkgs_dir, plugin_name, plugin_pname -def update_single_plugin(nixpkgs_dir: str, plugin_name: str, plugin_pname: str) -> Optional[Dict[str, str]]: +def update_single_plugin(nixpkgs_dir: str, plugin_name: str, plugin_pname: str) -> dict[str, str] | None: """Update a single Yazi plugin Returns: - Dict with update info including old_version, new_version, etc. or None if no change + dict with update info including old_version, new_version, etc. or None if no change """ plugin_dir = f"{nixpkgs_dir}/pkgs/by-name/ya/yazi/plugins/{plugin_name}" default_nix_path = f"{plugin_dir}/default.nix" - # Get repository info nix_content = read_nix_file(default_nix_path) old_version_match = re.search(r'version = "([^"]*)"', nix_content) old_version = old_version_match.group(1) if old_version_match else "unknown" @@ -278,17 +254,13 @@ def update_single_plugin(nixpkgs_dir: str, plugin_name: str, plugin_pname: str) owner = plugin_info["owner"] repo = plugin_info["repo"] - # Get Yazi version separately yazi_version = get_yazi_version(nixpkgs_dir) - # Setup GitHub API headers headers = get_github_headers() - # Check plugin compatibility with current Yazi version plugin_content = fetch_plugin_content(owner, repo, plugin_pname, headers) required_version = check_version_compatibility(plugin_content, plugin_name, yazi_version) - # Get latest commit info latest_commit, commit_date = get_latest_commit(owner, repo, plugin_pname, headers) print(f"Checking {plugin_name} latest commit {latest_commit} ({commit_date})") @@ -298,14 +270,11 @@ def update_single_plugin(nixpkgs_dir: str, plugin_name: str, plugin_pname: str) print(f"Updating {plugin_name} from commit {old_commit} to {latest_commit}") - # Generate new version string new_version = f"{required_version}-unstable-{commit_date}" - # Calculate hash for the plugin new_hash = calculate_sri_hash(owner, repo, latest_commit) print(f"Generated SRI hash: {new_hash}") - # Update the default.nix file update_nix_file(default_nix_path, latest_commit, new_version, new_hash) print(f"Successfully updated {plugin_name} from {old_version} to {new_version}") @@ -319,11 +288,11 @@ def update_single_plugin(nixpkgs_dir: str, plugin_name: str, plugin_pname: str) } -def update_all_plugins(nixpkgs_dir: str) -> List[Dict[str, str]]: +def update_all_plugins(nixpkgs_dir: str) -> list[dict[str, str]]: """Update all available Yazi plugins Returns: - List[Dict[str, str]]: List of successfully updated plugin info dicts + list[dict[str, str]]: List of successfully updated plugin info dicts """ plugins = get_all_plugins(nixpkgs_dir) updated_plugins = [] @@ -366,7 +335,6 @@ def update_all_plugins(nixpkgs_dir: str) -> List[Dict[str, str]]: failed_plugins.append({"name": plugin_name, "error": str(e)}) continue - # Print summary print(f"\n{'=' * 50}") print(f"Update summary: {updated_count} plugins updated out of {checked_count} checked") @@ -383,7 +351,7 @@ def update_all_plugins(nixpkgs_dir: str) -> List[Dict[str, str]]: return updated_plugins -def commit_changes(updated_plugins: List[Dict[str, str]]) -> None: +def commit_changes(updated_plugins: list[dict[str, str]]) -> None: """Commit all changes after updating plugins""" if not updated_plugins: print("No plugins were updated, skipping commit")