import os import click from flask import Flask, jsonify, redirect, render_template, request, url_for from flask_login import ( LoginManager, current_user, login_required, login_user, logout_user, ) from sqlalchemy import func from models import Sensor, SensorReadout, User, db login_manager = LoginManager() def create_app(): app = Flask(__name__) db_path = os.path.join(os.path.dirname(__file__), "data.db") app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{db_path}" app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False app.config["SECRET_KEY"] = os.environ.get("SECRET_KEY", "dev-secret-key-change-me") db.init_app(app) login_manager.init_app(app) login_manager.login_view = "login" @login_manager.user_loader def load_user(user_id): return db.session.get(User, int(user_id)) @login_manager.unauthorized_handler def unauthorized(): # Dla API zwróć JSON zamiast przekierowania na stronę logowania. if request.path.startswith("/api/"): return jsonify({"error": "authentication required"}), 401 next_page = request.path login_url = url_for("login") if next_page and next_page != login_url: login_url += f"?next={next_page}" return redirect(login_url) def _is_safe_next(target: str | None) -> bool: return bool(target) and target.startswith("/") and not target.startswith("//") # --- Strony --- @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 = "Nieprawidłowa nazwa użytkownika lub hasło." 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") @app.route("/api/docs") @login_required def api_docs(): return render_template("docs.html") # --- API: odczyt (wymaga zalogowania, sesja z przeglądarki wystarcza) --- @app.route("/api/sensors") @login_required def api_sensors(): sensors = Sensor.query.order_by(Sensor.name).all() return jsonify([{"id": s.id, "name": s.name} for s in sensors]) @app.route("/api/sensors//types") @login_required def api_sensor_types(sensor_id): types = ( db.session.query(SensorReadout.type) .filter_by(sensor_id=sensor_id) .distinct() .order_by(SensorReadout.type) .all() ) return jsonify([t[0] for t in types]) @app.route("/api/sensors//readouts") @login_required def api_sensor_readouts(sensor_id): type_ = request.args.get("type") query = SensorReadout.query.filter_by(sensor_id=sensor_id) if type_: query = query.filter_by(type=type_) readouts = query.order_by(SensorReadout.timestamp.desc()).limit(300).all() return jsonify( [ {"value": r.value, "timestamp": r.timestamp.isoformat() + "Z"} for r in readouts ] ) @app.route("/api/sensors//readouts/hourly") @login_required def api_sensor_readouts_hourly(sensor_id): type_ = request.args.get("type") query = ( db.session.query( func.strftime("%Y-%m-%dT%H:00:00", SensorReadout.timestamp).label("hour"), func.avg(SensorReadout.value).label("avg_value"), ) .filter(SensorReadout.sensor_id == sensor_id) ) if type_: query = query.filter(SensorReadout.type == type_) rows = ( query.group_by("hour") .order_by("hour") .all() ) return jsonify( [{"timestamp": r.hour, "value": round(r.avg_value, 2)} for r in rows] ) # --- API: zapis (celowo bez logowania — wywołują je urządzenia/sensory) --- @app.route("/api/sensors", methods=["POST"]) def api_create_sensor(): data = request.get_json(silent=True) if not data or "name" not in data: return jsonify({"error": "name is required"}), 400 name = data["name"].strip() if not name: return jsonify({"error": "name cannot be empty"}), 400 if Sensor.query.filter_by(name=name).first(): return jsonify({"error": "sensor already exists"}), 409 sensor = Sensor(name=name) db.session.add(sensor) db.session.commit() return jsonify({"id": sensor.id, "name": sensor.name}), 201 @app.route("/api/sensors//readouts", methods=["POST"]) def api_create_readout(sensor_id): sensor = db.session.get(Sensor, sensor_id) if not sensor: return jsonify({"error": "sensor not found"}), 404 data = request.get_json(silent=True) if not data: return jsonify({"error": "request body is required"}), 400 missing = [f for f in ("type", "value") if f not in data] if missing: return jsonify({"error": f"missing fields: {', '.join(missing)}"}), 400 try: value = float(data["value"]) except (TypeError, ValueError): return jsonify({"error": "value must be a number"}), 400 readout = SensorReadout( sensor_id=sensor_id, type=str(data["type"]).strip(), value=value, ) db.session.add(readout) db.session.commit() return jsonify( { "id": readout.id, "sensor_id": readout.sensor_id, "type": readout.type, "value": readout.value, "timestamp": readout.timestamp.isoformat() + "Z", } ), 201 # --- Komenda do utworzenia pierwszego admina --- # Użycie: # flask --app app create-admin --username admin --password 'sekret' # ADMIN_USERNAME=admin ADMIN_PASSWORD=sekret flask --app app create-admin # flask --app app create-admin (zapyta interaktywnie) @app.cli.command("create-admin") @click.option("--username", default=None, help="Nazwa administratora") @click.option("--password", default=None, help="Hasło administratora") def create_admin(username, password): """Utwórz lub zaktualizuj konto administratora (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 db.session.commit() click.echo(f"Zaktualizowano istniejącego użytkownika '{username}' (nadano is_admin=True).") else: user = User(username=username, is_admin=True) user.set_password(password) db.session.add(user) db.session.commit() click.echo(f"Utworzono administratora '{username}'.") return app if __name__ == "__main__": app = create_app() app.run(debug=True, host='0.0.0.0')