537 lines
20 KiB
Python
537 lines
20 KiB
Python
import os
|
|
from flask import Flask, render_template, request, redirect, url_for, session, flash
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user
|
|
from flask_bcrypt import Bcrypt
|
|
from werkzeug.utils import secure_filename
|
|
from functools import wraps
|
|
from flask_migrate import Migrate
|
|
from pdf2image import convert_from_path
|
|
import subprocess
|
|
|
|
app = Flask(__name__, instance_relative_config=True)
|
|
|
|
# Set the secret key from environment variable or use a default value
|
|
app.config['SECRET_KEY'] = os.getenv('SECRET_KEY', 'Ana_Are_Multe_Mere-Si_Nu_Are_Pere')
|
|
|
|
# Configure the database location to be in the instance folder
|
|
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(app.instance_path, 'dashboard.db')
|
|
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
|
|
|
|
# Ensure the instance folder exists
|
|
os.makedirs(app.instance_path, exist_ok=True)
|
|
|
|
db = SQLAlchemy(app)
|
|
bcrypt = Bcrypt(app)
|
|
|
|
UPLOAD_FOLDER = 'static/uploads'
|
|
app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER
|
|
UPLOAD_FOLDERLOGO = 'static/resurse'
|
|
app.config['UPLOAD_FOLDERLOGO'] = UPLOAD_FOLDERLOGO
|
|
|
|
# Ensure the upload folder exists
|
|
if not os.path.exists(UPLOAD_FOLDER):
|
|
os.makedirs(UPLOAD_FOLDER)
|
|
os.makedirs(UPLOAD_FOLDERLOGO)
|
|
|
|
login_manager = LoginManager(app)
|
|
login_manager.login_view = 'login'
|
|
|
|
migrate = Migrate(app, db)
|
|
|
|
@login_manager.user_loader
|
|
def load_user(user_id):
|
|
return db.session.get(User, int(user_id))
|
|
|
|
# Modele pentru baza de date
|
|
class User(db.Model, UserMixin):
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
username = db.Column(db.String(80), unique=True, nullable=False)
|
|
password = db.Column(db.String(120), nullable=False)
|
|
role = db.Column(db.String(20), nullable=False, default='user')
|
|
theme = db.Column(db.String(10), nullable=False, default='light')
|
|
|
|
class Player(db.Model):
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
username = db.Column(db.String(80), unique=True, nullable=False)
|
|
hostname = db.Column(db.String(120), unique=True, nullable=False)
|
|
password = db.Column(db.String(120), nullable=False)
|
|
|
|
class Group(db.Model):
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
name = db.Column(db.String(80), unique=True, nullable=False)
|
|
players = db.relationship('Player', secondary='group_player', backref='groups')
|
|
|
|
group_player = db.Table('group_player',
|
|
db.Column('group_id', db.Integer, db.ForeignKey('group.id'), primary_key=True),
|
|
db.Column('player_id', db.Integer, db.ForeignKey('player.id'), primary_key=True)
|
|
)
|
|
|
|
class Content(db.Model):
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
file_name = db.Column(db.String(120), nullable=False)
|
|
duration = db.Column(db.Integer, nullable=False)
|
|
player_id = db.Column(db.Integer, db.ForeignKey('player.id'), nullable=True)
|
|
group_id = db.Column(db.Integer, db.ForeignKey('group.id'), nullable=True)
|
|
|
|
def admin_required(f):
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
if current_user.role != 'admin':
|
|
return redirect(url_for('dashboard'))
|
|
return f(*args, **kwargs)
|
|
return decorated_function
|
|
|
|
def convert_ppt_to_pdf(input_file, output_file):
|
|
command = ['libreoffice', '--headless', '--convert-to', 'pdf', '--outdir', os.path.dirname(output_file), input_file]
|
|
subprocess.run(command, check=True)
|
|
|
|
@app.route('/')
|
|
@login_required
|
|
def dashboard():
|
|
players = Player.query.all()
|
|
groups = Group.query.all()
|
|
logo_exists = os.path.exists(os.path.join(app.config['UPLOAD_FOLDERLOGO'], 'logo.png'))
|
|
return render_template('dashboard.html', players=players, groups=groups, logo_exists=logo_exists)
|
|
|
|
@app.route('/register', methods=['GET', 'POST'])
|
|
def register():
|
|
if request.method == 'POST':
|
|
username = request.form['username']
|
|
password = request.form['password']
|
|
hashed_password = bcrypt.generate_password_hash(password).decode('utf-8')
|
|
new_user = User(username=username, password=hashed_password, role='user')
|
|
db.session.add(new_user)
|
|
db.session.commit()
|
|
return redirect(url_for('login'))
|
|
return render_template('register.html')
|
|
|
|
@app.route('/login', methods=['GET', 'POST'])
|
|
def login():
|
|
if request.method == 'POST':
|
|
username = request.form['username']
|
|
password = request.form['password']
|
|
user = User.query.filter_by(username=username).first()
|
|
if user and bcrypt.check_password_hash(user.password, password):
|
|
login_user(user)
|
|
return redirect(url_for('dashboard'))
|
|
else:
|
|
flash('Login Unsuccessful. Please check username and password', 'danger')
|
|
|
|
login_picture_exists = os.path.exists(os.path.join(app.config['UPLOAD_FOLDERLOGO'], 'login_picture.png'))
|
|
return render_template('login.html', login_picture_exists=login_picture_exists)
|
|
|
|
@app.route('/logout')
|
|
@login_required
|
|
def logout():
|
|
logout_user()
|
|
return redirect(url_for('login'))
|
|
|
|
@app.route('/upload_content', methods=['GET', 'POST'])
|
|
@login_required
|
|
@admin_required
|
|
def upload_content():
|
|
if request.method == 'POST':
|
|
target_type = request.form['target_type']
|
|
target_id = int(request.form['target_id'])
|
|
files = request.files.getlist('files')
|
|
duration = int(request.form['duration'])
|
|
return_url = request.form['return_url']
|
|
media_type = request.form['media_type']
|
|
|
|
for file in files:
|
|
filename = secure_filename(file.filename)
|
|
file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
|
|
file.save(file_path)
|
|
|
|
if media_type == 'pdf':
|
|
# Convert PDF to PNG
|
|
images = convert_from_path(file_path)
|
|
for i, image in enumerate(images):
|
|
image_filename = f"{os.path.splitext(filename)[0]}_page_{i + 1}.png"
|
|
image_path = os.path.join(app.config['UPLOAD_FOLDER'], image_filename)
|
|
image.save(image_path, 'PNG')
|
|
new_content = Content(file_name=image_filename, duration=duration, player_id=target_id if target_type == 'player' else None, group_id=target_id if target_type == 'group' else None)
|
|
db.session.add(new_content)
|
|
os.remove(file_path) # Remove the original PDF file
|
|
elif media_type == 'ppt':
|
|
# Convert PPT/PPTX to PDF
|
|
pdf_filename = f"{os.path.splitext(filename)[0]}.pdf"
|
|
pdf_path = os.path.join(app.config['UPLOAD_FOLDER'], pdf_filename)
|
|
convert_ppt_to_pdf(file_path, pdf_path)
|
|
# Convert PDF to PNG
|
|
images = convert_from_path(pdf_path)
|
|
for i, image in enumerate(images):
|
|
image_filename = f"{os.path.splitext(filename)[0]}_page_{i + 1}.png"
|
|
image_path = os.path.join(app.config['UPLOAD_FOLDER'], image_filename)
|
|
image.save(image_path, 'PNG')
|
|
new_content = Content(file_name=image_filename, duration=duration, player_id=target_id if target_type == 'player' else None, group_id=target_id if target_type == 'group' else None)
|
|
db.session.add(new_content)
|
|
os.remove(file_path) # Remove the original PPT/PPTX file
|
|
os.remove(pdf_path) # Remove the intermediate PDF file
|
|
else:
|
|
new_content = Content(file_name=filename, duration=duration, player_id=target_id if target_type == 'player' else None, group_id=target_id if target_type == 'group' else None)
|
|
db.session.add(new_content)
|
|
|
|
db.session.commit()
|
|
return redirect(return_url)
|
|
|
|
target_type = request.args.get('target_type')
|
|
target_id = request.args.get('target_id')
|
|
return_url = request.args.get('return_url', url_for('dashboard'))
|
|
players = Player.query.all()
|
|
groups = Group.query.all()
|
|
return render_template('upload_content.html', target_type=target_type, target_id=target_id, players=players, groups=groups, return_url=return_url)
|
|
|
|
@app.route('/admin')
|
|
@login_required
|
|
@admin_required
|
|
def admin():
|
|
logo_exists = os.path.exists(os.path.join(app.config['UPLOAD_FOLDERLOGO'], 'logo.png'))
|
|
login_picture_exists = os.path.exists(os.path.join(app.config['UPLOAD_FOLDERLOGO'], 'login_picture.png'))
|
|
users = User.query.all()
|
|
return render_template('admin.html', users=users, logo_exists=logo_exists, login_picture_exists=login_picture_exists)
|
|
|
|
@app.route('/admin/change_role/<int:user_id>', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def change_role(user_id):
|
|
user = User.query.get_or_404(user_id)
|
|
new_role = request.form['role']
|
|
user.role = new_role
|
|
db.session.commit()
|
|
return redirect(url_for('admin'))
|
|
|
|
@app.route('/admin/delete_user/<int:user_id>', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def delete_user(user_id):
|
|
user = User.query.get_or_404(user_id)
|
|
db.session.delete(user)
|
|
db.session.commit()
|
|
return redirect(url_for('admin'))
|
|
|
|
@app.route('/admin/create_user', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def create_user():
|
|
username = request.form['username']
|
|
password = request.form['password']
|
|
role = request.form['role']
|
|
hashed_password = bcrypt.generate_password_hash(password).decode('utf-8')
|
|
new_user = User(username=username, password=hashed_password, role=role)
|
|
db.session.add(new_user)
|
|
db.session.commit()
|
|
return redirect(url_for('admin'))
|
|
|
|
@app.route('/group/<int:group_id>/manage')
|
|
@login_required
|
|
@admin_required
|
|
def manage_group(group_id):
|
|
group = Group.query.get_or_404(group_id)
|
|
available_players = Player.query.filter(~Player.groups.any(Group.id == group_id)).all()
|
|
return render_template('manage_group.html', group=group, available_players=available_players)
|
|
|
|
@app.route('/group/<int:group_id>/add_player', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def add_player_to_group(group_id):
|
|
group = Group.query.get_or_404(group_id)
|
|
player_id = request.form['player_id']
|
|
player = Player.query.get_or_404(player_id)
|
|
group.players.append(player)
|
|
db.session.commit()
|
|
return redirect(url_for('manage_group', group_id=group_id))
|
|
|
|
@app.route('/group/<int:group_id>/remove_player/<int:player_id>', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def remove_player_from_group(group_id, player_id):
|
|
group = Group.query.get_or_404(group_id)
|
|
player = Player.query.get_or_404(player_id)
|
|
group.players.remove(player)
|
|
db.session.commit()
|
|
return redirect(url_for('manage_group', group_id=group_id))
|
|
|
|
@app.route('/group/<int:group_id>/upload', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def upload_content_to_group(group_id):
|
|
group = Group.query.get_or_404(group_id)
|
|
files = request.files.getlist('files')
|
|
duration = int(request.form['duration'])
|
|
|
|
for file in files:
|
|
filename = secure_filename(file.filename)
|
|
file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
|
|
file.save(file_path)
|
|
new_content = Content(file_name=filename, duration=duration, group_id=group_id)
|
|
db.session.add(new_content)
|
|
|
|
db.session.commit()
|
|
return redirect(url_for('manage_group', group_id=group_id))
|
|
|
|
@app.route('/group/content/<int:content_id>/edit', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def edit_group_content(content_id):
|
|
content = Content.query.get_or_404(content_id)
|
|
new_duration = int(request.form['duration'])
|
|
content.duration = new_duration
|
|
db.session.commit()
|
|
return redirect(url_for('manage_group', group_id=content.group_id))
|
|
|
|
@app.route('/group/content/<int:content_id>/delete', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def delete_group_content(content_id):
|
|
content = Content.query.get_or_404(content_id)
|
|
group_id = content.group_id
|
|
db.session.delete(content)
|
|
db.session.commit()
|
|
return redirect(url_for('manage_group', group_id=group_id))
|
|
|
|
@app.route('/group/<int:group_id>/delete', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def delete_group(group_id):
|
|
group = db.session.get(Group, group_id)
|
|
db.session.delete(group)
|
|
db.session.commit()
|
|
return redirect(url_for('dashboard'))
|
|
|
|
@app.route('/player/<int:player_id>')
|
|
@login_required
|
|
def player_page(player_id):
|
|
player = db.session.get(Player, player_id)
|
|
content = Content.query.filter_by(player_id=player_id).all()
|
|
return render_template('player_page.html', player=player, content=content)
|
|
|
|
@app.route('/player/<int:player_id>/upload', methods=['POST'])
|
|
@login_required
|
|
def upload_content_to_player(player_id):
|
|
player = Player.query.get_or_404(player_id)
|
|
files = request.files.getlist('files')
|
|
duration = int(request.form['duration'])
|
|
|
|
for file in files:
|
|
filename = secure_filename(file.filename)
|
|
file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
|
|
file.save(file_path)
|
|
new_content = Content(file_name=filename, duration=duration, player_id=player_id)
|
|
db.session.add(new_content)
|
|
|
|
db.session.commit()
|
|
return redirect(url_for('player_page', player_id=player_id))
|
|
|
|
@app.route('/content/<int:content_id>/edit', methods=['POST'])
|
|
@login_required
|
|
def edit_content(content_id):
|
|
content = Content.query.get_or_404(content_id)
|
|
new_duration = int(request.form['duration'])
|
|
content.duration = new_duration
|
|
db.session.commit()
|
|
return redirect(url_for('player_page', player_id=content.player_id))
|
|
|
|
@app.route('/content/<int:content_id>/delete', methods=['POST'])
|
|
@login_required
|
|
def delete_content(content_id):
|
|
content = Content.query.get_or_404(content_id)
|
|
player_id = content.player_id
|
|
db.session.delete(content)
|
|
db.session.commit()
|
|
return redirect(url_for('player_page', player_id=player_id))
|
|
|
|
@app.route('/player/<int:player_id>/fullscreen', methods=['GET', 'POST'])
|
|
def player_fullscreen(player_id):
|
|
player = Player.query.get_or_404(player_id)
|
|
|
|
if request.method == 'POST':
|
|
hostname = request.form['hostname']
|
|
password = request.form['password']
|
|
if player.hostname == hostname and bcrypt.check_password_hash(player.password, password):
|
|
authenticated = True
|
|
else:
|
|
authenticated = False
|
|
else:
|
|
authenticated = False
|
|
|
|
if authenticated or current_user.is_authenticated:
|
|
if player.groups:
|
|
# If the player is part of a group, get the group's content
|
|
group = player.groups[0] # Assuming a player can only be in one group
|
|
content = Content.query.filter_by(group_id=group.id).all()
|
|
else:
|
|
# If the player is not part of a group, get the player's content
|
|
content = Content.query.filter_by(player_id=player_id).all()
|
|
return render_template('player_fullscreen.html', player=player, content=content)
|
|
else:
|
|
return render_template('player_auth.html', player_id=player_id)
|
|
|
|
@app.route('/group/<int:group_id>/fullscreen')
|
|
@login_required
|
|
def group_fullscreen(group_id):
|
|
group = Group.query.get_or_404(group_id)
|
|
content = Content.query.filter_by(group_id=group.id).all()
|
|
return render_template('group_fullscreen.html', group=group, content=content)
|
|
|
|
@app.route('/player/<int:player_id>/delete', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def delete_player(player_id):
|
|
player = Player.query.get_or_404(player_id)
|
|
|
|
# Delete all media related to the player
|
|
media_items = Content.query.filter_by(player_id=player_id).all()
|
|
for media in media_items:
|
|
db.session.delete(media)
|
|
|
|
# Delete the player
|
|
db.session.delete(player)
|
|
db.session.commit()
|
|
|
|
return redirect(url_for('dashboard'))
|
|
|
|
@app.route('/player/add', methods=['GET', 'POST'])
|
|
@login_required
|
|
@admin_required
|
|
def add_player():
|
|
if request.method == 'POST':
|
|
username = request.form['username']
|
|
hostname = request.form['hostname']
|
|
password = bcrypt.generate_password_hash(request.form['password']).decode('utf-8')
|
|
new_player = Player(username=username, hostname=hostname, password=password)
|
|
db.session.add(new_player)
|
|
db.session.commit()
|
|
return redirect(url_for('dashboard'))
|
|
return render_template('add_player.html')
|
|
|
|
@app.route('/group/add', methods=['GET', 'POST'])
|
|
@login_required
|
|
@admin_required
|
|
def add_group():
|
|
if request.method == 'POST':
|
|
name = request.form['name']
|
|
new_group = Group(name=name)
|
|
db.session.add(new_group)
|
|
db.session.commit()
|
|
return redirect(url_for('dashboard'))
|
|
return render_template('add_group.html')
|
|
|
|
@app.route('/integrate_player')
|
|
@login_required
|
|
@admin_required
|
|
def integrate_player():
|
|
players = Player.query.all()
|
|
groups = Group.query.all()
|
|
return render_template('integrate_player.html', players=players, groups=groups)
|
|
|
|
@app.route('/edit_player/<int:player_id>', methods=['GET', 'POST'])
|
|
@login_required
|
|
@admin_required
|
|
def edit_player(player_id):
|
|
player = Player.query.get_or_404(player_id)
|
|
if request.method == 'POST':
|
|
player.username = request.form['username']
|
|
player.hostname = request.form['hostname']
|
|
if request.form['password']:
|
|
player.password = bcrypt.generate_password_hash(request.form['password']).decode('utf-8')
|
|
db.session.commit()
|
|
return redirect(url_for('player_page', player_id=player.id))
|
|
|
|
return_url = request.args.get('return_url', url_for('player_page', player_id=player.id))
|
|
return render_template('edit_player.html', player=player, return_url=return_url)
|
|
|
|
@app.route('/change_theme', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def change_theme():
|
|
theme = request.form['theme']
|
|
current_user.theme = theme
|
|
db.session.commit()
|
|
return redirect(url_for('admin'))
|
|
|
|
@app.route('/upload_logo', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def upload_logo():
|
|
if 'logo' not in request.files:
|
|
return redirect(url_for('admin'))
|
|
|
|
file = request.files['logo']
|
|
if file.filename == '':
|
|
return redirect(url_for('admin'))
|
|
|
|
if file:
|
|
filename = secure_filename(file.filename)
|
|
file_path = os.path.join(app.config['UPLOAD_FOLDERLOGO'], 'logo.png')
|
|
file.save(file_path)
|
|
return redirect(url_for('admin'))
|
|
|
|
@app.route('/upload_personalization_pictures', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def upload_personalization_pictures():
|
|
logo_file = request.files.get('logo')
|
|
login_picture_file = request.files.get('login_picture')
|
|
|
|
if logo_file and logo_file.filename != '':
|
|
logo_filename = secure_filename(logo_file.filename)
|
|
logo_file_path = os.path.join(app.config['UPLOAD_FOLDERLOGO'], 'logo.png')
|
|
logo_file.save(logo_file_path)
|
|
|
|
if login_picture_file and login_picture_file.filename != '':
|
|
login_picture_filename = secure_filename(login_picture_file.filename)
|
|
login_picture_file_path = os.path.join(app.config['UPLOAD_FOLDERLOGO'], 'login_picture.png')
|
|
login_picture_file.save(login_picture_file_path)
|
|
|
|
return redirect(url_for('admin'))
|
|
|
|
@app.route('/clean_unused_files', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def clean_unused_files():
|
|
# Get all file names from the database
|
|
content_files = {content.file_name for content in Content.query.all()}
|
|
logo_file = 'resurse/logo.png'
|
|
login_picture_file = 'resurse/login_picture.png'
|
|
|
|
# Debugging: Print the content files from the database
|
|
print("Content files from database:", content_files)
|
|
|
|
# Get all files in the upload folder
|
|
all_files = set(os.listdir(app.config['UPLOAD_FOLDER']))
|
|
|
|
# Determine unused files
|
|
used_files = content_files | {logo_file, login_picture_file}
|
|
unused_files = all_files - used_files
|
|
|
|
# Debugging: Print the lists of files
|
|
print("All files:", all_files)
|
|
print("Used files:", used_files)
|
|
print("Unused files:", unused_files)
|
|
|
|
# Delete unused files
|
|
for file_name in unused_files:
|
|
file_path = os.path.join(app.config['UPLOAD_FOLDER'], file_name)
|
|
if os.path.isfile(file_path):
|
|
print(f"Deleting file: {file_path}") # Debugging: Print the file being deleted
|
|
os.remove(file_path)
|
|
|
|
flash('Unused files have been cleaned.', 'success')
|
|
return redirect(url_for('admin'))
|
|
|
|
|
|
@app.context_processor
|
|
def inject_theme():
|
|
if current_user.is_authenticated:
|
|
theme = current_user.theme
|
|
else:
|
|
theme = 'light'
|
|
return dict(theme=theme)
|
|
|
|
if __name__ == '__main__':
|
|
with app.app_context():
|
|
db.create_all() # Creează toate tabelele
|
|
app.run(debug=True)
|