open-source and leadership: The Ultimate Showdown guide for Scalability

import os
import time
import logging
from typing import Dict, List, Optional
from dataclasses import dataclass
import requests
from requests.exceptions import RequestException, HTTPError

# Configure logging for production visibility
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

@dataclass
class ContributorMetrics:
    """Data class to store per-contributor OSS metrics"""
    username: str
    total_prs: int
    merged_prs: int
    avg_pr_lead_time_hours: float
    critical_repo_contributions: int

class OSSContributionTracker:
    """Scalable tracker for OSS contribution metrics across orgs"""
    GITHUB_API_BASE = "https://api.github.com"
    # Rate limit buffer: stay 10% under GitHub's 5000 req/hour limit for enterprise
    RATE_LIMIT_BUFFER = 0.9

    def __init__(self, github_token: str, org_name: str, critical_repos: List[str]):
        self.github_token = github_token
        self.org_name = org_name
        self.critical_repos = critical_repos
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"token {self.github_token}",
            "Accept": "application/vnd.github.v3+json"
        })
        self.rate_limit_remaining = 5000
        self.rate_limit_reset = 0

    def _check_rate_limit(self, response: requests.Response) -> None:
        """Update rate limit state from GitHub API response"""
        self.rate_limit_remaining = int(response.headers.get("X-RateLimit-Remaining", 0))
        self.rate_limit_reset = int(response.headers.get("X-RateLimit-Reset", 0))
        if self.rate_limit_remaining < 100:
            sleep_time = max(self.rate_limit_reset - time.time(), 0) + 10
            logger.warning(f"Low rate limit: {self.rate_limit_remaining} remaining. Sleeping {sleep_time:.0f}s")
            time.sleep(sleep_time)

    def _make_github_request(self, endpoint: str, params: Optional[Dict] = None) -> List[Dict]:
        """Make authenticated GitHub API request with error handling and rate limiting"""
        url = f"{self.GITHUB_API_BASE}{endpoint}"
        try:
            response = self.session.get(url, params=params or {})
            response.raise_for_status()
            self._check_rate_limit(response)
            return response.json()
        except HTTPError as e:
            if e.response.status_code == 403:
                logger.error(f"Rate limit exceeded or insufficient permissions: {e}")
                raise
            logger.error(f"HTTP error for {url}: {e}")
            raise
        except RequestException as e:
            logger.error(f"Network error for {url}: {e}")
            raise

    def get_org_contributors(self) -> List[str]:
        """Fetch all contributors across critical org repos"""
        contributors = set()
        for repo in self.critical_repos:
            logger.info(f"Fetching contributors for {self.org_name}/{repo}")
            # Paginate through all contributors (GitHub returns max 30 per page by default)
            page = 1
            while True:
                repo_contributors = self._make_github_request(
                    f"/repos/{self.org_name}/{repo}/contributors",
                    params={"page": page, "per_page": 100}
                )
                if not repo_contributors:
                    break
                for contributor in repo_contributors:
                    contributors.add(contributor["login"])
                page += 1
        return list(contributors)

    def calculate_contributor_metrics(self, username: str) -> ContributorMetrics:
        """Calculate aggregated metrics for a single contributor"""
        total_prs = 0
        merged_prs = 0
        total_lead_time = 0.0
        critical_repo_prs = 0

        for repo in self.critical_repos:
            # Fetch PRs by this contributor for the repo
            prs = self._make_github_request(
                f"/repos/{self.org_name}/{repo}/pulls",
                params={"creator": username, "state": "all", "per_page": 100}
            )
            for pr in prs:
                total_prs += 1
                if pr["merged_at"]:
                    merged_prs += 1
                    # Calculate lead time: merged_at - created_at in hours
                    created = time.strptime(pr["created_at"], "%Y-%m-%dT%H:%M:%SZ")
                    merged = time.strptime(pr["merged_at"], "%Y-%m-%dT%H:%M:%SZ")
                    lead_time = time.mktime(merged) - time.mktime(created)
                    total_lead_time += lead_time / 3600  # Convert to hours
                critical_repo_prs += 1

        avg_lead_time = total_lead_time / merged_prs if merged_prs > 0 else 0.0
        return ContributorMetrics(
            username=username,
            total_prs=total_prs,
            merged_prs=merged_prs,
            avg_pr_lead_time_hours=avg_lead_time,
            critical_repo_contributions=critical_repo_prs
        )

