github_ingestor.py

import os
import json
import time
import logging
from datetime import datetime, timezone
from pathlib import Path

import requests
try:
    from dotenv import load_dotenv
except ModuleNotFoundError:
    def load_dotenv(*args, **kwargs):
        return False

load_dotenv()

# Config

GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
BRONZE_PATH   = Path("data/bronze/github")
LOGS_PATH     = Path("data/logs")

TARGET_PACKAGES = [
    {"pypi": "pandas",       "github": "pandas-dev/pandas",              "so_tag": "pandas"},
    {"pypi": "requests",     "github": "psf/requests",                   "so_tag": "python-requests"},
    {"pypi": "fastapi",      "github": "tiangolo/fastapi",               "so_tag": "fastapi"},
    {"pypi": "numpy",        "github": "numpy/numpy",                    "so_tag": "numpy"},
    {"pypi": "flask",        "github": "pallets/flask",                  "so_tag": "flask"},
    {"pypi": "django",       "github": "django/django",                  "so_tag": "django"},
    {"pypi": "scikit-learn", "github": "scikit-learn/scikit-learn",      "so_tag": "scikit-learn"},
    {"pypi": "torch",        "github": "pytorch/pytorch",                "so_tag": "pytorch"},
]

# Logging setup

LOGS_PATH.mkdir(parents=True, exist_ok=True)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    handlers=[
        logging.FileHandler(LOGS_PATH / "github_ingestor.log", encoding="utf-8"),
        logging.StreamHandler(stream=open(os.devnull, 'w')),  # suppress console
    ],
)
logger = logging.getLogger(__name__)


def log_event(event_type: str, package: str, status: str, details: dict):
    """Write one structured JSON log entry."""
    entry = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "event":     event_type,
        "package":   package,
        "status":    status,
        **details,
    }
    log_file = LOGS_PATH / "github_ingestor_structured.jsonl"
    with open(log_file, "a") as f:
        f.write(json.dumps(entry) + "\n")


# HTTP helper

def _get(url: str, params: dict = None) -> requests.Response:
    """Authenticated GET that sleeps if rate limit is getting low."""
    headers = {
        "Authorization": f"token {GITHUB_TOKEN}",
        "Accept":        "application/vnd.github+json",
    }
    response = requests.get(url, headers=headers, params=params, timeout=15)

    remaining = int(response.headers.get("X-RateLimit-Remaining", 1))
    reset_at   = int(response.headers.get("X-RateLimit-Reset",     0))

    if remaining < 10:
        wait = max(0, reset_at - int(time.time())) + 5
        logger.warning(f"Rate limit low ({remaining} left). Sleeping {wait}s.")
        time.sleep(wait)

    response.raise_for_status()
    return response


# Save helper

def _save(subfolder: str, filename: str, data: dict):
    """Append data to an existing JSON file (or create it)."""
    dest = BRONZE_PATH / subfolder
    dest.mkdir(parents=True, exist_ok=True)
    filepath = dest / filename

    records = []
    if filepath.exists():
        try:
            with open(filepath, "r", encoding="utf-8") as f:
                existing = json.load(f)
            if isinstance(existing, list):
                records = existing
            elif existing is None:
                records = []
            else:
                records = [existing]
        except Exception as exc:
            backup = filepath.with_suffix(filepath.suffix + ".corrupt")
            filepath.replace(backup)
            logger.warning(f"Corrupt JSON moved to {backup}: {exc}")
            records = []

    records.append(data)

    with open(filepath, "w", encoding="utf-8") as f:
        json.dump(records, f, indent=2, ensure_ascii=False)

    logger.info(f"Saved -> {filepath} (records={len(records)})")
    return filepath, len(records)


# Ingestion functions

def ingest_repo_metadata(pkg: dict) -> dict | None:
    """Fetch repo metadata (stars, forks, issues etc). Returns None on failure."""
    repo  = pkg["github"]
    pypi  = pkg["pypi"]
    url   = f"https://api.github.com/repos/{repo}"

    try:
        response = _get(url)
        raw      = response.json()

        envelope = {
            "_ingested_at":  datetime.now(timezone.utc).isoformat(),
            "_source":       "github_rest_api",
            "_pypi_package": pypi,
            "_endpoint":     url,
            "data":          raw,
        }

        filename = f"{pypi}.json"
        filepath, record_count = _save("repos", filename, envelope)

        log_event("repo_metadata", pypi, "success", {
            "file":       str(filepath),
            "bytes":      os.path.getsize(filepath),
            "stars":      raw.get("stargazers_count"),
            "forks":      raw.get("forks_count"),
            "open_issues": raw.get("open_issues_count"),
            "records_in_file": record_count,
        })
        return raw

    except requests.HTTPError as e:
        logger.error(f"[{pypi}] repo metadata failed: {e}")
        log_event("repo_metadata", pypi, "error", {"error": str(e)})
        return None


