"""routes/core.py — index, login, logout, dashboard.""" import sys, json, time, hashlib, hmac from db import (app, q, run, get_db, today, now, _log_action, _int, _float, _str, csrf_protect, MANAGER_ROLES, ALL_ROLES, MPESA_SECRET, _client_ip, _get_csrf_token, hash_pw, check_pw, _check_login_rate, _record_failed_login, _reset_login_attempts) from datetime import date, timedelta from auth import login_required, role_required, is_manager, is_gm, current_user from services import get_cash_position, get_outlet_low_stock, _build_reconciliation, calc_expected_revenue, cost_per_unit, calc_payroll, calc_ingredient_cost from flask import render_template, request, redirect, url_for, session, jsonify, flash @app.route('/') def index(): if 'user_id' not in session: return redirect(url_for('login')) role = session.get('role') if role in ('general_manager','manager'): return redirect(url_for('dashboard')) if role == 'chef': return redirect(url_for('production')) if role == 'cashier': return redirect(url_for('cashier')) if role == 'waiter': return redirect(url_for('waiter')) if role == 'storekeeper': return redirect(url_for('storekeeper')) if role == 'logistics': return redirect(url_for('logistics')) return redirect(url_for('dashboard')) @app.route('/login', methods=['GET','POST']) def login(): if request.method == 'POST': ip = _client_ip() username = _str(request.form.get('username', '')) password = request.form.get('password', '') # Guard against excessively long inputs if len(password) > 200 or len(username) > 100: flash('Invalid credentials.') return render_template('login.html') # Rate-limit on both IP and username to resist distributed attacks for identifier in (ip, f'user:{username}'): allowed, wait = _check_login_rate(identifier) if not allowed: flash(f'Too many failed attempts. Please wait {wait // 60} min {wait % 60} sec.') return render_template('login.html') # Look up by name OR employee-id (no dropdown — prevents user enumeration) user = q("SELECT * FROM users WHERE (name=? OR CAST(id AS TEXT)=?) AND active=1", (username, username), one=True) if user and check_pw(password, user['password']): _reset_login_attempts(ip) _reset_login_attempts(f'user:{username}') session.clear() session.permanent = True session['user_id'] = user['id'] session['role'] = user['role'] session['name'] = user['name'] session['outlet_id'] = user['outlet_id'] session['line'] = user['line'] if 'line' in user.keys() else None _get_csrf_token() # mint CSRF token for this session run("INSERT INTO clock_events (user_id,event_type) VALUES (?,?)", (user['id'], 'login')) return redirect(url_for('index')) _record_failed_login(ip) _record_failed_login(f'user:{username}') flash('Invalid credentials.') return render_template('login.html') @app.route('/logout') def logout(): if 'user_id' in session: run("INSERT INTO clock_events (user_id,event_type) VALUES (?,?)", (session['user_id'],'logout')) session.clear() return redirect(url_for('login')) @app.route('/dashboard') @login_required def dashboard(): user = current_user() d = today() role = session.get('role') outlets = [dict(r) for r in q("SELECT * FROM outlets WHERE active=1")] products = q("SELECT * FROM products WHERE active=1") rev_data = [] for outlet in outlets: rev = calc_expected_revenue(outlet['id'], d) low = get_outlet_low_stock(outlet['id'], d) rev_data.append({'outlet': dict(outlet), 'revenue': rev, 'low_stock': low}) total_rev = sum(r['revenue'] for r in rev_data) ing_cost = calc_ingredient_cost(d) other_cost = q("SELECT COALESCE(SUM(amount+tx_cost),0) as t FROM expenses WHERE date=?", (d,), one=True)['t'] daily_cost = ing_cost + other_cost profit = total_rev - daily_cost mpesa_total= q("SELECT COALESCE(SUM(amount),0) as t FROM mpesa_transactions WHERE created_at >= ? AND created_at < date(?, '+1 day')", (d, d), one=True)['t'] payroll = calc_payroll() if role == 'general_manager' else None # Weekly P&L for GM trend chart weekly_pnl = [] if role == 'general_manager': for i in range(7): wd = (date.today() - timedelta(days=6-i)).isoformat() wrev = sum(calc_expected_revenue(o['id'], wd) for o in outlets) wcost = calc_ingredient_cost(wd) wother = q("SELECT COALESCE(SUM(amount+tx_cost),0) as t FROM expenses WHERE date=?", (wd,), one=True)['t'] weekly_pnl.append({'date': wd, 'revenue': round(wrev), 'cost': round(wcost+wother), 'profit': round(wrev-wcost-wother)}) # Wastage alerts across all outlets today wastage_alerts = [] for p in products: pid = p['id'] threshold = p['wastage_threshold_pct'] if 'wastage_threshold_pct' in p.keys() else 10 for outlet in outlets: wt = q("""SELECT COALESCE(SUM(qty_out),0) as t FROM stock_events WHERE outlet_id=? AND product_id=? AND timestamp >= ? AND timestamp < date(?, '+1 day') AND event_type='WASTAGE'""",(outlet['id'],pid,d,d),one=True)['t'] ti = q("""SELECT COALESCE(SUM(qty_in),0) as t FROM stock_events WHERE outlet_id=? AND product_id=? AND timestamp >= ? AND timestamp < date(?, '+1 day') AND event_type IN ('OPENING_DISPATCH','TOPUP')""",(outlet['id'],pid,d,d),one=True)['t'] if ti > 0 and wt > 0: wpct = wt / ti * 100 if wpct > (threshold or 10): wastage_alerts.append({'product': p['name'], 'outlet': outlet['name'], 'pct': round(wpct, 1), 'threshold': threshold or 10}) return render_template('dashboard.html', user=user, rev_data=rev_data, total_rev=total_rev, daily_cost=daily_cost, profit=profit, ing_cost=ing_cost, other_cost=other_cost, mpesa_total=mpesa_total, outlets=outlets, today=d, payroll=payroll, role=role, weekly_pnl=weekly_pnl, wastage_alerts=wastage_alerts) # ── Production ──────────────────────────────────────────────────────────────── @app.route('/production') @login_required @role_required('general_manager','manager','chef') def production(): user = current_user() d = today() products = [dict(r) for r in q("SELECT * FROM products WHERE active=1 ORDER BY category,name")] role = session.get('role') chef_id = session['user_id'] if role in MANAGER_ROLES: # Managers see summed totals across all chefs prod_records = q("""SELECT product_id, SUM(portions_made) as portions, SUM(finished) as finished, SUM(wastage) as wastage FROM production_records WHERE date=? GROUP BY product_id""",(d,)) else: # Chefs see only their own records prod_records = q("""SELECT product_id, portions_made as portions, finished, wastage FROM production_records WHERE date=? AND chef_id=?""",(d, chef_id)) prod_map = {r['product_id']: dict(r) for r in prod_records} from datetime import date as date_cls, timedelta as td tomorrow_date = (date_cls.fromisoformat(d) + td(days=1)).isoformat() return render_template('production.html', user=user, products=products, today=d, prod_map=prod_map, role=role, tomorrow_date=tomorrow_date)