bellingcat/octosuite
1
0
Fork 0
mirror of https://github.com/bellingcat/octosuite.git synced 2026-06-07 19:08:36 +03:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #25 from bellingcat/4.0/beta

Browse Source 
4.0/beta
This commit is contained in:
Ritchie Mwewa
2026-01-04 05:54:20 +02:00
committed by GitHub
parent 86d0112691 d9e2dae71d
commit 9b0f490089
9 changed files with 730 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pyproject.toml
View File

@@ -1,6 +1,6 @@
[project]
name = "octosuite"
version = "4.0.0beta1"
version = "4.0.0beta2"
description = "TUI-based toolkit for GitHub data analysis."
readme = "README.md"
license = "MIT"

2
src/octosuite/__init__.py
View File

@@ -1,2 +1,2 @@
__pkg__ = "octosuite"
__version__ = "4.0.0beta1"
__version__ = "4.0.0beta2"

42
src/octosuite/core/cache.py
View File

@@ -8,30 +8,59 @@ class ResponseCache:
"""Simple in-memory cache for API responses."""
def __init__(self):
"""Initialise the ResponseCache with an empty cache dictionary."""
self._cache = {}
@staticmethod
def _generate_key(url: str, params: dict = None) -> str:
"""Generate a unique cache key from URL and params."""
"""
Generate a unique cache key from URL and parameters.
:param url: The URL to generate a key for.
:param params: Optional dictionary of parameters to include in the key.
:return: MD5 hash string representing the unique cache key.
"""
cache_data = f"{url}:{json.dumps(params or {}, sort_keys=True)}"
return hashlib.md5(cache_data.encode()).hexdigest()
def get(self, url: str, params: dict = None):
"""Get cached response."""
"""
Retrieve a cached response if it exists.
:param url: The URL to retrieve cached data for.
:param params: Optional dictionary of parameters used in the original request.
:return: Cached response data if found, None otherwise.
"""
key = self._generate_key(url, params)
return self._cache.get(key)
def set(self, url: str, data, params: dict = None):
"""Cache a response."""
"""
Store a response in the cache.
:param url: The URL to cache data for.
:param data: The response data to cache.
:param params: Optional dictionary of parameters used in the request.
"""
key = self._generate_key(url, params)
self._cache[key] = data
def clear(self):
"""Clear all cached responses."""
"""Clear all cached responses from memory."""
self._cache.clear()
def remove(self, url: str, params: dict = None):
"""Remove a specific cached response."""
"""
Remove a specific cached response.
:param url: The URL to remove cached data for.
:param params: Optional dictionary of parameters used in the original request.
"""
key = self._generate_key(url, params)
self._cache.pop(key, None)
@@ -40,5 +69,6 @@ cache = ResponseCache()
def clear_cache():
"""Clear all cached API responses."""
"""Clear all cached API responses from the global cache instance."""
cache.clear()

34
src/octosuite/core/github.py
View File

