"""Main analyzer orchestrator.

def __init__(self, config: AnalyzerConfig, fetch_pr_details: bool = False) -> None:

"""Initialize analyzer with configuration.

self._config = config

self._output = TerminalOutput(verbose=config.verbose)

self._client = GitHubClient(config)

self._exporter = CSVExporter(config.output_dir)

# Initialize analyzers

self._commit_analyzer = CommitAnalyzer(self._client)

self._pr_analyzer = PullRequestAnalyzer(self._client, fetch_details=fetch_pr_details)

self._issue_analyzer = IssueAnalyzer(self._client)

self._contributor_tracker = ContributorTracker()

# Storage for results

self._all_commits: list[Commit] = []

self._all_prs: list[PullRequest] = []

self._all_issues: list[Issue] = []

self._repo_stats: list[RepositoryStats] = []

self._quality_metrics: list[QualityMetrics] = []

def run(self, repositories: list[Repository]) -> None:

"""Run full analysis on all repositories.

since = datetime.now(timezone.utc) - timedelta(days=self._config.days)

self._output.log(f"Starting analysis for {len(repositories)} repositories")

self._output.log(f"Analysis period: {self._config.days} days (since {since.date()})")

# Analyze each repository

for idx, repo in enumerate(repositories, 1):

self._output.progress(idx, len(repositories), f"Analyzing {repo.full_name}")

try:

self._analyze_repository(repo, since)

except RateLimitError as e:

self._output.error("Rate limit exceeded", e.details)

break

except GitHubAnalyzerError as e:

self._output.log(f"Error analyzing {repo.full_name}: {e.message}", "warning")

continue

# Track contributors from collected data

self._track_contributors()

# Generate productivity analysis

productivity = self._contributor_tracker.generate_analysis(self._config.days)

# Export all data

files = self._export_all(productivity)

# Show summary

self._show_summary(files)

def _analyze_repository(self, repo: Repository, since: datetime) -> None:

"""Analyze a single repository.

self._output.log(f"Fetching commits for {repo.full_name}", "info")

commits = self._commit_analyzer.fetch_and_analyze(repo, since)

self._all_commits.extend(commits)

self._output.log(f"Fetching pull requests for {repo.full_name}", "info")

prs = self._pr_analyzer.fetch_and_analyze(repo, since)

self._all_prs.extend(prs)

self._output.log(f"Fetching issues for {repo.full_name}", "info")

issues = self._issue_analyzer.fetch_and_analyze(repo, since)

self._all_issues.extend(issues)

# Calculate repository stats

commit_stats = self._commit_analyzer.get_stats(commits)

pr_stats = self._pr_analyzer.get_stats(prs)

issue_stats = self._issue_analyzer.get_stats(issues)

repo_stat = RepositoryStats(

repository=repo.full_name,

total_commits=commit_stats["total"],

merge_commits=commit_stats["merge_commits"],

revert_commits=commit_stats["revert_commits"],

total_additions=commit_stats["total_additions"],

total_deletions=commit_stats["total_deletions"],

unique_authors=commit_stats["unique_authors"],

total_prs=pr_stats["total"],

merged_prs=pr_stats["merged"],

open_prs=pr_stats["open"],

avg_time_to_merge_hours=pr_stats["avg_time_to_merge_hours"],

total_issues=issue_stats["total"],

closed_issues=issue_stats["closed"],

open_issues=issue_stats["open"],

bug_issues=issue_stats["bugs"],

analysis_period_days=self._config.days,

)

self._repo_stats.append(repo_stat)

# Calculate quality metrics

quality = calculate_quality_metrics(repo, commits, prs)

self._quality_metrics.append(quality)

self._output.log(

f"{repo.full_name}: {len(commits)} commits, {len(prs)} PRs, {len(issues)} issues",

"success",

)

def _track_contributors(self) -> None:

"""Track contributor statistics from all data."""

for commit in self._all_commits:

self._contributor_tracker.record_commit(commit)

for pr in self._all_prs:

self._contributor_tracker.record_pr(pr)

for issue in self._all_issues:

self._contributor_tracker.record_issue(issue)

def _export_all(self, productivity: list) -> list[Path]:

