commit f0524ed15a6b81e5d7e796528cccfe97bcd8ca42
parent 1daa13e736a049dbb1ea4513a07d72edfe5e8218
Author: Amit Dutta <amitdutta4255@gmail.com>
Date: Thu, 5 Feb 2026 14:40:24 +0530
v0.1.0
Diffstat:
| M | README.md | | | 154 | ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++----------------- |
| A | pyproject.toml | | | 24 | ++++++++++++++++++++++++ |
| A | src/pyrgo/__init__.py | | | 11 | +++++++++++ |
| A | src/pyrgo/cli.py | | | 198 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
| A | src/pyrgo/core.py | | | 209 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
| A | src/pyrgo/network.py | | | 131 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
| A | src/pyrgo/utils.py | | | 184 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
| A | test/test_pyrgo.py | | | 122 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
8 files changed, 1001 insertions(+), 32 deletions(-)
diff --git a/README.md b/README.md
@@ -1,16 +1,24 @@
-# Pyrgo
+# Pyrgo
-> **Execute Python code directly from GitHub, explore remote repositories, and find files instantly.**
+> The Swiss Army Knife for Remote Python Execution.
-Pyrgo is a lightweight CLI tool that turns GitHub into your remote Python interpreter. Whether you want to quickly test a script without cloning a massive repo, inspect a file's content, or map out a project structure, Pyrgo handles it in seconds.
+Pyrgo is a powerful CLI tool and Python library that lets you run, explore, and install Python code directly from GitHub and Gists. It handles dependencies, private repositories, and even turns remote scripts into local command-line tools.
+
+[](https://badge.fury.io/py/pyrgo)
+[](https://opensource.org/licenses/MIT)
## Features
-* **Remote Execution**: Run Python scripts directly from raw GitHub URLs.
-* **Safety First**: Inspect code before execution to ensure no malicious activity.
-* **Repo Exploration**: View the contents of remote folders without leaving your terminal.
-* **Deep Search**: Find specific files inside complex repositories instantly.
-* **Smart Caching**: (Coming Soon) Caches files locally to save bandwidth.
+* **Remote Execution**: Run scripts from GitHub or Gist URLs instantly.
+* **Auto-Dependency Management**: Automatically creates temporary virtual environments and installs missing packages using the `--auto-install` flag.
+* **Private Repo Access**: Authenticate securely with GitHub tokens to access private code and increase API rate limits.
+* **Bookmarks**: Save long URLs as short aliases (e.g., `pyrgo run clean-db`).
+* **Tool Installation**: Install remote scripts as permanent local CLI commands available in your system path.
+* **Recursive Downloads**: Download entire folders or specific sub-directories from a repository.
+* **Interactive Search**: Search for files in a repo and run them immediately from the results.
+* **Smart Caching**: Caches API responses to speed up repeated searches and reduce API usage.
+
+---
## Installation
@@ -18,50 +26,132 @@ Pyrgo is a lightweight CLI tool that turns GitHub into your remote Python interp
pip install pyrgo
```
-## Usage
+---
+
+## CLI Usage
+
+### 1. Run Remote Code
+Execute a script directly from a URL.
+
+**Basic Execution:**
+```bash
+pyrgo run [https://github.com/user/repo/blob/main/script.py](https://github.com/user/repo/blob/main/script.py)
+```
+
+**Run Gists:**
+```bash
+pyrgo run [https://gist.github.com/user/1234567890abcdef](https://gist.github.com/user/1234567890abcdef)
+```
+
+**Auto-Install Dependencies:**
+If a remote script requires packages you do not have installed (e.g., pandas, requests), use this flag to run it in an isolated environment:
+```bash
+pyrgo run [https://github.com/user/repo/blob/main/data.py](https://github.com/user/repo/blob/main/data.py) --auto-install
+```
+
+**Inspect Code:**
+View the source code with syntax highlighting before running it (Safety Check):
+```bash
+pyrgo run [https://github.com/user/repo/blob/main/script.py](https://github.com/user/repo/blob/main/script.py) --inspect
+```
+
+### 2. Authentication (Private Repos & Rate Limits)
+GitHub limits unauthenticated requests to 60 per hour. Login to increase this limit to 5,000 and access private repositories.
+
+```bash
+pyrgo login ghp_YourPersonalAccessToken...
+```
+*The token is stored securely in `~/.pyrgo/config.json`.*
+
+### 3. Bookmarks
+Stop copy-pasting long URLs. Save them once, run them anywhere.
-### 1. Run a Script
-Don't clone the whole repo just to run one file.
+**Add a Bookmark:**
+```bash
+pyrgo bookmark add clean-db [https://github.com/user/repo/blob/main/utils/cleanup.py](https://github.com/user/repo/blob/main/utils/cleanup.py)
+```
+**Run a Bookmark:**
```bash
-# Basic run
-pyrgo run [https://github.com/notamitgamer/adpkg/blob/main/test1.py](https://github.com/notamitgamer/adpkg/blob/main/test1.py)
+pyrgo run clean-db
+```
-# Inspect source code before running (Recommended)
-pyrgo run [https://github.com/notamitgamer/adpkg/blob/main/test1.py](https://github.com/notamitgamer/adpkg/blob/main/test1.py) --inspect
+**List Bookmarks:**
+```bash
+pyrgo bookmark list
```
-### 2. Show Folder Structure
-Visualize what's inside a remote directory.
+### 4. Install as a Tool
+Turn a remote Python script into a command you can run from anywhere in your terminal.
```bash
-pyrgo show [https://github.com/notamitgamer/adpkg/tree/main/subfolder](https://github.com/notamitgamer/adpkg/tree/main/subfolder)
+pyrgo install [https://github.com/user/repo/blob/main/my-tool.py](https://github.com/user/repo/blob/main/my-tool.py) --name mytool
```
-### 3. Find a File
-Searching for a config file or specific logic?
+* **Windows:** Creates a `.bat` file in `~/.pyrgo/bin`.
+* **Linux/Mac:** Creates an executable shim in `~/.pyrgo/bin`.
+* *Note: You must add `~/.pyrgo/bin` to your system PATH.*
+
+### 5. Find & Search
+Search for files inside a remote repository without cloning it.
```bash
-# Syntax: pyrgo find <repo-url> <filename-or-extension>
-pyrgo find [https://github.com/notamitgamer/adpkg](https://github.com/notamitgamer/adpkg) "config.py"
-pyrgo find [https://github.com/notamitgamer/adpkg](https://github.com/notamitgamer/adpkg) ".json"
+# Search for files containing "config"
+pyrgo find [https://github.com/user/repo](https://github.com/user/repo) "config"
```
+*This command is interactive. You can select a result number to run it immediately.*
-## Development
+### 6. Download Files & Folders
+Download artifacts to your local machine.
-We use `poetry` for dependency management.
+**Download a single file:**
+```bash
+pyrgo download [https://github.com/user/repo/blob/main/script.py](https://github.com/user/repo/blob/main/script.py)
+```
-1. Clone the repo.
-2. Install dependencies: `poetry install`
-3. Run the CLI locally: `poetry run pyrgo --help`
+**Download a specific folder (Recursive):**
+```bash
+pyrgo download [https://github.com/user/repo/tree/main/src/utils](https://github.com/user/repo/tree/main/src/utils) --output ./local_utils
+```
-## Contributing
+### 7. Show Folder Contents
+List files in a remote directory to understand the structure.
+
+```bash
+pyrgo show [https://github.com/user/repo/tree/main/src](https://github.com/user/repo/tree/main/src)
+```
+
+---
+
+## Python API Usage
+
+You can use Pyrgo inside your own Python scripts.
+
+```python
+import pyrgo
+
+# 1. Search a repository
+results = pyrgo.search_repository("[https://github.com/user/repo](https://github.com/user/repo)", "test")
+for item in results:
+ print(item['path'], item['raw_url'])
+
+# 2. Download a file
+pyrgo.download_file("[https://github.com/user/repo/blob/main/script.py](https://github.com/user/repo/blob/main/script.py)", output_path="script.py")
+
+# 3. Download a full folder
+pyrgo.download_folder("[https://github.com/user/repo/tree/main/src](https://github.com/user/repo/tree/main/src)")
+
+# 4. Execute code programmatically
+exit_code = pyrgo.execute_remote_code("[https://github.com/user/repo/blob/main/script.py](https://github.com/user/repo/blob/main/script.py)", args=["--verbose"])
+```
-Contributions are welcome! Please read our [Contributing Guide](CONTRIBUTING.md) to get started.
+## Configuration
-## Security
+Pyrgo stores configuration and cache files in your home directory:
-Found a bug? Please check our [Security Policy](SECURITY.md).
+* **Config:** `~/.pyrgo/config.json` (Tokens, Bookmarks)
+* **Cache:** `~/.pyrgo/cache/` (API responses)
+* **Binaries:** `~/.pyrgo/bin/` (Installed tools)
## License
diff --git a/pyproject.toml b/pyproject.toml
@@ -0,0 +1,23 @@
+[tool.poetry]
+name = "pyrgo"
+version = "0.1.0"
+description = "A CLI tool to run Python scripts directly from GitHub URLs"
+authors = ["Amit Dutta <mail@amit.is-a.dev>"]
+readme = "README.md"
+packages = [{include = "pyrgo", from = "src"}]
+
+[tool.poetry.dependencies]
+python = "^3.9"
+# Cleaned up typer definition (removed deprecated 'all' extra)
+typer = "^0.12.0"
+# Pin click to <8.2.0 to prevent crash with newer API changes
+click = "<8.2.0"
+requests = "^2.31.0"
+rich = "^13.7.0"
+
+[tool.poetry.scripts]
+pyrgo = "pyrgo.cli:app"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"+
\ No newline at end of file
diff --git a/src/pyrgo/__init__.py b/src/pyrgo/__init__.py
@@ -0,0 +1,10 @@
+__version__ = "0.1.0"
+
+from .core import (
+ execute_remote_code,
+ search_repository,
+ get_folder_contents,
+ download_file,
+ download_folder,
+ install_tool
+)+
\ No newline at end of file
diff --git a/src/pyrgo/cli.py b/src/pyrgo/cli.py
@@ -0,0 +1,197 @@
+import typer
+from rich.console import Console
+from rich.syntax import Syntax
+from rich.table import Table
+from rich.prompt import Prompt
+from typing import Optional
+
+from .core import (
+ execute_remote_code,
+ search_repository,
+ get_folder_contents,
+ download_file,
+ download_folder,
+ install_tool,
+ login_github,
+ add_bookmark,
+ list_bookmarks
+)
+from .network import RateLimitError, fetch_url_content, fetch_gist_content, convert_to_raw_url
+from .utils import print_error, print_info, print_warning, print_success, ConfigManager
+
+app = typer.Typer(help="Pyrgo: Run Python code from GitHub instantly.")
+bookmark_app = typer.Typer(help="Manage bookmarks.")
+app.add_typer(bookmark_app, name="bookmark")
+
+console = Console()
+
+# --- MAIN COMMANDS ---
+
+@app.command()
+def login(token: str = typer.Argument(..., help="Your GitHub Personal Access Token.")):
+ """
+ Authenticate with GitHub to increase rate limits and access private repos.
+ """
+ login_github(token)
+ print_success("GitHub token saved successfully!")
+
+@app.command()
+def install(
+ url: str = typer.Argument(..., help="URL of the script."),
+ name: str = typer.Option(..., "--name", "-n", help="Name of the command tool.")
+):
+ """
+ Install a remote script as a local command-line tool.
+ """
+ try:
+ msg = install_tool(url, name)
+ print_success(msg)
+ except Exception as e:
+ print_error(str(e))
+
+@app.command()
+def run(
+ url: str = typer.Argument(..., help="GitHub URL, Gist URL, or Bookmark Name."),
+ inspect: bool = typer.Option(False, "--inspect", "-i", help="Print code without running."),
+ yes: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation."),
+ auto_install: bool = typer.Option(False, "--auto-install", help="Auto-install missing dependencies.")
+):
+ """Download and execute a remote Python script."""
+ try:
+ # Check bookmarks
+ b_url = ConfigManager.get_bookmark(url)
+ if b_url:
+ print_info(f"Resolved bookmark '{url}' to: {b_url}")
+ url = b_url
+
+ if inspect:
+ raw = convert_to_raw_url(url)
+ if raw.startswith("gist:"):
+ content = fetch_gist_content(raw.split(":")[1])
+ else:
+ content = fetch_url_content(raw)
+ if content:
+ console.print(Syntax(content, "python", theme="monokai", line_numbers=True))
+ return
+
+ if not yes:
+ console.print(f"\n[bold yellow]SECURITY WARNING:[/bold yellow] Running remote code from:")
+ console.print(f"[underline]{url}[/underline]\n")
+ if not typer.confirm("Execute this script?"):
+ raise typer.Exit()
+
+ execute_remote_code(url, auto_install=auto_install)
+
+ except RateLimitError:
+ print_error("GitHub API Rate Limit Exceeded (60 reqs/hr).")
+ if typer.confirm("Would you like to add an API Key to increase limits?"):
+ token = Prompt.ask("Enter GitHub Token")
+ login_github(token)
+ print_success("Token saved! Please try running the command again.")
+ except Exception as e:
+ print_error(str(e))
+ raise typer.Exit(1)
+
+# --- BOOKMARK COMMANDS ---
+
+@bookmark_app.command("add")
+def bookmark_add(name: str, url: str):
+ """Save a URL as a bookmark."""
+ add_bookmark(name, url)
+ print_success(f"Bookmark '{name}' added.")
+
+@bookmark_app.command("list")
+def bookmark_list():
+ """List all saved bookmarks."""
+ bookmarks = list_bookmarks()
+ if not bookmarks:
+ print_warning("No bookmarks found.")
+ return
+
+ table = Table(title="Bookmarks")
+ table.add_column("Name", style="cyan")
+ table.add_column("URL", style="green")
+ for name, url in bookmarks.items():
+ table.add_row(name, url)
+ console.print(table)
+
+# --- EXISTING COMMANDS (UPDATED) ---
+
+@app.command()
+def find(
+ repo_url: str = typer.Argument(..., help="GitHub repository URL."),
+ query: str = typer.Argument(..., help="Search query (filename)."),
+ interactive: bool = typer.Option(True, help="Enable interactive selection.")
+):
+ """Search for files in a repo."""
+ try:
+ with console.status("Scanning..."):
+ from .core import search_repository
+ results = search_repository(repo_url, query)
+
+ if not results:
+ print_warning("No results.")
+ return
+
+ table = Table(title=f"Results for {query}")
+ table.add_column("#", style="yellow")
+ table.add_column("Path", style="cyan")
+
+ for idx, r in enumerate(results):
+ table.add_row(str(idx+1), r['path'])
+ console.print(table)
+
+ if interactive:
+ choice = Prompt.ask("Select file # to Run (or 'q')")
+ if choice.isdigit():
+ idx = int(choice) - 1
+ if 0 <= idx < len(results):
+ execute_remote_code(results[idx]['raw_url'])
+
+ except RateLimitError:
+ print_error("Rate Limit Hit.")
+ print_info("Use 'pyrgo login <token>' to fix this.")
+ except Exception as e:
+ print_error(str(e))
+
+@app.command()
+def download(
+ url: str = typer.Argument(..., help="GitHub file or folder URL."),
+ output: Optional[str] = typer.Option(None, "--output", "-o", help="Output path.")
+):
+ """Download file or folder."""
+ try:
+ from .core import download_folder, download_file
+ if "/tree/" in url:
+ path = download_folder(url, output)
+ print_success(f"Downloaded folder: {path}")
+ else:
+ path = download_file(url, output)
+ print_success(f"Downloaded file: {path}")
+ except Exception as e:
+ print_error(str(e))
+
+@app.command()
+def show(url: str = typer.Argument(..., help="GitHub folder URL.")):
+ """Show folder contents."""
+ try:
+ from .core import get_folder_contents
+ items = get_folder_contents(url)
+
+ if not items:
+ print_error("Empty folder or invalid URL.")
+ return
+
+ table = Table(title="Contents")
+ table.add_column("Name")
+ table.add_column("Type")
+
+ for item in items:
+ itype = "Dir" if item['type'] == 'dir' else "File"
+ table.add_row(item['name'], itype)
+ console.print(table)
+ except Exception as e:
+ print_error(str(e))
+
+if __name__ == "__main__":
+ app()+
\ No newline at end of file
diff --git a/src/pyrgo/core.py b/src/pyrgo/core.py
@@ -0,0 +1,208 @@
+import sys
+import subprocess
+import os
+import re
+import platform
+import stat
+from pathlib import Path
+from typing import List, Dict, Optional
+from .network import (
+ convert_to_raw_url,
+ fetch_url_content,
+ fetch_gist_content,
+ get_repo_details,
+ fetch_tree_recursively,
+ fetch_folder_contents
+)
+from .utils import (
+ temp_python_file,
+ check_package_installed,
+ ConfigManager,
+ VenvManager,
+ BIN_DIR
+)
+
+def resolve_url(url: str) -> str:
+ """Resolves bookmarks to full URLs."""
+ bookmark = ConfigManager.get_bookmark(url)
+ if bookmark:
+ return bookmark
+ return url
+
+def execute_remote_code(url: str, args: List[str] = None, auto_install: bool = False) -> int:
+ """Downloads and executes a Python script."""
+
+ full_url = resolve_url(url)
+ raw_url = convert_to_raw_url(full_url)
+
+ if raw_url.startswith("gist:"):
+ code_content = fetch_gist_content(raw_url.split(":")[1])
+ else:
+ code_content = fetch_url_content(raw_url)
+
+ if not code_content:
+ raise ValueError(f"Could not retrieve content from {full_url}")
+
+ # Dependency Check
+ missing_deps = scan_dependencies(code_content)
+
+ if auto_install and missing_deps:
+ # --- VIRTUAL ENV EXECUTION FLOW ---
+ manager = VenvManager()
+ try:
+ manager.create()
+ manager.install(missing_deps)
+
+ # Save code to a file inside the temp dir (so it's near the venv)
+ script_path = manager.venv_dir / "remote_script.py"
+ with open(script_path, "w", encoding="utf-8") as f:
+ f.write(code_content)
+
+ cmd = [str(manager.python_exe), str(script_path)]
+ if args:
+ cmd.extend(args)
+
+ result = subprocess.run(cmd, capture_output=False)
+ return result.returncode
+ finally:
+ manager.cleanup()
+ else:
+ # --- STANDARD EXECUTION FLOW ---
+ if missing_deps:
+ print(f"\033[93m[Warning] Missing packages: {', '.join(missing_deps)}. Use --auto-install to fix.\033[0m")
+
+ with temp_python_file(code_content) as temp_file:
+ cmd = [sys.executable, temp_file]
+ if args:
+ cmd.extend(args)
+ result = subprocess.run(cmd, capture_output=False)
+ return result.returncode
+
+def install_tool(url: str, name: str) -> str:
+ """Installs a remote script as a local command."""
+ full_url = resolve_url(url)
+ raw_url = convert_to_raw_url(full_url)
+ content = fetch_url_content(raw_url)
+
+ if not content:
+ raise ValueError("Failed to download script.")
+
+ target_path = BIN_DIR / f"{name}.py"
+ with open(target_path, "w", encoding="utf-8") as f:
+ f.write(content)
+
+ # Create executable shim
+ if platform.system() == "Windows":
+ bat_path = BIN_DIR / f"{name}.bat"
+ with open(bat_path, "w") as f:
+ f.write(f'@echo off\n"{sys.executable}" "{target_path}" %*')
+ final_msg = f"Installed {name} to {BIN_DIR}.\nAdd this folder to your PATH to run it via `{name}`."
+ else:
+ # Linux/Mac
+ shim_path = BIN_DIR / name
+ with open(shim_path, "w") as f:
+ f.write(f'#!/bin/sh\n"{sys.executable}" "{target_path}" "$@"')
+ st = os.stat(shim_path)
+ os.chmod(shim_path, st.st_mode | stat.S_IEXEC)
+ final_msg = f"Installed {name} to {BIN_DIR}.\nAdd this folder to your PATH."
+
+ return final_msg
+
+def scan_dependencies(code: str) -> List[str]:
+ imports = set()
+ imports.update(re.findall(r'^\s*import\s+(\w+)', code, re.MULTILINE))
+ imports.update(re.findall(r'^\s*from\s+(\w+)', code, re.MULTILINE))
+
+ missing = []
+ for imp in imports:
+ if not check_package_installed(imp):
+ missing.append(imp)
+ return missing
+
+def search_repository(repo_url: str, query: str) -> List[Dict[str, str]]:
+ owner, repo = get_repo_details(repo_url)
+ if not owner: raise ValueError("Invalid Repo URL")
+
+ # Smart branch detection happens in network.py
+ data = fetch_tree_recursively(owner, repo)
+ if not data or "tree" not in data:
+ return []
+
+ results = []
+ query_lower = query.lower()
+
+ # Try to determine branch from data url or default to main for raw links
+ branch = "main"
+
+ for item in data["tree"]:
+ path = item["path"]
+ if query_lower in path.lower():
+ raw_link = f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}/{path}"
+ results.append({"path": path, "type": item["type"], "raw_url": raw_link})
+ return results
+
+def get_folder_contents(url: str) -> List[Dict]:
+ items = fetch_folder_contents(url)
+ return items if items else []
+
+def download_file(url: str, output_path: Optional[str] = None) -> str:
+ full_url = resolve_url(url)
+ raw_url = convert_to_raw_url(full_url)
+
+ if raw_url.startswith("gist:"):
+ content = fetch_gist_content(raw_url.split(":")[1])
+ filename = "gist_script.py"
+ else:
+ content = fetch_url_content(raw_url)
+ filename = raw_url.split("/")[-1]
+
+ if not content:
+ raise ValueError("Failed to download.")
+
+ if not output_path:
+ output_path = filename
+
+ with open(output_path, "w", encoding="utf-8") as f:
+ f.write(content)
+ return output_path
+
+def download_folder(url: str, output_dir: Optional[str] = None) -> str:
+ # Basic logic reused from previous step, ensuring resolve_url is called if we support folder bookmarks
+ full_url = resolve_url(url)
+ # ... (Rest of logic is same as previous, omitted for brevity but assumed present)
+ # Re-implementing specifically for clarity:
+ owner, repo = get_repo_details(full_url)
+ parts = urlparse(full_url).path.strip("/").split("/")
+ if len(parts) < 5 or parts[2] != "tree":
+ raise ValueError("Invalid tree URL")
+
+ target_path = "/".join(parts[4:])
+ if not output_dir: output_dir = parts[-1]
+
+ data = fetch_tree_recursively(owner, repo) # branch auto-detected
+ if not data: raise ValueError("Could not fetch tree")
+
+ count = 0
+ os.makedirs(output_dir, exist_ok=True)
+ for item in data["tree"]:
+ if item['path'].startswith(target_path) and item['type'] == 'blob':
+ rel = item['path'][len(target_path):].strip("/")
+ local = os.path.join(output_dir, rel)
+ os.makedirs(os.path.dirname(local), exist_ok=True)
+ # Fetch
+ raw = f"https://raw.githubusercontent.com/{owner}/{repo}/main/{item['path']}"
+ c = fetch_url_content(raw)
+ if c:
+ with open(local, "w", encoding="utf-8") as f: f.write(c)
+ count += 1
+ return f"{output_dir} ({count} files)"
+
+# Config Wrappers for CLI
+def login_github(token: str):
+ ConfigManager.set_api_key(token)
+
+def add_bookmark(name: str, url: str):
+ ConfigManager.add_bookmark(name, url)
+
+def list_bookmarks():
+ return ConfigManager.list_bookmarks()+
\ No newline at end of file
diff --git a/src/pyrgo/network.py b/src/pyrgo/network.py
@@ -0,0 +1,130 @@
+import requests
+from urllib.parse import urlparse
+from .utils import print_error, load_cache, save_cache, ConfigManager
+
+class RateLimitError(Exception):
+ pass
+
+def get_auth_headers():
+ token = ConfigManager.get_api_key()
+ if token:
+ return {"Authorization": f"token {token}", "Accept": "application/vnd.github.v3+json"}
+ return {"Accept": "application/vnd.github.v3+json"}
+
+def convert_to_raw_url(url: str) -> str:
+ """
+ Converts GitHub/Gist URLs to raw content URLs.
+ Supports bookmarks via ConfigManager in core.py, not here.
+ """
+ parsed = urlparse(url)
+
+ if "gist.github.com" in parsed.netloc:
+ parts = parsed.path.strip("/").split("/")
+ if len(parts) >= 2:
+ return f"gist:{parts[1]}"
+ return url
+
+ if "github.com" in parsed.netloc:
+ # Standard: user/repo/blob/BRANCH/path
+ parts = parsed.path.strip("/").split("/")
+ if len(parts) >= 4 and parts[2] == "blob":
+ # Reconstruction for raw.githubusercontent
+ # Format: user/repo/BRANCH/path
+ new_path = "/".join([parts[0], parts[1]] + parts[3:])
+ return f"https://raw.githubusercontent.com/{new_path}"
+
+ return url
+
+def fetch_url_content(url: str) -> str:
+ try:
+ # For raw.githubusercontent.com, headers usually aren't needed but auth helps with private repos
+ headers = get_auth_headers()
+ response = requests.get(url, headers=headers)
+
+ if response.status_code == 404 and "Authorization" in headers:
+ # Fallback: Sometimes raw links for private repos behave differently
+ # We might need to use the API to get download_url if this fails
+ pass
+
+ response.raise_for_status()
+ return response.text
+ except requests.RequestException as e:
+ if e.response is not None and e.response.status_code == 404:
+ print_error("File not found (404). If this is a private repo, ensure you are logged in.")
+ else:
+ print_error(f"Failed to download: {e}")
+ return None
+
+def fetch_gist_content(gist_id: str):
+ api_url = f"https://api.github.com/gists/{gist_id}"
+ data = _fetch_api(api_url)
+ if data:
+ files = data.get("files", {})
+ if files:
+ return list(files.values())[0].get("content")
+ return None
+
+def get_repo_details(url: str):
+ parts = urlparse(url).path.strip("/").split("/")
+ if len(parts) >= 2:
+ return parts[0], parts[1]
+ return None, None
+
+def fetch_tree_recursively(owner: str, repo: str, branch: str = None):
+ cache_key = f"{owner}_{repo}_{branch or 'default'}_tree"
+
+ cached = load_cache(cache_key)
+ if cached: return cached
+
+ # Determine branch if not provided
+ if not branch:
+ # Try to guess default branch
+ repo_info = _fetch_api(f"https://api.github.com/repos/{owner}/{repo}")
+ if repo_info:
+ branch = repo_info.get("default_branch", "main")
+ else:
+ branch = "main"
+
+ url = f"https://api.github.com/repos/{owner}/{repo}/git/trees/{branch}?recursive=1"
+ data = _fetch_api(url)
+
+ if data:
+ save_cache(cache_key, data)
+ return data
+ return None
+
+def fetch_folder_contents(url: str):
+ parsed = urlparse(url)
+ parts = parsed.path.strip("/").split("/")
+
+ # Check for branch in URL: github.com/user/repo/tree/BRANCH/path
+ if len(parts) < 4 or parts[2] != "tree":
+ # Root
+ if len(parts) == 2:
+ return _fetch_api(f"https://api.github.com/repos/{parts[0]}/{parts[1]}/contents")
+ return None
+
+ owner, repo, branch = parts[0], parts[1], parts[3]
+ path = "/".join(parts[4:])
+
+ api_url = f"https://api.github.com/repos/{owner}/{repo}/contents/{path}?ref={branch}"
+ return _fetch_api(api_url)
+
+def _fetch_api(url):
+ try:
+ headers = get_auth_headers()
+ resp = requests.get(url, headers=headers)
+
+ if resp.status_code == 403:
+ # Check rate limit
+ limit = resp.headers.get("X-RateLimit-Remaining")
+ if limit == "0":
+ raise RateLimitError("GitHub API Rate Limit Exceeded.")
+
+ if resp.status_code == 200:
+ return resp.json()
+ return None
+ except RateLimitError:
+ raise
+ except Exception:
+ return None+
\ No newline at end of file
diff --git a/src/pyrgo/utils.py b/src/pyrgo/utils.py
@@ -0,0 +1,183 @@
+import os
+import sys
+import tempfile
+import json
+import time
+import shutil
+import venv
+import platform
+import subprocess
+import importlib.util
+from pathlib import Path
+from typing import Optional, Dict
+from contextlib import contextmanager
+from rich.console import Console
+
+console = Console()
+
+# Paths
+APP_DIR = Path.home() / ".pyrgo"
+CACHE_DIR = APP_DIR / "cache"
+BIN_DIR = APP_DIR / "bin"
+CONFIG_FILE = APP_DIR / "config.json"
+CACHE_DURATION = 600 # 10 minutes
+
+def ensure_dirs():
+ APP_DIR.mkdir(exist_ok=True)
+ CACHE_DIR.mkdir(exist_ok=True)
+ BIN_DIR.mkdir(exist_ok=True)
+
+class ConfigManager:
+ """Manages persistent configuration (Tokens, Bookmarks)."""
+
+ @staticmethod
+ def load() -> Dict:
+ ensure_dirs()
+ if not CONFIG_FILE.exists():
+ return {"api_key": None, "bookmarks": {}}
+ try:
+ with open(CONFIG_FILE, "r") as f:
+ return json.load(f)
+ except:
+ return {"api_key": None, "bookmarks": {}}
+
+ @staticmethod
+ def save(data: Dict):
+ ensure_dirs()
+ with open(CONFIG_FILE, "w") as f:
+ json.dump(data, f, indent=4)
+
+ @staticmethod
+ def get_api_key() -> Optional[str]:
+ # 1. Check Environment Variable
+ if os.environ.get("GITHUB_TOKEN"):
+ return os.environ["GITHUB_TOKEN"]
+
+ # 2. Check .env file in current directory
+ env_path = Path.cwd() / ".env"
+ if env_path.exists():
+ try:
+ with open(env_path, "r") as f:
+ for line in f:
+ if line.strip().startswith("GITHUB_TOKEN="):
+ return line.split("=", 1)[1].strip().strip('"').strip("'")
+ except:
+ pass
+
+ # 3. Check Global Config
+ config = ConfigManager.load()
+ return config.get("api_key")
+
+ @staticmethod
+ def set_api_key(key: str):
+ config = ConfigManager.load()
+ config["api_key"] = key
+ ConfigManager.save(config)
+
+ @staticmethod
+ def add_bookmark(name: str, url: str):
+ config = ConfigManager.load()
+ if "bookmarks" not in config:
+ config["bookmarks"] = {}
+ config["bookmarks"][name] = url
+ ConfigManager.save(config)
+
+ @staticmethod
+ def get_bookmark(name: str) -> Optional[str]:
+ config = ConfigManager.load()
+ return config.get("bookmarks", {}).get(name)
+
+ @staticmethod
+ def list_bookmarks() -> Dict[str, str]:
+ return ConfigManager.load().get("bookmarks", {})
+
+# Cache Logic
+def load_cache(key: str):
+ ensure_dirs()
+ cache_file = CACHE_DIR / f"{key}.json"
+ if cache_file.exists():
+ try:
+ with open(cache_file, "r", encoding="utf-8") as f:
+ data = json.load(f)
+ if time.time() - data["timestamp"] < CACHE_DURATION:
+ return data["content"]
+ except:
+ pass
+ return None
+
+def save_cache(key: str, content: any):
+ ensure_dirs()
+ cache_file = CACHE_DIR / f"{key}.json"
+ try:
+ with open(cache_file, "w", encoding="utf-8") as f:
+ json.dump({"timestamp": time.time(), "content": content}, f)
+ except:
+ pass
+
+# Dependency & Venv Logic
+def check_package_installed(package_name: str) -> bool:
+ if package_name in sys.builtin_module_names:
+ return True
+ try:
+ return importlib.util.find_spec(package_name) is not None
+ except Exception:
+ return False
+
+class VenvManager:
+ """Manages temporary virtual environments for auto-installing dependencies."""
+
+ def __init__(self):
+ self.temp_dir = tempfile.mkdtemp(prefix="pyrgo_env_")
+ self.venv_dir = Path(self.temp_dir)
+ self.python_exe = self._get_python_exe()
+ self.pip_exe = self._get_pip_exe()
+
+ def _get_python_exe(self):
+ if platform.system() == "Windows":
+ return self.venv_dir / "Scripts" / "python.exe"
+ return self.venv_dir / "bin" / "python"
+
+ def _get_pip_exe(self):
+ if platform.system() == "Windows":
+ return self.venv_dir / "Scripts" / "pip.exe"
+ return self.venv_dir / "bin" / "pip"
+
+ def create(self):
+ """Creates the virtual environment."""
+ console.print("[blue]Creating temporary virtual environment...[/blue]")
+ venv.create(self.venv_dir, with_pip=True)
+
+ def install(self, packages: list):
+ """Installs packages into the venv."""
+ if not packages:
+ return
+ console.print(f"[blue]Installing dependencies: {', '.join(packages)}...[/blue]")
+ subprocess.check_call([str(self.pip_exe), "install"] + packages, stdout=subprocess.DEVNULL)
+
+ def cleanup(self):
+ """Deletes the virtual environment."""
+ shutil.rmtree(self.temp_dir, ignore_errors=True)
+
+# Standard Helpers
+def print_error(msg: str):
+ console.print(f"[bold red]Error:[/bold red] {msg}")
+
+def print_info(msg: str):
+ console.print(f"[bold blue]Info:[/bold blue] {msg}")
+
+def print_success(msg: str):
+ console.print(f"[bold green]Success:[/bold green] {msg}")
+
+def print_warning(msg: str):
+ console.print(f"[bold yellow]Warning:[/bold yellow] {msg}")
+
+@contextmanager
+def temp_python_file(content: str):
+ fd, path = tempfile.mkstemp(suffix=".py", text=True)
+ try:
+ with os.fdopen(fd, 'w', encoding='utf-8') as tmp:
+ tmp.write(content)
+ yield path
+ finally:
+ if os.path.exists(path):
+ os.remove(path)+
\ No newline at end of file
diff --git a/test/test_pyrgo.py b/test/test_pyrgo.py
@@ -0,0 +1,121 @@
+import os
+import sys
+import shutil
+import pyrgo
+from rich.console import Console
+from rich.panel import Panel
+
+console = Console()
+
+# --- CONFIGURATION ---
+# We use the repo you mentioned in the chat for testing
+TEST_REPO = "https://github.com/notamitgamer/adpkg"
+TEST_FILE = "https://github.com/notamitgamer/adpkg/blob/main/test1.py"
+TEST_FOLDER = "https://github.com/notamitgamer/adpkg/tree/main/"
+DOWNLOAD_TARGET = "test_download_artifact.py"
+
+def print_header(title):
+ console.print(Panel(f"[bold white]{title}[/bold white]", style="bold blue"))
+
+def test_cli_command(cmd, description):
+ console.print(f"\n[bold cyan]🔹 Testing CLI: {description}[/bold cyan]")
+ console.print(f"[dim]Command: {cmd}[/dim]")
+
+ exit_code = os.system(cmd)
+
+ if exit_code == 0:
+ console.print("[bold green]✔ Passed[/bold green]")
+ return True
+ else:
+ console.print("[bold red]✘ Failed[/bold red]")
+ return False
+
+def test_python_api():
+ print_header("Testing Python Source Code API")
+
+ # 1. Search Repository
+ console.print("\n[bold cyan]🔹 API: pyrgo.search_repository()[/bold cyan]")
+ try:
+ results = pyrgo.search_repository(TEST_REPO, "test1.py")
+ if results:
+ console.print(f"[green]✔ Success: Found {len(results)} matches.[/green]")
+ for item in results[:1]: # Show first match
+ console.print(f" Found: {item['path']} ({item['raw_url']})")
+ else:
+ console.print("[yellow]âš Warning: No results found (API limit might be hit).[/yellow]")
+ except Exception as e:
+ console.print(f"[bold red]✘ Error:[/bold red] {e}")
+
+ # 2. Get Folder Contents
+ console.print("\n[bold cyan]🔹 API: pyrgo.get_folder_contents()[/bold cyan]")
+ try:
+ items = pyrgo.get_folder_contents(TEST_FOLDER)
+ if items:
+ console.print(f"[green]✔ Success: Fetched {len(items)} items from folder.[/green]")
+ else:
+ console.print("[yellow]âš Warning: Folder appears empty or API error.[/yellow]")
+ except Exception as e:
+ console.print(f"[bold red]✘ Error:[/bold red] {e}")
+
+ # 3. Download File
+ console.print("\n[bold cyan]🔹 API: pyrgo.download_file()[/bold cyan]")
+ try:
+ path = pyrgo.download_file(TEST_FILE, output_path="api_download_test.py")
+ if os.path.exists(path):
+ console.print(f"[green]✔ Success: File downloaded to {path}[/green]")
+ os.remove(path) # Cleanup
+ else:
+ console.print("[bold red]✘ Failed: File not found after download call.[/bold red]")
+ except Exception as e:
+ console.print(f"[bold red]✘ Error:[/bold red] {e}")
+
+def main():
+ # Ensure we are in the right directory or package is installed
+ if not shutil.which("pyrgo"):
+ console.print("[bold red]CRITICAL: 'pyrgo' command not found.[/bold red]")
+ console.print("Please run [bold]pip install .[/bold] in the root directory first.")
+ return
+
+ print_header("Starting Pyrgo Feature Tests")
+
+ # --- CLI TESTS ---
+
+ # 1. Help Command
+ test_cli_command("pyrgo --help", "Help Menu")
+
+ # 2. Run Command (Non-interactive)
+ # --yes skips the confirmation prompt
+ test_cli_command(f"pyrgo run {TEST_FILE} --yes", "Run Remote Script")
+
+ # 3. Inspect Mode
+ # Should print code but not run it
+ test_cli_command(f"pyrgo run {TEST_FILE} --inspect", "Inspect Code")
+
+ # 4. Show Folder
+ test_cli_command(f"pyrgo show {TEST_FOLDER}", "Show Folder Contents")
+
+ # 5. Find File (Non-interactive)
+ # Typer converts `interactive=True` to a flag `--no-interactive`
+ test_cli_command(f"pyrgo find {TEST_REPO} test --no-interactive", "Find File")
+
+ # 6. Download
+ test_cli_command(f"pyrgo download {TEST_FILE} -o {DOWNLOAD_TARGET}", "Download File")
+ if os.path.exists(DOWNLOAD_TARGET):
+ console.print("[dim] Verifying file existence... OK (Deleting now)[/dim]")
+ os.remove(DOWNLOAD_TARGET)
+
+ # 7. Bookmarks
+ test_cli_command(f"pyrgo bookmark add test-bm {TEST_FILE}", "Add Bookmark")
+ test_cli_command("pyrgo bookmark list", "List Bookmarks")
+ test_cli_command("pyrgo run test-bm --yes", "Run from Bookmark")
+
+ # --- API TESTS ---
+ test_python_api()
+
+ print_header("Test Suite Completed")
+
+if __name__ == "__main__":
+ try:
+ main()
+ except KeyboardInterrupt:
+ console.print("\n[bold red]Tests aborted by user.[/bold red]")+
\ No newline at end of file
