272 lines
8.5 KiB
Python
272 lines
8.5 KiB
Python
from datetime import date
|
|
|
|
from flask import Flask, render_template, redirect, url_for, request, flash, jsonify
|
|
from flask_login import (
|
|
LoginManager,
|
|
login_user,
|
|
logout_user,
|
|
login_required,
|
|
current_user,
|
|
)
|
|
|
|
from config import Config
|
|
from models import db, User, Trip, Stop, VERKEHRSMITTEL_OPTIONEN
|
|
from stops_import import normalize_name, register_cli
|
|
|
|
login_manager = LoginManager()
|
|
login_manager.login_view = "login"
|
|
|
|
|
|
def create_app():
|
|
app = Flask(__name__)
|
|
app.config.from_object(Config)
|
|
|
|
db.init_app(app)
|
|
login_manager.init_app(app)
|
|
|
|
with app.app_context():
|
|
db.create_all()
|
|
|
|
register_routes(app)
|
|
register_cli(app)
|
|
return app
|
|
|
|
|
|
@login_manager.user_loader
|
|
def load_user(user_id):
|
|
return User.query.get(int(user_id))
|
|
|
|
|
|
def register_routes(app):
|
|
@app.route("/")
|
|
def index():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for("dashboard"))
|
|
return redirect(url_for("login"))
|
|
|
|
@app.route("/register", methods=["GET", "POST"])
|
|
def register():
|
|
if request.method == "POST":
|
|
username = request.form["username"].strip()
|
|
email = request.form["email"].strip().lower()
|
|
password = request.form["password"]
|
|
|
|
if User.query.filter(
|
|
(User.username == username) | (User.email == email)
|
|
).first():
|
|
flash("Benutzername oder E-Mail bereits vergeben.", "error")
|
|
return redirect(url_for("register"))
|
|
|
|
user = User(username=username, email=email)
|
|
user.set_password(password)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
|
|
login_user(user)
|
|
return redirect(url_for("dashboard"))
|
|
|
|
return render_template("register.html")
|
|
|
|
@app.route("/login", methods=["GET", "POST"])
|
|
def login():
|
|
if request.method == "POST":
|
|
username = request.form["username"].strip()
|
|
password = request.form["password"]
|
|
|
|
user = User.query.filter_by(username=username).first()
|
|
if user is None or not user.check_password(password):
|
|
flash("Login fehlgeschlagen. Bitte prüfe deine Daten.", "error")
|
|
return redirect(url_for("login"))
|
|
|
|
login_user(user)
|
|
return redirect(url_for("dashboard"))
|
|
|
|
return render_template("login.html")
|
|
|
|
@app.route("/logout")
|
|
@login_required
|
|
def logout():
|
|
logout_user()
|
|
return redirect(url_for("login"))
|
|
|
|
@app.route("/dashboard")
|
|
@login_required
|
|
def dashboard():
|
|
fahrten = (
|
|
Trip.query.filter_by(user_id=current_user.id)
|
|
.order_by(Trip.trip_date.desc())
|
|
.all()
|
|
)
|
|
return render_template("dashboard.html", fahrten=fahrten)
|
|
|
|
@app.route("/api/stops")
|
|
@login_required
|
|
def api_stops():
|
|
"""Autocomplete for the origin/destination fields."""
|
|
query = request.args.get("q", "").strip()
|
|
if len(query) < 2:
|
|
return jsonify([])
|
|
|
|
needle = normalize_name(query).replace("\\", "\\\\")
|
|
needle = needle.replace("%", r"\%").replace("_", r"\_")
|
|
order = (db.func.length(Stop.name), Stop.name)
|
|
|
|
results = (
|
|
Stop.query.filter(Stop.name_normalized.like(needle + "%", escape="\\"))
|
|
.order_by(*order)
|
|
.limit(8)
|
|
.all()
|
|
)
|
|
if len(results) < 8:
|
|
seen = {stop.id for stop in results}
|
|
for stop in (
|
|
Stop.query.filter(
|
|
Stop.name_normalized.like("%" + needle + "%", escape="\\")
|
|
)
|
|
.order_by(*order)
|
|
.limit(20)
|
|
.all()
|
|
):
|
|
if stop.id not in seen:
|
|
results.append(stop)
|
|
if len(results) >= 8:
|
|
break
|
|
|
|
return jsonify(
|
|
[
|
|
{
|
|
"id": stop.id,
|
|
"name": stop.name,
|
|
"type": stop.stop_type,
|
|
"municipality": stop.municipality,
|
|
"lat": stop.latitude,
|
|
"lon": stop.longitude,
|
|
}
|
|
for stop in results
|
|
]
|
|
)
|
|
|
|
def _line_sort_key(ref):
|
|
head = ""
|
|
i = 0
|
|
while i < len(ref) and not ref[i].isdigit():
|
|
head += ref[i]
|
|
i += 1
|
|
num = ref[i:]
|
|
digits = ""
|
|
while num and num[0].isdigit():
|
|
digits += num[0]
|
|
num = num[1:]
|
|
return (head.lower(), int(digits) if digits else -1, num.lower())
|
|
|
|
@app.route("/api/lines")
|
|
@login_required
|
|
def api_lines():
|
|
"""Line suggestions for the "Linie" field, scoped to the chosen stops."""
|
|
def stop_lines(param):
|
|
stop = db.session.get(Stop, int(param)) if (param or "").isdigit() else None
|
|
return set(stop.line_list()) if stop else set()
|
|
|
|
origin = stop_lines(request.args.get("origin_stop_id"))
|
|
destination = stop_lines(request.args.get("destination_stop_id"))
|
|
candidates = origin | destination
|
|
if not candidates:
|
|
return jsonify([])
|
|
|
|
needle = request.args.get("q", "").strip().lower()
|
|
both = origin & destination
|
|
refs = sorted(
|
|
(r for r in candidates if not needle or r.lower().startswith(needle)),
|
|
key=lambda r: (r not in both, _line_sort_key(r)),
|
|
)
|
|
return jsonify(
|
|
[{"ref": r, "both": r in both} for r in refs[:20]]
|
|
)
|
|
|
|
def parse_trip_date(raw_value):
|
|
raw_value = (raw_value or "").strip()
|
|
try:
|
|
return date.fromisoformat(raw_value) if raw_value else date.today()
|
|
except ValueError:
|
|
return date.today()
|
|
|
|
def resolve_stop_id(raw_id):
|
|
"""Return a valid Stop.id for a submitted value, or None (free text)."""
|
|
if not raw_id:
|
|
return None
|
|
try:
|
|
stop = db.session.get(Stop, int(raw_id))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return stop.id if stop is not None else None
|
|
|
|
def fill_trip_from_form(fahrt):
|
|
fahrt.transport_mode = request.form["transport_mode"]
|
|
fahrt.line = request.form.get("line", "").strip()
|
|
fahrt.origin = request.form["origin"].strip()
|
|
fahrt.origin_stop_id = resolve_stop_id(request.form.get("origin_stop_id"))
|
|
fahrt.destination = request.form["destination"].strip()
|
|
fahrt.destination_stop_id = resolve_stop_id(
|
|
request.form.get("destination_stop_id")
|
|
)
|
|
fahrt.trip_date = parse_trip_date(request.form.get("trip_date"))
|
|
fahrt.rating = int(request.form["rating"])
|
|
fahrt.comment = request.form.get("comment", "").strip()
|
|
|
|
@app.route("/fahrt/neu", methods=["GET", "POST"])
|
|
@login_required
|
|
def neue_fahrt():
|
|
if request.method == "POST":
|
|
fahrt = Trip(user_id=current_user.id)
|
|
fill_trip_from_form(fahrt)
|
|
db.session.add(fahrt)
|
|
db.session.commit()
|
|
flash("Fahrt gespeichert.", "success")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
return render_template(
|
|
"add_entry.html",
|
|
verkehrsmittel_optionen=VERKEHRSMITTEL_OPTIONEN,
|
|
heute=date.today().isoformat(),
|
|
fahrt=None,
|
|
)
|
|
|
|
@app.route("/fahrt/<int:fahrt_id>/bearbeiten", methods=["GET", "POST"])
|
|
@login_required
|
|
def fahrt_bearbeiten(fahrt_id):
|
|
fahrt = Trip.query.get_or_404(fahrt_id)
|
|
if fahrt.user_id != current_user.id:
|
|
flash("Nicht erlaubt.", "error")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
if request.method == "POST":
|
|
fill_trip_from_form(fahrt)
|
|
db.session.commit()
|
|
flash("Fahrt aktualisiert.", "success")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
return render_template(
|
|
"add_entry.html",
|
|
verkehrsmittel_optionen=VERKEHRSMITTEL_OPTIONEN,
|
|
heute=fahrt.trip_date.isoformat(),
|
|
fahrt=fahrt,
|
|
)
|
|
|
|
@app.route("/fahrt/<int:fahrt_id>/loeschen", methods=["POST"])
|
|
@login_required
|
|
def fahrt_loeschen(fahrt_id):
|
|
fahrt = Trip.query.get_or_404(fahrt_id)
|
|
if fahrt.user_id != current_user.id:
|
|
flash("Nicht erlaubt.", "error")
|
|
return redirect(url_for("dashboard"))
|
|
db.session.delete(fahrt)
|
|
db.session.commit()
|
|
flash("Fahrt gelöscht.", "success")
|
|
return redirect(url_for("dashboard"))
|
|
|
|
|
|
app = create_app()
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=5000, debug=True)
|