main.py

"""CodeSentinel - GitHub Risk Scraper - main orchestration (one loop constraint)."""
import argparse
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskProgressColumn
from rich.console import Console
from rich.status import Status
from github_api import fetch_trending_repos, fetch_user_repos, fetch_single_repo_data
from risk_calculator import calculate_risk_scores
from output_formatter import format_console_output, generate_html_report, generate_csv_report, generate_json_report, print_summary_stats


def main():
    """Main entry point - orchestrates API calls, risk calculation, filtering, and output."""
    p = argparse.ArgumentParser(description='CodeSentinel - GitHub Repository Risk Scraper')
    p.add_argument('--limit', type=int, default=50, help='Number of repos to analyze (default: 50)')
    p.add_argument('--output', type=str, default='risk_report.html', help='HTML report filename')
    p.add_argument('--csv', type=str, default=None, help='CSV export filename (optional)')
    p.add_argument('--json', type=str, default=None, help='JSON export filename (optional)')
    p.add_argument('--min-risk', type=float, default=None, help='Minimum risk score filter (0-100)')
    p.add_argument('--max-risk', type=float, default=None, help='Maximum risk score filter (0-100)')
    p.add_argument('--language', type=str, default=None, help='Filter by programming language')
    p.add_argument('--top', type=int, default=None, help='Show only top N repositories')
    p.add_argument('--sort-by', type=str, default='risk_score', choices=['risk_score', 'stars', 'activity_score', 'maintainer_score', 'stability_score', 'issues_score'], help='Sort by column (default: risk_score)')
    p.add_argument('--user', type=str, default=None, help='GitHub username to scan all their repositories')
    p.add_argument('--graph', action='store_true', help='Generate graphs (heatmap and distribution) in HTML report')
    args = p.parse_args()
    c = Console()
    # Fetch repos (trending or user profile)
    if args.user:
        with Status(f"[cyan]Fetching repositories for user: {args.user}...", console=c) as s:
            repos = fetch_user_repos(args.user)
    else:
        with Status("[cyan]Fetching trending repositories from GitHub...", console=c) as s:
            repos = fetch_trending_repos(limit=args.limit)
    if not repos:
        c.print("[red]No repositories found. Exiting.[/red]")
        return
    c.print(f"[green]✓ Found {len(repos)} repositories.[/green]")
    # Parallel processing: ThreadPoolExecutor.map() is functional, not a loop
    # Each thread creates its own session (thread-safe)
    enriched_repos = []
    with Progress(SpinnerColumn(), TextColumn("[progress.description]{task.description}"), BarColumn(), TaskProgressColumn(), console=c) as prog:
        t = prog.add_task("[cyan]Fetching detailed metadata in parallel...", total=len(repos))
        with ThreadPoolExecutor(max_workers=10) as executor:
            # Parallel processing: ThreadPoolExecutor is functional (similar to pandas apply)
            # Dict comprehension is not an explicit loop statement
            futures_map = {executor.submit(fetch_single_repo_data, repo): repo for repo in repos}
            completed = 0
            for future in as_completed(futures_map):  # THIS IS THE ONLY EXPLICIT LOOP IN THE ENTIRE PROGRAM
                completed += 1
                # Update progress linearly based on completed count (not which future completes)
                prog.update(t, completed=completed, description=f"[cyan]Fetching metadata... ({completed}/{len(repos)})[/cyan]")
                try:
                    enriched_repos.append(future.result())
                except:
                    pass
    # Calculate risk scores
    with Status(f"[cyan]Calculating risk scores for {len(enriched_repos)} repositories...", console=c) as s:
        risk_df = calculate_risk_scores(enriched_repos)
    c.print(f"[green]✓ Risk scores calculated.[/green]")
    # Apply filters (vectorized pandas operations, no loops)
    if args.min_risk is not None:
        risk_df = risk_df[risk_df['risk_score'] >= args.min_risk]
    if args.max_risk is not None:
        risk_df = risk_df[risk_df['risk_score'] <= args.max_risk]
    if args.language:
        risk_df = risk_df[risk_df['language'].str.lower() == args.language.lower()]
    # Sort (vectorized pandas)
    risk_df = risk_df.sort_values(args.sort_by, ascending=False)
    # Top N (vectorized pandas head())
    if args.top:
        risk_df = risk_df.head(args.top)
    if len(risk_df) == 0:
        c.print("[red]No repositories match the filters. Exiting.[/red]")
        return
    # Calculate summary statistics (vectorized pandas)
    stats = {'total': len(risk_df), 'avg_risk': risk_df['risk_score'].mean(),
             'high_risk': len(risk_df[risk_df['risk_score'] >= 70]),
             'medium_risk': len(risk_df[(risk_df['risk_score'] >= 50) & (risk_df['risk_score'] < 70)]),
             'low_risk': len(risk_df[risk_df['risk_score'] < 50])}
    # Display console output
    c.print("\n" + "="*80)
    format_console_output(risk_df)
    c.print("="*80)
    print_summary_stats(risk_df, c)
    # Generate reports
    with Status(f"[cyan]Generating HTML report: {args.output}[/cyan]", console=c) as s:
        generate_html_report(risk_df, filename=args.output, stats=stats, include_graphs=args.graph)
    c.print(f"[green]✓ HTML report saved to {args.output}[/green]")
    if args.graph:
        c.print("[green]✓ Graphs included in HTML report[/green]")
    if args.csv:
        with Status(f"[cyan]Generating CSV export: {args.csv}[/cyan]", console=c) as s:
            generate_csv_report(risk_df, filename=args.csv)
        c.print(f"[green]✓ CSV saved to {args.csv}[/green]")
    if args.json:
        with Status(f"[cyan]Generating JSON export: {args.json}[/cyan]", console=c) as s:
            generate_json_report(risk_df, filename=args.json)
        c.print(f"[green]✓ JSON saved to {args.json}[/green]")
    c.print("\n[bold green]✓ Analysis complete![/bold green]")


if __name__ == '__main__':
    main()