Files
Projekcija/web/server.py

358 lines
11 KiB
Python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# Projekcija - Lyrics Projector (Web Server)
#
# Izvorna zasnova in implementacija: Uroš Urbanija
# Nadgradnje in vzdrževanje: Valentin Korenjak
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Flask web server for the song projector application.
HTML, CSS and JavaScript are now stored in separate template/static files rather
than being generated inside Python code.
"""
import threading
import os
import json
import queue
import time
from flask import Flask, render_template, request, jsonify, Response
# create Flask app with proper folders relative to this file
app = Flask(__name__, static_folder="static", template_folder="templates")
# Globalna referenca na SongProjector aplikaciju
_projector_app = None
# List of queues for SSE clients
_sse_clients = []
_sse_lock = threading.Lock()
def notify_clients():
"""Notify all connected SSE clients to refresh content"""
with _sse_lock:
for q in _sse_clients:
q.put("refresh content")
def set_projector_app(projector_app):
"""Postavi referenco na SongProjector aplikacijo"""
global _projector_app
_projector_app = projector_app
# ----------------------------------------------------------
# Flask rute
# ----------------------------------------------------------
@app.route("/", methods=["GET"])
def index():
"""Prikaže glavno HTML stranko"""
# predaja nadzora Jinja2, ki poišče web/templates/index.html
return render_template("index.html")
@app.route('/api/state', methods=['GET'])
def get_state():
"""Vrati trenutno stanje aplikacije"""
if _projector_app is None:
return jsonify({
'current_text': 'Napaka: Aplikacija ni inicijalizirana',
'page_info': '',
'caps_mode': False,
'split_by_stanza': False,
'can_prev': False,
'can_next': False
})
current_text = ""
page_info = ""
can_prev = False
can_next = False
if _projector_app.pages:
current_text = _projector_app.pages[_projector_app.current_page_index]
if _projector_app.all_caps_mode:
current_text = current_text.upper()
current_page = _projector_app.current_page_index + 1
total_pages = len(_projector_app.pages)
page_info = f"{_projector_app.song_number_last} {current_page}/{total_pages}"
can_prev = _projector_app.current_page_index > 0
can_next = _projector_app.current_page_index + 1 < len(_projector_app.pages)
else:
current_text = "Pripravljeno. Vpiši številko pesmi ali drugega besedila."
return jsonify({
'current_text': current_text,
'page_info': page_info,
'caps_mode': _projector_app.all_caps_mode,
'split_by_stanza': _projector_app.settings.get("split_by_stanza", False),
'can_prev': can_prev,
'can_next': can_next
})
@app.route('/api/events')
def sse_events():
"""SSE endpoint for real-time updates"""
def event_stream():
q = queue.Queue()
with _sse_lock:
_sse_clients.append(q)
try:
# Send initial refresh command on connection
yield "data: refresh content\n\n"
while True:
msg = q.get()
yield f"data: {msg}\n\n"
finally:
with _sse_lock:
_sse_clients.remove(q)
return Response(event_stream(), mimetype="text/event-stream")
@app.route('/api/load_song', methods=['POST'])
def load_song():
"""Naloži pesem po številki"""
if _projector_app is None:
return jsonify({'status': 'error', 'message': 'Aplikacija ni inicijalizirana'})
data = request.get_json()
song_number = data.get('song_number', '').strip()
if song_number:
_projector_app.song_number = song_number
_projector_app.load_song()
if hasattr(_projector_app, 'show_page'):
_projector_app.show_page()
notify_clients()
return jsonify({'status': 'ok'})
@app.route('/api/next_page', methods=['POST'])
def next_page():
"""Naslednja stran"""
if _projector_app is not None:
_projector_app.next_page()
notify_clients()
return jsonify({'status': 'ok'})
@app.route('/api/prev_page', methods=['POST'])
def prev_page():
"""Prejšnja stran"""
if _projector_app is not None:
_projector_app.prev_page()
notify_clients()
return jsonify({'status': 'ok'})
@app.route('/api/clear_screen', methods=['POST'])
def clear_screen():
"""Zatemniti ekran"""
if _projector_app is not None:
_projector_app.clear_screen()
notify_clients()
return jsonify({'status': 'ok'})
@app.route('/api/toggle_caps', methods=['POST'])
def toggle_caps():
"""Preklop med velikimi in malimi črkami"""
if _projector_app is not None:
_projector_app.all_caps_mode = not _projector_app.all_caps_mode
_projector_app.show_page()
notify_clients()
return jsonify({'status': 'ok'})
@app.route('/api/toggle_split', methods=['POST'])
def toggle_split():
"""Preklop med načinom preloma po kiticah in prostim prelomom"""
if _projector_app is not None:
_projector_app.toggle_split_mode()
notify_clients()
return jsonify({'status': 'ok'})
@app.route('/api/app_info', methods=['GET'])
def get_app_info():
"""Vrne informacije o aplikaciji iz appinfo.json"""
try:
# appinfo.json je v korenskem imeniku projekta (nad web/)
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
appinfo_path = os.path.join(base_dir, 'appinfo.json')
with open(appinfo_path, 'r', encoding='utf-8') as f:
info = json.load(f)
# Dodaj število pesmi v bazi
if _projector_app is not None:
_projector_app.cursor.execute("SELECT COUNT(*) FROM songs")
count = _projector_app.cursor.fetchone()[0]
info['song_count'] = count
return jsonify(info)
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/search_songs', methods=['POST'])
def search_songs():
"""Iskanje besedil po naslovu"""
if _projector_app is None:
return jsonify({'results': []})
data = request.get_json()
query = data.get('query', '').strip()
if not query:
return jsonify({'results': []})
try:
_projector_app.cursor.execute(
"SELECT id, title FROM songs WHERE title LIKE ? COLLATE NOCASE",
(f"%{query}%",)
)
results = _projector_app.cursor.fetchall()
return jsonify({'results': results})
except Exception as e:
return jsonify({'error': str(e)})
@app.route('/api/get_song_details', methods=['GET'])
def get_song_details():
"""Vrne podrobnosti trenutno naložene pesmi ali pesmi po ID-ju"""
if _projector_app is None:
return jsonify({'status': 'error', 'message': 'Aplikacija ni inicijalizirana'})
song_id_param = request.args.get('id')
if song_id_param:
try:
song_id = int(song_id_param)
except ValueError:
return jsonify({'status': 'error', 'message': 'Neveljaven ID pesmi'})
elif _projector_app.song_number_last:
try:
song_id = int(_projector_app.song_number_last)
except ValueError:
return jsonify({'status': 'error', 'message': 'Neveljavna številka naložene pesmi'})
else:
return jsonify({'status': 'error', 'message': 'Nobena pesem ni naložena'})
try:
_projector_app.cursor.execute("SELECT id, title, lyrics FROM songs WHERE id=?", (song_id,))
result = _projector_app.cursor.fetchone()
if result:
return jsonify({
'status': 'ok',
'song': {
'id': result[0],
'title': result[1],
'lyrics': result[2]
}
})
else:
return jsonify({'status': 'error', 'message': f'Pesem s številko {song_id} ni najdena v bazi'})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
@app.route('/api/update_song', methods=['POST'])
def update_song():
"""Posodobi naslov in besedilo pesmi ali ustvari novo"""
if _projector_app is None:
return jsonify({'status': 'error', 'message': 'Aplikacija ni inicijalizirana'})
data = request.get_json()
song_id = data.get('id')
title = data.get('title', '').strip()
lyrics = data.get('lyrics', '').strip()
if not song_id or not title or not lyrics:
return jsonify({'status': 'error', 'message': 'Manjkajoči podatki'})
try:
if song_id == 'new':
# Pridobi prvo naslednjo prosto številko
_projector_app.cursor.execute("SELECT MAX(id) FROM songs")
max_id = _projector_app.cursor.fetchone()[0]
new_id = (max_id or 0) + 1
_projector_app.cursor.execute(
"INSERT INTO songs (id, title, lyrics) VALUES (?, ?, ?)",
(new_id, title, lyrics)
)
_projector_app.conn.commit()
# Naloži novo pesem
_projector_app.song_number = str(new_id)
_projector_app.load_song()
# Pošlji ntfy obvestilo
if hasattr(_projector_app, 'send_ntfy_notification'):
_projector_app.send_ntfy_notification(new_id, title, lyrics)
return jsonify({'status': 'ok', 'new_id': new_id})
# Obstoječa pesem
_projector_app.cursor.execute("UPDATE songs SET title=?, lyrics=? WHERE id=?", (title, lyrics, song_id))
_projector_app.conn.commit()
# Pošlji ntfy obvestilo
if hasattr(_projector_app, 'send_ntfy_notification'):
_projector_app.send_ntfy_notification(song_id, title, lyrics)
# Osvežimo trenutno pesem, če je to ta, ki smo jo pravkar uredili
if str(_projector_app.song_number_last) == str(song_id):
_projector_app.song_number = str(song_id)
_projector_app.load_song()
notify_clients()
return jsonify({'status': 'ok'})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)})
def run_server(host='127.0.0.1', port=5000):
"""Zaženi Flask server"""
app.run(host=host, port=port, debug=False, use_reloader=False, threaded=True)
def start_server_thread(projector_app, host='127.0.0.1', port=5000):
"""Zaženi server v zvjenoj niti"""
set_projector_app(projector_app)
server_thread = threading.Thread(
target=run_server,
args=(host, port),
daemon=True
)
server_thread.start()
return server_thread