import fcntl import os import uuid from contextlib import contextmanager from datetime import date, datetime, timedelta, timezone from pathlib import Path from zoneinfo import ZoneInfo import click from flask import ( Flask, flash, jsonify, redirect, render_template, request, send_file, url_for, ) from flask_login import current_user, login_required, login_user, logout_user from sqlalchemy import func from auth import init_login_manager from config import ( clipboard_dir, clipboard_max_file_bytes, clipboard_max_total_bytes, load_secret_key, sensors_api_url, users_database_uri, ) from sensors_client import SensorsAPIError, SensorsClient from user_models import ( ClipboardFile, Note, NoteItem, Notification, User, user_db, ) def format_bytes(value: int) -> str: size = float(value) for unit in ("B", "KB", "MB", "GB", "TB"): if size < 1024 or unit == "TB": if unit == "B": return f"{int(size)} B" return f"{size:.1f} {unit}" size /= 1024 return f"{int(size)} B" LOCAL_TZ_DEFAULT = "Europe/Warsaw" def _calendar_tz(name): try: return ZoneInfo(name) except Exception: return ZoneInfo(LOCAL_TZ_DEFAULT) def _parse_date_arg(raw): value = (raw or "").strip() if not value: return None try: return date.fromisoformat(value) except ValueError: return None def _day_bounds_utc(day: date, tz): start_local = datetime(day.year, day.month, day.day) end_local = start_local + timedelta(days=1) start_utc = start_local.replace(tzinfo=tz).astimezone(timezone.utc).replace(tzinfo=None) end_utc = end_local.replace(tzinfo=tz).astimezone(timezone.utc).replace(tzinfo=None) return start_utc, end_utc def create_app(): app = Flask(__name__) app.config["SECRET_KEY"] = load_secret_key() app.config["SQLALCHEMY_DATABASE_URI"] = users_database_uri() app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False app.config["SENSORS_API_URL"] = sensors_api_url() app.config["CLIPBOARD_DIR"] = str(clipboard_dir()) app.config["CLIPBOARD_MAX_FILE_BYTES"] = clipboard_max_file_bytes() app.config["CLIPBOARD_MAX_TOTAL_BYTES"] = clipboard_max_total_bytes() app.config["MAX_CONTENT_LENGTH"] = ( app.config["CLIPBOARD_MAX_FILE_BYTES"] + 1024 * 1024 ) Path(app.config["CLIPBOARD_DIR"]).mkdir(parents=True, exist_ok=True) app.add_template_filter(format_bytes, "filesize") user_db.init_app(app) sensors = SensorsClient(app.config["SENSORS_API_URL"]) def _login_redirect(req): login_url = url_for("login") if req.path and req.path != login_url: return url_for("login", next=req.path) return login_url def _load_user(user_id): try: return user_db.session.get(User, int(user_id)) except (TypeError, ValueError): return None init_login_manager(app, _login_redirect, _load_user) def _is_safe_next(target: str | None) -> bool: return bool(target) and target.startswith("/") and not target.startswith("//") def _proxy(call): try: status, data = call() except SensorsAPIError as exc: return jsonify({"error": str(exc)}), 502 return jsonify(data), status @app.context_processor def inject_globals(): unread_notifications = 0 if current_user.is_authenticated: unread_notifications = Notification.query.filter_by( user_id=current_user.id, is_read=False ).count() return { "main_api_url": request.url_root.rstrip("/") + "/api", "unread_notifications": unread_notifications, } # --- Pages --- @app.route("/") def index(): if current_user.is_authenticated: return redirect(url_for("home")) return redirect(url_for("login")) @app.route("/login", methods=["GET", "POST"]) def login(): if current_user.is_authenticated: return redirect(url_for("home")) error = None if request.method == "POST": username = request.form.get("username", "").strip() password = request.form.get("password", "") user = User.query.filter_by(username=username).first() if user and user.check_password(password): login_user(user) next_page = request.args.get("next") or request.form.get("next") if not _is_safe_next(next_page): next_page = url_for("home") return redirect(next_page) error = "Invalid username or password." return render_template("login.html", error=error, next=request.args.get("next", "")) @app.route("/logout") @login_required def logout(): logout_user() return redirect(url_for("login")) @app.route("/home") @login_required def home(): return render_template("home.html", user=current_user) @app.route("/sensors") @login_required def sensors_page(): return render_template("sensors.html", user=current_user) @app.route("/api/docs") @login_required def api_docs(): return render_template("docs.html", user=current_user) # --- Clipboard --- def _clipboard_path(name): return Path(app.config["CLIPBOARD_DIR"]) / name @contextmanager def _clipboard_lock(): lock_path = Path(app.config["CLIPBOARD_DIR"]) / ".clipboard.lock" with open(lock_path, "a+") as lock_file: fcntl.flock(lock_file, fcntl.LOCK_EX) try: yield finally: fcntl.flock(lock_file, fcntl.LOCK_UN) def _clipboard_file_or_404(file_id): return ClipboardFile.query.filter_by( id=file_id, user_id=current_user.id ).first_or_404() @app.route("/clipboard") @login_required def clipboard_page(): files = ( ClipboardFile.query.filter_by(user_id=current_user.id) .order_by(ClipboardFile.created_at.desc(), ClipboardFile.id.desc()) .all() ) total_used = ( user_db.session.query( func.coalesce(func.sum(ClipboardFile.size), 0) ) .filter(ClipboardFile.user_id == current_user.id) .scalar() or 0 ) max_total = app.config["CLIPBOARD_MAX_TOTAL_BYTES"] return render_template( "clipboard.html", user=current_user, files=files, total_used=total_used, max_total=max_total, remaining=max(0, max_total - total_used), max_file=app.config["CLIPBOARD_MAX_FILE_BYTES"], ) @app.errorhandler(413) def request_too_large(_error): flash( "Upload too large. Maximum file size is " f"{format_bytes(app.config['CLIPBOARD_MAX_FILE_BYTES'])}.", "error", ) return redirect(url_for("clipboard_page")) @app.route("/clipboard/upload", methods=["POST"]) @login_required def clipboard_upload(): upload = request.files.get("file") if not upload or not upload.filename: flash("Choose a file to upload.", "error") return redirect(url_for("clipboard_page")) original_name = ( os.path.basename(upload.filename.replace("\\", "/")) .strip() .replace("\x00", "") ) if not original_name: flash("The file name is invalid.", "error") return redirect(url_for("clipboard_page")) original_name = original_name[:255] directory = Path(app.config["CLIPBOARD_DIR"]) stored_name = uuid.uuid4().hex temp_path = directory / f".{stored_name}.tmp" final_path = directory / stored_name try: upload.save(str(temp_path)) size = temp_path.stat().st_size except OSError: temp_path.unlink(missing_ok=True) flash("Could not save the uploaded file.", "error") return redirect(url_for("clipboard_page")) if size <= 0: temp_path.unlink(missing_ok=True) flash("The file is empty.", "error") return redirect(url_for("clipboard_page")) max_file = app.config["CLIPBOARD_MAX_FILE_BYTES"] if size > max_file: temp_path.unlink(missing_ok=True) flash( f"File is too large. Maximum size is {format_bytes(max_file)}.", "error", ) return redirect(url_for("clipboard_page")) with _clipboard_lock(): total_used = ( user_db.session.query( func.coalesce(func.sum(ClipboardFile.size), 0) ) .filter(ClipboardFile.user_id == current_user.id) .scalar() or 0 ) max_total = app.config["CLIPBOARD_MAX_TOTAL_BYTES"] if total_used + size > max_total: temp_path.unlink(missing_ok=True) flash( f"Clipboard storage is full " f"({format_bytes(max_total)} limit).", "error", ) return redirect(url_for("clipboard_page")) try: os.replace(temp_path, final_path) user_db.session.add( ClipboardFile( user_id=current_user.id, original_name=original_name, stored_name=stored_name, size=size, content_type=( upload.mimetype or "application/octet-stream" )[:255], ) ) user_db.session.commit() except Exception: user_db.session.rollback() final_path.unlink(missing_ok=True) temp_path.unlink(missing_ok=True) flash("Could not save the uploaded file.", "error") return redirect(url_for("clipboard_page")) flash(f"Uploaded {original_name}.", "success") return redirect(url_for("clipboard_page")) @app.route("/clipboard//download") @login_required def clipboard_download(file_id): entry = _clipboard_file_or_404(file_id) path = _clipboard_path(entry.stored_name) if not path.is_file(): flash("The file is missing from the server.", "error") return redirect(url_for("clipboard_page")) return send_file( path, as_attachment=True, download_name=entry.original_name, mimetype=entry.content_type or "application/octet-stream", ) @app.route("/clipboard//delete", methods=["POST"]) @login_required def clipboard_delete(file_id): entry = _clipboard_file_or_404(file_id) name = entry.original_name path = _clipboard_path(entry.stored_name) try: path.unlink(missing_ok=True) except OSError: flash("Could not delete the file from disk.", "error") return redirect(url_for("clipboard_page")) user_db.session.delete(entry) user_db.session.commit() flash(f"Deleted {name}.", "success") return redirect(url_for("clipboard_page")) @app.route("/api/sensors//readouts/range") @login_required def api_sensor_readouts_range(sensor_id): return _proxy( lambda: sensors.sensor_readouts_range(sensor_id, request.args.to_dict()) ) @app.route("/calendar") @login_required def calendar_page(): view = (request.args.get("view") or "month").strip().lower() if view not in ("month", "week"): view = "month" tz = _calendar_tz(request.args.get("tz") or LOCAL_TZ_DEFAULT) now_local = datetime.now(tz) year = now_local.year month = now_local.month week_start = None week_end = None week_prev = None week_next = None days = [] month_weeks = None if view == "week": week_date = _parse_date_arg(request.args.get("date")) or now_local.date() week_start = week_date - timedelta(days=week_date.weekday()) week_end = week_start + timedelta(days=6) week_prev = week_start - timedelta(days=7) week_next = week_start + timedelta(days=7) day_list = [week_start + timedelta(days=i) for i in range(7)] range_from = _day_bounds_utc(week_start, tz)[0] range_to = _day_bounds_utc(week_end + timedelta(days=1), tz)[0] else: year = int(request.args.get("year") or now_local.year) month = int(request.args.get("month") or now_local.month) if not 1 <= month <= 12: month = 1 first_day = date(year, month, 1) next_year = year + 1 if month == 12 else year next_month = 1 if month == 12 else month + 1 last_day = date(next_year, next_month, 1) - timedelta(days=1) day_list = [] current = first_day while current <= last_day: day_list.append(current) current += timedelta(days=1) range_from, _ = _day_bounds_utc(first_day, tz) range_to, _ = _day_bounds_utc(last_day, tz) week_start = first_day - timedelta(days=first_day.weekday()) week_end = last_day readouts_by_day = {day: [] for day in day_list} sensor_names = {} try: listed = sensors.list_sensors() for sensor in listed[1] or []: sensor_id = int(sensor["id"]) sensor_names[sensor_id] = sensor["name"] status, payload = sensors.sensor_readouts_range( sensor_id, {"from": range_from.isoformat(), "to": range_to.isoformat()}, ) if status != 200: continue for item in payload or []: raw_ts = (item.get("timestamp") or "").rstrip("Z") ts = datetime.fromisoformat(raw_ts) if raw_ts else None if ts is None: continue local_day = ts.replace(tzinfo=timezone.utc).astimezone(tz).date() if local_day in readouts_by_day: readouts_by_day[local_day].append( { "id": item.get("id"), "sensor_id": item.get("sensor_id") or sensor_id, "type": item.get("type"), "value": item.get("value"), "timestamp": ts, } ) except SensorsAPIError: pass days = [ {"date": day, "readouts": readouts_by_day.get(day, [])} for day in day_list ] month_weeks = None if view == "month": first_monday = week_start last_sunday = week_end + timedelta(days=6 - week_end.weekday()) end_exclusive = last_sunday + timedelta(days=1) weeks = [] week = [] current = first_monday while current < end_exclusive: week.append( { "date": current, "in_month": current.month == month, "readouts": readouts_by_day.get(current, []), } ) if current.weekday() == 6: weeks.append(week) week = [] current += timedelta(days=1) if week: weeks.append(week) month_weeks = weeks if month == 1: prev_year, prev_month = year - 1, 12 else: prev_year, prev_month = year, month - 1 if month == 12: next_year, next_month = year + 1, 1 else: next_year, next_month = year, month + 1 month_names = ( "January", "February", "March", "April", "May", "June", "July", "August", "September", "October", "November", "December", ) month_label = f"{month_names[month - 1]} {year}" return render_template( "calendar.html", user=current_user, view=view, year=year, month=month, week_start=week_start, week_end=week_end, days=days, month_weeks=month_weeks, sensor_names=sensor_names, prev_year=prev_year, prev_month=prev_month, next_year=next_year, next_month=next_month, week_prev=week_prev, week_next=week_next, today=now_local.date(), ) # --- Notifications (stored in main_app.db) --- @app.route("/notifications") @login_required def notifications_page(): notifications = ( Notification.query.filter_by(user_id=current_user.id) .order_by( Notification.is_read.asc(), Notification.created_at.desc(), Notification.id.desc(), ) .all() ) return render_template( "notifications.html", user=current_user, notifications=notifications ) @app.route("/notifications//read") @login_required def notification_read(notification_id): notification = Notification.query.filter_by( id=notification_id, user_id=current_user.id ).first_or_404() if not notification.is_read: notification.mark_read() user_db.session.commit() return redirect( url_for( "notifications_page", _anchor=f"notification-{notification.id}", ) ) @app.route("/notifications//delete", methods=["POST"]) @login_required def notification_delete(notification_id): notification = Notification.query.filter_by( id=notification_id, user_id=current_user.id ).first_or_404() user_db.session.delete(notification) user_db.session.commit() flash("Deleted notification.", "success") return redirect(url_for("notifications_page")) # --- Notes (stored in main_app.db) --- def _note_or_404(note_id): return Note.query.filter_by( id=note_id, user_id=current_user.id ).first_or_404() def _note_redirect(note_id): return redirect( url_for("notes_page", _anchor=f"note-{note_id}") ) @app.route("/notes") @login_required def notes_page(): notes = ( Note.query.filter_by(user_id=current_user.id) .order_by(Note.updated_at.desc(), Note.id.desc()) .all() ) return render_template("notes.html", user=current_user, notes=notes) @app.route("/notes", methods=["POST"]) @login_required def notes_create(): title = request.form.get("title", "").strip() content = request.form.get("content", "").strip() if not title: return redirect(url_for("notes_page")) note = Note( user_id=current_user.id, title=title[:200], content=content[:2000], ) user_db.session.add(note) user_db.session.commit() return _note_redirect(note.id) @app.route("/notes/autosave", methods=["POST"]) @login_required def notes_autosave(): data = request.get_json(silent=True) if not data: return jsonify({"error": "json required"}), 400 note_id = data.get("id") title = (data.get("title") or "").strip()[:200] content = (data.get("content") or "").strip()[:2000] if not title: return jsonify({"error": "title required"}), 400 if note_id: note = _note_or_404(int(note_id)) note.title = title note.content = content note.touch() user_db.session.commit() return jsonify({"id": note.id, "status": "saved"}) note = Note( user_id=current_user.id, title=title, content=content, ) user_db.session.add(note) user_db.session.commit() return jsonify({"id": note.id, "status": "created"}) @app.route("/notes/", methods=["POST"]) @login_required def notes_update(note_id): note = _note_or_404(note_id) title = request.form.get("title", "").strip() content = request.form.get("content", "").strip() if title: note.title = title[:200] note.content = content[:2000] note.touch() user_db.session.commit() return _note_redirect(note.id) @app.route("/notes//delete", methods=["POST"]) @login_required def notes_delete(note_id): note = _note_or_404(note_id) user_db.session.delete(note) user_db.session.commit() return redirect(url_for("notes_page")) @app.route("/notes//items", methods=["POST"]) @login_required def notes_add_item(note_id): note = _note_or_404(note_id) text = request.form.get("text", "").strip() if text: item = NoteItem(note=note, text=text[:255]) note.touch() user_db.session.add(item) user_db.session.commit() return _note_redirect(note.id) @app.route( "/notes//items/", methods=["POST"] ) @login_required def notes_update_item(note_id, item_id): note = _note_or_404(note_id) item = NoteItem.query.filter_by(id=item_id, note_id=note.id).first_or_404() if request.is_json: data = request.get_json(silent=True) or {} text = (data.get("text") or "").strip() else: text = request.form.get("text", "").strip() if text: item.text = text[:255] note.touch() user_db.session.commit() if request.is_json: return jsonify({"id": item.id, "status": "saved"}) return _note_redirect(note.id) @app.route( "/notes//items//toggle", methods=["POST"] ) @login_required def notes_toggle_item(note_id, item_id): note = _note_or_404(note_id) item = NoteItem.query.filter_by(id=item_id, note_id=note.id).first_or_404() item.is_done = request.form.get("is_done") in {"1", "true", "on", "yes"} note.touch() user_db.session.commit() return _note_redirect(note.id) @app.route( "/notes//items//delete", methods=["POST"] ) @login_required def notes_delete_item(note_id, item_id): note = _note_or_404(note_id) item = NoteItem.query.filter_by(id=item_id, note_id=note.id).first_or_404() user_db.session.delete(item) note.touch() user_db.session.commit() return _note_redirect(note.id) # --- API: proxied to the sensors service --- # The browser only talks to this app; the sensors service owns the # sensor database and the actual API implementation. @app.route("/api/sensors") def api_sensors(): return _proxy(sensors.list_sensors) @app.route("/api/sensors", methods=["POST"]) def api_create_sensor(): return _proxy(lambda: sensors.create_sensor(request.get_json(silent=True))) @app.route("/api/sensors//readouts", methods=["POST"]) def api_create_readout(sensor_id): return _proxy( lambda: sensors.create_readout(sensor_id, request.get_json(silent=True)) ) @app.route("/api/sensors//types") @login_required def api_sensor_types(sensor_id): return _proxy(lambda: sensors.sensor_types(sensor_id)) @app.route("/api/sensors//readouts") @login_required def api_sensor_readouts(sensor_id): return _proxy( lambda: sensors.sensor_readouts( sensor_id, {"type": request.args.get("type")} ) ) @app.route("/api/sensors//readouts/hourly") @login_required def api_sensor_readouts_hourly(sensor_id): return _proxy( lambda: sensors.sensor_readouts_hourly(sensor_id, request.args.to_dict()) ) # --- Command for creating the first admin --- # Usage: # flask --app app create-admin --username admin --password 'secret' # ADMIN_USERNAME=admin ADMIN_PASSWORD=secret flask --app app create-admin @app.cli.command("create-admin") @click.option("--username", default=None, help="Administrator username") @click.option("--password", default=None, help="Administrator password") def create_admin(username, password): """Create or update an administrator account (is_admin=True).""" username = username or os.environ.get("ADMIN_USERNAME") if not username: username = click.prompt("Admin username") username = username.strip() if not username: raise click.ClickException("Username cannot be empty.") password = password or os.environ.get("ADMIN_PASSWORD") if not password: password = click.prompt("Admin password", hide_input=True, confirmation_prompt=True) if not password: raise click.ClickException("Password cannot be empty.") user = User.query.filter_by(username=username).first() if user: user.set_password(password) user.is_admin = True user_db.session.commit() click.echo(f"Updated existing user '{username}' (granted is_admin=True).") else: user = User(username=username, is_admin=True) user.set_password(password) user_db.session.add(user) user_db.session.commit() click.echo(f"Created administrator '{username}'.") @app.cli.command("create-notification") @click.option("--username", default=None, help="Recipient username") @click.option("--all", "all_users", is_flag=True, help="Send to all users") @click.option("--message", default=None, help="Notification text") def create_notification(username, all_users, message): """Create a notification for one user or for all users.""" if all_users and username: raise click.ClickException("Use either --username or --all, not both.") if not all_users and not username: raise click.ClickException("Provide --username or --all.") message = (message or "").strip() if not message: message = click.prompt("Message").strip() if not message: raise click.ClickException("Message cannot be empty.") message = message[:500] if all_users: users = User.query.all() else: user = User.query.filter_by(username=username).first() if not user: raise click.ClickException(f"User '{username}' not found.") users = [user] for user in users: user_db.session.add(Notification(user_id=user.id, message=message)) user_db.session.commit() click.echo(f"Created {len(users)} notification(s).") return app if __name__ == "__main__": app = create_app() app.run(debug=True, host='0.0.0.0', port=8000)