path: root/api/app.py

# encoding: utf-8
import logging
import os
import subprocess
from logging.handlers import RotatingFileHandler
from flask_babel import gettext, Babel
from flask import Flask, render_template, g, send_from_directory, jsonify, safe_join, request, flash, Response
from flask_bootstrap import Bootstrap
from flask_mail import Mail
from flask_migrate import Migrate
from flask_security import Security
from flask_security import signals as FlaskSecuritySignals
from flask_uploads import configure_uploads, UploadSet, AUDIO, patch_request_class
from app_oauth import config_oauth
from flask_cors import CORS, cross_origin
from cachetools import cached, TTLCache
from flasgger import Swagger
from werkzeug.middleware.proxy_fix import ProxyFix
from werkzeug.utils import import_string
import requests
from utils.defaults import Reel2bitsDefaults
import commands
from utils.flake_id import FlakeId
import html
from utils.meta_tags import get_default_head_tags, get_request_head_tags
from sqlalchemy import event
from sqlalchemy.engine import Engine
import time

from models import db, Config, user_datastore, create_actor
from utils.various import InvalidUsage, is_admin, add_user_log, join_url

from little_boxes import activitypub as ap
from activitypub.backend import Reel2BitsBackend

from celery import Celery

from version import VERSION

__VERSION__ = VERSION

AVAILABLE_LOCALES = ["fr", "fr_FR", "en", "en_US", "pl"]

spa_cache = TTLCache(maxsize=100, ttl=900)

try:
    import sentry_sdk
    from sentry_sdk.integrations.flask import FlaskIntegration as SentryFlaskIntegration
    from sentry_sdk.integrations.celery import CeleryIntegration as SentryCeleryIntegration

    print(" * Sentry Flask/Celery support have been loaded")
    HAS_SENTRY = True
except ImportError:
    print(" * No Sentry Flask/Celery support available")
    HAS_SENTRY = False

GIT_VERSION = ""
gitpath = os.path.join(os.getcwd(), "../.git")
if os.path.isdir(gitpath):
    GIT_VERSION = subprocess.check_output(["git", "rev-parse", "--short", "HEAD"])
    if GIT_VERSION:
        GIT_VERSION = GIT_VERSION.strip().decode("UTF-8")


@cached(spa_cache)
def get_spa_html(spa_url):
    if spa_url.startswith("/") or spa_url.startswith("../"):
        print("~~~ SPA: file", spa_url)
        with open(spa_url) as f:
            return f.read()
    print("~~~ SPA: url", spa_url)
    response = requests.get(join_url(spa_url, "index.html"), verify=False)
    response.raise_for_status()
    content = response.text
    return content


def make_celery(remoulade):
    celery = Celery(remoulade.import_name, broker=remoulade.config["CELERY_BROKER_URL"])
    celery.conf.update(remoulade.config)
    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            with remoulade.app_context():
                return TaskBase.__call__(self, *args, **kwargs)

    celery.Task = ContextTask
    return celery  # omnomnom