"""Export all data to CSV files.

self._output.log("Exporting data to CSV files", "info")

files = []

files.append(self._exporter.export_commits(self._all_commits))

files.append(self._exporter.export_pull_requests(self._all_prs))

files.append(self._exporter.export_issues(self._all_issues))

files.append(self._exporter.export_repository_summary(self._repo_stats))

files.append(self._exporter.export_quality_metrics(self._quality_metrics))

files.append(self._exporter.export_productivity(productivity))

files.append(self._exporter.export_contributors(self._contributor_tracker.get_stats()))

return files

def _show_summary(self, files: list[Path]) -> None:

"""Show analysis summary.

commit_stats = self._commit_analyzer.get_stats(self._all_commits)

pr_stats = self._pr_analyzer.get_stats(self._all_prs)

issue_stats = self._issue_analyzer.get_stats(self._all_issues)

self._output.summary({

"repositories": len(self._repo_stats),

"commits": commit_stats,

"prs": pr_stats,

"issues": issue_stats,

"files": [str(f) for f in files],

})

self._output.success("Analysis complete!")

def close(self) -> None:

"""Clean up resources."""

"""Parse command-line arguments.

parser = argparse.ArgumentParser(

description="Analyze GitHub repositories and Jira projects, export metrics to CSV.",

formatter_class=argparse.RawDescriptionHelpFormatter,

epilog="""

""",

)

parser.add_argument(

"--sources", "-s",

type=str,

default="auto",

help="Data sources to analyze: auto, github, jira, or github,jira (default: auto)",

)

parser.add_argument(

"--days", "-d",

type=int,

default=None,

help="Number of days to analyze (default: 30, or GITHUB_ANALYZER_DAYS env var)",

)

parser.add_argument(

"--output", "-o",

type=str,

default=None,

help="Output directory for CSV files (default: github_export)",

)

parser.add_argument(

"--repos", "-r",

type=str,

default=None,

help="Path to repos.txt file (default: repos.txt)",

)

parser.add_argument(

"--jira-projects", "-j",

type=str,

default=None,

help="Path to jira_projects.txt file (default: jira_projects.txt)",

)

parser.add_argument(

"--quiet", "-q",

action="store_true",

help="Suppress verbose output",

)

parser.add_argument(

"--verbose", "-v",

action="store_true",

help="Enable verbose mode with API audit logging (Feature 006, FR-009)",

)

parser.add_argument(

"--full",

action="store_true",

help="Fetch full PR details (slower, includes additions/deletions per PR)",

)

"""Prompt user for yes/no answer.

default_hint = "[Y/n]" if default else "[y/N]"

try:

answer = input(f"{question} {default_hint}: ").strip().lower()

if not answer:

return default

return answer in ("y", "yes", "s", "si", "sì")

except (EOFError, KeyboardInterrupt):

print()

"""Prompt user for integer value.

try:

answer = input(f"{question} [{default}]: ").strip()

if not answer:

return default

return int(answer)

except ValueError:

print(f"Invalid number, using default: {default}")

return default

except (EOFError, KeyboardInterrupt):

print()

"""Parse sources string to list of DataSource.

sources = []

for name in sources_str.lower().split(","):

name = name.strip()

if name == "github":

sources.append(DataSource.GITHUB)

elif name == "jira":

sources.append(DataSource.JIRA)

elif name:

raise ValueError(f"Unknown source: {name}. Valid sources: github, jira")

"""Auto-detect available data sources from environment.

sources = []

# Check for GitHub token

if os.environ.get("GITHUB_TOKEN", "").strip():

sources.append(DataSource.GITHUB)

# Check for Jira credentials (all required)

jira_url = os.environ.get("JIRA_URL", "").strip()

jira_email = os.environ.get("JIRA_EMAIL", "").strip()

jira_token = os.environ.get("JIRA_API_TOKEN", "").strip()

if jira_url and jira_email and jira_token:

sources.append(DataSource.JIRA)

"""Validate that required credentials exist for sources.

for source in sources:

if source == DataSource.GITHUB:

if not os.environ.get("GITHUB_TOKEN", "").strip():

raise ValueError(

"GitHub source requested but GITHUB_TOKEN environment variable not set"

)

elif source == DataSource.JIRA:

jira_url = os.environ.get("JIRA_URL", "").strip()