@@ -14,6 +14,8 @@ __all__ = ["BASE_URL", "GitHub"]
class GitHub:
"""Handles GitHub API requests with caching and response sanitisation."""
def __init__(
self,
user_agent: str = (
@@ -22,6 +24,12 @@ class GitHub:
f"https://github.com/bellingcat/octosuite) requests/{requests.__version__}"
),
):
"""
Initialise the GitHub API client.
:param user_agent: Custom User-Agent string for API requests.
"""
self.user_agent = user_agent
self.cache = cache
@@ -32,6 +40,16 @@ class GitHub:
return_response: bool = False,
use_cache: bool = True,
) -> t.Union[dict, list, Response]:
"""
Make a GET request to the GitHub API.
:param url: The API endpoint URL.
:param params: Optional query parameters for the request.
:param return_response: If True, return the raw Response object instead of JSON data.
:param use_cache: If True, use cached responses when available.
:return: Dictionary, list, or Response object depending on the request and parameters.
"""
if use_cache and not return_response:
cached = self.cache.get(url, params)
if cached is not None:
@@ -58,7 +76,14 @@ class GitHub:
def is_valid_entity(
self, _type: t.Literal["user", "org", "repo"], **kwargs
) -> bool:
"""Validate if a GitHub entity exists."""
"""
Validate whether a GitHub entity exists.
:param _type: Type of entity to validate ("user", "org", or "repo").
:param kwargs: Entity identifiers (username for user/org, repo_owner and repo_name for repo).
:return: True if the entity exists, False otherwise.
"""
try:
type_map = {
"user": f"https://api.github.com/users/{kwargs.get('username')}",
@@ -86,6 +111,13 @@ class GitHub:
return False
def sanitise_response(self, response: t.Union[dict, list]) -> t.Union[dict, list]:
"""
Remove API URLs and null values from response data recursively.
:param response: The response data to sanitise (dict or list).
:return: Sanitised response with API URLs and null values removed.
"""
pattern = re.compile(r"https://api\.github\.com")
if isinstance(response, list):

343
src/octosuite/core/models.py
View File

@@ -8,14 +8,25 @@ __all__ = ["User", "Org", "Repo", "Search"]
class GitHubEntity:
"""Base class for GitHub entities with common functionality."""
"""
Base class for GitHub entities with common functionality."""
def __init__(self, source: str):
"""
Initialise a GitHub entity.
:param source: The source identifier for the entity.
"""
self.endpoint = None
self.source = source
def exists(self) -> tuple[bool, dict]:
"""Check if the entity exists on GitHub."""
"""
Check if the entity exists on GitHub.
:return: Tuple of (exists, response_data) where exists is True if the entity is found.
"""
# Check cache first
cached = github.cache.get(self.endpoint)
if cached is not None:
@@ -38,58 +49,150 @@ class GitHubEntity:
class User(GitHubEntity):
"""Represents a GitHub user with methods to query user data."""
def __init__(self, name: str):
"""
Initialise a User instance.
:param name: The GitHub username.
"""
super().__init__(source=name)
self.name = name
self.endpoint = f"{BASE_URL}/users/{name}"
def profile(self) -> dict:
"""
Retrieve the user's profile information.
:return: Dictionary containing user profile data.
"""
profile = github.get(url=self.endpoint)
return profile
def repos(self, page: int, per_page: int) -> list:
"""
Retrieve the user's public repositories.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of repository dictionaries.
"""
params = {"page": page, "per_page": per_page}
repos = github.get(url=f"{self.endpoint}/repos", params=params)
return repos
def subscriptions(self, page: int, per_page: int) -> list:
"""
Retrieve repositories the user is subscribed to.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of repository subscription dictionaries.
"""
params = {"page": page, "per_page": per_page}
subscriptions = github.get(url=f"{self.endpoint}/subscriptions", params=params)
return subscriptions
def starred(self, page: int, per_page: int) -> list:
"""
Retrieve repositories the user has starred.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of starred repository dictionaries.
"""
params = {"page": page, "per_page": per_page}
starred = github.get(url=f"{self.endpoint}/starred", params=params)
return starred
def followers(self, page: int, per_page: int) -> list:
"""
Retrieve the user's followers.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of follower user dictionaries.
"""
params = {"page": page, "per_page": per_page}
users = github.get(url=f"{self.endpoint}/followers", params=params)
return users
def following(self, page: int, per_page: int) -> list:
"""
Retrieve users that this user is following.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of followed user dictionaries.
"""
params = {"page": page, "per_page": per_page}
users = github.get(url=f"{self.endpoint}/following", params=params)
return users
def follows(self, user: str) -> bool: ...
def follows(self, user: str) -> bool:
"""Check if this user follows another user.
:param user: The username to check.
:return: True if this user follows the specified user.
"""
...
def orgs(self, page: int, per_page: int) -> list:
"""
Retrieve organisations the user belongs to.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of organisation dictionaries.
"""
params = {"page": page, "per_page": per_page}
orgs = github.get(url=f"{self.endpoint}/orgs", params=params)
return orgs
def gists(self, page: int, per_page: int) -> list:
"""
Retrieve the user's gists.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of gist dictionaries.
"""
params = {"page": page, "per_page": per_page}
gists = github.get(url=f"{self.endpoint}/gists", params=params)
return gists
def events(self, page: int, per_page: int) -> list:
"""
Retrieve the user's public events.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of event dictionaries.
"""
params = {"page": page, "per_page": per_page}
events = github.get(url=f"{self.endpoint}/events", params=params)
return events
def received_events(self, page: int, per_page: int) -> list:
"""
Retrieve events received by the user.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of received event dictionaries.
"""
params = {"page": page, "per_page": per_page}
received_events = github.get(
url=f"{self.endpoint}/received_events", params=params
@@ -98,133 +201,337 @@ class User(GitHubEntity):
class Org(GitHubEntity):
"""Represents a GitHub organisation with methods to query organisation data."""
def __init__(self, name: str):
"""
Initialise an Org instance.
:param name: The GitHub organisation name.
"""
super().__init__(source=name)
self.name = name
self.endpoint = f"{BASE_URL}/orgs/{name}"
def profile(self) -> dict:
"""
Retrieve the organisation's profile information.
:return: Dictionary containing organisation profile data.
"""
profile = github.get(url=self.endpoint)
return profile
def repos(self, page: int, per_page: int) -> list:
"""
Retrieve the organisation's public repositories.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of repository dictionaries.
"""
params = {"page": page, "per_page": per_page}
repos = github.get(url=f"{self.endpoint}/repos", params=params)
return repos
def events(self, page: int, per_page: int) -> list:
"""
Retrieve the organisation's public events.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of event dictionaries.
"""
params = {"page": page, "per_page": per_page}
events = github.get(url=f"{self.endpoint}/events", params=params)
return events
def hooks(self, page: int, per_page: int) -> list:
"""
Retrieve the organisation's webhooks.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of webhook dictionaries.
"""
params = {"page": page, "per_page": per_page}
hooks = github.get(url=f"{self.endpoint}/hooks", params=params)
return hooks
def issues(self, page: int, per_page: int) -> list:
"""
Retrieve the organisation's issues.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of issue dictionaries.
"""
params = {"page": page, "per_page": per_page}
issues = github.get(url=f"{self.endpoint}/issues", params=params)
return issues
def members(self, page: int, per_page: int) -> list:
"""
Retrieve the organisation's public members.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of member user dictionaries.
"""
params = {"page": page, "per_page": per_page}
members = github.get(url=f"{self.endpoint}/members", params=params)
return members
class Repo(GitHubEntity):
"""Represents a GitHub repository with methods to query repository data."""
def __init__(self, name: str, owner: str):
"""
Initialise a Repo instance.
:param name: The repository name.
:param owner: The repository owner's username.
"""
super().__init__(source=f"{owner}/{name}")
self.name = name
self.owner = owner
self.endpoint = f"{BASE_URL}/repos/{owner}/{name}"
def profile(self) -> dict:
"""
Retrieve the repository's information.
:return: Dictionary containing repository data.
"""
profile = github.get(url=self.endpoint)
return profile
def forks(self, page: int, per_page: int) -> list:
"""
Retrieve the repository's forks.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of fork repository dictionaries.
"""
params = {"page": page, "per_page": per_page}
forks = github.get(url=f"{self.endpoint}/forks", params=params)
return forks
def issue_events(self, page: int, per_page: int) -> list:
"""
Retrieve the repository's issue events.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of issue event dictionaries.
"""
params = {"page": page, "per_page": per_page}
issue_events = github.get(url=f"{self.endpoint}/issue_events", params=params)
return issue_events
def events(self, page: int, per_page: int) -> list:
"""
Retrieve the repository's events.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of event dictionaries.
"""
params = {"page": page, "per_page": per_page}
events = github.get(url=f"{self.endpoint}/events", params=params)
return events
def assignees(self, page: int, per_page: int) -> list:
"""
Retrieve the repository's available assignees.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of assignee user dictionaries.
"""
params = {"page": page, "per_page": per_page}
assignees = github.get(url=f"{self.endpoint}/assignees", params=params)
return assignees
def branches(self, page: int, per_page: int) -> list:
"""
Retrieve the repository's branches.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of branch dictionaries.
"""
params = {"page": page, "per_page": per_page}
branches = github.get(url=f"{self.endpoint}/branches", params=params)
return branches
def tags(self, page: int, per_page: int) -> list:
"""
Retrieve the repository's tags.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of tag dictionaries.
"""
params = {"page": page, "per_page": per_page}
tags = github.get(url=f"{self.endpoint}/tags", params=params)
return tags
def languages(self) -> dict:
"""
Retrieve the programming languages used in the repository.
:return: Dictionary mapping language names to bytes of code.
"""
languages = github.get(url=f"{self.endpoint}/languages")
return languages
def stargazers(self, page: int, per_page: int) -> list:
"""
Retrieve users who have starred the repository.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of stargazer user dictionaries.
"""
params = {"page": page, "per_page": per_page}
stargazers = github.get(url=f"{self.endpoint}/stargazers", params=params)
return stargazers
def subscribers(self, page: int, per_page: int) -> list:
"""
Retrieve users subscribed to the repository.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of subscriber user dictionaries.
"""
params = {"page": page, "per_page": per_page}
subscribers = github.get(url=f"{self.endpoint}/subscribers", params=params)
return subscribers
def commits(self, page: int, per_page: int) -> list:
"""
Retrieve the repository's commits.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of commit dictionaries.
"""
params = {"page": page, "per_page": per_page}
commits = github.get(url=f"{self.endpoint}/commits", params=params)
return commits
def comments(self, page: int, per_page: int) -> list:
"""
Retrieve the repository's commit comments.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of comment dictionaries.
"""
params = {"page": page, "per_page": per_page}
comments = github.get(url=f"{self.endpoint}/comments", params=params)
return comments
def contents(self, path: str) -> list:
"""
Retrieve the contents of a file or directory in the repository.
:param path: Path to the file or directory.
:return: List of content item dictionaries.
"""
contents = github.get(url=f"{self.endpoint}/contents/{path}")
return contents
def issues(self, page: int, per_page: int) -> list:
"""
Retrieve the repository's issues.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of issue dictionaries.
"""
params = {"page": page, "per_page": per_page}
issues = github.get(url=f"{self.endpoint}/issues", params=params)
return issues
def releases(self, page: int, per_page: int) -> list:
"""
Retrieve the repository's releases.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of release dictionaries.
"""
params = {"page": page, "per_page": per_page}
releases = github.get(url=f"{self.endpoint}/releases", params=params)
return releases
def deployments(self, page: int, per_page: int) -> list:
"""
Retrieve the repository's deployments.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of deployment dictionaries.
"""
params = {"page": page, "per_page": per_page}
deployments = github.get(url=f"{self.endpoint}/deployments", params=params)
return deployments
def labels(self, page: int, per_page: int) -> list:
"""
Retrieve the repository's labels.
:param page: Page number for pagination.
:param per_page: Number of results per page.
:return: List of label dictionaries.
"""
params = {"page": page, "per_page": per_page}
labels = github.get(url=f"{self.endpoint}/labels", params=params)
return labels
class Search:
"""Provides methods to search GitHub for various entity types."""
def __init__(self, query: str, page: int, per_page: int):
"""
Initialise a Search instance.
:param query: The search query string.
:param page: Page number for pagination.
:param per_page: Number of results per page.
"""
self.query = query
self.page = page
self.per_page = per_page
@@ -232,21 +539,51 @@ class Search:
self.params = {"q": query, "page": page, "per_page": per_page}
def repos(self) -> list:
"""
Search for repositories matching the query.
:return: List of repository search result dictionaries.
"""
repos = github.get(url=f"{self.endpoint}/repositories", params=self.params)
return repos
def users(self) -> list:
"""
Search for users matching the query.
:return: List of user search result dictionaries.
"""
users = github.get(url=f"{self.endpoint}/users", params=self.params)
return users
def commits(self) -> list:
"""
Search for commits matching the query.
:return: List of commit search result dictionaries.
"""
commits = github.get(url=f"{self.endpoint}/commits", params=self.params)
return commits
def issues(self) -> list:
"""
Search for issues and pull requests matching the query.
:return: List of issue search result dictionaries.
"""
issues = github.get(url=f"{self.endpoint}/issues", params=self.params)
return issues
def topics(self) -> list:
"""
Search for topics matching the query.
:return: List of topic search result dictionaries.
"""
topics = github.get(url=f"{self.endpoint}/topics", params=self.params)
return topics

42
src/octosuite/lib.py
View File

@@ -31,6 +31,14 @@ console = Console(log_time=False)
def preview_response(data: t.Union[dict, list], source: str, _type: str):
"""
Display a preview of response data in a tree structure.
:param data: The data to preview (dict or list).
:param source: The source identifier of the data.
:param _type: The type of data being previewed.
"""
if isinstance(data, dict):
tree = Tree(
label=f"\n[bold]{data.get('name') or data.get('login') or data.get('id') or 'Data'}[/bold]",
@@ -44,7 +52,7 @@ def preview_response(data: t.Union[dict, list], source: str, _type: str):
elif isinstance(data, list):
preview_data = data[:5]
tree = Tree(
label=f"\nPreview: First {len(preview_data)} of {len(data)} {_type} from {source}",
label=f"\n[bold]First {len(preview_data)} of {len(data)} {_type} for '{source}'[/bold]",
guide_style="#444444",
highlight=True,
)
@@ -74,7 +82,17 @@ def export_response(
file_formats: list,
output_dir: str = "../exports",
):
"""Export data to selected formats using built-in libraries."""
"""
Export response data to one or more file formats.
:param data: The data to export (dict or list).
:param data_type: The type of data being exported.
:param source: The source identifier of the data.
:param file_formats: List of file formats to export to (json, csv, html).
:param output_dir: Directory path where exported files will be saved.
:return: List of exported file paths.
"""
# Create output directory if it doesn't exist
output_dir = Path(output_dir)
output_dir.mkdir(exist_ok=True)
@@ -165,6 +183,14 @@ def export_response(
def fill_tree(tree: Tree, data: t.Union[dict, list]) -> Tree:
"""
Recursively populate a Rich Tree with data.
:param tree: The Tree object to populate.
:param data: The data to add to the tree (dict or list).
:return: The populated Tree object.
"""
if isinstance(data, dict):
for key, value in data.items():
if isinstance(value, dict) or isinstance(value, list):
@@ -187,6 +213,8 @@ def fill_tree(tree: Tree, data: t.Union[dict, list]) -> Tree:
def check_updates():
"""Check for available package updates and display the result."""
with console.status("[dim]Checking for updates...[/dim]") as status:
checker = UpdateChecker()
result = checker.check(__pkg__, __version__)
@@ -207,6 +235,10 @@ def clear_screen():
def ascii_banner(text: str):
"""Display a colourful ASCII art banner with gradient styling.
:param text: The text to convert to ASCII art.
"""
clear_screen()
ascii_text = pyfiglet.figlet_format(text=text, font="chunky")
@@ -224,6 +256,12 @@ def ascii_banner(text: str):
def set_menu_title(menu_type: t.Literal["home", "user", "org", "repo", "search"]):
"""
Set the terminal window title based on the current menu.
:param menu_type: The type of menu being displayed.
"""
title: str = __pkg__.title()
title += f" | {menu_type.title()}"
console.set_window_title(title=title)

27
src/octosuite/tui/dialogs.py
View File

@@ -26,10 +26,22 @@ __all__ = ["Dialogs"]
class Dialogs:
def __init__(self): ...
"""Provides interactive dialogue boxes for user confirmations and information display."""
def __init__(self):
"""Initialise the Dialogs class."""
...
@staticmethod
def _boolean(title: str, text: str) -> bool:
"""
Display a Yes/No dialogue and return the user's choice.
:param title: The title of the dialogue box.
:param text: The message text to display.
:return: True if user selects Yes, False if No or cancelled.
"""
try:
result = button_dialog(
title=title,
@@ -41,11 +53,23 @@ class Dialogs:
return True
def quit(self) -> bool:
"""
Display a confirmation dialogue for quitting the application.
:return: True if user confirms quit, False otherwise.
"""
return self._boolean(
title="Quit", text="This will close the session, continue?"
)
def clear_cache(self) -> bool:
"""
Display a confirmation dialogue for clearing the cache.
:return: True if user confirms cache clearing, False otherwise.
"""
return self._boolean(
title="Clear Cache",
text="This will clear all octosuite caches, continue?",
@@ -53,4 +77,5 @@ class Dialogs:
@staticmethod
def license():
"""Display the MIT license notice in a dialogue box."""
message_dialog(title="MIT License", text=LICENSE_NOTICE).run()

397
src/octosuite/tui/menus.py
View File

@@ -21,7 +21,6 @@ CUSTOM_STYLE = Style(
INSTRUCTIONS = "↑↓ [move] • ⮠ [select]"
EXPORT_INSTRUCTIONS = "↑↓ [move] • ⮠ [confirm] • spacebar [check]"
POINTER: str = "🖝 "
dialogs = Dialogs()
prompts = Prompts()
@@ -30,9 +29,15 @@ __all__ = ["Menus"]
class BaseMenu:
"""Base class with common menu functionality."""
"""Base class providing common menu functionality and response handling for GitHub data queries."""
def __init__(self, main_menu: t.Callable):
"""
Initialise the BaseMenu with pagination and search method configurations.
:param main_menu: Callable reference to the main menu function.
"""
# Define which methods require pagination
self.paginated_methods = {
"repos",
@@ -76,17 +81,16 @@ class BaseMenu:
callback: t.Callable,
*callback_args,
) -> bool:
"""Validate if an entity exists on GitHub.
Args:
identifier: Name/identifier of the entity
instance: The instance with an exists() method (User, Repo, or Org)
callback: Function to call if validation fails
*callback_args: Arguments to pass to callback
Returns:
bool: True if entity exists, False otherwise
"""
Validate whether a GitHub entity exists.
:param identifier: Name or identifier of the entity to validate.
:param instance: The instance with an exists() method (User, Repo, or Org).
:param callback: Function to call if validation fails.
:param callback_args: Arguments to pass to the callback function.
:return: True if the entity exists, False otherwise.
"""
if isinstance(instance, User):
target_type = "user"
elif isinstance(instance, Repo):
@@ -95,18 +99,12 @@ class BaseMenu:
target_type = "org"
with Status(
f"[dim]Validating {target_type}'s ({identifier}) existence...[/dim]",
f"[dim]Validating {target_type} ({identifier})[/dim]...",
console=console,
) as status:
exists, response = instance.exists()
if exists:
ascii_banner(text=identifier)
console.print(
f"[bold][green]✔[/green] {target_type.capitalize()} ({identifier}) exists on GitHub[/bold]"
)
return True
else:
if not exists:
status.stop()
if response and "message" in response:
console.print(
@@ -116,6 +114,8 @@ class BaseMenu:
callback(*callback_args)
return False
return True
def execute_and_handle_response(
self,
instance: t.Union[User, Repo, Org, Search],
@@ -123,18 +123,19 @@ class BaseMenu:
target_type: t.Literal["user", "repo", "org"],
source: str,
):
"""Execute a method and handle its response.
Args:
instance: The instance to execute the method on
method_name: Name of the method to execute
target_type: Type of entity ("user", "repo", "org")
source: Source identifier for response handling
"""
Execute a method on an instance and handle the resulting data.
:param instance: The instance to execute the method on.
:param method_name: Name of the method to execute.
:param target_type: Type of entity ("user", "repo", or "org").
:param source: Source identifier for response handling.
"""
valid_methods = self.paginated_methods | self.non_paginated_methods
if method_name in valid_methods:
with Status(
status=f"[dim]Initialising {target_type} {method_name}...[/dim]",
status=f"[dim]Initialising {target_type} {method_name}[/dim]...",
console=console,
) as status:
data = self.execute_selection(
@@ -148,7 +149,14 @@ class BaseMenu:
self.response_handler(data=data, data_type=method_name, source=source)
def execute_selection(self, status: Status, **kwargs):
"""Execute a method on an instance, prompting for pagination if needed."""
"""
Execute a method on an instance with pagination if required.
:param status: Rich Status object for displaying progress.
:param kwargs: Must include 'instance', 'method_name', and 'source'.
:return: The data returned by the executed method.
"""
instance = kwargs.get("instance")
method_name = kwargs.get("method_name")
source = kwargs.get("source")
@@ -160,88 +168,111 @@ class BaseMenu:
prompts.pagination_params() if method_name in self.paginated_methods else {}
)
status.start()
status.update(f"[dim]Getting {method_name} from {source}...[/dim]")
status.update(f"[dim]Getting {method_name} from {source}[/dim]...")
return method(**params)
def response_handler(self, data: t.Union[dict, list], data_type: str, source: str):
"""Export data to file in user-selected format(s)."""
preview_response(data=data, source=source, _type=data_type)
"""
Handle retrieved data by previewing and offering export options.
:param data: The data to handle (dict or list).
:param data_type: Type of data retrieved.
:param source: Source identifier of the data.
"""
try:
export_choice = q.select(
"What would you like to do?",
choices=[
q.Choice(
title="Export",
value="export",
description="Export the data",
shortcut_key="e",
),
q.Choice(
title="Skip",
value="skip",
description="Do nothing, and go back to previous menu",
shortcut_key="x",
),
q.Choice(
title="Quit",
value="quit",
description="Close this session",
),
],
pointer=POINTER,
style=CUSTOM_STYLE,
instruction=INSTRUCTIONS,
).ask()
if not data:
console.print(
f"[bold][yellow]✘[/yellow] No data found for '{source}'[/bold]"
)
else:
preview_response(data=data, source=source, _type=data_type)
export_choice = q.select(
"What would you like to do?",
choices=[
q.Choice(
title="Export",
value="export",
description="Export the data",
shortcut_key="e",
),
q.Choice(
title="Skip",
value="skip",
description="Do nothing, and go back to previous menu",
shortcut_key="x",
),
q.Choice(
title="Quit",
value="quit",
description="Close this session",
),
],
style=CUSTOM_STYLE,
instruction=INSTRUCTIONS,
).ask()
if export_choice == "skip":
return
if export_choice == "quit":
if dialogs.quit():
sys.exit()
else:
self.response_handler(
data=data,
data_type=data_type,
source=source,
)
if export_choice == "skip":
return
file_formats = q.checkbox(
"Select export format(s)",
choices=[
q.Choice(
title="JSON",
value="json",
description="Export as JSON file",
),
q.Choice(
title="CSV",
value="csv",
description="Export as CSV file",
),
q.Choice(
title="HTML",
value="html",
description="Export as HTML table",
),
],
pointer=POINTER,
style=CUSTOM_STYLE,
instruction=EXPORT_INSTRUCTIONS,
validate=lambda x: len(x) > 0 or "Please select at least one format",
).ask()
if export_choice == "quit":
if dialogs.quit():
sys.exit()
else:
clear_screen()
self.response_handler(
data=data,
data_type=data_type,
source=source,
)
return
export_response(
data=data, data_type=data_type, source=source, file_formats=file_formats
)
file_formats = q.checkbox(
"Select export format(s)",
choices=[
q.Choice(
title="JSON",
value="json",
description="Export as JSON file",
),
q.Choice(
title="CSV",
value="csv",
description="Export as CSV file",
),
q.Choice(
title="HTML",
value="html",
description="Export as HTML table",
),
],
style=CUSTOM_STYLE,
instruction=EXPORT_INSTRUCTIONS,
validate=lambda x: len(x) > 0
or "Please select at least one format",
).ask()
export_response(
data=data,
data_type=data_type,
source=source,
file_formats=file_formats,
)
except KeyboardInterrupt:
print("\nExport cancelled")
console.print("\nExport cancelled")
finally:
console.input(" Press [bold]ENTER[/bold] to continue ...")
def navigation_handler(self, option: str, callback: t.Callable, *callback_args):
"""Handle navigation options (back, quit, change settings)."""
"""
Handle common navigation options across menus.
:param option: The selected navigation option.
:param callback: Function to call for certain navigation actions.
:param callback_args: Arguments to pass to the callback function.
:return: True if navigation was handled, False otherwise.
"""
navigation_handlers = {
"back": lambda: self.main_menu(),
"quit": lambda: (
@@ -257,7 +288,11 @@ class BaseMenu:
class Menus(BaseMenu):
"""Main menu system providing interactive interfaces for GitHub data queries."""
def __init__(self):
"""Initialise the Menus class with mode handlers for different query types."""
super().__init__(main_menu=self.main)
self.mode_handlers = {
@@ -268,7 +303,8 @@ class Menus(BaseMenu):
}
def main(self):
"""Main menu to select mode."""
"""Display the main menu for selecting query mode."""
set_menu_title(menu_type="home")
clear_screen()
try:
@@ -318,7 +354,6 @@ class Menus(BaseMenu):
description="Close this session",
),
],
pointer=POINTER,
style=CUSTOM_STYLE,
instruction=INSTRUCTIONS,
).ask()
@@ -350,7 +385,12 @@ class Menus(BaseMenu):
sys.exit()
def search(self, query: t.Optional[str] = None):
"""Search menu for querying GitHub."""
"""
Display the search menu for querying GitHub content.
:param query: Optional search query string. If None, prompts user for input.
"""
set_menu_title(menu_type="search")
clear_screen()
if query is None:
@@ -409,7 +449,6 @@ class Menus(BaseMenu):
shortcut_key="q",
),
],
pointer=POINTER,
style=CUSTOM_STYLE,
instruction=INSTRUCTIONS,
use_shortcuts=True,
@@ -431,7 +470,7 @@ class Menus(BaseMenu):
# Execute search if it's a valid method
if option in self.search_methods:
with Status(
status=f"[dim]Initialising {option} search...[/dim]", console=console
status=f"[dim]Initialising {option} search[/dim]...", console=console
) as status:
status.stop()
params = prompts.pagination_params()
@@ -445,7 +484,7 @@ class Menus(BaseMenu):
)
method = getattr(search, option)
status.update(f"[dim]Searching {option} for {query}...[/dim]")
status.update(f"[dim]Searching {option} for {query}[/dim]...")
data = method()
if data:
@@ -461,7 +500,13 @@ class Menus(BaseMenu):
self.search(query=query)
def user(self, username: t.Optional[str] = None, is_validated: bool = False):
"""User menu for querying user data."""
"""
Display the user menu for querying GitHub user data.
:param username: Optional GitHub username. If None, prompts user for input.
:param is_validated: Whether the username has already been validated.
"""
set_menu_title(menu_type="user")
clear_screen()
if username is None:
@@ -475,14 +520,14 @@ class Menus(BaseMenu):
user = User(name=username)
if not is_validated:
if not self.target_validator(
identifier=username,
instance=user,
callback=self.user,
):
return
if not self.target_validator(
identifier=username,
instance=user,
callback=self.user,
):
return
ascii_banner(text=username)
option = q.select(
"What would you like to do/get?",
choices=[
@@ -555,7 +600,6 @@ class Menus(BaseMenu):
shortcut_key="q",
),
],
pointer=POINTER,
style=CUSTOM_STYLE,
instruction=INSTRUCTIONS,
use_shortcuts=True,
@@ -566,32 +610,37 @@ class Menus(BaseMenu):
return
# Handle navigation
if self.navigation_handler(option, self.user, username):
elif self.navigation_handler(option, self.user, username):
return
# Handle change username
if option == "change_username":
elif option == "change_username":
self.user()
return
else:
# Execute action and handle response
self.execute_and_handle_response(
instance=user,
method_name=option,
target_type="user",
source=username,
)
# Execute action and handle response
self.execute_and_handle_response(
instance=user,
method_name=option,
target_type="user",
source=username,
)
# After handling response, show menu again WITHOUT re-validating
self.user(username=username, is_validated=True)
# After handling response, show menu again WITHOUT re-validating
self.user(username=username, is_validated=True)
def repo(
self,
name: t.Optional[str] = None,
owner: t.Optional[str] = None,
is_validated: bool = False,
):
"""Repository menu for querying repo data."""
"""
Display the repository menu for querying GitHub repository data.
:param name: Optional repository name. If None, prompts user for input.
:param owner: Optional repository owner. If None, prompts user for input.
"""
set_menu_title(menu_type="repo")
clear_screen()
if name is None or owner is None:
@@ -614,14 +663,14 @@ class Menus(BaseMenu):
source = f"{owner}/{name}"
# Only validate if not already validated
if not is_validated:
if not self.target_validator(
identifier=source,
instance=repo,
callback=self.repo,
):
return
if not self.target_validator(
identifier=source,
instance=repo,
callback=self.repo,
):
return
ascii_banner(text=source)
option = q.select(
"What would you like to do/get?",
choices=[
@@ -733,20 +782,11 @@ class Menus(BaseMenu):
shortcut_key="q",
),
],
pointer=POINTER,
style=CUSTOM_STYLE,
instruction=INSTRUCTIONS,
use_shortcuts=True,
).ask()
if option is None:
self.main()
return
# Handle navigation
if self.navigation_handler(option, self.repo, name, owner):
return
# Handle change options
change_handlers = {
"change_repo_name": lambda: self.repo(owner=owner),
@@ -754,23 +794,36 @@ class Menus(BaseMenu):
"change_both": lambda: self.repo(),
}
if option in change_handlers:
change_handlers[option]()
if option is None:
self.main()
return
# Execute action and handle response
self.execute_and_handle_response(
instance=repo,
method_name=option,
target_type="repo",
source=source,
)
# Handle navigation
elif self.navigation_handler(option, self.repo, name, owner):
return
# After handling response, show menu again WITHOUT re-validating
self.repo(name=name, owner=owner, is_validated=True)
elif option in change_handlers:
change_handlers[option]()
return
else:
# Execute action and handle response
self.execute_and_handle_response(
instance=repo,
method_name=option,
target_type="repo",
source=source,
)
# After handling response, show menu again WITHOUT re-validating
self.repo(name=name, owner=owner)
def org(self, name: t.Optional[str] = None):
"""
Display the organisation menu for querying GitHub organisation data.
:param name: Optional organisation name. If None, prompts user for input.
"""
def org(self, name: t.Optional[str] = None, is_validated: bool = False):
"""Organisation menu for querying org data."""
set_menu_title(menu_type="org")
clear_screen()
if name is None:
@@ -784,14 +837,12 @@ class Menus(BaseMenu):
org = Org(name=name)
# Only validate if not already validated
if not is_validated:
if not self.target_validator(
identifier=name,
instance=org,
callback=self.org,
):
return
if not self.target_validator(
identifier=name,
instance=org,
callback=self.org,
):
return
option = q.select(
"What would you like to do?",
@@ -844,7 +895,6 @@ class Menus(BaseMenu):
shortcut_key="q",
),
],
pointer=POINTER,
style=CUSTOM_STYLE,
instruction=INSTRUCTIONS,
use_shortcuts=True,
@@ -855,18 +905,19 @@ class Menus(BaseMenu):
return
# Handle navigation
if self.navigation_handler(option, self.org, name):
elif self.navigation_handler(option, self.org, name):
return
# Handle change org
if option == "change_org":
elif option == "change_org":
self.org()
return
# Execute action and handle response
self.execute_and_handle_response(
instance=org, method_name=option, target_type="org", source=name
)
else:
# Execute action and handle response
self.execute_and_handle_response(
instance=org, method_name=option, target_type="org", source=name
)
# After handling response, show menu again WITHOUT re-validating
self.org(name=name, is_validated=True)
# After handling response, show menu again WITHOUT re-validating
self.org(name=name)

32
src/octosuite/tui/prompts.py
View File

@@ -7,7 +7,10 @@ __all__ = ["Prompts"]
class Prompts:
"""Provides interactive prompts for user input collection."""
def __init__(self):
"""Initialise the Prompts class."""
pass
@staticmethod
@@ -17,17 +20,40 @@ class Prompts:
style: t.Optional[Style] = None,
qmark: t.Optional[str] = "?",
) -> str:
return q.text(
"""
Display a text input prompt and validate non-empty input.
:param message: The prompt message to display.
:param instruction: Optional instruction text to guide the user.
:param style: Optional questionary Style object for custom styling.
:param qmark: Optional custom question mark character or string.
:return: The user's input string.
:raises KeyboardInterrupt: If the user cancels the prompt with CTRL+C.
"""
result = q.text(
message=message,
instruction=instruction,
style=style,
qmark=qmark,
validate=lambda text: len(text.strip()) > 0 or "Input cannot be empty",
validate=lambda text: len(text.strip()) > 0
or None
or "Input cannot be empty",
).ask()
if result is None:
raise KeyboardInterrupt
return result
@staticmethod
def pagination_params() -> dict:
"""Prompt user for pagination parameters."""
"""
Prompt the user for pagination parameters.
:return: Dictionary containing 'page' and 'per_page' integer values.
"""
try:
page = q.text(message="Page", default="1", qmark="n").ask()
per_page = q.text(