main.py

import os
import requests
import time
from datetime import datetime, timedelta, timezone
from dotenv import load_dotenv
load_dotenv()

is_github_actions = os.environ.get("GITHUB_ACTIONS") == "true"

class GitHubStatsError(Exception):
    pass

def graphql(query: str, variables: dict, headers: dict) -> dict:
    res = requests.post('https://api.github.com/graphql', json={"query": query, "variables": variables}, headers=headers, timeout=15)
    res.raise_for_status()
    body = res.json()
    if "errors" in body:
        raise GitHubStatsError(f"GraphQL error: {body['errors']}")
    return body.get("data", {})


def get_user_info(headers: dict) -> dict:
    username = os.environ.get("GH_USERNAME")
    if not username:
        raise GitHubStatsError("GH_USERNAME is not set")
    query = """
    query($username: String!) {
      user(login: $username) {
        createdAt
        followers { totalCount }
        organizations { totalCount }
        repositories(first: 1) { totalCount }
        repositoriesContributedTo(first: 1, contributionTypes: [COMMIT, ISSUE, PULL_REQUEST]) {
          totalCount
        }
      }
    }
    """
    data = graphql(query, {"username": username}, headers)
    user = data.get("user")
    if not user:
        raise GitHubStatsError("User not found in GraphQL response")
    return user


def get_commits_in_range(headers: dict, from_dt: datetime, to_dt: datetime) -> tuple[int, int]:
    username = os.environ.get("GH_USERNAME")
    if not username:
        raise GitHubStatsError("GH_USERNAME is not set")
    query = """
    query($username: String!, $from: DateTime!, $to: DateTime!) {
      user(login: $username) {
        contributionsCollection(from: $from, to: $to) {
          totalCommitContributions
          restrictedContributionsCount
        }
      }
    }
    """
    data = graphql(query, {
        "username": username,
        "from": from_dt.strftime("%Y-%m-%dT%H:%M:%SZ"),
        "to": to_dt.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }, headers)
    collection = data.get("user", {}).get("contributionsCollection", {})
    return collection.get("totalCommitContributions", 0), collection.get("restrictedContributionsCount", 0)


def get_commit_search_total(headers: dict) -> int:
    username = os.environ.get("GH_USERNAME")
    if not username:
        raise GitHubStatsError("GH_USERNAME is not set")
    res = requests.get(
        f"https://api.github.com/search/commits",
        params={"q": f"author:{username}"},
        headers={**headers, "Accept": "application/vnd.github+json"},
        timeout=15,
    )
    res.raise_for_status()
    return res.json().get("total_count", 0)


def get_github_stats() -> dict:
    token = os.environ.get("GH_TOKEN")
    if not token:
        raise GitHubStatsError("GH_TOKEN is not set")

    headers = {
        "Authorization": f"bearer {token}",
        "Content-Type": "application/json",
    }

    user = get_user_info(headers)

    followers = user["followers"]["totalCount"]
    orgs = user["organizations"]["totalCount"]
    owned = user["repositories"]["totalCount"]
    contributed = user["repositoriesContributedTo"]["totalCount"]
    total_repos = owned + contributed

    account_start = datetime.strptime(user["createdAt"], "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
    now = datetime.now(timezone.utc)

    total_commits = 0
    total_restricted = 0
    cursor = account_start

    while cursor < now:
        chunk_end = min(cursor + timedelta(weeks=6), now)
        dto = (chunk_end - timedelta(hours=1)).replace(minute=59, second=59, microsecond=0)
        commits, restricted = get_commits_in_range(headers, cursor, dto)
        total_commits += commits
        total_restricted += restricted
        cursor = chunk_end

    if total_restricted:
        print(f"{total_restricted} restricted contributions excluded, your token may lack the 'repo' scope or an organization hasn't granted access")

    search_total = get_commit_search_total(headers)
    if search_total > total_commits:
        total_commits = search_total

    return {
        "commits": str(total_commits),
        "repos": str(total_repos),
        "orgs": str(orgs),
        "followers": str(followers),
    }


def build_widget_payload(stats: dict) -> dict:
    username = os.environ.get("GH_USERNAME")
    if not username:
        raise GitHubStatsError("GH_USERNAME is not set")
    def text(value: str) -> dict:
        return {"value_type": "custom_string", "presentation_type": "text", "value": value}

    def asset(value: str) -> dict:
        return {"value_type": "application_asset", "presentation_type": "image", "value": value}

    def item(name: str, description: str, image: str) -> dict:
        return {"fields": {"name": text(name), "description": text(description), "image": asset(image)}}

    return {
        "surfaces": {
            "widget_top": {
                "layout": "widget_top_hero",
                "components": {
                    "title": {"fields": {"text": text("GitHub Stats")}},
                    "subtitle_1": {"fields": {"text": text(f"stats for github.com/{username}")}},
                    "hero_image": {"fields": {"image": asset(os.environ.get("HERO_IMAGE"))}},
                    "subtitle_2": {"fields": {"text": text("autoupdates every hour, made by lezetho")}},
                },
            },
            "widget_bottom": {
                "layout": "widget_bottom_collection",
                "components": {
                    "item_1": item("total commits", stats["commits"], os.environ.get("COMMITS_TOOLTIP_IMAGE")),
                    "item_2": item("number of orgs joined", stats["orgs"], os.environ.get("ORGS_TOOLTIP_IMAGE")),
                    "item_3": item("number of repos contributed to", stats["repos"], os.environ.get("REPO_TOOLTIP_IMAGE")),
                    "item_4": item("number of followers", stats["followers"], os.environ.get("FOLLOWERS_TOOLTIP_IMAGE")),
                },
            },
            "add_widget_preview": {
                "layout": "add_widget_preview_hero",
                "components": {
                    "hero_image": {"fields": {"image": asset(os.environ.get("HERO_IMAGE"))}},
                },
            },
        }
    }


def update_discord_widget(stats: dict) -> None:
    application_id = os.environ.get("APPLICATION_ID")
    if not application_id:
        raise GitHubStatsError("APPLICATION_ID is not set.")
    token = os.environ.get("DISCORD_BOT_TOKEN")
    if not token:
        raise GitHubStatsError("DISCORD_BOT_TOKEN is not set.")
    widget_config_id = os.environ.get("WIDGET_CONFIG_ID")
    if not widget_config_id:
        raise GitHubStatsError("WIDGET_CONFIG_ID is not set.")
    headers = {
        "Authorization": f"Bot {token}",
        "Content-Type": "application/json",
    }
    base_url = f"https://discord.com/api/v9/applications/{application_id}/widget-configs/{widget_config_id}"

    res = requests.patch(base_url, json=build_widget_payload(stats), headers=headers, timeout=15)
    res.raise_for_status()

    pub = requests.post(f"{base_url}/publish", headers=headers, timeout=15)
    if pub.status_code not in (200, 204):
        raise GitHubStatsError(f"Widget publish failed: HTTP {pub.status_code} — {pub.text}")


if __name__ == "__main__":
    if is_github_actions:
        try:
            stats = get_github_stats()
            update_discord_widget(stats)
            print("Widget updated successfully")
        except GitHubStatsError as e:
            print(f"Error: {e}")
            raise SystemExit(1)
        except requests.HTTPError as e:
            print(f"HTTP error: {e.response.status_code} — {e.response.text}")
            raise SystemExit(1)
    else:
        while True:
            try:
                stats = get_github_stats()
                update_discord_widget(stats)
                print("Widget updated successfully. Next update in 1 hour.")
            except (GitHubStatsError, requests.HTTPError) as e:
                print(f"Error: {e}")
            time.sleep(3600)