if __name__ == "__main__":
    # Example usage: Replace with your org's details
    # Get token from https://github.com/settings/tokens (need repo scope)
    GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
    if not GITHUB_TOKEN:
        raise ValueError("Missing GITHUB_TOKEN environment variable")

    tracker = OSSContributionTracker(
        github_token=GITHUB_TOKEN,
        org_name="your-org-name",
        critical_repos=["web-app", "api-gateway", "data-pipeline"]  # From https://github.com/your-org/web-app etc.
    )

    logger.info("Starting contributor metrics collection")
    contributors = tracker.get_org_contributors()
    logger.info(f"Found {len(contributors)} total contributors")

    metrics = []
    for username in contributors[:10]:  # Limit to first 10 for demo
        try:
            user_metrics = tracker.calculate_contributor_metrics(username)
            metrics.append(user_metrics)
            logger.info(f"Processed {username}: {user_metrics.merged_prs} merged PRs")
        except Exception as e:
            logger.error(f"Failed to process {username}: {e}")

    # Output top contributors by merged PRs
    metrics.sort(key=lambda x: x.merged_prs, reverse=True)
    print("\nTop OSS Contributors:")
    for m in metrics:
        print(f"{m.username}: {m.merged_prs} merged PRs, {m.avg_pr_lead_time_hours:.1f}h avg lead time")

import os
import json
import logging
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
import requests
from requests.exceptions import RequestException
import semver

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

@dataclass
class DependencyRisk:
    """Risk assessment for a single OSS dependency"""
    name: str
    current_version: str
    latest_version: str
    cve_count: int
    license: str
    is_compliant_license: bool
    risk_score: int  # 0-100, higher = riskier

class OSSDependencyScanner:
    """Scalable OSS dependency risk scanner with CVE and license checks"""
    NPM_REGISTRY = "https://registry.npmjs.org"
    PYPI_REGISTRY = "https://pypi.org/pypi"
    # Approved licenses: OSI-approved, no copyleft for internal tools
    APPROVED_LICENSES = {"MIT", "Apache-2.0", "BSD-3-Clause", "BSD-2-Clause", "ISC"}

    def __init__(self, repo_path: str, ecosystem: str):
        self.repo_path = repo_path
        self.ecosystem = ecosystem  # "npm" or "pypi"
        self.session = requests.Session()
        self.session.headers.update({"Accept": "application/json"})

    def _parse_dependencies(self) -> List[Tuple[str, str]]:
        """Parse dependency file based on ecosystem"""
        deps = []
        try:
            if self.ecosystem == "npm":
                lock_path = os.path.join(self.repo_path, "package-lock.json")
                with open(lock_path, "r") as f:
                    lock_data = json.load(f)
                    for name, dep in lock_data.get("dependencies", {}).items():
                        deps.append((name, dep.get("version", "0.0.0")))
            elif self.ecosystem == "pypi":
                lock_path = os.path.join(self.repo_path, "requirements.txt")
                with open(lock_path, "r") as f:
                    for line in f:
                        line = line.strip()
                        if not line or line.startswith("#"):
                            continue
                        # Parse == pinned versions
                        if "==" in line:
                            name, version = line.split("==")
                            deps.append((name.strip(), version.strip()))
                        else:
                            logger.warning(f"Unpinned dependency: {line}")
            else:
                raise ValueError(f"Unsupported ecosystem: {self.ecosystem}")
        except FileNotFoundError as e:
            logger.error(f"Dependency file not found: {e}")
            raise
        except json.JSONDecodeError as e:
            logger.error(f"Invalid dependency file: {e}")
            raise
        return deps

    def _get_latest_version(self, dep_name: str) -> Optional[str]:
        """Fetch latest version from package registry"""
        try:
            if self.ecosystem == "npm":
                response = self.session.get(f"{self.NPM_REGISTRY}/{dep_name}/latest")
                response.raise_for_status()
                return response.json().get("version")
            elif self.ecosystem == "pypi":
                response = self.session.get(f"{self.PYPI_REGISTRY}/{dep_name}/json")
                response.raise_for_status()
                return response.json().get("info", {}).get("version")
        except RequestException as e:
            logger.error(f"Failed to fetch latest version for {dep_name}: {e}")
            return None

    def _get_cve_count(self, dep_name: str, version: str) -> int:
        """Check CVE count via OSV API (https://osv.dev)"""
        try:
            response = self.session.post(
                "https://api.osv.dev/v1/query",
                json={"package": {"name": dep_name, "ecosystem": self.ecosystem}, "version": version}
            )
            response.raise_for_status()
            return len(response.json().get("vulns", []))
        except RequestException as e:
            logger.error(f"Failed to check CVEs for {dep_name}@{version}: {e}")
            return 0

    def _check_license_compliance(self, dep_name: str) -> Tuple[str, bool]:
        """Check dependency license against approved list"""
        try:
            if self.ecosystem == "npm":
                response = self.session.get(f"{self.NPM_REGISTRY}/{dep_name}/latest")
                response.raise_for_status()
                license = response.json().get("license", "Unknown")
            elif self.ecosystem == "pypi":
                response = self.session.get(f"{self.PYPI_REGISTRY}/{dep_name}/json")
                response.raise_for_status()
                license = response.json().get("info", {}).get("license", "Unknown")
            is_compliant = license in self.APPROVED_LICENSES
            return license, is_compliant
        except RequestException as e:
            logger.error(f"Failed to check license for {dep_name}: {e}")
            return "Unknown", False

    def scan_dependencies(self) -> List[DependencyRisk]:
        """Run full dependency risk scan"""
        deps = self._parse_dependencies()
        risks = []
        for dep_name, current_version in deps:
            logger.info(f"Scanning {dep_name}@{current_version}")
            latest_version = self._get_latest_version(dep_name)
            cve_count = self._get_cve_count(dep_name, current_version)
            license, is_compliant = self._check_license_compliance(dep_name)

            # Calculate risk score: 40% CVEs, 30% outdated, 30% license
            risk_score = 0
            if cve_count > 0:
                risk_score += min(cve_count * 10, 40)  # Max 40 for CVEs
            if latest_version:
                try:
                    # Compare semver versions, skip if invalid
                    if semver.compare(current_version, latest_version) < 0:
                        risk_score += 30  # Outdated by any amount
                except ValueError:
                    logger.warning(f"Invalid semver for {dep_name}: {current_version}")
            if not is_compliant:
                risk_score += 30

            risks.append(DependencyRisk(
                name=dep_name,
                current_version=current_version,
                latest_version=latest_version or "Unknown",
                cve_count=cve_count,
                license=license,
                is_compliant_license=is_compliant,
                risk_score=risk_score
            ))
        return risks

