github2file.py - master - dpavlov/datagrid_reporter

github2file.py
218 строк9 KB
Dmitriy Pavlov
new scripts skeleton and MVP for extended activity view
13 май 2026, 00:28
13 май 2026, 00:28500f869
import json
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional

try:
    from datagrid_reporter.config import GithubConfig
    from datagrid_reporter.dates import github_timestamp_to_day, horizon_start, utc_today
    from datagrid_reporter.github_service import BASE_URL, get_commit, iter_pr_commits, iter_recent_prs
    from datagrid_reporter.http_json import get_json, github_headers
    from datagrid_reporter.jira_git_common import save_activity
except ModuleNotFoundError:
    import sys

    sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
    from datagrid_reporter.config import GithubConfig
    from datagrid_reporter.dates import github_timestamp_to_day, horizon_start, utc_today
    from datagrid_reporter.github_service import BASE_URL, get_commit, iter_pr_commits, iter_recent_prs
    from datagrid_reporter.http_json import get_json, github_headers
    from datagrid_reporter.jira_git_common import save_activity


def extract_pr_number(url: str) -> Optional[int]:
    match = re.search(r"/pull/(\d+)", url or "")
    return int(match.group(1)) if match else None


def calculate_full_age_in_days(comment_date: str) -> int:
    dt = datetime.strptime(comment_date, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
    return (datetime.now(timezone.utc) - dt).days


def get_pr_details(repo_owner: str, repo_name: str, pr_number: int, days: int) -> Optional[Dict[str, Any]]:
    cfg = _cfg(repo_owner, repo_name)
    pr = get_json(f"{BASE_URL}/repos/{repo_owner}/{repo_name}/pulls/{pr_number}", headers=github_headers())
    if not pr:
        return None

    number = pr["number"]
    author = (pr.get("user") or {}).get("login")
    title = pr.get("title")
    created = pr.get("created_at")
    updated = pr.get("updated_at")
    merged = pr.get("merged_at")

    if in_date_range(created, days):
        save_activity(f"PR #{number} '{title}' by {author} created at {created}")
    elif in_date_range(updated, days):
        save_activity(f"PR #{number} '{title}' by {author} updated at {updated}")
    if in_date_range(merged, days):
        save_activity(f"PR #{number} '{title}' by {author} merged at {merged}")

    return {
        "pr_data": {
            "number": number,
            "title": title,
            "body": pr.get("body"),
            "state": pr.get("state"),
            "created_at": created,
            "updated_at": updated,
            "merged_at": merged,
            "patch_url": pr.get("patch_url"),
            "closed_at": pr.get("closed_at"),
            "author": author,
            "html_url": pr.get("html_url"),
        },
        "comments": get_pr_comments(f"{BASE_URL}/repos/{repo_owner}/{repo_name}/pulls/{pr_number}", pr_number, days),
        "commits": get_pr_commits(cfg, pr_number, days),
        "approvals": get_pr_approvals(f"{BASE_URL}/repos/{repo_owner}/{repo_name}/pulls/{pr_number}", pr_number, days),
    }


def get_pr_comments(base_url: str, pr_number: int, days: int) -> List[Dict[str, Any]]:
    comments = get_json(f"{base_url}/comments", headers=github_headers()) or []
    result = []
    for comment in comments:
        created_at = comment.get("created_at")
        if not in_date_range(created_at, days):
            continue
        diff_hunk = comment.get("diff_hunk") or ""
        relevant_diff = "\n".join(diff_hunk.splitlines()[:5])
        login = (comment.get("user") or {}).get("login")
        body = comment.get("body") or ""
        result.append(
            {
                "author": login,
                "body": body,
                "created_at": created_at,
                "updated_at": comment.get("updated_at"),
                "relevant_diff": relevant_diff,
            }
        )
        save_activity(f"\tPR #{pr_number} Commented by {login} at {created_at}: {body} {relevant_diff}")
    return result


def get_pr_commits(cfg_or_base_url: Any, pr_number: int, days: int) -> List[Dict[str, Any]]:
    cfg = cfg_or_base_url if isinstance(cfg_or_base_url, GithubConfig) else _cfg_from_pull_url(str(cfg_or_base_url))
    start_day = horizon_start(days, utc_today())
    commits = []
    for commit_ref in iter_pr_commits(cfg, pr_number, start_day):
        commit = get_commit(cfg, commit_ref["sha"])
        commit_data = commit.get("commit") or {}
        author_data = commit_data.get("author") or {}
        committer_data = commit_data.get("committer") or {}
        date = author_data.get("date") or committer_data.get("date")
        if date and github_timestamp_to_day(date) < start_day:
            continue
        author_user = commit.get("author") or {}
        committer_user = commit.get("committer") or {}
        item = {
            "sha": commit.get("sha"),
            "message": commit_data.get("message"),
            "author_login": author_user.get("login") or "",
            "author_name": author_data.get("name"),
            "author_email": author_data.get("email"),
            "committer_login": committer_user.get("login") or "",
            "committer_name": committer_data.get("name"),
            "committer_email": committer_data.get("email"),
            "date": date,
        }
        commits.append(item)
        display = item["author_login"] or item["committer_login"] or item["author_email"] or item["committer_email"]
        save_activity(f"\tPR #{pr_number} commit added by {display} at {date}: {item['message']}")
    return commits


def get_pr_approvals(base_url: str, pr_number: int, days: int) -> List[Dict[str, Any]]:
    reviews = get_json(f"{base_url}/reviews", headers=github_headers()) or []
    approvals = []
    for review in reviews:
        submitted_at = review.get("submitted_at")
        if not submitted_at or not in_date_range(submitted_at, days):
            continue
        if (review.get("state") or "").upper() != "APPROVED":
            continue
        user = review.get("user") or {}
        body = review.get("body") or ""
        login = user.get("login")
        save_activity(f"\tPR #{pr_number} approved by {login} at {submitted_at}: {body}")
        approvals.append(
            {
                "author": login,
                "author_url": user.get("html_url"),
                "body": body,
                "submitted_at": submitted_at,
                "state": review.get("state"),
            }
        )
    return approvals


def in_date_range(value: Optional[str], days: int) -> bool:
    if not value:
        return False
    return github_timestamp_to_day(value) >= horizon_start(days, utc_today())


def save_to_file(data: Any, file_name: str) -> None:
    Path(file_name).write_text(json.dumps(data, indent=4, ensure_ascii=False), encoding="utf-8")
    print(f"Data saved to file: {file_name}")


def check_rate_limit() -> int:
    data = get_json(f"{BASE_URL}/rate_limit", headers=github_headers())
    remaining = int((data.get("rate") or {}).get("remaining") or 0)
    reset = (data.get("rate") or {}).get("reset")
    print(f"Rate Limit: {remaining} requests remaining. Resets at {reset}.")
    return remaining


def get_recent_prs(repo_owner: str, repo_name: str, days: int, state: str = "all", per_page: int = 50) -> List[Dict[str, Any]]:
    cfg = _cfg(repo_owner, repo_name, state=state, per_page=per_page)
    start_day = horizon_start(days, utc_today())
    result = []
    for pr in iter_recent_prs(cfg, start_day):
        result.extend(filter_prs_by_last_week([pr], days))
    return result


def filter_prs_by_last_week(prs: List[Dict[str, Any]], days: int) -> List[Dict[str, Any]]:
    start_day = horizon_start(days, utc_today())
    result = []
    for pr in prs:
        if github_timestamp_to_day(pr["updated_at"]) >= start_day:
            result.append(
                {
                    "number": pr["number"],
                    "title": pr["title"],
                    "author": (pr.get("user") or {}).get("login"),
                    "updated_at": pr["updated_at"],
                    "state": pr["state"],
                    "html_url": pr["html_url"],
                }
            )
    return result


def get_recent_prs_list(repo_owner: str, repo_name: str, days: int) -> List[Dict[str, Any]]:
    return get_recent_prs(repo_owner, repo_name, days, state="all", per_page=50)


def _cfg(repo_owner: str, repo_name: str, state: str = "all", per_page: int = 50) -> GithubConfig:
    return GithubConfig(owner=repo_owner, repo=repo_name, window_days=7, state=state, per_page=per_page)


def _cfg_from_pull_url(base_url: str) -> GithubConfig:
    match = re.search(r"/repos/([^/]+)/([^/]+)/pulls/", base_url)
    if not match:
        raise ValueError(f"Cannot infer GitHub repo from URL: {base_url}")
    return _cfg(match.group(1), match.group(2))


if __name__ == "__main__":
    recent_prs = get_recent_prs_list("apache", "ignite", 1)
    save_to_file(recent_prs, "recent_prs_last_week.json")
    print(f"Found {len(recent_prs)} PRs updated in the last day.")