""" Script to scan specific repositories against SonarQube. """ import sys import logging import argparse from pathlib import Path # Add parent directory to path for imports parent_dir = Path(__file__).parent.parent sys.path.insert(0, str(parent_dir)) from src.config_manager import ConfigManager from src.bitbucket_client import BitbucketClient from src.git_manager import GitManager from src.sonarqube_scanner import SonarQubeScanner from src.report_generator import ReportGenerator from src.scan_recent import _get_additional_repositories import json import time # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) class SpecificRepoScanner: """Scanner for specific repositories.""" def __init__(self, config: ConfigManager): """Initialize scanner with configuration.""" self.config = config # Initialize components 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 _update_scan_history(self, repo_slug: str, success: bool, report_path: Path = None) -> None: """Update scan history for a repository.""" from datetime import datetime 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 scan_repositories(self, repo_names: list) -> None: """ Scan specific repositories. Args: repo_names: List of repository names/slugs to scan """ logger.info(f"Starting scan for {len(repo_names)} repositories") scanned_count = 0 skipped_count = 0 failed_count = 0 for repo_slug in repo_names: logger.info(f"\n{'='*60}") logger.info(f"Processing repository: {repo_slug}") logger.info(f"{'='*60}") try: # Get repository info from Bitbucket, or from scanning.additional_repositories (e.g. GitHub) repo_info = self.bitbucket.get_repository_info(repo_slug) if not repo_info: for ar in _get_additional_repositories(self.config): if (ar.get("name") or ar.get("slug")) == repo_slug: repo_info = ar logger.info(f"Resolved '{repo_slug}' from scanning.additional_repositories") break if not repo_info: logger.warning(f"Repository '{repo_slug}' not in Bitbucket or additional_repositories, skipping") skipped_count += 1 continue # Get clone URL 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 # Clone or update repository logger.info(f"Cloning/updating repository: {repo_slug}") 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/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 = True # Check for sonar-project.properties sonar_props_path = self.git_manager.get_sonar_properties_path(repo_path) if not sonar_props_path: # Generate project key workspace = self.config.get("bitbucket.workspace") project_key = f"{workspace}_{repo_slug}".lower().replace("-", "_").replace(" ", "_") import re project_key = re.sub(r'[^a-z0-9_]', '_', project_key) project_name = repo_info.get("name") or repo_slug logger.info(f"Generating sonar-project.properties for {repo_slug} (project key: {project_key})") try: 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 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 if not self.sonar_scanner.check_project_exists(project_key): logger.info(f"Project '{project_key}' does not exist in SonarQube, creating it...") project_name = repo_info.get("name") or repo_slug description = repo_info.get("description", "") or "" 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 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 logger.info(f"Running SonarQube scan for {repo_slug}...") scan_results = self.sonar_scanner.run_scan(repo_path, sonar_props_path) # Fetch results from SonarQube API sonar_results = None if project_key and scan_results.get("success"): logger.info("Waiting for scan to process...") 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"[SUCCESS] Completed scan for {repo_slug}") else: failed_count += 1 logger.error(f"[FAILED] Scan failed for {repo_slug}: {scan_results.get('error', 'Unknown 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(f"\n{'='*60}") logger.info("Scan Summary") logger.info(f"{'='*60}") logger.info(f"Successfully scanned: {scanned_count}") logger.info(f"Skipped: {skipped_count}") logger.info(f"Failed: {failed_count}") logger.info(f"Total: {len(repo_names)}") def main(): """Main entry point.""" parser = argparse.ArgumentParser( description="Scan specific repositories against SonarQube", formatter_class=argparse.RawDescriptionHelpFormatter ) parser.add_argument( "repos", nargs="+", help="Repository names/slugs to scan" ) args = parser.parse_args() try: # Load configuration logger.info("Loading configuration...") config = ConfigManager() # Initialize scanner scanner = SpecificRepoScanner(config) # Scan repositories scanner.scan_repositories(args.repos) except KeyboardInterrupt: logger.info("Interrupted by user") sys.exit(1) except Exception as e: logger.error(f"Unexpected error: {e}", exc_info=True) sys.exit(1) if __name__ == "__main__": main()