STRATA — DC2: Ecosystem Artifacts (Issues, PRs, CI) — Data Collection II

from datetime import datetime, timezone
from typing import Optional, Iterable, Dict, Any

# =============================
# DC2: Ecosystem Artifacts
# =============================

def _normalize_timestamp_to_utc(value: Any) -> datetime:
    """Coerce timestamps to timezone-aware UTC datetime."""
    if value is None:
        return None
    if isinstance(value, datetime):
        # If already datetime, ensure it's UTC-aware
        if value.tzinfo is None:
            return value.replace(tzinfo=timezone.utc)
        return value.astimezone(timezone.utc)
    
    # Parse string timestamp
    v = str(value)
    
    # Try common formats with explicit UTC 'Z' suffix
    for fmt in ("%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%dT%H:%M:%SZ"):
        try:
            dt = datetime.strptime(v, fmt)
            return dt.replace(tzinfo=timezone.utc)
        except ValueError:
            continue
    
    # Try formats without timezone (assume UTC)
    for fmt in ("%Y-%m-%dT%H:%M:%S.%f", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
        try:
            dt = datetime.strptime(v, fmt)
            return dt.replace(tzinfo=timezone.utc)
        except ValueError:
            continue
    
    # Fallback: fromisoformat
    try:
        dt = datetime.fromisoformat(v.replace('Z', '+00:00'))
        return dt.astimezone(timezone.utc) if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
    except Exception:
        raise ValueError(f"Unrecognized timestamp: {value}")

# ---- Issues ----

def upsert_issue(provider: str, repo: str, issue: Dict[str, Any]) -> int:
    """Insert/update a single issue idempotently; return db id.
    
    Implementation guidance:
    - Build an INSERT ... ON CONFLICT (provider, repo, issue_number) DO UPDATE statement.
    - Extract: provider, repo, issue["number"], issue["title"], issue["author"], 
      issue["state"], issue["created_at"], issue["closed_at"].
    - Use _normalize_timestamp_to_utc() for timestamp fields.
    - ON UPDATE: set title=EXCLUDED.title, state=EXCLUDED.state,
      author=COALESCE(EXCLUDED.author, issues.author),
      created_at=LEAST(issues.created_at, EXCLUDED.created_at),
      closed_at=COALESCE(EXCLUDED.closed_at, issues.closed_at).
    - Use RETURNING id to get the row id.
    - If no rows returned (shouldn't happen with RETURNING), fallback SELECT.
    """
    # TODO: Implement upsert logic
    pass

def ingest_issues(provider: str, repo: str, issues: Iterable[Dict[str, Any]]) -> int:
    """Ingest multiple issues; return count processed.
    
    Implementation guidance:
    - Loop over issues iterable.
    - Call upsert_issue(provider, repo, issue) for each.
    - Return total count processed.
    """
    # TODO: Implement
    pass

# ---- Pull Requests ----

def upsert_pull_request(provider: str, repo: str, pr: Dict[str, Any]) -> int:
    """Insert/update a single pull/merge request idempotently; return db id.
    
    Implementation guidance:
    - Similar to upsert_issue, but for pull_requests table.
    - Extract: provider, repo, pr["number"], pr["title"], pr["author"],
      pr["state"], pr["created_at"], pr["merged_at"], pr["closed_at"].
    - Use _normalize_timestamp_to_utc() for timestamp fields.
    - ON CONFLICT (provider, repo, pr_number) DO UPDATE with similar COALESCE/LEAST logic.
    - Use RETURNING id; fallback SELECT if needed.
    """
    # TODO: Implement upsert logic
    pass

def ingest_pull_requests(provider: str, repo: str, prs: Iterable[Dict[str, Any]]) -> int:
    """Ingest multiple pull requests; return count processed.
    
    Implementation guidance:
    - Loop over prs iterable.
    - Call upsert_pull_request(provider, repo, pr) for each.
    - Return total count processed.
    """
    # TODO: Implement
    pass

# ---- CI Pipelines & Jobs ----

def upsert_ci_pipeline(provider: str, repo: str, pipe: Dict[str, Any]) -> int:
    """Insert/update a CI pipeline idempotently; return db id.
    
    Implementation guidance:
    - INSERT into ci_pipelines with fields: provider, repo, pipeline_id,
      status, created_at, updated_at, sha.
    - Use str(pipe["pipeline_id"]) to handle large integers.
    - Use _normalize_timestamp_to_utc() for timestamp fields.
    - ON CONFLICT (provider, repo, pipeline_id) DO UPDATE:
      status=EXCLUDED.status,
      created_at=LEAST(ci_pipelines.created_at, EXCLUDED.created_at),
      updated_at=COALESCE(EXCLUDED.updated_at, ci_pipelines.updated_at),
      sha=COALESCE(EXCLUDED.sha, ci_pipelines.sha).
    - Use RETURNING id; fallback SELECT if needed.
    """
    # TODO: Implement upsert logic
    pass

def upsert_ci_job(provider: str, repo: str, job: Dict[str, Any]) -> int:
    """Insert/update a CI job idempotently; return db id.
    
    Implementation guidance:
    - INSERT into ci_jobs with fields: provider, repo, pipeline_id, job_id,
      name, status, started_at, finished_at, duration_seconds.
    - Convert job_id and pipeline_id to strings.
    - Use _normalize_timestamp_to_utc() for timestamp fields (if not None).
    - For duration_seconds: int(job.get("duration_seconds", 0)) if not None else None.
    - ON CONFLICT (provider, repo, job_id) DO UPDATE with COALESCE logic.
    - Use RETURNING id; fallback SELECT if needed.
    """
    # TODO: Implement upsert logic
    pass

def ingest_ci(provider: str, repo: str, pipelines: Iterable[Dict[str, Any]], 
              jobs_by_pipeline: Optional[Dict[str, Iterable[Dict[str, Any]]]] = None) -> int:
    """Ingest pipelines and their jobs. Returns number of pipelines processed.
    
    Implementation guidance:
    - Loop over pipelines iterable.
    - For each pipeline, call upsert_ci_pipeline(provider, repo, pipe).
    - If jobs_by_pipeline is provided:
      - Get pipeline_id as str(pipe["pipeline_id"]).
      - For each job in jobs_by_pipeline.get(pipeline_id, []):
        - Ensure job has "pipeline_id" set
        - Call upsert_ci_job(provider, repo, job).
    - Return total count of pipelines processed.
    """
    # TODO: Implement
    pass

# ---------------------------------------------
# Helper Functions -- some are provided, others are stubbed; you can implement them or decide on your own way of providing similar functionality
# ---------------------------------------------

import requests

def fetch_json(url: str, headers: Optional[Dict[str, str]] = None, 
               params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """Lightweight GET JSON wrapper."""
    resp = requests.get(url, headers=headers or {}, params=params or {}, timeout=30)
    resp.raise_for_status()
    return resp.json()

def collect_github_issues(owner_repo: str, state: str = "all", 
                          token: Optional[str] = None, 
                          per_page: int = 100, max_pages: int = 1) -> Iterable[Dict[str, Any]]:
    """Generator yielding normalized issue dicts from GitHub REST v3."""
    owner, repo = owner_repo.split("/", 1)
    headers = {"Accept": "application/vnd.github+json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    url = f"https://api.github.com/repos/{owner}/{repo}/issues"
    page = 1
    while page <= max_pages:
        data = fetch_json(url, headers=headers, params={"state": state, "per_page": per_page, "page": page})
        if not data:
            break
        for it in data:
            if "pull_request" in it:
                # skip PRs here; use collect_github_pulls for PR details
                continue
            yield {
                "number": it["number"],
                "title": it.get("title", ""),
                "author": (it.get("user") or {}).get("login"),
                "state": it.get("state", "open"),
                "created_at": it.get("created_at"),
                "closed_at": it.get("closed_at"),
            }
        page += 1

def collect_github_pulls(owner_repo: str, state: str = "all", 
                        token: Optional[str] = None, 
                        per_page: int = 100, max_pages: int = 1) -> Iterable[Dict[str, Any]]:
    """Generator yielding normalized pull request dicts from GitHub REST v3.
    
    Implementation guidance:
    - URL: https://api.github.com/repos/{owner}/{repo}/pulls
    - Headers: Accept: application/vnd.github+json, Authorization if token provided
    - Query params: state={state}, per_page={per_page}, page={page}
    - Normalize state: if pr.get("merged_at") then state="merged", else pr.get("state", "open")
    - Yield dict with: number, title, author (from user.login), state, 
      created_at, merged_at, closed_at
    """
    # TODO: Implement (similar pattern to collect_github_issues)
    pass

def collect_github_actions_runs(owner_repo: str, token: Optional[str] = None, 
                                per_page: int = 100, max_pages: int = 1) -> Iterable[Dict[str, Any]]:
    """Yield pipelines from GitHub Actions workflow runs.
    
    Implementation guidance:
    - URL: https://api.github.com/repos/{owner}/{repo}/actions/runs
    - Headers: Accept: application/vnd.github+json, Authorization if token provided
    - Response is a dict with "workflow_runs" key containing array
    - For each run r, yield dict with:
      pipeline_id: r["id"]
      status: r.get("conclusion") or r.get("status") or "unknown"
      created_at: r.get("created_at")
      updated_at: r.get("updated_at")
      sha: r.get("head_sha")
    """
    # TODO: Implement
    pass

def collect_github_actions_jobs(owner_repo: str, run_id: str, 
                                token: Optional[str] = None, 
                                per_page: int = 100, max_pages: int = 1) -> Iterable[Dict[str, Any]]:
    """Yield jobs for a specific GitHub Actions run.
    
    Implementation guidance:
    - URL: https://api.github.com/repos/{owner}/{repo}/actions/runs/{run_id}/jobs
    - Headers: Accept: application/vnd.github+json, Authorization if token provided
    - Response is a dict with "jobs" key containing array
    - For each job j, yield dict with:
      pipeline_id: str(run_id)
      job_id: j["id"]
      name: j.get("name")
      status: j.get("conclusion") or j.get("status")
      started_at: j.get("started_at")
      finished_at: j.get("completed_at")
      duration_seconds: j.get("duration_ms", 0) // 1000 if j.get("duration_ms") else None
    """
    # TODO: Implement
    pass

-- Issues (DC2)
CREATE TABLE IF NOT EXISTS issues (
    id SERIAL PRIMARY KEY,
    provider TEXT NOT NULL,                 -- e.g., 'github', 'gitlab'
    repo TEXT NOT NULL,                     -- 'owner/repo' or 'group/project'
    issue_number INTEGER NOT NULL,          
    title TEXT NOT NULL,
    author TEXT,
    state TEXT NOT NULL,                    -- 'open', 'closed', etc.
    created_at TIMESTAMPTZ NOT NULL,
    closed_at TIMESTAMPTZ
);
CREATE UNIQUE INDEX IF NOT EXISTS uq_issue_identity ON issues(provider, repo, issue_number);

-- Pull / Merge Requests (DC2)
CREATE TABLE IF NOT EXISTS pull_requests (
    id SERIAL PRIMARY KEY,
    provider TEXT NOT NULL,
    repo TEXT NOT NULL,
    pr_number INTEGER NOT NULL,
    title TEXT NOT NULL,
    author TEXT,
    state TEXT NOT NULL,                    -- 'open', 'closed', 'merged' (normalized, see notes in readme)
    created_at TIMESTAMPTZ NOT NULL,
    merged_at TIMESTAMPTZ,
    closed_at TIMESTAMPTZ
);
CREATE UNIQUE INDEX IF NOT EXISTS uq_pr_identity ON pull_requests(provider, repo, pr_number);

-- CI Pipelines (DC2)
CREATE TABLE IF NOT EXISTS ci_pipelines (
    id SERIAL PRIMARY KEY,
    provider TEXT NOT NULL,
    repo TEXT NOT NULL,
    pipeline_id TEXT NOT NULL,              -- provider-visible id (string to handle big ints)
    status TEXT NOT NULL,
    created_at TIMESTAMPTZ NOT NULL,
    updated_at TIMESTAMPTZ,
    sha TEXT                                -- commit hash the pipeline ran for (if known)
);
CREATE UNIQUE INDEX IF NOT EXISTS uq_pipeline_identity ON ci_pipelines(provider, repo, pipeline_id);

-- CI Jobs (DC2)
CREATE TABLE IF NOT EXISTS ci_jobs (
    id SERIAL PRIMARY KEY,
    provider TEXT NOT NULL,
    repo TEXT NOT NULL,
    pipeline_id TEXT NOT NULL,
    job_id TEXT NOT NULL,
    name TEXT,
    status TEXT,
    started_at TIMESTAMPTZ,
    finished_at TIMESTAMPTZ,
    duration_seconds INTEGER
);
CREATE UNIQUE INDEX IF NOT EXISTS uq_job_identity ON ci_jobs(provider, repo, job_id);