Nirdoan/Tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 6 Wiki Activity

Fix bugs, add log rotation, and optimize hot paths

Browse Source 
- Fix AttributeError crash on empty request body in md5, hasher, textdiff,
  jwtdecoder, timestamp, passwordgen (get_json without silent=True / or {})
- Fix memory exhaustion in ipcalc: replace list(network.hosts()) with direct
  arithmetic — safe for /8 and larger networks
- Fix O(1M) loop in cronexplainer.get_next_runs: rewrite to skip by
  month/day/hour instead of iterating every minute
- Fix connection leak in notes.ensure_table: add try/finally around conn.close
- Fix admin._ensure_tables / notes._ensure_table running DDL on every request:
  guard with module-level flags (_tables_initialized, _table_ready)
- Fix update_website returning 200 when no row matched; delete_website returning
  success when nothing was deleted; add rowcount checks for both
- Add role validation in admin create_user / update_user (_VALID_ROLES guard)
- Add delimiter length guard in csvviewer (csv.reader requires single char)
- Fix loremipsum: wrap int(count) in try/except ValueError → 400 response
- Fix auth/token: use auth_header[7:] instead of fragile .replace()
- Fix app.py: remove duplicate import sys; cache DB liveness check with 30s TTL
  to avoid a new TCP connection on every frontend page load; move api/setup
  path guard before DB check
- Replace FileHandler with RotatingFileHandler (5 MB / 3 backups) in logger;
  fix relative log paths to absolute paths anchored to __file__