jira_email = os.environ.get("JIRA_EMAIL", "").strip()

jira_token = os.environ.get("JIRA_API_TOKEN", "").strip()

if not (jira_url and jira_email and jira_token):

raise ValueError(

"Jira source requested but Jira credentials incomplete. "

"Set JIRA_URL, JIRA_EMAIL, and JIRA_API_TOKEN environment variables."

"""Validate repository name format (owner/repo).

if not repo or not repo.strip():

return False

"""Validate organization name format.

if not org or not org.strip():

return False

org = org.strip()

if len(org) > 39:

return False

# Double hyphens not allowed

if "--" in org:

return False

# Single character is valid

if len(org) == 1:

return org.isalnum()

"""Calculate activity cutoff date from number of days (Feature 005 - T007).

"""Filter repositories by pushed_at date (Feature 005 - T008).

active_repos = []

for repo in repos:

pushed_at_str = repo.get("pushed_at")

if not pushed_at_str:

# Skip repos without pushed_at (treat as inactive)

continue

try:

# Parse ISO 8601 timestamp (e.g., "2025-11-28T10:00:00Z")

pushed_at = datetime.fromisoformat(pushed_at_str.replace("Z", "+00:00"))

repo_date = pushed_at.date()

# Include if pushed_at >= cutoff (inclusive boundary per spec)

if repo_date >= cutoff:

active_repos.append(repo)

except (ValueError, AttributeError):

# Skip repos with invalid date format

continue

"""Display repository activity statistics (Feature 005 - T009).

"""Format GitHub repositories for display.

lines = []

for idx, repo in enumerate(repos, 1):

full_name = repo.get("full_name", "unknown")

is_private = repo.get("private", False)

description = repo.get("description") or ""

# Truncate description to 50 chars

if len(description) > 50:

description = description[:47] + "..."

# Build display line

if is_private:

if description:

lines.append(f" {idx}. {full_name} - [private] {description}")

else:

lines.append(f" {idx}. {full_name} - [private]")

else:

if description:

lines.append(f" {idx}. {full_name} - {description}")

else:

lines.append(f" {idx}. {full_name}")

"""Load repository names from repos.txt file.

try:

path = Path(repos_file)

if not path.exists():

return []

content = path.read_text().strip()

if not content:

return []

repos = []

for line in content.splitlines():

line = line.strip()

# Skip comments and empty lines

if not line or line.startswith("#"):

continue

# Handle full URLs

if line.startswith("http"):

# Extract owner/repo from URL

# https://github.com/owner/repo or https://github.com/owner/repo.git

parts = line.rstrip("/").rstrip(".git").split("/")

if len(parts) >= 2:

repos.append(f"{parts[-2]}/{parts[-1]}")

else:

repos.append(line)

return repos

except OSError:

"""Handle rate limit error with wait time display (FR-008, T049).

import time as time_module

if e.reset_time:

wait_seconds = max(0, e.reset_time - int(time_module.time()))

log(f"Rate limit exceeded. Waiting {wait_seconds} seconds...", "warning")

else:

repos: list[dict],

days: int,

total_count: int,

) -> tuple[list[dict] | None, int | None, bool]:

"""Handle zero active repositories scenario with options (FR-009).

print(f"⚠️ No repositories have been pushed to in the last {days} days.")

print("Options:")

print(f" [1] Include all {total_count} repositories anyway")

print(f" [2] Adjust timeframe (currently {days} days)")

print(" [3] Cancel")

try:

zero_choice = input("\nYour choice: ").strip()

except (EOFError, KeyboardInterrupt):

return None, None, True # Exit completely

if zero_choice == "1":

# Include all repos

return repos, days, False

elif zero_choice == "2":

# Adjust timeframe

try:

new_days_input = input("Enter new timeframe in days: ").strip()

new_days = int(new_days_input)

if new_days <= 0:

print("Invalid number of days. Must be positive.")

return None, None, False # Return to menu

except ValueError:

print("Invalid input. Please enter a number.")

return None, None, False # Return to menu

except (EOFError, KeyboardInterrupt):

return None, None, True # Exit completely

# Refilter with new timeframe

print("\nRechecking...")

new_cutoff = get_cutoff_date(new_days)

new_active = filter_by_activity(repos, new_cutoff)

display_activity_stats(total=total_count, active=len(new_active), days=new_days)

return new_active, new_days, False

else:

# Cancel (option 3 or any other input)

repos_file: str,

github_token: str,

interactive: bool = True,

output: TerminalOutput | None = None,

days: int = 30,

) -> list[str]:

"""Select GitHub repositories from file or interactively (Feature 004 + 005).

