807 lines
28 KiB
Python
807 lines
28 KiB
Python
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/<int:file_id>/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/<int:file_id>/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/<int:sensor_id>/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/<int:notification_id>/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/<int:notification_id>/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/<int:note_id>", 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/<int:note_id>/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/<int:note_id>/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/<int:note_id>/items/<int:item_id>", 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/<int:note_id>/items/<int:item_id>/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/<int:note_id>/items/<int:item_id>/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/<int:sensor_id>/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/<int:sensor_id>/types")
|
|
@login_required
|
|
def api_sensor_types(sensor_id):
|
|
return _proxy(lambda: sensors.sensor_types(sensor_id))
|
|
|
|
@app.route("/api/sensors/<int:sensor_id>/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/<int:sensor_id>/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)
|