Mateusz Gruszczyński
553 lines
33 KiB
Python
553 lines
33 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, or_
|
||
from sqlalchemy.exc import IntegrityError
|
||
from .extensions import db
|
||
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):
|
||
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()
|
||
settings = current_user.company or CompanySettings.query.first()
|
||
if settings is None:
|
||
flash("Najpierw utwórz firmę i przypisz do niej użytkownika, aby korzystać z tankowań.", "warning")
|
||
if current_user.role in ("admin", "boss"):
|
||
return redirect(url_for("main.admin_companies"))
|
||
return redirect(url_for("main.dashboard"))
|
||
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 settings
|
||
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(); response = ok_response("Zapisano tankowanie", redirect=url_for("main.dashboard")); return response or redirect(url_for("main.dashboard"))
|
||
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()
|
||
today = date.today()
|
||
mode = request.args.get("mode", "history")
|
||
if mode not in {"history", "compare"}:
|
||
mode = "history"
|
||
|
||
include_vat = request.args.get("gross") == "1"
|
||
available_years = [
|
||
year for (year,) in db.session.query(extract("year", OrlenPrice.effective_date))
|
||
.distinct().order_by(extract("year", OrlenPrice.effective_date).desc()).all()
|
||
if year is not None
|
||
]
|
||
available_years = [int(year) for year in available_years]
|
||
if today.year not in available_years:
|
||
available_years.insert(0, today.year)
|
||
|
||
selected_year = request.args.get("year", type=int) or today.year
|
||
selected_fuels = [f.upper() for f in request.args.getlist("fuel") if f.upper() in FUEL_TYPES] or ["PB95"]
|
||
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()]
|
||
|
||
rows = []
|
||
chart = {}
|
||
comparison_summary = []
|
||
comparison_type = request.args.get("comparison_type", "years")
|
||
compare_fuel = request.args.get("compare_fuel", "PB95").upper()
|
||
if compare_fuel not in FUEL_TYPES:
|
||
compare_fuel = "PB95"
|
||
compare_region = request.args.get("compare_region", settings.region).lower()
|
||
if compare_region not in POLISH_REGIONS:
|
||
compare_region = settings.region.lower()
|
||
compare_year = request.args.get("compare_year", type=int) or (selected_year - 1)
|
||
|
||
def priced_value(row):
|
||
multiplier = 1 + float(effective_vat_rate(settings.vat_rate, row.effective_date)) / 100 if include_vat else 1
|
||
return round(float(row.price_per_liter) * multiplier, 4)
|
||
|
||
def summary(label, source_rows):
|
||
values = [priced_value(row) for row in source_rows]
|
||
return {
|
||
"label": label,
|
||
"count": len(values),
|
||
"average": round(sum(values) / len(values), 4) if values else None,
|
||
"minimum": min(values) if values else None,
|
||
"maximum": max(values) if values else None,
|
||
"first": values[0] if values else None,
|
||
"last": values[-1] if values else None,
|
||
"change": round(values[-1] - values[0], 4) if len(values) > 1 else None,
|
||
}
|
||
|
||
if mode == "compare":
|
||
if comparison_type not in {"years", "periods"}:
|
||
comparison_type = "years"
|
||
base_query = OrlenPrice.query.filter_by(fuel_type=compare_fuel)
|
||
if compare_fuel == "LPG":
|
||
base_query = base_query.filter_by(region=compare_region)
|
||
|
||
if comparison_type == "years":
|
||
year_a, year_b = selected_year, compare_year
|
||
all_rows = base_query.filter(extract("year", OrlenPrice.effective_date).in_([year_a, year_b])).order_by(OrlenPrice.effective_date).all()
|
||
for year in (year_a, year_b):
|
||
period_rows = [row for row in all_rows if row.effective_date.year == year]
|
||
chart[str(year)] = [{"date": row.effective_date.strftime("%m-%d"), "value": priced_value(row)} for row in period_rows]
|
||
comparison_summary.append(summary(str(year), period_rows))
|
||
rows = all_rows
|
||
period_a_from = date(year_a, 1, 1)
|
||
period_a_to = date(year_a, 12, 31)
|
||
period_b_from = date(year_b, 1, 1)
|
||
period_b_to = date(year_b, 12, 31)
|
||
else:
|
||
def parse_date_arg(name, fallback):
|
||
try:
|
||
return date.fromisoformat(request.args.get(name, ""))
|
||
except ValueError:
|
||
return fallback
|
||
|
||
period_a_from = parse_date_arg("period_a_from", today.replace(month=1, day=1))
|
||
period_a_to = parse_date_arg("period_a_to", today)
|
||
previous_year = today.year - 1
|
||
period_b_from = parse_date_arg("period_b_from", date(previous_year, period_a_from.month, min(period_a_from.day, 28)))
|
||
period_b_to = parse_date_arg("period_b_to", date(previous_year, period_a_to.month, min(period_a_to.day, 28)))
|
||
if period_a_from > period_a_to:
|
||
period_a_from, period_a_to = period_a_to, period_a_from
|
||
if period_b_from > period_b_to:
|
||
period_b_from, period_b_to = period_b_to, period_b_from
|
||
|
||
ranges = [
|
||
("Okres A", period_a_from, period_a_to),
|
||
("Okres B", period_b_from, period_b_to),
|
||
]
|
||
all_rows = base_query.filter(
|
||
or_(
|
||
OrlenPrice.effective_date.between(period_a_from, period_a_to),
|
||
OrlenPrice.effective_date.between(period_b_from, period_b_to),
|
||
)
|
||
).order_by(OrlenPrice.effective_date).all()
|
||
for label, start, end in ranges:
|
||
period_rows = [row for row in all_rows if start <= row.effective_date <= end]
|
||
chart[label] = [{"date": f"Dzień {(row.effective_date - start).days + 1:03d}", "value": priced_value(row)} for row in period_rows]
|
||
comparison_summary.append(summary(f"{label}: {start.isoformat()} – {end.isoformat()}", period_rows))
|
||
rows = all_rows
|
||
else:
|
||
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()
|
||
for fuel in selected_fuels:
|
||
fuel_rows = [row for row in rows if row.fuel_type == fuel and (fuel != "LPG" or row.region in selected_regions)]
|
||
if fuel == "LPG":
|
||
for region in selected_regions:
|
||
chart[f"LPG · {region}"] = [{"date": row.effective_date.isoformat(), "value": priced_value(row)} for row in fuel_rows if row.region == region]
|
||
else:
|
||
chart[fuel] = [{"date": row.effective_date.isoformat(), "value": priced_value(row)} for row in fuel_rows]
|
||
period_a_from = period_a_to = period_b_from = period_b_to = None
|
||
|
||
summary_difference = None
|
||
if len(comparison_summary) == 2 and comparison_summary[0]["average"] is not None and comparison_summary[1]["average"] is not None:
|
||
first_average = comparison_summary[0]["average"]
|
||
second_average = comparison_summary[1]["average"]
|
||
summary_difference = {
|
||
"value": round(first_average - second_average, 4),
|
||
"percent": round((first_average - second_average) / second_average * 100, 2) if second_average else None,
|
||
}
|
||
|
||
return render_template(
|
||
"orlen.html", rows=rows, chart=chart, mode=mode,
|
||
selected_fuels=selected_fuels, selected_year=selected_year,
|
||
compare_year=compare_year, compare_fuel=compare_fuel,
|
||
comparison_type=comparison_type, comparison_summary=comparison_summary,
|
||
summary_difference=summary_difference, compare_region=compare_region,
|
||
period_a_from=period_a_from, period_a_to=period_a_to,
|
||
period_b_from=period_b_from, period_b_to=period_b_to,
|
||
settings=settings, fuel_types=FUEL_TYPES, include_vat=include_vat,
|
||
regions=POLISH_REGIONS, selected_regions=selected_regions,
|
||
available_years=available_years,
|
||
)
|
||
|
||
@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])
|