- Wrap all DB connections in try/finally conn.close() throughout admin and notes

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Nirodan
2026-05-06 10:06:29 +02:00
parent 31494c9dab
commit 98bb34f094
15 changed files with 363 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/admin.py
+124 -88
View File
@@ -6,6 +6,9 @@ from util.logger import logger
admin_bp = Blueprint("admin", __name__)
_VALID_ROLES = {"user", "admin"}
_tables_initialized = False
def _require_admin():
user = verify_token()
@@ -18,6 +21,9 @@ def _require_admin():
def _ensure_tables(cur):
global _tables_initialized
if _tables_initialized:
return
cur.execute(
"""
CREATE TABLE IF NOT EXISTS users (
@@ -38,6 +44,7 @@ def _ensure_tables(cur):
)
"""
)
_tables_initialized = True
@admin_bp.route("/api/admin/users", methods=["GET"])
@@ -47,12 +54,14 @@ def list_users():
return err
try:
conn = get_connection()
cur = conn.cursor(dictionary=True)
_ensure_tables(cur)
cur.execute("SELECT id, username, role FROM users ORDER BY username ASC")
users = cur.fetchall()
cur.close()
conn.close()
try:
cur = conn.cursor(dictionary=True)
_ensure_tables(cur)
cur.execute("SELECT id, username, role FROM users ORDER BY username ASC")
users = cur.fetchall()
cur.close()
finally:
conn.close()
return jsonify(users)
except Exception as e:
logger.error(f"[Admin list_users] {e}")
@@ -64,29 +73,32 @@ def create_user():
admin, err = _require_admin()
if err:
return err
data = request.get_json() or {}
data = request.get_json(silent=True) or {}
username = data.get("username", "").strip()
password = data.get("password", "")
role = data.get("role", "user")
if not username or not password:
return jsonify({"message": "Username und Passwort erforderlich"}), 400
if role not in _VALID_ROLES:
return jsonify({"message": f"Ungültige Rolle. Erlaubt: {', '.join(_VALID_ROLES)}"}), 400
try:
conn = get_connection()
cur = conn.cursor(dictionary=True)
_ensure_tables(cur)
cur.execute("SELECT id FROM users WHERE username=%s", (username,))
if cur.fetchone():
try:
cur = conn.cursor(dictionary=True)
_ensure_tables(cur)
cur.execute("SELECT id FROM users WHERE username=%s", (username,))
if cur.fetchone():
cur.close()
return jsonify({"message": "Nutzer existiert bereits"}), 409
cur.execute(
"INSERT INTO users (username, password, role) VALUES (%s, %s, %s)",
(username, generate_password_hash(password), role)
)
conn.commit()
new_id = cur.lastrowid
cur.close()
finally:
conn.close()
return jsonify({"message": "Nutzer existiert bereits"}), 409
cur.execute(
"INSERT INTO users (username, password, role) VALUES (%s, %s, %s)",
(username, generate_password_hash(password), role)
)
conn.commit()
new_id = cur.lastrowid
cur.close()
conn.close()
logger.info(f"✅ User erstellt: {username} durch {admin['username']}")
return jsonify({"id": new_id, "username": username, "role": role}), 201
except Exception as e:
@@ -99,25 +111,29 @@ def update_user(user_id):
admin, err = _require_admin()
if err:
return err
data = request.get_json() or {}
data = request.get_json(silent=True) or {}
role = data.get("role")
password = data.get("password")
if role is None and password is None:
return jsonify({"message": "Nichts zu aktualisieren"}), 400
if role is not None and role not in _VALID_ROLES:
return jsonify({"message": f"Ungültige Rolle. Erlaubt: {', '.join(_VALID_ROLES)}"}), 400
try:
conn = get_connection()
cur = conn.cursor()
_ensure_tables(cur)
if role:
cur.execute("UPDATE users SET role=%s WHERE id=%s", (role, user_id))
if password:
cur.execute(
"UPDATE users SET password=%s WHERE id=%s",
(generate_password_hash(password), user_id)
)
conn.commit()
cur.close()
conn.close()
try:
cur = conn.cursor()
_ensure_tables(cur)
if role:
cur.execute("UPDATE users SET role=%s WHERE id=%s", (role, user_id))
if password:
cur.execute(
"UPDATE users SET password=%s WHERE id=%s",
(generate_password_hash(password), user_id)
)
conn.commit()
cur.close()
finally:
conn.close()
logger.info(f"✏️ User aktualisiert (id={user_id}) durch {admin['username']}")
return jsonify({"message": "Aktualisiert"}), 200
except Exception as e:
@@ -132,24 +148,23 @@ def delete_user(user_id):
return err
try:
conn = get_connection()
cur = conn.cursor()
_ensure_tables(cur)
# Schutz: Admin darf sich nicht selbst löschen
cur.execute("SELECT username FROM users WHERE id=%s", (user_id,))
row = cur.fetchone()
if not row:
try:
cur = conn.cursor()
_ensure_tables(cur)
cur.execute("SELECT username FROM users WHERE id=%s", (user_id,))
row = cur.fetchone()
if not row:
cur.close()
return jsonify({"message": "Nicht gefunden"}), 404
username = row[0]
if username == admin["username"]:
cur.close()
return jsonify({"message": "Du kannst dich nicht selbst löschen"}), 400
cur.execute("DELETE FROM users WHERE id=%s", (user_id,))
conn.commit()
cur.close()
finally:
conn.close()
return jsonify({"message": "Nicht gefunden"}), 404
username = row[0]
if username == admin["username"]:
cur.close()
conn.close()
return jsonify({"message": "Du kannst dich nicht selbst löschen"}), 400
cur.execute("DELETE FROM users WHERE id=%s", (user_id,))
conn.commit()
cur.close()
conn.close()
logger.info(f"🗑️ User gelöscht (id={user_id}) durch {admin['username']}")
return jsonify({"message": "Gelöscht"}), 200
except Exception as e:
@@ -166,12 +181,14 @@ def list_websites_admin():
return err
try:
conn = get_connection()
cur = conn.cursor(dictionary=True)
_ensure_tables(cur)
cur.execute("SELECT id, name, url, description FROM websites ORDER BY name ASC")
rows = cur.fetchall()
cur.close()
conn.close()
try:
cur = conn.cursor(dictionary=True)
_ensure_tables(cur)
cur.execute("SELECT id, name, url, description FROM websites ORDER BY name ASC")
rows = cur.fetchall()
cur.close()
finally:
conn.close()
return jsonify(rows)
except Exception as e:
logger.error(f"[Admin list_websites] {e}")
@@ -183,7 +200,7 @@ def create_website():
_, err = _require_admin()
if err:
return err
data = request.get_json() or {}
data = request.get_json(silent=True) or {}
name = data.get("name", "").strip()
url = data.get("url", "").strip()
description = data.get("description", "").strip()
@@ -191,16 +208,18 @@ def create_website():
return jsonify({"message": "Name und URL erforderlich"}), 400
try:
conn = get_connection()
cur = conn.cursor()
_ensure_tables(cur)
cur.execute(
"INSERT INTO websites (name, url, description) VALUES (%s, %s, %s)",
(name, url, description),
)
conn.commit()
new_id = cur.lastrowid
cur.close()
conn.close()
try:
cur = conn.cursor()
_ensure_tables(cur)
cur.execute(
"INSERT INTO websites (name, url, description) VALUES (%s, %s, %s)",
(name, url, description),
)
conn.commit()
new_id = cur.lastrowid
cur.close()
finally:
conn.close()
return jsonify({"id": new_id, "name": name, "url": url, "description": description}), 201
except Exception as e:
logger.error(f"[Admin create_website] {e}")
@@ -212,18 +231,28 @@ def update_website(item_id):
_, err = _require_admin()
if err:
return err
data = request.get_json() or {}
data = request.get_json(silent=True) or {}
name = data.get("name", "").strip()
url = data.get("url", "").strip()
if not name or not url:
return jsonify({"message": "Name und URL erforderlich"}), 400
description = data.get("description", "").strip()
try:
conn = get_connection()
cur = conn.cursor()
_ensure_tables(cur)
cur.execute(
"UPDATE websites SET name=%s, url=%s, description=%s WHERE id=%s",
(data.get("name"), data.get("url"), data.get("description", ""), item_id),
)
conn.commit()
cur.close()
conn.close()
try:
cur = conn.cursor()
_ensure_tables(cur)
cur.execute(
"UPDATE websites SET name=%s, url=%s, description=%s WHERE id=%s",
(name, url, description, item_id),
)
conn.commit()
affected = cur.rowcount
cur.close()
finally:
conn.close()
if affected == 0:
return jsonify({"message": "Nicht gefunden"}), 404
return jsonify({"message": "Aktualisiert"}), 200
except Exception as e:
logger.error(f"[Admin update_website] {e}")
@@ -237,12 +266,17 @@ def delete_website(item_id):
return err
try:
conn = get_connection()
cur = conn.cursor()
_ensure_tables(cur)
cur.execute("DELETE FROM websites WHERE id=%s", (item_id,))
conn.commit()
cur.close()
conn.close()
try:
cur = conn.cursor()
_ensure_tables(cur)
cur.execute("DELETE FROM websites WHERE id=%s", (item_id,))
conn.commit()
affected = cur.rowcount
cur.close()
finally:
conn.close()
if affected == 0:
return jsonify({"message": "Nicht gefunden"}), 404
return jsonify({"message": "Gelöscht"}), 200
except Exception as e:
logger.error(f"[Admin delete_website] {e}")
@@ -258,12 +292,14 @@ def list_websites_public():
return jsonify({"message": "Nicht autorisiert"}), 401
try:
conn = get_connection()
cur = conn.cursor(dictionary=True)
_ensure_tables(cur)
cur.execute("SELECT id, name, url, description FROM websites ORDER BY name ASC")
rows = cur.fetchall()
cur.close()
conn.close()
try:
cur = conn.cursor(dictionary=True)
_ensure_tables(cur)
cur.execute("SELECT id, name, url, description FROM websites ORDER BY name ASC")
rows = cur.fetchall()
cur.close()
finally:
conn.close()
return jsonify(rows)
except Exception as e:
logger.error(f"[Public list_websites] {e}")
backend/app.py
+22 -11
View File
@@ -1,10 +1,11 @@
import os
import sys
import time as _time
if __name__ != '__main__':
import sys
sys.path.append(os.path.dirname(__file__))
from flask import Flask, send_from_directory, redirect
from flask import Flask, send_from_directory, redirect, abort
from util.logger import logger
from util.db_config import is_configured, load_config, test_connection
from util.setup_routes import setup_blueprint
@@ -37,7 +38,6 @@ from admin import admin_bp
app = Flask(__name__, template_folder="templates")
limiter.init_app(app)
# Blueprints registrieren
app.register_blueprint(setup_blueprint)
app.register_blueprint(auth_bp)
app.register_blueprint(md5_blueprint)
@@ -62,24 +62,35 @@ app.register_blueprint(csv_blueprint)
app.register_blueprint(notes_blueprint)
app.register_blueprint(admin_bp)
# 🌐 React-Frontend ausliefern
# Cache DB liveness check so we don't open a new TCP connection on every page load.
_db_check = {"ok": False, "ts": 0.0}
_DB_CHECK_TTL = 30.0 # seconds
def _is_db_ready():
now = _time.monotonic()
if now - _db_check["ts"] > _DB_CHECK_TTL:
_db_check["ok"] = is_configured() and bool(test_connection(load_config()))
_db_check["ts"] = now
return _db_check["ok"]
@app.route('/', defaults={'path': ''})
@app.route('/<path:path>')
def serve_frontend(path):
if not is_configured() or not test_connection(load_config()):
return redirect('/setup')
if path.startswith('setup') or path.startswith('api'):
from flask import abort
# Unmatched API / setup paths get a clean 404 before any DB work.
if path.startswith('api') or path.startswith('setup'):
abort(404)
if not _is_db_ready():
return redirect('/setup')
dist_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'frontend', 'dist'))
file_path = os.path.join(dist_dir, path)
if path and os.path.exists(file_path):
return send_from_directory(dist_dir, path)
else:
return send_from_directory(dist_dir, 'index.html')
return send_from_directory(dist_dir, 'index.html')
if __name__ == '__main__':
backend/auth/token.py
+1 -1
View File
@@ -18,7 +18,7 @@ def verify_token():
logger.warning("🔐 Invalid Bearer header")
return None
token = auth_header.replace("Bearer ", "")
token = auth_header[7:]
try:
decoded = decode(token, SECRET_KEY, algorithms=["HS256"])
return decoded
backend/tools/cronexplainer.py
+82 -12
View File
@@ -96,27 +96,97 @@ def build_summary(minute, hour, day, month, weekday):
def get_next_runs(minute_vals, hour_vals, day_vals, month_vals, weekday_vals):
# cron weekday 0=Sun..6=Sat,7=Sun -> python weekday 0=Mon..6=Sun
"""Return up to 5 upcoming run times.
Uses targeted jumps (skip month, day, hour) instead of iterating every
minute, so performance is acceptable even for rare expressions like
'0 0 29 2 *' (Feb 29 at midnight).
"""
# cron weekday 0=Sun..6=Sat,7=Sun → Python weekday 0=Mon..6=Sun
py_weekdays = set((wd + 6) % 7 for wd in weekday_vals)
minute_set = set(minute_vals)
hour_set = set(hour_vals)
day_set = set(day_vals)
month_set = set(month_vals)
minute_list = sorted(minute_vals)
hour_list = sorted(hour_vals)
month_list = sorted(month_vals)
if not minute_list or not hour_list or not month_list:
return []
now = datetime.now().replace(second=0, microsecond=0) + timedelta(minutes=1)
max_date = now + timedelta(days=4 * 366)
results = []
current = now
for _ in range(2 * 366 * 24 * 60):
if len(results) >= 5:
break
if (current.month in month_set and
current.day in day_set and
current.weekday() in py_weekdays and
current.hour in hour_set and
current.minute in minute_set):
results.append(current.isoformat())
current += timedelta(minutes=1)
def _advance_from_results():
"""Move current to the next candidate after a match."""
nonlocal current
next_mins = [m for m in minute_list if m > current.minute]
if next_mins:
current = current.replace(minute=next_mins[0])
return
next_hours = [h for h in hour_list if h > current.hour]
if next_hours:
current = current.replace(hour=next_hours[0], minute=minute_list[0])
return
current = (current + timedelta(days=1)).replace(hour=hour_list[0], minute=minute_list[0])
while current <= max_date and len(results) < 5:
# ── month ──────────────────────────────────────────────────────────
if current.month not in month_set:
next_months = [m for m in month_list if m > current.month]
if next_months:
current = current.replace(
month=next_months[0], day=1,
hour=hour_list[0], minute=minute_list[0]
)
else:
current = current.replace(
year=current.year + 1, month=month_list[0], day=1,
hour=hour_list[0], minute=minute_list[0]
)
continue
# ── day + weekday ──────────────────────────────────────────────────
if current.day not in day_set or current.weekday() not in py_weekdays:
current = (current + timedelta(days=1)).replace(
hour=hour_list[0], minute=minute_list[0]
)
continue
# ── hour ───────────────────────────────────────────────────────────
if current.hour not in hour_set:
next_hours = [h for h in hour_list if h > current.hour]
if next_hours:
current = current.replace(hour=next_hours[0], minute=minute_list[0])
else:
current = (current + timedelta(days=1)).replace(
hour=hour_list[0], minute=minute_list[0]
)
continue
# ── minute ─────────────────────────────────────────────────────────
if current.minute not in minute_set:
next_mins = [m for m in minute_list if m > current.minute]
if next_mins:
current = current.replace(minute=next_mins[0])
else:
next_hours = [h for h in hour_list if h > current.hour]
if next_hours:
current = current.replace(hour=next_hours[0], minute=minute_list[0])
else:
current = (current + timedelta(days=1)).replace(
hour=hour_list[0], minute=minute_list[0]
)
continue
# ── match ──────────────────────────────────────────────────────────
results.append(current.isoformat())
_advance_from_results()
return results
@@ -127,7 +197,7 @@ def explain_cron():
if not user:
return jsonify({"message": "Nicht autorisiert"}), 401
try:
data = request.get_json() or {}
data = request.get_json(silent=True) or {}
expression = data.get("expression", "").strip()
fields = expression.split()
backend/tools/csvviewer.py
+3 -1
View File
@@ -15,7 +15,7 @@ def parse_csv():
if not user:
return jsonify({"message": "Nicht autorisiert"}), 401
try:
data = request.get_json() or {}
data = request.get_json(silent=True) or {}
text = data.get("text", "")
delimiter = data.get("delimiter", ",")
@@ -24,6 +24,8 @@ def parse_csv():
delimiter = "\t"
if not delimiter:
delimiter = ","
if len(delimiter) != 1:
return jsonify({"message": "Delimiter muss genau ein Zeichen sein"}), 400
reader = csv.reader(io.StringIO(text), delimiter=delimiter)
all_rows = list(reader)
backend/tools/hasher.py
+2 -2
View File
@@ -13,7 +13,7 @@ def hash_sha256():
if not user:
return jsonify({"message": "Nicht autorisiert"}), 401
try:
data = request.get_json()
data = request.get_json(silent=True) or {}
text = data.get("text", "")
result = hashlib.sha256(text.encode()).hexdigest()
logger.info(f"SHA256 erstellt von {user['username']}")
@@ -29,7 +29,7 @@ def hash_bcrypt():
if not user:
return jsonify({"message": "Nicht autorisiert"}), 401
try:
data = request.get_json()
data = request.get_json(silent=True) or {}
text = data.get("text", "")
hashed = bcrypt.hashpw(text.encode(), bcrypt.gensalt()).decode()
logger.info(f"bcrypt erstellt von {user['username']}")
backend/tools/ipcalc.py
+20 -10
View File
@@ -12,7 +12,7 @@ def ip_calculate():
if not user:
return jsonify({"message": "Nicht autorisiert"}), 401
try:
data = request.get_json() or {}
data = request.get_json(silent=True) or {}
cidr = data.get("cidr", "").strip()
try:
@@ -20,16 +20,26 @@ def ip_calculate():
except ValueError as e:
return jsonify({"message": f"Ungültige CIDR-Notation: {e}"}), 400
hosts = list(network.hosts())
first_host = str(hosts[0]) if hosts else str(network.network_address)
last_host = str(hosts[-1]) if hosts else str(network.broadcast_address)
total_hosts = len(hosts)
prefix = network.prefixlen
net_int = int(network.network_address)
bcast_int = int(network.broadcast_address)
# Avoid materialising millions of host objects for large networks.
if prefix == 32:
total_hosts = 1
first_host = str(network.network_address)
last_host = str(network.network_address)
elif prefix == 31:
total_hosts = 2
first_host = str(network.network_address)
last_host = str(network.broadcast_address)
else:
total_hosts = network.num_addresses - 2
first_host = str(ipaddress.IPv4Address(net_int + 1))
last_host = str(ipaddress.IPv4Address(bcast_int - 1))
# Wildcard = inverse of netmask
netmask_int = int(network.netmask)
wildcard_int = (~netmask_int) & 0xFFFFFFFF
wildcard = str(ipaddress.IPv4Address(wildcard_int))
wildcard = str(ipaddress.IPv4Address((~netmask_int) & 0xFFFFFFFF))
ip_class = "Privat" if network.is_private else "Öffentlich"
return jsonify({
@@ -40,7 +50,7 @@ def ip_calculate():
"first_host": first_host,
"last_host": last_host,
"total_hosts": total_hosts,
"prefix_length": network.prefixlen,
"prefix_length": prefix,
"ip_class": ip_class,
})
backend/tools/jwtdecoder.py
+1 -1
View File
@@ -13,7 +13,7 @@ def decode_jwt():
if not user:
return jsonify({"message": "Nicht autorisiert"}), 401
try:
data = request.get_json()
data = request.get_json(silent=True) or {}
token = data.get("token", "").strip()
header = jwt.get_unverified_header(token)
payload = jwt.decode(token, options={"verify_signature": False}, algorithms=["HS256", "HS384", "HS512", "RS256", "RS384", "RS512", "ES256", "ES384", "ES512"])
backend/tools/loremipsum.py
+4 -1
View File
@@ -44,7 +44,10 @@ def generate_lorem():
try:
data = request.get_json() or {}
gen_type = data.get("type", "sentences")
count = int(data.get("count", 3))
try:
count = int(data.get("count", 3))
except (ValueError, TypeError):
return jsonify({"message": "count muss eine ganze Zahl sein"}), 400
count = max(1, min(20, count))
if gen_type == "words":
backend/tools/md5.py
+1 -2
View File
@@ -15,8 +15,7 @@ def hash_md5():
return jsonify({"message": "Nicht autorisiert"}), 401
try:
data = request.get_json()
logger.debug(f"📩 Payload: {data}")
data = request.get_json(silent=True) or {}
password = data.get("password", "")
result = hashlib.md5(password.encode()).hexdigest()
backend/tools/notes.py
+73 -55
View File
@@ -6,24 +6,32 @@ from auth.token import verify_token
notes_blueprint = Blueprint('notes_tool', __name__)
_table_ready = False
def ensure_table():
def _ensure_table():
global _table_ready
if _table_ready:
return
conn = get_connection()
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS notes (
id INT AUTO_INCREMENT PRIMARY KEY,
user_id INT NOT NULL,
title VARCHAR(255) NOT NULL,
content TEXT,
language VARCHAR(50) DEFAULT 'text',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
)
""")
conn.commit()
cursor.close()
conn.close()
try:
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS notes (
id INT AUTO_INCREMENT PRIMARY KEY,
user_id INT NOT NULL,
title VARCHAR(255) NOT NULL,
content TEXT,
language VARCHAR(50) DEFAULT 'text',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
)
""")
conn.commit()
cursor.close()
_table_ready = True
finally:
conn.close()
@notes_blueprint.route('/api/notes', methods=['GET'])
@@ -32,16 +40,18 @@ def get_notes():
if not user:
return jsonify({"message": "Nicht autorisiert"}), 401
try:
ensure_table()
_ensure_table()
conn = get_connection()
cursor = conn.cursor(dictionary=True)
cursor.execute(
"SELECT id, title, content, language, created_at, updated_at FROM notes WHERE user_id = %s ORDER BY updated_at DESC",
(user['id'],)
)
notes = cursor.fetchall()
cursor.close()
conn.close()
try:
cursor = conn.cursor(dictionary=True)
cursor.execute(
"SELECT id, title, content, language, created_at, updated_at FROM notes WHERE user_id = %s ORDER BY updated_at DESC",
(user['id'],)
)
notes = cursor.fetchall()
cursor.close()
finally:
conn.close()
for n in notes:
if n.get('created_at'):
n['created_at'] = n['created_at'].isoformat()
@@ -59,22 +69,24 @@ def create_note():
if not user:
return jsonify({"message": "Nicht autorisiert"}), 401
try:
ensure_table()
data = request.get_json() or {}
_ensure_table()
data = request.get_json(silent=True) or {}
title = data.get("title", "Neue Notiz").strip() or "Neue Notiz"
content = data.get("content", "")
language = data.get("language", "text")
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO notes (user_id, title, content, language) VALUES (%s, %s, %s, %s)",
(user['id'], title, content, language)
)
conn.commit()
note_id = cursor.lastrowid
cursor.close()
conn.close()
try:
cursor = conn.cursor()
cursor.execute(
"INSERT INTO notes (user_id, title, content, language) VALUES (%s, %s, %s, %s)",
(user['id'], title, content, language)
)
conn.commit()
note_id = cursor.lastrowid
cursor.close()
finally:
conn.close()
logger.info(f"Notiz erstellt von {user['username']}: id={note_id}")
return jsonify({"id": note_id, "title": title, "content": content, "language": language})
@@ -89,22 +101,25 @@ def update_note(note_id):
if not user:
return jsonify({"message": "Nicht autorisiert"}), 401
try:
data = request.get_json() or {}
_ensure_table()
data = request.get_json(silent=True) or {}
title = data.get("title", "").strip() or "Neue Notiz"
content = data.get("content", "")
language = data.get("language", "text")
now = datetime.now(timezone.utc).replace(tzinfo=None)
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"UPDATE notes SET title=%s, content=%s, language=%s, updated_at=%s WHERE id=%s AND user_id=%s",
(title, content, language, now, note_id, user['id'])
)
conn.commit()
affected = cursor.rowcount
cursor.close()
conn.close()
try:
cursor = conn.cursor()
cursor.execute(
"UPDATE notes SET title=%s, content=%s, language=%s, updated_at=%s WHERE id=%s AND user_id=%s",
(title, content, language, now, note_id, user['id'])
)
conn.commit()
affected = cursor.rowcount
cursor.close()
finally:
conn.close()
if affected == 0:
return jsonify({"message": "Notiz nicht gefunden"}), 404
@@ -120,16 +135,19 @@ def delete_note(note_id):
if not user:
return jsonify({"message": "Nicht autorisiert"}), 401
try:
_ensure_table()
conn = get_connection()
cursor = conn.cursor()
cursor.execute(
"DELETE FROM notes WHERE id=%s AND user_id=%s",
(note_id, user['id'])
)
conn.commit()
affected = cursor.rowcount
cursor.close()
conn.close()
try:
cursor = conn.cursor()
cursor.execute(
"DELETE FROM notes WHERE id=%s AND user_id=%s",
(note_id, user['id'])
)
conn.commit()
affected = cursor.rowcount
cursor.close()
finally:
conn.close()
if affected == 0:
return jsonify({"message": "Notiz nicht gefunden"}), 404
backend/tools/passwordgen.py
+1 -1
View File
@@ -13,7 +13,7 @@ def generate_password():
if not user:
return jsonify({"message": "Nicht autorisiert"}), 401
try:
data = request.get_json()
data = request.get_json(silent=True) or {}
length = min(max(int(data.get("length", 16)), 8), 64)
use_uppercase = data.get("uppercase", True)
use_lowercase = data.get("lowercase", True)
backend/tools/textdiff.py
+1 -1
View File
@@ -12,7 +12,7 @@ def text_diff():
if not user:
return jsonify({"message": "Nicht autorisiert"}), 401
try:
data = request.get_json()
data = request.get_json(silent=True) or {}
text1 = data.get("text1", "")
text2 = data.get("text2", "")
backend/tools/timestamp.py
+1 -1
View File
@@ -20,7 +20,7 @@ def convert_timestamp():
if not user:
return jsonify({"message": "Nicht autorisiert"}), 401
try:
data = request.get_json()
data = request.get_json(silent=True) or {}
value = data.get("value", "").strip()
direction = data.get("direction", "unix_to_date")
backend/util/logger.py
+27 -9
View File
@@ -1,21 +1,39 @@
import logging
import os
from logging.handlers import RotatingFileHandler
os.makedirs("logs", exist_ok=True)
_LOG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "logs")
os.makedirs(_LOG_DIR, exist_ok=True)
fmt = "%(asctime)s [%(levelname)s] %(message)s"
_FMT = "%(asctime)s [%(levelname)s] %(message)s"
_formatter = logging.Formatter(_FMT)
error_handler = logging.FileHandler("logs/error.log")
error_handler.setLevel(logging.ERROR)
_MAX_BYTES = 5 * 1024 * 1024 # 5 MB per file
_BACKUP_COUNT = 3
def _rotating(filename, level):
h = RotatingFileHandler(
os.path.join(_LOG_DIR, filename),
maxBytes=_MAX_BYTES,
backupCount=_BACKUP_COUNT,
encoding="utf-8",
)
h.setLevel(level)
h.setFormatter(_formatter)
return h
_console = logging.StreamHandler()
_console.setFormatter(_formatter)
logging.basicConfig(
level=logging.INFO,
format=fmt,
handlers=[
logging.FileHandler("logs/app.log"),
error_handler,
logging.StreamHandler()
]
_rotating("app.log", logging.INFO),
_rotating("error.log", logging.ERROR),
_console,
],
)
logger = logging.getLogger("main")