Upozornění:
Veškerý zde vystavený kód je mým autorským dílem a je zveřejněn výhradně za účelem ukázky stylu mé práce.
Kód slouží pouze k nahlédnutí – jakékoliv kopírování, šíření, úpravy či jiné využití bez mého výslovného souhlasu nejsou povoleny.
Zůstává mým duševním vlastnictvím ve smyslu autorského zákona.
Děkuji za pochopení.
← Zpět na seznam
RoSaaRi (Python)
Navrh, vyvinul a nasadil jsem vlastní aplikaci, kterou používáme v práci. Je dost dobrá, vím co je potřeba, jsem řidič sanitky :)
Jj, omlouvám se, ale ještě to mám namrskané vše v jednom kódu. Tohle je verze na které stále pracuji.
from flask import Flask, request, jsonify, send_from_directory
from datetime import datetime, timedelta, date
import sqlite3
import os
import json
import secrets
import smtplib
from email.message import EmailMessage
from dotenv import load_dotenv
import bcrypt
import csv
from itsdangerous import URLSafeTimedSerializer, BadSignature, SignatureExpired
from functools import wraps
from zoneinfo import ZoneInfo
from werkzeug.exceptions import HTTPException
load_dotenv()
app = Flask(__name__, instance_relative_config=True)
os.makedirs(app.instance_path, exist_ok=True)
DB_NAME = os.path.join(app.instance_path, "zaznamy.db")
app.config["SECRET_KEY"] = os.getenv("SECRET_KEY")
if not app.config["SECRET_KEY"]:
raise RuntimeError("Chybí SECRET_KEY v .env")
serializer = URLSafeTimedSerializer(app.config["SECRET_KEY"])
TOKEN_MAX_AGE = 60 * 60 * 48 # 48 hodin
RESET_TOKEN_MAX_AGE = 60 * 30 # 30 minut
APP_URL = os.getenv("APP_URL", "http://localhost:8050")
def get_db():
conn = sqlite3.connect(DB_NAME, timeout=5)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=5000")
conn.execute("PRAGMA foreign_keys=ON")
return conn
def init_db():
with get_db() as conn:
c = conn.cursor()
c.execute('''
CREATE TABLE IF NOT EXISTS zaznamy (
id TEXT PRIMARY KEY,
typ TEXT,
kod INTEGER,
cas_nalozeni TEXT,
mesto_nalozeni TEXT,
poznamka TEXT,
cas_vylozeni TEXT,
mesto_vylozeni TEXT,
tachometr TEXT,
ujeto_km INTEGER,
smena TEXT,
datum TEXT,
ridic TEXT,
vozidlo TEXT,
personal_id TEXT,
pojistovna TEXT,
ulozeno TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
c.execute('''
CREATE TABLE IF NOT EXISTS karta_vozidla (
id INTEGER PRIMARY KEY AUTOINCREMENT,
datum TEXT,
poradove_cislo INTEGER,
vozidlo TEXT,
smena TEXT,
ujeto_km INTEGER,
konec_km INTEGER,
natankovano_l REAL,
stav_v_nadrzi_l REAL,
pkzt INTEGER DEFAULT 0,
poznamka TEXT,
ridic TEXT,
zapsano TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
c.execute("""
CREATE TABLE IF NOT EXISTS mesicni_souhrny_ridicu (
id INTEGER PRIMARY KEY AUTOINCREMENT,
rok_mesic TEXT NOT NULL,
ridic TEXT NOT NULL,
ujeto_km INTEGER DEFAULT 0,
vytvoril TEXT,
vytvoreno TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
poznamka TEXT,
UNIQUE (rok_mesic, ridic)
)
""")
c.execute("""
CREATE TABLE IF NOT EXISTS mesicni_souhrny_vozidel (
id INTEGER PRIMARY KEY AUTOINCREMENT,
rok_mesic TEXT NOT NULL,
vozidlo TEXT NOT NULL,
ujeto_km INTEGER DEFAULT 0,
natankovano_l REAL DEFAULT 0,
spotreba_l_100km REAL,
start_km INTEGER,
konec_km INTEGER,
vytvoril TEXT,
vytvoreno TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
poznamka TEXT,
UNIQUE (rok_mesic, vozidlo)
)
""")
c.execute('''
CREATE TABLE IF NOT EXISTS uzivatele (
username TEXT PRIMARY KEY,
heslo TEXT,
ridic TEXT,
vozidlo TEXT,
role TEXT DEFAULT 'ridic',
email TEXT,
sluzeb_telefon TEXT
)
''')
c.execute('''
CREATE TABLE IF NOT EXISTS vozidla (
spz TEXT PRIMARY KEY,
spotreba REAL,
stav TEXT DEFAULT 'volne',
vytvoreno TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
vyrazeno TIMESTAMP
)
''')
# Zapis ujetych km
c.execute("PRAGMA table_info(zaznamy)")
zaznamy_sloupce = {row[1] for row in c.fetchall()}
if "ujeto_km" not in zaznamy_sloupce:
c.execute("ALTER TABLE zaznamy ADD COLUMN ujeto_km INTEGER")
c.execute("PRAGMA table_info(karta_vozidla)")
karta_sloupce = {row[1] for row in c.fetchall()}
if "pkzt" not in karta_sloupce:
c.execute("ALTER TABLE karta_vozidla ADD COLUMN pkzt INTEGER DEFAULT 0")
c.execute("PRAGMA table_info(vozidla)")
vozidla_sloupce = {row[1] for row in c.fetchall()}
if "vytvoreno" not in vozidla_sloupce:
c.execute("""
ALTER TABLE vozidla
ADD COLUMN vytvoreno TIMESTAMP
""")
if "vyrazeno" not in vozidla_sloupce:
c.execute("""
ALTER TABLE vozidla
ADD COLUMN vyrazeno TIMESTAMP
""")
# Migrace tabulky uzivatele
c.execute("PRAGMA table_info(uzivatele)")
uzivatele_sloupce = {row[1] for row in c.fetchall()}
for nazev, definice in {
"email": "TEXT",
"sluzeb_telefon": "TEXT"
}.items():
if nazev not in uzivatele_sloupce:
c.execute(f"ALTER TABLE uzivatele ADD COLUMN {nazev} {definice}")
conn.commit()
def hash_hesla(heslo):
return bcrypt.hashpw(heslo.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
def over_heslo(heslo, hash_z_db):
if not heslo or not hash_z_db:
return False
return bcrypt.checkpw(
heslo.encode("utf-8"),
hash_z_db.encode("utf-8")
)
def vytvor_token(username, role, ridic, vozidlo):
return serializer.dumps({
"username": username,
"role": role or "ridic",
"ridic": ridic,
"vozidlo": vozidlo
})
def vytvor_reset_token(username, email):
return serializer.dumps({
"type": "password_reset",
"username": username,
"email": email
})
def nacti_token():
auth = request.headers.get("Authorization", "")
if not auth.startswith("Bearer "):
return None
token = auth.replace("Bearer ", "", 1)
try:
return serializer.loads(token, max_age=TOKEN_MAX_AGE)
except SignatureExpired:
return None
except BadSignature:
return None
def login_required(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
user = nacti_token()
if not user:
return jsonify({"error": "Neoprávněný přístup"}), 401
request.user = user
return fn(*args, **kwargs)
return wrapper
def admin_required(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
user = nacti_token()
if not user:
return jsonify({"error": "Neoprávněný přístup"}), 401
if user.get("role") != "admin":
return jsonify({"error": "Nedostatečná oprávnění"}), 403
request.user = user
return fn(*args, **kwargs)
return wrapper
def zaloguj_udalost(typ, uzivatel, ip_adresa):
zaznam = {
"typ": typ,
"uzivatel": uzivatel,
"cas": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"ip": ip_adresa
}
file_exists = os.path.exists("logs.csv")
with open("logs.csv", mode="a", newline="", encoding="utf-8") as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=["typ", "uzivatel", "cas", "ip"], delimiter=";")
if not file_exists:
writer.writeheader()
writer.writerow(zaznam)
def get_smena_a_km(vozidlo):
today = datetime.now(ZoneInfo("Europe/Prague")).strftime("%d.%m.%Y")
with get_db() as conn:
c = conn.cursor()
c.execute("SELECT konec_km FROM karta_vozidla WHERE vozidlo = ? ORDER BY datum DESC, id DESC LIMIT 1", (vozidlo,))
result = c.fetchone()
start_km = result[0] if result and result[0] else 0
return {"datum": today, "start_km": start_km}
# === Funkce pro odeslání e-mailu s volitelnými přílohami ===
def odesli_email(prijemce, predmet, zprava, prilohy=None):
SMTP_SERVER = "smtp.gmail.com"
SMTP_PORT = 587
SMTP_USER = os.getenv("SMTP_USER")
SMTP_PASSWORD = os.getenv("SMTP_PASSWORD")
try:
msg = EmailMessage()
msg['Subject'] = predmet
msg['From'] = SMTP_USER
msg['To'] = prijemce
msg.set_content(zprava)
if prilohy:
for pril in prilohy:
with open(pril, 'rb') as f:
file_data = f.read()
file_name = os.path.basename(pril)
msg.add_attachment(file_data, maintype='application', subtype='octet-stream', filename=file_name)
with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
server.starttls()
server.login(SMTP_USER, SMTP_PASSWORD)
server.send_message(msg)
print(f"[EMAIL] E-mail odeslán na {prijemce}")
return True
except Exception as e:
print(f"[EMAIL] Chyba při odesílání na {prijemce}: {e}")
return False
@app.route("/api/vozidla/")
@login_required
def get_spotreba_vozidla(spz):
with get_db() as conn:
c = conn.cursor()
c.execute("SELECT spotreba FROM vozidla WHERE spz = ?", (spz,))
row = c.fetchone()
if row:
return jsonify({"spotreba": row[0]})
else:
return jsonify({"error": "Vozidlo nenalezeno"}), 404
@app.route("/api/login", methods=["POST"])
def login():
data = request.get_json()
username = data.get("username")
password = data.get("password")
with get_db() as conn:
c = conn.cursor()
c.execute("SELECT heslo, ridic, vozidlo, role FROM uzivatele WHERE username = ?", (username,))
row = c.fetchone()
if not row:
return jsonify({"success": False, "msg": "Uživatel nenalezen"}), 401
heslo_db, ridic, vozidlo, role = row
if not over_heslo(password, heslo_db):
return jsonify({"success": False, "msg": "Neplatné heslo"}), 401
smena_info = get_smena_a_km(vozidlo)
zaloguj_udalost("prihlaseni", username, request.remote_addr)
token = vytvor_token(username, role, ridic, vozidlo)
return jsonify({
"success": True,
"token": token,
"ridic": ridic,
"vozidlo": vozidlo,
"datum": smena_info["datum"],
"start_km": smena_info["start_km"],
"role": role or "ridic"
})
@app.route("/api/poradove_cislo")
@login_required
def get_poradove_cislo():
vozidlo = request.args.get("vozidlo") or request.user.get("vozidlo")
today = datetime.now(ZoneInfo("Europe/Prague")).strftime("%Y-%m-%d")
month_prefix = today[:7]
with get_db() as conn:
c = conn.cursor()
c.execute("""
SELECT MAX(poradove_cislo)
FROM karta_vozidla
WHERE vozidlo = ? AND datum LIKE ?
""", (vozidlo, f"{month_prefix}%"))
posledni = c.fetchone()[0] or 0
return jsonify({
"poradove_cislo": posledni + 1,
"datum_iso": today
})
@app.route("/api/zaznamy", methods=["POST"])
@login_required
def prijmi_zaznamy():
user = request.user
data = request.get_json()
if not isinstance(data, list):
return jsonify({"error": "Data musí být seznam záznamů"}), 400
with get_db() as conn:
c = conn.cursor()
for z in data:
try:
c.execute('''
INSERT INTO zaznamy (
id, typ, kod, cas_nalozeni, mesto_nalozeni, poznamka,
cas_vylozeni, mesto_vylozeni, tachometr, ujeto_km, smena, datum,
ridic, vozidlo, personal_id, pojistovna
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''', (
z.get("id"), z.get("typ"), z.get("kod"), z.get("cas_nalozeni"),
z.get("mesto_nalozeni"), z.get("poznamka"), z.get("cas_vylozeni"),
z.get("mesto_vylozeni"), z.get("tachometr"), z.get("ujeto_km"), z.get("smena"),
z.get("datum"), user.get("ridic"), user.get("vozidlo"),
z.get("personal_id"), z.get("pojistovna")))
except sqlite3.IntegrityError:
pass
conn.commit()
return jsonify({"status": "OK", "count": len(data)})
@app.route("/api/karta", methods=["POST"])
@login_required
def uloz_kartu():
z = request.get_json()
if not isinstance(z, dict):
return jsonify({"error": "Chybná data"}), 400
datum = z.get("datum")
vozidlo = z.get("vozidlo")
poradove = z.get("poradove_cislo")
if not datum or not vozidlo:
return jsonify({"error": "Chybí datum nebo vozidlo"}), 400
with get_db() as conn:
c = conn.cursor()
if not poradove:
month_prefix = datum[:7]
c.execute("""
SELECT MAX(poradove_cislo)
FROM karta_vozidla
WHERE vozidlo = ? AND datum LIKE ?
""", (vozidlo, f"{month_prefix}%"))
poradove = (c.fetchone()[0] or 0) + 1
c.execute("""
INSERT INTO karta_vozidla (
datum, poradove_cislo, vozidlo, smena, ujeto_km, konec_km,
natankovano_l, stav_v_nadrzi_l, pkzt, poznamka, ridic
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
datum, poradove, vozidlo, z.get("smena"), z.get("ujeto_km"),
z.get("konec_km"), z.get("natankovano_l"), z.get("stav_v_nadrzi_l"),
z.get("pkzt", 0), z.get("poznamka"), z.get("ridic")
))
conn.commit()
return jsonify({"status": "Karta uložena", "poradove": poradove}), 200
@app.route("/tabulky")
@login_required
def get_tabulky():
with get_db() as conn:
c = conn.cursor()
c.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")
tabulky = [row[0] for row in c.fetchall()]
return jsonify(tabulky)
@app.route("/filtry")
@login_required
def get_filtry():
tabulka = request.args.get("tabulka")
if tabulka not in ["zaznamy", "karta_vozidla", "uzivatele", "vozidla", "mesicni_souhrny_ridicu", "mesicni_souhrny_vozidel"]:
return jsonify({"error": "Neznámá tabulka"}), 400
with get_db() as conn:
c = conn.cursor()
try:
c.execute(f"SELECT DISTINCT ridic FROM {tabulka} WHERE ridic IS NOT NULL")
jmena = sorted(set(row[0] for row in c.fetchall() if row[0]))
except sqlite3.OperationalError:
jmena = []
try:
if tabulka == "vozidla":
c.execute("SELECT spz FROM vozidla WHERE stav = 'volne'")
spz = sorted(set(row[0] for row in c.fetchall()))
else:
c.execute(f"SELECT DISTINCT vozidlo FROM {tabulka} WHERE vozidlo IS NOT NULL")
spz = sorted(set(str(row[0]) for row in c.fetchall() if row[0]))
except sqlite3.OperationalError:
spz = []
return jsonify({"spz": spz, "jmeno": jmena})
@app.route("/api/vozidla", methods=["POST"])
@admin_required
def pridej_vozidlo():
data = request.get_json() or {}
spz = (data.get("spz") or "").strip()
spotreba = data.get("spotreba")
stav = data.get("stav") or "volne"
start_km = data.get("start_km")
stav_v_nadrzi_l = data.get("stav_v_nadrzi_l")
if not spz:
return jsonify({"error": "Chybí SPZ"}), 400
if stav not in ["volne", "obsazene", "servis", "vyrazeno"]:
return jsonify({"error": "Neplatný stav vozidla"}), 400
try:
spotreba = float(spotreba) if spotreba not in [None, ""] else None
except ValueError:
return jsonify({"error": "Spotřeba musí být číslo"}), 400
try:
with get_db() as conn:
c = conn.cursor()
c.execute("""
INSERT INTO vozidla (spz, spotreba, stav)
VALUES (?, ?, ?)
""", (spz, spotreba, stav))
if start_km not in [None, ""] or stav_v_nadrzi_l not in [None, ""]:
c.execute("""
INSERT INTO karta_vozidla (
datum, poradove_cislo, vozidlo, smena, ujeto_km, konec_km,
natankovano_l, stav_v_nadrzi_l, pkzt, poznamka, ridic
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
datetime.now(ZoneInfo("Europe/Prague")).strftime("%Y-%m-%d"),
0,
spz,
"START",
0,
int(start_km) if start_km not in [None, ""] else None,
0,
float(stav_v_nadrzi_l) if stav_v_nadrzi_l not in [None, ""] else None,
0,
"Počáteční stav vozidla",
request.user.get("ridic")
))
conn.commit()
except sqlite3.IntegrityError:
return jsonify({"error": "Vozidlo s touto SPZ už existuje"}), 409
return jsonify({
"status": "vytvoreno",
"spz": spz,
"spotreba": spotreba,
"stav": stav
}), 201
@app.route("/api/stav_vozidla")
@login_required
def stav_vozidla():
spz = request.args.get("spz")
if not spz:
return jsonify({"error": "Chybí parametr SPZ"}), 400
with get_db() as conn:
c = conn.cursor()
c.execute("SELECT stav FROM vozidla WHERE spz = ?", (spz,))
row = c.fetchone()
if not row:
return jsonify({"error": "Vozidlo neexistuje"}), 404
return jsonify({"stav": row[0]})
@app.route("/api/rezervuj_vozidlo", methods=["POST"])
@login_required
def rezervuj_vozidlo():
data = request.get_json()
spz = data.get("vozidlo")
if not spz:
return jsonify({"error": "Chybí parametr vozidlo"}), 400
with get_db() as conn:
c = conn.cursor()
c.execute("SELECT stav FROM vozidla WHERE spz = ?", (spz,))
row = c.fetchone()
if not row:
return jsonify({"error": "Vozidlo neexistuje"}), 404
if row[0] == "obsazene":
return jsonify({"error": "Vozidlo je již obsazeno"}), 409
c.execute("UPDATE vozidla SET stav = 'obsazene' WHERE spz = ?", (spz,))
conn.commit()
return jsonify({"ok": True})
@app.route("/api/uvolni_vozidlo", methods=["POST"])
@login_required
def uvolni_vozidlo():
data = request.get_json()
spz = data.get("vozidlo")
if not spz:
return jsonify({"error": "Chybí parametr vozidlo"}), 400
with get_db() as conn:
c = conn.cursor()
c.execute("SELECT stav FROM vozidla WHERE spz = ?", (spz,))
row = c.fetchone()
if not row:
return jsonify({"error": "Vozidlo neexistuje"}), 404
if row[0] == "volne":
return jsonify({"ok": True, "msg": "Vozidlo je již volné"})
c.execute("UPDATE vozidla SET stav = 'volne' WHERE spz = ?", (spz,))
conn.commit()
return jsonify({"ok": True, "msg": "Vozidlo bylo uvolněno"})
@app.route("/data")
@login_required
def get_data():
tabulka = request.args.get("tabulka")
zobrazit_vyrazena = request.args.get("zobrazit_vyrazena") == "true"
povolene_tabulky = ["zaznamy", "karta_vozidla", "vozidla", "uzivatele", "mesicni_souhrny_ridicu", "mesicni_souhrny_vozidel"]
if tabulka not in povolene_tabulky:
return jsonify({"error": "Nepovolená tabulka"}), 403
spz = request.args.get("spz")
jmeno = request.args.get("jmeno")
datum = request.args.get("datum")
filters = []
params = []
if jmeno:
filters.append("ridic = ?")
params.append(jmeno)
spz_sloupec = "vozidlo" if tabulka in ["karta_vozidla", "zaznamy"] else "spz"
if spz and spz_sloupec:
filters.append(f"{spz_sloupec} = ?")
params.append(spz)
if datum:
filters.append("datum = ?")
params.append(datum)
where_clause = " WHERE " + " AND ".join(filters) if filters else ""
if tabulka == "karta_vozidla":
order_clause = " ORDER BY datum DESC, id DESC"
elif tabulka == "zaznamy":
order_clause = """
ORDER BY
date(
substr(datum, instr(datum, '.') + 1 + instr(substr(datum, instr(datum, '.') + 1), '.')) || '-' ||
printf('%02d', CAST(substr(datum, instr(datum, '.') + 1, instr(substr(datum, instr(datum, '.') + 1), '.') - 1) AS INTEGER)) || '-' ||
printf('%02d', CAST(substr(datum, 1, instr(datum, '.') - 1) AS INTEGER))
) DESC,
ulozeno DESC
"""
else:
order_clause = ""
with get_db() as conn:
conn.row_factory = sqlite3.Row
c = conn.cursor()
if tabulka == "vozidla" and not zobrazit_vyrazena:
if filters:
filters.append("stav != 'vyrazeno'")
where_clause = " WHERE " + " AND ".join(filters)
else:
where_clause = " WHERE stav != 'vyrazeno'"
c.execute(f"SELECT * FROM {tabulka}{where_clause}{order_clause}", params)
rows = [dict(row) for row in c.fetchall()]
return jsonify(rows)
@app.route("/update", methods=["PUT"])
@admin_required
def update_zaznam():
tabulka = request.args.get("tabulka")
data = request.get_json()
if tabulka not in ["zaznamy", "karta_vozidla", "vozidla", "uzivatele"]:
return jsonify({"error": "Nepovolená tabulka"}), 400
if not data:
return jsonify({"error": "Chybí data"}), 400
if tabulka == "vozidla":
if "spz" not in data:
return jsonify({"error": "Chybí SPZ"}), 400
spz = data["spz"]
fields = [k for k in data.keys() if k != "spz"]
if not fields:
return jsonify({"error": "Není co upravit"}), 400
values = [data[k] for k in fields]
with get_db() as conn:
c = conn.cursor()
c.execute(
f"UPDATE vozidla SET {', '.join([f'{k} = ?' for k in fields])} WHERE spz = ?",
(*values, spz)
)
conn.commit()
return jsonify({"status": "upraveno"})
if tabulka == "uzivatele":
if "username" not in data:
return jsonify({"error": "Chybí username"}), 400
username = data["username"]
zakazane = {"heslo"}
fields = [k for k in data.keys() if k not in zakazane and k != "username"]
if not fields:
return jsonify({"error": "Není co upravit"}), 400
values = [data[k] for k in fields]
with get_db() as conn:
c = conn.cursor()
c.execute(
f"UPDATE uzivatele SET {', '.join([f'{k} = ?' for k in fields])} WHERE username = ?",
(*values, username)
)
conn.commit()
return jsonify({"status": "upraveno"})
keys = list(data.keys())
if "id" in data:
where = "id = ?"
where_value = data["id"]
elif "poradove_cislo" in data and "datum" in data:
where = "poradove_cislo = ? AND datum = ?"
where_value = (data["poradove_cislo"], data["datum"])
else:
return jsonify({"error": "Nelze určit unikátní záznam"}), 400
fields = [f"{k} = ?" for k in keys if k not in ["id", "poradove_cislo", "datum"]]
values = [data[k] for k in keys if k not in ["id", "poradove_cislo", "datum"]]
with get_db() as conn:
c = conn.cursor()
if isinstance(where_value, tuple):
c.execute(f"UPDATE {tabulka} SET {', '.join(fields)} WHERE {where}", (*values, *where_value))
else:
c.execute(f"UPDATE {tabulka} SET {', '.join(fields)} WHERE {where}", (*values, where_value))
conn.commit()
return jsonify({"status": "upraveno"})
@app.route("/delete", methods=["DELETE"])
@admin_required
def delete_zaznam():
tabulka = request.args.get("tabulka")
data = request.get_json()
if tabulka not in ["zaznamy", "karta_vozidla", "vozidla"]:
return jsonify({"error": "Nepovolená tabulka"}), 400
if not data:
return jsonify({"error": "Chybí data"}), 400
if tabulka == "vozidla":
spz = data.get("spz")
if not spz:
return jsonify({"error": "Chybí SPZ"}), 400
with get_db() as conn:
c = conn.cursor()
c.execute("""
UPDATE vozidla
SET stav = 'vyrazeno',
vyrazeno = CURRENT_TIMESTAMP
WHERE spz = ?
""", (spz,))
conn.commit()
return jsonify({"status": "vyrazeno"})
if "id" in data:
where = "id = ?"
params = (data["id"],)
elif "poradove_cislo" in data and "datum" in data:
where = "poradove_cislo = ? AND datum = ?"
params = (data["poradove_cislo"], data["datum"])
else:
return jsonify({"error": "Nelze určit záznam pro smazání"}), 400
with get_db() as conn:
c = conn.cursor()
c.execute(f"DELETE FROM {tabulka} WHERE {where}", params)
conn.commit()
return jsonify({"status": "smazano"})
# Vytvoreni uzivatele
@app.route("/api/uzivatele", methods=["POST"])
@admin_required
def pridej_uzivatele():
data = request.get_json()
if not isinstance(data, dict):
return jsonify({"error": "Data musí být objekt"}), 400
username = (data.get("username") or "").strip()
heslo = (data.get("heslo") or "").strip()
ridic = (data.get("ridic") or "").strip()
vozidlo = (data.get("vozidlo") or "").strip()
role = (data.get("role") or "ridic").strip()
email = (data.get("email") or "").strip()
sluzeb_telefon = (data.get("sluzeb_telefon") or "").strip()
if not username or not heslo or not ridic:
return jsonify({"error": "Vyplň username, heslo a jméno zaměstnance."}), 400
if role not in ["ridic", "admin"]:
return jsonify({"error": "Role musí být ridic nebo admin."}), 400
if vozidlo:
with get_db() as conn:
c = conn.cursor()
c.execute("SELECT spz FROM vozidla WHERE spz = ? AND stav != 'vyrazeno'", (vozidlo,))
if not c.fetchone():
return jsonify({
"error": f"Vozidlo '{vozidlo}' neexistuje. Nejdřív ho vytvoř v sekci Přidat vozidlo."
}), 400
hashed = hash_hesla(heslo)
try:
with get_db() as conn:
c = conn.cursor()
c.execute("""
INSERT INTO uzivatele (
username, heslo, ridic, vozidlo, role,
email, sluzeb_telefon
)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (
username, hashed, ridic, vozidlo, role,
email, sluzeb_telefon
))
conn.commit()
except sqlite3.IntegrityError:
return jsonify({"error": "Uživatel s tímto username už existuje."}), 409
return jsonify({
"status": "vytvoreno",
"username": username,
"ridic": ridic,
"role": role
}), 201
@app.route("/api/ridici")
@login_required
def ridici():
con = sqlite3.connect(DB_NAME)
cur = con.cursor()
cur.execute("SELECT ridic FROM uzivatele WHERE role = 'ridic' AND ridic IS NOT NULL")
data = [row[0] for row in cur.fetchall()]
con.close()
return jsonify(data)
@app.route("/api/zapomenute-heslo", methods=["POST"])
def zapomenute_heslo():
data = request.get_json() or {}
email = (data.get("email") or "").strip().lower()
if not email:
return jsonify({"success": False, "msg": "Chybí e-mail"}), 400
with get_db() as conn:
c = conn.cursor()
c.execute("SELECT username, email FROM uzivatele WHERE lower(email) = ?", (email,))
row = c.fetchone()
if row:
token = vytvor_reset_token(row["username"], row["email"])
reset_link = f"{APP_URL}/reset_hesla.html?token={token}"
predmet = "Obnova hesla - Sanitka systém"
zprava = f"""Dobrý den,
pro nastavení nového hesla klikněte na tento odkaz:
{reset_link}
Odkaz platí 30 minut.
Pokud jste o obnovu hesla nežádal/a, tento e-mail ignorujte.
"""
odesli_email(row["email"], predmet, zprava)
zaloguj_udalost("zadost_obnova_hesla", row["username"], request.remote_addr)
return jsonify({
"success": True,
"msg": "Pokud účet s tímto e-mailem existuje, poslali jsme odkaz pro obnovu hesla."
})
@app.route("/api/reset-password", methods=["POST"])
def reset_password():
data = request.get_json() or {}
token = data.get("token") or ""
nove_heslo = (data.get("nove_heslo") or "").strip()
if not token:
return jsonify({"success": False, "msg": "Chybí resetovací token."}), 400
if len(nove_heslo) < 6:
return jsonify({"success": False, "msg": "Heslo musí mít alespoň 6 znaků."}), 400
try:
payload = serializer.loads(token, max_age=RESET_TOKEN_MAX_AGE)
except SignatureExpired:
return jsonify({"success": False, "msg": "Odkaz pro obnovu hesla vypršel."}), 400
except BadSignature:
return jsonify({"success": False, "msg": "Neplatný odkaz pro obnovu hesla."}), 400
if payload.get("type") != "password_reset":
return jsonify({"success": False, "msg": "Neplatný typ tokenu."}), 400
username = payload.get("username")
with get_db() as conn:
c = conn.cursor()
c.execute(
"UPDATE uzivatele SET heslo = ? WHERE username = ?",
(hash_hesla(nove_heslo), username)
)
conn.commit()
zaloguj_udalost("reset_hesla", username, request.remote_addr)
return jsonify({
"success": True,
"msg": "Heslo bylo změněno. Teď se můžete přihlásit."
})
@app.route("/api/zmenit-heslo", methods=["POST"])
@login_required
def zmenit_heslo():
data = request.get_json()
ridic = data.get("username")
stare_heslo = data.get("stare_heslo")
nove_heslo = data.get("nove_heslo")
if not all([ridic, stare_heslo, nove_heslo]):
return jsonify({"success": False, "msg": "Chybí údaje"}), 400
with get_db() as conn:
c = conn.cursor()
c.execute("SELECT heslo FROM uzivatele WHERE ridic = ?", (ridic,))
row = c.fetchone()
if not row:
return jsonify({"success": False, "msg": "Uživatel nenalezen"}), 404
stary_hash = row[0]
if not over_heslo(stare_heslo, stary_hash):
return jsonify({"success": False, "msg": "Špatné staré heslo"}), 401
novy_hash = hash_hesla(nove_heslo)
with get_db() as conn:
c = conn.cursor()
c.execute("UPDATE uzivatele SET heslo = ? WHERE ridic = ?", (novy_hash, ridic))
conn.commit()
return jsonify({"success": True, "msg": "Heslo změněno."})
@app.route("/api/souhrny/ridic", methods=["GET", "POST"])
@admin_required
def souhrn_ridice():
if request.method == "GET":
rok_mesic = request.args.get("rok_mesic")
ridic = request.args.get("ridic")
with get_db() as conn:
c = conn.cursor()
c.execute("""
SELECT *
FROM mesicni_souhrny_ridicu
WHERE rok_mesic = ? AND ridic = ?
""", (rok_mesic, ridic))
row = c.fetchone()
return jsonify(dict(row) if row else None)
data = request.get_json()
with get_db() as conn:
c = conn.cursor()
c.execute("""
INSERT INTO mesicni_souhrny_ridicu
(rok_mesic, ridic, ujeto_km, vytvoril, poznamka)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(rok_mesic, ridic) DO UPDATE SET
ujeto_km = excluded.ujeto_km,
vytvoril = excluded.vytvoril,
vytvoreno = CURRENT_TIMESTAMP,
poznamka = excluded.poznamka
""", (
data.get("rok_mesic"),
data.get("ridic"),
data.get("ujeto_km"),
request.user.get("username"),
data.get("poznamka")
))
conn.commit()
return jsonify({"status": "ulozeno"})
@app.route("/api/souhrny/vozidlo", methods=["GET", "POST"])
@admin_required
def souhrn_vozidla():
if request.method == "GET":
rok_mesic = request.args.get("rok_mesic")
vozidlo = request.args.get("vozidlo")
with get_db() as conn:
c = conn.cursor()
c.execute("""
SELECT *
FROM mesicni_souhrny_vozidel
WHERE rok_mesic = ? AND vozidlo = ?
""", (rok_mesic, vozidlo))
row = c.fetchone()
return jsonify(dict(row) if row else None)
data = request.get_json()
with get_db() as conn:
c = conn.cursor()
c.execute("""
INSERT INTO mesicni_souhrny_vozidel
(rok_mesic, vozidlo, ujeto_km, natankovano_l, spotreba_l_100km, start_km, konec_km, vytvoril, poznamka)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(rok_mesic, vozidlo) DO UPDATE SET
ujeto_km = excluded.ujeto_km,
natankovano_l = excluded.natankovano_l,
spotreba_l_100km = excluded.spotreba_l_100km,
start_km = excluded.start_km,
konec_km = excluded.konec_km,
vytvoril = excluded.vytvoril,
vytvoreno = CURRENT_TIMESTAMP,
poznamka = excluded.poznamka
""", (
data.get("rok_mesic"),
data.get("vozidlo"),
data.get("ujeto_km"),
data.get("natankovano_l"),
data.get("spotreba_l_100km"),
data.get("start_km"),
data.get("konec_km"),
request.user.get("username"),
data.get("poznamka")
))
conn.commit()
return jsonify({"status": "ulozeno"})
@app.route('/')
def static_files(filename):
return send_from_directory('.', filename)
@app.route("/")
def index():
return send_from_directory('.', 'index.html')
@app.errorhandler(Exception)
def handle_error(e):
import traceback
if isinstance(e, HTTPException):
return e
traceback.print_exc()
return jsonify({
"error": f"Interní chyba serveru: {str(e)}"
}), 500
if __name__ == "__main__":
init_db()
app.run(host="0.0.0.0", port=8050, debug=False)
← Zpět na seznam