def ingest_readme(pkg: dict):
    """Fetch and decode the README (base64 -> plain text)."""
    repo  = pkg["github"]
    pypi  = pkg["pypi"]
    url   = f"https://api.github.com/repos/{repo}/readme"

    try:
        response  = _get(url)
        raw       = response.json()

        import base64
        content_b64  = raw.get("content", "")
        readme_text  = base64.b64decode(content_b64).decode("utf-8", errors="replace")

        envelope = {
            "_ingested_at":  datetime.now(timezone.utc).isoformat(),
            "_source":       "github_rest_api",
            "_pypi_package": pypi,
            "_endpoint":     url,
            "raw_api_response": raw,   # original envelope preserved
            "readme_text":   readme_text,   # decoded for convenience
        }

        filename = f"{pypi}.json"
        filepath, record_count = _save("readmes", filename, envelope)

        log_event("readme", pypi, "success", {
            "file":           str(filepath),
            "readme_chars":   len(readme_text),
            "readme_encoding": raw.get("encoding"),
            "records_in_file": record_count,
        })

    except requests.HTTPError as e:
        logger.error(f"[{pypi}] README fetch failed: {e}")
        log_event("readme", pypi, "error", {"error": str(e)})


def ingest_events(per_page: int = 100):
    """Grab one page from the public Events API."""
    url = "https://api.github.com/events"

    try:
        response = _get(url, params={"per_page": per_page})
        events   = response.json()

        envelope = {
            "_ingested_at": datetime.now(timezone.utc).isoformat(),
            "_source":      "github_events_api",
            "_endpoint":    url,
            "_event_count": len(events),
            "events":       events,
        }

        filename = "events.json"
        filepath, record_count = _save("events", filename, envelope)

        type_counts = {}
        for ev in events:
            t = ev.get("type", "unknown")
            type_counts[t] = type_counts.get(t, 0) + 1

        log_event("events_poll", "global", "success", {
            "file":         str(filepath),
            "event_count":  len(events),
            "type_breakdown": type_counts,
            "records_in_file": record_count,
        })

    except requests.HTTPError as e:
        logger.error(f"Events API failed: {e}")
        log_event("events_poll", "global", "error", {"error": str(e)})


def ingest_contributors(pkg: dict):
    """Fetch top-100 contributors for a repo."""
    repo  = pkg["github"]
    pypi  = pkg["pypi"]
    url   = f"https://api.github.com/repos/{repo}/contributors"

    try:
        response     = _get(url, params={"per_page": 100, "anon": "false"})
        contributors = response.json()

        envelope = {
            "_ingested_at":  datetime.now(timezone.utc).isoformat(),
            "_source":       "github_rest_api",
            "_pypi_package": pypi,
            "_endpoint":     url,
            "contributors":  contributors,
        }

        filename = f"{pypi}.json"
        filepath, record_count = _save("contributors", filename, envelope)

        log_event("contributors", pypi, "success", {
            "file":              str(filepath),
            "contributor_count": len(contributors),
            "records_in_file": record_count,
        })

    except requests.HTTPError as e:
        logger.error(f"[{pypi}] contributors failed: {e}")
        log_event("contributors", pypi, "error", {"error": str(e)})


# Main

def run():
    run_ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    logging.info("-" * 60)
    logger.info(f" GitHub Bronze ingestion started | run={run_ts}")
    logging.info("-" * 60)

    stats = {"success": 0, "error": 0}

    for pkg in TARGET_PACKAGES:
        pypi = pkg["pypi"]
        logger.info(f"[LOG] Processing: {pypi}")

        result = ingest_repo_metadata(pkg)
        if result:
            stats["success"] += 1
        else:
            stats["error"] += 1

        ingest_readme(pkg)
        ingest_contributors(pkg)

        time.sleep(1)

    ingest_events()

    logging.info("-" * 60)
    logger.info(
        f"Run complete | success={stats['success']} errors={stats['error']}"
    )
    logging.info("-" * 60)
    log_event("run_complete", "all", "success", stats)


if __name__ == "__main__":
    if not GITHUB_TOKEN:
        raise EnvironmentError("GITHUB_TOKEN not set in .env")
    run()