if __name__ == "__main__":
    # Example usage: Scan a local repo's dependencies
    # For demo, assumes repo is cloned from https://github.com/your-org/web-app
    repo_path = os.getenv("REPO_PATH", "./web-app")
    ecosystem = os.getenv("ECOSYSTEM", "npm")

    scanner = OSSDependencyScanner(repo_path=repo_path, ecosystem=ecosystem)
    try:
        risks = scanner.scan_dependencies()
        # Sort by risk score descending
        risks.sort(key=lambda x: x.risk_score, reverse=True)
        print("\nOSS Dependency Risk Report:")
        print(f"{'Dependency':<30} {'Current':<10} {'Latest':<10} {'CVEs':<5} {'License':<12} {'Compliant':<10} {'Risk Score':<10}")
        for r in risks:
            print(f"{r.name:<30} {r.current_version:<10} {r.latest_version:<10} {r.cve_count:<5} {r.license:<12} {r.is_compliant_license:<10} {r.risk_score:<10}")
    except Exception as e:
        logger.error(f"Scan failed: {e}")
        raise

import React, { useState, useEffect } from "react";
import { LineChart, Line, XAxis, YAxis, CartesianGrid, Tooltip, Legend, ResponsiveContainer } from "recharts";
import { api } from "./api";  // Assume api client is configured with https://github.com/your-org/oss-tracker-api

// Types matching the backend ContributorMetrics from first code example
interface ContributorMetrics {
  username: string;
  total_prs: number;
  merged_prs: number;
  avg_pr_lead_time_hours: number;
  critical_repo_contributions: number;
}

interface OSSLeadershipDashboardProps {
  orgName: string;
  criticalRepos: string[];
  githubToken: string;
}

// Error boundary for dashboard failures
interface ErrorBoundaryState {
  hasError: boolean;
  error: Error | null;
}

class ErrorBoundary extends React.Component<{ children: React.ReactNode }, ErrorBoundaryState> {
  constructor(props: { children: React.ReactNode }) {
    super(props);
    this.state = { hasError: false, error: null };
  }

  static getDerivedStateFromError(error: Error) {
    return { hasError: true, error };
  }

  componentDidCatch(error: Error, errorInfo: React.ErrorInfo) {
    console.error("Dashboard error:", error, errorInfo);
  }

  render() {
    if (this.state.hasError) {
      return (

          OSS Dashboard Failed to Load
          {this.state.error?.message || "Unknown error"}
           this.setState({ hasError: false, error: null })}>
            Retry


      );
    }
    return this.props.children;
  }
}

