diff --git a/README.md b/README.md index 5eaa65e..2ddc604 100644 --- a/README.md +++ b/README.md @@ -164,6 +164,160 @@ if __name__ == "__main__": ``` --- +## Social Media + +- Twitter API Tool + +```python +import os +from time import time + +from swarm_models import OpenAIChat +from swarms import Agent +from dotenv import load_dotenv + +from swarms_tools.social_media.twitter_tool import TwitterTool + +load_dotenv() + +model_name = "gpt-4o" + +model = OpenAIChat( + model_name=model_name, + max_tokens=3000, + openai_api_key=os.getenv("OPENAI_API_KEY"), +) + + +medical_coder = Agent( + agent_name="Medical Coder", + system_prompt=""" + You are a highly experienced and certified medical coder with extensive knowledge of ICD-10 coding guidelines, clinical documentation standards, and compliance regulations. Your responsibility is to ensure precise, compliant, and well-documented coding for all clinical cases. + + ### Primary Responsibilities: + 1. **Review Clinical Documentation**: Analyze all available clinical records, including specialist inputs, physician notes, lab results, imaging reports, and discharge summaries. + 2. **Assign Accurate ICD-10 Codes**: Identify and assign appropriate codes for primary diagnoses, secondary conditions, symptoms, and complications. + 3. **Ensure Coding Compliance**: Follow the latest ICD-10-CM/PCS coding guidelines, payer-specific requirements, and organizational policies. + 4. **Document Code Justification**: Provide clear, evidence-based rationale for each assigned code. + + ### Detailed Coding Process: + - **Review Specialist Inputs**: Examine all relevant documentation to capture the full scope of the patient's condition and care provided. + - **Identify Diagnoses**: Determine the primary and secondary diagnoses, as well as any symptoms or complications, based on the documentation. + - **Assign ICD-10 Codes**: Select the most accurate and specific ICD-10 codes for each identified diagnosis or condition. + - **Document Supporting Evidence**: Record the documentation source (e.g., lab report, imaging, or physician note) for each code to justify its assignment. + - **Address Queries**: Note and flag any inconsistencies, missing information, or areas requiring clarification from providers. + + ### Output Requirements: + Your response must be clear, structured, and compliant with professional standards. Use the following format: + + 1. **Primary Diagnosis Codes**: + - **ICD-10 Code**: [e.g., E11.9] + - **Description**: [e.g., Type 2 diabetes mellitus without complications] + - **Supporting Documentation**: [e.g., Physician's note dated MM/DD/YYYY] + + 2. **Secondary Diagnosis Codes**: + - **ICD-10 Code**: [Code] + - **Description**: [Description] + - **Order of Clinical Significance**: [Rank or priority] + + 3. **Symptom Codes**: + - **ICD-10 Code**: [Code] + - **Description**: [Description] + + 4. **Complication Codes**: + - **ICD-10 Code**: [Code] + - **Description**: [Description] + - **Relevant Documentation**: [Source of information] + + 5. **Coding Notes**: + - Observations, clarifications, or any potential issues requiring provider input. + + ### Additional Guidelines: + - Always prioritize specificity and compliance when assigning codes. + - For ambiguous cases, provide a brief note with reasoning and flag for clarification. + - Ensure the output format is clean, consistent, and ready for professional use. + """, + llm=model, + max_loops=1, + dynamic_temperature_enabled=True, +) + + +# Define your options with the necessary credentials +options = { + "id": "29998836", + "name": "mcsswarm", + "description": "An example Twitter Plugin for testing.", + "credentials": { + "apiKey": os.getenv("TWITTER_API_KEY"), + "apiSecretKey": os.getenv("TWITTER_API_SECRET_KEY"), + "accessToken": os.getenv("TWITTER_ACCESS_TOKEN"), + "accessTokenSecret": os.getenv("TWITTER_ACCESS_TOKEN_SECRET"), + }, +} + +# Initialize the TwitterTool with your options +twitter_plugin = TwitterTool(options) + +# # Post a tweet +# post_tweet_fn = twitter_plugin.get_function('post_tweet') +# post_tweet_fn("Hello world!") + + +# Assuming `twitter_plugin` and `medical_coder` are already initialized +post_tweet = twitter_plugin.get_function("post_tweet") + +# Set to track posted tweets and avoid duplicates +posted_tweets = set() + + +def post_unique_tweet(): + """ + Generate and post a unique tweet. Skip duplicates. + """ + tweet_prompt = ( + "Share an intriguing, lesser-known fact about a medical disease, and include an innovative, fun, or surprising way to manage or cure it! " + "Make the response playful, engaging, and inspiring—something that makes people smile while learning. No markdown, just plain text!" + ) + + # Generate a new tweet text + tweet_text = medical_coder.run(tweet_prompt) + + # Check for duplicates + if tweet_text in posted_tweets: + print("Duplicate tweet detected. Skipping...") + return + + # Post the tweet + try: + post_tweet(tweet_text) + print(f"Posted tweet: {tweet_text}") + # Add the tweet to the set of posted tweets + posted_tweets.add(tweet_text) + except Exception as e: + print(f"Error posting tweet: {e}") + + +# Loop to post tweets every 10 seconds +def start_tweet_loop(interval=10): + """ + Continuously post tweets every `interval` seconds. + + Args: + interval (int): Time in seconds between tweets. + """ + print("Starting tweet loop...") + while True: + post_unique_tweet() + time.sleep(interval) + + +# Start the loop +start_tweet_loop(10) + + +``` + ## 🧩 Standardized Schema Every tool in **Swarms Tools** adheres to a strict schema for maintainability and interoperability: diff --git a/mcs_agent.py b/mcs_agent.py new file mode 100644 index 0000000..d33e291 --- /dev/null +++ b/mcs_agent.py @@ -0,0 +1,145 @@ +import os +from time import time + +from swarm_models import OpenAIChat +from swarms import Agent +from dotenv import load_dotenv + +from swarms_tools.social_media.twitter_tool import TwitterTool + +load_dotenv() + +model_name = "gpt-4o" + +model = OpenAIChat( + model_name=model_name, + max_tokens=3000, + openai_api_key=os.getenv("OPENAI_API_KEY"), +) + + +medical_coder = Agent( + agent_name="Medical Coder", + system_prompt=""" + You are a highly experienced and certified medical coder with extensive knowledge of ICD-10 coding guidelines, clinical documentation standards, and compliance regulations. Your responsibility is to ensure precise, compliant, and well-documented coding for all clinical cases. + + ### Primary Responsibilities: + 1. **Review Clinical Documentation**: Analyze all available clinical records, including specialist inputs, physician notes, lab results, imaging reports, and discharge summaries. + 2. **Assign Accurate ICD-10 Codes**: Identify and assign appropriate codes for primary diagnoses, secondary conditions, symptoms, and complications. + 3. **Ensure Coding Compliance**: Follow the latest ICD-10-CM/PCS coding guidelines, payer-specific requirements, and organizational policies. + 4. **Document Code Justification**: Provide clear, evidence-based rationale for each assigned code. + + ### Detailed Coding Process: + - **Review Specialist Inputs**: Examine all relevant documentation to capture the full scope of the patient's condition and care provided. + - **Identify Diagnoses**: Determine the primary and secondary diagnoses, as well as any symptoms or complications, based on the documentation. + - **Assign ICD-10 Codes**: Select the most accurate and specific ICD-10 codes for each identified diagnosis or condition. + - **Document Supporting Evidence**: Record the documentation source (e.g., lab report, imaging, or physician note) for each code to justify its assignment. + - **Address Queries**: Note and flag any inconsistencies, missing information, or areas requiring clarification from providers. + + ### Output Requirements: + Your response must be clear, structured, and compliant with professional standards. Use the following format: + + 1. **Primary Diagnosis Codes**: + - **ICD-10 Code**: [e.g., E11.9] + - **Description**: [e.g., Type 2 diabetes mellitus without complications] + - **Supporting Documentation**: [e.g., Physician's note dated MM/DD/YYYY] + + 2. **Secondary Diagnosis Codes**: + - **ICD-10 Code**: [Code] + - **Description**: [Description] + - **Order of Clinical Significance**: [Rank or priority] + + 3. **Symptom Codes**: + - **ICD-10 Code**: [Code] + - **Description**: [Description] + + 4. **Complication Codes**: + - **ICD-10 Code**: [Code] + - **Description**: [Description] + - **Relevant Documentation**: [Source of information] + + 5. **Coding Notes**: + - Observations, clarifications, or any potential issues requiring provider input. + + ### Additional Guidelines: + - Always prioritize specificity and compliance when assigning codes. + - For ambiguous cases, provide a brief note with reasoning and flag for clarification. + - Ensure the output format is clean, consistent, and ready for professional use. + """, + llm=model, + max_loops=1, + dynamic_temperature_enabled=True, +) + + +# Define your options with the necessary credentials +options = { + "id": "29998836", + "name": "mcsswarm", + "description": "An example Twitter Plugin for testing.", + "credentials": { + "apiKey": os.getenv("TWITTER_API_KEY"), + "apiSecretKey": os.getenv("TWITTER_API_SECRET_KEY"), + "accessToken": os.getenv("TWITTER_ACCESS_TOKEN"), + "accessTokenSecret": os.getenv("TWITTER_ACCESS_TOKEN_SECRET"), + }, +} + +# Initialize the TwitterTool with your options +twitter_plugin = TwitterTool(options) + +# # Post a tweet +# post_tweet_fn = twitter_plugin.get_function('post_tweet') +# post_tweet_fn("Hello world!") + + +# Assuming `twitter_plugin` and `medical_coder` are already initialized +post_tweet = twitter_plugin.get_function("post_tweet") + +# Set to track posted tweets and avoid duplicates +posted_tweets = set() + + +def post_unique_tweet(): + """ + Generate and post a unique tweet. Skip duplicates. + """ + tweet_prompt = ( + "Share an intriguing, lesser-known fact about a medical disease, and include an innovative, fun, or surprising way to manage or cure it! " + "Make the response playful, engaging, and inspiring—something that makes people smile while learning. No markdown, just plain text!" + ) + + # Generate a new tweet text + tweet_text = medical_coder.run(tweet_prompt) + + # Check for duplicates + if tweet_text in posted_tweets: + print("Duplicate tweet detected. Skipping...") + return + + # Post the tweet + try: + post_tweet(tweet_text) + print(f"Posted tweet: {tweet_text}") + # Add the tweet to the set of posted tweets + posted_tweets.add(tweet_text) + except Exception as e: + print(f"Error posting tweet: {e}") + + +# Loop to post tweets every 10 seconds +def start_tweet_loop(interval=10): + """ + Continuously post tweets every `interval` seconds. + + Args: + interval (int): Time in seconds between tweets. + """ + print("Starting tweet loop...") + while True: + post_unique_tweet() + time.sleep(interval) + + +# Start the loop +start_tweet_loop(10) diff --git a/swarms_tools/__init__.py b/swarms_tools/__init__.py index b82d508..32cc161 100644 --- a/swarms_tools/__init__.py +++ b/swarms_tools/__init__.py @@ -4,3 +4,4 @@ from swarms_tools.finance import * from swarms_tools.structs import * +from swarms_tools.social_media import * \ No newline at end of file diff --git a/swarms_tools/social_media/__init__.py b/swarms_tools/social_media/__init__.py index e69de29..a3e0134 100644 --- a/swarms_tools/social_media/__init__.py +++ b/swarms_tools/social_media/__init__.py @@ -0,0 +1,3 @@ +from swarms_tools.social_media.twitter_tool import TwitterTool + +__all__ = ["TwitterTool"] \ No newline at end of file diff --git a/swarms_tools/social_media/twitter_api.py b/swarms_tools/social_media/twitter_api.py deleted file mode 100644 index 3273193..0000000 --- a/swarms_tools/social_media/twitter_api.py +++ /dev/null @@ -1,137 +0,0 @@ -import os -from typing import Callable -import tweepy -from dotenv import load_dotenv -from loguru import logger - -load_dotenv() - - -class TwitterBot: - """Twitter Bot to auto-respond to DMs and mentions.""" - - def __init__(self, response_callback: Callable[[str], str]): - try: - # Load credentials from environment variables - api_key = os.getenv("TWITTER_API_KEY") - api_secret = os.getenv("TWITTER_API_SECRET") - access_token = os.getenv("TWITTER_ACCESS_TOKEN") - access_token_secret = os.getenv( - "TWITTER_ACCESS_TOKEN_SECRET" - ) - - if not all( - [ - api_key, - api_secret, - access_token, - access_token_secret, - ] - ): - raise ValueError("Missing Twitter API credentials") - - # Authenticate and initialize Tweepy clients - self.client = tweepy.Client( - consumer_key=api_key, - consumer_secret=api_secret, - access_token=access_token, - access_token_secret=access_token_secret, - wait_on_rate_limit=True, - ) - auth = tweepy.OAuth1UserHandler( - api_key, api_secret, access_token, access_token_secret - ) - self.api = tweepy.API(auth) - - # Verify credentials - self.me = self.client.get_me().data - logger.info( - f"Twitter Bot initialized for user @{self.me.username}" - ) - - self.response_callback = response_callback - self.processed_ids = set() - - except Exception as e: - logger.error( - f"Failed to initialize Twitter Bot: {str(e)}" - ) - raise - - def auto_respond_mentions(self): - """Fetch and respond to mentions.""" - try: - mentions = self.client.get_users_mentions( - self.me.id, - tweet_fields=["created_at"], - max_results=100, - ) - if not mentions.data: - logger.info("No new mentions to process.") - return - - for mention in mentions.data: - if mention.id in self.processed_ids: - continue - - logger.info( - f"Processing mention {mention.id}: {mention.text}" - ) - response = self.response_callback(mention.text) - if response: - self.client.create_tweet( - text=response, in_reply_to_tweet_id=mention.id - ) - logger.info(f"Replied to mention {mention.id}.") - self.processed_ids.add(mention.id) - - except Exception as e: - logger.error( - f"Error while responding to mentions: {str(e)}" - ) - - def auto_respond_dms(self): - """Fetch and respond to DMs.""" - try: - logger.warning( - "DM functionality is limited by API access level. Upgrade required." - ) - - # If access is granted, implement DM handling logic for v2 or elevated v1.1 - # Example for v1.1 (requires elevated access) - messages = self.api.get_direct_messages() - for message in messages: - message_id = message.id - if message_id in self.processed_ids: - continue - - text = message.message_create["message_data"]["text"] - sender_id = message.message_create["sender_id"] - - logger.info( - f"Processing DM {message_id} from {sender_id}: {text}" - ) - response = self.response_callback(text) - if response: - self.api.send_direct_message(sender_id, response) - logger.info(f"Replied to DM {message_id}.") - self.processed_ids.add(message_id) - - except Exception as e: - logger.error(f"Error while responding to DMs: {str(e)}") - - def run(self): - """Run the bot to handle mentions and DMs.""" - logger.info("Starting Twitter Bot...") - self.auto_respond_mentions() - self.auto_respond_dms() - - -# if __name__ == "__main__": - -# def generate_response(message: str) -> str: -# """Simple response generator.""" -# return f"Thank you for your message! You said: {message}" - -# bot = TwitterBot(response_callback=generate_response) -# bot.run() diff --git a/swarms_tools/social_media/twitter_tool.py b/swarms_tools/social_media/twitter_tool.py new file mode 100644 index 0000000..e9fc7db --- /dev/null +++ b/swarms_tools/social_media/twitter_tool.py @@ -0,0 +1,169 @@ +import logging +import subprocess +from typing import Any, Callable, Dict, List, Optional + +try: + import tweepy +except ImportError: + print("Tweepy is not installed. Please install it using 'pip install tweepy'.") + subprocess.run(["pip", "install", "tweepy"]) + raise + + +class TwitterTool: + """ + A plugin that interacts with Twitter to perform various actions such as posting, replying, quoting, and liking tweets, as well as fetching metrics. + """ + + def __init__(self, options: Dict[str, Any]) -> None: + """ + Initializes the TwitterTool with the provided options. + + Args: + options (Dict[str, Any]): A dictionary containing plugin configuration options. + """ + self.id: str = options.get("id", "twitter_plugin") + self.name: str = options.get("name", "Twitter Plugin") + self.description: str = options.get( + "description", + "A plugin that executes tasks within Twitter, capable of posting, replying, quoting, and liking tweets, and getting metrics.", + ) + # Ensure credentials are provided + credentials: Optional[Dict[str, str]] = options.get( + "credentials" + ) + if not credentials: + raise ValueError("Twitter API credentials are required.") + + self.twitter_client: tweepy.Client = tweepy.Client( + consumer_key=credentials.get("apiKey"), + consumer_secret=credentials.get("apiSecretKey"), + access_token=credentials.get("accessToken"), + access_token_secret=credentials.get("accessTokenSecret"), + ) + # Define internal function mappings + self._functions: Dict[str, Callable[..., Any]] = { + "get_metrics": self._get_metrics, + "reply_tweet": self._reply_tweet, + "post_tweet": self._post_tweet, + "like_tweet": self._like_tweet, + "quote_tweet": self._quote_tweet, + } + + # Configure logging + logging.basicConfig(level=logging.INFO) + self.logger: logging.Logger = logging.getLogger(__name__) + + @property + def available_functions(self) -> List[str]: + """ + Get list of available function names. + + Returns: + List[str]: A list of available function names. + """ + return list(self._functions.keys()) + + def get_function(self, fn_name: str) -> Callable: + """ + Get a specific function by name. + + Args: + fn_name (str): Name of the function to retrieve + + Raises: + ValueError: If function name is not found + + Returns: + Function object + """ + if fn_name not in self._functions: + raise ValueError( + f"Function '{fn_name}' not found. Available functions: {', '.join(self.available_functions)}" + ) + return self._functions[fn_name] + + def _get_metrics(self) -> Dict[str, int]: + """ + Fetches user metrics such as followers, following, and tweets count. + + Returns: + Dict[str, int]: A dictionary containing user metrics. + """ + try: + user = self.twitter_client.get_me( + user_fields=["public_metrics"] + ) + if not user or not user.data: + self.logger.warning("Failed to fetch user metrics.") + return {} + public_metrics = user.data.public_metrics + return { + "followers": public_metrics.get("followers_count", 0), + "following": public_metrics.get("following_count", 0), + "tweets": public_metrics.get("tweet_count", 0), + } + except tweepy.TweepyException as e: + print(f"Failed to fetch metrics: {e}") + return {} + + def _reply_tweet(self, tweet_id: int, reply: str) -> None: + """ + Replies to a tweet with the given reply text. + + Args: + tweet_id (int): The ID of the tweet to reply to. + reply (str): The text of the reply. + """ + try: + self.twitter_client.create_tweet( + in_reply_to_tweet_id=tweet_id, text=reply + ) + print(f"Successfully replied to tweet {tweet_id}.") + except tweepy.TweepyException as e: + print(f"Failed to reply to tweet {tweet_id}: {e}") + + def _post_tweet(self, tweet: str) -> Dict[str, Any]: + """ + Posts a new tweet with the given text. + + Args: + tweet (str): The text of the tweet to post. + + Returns: + Dict[str, Any]: The response from Twitter. + """ + try: + self.twitter_client.create_tweet(text=tweet) + print("Tweet posted successfully.") + except tweepy.TweepyException as e: + print(f"Failed to post tweet: {e}") + + def _like_tweet(self, tweet_id: int) -> None: + """ + Likes a tweet with the given ID. + + Args: + tweet_id (int): The ID of the tweet to like. + """ + try: + self.twitter_client.like(tweet_id) + print(f"Tweet {tweet_id} liked successfully.") + except tweepy.TweepyException as e: + print(f"Failed to like tweet {tweet_id}: {e}") + + def _quote_tweet(self, tweet_id: int, quote: str) -> None: + """ + Quotes a tweet with the given ID and adds a quote text. + + Args: + tweet_id (int): The ID of the tweet to quote. + quote (str): The text of the quote. + """ + try: + self.twitter_client.create_tweet( + quote_tweet_id=tweet_id, text=quote + ) + print(f"Successfully quoted tweet {tweet_id}.") + except tweepy.TweepyException as e: + print(f"Failed to quote tweet {tweet_id}: {e}") diff --git a/swarms_tools/tech/__init__.py b/swarms_tools/tech/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/swarms_tools/tech/github.py b/swarms_tools/tech/github.py new file mode 100644 index 0000000..438e961 --- /dev/null +++ b/swarms_tools/tech/github.py @@ -0,0 +1,255 @@ +import os +import requests +from dotenv import load_dotenv +from loguru import logger +from typing import Any, Dict, List, Optional + +# Load environment variables +load_dotenv() +GITHUB_TOKEN = os.getenv("GITHUB_TOKEN") +GITHUB_API_URL = "https://api.github.com" + +if not GITHUB_TOKEN: + logger.error("Missing GITHUB_TOKEN in .env file") + raise ValueError( + "GITHUB_TOKEN not found in environment variables" + ) + +headers = { + "Authorization": f"Bearer {GITHUB_TOKEN}", + "Accept": "application/vnd.github.v3+json", +} + + +def get_user_info(username: str) -> Dict[str, Any]: + """ + Fetch information about a GitHub user. + + Args: + username (str): GitHub username. + + Returns: + Dict[str, Any]: User information. + """ + url = f"{GITHUB_API_URL}/users/{username}" + logger.info(f"Fetching user info for {username}") + response = requests.get(url, headers=headers) + response.raise_for_status() + return response.json() + + +def list_repo_issues( + owner: str, repo: str, state: str = "open" +) -> List[Dict[str, Any]]: + """ + List issues for a repository. + + Args: + owner (str): Repository owner. + repo (str): Repository name. + state (str): Issue state (open, closed, all). Defaults to "open". + + Returns: + List[Dict[str, Any]]: List of issues. + """ + url = f"{GITHUB_API_URL}/repos/{owner}/{repo}/issues" + params = {"state": state} + logger.info(f"Listing {state} issues for {owner}/{repo}") + response = requests.get(url, headers=headers, params=params) + response.raise_for_status() + return response.json() + + +def create_issue( + owner: str, + repo: str, + title: str, + body: Optional[str] = None, + labels: Optional[List[str]] = None, +) -> Dict[str, Any]: + """ + Create an issue in a repository. + + Args: + owner (str): Repository owner. + repo (str): Repository name. + title (str): Issue title. + body (Optional[str]): Issue description. Defaults to None. + labels (Optional[List[str]]): List of labels for the issue. Defaults to None. + + Returns: + Dict[str, Any]: Created issue details. + """ + url = f"{GITHUB_API_URL}/repos/{owner}/{repo}/issues" + payload = {"title": title, "body": body, "labels": labels} + logger.info( + f"Creating issue in {owner}/{repo} with title: {title}" + ) + response = requests.post(url, headers=headers, json=payload) + response.raise_for_status() + return response.json() + + +def list_open_prs(owner: str, repo: str) -> List[Dict[str, Any]]: + """ + List open pull requests for a repository. + + Args: + owner (str): Repository owner. + repo (str): Repository name. + + Returns: + List[Dict[str, Any]]: List of open pull requests. + """ + url = f"{GITHUB_API_URL}/repos/{owner}/{repo}/pulls" + logger.info(f"Listing open pull requests for {owner}/{repo}") + response = requests.get(url, headers=headers) + response.raise_for_status() + return response.json() + + +def get_repo_details(owner: str, repo: str) -> Dict[str, Any]: + """ + Get details about a repository. + + Args: + owner (str): Repository owner. + repo (str): Repository name. + + Returns: + Dict[str, Any]: Repository details. + """ + url = f"{GITHUB_API_URL}/repos/{owner}/{repo}" + logger.info(f"Fetching details for repository {owner}/{repo}") + response = requests.get(url, headers=headers) + response.raise_for_status() + return response.json() + + +def close_issue( + owner: str, repo: str, issue_number: int +) -> Dict[str, Any]: + """ + Close an issue in a repository. + + Args: + owner (str): Repository owner. + repo (str): Repository name. + issue_number (int): Issue number. + + Returns: + Dict[str, Any]: Updated issue details. + """ + url = ( + f"{GITHUB_API_URL}/repos/{owner}/{repo}/issues/{issue_number}" + ) + payload = {"state": "closed"} + logger.info(f"Closing issue #{issue_number} in {owner}/{repo}") + response = requests.patch(url, headers=headers, json=payload) + response.raise_for_status() + return response.json() + + +def create_pull_request( + owner: str, + repo: str, + title: str, + head: str, + base: str, + body: Optional[str] = None, +) -> Dict[str, Any]: + """ + Create a pull request in a repository. + + Args: + owner (str): Repository owner. + repo (str): Repository name. + title (str): Pull request title. + head (str): Branch where changes are implemented. + base (str): Branch into which the pull request will be merged. + body (Optional[str]): Pull request description. Defaults to None. + + Returns: + Dict[str, Any]: Created pull request details. + """ + url = f"{GITHUB_API_URL}/repos/{owner}/{repo}/pulls" + payload = { + "title": title, + "head": head, + "base": base, + "body": body, + } + logger.info( + f"Creating pull request in {owner}/{repo} from {head} to {base} with title: {title}" + ) + response = requests.post(url, headers=headers, json=payload) + response.raise_for_status() + return response.json() + + +def merge_pull_request( + owner: str, repo: str, pr_number: int +) -> Dict[str, Any]: + """ + Merge a pull request in a repository. + + Args: + owner (str): Repository owner. + repo (str): Repository name. + pr_number (int): Pull request number. + + Returns: + Dict[str, Any]: Merged pull request details. + """ + url = f"{GITHUB_API_URL}/repos/{owner}/{repo}/pulls/{pr_number}/merge" + logger.info( + f"Merging pull request #{pr_number} in {owner}/{repo}" + ) + response = requests.put(url, headers=headers) + response.raise_for_status() + return response.json() + + +def list_repo_collaborators( + owner: str, repo: str +) -> List[Dict[str, Any]]: + """ + List collaborators for a repository. + + Args: + owner (str): Repository owner. + repo (str): Repository name. + + Returns: + List[Dict[str, Any]]: List of collaborators. + """ + url = f"{GITHUB_API_URL}/repos/{owner}/{repo}/collaborators" + logger.info(f"Listing collaborators for {owner}/{repo}") + response = requests.get(url, headers=headers) + response.raise_for_status() + return response.json() + + +def add_repo_collaborator( + owner: str, repo: str, username: str, permission: str = "push" +) -> Dict[str, Any]: + """ + Add a collaborator to a repository. + + Args: + owner (str): Repository owner. + repo (str): Repository name. + username (str): Collaborator's GitHub username. + permission (str): Permission level (pull, push, admin). Defaults to "push". + + Returns: + Dict[str, Any]: Response details. + """ + url = f"{GITHUB_API_URL}/repos/{owner}/{repo}/collaborators/{username}" + payload = {"permission": permission} + logger.info( + f"Adding {username} as a collaborator to {owner}/{repo} with permission: {permission}" + ) + response = requests.put(url, headers=headers, json=payload) + response.raise_for_status() + return response.json() diff --git a/tests/test_github.py b/tests/test_github.py new file mode 100644 index 0000000..a9be536 --- /dev/null +++ b/tests/test_github.py @@ -0,0 +1,188 @@ +from loguru import logger + + +def test_get_user_info(): + try: + from swarms_tools.tech.github import get_user_info + + username = "octocat" + logger.info( + f"Testing get_user_info with username: {username}" + ) + result = get_user_info(username) + assert "login" in result, "Key 'login' not found in result" + logger.success("get_user_info test passed!") + except Exception as e: + logger.error(f"get_user_info test failed: {e}") + + +def test_list_repo_issues(): + try: + from swarms_tools.tech.github import list_repo_issues + + owner, repo = "octocat", "Hello-World" + logger.info(f"Testing list_repo_issues for {owner}/{repo}") + result = list_repo_issues(owner, repo) + assert isinstance(result, list), "Result is not a list" + logger.success("list_repo_issues test passed!") + except Exception as e: + logger.error(f"list_repo_issues test failed: {e}") + + +def test_create_issue(): + try: + from swarms_tools.tech.github import create_issue + + owner, repo = "octocat", "Hello-World" + title = "Test Issue" + logger.info( + f"Testing create_issue for {owner}/{repo} with title: {title}" + ) + result = create_issue(owner, repo, title, body="Test Body") + assert ( + "title" in result and result["title"] == title + ), "Issue creation failed" + logger.success("create_issue test passed!") + except Exception as e: + logger.error(f"create_issue test failed: {e}") + + +def test_list_open_prs(): + try: + from swarms_tools.tech.github import list_open_prs + + owner, repo = "octocat", "Hello-World" + logger.info(f"Testing list_open_prs for {owner}/{repo}") + result = list_open_prs(owner, repo) + assert isinstance(result, list), "Result is not a list" + logger.success("list_open_prs test passed!") + except Exception as e: + logger.error(f"list_open_prs test failed: {e}") + + +def test_get_repo_details(): + try: + from swarms_tools.tech.github import get_repo_details + + owner, repo = "octocat", "Hello-World" + logger.info(f"Testing get_repo_details for {owner}/{repo}") + result = get_repo_details(owner, repo) + assert ( + "name" in result and result["name"] == repo + ), "Repository details mismatch" + logger.success("get_repo_details test passed!") + except Exception as e: + logger.error(f"get_repo_details test failed: {e}") + + +def test_close_issue(): + try: + from swarms_tools.tech.github import close_issue, create_issue + + owner, repo = "octocat", "Hello-World" + title = "Test Issue to Close" + logger.info(f"Testing close_issue for {owner}/{repo}") + issue = create_issue(owner, repo, title, body="Test Body") + issue_number = issue["number"] + result = close_issue(owner, repo, issue_number) + assert ( + result["state"] == "closed" + ), "Issue not closed successfully" + logger.success("close_issue test passed!") + except Exception as e: + logger.error(f"close_issue test failed: {e}") + + +def test_create_pull_request(): + try: + from swarms_tools.tech.github import create_pull_request + + owner, repo = "octocat", "Hello-World" + title, head, base = "Test PR", "feature-branch", "main" + logger.info(f"Testing create_pull_request for {owner}/{repo}") + result = create_pull_request( + owner, repo, title, head, base, body="Test PR Body" + ) + assert ( + "title" in result and result["title"] == title + ), "Pull request creation failed" + logger.success("create_pull_request test passed!") + except Exception as e: + logger.error(f"create_pull_request test failed: {e}") + + +def test_merge_pull_request(): + try: + from swarms_tools.tech.github import ( + merge_pull_request, + create_pull_request, + ) + + owner, repo = "octocat", "Hello-World" + title, head, base = ( + "Test PR to Merge", + "feature-branch", + "main", + ) + logger.info(f"Testing merge_pull_request for {owner}/{repo}") + pr = create_pull_request( + owner, repo, title, head, base, body="Test PR Body" + ) + pr_number = pr["number"] + result = merge_pull_request(owner, repo, pr_number) + assert result[ + "merged" + ], "Pull request not merged successfully" + logger.success("merge_pull_request test passed!") + except Exception as e: + logger.error(f"merge_pull_request test failed: {e}") + + +def test_list_repo_collaborators(): + try: + from swarms_tools.tech.github import list_repo_collaborators + + owner, repo = "octocat", "Hello-World" + logger.info( + f"Testing list_repo_collaborators for {owner}/{repo}" + ) + result = list_repo_collaborators(owner, repo) + assert isinstance(result, list), "Result is not a list" + logger.success("list_repo_collaborators test passed!") + except Exception as e: + logger.error(f"list_repo_collaborators test failed: {e}") + + +def test_add_repo_collaborator(): + try: + from swarms_tools.tech.github import add_repo_collaborator + + owner, repo = "octocat", "Hello-World" + username = "test-collaborator" + logger.info( + f"Testing add_repo_collaborator for {owner}/{repo} and username: {username}" + ) + result = add_repo_collaborator(owner, repo, username) + assert result["permissions"].get( + "push", False + ), "Collaborator not added successfully" + logger.success("add_repo_collaborator test passed!") + except Exception as e: + logger.error(f"add_repo_collaborator test failed: {e}") + + +if __name__ == "__main__": + logger.info("Starting GitHub function tests...") + + test_get_user_info() + test_list_repo_issues() + test_create_issue() + test_list_open_prs() + test_get_repo_details() + test_close_issue() + test_create_pull_request() + test_merge_pull_request() + test_list_repo_collaborators() + test_add_repo_collaborator() + + logger.info("All tests completed!")