app.py

from flask import Flask, render_template, redirect, url_for, request, Response
from pathlib import Path
from db import db, initialize_database
from models import Product, Customer, Category, Order, ProductOrder
from sqlalchemy import select
from datetime import datetime as dt
from routers import api_bp
import csv
import io

app = Flask(__name__)


app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///store.db"
app.instance_path = Path("./databases").resolve()


db.init_app(app)
with app.app_context():
    initialize_database(Product, Category, Customer, Order, ProductOrder)


app.register_blueprint(api_bp, url_prefix="/api")

# Routes
@app.route("/")
def home():
    try:
        product_count = db.session.execute(db.select(db.func.count(Product.id))).scalar_one()
        category_count = db.session.execute(db.select(db.func.count(Category.id))).scalar_one()
        customer_count = db.session.execute(db.select(db.func.count(Customer.id))).scalar_one()
        order_count = db.session.execute(db.select(db.func.count(Order.id))).scalar_one()
        pending_orders = db.session.execute(db.select(db.func.count(Order.id)).where(Order.complete == None)).scalar_one()
        total_revenue = db.session.execute(db.select(db.func.sum(Order.amount)).where(Order.complete != None)).scalar_one() or 0
        low_stock_count = db.session.execute(
            db.select(db.func.count(Product.id)).where(Product.available < 5)
        ).scalar_one()
    except:
        product_count = category_count = customer_count = order_count = 0
        pending_orders = total_revenue = low_stock_count = 0

    return render_template(
        "index.html",
        product_count=product_count,
        category_count=category_count,
        customer_count=customer_count,
        order_count=order_count,
        pending_orders=pending_orders,
        total_revenue=total_revenue,
        low_stock_count=low_stock_count,
    )


@app.route("/products")
def products():
    search_query = request.args.get("search", "")
    sort = request.args.get("sort")
    min_price = request.args.get("min_price", type=float)
    max_price = request.args.get("max_price", type=float)
    in_stock = request.args.get("in_stock") == "on"
    category_filter = request.args.get("category")

    statement = select(Product).where(Product.category_rel.has())

    if search_query:
        statement = statement.where(Product.name.ilike(f"%{search_query}%"))
    if min_price is not None:
        statement = statement.where(Product.price >= min_price)
    if max_price is not None:
        statement = statement.where(Product.price <= max_price)
    if in_stock:
        statement = statement.where(Product.available > 0)
    if category_filter:
        statement = statement.where(Product.category_rel.has(Category.name == category_filter))

    if sort == "price_low":
        statement = statement.order_by(Product.price)
    elif sort == "price_high":
        statement = statement.order_by(Product.price.desc())
    elif sort == "name":
        statement = statement.order_by(Product.name)
    elif sort == "availability":
        statement = statement.order_by(Product.available.desc())

    products_select = db.session.execute(statement).scalars().all()

    for product in products_select:
        try:
            product.category = product.category_rel.name
        except AttributeError:
            product.category = "Uncategorized"

    categories = db.session.execute(select(Category.name)).scalars().all()

    return render_template(
        "products.html",
        products=products_select,
        categories=categories,
        search_query=search_query,
        min_price=min_price,
        max_price=max_price,
        in_stock=in_stock,
        category_filter=category_filter,
        sort=sort,
    )


@app.route("/categories")
def categories():
    categories_select = db.session.execute(select(Category)).scalars().all()
    category_names = [category.name for category in categories_select]

    return render_template("categories.html", categories=category_names)


@app.route("/categories/<category>")
def category(category):
    products_select = db.session.execute(
        select(Product).where(Product.category_rel.has(Category.name == category))
    ).scalars().all()

    for product in products_select:
        try:
            product.category = product.category_rel.name
        except AttributeError:
            product.category = "Uncategorized"

    return render_template("category.html", category=category, products=products_select)


@app.route("/customers")
def customers():
    search_query = request.args.get("search", "")
    sort = request.args.get("sort")

    statement = select(Customer)

    if search_query:
        statement = statement.where(
            (Customer.nickname.ilike(f"%{search_query}%"))
            | (Customer.email.ilike(f"%{search_query}%"))
        )

    if sort == "nickname":
        statement = statement.order_by(Customer.nickname)
    elif sort == "email":
        statement = statement.order_by(Customer.email)
    elif sort == "date":
        statement = statement.order_by(Customer.created_at.desc())
    elif sort == "id":
        statement = statement.order_by(Customer.id)

    customers = db.session.execute(statement).scalars().all()

    return render_template(
        "customers.html", customers=customers, search_query=search_query, sort=sort
    )


