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

app.py - optymalizacje

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2025-07-29 11:40:28 +02:00
parent 67d4fd0024
commit 35396afecb
3 changed files with 250 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

357
app.py
View File

@@ -27,9 +27,7 @@ from flask import (
abort,
session,
jsonify,
make_response,
)
from markupsafe import Markup
from flask_sqlalchemy import SQLAlchemy
from flask_login import (
LoginManager,
@@ -46,7 +44,7 @@ from config import Config
from PIL import Image, ExifTags, ImageFilter, ImageOps
from werkzeug.utils import secure_filename
from werkzeug.middleware.proxy_fix import ProxyFix
from sqlalchemy import func, extract, inspect, or_
from sqlalchemy import func, extract, inspect, or_, case
from sqlalchemy.orm import joinedload
from collections import defaultdict, deque
from functools import wraps
@@ -54,12 +52,9 @@ from flask_talisman import Talisman
# OCR
import pytesseract
from collections import Counter
from pytesseract import Output
import logging
app = Flask(__name__)
app.config.from_object(Config)
@@ -173,6 +168,11 @@ class ShoppingList(db.Model):
is_archived = db.Column(db.Boolean, default=False)
is_public = db.Column(db.Boolean, default=True)
# Relacje
items = db.relationship("Item", back_populates="shopping_list", lazy="select")
receipts = db.relationship("Receipt", back_populates="shopping_list", lazy="select")
expenses = db.relationship("Expense", back_populates="shopping_list", lazy="select")
class Item(db.Model):
id = db.Column(db.Integer, primary_key=True)
@@ -193,6 +193,8 @@ class Item(db.Model):
not_purchased_reason = db.Column(db.Text, nullable=True)
position = db.Column(db.Integer, default=0)
shopping_list = db.relationship("ShoppingList", back_populates="items")
class SuggestedProduct(db.Model):
id = db.Column(db.Integer, primary_key=True)
@@ -206,7 +208,8 @@ class Expense(db.Model):
amount = db.Column(db.Float, nullable=False)
added_at = db.Column(db.DateTime, default=datetime.utcnow)
receipt_filename = db.Column(db.String(255), nullable=True)
list = db.relationship("ShoppingList", backref="expenses", lazy=True)
shopping_list = db.relationship("ShoppingList", back_populates="expenses")
class Receipt(db.Model):
@@ -214,10 +217,18 @@ class Receipt(db.Model):
list_id = db.Column(db.Integer, db.ForeignKey("shopping_list.id"), nullable=False)
filename = db.Column(db.String(255), nullable=False)
uploaded_at = db.Column(db.DateTime, default=datetime.utcnow)
shopping_list = db.relationship("ShoppingList", backref="receipts", lazy=True)
filesize = db.Column(db.Integer, nullable=True)
file_hash = db.Column(db.String(64), nullable=True, unique=True)
shopping_list = db.relationship("ShoppingList", back_populates="receipts")
if app.config["SQLALCHEMY_DATABASE_URI"].startswith("sqlite:///"):
db_path = app.config["SQLALCHEMY_DATABASE_URI"].replace("sqlite:///", "", 1)
db_dir = os.path.dirname(db_path)
if db_dir and not os.path.exists(db_dir):
os.makedirs(db_dir, exist_ok=True)
print(f"Utworzono katalog bazy: {db_dir}")
with app.app_context():
db.create_all()
@@ -287,17 +298,33 @@ def allowed_file(filename):
def get_list_details(list_id):
shopping_list = ShoppingList.query.get_or_404(list_id)
items = Item.query.filter_by(list_id=list_id).order_by(Item.position.asc()).all()
expenses = Expense.query.filter_by(list_id=list_id).all()
total_expense = sum(e.amount for e in expenses)
shopping_list = ShoppingList.query.options(
joinedload(ShoppingList.items).joinedload(Item.added_by_user),
joinedload(ShoppingList.expenses),
joinedload(ShoppingList.receipts),
).get_or_404(list_id)
receipts = Receipt.query.filter_by(list_id=list_id).all()
receipt_files = [r.filename for r in receipts]
items = sorted(shopping_list.items, key=lambda i: i.position or 0)
expenses = shopping_list.expenses
total_expense = sum(e.amount for e in expenses) if expenses else 0
receipt_files = [r.filename for r in shopping_list.receipts]
return shopping_list, items, receipt_files, expenses, total_expense
def get_total_expense_for_list(list_id, start_date=None, end_date=None):
query = db.session.query(func.sum(Expense.amount)).filter(
Expense.list_id == list_id
)
if start_date and end_date:
query = query.filter(
Expense.added_at >= start_date, Expense.added_at < end_date
)
return query.scalar() or 0
def generate_share_token(length=8):
return secrets.token_hex(length // 2)
@@ -310,17 +337,26 @@ def check_list_public(shopping_list):
def enrich_list_data(l):
items = Item.query.filter_by(list_id=l.id).all()
l.total_count = len(items)
l.purchased_count = len([i for i in items if i.purchased])
expenses = Expense.query.filter_by(list_id=l.id).all()
l.total_expense = sum(e.amount for e in expenses)
counts = (
db.session.query(
func.count(Item.id),
func.sum(case((Item.purchased == True, 1), else_=0)),
func.sum(Expense.amount),
)
.outerjoin(Expense, Expense.list_id == Item.list_id)
.filter(Item.list_id == l.id)
.first()
)
l.total_count = counts[0] or 0
l.purchased_count = counts[1] or 0
l.total_expense = counts[2] or 0
return l
def save_resized_image(file, path):
try:
# Otwórz i sprawdź poprawność pliku
image = Image.open(file)
image.verify()
file.seek(0)
@@ -364,9 +400,17 @@ def admin_required(f):
def get_progress(list_id):
items = Item.query.filter_by(list_id=list_id).order_by(Item.position.asc()).all()
total_count = len(items)
purchased_count = len([i for i in items if i.purchased])
total_count, purchased_count = (
db.session.query(
func.count(Item.id), func.sum(case((Item.purchased == True, 1), else_=0))
)
.filter(Item.list_id == list_id)
.first()
)
total_count = total_count or 0
purchased_count = purchased_count or 0
percent = (purchased_count / total_count * 100) if total_count > 0 else 0
return purchased_count, total_count, percent
@@ -452,7 +496,7 @@ def handle_crop_receipt(receipt_id, file):
return {"success": False, "error": str(e)}
def get_expenses_aggregated_by_list_created_at(
def get_total_expenses_grouped_by_list_created_at(
user_only=False,
admin=False,
show_all=False,
@@ -461,14 +505,10 @@ def get_expenses_aggregated_by_list_created_at(
end_date=None,
user_id=None,
):
"""
Wspólna logika: sumujemy najnowszy wydatek z każdej listy,
ale do agregacji/filtra bierzemy ShoppingList.created_at!
"""
lists_query = ShoppingList.query
# Uprawnienia
if admin:
# admin widzi wszystko, ewentualnie: dodać filtrowanie wg potrzeb
pass
elif show_all:
lists_query = lists_query.filter(
@@ -480,7 +520,7 @@ def get_expenses_aggregated_by_list_created_at(
else:
lists_query = lists_query.filter(ShoppingList.owner_id == user_id)
# Filtrowanie po created_at listy
# Filtr daty utworzenia listy
if start_date and end_date:
try:
dt_start = datetime.strptime(start_date, "%Y-%m-%d")
@@ -490,34 +530,40 @@ def get_expenses_aggregated_by_list_created_at(
lists_query = lists_query.filter(
ShoppingList.created_at >= dt_start, ShoppingList.created_at < dt_end
)
lists = lists_query.all()
if not lists:
return {"labels": [], "expenses": []}
# Najnowszy wydatek każdej listy
data = []
for sl in lists:
latest_exp = (
Expense.query.filter_by(list_id=sl.id)
.order_by(Expense.added_at.desc())
.first()
list_ids = [l.id for l in lists]
# Suma wszystkich wydatków dla każdej listy
total_expenses = (
db.session.query(
Expense.list_id, func.sum(Expense.amount).label("total_amount")
)
if latest_exp:
data.append({"created_at": sl.created_at, "amount": latest_exp.amount})
.filter(Expense.list_id.in_(list_ids))
.group_by(Expense.list_id)
.all()
)
expense_map = {lid: amt for lid, amt in total_expenses}
# Grupowanie po wybranym zakresie wg utworzenia listy
grouped = defaultdict(float)
for rec in data:
ts = rec["created_at"] or datetime.now(timezone.utc)
if range_type == "monthly":
key = ts.strftime("%Y-%m")
elif range_type == "quarterly":
key = f"{ts.year}-Q{((ts.month - 1) // 3 + 1)}"
elif range_type == "halfyearly":
key = f"{ts.year}-H{1 if ts.month <= 6 else 2}"
elif range_type == "yearly":
key = str(ts.year)
else:
key = ts.strftime("%Y-%m-%d")
grouped[key] += rec["amount"]
for sl in lists:
if sl.id in expense_map:
ts = sl.created_at or datetime.now(timezone.utc)
if range_type == "monthly":
key = ts.strftime("%Y-%m")
elif range_type == "quarterly":
key = f"{ts.year}-Q{((ts.month - 1) // 3 + 1)}"
elif range_type == "halfyearly":
key = f"{ts.year}-H{1 if ts.month <= 6 else 2}"
elif range_type == "yearly":
key = str(ts.year)
else:
key = ts.strftime("%Y-%m-%d")
grouped[key] += expense_map[sl.id]
labels = sorted(grouped)
expenses = [round(grouped[l], 2) for l in labels]
@@ -882,6 +928,7 @@ def main_page():
)
return query
# Pobranie list
if current_user.is_authenticated:
user_lists = (
date_filter(
@@ -934,8 +981,47 @@ def main_page():
.all()
)
for l in user_lists + public_lists + archived_lists:
enrich_list_data(l)
all_lists = user_lists + public_lists + archived_lists
all_ids = [l.id for l in all_lists]
if all_ids:
# statystyki produktów
stats = (
db.session.query(
Item.list_id,
func.count(Item.id).label("total_count"),
func.sum(case((Item.purchased == True, 1), else_=0)).label(
"purchased_count"
),
)
.filter(Item.list_id.in_(all_ids))
.group_by(Item.list_id)
.all()
)
stats_map = {
s.list_id: (s.total_count or 0, s.purchased_count or 0) for s in stats
}
# ostatnia kwota (w tym przypadku max = suma z ostatniego zapisu)
latest_expenses_map = dict(
db.session.query(
Expense.list_id, func.coalesce(func.max(Expense.amount), 0)
)
.filter(Expense.list_id.in_(all_ids))
.group_by(Expense.list_id)
.all()
)
for l in all_lists:
total_count, purchased_count = stats_map.get(l.id, (0, 0))
l.total_count = total_count
l.purchased_count = purchased_count
l.total_expense = latest_expenses_map.get(l.id, 0)
else:
for l in all_lists:
l.total_count = 0
l.purchased_count = 0
l.total_expense = 0
return render_template(
"main.html",
@@ -1197,7 +1283,9 @@ def view_list(list_id):
for item in items:
if item.added_by != shopping_list.owner_id:
item.added_by_display = item.added_by_user.username if item.added_by_user else "?"
item.added_by_display = (
item.added_by_user.username if item.added_by_user else "?"
)
else:
item.added_by_display = None
@@ -1226,11 +1314,12 @@ def user_expenses():
start = None
end = None
expenses_query = Expense.query.join(
ShoppingList, Expense.list_id == ShoppingList.id
).options(joinedload(Expense.list))
expenses_query = Expense.query.options(
joinedload(Expense.shopping_list).joinedload(ShoppingList.owner),
joinedload(Expense.shopping_list).joinedload(ShoppingList.expenses),
).join(ShoppingList, Expense.list_id == ShoppingList.id)
# Jeśli show_all to False, filtruj tylko po bieżącym użytkowniku
# Filtry dostępu
if not show_all:
expenses_query = expenses_query.filter(ShoppingList.owner_id == current_user.id)
else:
@@ -1240,6 +1329,7 @@ def user_expenses():
)
)
# Filtr daty
if start_date_str and end_date_str:
try:
start = datetime.strptime(start_date_str, "%Y-%m-%d")
@@ -1250,37 +1340,43 @@ def user_expenses():
except ValueError:
flash("Błędny zakres dat", "danger")
# Pobranie wszystkich wydatków z powiązanymi listami
expenses = expenses_query.order_by(Expense.added_at.desc()).all()
# Zbiorcze sumowanie wydatków per lista w SQL
list_ids = {e.list_id for e in expenses}
lists = (
ShoppingList.query.filter(ShoppingList.id.in_(list_ids))
.order_by(ShoppingList.created_at.desc())
.all()
)
totals_map = {}
if list_ids:
totals = (
db.session.query(
Expense.list_id, func.sum(Expense.amount).label("total_expense")
)
.filter(Expense.list_id.in_(list_ids))
.group_by(Expense.list_id)
.all()
)
totals_map = {t.list_id: t.total_expense or 0 for t in totals}
# Tabela wydatków
expense_table = [
{
"title": e.list.title if e.list else "Nieznana",
"title": e.shopping_list.title if e.shopping_list else "Nieznana",
"amount": e.amount,
"added_at": e.added_at,
}
for e in expenses
]
# Lista z danymi i sumami
lists_data = [
{
"id": l.id,
"title": l.title,
"created_at": l.created_at,
"total_expense": sum(
e.amount
for e in l.expenses
if (not start or not end) or (e.added_at >= start and e.added_at < end)
),
"total_expense": totals_map.get(l.id, 0),
"owner_username": l.owner.username if l.owner else "?",
}
for l in lists
for l in {e.shopping_list for e in expenses if e.shopping_list}
]
return render_template(
@@ -1299,7 +1395,7 @@ def user_expenses_data():
end_date = request.args.get("end_date")
show_all = request.args.get("show_all", "false").lower() == "true"
result = get_expenses_aggregated_by_list_created_at(
result = get_total_expenses_grouped_by_list_created_at(
user_only=True,
admin=False,
show_all=show_all,
@@ -1324,13 +1420,16 @@ def shared_list(token=None, list_id=None):
list_id = shopping_list.id
total_expense = get_total_expense_for_list(list_id)
shopping_list, items, receipt_files, expenses, total_expense = get_list_details(
list_id
)
for item in items:
if item.added_by != shopping_list.owner_id:
item.added_by_display = item.added_by_user.username if item.added_by_user else "?"
item.added_by_display = (
item.added_by_user.username if item.added_by_user else "?"
)
else:
item.added_by_display = None
@@ -1456,7 +1555,7 @@ def upload_receipt(list_id):
except ValueError as e:
return _receipt_error(str(e))
filesize = os.path.getsize(file_path) if os.path.exists(file_path) else None
filesize = os.path.getsize(file_path)
uploaded_at = datetime.now(timezone.utc)
new_receipt = Receipt(
@@ -1625,21 +1724,59 @@ def crop_receipt_user():
def admin_panel():
now = datetime.now(timezone.utc)
# Liczniki globalne
user_count = User.query.count()
list_count = ShoppingList.query.count()
item_count = Item.query.count()
all_lists = ShoppingList.query.options(db.joinedload(ShoppingList.owner)).all()
all_files = os.listdir(app.config["UPLOAD_FOLDER"])
all_lists = ShoppingList.query.options(
joinedload(ShoppingList.owner),
joinedload(ShoppingList.items),
joinedload(ShoppingList.receipts),
joinedload(ShoppingList.expenses),
).all()
all_ids = [l.id for l in all_lists]
stats_map = {}
latest_expenses_map = {}
if all_ids:
# Statystyki produktów
stats = (
db.session.query(
Item.list_id,
func.count(Item.id).label("total_count"),
func.sum(case((Item.purchased == True, 1), else_=0)).label(
"purchased_count"
),
)
.filter(Item.list_id.in_(all_ids))
.group_by(Item.list_id)
.all()
)
stats_map = {
s.list_id: (s.total_count or 0, s.purchased_count or 0) for s in stats
}
# Pobranie ostatnich kwot dla wszystkich list w jednym zapytaniu
latest_expenses_map = dict(
db.session.query(
Expense.list_id, func.coalesce(func.max(Expense.amount), 0)
)
.filter(Expense.list_id.in_(all_ids))
.group_by(Expense.list_id)
.all()
)
enriched_lists = []
for l in all_lists:
enrich_list_data(l)
items = Item.query.filter_by(list_id=l.id).all()
total_count = l.total_count
purchased_count = l.purchased_count
total_count, purchased_count = stats_map.get(l.id, (0, 0))
percent = (purchased_count / total_count * 100) if total_count > 0 else 0
comments_count = len([i for i in items if i.note and i.note.strip() != ""])
receipts_count = Receipt.query.filter_by(list_id=l.id).count()
comments_count = sum(1 for i in l.items if i.note and i.note.strip() != "")
receipts_count = len(l.receipts)
total_expense = latest_expenses_map.get(l.id, 0)
if l.is_temporary and l.expires_at:
expires_at = l.expires_at
@@ -1657,11 +1794,12 @@ def admin_panel():
"percent": round(percent),
"comments_count": comments_count,
"receipts_count": receipts_count,
"total_expense": l.total_expense,
"total_expense": total_expense,
"expired": is_expired,
}
)
# Top produkty
top_products = (
db.session.query(Item.name, func.count(Item.id).label("count"))
.filter(Item.purchased.is_(True))
@@ -1672,8 +1810,9 @@ def admin_panel():
)
purchased_items_count = Item.query.filter_by(purchased=True).count()
total_expense_sum = db.session.query(func.sum(Expense.amount)).scalar() or 0
# 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
@@ -1682,21 +1821,19 @@ def admin_panel():
db.session.query(func.sum(Expense.amount))
.filter(extract("year", Expense.added_at) == current_year)
.scalar()
or 0
)
) 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
)
) or 0
# Statystyki systemowe
process = psutil.Process(os.getpid())
app_mem = process.memory_info().rss // (1024 * 1024) # MB
# Engine info
db_engine = db.engine
db_info = {
"engine": db_engine.name,
@@ -1704,11 +1841,9 @@ def admin_panel():
"url": str(db_engine.url).split("?")[0],
}
# Tabele
inspector = inspect(db_engine)
table_count = len(inspector.get_table_names())
# Rekordy (szybkie zliczenie)
record_total = (
db.session.query(func.count(User.id)).scalar()
+ db.session.query(func.count(ShoppingList.id)).scalar()
@@ -1717,7 +1852,6 @@ def admin_panel():
+ db.session.query(func.count(Expense.id)).scalar()
)
# Uptime
uptime_minutes = int(
(datetime.now(timezone.utc) - app_start_time).total_seconds() // 60
)
@@ -1999,22 +2133,23 @@ def delete_selected_lists():
@login_required
@admin_required
def edit_list(list_id):
l = db.session.get(ShoppingList, list_id)
# Pobieramy listę z powiązanymi danymi jednym zapytaniem
l = (
db.session.query(ShoppingList)
.options(
joinedload(ShoppingList.expenses),
joinedload(ShoppingList.receipts),
joinedload(ShoppingList.owner),
joinedload(ShoppingList.items),
)
.get(list_id)
)
if l is None:
abort(404)
expenses = Expense.query.filter_by(list_id=list_id).all()
total_expense = sum(e.amount for e in expenses)
users = User.query.all()
items = (
db.session.query(Item).filter_by(list_id=list_id).order_by(Item.id.desc()).all()
)
receipts = (
Receipt.query.filter_by(list_id=list_id)
.order_by(Receipt.uploaded_at.desc())
.all()
)
# Suma wydatków z listy
total_expense = get_total_expense_for_list(l.id)
if request.method == "POST":
action = request.form.get("action")
@@ -2064,7 +2199,7 @@ def edit_list(list_id):
if new_amount_str:
try:
new_amount = float(new_amount_str)
for expense in expenses:
for expense in l.expenses:
db.session.delete(expense)
db.session.commit()
db.session.add(Expense(list_id=list_id, amount=new_amount))
@@ -2184,6 +2319,11 @@ def edit_list(list_id):
flash("Nie znaleziono produktu", "danger")
return redirect(url_for("edit_list", list_id=list_id))
# Dane do widoku
users = User.query.all()
items = l.items
receipts = l.receipts
return render_template(
"admin/edit_list.html",
list=l,
@@ -2191,7 +2331,6 @@ def edit_list(list_id):
users=users,
items=items,
receipts=receipts,
upload_folder=app.config["UPLOAD_FOLDER"],
)
@@ -2269,7 +2408,7 @@ def admin_expenses_data():
start_date = request.args.get("start_date")
end_date = request.args.get("end_date")
result = get_expenses_aggregated_by_list_created_at(
result = get_total_expenses_grouped_by_list_created_at(
user_only=False,
admin=True,
show_all=True,