Compare commits

..

15 Commits

Author SHA1 Message Date
Ritchie Mwewa
9b0f490089 Merge pull request #25 from bellingcat/4.0/beta
4.0/beta
2026-01-04 05:54:20 +02:00
Ritchie Mwewa
d9e2dae71d Merge branch 'master' into 4.0/beta 2026-01-04 05:54:11 +02:00
Ritchie Mwewa
0c06b95d18 Merge branch '4.0/beta' of github.com:bellingcat/octosuite into 4.0/beta 2026-01-04 05:52:23 +02:00
Ritchie Mwewa
ba8ed6bd4e Patches in menus 2026-01-04 05:52:01 +02:00
Ritchie Mwewa
86d0112691 Merge pull request #24 from bellingcat/4.0/beta
4.0/beta
2026-01-03 23:04:21 +02:00
Ritchie Mwewa
39f82ad775 Merge branch 'master' into 4.0/beta 2026-01-03 23:04:11 +02:00
Ritchie Mwewa
b5ee95db2b Validating targets will instantly cache their profile data since we're hitting the profile api to validate 2026-01-03 23:02:58 +02:00
Ritchie Mwewa
77efa0982f Merge branch '4.0/beta' of github.com:bellingcat/octosuite into 4.0/beta 2026-01-03 22:52:22 +02:00
Ritchie Mwewa
38a9a4b6b3 Allow response caching 2026-01-03 22:51:52 +02:00
Ritchie Mwewa
ab3cc398d0 Merge pull request #23 from bellingcat/4.0/beta
4.0/beta
2026-01-03 17:18:57 +02:00
Ritchie Mwewa
0a453dade6 Merge branch 'master' into 4.0/beta 2026-01-03 17:18:45 +02:00
Ritchie Mwewa
66abbcbef6 Update README.md 2026-01-03 17:17:33 +02:00
Ritchie Mwewa
04e35dbc98 Update README.md 2026-01-03 16:56:07 +02:00
Ritchie Mwewa
f4d7fa81b8 4.0.0-beta1 2026-01-03 16:39:29 +02:00
Ritchie Mwewa
7bf04cffd4 4.0.0-beta1 2026-01-03 16:36:07 +02:00
11 changed files with 1012 additions and 262 deletions

View File