def create_app(config_filename="config.development.Config", app_name=None, register_blueprints=True):
    # App configuration
    app = Flask(app_name or __name__)
    app_settings = os.getenv("APP_SETTINGS", config_filename)
    print(f" * Loading config: '{app_settings}'")
    try:
        cfg = import_string(app_settings)()
    except ImportError:
        print(" *** Cannot import config ***")
        cfg = import_string("config.config.BaseConfig")
        print(" *** Default config loaded, expect problems ***")
    if hasattr(cfg, "post_load"):
        print(" *** Doing some magic")
        cfg.post_load()
    app.config.from_object(cfg)

    app.wsgi_app = ProxyFix(app.wsgi_app, x_proto=1, x_host=1)

    Bootstrap(app)

    app.jinja_env.add_extension("jinja2.ext.with_")
    app.jinja_env.add_extension("jinja2.ext.do")
    app.jinja_env.globals.update(is_admin=is_admin)

    if HAS_SENTRY:
        sentry_sdk.init(
            app.config["SENTRY_DSN"],
            integrations=[SentryFlaskIntegration(), SentryCeleryIntegration()],
            release=f"{VERSION} ({GIT_VERSION})",
        )
        print(" * Sentry Flask/Celery support activated")
        print(" * Sentry DSN: %s" % app.config["SENTRY_DSN"])

    if app.debug:
        app.jinja_env.auto_reload = True
        logging.basicConfig(level=logging.DEBUG)

    # Logging
    if not app.debug:
        formatter = logging.Formatter("%(asctime)s %(levelname)s: %(message)s " "[in %(pathname)s:%(lineno)d]")
        file_handler = RotatingFileHandler("%s/errors_app.log" % os.getcwd(), "a", 1000000, 1)
        file_handler.setLevel(logging.INFO)
        file_handler.setFormatter(formatter)
        app.logger.addHandler(file_handler)

    dbLogger = logging.getLogger("reel2bits.sqltime")
    dbLogger.setLevel(logging.DEBUG)

    CORS(app, origins=["*"])

    if app.debug:
        logging.getLogger("flask_cors.extension").level = logging.DEBUG

    mail = Mail(app)  # noqa: F841
    migrate = Migrate(app, db)  # noqa: F841 lgtm [py/unused-local-variable]
    babel = Babel(app)  # noqa: F841
    app.babel = babel

    template = {
        "swagger": "2.0",
        "info": {"title": "reel2bits API", "description": "API instance", "version": VERSION},
        "host": app.config["AP_DOMAIN"],
        "basePath": "/",
        "schemes": ["https"],
        "securityDefinitions": {
            "OAuth2": {
                "type": "oauth2",
                "flows": {
                    "authorizationCode": {
                        "authorizationUrl": f"https://{app.config['AP_DOMAIN']}/oauth/authorize",
                        "tokenUrl": f"https://{app.config['AP_DOMAIN']}/oauth/token",
                        "scopes": {
                            "read": "Grants read access",
                            "write": "Grants write access",
                            "admin": "Grants admin operations",
                        },
                    }
                },
            }
        },
        "consumes": ["application/json", "application/jrd+json"],
        "produces": ["application/json", "application/jrd+json"],
    }

    db.init_app(app)

    # ActivityPub backend
    back = Reel2BitsBackend()
    ap.use_backend(back)

    # Oauth
    config_oauth(app)

    # Setup Flask-Security
    security = Security(app, user_datastore)  # noqa: F841 lgtm [py/unused-local-variable]

    @FlaskSecuritySignals.password_reset.connect_via(app)
    @FlaskSecuritySignals.password_changed.connect_via(app)
    def log_password_reset(sender, user):
        if not user:
            return
        add_user_log(user.id, user.id, "user", "info", "Your password has been changed !")

    @FlaskSecuritySignals.reset_password_instructions_sent.connect_via(app)
    def log_reset_password_instr(sender, user, token):
        if not user:
            return
        add_user_log(user.id, user.id, "user", "info", "Password reset instructions sent.")

    @FlaskSecuritySignals.user_registered.connect_via(app)
    def create_actor_for_registered_user(app, user, confirm_token):
        if not user:

            return
        actor = create_actor(user)
        actor.user = user
        actor.user_id = user.id
        db.session.add(actor)
        db.session.commit()

    @security.mail_context_processor
    def mail_ctx_proc():
        _config = Config.query.first()
        if not _config:
            print("ERROR: cannot get instance Config from database")
        instance = {"name": None, "url": None}
        if _config:
            instance["name"] = _config.app_name
        instance["url"] = app.config["REEL2BITS_URL"]
        return dict(instance=instance)

    @babel.localeselector
    def get_locale():
        # if a user is logged in, use the locale from the user settings
        identity = getattr(g, "identity", None)
        if identity is not None and identity.id:
            return identity.user.locale
        # otherwise try to guess the language from the user accept
        # header the browser transmits.  We support fr/en in this
        # example.  The best match wins.
        return request.accept_languages.best_match(AVAILABLE_LOCALES)

    @babel.timezoneselector
    def get_timezone():
        identity = getattr(g, "identity", None)
        if identity is not None and identity.id:
            return identity.user.timezone

    @app.be