import common import time import os import config import email_send import shutil from celery import shared_task from app import db, _ from smtplib import SMTP from celery.utils.log import get_task_logger from sqlalchemy.orm import make_transient from datetime import datetime @shared_task(ignore_result=False) def send_notification(user_notification_id): from models import UserNotification, Commit, Post, PullRequest user_notification = db.session.get(UserNotification, user_notification_id) user = user_notification.user notification = user_notification.notification if user.email: with (SMTP(config.MAIL_SERVER) as mail): match notification.data.get("type"): case "welcome": message = email_send.make_multipart_message( f"[system] Welcome, {user.username}", config.NOTIFICATION_EMAIL, user.email, "welcome", username=user.username ) case "commit": commit = db.session.get(Commit, notification.data.get("commit")) line_separator = "\n\n" # hack so it works in older Pythons newline = "\n" message = email_send.make_multipart_message( f"[commit in {notification.data.get('repo')}] {commit.message.partition(line_separator)[0].replace(newline, ' ')}", config.NOTIFICATION_EMAIL, user.email, "commit", username=user.username, commit=commit, url="https://" if config.suggest_https else "http://" + config.BASE_DOMAIN + "/repo/" + notification.data.get("repo") + "/commit/" + notification.data.get("commit") ) case "post": post = db.session.get(Post, notification.data.get("post")) message = email_send.make_multipart_message( f"[post in {notification.data.get('repo')}] {post.subject}", config.NOTIFICATION_EMAIL, user.email, "forum", username=user.username, post=post, url="https://" if config.suggest_https else "http://" + config.BASE_DOMAIN + "/repo/" + notification.data.get("repo") + "/post/" + notification.data.get("post") ) case "pr": pr = db.session.get(PullRequest, notification.data.get("pr")) message = email_send.make_multipart_message( f"[PR in {notification.data.get('repo')}] {pr.head_route}:{pr.head_branch} -> {pr.base_route}:{pr.base_branch}", config.NOTIFICATION_EMAIL, user.email, "pr", username=user.username, pr=pr, url="https://" if config.suggest_https else "http://" + config.BASE_DOMAIN + notification.data.get("base") + "/prs/" ) mail.sendmail(config.NOTIFICATION_EMAIL, user.email, message) return 0 # notification sent successfully @shared_task(ignore_result=False) def merge_heads(head_route, head_branch, base_route, base_branch, pr_id, simulate=True): from models import Repo, Commit, PullRequest server_repo_location = os.path.join(config.REPOS_PATH, base_route.lstrip("/")) pull_request = db.session.get(PullRequest, pr_id) if not os.path.isdir(server_repo_location): raise FileNotFoundError(f"Repo {server_repo_location} not found, cannot merge.") if base_route == head_route: common.git_command(server_repo_location, b"", "checkout", f"{base_branch}") if simulate: out, err, merge_exit = common.git_command(server_repo_location, b"", "merge", "--no-commit", "--no-ff", f"heads/{head_branch}", return_err=True, return_exit=True) # Undo the merge. common.git_command(server_repo_location, b"", "merge", "--abort") else: out, err, merge_exit = common.git_command(server_repo_location, b"", "merge", f"heads/{head_branch}", return_err=True, return_exit=True) pull_request.state = 1 for resolves in pull_request.resolves: resolves.post.state = 0 db.session.commit() new_commits = common.git_command(server_repo_location, b"", "log", "--oneline", f"heads/{base_branch}..heads/{head_branch}") return "merge_simulator" if simulate else "merge", out, err, head_route, head_branch, base_route, base_branch, merge_exit, new_commits # Otherwise, we need to fetch the head repo. remote_url = "https://" if config.suggest_https else "http://" + os.path.join(config.BASE_DOMAIN + f":{config.port}" if config.port not in {80, 443} else "", "git", head_route.lstrip("/")) out, err = b"", b"" part_out, part_err = common.git_command(server_repo_location, b"", "remote", "add", "NEW", remote_url, return_err=True) out += part_out err += part_err part_out, part_err = common.git_command(server_repo_location, b"", "remote", "update", return_err=True) out += part_out err += part_err part_out, part_err = common.git_command(server_repo_location, b"", "fetch", "NEW", f"{head_branch}", return_err=True) out += part_out err += part_err part_out, part_err = common.git_command(server_repo_location, b"", "checkout", f"{base_branch}", return_err=True) out += part_out err += part_err new_commits, part_err = common.git_command(server_repo_location, b"", "log", "--pretty=format:\"%H\"", f"heads/{base_branch}..NEW/{head_branch}", "--", return_err=True) new_commits = new_commits.decode().splitlines() err += part_err if simulate: part_out, part_err, merge_exit = common.git_command(server_repo_location, b"", "merge", "--allow-unrelated-histories", "--no-commit", "--no-ff", f"NEW/{head_branch}", return_err=True, return_exit=True) else: part_out, part_err, merge_exit = common.git_command(server_repo_location, b"", "merge", "--allow-unrelated-histories", f"NEW/{head_branch}", return_err=True, return_exit=True) diff, diff_exit = common.git_command(server_repo_location, b"", "diff", "--check", return_exit=True) out += part_out err += part_err part_out, part_err = common.git_command(server_repo_location, b"", "remote", "rm", "NEW", return_err=True) out += part_out err += part_err if simulate: # Undo the merge. common.git_command(server_repo_location, b"", "merge", "--abort") else: # Copy the commits rows from the head repo to the base repo for commit in new_commits: commit_data = Commit.query.filter_by(repo_name=head_route, sha=commit).first() db.session.expunge(commit_data) make_transient(commit_data) commit_data.repo_name = base_route commit_data.identifier = f"{base_route}/{commit_data.sha}" commit_data.receive_date = datetime.now() db.session.add(commit_data) # Consider the PR merged. pull_request.state = 1 for resolves in pull_request.resolves: resolves.post.state = 0 db.session.commit() return "merge_simulator" if simulate else "merge", out, err, head_route, head_branch, base_route, base_branch, merge_exit, new_commits @shared_task(ignore_result=False) def copy_site(route): from models import Repo repo = db.session.get(Repo, route) server_repo_location = os.path.join(config.REPOS_PATH, route.lstrip("/")) subdomain = repo.owner.username subpath = repo.name if repo.has_site != 2 else "" site_location = os.path.join(config.SITE_PATH, subdomain, subpath) # Get the branch to be used for the site; if it somehow doesn't exist, use the default branch. branch = repo.site_branch or repo.default_branch # Make a shallow clone of the repo; this prevents getting the full git database when it's not needed. if os.path.isdir(site_location): # Delete the old site. shutil.rmtree(site_location) common.git_command(config.SITE_PATH, b"", "clone", "--depth=1", f"--branch={branch}", os.path.join(os.getcwd(), server_repo_location), os.path.join(subdomain, subpath)) @shared_task(ignore_result=False) def delete_site(route): from models import Repo repo = db.session.get(Repo, route) subdomain = repo.owner.username subpath = repo.name if repo.has_site != 2 else "." site_location = os.path.join(config.SITE_PATH, subdomain, subpath) if os.path.isdir(site_location): shutil.rmtree(site_location) # Redo the primary site. primary_site = Repo.query.filter_by(owner=repo.owner, has_site=2).first() if primary_site: copy_site(primary_site.route) @shared_task(ignore_result=False) def request_email_change(username, email): from models import User, EmailChangeRequest user = db.session.get(User, username) request = EmailChangeRequest(user, email) db.session.add(request) db.session.commit() message = email_send.make_multipart_message( "Email change request for {username}".format(username=username), config.NOTIFICATION_EMAIL, email, "email-change", username=username, code=request.code, new_email=email, url="https://" if config.suggest_https else "http://" + config.BASE_DOMAIN + "/settings/confirm-email/" + request.code ) with (SMTP(config.MAIL_SERVER) as mail): mail.sendmail(config.NOTIFICATION_EMAIL, email, message)