203 lines
7.2 KiB
Python
203 lines
7.2 KiB
Python
from flask import Flask, jsonify, request
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from sqlalchemy.sql.expression import func
|
|
import secrets
|
|
|
|
|
|
################################ CONFIG ################################
|
|
app = Flask(__name__)
|
|
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///citations.db'
|
|
db = SQLAlchemy(app)
|
|
|
|
################################ CLASS ################################
|
|
# CLASS CITATION
|
|
class Citation(db.Model):
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
auteur = db.Column(db.String(50), nullable=False)
|
|
citation = db.Column(db.String(500), nullable=False)
|
|
|
|
# CLASS USER
|
|
class User(db.Model):
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
username = db.Column(db.String(50), unique=True, nullable=False)
|
|
api_key = db.Column(db.String(64), unique=True, nullable=False)
|
|
is_admin = db.Column(db.Boolean, default=False)
|
|
|
|
|
|
# CREATE ADMIN FIRST TIME
|
|
def initialize_admin_api_key():
|
|
# Verif si l'admin existe déjà
|
|
admin_user = User.query.filter_by(username='admin').first()
|
|
|
|
if not admin_user:
|
|
# Si l'admin n'existe pas, générer une nouvelle clé API et l'ajouter à la bdd
|
|
admin_api_key = secrets.token_urlsafe(32)
|
|
admin_user = User(username='admin', api_key=admin_api_key, is_admin=True)
|
|
db.session.add(admin_user)
|
|
db.session.commit()
|
|
print(f"Clé API de l'administrateur : {admin_api_key}")
|
|
|
|
|
|
# REQUIRE API KEY WRAPPER
|
|
def require_api_key(*api_types):
|
|
def decorator(func):
|
|
def wrapper(*args, **kwargs):
|
|
|
|
provided_key = request.headers.get('API-Key')
|
|
user = User.query.filter_by(api_key=provided_key).first()
|
|
|
|
if user:
|
|
if 'admin' in api_types and user.is_admin:
|
|
return func(*args, **kwargs)
|
|
elif 'user' in api_types:
|
|
return func(*args, **kwargs)
|
|
|
|
return jsonify({"error": "Invalid API key or insufficient permissions"}), 401
|
|
|
|
return wrapper
|
|
return decorator
|
|
|
|
|
|
# CREATE DB
|
|
with app.app_context():
|
|
db.create_all()
|
|
initialize_admin_api_key()
|
|
|
|
|
|
################################ ROUTES ################################
|
|
## RANDOM CITATION
|
|
@app.route('/api/citations/random', methods=['GET'], endpoint='random')
|
|
@require_api_key('user', 'admin')
|
|
def get_citation():
|
|
citation = Citation.query.order_by(func.random()).first()
|
|
citation = {"auteur": citation.auteur, "citation": citation.citation}
|
|
|
|
response = jsonify({"citations": citation})
|
|
response.headers['Content-Type'] = 'application/json; charset=utf-8'
|
|
return response
|
|
|
|
## GET ALL CITATION
|
|
@app.route('/api/citations/all', methods=['GET'], endpoint='all')
|
|
@require_api_key('admin')
|
|
def get_all_citations():
|
|
citations = Citation.query.all()
|
|
citations_list = [{"auteur": citation.auteur, "citation": citation.citation} for citation in citations]
|
|
|
|
response = jsonify({"citations": citations_list})
|
|
response.headers['Content-Type'] = 'application/json; charset=utf-8'
|
|
return response
|
|
|
|
## AJOUT CITATION
|
|
@app.route('/api/citations/add', methods=['POST'], endpoint='ajout')
|
|
@require_api_key('admin')
|
|
def add_citation():
|
|
try:
|
|
data = request.get_json()
|
|
new_auteur = data['auteur']
|
|
new_citation_text = data['citation']
|
|
existing_citation = Citation.query.filter_by(auteur=new_auteur, citation=new_citation_text).first()
|
|
|
|
if existing_citation:
|
|
return jsonify({"error": "Cette citation existe deja dans la base de donnees"}), 400
|
|
|
|
new_citation = Citation(auteur=data['auteur'], citation=data['citation'])
|
|
db.session.add(new_citation)
|
|
db.session.commit()
|
|
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
response = jsonify({"message": "Citation ajoutee avec succes!"})
|
|
response.headers['Content-Type'] = 'application/json; charset=utf-8'
|
|
return response
|
|
|
|
## AJOUT LISTE DE CITATIONS
|
|
@app.route('/api/citations/add_list', methods=['POST'], endpoint='ajout_liste')
|
|
@require_api_key('admin')
|
|
def add_citations_from_list():
|
|
try:
|
|
data = request.get_json()
|
|
citations_to_add = data.get('citations')
|
|
|
|
if not citations_to_add or not isinstance(citations_to_add, list):
|
|
response = jsonify({"error": "Veuillez fournir une liste valide de citations"}), 400
|
|
return response
|
|
|
|
citations = []
|
|
for citation_data in citations_to_add:
|
|
new_auteur = citation_data.get('auteur')
|
|
new_citation_text = citation_data.get('citation')
|
|
|
|
if not new_auteur or not new_citation_text:
|
|
response = jsonify({"error": "Chaque element de la liste doit avoir des cles 'auteur' et 'citation'"}), 400
|
|
return response
|
|
|
|
existing_citation = Citation.query.filter_by(auteur=new_auteur, citation=new_citation_text).first()
|
|
if existing_citation:
|
|
response = jsonify({"error": f"La citation de {new_auteur} existe deja dans la base de donnees"}), 400
|
|
return response
|
|
|
|
new_citation = Citation(auteur=new_auteur, citation=new_citation_text)
|
|
citations.append(new_citation)
|
|
|
|
db.session.add_all(citations)
|
|
db.session.commit()
|
|
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
response = jsonify({"message": "Citations ajoutees avec succes depuis la liste!"})
|
|
response.headers['Content-Type'] = 'application/json; charset=utf-8'
|
|
return response
|
|
|
|
################################ AJOUT USER ################################
|
|
@app.route('/api/users/add', methods=['POST'], endpoint='add_user')
|
|
@require_api_key('admin')
|
|
def add_user():
|
|
try:
|
|
data = request.get_json()
|
|
new_username = data['username']
|
|
is_admin = data.get('is_admin', False)
|
|
existing_user = User.query.filter_by(username=new_username).first()
|
|
|
|
if existing_user:
|
|
return jsonify({"error": "Cet utilisateur existe déjà"}), 400
|
|
|
|
new_api_key = secrets.token_urlsafe(32)
|
|
|
|
new_user = User(username=new_username, api_key=new_api_key, is_admin=is_admin)
|
|
db.session.add(new_user)
|
|
db.session.commit()
|
|
|
|
return jsonify({"username": new_username, "api_key": new_api_key, "is_admin": is_admin})
|
|
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
################################ DEL USER ################################
|
|
@app.route('/api/users/delete', methods=['DELETE'], endpoint='delete_user')
|
|
@require_api_key('admin')
|
|
def delete_user():
|
|
try:
|
|
data = request.get_json()
|
|
username_to_delete = data.get('username')
|
|
|
|
if not username_to_delete:
|
|
return jsonify({"error": "Le nom d'utilisateur est requis"}), 400
|
|
|
|
user_to_delete = User.query.filter_by(username=username_to_delete).first()
|
|
|
|
if not user_to_delete:
|
|
return jsonify({"error": "Cet utilisateur n'existe pas"}), 404
|
|
|
|
db.session.delete(user_to_delete)
|
|
db.session.commit()
|
|
|
|
return jsonify({"message": "Utilisateur " + username_to_delete + " supprimé avec succès"})
|
|
|
|
except Exception as e:
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
################################ MAIN ################################
|
|
if __name__ == '__main__':
|
|
app.run(debug=True) |
