225 lines
7.0 KiB
Python
225 lines
7.0 KiB
Python
import logging as lg
|
||
import os
|
||
|
||
import requests
|
||
from django.conf import settings
|
||
from django.db.models.signals import post_save
|
||
from django.dispatch import receiver
|
||
|
||
from home.models.pages import CoursePage, ModuleLessonPage
|
||
|
||
GITEA_ORG_NAME = "Studio77"
|
||
|
||
logger = lg.getLogger(__name__)
|
||
|
||
|
||
@receiver(post_save, sender=CoursePage)
|
||
def create_gitea_team_on_course_creation(sender, instance, created, **kwargs):
|
||
if not instance.live:
|
||
logger.debug(
|
||
f"Course {instance.title} is not live, skipping Gitea team creation"
|
||
)
|
||
return
|
||
|
||
course = instance
|
||
team_name = f"course-{course.id}"
|
||
api_url = getattr(settings, "GITEA_URL", None)
|
||
|
||
if not api_url:
|
||
logger.debug("GITEA_URL is not set, skipping Gitea team creation")
|
||
return
|
||
|
||
# check if team already exists
|
||
try:
|
||
response = requests.get(
|
||
f"{api_url}/orgs/{GITEA_ORG_NAME}/teams",
|
||
timeout=5,
|
||
headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
|
||
)
|
||
response.raise_for_status()
|
||
teams = response.json()
|
||
if any(team["name"] == team_name for team in teams):
|
||
logger.info(f"Gitea team {team_name} already exists, skipping creation")
|
||
return
|
||
except Exception as e:
|
||
logger.exception(
|
||
f"Failed to check existing Gitea teams: {e}\n{response.text}",
|
||
e,
|
||
)
|
||
return
|
||
|
||
url = f"{api_url}/orgs/{GITEA_ORG_NAME}/teams"
|
||
payload = {
|
||
"can_create_org_repo": False,
|
||
"description": f"Team for course {course.title}",
|
||
"includes_all_repositories": False,
|
||
"name": team_name,
|
||
"permission": "read",
|
||
"units": [
|
||
# "repo.actions",
|
||
"repo.code",
|
||
# "repo.issues",
|
||
# "repo.ext_issues",
|
||
# "repo.wiki",
|
||
# "repo.ext_wiki",
|
||
# "repo.pulls",
|
||
# "repo.releases",
|
||
# "repo.projects",
|
||
# "repo.ext_wiki",
|
||
],
|
||
}
|
||
try:
|
||
response = requests.post(
|
||
url,
|
||
json=payload,
|
||
timeout=5,
|
||
headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
|
||
)
|
||
response.raise_for_status()
|
||
logger.info(f"Successfully created Gitea team for course {course.title}")
|
||
except Exception as e:
|
||
logger.exception(
|
||
f"Failed to create Gitea team for course {course.title}: {e}\n{response.text}",
|
||
e,
|
||
)
|
||
|
||
# def repo():
|
||
# # check if repository already exists
|
||
# try:
|
||
# response = requests.get(
|
||
# f"{api_url}/orgs/{GITEA_ORG_NAME}/repos",
|
||
# timeout=5,
|
||
# headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
|
||
# )
|
||
# response.raise_for_status()
|
||
# repos = response.json()
|
||
# if any(repo["name"] == team_name for repo in repos):
|
||
# logger.debug(
|
||
# f"Gitea repository {team_name} already exists, skipping creation"
|
||
# )
|
||
# return
|
||
# except Exception as e:
|
||
# logger.exception(
|
||
# f"Failed to check existing Gitea repositories: {e}\n{response.text}",
|
||
# e,
|
||
# )
|
||
# return
|
||
#
|
||
# # create course repository
|
||
# repo_name = f"course-{course.id}"
|
||
# url = f"{api_url}/orgs/{GITEA_ORG_NAME}/repos"
|
||
# payload = {
|
||
# "auto_init": True,
|
||
# "default_branch": "main",
|
||
# "description": f"{course.title}",
|
||
# "name": repo_name,
|
||
# "private": True,
|
||
# }
|
||
#
|
||
# try:
|
||
# response = requests.post(
|
||
# url,
|
||
# json=payload,
|
||
# timeout=5,
|
||
# headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
|
||
# )
|
||
# response.raise_for_status()
|
||
# repo_url = response.json().get("url", None)
|
||
# course.repository_url = repo_url
|
||
# course.save(update_fields=["repository_url"])
|
||
# logger.info(
|
||
# f"Successfully created Gitea repository for course {course.title}"
|
||
# )
|
||
# except Exception as e:
|
||
# logger.exception(
|
||
# f"Failed to create Gitea repository for course {course.title}: {e}\n{response.text}",
|
||
# e,
|
||
# )
|
||
#
|
||
#
|
||
|
||
|
||
@receiver(post_save, sender=ModuleLessonPage)
|
||
def create_gitea_repo_on_lesson_creation(
|
||
sender, instance: ModuleLessonPage, created, **kwargs
|
||
):
|
||
if not instance.live:
|
||
logger.debug(
|
||
f"Lesson {instance.title} is not live, skipping Gitea repository creation"
|
||
)
|
||
return
|
||
|
||
course = instance.module.course
|
||
repo_name = f"course-{course.id}-lesson-{instance.id}"
|
||
|
||
if not instance.create_gitea_repo:
|
||
logger.debug(
|
||
f"Lesson {instance.title} is not marked for Gitea repository creation, skipping"
|
||
)
|
||
return
|
||
|
||
if not course.live:
|
||
logger.debug(
|
||
f"Course {course.title} is not live, skipping Gitea repository creation for lesson {instance.title}"
|
||
)
|
||
return
|
||
|
||
api_url = getattr(settings, "GITEA_URL", None)
|
||
|
||
if not api_url:
|
||
logger.debug("GITEA_URL is not set, skipping Gitea repository creation")
|
||
return
|
||
|
||
# check if repository already exists
|
||
try:
|
||
response = requests.get(
|
||
f"{api_url}/orgs/{GITEA_ORG_NAME}/repos",
|
||
timeout=5,
|
||
headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
|
||
)
|
||
response.raise_for_status()
|
||
repos = response.json()
|
||
if any(repo["name"] == repo_name for repo in repos):
|
||
logger.info(
|
||
f"Gitea repository {repo_name} already exists, skipping creation"
|
||
)
|
||
return
|
||
except Exception as e:
|
||
logger.exception(
|
||
f"Failed to check existing Gitea repositories: {e}\n{response.text}",
|
||
e,
|
||
)
|
||
return
|
||
|
||
# create lesson repository
|
||
url = f"{api_url}/orgs/{GITEA_ORG_NAME}/repos"
|
||
payload = {
|
||
"auto_init": True,
|
||
"default_branch": "main",
|
||
"description": f"{instance.module.course} – {instance.module}: {instance.title}",
|
||
"name": repo_name,
|
||
"private": True,
|
||
}
|
||
|
||
try:
|
||
response = requests.post(
|
||
url,
|
||
json=payload,
|
||
timeout=5,
|
||
headers={"Authorization": f"token {os.getenv('GITEA_API_TOKEN')}"},
|
||
)
|
||
response.raise_for_status()
|
||
repo_url = response.json().get("html_url")
|
||
|
||
instance.gitea_repo_url = repo_url
|
||
instance.save(update_fields=["gitea_repo_url"])
|
||
|
||
logger.info(
|
||
f"Successfully created Gitea repository for lesson {instance.title}: {repo_url}"
|
||
)
|
||
except Exception as e:
|
||
logger.exception(
|
||
f"Failed to create Gitea repository for lesson {instance.title}: {e}\n{response.text}",
|
||
e,
|
||
)
|