""" Webdist Flask backend """ import logging import subprocess import sqlite3 from datetime import datetime from flask import Flask, render_template # Enable logging logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO) logger = logging.getLogger(__name__) def create_app(config): """ creates and returns the Flask app """ _app = Flask(__name__) _app.config.from_pyfile(config) # initialize moment on the app within create_app() return _app app = create_app("/etc/webdist/main.cfg") def cursor_row_to_dict(cursor, row): """ Converts a sqlite3 cursor row into a dict """ items = {} for idx, key in enumerate(cursor.description): if key[0] in ['buildtime']: items[key[0]] = datetime.fromtimestamp(row[idx]).isoformat() else: items[key[0]] = row[idx] return items def latest(): """ Returns a dict of latest packages """ response = {} for repo in app.config['REPOS']: response[repo] = [] REPO_SOURCES_DB = "file:" + app.config['REPOS'][repo]['path'] + \ f"{app.config['REPOS'][repo]['dbname']}-sources.db?mode=ro" logger.info(REPO_SOURCES_DB) conn = sqlite3.connect(REPO_SOURCES_DB, uri=True) cursor = conn.cursor() cursor.execute( 'SELECT * FROM sources ORDER BY buildtime DESC limit 100') for row in cursor: response[repo].append(cursor_row_to_dict(cursor, row)) return response def query(querystring): """ Performs a query of a package """ logger.info("Query of %s", querystring) response = {} for repo in app.config['REPOS']: response[repo] = [] REPO_SOURCES_DB = "file:" + app.config['REPOS'][repo]['path'] + \ f"{app.config['REPOS'][repo]['dbname']}-sources.db?mode=ro" conn = sqlite3.connect(REPO_SOURCES_DB, uri=True) cursor = conn.cursor() cursor.execute( f'SELECT * FROM sources where name like "%%{querystring}%%" ' 'or summary like "%%{querystring}%%" ' 'ORDER BY name="{querystring}" DESC, name like "{qs}%%" DESC, ' 'name like "%%{querystring}%%" DESC limit 10') for row in cursor: response[repo].append(cursor_row_to_dict(cursor, row)) return response @app.route("/") def _search(): return render_template("index.html") @app.route("/latest") def _latest(): data = latest() return render_template("latest.html", data=data) @app.route("/button/") def _button(name): script = app.config['BUTTONS'][name]['script'] output = subprocess.check_output([script]) return render_template("button.html",output=output.decode()) @app.route("/query/") def _query(querystring): data = query(querystring) return render_template("query.html", data=data) @app.route("/api/v1/latest") def _hello(): data = latest() return {'result': data }