From 190794b2c1dc64028a9d65f14daffdf0a323ade4 Mon Sep 17 00:00:00 2001
From: root <root@localhost>
Date: Wed, 12 Mar 2025 13:58:24 +0100
Subject: [PATCH] Claude
---
CLAUDE.md | 25 +++++++++++++++++++
backend/app.py | 65 +++++++++++++++++++++++++++++++++++++++++++++++---
2 files changed, 87 insertions(+), 3 deletions(-)
create mode 100644 CLAUDE.md
diff --git a/CLAUDE.md b/CLAUDE.md
new file mode 100644
index 0000000..1bfca86
--- /dev/null
+++ b/CLAUDE.md
@@ -0,0 +1,25 @@
+# MYP Project Development Guidelines
+
+## Build/Run Commands
+- Backend: `cd backend && source venv/bin/activate && python app.py`
+- Frontend: `cd packages/reservation-platform && pnpm dev`
+- Run tests: `cd backend && python -m unittest development/tests/tests.py`
+- Run single test: `cd backend && python -m unittest development.tests.tests.MYPBackendTestCase.test_name`
+- Check jobs manually: `cd backend && source venv/bin/activate && flask check-jobs`
+- Lint frontend: `cd packages/reservation-platform && pnpm lint`
+- Format frontend: `cd packages/reservation-platform && npx @biomejs/biome format --write ./src`
+
+## Code Style
+- **Python Backend**:
+ - Use PEP 8 conventions, 4-space indentation
+ - Line width: 100 characters max
+ - Add docstrings to functions and classes
+ - Error handling: Use try/except with specific exceptions
+ - Naming: snake_case for functions/variables, PascalCase for classes
+
+- **Frontend (Next.js/TypeScript)**:
+ - Use Biome for formatting and linting (line width: 120 chars)
+ - Organize imports automatically with Biome
+ - Use TypeScript types for all code
+ - Use React hooks for state management
+ - Naming: camelCase for functions/variables, PascalCase for components
\ No newline at end of file
diff --git a/backend/app.py b/backend/app.py
index 0f7c71c..5dcb0f7 100755
--- a/backend/app.py
+++ b/backend/app.py
@@ -9,6 +9,8 @@ import json
import logging
import uuid
import sqlite3
+import threading
+import time
from logging.handlers import RotatingFileHandler
from datetime import timedelta
from PyP100 import PyP100
@@ -28,6 +30,7 @@ app.config['SESSION_COOKIE_HTTPONLY'] = True
app.config['SESSION_COOKIE_SECURE'] = os.environ.get('FLASK_ENV') == 'production'
app.config['SESSION_COOKIE_SAMESITE'] = 'Lax'
app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(days=7)
+app.config['JOB_CHECK_INTERVAL'] = int(os.environ.get('JOB_CHECK_INTERVAL', '60')) # Sekunden
# Steckdosen-Konfiguration
TAPO_USERNAME = os.environ.get('TAPO_USERNAME')
@@ -124,10 +127,19 @@ def init_printers():
if ip_address in existing_ips:
app.logger.info(f"Drucker mit IP {ip_address} existiert bereits in der Datenbank")
+ # Setze den Status des existierenden Druckers auf 0 (verfügbar)
+ socket_id = existing_ips[ip_address]
+ update_socket(socket_id, status=0)
+ # Stelle sicher, dass die Steckdose wirklich ausgeschaltet ist
+ turn_off_socket(ip_address)
+ app.logger.info(f"Steckdose mit IP {ip_address} wurde beim Start ausgeschaltet")
else:
# Neuen Drucker eintragen
- create_socket(name=printer_name, description=description, ip_address=ip_address, status=0)
+ new_socket = create_socket(name=printer_name, description=description, ip_address=ip_address, status=0)
app.logger.info(f"Neuer Drucker angelegt: {printer_name} mit IP {ip_address}")
+ # Stelle sicher, dass die Steckdose wirklich ausgeschaltet ist
+ turn_off_socket(ip_address)
+ app.logger.info(f"Neue Steckdose mit IP {ip_address} wurde beim Start ausgeschaltet")
# Benutzerverwaltung
def get_user_by_id(user_id):
@@ -1101,7 +1113,6 @@ def stats():
})
# Regelmäßige Überprüfung der Jobs und automatische Abschaltung der Steckdosen
-@app.cli.command("check-jobs")
def check_jobs():
"""Überprüft abgelaufene Jobs und schaltet Steckdosen automatisch aus."""
with app.app_context():
@@ -1132,11 +1143,29 @@ def check_jobs():
# Nur wenn es nicht der letzte Versuch war, kurz warten und neu versuchen
if attempt < max_attempts:
- import time
time.sleep(1)
app.logger.info(f"{len(expired_jobs)} abgelaufene Jobs überprüft, {handled_jobs} Steckdosen aktualisiert.")
+# Hintergrund-Thread für das Job-Polling
+def background_job_checker():
+ """Hintergrund-Thread, der regelmäßig abgelaufene Jobs überprüft."""
+ app.logger.info("Starte Hintergrund-Thread für Job-Überprüfung")
+ while True:
+ try:
+ check_jobs()
+ except Exception as e:
+ app.logger.error(f"Fehler im Hintergrund-Thread für Job-Überprüfung: {e}")
+
+ # Pause zwischen den Überprüfungen
+ time.sleep(app.config['JOB_CHECK_INTERVAL'])
+
+# CLI-Befehl für manuelle Ausführung
+@app.cli.command("check-jobs")
+def cli_check_jobs():
+ """CLI-Befehl zur manuellen Überprüfung abgelaufener Jobs."""
+ check_jobs()
+
@app.route('/api/job/<job_id>/status', methods=['GET'])
def job_status(job_id):
"""Endpunkt zum Überprüfen des Status eines Jobs für Frontend-Polling."""
@@ -1280,10 +1309,40 @@ def stats_page():
return redirect(url_for('index'))
return render_template('stats.html', current_user=current_user, active_page='stats')
+# Initialisierung und Start des Hintergrund-Threads beim ersten Request
+with app.app_context():
+ # Diese Funktion wird nach dem App-Start aber vor dem ersten Request ausgeführt
+ @app.before_request
+ def initialize_background_tasks():
+ """Startet den Hintergrund-Thread für Job-Überprüfung beim ersten Request."""
+ # Überprüfung, ob dieser Handler bereits ausgeführt wurde
+ if getattr(app, '_job_thread_initialized', False):
+ return
+
+ # Starte den Hintergrund-Thread nur, wenn er noch nicht läuft
+ for thread in threading.enumerate():
+ if thread.name == 'job_checker_thread':
+ app.logger.info("Hintergrund-Thread für Job-Überprüfung läuft bereits")
+ app._job_thread_initialized = True
+ return
+
+ # Thread starten
+ job_thread = threading.Thread(target=background_job_checker, daemon=True, name='job_checker_thread')
+ job_thread.start()
+ app.logger.info("Hintergrund-Thread für Job-Überprüfung beim ersten Request gestartet")
+ app._job_thread_initialized = True
+
# Server starten
if __name__ == '__main__':
with app.app_context():
init_db()
if PRINTERS:
+ # Initialisiere Drucker und schalte alle Steckdosen beim Start aus
init_printers()
+
+ # Starte den Hintergrund-Thread für die Job-Überprüfung
+ job_thread = threading.Thread(target=background_job_checker, daemon=True, name='job_checker_thread')
+ job_thread.start()
+ app.logger.info("Hintergrund-Thread für Job-Überprüfung gestartet")
+
app.run(debug=True, host='0.0.0.0')
\ No newline at end of file