Python script, Unicode text, UTF-8 text executable
1
__version__ = "0.3.1"
2
import os
4
import shutil
5
import random
6
import subprocess
7
import platform
8
import PIL
10
import git
11
import mimetypes
12
import magic
13
import flask
14
import cairosvg
15
import celery
16
import shlex
17
from functools import wraps
18
from datetime import datetime
19
from enum import Enum
20
from cairosvg import svg2png
21
from flask_sqlalchemy import SQLAlchemy
22
from flask_bcrypt import Bcrypt
23
from markupsafe import escape, Markup
24
from flask_migrate import Migrate
25
from PIL import Image
26
from flask_httpauth import HTTPBasicAuth
27
import config
28
from common import git_command
29
from flask_babel import Babel, gettext, ngettext, force_locale
30
from jinja2_fragments.flask import render_block
31
import logging
33
class No304(logging.Filter):
36
def filter(self, record):
37
return not record.getMessage().strip().endswith("304 -")
38
logging.getLogger("werkzeug").addFilter(No304())
41
_ = gettext
43
n_ = ngettext
44
app = flask.Flask(__name__)
46
app.config.from_mapping(
47
CELERY=dict(
48
broker_url=config.REDIS_URI,
49
result_backend=config.REDIS_URI,
50
task_ignore_result=True,
51
),
52
)
53
auth = HTTPBasicAuth()
55
app.config["SQLALCHEMY_DATABASE_URI"] = config.DB_URI
57
app.config["SECRET_KEY"] = config.DB_PASSWORD
58
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
59
app.config["BABEL_TRANSLATION_DIRECTORIES"] = "i18n"
60
app.config["MAX_CONTENT_LENGTH"] = config.MAX_PAYLOAD_SIZE
61
app.config["SESSION_COOKIE_SAMESITE"] = "Lax"
62
app.config["SESSION_COOKIE_SECURE"] = config.suggest_https # only send cookies over HTTPS if the server is configured for it
63
app.config["SESSION_COOKIE_HTTPONLY"] = True # don't allow JS to access the cookie
64
if config.restrict_cookie_domain:
65
app.config["SESSION_COOKIE_DOMAIN"] = config.BASE_DOMAIN # don't share across subdomains, since user content is hosted there
66
db = SQLAlchemy(app)
68
bcrypt = Bcrypt(app)
69
migrate = Migrate(app, db)
70
from misc_utils import *
72
import git_http
74
import api
75
import jinja_utils
76
import celery_tasks
77
from celery import Celery, Task
78
import celery_integration
79
import pathlib
80
from models import *
82
babel = Babel(app)
84
def get_locale():
87
if flask.request.cookies.get("language"):
88
return flask.request.cookies.get("language")
89
return flask.request.accept_languages.best_match(config.available_locales)
90
babel.init_app(app, locale_selector=get_locale)
93
with app.app_context():
95
locale_names = {}
96
for language in config.available_locales:
97
with force_locale(language):
98
# NOTE: Translate this to the language's name in that language, for example in French you would use français
99
locale_names[language] = gettext("English")
100
worker = celery_integration.init_celery_app(app)
102
repositories = flask.Blueprint("repository", __name__, template_folder="templates/repository/")
104
app.jinja_env.add_extension("jinja2.ext.do")
106
app.jinja_env.add_extension("jinja2.ext.loopcontrols")
107
app.jinja_env.add_extension("jinja2.ext.debug")
108
@app.context_processor
111
def default():
112
username = flask.session.get("username")
113
user_object = User.query.filter_by(username=username).first()
115
return {
117
"logged_in_user": username,
118
"user_object": user_object,
119
"Notification": Notification,
120
"unread": UserNotification.query.filter_by(user_username=username).filter(
121
UserNotification.attention_level > 0).count(),
122
"config": config,
123
"Markup": Markup,
124
"locale_names": locale_names,
125
"set": set, # since using {} is impossible in Jinja
126
"request": flask.request,
127
}
128
@app.route("/")
131
def main():
132
if flask.session.get("username"):
133
return flask.render_template("home.html")
134
else:
135
return flask.render_template("no-home.html")
136
@app.route("/userstyle")
139
def userstyle():
140
if flask.session.get("username") and os.path.exists(
141
os.path.join(config.REPOS_PATH, flask.session.get("username"), ".config",
142
"theme.css")):
143
return flask.send_from_directory(
144
os.path.join(config.REPOS_PATH, flask.session.get("username"), ".config"),
145
"theme.css")
146
else:
147
return flask.Response("", mimetype="text/css")
148
@app.route("/about/")
151
def about():
152
return flask.render_template("about.html", platform=platform, version=__version__)
153
@app.route("/search")
156
def search():
157
query = flask.request.args.get("q")
158
page_number = flask.request.args.get("page", 1, type=int)
159
if flask.session.get("username"):
160
default_page_length = db.session.get(User, flask.session.get("username")).default_page_length
161
else:
162
default_page_length = 16
163
page_length = flask.request.args.get("per_page", default_page_length, type=int)
164
if not query:
165
flask.abort(400)
166
results = Repo.query.filter(Repo.name.ilike(f"%{query}%")).filter_by(visibility=2).paginate(
168
page=page_number, per_page=page_length)
169
if results.has_next:
171
next_page = results.next_num
172
else:
173
next_page = None
174
if results.has_prev:
176
prev_page = results.prev_num
177
else:
178
prev_page = None
179
return flask.render_template("search.html", results=results, query=query,
181
page_number=page_number,
182
page_length=page_length,
183
next_page=next_page,
184
prev_page=prev_page,
185
num_pages=results.pages)
186
@app.route("/language", methods=["POST"])
189
def set_locale():
190
response = flask.redirect(flask.request.referrer if flask.request.referrer else "/",
191
code=303)
192
if not flask.request.form.get("language"):
193
response.delete_cookie("language")
194
else:
195
response.set_cookie("language", flask.request.form.get("language"))
196
return response
198
@app.route("/cookie-dismiss")
201
def dismiss_banner():
202
response = flask.redirect(flask.request.referrer if flask.request.referrer else "/",
203
code=303)
204
response.set_cookie("cookie-banner", "1")
205
return response
206
@app.route("/help/")
209
def help_redirect():
210
return flask.redirect(config.help_url, code=302)
211
@app.route("/settings/")
214
def settings():
215
if not flask.session.get("username"):
216
flask.abort(401)
217
user = User.query.filter_by(username=flask.session.get("username")).first()
218
return flask.render_template("user-settings.html", user=user)
220
@app.route("/settings/confirm-email/<code>")
223
def confirm_email(code):
224
request = EmailChangeRequest.query.filter_by(code=code).first()
225
if not request:
226
flask.abort(404)
227
user = db.session.get(User, request.user_username)
229
user.email = request.new_email
230
db.session.delete(request)
231
db.session.commit()
232
return flask.redirect("/settings", code=303)
234
@app.route("/settings/profile", methods=["POST"])
237
def settings_profile():
238
user = User.query.filter_by(username=flask.session.get("username")).first()
239
user.display_name = flask.request.form["displayname"]
241
user.URL = flask.request.form["url"]
242
user.company = flask.request.form["company"]
243
user.company_URL = flask.request.form["companyurl"]
244
if not flask.request.form.get("email"):
245
# Deleting the email can be instant; no need to confirm
246
user.email = ""
247
elif flask.request.form.get("email") != user.email:
248
# Changing the email requires confirmation from the address holder
249
celery_tasks.request_email_change.delay(user.username, flask.request.form["email"])
250
user.location = flask.request.form["location"]
251
user.show_mail = True if flask.request.form.get("showmail") else False
252
user.bio = flask.request.form.get("bio")
253
db.session.commit()
255
flask.flash(
257
Markup("<iconify-icon icon='mdi:check'></iconify-icon>" + _("Settings saved")),
258
category="success")
259
return flask.redirect(f"/{flask.session.get('username')}", code=303)
260
@app.route("/settings/preferences", methods=["POST"])
263
def settings_prefs():
264
user = User.query.filter_by(username=flask.session.get("username")).first()
265
user.default_page_length = int(flask.request.form["page_length"])
267
user.max_post_nesting = int(flask.request.form["max_post_nesting"])
268
db.session.commit()
270
flask.flash(
272
Markup("<iconify-icon icon='mdi:check'></iconify-icon>" + _("Settings saved")),
273
category="success")
274
return flask.redirect(f"/{flask.session.get('username')}", code=303)
275
@app.route("/favourites/", methods=["GET", "POST"])
278
def favourites():
279
if not flask.session.get("username"):
280
flask.abort(401)
281
if flask.request.method == "GET":
282
relationships = RepoFavourite.query.filter_by(
283
user_username=flask.session.get("username"))
284
return flask.render_template("favourites.html", favourites=relationships)
286
@app.route("/favourites/<int:id>", methods=["POST"])
289
def favourite_edit(id):
290
if not flask.session.get("username"):
291
flask.abort(401)
292
favourite = db.session.get(RepoFavourite, id)
293
if favourite.user_username != flask.session.get("username"):
294
flask.abort(403)
295
data = flask.request.form
296
# print(data)
297
favourite.notify_commit = js_to_bool(data.get("commit"))
298
favourite.notify_forum = js_to_bool(data.get("forum"))
299
favourite.notify_pr = js_to_bool(data.get("pull_request"))
300
favourite.notify_admin = js_to_bool(data.get("administrative"))
301
print(favourite.notify_commit, favourite.notify_forum, favourite.notify_pr,
302
favourite.notify_admin)
303
db.session.commit()
304
return flask.render_template_string(
305
"""
306
<tr hx-post="/favourites/{{ favourite.id }}" hx-trigger="change" hx-include="#commit-{{ favourite.id }}, #forum-{{ favourite.id }}, #pull_request-{{ favourite.id }}, #administrative-{{ favourite.id }}" hx-headers='{"Content-Type": "application/json"}' hx-swap="outerHTML">
307
<td><a href="{{ favourite.repo.route }}">{{ favourite.repo.owner.username }}/{{ favourite.repo.name }}</a></td>
308
<td style="text-align: center;"><input type="checkbox" name="commit" id="commit-{{ favourite.id }}" value="true" {% if favourite.notify_commit %}checked{% endif %}></td>
309
<td style="text-align: center;"><input type="checkbox" name="forum" id="forum-{{ favourite.id }}" value="true" {% if favourite.notify_forum %}checked{% endif %}></td>
310
<td style="text-align: center;"><input type="checkbox" name="pull_request" id="pull_request-{{ favourite.id }}" value="true" {% if favourite.notify_pr %}checked{% endif %}></td>
311
<td style="text-align: center;"><input type="checkbox" name="administrative" id="administrative-{{ favourite.id }}" value="true" {% if favourite.notify_admin %}checked{% endif %}></td>
312
</tr>
313
""",
314
favourite=favourite
315
)
316
@app.route("/notifications/", methods=["GET", "POST"])
319
def notifications():
320
if not flask.session.get("username"):
321
flask.abort(401)
322
if flask.request.method == "GET":
323
return flask.render_template("notifications.html",
324
notifications=UserNotification.query.filter_by(
325
user_username=flask.session.get("username")
326
).order_by(UserNotification.id.desc()),
327
db=db, Commit=Commit, Post=Post
328
)
329
@app.route("/notifications/<int:notification_id>/read", methods=["POST"])
332
def mark_read(notification_id):
333
if not flask.session.get("username"):
334
flask.abort(401)
335
notification = UserNotification.query.filter_by(id=notification_id).first()
336
if notification.user_username != flask.session.get("username"):
337
flask.abort(403)
338
notification.mark_read()
339
db.session.commit()
340
return flask.render_template_string(
341
"<button hx-post='/notifications/{{ notification.id }}/unread' hx-swap='outerHTML'>Mark as unread</button>",
342
notification=notification), 200
343
@app.route("/notifications/<int:notification_id>/unread", methods=["POST"])
346
def mark_unread(notification_id):
347
if not flask.session.get("username"):
348
flask.abort(401)
349
notification = UserNotification.query.filter_by(id=notification_id).first()
350
if notification.user_username != flask.session.get("username"):
351
flask.abort(403)
352
notification.mark_unread()
353
db.session.commit()
354
return flask.render_template_string(
355
"<button hx-post='/notifications/{{ notification.id }}/read' hx-swap='outerHTML'>Mark as read</button>",
356
notification=notification), 200
357
@app.route("/notifications/mark-all-read", methods=["POST"])
360
def mark_all_read():
361
if not flask.session.get("username"):
362
flask.abort(401)
363
notifications = UserNotification.query.filter_by(
365
user_username=flask.session.get("username"))
366
for notification in notifications:
367
notification.mark_read()
368
db.session.commit()
369
return flask.redirect("/notifications/", code=303)
370
@app.route("/accounts/", methods=["GET", "POST"])
373
def login():
374
if flask.request.method == "GET":
375
return flask.render_template("login.html")
376
else:
377
if "login" in flask.request.form:
378
username = flask.request.form["username"]
379
password = flask.request.form["password"]
380
user = User.query.filter_by(username=username).first()
382
if user and bcrypt.check_password_hash(user.password_hashed, password):
384
flask.session["username"] = user.username
385
flask.flash(
386
Markup("<iconify-icon icon='mdi:account'></iconify-icon>" + _(
387
"Successfully logged in as {username}").format(
388
username=username)),
389
category="success")
390
return flask.redirect("/", code=303)
391
elif not user:
392
flask.flash(Markup(
393
"<iconify-icon icon='mdi:account-question'></iconify-icon>" + _(
394
"User not found")),
395
category="alert")
396
return flask.render_template("login.html")
397
else:
398
flask.flash(Markup(
399
"<iconify-icon icon='mdi:account-question'></iconify-icon>" + _(
400
"Invalid password")),
401
category="error")
402
return flask.render_template("login.html")
403
if "signup" in flask.request.form:
404
username = flask.request.form["username"]
405
password = flask.request.form["password"]
406
password2 = flask.request.form["password2"]
407
email = flask.request.form.get("email")
408
email2 = flask.request.form.get("email2") # repeat email is a honeypot
409
name = flask.request.form.get("name")
410
if not only_chars(username,
412
"0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-"):
413
flask.flash(Markup(
414
_("Usernames may only contain Latin alphabet, numbers and '-'")),
415
category="error")
416
return flask.render_template("login.html")
417
if "--" in username:
418
flask.flash(Markup(
419
_("Usernames may not contain consecutive hyphens")),
420
category="error")
421
return flask.render_template("login.html")
422
if username.startswith("-") or username.endswith("-"):
423
flask.flash(Markup(
424
_("Usernames may not start or end with a hyphen")),
425
category="error")
426
return flask.render_template("login.html")
427
if username in config.RESERVED_NAMES:
428
flask.flash(
429
Markup(
430
_("Sorry, {username} is a system path").format(
431
username=username)),
432
category="error")
433
return flask.render_template("login.html")
434
if not username.islower():
436
if not name: # infer display name from the wanted username if not customised
437
display_name = username
438
username = username.lower()
439
flask.flash(Markup(
440
_("Usernames must be lowercase, so it's been converted automatically")),
441
category="info")
442
user_check = User.query.filter_by(username=username).first()
444
if user_check or email2: # make the honeypot look like a normal error
445
flask.flash(
446
Markup(
447
_(
448
"The username {username} is taken").format(
449
username=username)),
450
category="error")
451
return flask.render_template("login.html")
452
if password2 != password:
454
flask.flash(Markup(_(
455
"Make sure the passwords match")),
456
category="error")
457
return flask.render_template("login.html")
458
user = User(username, password, email, name)
460
db.session.add(user)
461
db.session.commit()
462
flask.session["username"] = user.username
463
flask.flash(Markup(
464
_(
465
"Successfully created and logged in as {username}").format(
466
username=username)),
467
category="success")
468
notification = Notification({"type": "welcome"})
470
db.session.add(notification)
471
db.session.commit()
472
return flask.redirect("/", code=303)
474
@app.route("/newrepo/", methods=["GET", "POST"])
477
def new_repo():
478
if not flask.session.get("username"):
479
flask.abort(401)
480
if flask.request.method == "GET":
481
return flask.render_template("new-repo.html")
482
else:
483
name = flask.request.form["name"]
484
visibility = int(flask.request.form["visibility"])
485
if not only_chars(name,
487
"0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-_"):
488
flask.flash(Markup(
489
"<iconify-icon icon='mdi:error'></iconify-icon>" + _(
490
"Repository names may only contain Latin alphabet, numbers, '-' and '_'")),
491
category="error")
492
return flask.render_template("new-repo.html")
493
user = User.query.filter_by(username=flask.session.get("username")).first()
495
repo = Repo(user, name, visibility)
497
db.session.add(repo)
498
db.session.commit()
499
flask.flash(Markup(_("Successfully created repository {name}").format(name=name)),
501
category="success")
502
return flask.redirect(repo.route, code=303)
503
@app.route("/logout")
506
def logout():
507
flask.session.clear()
508
# print("Logged out")
509
flask.flash(Markup(
510
"<iconify-icon icon='mdi:account'></iconify-icon>" + _("Successfully logged out")),
511
category="info")
512
return flask.redirect("/", code=303)
513
@app.route("/<username>/", methods=["GET", "POST"])
516
def user_profile(username):
517
if db.session.get(User, username) is None:
518
flask.abort(404)
519
old_relationship = UserFollow.query.filter_by(
520
follower_username=flask.session.get("username"),
521
followed_username=username).first()
522
if flask.request.method == "GET":
523
user = User.query.filter_by(username=username).first()
524
match flask.request.args.get("action"):
525
case "repositories":
526
repos = Repo.query.filter_by(owner_name=username, visibility=2)
527
return flask.render_template("user-profile-repositories.html", user=user,
528
repos=repos,
529
relationship=old_relationship)
530
case "followers":
531
return flask.render_template("user-profile-followers.html", user=user,
532
relationship=old_relationship)
533
case "follows":
534
return flask.render_template("user-profile-follows.html", user=user,
535
relationship=old_relationship)
536
case _:
537
return flask.render_template("user-profile-overview.html", user=user,
538
relationship=old_relationship)
539
elif flask.request.method == "POST":
541
match flask.request.args.get("action"):
542
case "follow":
543
if username == flask.session.get("username"):
544
flask.abort(403)
545
if old_relationship:
546
db.session.delete(old_relationship)
547
else:
548
relationship = UserFollow(
549
flask.session.get("username"),
550
username
551
)
552
db.session.add(relationship)
553
db.session.commit()
554
user = db.session.get(User, username)
556
author = db.session.get(User, flask.session.get("username"))
557
notification = Notification({"type": "update", "version": "0.0.0"})
558
db.session.add(notification)
559
db.session.commit()
560
db.session.commit()
562
return flask.redirect("?", code=303)
563
@app.route("/<username>/<repository>/")
566
def repository_index(username, repository):
567
return flask.redirect("./tree", code=302)
568
@app.route("/info/<username>/avatar")
571
def user_avatar(username):
572
server_userdata_location = os.path.join(config.USERDATA_PATH, username)
573
if not os.path.exists(server_userdata_location):
574
return flask.render_template("errors/not-found.html"), 404
575
return flask.send_from_directory(server_userdata_location, "avatar.png")
577
@app.route("/info/<username>/avatar", methods=["POST"])
580
def user_avatar_upload(username):
581
server_userdata_location = os.path.join(config.USERDATA_PATH, username)
582
if not os.path.exists(server_userdata_location):
584
flask.abort(404)
585
if not flask.session.get("username") == username:
586
flask.abort(403)
587
# Convert image to PNG
589
try:
590
image = Image.open(flask.request.files["avatar"])
591
except PIL.UnidentifiedImageError:
592
flask.abort(400)
593
image.save(os.path.join(server_userdata_location, "avatar.png"))
594
return flask.redirect(f"/{username}", code=303)
596
@app.route("/<username>/<repository>/raw/<branch>/<path:subpath>")
599
def repository_raw(username, repository, branch, subpath):
600
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
601
if not os.path.exists(server_repo_location):
602
app.logger.error(f"Cannot load {server_repo_location}")
603
flask.abort(404)
604
if not (get_visibility(username, repository) or get_permission_level(
605
flask.session.get("username"), username,
606
repository) is not None):
607
flask.abort(403)
608
app.logger.info(f"Loading {server_repo_location}")
610
if not os.path.exists(server_repo_location):
612
app.logger.error(f"Cannot load {server_repo_location}")
613
return flask.render_template("errors/not-found.html"), 404
614
repo = git.Repo(server_repo_location)
616
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
617
if not repo_data.default_branch:
618
if repo.heads:
619
repo_data.default_branch = repo.heads[0].name
620
else:
621
return flask.render_template("empty.html",
622
remote=f"http://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
623
if not branch:
624
branch = repo_data.default_branch
625
return flask.redirect(f"./{branch}", code=302)
626
if branch.startswith("tag:"):
628
ref = f"tags/{branch[4:]}"
629
elif branch.startswith("~"):
630
ref = branch[1:]
631
else:
632
ref = f"heads/{branch}"
633
ref = ref.replace("~", "/") # encode slashes for URL support
635
try:
637
repo.git.checkout("-f", ref)
638
except git.exc.GitCommandError:
639
return flask.render_template("errors/not-found.html"), 404
640
return flask.send_from_directory(config.REPOS_PATH,
642
os.path.join(username, repository, subpath))
643
@repositories.route("/<username>/<repository>/tree/", defaults={"branch": None, "subpath": ""})
646
@repositories.route("/<username>/<repository>/tree/<branch>/", defaults={"subpath": ""})
647
@repositories.route("/<username>/<repository>/tree/<branch>/<path:subpath>")
648
def repository_tree(username, repository, branch, subpath):
649
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
650
if not os.path.exists(server_repo_location):
651
app.logger.error(f"Cannot load {server_repo_location}")
652
flask.abort(404)
653
if not (get_visibility(username, repository) or get_permission_level(
654
flask.session.get("username"), username,
655
repository) is not None):
656
flask.abort(403)
657
app.logger.info(f"Loading {server_repo_location}")
659
repo = git.Repo(server_repo_location)
661
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
662
if not repo_data.default_branch:
663
if repo.heads:
664
repo_data.default_branch = repo.heads[0].name
665
else:
666
return flask.render_template("empty.html",
667
remote=f"{config.www_protocol}://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
668
if not branch:
669
branch = repo_data.default_branch
670
return flask.redirect(f"./{branch}", code=302)
671
if branch.startswith("tag:"):
673
ref = f"tags/{branch[4:]}"
674
elif branch.startswith("~"):
675
ref = branch[1:]
676
else:
677
ref = f"heads/{branch}"
678
ref = ref.replace("~", "/") # encode slashes for URL support
680
try:
682
repo.git.checkout("-f", ref)
683
except git.exc.GitCommandError:
684
return flask.render_template("errors/not-found.html"), 404
685
branches = repo.heads
687
all_refs = []
689
for ref in repo.heads:
690
all_refs.append((ref, "head"))
691
for ref in repo.tags:
692
all_refs.append((ref, "tag"))
693
if os.path.isdir(os.path.join(server_repo_location, subpath)):
695
files = []
696
blobs = []
697
for entry in os.listdir(os.path.join(server_repo_location, subpath)):
699
if not os.path.basename(entry) == ".git":
700
files.append(os.path.join(subpath, entry))
701
infos = []
703
for file in files:
705
path = os.path.join(server_repo_location, file)
706
mimetype = guess_mime(path)
707
text = git_command(server_repo_location, None, "log", "--format='%H\n'",
709
shlex.quote(file)).decode()
710
sha = text.split("\n")[0]
712
identifier = f"/{username}/{repository}/{sha}"
713
last_commit = db.session.get(Commit, identifier)
715
info = {
717
"name": os.path.basename(file),
718
"serverPath": path,
719
"relativePath": file,
720
"link": os.path.join(f"/{username}/{repository}/tree/{branch}/", file),
721
"size": human_size(os.path.getsize(path)),
722
"mimetype": f"{mimetype}{f' ({mimetypes.guess_type(path)[1]})' if mimetypes.guess_type(path)[1] else ''}",
723
"commit": last_commit,
724
"shaSize": 7,
725
}
726
special_icon = config.match_icon(os.path.basename(file))
728
if special_icon:
729
info["icon"] = special_icon
730
elif os.path.isdir(path):
731
info["icon"] = config.folder_icon
732
elif mimetypes.guess_type(path)[0] in config.file_icons:
733
info["icon"] = config.file_icons[mimetypes.guess_type(path)[0]]
734
else:
735
info["icon"] = config.unknown_icon
736
if os.path.isdir(path):
738
infos.insert(0, info)
739
else:
740
infos.append(info)
741
return flask.render_template(
743
"repo-tree.html",
744
username=username,
745
repository=repository,
746
files=infos,
747
subpath=os.path.join("/", subpath),
748
branches=all_refs,
749
current=branch,
750
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
751
is_favourite=get_favourite(flask.session.get("username"), username, repository),
752
repo_data=repo_data,
753
)
754
else:
755
path = os.path.join(server_repo_location, subpath)
756
if not os.path.exists(path):
758
return flask.render_template("errors/not-found.html"), 404
759
mimetype = guess_mime(path)
761
mode = mimetype.split("/", 1)[0]
762
size = human_size(os.path.getsize(path))
763
special_icon = config.match_icon(os.path.basename(path))
765
if special_icon:
766
icon = special_icon
767
elif os.path.isdir(path):
768
icon = config.folder_icon
769
elif mimetypes.guess_type(path)[0] in config.file_icons:
770
icon = config.file_icons[mimetypes.guess_type(path)[0]]
771
else:
772
icon = config.unknown_icon
773
contents = None
775
if mode == "text":
776
contents = convert_to_html(path)
777
return flask.render_template(
779
"repo-file.html",
780
username=username,
781
repository=repository,
782
file=os.path.join(f"/{username}/{repository}/raw/{branch}/", subpath),
783
branches=all_refs,
784
current=branch,
785
mode=mode,
786
mimetype=mimetype,
787
detailedtype=magic.from_file(path),
788
size=size,
789
icon=icon,
790
subpath=os.path.join("/", subpath),
791
extension=pathlib.Path(path).suffix,
792
basename=os.path.basename(path),
793
contents=contents,
794
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
795
is_favourite=get_favourite(flask.session.get("username"), username, repository),
796
repo_data=repo_data,
797
)
798
@repositories.route("/<username>/<repository>/commit/<sha>")
801
def repository_commit(username, repository, sha):
802
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
803
if not os.path.exists(server_repo_location):
804
app.logger.error(f"Cannot load {server_repo_location}")
805
flask.abort(404)
806
if not (get_visibility(username, repository) or get_permission_level(
807
flask.session.get("username"), username,
808
repository) is not None):
809
flask.abort(403)
810
app.logger.info(f"Loading {server_repo_location}")
812
if not os.path.exists(server_repo_location):
814
app.logger.error(f"Cannot load {server_repo_location}")
815
return flask.render_template("errors/not-found.html"), 404
816
repo = git.Repo(server_repo_location)
818
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
819
files = git_command(os.path.join(server_repo_location, ".git"), None, "diff-tree", "-r",
821
"--name-only", "--no-commit-id", sha).decode().split("\n")[:-1]
822
# print(files)
824
return flask.render_template(
826
"repo-commit.html",
827
username=username,
828
repository=repository,
829
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
830
is_favourite=get_favourite(flask.session.get("username"), username, repository),
831
diff={file: git_command(os.path.join(server_repo_location, ".git"), None, "diff",
832
str(sha) + "^!", "--", file).decode().split("\n") for
833
file in files},
834
data=db.session.get(Commit, f"/{username}/{repository}/{sha}"),
835
repo_data=repo_data,
836
comment_query=Comment.query,
837
permission_level=get_permission_level(flask.session.get("username"), username, repository),
838
)
839
@repositories.route("/<username>/<repository>/commit/<sha>/add_comment", methods=["POST"])
842
def repository_commit_add_comment(username, repository, sha):
843
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
844
if not os.path.exists(server_repo_location):
845
app.logger.error(f"Cannot load {server_repo_location}")
846
flask.abort(404)
847
if not (get_visibility(username, repository) or get_permission_level(
848
flask.session.get("username"), username,
849
repository) is not None):
850
flask.abort(403)
851
comment = Comment(
853
db.session.get(User, flask.session.get("username")),
854
db.session.get(Repo, f"/{username}/{repository}"),
855
db.session.get(Commit, f"/{username}/{repository}/{sha}"),
856
flask.request.form["comment"],
857
flask.request.form["file"],
858
flask.request.form["line"],
859
)
860
db.session.add(comment)
862
db.session.commit()
863
return flask.redirect(
865
flask.url_for(".repository_commit", username=username, repository=repository, sha=sha),
866
code=303
867
)
868
@repositories.route("/<username>/<repository>/commit/<sha>/delete_comment/<int:id>", methods=["POST"])
871
def repository_commit_delete_comment(username, repository, sha, id):
872
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
873
# print(f"/{username}/{repository}/{flask.request.form.get('id')}")
874
comment = Comment.query.filter_by(identifier=f"/{username}/{repository}/{id}").first()
875
commit = Commit.query.filter_by(identifier=f"/{username}/{repository}/{sha}").first()
876
if (
877
comment.owner.username == flask.session.get("username")
878
or get_permission_level(flask.session.get("username"), username, repository) >= 2
879
or comment.commit.owner.username == flask.session.get("username")
880
):
881
db.session.delete(comment)
882
db.session.commit()
883
return flask.redirect(
885
flask.url_for(".repository_commit", username=username, repository=repository, sha=sha),
886
code=303
887
)
888
@repositories.route("/<username>/<repository>/commit/<sha>/resolve_comment/<int:id>", methods=["POST"])
891
def repository_commit_resolve_comment(username, repository, sha, id):
892
comment = Comment.query.filter_by(identifier=f"/{username}/{repository}/{id}").first()
893
if (
894
comment.commit.owner.username == flask.session.get("username")
895
or get_permission_level(flask.session.get("username"), username, repository) >= 2
896
or comment.owner.username == flask.session.get("username")
897
):
898
comment.state = int(not comment.state)
899
db.session.commit()
900
return flask.redirect(
902
flask.url_for(".repository_commit", username=username, repository=repository, sha=sha),
903
code=303
904
)
905
@repositories.route("/<username>/<repository>/forum/")
908
def repository_forum(username, repository):
909
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
910
if not os.path.exists(server_repo_location):
911
app.logger.error(f"Cannot load {server_repo_location}")
912
flask.abort(404)
913
if not (get_visibility(username, repository) or get_permission_level(
914
flask.session.get("username"), username,
915
repository) is not None):
916
flask.abort(403)
917
app.logger.info(f"Loading {server_repo_location}")
919
if not os.path.exists(server_repo_location):
921
app.logger.error(f"Cannot load {server_repo_location}")
922
return flask.render_template("errors/not-found.html"), 404
923
repo = git.Repo(server_repo_location)
925
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
926
user = User.query.filter_by(username=flask.session.get("username")).first()
927
relationships = RepoAccess.query.filter_by(repo=repo_data)
928
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
929
return flask.render_template(
931
"repo-forum.html",
932
username=username,
933
repository=repository,
934
repo_data=repo_data,
935
relationships=relationships,
936
repo=repo,
937
user_relationship=user_relationship,
938
Post=Post,
939
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
940
is_favourite=get_favourite(flask.session.get("username"), username, repository),
941
default_branch=repo_data.default_branch
942
)
943
@repositories.route("/<username>/<repository>/forum/topic/<int:id>")
946
def repository_forum_topic(username, repository, id):
947
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
948
if not os.path.exists(server_repo_location):
949
app.logger.error(f"Cannot load {server_repo_location}")
950
flask.abort(404)
951
if not (get_visibility(username, repository) or get_permission_level(
952
flask.session.get("username"), username,
953
repository) is not None):
954
flask.abort(403)
955
app.logger.info(f"Loading {server_repo_location}")
957
if not os.path.exists(server_repo_location):
959
app.logger.error(f"Cannot load {server_repo_location}")
960
return flask.render_template("errors/not-found.html"), 404
961
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
963
user = User.query.filter_by(username=flask.session.get("username")).first()
964
relationships = RepoAccess.query.filter_by(repo=repo_data)
965
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
966
post = Post.query.filter_by(id=id).first()
968
return flask.render_template(
970
"repo-topic.html",
971
username=username,
972
repository=repository,
973
repo_data=repo_data,
974
relationships=relationships,
975
user_relationship=user_relationship,
976
post=post,
977
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
978
is_favourite=get_favourite(flask.session.get("username"), username, repository),
979
default_branch=repo_data.default_branch
980
)
981
@repositories.route("/<username>/<repository>/forum/new", methods=["POST", "GET"])
984
def repository_forum_new(username, repository):
985
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
986
if not os.path.exists(server_repo_location):
987
app.logger.error(f"Cannot load {server_repo_location}")
988
flask.abort(404)
989
if not (get_visibility(username, repository) or get_permission_level(
990
flask.session.get("username"), username,
991
repository) is not None):
992
flask.abort(403)
993
app.logger.info(f"Loading {server_repo_location}")
995
if not os.path.exists(server_repo_location):
997
app.logger.error(f"Cannot load {server_repo_location}")
998
return flask.render_template("errors/not-found.html"), 404
999
repo = git.Repo(server_repo_location)
1001
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1002
user = User.query.filter_by(username=flask.session.get("username")).first()
1003
relationships = RepoAccess.query.filter_by(repo=repo_data)
1004
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
1005
post = Post(user, repo_data, None, flask.request.form["subject"],
1007
flask.request.form["message"])
1008
db.session.add(post)
1010
db.session.commit()
1011
return flask.redirect(
1013
flask.url_for(".repository_forum_thread", username=username, repository=repository,
1014
post_id=post.number),
1015
code=303)
1016
@repositories.route("/<username>/<repository>/forum/<int:post_id>")
1019
def repository_forum_thread(username, repository, post_id):
1020
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1021
if not os.path.exists(server_repo_location):
1022
app.logger.error(f"Cannot load {server_repo_location}")
1023
flask.abort(404)
1024
if not (get_visibility(username, repository) or get_permission_level(
1025
flask.session.get("username"), username,
1026
repository) is not None):
1027
flask.abort(403)
1028
app.logger.info(f"Loading {server_repo_location}")
1030
if not os.path.exists(server_repo_location):
1032
app.logger.error(f"Cannot load {server_repo_location}")
1033
return flask.render_template("errors/not-found.html"), 404
1034
repo = git.Repo(server_repo_location)
1036
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1037
user = User.query.filter_by(username=flask.session.get("username")).first()
1038
relationships = RepoAccess.query.filter_by(repo=repo_data)
1039
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
1040
if user:
1042
max_post_nesting = user.max_post_nesting
1043
else:
1044
max_post_nesting = 2
1045
return flask.render_template(
1047
"repo-forum-thread.html",
1048
username=username,
1049
repository=repository,
1050
repo_data=repo_data,
1051
relationships=relationships,
1052
repo=repo,
1053
Post=Post,
1054
user_relationship=user_relationship,
1055
post_id=post_id,
1056
max_post_nesting=max_post_nesting,
1057
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1058
is_favourite=get_favourite(flask.session.get("username"), username, repository),
1059
parent=Post.query.filter_by(repo=repo_data, number=post_id).first(),
1060
has_permission=not ((not get_permission_level(flask.session.get("username"), username,
1061
repository)) and db.session.get(Post,
1062
f"/{username}/{repository}/{post_id}").owner.username != flask.session.get("username")),
1063
)
1064
@repositories.route("/<username>/<repository>/forum/<int:post_id>/change-state",
1067
methods=["POST"])
1068
def repository_forum_change_state(username, repository, post_id):
1069
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1070
if not os.path.exists(server_repo_location):
1071
app.logger.error(f"Cannot load {server_repo_location}")
1072
flask.abort(404)
1073
if (not get_permission_level(flask.session.get("username"), username, repository)) and db.session.get(Post, f"/{username}/{repository}/{post_id}").owner.username != flask.session.get("username"):
1074
flask.abort(403)
1075
app.logger.info(f"Loading {server_repo_location}")
1077
repo = git.Repo(server_repo_location)
1079
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1080
user = User.query.filter_by(username=flask.session.get("username")).first()
1081
relationships = RepoAccess.query.filter_by(repo=repo_data)
1082
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
1083
post = Post.query.filter_by(identifier=f"/{username}/{repository}/{post_id}").first()
1085
if not post:
1087
flask.abort(404)
1088
post.state = int(flask.request.form["new-state"])
1090
db.session.commit()
1092
return flask.redirect(
1094
flask.url_for(".repository_forum_thread", username=username, repository=repository,
1095
post_id=post_id),
1096
code=303)
1097
@repositories.route("/<username>/<repository>/forum/<int:post_id>/reply", methods=["POST"])
1100
def repository_forum_reply(username, repository, post_id):
1101
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1102
if not os.path.exists(server_repo_location):
1103
app.logger.error(f"Cannot load {server_repo_location}")
1104
flask.abort(404)
1105
if not (get_visibility(username, repository) or get_permission_level(
1106
flask.session.get("username"), username,
1107
repository) is not None):
1108
flask.abort(403)
1109
app.logger.info(f"Loading {server_repo_location}")
1111
if not os.path.exists(server_repo_location):
1113
app.logger.error(f"Cannot load {server_repo_location}")
1114
return flask.render_template("errors/not-found.html"), 404
1115
repo = git.Repo(server_repo_location)
1117
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1118
user = User.query.filter_by(username=flask.session.get("username")).first()
1119
relationships = RepoAccess.query.filter_by(repo=repo_data)
1120
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
1121
if not user:
1122
flask.abort(401)
1123
parent = Post.query.filter_by(identifier=f"/{username}/{repository}/{post_id}").first()
1125
post = Post(user, repo_data, parent, flask.request.form["subject"],
1126
flask.request.form["message"])
1127
db.session.add(post)
1129
post.update_date()
1130
db.session.commit()
1131
return flask.redirect(
1133
flask.url_for(".repository_forum_thread", username=username, repository=repository,
1134
post_id=post_id),
1135
code=303)
1136
@repositories.route("/<username>/<repository>/forum/<int:post_id>/voteup",
1139
defaults={"score": 1})
1140
@repositories.route("/<username>/<repository>/forum/<int:post_id>/votedown",
1141
defaults={"score": -1})
1142
@repositories.route("/<username>/<repository>/forum/<int:post_id>/votes", defaults={"score": 0})
1143
def repository_forum_vote(username, repository, post_id, score):
1144
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1145
if not os.path.exists(server_repo_location):
1146
app.logger.error(f"Cannot load {server_repo_location}")
1147
flask.abort(404)
1148
if not (get_visibility(username, repository) or get_permission_level(
1149
flask.session.get("username"), username,
1150
repository) is not None):
1151
flask.abort(403)
1152
app.logger.info(f"Loading {server_repo_location}")
1154
if not os.path.exists(server_repo_location):
1156
app.logger.error(f"Cannot load {server_repo_location}")
1157
return flask.render_template("errors/not-found.html"), 404
1158
repo = git.Repo(server_repo_location)
1160
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1161
user = User.query.filter_by(username=flask.session.get("username")).first()
1162
relationships = RepoAccess.query.filter_by(repo=repo_data)
1163
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
1164
if not user:
1165
flask.abort(401)
1166
post = Post.query.filter_by(identifier=f"/{username}/{repository}/{post_id}").first()
1168
if score:
1170
old_relationship = PostVote.query.filter_by(user_username=user.username,
1171
post_identifier=post.identifier).first()
1172
if old_relationship:
1173
if score == old_relationship.vote_score:
1174
db.session.delete(old_relationship)
1175
post.vote_sum -= old_relationship.vote_score
1176
else:
1177
post.vote_sum -= old_relationship.vote_score
1178
post.vote_sum += score
1179
old_relationship.vote_score = score
1180
else:
1181
relationship = PostVote(user, post, score)
1182
post.vote_sum += score
1183
db.session.add(relationship)
1184
db.session.commit()
1186
user_vote = PostVote.query.filter_by(user_username=user.username,
1188
post_identifier=post.identifier).first()
1189
response = flask.make_response(
1190
str(post.vote_sum) + " " + str(user_vote.vote_score if user_vote else 0))
1191
response.content_type = "text/plain"
1192
return response
1194
@repositories.route("/<username>/<repository>/favourite")
1197
def repository_favourite(username, repository):
1198
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1199
if not os.path.exists(server_repo_location):
1200
app.logger.error(f"Cannot load {server_repo_location}")
1201
flask.abort(404)
1202
if not (get_visibility(username, repository) or get_permission_level(
1203
flask.session.get("username"), username,
1204
repository) is not None):
1205
flask.abort(403)
1206
app.logger.info(f"Loading {server_repo_location}")
1208
if not os.path.exists(server_repo_location):
1210
app.logger.error(f"Cannot load {server_repo_location}")
1211
return flask.render_template("errors/not-found.html"), 404
1212
repo = git.Repo(server_repo_location)
1214
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1215
user = User.query.filter_by(username=flask.session.get("username")).first()
1216
relationships = RepoAccess.query.filter_by(repo=repo_data)
1217
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
1218
if not user:
1219
flask.abort(401)
1220
old_relationship = RepoFavourite.query.filter_by(user_username=user.username,
1222
repo_route=repo_data.route).first()
1223
if old_relationship:
1224
db.session.delete(old_relationship)
1225
else:
1226
relationship = RepoFavourite(user, repo_data)
1227
db.session.add(relationship)
1228
db.session.commit()
1230
return flask.redirect(flask.url_for("favourites"), code=303)
1232
@repositories.route("/<username>/<repository>/users/", methods=["GET", "POST"])
1235
def repository_users(username, repository):
1236
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1237
if not os.path.exists(server_repo_location):
1238
app.logger.error(f"Cannot load {server_repo_location}")
1239
flask.abort(404)
1240
if not (get_visibility(username, repository) or get_permission_level(
1241
flask.session.get("username"), username,
1242
repository) is not None):
1243
flask.abort(403)
1244
app.logger.info(f"Loading {server_repo_location}")
1246
if not os.path.exists(server_repo_location):
1248
app.logger.error(f"Cannot load {server_repo_location}")
1249
return flask.render_template("errors/not-found.html"), 404
1250
repo = git.Repo(server_repo_location)
1252
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1253
user = User.query.filter_by(username=flask.session.get("username")).first()
1254
relationships = RepoAccess.query.filter_by(repo=repo_data)
1255
user_relationship = RepoAccess.query.filter_by(repo=repo_data, user=user).first()
1256
if flask.request.method == "GET":
1258
return flask.render_template(
1259
"repo-users.html",
1260
username=username,
1261
repository=repository,
1262
repo_data=repo_data,
1263
relationships=relationships,
1264
repo=repo,
1265
user_relationship=user_relationship,
1266
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1267
is_favourite=get_favourite(flask.session.get("username"), username, repository)
1268
)
1269
else:
1270
if get_permission_level(flask.session.get("username"), username, repository) != 2:
1271
flask.abort(401)
1272
if flask.request.form.get("new-username"):
1274
# Create new relationship
1275
new_user = User.query.filter_by(
1276
username=flask.request.form.get("new-username")).first()
1277
relationship = RepoAccess(new_user, repo_data, flask.request.form.get("new-level"))
1278
db.session.add(relationship)
1279
db.session.commit()
1280
if flask.request.form.get("update-username"):
1281
# Create new relationship
1282
updated_user = User.query.filter_by(
1283
username=flask.request.form.get("update-username")).first()
1284
relationship = RepoAccess.query.filter_by(repo=repo_data, user=updated_user).first()
1285
if flask.request.form.get("update-level") == -1:
1286
relationship.delete()
1287
else:
1288
relationship.access_level = flask.request.form.get("update-level")
1289
db.session.commit()
1290
return flask.redirect(
1292
app.url_for(".repository_users", username=username, repository=repository))
1293
@repositories.route("/<username>/<repository>/branches/")
1296
def repository_branches(username, repository):
1297
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1298
if not os.path.exists(server_repo_location):
1299
app.logger.error(f"Cannot load {server_repo_location}")
1300
flask.abort(404)
1301
if not (get_visibility(username, repository) or get_permission_level(
1302
flask.session.get("username"), username,
1303
repository) is not None):
1304
flask.abort(403)
1305
app.logger.info(f"Loading {server_repo_location}")
1307
if not os.path.exists(server_repo_location):
1309
app.logger.error(f"Cannot load {server_repo_location}")
1310
return flask.render_template("errors/not-found.html"), 404
1311
repo = git.Repo(server_repo_location)
1313
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1314
return flask.render_template(
1316
"repo-branches.html",
1317
username=username,
1318
repository=repository,
1319
repo_data=repo_data,
1320
repo=repo,
1321
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1322
is_favourite=get_favourite(flask.session.get("username"), username, repository)
1323
)
1324
@repositories.route("/<username>/<repository>/log/", defaults={"branch": None})
1327
@repositories.route("/<username>/<repository>/log/<branch>/")
1328
def repository_log(username, repository, branch):
1329
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1330
if not os.path.exists(server_repo_location):
1331
app.logger.error(f"Cannot load {server_repo_location}")
1332
flask.abort(404)
1333
if not (get_visibility(username, repository) or get_permission_level(
1334
flask.session.get("username"), username,
1335
repository) is not None):
1336
flask.abort(403)
1337
app.logger.info(f"Loading {server_repo_location}")
1339
if not os.path.exists(server_repo_location):
1341
app.logger.error(f"Cannot load {server_repo_location}")
1342
return flask.render_template("errors/not-found.html"), 404
1343
repo = git.Repo(server_repo_location)
1345
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1346
if not repo_data.default_branch:
1347
if repo.heads:
1348
repo_data.default_branch = repo.heads[0].name
1349
else:
1350
return flask.render_template("empty.html",
1351
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}"), 200
1352
if not branch:
1353
branch = repo_data.default_branch
1354
return flask.redirect(f"./{branch}", code=302)
1355
if branch.startswith("tag:"):
1357
ref = f"tags/{branch[4:]}"
1358
elif branch.startswith("~"):
1359
ref = branch[1:]
1360
else:
1361
ref = f"heads/{branch}"
1362
ref = ref.replace("~", "/") # encode slashes for URL support
1364
try:
1366
repo.git.checkout("-f", ref)
1367
except git.exc.GitCommandError:
1368
return flask.render_template("errors/not-found.html"), 404
1369
branches = repo.heads
1371
all_refs = []
1373
for ref in repo.heads:
1374
all_refs.append((ref, "head"))
1375
for ref in repo.tags:
1376
all_refs.append((ref, "tag"))
1377
commit_list = [f"/{username}/{repository}/{sha}" for sha in
1379
git_command(server_repo_location, None, "log",
1380
"--format='%H'").decode().split("\n")]
1381
commits = Commit.query.filter(Commit.identifier.in_(commit_list)).order_by(Commit.author_date.desc())
1383
page_number = flask.request.args.get("page", 1, type=int)
1384
if flask.session.get("username"):
1385
default_page_length = db.session.get(User, flask.session.get("username")).default_page_length
1386
else:
1387
default_page_length = 16
1388
page_length = flask.request.args.get("per_page", default_page_length, type=int)
1389
page_listing = db.paginate(commits, page=page_number, per_page=page_length)
1390
if page_listing.has_next:
1392
next_page = page_listing.next_num
1393
else:
1394
next_page = None
1395
if page_listing.has_prev:
1397
prev_page = page_listing.prev_num
1398
else:
1399
prev_page = None
1400
return flask.render_template(
1402
"repo-log.html",
1403
username=username,
1404
repository=repository,
1405
branches=all_refs,
1406
current=branch,
1407
repo_data=repo_data,
1408
repo=repo,
1409
commits=page_listing,
1410
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1411
is_favourite=get_favourite(flask.session.get("username"), username, repository),
1412
page_number=page_number,
1413
page_length=page_length,
1414
next_page=next_page,
1415
prev_page=prev_page,
1416
num_pages=page_listing.pages
1417
)
1418
@repositories.route("/<username>/<repository>/prs/", methods=["GET", "POST"])
1421
def repository_prs(username, repository):
1422
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1423
if not os.path.exists(server_repo_location):
1424
app.logger.error(f"Cannot load {server_repo_location}")
1425
flask.abort(404)
1426
if not (get_visibility(username, repository) or get_permission_level(
1427
flask.session.get("username"), username,
1428
repository) is not None):
1429
flask.abort(403)
1430
app.logger.info(f"Loading {server_repo_location}")
1432
if not os.path.exists(server_repo_location):
1434
app.logger.error(f"Cannot load {server_repo_location}")
1435
return flask.render_template("errors/not-found.html"), 404
1436
if flask.request.method == "GET":
1438
repo = git.Repo(server_repo_location)
1439
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1440
user = User.query.filter_by(username=flask.session.get("username")).first()
1441
return flask.render_template(
1443
"repo-prs.html",
1444
username=username,
1445
repository=repository,
1446
repo_data=repo_data,
1447
repo=repo,
1448
PullRequest=PullRequest,
1449
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1450
is_favourite=get_favourite(flask.session.get("username"), username, repository),
1451
default_branch=repo_data.default_branch,
1452
branches=repo.branches
1453
)
1454
else:
1456
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1457
head = flask.request.form.get("head")
1458
head_route = flask.request.form.get("headroute")
1459
base = flask.request.form.get("base")
1460
if not head and base and head_route:
1462
return flask.redirect(".", 400)
1463
head_repo = git.Repo(os.path.join(config.REPOS_PATH, head_route.lstrip("/")))
1465
base_repo = git.Repo(server_repo_location)
1466
# print(head_repo)
1467
if head not in head_repo.branches or base not in base_repo.branches:
1469
flask.flash(Markup(
1470
"<iconify-icon icon='mdi:error'></iconify-icon>" + _("Bad branch name")),
1471
category="error")
1472
return flask.redirect(".", 303)
1473
head_data = db.session.get(Repo, head_route)
1475
if not head_data.visibility:
1476
flask.flash(Markup(
1477
"<iconify-icon icon='mdi:error'></iconify-icon>" + _(
1478
"Head can't be restricted")),
1479
category="error")
1480
return flask.redirect(".", 303)
1481
pull_request = PullRequest(head_data, head, repo_data, base,
1483
db.session.get(User, flask.session["username"]))
1484
db.session.add(pull_request)
1486
db.session.commit()
1487
return flask.redirect(".", 303)
1489
@repositories.route("/<username>/<repository>/prs/merge", methods=["POST"])
1492
def repository_prs_merge(username, repository):
1493
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1494
if not os.path.exists(server_repo_location):
1495
app.logger.error(f"Cannot load {server_repo_location}")
1496
flask.abort(404)
1497
if not (get_visibility(username, repository) or get_permission_level(
1498
flask.session.get("username"), username,
1499
repository) is not None):
1500
flask.abort(403)
1501
if not get_permission_level(flask.session.get("username"), username, repository):
1503
flask.abort(401)
1504
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1506
repo = git.Repo(server_repo_location)
1507
id = flask.request.form.get("id")
1508
pull_request = db.session.get(PullRequest, id)
1510
if pull_request:
1512
result = celery_tasks.merge_heads.delay(
1513
pull_request.head_route,
1514
pull_request.head_branch,
1515
pull_request.base_route,
1516
pull_request.base_branch,
1517
simulate=True
1518
)
1519
task_result = worker.AsyncResult(result.id)
1520
return flask.redirect(f"/task/{result.id}?pr-id={id}", 303) # should be 202 Accepted but we must use a redirect
1522
# db.session.delete(pull_request)
1523
# db.session.commit()
1524
else:
1525
flask.abort(400)
1526
@repositories.route("/<username>/<repository>/prs/<int:id>/merge")
1529
def repository_prs_merge_stage_two(username, repository, id):
1530
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1531
if not os.path.exists(server_repo_location):
1532
app.logger.error(f"Cannot load {server_repo_location}")
1533
flask.abort(404)
1534
if not (get_visibility(username, repository) or get_permission_level(
1535
flask.session.get("username"), username,
1536
repository) is not None):
1537
flask.abort(403)
1538
if not get_permission_level(flask.session.get("username"), username, repository):
1540
flask.abort(401)
1541
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1543
repo = git.Repo(server_repo_location)
1544
pull_request = db.session.get(PullRequest, id)
1546
if pull_request:
1548
result = celery_tasks.merge_heads.delay(
1549
pull_request.head_route,
1550
pull_request.head_branch,
1551
pull_request.base_route,
1552
pull_request.base_branch,
1553
simulate=False
1554
)
1555
task_result = worker.AsyncResult(result.id)
1556
pull_request.state = 1
1558
db.session.commit()
1559
return flask.redirect(f"/task/{result.id}?pr-id={id}", 303)
1561
# db.session.delete(pull_request)
1562
else:
1563
flask.abort(400)
1564
@app.route("/task/<task_id>")
1567
def task_monitor(task_id):
1568
task_result = worker.AsyncResult(task_id)
1569
if flask.request.args.get("partial"):
1571
# htmx partial update
1572
return render_block("task-monitor.html", "content", result=task_result, query_string=flask.request.query_string.decode(), delay=1000)
1573
# Since most tasks finish rather quickly, the initial delay is faster, so it doesn't wait for too long
1575
return flask.render_template("task-monitor.html", result=task_result, query_string=flask.request.query_string.decode(), delay=125)
1576
@repositories.route("/<username>/<repository>/prs/delete", methods=["POST"])
1579
def repository_prs_delete(username, repository):
1580
server_repo_location = os.path.join(config.REPOS_PATH, username, repository)
1581
if not os.path.exists(server_repo_location):
1582
app.logger.error(f"Cannot load {server_repo_location}")
1583
flask.abort(404)
1584
if not (get_visibility(username, repository) or get_permission_level(
1585
flask.session.get("username"), username,
1586
repository) is not None):
1587
flask.abort(403)
1588
if not get_permission_level(flask.session.get("username"), username, repository):
1590
flask.abort(401)
1591
repo_data = Repo.query.filter_by(route=f"/{username}/{repository}").first()
1593
repo = git.Repo(server_repo_location)
1594
id = flask.request.form.get("id")
1595
pull_request = db.session.get(PullRequest, id)
1597
if pull_request:
1599
pull_request.state = 2
1600
db.session.commit()
1601
return flask.redirect(".", 303)
1603
@repositories.route("/<username>/<repository>/settings/")
1606
def repository_settings(username, repository):
1607
if get_permission_level(flask.session.get("username"), username, repository) != 2:
1608
flask.abort(401)
1609
repo = git.Repo(os.path.join(config.REPOS_PATH, username, repository))
1611
site_link = Markup(f"<code>http{'s' if config.suggest_https else ''}://{username}.{config.BASE_DOMAIN}/{repository}</code>")
1613
primary_site_link = Markup(f"<code>http{'s' if config.suggest_https else ''}://{username}.{config.BASE_DOMAIN}/</code>")
1614
return flask.render_template("repo-settings.html", username=username, repository=repository,
1616
repo_data=db.session.get(Repo, f"/{username}/{repository}"),
1617
branches=[branch.name for branch in repo.branches],
1618
site_link=site_link, primary_site_link=primary_site_link,
1619
remote=f"http{'s' if config.suggest_https else ''}://{config.BASE_DOMAIN}/git/{username}/{repository}",
1620
is_favourite=get_favourite(flask.session.get("username"), username, repository),
1621
)
1622
@repositories.route("/<username>/<repository>/settings/", methods=["POST"])
1625
def repository_settings_post(username, repository):
1626
if get_permission_level(flask.session.get("username"), username, repository) != 2:
1627
flask.abort(401)
1628
repo = db.session.get(Repo, f"/{username}/{repository}")
1630
repo.visibility = flask.request.form.get("visibility", type=int)
1632
repo.info = flask.request.form.get("description")
1633
repo.default_branch = flask.request.form.get("default_branch")
1634
repo.url = flask.request.form.get("url")
1635
# Update site settings
1637
had_site = repo.has_site
1638
old_branch = repo.site_branch
1639
if flask.request.form.get("site_branch"):
1640
repo.site_branch = flask.request.form.get("site_branch")
1641
if flask.request.form.get("primary_site"):
1642
if had_site != 2:
1643
# Remove primary site from other repos
1644
for other_repo in Repo.query.filter_by(owner=repo.owner, has_site=2):
1645
other_repo.has_site = 1 # switch it to a regular site
1646
flask.flash(Markup(
1647
_("Your repository {repository} has been demoted from a primary site to a regular site because there can only be one primary site per user.").format(
1648
repository=other_repo.route
1649
)), category="warning")
1650
repo.has_site = 2
1651
else:
1652
repo.has_site = 1
1653
else:
1654
repo.site_branch = None
1655
repo.has_site = 0
1656
db.session.commit()
1658
if not (had_site, old_branch) == (repo.has_site, repo.site_branch):
1660
# Deploy the newly activated site
1661
result = celery_tasks.copy_site.delay(repo.route)
1662
if had_site and not repo.has_site:
1664
# Remove the site
1665
result = celery_tasks.delete_site.delay(repo.route)
1666
if repo.has_site == 2 or (had_site == 2 and had_site != repo.has_site):
1668
# Deploy all other sites which were destroyed by the primary site
1669
for other_repo in Repo.query.filter_by(owner=repo.owner, has_site=1):
1670
result = celery_tasks.copy_site.delay(other_repo.route)
1671
return flask.redirect(f"/{username}/{repository}/settings", 303)
1673
@app.errorhandler(404)
1676
def e404(error):
1677
return flask.render_template("errors/not-found.html"), 404
1678
@app.errorhandler(401)
1681
def e401(error):
1682
return flask.render_template("errors/unauthorised.html"), 401
1683
@app.errorhandler(403)
1686
def e403(error):
1687
return flask.render_template("errors/forbidden.html"), 403
1688
@app.errorhandler(418)
1691
def e418(error):
1692
return flask.render_template("errors/teapot.html"), 418
1693
@app.errorhandler(405)
1696
def e405(error):
1697
return flask.render_template("errors/method-not-allowed.html"), 405
1698
@app.errorhandler(500)
1701
def e500(error):
1702
return flask.render_template("errors/server-error.html"), 500
1703
@app.errorhandler(400)
1706
def e400(error):
1707
return flask.render_template("errors/bad-request.html"), 400
1708
@app.errorhandler(410)
1711
def e410(error):
1712
return flask.render_template("errors/gone.html"), 410
1713
@app.errorhandler(415)
1716
def e415(error):
1717
return flask.render_template("errors/media-type.html"), 415
1718
if __name__ == "__main__":
1721
app.run(debug=True, port=8080, host="0.0.0.0")
1722
app.register_blueprint(repositories)
1724