feat: initial project structure

- GitLabClient: REST Events API z paginacją, pobieranie członków grupy
- Aggregator: zliczanie commitów, MR, komentarzy per użytkownik
- Exporter: generowanie pliku Excel (openpyxl) ze stylami
- main.py: CLI (click) + .env support
- README, .env.example, requirements.txt, .gitignore
This commit is contained in:
2026-03-04 18:27:01 +00:00
commit 066539c89c
8 changed files with 484 additions and 0 deletions

142
gitlab_client.py Normal file
View File

@@ -0,0 +1,142 @@
"""
GitLab GraphQL client — pobiera events dla grupy w zadanym zakresie dat.
"""
import requests
from datetime import date
from typing import Generator
EVENTS_QUERY = """
query GroupContributions($fullPath: ID!, $after: String) {
group(fullPath: $fullPath) {
members(first: 100, after: $after) {
pageInfo {
hasNextPage
endCursor
}
nodes {
user {
id
name
username
contributionsCollection {
totalCommitContributions
totalPullRequestContributions: totalPullRequestContributions
}
}
}
}
}
}
"""
# Events API (REST) — GraphQL nie udostępnia pełnych events per user per group
# Używamy REST Events API z paginacją
EVENTS_REST_PATH = "/api/v4/groups/{group_id}/events"
class GitLabClient:
def __init__(self, url: str, token: str):
self.url = url.rstrip("/")
self.token = token
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
})
def get_group_id(self, group_path: str) -> int:
"""Pobiera numeryczne ID grupy po jej ścieżce."""
resp = self.session.get(
f"{self.url}/api/v4/groups/{requests.utils.quote(group_path, safe='')}",
)
resp.raise_for_status()
return resp.json()["id"]
def get_group_members(self, group_id: int) -> list[dict]:
"""Zwraca wszystkich członków grupy (z podgrupami)."""
members = []
page = 1
while True:
resp = self.session.get(
f"{self.url}/api/v4/groups/{group_id}/members/all",
params={"per_page": 100, "page": page},
)
resp.raise_for_status()
batch = resp.json()
if not batch:
break
members.extend(batch)
if len(batch) < 100:
break
page += 1
return members
def get_user_events(
self,
user_id: int,
date_from: date,
date_to: date,
) -> list[dict]:
"""
Pobiera eventy użytkownika w zadanym zakresie dat.
Filtruje po: pushed (commity), merged (MR), commented, created.
"""
events = []
page = 1
while True:
resp = self.session.get(
f"{self.url}/api/v4/users/{user_id}/events",
params={
"after": date_from.isoformat(),
"before": date_to.isoformat(),
"per_page": 100,
"page": page,
},
)
resp.raise_for_status()
batch = resp.json()
if not batch:
break
events.extend(batch)
if len(batch) < 100:
break
page += 1
return events
def get_group_events(
self,
group_id: int,
date_from: date,
date_to: date,
action: str | None = None,
) -> Generator[dict, None, None]:
"""
Pobiera eventy grupy w zadanym zakresie dat (paginacja).
action: pushed | merged | created | commented | None (wszystkie)
"""
page = 1
params = {
"after": date_from.isoformat(),
"before": date_to.isoformat(),
"per_page": 100,
"page": page,
}
if action:
params["action"] = action
while True:
params["page"] = page
resp = self.session.get(
f"{self.url}/api/v4/groups/{group_id}/events",
params=params,
)
resp.raise_for_status()
batch = resp.json()
if not batch:
break
yield from batch
if len(batch) < 100:
break
page += 1