__all__ = [ "RepoAccess", "RepoFavourite", "Repo", "UserFollow", "UserNotification", "User", "Notification", "PostVote", "Post", "Commit", "PullRequest", "EmailChangeRequest", "Comment", ] import secrets import subprocess import markdown from app import app, db, bcrypt import git from datetime import datetime, timedelta from enum import Enum from PIL import Image from cairosvg import svg2png import os import config import cairosvg import random import celery_tasks with (app.app_context()): class RepoAccess(db.Model): id = db.Column(db.Integer, primary_key=True) user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) repo_route = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False) access_level = db.Column(db.SmallInteger(), nullable=False) # 0 read-only, 1 read-write, 2 admin user = db.relationship("User", back_populates="repo_access") repo = db.relationship("Repo", back_populates="repo_access") __table_args__ = (db.UniqueConstraint("user_username", "repo_route", name="_user_repo_uc"),) def __init__(self, user, repo, level): self.user_username = user.username self.repo_route = repo.route self.access_level = level class RepoFavourite(db.Model): id = db.Column(db.Integer, primary_key=True) user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) repo_route = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False) notify_commit = db.Column(db.Boolean, default=False, nullable=False) notify_forum = db.Column(db.Boolean, default=False, nullable=False) notify_pr = db.Column(db.Boolean, default=False, nullable=False) notify_admin = db.Column(db.Boolean, default=False, nullable=False) user = db.relationship("User", back_populates="favourites") repo = db.relationship("Repo", back_populates="favourites") __table_args__ = (db.UniqueConstraint("user_username", "repo_route", name="_user_repo_uc1"),) def __init__(self, user, repo): self.user_username = user.username self.repo_route = repo.route class PostVote(db.Model): id = db.Column(db.Integer, primary_key=True) user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) post_identifier = db.Column(db.String(109), db.ForeignKey("post.identifier"), nullable=False) vote_score = db.Column(db.SmallInteger(), nullable=False) user = db.relationship("User", back_populates="votes") post = db.relationship("Post", back_populates="votes") __table_args__ = (db.UniqueConstraint("user_username", "post_identifier", name="_user_post_uc"),) def __init__(self, user, post, score): self.user_username = user.username self.post_identifier = post.identifier self.vote_score = score class User(db.Model): username = db.Column(db.String(32), unique=True, nullable=False, primary_key=True) display_name = db.Column(db.Unicode(128), unique=False, nullable=True) bio = db.Column(db.Unicode(16384), unique=False, nullable=True) password_hashed = db.Column(db.String(60), nullable=False) email = db.Column(db.String(254), nullable=True) company = db.Column(db.Unicode(64), nullable=True) company_url = db.Column(db.String(256), nullable=True) url = db.Column(db.String(256), nullable=True) show_mail = db.Column(db.Boolean, default=False, nullable=False) location = db.Column(db.Unicode(64), nullable=True) creation_date = db.Column(db.DateTime, default=datetime.utcnow) default_page_length = db.Column(db.SmallInteger, nullable=False, default=32, server_default="32") max_post_nesting = db.Column(db.SmallInteger, nullable=False, default=3, server_default="3") repositories = db.relationship("Repo", back_populates="owner") followers = db.relationship("UserFollow", back_populates="followed", foreign_keys="[UserFollow.followed_username]") follows = db.relationship("UserFollow", back_populates="follower", foreign_keys="[UserFollow.follower_username]") email_change_requests = db.relationship("EmailChangeRequest", back_populates="user") repo_access = db.relationship("RepoAccess", back_populates="user") votes = db.relationship("PostVote", back_populates="user") favourites = db.relationship("RepoFavourite", back_populates="user") commits = db.relationship("Commit", back_populates="owner") posts = db.relationship("Post", back_populates="owner") comments = db.relationship("Comment", back_populates="owner") prs = db.relationship("PullRequest", back_populates="owner") notifications = db.relationship("UserNotification", back_populates="user") def __init__(self, username, password, email=None, display_name=None): self.username = username self.password_hashed = bcrypt.generate_password_hash(password, config.HASHING_ROUNDS).decode("utf-8") self.email = "" if email: email_change_request = EmailChangeRequest(self, email) db.session.add(email_change_request) db.session.flush() self.display_name = display_name # Create the user's directory if not os.path.exists(os.path.join(config.REPOS_PATH, username)): os.makedirs(os.path.join(config.REPOS_PATH, username)) if not os.path.exists(os.path.join(config.USERDATA_PATH, username)): os.makedirs(os.path.join(config.USERDATA_PATH, username)) avatar_name = random.choice(os.listdir(config.DEFAULT_AVATARS_PATH)) if os.path.join(config.DEFAULT_AVATARS_PATH, avatar_name).endswith(".svg"): cairosvg.svg2png(url=os.path.join(config.DEFAULT_AVATARS_PATH, avatar_name), write_to="/tmp/roundabout-avatar.png") avatar = Image.open("/tmp/roundabout-avatar.png") else: avatar = Image.open(os.path.join(config.DEFAULT_AVATARS_PATH, avatar_name)) avatar.thumbnail(config.AVATAR_SIZE) avatar.save(os.path.join(config.USERDATA_PATH, username, "avatar.png")) # Create the configuration repo config_repo = Repo(self, ".config", 0) db.session.add(config_repo) notification = Notification({"type": "welcome"}) db.session.add(notification) db.session.commit() user_notification = UserNotification(self, notification, 1) db.session.add(user_notification) db.session.commit() celery_tasks.send_notification.apply_async(args=[user_notification.id]) class Repo(db.Model): route = db.Column(db.String(98), unique=True, nullable=False, primary_key=True) owner_name = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) name = db.Column(db.String(64), nullable=False) owner = db.relationship("User", back_populates="repositories") visibility = db.Column(db.SmallInteger(), nullable=False) info = db.Column(db.Unicode(512), nullable=True) url = db.Column(db.String(256), nullable=True) creation_date = db.Column(db.DateTime, default=datetime.utcnow) default_branch = db.Column(db.String(64), nullable=True, default="") commits = db.relationship("Commit", back_populates="repo") posts = db.relationship("Post", back_populates="repo") comments = db.relationship("Comment", back_populates="repo") repo_access = db.relationship("RepoAccess", back_populates="repo") favourites = db.relationship("RepoFavourite", back_populates="repo") heads = db.relationship("PullRequest", back_populates="head", foreign_keys="[PullRequest.head_route]") bases = db.relationship("PullRequest", back_populates="base", foreign_keys="[PullRequest.base_route]") has_site = db.Column(db.SmallInteger, nullable=False, default=0, # 0 means no site, 1 means it's got a site, 2 means it's the user's primary site server_default="0") # (the one accessible at username.localhost) site_branch = db.Column(db.String(64), nullable=True) last_post_id = db.Column(db.Integer, nullable=False, default=0, server_default="0") last_comment_id = db.Column(db.Integer, nullable=False, default=0, server_default="0") def __init__(self, owner, name, visibility): self.route = f"/{owner.username}/{name}" self.name = name self.owner_name = owner.username self.owner = owner self.visibility = visibility # Add the owner as an admin repo_access = RepoAccess(owner, self, 2) db.session.add(repo_access) # Create the directory if not os.path.exists(os.path.join(config.REPOS_PATH, self.owner_name, self.name)): subprocess.run(["git", "init", self.name], cwd=os.path.join(config.REPOS_PATH, self.owner_name)) class Commit(db.Model): identifier = db.Column(db.String(227), unique=True, nullable=False, primary_key=True) sha = db.Column(db.String(128), nullable=False) repo_name = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False) owner_name = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) owner_identity = db.Column(db.String(321)) receive_date = db.Column(db.DateTime, default=datetime.now) author_date = db.Column(db.DateTime) message = db.Column(db.UnicodeText) repo = db.relationship("Repo", back_populates="commits") owner = db.relationship("User", back_populates="commits") comments = db.relationship("Comment", back_populates="commit") def __init__(self, sha, owner, repo, date, message, owner_identity): self.identifier = f"{repo.route}/{sha}" self.sha = sha self.repo_name = repo.route self.repo = repo self.owner_name = owner.username self.owner = owner self.author_date = datetime.fromtimestamp(int(date)) self.message = message self.owner_identity = owner_identity notification = Notification({"type": "commit", "repo": repo.route, "commit": sha}) db.session.add(notification) db.session.commit() # save the notification to get the ID # Send a notification to all users who have enabled commit notifications for this repo for relationship in RepoFavourite.query.filter_by(repo_route=repo.route, notify_commit=True).all(): user = relationship.user user_notification = UserNotification(user, notification, 1) db.session.add(user_notification) db.session.commit() celery_tasks.send_notification.apply_async(args=[user_notification.id]) class Post(db.Model): identifier = db.Column(db.String(109), unique=True, nullable=False, primary_key=True) number = db.Column(db.Integer, nullable=False) repo_name = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False) owner_name = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) votes = db.relationship("PostVote", back_populates="post") vote_sum = db.Column(db.Integer, nullable=False, default=0) parent_id = db.Column(db.String(109), db.ForeignKey("post.identifier"), nullable=True) root_id = db.Column(db.String(109), db.ForeignKey("post.identifier"), nullable=True) state = db.Column(db.SmallInteger, nullable=True, default=1) date = db.Column(db.DateTime, default=datetime.now) last_updated = db.Column(db.DateTime, default=datetime.now) subject = db.Column(db.Unicode(384)) message = db.Column(db.UnicodeText) html = db.Column(db.UnicodeText) repo = db.relationship("Repo", back_populates="posts") owner = db.relationship("User", back_populates="posts") parent = db.relationship("Post", back_populates="children", primaryjoin="Post.parent_id==Post.identifier", foreign_keys="[Post.parent_id]", remote_side="Post.identifier") root = db.relationship("Post", primaryjoin="Post.root_id==Post.identifier", foreign_keys="[Post.root_id]", remote_side="Post.identifier") children = db.relationship("Post", remote_side="Post.parent_id", primaryjoin="Post.identifier==Post.parent_id", foreign_keys="[Post.parent_id]") def __init__(self, owner, repo, parent, subject, message): self.identifier = f"{repo.route}/{repo.last_post_id}" self.number = repo.last_post_id self.repo_name = repo.route self.repo = repo self.owner_name = owner.username self.owner = owner self.subject = subject self.message = message self.html = markdown.markdown2html(message).prettify() self.parent = parent if parent and parent.parent: self.root = parent.parent elif parent: self.root = parent else: self.root = None repo.last_post_id += 1 notification = Notification({"type": "post", "repo": repo.route, "post": self.identifier}) db.session.add(notification) db.session.commit() # save the notification to get the ID # Send a notification to all users who have enabled forum notifications for this repo for relationship in RepoFavourite.query.filter_by(repo_route=repo.route, notify_forum=True).all(): user = relationship.user user_notification = UserNotification(user, notification, 1) db.session.add(user_notification) db.session.commit() celery_tasks.send_notification.apply_async(args=[user_notification.id]) def update_date(self): self.last_updated = datetime.now() with db.session.no_autoflush: if self.parent is not None: self.parent.update_date() class Comment(db.Model): identifier = db.Column(db.String(109), unique=True, nullable=False, primary_key=True) number = db.Column(db.Integer, nullable=False) repo_name = db.Column(db.String(98), db.ForeignKey("repo.route"), nullable=False) owner_name = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) commit_identifier = db.Column(db.String(227), db.ForeignKey("commit.identifier"), nullable=False) pr_id = db.Column(db.BigInteger, db.ForeignKey("pull_request.id"), nullable=True) file = db.Column(db.String(256), nullable=True) line_number = db.Column(db.Integer, nullable=True) line_type = db.Column(db.SmallInteger, nullable=True, default=0, server_default="0") # 0 is deleted, 1 is modified state = db.Column(db.SmallInteger, nullable=True, default=1) review = db.Column(db.SmallInteger, nullable=True, default=0) date = db.Column(db.DateTime, default=datetime.now) message = db.Column(db.UnicodeText) html = db.Column(db.UnicodeText) repo = db.relationship("Repo", back_populates="comments") owner = db.relationship("User", back_populates="comments") commit = db.relationship("Commit", back_populates="comments") def __init__(self, owner, repo, commit, message, file, line_number, pr=None): self.identifier = f"{repo.route}/{repo.last_comment_id}" self.number = repo.last_comment_id self.repo_name = repo.route self.repo = repo self.owner_name = owner.username self.owner = owner self.commit_identifier = commit.identifier self.commit = commit self.message = message self.html = markdown.markdown2html(message).prettify() self.file = file self.line_number = int(line_number[1:]) self.line_type = int(line_number[0] == "+") if pr: self.pr = pr repo.last_comment_id += 1 @property def text(self): return self.html @text.setter def text(self, value): self.html = markdown.markdown2html(value).prettify() self.message = value # message is stored in markdown format for future editing or plaintext display class UserNotification(db.Model): id = db.Column(db.Integer, primary_key=True) user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) notification_id = db.Column(db.BigInteger, db.ForeignKey("notification.id")) attention_level = db.Column(db.SmallInteger, nullable=False) # 0 is read read_time = db.Column(db.DateTime, nullable=True) user = db.relationship("User", back_populates="notifications") notification = db.relationship("Notification", back_populates="notifications") __table_args__ = (db.UniqueConstraint("user_username", "notification_id", name="_user_notification_uc"),) def __init__(self, user, notification, level): self.user_username = user.username self.notification_id = notification.id self.attention_level = level def mark_read(self): self.read_time = datetime.utcnow() self.attention_level = 0 def mark_unread(self): self.attention_level = 4 class UserFollow(db.Model): id = db.Column(db.Integer, primary_key=True) follower_username = db.Column(db.String(32), db.ForeignKey("user.username", ondelete="CASCADE"), nullable=False) followed_username = db.Column(db.String(32), db.ForeignKey("user.username", ondelete="CASCADE"), nullable=False) follower = db.relationship("User", back_populates="followers", foreign_keys=[follower_username]) followed = db.relationship("User", back_populates="follows", foreign_keys=[followed_username]) def __init__(self, follower_username, followed_username): self.follower_username = follower_username self.followed_username = followed_username class Notification(db.Model): id = db.Column(db.BigInteger, primary_key=True, autoincrement=True) data = db.Column(db.dialects.postgresql.JSONB, nullable=False, default={}) notifications = db.relationship("UserNotification", back_populates="notification") timestamp = db.Column(db.DateTime, nullable=False, default=datetime.now) def __init__(self, json): self.data = json class PullRequest(db.Model): id = db.Column(db.BigInteger, primary_key=True, autoincrement=True) head_route = db.Column(db.String(98), db.ForeignKey("repo.route", ondelete="CASCADE"), nullable=False) base_route = db.Column(db.String(98), db.ForeignKey("repo.route", ondelete="CASCADE"), nullable=False) owner_name = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) state = db.Column(db.SmallInteger, nullable=False, default=0) # 0 pending, 1 merged, 2 rejected head = db.relationship("Repo", back_populates="heads", foreign_keys=[head_route]) base = db.relationship("Repo", back_populates="bases", foreign_keys=[base_route]) head_branch = db.Column(db.String(64), nullable=False) base_branch = db.Column(db.String(64), nullable=False) owner = db.relationship("User", back_populates="prs") timestamp = db.Column(db.DateTime, nullable=False, default=datetime.now) def __init__(self, head, head_branch, base, base_branch, owner): self.head = head self.base = base self.head_branch = head_branch self.base_branch = base_branch self.owner = owner class EmailChangeRequest(db.Model): id = db.Column(db.BigInteger, primary_key=True, autoincrement=True) user_username = db.Column(db.String(32), db.ForeignKey("user.username"), nullable=False) new_email = db.Column(db.String(254), nullable=False) code = db.Column(db.String(64), nullable=False) expires_on = db.Column(db.DateTime, nullable=False) user = db.relationship("User", back_populates="email_change_requests") def __init__(self, user, new_email): self.user = user self.new_email = new_email self.code = hex(secrets.randbits(256)).removeprefix("0x") self.expires_on = datetime.now() + timedelta(days=1)