diff --git a/tools/github_ci_tool.py b/tools/github_ci_tool.py index 7173453..29edd43 100644 --- a/tools/github_ci_tool.py +++ b/tools/github_ci_tool.py @@ -20,8 +20,8 @@ class GitHubCIHelper(BaseTool): # Inherits from BaseTool Initializes the GitHubCIHelper. Args: - repo_owner (str): The owner of the GitHub repository (e.g., \'\'\'bucolucas\'\'\'). - repo_name (str): The name of the GitHub repository (e.g., \'\'\'cyclop\'\'\'). + repo_owner (str): The owner of the GitHub repository (e.g., '''bucolucas'''). + repo_name (str): The name of the GitHub repository (e.g., '''cyclop'''). github_token (str, optional): A GitHub Personal Access Token (PAT) for API authentication. Recommended for private repos or higher rate limits. @@ -32,7 +32,7 @@ class GitHubCIHelper(BaseTool): # Inherits from BaseTool self.repo_owner = repo_owner self.repo_name = repo_name self.base_url = f"https://api.github.com/repos/{self.repo_owner}/{self.repo_name}" - self._token = github_token or os.environ.get(\'GITHUB_TOKEN\') # Renamed to _token for consistency + self._token = github_token or os.environ.get("GITHUB_TOKEN") # Renamed to _token for consistency self.headers = { "Accept": "application/vnd.github.v3+json" @@ -75,7 +75,7 @@ class GitHubCIHelper(BaseTool): # Inherits from BaseTool "type": "object", "properties": { "pull_request_number": {"type": "integer", "description": "The number of the pull request."}, - "workflow_name": {"type": "string", "description": "The display name of the workflow (e.g., \'\'\'Python CI\'\'\').", "default": "Python CI"} + "workflow_name": {"type": "string", "description": "The display name of the workflow (e.g., '''Python CI''').", "default": "Python CI"} }, "required": ["pull_request_number"] } @@ -90,7 +90,7 @@ class GitHubCIHelper(BaseTool): # Inherits from BaseTool "type": "object", "properties": { "run_id": {"type": "integer", "description": "The ID of the workflow run."}, - "job_name": {"type": "string", "description": "The name of the job (e.g., \'\'\'test\'\'\').", "default": "test"} + "job_name": {"type": "string", "description": "The name of the job (e.g., '''test''').", "default": "test"} }, "required": ["run_id"] } @@ -129,7 +129,7 @@ class GitHubCIHelper(BaseTool): # Inherits from BaseTool return error_message def clear(self): - """Clears any sensitive state if necessary. For this tool, it\'s a no-op but present for interface consistency.""" + """Clears any sensitive state if necessary. For this tool, it's a no-op but present for interface consistency.""" self.logger.info("GitHubCIHelper state cleared (no specific state to clear).") @@ -140,13 +140,13 @@ class GitHubCIHelper(BaseTool): # Inherits from BaseTool # Use self.session instead of requests directly response = self.session.request(method, url, headers=self.headers, **kwargs) response.raise_for_status() - if response.content and response.headers.get(\'Content-Type\', \'\').startswith(\'application/json\'): + if response.content and response.headers.get("Content-Type", "").startswith("application/json"): return response.json() elif response.content: # For non-JSON content like zip files or plain text logs return response return None except requests.exceptions.HTTPError as e: - self.logger.error(f"HTTP error occurred: {e} - {e.response.text if e.response else \'No response text\'}") # Use self.logger + self.logger.error(f"HTTP error occurred: {e} - {e.response.text if e.response else "No response text"}") # Use self.logger raise except requests.exceptions.RequestException as e: self.logger.error(f"Request failed: {e}") # Use self.logger @@ -163,10 +163,10 @@ class GitHubCIHelper(BaseTool): # Inherits from BaseTool pr_response = self._make_request("GET", pr_url) # this returns a response object or parsed JSON pr_data = pr_response if isinstance(pr_response, dict) else pr_response.json() # Ensure pr_data is dict - if not pr_data or \'head\' not in pr_data or \'sha\' not in pr_data[\'head\']: + if not pr_data or "head" not in pr_data or "sha" not in pr_data["head"]: self.logger.error(f"Could not get head SHA for PR {pull_request_number}. Response: {pr_data}") return None - head_sha = pr_data[\'head\'][\'sha\'] + head_sha = pr_data["head"]["sha"] runs_url = f"{self.base_url}/actions/runs?event=pull_request&head_sha={head_sha}" runs_response = self._make_request("GET", runs_url) @@ -183,17 +183,17 @@ class GitHubCIHelper(BaseTool): # Inherits from BaseTool """ Gets the latest failed workflow run for a specific pull request and workflow name. """ - self.logger.info(f"Getting latest failed run for PR #{pull_request_number}, workflow: \'{workflow_name}\'") + self.logger.info(f"Getting latest failed run for PR #{pull_request_number}, workflow: '{workflow_name}'") runs = self.get_pr_workflow_runs(pull_request_number) if not runs: self.logger.info(f"No runs found for PR #{pull_request_number} to check for failures.") return None - for run in sorted(runs, key=lambda r: r[\'created_at\'], reverse=True): - if run[\'name\'] == workflow_name and run[\'conclusion\'] == \'failure\': - self.logger.info(f"Found failed run {run[\'id\']} for workflow \'{workflow_name}\' in PR #{pull_request_number}") + for run in sorted(runs, key=lambda r: r["created_at"], reverse=True): + if run["name"] == workflow_name and run["conclusion"] == "failure": + self.logger.info(f"Found failed run {run['id']} for workflow '{workflow_name}' in PR #{pull_request_number}") return run - self.logger.info(f"No failed run for workflow \'{workflow_name}\' found for PR #{pull_request_number}") # Use self.logger + self.logger.info(f"No failed run for workflow '{workflow_name}' found for PR #{pull_request_number}") # Use self.logger return None @metrics.measure @@ -201,9 +201,9 @@ class GitHubCIHelper(BaseTool): # Inherits from BaseTool """ Downloads and returns the logs for a specific job within a workflow run. """ - self.logger.info(f"Getting job logs for run ID {run_id}, job name \'{job_name}\'") + self.logger.info(f"Getting job logs for run ID {run_id}, job name '{job_name}'") jobs_url = f"{self.base_url}/actions/runs/{run_id}/jobs" - target_job = None # Initialize target_job here to ensure it\'s defined for later logging + target_job = None # Initialize target_job here to ensure it's defined for later logging try: jobs_response = self._make_request("GET", jobs_url) jobs_data = jobs_response if isinstance(jobs_response, dict) else jobs_response.json() @@ -218,26 +218,27 @@ class GitHubCIHelper(BaseTool): # Inherits from BaseTool break if not target_job: - self.logger.error(f"Job \'{job_name}\' not found in run ID {run_id}") + self.logger.error(f"Job '{job_name}' not found in run ID {run_id}") return None - if target_job[\'status\'] != \'completed\': - self.logger.info(f"Job \'{job_name}\' in run ID {run_id} has not completed. Status: {target_job[\'status\']}") - return f"Job \'{job_name}\' not yet completed (status: {target_job[\'status\']}). Logs may be unavailable." + if target_job["status"] != "completed": + self.logger.info(f"Job '{job_name}' in run ID {run_id} has not completed. Status: {target_job['status']}") + return f"Job '{job_name}' not yet completed (status: {target_job['status']}). Logs may be unavailable." - logs_url = f"{self.base_url}/actions/jobs/{target_job[\'id\']}/logs" + logs_url = f"{self.base_url}/actions/jobs/{target_job['id']}/logs" self.logger.info(f"Attempting to download logs from: {logs_url}") log_response = self.session.get(logs_url, headers=self.headers, allow_redirects=True, stream=True) log_response.raise_for_status() - if \'application/zip\' in log_response.headers.get(\'Content-Type\', \'\'): - self.logger.info(f"Received zip file for logs of job ID {target_job[\'id\']}.") + if 'application/zip' in log_response.headers.get('Content-Type', ''): + self.logger.info(f"Received zip file for logs of job ID {target_job['id']}.") with zipfile.ZipFile(io.BytesIO(log_response.content)) as zf: - log_file_names = [name for name in zf.namelist() if not name.endswith(\'/\')] + log_file_names = [name for name in zf.namelist() if not name.endswith('/')] + if not log_file_names: - self.logger.error(f"No files found in the downloaded log zip for job ID {target_job[\'id\']}.") + self.logger.error(f"No files found in the downloaded log zip for job ID {target_job['id']}.") return None actual_log_file_name = log_file_names[0] @@ -247,28 +248,28 @@ class GitHubCIHelper(BaseTool): # Inherits from BaseTool actual_log_file_name = name break - self.logger.info(f"Extracting log file: {actual_log_file_name} from zip for job ID {target_job[\'id\']}.") + self.logger.info(f"Extracting log file: {actual_log_file_name} from zip for job ID {target_job['id']}.") with zf.open(actual_log_file_name) as log_file: - return log_file.read().decode(\'utf-8\') + return log_file.read().decode("utf-8") else: - self.logger.info(f"Received plain text logs for job ID {target_job[\'id\']}.") + self.logger.info(f"Received plain text logs for job ID {target_job['id']}.") return log_response.text except requests.exceptions.HTTPError as e: - self.logger.error(f"HTTP error downloading logs for job ID {target_job.get(\'id\', \'unknown\') if target_job else \'unknown\'}: {e} - {e.response.text if e.response else \'No response text\'}", exc_info=True) + self.logger.error(f"HTTP error downloading logs for job ID {target_job.get('id', 'unknown') if target_job else 'unknown'}: {e} - {e.response.text if e.response else 'No response text'}", exc_info=True) if e.response and e.response.status_code == 404: self.logger.error("Log download URL might be invalid or logs expired.") return f"Error downloading logs: {e}" except requests.exceptions.RequestException as e: - self.logger.error(f"Request failed downloading logs for job ID {target_job.get(\'id\', \'unknown\') if target_job else \'unknown\'}: {e}", exc_info=True) + self.logger.error(f"Request failed downloading logs for job ID {target_job.get('id', 'unknown') if target_job else 'unknown'}: {e}", exc_info=True) return f"Error during log download request: {e}" except zipfile.BadZipFile: - self.logger.error(f"Failed to unzip logs for job ID {target_job.get(\'id\', \'unknown\') if target_job else \'unknown\'}.", exc_info=True) + self.logger.error(f"Failed to unzip logs for job ID {target_job.get('id', 'unknown') if target_job else 'unknown'}.", exc_info=True) # Adding response text for BadZipFile can be risky if it's large binary data. # Consider logging only a snippet or specific headers if this occurs frequently. return "Failed to unzip logs." except Exception as e: - self.logger.error(f"An unexpected error occurred while processing logs for job {target_job.get(\'id\', \'unknown\') if target_job else \'unknown\'}: {e}", exc_info=True) + self.logger.error(f"An unexpected error occurred while processing logs for job {target_job.get('id', 'unknown') if target_job else 'unknown'}: {e}", exc_info=True) return f"Unexpected error processing logs: {e}" @@ -356,7 +357,7 @@ class GitHubCIHelper(BaseTool): # Inherits from BaseTool # --- Example Usage (Illustrative) --- if __name__ == "__main__": # This example assumes you have GITHUB_TOKEN environment variable set - # And that \'requests\' is installed. + # And that 'requests' is installed. # Replace with your actual repo owner, name, and PR number. pr_number = 206 # Example PR repo_owner = "bucolucas" # Example owner @@ -364,7 +365,7 @@ if __name__ == "__main__": # Setup basic logging for the example # In a real app, logger would be configured externally - logging.basicConfig(level=logging.INFO, format=\'%(asctime)s - %(name)s - %(levelname)s - %(message)s\') + logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') example_logger = logging.getLogger("GitHubCIHelperExample") @@ -375,10 +376,10 @@ if __name__ == "__main__": failed_run = helper.get_latest_failed_run_for_pr(pull_request_number=pr_number, workflow_name="Python CI") if failed_run: - example_logger.info(f"Found failed run: ID {failed_run[\'id\']}, Status {failed_run[\'conclusion\']}") - example_logger.info(f"Attempting to download logs for job \'test\' in run {failed_run[\'id\']}...") + example_logger.info(f"Found failed run: ID {failed_run['id']}, Status {failed_run['conclusion']}") + example_logger.info(f"Attempting to download logs for job 'test' in run {failed_run['id']}...") - log_content = helper.get_job_logs_for_run(run_id=failed_run[\'id\'], job_name="test") + log_content = helper.get_job_logs_for_run(run_id=failed_run['id'], job_name="test") if isinstance(log_content, str) and not log_content.startswith("Error") and not log_content.startswith("Job") and not log_content.startswith("Failed"): example_logger.info(f"Successfully downloaded logs (length: {len(log_content)} characters).") @@ -394,9 +395,9 @@ if __name__ == "__main__": # print(f"Log start:\n{log_content[:2000]}") elif log_content is None: example_logger.error("Could not retrieve log content (returned None).") - else: # If it\'s an error message string from the function itself + else: # If it's an error message string from the function itself example_logger.error(f"Failed to get/process logs: {log_content}") else: - example_logger.info(f"No failed \'Python CI\' workflow run found for PR #{pr_number} or the PR doesn\'t exist/no runs yet.") + example_logger.info(f"No failed 'Python CI' workflow run found for PR #{pr_number} or the PR doesn't exist/no runs yet.")