@app.route("/customers/<customer_id>")
def customer(customer_id):
    customer = db.session.execute(select(Customer).where(Customer.id == customer_id)).scalar_one()
    return render_template("customer.html", customer=customer)


@app.route("/products/<product_id>")
def product(product_id):
    product = db.session.execute(select(Product).where(Product.id == product_id)).scalar_one()
    try:
        product.category = product.category_rel.name
    except AttributeError:
        product.category = "Uncategorized"

    return render_template("product.html", product=product)


@app.route("/orders")
def orders():
    sort = request.args.get("sort")
    filter_status = request.args.get("filter")

    statement = db.select(Order)

    if filter_status == "completed":
        statement = statement.where(Order.complete != None)
    elif filter_status == "pending":
        statement = statement.where(Order.complete == None)

    if sort == "id":
        statement = statement.order_by(Order.id)
    elif sort == "date":
        statement = statement.order_by(Order.created.desc())
    elif sort == "amount":
        statement = statement.order_by(Order.amount.desc())
    elif sort == "status":
        statement = statement.order_by(Order.complete.is_(None))

    orders = db.session.execute(statement).scalars().all()

    return render_template("orders.html", orders=orders)


@app.route("/orders/<order_id>")
def order(order_id):
    order = db.session.execute(select(Order).where(Order.id == order_id)).scalar_one()
    return render_template("order.html", order=order)


@app.route("/orders/<int:id>/complete", methods=["POST"])
def complete_order(id):
    print("Completing order with ID:", id)

    order = db.session.execute(select(Order).where(Order.id == id)).scalar_one()

    for po in order.products:
        if po.product.available < po.quantity:
            return redirect(url_for("order", order_id=id))

    estimated_amount = 0
    for po in order.products:
        estimated_amount += po.product.price * po.quantity
        po.product.available -= po.quantity

    order.complete = dt.now()
    order.amount = estimated_amount

    db.session.commit()

    return redirect(url_for("order", order_id=id))


@app.route("/export/products")
def export_products():
    products = db.session.execute(select(Product).where(Product.category_rel.has())).scalars().all()

    output = io.StringIO()
    writer = csv.writer(output)

    writer.writerow(["ID", "Name", "Price", "Available", "Category"])

    for product in products:
        try:
            category = product.category_rel.name
        except AttributeError:
            category = "Uncategorized"

        writer.writerow(
            [product.id, product.name, f"{product.price:.2f}", product.available, category]
        )

    output.seek(0)
    return Response(
        output,
        mimetype="text/csv",
        headers={"Content-Disposition": "attachment;filename=products.csv"},
    )


@app.route("/export/orders")
def export_orders():
    orders = db.session.execute(select(Order)).scalars().all()

    output = io.StringIO()
    writer = csv.writer(output)

    writer.writerow(["Order ID", "Customer", "Date", "Status", "Amount", "Products"])

    for order in orders:
        products_list = ", ".join(
            [f"{po.product.name} (x{po.quantity})" for po in order.products]
        )

        writer.writerow(
            [
                order.id,
                order.customer.name,
                order.created.strftime("%Y-%m-%d %H:%M"),
                "Completed" if order.complete else "Pending",
                f"{order.amount:.2f}" if order.complete else f"{order.estimate():.2f} (est.)",
                products_list,
            ]
        )

    output.seek(0)
    return Response(
        output,
        mimetype="text/csv",
        headers={"Content-Disposition": "attachment;filename=orders.csv"},
    )


@app.route("/export/customers")
def export_customers():
    customers = db.session.execute(select(Customer)).scalars().all()

    output = io.StringIO()
    writer = csv.writer(output)

    writer.writerow(["ID", "Nickname", "Email", "Orders"])

    for customer in customers:
        writer.writerow(
            [customer.id, customer.nickname, customer.email, len(customer.orders)]
        )

    output.seek(0)
    return Response(
        output,
        mimetype="text/csv",
        headers={"Content-Disposition": "attachment;filename=customers.csv"},
    )


if __name__ == "__main__":
    app.run(debug=False, port=8888)