# Helper for consistent output

def log(msg: str, level: str = "info") -> None:

if output:

output.log(msg, level)

else:

print(msg)

# Try loading from file first (FR-001)

file_repos = load_github_repos_from_file(repos_file)

if file_repos:

return file_repos

# No file or empty - need to prompt or skip

if not interactive:

# FR-013, FR-014: Non-interactive mode skips prompts

log("No repos.txt found. Skipping GitHub analysis in non-interactive mode.", "info")

return []

# Display menu per FR-002

print("\nOptions:")

print(" [A] Analyze ALL accessible repositories")

print(" [S] Specify repository names manually (owner/repo format)")

print(" [O] Analyze organization repositories")

print(" [L] Select from list by number (e.g., 1,3,5 or 1-3)")

print(" [Q] Quit/Skip GitHub analysis")

# Create client for API calls - use provided token

config = AnalyzerConfig(github_token=github_token)

client = GitHubClient(config)

try:

while True:

try:

choice = input("\nYour choice [A/S/O/L/Q]: ").strip().upper()

except (EOFError, KeyboardInterrupt):

# FR-004: Handle EOF/KeyboardInterrupt gracefully

log("GitHub analysis skipped.", "warning")

return []

if choice == "A":

# FR-005: List all user repos with activity filtering (Feature 005)

log("Fetching your repositories...", "info")

try:

repos = client.list_user_repos()

if not repos:

log("No repositories found for your account.", "warning")

continue

# Feature 005: Apply activity filtering

cutoff = get_cutoff_date(days)

active_repos = filter_by_activity(repos, cutoff)

# Display activity statistics (FR-007)

display_activity_stats(total=len(repos), active=len(active_repos), days=days)

# Handle zero active repos (FR-009)

current_days = days

if not active_repos:

result_repos, new_days, should_exit = _handle_zero_active_repos(

repos, current_days, len(repos)

)

if should_exit:

return [] # EOF/KeyboardInterrupt - exit completely

if result_repos is None:

continue # User cancelled - return to menu

active_repos = result_repos

if new_days is not None:

current_days = new_days

# Confirmation prompt (FR-006)

try:

confirm = input(f"Proceed with {len(active_repos)} active repositories? [Y/n/all]: ").strip().lower()

except (EOFError, KeyboardInterrupt):

log("GitHub analysis skipped.", "warning")

return []

if confirm == "n":

continue # Return to menu

elif confirm == "all":

# Use all repos (bypass filter)

repo_names = [r["full_name"] for r in repos]

else:

# Default (Y or Enter): use active repos only

repo_names = [r["full_name"] for r in active_repos]

log(f"Using {len(repo_names)} repositories.", "success")

return repo_names

except RateLimitError as e:

_handle_rate_limit(e, log)

continue

except GitHubAnalyzerError as e:

log(f"Error fetching repositories: {e.message}", "error")

continue

elif choice == "S":

# FR-009: Manual specification

try:

manual_input = input("Enter repository names (owner/repo, comma-separated): ").strip()

except (EOFError, KeyboardInterrupt):

log("GitHub analysis skipped.", "warning")

return []

if not manual_input:

log("No repositories entered.", "warning")

continue

# Parse and validate (FR-011, FR-012)

manual_repos = [r.strip() for r in manual_input.split(",") if r.strip()]

valid_repos = [r for r in manual_repos if validate_repo_format(r)]

invalid_repos = [r for r in manual_repos if not validate_repo_format(r)]

if invalid_repos:

log(f"Invalid repository format ignored: {', '.join(invalid_repos)}", "warning")

if valid_repos:

log(f"Selected {len(valid_repos)} repositories: {', '.join(valid_repos)}", "success")

return valid_repos

else:

log("No valid repository names entered. Try again.", "warning")

elif choice == "O":

# FR-006: Organization repos with Search API filtering (Feature 005)

try:

org_name = input("Enter organization name: ").strip()