const OSSLeadershipDashboard: React.FC = ({
  orgName,
  criticalRepos,
  githubToken
}) => {
  const [metrics, setMetrics] = useState([]);
  const [loading, setLoading] = useState(true);
  const [error, setError] = useState(null);
  const [timeRange, setTimeRange] = useState<"7d" | "30d" | "90d">("30d");

  // Fetch metrics from backend API (deployed from first code example)
  const fetchMetrics = async () => {
    setLoading(true);
    setError(null);
    try {
      const response = await api.get(`/oss-metrics`, {
        params: {
          org: orgName,
          repos: criticalRepos.join(","),
          time_range: timeRange,
          token: githubToken
        }
      });
      // Validate response data
      if (!Array.isArray(response.data)) {
        throw new Error("Invalid metrics response: expected array");
      }
      setMetrics(response.data);
    } catch (err) {
      const message = err instanceof Error ? err.message : "Failed to fetch OSS metrics";
      console.error("Metrics fetch error:", message);
      setError(message);
    } finally {
      setLoading(false);
    }
  };

  useEffect(() => {
    fetchMetrics();
    // Refresh every 5 minutes
    const interval = setInterval(fetchMetrics, 5 * 60 * 1000);
    return () => clearInterval(interval);
  }, [orgName, criticalRepos, timeRange, githubToken]);

  // Prepare chart data: merged PRs per contributor
  const chartData = metrics.map(m => ({
    username: m.username,
    mergedPRs: m.merged_prs,
    avgLeadTime: m.avg_pr_lead_time_hours
  }));

  if (loading) {
    return Loading OSS Leadership Metrics...;
  }

  if (error) {
    return (

        Error Loading Metrics
        {error}
        Retry

    );
  }

  return (



          OSS Leadership Dashboard: {orgName}

            Time Range:
             setTimeRange(e.target.value as "7d" | "30d" | "90d")}
            >
              Last 7 Days
              Last 30 Days
              Last 90 Days






            Total Contributors
            {metrics.length}


            Total Merged PRs
            {metrics.reduce((sum, m) => sum + m.merged_prs, 0)}


            Avg PR Lead Time

              {(metrics.reduce((sum, m) => sum + m.avg_pr_lead_time_hours, 0) / metrics.length || 0).toFixed(1)}h





          Merged PRs per Contributor














          Contributor Details

              {metrics.map(m => (

              ))}



                Username
                Total PRs
                Merged PRs
                Avg Lead Time (h)
                Critical Repo Contributions



                  {m.username}
                  {m.total_prs}
                  {m.merged_prs}
                  {m.avg_pr_lead_time_hours.toFixed(1)}
                  {m.critical_repo_contributions}





          Data sourced from OSS Tracker API
          Last updated: {new Date().toLocaleString()}



  );
};

export default OSSLeadershipDashboard;

# Example CODEOWNERS file for https://github.com/your-org/api-gateway
# Auth module owners
/src/auth/* @your-org/auth-team
# OSS dependency upgrades
/package.json @your-org/oss-lead
/requirements.txt @your-org/oss-lead
# Docs
/docs/* @your-org/docs-team
# All other files fallback to OSS lead
* @your-org/oss-lead

# Example GitHub Actions workflow for OSS compliance
name: OSS Compliance Check
on: [pull_request]

jobs:
  oss-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run OSS Dependency Scanner
        uses: your-org/oss-scanner-action@v1
        with:
          ecosystem: npm
          repo-path: .
          github-token: ${{ secrets.GITHUB_TOKEN }}
      - name: Check CVEs
        uses: actions/dependency-review-action@v3
        with:
          fail-on-severity: high

# Example Prometheus metrics exported by OSS Contribution Tracker
oss_pr_merge_time_seconds{repo="api-gateway", percentile="99"} 110
oss_pr_merge_time_seconds{repo="api-gateway", percentile="50"} 24
oss_contributor_count{org="your-org"} 42
oss_compliance_violations{org="your-org", quarter="2024Q2"} 0

oss-scalability-guide/
├── LICENSE
├── README.md
├── contribution-tracker/       # First code example: OSS contribution metrics
│   ├── requirements.txt
│   ├── tracker.py
│   └── Dockerfile
├── dependency-scanner/         # Second code example: OSS dependency risk scanner
│   ├── requirements.txt
│   ├── scanner.py
│   └── Dockerfile
├── leadership-dashboard/       # Third code example: React OSS dashboard
│   ├── src/
│   │   ├── OSSLeadershipDashboard.tsx
│   │   └── api.ts
│   ├── package.json
│   └── tsconfig.json
├── .github/                    # CI/CD workflows
│   └── workflows/
│       └── oss-compliance.yml
└── docs/                       # Additional guides
    ├── CODEOWNERS-guide.md
    └── metrics-guide.md