Mateusz Gruszczyński
434 lines
27 KiB
Python
434 lines
27 KiB
Python
from collections import defaultdict
|
|
from datetime import datetime, date
|
|
from decimal import Decimal
|
|
from functools import wraps
|
|
from flask import Blueprint, render_template, request, redirect, url_for, flash, jsonify, abort
|
|
from flask_login import login_required, current_user
|
|
from sqlalchemy import extract
|
|
from sqlalchemy.exc import IntegrityError
|
|
from .extensions import db, socketio
|
|
from .models import User, Vehicle, FuelEntry, CompanySettings, AppSetting, OrlenPrice, FuelCardPolicy, FuelStationCompany, FuelStationPoint, FuelCard, FuelCardStationRule
|
|
from .services import calculate_costs, fetch_orlen_price, fetch_orlen_range, invoice_period, fetch_ure_stations, aggregate_ure_companies, POLISH_REGIONS
|
|
from .station_catalog import sync_station_catalog
|
|
from . import THEMES
|
|
from .vat import effective_vat_rate, global_vat_override, save_global_vat_override
|
|
|
|
main = Blueprint("main", __name__)
|
|
FUEL_TYPES = ("PB95", "PB98", "DIESEL", "LPG")
|
|
|
|
|
|
def wants_json():
|
|
return request.headers.get("X-Requested-With") == "XMLHttpRequest" or request.is_json
|
|
|
|
def ok_response(message, **extra):
|
|
socketio.emit("app_changed", {"message": message})
|
|
if wants_json():
|
|
return jsonify({"ok": True, "message": message, **extra})
|
|
flash(message, "success")
|
|
return None
|
|
|
|
def error_response(message, status=400):
|
|
if wants_json():
|
|
return jsonify({"ok": False, "error": message}), status
|
|
flash(message, "danger")
|
|
return None
|
|
|
|
def roles(*allowed):
|
|
def deco(fn):
|
|
@wraps(fn)
|
|
def wrapped(*a, **kw):
|
|
if current_user.role not in allowed:
|
|
flash("Brak uprawnień", "danger")
|
|
return redirect(url_for("main.dashboard"))
|
|
return fn(*a, **kw)
|
|
return wrapped
|
|
return deco
|
|
|
|
def accessible_vehicles():
|
|
if current_user.role == "admin":
|
|
return Vehicle.query.order_by(Vehicle.name).all()
|
|
if current_user.role == "boss" and current_user.company_id:
|
|
return Vehicle.query.filter_by(company_id=current_user.company_id).order_by(Vehicle.name).all()
|
|
ids = {v.id for v in current_user.owned_vehicles + current_user.shared_vehicles}
|
|
return Vehicle.query.filter(Vehicle.id.in_(ids)).order_by(Vehicle.name).all() if ids else []
|
|
|
|
@main.route("/")
|
|
@login_required
|
|
def dashboard():
|
|
settings = CompanySettings.query.first(); vehicles = accessible_vehicles(); ids = [v.id for v in vehicles]
|
|
split_enabled = (db.session.get(AppSetting, "invoice_split_enabled") or AppSetting(key="invoice_split_enabled", value="1")).value == "1"
|
|
month = request.args.get("month", datetime.now().strftime("%Y-%m"))
|
|
try: year, mon = map(int, month.split("-"))
|
|
except ValueError: year, mon = datetime.now().year, datetime.now().month; month=f"{year:04d}-{mon:02d}"
|
|
entries = FuelEntry.query.filter(FuelEntry.vehicle_id.in_(ids), extract("year", FuelEntry.fueled_at)==year, extract("month", FuelEntry.fueled_at)==mon).order_by(FuelEntry.fueled_at.desc()).all() if ids else []
|
|
totals = {"gross": Decimal("0"), "deductible_vat": Decimal("0"), "final_cost": Decimal("0"), "liters": Decimal("0")}
|
|
by_vehicle = defaultdict(lambda: {"gross": 0, "payable": 0, "liters": 0}); invoices = defaultdict(lambda: {"gross": 0, "payable": 0, "count": 0}); settlements = {}
|
|
for e in entries:
|
|
vat_rate = effective_vat_rate(settings.vat_rate, e.fueled_at)
|
|
c = calculate_costs(e.gross, vat_rate, settings.vat_deduction_percent)
|
|
totals["gross"] += c["gross"]; totals["deductible_vat"] += c["deductible_vat"]; totals["final_cost"] += c["final_cost"]; totals["liters"] += Decimal(e.liters)
|
|
station_policy = e.station_company
|
|
normal_price = float(e.price_per_liter)
|
|
payable_price = normal_price
|
|
rule = FuelCardStationRule.query.filter_by(fuel_card_id=e.fuel_card_id, station_company_id=e.station_company_id).first() if e.used_fuel_card and e.fuel_card_id and e.station_company_id else None
|
|
uses_station_last_price = bool(rule and rule.use_orlen_last_price and e.wholesale_price is not None)
|
|
if uses_station_last_price:
|
|
payable_price = float(e.wholesale_price) * (1 + float(vat_rate) / 100)
|
|
if rule:
|
|
net = payable_price / (1 + float(vat_rate) / 100)
|
|
net = net * (1 - float(rule.discount_net_percent or 0) / 100) + float(rule.surcharge_net_per_liter or 0)
|
|
payable_price = net * (1 + float(vat_rate) / 100)
|
|
payable_price = max(payable_price, 0)
|
|
payable_gross = float(e.liters) * payable_price
|
|
settlements[e.id] = {"normal_price": normal_price, "payable_price": payable_price, "normal_gross": e.gross, "payable_gross": payable_gross, "uses_last_price": uses_station_last_price, "vat_rate": float(vat_rate)}
|
|
by_vehicle[e.vehicle.name]["gross"] += float(c["gross"]); by_vehicle[e.vehicle.name]["payable"] += payable_gross; by_vehicle[e.vehicle.name]["liters"] += float(e.liters)
|
|
if e.used_fuel_card:
|
|
key = invoice_period(e.fueled_at, settings.invoice_split_day) if split_enabled else "Cały miesiąc"
|
|
else:
|
|
key = "Poza kartą"
|
|
invoices[key]["gross"] += float(c["gross"]); invoices[key]["payable"] += payable_gross; invoices[key]["count"] += 1
|
|
return render_template("dashboard.html", vehicles=vehicles, entries=entries, totals=totals, month=month, by_vehicle=dict(by_vehicle), invoices=dict(invoices), settings=settings, split_enabled=split_enabled, settlements=settlements, vat_override=global_vat_override())
|
|
|
|
@main.route("/vehicles", methods=["GET", "POST"])
|
|
@login_required
|
|
def vehicles():
|
|
if request.method == "POST":
|
|
v = Vehicle(name=request.form["name"], make=request.form["make"], model=request.form["model"], registration=request.form["registration"].upper(), fuel_type=request.form["fuel_type"], owner_id=current_user.id, has_fuel_card="has_fuel_card" in request.form, card_provider=request.form.get("card_provider"), current_odometer=int(request.form.get("current_odometer") or 0))
|
|
db.session.add(v); db.session.commit(); response = ok_response("Dodano pojazd", vehicle_id=v.id); return response or redirect(url_for("main.vehicles"))
|
|
company_id=current_user.company_id
|
|
users_q=User.query.filter_by(active=True)
|
|
cards_q=FuelCard.query.filter_by(active=True)
|
|
if current_user.role=='boss' and company_id:
|
|
users_q=users_q.filter_by(company_id=company_id);cards_q=cards_q.filter_by(company_id=company_id)
|
|
return render_template("vehicles.html", vehicles=accessible_vehicles(), fuel_cards=cards_q.order_by(FuelCard.name).all(), assignable_users=users_q.order_by(User.name).all())
|
|
|
|
@main.route("/vehicles/<int:vehicle_id>/share", methods=["POST"])
|
|
@login_required
|
|
def share_vehicle(vehicle_id):
|
|
vehicle = Vehicle.query.get_or_404(vehicle_id)
|
|
if current_user.role != "admin" and vehicle.owner_id != current_user.id:
|
|
response = error_response("Tylko właściciel lub admin może udostępniać auto", 403); return response or redirect(url_for("main.vehicles"))
|
|
user = User.query.filter_by(email=request.form["email"].strip().lower()).first()
|
|
if not user:
|
|
response = error_response("Nie znaleziono użytkownika", 404); return response or redirect(url_for("main.vehicles"))
|
|
if user not in vehicle.drivers: vehicle.drivers.append(user); db.session.commit()
|
|
response = ok_response("Udostępniono pojazd"); return response or redirect(url_for("main.vehicles"))
|
|
|
|
|
|
@main.post("/vehicles/<int:vehicle_id>/unshare/<int:user_id>")
|
|
@login_required
|
|
def unshare_vehicle(vehicle_id, user_id):
|
|
vehicle = Vehicle.query.get_or_404(vehicle_id)
|
|
if current_user.role != "admin" and vehicle.owner_id != current_user.id:
|
|
return error_response("Tylko właściciel lub admin może odebrać dostęp", 403)
|
|
user = User.query.get_or_404(user_id)
|
|
if user in vehicle.drivers:
|
|
vehicle.drivers.remove(user); db.session.commit()
|
|
response = ok_response("Odebrano dostęp do pojazdu")
|
|
return response or redirect(url_for("main.vehicles"))
|
|
|
|
@main.post("/vehicles/<int:vehicle_id>/card-policy")
|
|
@login_required
|
|
def update_card_policy(vehicle_id):
|
|
vehicle = Vehicle.query.get_or_404(vehicle_id)
|
|
if current_user.role != "admin" and vehicle.owner_id != current_user.id:
|
|
return error_response("Tylko właściciel lub admin może zmienić zasady rozliczeń", 403)
|
|
policy = vehicle.fuel_card_policy or FuelCardPolicy(vehicle=vehicle)
|
|
policy.use_orlen_last_price = "use_orlen_last_price" in request.form
|
|
policy.discount_percent = Decimal(request.form.get("discount_percent") or 0)
|
|
policy.surcharge_per_liter = Decimal(request.form.get("surcharge_per_liter") or 0)
|
|
policy.description = request.form.get("description", "").strip() or None
|
|
db.session.add(policy); db.session.commit()
|
|
response = ok_response("Zapisano zasady rozliczeń karty")
|
|
return response or redirect(url_for("main.vehicles"))
|
|
|
|
@main.route("/fuel", methods=["GET", "POST"])
|
|
@login_required
|
|
def fuel():
|
|
vehicles = accessible_vehicles()
|
|
if request.method == "POST":
|
|
vehicle = next((v for v in vehicles if v.id == int(request.form["vehicle_id"])), None)
|
|
if not vehicle:
|
|
response = error_response("Brak dostępu do pojazdu", 403); return response or redirect(url_for("main.fuel"))
|
|
odometer = int(request.form["odometer"])
|
|
if odometer < vehicle.current_odometer:
|
|
response = error_response("Licznik nie może być niższy od poprzedniego"); return response or redirect(url_for("main.fuel"))
|
|
fueled_at = datetime.fromisoformat(request.form["fueled_at"]); wholesale = None; source = None
|
|
station_company_id = request.form.get("station_company_id", type=int)
|
|
station_company = db.session.get(FuelStationCompany, station_company_id) if station_company_id else None
|
|
settings = vehicle.company or current_user.company or CompanySettings.query.first()
|
|
access_mode = settings.station_access_mode or "all_prefer_favorites"
|
|
if station_company and access_mode == "allowed_only" and station_company not in settings.allowed_stations:
|
|
response = error_response("Ta stacja nie jest dozwolona przez politykę firmy", 403); return response or redirect(url_for("main.fuel"))
|
|
if station_company and access_mode == "favorites_allowed_only" and station_company not in settings.allowed_stations and station_company not in settings.favorite_stations:
|
|
response = error_response("Ta stacja nie jest ulubiona ani dozwolona przez politykę firmy", 403); return response or redirect(url_for("main.fuel"))
|
|
price_region = station_company.selected_region if station_company and station_company.selected_region else settings.region
|
|
try:
|
|
p = fetch_orlen_price(request.form["fuel_type"], fueled_at.date(), price_region); wholesale=p["price_per_liter"]; source=p["source"]
|
|
except Exception: pass
|
|
station_label = station_company.company_name if station_company else request.form.get("station")
|
|
e = FuelEntry(vehicle_id=vehicle.id, user_id=current_user.id, fueled_at=fueled_at, liters=Decimal(request.form["liters"]), price_per_liter=Decimal(request.form["price_per_liter"]), fuel_type=request.form["fuel_type"], odometer=odometer, station=station_label, station_company_id=station_company.id if station_company else None, invoice_number=request.form.get("invoice_number"), used_fuel_card="used_fuel_card" in request.form, wholesale_price=wholesale, wholesale_source=source)
|
|
vehicle.current_odometer = odometer; db.session.add(e); db.session.commit(); socketio.emit("fuel_added", {"vehicle": vehicle.name, "gross": round(e.gross,2)}); response = ok_response("Zapisano tankowanie", redirect=url_for("main.dashboard")); return response or redirect(url_for("main.dashboard"))
|
|
settings=current_user.company or CompanySettings.query.first()
|
|
allowed=list(settings.allowed_stations)
|
|
company_favorites=list(settings.favorite_stations)
|
|
access_mode=settings.station_access_mode or "all_prefer_favorites"
|
|
base_query=FuelStationCompany.query.filter_by(active=True)
|
|
if access_mode == "allowed_only":
|
|
allowed_ids=[x.id for x in allowed]
|
|
stations=base_query.filter(FuelStationCompany.id.in_(allowed_ids)).all() if allowed_ids else []
|
|
elif access_mode == "favorites_allowed_only":
|
|
scope_ids={x.id for x in allowed}|{x.id for x in company_favorites}
|
|
stations=base_query.filter(FuelStationCompany.id.in_(scope_ids)).all() if scope_ids else []
|
|
else:
|
|
stations=base_query.all()
|
|
stations=sorted(stations,key=lambda x:(-x.station_count,x.display_name.lower()))
|
|
user_favorites=[x for x in current_user.favorite_stations if x in stations]
|
|
user_favorite_ids={x.id for x in current_user.favorite_stations}
|
|
favorites=list(user_favorites)
|
|
favorite_scope="user"
|
|
if not favorites and access_mode in ("all_prefer_favorites", "favorites_allowed_only") and company_favorites:
|
|
favorites=[x for x in company_favorites if x in stations];favorite_scope="company"
|
|
if not favorites and access_mode == "all_prefer_favorites": favorites=stations[:10];favorite_scope="default"
|
|
favorites=sorted(favorites,key=lambda x:(-x.station_count,x.display_name.lower()))
|
|
favorite_ids={x.id for x in favorites}
|
|
top_stations=sorted((x for x in stations if x.id not in favorite_ids),key=lambda x:(-x.station_count,x.display_name.lower()))[:10]
|
|
proposed_ids=favorite_ids|{x.id for x in top_stations}
|
|
favorite_catalog=sorted(stations,key=lambda x:(0 if x.id in user_favorite_ids else 1,-x.station_count,x.display_name.lower()))
|
|
other_stations=sorted((x for x in stations if x.id not in proposed_ids),key=lambda x:x.display_name.lower())
|
|
return render_template("fuel_form.html", vehicles=vehicles, stations=stations, favorite_stations=favorites, top_stations=top_stations, favorite_catalog=favorite_catalog, other_stations=other_stations, now=datetime.now().strftime("%Y-%m-%dT%H:%M"), restricted=access_mode in ("allowed_only", "favorites_allowed_only"), access_mode=access_mode, user_favorite_ids=user_favorite_ids, favorite_scope=favorite_scope, company=settings, fuel_cards=FuelCard.query.filter_by(active=True,company_id=settings.id).order_by(FuelCard.name).all())
|
|
|
|
@main.route("/api/orlen-price")
|
|
@login_required
|
|
def orlen_price():
|
|
try: return jsonify(fetch_orlen_price(request.args["fuel_type"], datetime.fromisoformat(request.args["date"]).date(), CompanySettings.query.first().region))
|
|
except Exception as exc: return jsonify({"error": str(exc)}), 502
|
|
|
|
@main.route("/orlen")
|
|
@login_required
|
|
def orlen_data():
|
|
settings = CompanySettings.query.first()
|
|
selected_year = request.args.get("year", type=int) or date.today().year
|
|
selected_fuels = [f.upper() for f in request.args.getlist("fuel") if f.upper() in FUEL_TYPES] or ["PB95"]
|
|
include_vat = request.args.get("gross") == "1"
|
|
show_lpg_regions = request.args.get('show_lpg_regions') == '1'
|
|
selected_regions = [r.lower() for r in request.args.getlist("region") if r.lower() in POLISH_REGIONS]
|
|
if not selected_regions: selected_regions=[settings.region.lower()]
|
|
if not show_lpg_regions: selected_regions=[settings.region.lower()]
|
|
query = OrlenPrice.query.filter(OrlenPrice.fuel_type.in_(selected_fuels), extract("year", OrlenPrice.effective_date)==selected_year)
|
|
rows = query.order_by(OrlenPrice.effective_date, OrlenPrice.fuel_type).all()
|
|
chart = {}
|
|
for fuel in selected_fuels:
|
|
fuel_rows = [r for r in rows if r.fuel_type == fuel and (fuel != "LPG" or r.region in selected_regions)]
|
|
chart_key = fuel if fuel != "LPG" else None
|
|
if fuel == "LPG":
|
|
for region in selected_regions:
|
|
chart[f"LPG · {region}"] = [{"date": r.effective_date.isoformat(), "value": round(float(r.price_per_liter) * (1 + float(effective_vat_rate(settings.vat_rate, r.effective_date)) / 100 if include_vat else 1), 4)} for r in fuel_rows if r.region == region]
|
|
else:
|
|
chart[fuel] = [{"date": r.effective_date.isoformat(), "value": round(float(r.price_per_liter) * (1 + float(effective_vat_rate(settings.vat_rate, r.effective_date)) / 100 if include_vat else 1), 4)} for r in fuel_rows]
|
|
return render_template("orlen.html", rows=rows, chart=chart, selected_fuels=selected_fuels, selected_year=selected_year, settings=settings, fuel_types=FUEL_TYPES, include_vat=include_vat, regions=POLISH_REGIONS, selected_regions=selected_regions, show_lpg_regions=show_lpg_regions)
|
|
|
|
@main.post("/orlen/sync")
|
|
@login_required
|
|
@roles("boss", "admin")
|
|
def orlen_sync():
|
|
payload = request.get_json(silent=True) or request.form
|
|
fuels = payload.get("fuels", [])
|
|
if isinstance(fuels, str): fuels = [fuels]
|
|
fuels = [f.upper() for f in fuels if f.upper() in FUEL_TYPES]
|
|
year = int(payload.get("year", date.today().year))
|
|
if not fuels: return jsonify({"ok": False, "error": "Wybierz co najmniej jedno paliwo."}), 400
|
|
settings = CompanySettings.query.first(); start = date(year,1,1); end = min(date(year,12,31), date.today())
|
|
result = {}; total = 0
|
|
try:
|
|
for fuel in fuels:
|
|
imported = 0; updated = 0
|
|
for row in fetch_orlen_range(fuel, start, end, "all" if fuel == "LPG" else settings.region):
|
|
existing = OrlenPrice.query.filter_by(fuel_type=row["fuel_type"], effective_date=row["effective_date"], region=row["region"]).first()
|
|
if existing:
|
|
existing.price_per_liter=row["price_per_liter"]; existing.raw_value=row["raw_value"]; existing.product_name=row["product_name"]; existing.source=row["source"]; existing.fetched_at=datetime.utcnow(); updated += 1
|
|
else:
|
|
db.session.add(OrlenPrice(**row)); imported += 1
|
|
result[fuel] = {"added": imported, "updated": updated}; total += imported + updated
|
|
db.session.commit()
|
|
return jsonify({"ok": True, "message": f"Przetworzono {total} rekordów.", "result": result})
|
|
except Exception as exc:
|
|
db.session.rollback(); return jsonify({"ok": False, "error": str(exc)}), 502
|
|
|
|
@main.route("/admin", methods=["GET","POST"])
|
|
@login_required
|
|
@roles("boss", "admin")
|
|
def admin():
|
|
if request.method=="POST":
|
|
settings=current_user.company or CompanySettings.query.first()
|
|
settings.name=request.form.get("name",settings.name);settings.entity_type=request.form.get("entity_type",settings.entity_type);settings.vat_rate=Decimal(request.form.get("vat_rate",settings.vat_rate));settings.vat_deduction_percent=Decimal(request.form.get("vat_deduction_percent",settings.vat_deduction_percent));settings.region=request.form.get("region",settings.region);settings.invoice_split_day=int(request.form.get("invoice_split_day",settings.invoice_split_day))
|
|
theme=request.form.get("theme")
|
|
if theme in THEMES:db.session.merge(AppSetting(key="theme",value=theme))
|
|
db.session.commit();return ok_response("Zapisano") or redirect(url_for("main.admin_companies"))
|
|
return redirect(url_for("main.admin_companies"))
|
|
|
|
@main.get("/admin/companies")
|
|
@login_required
|
|
@roles("boss", "admin")
|
|
def admin_companies():
|
|
q = request.args.get("q", "").strip()
|
|
page = request.args.get("page", 1, type=int)
|
|
selected_id = request.args.get("company_id", type=int)
|
|
companies_query = CompanySettings.query
|
|
if current_user.role == "boss":
|
|
companies_query = companies_query.filter_by(id=current_user.company_id)
|
|
elif q:
|
|
companies_query = companies_query.filter(CompanySettings.name.ilike(f"%{q}%"))
|
|
companies_page = companies_query.order_by(CompanySettings.name.asc()).paginate(
|
|
page=page, per_page=20, error_out=False
|
|
)
|
|
visible_companies = companies_page.items
|
|
if current_user.role == "boss":
|
|
selected_company = current_user.company
|
|
else:
|
|
selected_company = db.session.get(CompanySettings, selected_id) if selected_id else None
|
|
if selected_company is None:
|
|
selected_company = visible_companies[0] if visible_companies else CompanySettings.query.order_by(CompanySettings.name.asc()).first()
|
|
if selected_company and current_user.role == "boss" and selected_company.id != current_user.company_id:
|
|
abort(403)
|
|
fuel_cards = FuelCard.query.filter_by(company_id=selected_company.id).order_by(FuelCard.name.asc()).all() if selected_company else []
|
|
stations = FuelStationCompany.query.filter_by(active=True).order_by(
|
|
FuelStationCompany.station_count.desc(), FuelStationCompany.brand_name.asc()
|
|
).all()
|
|
return render_template(
|
|
"admin_companies.html",
|
|
companies_page=companies_page,
|
|
selected_company=selected_company,
|
|
fuel_cards=fuel_cards,
|
|
q=q,
|
|
regions=POLISH_REGIONS,
|
|
stations=stations,
|
|
vat_override=global_vat_override(),
|
|
)
|
|
|
|
|
|
@main.post("/admin/global-vat")
|
|
@login_required
|
|
@roles("admin")
|
|
def update_global_vat():
|
|
try:
|
|
save_global_vat_override(request.form)
|
|
db.session.commit()
|
|
response = ok_response("Zapisano globalne ustawienia VAT")
|
|
return response or redirect(url_for("main.admin_companies"))
|
|
except ValueError as exc:
|
|
db.session.rollback()
|
|
response = error_response(str(exc))
|
|
return response or redirect(url_for("main.admin_companies"))
|
|
|
|
@main.get("/admin/application")
|
|
@login_required
|
|
@roles("admin")
|
|
def admin_application():
|
|
return render_template("admin_application.html")
|
|
|
|
@main.get("/admin/users")
|
|
@login_required
|
|
@roles("boss", "admin")
|
|
def admin_users():
|
|
page=request.args.get("page",1,type=int);q=request.args.get("q","").strip();users_q=User.query
|
|
if current_user.role=="boss": users_q=users_q.filter_by(company_id=current_user.company_id)
|
|
if q: users_q=users_q.filter(db.or_(User.name.ilike(f"%{q}%"),User.email.ilike(f"%{q}%")))
|
|
users_page=users_q.order_by(User.name).paginate(page=page,per_page=25,error_out=False)
|
|
companies=CompanySettings.query.order_by(CompanySettings.name).all() if current_user.role=="admin" else [current_user.company]
|
|
cards=FuelCard.query.order_by(FuelCard.name).all()
|
|
if current_user.role=="boss": cards=cards.filter_by(company_id=current_user.company_id).all()
|
|
return render_template("admin_users.html",users_page=users_page,q=q,companies=[c for c in companies if c],fuel_cards=cards)
|
|
|
|
@main.post("/admin/users/<int:user_id>")
|
|
@login_required
|
|
@roles("admin")
|
|
def edit_user(user_id):
|
|
user = User.query.get_or_404(user_id)
|
|
try:
|
|
user.name = request.form["name"].strip(); user.email = request.form["email"].strip().lower(); user.role = request.form["role"]
|
|
user.active = "active" in request.form
|
|
password = request.form.get("password", "")
|
|
if password:
|
|
if len(password) < 8: raise ValueError("Nowe hasło musi mieć co najmniej 8 znaków.")
|
|
user.set_password(password)
|
|
db.session.commit(); response = ok_response(f"Zaktualizowano użytkownika {user.email}"); return response or redirect(url_for("main.admin"))
|
|
except (ValueError, IntegrityError) as exc:
|
|
db.session.rollback(); message = str(exc.orig) if isinstance(exc, IntegrityError) else str(exc); response = error_response(message); return response or redirect(url_for("main.admin", _anchor=f"user-{user_id}"))
|
|
|
|
|
|
@main.route("/stations")
|
|
@login_required
|
|
@roles("boss", "admin")
|
|
def stations():
|
|
q = request.args.get("q", "").strip()
|
|
page = request.args.get("page", 1, type=int)
|
|
sort = request.args.get("sort", "company_name")
|
|
direction = request.args.get("direction", "asc")
|
|
companies_q = CompanySettings.query.filter_by(active=True).order_by(CompanySettings.name.asc())
|
|
if current_user.role == "boss":
|
|
companies_q = companies_q.filter_by(id=current_user.company_id)
|
|
companies = companies_q.all()
|
|
requested_company_id = request.args.get("company_id", type=int)
|
|
if current_user.role == "boss":
|
|
company = current_user.company
|
|
elif requested_company_id is None:
|
|
return render_template("stations_company_select.html", companies=companies)
|
|
else:
|
|
company = db.session.get(CompanySettings, requested_company_id)
|
|
if company not in companies:
|
|
abort(404, description="Nie znaleziono aktywnej firmy")
|
|
if company is None:
|
|
abort(404, description="Brak aktywnej firmy do konfiguracji")
|
|
|
|
query = FuelStationCompany.query.filter_by(active=True)
|
|
if q:
|
|
terms = [term for term in q.split() if term]
|
|
for term in terms:
|
|
pattern = f"%{term}%"
|
|
query = query.filter(db.or_(FuelStationCompany.brand_name.ilike(pattern), FuelStationCompany.company_name.ilike(pattern), FuelStationCompany.nip.ilike(pattern), FuelStationCompany.regon.ilike(pattern), FuelStationCompany.station_names.ilike(pattern)))
|
|
columns = {"company_name": FuelStationCompany.brand_name, "station_count": FuelStationCompany.station_count, "nip": FuelStationCompany.nip, "regon": FuelStationCompany.regon, "selected_region": FuelStationCompany.selected_region}
|
|
column = columns.get(sort, FuelStationCompany.company_name)
|
|
query = query.order_by(column.desc() if direction == "desc" else column.asc(), FuelStationCompany.brand_name.asc())
|
|
station_page = query.paginate(page=page, per_page=30, error_out=False)
|
|
cards_q=FuelCard.query.filter_by(active=True, company_id=company.id)
|
|
cards=cards_q.order_by(FuelCard.name).all()
|
|
rules={(r.fuel_card_id,r.station_company_id):r for r in FuelCardStationRule.query.filter(FuelCardStationRule.fuel_card_id.in_([c.id for c in cards] or [-1])).all()}
|
|
return render_template("stations.html", station_page=station_page, q=q, regions=POLISH_REGIONS, sort=sort, direction=direction, allowed_ids={x.id for x in company.allowed_stations}, favorite_ids={x.id for x in company.favorite_stations}, company=company, companies=companies, fuel_cards=cards, card_rules=rules)
|
|
|
|
@main.post("/stations/sync")
|
|
@login_required
|
|
@roles("boss", "admin")
|
|
def stations_sync():
|
|
try:
|
|
companies = aggregate_ure_companies(fetch_ure_stations())
|
|
added, updated = sync_station_catalog(companies)
|
|
db.session.commit()
|
|
response = ok_response(f"Katalog URE: dodano {added}, zaktualizowano {updated}.")
|
|
return response or redirect(url_for("main.stations"))
|
|
except Exception as exc:
|
|
db.session.rollback(); response = error_response(f"Nie udało się pobrać danych URE: {exc}", 502); return response or redirect(url_for("main.stations"))
|
|
|
|
@main.post("/stations/<int:station_id>")
|
|
@login_required
|
|
@roles("boss", "admin")
|
|
def update_station(station_id):
|
|
station = FuelStationCompany.query.get_or_404(station_id)
|
|
region = request.form.get("selected_region", "").strip().lower()
|
|
if region and region not in station.region_list:
|
|
return error_response("Wybrane województwo nie występuje w danych tej firmy.")
|
|
station.selected_region = region
|
|
station.active = "active" in request.form
|
|
db.session.commit(); response = ok_response(f"Zapisano ustawienia firmy {station.company_name}"); return response or redirect(url_for("main.stations"))
|
|
|
|
@main.get("/api/stations")
|
|
@login_required
|
|
def station_suggestions():
|
|
q = request.args.get("q", "").strip()
|
|
query = FuelStationCompany.query.filter_by(active=True)
|
|
if q:
|
|
pattern = f"%{q}%"
|
|
query = query.filter(db.or_(FuelStationCompany.brand_name.ilike(pattern), FuelStationCompany.company_name.ilike(pattern), FuelStationCompany.station_names.ilike(pattern), FuelStationCompany.nip.ilike(pattern)))
|
|
rows = query.order_by(FuelStationCompany.company_name).limit(30).all()
|
|
return jsonify([{"id": x.id, "name": x.company_name, "brands": x.station_name_list[:5], "region": x.selected_region, "last_price": x.use_orlen_last_price} for x in rows])
|