except (EOFError, KeyboardInterrupt):

log("GitHub analysis skipped.", "warning")

return []

if not validate_org_name(org_name):

log("Invalid organization name format.", "warning")

continue

log(f"Fetching repositories for organization '{org_name}'...", "info")

try:

# Feature 005: Use Search API for efficient org filtering

cutoff = get_cutoff_date(days)

cutoff_str = cutoff.isoformat()

# Get total org repos count for stats

all_org_repos = client.list_org_repos(org_name)

total_count = len(all_org_repos)

if total_count == 0:

log(f"No repositories found in organization '{org_name}'.", "warning")

continue

# Get active repos via Search API

search_result = client.search_active_org_repos(org_name, cutoff_str)

active_repos = search_result.get("items", [])

incomplete = search_result.get("incomplete_results", False)

# Display activity statistics (FR-007)

display_activity_stats(total=total_count, active=len(active_repos), days=days)

# Warn if results may be incomplete

if incomplete:

print("⚠️ Results may be incomplete due to API limitations.")

# Handle zero active repos (FR-009)

current_days = days

if not active_repos:

result_repos, new_days, should_exit = _handle_zero_active_repos(

all_org_repos, current_days, total_count

)

if should_exit:

return [] # EOF/KeyboardInterrupt - exit completely

if result_repos is None:

continue # User cancelled - return to menu

active_repos = result_repos

if new_days is not None:

current_days = new_days

# Confirmation prompt (FR-006)

try:

confirm = input(f"Show {len(active_repos)} active repositories for selection? [Y/n/all]: ").strip().lower()

except (EOFError, KeyboardInterrupt):

log("GitHub analysis skipped.", "warning")

return []

if confirm == "n":

continue # Return to menu

elif confirm == "all":

# Show all repos (bypass filter)

display_repos = all_org_repos

else:

# Default (Y or Enter): show active repos only

display_repos = active_repos

log(f"Showing {len(display_repos)} repositories:", "info")

print(format_repo_list(display_repos))

# Ask for selection

try:

selection_input = input("\nSelect (e.g., 1,3,5 or 1-3 or 'all'): ").strip()

except (EOFError, KeyboardInterrupt):

log("GitHub analysis skipped.", "warning")

return []

indices = parse_project_selection(selection_input, len(display_repos))

if indices:

selected = [display_repos[i]["full_name"] for i in indices]

log(f"Selected {len(selected)} repositories.", "success")

return selected

else:

log("Invalid selection.", "warning")

except RateLimitError as e:

# Feature 005 FR-008: Fallback to unfiltered mode on rate limit

_handle_rate_limit(e, log)

log("Showing all repositories without activity filter.", "info")

try:

all_org_repos = client.list_org_repos(org_name)

if all_org_repos:

log(f"Showing {len(all_org_repos)} repositories (unfiltered):", "info")

print(format_repo_list(all_org_repos))

selection_input = input("\nSelect (e.g., 1,3,5 or 1-3 or 'all'): ").strip()

indices = parse_project_selection(selection_input, len(all_org_repos))

if indices:

selected = [all_org_repos[i]["full_name"] for i in indices]

log(f"Selected {len(selected)} repositories.", "success")

return selected

except (EOFError, KeyboardInterrupt, GitHubAnalyzerError):

pass

continue

except GitHubAnalyzerError as e:

if "404" in str(e):

log(f"Organization '{org_name}' not found or not accessible.", "warning")

else:

log(f"Error fetching organization repos: {e.message}", "error")

continue

elif choice == "L":

# FR-010: Select from personal list with activity filtering (Feature 005)

log("Fetching your repositories...", "info")

try:

repos = client.list_user_repos()

if not repos:

log("No repositories found for your account.", "warning")

continue

# Feature 005: Apply activity filtering

cutoff = get_cutoff_date(days)

active_repos = filter_by_activity(repos, cutoff)

# Display activity statistics (FR-007)

display_activity_stats(total=len(repos), active=len(active_repos), days=days)

# Handle zero active repos (FR-009)

current_days = days

if not active_repos:

result_repos, new_days, should_exit = _handle_zero_active_repos(

repos, current_days, len(repos)

)

if should_exit:

return [] # EOF/KeyboardInterrupt - exit completely