""" Quick scan orchestration script for Bitbucket SonarQube automation. Scans only the top 20 repositories with the most recent commits. """ import sys import json import logging from pathlib import Path from datetime import datetime from typing import Dict, Optional, List, Tuple # 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 for quick scans (top 20 repos).""" 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/"): 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 _get_repos_with_recent_commits(self, repos: List[Dict], limit: int = 20) -> List[Dict]: """ Get repositories sorted by most recent commit date. Args: repos: List of repository info dictionaries limit: Maximum number of repos to return Returns: List of top N repositories sorted by commit date (most recent first) """ logger.info(f"Fetching commit dates for {len(repos)} repositories to find the {limit} most recent...") repos_with_dates: List[Tuple[datetime, Dict]] = [] repos_without_dates: List[Dict] = [] for repo_info in repos: repo_slug = repo_info.get("name") or repo_info.get("slug") if not repo_slug: continue # Only process repos from the configured workspace (skip additional repos for sorting) workspace_slug = repo_info.get("workspace", {}).get("slug") if workspace_slug != self.config.get("bitbucket.workspace"): # Additional repos - add to list without date (will be included separately) repos_without_dates.append(repo_info) continue try: # Fetch latest commit for this repository latest_commit = self.bitbucket.get_latest_commit(repo_slug) if latest_commit and latest_commit.get("date"): try: # Parse ISO 8601 date from Bitbucket commit_date_str = latest_commit.get("date") commit_date = datetime.fromisoformat(commit_date_str.replace("Z", "+00:00")) # Convert to UTC if timezone-aware if commit_date.tzinfo: commit_date = commit_date.replace(tzinfo=None) repos_with_dates.append((commit_date, repo_info)) logger.debug(f"Found commit date for {repo_slug}: {commit_date}") except Exception as e: logger.warning(f"Failed to parse commit date for {repo_slug}: {e}") # Use a very old date as fallback so it goes to the end repos_with_dates.append((datetime.min, repo_info)) else: logger.warning(f"Could not fetch commit date for {repo_slug}, using fallback") # Use a very old date as fallback repos_with_dates.append((datetime.min, repo_info)) except Exception as e: logger.warning(f"Error fetching commit date for {repo_slug}: {e}") # Use a very old date as fallback repos_with_dates.append((datetime.min, repo_info)) # Sort by commit date descending (most recent first) repos_with_dates.sort(key=lambda x: x[0], reverse=True) # Take top N repos selected_repos = [repo_info for _, repo_info in repos_with_dates[:limit]] logger.info(f"Selected {len(selected_repos)} repositories with most recent commits") if selected_repos: # Log the selected repos for i, repo_info in enumerate(selected_repos[:5], 1): # Log first 5 repo_slug = repo_info.get("name") or repo_info.get("slug") logger.info(f" {i}. {repo_slug}") if len(selected_repos) > 5: logger.info(f" ... and {len(selected_repos) - 5} more") # Add additional repos (they don't count toward the limit) if repos_without_dates: logger.info(f"Adding {len(repos_without_dates)} additional repositories from config") selected_repos.extend(repos_without_dates) return selected_repos def run_scan(self) -> None: """Run the complete scanning workflow for top 20 repos.""" logger.info("Starting SonarQube quick scan automation (top 20 repos with recent commits)") # Get all repositories from workspace all_repos = self.bitbucket.get_all_repositories() or [] if not all_repos: logger.warning("No repositories found in workspace") return logger.info(f"Found {len(all_repos)} repositories in workspace") # Get top 20 repos with most recent commits repos = self._get_repos_with_recent_commits(all_repos, limit=20) if not repos: logger.warning("No repositories selected for scanning") return logger.info(f"Processing {len(repos)} repositories (top 20 + additional repos)") 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("Quick Scan Summary:") logger.info(f" Total repositories processed: {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()