from typing import Dict, List
import focs_gitea
from canvasapi.paginated_list import PaginatedList
from mattermostdriver import Driver
from joint_teapot.config import settings
from joint_teapot.utils.logger import logger
from joint_teapot.workers.gitea import Gitea
class Mattermost:
def __init__(
self,
access_token: str = "", # nosec
team_name: str = "",
domain_name: str = "",
suffix: str = "",
):
access_token = access_token or settings.mattermost_access_token
team_name = team_name or settings.mattermost_team
domain_name = domain_name or settings.mattermost_domain_name
suffix = suffix or settings.mattermost_suffix
self.url = domain_name
self.url_suffix = suffix
self.endpoint = Driver(
{
"url": domain_name,
"port": 443,
"basepath": suffix + "/api/v4",
"token": access_token,
}
)
try:
operator = self.endpoint.login()
except Exception:
logger.error("Cannot login to Mattermost")
return
if "admin" not in operator["roles"]:
logger.error("Please make sure you have enough permission")
try:
self.team = self.endpoint.teams.get_team_by_name(team_name)
except Exception as e:
logger.error(f"Cannot get team {team_name}: {e}")
return
def create_channels_for_groups(
self,
groups: Dict[str, List[str]],
suffix: str = "",
invite_teaching_team: bool = True,
) -> None:
for group_name, members in groups.items():
channel_name = group_name + suffix
try:
channel = self.endpoint.channels.create_channel(
{
"team_id": self.team["id"],
"name": channel_name,
"display_name": channel_name,
"type": "P", # create private channels
}
)
logger.info(f"Added group {channel_name} to Mattermost")
except Exception as e:
logger.warning(
f"Error when creating channel {channel_name}: {e} Perhaps channel already exists?"
)
continue
if invite_teaching_team:
members.extend(settings.mattermost_teaching_team)
for member in members:
try:
mmuser = self.endpoint.users.get_user_by_username(member)
except Exception:
logger.warning(
f"User {member} is not found on the Mattermost server"
)
self.endpoint.posts.create_post(
{
"channel_id": channel["id"],
"message": f"User {member} is not found on the Mattermost server",
}
)
continue
# code for adding student to mm, disabled since there is no need to do that
# try:
# mmuser = self.endpoint.users.create_user({'email':f"{member}@sjtu.edu.cn", 'username':member, auth_service:"gitlab"})
# except e:
# logger.error(f"Error creating user {member}")
# continue
try:
self.endpoint.channels.add_user(
channel["id"], {"user_id": mmuser["id"]}
)
except Exception:
logger.warning(f"User {member} is not in the team")
self.endpoint.posts.create_post(
{
"channel_id": channel["id"],
"message": f"User {member} is not in the team",
}
)
logger.info(f"Added member {member} to channel {channel_name}")
def create_channels_for_individuals(
self,
students: PaginatedList,
invite_teaching_team: bool = True,
) -> None:
for student in students:
display_name = student.name
channel_name = student.sis_id
try:
channel = self.endpoint.channels.create_channel(
{
"team_id": self.team["id"],
"name": channel_name,
"display_name": display_name,
"type": "P", # create private channels
}
)
logger.info(
f"Added channel {display_name} ({channel_name}) to Mattermost"
)
except Exception as e:
logger.warning(
f"Error when creating channel {channel_name}: {e} Perhaps channel already exists?"
)
continue
members = [student.login_id]
if invite_teaching_team:
members.extend(settings.mattermost_teaching_team)
for member in members:
try:
mmuser = self.endpoint.users.get_user_by_username(member)
except Exception:
logger.warning(
f"User {member} is not found on the Mattermost server"
)
self.endpoint.posts.create_post(
{
"channel_id": channel["id"],
"message": f"User {member} is not found on the Mattermost server",
}
)
continue
# code for adding student to mm, disabled since there is no need to do that
# try:
# mmuser = self.endpoint.users.create_user({'email':f"{member}@sjtu.edu.cn", 'username':member, auth_service:"gitlab"})
# except e:
# logger.error(f"Error creating user {member}")
# continue
try:
self.endpoint.channels.add_user(
channel["id"], {"user_id": mmuser["id"]}
)
except Exception:
logger.warning(f"User {member} is not in the team")
self.endpoint.posts.create_post(
{
"channel_id": channel["id"],
"message": f"User {member} is not in the team",
}
)
logger.info(f"Added member {member} to channel {channel_name}")
def create_webhooks_for_repos(
self, repos: List[str], gitea: Gitea, git_suffix: bool
) -> None:
# one group corresponds to one repo so these concepts can be used interchangeably
for repo in repos:
logger.info(f"Creating webhooks for repo {gitea.org_name}/{repo}")
channel_name = f"{repo}-git" if git_suffix else repo
try:
mm_channel = self.endpoint.channels.get_channel_by_name(
self.team["id"], channel_name
)
except Exception as e:
logger.warning(
f"Error when getting channel {channel_name} from Mattermost team {self.team['name']}: {e}"
)
continue
try:
mm_webhook = self.endpoint.webhooks.create_incoming_hook(
{
"channel_id": mm_channel["id"],
"display_name": f"Gitea integration for {self.team['name']}/{repo}",
"channel_locked": True,
}
)
except Exception as e:
logger.error(f"Error when creating incoming webhook at Mattermost: {e}")
continue
try:
gitea.repository_api.repo_create_hook(
gitea.org_name,
repo,
body=focs_gitea.CreateHookOption(
active=True,
type="slack",
events=[
"issues_only",
"issue_comment",
"pull_request_only",
"pull_request_comment",
"pull_request_review",
],
config={
"url": f"https://{self.url}{self.url_suffix}/hooks/{mm_webhook['id']}",
"username": "FOCS Gitea",
"icon_url": f"https://{self.url}{self.url_suffix}/api/v4/brand/image",
"content_type": "json",
"channel": channel_name,
},
),
)
except Exception as e:
logger.warning(f"Error when creating outgoing webhook at Gitea: {e}")
# unused since we can give students invitation links instead
def invite_students_to_team(self, students: List[str]) -> None:
for student in students:
try:
mmuser = self.endpoint.users.get_user_by_username(student)
except Exception:
logger.warning(f"User {student} is not found on the Mattermost server")
continue
self.endpoint.teams.add_user_to_team(
self.team["id"], {"user_id": mmuser["id"], "team_id": self.team["id"]}
)
logger.info(f"Added user {student} to team {self.team['name']}")