651 lines
24 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""
gitea-classroom — GitHub Classroom Ersatz für Gitea
====================================================
Konfiguration:
export GITEA_URL="https://gitea.meine-uni.de"
export GITEA_TOKEN="dein_api_token"
Befehle:
gitea-classroom add-students --org MeinKurs --csv students.csv
gitea-classroom create-assignment --org MeinKurs --assignment ueb01 --template uebung01-template --csv students.csv
gitea-classroom create-group-assignment --org MeinKurs --assignment proj01 --template proj-template --csv groups.csv
gitea-classroom list-submissions --org MeinKurs --assignment ueb01
gitea-classroom list-group-submissions --org MeinKurs --assignment proj01
gitea-classroom clone-submissions --org MeinKurs --assignment ueb01 --dir ./abgaben
gitea-classroom delete-assignment --org MeinKurs --assignment ueb01
"""
import argparse
import base64
import csv
import os
import subprocess
import sys
from collections import defaultdict
from datetime import datetime
from urllib.parse import urlparse, urlunparse
import requests
from gitea import Gitea, Organization, Team
# ──────────────────────────────────────────────
# Konfiguration
# ──────────────────────────────────────────────
GITEA_URL = os.environ.get("GITEA_URL", "https://git.efi.th-nuernberg.de")
GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "DEIN_API_TOKEN_HIER")
SEPARATOR = "-" # Trennzeichen: ueb01-max.mustermann / proj01-team-a
# ──────────────────────────────────────────────
# Hilfsfunktionen (allgemein)
# ──────────────────────────────────────────────
def get_gitea() -> Gitea:
g = Gitea(GITEA_URL, GITEA_TOKEN)
try:
g.get_version()
except Exception as e:
print(f"[FEHLER] Verbindung zu {GITEA_URL} fehlgeschlagen: {e}")
sys.exit(1)
return g
def get_org(g: Gitea, org_name: str) -> Organization:
try:
return Organization.request(g, org_name)
except Exception:
print(f"[FEHLER] Organisation '{org_name}' nicht gefunden.")
sys.exit(1)
def headers() -> dict:
return {"Authorization": f"token {GITEA_TOKEN}", "Content-Type": "application/json"}
def api(method: str, path: str, **kwargs) -> requests.Response:
"""Sendet einen API-Request an die Gitea-Instanz."""
url = f"{GITEA_URL}/api/v1{path}"
return requests.request(method, url, headers=headers(), **kwargs)
def api_message(r: requests.Response) -> str:
"""Gibt die Fehlermeldung einer API-Antwort zurück (robust gegen leere Body)."""
try:
return r.json().get("message", r.text) if r.text else str(r.status_code)
except Exception:
return r.text or str(r.status_code)
def repo_exists(org: str, name: str) -> bool:
return api("GET", f"/repos/{org}/{name}").status_code == 200
def list_org_repos(org_name: str) -> list[dict]:
"""Gibt alle Repos einer Organisation zurück (paginiert)."""
repos, page = [], 1
while True:
r = api("GET", f"/orgs/{org_name}/repos", params={"limit": 50, "page": page})
batch = r.json()
if not batch:
break
repos.extend(batch)
page += 1
return repos
def copy_repo(template_owner: str, template_repo: str,
dest_org: str, dest_name: str, private: bool = True) -> bool:
"""
Kopiert ein Template-Repo in die Ziel-Org.
Versucht zuerst die Generate-API (funktioniert für Repos auf derselben Instanz,
erfordert dass das Repo in Gitea als Template markiert ist).
Fällt auf die Migrate-API zurück (für externe Repos).
"""
r = api("POST", f"/repos/{template_owner}/{template_repo}/generate", json={
"owner": dest_org,
"name": dest_name,
"private": private,
"git_content": True,
})
if r.status_code in (200, 201):
return True
r = api("POST", "/repos/migrate", json={
"clone_addr": f"{GITEA_URL}/{template_owner}/{template_repo}",
"repo_owner": dest_org,
"repo_name": dest_name,
"mirror": False,
"private": private,
"auth_token": GITEA_TOKEN,
})
if r.status_code in (200, 201):
return True
print(f" [WARNUNG] '{dest_name}' konnte nicht erstellt werden: "
f"{r.status_code} {api_message(r)}")
return False
def get_last_commit(org: str, repo_name: str) -> dict | None:
r = api("GET", f"/repos/{org}/{repo_name}/commits", params={"limit": 1, "page": 1})
if r.status_code == 200:
commits = r.json()
if commits:
return commits[0]
return None
def format_commit(commit: dict | None) -> tuple[str, str, str]:
if not commit:
return " kein Commit ", "", ""
raw_date = commit["commit"]["author"]["date"]
try:
dt = datetime.fromisoformat(raw_date.replace("Z", "+00:00"))
date = dt.strftime("%Y-%m-%d %H:%M")
except Exception:
date = raw_date[:16]
author = commit["commit"]["author"]["name"][:18]
message = commit["commit"]["message"].splitlines()[0][:40]
return date, author, message
# ──────────────────────────────────────────────
# Feedback-PR
# ──────────────────────────────────────────────
def create_feedback_pr(org: str, repo_name: str) -> bool:
"""
Erstellt einen Feedback-PR der Student-Commits sichtbar macht:
- feedback-Branch = eingefrorener Template-Stand (Referenzpunkt)
- default-Branch bekommt einen .classroom-Marker-Commit
- PR default → feedback wächst mit jedem Student-Push
"""
r = api("GET", f"/repos/{org}/{repo_name}")
if r.status_code != 200:
return False
default_branch = r.json().get("default_branch", "main")
# feedback-Branch am aktuellen HEAD einfrieren
r = api("POST", f"/repos/{org}/{repo_name}/branches",
json={"new_branch_name": "feedback", "old_branch_name": default_branch})
if r.status_code not in (200, 201):
print(f" [WARNUNG] Branch 'feedback' konnte nicht erstellt werden: {api_message(r)}")
return False
# Marker-Commit auf default-Branch damit der PR sofort einen Diff hat
content = base64.b64encode(
"Dieses Repository wird über gitea-classroom verwaltet.\n".encode()
).decode()
r = api("POST", f"/repos/{org}/{repo_name}/contents/.classroom",
json={"message": "Classroom initialisieren", "content": content,
"branch": default_branch})
if r.status_code not in (200, 201):
print(f" [WARNUNG] .classroom konnte nicht erstellt werden: {api_message(r)}")
return False
# PR: default → feedback (zeigt alle Student-Commits)
r = api("POST", f"/repos/{org}/{repo_name}/pulls", json={
"title": "Feedback",
"head": default_branch,
"base": "feedback",
"body": "Dieser Pull Request zeigt alle Abgabe-Commits und dient als Feedback-Kanal.",
})
if r.status_code in (200, 201):
return True
print(f" [WARNUNG] Feedback-PR konnte nicht erstellt werden: {api_message(r)}")
return False
# ──────────────────────────────────────────────
# Gitea-Team-Hilfsfunktionen
# ──────────────────────────────────────────────
def ensure_org_team(org: Organization, team_name: str,
permission: str = "write") -> Team:
"""Gibt ein Team zurück (erstellt es falls nötig)."""
for team in org.get_teams():
if team.name == team_name:
return team
payload = {
"name": team_name,
"permission": permission,
"units": ["repo.code", "repo.issues", "repo.pulls"],
"includes_all_repositories": False,
}
r = api("POST", f"/orgs/{org.name}/teams", json=payload)
if r.status_code not in (200, 201):
print(f"[FEHLER] Team '{team_name}' konnte nicht erstellt werden: {r.text}")
sys.exit(1)
team_id = r.json()["id"]
for team in org.get_teams():
if team.id == team_id:
return team
raise RuntimeError("Team erstellt, aber nicht auffindbar.")
def add_member_to_team(team: Team, username: str) -> bool:
r = api("PUT", f"/teams/{team.id}/members/{username}")
return r.status_code in (200, 204)
def add_repo_to_team(team: Team, org_name: str, repo_name: str) -> bool:
r = api("PUT", f"/teams/{team.id}/repos/{org_name}/{repo_name}")
return r.status_code in (200, 204)
def add_collaborator(org: str, repo: str, username: str,
permission: str = "write") -> bool:
r = api("PUT", f"/repos/{org}/{repo}/collaborators/{username}",
json={"permission": permission})
return r.status_code in (200, 204)
def delete_org_team_by_name(org: Organization, team_name: str):
"""Löscht ein Team der Org anhand des Namens (ignoriert Fehler)."""
for team in org.get_teams():
if team.name == team_name:
api("DELETE", f"/teams/{team.id}")
return
# ──────────────────────────────────────────────
# CSV lesen
# ──────────────────────────────────────────────
def read_students_csv(path: str) -> list[dict]:
"""CSV mit Spalte 'username' (+ optional 'name', 'email')."""
if not os.path.exists(path):
print(f"[FEHLER] CSV '{path}' nicht gefunden.")
sys.exit(1)
with open(path, newline="", encoding="utf-8") as f:
rows = list(csv.DictReader(f))
if not rows or "username" not in rows[0]:
print("[FEHLER] CSV benötigt eine Spalte 'username'.")
sys.exit(1)
return rows
def read_groups_csv(path: str) -> dict[str, list[str]]:
"""
CSV mit Spalten 'group' und 'username'.
Gibt dict { group_name -> [username, ...] } zurück.
"""
if not os.path.exists(path):
print(f"[FEHLER] CSV '{path}' nicht gefunden.")
sys.exit(1)
with open(path, newline="", encoding="utf-8") as f:
rows = list(csv.DictReader(f))
if not rows or "group" not in rows[0] or "username" not in rows[0]:
print("[FEHLER] Gruppen-CSV benötigt Spalten 'group' und 'username'.")
sys.exit(1)
groups: dict[str, list[str]] = defaultdict(list)
for row in rows:
groups[row["group"].strip()].append(row["username"].strip())
return dict(groups)
# ──────────────────────────────────────────────
# Befehle
# ──────────────────────────────────────────────
def cmd_add_students(args):
"""Fügt Studenten zur Org hinzu und legt sie ins 'students'-Team."""
g = get_gitea()
org = get_org(g, args.org)
team = ensure_org_team(org, "students")
students = read_students_csv(args.csv)
print(f"\n{len(students)} Studenten → Org '{args.org}' / Team 'students'\n")
ok = skip = 0
for row in students:
username = row["username"].strip()
if add_member_to_team(team, username):
print(f"{username}")
ok += 1
else:
print(f"{username} (Team-Zuweisung fehlgeschlagen)")
skip += 1
print(f"\nFertig: {ok} hinzugefügt, {skip} übersprungen.")
def cmd_create_assignment(args):
"""Erstellt pro Student ein eigenes Repo (Kopie des Templates)."""
g = get_gitea()
get_org(g, args.org)
students = read_students_csv(args.csv)
template_owner = args.template_owner or args.org
print(f"\n→ Einzel-Assignment '{args.assignment}' für {len(students)} Studenten")
print(f" Template : {template_owner}/{args.template}")
print(f" Muster : {args.assignment}{SEPARATOR}<username>\n")
ok = skip = 0
for row in students:
username = row["username"].strip()
rname = f"{args.assignment}{SEPARATOR}{username}"
if repo_exists(args.org, rname):
print(f" {rname} (bereits vorhanden)")
skip += 1
continue
if copy_repo(template_owner, args.template, args.org, rname,
private=not args.public):
add_collaborator(args.org, rname, username)
pr_ok = create_feedback_pr(args.org, rname)
print(f"{rname}" + (" (inkl. Feedback-PR)" if pr_ok else ""))
ok += 1
else:
skip += 1
print(f"\nFertig: {ok} Repos erstellt, {skip} übersprungen/fehlgeschlagen.")
print(f"Repos unter: {GITEA_URL}/{args.org}")
def cmd_create_group_assignment(args):
"""
Erstellt pro Gruppe ein gemeinsames Repo und ein eigenes Gitea-Team.
Für jede Gruppe wird:
1. Ein Repo <assignment>-<group> angelegt (Kopie des Templates)
2. Ein Gitea-Team <assignment>-<group> angelegt
3. Alle Gruppenmitglieder ins Team eingetragen
4. Das Repo dem Gruppen-Team und dem 'students'-Team zugewiesen
"""
g = get_gitea()
org = get_org(g, args.org)
template_owner = args.template_owner or args.org
groups = read_groups_csv(args.csv)
total_members = sum(len(m) for m in groups.values())
print(f"\n→ Gruppen-Assignment '{args.assignment}'")
print(f" {len(groups)} Gruppen, {total_members} Mitglieder gesamt")
print(f" Template : {template_owner}/{args.template}\n")
ok = skip = 0
for group_name, members in sorted(groups.items()):
rname = f"{args.assignment}{SEPARATOR}{group_name}"
team_name = rname # Gitea-Team trägt denselben Namen wie das Repo
print(f" Gruppe '{group_name}' → Repo '{rname}' ({len(members)} Mitglieder)")
# 1. Repo erstellen
if repo_exists(args.org, rname):
print(f" Repo bereits vorhanden, übersprungen")
skip += 1
else:
if not copy_repo(template_owner, args.template, args.org, rname,
private=not args.public):
skip += 1
continue
# 2. Gruppen-Team anlegen / holen
group_team = ensure_org_team(org, team_name, permission="write")
# 3. Mitglieder ins Gruppen-Team eintragen
for username in members:
if add_member_to_team(group_team, username):
print(f"{username}")
else:
print(f"{username} (Team-Zuweisung fehlgeschlagen)")
# 4. Repo den Teams zuweisen
add_repo_to_team(group_team, args.org, rname)
pr_ok = create_feedback_pr(args.org, rname)
print(f"{GITEA_URL}/{args.org}/{rname}" + (" (inkl. Feedback-PR)" if pr_ok else ""))
ok += 1
print(f"\nFertig: {ok} Gruppen-Repos erstellt, {skip} übersprungen/fehlgeschlagen.")
def cmd_list_submissions(args):
"""Zeigt letzten Commit je Repo für ein Assignment (Einzel oder Gruppe)."""
get_gitea()
repos = list_org_repos(args.org)
prefix = f"{args.assignment}{SEPARATOR}"
asgn_repos = sorted(
[r for r in repos if r["name"].startswith(prefix)],
key=lambda r: r["name"]
)
if not asgn_repos:
print(f"Keine Repos für '{args.assignment}' in '{args.org}' gefunden.")
return
label_w = 28
print(f"\n{'Team / Student':<{label_w}} {'Letzter Commit':<22} {'Autor':<20} {'Nachricht'}")
print("" * 95)
for repo in asgn_repos:
label = repo["name"][len(prefix):]
commit = get_last_commit(args.org, repo["name"])
date, author, message = format_commit(commit)
print(f" {label:<{label_w-2}} {date:<22} {author:<20} {message}")
print(f"\nGesamt: {len(asgn_repos)} Repos")
def cmd_list_group_submissions(args):
"""Wie list-submissions, zeigt zusätzlich alle Commits pro Gruppe."""
get_gitea()
repos = list_org_repos(args.org)
prefix = f"{args.assignment}{SEPARATOR}"
asgn_repos = sorted(
[r for r in repos if r["name"].startswith(prefix)],
key=lambda r: r["name"]
)
if not asgn_repos:
print(f"Keine Repos für '{args.assignment}' in '{args.org}' gefunden.")
return
for repo in asgn_repos:
group = repo["name"][len(prefix):]
print(f"\n{''*60}")
print(f" Gruppe: {group}{GITEA_URL}/{args.org}/{repo['name']}")
print(f"{''*60}")
# Die letzten 5 Commits des Repos abrufen
r = api("GET", f"/repos/{args.org}/{repo['name']}/commits",
params={"limit": 5, "page": 1})
commits = r.json() if r.status_code == 200 else []
if not commits:
print(" (noch keine Commits)")
continue
print(f" {'Datum':<20} {'Autor':<22} Nachricht")
print(f" {''*20} {''*22} {''*35}")
for c in commits:
raw = c["commit"]["author"]["date"]
try:
dt = datetime.fromisoformat(raw.replace("Z", "+00:00"))
date = dt.strftime("%Y-%m-%d %H:%M")
except Exception:
date = raw[:16]
author = c["commit"]["author"]["name"][:20]
message = c["commit"]["message"].splitlines()[0][:40]
print(f" {date:<20} {author:<22} {message}")
print()
def cmd_clone_submissions(args):
"""Klont alle Repos eines Assignments lokal (oder aktualisiert sie per git pull)."""
get_gitea()
repos = list_org_repos(args.org)
prefix = f"{args.assignment}{SEPARATOR}"
asgn_repos = sorted(
[r for r in repos if r["name"].startswith(prefix)],
key=lambda r: r["name"],
)
if not asgn_repos:
print(f"Keine Repos für '{args.assignment}' in '{args.org}' gefunden.")
return
base_dir = os.path.join(args.dir, args.assignment)
os.makedirs(base_dir, exist_ok=True)
parsed = urlparse(GITEA_URL)
auth_base = urlunparse(parsed._replace(netloc=f"token:{GITEA_TOKEN}@{parsed.netloc}"))
print(f"\n{len(asgn_repos)} Repos → {os.path.abspath(base_dir)}\n")
ok = updated = failed = 0
for repo in asgn_repos:
name = repo["name"]
label = name[len(prefix):]
dest = os.path.join(base_dir, label)
clone_url = f"{auth_base}/{args.org}/{name}.git"
if os.path.isdir(os.path.join(dest, ".git")):
result = subprocess.run(
["git", "-C", dest, "pull", "--ff-only"],
capture_output=True, text=True,
)
if result.returncode == 0:
print(f"{label} (aktualisiert)")
updated += 1
else:
print(f"{label} (pull fehlgeschlagen): {result.stderr.strip()}")
failed += 1
else:
result = subprocess.run(
["git", "clone", clone_url, dest],
capture_output=True, text=True,
)
if result.returncode == 0:
print(f"{label}")
ok += 1
else:
print(f"{label} (clone fehlgeschlagen): {result.stderr.strip()}")
failed += 1
print(f"\nFertig: {ok} geklont, {updated} aktualisiert, {failed} fehlgeschlagen.")
print(f"Verzeichnis: {os.path.abspath(base_dir)}")
def cmd_delete_assignment(args):
"""Löscht alle Repos (und Gruppen-Teams) eines Assignments."""
g = get_gitea()
org = get_org(g, args.org)
repos = list_org_repos(args.org)
prefix = f"{args.assignment}{SEPARATOR}"
targets = [r["name"] for r in repos if r["name"].startswith(prefix)]
if not targets:
print(f"Keine Repos für '{args.assignment}' gefunden.")
return
print(f"\nFolgende {len(targets)} Repos werden gelöscht:")
for name in targets:
print(f"{args.org}/{name}")
confirm = input("\nSicher? Tippe den Assignment-Namen zur Bestätigung: ").strip()
if confirm != args.assignment:
print("Abgebrochen.")
return
ok = 0
for name in targets:
r = api("DELETE", f"/repos/{args.org}/{name}")
if r.status_code == 204:
print(f" ✓ Repo gelöscht: {name}")
ok += 1
else:
print(f" ✗ Fehler bei {name}: {r.status_code}")
# Gruppen-Team löschen (falls vorhanden, bei Gruppen-Assignments)
delete_org_team_by_name(org, name)
print(f"\n{ok}/{len(targets)} Repos gelöscht.")
# ──────────────────────────────────────────────
# CLI
# ──────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(
description="Gitea Classroom GitHub Classroom Ersatz via Gitea API",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__,
)
sub = parser.add_subparsers(dest="command", required=True)
# ── add-students ──
p = sub.add_parser("add-students", help="Studenten zur Org hinzufügen")
p.add_argument("--org", required=True)
p.add_argument("--csv", required=True, help="students.csv mit Spalte 'username'")
# ── create-assignment (Einzel) ──
p = sub.add_parser("create-assignment", help="Einzel-Repos pro Student erstellen")
p.add_argument("--org", required=True)
p.add_argument("--assignment", required=True, help="z. B. ueb01")
p.add_argument("--template", required=True, help="Name des Template-Repos")
p.add_argument("--template-owner", default=None, help="Owner des Templates (Standard: --org)")
p.add_argument("--csv", required=True, help="students.csv")
p.add_argument("--public", action="store_true")
# ── create-group-assignment ──
p = sub.add_parser("create-group-assignment",
help="Gruppen-Repos erstellen (ein Repo pro Gruppe)")
p.add_argument("--org", required=True)
p.add_argument("--assignment", required=True, help="z. B. proj01")
p.add_argument("--template", required=True, help="Name des Template-Repos")
p.add_argument("--template-owner", default=None)
p.add_argument("--csv", required=True,
help="groups.csv mit Spalten 'group' und 'username'")
p.add_argument("--public", action="store_true")
# ── list-submissions ──
p = sub.add_parser("list-submissions",
help="Letzter Commit je Repo (Einzel & Gruppen)")
p.add_argument("--org", required=True)
p.add_argument("--assignment", required=True)
# ── list-group-submissions ──
p = sub.add_parser("list-group-submissions",
help="Letzte 5 Commits pro Gruppen-Repo")
p.add_argument("--org", required=True)
p.add_argument("--assignment", required=True)
# ── clone-submissions ──
p = sub.add_parser("clone-submissions",
help="Alle Abgabe-Repos lokal klonen / aktualisieren")
p.add_argument("--org", required=True)
p.add_argument("--assignment", required=True)
p.add_argument("--dir", default=".", help="Zielverzeichnis (Standard: .)")
# ── delete-assignment ──
p = sub.add_parser("delete-assignment",
help="Alle Repos + Gruppen-Teams eines Assignments löschen")
p.add_argument("--org", required=True)
p.add_argument("--assignment", required=True)
args = parser.parse_args()
dispatch = {
"add-students": cmd_add_students,
"create-assignment": cmd_create_assignment,
"create-group-assignment": cmd_create_group_assignment,
"list-submissions": cmd_list_submissions,
"list-group-submissions": cmd_list_group_submissions,
"clone-submissions": cmd_clone_submissions,
"delete-assignment": cmd_delete_assignment,
}
dispatch[args.command](args)
if __name__ == "__main__":
main()