417 lines
15 KiB
Python
417 lines
15 KiB
Python
import pytz
|
|
import random
|
|
from infra.log.module_logger import ModuleLogger
|
|
from typing import Optional
|
|
|
|
from app.authentication.backend.common.config.app_settings import app_settings
|
|
|
|
from app.authentication.backend.models.constants import DepotStatus
|
|
from app.authentication.backend.infra.code_management.gitea.gitea import Gitea
|
|
from app.authentication.backend.models.gitea.models import CodeDepotDoc
|
|
from infra.exception.exceptions import InvalidOperationError
|
|
from infra.utils.date import get_sunday
|
|
from datetime import datetime
|
|
from datetime import datetime, timedelta, timezone
|
|
from dateutil.parser import parse
|
|
|
|
|
|
class CodeDepotHandler:
|
|
def __init__(self) -> None:
|
|
self.gitea_url = app_settings.GITEA_URL
|
|
self.gitea_token = app_settings.GITEA_TOKEN
|
|
self.gitea_org = app_settings.GITEA_DEPOT_ORGANIZATION
|
|
|
|
self.code_depot_domain_name = app_settings.CODE_DEPOT_DOMAIN_NAME
|
|
self.code_depot_ssh_port = app_settings.CODE_DEPOT_SSH_PORT
|
|
self.code_depot_http_port = app_settings.CODE_DEPOT_HTTP_PORT
|
|
|
|
self.gitea_admin = Gitea(
|
|
self.gitea_url, token_text=self.gitea_token, auth=None, verify=False
|
|
)
|
|
self.product_org = self.gitea_admin.get_org_by_name(self.gitea_org)
|
|
self.module_logger = ModuleLogger(sender_id="CodeDepotManager")
|
|
|
|
async def check_depot_name_availabe(self, code_depot_name: str) -> bool:
|
|
"""Return True if the depot name is available, otherwise return False
|
|
|
|
Parameters:
|
|
code_depot_name (str): the name of the code depot
|
|
|
|
Returns:
|
|
bool: True if the depot name is availabe, otherwise return False
|
|
"""
|
|
|
|
result = await CodeDepotDoc.find_one(CodeDepotDoc.depot_name == code_depot_name)
|
|
|
|
if result:
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
def check_code_depot_exist(self, code_depot_name: str) -> bool:
|
|
"""Check and return True if the code deport with the name exist in Gitea, otherwise return False
|
|
|
|
Parameters:
|
|
code_depot_name (str): the name of the code depot
|
|
|
|
Returns:
|
|
bool: True if the code depot exist, otherwise return False
|
|
"""
|
|
all_depots = self.product_org.get_repositories()
|
|
for depot in all_depots:
|
|
if depot.name.lower() == code_depot_name.lower():
|
|
return True
|
|
|
|
return False
|
|
|
|
async def __generate_new_code_depot_name(
|
|
self, code_depot_name: str, gitea_code_depot_names: list[str]
|
|
) -> str:
|
|
"""Generate a new code depot name if the code depot name already exists
|
|
|
|
Parameters:
|
|
code_depot_name (str): the name of the code depot
|
|
|
|
Returns:
|
|
str: the new code depot name
|
|
"""
|
|
code_depot_name = code_depot_name.lower()
|
|
depot_doc = await CodeDepotDoc.find_one(
|
|
CodeDepotDoc.depot_name == code_depot_name
|
|
)
|
|
|
|
# if the depot name already exists in the database or Gitea, generate a new name
|
|
if depot_doc or (code_depot_name in gitea_code_depot_names):
|
|
new_code_depot_name = "{}-{}".format(
|
|
code_depot_name, random.randint(10000, 99999)
|
|
)
|
|
while new_code_depot_name in gitea_code_depot_names:
|
|
new_code_depot_name = "{}-{}".format(
|
|
code_depot_name, random.randint(10000, 99999)
|
|
)
|
|
|
|
return await self.__generate_new_code_depot_name(
|
|
new_code_depot_name, gitea_code_depot_names
|
|
)
|
|
else:
|
|
return code_depot_name
|
|
|
|
async def create_code_depot(self, product_id, code_depot_name) -> Optional[str]:
|
|
"""Create a new git code depot
|
|
|
|
Parameters:
|
|
product_id (str): the id of the product
|
|
code_depot_name (str): the name of the code depot
|
|
|
|
Returns:
|
|
str: return code depot id if it's created successfully, else return None
|
|
"""
|
|
gitea_code_depot_names = [
|
|
depot.name.lower() for depot in self.product_org.get_repositories()
|
|
]
|
|
code_depot_name_to_be_created = await self.__generate_new_code_depot_name(
|
|
code_depot_name, gitea_code_depot_names
|
|
)
|
|
|
|
new_depot_doc = CodeDepotDoc(
|
|
depot_name=code_depot_name_to_be_created,
|
|
product_id=product_id,
|
|
depot_status=DepotStatus.TO_BE_CREATED,
|
|
)
|
|
await new_depot_doc.create()
|
|
|
|
try:
|
|
self.product_org.create_repo(
|
|
repoName=code_depot_name_to_be_created,
|
|
description=product_id,
|
|
private=True,
|
|
autoInit=True,
|
|
default_branch="main",
|
|
)
|
|
except Exception as e:
|
|
await self.module_logger.log_exception(
|
|
exception=e,
|
|
text=f"Failed to create depot {code_depot_name_to_be_created} in Gitea. Error: {e}",
|
|
properties={
|
|
"code_depot_name_to_be_created": code_depot_name_to_be_created
|
|
},
|
|
)
|
|
raise InvalidOperationError(
|
|
f"Failed to create depot {code_depot_name_to_be_created} in Gitea"
|
|
)
|
|
|
|
await new_depot_doc.set({CodeDepotDoc.depot_status: DepotStatus.CREATED})
|
|
await new_depot_doc.save()
|
|
|
|
return str(new_depot_doc.id)
|
|
|
|
def get_depot_ssh_url(self, code_depot_name: str) -> str:
|
|
"""Return the ssh url of the code depot
|
|
|
|
Parameters:
|
|
depot_name (str): the name of the depot
|
|
|
|
Returns:
|
|
str: the ssh url of the code depot
|
|
"""
|
|
if self.code_depot_ssh_port != "22":
|
|
return f"git@{self.code_depot_domain_name}:{self.code_depot_ssh_port}/{self.gitea_org}/{code_depot_name}.git"
|
|
else:
|
|
return f"git@{self.code_depot_domain_name}/{self.gitea_org}/{code_depot_name}.git"
|
|
|
|
def get_depot_http_url(self, code_depot_name: str) -> str:
|
|
"""Return the http url of the code depot
|
|
|
|
Parameters:
|
|
depot_name (str): the name of the depot
|
|
|
|
Returns:
|
|
str: the http url of the code depot
|
|
"""
|
|
if self.code_depot_http_port in ["443", "3443"]:
|
|
return f"https://{self.code_depot_domain_name}:{self.code_depot_http_port}/{self.gitea_org}/{code_depot_name}.git"
|
|
else:
|
|
return f"http://{self.code_depot_domain_name}:{self.code_depot_http_port}/{self.gitea_org}/{code_depot_name}.git"
|
|
|
|
def get_depot_http_url_with_user_name(
|
|
self, code_depot_name: str, user_name: str
|
|
) -> str:
|
|
"""Return the http url of the code depot
|
|
|
|
Parameters:
|
|
depot_name (str): the name of the depot
|
|
|
|
Returns:
|
|
str: the http url of the code depot
|
|
"""
|
|
if self.code_depot_http_port in ["443", "3443"]:
|
|
return f"https://{user_name}@{self.code_depot_domain_name}:{self.code_depot_http_port}/{self.gitea_org}/{code_depot_name}.git"
|
|
else:
|
|
return f"http://{user_name}@{self.code_depot_domain_name}:{self.code_depot_http_port}/{self.gitea_org}/{code_depot_name}.git"
|
|
|
|
async def get_depot_users(self, code_depot_name: str) -> list[str]:
|
|
"""Return list of user names have permission to access the depot
|
|
|
|
Parameters:
|
|
depot_name (str): the name of the depot
|
|
|
|
Returns:
|
|
list: list of user names
|
|
"""
|
|
|
|
result = await CodeDepotDoc.find_one(CodeDepotDoc.depot_name == code_depot_name)
|
|
|
|
if result:
|
|
return result.collaborators
|
|
else:
|
|
return []
|
|
|
|
async def update_depot_user_password(self, user_name: str, password: str) -> bool:
|
|
"""Update the password of the user in Gitea
|
|
|
|
Parameters:
|
|
user_name (str): the name of the user
|
|
password (str): the new password of the user
|
|
|
|
Returns:
|
|
bool: True if operations succeed, otherwise return False
|
|
"""
|
|
depot_user = self.gitea_admin.get_user_by_name(user_name)
|
|
|
|
if depot_user:
|
|
try:
|
|
self.gitea_admin.update_user_password(user_name, password)
|
|
except Exception as e:
|
|
await self.module_logger.log_exception(
|
|
exception=e,
|
|
text=f"Failed to update password for user {user_name} in Gitea.",
|
|
properties={"user_name": user_name},
|
|
)
|
|
return False
|
|
else:
|
|
await self.module_logger.log_error(
|
|
error=f"User {user_name} does not exist in Gitea.",
|
|
properties={"user_name": user_name},
|
|
)
|
|
return False
|
|
|
|
return True
|
|
|
|
async def create_depot_user(
|
|
self, user_name: str, password: str, email: str
|
|
) -> bool:
|
|
"""Create a new user in Gitea
|
|
Parameters:
|
|
user_name (str): the name of the user
|
|
password (str): the password of the user
|
|
email (str): email address of the user
|
|
|
|
Returns:
|
|
bool: True if operations succeed, otherwise return False
|
|
"""
|
|
depot_user = self.gitea_admin.get_user_by_name(user_name)
|
|
|
|
if depot_user:
|
|
await self.module_logger.log_info(
|
|
info=f"User {user_name} exist in Gitea.",
|
|
properties={"user_name": user_name},
|
|
)
|
|
return True
|
|
else:
|
|
try:
|
|
await self.module_logger.log_info(
|
|
info=f"Create user {user_name} in Gitea with password.",
|
|
properties={"user_name": user_name},
|
|
)
|
|
depot_user = self.gitea_admin.create_user(
|
|
user_name=user_name,
|
|
login_name=user_name,
|
|
password=password,
|
|
email=email,
|
|
change_pw=False,
|
|
send_notify=False,
|
|
)
|
|
except Exception as e:
|
|
await self.module_logger.log_exception(
|
|
exception=e,
|
|
text=f"Failed to create user {user_name} in Gitea.",
|
|
properties={"user_name": user_name},
|
|
)
|
|
return False
|
|
|
|
return True
|
|
|
|
async def grant_user_depot_access(
|
|
self, user_name: str, code_depot_name: str
|
|
) -> bool:
|
|
"""Grant user access to the code depot in Gitea
|
|
|
|
Parameters:
|
|
user_name (str): the name of the user
|
|
code_depot_name (str): the name of the code depot
|
|
|
|
Returns:
|
|
bool: True if operations succeed, otherwise return False
|
|
"""
|
|
try:
|
|
code_depot = self.product_org.get_repository(code_depot_name)
|
|
except Exception as e:
|
|
await self.module_logger.log_exception(
|
|
exception=e,
|
|
text=f"Failed to get depot {code_depot_name} in Gitea.",
|
|
properties={"code_depot_name": code_depot_name},
|
|
)
|
|
return False
|
|
|
|
try:
|
|
if code_depot.add_collaborator(user_name=user_name, permission="Write"):
|
|
code_depot_doc = await CodeDepotDoc.find_one(
|
|
CodeDepotDoc.depot_name == code_depot_name
|
|
)
|
|
|
|
if code_depot_doc:
|
|
if user_name not in code_depot_doc.collaborators:
|
|
code_depot_doc.collaborators.append(user_name)
|
|
await code_depot_doc.save()
|
|
except Exception as e:
|
|
await self.module_logger.log_exception(
|
|
exception=e,
|
|
text=f"Failed to grant permission for user {user_name} to access code depot {code_depot_name}.",
|
|
properties={"user_name": user_name, "code_depot_name": code_depot_name},
|
|
)
|
|
return False
|
|
|
|
return True
|
|
|
|
async def generate_statistic_result(
|
|
self, code_depot_name: str
|
|
) -> Optional[dict[str, any]]:
|
|
"""Call Gitea API and collect statistic result of the repository.
|
|
|
|
Args:
|
|
code_depot_name (str): the name of the code depot
|
|
|
|
Returns:
|
|
dict[str, any]: statistic result
|
|
"""
|
|
|
|
try:
|
|
code_depot = self.product_org.get_repository(code_depot_name)
|
|
except Exception as e:
|
|
await self.module_logger.log_exception(
|
|
exception=e,
|
|
text=f"Failed to get depot {code_depot_name} in Gitea.",
|
|
properties={"code_depot_name": code_depot_name},
|
|
)
|
|
return None
|
|
|
|
commits = code_depot.get_commits()
|
|
|
|
if len(commits) > 0:
|
|
commit_dates = []
|
|
for commit in commits:
|
|
commit_dates.append(parse(commit.created).date())
|
|
|
|
today = datetime.now(timezone.utc)
|
|
last_sunday = get_sunday(today)
|
|
# only get commits statistic of the last 20 weeks
|
|
last_20_sundays = [last_sunday - timedelta(weeks=i) for i in range(20)]
|
|
|
|
weekly_commits = {}
|
|
for date in last_20_sundays:
|
|
weekly_commits[date.strftime("%Y-%m-%d")] = 0
|
|
|
|
for commit_date in commit_dates:
|
|
commit_date_sunday = get_sunday(commit_date).strftime("%Y-%m-%d")
|
|
if commit_date_sunday in weekly_commits:
|
|
weekly_commits[commit_date_sunday] += 1
|
|
|
|
last_update = parse(commits[-1].created)
|
|
results = {
|
|
"total_commits": len(commits),
|
|
"last_commiter": commits[-1].commit["committer"]["name"],
|
|
"last_update": last_update.astimezone(pytz.UTC),
|
|
"weekly_commits": weekly_commits,
|
|
}
|
|
else:
|
|
results = {
|
|
"total_commits": 0,
|
|
"last_commiter": "",
|
|
"last_update": None,
|
|
"weekly_commits": {},
|
|
}
|
|
|
|
return results
|
|
|
|
async def analyze_code_depots(self):
|
|
code_depot_docs = await CodeDepotDoc.find(
|
|
CodeDepotDoc.depot_status == DepotStatus.CREATED
|
|
).to_list()
|
|
|
|
for code_depot_doc in code_depot_docs:
|
|
await self.module_logger.log_info(
|
|
info="Start to analyze code depot {}".format(code_depot_doc.depot_name),
|
|
properties={"code_depot_name": code_depot_doc.depot_name},
|
|
)
|
|
statistic_result = await self.generate_statistic_result(
|
|
code_depot_doc.depot_name
|
|
)
|
|
|
|
# cannot get the statistic result, probably it cannot find the
|
|
# git repository in gitea
|
|
if not statistic_result:
|
|
continue
|
|
|
|
code_depot_doc.total_commits = statistic_result["total_commits"]
|
|
code_depot_doc.last_commiter = statistic_result["last_commiter"]
|
|
code_depot_doc.last_update = statistic_result["last_update"]
|
|
code_depot_doc.weekly_commits = statistic_result["weekly_commits"]
|
|
await code_depot_doc.save()
|
|
|
|
async def fetch_code_depot(self, code_depot_id: str) -> Optional[dict[str, any]]:
|
|
code_depot_doc = await CodeDepotDoc.get(code_depot_id)
|
|
if code_depot_doc:
|
|
return code_depot_doc.model_dump()
|
|
else:
|
|
return None
|