Python script, ASCII text executable
1
import common
2
import time
3
import os
4
import config
5
import email_send
6
import shutil
7
from celery import shared_task
8
from app import db
9
from smtplib import SMTP
10
from celery.utils.log import get_task_logger
11
@shared_task(ignore_result=False)
14
def send_notification(user_notification_id):
15
from models import UserNotification
16
user_notification = db.session.get(UserNotification, user_notification_id)
17
user = user_notification.user
18
notification = user_notification.notification
19
if user.email:
21
with (SMTP(config.MAIL_SERVER) as mail):
22
if notification.data.get("type") == "welcome":
23
message = email_send.make_multipart_message(
24
f"Welcome, {user.username}",
25
config.NOTIFICATION_EMAIL,
26
user.email,
27
"welcome",
28
username=user.username)
29
mail.sendmail(config.NOTIFICATION_EMAIL, user.email, message)
31
return 0 # notification sent successfully
33
@shared_task(ignore_result=False)
36
def merge_heads(head_route, head_branch, base_route, base_branch, simulate=True):
37
from models import Repo
38
server_repo_location = os.path.join(config.REPOS_PATH, base_route.lstrip("/"))
39
if not os.path.isdir(server_repo_location):
40
raise FileNotFoundError(f"Repo {server_repo_location} not found, cannot merge.")
41
if base_route == head_route:
43
common.git_command(server_repo_location, b"", "checkout", f"{base_branch}")
44
if simulate:
45
out, err, merge_exit = common.git_command(server_repo_location, b"", "merge", "--no-commit", "--no-ff", f"heads/{head_branch}",
46
return_err=True, return_exit=True)
47
# Undo the merge.
49
common.git_command(server_repo_location, b"", "merge", "--abort")
50
else:
51
out, err, merge_exit = common.git_command(server_repo_location, b"", "merge", f"heads/{head_branch}",
52
return_err=True, return_exit=True)
53
return "merge_simulator" if simulate else "merge", out, err, head_route, head_branch, base_route, base_branch, merge_exit
55
# Otherwise, we need to fetch the head repo.
57
remote_url = os.path.join(config.BASE_DOMAIN, "git", base_route.lstrip("/"))
58
out, err = b"", b""
60
part_out, part_err = common.git_command(server_repo_location, b"", "remote", "add", "NEW", remote_url, return_err=True)
61
out += part_out
62
err += part_err
63
part_out, part_err = common.git_command(server_repo_location, b"", "remote", "update", return_err=True)
64
out += part_out
65
err += part_err
66
part_out, part_err = common.git_command(server_repo_location, b"", "checkout", f"{base_branch}", return_err=True)
67
out += part_out
68
err += part_err
69
if simulate:
70
part_out, part_err, merge_exit = common.git_command(server_repo_location, b"", "merge", "--allow-unrelated-histories",
71
"--no-commit", "--no-ff", f"NEW/{head_branch}", return_err=True, return_exit=True)
72
else:
73
part_out, part_err, merge_exit = common.git_command(server_repo_location, b"", "merge", "--allow-unrelated-histories",
74
f"NEW/{head_branch}", return_err=True, return_exit=True)
75
diff, diff_exit = common.git_command(server_repo_location, b"", "diff", "--check", return_exit=True)
77
out += part_out
79
err += part_err
80
part_out, part_err = common.git_command(server_repo_location, b"", "remote", "rm", "NEW", return_err=True)
81
out += part_out
82
err += part_err
83
if simulate:
84
# Undo the merge.
85
common.git_command(server_repo_location, b"", "merge", "--abort")
86
return "merge_simulator" if simulate else "merge", out, err, head_route, head_branch, base_route, base_branch, merge_exit
88
@shared_task(ignore_result=False)
91
def copy_site(route):
92
from models import Repo
93
repo = db.session.get(Repo, route)
94
server_repo_location = os.path.join(config.REPOS_PATH, route.lstrip("/"))
95
if repo.has_site == 2:
96
subdomain = repo.owner.username
97
else:
98
subdomain = repo.name + "." + repo.owner.username
99
site_location = os.path.join(config.SITE_PATH, subdomain)
100
# Get the branch to be used for the site; if it somehow doesn't exist, use the default branch.
101
branch = repo.site_branch or repo.default_branch
102
# Make a shallow clone of the repo; this prevents getting the full git database when it's not needed.
103
if os.path.isdir(site_location):
104
# Delete the old site.
105
shutil.rmtree(site_location)
106
common.git_command(config.SITE_PATH, b"", "clone", "--depth=1", f"--branch={branch}", os.path.join(os.getcwd(), server_repo_location), subdomain)
108
@shared_task(ignore_result=False)
111
def delete_site(route):
112
from models import Repo
113
repo = db.session.get(Repo, route)
114
if repo.has_site == 2:
115
subdomain = repo.owner.username
116
else:
117
subdomain = repo.name + "." + repo.owner.username
118
site_location = os.path.join(config.SITE_PATH, subdomain)
119
if os.path.isdir(site_location):
120
shutil.rmtree(site_location)
121