gru/lista_zakupowa_live
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity

poprawki w panelu, kategorie na wykresach i inne

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2025-07-31 10:37:44 +02:00
parent 29ccd252b8
commit 0a44753eb2
6 changed files with 621 additions and 292 deletions
Download Patch File Download Diff File Expand all files Collapse all files

356
app.py
View File

@@ -89,9 +89,9 @@ referrer_policy = app.config.get("REFERRER_POLICY")
if referrer_policy:
talisman_kwargs["referrer_policy"] = referrer_policy
talisman = Talisman(app,
session_cookie_secure=app.config["SESSION_COOKIE_SECURE"],
**talisman_kwargs)
talisman = Talisman(
app, session_cookie_secure=app.config["SESSION_COOKIE_SECURE"], **talisman_kwargs
)
register_heif_opener() # pillow_heif dla HEIC
SQLALCHEMY_ECHO = True
@@ -109,7 +109,7 @@ SESSION_COOKIE_SECURE = app.config.get("SESSION_COOKIE_SECURE")
app.config["COMPRESS_ALGORITHM"] = ["zstd", "br", "gzip", "deflate"]
app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(minutes=SESSION_TIMEOUT_MINUTES)
#app.config["SESSION_COOKIE_SECURE"] = True if app.config.get("SESSION_COOKIE_SECURE") is True else False
# app.config["SESSION_COOKIE_SECURE"] = True if app.config.get("SESSION_COOKIE_SECURE") is True else False
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1)
DEBUG_MODE = app.config.get("DEBUG_MODE", False)
@@ -166,14 +166,23 @@ class User(UserMixin, db.Model):
# Tabela pośrednia
shopping_list_category = db.Table(
"shopping_list_category",
db.Column("shopping_list_id", db.Integer, db.ForeignKey("shopping_list.id"), primary_key=True),
db.Column("category_id", db.Integer, db.ForeignKey("category.id"), primary_key=True)
db.Column(
"shopping_list_id",
db.Integer,
db.ForeignKey("shopping_list.id"),
primary_key=True,
),
db.Column(
"category_id", db.Integer, db.ForeignKey("category.id"), primary_key=True
),
)
class Category(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100), unique=True, nullable=False)
class ShoppingList(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(150), nullable=False)
@@ -198,9 +207,10 @@ class ShoppingList(db.Model):
categories = db.relationship(
"Category",
secondary=shopping_list_category,
backref=db.backref("shopping_lists", lazy="dynamic")
backref=db.backref("shopping_lists", lazy="dynamic"),
)
class Item(db.Model):
id = db.Column(db.Integer, primary_key=True)
list_id = db.Column(db.Integer, db.ForeignKey("shopping_list.id"))
@@ -255,9 +265,14 @@ def handle_db_error(e):
app.logger.error(f"[Błąd DB] {e}")
if request.accept_mimetypes.best == "application/json":
return jsonify({
"error": "Baza danych jest obecnie niedostępna. Spróbuj ponownie później."
}), 503
return (
jsonify(
{
"error": "Baza danych jest obecnie niedostępna. Spróbuj ponownie później."
}
),
503,
)
return (
render_template(
@@ -288,6 +303,7 @@ def check_password(stored_hash, password_input):
return False
return False
def set_authorized_cookie(response):
secure_flag = app.config["SESSION_COOKIE_SECURE"] # wartość z config.py
@@ -301,7 +317,7 @@ def set_authorized_cookie(response):
secure=secure_flag,
httponly=True,
samesite="Lax",
path="/"
path="/",
)
return response
@@ -331,32 +347,25 @@ with app.app_context():
print(f"[INFO] Zmieniono hasło admina '{admin_username}' z konfiguracji.")
db.session.commit()
else:
db.session.add(User(
username=admin_username,
password_hash=password_hash,
is_admin=True
))
db.session.add(
User(username=admin_username, password_hash=password_hash, is_admin=True)
)
db.session.commit()
# --- Predefiniowane kategorie ---
default_categories = [
"Spożywcze", "Budowlane", "Zabawki", "Chemia", "Inne",
"Elektronika", "Odzież i obuwie", "Artykuły biurowe",
"Kosmetyki i higiena", "Motoryzacja", "Ogród i rośliny",
"Zwierzęta", "Sprzęt sportowy", "Książki i prasa",
"Narzędzia i majsterkowanie", "RTV / AGD", "Apteka i suplementy",
"Artykuły dekoracyjne", "Gry i hobby", "Usługi", "Pieczywo"
]
default_categories = app.config["DEFAULT_CATEGORIES"]
# Pobierz istniejące nazwy z bazy, ignorując puste/niewłaściwe rekordy
existing_names = {
c.name for c in Category.query.filter(Category.name.isnot(None)).all()
}
# Znajdź brakujące
missing = [cat for cat in default_categories if cat not in existing_names]
# ignorujemy wielkość liter przy porównaniu
existing_names_lower = {name.lower() for name in existing_names}
missing = [
cat for cat in default_categories if cat.lower() not in existing_names_lower
]
# Dodaj tylko brakujące
if missing:
db.session.add_all(Category(name=cat) for cat in missing)
db.session.commit()
@@ -656,7 +665,7 @@ def get_total_expenses_grouped_by_list_created_at(
if category_id:
lists_query = lists_query.join(
shopping_list_category,
shopping_list_category.c.shopping_list_id == ShoppingList.id
shopping_list_category.c.shopping_list_id == ShoppingList.id,
).filter(shopping_list_category.c.category_id == category_id)
if start_date and end_date:
@@ -737,6 +746,158 @@ def recalculate_filesizes(receipt_id: int = None):
return updated, unchanged, not_found
def get_admin_expense_summary():
now = datetime.now(timezone.utc)
current_year = now.year
current_month = now.month
def calc_sum(base_query):
total = base_query.scalar() or 0
year_total = (
base_query.filter(
extract("year", Expense.added_at) == current_year
).scalar()
or 0
)
month_total = (
base_query.filter(extract("year", Expense.added_at) == current_year)
.filter(extract("month", Expense.added_at) == current_month)
.scalar()
or 0
)
return {"total": total, "year": year_total, "month": month_total}
# baza wspólna
base = db.session.query(func.sum(Expense.amount)).join(
ShoppingList, ShoppingList.id == Expense.list_id
)
# wszystkie listy
all_lists = calc_sum(base)
# aktywne listy
active_lists = calc_sum(
base.filter(
ShoppingList.is_archived == False,
or_(ShoppingList.expires_at == None, ShoppingList.expires_at > now),
)
)
# archiwalne
archived_lists = calc_sum(base.filter(ShoppingList.is_archived == True))
# wygasłe
expired_lists = calc_sum(
base.filter(
ShoppingList.is_archived == False,
ShoppingList.expires_at != None,
ShoppingList.expires_at <= now,
)
)
return {
"all": all_lists,
"active": active_lists,
"archived": archived_lists,
"expired": expired_lists,
}
def category_to_color(name):
"""Generuje powtarzalny pastelowy kolor HEX na podstawie nazwy kategorii."""
hash_val = int(hashlib.md5(name.encode("utf-8")).hexdigest(), 16)
r = (hash_val & 0xFF0000) >> 16
g = (hash_val & 0x00FF00) >> 8
b = hash_val & 0x0000FF
# Rozjaśnienie (pastel)
r = (r + 255) // 2
g = (g + 255) // 2
b = (b + 255) // 2
return f"#{r:02x}{g:02x}{b:02x}"
def get_total_expenses_grouped_by_category(
show_all, range_type, start_date, end_date, user_id
):
lists_query = ShoppingList.query
if show_all:
lists_query = lists_query.filter(
or_(ShoppingList.owner_id == user_id, ShoppingList.is_public == True)
)
else:
lists_query = lists_query.filter(ShoppingList.owner_id == user_id)
if start_date and end_date:
try:
dt_start = datetime.strptime(start_date, "%Y-%m-%d")
dt_end = datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1)
except Exception:
return {"error": "Błędne daty"}
lists_query = lists_query.filter(
ShoppingList.created_at >= dt_start, ShoppingList.created_at < dt_end
)
lists = lists_query.options(joinedload(ShoppingList.categories)).all()
if not lists:
return {"labels": [], "datasets": []}
data_map = defaultdict(lambda: defaultdict(float))
all_labels = set()
for l in lists:
total_expense = (
db.session.query(func.sum(Expense.amount))
.filter(Expense.list_id == l.id)
.scalar()
) or 0
if total_expense <= 0:
continue
if range_type == "monthly":
key = l.created_at.strftime("%Y-%m")
elif range_type == "quarterly":
key = f"{l.created_at.year}-Q{((l.created_at.month - 1) // 3 + 1)}"
elif range_type == "halfyearly":
key = f"{l.created_at.year}-H{1 if l.created_at.month <= 6 else 2}"
elif range_type == "yearly":
key = str(l.created_at.year)
else:
key = l.created_at.strftime("%Y-%m-%d")
all_labels.add(key)
if not l.categories:
data_map[key]["Inne"] += total_expense
else:
for c in l.categories:
data_map[key][c.name] += total_expense
labels = sorted(all_labels)
categories_with_expenses = sorted(
{
cat
for cat_data in data_map.values()
for cat, value in cat_data.items()
if value > 0
}
)
datasets = []
for cat in categories_with_expenses:
datasets.append(
{
"label": cat,
"data": [round(data_map[label].get(cat, 0), 2) for label in labels],
"backgroundColor": category_to_color(cat),
}
)
return {"labels": labels, "datasets": datasets}
############# OCR ###########################
@@ -1143,8 +1304,7 @@ def main_page():
# ostatnia kwota (w tym przypadku max = suma z ostatniego zapisu)
latest_expenses_map = dict(
db.session.query(
Expense.list_id,
func.coalesce(func.sum(Expense.amount), 0)
Expense.list_id, func.coalesce(func.sum(Expense.amount), 0)
)
.filter(Expense.list_id.in_(all_ids))
.group_by(Expense.list_id)
@@ -1185,7 +1345,10 @@ def system_auth():
next_page = request.args.get("next") or url_for("main_page")
if is_ip_blocked(ip):
flash("Przekroczono limit prób logowania. Dostęp zablokowany na 1 godzinę.", "danger")
flash(
"Przekroczono limit prób logowania. Dostęp zablokowany na 1 godzinę.",
"danger",
)
return render_template("system_auth.html"), 403
if request.method == "POST":
@@ -1196,7 +1359,10 @@ def system_auth():
else:
register_failed_attempt(ip)
if is_ip_blocked(ip):
flash("Przekroczono limit prób logowania. Dostęp zablokowany na 1 godzinę.", "danger")
flash(
"Przekroczono limit prób logowania. Dostęp zablokowany na 1 godzinę.",
"danger",
)
return render_template("system_auth.html"), 403
remaining = attempts_remaining(ip)
flash(f"Nieprawidłowe hasło. Pozostało {remaining} prób.", "warning")
@@ -1306,11 +1472,10 @@ def edit_my_list(list_id):
list=l,
receipts=receipts,
categories=categories,
selected_categories=selected_categories_ids
selected_categories=selected_categories_ids,
)
@app.route("/delete_user_list/<int:list_id>", methods=["POST"])
@login_required
def delete_user_list(list_id):
@@ -1452,7 +1617,28 @@ def user_expenses():
category_id = request.args.get("category_id", type=int)
show_all = request.args.get("show_all", "true").lower() == "true"
categories = Category.query.order_by(Category.name.asc()).all()
categories = (
Category.query.join(
shopping_list_category, shopping_list_category.c.category_id == Category.id
)
.join(
ShoppingList, ShoppingList.id == shopping_list_category.c.shopping_list_id
)
.join(Expense, Expense.list_id == ShoppingList.id)
.filter(
or_(
ShoppingList.owner_id == current_user.id,
(
ShoppingList.is_public == True
if show_all
else ShoppingList.owner_id == current_user.id
),
)
)
.distinct()
.order_by(Category.name.asc())
.all()
)
start = None
end = None
@@ -1468,15 +1654,14 @@ def user_expenses():
else:
expenses_query = expenses_query.filter(
or_(
ShoppingList.owner_id == current_user.id,
ShoppingList.is_public == True
ShoppingList.owner_id == current_user.id, ShoppingList.is_public == True
)
)
if category_id:
expenses_query = expenses_query.join(
shopping_list_category,
shopping_list_category.c.shopping_list_id == ShoppingList.id
shopping_list_category.c.shopping_list_id == ShoppingList.id,
).filter(shopping_list_category.c.category_id == category_id)
if start_date_str and end_date_str:
@@ -1522,7 +1707,7 @@ def user_expenses():
"created_at": l.created_at,
"total_expense": totals_map.get(l.id, 0),
"owner_username": l.owner.username if l.owner else "?",
"categories": [c.id for c in l.categories]
"categories": [c.id for c in l.categories],
}
for l in {e.shopping_list for e in expenses if e.shopping_list}
]
@@ -1545,17 +1730,27 @@ def user_expenses_data():
end_date = request.args.get("end_date")
show_all = request.args.get("show_all", "true").lower() == "true"
category_id = request.args.get("category_id", type=int)
by_category = request.args.get("by_category", "false").lower() == "true"
result = get_total_expenses_grouped_by_list_created_at(
user_only=True,
admin=False,
show_all=show_all,
range_type=range_type,
start_date=start_date,
end_date=end_date,
user_id=current_user.id,
category_id=category_id
)
if by_category:
result = get_total_expenses_grouped_by_category(
show_all=show_all,
range_type=range_type,
start_date=start_date,
end_date=end_date,
user_id=current_user.id,
)
else:
result = get_total_expenses_grouped_by_list_created_at(
user_only=True,
admin=False,
show_all=show_all,
range_type=range_type,
start_date=start_date,
end_date=end_date,
user_id=current_user.id,
category_id=category_id,
)
if "error" in result:
return jsonify({"error": result["error"]}), 400
@@ -1640,12 +1835,14 @@ def all_products():
)
top_products = (
top_products_query.order_by(
SuggestedProduct.name.asc(), # musi być pierwsze
SuggestedProduct.usage_count.desc(),
db.session.query(
func.lower(Item.name).label("name"), func.sum(Item.quantity).label("count")
)
.distinct(SuggestedProduct.name)
.limit(20)
.join(ShoppingList, ShoppingList.id == Item.list_id)
.filter(Item.purchased.is_(True))
.group_by(func.lower(Item.name))
.order_by(func.sum(Item.quantity).desc())
.limit(5)
.all()
)
@@ -1887,6 +2084,7 @@ def admin_panel():
joinedload(ShoppingList.items),
joinedload(ShoppingList.receipts),
joinedload(ShoppingList.expenses),
joinedload(ShoppingList.categories),
).all()
all_ids = [l.id for l in all_lists]
@@ -1914,15 +2112,13 @@ def admin_panel():
latest_expenses_map = dict(
db.session.query(
Expense.list_id,
func.coalesce(func.sum(Expense.amount), 0)
Expense.list_id, func.coalesce(func.sum(Expense.amount), 0)
)
.filter(Expense.list_id.in_(all_ids))
.group_by(Expense.list_id)
.all()
)
enriched_lists = []
for l in all_lists:
total_count, purchased_count = stats_map.get(l.id, (0, 0))
@@ -1949,6 +2145,7 @@ def admin_panel():
"receipts_count": receipts_count,
"total_expense": total_expense,
"expired": is_expired,
"categories": l.categories,
}
)
@@ -1964,24 +2161,8 @@ def admin_panel():
purchased_items_count = Item.query.filter_by(purchased=True).count()
# Podsumowania wydatków globalnych
total_expense_sum = db.session.query(func.sum(Expense.amount)).scalar() or 0
current_time = datetime.now(timezone.utc)
current_year = current_time.year
current_month = current_time.month
year_expense_sum = (
db.session.query(func.sum(Expense.amount))
.filter(extract("year", Expense.added_at) == current_year)
.scalar()
) or 0
month_expense_sum = (
db.session.query(func.sum(Expense.amount))
.filter(extract("year", Expense.added_at) == current_year)
.filter(extract("month", Expense.added_at) == current_month)
.scalar()
) or 0
# Nowe podsumowanie wydatków
expense_summary = get_admin_expense_summary()
# Statystyki systemowe
process = psutil.Process(os.getpid())
@@ -1996,9 +2177,7 @@ def admin_panel():
inspector = inspect(db_engine)
table_count = len(inspector.get_table_names())
record_total = get_total_records()
uptime_minutes = int(
(datetime.now(timezone.utc) - app_start_time).total_seconds() // 60
)
@@ -2011,9 +2190,7 @@ def admin_panel():
purchased_items_count=purchased_items_count,
enriched_lists=enriched_lists,
top_products=top_products,
total_expense_sum=total_expense_sum,
year_expense_sum=year_expense_sum,
month_expense_sum=month_expense_sum,
expense_summary=expense_summary,
now=now,
python_version=sys.version,
system_info=platform.platform(),
@@ -2025,7 +2202,6 @@ def admin_panel():
)
@app.route("/admin/delete_list/<int:list_id>")
@login_required
@admin_required
@@ -2291,7 +2467,7 @@ def edit_list(list_id):
joinedload(ShoppingList.owner),
joinedload(ShoppingList.items),
joinedload(ShoppingList.categories),
]
],
)
if l is None:
@@ -2378,7 +2554,7 @@ def edit_list(list_id):
db.session.commit()
flash("Zapisano zmiany listy", "success")
return redirect(url_for("edit_list", list_id=list_id))
elif action == "add_item":
item_name = request.form.get("item_name", "").strip()
quantity_str = request.form.get("quantity", "1")
@@ -2487,7 +2663,7 @@ def edit_list(list_id):
items=items,
receipts=receipts,
categories=categories,
selected_categories=selected_categories_ids
selected_categories=selected_categories_ids,
)
@@ -2640,7 +2816,11 @@ def recalculate_filesizes_all():
@login_required
@admin_required
def admin_mass_edit_categories():
lists = ShoppingList.query.options(joinedload(ShoppingList.categories)).order_by(ShoppingList.created_at.desc()).all()
lists = (
ShoppingList.query.options(joinedload(ShoppingList.categories))
.order_by(ShoppingList.created_at.desc())
.all()
)
categories = Category.query.order_by(Category.name.asc()).all()
if request.method == "POST":
@@ -2654,7 +2834,9 @@ def admin_mass_edit_categories():
flash("Zaktualizowano kategorie dla wybranych list", "success")
return redirect(url_for("admin_mass_edit_categories"))
return render_template("admin/mass_edit_categories.html", lists=lists, categories=categories)
return render_template(
"admin/mass_edit_categories.html", lists=lists, categories=categories
)
@app.route("/healthcheck")