@@ -1,4 +1,4 @@
![octosuite](img/octosuite.png)
![octosuite](https://raw.githubusercontent.com/bellingcat/octosuite/refs/heads/master/img/octosuite.png)
TUI-based toolkit for GitHub data analysis.
@@ -7,6 +7,7 @@ TUI-based toolkit for GitHub data analysis.
![Code Size](https://img.shields.io/github/languages/code-size/bellingcat/octosuite)
![Release Date](https://img.shields.io/github/release-date/bellingcat/octosuite)
![Build Status](https://img.shields.io/github/actions/workflow/status/bellingcat/octosuite/python-publish.yml)
[![CodeQL Advanced](https://github.com/bellingcat/octosuite/actions/workflows/codeql.yml/badge.svg)](https://github.com/bellingcat/octosuite/actions/workflows/codeql.yml)
![License](https://img.shields.io/github/license/bellingcat/octosuite)
## Overview
@@ -16,11 +17,56 @@ repositories, organizations, and search across GitHub's platform.
## Features
- **User Data** - View profiles, repositories, followers, organizations, and activity, e.t.c.
- **Repository Data** - Access repository details, commits, issues, releases, and contributors
- **Organisation Data** - Explore organisation profiles, members, and repositories
- **Search** - Search across repositories, users, commits, issues, and topics
- **Export** - Save data in JSON, CSV, or HTML formats
<details>
<summary><strong>See details</strong></summary>
- **User** - Get user data
- Profile
- Repositories
- Subscriptions
- Starred
- Followers
- Following
- Organizations
- Gists
- Events
- Received Events
- **Repository** - Get repository data
- Profile
- Forks
- Issues
- Issue Events
- Events
- Assignees
- Branches
- Tags
- Languages
- Stargazers
- Subscribers
- Commits
- Comments
- Releases
- Deployments
- Labels
- **Organisation** - Get organisation data
- Profile
- Repositories
- Events
- Hooks
- Issues
- Members
- **Search** - Search GitHub
- Repositories
- Users
- Commits
- Issues
- Topics
- **Export** - Export data
- JSON
- CSV
- HTML
</details>
## Installation
@@ -50,17 +96,20 @@ octosuite
```
> [!Note]
> You can run octosuite with commands `octosuite`, or `ocs`
> You can then run octosuite with command `octosuite`
## Usage
Navigate using <kbd>UP</kbd><kbd>DOWN</kbd> and <kbd>ENTER</kbd> to select options. The interface guides you through
Navigate using <kbd>UP</kbd><kbd>DOWN</kbd> and <kbd>ENTER</kbd> to select options. In the export menu, you should
use <kbd>SPACE</kbd> to check the format you want.
The interface guides you through
selecting a
data source
and
choosing what information to retrieve. Preview the results and optionally export them in your preferred format.
![home](img/menu.png)
![home](https://raw.githubusercontent.com/bellingcat/octosuite/refs/heads/master/img/menu.png)
## License

View File

@@ -1,6 +1,6 @@
[project]
name = "octosuite"
version = "4.0.0-beta"
version = "4.0.0beta2"
description = "TUI-based toolkit for GitHub data analysis."
readme = "README.md"
license = "MIT"
@@ -33,5 +33,4 @@ dev = [
]
[project.scripts]
octosuite = "octosuite.app:start"
ocs = "octosuite.app:start"
octosuite = "octosuite.app:start"

View File

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

View File

@@ -1,5 +1,6 @@
import sys
from .core.cache import cache
from .lib import console, __pkg__, __version__
from .tui.menus import Menus
@@ -11,3 +12,5 @@ def start():
menu.main()
except KeyboardInterrupt:
sys.exit()
finally:
cache.clear()

View File

@@ -0,0 +1,74 @@
import hashlib
import json
__all__ = ["cache"]
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 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):
"""
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):
"""
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 from memory."""
self._cache.clear()
def remove(self, url: str, params: dict = None):
"""
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)
cache = ResponseCache()
def clear_cache():
"""Clear all cached API responses from the global cache instance."""
cache.clear()

View File

@@ -5,6 +5,7 @@ import typing as t
import requests
from requests import Response
from .cache import cache
from ..lib import __version__
BASE_URL = "https://api.github.com"
@@ -13,6 +14,8 @@ __all__ = ["BASE_URL", "GitHub"]
class GitHub:
"""Handles GitHub API requests with caching and response sanitisation."""
def __init__(
self,
user_agent: str = (
@@ -21,11 +24,37 @@ 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
def get(
self, url: str, params: t.Optional[dict] = None, return_response: bool = False
self,
url: str,
params: t.Optional[dict] = None,
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:
return cached
response = requests.get(
url=url, params=params, headers={"User-Agent": self.user_agent}
)
@@ -34,33 +63,65 @@ class GitHub:
return response
if response.status_code == 200:
return self._sanitise_response(response=response.json())
sanitised = self.sanitise_response(response=response.json())
# Cache the successful response
if use_cache:
self.cache.set(url, sanitised, params)
return sanitised
return []
def is_valid_entity(
self, entity_type: t.Literal["user", "org", "repo"], **kwargs
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:
if entity_type == "user":
url = f"https://api.github.com/users/{kwargs.get('username')}"
elif entity_type == "org":
url = f"https://api.github.com/orgs/{kwargs.get('username')}"
elif entity_type == "repo":
url = f"https://api.github.com/repos/{kwargs.get('repo_owner')}/{kwargs.get('repo_name')}"
else:
type_map = {
"user": f"https://api.github.com/users/{kwargs.get('username')}",
"org": f"https://api.github.com/orgs/{kwargs.get('username')}",
"repo": f"https://api.github.com/repos/{kwargs.get('repo_owner')}/{kwargs.get('repo_name')}",
}
url = type_map[_type]
# Check cache first
cached = self.cache.get(url)
if cached is not None:
return True # If cached, entity exists
response = requests.get(url=url, headers={"User-Agent": self.user_agent})
# Only cache if entity exists (status 200)
if response.status_code == 200:
sanitised = self.sanitise_response(response.json())
self.cache.set(url, sanitised)
return True
response = self.get(url=url, return_response=True)
return response.status_code == 200
return False
except requests.RequestException:
return False
def _sanitise_response(self, response: t.Union[dict, list]) -> t.Union[dict, list]:
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):
return [self._sanitise_response(response=item) for item in response]
return [self.sanitise_response(response=item) for item in response]
if isinstance(response, dict):
keys_to_remove = [
@@ -74,6 +135,6 @@ class GitHub:
# Recursively clean nested dicts/lists
for key, value in response.items():
if isinstance(value, (dict, list)):
response[key] = self._sanitise_response(response=value)
response[key] = self.sanitise_response(response=value)
return response

View File

@@ -8,74 +8,191 @@ __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) -> bool:
"""Check if the entity exists on GitHub."""
def exists(self) -> tuple[bool, dict]:
"""
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:
return True, cached
try:
response = github.get(url=self.endpoint, return_response=True)
return response.status_code == 200
except exceptions.RequestException as err:
return False
if response.status_code == 200:
data = response.json()
# Sanitise the data
sanitised = github.sanitise_response(data)
# Cache the sanitised response
github.cache.set(self.endpoint, sanitised)
return True, sanitised
return False, response.json()
except exceptions.RequestException:
return False, {}
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
@@ -84,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
@@ -218,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

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,10 +213,12 @@ 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__)
if result:
if result is not None:
status.stop()
message_dialog(title="Update Available", text=result).run()
else:
@@ -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)

View File

@@ -26,20 +26,56 @@ __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 quit() -> bool:
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="Quit",
text="This will close the session, continue?",
title=title,
text=text,
buttons=[("Yes", True), ("No", False)],
).run()
return result if result is not None else False
except KeyboardInterrupt:
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?",
)
@staticmethod
def license():
"""Display the MIT license notice in a dialogue box."""
message_dialog(title="MIT License", text=LICENSE_NOTICE).run()

View File

@@ -7,6 +7,7 @@ from rich.status import Status
from .dialogs import Dialogs
from .prompts import Prompts
from ..core.cache import cache
from ..core.models import User, Repo, Org, Search
from ..lib import check_updates, preview_response, export_response, set_menu_title
from ..lib import console, clear_screen, ascii_banner
@@ -20,7 +21,6 @@ CUSTOM_STYLE = Style(
INSTRUCTIONS = "↑↓ [move] • ⮠ [select]"
EXPORT_INSTRUCTIONS = "↑↓ [move] • ⮠ [confirm] • spacebar [check]"
POINTER: str = "🖝 "
dialogs = Dialogs()
prompts = Prompts()
@@ -28,8 +28,16 @@ prompts = Prompts()
__all__ = ["Menus"]
class Menus:
def __init__(self):
class BaseMenu:
"""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",
@@ -64,15 +72,91 @@ class Menus:
# Search methods (all require pagination)
self.search_methods = {"repos", "users", "commits", "issues", "topics"}
self.mode_handlers = {
"user": self.user,
"repo": self.repo,
"org": self.org,
"search": self.search,
}
self.main_menu = main_menu
@staticmethod
def target_validator(
identifier: str,
instance: t.Union[User, Repo, Org],
callback: t.Callable,
*callback_args,
) -> bool:
"""
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):
target_type = "repo"
else:
target_type = "org"
with Status(
f"[dim]Validating {target_type} ({identifier})[/dim]...",
console=console,
) as status:
exists, response = instance.exists()
if not exists:
status.stop()
if response and "message" in response:
console.print(
f"[bold][yellow]✘[/yellow] {response['message']}[/bold]"
)
console.input(" Press [bold]ENTER[/bold] to continue ...")
callback(*callback_args)
return False
return True
def execute_and_handle_response(
self,
instance: t.Union[User, Repo, Org, Search],
method_name: str,
target_type: t.Literal["user", "repo", "org"],
source: str,
):
"""
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]...",
console=console,
) as status:
data = self.execute_selection(
source=source,
instance=instance,
method_name=method_name,
status=status,
)
status.stop()
self.response_handler(data=data, data_type=method_name, source=source)
def execute_selection(self, status: Status, **kwargs):
"""
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.
"""
def _execute_selection(self, status: Status, **kwargs):
"""Execute a method on an instance, prompting for pagination if needed."""
instance = kwargs.get("instance")
method_name = kwargs.get("method_name")
source = kwargs.get("source")
@@ -84,13 +168,113 @@ class Menus:
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 _navigation(self, option: str, callback: t.Callable, *callback_args):
"""Handle navigation options (back, quit, change settings)."""
def response_handler(self, data: t.Union[dict, list], data_type: str, source: str):
"""
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:
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:
clear_screen()
self.response_handler(
data=data,
data_type=data_type,
source=source,
)
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",
),
],
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:
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 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(),
"back": lambda: self.main_menu(),
"quit": lambda: (
sys.exit() if dialogs.quit() else callback(*callback_args)
),
@@ -102,83 +286,25 @@ class Menus:
return True
return False
def response_handling(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)
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()
class Menus(BaseMenu):
"""Main menu system providing interactive interfaces for GitHub data queries."""
if export_choice == "skip":
return
def __init__(self):
"""Initialise the Menus class with mode handlers for different query types."""
if export_choice == "quit":
if dialogs.quit():
sys.exit()
else:
self.response_handling(
data=data, data_type=data_type, source=source
)
return
super().__init__(main_menu=self.main)
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()
export_response(
data=data, data_type=data_type, source=source, file_formats=file_formats
)
except KeyboardInterrupt:
print("\nExport cancelled")
self.mode_handlers = {
"user": self.user,
"repo": self.repo,
"org": self.org,
"search": self.search,
}
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:
@@ -207,6 +333,11 @@ class Menus:
value="search",
description="Search GitHub",
),
q.Choice(
title="Clear cache",
value="clear_cache",
description="Clear all in-memory cache from Octosuite",
),
q.Choice(
title="Updates",
value="updates",
@@ -223,7 +354,6 @@ class Menus:
description="Close this session",
),
],
pointer=POINTER,
style=CUSTOM_STYLE,
instruction=INSTRUCTIONS,
).ask()
@@ -239,6 +369,10 @@ class Menus:
elif action == "updates":
check_updates()
self.main()
elif action == "clear_cache":
if dialogs.clear_cache():
cache.clear()
self.main()
elif action is None:
sys.exit()
else:
@@ -251,7 +385,12 @@ class Menus:
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:
@@ -262,9 +401,7 @@ class Menus:
style=CUSTOM_STYLE,
)
clear_screen()
ascii_banner(text=query)
option = q.select(
"What would you like to do/search?",
choices=[
@@ -302,7 +439,7 @@ class Menus:
q.Choice(
title="Go Back",
value="back",
description="Go back to ewous menu",
description="Go back to previous menu",
shortcut_key="b",
),
q.Choice(
@@ -312,7 +449,6 @@ class Menus:
shortcut_key="q",
),
],
pointer=POINTER,
style=CUSTOM_STYLE,
instruction=INSTRUCTIONS,
use_shortcuts=True,
@@ -320,9 +456,10 @@ class Menus:
if option is None:
self.main()
return
# Handle navigation
if self._navigation(option, self.search, query):
if self.navigation_handler(option, self.search, query):
return
# Handle change query
@@ -332,11 +469,9 @@ class Menus:
# Execute search if it's a valid method
if option in self.search_methods:
with console.status(
status=f"[dim]Initialising {option} search...[/dim]"
with Status(
status=f"[dim]Initialising {option} search[/dim]...", console=console
) as status:
# Get pagination params
status.stop()
params = prompts.pagination_params()
status.start()
@@ -349,22 +484,29 @@ class Menus:
)
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:
items = data.get("items")
status.stop()
self.response_handling(
self.response_handler(
data=items if items is not None else data,
data_type=option,
source=query,
)
# After handling response, show menu again
self.search(query=query)
def user(self, username: t.Optional[str] = None):
"""User menu for querying user data."""
def user(self, username: t.Optional[str] = None, is_validated: bool = False):
"""
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:
@@ -376,9 +518,16 @@ class Menus:
qmark="@",
)
clear_screen()
ascii_banner(text=username)
user = User(name=username)
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=[
@@ -451,7 +600,6 @@ class Menus:
shortcut_key="q",
),
],
pointer=POINTER,
style=CUSTOM_STYLE,
instruction=INSTRUCTIONS,
use_shortcuts=True,
@@ -459,50 +607,40 @@ class Menus:
if option is None:
self.main()
return
# Handle navigation
if self._navigation(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 if it's a valid method
valid_methods = self.paginated_methods | self.non_paginated_methods
if option in valid_methods:
with Status(
status=f"[dim]Initialising user {option}...[/dim]",
console=console,
) as status:
user = User(name=username)
# After handling response, show menu again WITHOUT re-validating
self.user(username=username, is_validated=True)
status.update(f"[dim]Validating user's ({username}) existence...[/dim]")
if user.exists():
console.print(
f"[bold][green]✔[/green] User ({username}) exists on GitHub[/bold]"
)
data = self._execute_selection(
source=username,
instance=user,
method_name=option,
status=status,
)
def repo(
self,
name: t.Optional[str] = None,
owner: t.Optional[str] = None,
):
"""
Display the repository menu for querying GitHub repository data.
status.stop()
self.response_handling(data=data, data_type=option, source=username)
else:
status.stop()
console.print(
f"[bold][yellow]✘[/yellow] User ({username}) doesn't exist on GitHub[/bold]"
)
console.input(" Press [bold]ENTER[/bold] to continue ...")
self.user()
:param name: Optional repository name. If None, prompts user for input.
:param owner: Optional repository owner. If None, prompts user for input.
"""
self.user(username=username)
def repo(self, name: t.Optional[str] = None, owner: t.Optional[str] = None):
"""Repository menu for querying repo data."""
set_menu_title(menu_type="repo")
clear_screen()
if name is None or owner is None:
@@ -521,9 +659,18 @@ class Menus:
qmark="@",
)
clear_screen()
ascii_banner(text=f"{owner}/{name}")
repo = Repo(name=name, owner=owner)
source = f"{owner}/{name}"
# Only validate if not already validated
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=[
@@ -635,19 +782,11 @@ class Menus:
shortcut_key="q",
),
],
pointer=POINTER,
style=CUSTOM_STYLE,
instruction=INSTRUCTIONS,
use_shortcuts=True,
).ask()
if option is None:
self.main()
# Handle navigation
if self._navigation(option, self.repo, name, owner):
return
# Handle change options
change_handlers = {
"change_repo_name": lambda: self.repo(owner=owner),
@@ -655,45 +794,36 @@ class Menus:
"change_both": lambda: self.repo(),
}
if option in change_handlers:
change_handlers[option]()
if option is None:
self.main()
return
# Execute action if it's a valid method
valid_methods = self.paginated_methods | self.non_paginated_methods
if option in valid_methods:
source = f"{owner}/{name}"
with Status(
status=f"[dim]Initialising repository {option}...[/dim]",
console=console,
) as status:
repo = Repo(name=name, owner=owner)
# Handle navigation
elif self.navigation_handler(option, self.repo, name, owner):
return
status.update(
f"[dim]Validating repository's ({source}) existence...[/dim]"
)
if repo.exists():
console.print(
f"[bold][green]✔[/green] Repository ({source}) exists on GitHub[/bold]"
)
data = self._execute_selection(
source=source, instance=repo, method_name=option, status=status
)
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,
)
status.stop()
self.response_handling(data=data, data_type=option, source=source)
else:
status.stop()
console.print(
f"[bold][yellow]✘[/yellow] Repository ({source}) doesn't exist on GitHub[/bold]"
)
console.input(" Press [bold]ENTER[/bold] to continue ...")
self.repo()
self.repo(name=name, owner=owner)
# After handling response, show menu again WITHOUT re-validating
self.repo(name=name, owner=owner)
def org(self, name: t.Optional[str] = None):
"""Organisation menu for querying org data."""
"""
Display the organisation menu for querying GitHub organisation data.
:param name: Optional organisation name. If None, prompts user for input.
"""
set_menu_title(menu_type="org")
clear_screen()
if name is None:
@@ -705,8 +835,14 @@ class Menus:
qmark="@",
)
clear_screen()
ascii_banner(text=name)
org = Org(name=name)
if not self.target_validator(
identifier=name,
instance=org,
callback=self.org,
):
return
option = q.select(
"What would you like to do?",
@@ -766,45 +902,22 @@ class Menus:
if option is None:
self.main()
return
# Handle navigation
if self._navigation(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 if it's a valid method
valid_methods = self.paginated_methods | self.non_paginated_methods
if option in valid_methods:
with Status(
status=f"[dim]Initialising organisation {option}...[/dim]",
console=console,
) as status:
org = Org(name=name)
else:
# Execute action and handle response
self.execute_and_handle_response(
instance=org, method_name=option, target_type="org", source=name
)
status.update(
f"[dim]Validating organisation's ({name}) existence...[/dim]"
)
if org.exists():
console.print(
f"[bold][green]✔[/green] Organisation ({name}) exists on GitHub[/bold]"
)
data = self._execute_selection(
source=name, instance=org, method_name=option, status=status
)
status.stop()
self.response_handling(data=data, data_type=option, source=name)
else:
status.stop()
console.print(
f"[bold][yellow]✘[/yellow] Organisation ({name}) doesn't exist on GitHub[/bold]"
)
console.input(" Press [bold]ENTER[/bold] to continue ...")
self.org()
self.org(name=name)
# After handling response, show menu again WITHOUT re-validating
self.org(name=name)

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(