119 lines
3.6 KiB
Python
119 lines
3.6 KiB
Python
from flask import Flask, request, jsonify
|
|
from flask_httpauth import HTTPBasicAuth
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
import json
|
|
import sqlite3
|
|
import os
|
|
import subprocess
|
|
|
|
DB_PATH = os.path.join(os.path.dirname(__file__), 'links.db')
|
|
|
|
app = Flask(__name__)
|
|
auth = HTTPBasicAuth('CustomBasic')
|
|
|
|
users = {
|
|
"admin": generate_password_hash("test"),
|
|
}
|
|
|
|
def get_data_from_query(query):
|
|
con = sqlite3.connect(DB_PATH)
|
|
con.row_factory = sqlite3.Row
|
|
cur = con.cursor()
|
|
cur.execute(query)
|
|
|
|
links_list = []
|
|
for row in cur.fetchall():
|
|
d = dict(zip(row.keys(), row))
|
|
links_list.append(d)
|
|
|
|
con.close()
|
|
return links_list
|
|
|
|
def regen_JSON():
|
|
"""Gets links from DB and outputs them in JSON"""
|
|
links_list = get_data_from_query('SELECT url, name FROM links WHERE active=1 ORDER BY position')
|
|
links_json = json.dumps(links_list, indent=4)
|
|
return links_json
|
|
|
|
@auth.verify_password
|
|
def verify_password(username, password):
|
|
if username in users and \
|
|
check_password_hash(users.get(username), password):
|
|
return username
|
|
|
|
@app.route('/links', methods = ['GET'])
|
|
def get_links():
|
|
links_list = get_data_from_query('SELECT url, name FROM links WHERE active=1 ORDER BY position')
|
|
return jsonify(links_list)
|
|
|
|
@app.route('/editor/links', methods = ['POST'])
|
|
@auth.login_required
|
|
def update_links():
|
|
con = sqlite3.connect(DB_PATH)
|
|
cur = con.cursor()
|
|
try:
|
|
cur.execute("begin")
|
|
cur.execute('DELETE FROM links')
|
|
|
|
links = []
|
|
data = request.json['links']
|
|
items = 'url', 'name', 'clicks', 'active'
|
|
for i in range(len(data)):
|
|
if not(all(e in data[i] for e in items)):
|
|
return "Bad request, some items missing from link object", 400
|
|
|
|
url = data[i]['url']
|
|
name = data[i]['name']
|
|
clicks = data[i]['clicks']
|
|
active = data[i]['active']
|
|
position = i
|
|
|
|
newlink = (url, name, clicks, position, active)
|
|
links.append(newlink)
|
|
|
|
cur.executemany('INSERT INTO links VALUES (?,?,?,?,?)', links)
|
|
con.commit()
|
|
con.close()
|
|
|
|
frontend_path = os.environ.get('FRONTEND_PATH', None)
|
|
if frontend_path is None:
|
|
raise Exception('FRONTEND_PATH is not defined')
|
|
|
|
update_command = f"cd '{frontend_path}' && ./update.sh"
|
|
status = subprocess.call(update_command, shell=True)
|
|
|
|
if status != 0:
|
|
raise Exception(f"`{update_command}` exited with an error ({status})")
|
|
|
|
except Exception as e:
|
|
cur.execute("rollback")
|
|
con.close()
|
|
raise e
|
|
|
|
links_list = get_data_from_query('SELECT name, url, clicks, active FROM links ORDER BY position')
|
|
return jsonify(links_list)
|
|
|
|
@app.route('/editor/links', methods = ['GET'])
|
|
@auth.login_required
|
|
def get_editor_links():
|
|
"""endpoint lists all URLs and clicks, returns json object for editor."""
|
|
links_list = get_data_from_query('SELECT name, url, clicks, active FROM links ORDER BY position')
|
|
return jsonify(links_list)
|
|
|
|
@app.route('/clicks', methods=['POST'])
|
|
def update_clicks():
|
|
if ('url' not in request.json or 'name' not in request.json):
|
|
return 'url and/or name not found', 500
|
|
else:
|
|
url_id = request.json['url']
|
|
url_name = request.json['name']
|
|
con = sqlite3.connect(DB_PATH)
|
|
cur = con.cursor()
|
|
cur.execute("UPDATE links SET clicks=clicks + 1 WHERE url=? AND name=?", [url_id, url_name])
|
|
con.commit()
|
|
con.close()
|
|
return 'ok'
|
|
|
|
if __name__ == "__main__":
|
|
app.run(debug=True)
|