""" Main orchestration script for Bitbucket SonarQube automation. """ import sys import json import logging from pathlib import Path from datetime import datetime from typing import Dict, Optional # Import modules - use absolute imports for script execution try: from config_manager import ConfigManager from bitbucket_client import BitbucketClient from git_manager import GitManager from sonarqube_scanner import SonarQubeScanner from report_generator import ReportGenerator except ImportError: # If running as module, use relative imports from .config_manager import ConfigManager from .bitbucket_client import BitbucketClient from .git_manager import GitManager from .sonarqube_scanner import SonarQubeScanner from .report_generator import ReportGenerator # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('sonar_scan.log'), logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger(__name__) class ScanOrchestrator: """Orchestrates the entire scanning workflow.""" def __init__(self, config: ConfigManager): """ Initialize orchestrator. Args: config: Configuration manager instance """ self.config = config # Initialize components # Support both API token and app password authentication api_token = config.get("bitbucket.api_token") app_password = config.get("bitbucket.app_password") self.bitbucket = BitbucketClient( workspace=config.get("bitbucket.workspace"), username=config.get("bitbucket.username"), app_password=app_password if not api_token else None, api_token=api_token ) self.git_manager = GitManager( repos_dir=config.get("scanning.repos_dir") ) self.sonar_scanner = SonarQubeScanner( sonar_url=config.get("sonarqube.url"), sonar_token=config.get("sonarqube.token"), scanner_path=config.get("scanning.sonar_scanner_path") ) self.report_generator = ReportGenerator( reports_dir=config.get("scanning.reports_dir") ) # Load scan history self.history_file = Path(config.get("scanning.history_file")) self.scan_history = self._load_history() def _load_history(self) -> Dict: """Load scan history from JSON file.""" if self.history_file.exists(): try: with open(self.history_file, "r") as f: return json.load(f) except Exception as e: logger.warning(f"Failed to load scan history: {e}") return {} def _save_history(self) -> None: """Save scan history to JSON file.""" try: with open(self.history_file, "w") as f: json.dump(self.scan_history, f, indent=2) except Exception as e: logger.error(f"Failed to save scan history: {e}") def _get_last_scan_date(self, repo_slug: str) -> Optional[datetime]: """ Get last scan date for a repository. Args: repo_slug: Repository slug/name Returns: Last scan datetime or None """ if repo_slug in self.scan_history: date_str = self.scan_history[repo_slug].get("last_scan") if date_str: try: return datetime.fromisoformat(date_str) except Exception: pass return None def _has_report_for_today(self, repo_slug: str) -> bool: """ Check if a report exists for today's date for the repository. Args: repo_slug: Repository slug/name Returns: True if a report exists for today, False otherwise """ reports_dir = Path(self.config.get("scanning.reports_dir")) if not reports_dir.exists(): return False # Today's date in format YYYYMMDD today_str = datetime.now().strftime("%Y%m%d") # Look for reports matching pattern: {repo_slug}_{YYYYMMDD}_*.pdf pattern = f"{repo_slug}_{today_str}_*.pdf" # Check if any files match the pattern matching_files = list(reports_dir.glob(pattern)) return len(matching_files) > 0 def _update_scan_history(self, repo_slug: str, success: bool, report_path: Optional[str] = None) -> None: """ Update scan history for a repository. Args: repo_slug: Repository slug/name success: Whether scan was successful report_path: Path to generated report """ self.scan_history[repo_slug] = { "last_scan": datetime.now().isoformat(), "success": success, "report_path": str(report_path) if report_path else None } self._save_history() def _get_additional_repositories(self) -> list: """ Get additional repositories from configuration. These can be from different workspaces. Returns: List of repository info dictionaries """ additional_repos_config = self.config.get("scanning.additional_repositories", []) if not additional_repos_config: return [] repos = [] for repo_config in additional_repos_config: if isinstance(repo_config, str): # Simple string format: "workspace/repo" or SSH URL # Parse SSH URL: git@bitbucket.org:workspace/repo.git if repo_config.startswith("git@bitbucket.org:"): # Extract workspace and repo name path = repo_config.replace("git@bitbucket.org:", "").replace(".git", "") parts = path.split("/") if len(parts) == 2: workspace, repo_name = parts repos.append({ "name": repo_name, "slug": repo_name, "workspace": {"slug": workspace}, "links": { "clone": [ {"name": "ssh", "href": repo_config} ] } }) elif repo_config.startswith("https://github.com/"): # Before workspace/repo: HTTPS URLs also contain "/" path = repo_config.replace("https://github.com/", "").replace(".git", "") parts = path.split("/") if len(parts) == 2: org, repo_name = parts repos.append({ "name": repo_name, "slug": repo_name, "workspace": {"slug": org}, "links": { "clone": [ {"name": "ssh", "href": repo_config}, {"name": "https", "href": repo_config}, ] } }) elif "/" in repo_config: # Format: "workspace/repo" parts = repo_config.split("/") if len(parts) == 2: workspace, repo_name = parts ssh_url = f"git@bitbucket.org:{workspace}/{repo_name}.git" repos.append({ "name": repo_name, "slug": repo_name, "workspace": {"slug": workspace}, "links": { "clone": [ {"name": "ssh", "href": ssh_url} ] } }) elif isinstance(repo_config, dict): # Dictionary format with more details repo_name = repo_config.get("name") or repo_config.get("slug") workspace = repo_config.get("workspace") ssh_url = repo_config.get("ssh_url") or repo_config.get("clone_url") if repo_name and ssh_url: if isinstance(workspace, str): workspace = {"slug": workspace} elif not workspace: # Extract workspace from SSH URL if "git@bitbucket.org:" in ssh_url: path = ssh_url.replace("git@bitbucket.org:", "").replace(".git", "") ws_name = path.split("/")[0] if "/" in path else None workspace = {"slug": ws_name} if ws_name else {} repos.append({ "name": repo_name, "slug": repo_name, "workspace": workspace or {}, "links": { "clone": [ {"name": "ssh", "href": ssh_url} ] } }) return repos def run_scan(self) -> None: """Run the complete scanning workflow.""" logger.info("Starting SonarQube scan automation") # Get all repositories from workspace repos = self.bitbucket.get_all_repositories() or [] # Get additional manually configured repositories additional_repos = self._get_additional_repositories() if additional_repos: logger.info(f"Found {len(additional_repos)} additional repositories from config") repos.extend(additional_repos) if not repos: logger.warning("No repositories found in workspace or additional repositories configured") return logger.info(f"Found {len(repos)} repositories to check") scanned_count = 0 skipped_count = 0 failed_count = 0 # Process each repository for repo_info in repos: repo_slug = repo_info.get("name") or repo_info.get("slug") if not repo_slug: logger.warning(f"Skipping repository with no name/slug: {repo_info}") continue logger.info(f"Processing repository: {repo_slug}") try: # Step 1: Check if repository folder is empty or missing repos_dir = Path(self.config.get("scanning.repos_dir")) repo_path = repos_dir / repo_slug is_empty_or_missing = not repo_path.exists() or not (repo_path / ".git").exists() # Step 2: If folder is empty, clone the repository if is_empty_or_missing: logger.info(f"Repository folder is empty or missing for {repo_slug}, cloning...") # Get clone URL - use SSH since key is already set up clone_url = self.bitbucket.get_clone_url(repo_info, protocol="ssh") if not clone_url: logger.warning(f"No SSH clone URL found for {repo_slug}, skipping") skipped_count += 1 continue # SSH cloning doesn't require username/password - uses SSH keys clone_result = self.git_manager.clone_or_update( repo_slug=repo_slug, clone_url=clone_url ) if not clone_result or (isinstance(clone_result, tuple) and not clone_result[0]): logger.error(f"Failed to clone {repo_slug}") failed_count += 1 self._update_scan_history(repo_slug, False) continue # Handle return format if isinstance(clone_result, tuple): repo_path, is_first_clone = clone_result else: repo_path = clone_result is_first_clone = True else: # Repository exists, check if we need to scan # First check if there's a report for today - if not, always scan has_today_report = self._has_report_for_today(repo_slug) if not has_today_report: logger.info(f"No report found for today for {repo_slug}, will scan") # Force update and scan workspace_slug = repo_info.get("workspace", {}).get("slug") is_additional_repo = workspace_slug != self.config.get("bitbucket.workspace") if not is_additional_repo: logger.info(f"Updating repository: {repo_slug}") else: logger.info(f"Updating additional repository: {repo_slug}") else: # Report exists for today, check for changes before updating # For additional repos (not from API), always update workspace_slug = repo_info.get("workspace", {}).get("slug") is_additional_repo = workspace_slug != self.config.get("bitbucket.workspace") if not is_additional_repo: # Only check for changes if it's from the configured workspace last_scan_date = self._get_last_scan_date(repo_slug) has_changes = self.bitbucket.has_changes_since(repo_slug, last_scan_date) if not has_changes: logger.info(f"No changes detected for {repo_slug}, skipping (report exists for today)") skipped_count += 1 continue logger.info(f"Changes detected for {repo_slug}, updating...") else: logger.info(f"Updating additional repository: {repo_slug}") # Get clone URL - use SSH since key is already set up clone_url = self.bitbucket.get_clone_url(repo_info, protocol="ssh") if not clone_url: logger.warning(f"No SSH clone URL found for {repo_slug}, skipping") skipped_count += 1 continue # SSH cloning doesn't require username/password - uses SSH keys clone_result = self.git_manager.clone_or_update( repo_slug=repo_slug, clone_url=clone_url ) if not clone_result or (isinstance(clone_result, tuple) and not clone_result[0]): logger.error(f"Failed to update {repo_slug}") failed_count += 1 self._update_scan_history(repo_slug, False) continue # Handle return format if isinstance(clone_result, tuple): repo_path, is_first_clone = clone_result else: repo_path = clone_result is_first_clone = False # Step 3: Check for sonar-project.properties and generate if missing sonar_props_path = self.git_manager.get_sonar_properties_path(repo_path) if not sonar_props_path: # Generate project key from workspace and repo slug workspace = self.config.get("bitbucket.workspace") project_key = f"{workspace}_{repo_slug}".lower().replace("-", "_").replace(" ", "_") # Remove any special characters import re project_key = re.sub(r'[^a-z0-9_]', '_', project_key) # Get project name from repository info project_name = repo_info.get("name") or repo_slug description = repo_info.get("description", "") or "" logger.info(f"Generating sonar-project.properties for {repo_slug} (project key: {project_key})") try: # Generate sonar-project.properties file sonar_props_path = self.sonar_scanner.generate_sonar_properties( repo_path=repo_path, project_key=project_key, project_name=project_name ) except Exception as e: logger.error(f"Failed to generate sonar-project.properties for {repo_slug}: {e}") skipped_count += 1 continue # Get project key from properties file project_key = self.sonar_scanner.get_project_key(sonar_props_path) if not project_key: logger.warning(f"Could not extract project key from sonar-project.properties for {repo_slug}") skipped_count += 1 continue # Check if project exists in SonarQube, create if not (especially on first clone) if not self.sonar_scanner.check_project_exists(project_key): logger.info(f"Project '{project_key}' does not exist in SonarQube, creating it...") # Get project name and description from repository project_name = repo_info.get("name") or repo_slug description = repo_info.get("description", "") or "" # Try to read project name from properties if available try: with open(sonar_props_path, "r", encoding="utf-8") as f: for line in f: if line.strip().startswith("sonar.projectName"): project_name = line.split("=", 1)[1].strip() break except Exception: pass # Create project in SonarQube if self.sonar_scanner.create_project( project_key=project_key, project_name=project_name, description=description, ): logger.info(f"Successfully created SonarQube project: {project_key} ({project_name})") else: logger.warning(f"Failed to create SonarQube project '{project_key}', but continuing with scan") else: logger.debug(f"Project '{project_key}' already exists in SonarQube") # Run SonarQube scan scan_results = self.sonar_scanner.run_scan(repo_path, sonar_props_path) # Get project key and fetch results from SonarQube API project_key = self.sonar_scanner.get_project_key(sonar_props_path) sonar_results = None if project_key and scan_results.get("success"): # Wait a bit for scan to process import time time.sleep(5) sonar_results = self.sonar_scanner.get_scan_results(project_key) # Generate PDF report try: report_path = self.report_generator.generate_report( repo_name=repo_slug, scan_results=scan_results, sonar_results=sonar_results, scan_output=scan_results.get("output") ) logger.info(f"Report generated: {report_path}") except Exception as e: logger.error(f"Failed to generate report for {repo_slug}: {e}") report_path = None # Update history self._update_scan_history( repo_slug, scan_results.get("success", False), report_path ) if scan_results.get("success"): scanned_count += 1 logger.info(f"Successfully scanned {repo_slug}") else: failed_count += 1 logger.error(f"Scan failed for {repo_slug}: {scan_results.get('error')}") except Exception as e: logger.error(f"Unexpected error processing {repo_slug}: {e}", exc_info=True) failed_count += 1 self._update_scan_history(repo_slug, False) # Summary logger.info("=" * 60) logger.info("Scan Summary:") logger.info(f" Total repositories: {len(repos)}") logger.info(f" Scanned: {scanned_count}") logger.info(f" Skipped: {skipped_count}") logger.info(f" Failed: {failed_count}") logger.info("=" * 60) def main(): """Main entry point.""" try: # Load configuration config = ConfigManager() # Validate required configuration if not config.get("bitbucket.workspace"): logger.error("Bitbucket workspace not configured") sys.exit(1) # Check for either API token or app password has_api_token = bool(config.get("bitbucket.api_token")) has_app_password = bool(config.get("bitbucket.app_password")) if not config.get("bitbucket.username"): logger.error("Bitbucket username/email not configured") sys.exit(1) if not has_api_token and not has_app_password: logger.error("Bitbucket authentication not configured (need either API token or app password)") sys.exit(1) if not config.get("sonarqube.token"): logger.error("SonarQube token not configured") sys.exit(1) # Save configuration (without sensitive data) config.save_config() # Run orchestrator orchestrator = ScanOrchestrator(config) orchestrator.run_scan() except KeyboardInterrupt: logger.info("Scan interrupted by user") sys.exit(1) except Exception as e: logger.error(f"Fatal error: {e}", exc_info=True) sys.exit(1) if __name__ == "__main__": main()