import json from datetime import datetime from typing import Optional from flask import render_template, url_for, flash, redirect, Blueprint, Response from flask.typing import ResponseReturnValue from flask_wtf import FlaskForm from sqlalchemy import exc from wtforms import SelectField, StringField, SubmitField from wtforms.validators import DataRequired from app.extensions import db from app.lists.bc2 import mirror_sites from app.lists.bridgelines import bridgelines from app.lists.mirror_mapping import mirror_mapping from app.models.base import MirrorList from app.portal.util import response_404, view_lifecycle bp = Blueprint("list", __name__) @bp.app_template_filter("provider_name") def list_provider_name(s: str) -> str: return MirrorList.providers_supported.get(s, "Unknown") @bp.app_template_filter("format_name") def list_format_name(s: str) -> str: return MirrorList.formats_supported.get(s, "Unknown") @bp.app_template_filter("list_encoding_name") def list_encoding_name(s: str) -> str: return MirrorList.encodings_supported.get(s, "Unknown") @bp.route('/list') def list_list() -> ResponseReturnValue: lists = MirrorList.query.filter(MirrorList.destroyed.is_(None)).all() return render_template("list.html.j2", section="list", title="Mirror Lists", item="mirror list", new_link=url_for("portal.list.list_new"), items=lists, extra_buttons=[ { "link": url_for("portal.list.list_preview", format_=k), "text": f"Preview {v}", "style": "secondary" } for k, v in MirrorList.formats_supported.items() ] ) @bp.route('/preview/') def list_preview(format_: str) -> ResponseReturnValue: if format_ == "bca": return Response(json.dumps(mirror_mapping()), content_type="application/json") if format_ == "bc2": return Response(json.dumps(mirror_sites()), content_type="application/json") if format_ == "bridgelines": return Response(json.dumps(bridgelines()), content_type="application/json") return response_404(message="Format not found") @bp.route("/destroy/", methods=['GET', 'POST']) def list_destroy(list_id: int) -> ResponseReturnValue: list_ = MirrorList.query.filter(MirrorList.id == list_id, MirrorList.destroyed.is_(None)).first() if list_ is None: return response_404("The requested bridge configuration could not be found.") return view_lifecycle( header="Destroy mirror list?", message=list_.description, success_view="portal.list.list_list", success_message="This list will no longer be updated and may be deleted depending on the provider.", section="list", resource=list_, action="destroy" ) @bp.route("/new", methods=['GET', 'POST']) @bp.route("/new/", methods=['GET', 'POST']) def list_new(group_id: Optional[int] = None) -> ResponseReturnValue: form = NewMirrorListForm() form.provider.choices = list(MirrorList.providers_supported.items()) form.format.choices = list(MirrorList.formats_supported.items()) form.encoding.choices = list(MirrorList.encodings_supported.items()) if form.validate_on_submit(): list_ = MirrorList() list_.provider = form.provider.data list_.format = form.format.data list_.encoding = form.encoding.data list_.description = form.description.data list_.container = form.container.data list_.branch = form.branch.data list_.role = form.role.data list_.filename = form.filename.data list_.created = datetime.utcnow() list_.updated = datetime.utcnow() try: db.session.add(list_) db.session.commit() flash("Created new mirror list.", "success") return redirect(url_for("portal.list.list_list")) except exc.SQLAlchemyError as e: print(e) flash("Failed to create new mirror list.", "danger") return redirect(url_for("portal.list.list_list")) if group_id: form.group.data = group_id return render_template("new.html.j2", section="list", form=form) class NewMirrorListForm(FlaskForm): # type: ignore provider = SelectField('Provider', validators=[DataRequired()]) format = SelectField('Distribution Method', validators=[DataRequired()]) encoding = SelectField('Encoding', validators=[DataRequired()]) description = StringField('Description', validators=[DataRequired()]) container = StringField('Container', validators=[DataRequired()], description="GitHub Project, GitLab Project or AWS S3 bucket name.") branch = StringField('Git Branch/AWS Region', validators=[DataRequired()], description="For GitHub/GitLab, set this to the desired branch name, e.g. main. For AWS S3, " "set this field to the desired region, e.g. us-east-1.") role = StringField('Role ARN', description="(Optional) ARN for IAM role to assume for interaction with the S3 bucket.") filename = StringField('Filename', validators=[DataRequired()]) submit = SubmitField('Save Changes')