from flask import Flask, render_template, request, redirect, url_for, flash, send_file, session, jsonify, send_from_directory import sqlite3 import io import math import secrets import json import logging from logging.handlers import RotatingFileHandler import os import shutil from pathlib import Path from werkzeug.security import generate_password_hash, check_password_hash from functools import wraps from contextlib import contextmanager from datetime import datetime, timedelta, date # ============================================================================ # APPLICATION SETUP # ============================================================================ app = Flask(__name__) app.secret_key = os.environ.get("SECRET_KEY", "change-this-secret-key-before-production") BASE_DIR = Path(__file__).resolve().parent DB_PATH = BASE_DIR / "orders.db" PER_PAGE = 10 # ============================================================================ # PRODUCTION STABILITY SETUP # ============================================================================ LOG_DIR = BASE_DIR / "logs" BACKUP_DIR = BASE_DIR / "backups" LOG_DIR.mkdir(exist_ok=True) BACKUP_DIR.mkdir(exist_ok=True) def setup_logging(): """Save server-side errors to logs/error.log.""" handler = RotatingFileHandler( LOG_DIR / "error.log", maxBytes=1024 * 1024, backupCount=5, encoding="utf-8" ) handler.setLevel(logging.ERROR) handler.setFormatter(logging.Formatter( "%(asctime)s | %(levelname)s | %(message)s [in %(pathname)s:%(lineno)d]" )) app.logger.addHandler(handler) app.logger.setLevel(logging.ERROR) setup_logging() BACKUP_DIR = BASE_DIR / "backups" CSRF_SESSION_KEY = "csrf_token" UPLOAD_DIR = Path("uploads") ERROR_LOG_FILE = Path("error.log") MAX_LOGIN_ATTEMPTS = 5 LOGIN_LOCK_MINUTES = 15 def safe_money(value): """Return a safe numeric value for formatting and calculations.""" try: return float(value or 0) except (TypeError, ValueError): return 0.0 def safe_sort_value(value): """Return a safe sortable string for dates/text values.""" return value or "" def current_user_name(): """Return current logged-in username for logs.""" return session.get("username", "system") def add_activity(action, table_name="", record_id=None, description=""): """Write a user activity entry without crashing the main action.""" try: with get_db() as db: db.execute(""" INSERT INTO activity_log (user_id, username, action, table_name, record_id, description, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( session.get("user_id"), current_user_name(), action, table_name, record_id, description, datetime.now().strftime('%Y-%m-%d %H:%M:%S') )) except Exception: pass def record_activity(db, action, table_name="", record_id=None, description=""): """Write an activity entry using the current open database connection.""" db.execute(""" INSERT INTO activity_log (user_id, username, action, table_name, record_id, description, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( session.get("user_id"), current_user_name(), action, table_name, record_id, description, datetime.now().strftime('%Y-%m-%d %H:%M:%S') )) def warranty_expiry_from_days(order_date_value, days_value): """Calculate expiry date from order date + days.""" if not order_date_value or not days_value: return None try: return (datetime.strptime(order_date_value, '%Y-%m-%d') + timedelta(days=int(days_value))).strftime('%Y-%m-%d') except Exception: return None def generate_invoice_no(db): """Generate invoice number like INV-2026-0001.""" year = datetime.now().year prefix = f"INV-{year}-" row = db.execute("SELECT invoice_no FROM orders WHERE invoice_no LIKE ? ORDER BY id DESC LIMIT 1", (prefix + '%',)).fetchone() if row and row['invoice_no']: try: next_no = int(row['invoice_no'].split('-')[-1]) + 1 except Exception: next_no = 1 else: next_no = 1 return f"{prefix}{next_no:04d}" def ensure_order_items_for_order(db, order): """Backfill item lines for older single-item orders.""" if not order: return [] existing = db.execute("SELECT * FROM order_items WHERE order_id = ? ORDER BY id", (order["id"],)).fetchall() if existing: return existing db.execute(""" INSERT INTO order_items (order_id, stock_id, item_name, item_type, serial, imei, qty, unit_price, total_price, cost, product_warranty_days, battery_warranty_days) VALUES (?, ?, ?, 'product', ?, ?, ?, ?, ?, ?, ?, ?) """, ( order["id"], order["stock_id"], order["item_name"] or "Item", order["serial"], order["imei"], order["qty"] or 1, safe_money(order["total_price"]) / max(1, int(order["qty"] or 1)), safe_money(order["total_price"]), safe_money(order["cost"]), order["product_warranty_days"] or 0, order["battery_warranty_days"] or 0 )) return db.execute("SELECT * FROM order_items WHERE order_id = ? ORDER BY id", (order["id"],)).fetchall() def summarize_order_items(items): """Return invoice totals for a list of SQLite rows/dicts.""" total = sum(safe_money(item["total_price"]) for item in items) cost = sum(safe_money(item["cost"]) for item in items) qty = sum(to_int(item["qty"], 0) for item in items) product_count = sum(1 for item in items if (item["item_type"] or "product") == "product") service_count = sum(1 for item in items if (item["item_type"] or "product") == "service") return {"total": total, "cost": cost, "qty": qty, "product_count": product_count, "service_count": service_count} def make_daily_backup(force=False): """Create one dated backup of orders.db per day.""" if not DB_PATH.exists(): return None BACKUP_DIR.mkdir(exist_ok=True) backup_name = f"orders_{datetime.now().strftime('%Y-%m-%d')}.db" backup_path = BACKUP_DIR / backup_name if force or not backup_path.exists(): shutil.copy2(DB_PATH, backup_path) return backup_path def keep_recent_backups(limit=30): """Keep only latest backup files.""" if not BACKUP_DIR.exists(): return backups = sorted(BACKUP_DIR.glob('orders_*.db'), key=lambda x: x.stat().st_mtime, reverse=True) for old in backups[limit:]: try: old.unlink() except Exception: pass def client_ip(): """Return best available client IP for login protection.""" return (request.headers.get("X-Forwarded-For", request.remote_addr or "unknown").split(",")[0].strip()) def is_login_locked(db, username, ip_address): """Check temporary lock status after too many failed attempts.""" cutoff = (datetime.now() - timedelta(minutes=LOGIN_LOCK_MINUTES)).strftime('%Y-%m-%d %H:%M:%S') row = db.execute(""" SELECT COUNT(*) AS attempts FROM login_attempts WHERE success = 0 AND attempted_at >= ? AND (username = ? OR ip_address = ?) """, (cutoff, username, ip_address)).fetchone() attempts = row["attempts"] if row else 0 return attempts >= MAX_LOGIN_ATTEMPTS, attempts def record_login_attempt(db, username, success): """Store login success/failure for anti-hack protection.""" db.execute(""" INSERT INTO login_attempts (username, ip_address, success, attempted_at) VALUES (?, ?, ?, ?) """, (username, client_ip(), 1 if success else 0, datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) def clear_failed_login_attempts(db, username, ip_address): """Clear failed attempts after successful login or admin reset.""" db.execute("DELETE FROM login_attempts WHERE success = 0 AND (username = ? OR ip_address = ?)", (username, ip_address)) def validate_password_strength(password): """Business-safe password rules.""" if len(password or "") < 8: return False, "Password must be at least 8 characters long." if not any(c.isupper() for c in password): return False, "Password must include at least one uppercase letter." if not any(c.islower() for c in password): return False, "Password must include at least one lowercase letter." if not any(c.isdigit() for c in password): return False, "Password must include at least one number." if not any(not c.isalnum() for c in password): return False, "Password must include at least one symbol." return True, "" # ============================================================================ # DECORATORS # ============================================================================ def login_required(view_func): """Decorator to require login for protected routes""" @wraps(view_func) def wrapper(*args, **kwargs): if "user_id" not in session: flash("🔒 Please log in to access this page.") return redirect(url_for("login")) return view_func(*args, **kwargs) return wrapper def admin_required(view_func): """Decorator to require admin role for admin-only routes""" @wraps(view_func) def wrapper(*args, **kwargs): if session.get("role") != "admin": flash("❌ You do not have permission to perform this action.") return redirect(url_for("index")) return view_func(*args, **kwargs) return wrapper @app.before_request def protect_post_requests(): """Simple CSRF protection for all POST forms.""" if request.method == "POST": token = session.get(CSRF_SESSION_KEY) form_token = request.form.get("csrf_token") or request.headers.get("X-CSRFToken") if not token or token != form_token: flash("❌ Security token expired. Please try again.") return redirect(request.referrer or url_for("login")) @app.context_processor def inject_csrf_token(): def csrf_token(): token = session.get(CSRF_SESSION_KEY) if not token: token = secrets.token_urlsafe(32) session[CSRF_SESSION_KEY] = token return token return dict(csrf_token=csrf_token) @app.before_request def ensure_csrf_token_exists(): if CSRF_SESSION_KEY not in session: session[CSRF_SESSION_KEY] = secrets.token_urlsafe(32) @app.before_request def require_login(): """Protect all pages. Only login and static files are public.""" public_endpoints = {"login", "static", "service_worker", "favicon"} if request.endpoint in public_endpoints: return None if request.endpoint is None: return None if "user_id" not in session: return redirect(url_for("login")) return None # ============================================================================ # DATABASE FUNCTIONS # ============================================================================ @contextmanager def get_db(): """Database connection context manager""" conn = sqlite3.connect(DB_PATH, timeout=2000.0, check_same_thread=False) conn.row_factory = sqlite3.Row try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() def init_db(): """Initialize database tables and default users""" with get_db() as db: db.executescript(""" CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_date TEXT, stock_id INTEGER, invoice_no TEXT, order_id TEXT, serial TEXT, imei TEXT, contact1 TEXT, contact2 TEXT, order_via TEXT, store TEXT, item_name TEXT, qty INTEGER, total_price REAL, cash_received REAL DEFAULT 0, cost REAL, product_warranty_days INTEGER, battery_warranty_days INTEGER ); CREATE TABLE IF NOT EXISTS order_items ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL, stock_id INTEGER, item_name TEXT NOT NULL, item_type TEXT CHECK(item_type IN ('product','service')) DEFAULT 'product', serial TEXT, imei TEXT, qty INTEGER DEFAULT 1, unit_price REAL DEFAULT 0, total_price REAL DEFAULT 0, cost REAL DEFAULT 0, product_warranty_days INTEGER DEFAULT 0, battery_warranty_days INTEGER DEFAULT 0, notes TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS stock ( id INTEGER PRIMARY KEY AUTOINCREMENT, stock_id TEXT NOT NULL, item_name TEXT NOT NULL, item_category TEXT, date TEXT, order_price REAL, logistic_charges REAL, shipping_charges REAL, unit_cost REAL, qty INTEGER DEFAULT 0, ordered_qty INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS expenses ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL, reason TEXT NOT NULL, amount REAL NOT NULL, payment_method TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS cash_book ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL, reason TEXT NOT NULL, amount REAL NOT NULL, type TEXT CHECK(type IN ('credit', 'debit')) NOT NULL, payment_method TEXT CHECK(payment_method IN ('Cash', 'Bank Transfer', 'Card', 'Other')) NOT NULL, payment_done_by TEXT CHECK(payment_done_by IN ('Kalana', 'Eshan', 'Yasiru', '3rd Party')) NOT NULL, description TEXT, total_balance REAL ); CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, password TEXT NOT NULL, role TEXT CHECK(role IN ('admin', 'viewer')) NOT NULL, full_name TEXT, email TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP, last_login TEXT, is_active INTEGER DEFAULT 1 ); CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, description TEXT, priority TEXT CHECK(priority IN ('Low', 'Medium', 'High', 'Urgent')) DEFAULT 'Medium', status TEXT CHECK(status IN ('Todo', 'In Progress', 'Completed', 'On Hold')) DEFAULT 'Todo', assigned_to TEXT, created_by TEXT, due_date TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP, completed_at TEXT, category TEXT, tags TEXT ); CREATE TABLE IF NOT EXISTS assets ( id INTEGER PRIMARY KEY AUTOINCREMENT, asset_name TEXT NOT NULL, category TEXT CHECK(category IN ('Office Equipment', 'Furniture', 'Electronics', 'Vehicles', 'Software', 'Property', 'Tools', 'Other')) NOT NULL, purchase_date TEXT NOT NULL, purchase_price REAL NOT NULL, current_value REAL NOT NULL, depreciation_rate REAL DEFAULT 0, location TEXT, assigned_to TEXT, warranty_expiry TEXT, maintenance_date TEXT, status TEXT CHECK(status IN ('Active', 'In Repair', 'Retired', 'Sold')) DEFAULT 'Active', serial_number TEXT, supplier TEXT, notes TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS imei_devices ( id INTEGER PRIMARY KEY AUTOINCREMENT, imei TEXT UNIQUE, serial TEXT, stock_id INTEGER, item_name TEXT, status TEXT CHECK(status IN ('in_stock','sold','returned','faulty')) DEFAULT 'in_stock', product_warranty_expiry TEXT, battery_warranty_expiry TEXT, trcsl_required INTEGER DEFAULT 0, trcsl_status TEXT CHECK(trcsl_status IN ('Not Required','Approved','Pending','Not Approved')) DEFAULT 'Not Required', trcsl_reference TEXT, import_batch TEXT, location TEXT DEFAULT 'Shop', order_id INTEGER, created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS order_returns ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id INTEGER NOT NULL, return_date TEXT NOT NULL, return_type TEXT CHECK(return_type IN ('Return','Replacement','Refund')) NOT NULL, refund_amount REAL DEFAULT 0, replacement_item TEXT, reason TEXT NOT NULL, created_by TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS activity_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, username TEXT, action TEXT NOT NULL, table_name TEXT, record_id INTEGER, description TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS login_attempts ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT, ip_address TEXT, success INTEGER DEFAULT 0, attempted_at TEXT DEFAULT CURRENT_TIMESTAMP ); """) # Safe migrations for existing databases for sql in [ "ALTER TABLE stock ADD COLUMN reorder_level INTEGER DEFAULT 5", "ALTER TABLE stock ADD COLUMN customs_charges REAL DEFAULT 0", "ALTER TABLE stock ADD COLUMN location TEXT DEFAULT 'Shop'", "ALTER TABLE orders ADD COLUMN delivery_courier TEXT", "ALTER TABLE orders ADD COLUMN tracking_no TEXT", "ALTER TABLE orders ADD COLUMN delivery_status TEXT DEFAULT 'Pending'", "ALTER TABLE users ADD COLUMN must_change_password INTEGER DEFAULT 0", "ALTER TABLE order_returns ADD COLUMN status TEXT DEFAULT 'Pending'", "ALTER TABLE order_returns ADD COLUMN condition_note TEXT", "ALTER TABLE order_returns ADD COLUMN action_taken TEXT", "ALTER TABLE order_returns ADD COLUMN resolved_at TEXT", "ALTER TABLE orders ADD COLUMN deleted_at TEXT", "ALTER TABLE orders ADD COLUMN deleted_by TEXT" ]: try: db.execute(sql) except sqlite3.OperationalError: pass # Phase 2 business tables db.executescript(""" CREATE TABLE IF NOT EXISTS suppliers ( id INTEGER PRIMARY KEY AUTOINCREMENT, supplier_name TEXT NOT NULL, contact_person TEXT, phone TEXT, email TEXT, address TEXT, notes TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS purchase_orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, supplier_id INTEGER, po_date TEXT NOT NULL, item_name TEXT NOT NULL, qty INTEGER DEFAULT 0, product_cost REAL DEFAULT 0, shipping_cost REAL DEFAULT 0, customs_cost REAL DEFAULT 0, other_cost REAL DEFAULT 0, total_cost REAL DEFAULT 0, unit_cost REAL DEFAULT 0, amount_paid REAL DEFAULT 0, status TEXT CHECK(status IN ('Pending','Partially Paid','Paid','Received','Cancelled')) DEFAULT 'Pending', notes TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS reminders ( id INTEGER PRIMARY KEY AUTOINCREMENT, reminder_type TEXT NOT NULL, title TEXT NOT NULL, related_table TEXT, related_id INTEGER, due_date TEXT, status TEXT CHECK(status IN ('Pending','Done','Cancelled')) DEFAULT 'Pending', notes TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS stock_locations ( id INTEGER PRIMARY KEY AUTOINCREMENT, location_name TEXT UNIQUE NOT NULL, description TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS soft_deleted_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, table_name TEXT NOT NULL, record_id INTEGER NOT NULL, record_json TEXT, deleted_by TEXT, deleted_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS system_errors ( id INTEGER PRIMARY KEY AUTOINCREMENT, error_type TEXT, message TEXT, path TEXT, method TEXT, username TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); """) # Create default users hashed_kalana = generate_password_hash(os.environ.get("DEFAULT_ADMIN_PASSWORD", "K@lana#2025!")) hashed_yasiru = generate_password_hash(os.environ.get("DEFAULT_YASIRU_PASSWORD", "Ya$iru@1234")) hashed_eshan = generate_password_hash(os.environ.get("DEFAULT_ESHAN_PASSWORD", "Eshan#Secure1")) db.execute(""" INSERT OR IGNORE INTO users (id, username, password, role, full_name, is_active) VALUES (1, 'kalana', ?, 'admin', 'Kalana Admin', 1), (2, 'yasiru', ?, 'viewer', 'Yasiru User', 1), (3, 'eshan', ?, 'viewer', 'Eshan User', 1); """, (hashed_kalana, hashed_yasiru, hashed_eshan)) # ============================================================================ # UTILITY FUNCTIONS # ============================================================================ def get_current_period(): """Get current billing period (27th to 26th)""" today = datetime.now() if today.day >= 27: start_date = datetime(today.year, today.month, 27) if today.month == 12: end_date = datetime(today.year + 1, 1, 26) else: end_date = datetime(today.year, today.month + 1, 26) else: if today.month == 1: start_date = datetime(today.year - 1, 12, 27) else: start_date = datetime(today.year, today.month - 1, 27) end_date = datetime(today.year, today.month, 26) return { 'start_date': start_date.strftime('%Y-%m-%d'), 'end_date': end_date.strftime('%Y-%m-%d'), 'start': start_date, 'end': end_date } def to_int(value, default=0): """Safely convert form/query values to int.""" try: if value is None or value == "": return default return int(value) except (TypeError, ValueError): return default def to_float(value, default=0.0): """Safely convert form/query values to float.""" try: if value is None or value == "": return default return float(value) except (TypeError, ValueError): return default def get_page(): """Read current page safely.""" return max(1, to_int(request.args.get("page", 1), 1)) @app.route("/") def index(): return redirect(url_for("login")) # ============================================================================ # GLOBAL ERROR HANDLERS # ============================================================================ @app.errorhandler(404) def handle_404(error): app.logger.warning(f"404 Not Found: {request.path}") return render_template("errors/404.html"), 404 @app.errorhandler(403) def handle_403(error): app.logger.warning(f"403 Forbidden: {request.path}") return render_template("errors/403.html"), 403 @app.errorhandler(503) def handle_503(error): app.logger.warning("503 Service Unavailable") return render_template("errors/503.html"), 503 @app.errorhandler(500) def handle_500(error): app.logger.exception("Internal server error") return render_template("errors/500.html"), 500 @app.errorhandler(sqlite3.Error) def handle_database_error(error): app.logger.exception("Database error") flash("Database error occurred. Please try again or contact administrator.") return render_template("errors/500.html"), 500 @app.errorhandler(Exception) def handle_unexpected_error(error): from werkzeug.exceptions import HTTPException if isinstance(error, HTTPException): return error app.logger.exception("Unexpected application error") return render_template("errors/error.html"), 500 # ============================================================================ # AUTHENTICATION ROUTES # ============================================================================ @app.route("/login", methods=["GET", "POST"]) def login(): if "user_id" in session: return redirect(url_for("dashboard")) """User login with failed-attempt lockout protection.""" if request.method == "POST": username = request.form.get("username", "").strip() password = request.form.get("password", "") ip_address = client_ip() with get_db() as db: locked, attempts = is_login_locked(db, username, ip_address) if locked: flash(f"❌ Too many failed login attempts. Try again after {LOGIN_LOCK_MINUTES} minutes or ask an admin to reset your password.") return render_template("login.html") user = db.execute( "SELECT * FROM users WHERE username = ?;", (username,) ).fetchone() if user and not user["is_active"]: record_login_attempt(db, username, False) flash("❌ Your account has been deactivated. Contact an administrator.") elif user and check_password_hash(user["password"], password): clear_failed_login_attempts(db, username, ip_address) record_login_attempt(db, username, True) db.execute( "UPDATE users SET last_login = ? WHERE id = ?", (datetime.now().strftime('%Y-%m-%d %H:%M:%S'), user["id"]) ) session["user_id"] = user["id"] session["username"] = user["username"] session["role"] = user["role"] session["full_name"] = user["full_name"] or user["username"] flash("✅ Logged in successfully.") return redirect(url_for("dashboard")) else: record_login_attempt(db, username, False) remaining = max(0, MAX_LOGIN_ATTEMPTS - attempts - 1) flash(f"❌ Invalid credentials. Attempts remaining: {remaining}") return render_template("login.html") @app.route("/logout") @login_required def logout(): """User logout""" session.clear() flash("👋 Logged out.") return redirect(url_for("login")) # ============================================================================ # USER MANAGEMENT ROUTES # ============================================================================ @app.route("/users") @login_required @admin_required def manage_users(): """User management page (admin only)""" with get_db() as db: users = db.execute(""" SELECT id, username, role, full_name, email, created_at, last_login, is_active FROM users ORDER BY id ASC """).fetchall() return render_template("users.html", users=users) @app.route("/users/add", methods=["POST"]) @login_required @admin_required def add_user(): """Add a new user (admin only)""" username = request.form.get("username", "").strip() password = request.form.get("password", "").strip() role = request.form.get("role", "viewer") full_name = request.form.get("full_name", "").strip() email = request.form.get("email", "").strip() if not username or not password: flash("❌ Username and password are required.") return redirect(url_for("manage_users")) ok, msg = validate_password_strength(password) if not ok: flash(f"❌ {msg}") return redirect(url_for("manage_users")) with get_db() as db: # Check if username already exists existing = db.execute( "SELECT id FROM users WHERE username = ?", (username,) ).fetchone() if existing: flash("❌ Username already exists.") return redirect(url_for("manage_users")) # Create new user hashed_password = generate_password_hash(password) db.execute(""" INSERT INTO users (username, password, role, full_name, email, is_active) VALUES (?, ?, ?, ?, ?, 1) """, (username, hashed_password, role, full_name, email)) flash(f"✅ User '{username}' created successfully.") return redirect(url_for("manage_users")) @app.route("/users/edit/", methods=["POST"]) @login_required @admin_required def edit_user(user_id): """Edit user details (admin only)""" role = request.form.get("role", "viewer") full_name = request.form.get("full_name", "").strip() email = request.form.get("email", "").strip() is_active = 1 if request.form.get("is_active") == "on" else 0 with get_db() as db: # Prevent deactivating the last admin if is_active == 0: user = db.execute("SELECT role FROM users WHERE id = ?", (user_id,)).fetchone() if user and user["role"] == "admin": active_admins = db.execute( "SELECT COUNT(*) as count FROM users WHERE role = 'admin' AND is_active = 1" ).fetchone()["count"] if active_admins <= 1: flash("❌ Cannot deactivate the last active admin user.") return redirect(url_for("manage_users")) db.execute(""" UPDATE users SET role = ?, full_name = ?, email = ?, is_active = ? WHERE id = ? """, (role, full_name, email, is_active, user_id)) flash("✅ User updated successfully.") return redirect(url_for("manage_users")) @app.route("/users/delete/", methods=["POST"]) @login_required @admin_required def delete_user(user_id): """Delete a user (admin only)""" if user_id == session.get("user_id"): flash("❌ You cannot delete your own account.") return redirect(url_for("manage_users")) with get_db() as db: # Check if it's the last admin user = db.execute("SELECT role FROM users WHERE id = ?", (user_id,)).fetchone() if user and user["role"] == "admin": active_admins = db.execute( "SELECT COUNT(*) as count FROM users WHERE role = 'admin' AND is_active = 1" ).fetchone()["count"] if active_admins <= 1: flash("❌ Cannot delete the last admin user.") return redirect(url_for("manage_users")) db.execute("DELETE FROM users WHERE id = ?", (user_id,)) flash("✅ User deleted successfully.") return redirect(url_for("manage_users")) @app.route("/users/reset-password/", methods=["POST"]) @login_required @admin_required def reset_user_password(user_id): """Reset user password (admin only)""" new_password = request.form.get("new_password", "").strip() ok, msg = validate_password_strength(new_password) if not new_password or not ok: flash(f"❌ {msg or 'Password is required.'}") return redirect(url_for("manage_users")) with get_db() as db: hashed_password = generate_password_hash(new_password) target = db.execute("SELECT username FROM users WHERE id = ?", (user_id,)).fetchone() db.execute( "UPDATE users SET password = ?, must_change_password = 1 WHERE id = ?", (hashed_password, user_id) ) if target: db.execute("DELETE FROM login_attempts WHERE username = ?", (target["username"],)) record_activity(db, "Reset password", "users", user_id, f"Password reset for {target['username']}") flash("✅ Password reset successfully. User should change it after login.") return redirect(url_for("manage_users")) @app.route("/profile") @login_required def user_profile(): """User profile page""" with get_db() as db: user = db.execute( "SELECT id, username, role, full_name, email, created_at, last_login FROM users WHERE id = ?", (session.get("user_id"),) ).fetchone() return render_template("profile.html", user=user) @app.route("/profile/update", methods=["POST"]) @login_required def update_profile(): """Update user profile""" full_name = request.form.get("full_name", "").strip() email = request.form.get("email", "").strip() with get_db() as db: db.execute( "UPDATE users SET full_name = ?, email = ? WHERE id = ?", (full_name, email, session.get("user_id")) ) session["full_name"] = full_name or session.get("username") flash("✅ Profile updated successfully.") return redirect(url_for("user_profile")) @app.route("/profile/change-password", methods=["POST"]) @login_required def change_password(): """Change user password""" current_password = request.form.get("current_password", "") new_password = request.form.get("new_password", "") confirm_password = request.form.get("confirm_password", "") if not current_password or not new_password: flash("❌ All fields are required.") return redirect(url_for("user_profile")) if new_password != confirm_password: flash("❌ New passwords do not match.") return redirect(url_for("user_profile")) ok, msg = validate_password_strength(new_password) if not ok: flash(f"❌ {msg}") return redirect(url_for("user_profile")) with get_db() as db: user = db.execute( "SELECT password FROM users WHERE id = ?", (session.get("user_id"),) ).fetchone() if not check_password_hash(user["password"], current_password): flash("❌ Current password is incorrect.") return redirect(url_for("user_profile")) hashed_password = generate_password_hash(new_password) db.execute( "UPDATE users SET password = ?, must_change_password = 0 WHERE id = ?", (hashed_password, session.get("user_id")) ) record_activity(db, "Change password", "users", session.get("user_id"), "User changed own password") flash("✅ Password changed successfully.") return redirect(url_for("user_profile")) # ============================================================================ # DASHBOARD ROUTE # ============================================================================ @app.route("/dashboard") @login_required def dashboard(): """Dashboard with statistics""" current_period = get_current_period() with get_db() as db: # Get current period stats stats = db.execute(""" SELECT COUNT(*) as total_orders, COALESCE(SUM(cash_received), 0) as total_revenue, COALESCE(SUM(cost), 0) as total_cost FROM orders WHERE order_date BETWEEN ? AND ? """, (current_period['start_date'], current_period['end_date'])).fetchone() # Get expenses expenses = db.execute(""" SELECT COALESCE(SUM(amount), 0) as total_expenses FROM expenses WHERE date BETWEEN ? AND ? """, (current_period['start_date'], current_period['end_date'])).fetchone() # Get stock stats stock_stats = db.execute(""" SELECT COUNT(*) as total_items, COALESCE(SUM(qty), 0) as total_quantity, COUNT(CASE WHEN qty BETWEEN 1 AND 5 THEN 1 END) as low_stock_items, COUNT(CASE WHEN qty <= 0 THEN 1 END) as out_stock_items FROM stock """).fetchone() # Get platform data for chart platform_data = db.execute(""" SELECT order_via, COUNT(*) as count FROM orders WHERE order_date BETWEEN ? AND ? GROUP BY order_via """, (current_period['start_date'], current_period['end_date'])).fetchall() platform_labels = [row['order_via'] or 'Unknown' for row in platform_data] platform_values = [row['count'] for row in platform_data] # Get previous period for comparison prev_start = (datetime.strptime(current_period['start_date'], '%Y-%m-%d') - timedelta(days=30)).strftime('%Y-%m-%d') prev_end = (datetime.strptime(current_period['end_date'], '%Y-%m-%d') - timedelta(days=30)).strftime('%Y-%m-%d') prev_stats = db.execute(""" SELECT COUNT(*) as total_orders, COALESCE(SUM(cash_received), 0) as total_revenue FROM orders WHERE order_date BETWEEN ? AND ? """, (prev_start, prev_end)).fetchone() prev_expenses = db.execute(""" SELECT COALESCE(SUM(amount), 0) as total_expenses FROM expenses WHERE date BETWEEN ? AND ? """, (prev_start, prev_end)).fetchone() # Calculate changes orders_change = round(((stats['total_orders'] - prev_stats['total_orders']) / max(prev_stats['total_orders'], 1)) * 100, 1) revenue_change = round(((stats['total_revenue'] - prev_stats['total_revenue']) / max(prev_stats['total_revenue'], 1)) * 100, 1) expenses_change = round(((expenses['total_expenses'] - prev_expenses['total_expenses']) / max(prev_expenses['total_expenses'], 1)) * 100, 1) net_profit = stats['total_revenue'] - stats['total_cost'] - expenses['total_expenses'] prev_profit = prev_stats['total_revenue'] - prev_expenses['total_expenses'] profit_change = round(((net_profit - prev_profit) / max(abs(prev_profit), 1)) * 100, 1) # Get recent activities recent_activities = [] recent_orders = db.execute(""" SELECT order_date, item_name, cash_received FROM orders ORDER BY id DESC LIMIT 3 """).fetchall() for order in recent_orders: recent_activities.append({ 'type': 'order', 'time': order['order_date'], 'description': f"New order: {order['item_name'] or 'Unknown item'} - Rs. {safe_money(order['cash_received']):,.2f}" }) recent_expenses = db.execute(""" SELECT date, reason, amount FROM expenses ORDER BY id DESC LIMIT 2 """).fetchall() for expense in recent_expenses: recent_activities.append({ 'type': 'expense', 'time': expense['date'], 'description': f"Expense: {expense['reason'] or 'Expense'} - Rs. {safe_money(expense['amount']):,.2f}" }) # Sort by time recent_activities.sort(key=lambda x: safe_sort_value(x.get('time')), reverse=True) recent_activities = recent_activities[:5] dashboard_stats = { 'total_orders': stats['total_orders'], 'total_revenue': stats['total_revenue'], 'total_expenses': expenses['total_expenses'], 'net_profit': net_profit, 'total_stock_items': stock_stats['total_items'], 'total_stock_quantity': stock_stats['total_quantity'], 'low_stock_items': stock_stats['low_stock_items'], 'out_stock_items': stock_stats['out_stock_items'], 'gross_profit': stats['total_revenue'] - stats['total_cost'], 'orders_change': orders_change, 'revenue_change': revenue_change, 'expenses_change': expenses_change, 'profit_change': profit_change } return render_template( "dashboard.html", stats=dashboard_stats, current_period=current_period, platform_labels_json=json.dumps(platform_labels), platform_values_json=json.dumps(platform_values), recent_activities=recent_activities ) # ============================================================================ # ORDERS ROUTES # ============================================================================ @app.route("/") @app.route("/orders") @login_required def orders(): """Orders page with search, filters, edit/delete links, and pagination.""" order_methods = ['Daraz', 'Direct', 'Tudo', 'Payzy', 'Other'] stores = ['Main Store', 'Online Store', 'Daraz Store', 'Other'] with get_db() as db: delete_id = request.args.get("delete") if delete_id and session.get("role") == "admin": order = db.execute("SELECT * FROM orders WHERE id = ?", (delete_id,)).fetchone() if order: db.execute("UPDATE stock SET qty = qty + ? WHERE id = ?", (order["qty"] or 0, order["stock_id"])) db.execute("DELETE FROM orders WHERE id = ?", (delete_id,)) record_activity(db, "delete", "orders", to_int(delete_id), f"Deleted order {order['invoice_no'] or delete_id}") flash("🗑️ Order deleted successfully.") else: flash("❌ Order not found.") return redirect(url_for("index", page=request.args.get("page", 1))) edit_id = request.args.get("edit") edit_row = db.execute("SELECT * FROM orders WHERE id = ?", (edit_id,)).fetchone() if edit_id else None search_query = request.args.get("q", "").strip() filter_start_date = request.args.get("start_date") or "" filter_end_date = request.args.get("end_date") or "" filter_order_via = request.args.get("order_via") or "All" filter_store = request.args.get("store") or "All" filter_cash_status = request.args.get("cash_status") or "All" query = "SELECT * FROM orders WHERE 1=1" params = [] if search_query: query += " AND (invoice_no LIKE ? OR order_id LIKE ? OR serial LIKE ? OR imei LIKE ? OR contact1 LIKE ? OR item_name LIKE ?)" like_query = f"%{search_query}%" params.extend([like_query] * 6) if filter_start_date: query += " AND order_date >= ?" params.append(filter_start_date) if filter_end_date: query += " AND order_date <= ?" params.append(filter_end_date) if filter_order_via != "All": query += " AND order_via = ?" params.append(filter_order_via) if filter_store != "All": query += " AND store = ?" params.append(filter_store) if filter_cash_status == "Pending": query += " AND COALESCE(cash_received, 0) < COALESCE(total_price, 0)" elif filter_cash_status == "Received": query += " AND COALESCE(cash_received, 0) >= COALESCE(total_price, 0)" page = get_page() total_orders = db.execute(query.replace("SELECT *", "SELECT COUNT(*)"), params).fetchone()[0] total_pages = max(1, math.ceil(total_orders / PER_PAGE)) page = min(page, total_pages) offset = (page - 1) * PER_PAGE orders = db.execute(query + " ORDER BY id DESC LIMIT ? OFFSET ?", params + [PER_PAGE, offset]).fetchall() stocks = db.execute("SELECT id, stock_id, item_name, unit_cost, qty FROM stock ORDER BY item_name").fetchall() return render_template( "orders.html", orders=orders, stocks=stocks, stock_items=stocks, edit_row=edit_row, order_methods=order_methods, stores=stores, search_query=search_query, filter_start_date=filter_start_date, filter_end_date=filter_end_date, filter_order_via=filter_order_via, filter_store=filter_store, filter_cash_status=filter_cash_status, total_pages=total_pages, current_page=page ) @app.route("/add_order", methods=["POST"]) @login_required @admin_required def add_order(): """Add a new order and reduce stock.""" stock_id = to_int(request.form.get("stock_id")) qty = to_int(request.form.get("qty"), 1) if stock_id <= 0 or qty <= 0: flash("❌ Please select a valid stock item and quantity.") return redirect(url_for("index")) with get_db() as db: stock = db.execute("SELECT * FROM stock WHERE id = ?", (stock_id,)).fetchone() if not stock: flash("❌ Selected stock item was not found.") return redirect(url_for("index")) if (stock["qty"] or 0) < qty: flash("❌ Insufficient stock.") return redirect(url_for("index")) db.execute("UPDATE stock SET qty = qty - ? WHERE id = ?", (qty, stock_id)) unit_cost = to_float(stock["unit_cost"]) total_price = unit_cost * qty cost = unit_cost * qty order_date_value = request.form.get("order_date") or datetime.now().strftime('%Y-%m-%d') invoice_value = (request.form.get("invoice_no") or "").strip() or generate_invoice_no(db) imei_value = (request.form.get("imei") or "").strip() serial_value = (request.form.get("serial") or "").strip() product_warranty_days = to_int(request.form.get("product_warranty_days")) battery_warranty_days = to_int(request.form.get("battery_warranty_days")) cur = db.execute(""" INSERT INTO orders (order_date, stock_id, invoice_no, order_id, serial, imei, contact1, contact2, order_via, store, item_name, qty, total_price, cash_received, cost, product_warranty_days, battery_warranty_days) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( order_date_value, stock_id, invoice_value, request.form.get("order_id"), serial_value, imei_value, request.form.get("contact1"), request.form.get("contact2"), request.form.get("order_via"), request.form.get("store"), stock["item_name"], qty, total_price, to_float(request.form.get("cash_received")), cost, product_warranty_days, battery_warranty_days )) new_order_id = cur.lastrowid if imei_value: existing_device = db.execute("SELECT id, status FROM imei_devices WHERE imei = ?", (imei_value,)).fetchone() if existing_device and existing_device["status"] == "sold": flash("⚠️ Warning: this IMEI is already marked as sold. Please check duplicate sale.") db.execute(""" INSERT INTO imei_devices (imei, serial, stock_id, item_name, status, product_warranty_expiry, battery_warranty_expiry, order_id, updated_at) VALUES (?, ?, ?, ?, 'sold', ?, ?, ?, ?) ON CONFLICT(imei) DO UPDATE SET serial=excluded.serial, stock_id=excluded.stock_id, item_name=excluded.item_name, status='sold', product_warranty_expiry=excluded.product_warranty_expiry, battery_warranty_expiry=excluded.battery_warranty_expiry, order_id=excluded.order_id, updated_at=excluded.updated_at """, ( imei_value, serial_value, stock_id, stock["item_name"], warranty_expiry_from_days(order_date_value, product_warranty_days), warranty_expiry_from_days(order_date_value, battery_warranty_days), new_order_id, datetime.now().strftime('%Y-%m-%d %H:%M:%S') )) record_activity(db, "add", "orders", new_order_id, f"Added order {invoice_value} for {stock['item_name']}") flash("✅ Order added successfully.") return redirect(url_for("index")) @app.route("/edit_order/", methods=["GET", "POST"]) @login_required @admin_required def edit_order(order_id): """Edit an existing order and adjust stock correctly.""" with get_db() as db: old_order = db.execute("SELECT * FROM orders WHERE id = ?", (order_id,)).fetchone() if not old_order: flash("❌ Order not found.") return redirect(url_for("index")) if request.method == "POST": new_stock_id = to_int(request.form.get("stock_id")) new_qty = to_int(request.form.get("qty"), 1) if new_stock_id <= 0 or new_qty <= 0: flash("❌ Please select a valid stock item and quantity.") return redirect(url_for("edit_order", order_id=order_id)) # Return old quantity before checking the new stock quantity. db.execute("UPDATE stock SET qty = qty + ? WHERE id = ?", (old_order["qty"] or 0, old_order["stock_id"])) stock = db.execute("SELECT * FROM stock WHERE id = ?", (new_stock_id,)).fetchone() if not stock: db.execute("UPDATE stock SET qty = qty - ? WHERE id = ?", (old_order["qty"] or 0, old_order["stock_id"])) flash("❌ Selected stock item was not found.") return redirect(url_for("edit_order", order_id=order_id)) if (stock["qty"] or 0) < new_qty: # Restore old deduction before leaving. db.execute("UPDATE stock SET qty = qty - ? WHERE id = ?", (old_order["qty"] or 0, old_order["stock_id"])) flash("❌ Insufficient stock.") return redirect(url_for("edit_order", order_id=order_id)) db.execute("UPDATE stock SET qty = qty - ? WHERE id = ?", (new_qty, new_stock_id)) unit_cost = to_float(stock["unit_cost"]) total_price = unit_cost * new_qty cost = unit_cost * new_qty db.execute(""" UPDATE orders SET order_date=?, stock_id=?, invoice_no=?, order_id=?, serial=?, imei=?, contact1=?, contact2=?, order_via=?, store=?, item_name=?, qty=?, total_price=?, cash_received=?, cost=?, product_warranty_days=?, battery_warranty_days=? WHERE id=? """, ( request.form.get("order_date"), new_stock_id, request.form.get("invoice_no"), request.form.get("order_id"), request.form.get("serial"), request.form.get("imei"), request.form.get("contact1"), request.form.get("contact2"), request.form.get("order_via"), request.form.get("store"), stock["item_name"], new_qty, total_price, to_float(request.form.get("cash_received")), cost, to_int(request.form.get("product_warranty_days")), to_int(request.form.get("battery_warranty_days")), order_id )) record_activity(db, "edit", "orders", order_id, f"Edited order {request.form.get('invoice_no') or order_id}") flash("✅ Order updated successfully.") return redirect(url_for("index")) stocks = db.execute("SELECT id, stock_id, item_name, unit_cost, qty FROM stock ORDER BY item_name").fetchall() return render_template( "orders.html", orders=[], stocks=stocks, stock_items=stocks, edit_row=old_order, edit_order=old_order, order_methods=['Daraz', 'Direct', 'Tudo', 'Payzy', 'Other'], stores=['Main Store', 'Online Store', 'Daraz Store', 'Other'], filter_start_date="", filter_end_date="", filter_order_via="All", filter_store="All", filter_cash_status="All", total_pages=1, current_page=1, search_query="" ) @app.route("/delete_order/") @login_required @admin_required def delete_order(order_id): """Delete an order and safely restore stock.""" with get_db() as db: order = db.execute("SELECT * FROM orders WHERE id = ?", (order_id,)).fetchone() if not order: flash("❌ Order not found.") return redirect(url_for("index", page=request.args.get("page", 1))) db.execute("UPDATE stock SET qty = qty + ? WHERE id = ?", (order["qty"] or 0, order["stock_id"])) db.execute("DELETE FROM orders WHERE id = ?", (order_id,)) record_activity(db, "delete", "orders", order_id, f"Deleted order {order['invoice_no'] or order_id}") flash("🗑️ Order deleted successfully.") current_page = request.args.get("page", 1) return redirect(url_for("index", page=current_page)) @app.route("/invoice/new", methods=["GET", "POST"]) @login_required @admin_required def new_invoice(): """Create a professional multi-line invoice with products and services.""" order_methods = ['Daraz', 'Direct', 'Tudo', 'Payzy', 'Other'] stores = ['Main Store', 'Online Store', 'Daraz Store', 'Other'] with get_db() as db: stocks = db.execute("SELECT id, stock_id, item_name, unit_cost, qty FROM stock ORDER BY item_name").fetchall() if request.method == "POST": order_date_value = request.form.get("order_date") or datetime.now().strftime('%Y-%m-%d') invoice_value = (request.form.get("invoice_no") or "").strip() or generate_invoice_no(db) names = request.form.getlist("item_name[]") types = request.form.getlist("item_type[]") stock_ids = request.form.getlist("stock_id[]") serials = request.form.getlist("serial[]") imeis = request.form.getlist("imei[]") qtys = request.form.getlist("qty[]") unit_prices = request.form.getlist("unit_price[]") costs = request.form.getlist("cost[]") product_warranties = request.form.getlist("product_warranty_days[]") battery_warranties = request.form.getlist("battery_warranty_days[]") notes_list = request.form.getlist("notes[]") line_items = [] for i, raw_name in enumerate(names): item_type = (types[i] if i < len(types) else "product") or "product" item_type = "service" if item_type == "service" else "product" stock_id = to_int(stock_ids[i] if i < len(stock_ids) else 0) stock_row = db.execute("SELECT * FROM stock WHERE id = ?", (stock_id,)).fetchone() if stock_id else None item_name = (raw_name or "").strip() or (stock_row["item_name"] if stock_row else "") qty = to_int(qtys[i] if i < len(qtys) else 1, 1) unit_price = to_float(unit_prices[i] if i < len(unit_prices) else 0) line_cost = to_float(costs[i] if i < len(costs) else 0) if not item_name or qty <= 0: continue if item_type == "product" and stock_row: if (stock_row["qty"] or 0) < qty: flash(f"❌ Insufficient stock for {stock_row['item_name']}.") return redirect(url_for("new_invoice")) if unit_price <= 0: unit_price = to_float(stock_row["unit_cost"]) if line_cost <= 0: line_cost = to_float(stock_row["unit_cost"]) * qty elif item_type == "service": stock_id = None line_cost = line_cost or 0 line_items.append({ "stock_id": stock_id, "item_name": item_name, "item_type": item_type, "serial": (serials[i] if i < len(serials) else "").strip(), "imei": (imeis[i] if i < len(imeis) else "").strip(), "qty": qty, "unit_price": unit_price, "total_price": unit_price * qty, "cost": line_cost, "product_warranty_days": to_int(product_warranties[i] if i < len(product_warranties) else 0), "battery_warranty_days": to_int(battery_warranties[i] if i < len(battery_warranties) else 0), "notes": (notes_list[i] if i < len(notes_list) else "").strip() }) if not line_items: flash("❌ Add at least one product or service line.") return redirect(url_for("new_invoice")) total_price = sum(item["total_price"] for item in line_items) total_cost = sum(item["cost"] for item in line_items) first = line_items[0] summary_name = first["item_name"] if len(line_items) == 1 else f"{first['item_name']} + {len(line_items)-1} more" cur = db.execute(""" INSERT INTO orders (order_date, stock_id, invoice_no, order_id, serial, imei, contact1, contact2, order_via, store, item_name, qty, total_price, cash_received, cost, product_warranty_days, battery_warranty_days) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( order_date_value, first["stock_id"], invoice_value, request.form.get("order_id"), first["serial"], first["imei"], request.form.get("contact1"), request.form.get("contact2"), request.form.get("order_via"), request.form.get("store"), summary_name, sum(item["qty"] for item in line_items), total_price, to_float(request.form.get("cash_received")), total_cost, first["product_warranty_days"], first["battery_warranty_days"] )) new_order_id = cur.lastrowid for item in line_items: if item["item_type"] == "product" and item["stock_id"]: db.execute("UPDATE stock SET qty = qty - ? WHERE id = ?", (item["qty"], item["stock_id"])) db.execute(""" INSERT INTO order_items (order_id, stock_id, item_name, item_type, serial, imei, qty, unit_price, total_price, cost, product_warranty_days, battery_warranty_days, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( new_order_id, item["stock_id"], item["item_name"], item["item_type"], item["serial"], item["imei"], item["qty"], item["unit_price"], item["total_price"], item["cost"], item["product_warranty_days"], item["battery_warranty_days"], item["notes"] )) if item["imei"]: db.execute(""" INSERT INTO imei_devices (imei, serial, stock_id, item_name, status, product_warranty_expiry, battery_warranty_expiry, order_id, updated_at) VALUES (?, ?, ?, ?, 'sold', ?, ?, ?, ?) ON CONFLICT(imei) DO UPDATE SET serial=excluded.serial, stock_id=excluded.stock_id, item_name=excluded.item_name, status='sold', product_warranty_expiry=excluded.product_warranty_expiry, battery_warranty_expiry=excluded.battery_warranty_expiry, order_id=excluded.order_id, updated_at=excluded.updated_at """, ( item["imei"], item["serial"], item["stock_id"], item["item_name"], warranty_expiry_from_days(order_date_value, item["product_warranty_days"]), warranty_expiry_from_days(order_date_value, item["battery_warranty_days"]), new_order_id, datetime.now().strftime('%Y-%m-%d %H:%M:%S') )) record_activity(db, "add", "orders", new_order_id, f"Created multi-item invoice {invoice_value}") flash("✅ Invoice created successfully.") return redirect(url_for("invoice", order_id=new_order_id)) return render_template("invoice_builder.html", stocks=stocks, order_methods=order_methods, stores=stores) @app.route("/invoice/") @login_required def invoice(order_id): """Generate invoice for an order""" with get_db() as db: order = db.execute("SELECT * FROM orders WHERE id = ?", (order_id,)).fetchone() if not order: flash("❌ Order not found.") return redirect(url_for("index")) items = ensure_order_items_for_order(db, order) totals = summarize_order_items(items) return render_template("invoice.html", order=order, items=items, totals=totals, print_size=request.args.get("size", "a6")) @app.route("/generate_invoice/") @login_required def generate_invoice(order_id): """Backward-compatible invoice endpoint used by older templates.""" return invoice(order_id) # ============================================================================ # STOCK ROUTES # ============================================================================ @app.route("/stock", methods=["GET", "POST"]) @login_required def stock_manage(): """Stock management with search, filters, edit/delete and pagination.""" with get_db() as db: if request.method == "POST": edit_id = request.form.get("edit_id") values = ( request.form.get("stock_id", "").strip(), request.form.get("item_name", "").strip(), request.form.get("item_category", "").strip(), request.form.get("date") or None, to_float(request.form.get("order_price")), to_float(request.form.get("logistic_charges")), to_float(request.form.get("shipping_charges")), to_float(request.form.get("unit_cost")), to_int(request.form.get("qty")), to_int(request.form.get("ordered_qty")) ) if edit_id: db.execute(""" UPDATE stock SET stock_id=?, item_name=?, item_category=?, date=?, order_price=?, logistic_charges=?, shipping_charges=?, unit_cost=?, qty=?, ordered_qty=? WHERE id=? """, values + (edit_id,)) record_activity(db, "edit", "stock", to_int(edit_id), f"Edited stock {values[0]} - {values[1]}") flash("✅ Stock updated successfully.") else: db.execute(""" INSERT INTO stock (stock_id, item_name, item_category, date, order_price, logistic_charges, shipping_charges, unit_cost, qty, ordered_qty) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, values) record_activity(db, "add", "stock", None, f"Added stock {values[0]} - {values[1]}") flash("✅ Stock added successfully.") return redirect(url_for("stock_manage")) delete_id = request.args.get("delete") if delete_id: db.execute("DELETE FROM stock WHERE id=?", (delete_id,)) record_activity(db, "delete", "stock", to_int(delete_id), "Deleted stock item") flash("🗑️ Stock item deleted successfully.") return redirect(url_for("stock_manage", page=request.args.get("page", 1))) edit_id = request.args.get("edit") edit_row = db.execute("SELECT * FROM stock WHERE id=?", (edit_id,)).fetchone() if edit_id else None search_query = request.args.get("q", "").strip() filter_category = request.args.get("category") or "All" filter_stock_level = request.args.get("stock_level") or "All" filter_available_only = request.args.get("available_only") == "on" query = "SELECT * FROM stock WHERE 1=1" params = [] if search_query: query += " AND (stock_id LIKE ? OR item_name LIKE ? OR item_category LIKE ?)" like_query = f"%{search_query}%" params.extend([like_query, like_query, like_query]) if filter_category != "All": query += " AND item_category = ?" params.append(filter_category) if filter_available_only: query += " AND qty > 0" if filter_stock_level == "High": query += " AND qty > 10" elif filter_stock_level == "Medium": query += " AND qty BETWEEN 1 AND 10" elif filter_stock_level == "Low": query += " AND qty BETWEEN 1 AND 5" elif filter_stock_level == "Out": query += " AND qty <= 0" page = get_page() total_stock = db.execute(query.replace("SELECT *", "SELECT COUNT(*)"), params).fetchone()[0] total_pages = max(1, math.ceil(total_stock / PER_PAGE)) page = min(page, total_pages) offset = (page - 1) * PER_PAGE stocks = db.execute(query + " ORDER BY id DESC LIMIT ? OFFSET ?", params + [PER_PAGE, offset]).fetchall() categories = db.execute("SELECT DISTINCT item_category FROM stock WHERE item_category IS NOT NULL AND item_category != '' ORDER BY item_category").fetchall() return render_template( "stock.html", stocks=stocks, stock=stocks, categories=categories, edit_row=edit_row, search_query=search_query, filter_category=filter_category, filter_stock_level=filter_stock_level, filter_available_only=filter_available_only, total_pages=total_pages, current_page=page ) # ============================================================================ # EXPENSES ROUTES # ============================================================================ @app.route("/expenses", methods=["GET", "POST"]) @login_required def manage_expenses(): """Expenses management with search, filters, edit/delete and pagination.""" payment_methods = ['Cash', 'Bank Transfer', 'Card', 'Other'] with get_db() as db: if request.method == "POST": edit_id = request.form.get("edit_id") values = ( request.form.get("date"), request.form.get("reason", "").strip(), to_float(request.form.get("amount")), request.form.get("payment_method") or "Cash" ) if edit_id: db.execute("UPDATE expenses SET date=?, reason=?, amount=?, payment_method=? WHERE id=?", values + (edit_id,)) flash("✅ Expense updated successfully.") else: db.execute("INSERT INTO expenses (date, reason, amount, payment_method) VALUES (?, ?, ?, ?)", values) flash("✅ Expense added successfully.") return redirect(url_for("manage_expenses")) delete_id = request.args.get("delete") if delete_id: db.execute("DELETE FROM expenses WHERE id=?", (delete_id,)) flash("🗑️ Expense deleted successfully.") return redirect(url_for("manage_expenses", page=request.args.get("page", 1))) edit_id = request.args.get("edit") edit_row = db.execute("SELECT * FROM expenses WHERE id=?", (edit_id,)).fetchone() if edit_id else None search_query = request.args.get("q", "").strip() filter_start_date = request.args.get("start_date") or "" filter_end_date = request.args.get("end_date") or "" filter_payment_method = request.args.get("payment_method") or "All" query = "SELECT * FROM expenses WHERE 1=1" params = [] if search_query: query += " AND (reason LIKE ? OR payment_method LIKE ?)" like_query = f"%{search_query}%" params.extend([like_query, like_query]) if filter_start_date: query += " AND date >= ?" params.append(filter_start_date) if filter_end_date: query += " AND date <= ?" params.append(filter_end_date) if filter_payment_method != "All": query += " AND payment_method = ?" params.append(filter_payment_method) page = get_page() total_expenses = db.execute(query.replace("SELECT *", "SELECT COUNT(*)"), params).fetchone()[0] total_pages = max(1, math.ceil(total_expenses / PER_PAGE)) page = min(page, total_pages) offset = (page - 1) * PER_PAGE expenses = db.execute(query + " ORDER BY date DESC, id DESC LIMIT ? OFFSET ?", params + [PER_PAGE, offset]).fetchall() return render_template( "expenses.html", expenses=expenses, edit_row=edit_row, payment_methods=payment_methods, search_query=search_query, filter_start_date=filter_start_date, filter_end_date=filter_end_date, filter_payment_method=filter_payment_method, total_pages=total_pages, current_page=page ) # ============================================================================ # PROFIT ROUTES # ============================================================================ @app.route("/profit", methods=["GET", "POST"]) @login_required def calculate_profit(): """Calculate profit for a date range""" result = None if request.method == "POST": start_date = request.form.get("start_date") end_date = request.form.get("end_date") with get_db() as db: order_data = db.execute(""" SELECT COALESCE(SUM(cash_received), 0) AS total_cash, COALESCE(SUM(cost), 0) AS total_cost FROM orders WHERE order_date BETWEEN ? AND ? AND cash_received > 0 """, (start_date, end_date)).fetchone() expense_data = db.execute(""" SELECT COALESCE(SUM(amount), 0) AS total_expenses FROM expenses WHERE date BETWEEN ? AND ? """, (start_date, end_date)).fetchone() total_cash = order_data["total_cash"] total_cost = order_data["total_cost"] total_expenses = expense_data["total_expenses"] monthly_profit = total_cash - total_cost final_profit = monthly_profit - total_expenses def format_currency(value): return "Rs. {:,.2f}".format(value) result = { "start_date": start_date, "end_date": end_date, "cash_received": format_currency(total_cash), "total_cost": format_currency(total_cost), "monthly_profit": format_currency(monthly_profit), "expenses": format_currency(total_expenses), "final_profit": format_currency(final_profit), "is_profit": final_profit >= 0 } return render_template("profit.html", result=result) # ============================================================================ # CASH BOOK ROUTES # ============================================================================ @app.route("/cashbook", methods=["GET", "POST"]) @login_required def manage_cashbook(): """Cash book management with search and filters""" with get_db() as db: types = ['credit', 'debit'] payment_methods = ['Cash', 'Bank Transfer', 'Card', 'Other'] payment_done_bys = ['Kalana', 'Eshan', 'Yasiru', '3rd Party'] if request.method == "POST": edit_id = request.form.get("edit_id") date = request.form.get("date") reason = request.form.get("reason") amount = to_float(request.form.get("amount")) ttype = request.form.get("type") payment_method = request.form.get("payment_method") payment_done_by = request.form.get("payment_done_by") description = request.form.get("description") if edit_id: db.execute(""" UPDATE cash_book SET date=?, reason=?, amount=?, type=?, payment_method=?, payment_done_by=?, description=? WHERE id=?; """, (date, reason, amount, ttype, payment_method, payment_done_by, description, edit_id)) flash("✅ Transaction updated.") else: db.execute(""" INSERT INTO cash_book (date, reason, amount, type, payment_method, payment_done_by, description) VALUES (?, ?, ?, ?, ?, ?, ?); """, (date, reason, amount, ttype, payment_method, payment_done_by, description)) flash("✅ Transaction added.") return redirect(url_for("manage_cashbook")) delete_id = request.args.get("delete") if delete_id: db.execute("DELETE FROM cash_book WHERE id=?;", (delete_id,)) flash("🗑️ Transaction deleted.") return redirect(url_for("manage_cashbook", page=request.args.get("page", 1))) edit_id = request.args.get("edit") edit_row = db.execute("SELECT * FROM cash_book WHERE id=?;", (edit_id,)).fetchone() if edit_id else None # Get search query search_query = request.args.get("q", "").strip() # Get filter parameters filter_start_date = request.args.get("start_date") filter_end_date = request.args.get("end_date") filter_payment_method = request.args.get("payment_method") or "All" filter_payment_done_by = request.args.get("payment_done_by") or "All" query = "SELECT * FROM cash_book WHERE 1=1" params = [] # Add search condition if search_query: query += " AND (reason LIKE ? OR description LIKE ?)" like_query = f"%{search_query}%" params.extend([like_query, like_query]) # Add date filters if filter_start_date: query += " AND date >= ?" params.append(filter_start_date) if filter_end_date: query += " AND date <= ?" params.append(filter_end_date) # Add payment method filter if filter_payment_method != "All": query += " AND payment_method = ?" params.append(filter_payment_method) # Add payment done by filter if filter_payment_done_by != "All": query += " AND payment_done_by = ?" params.append(filter_payment_done_by) # Get pagination page = get_page() count_query = query.replace("SELECT *", "SELECT COUNT(*)") total_transactions = db.execute(count_query, params).fetchone()[0] total_pages = max(1, math.ceil(total_transactions / PER_PAGE)) page = min(page, total_pages) offset = (page - 1) * PER_PAGE query += " ORDER BY date DESC, id DESC LIMIT ? OFFSET ?" transactions = db.execute(query, params + [PER_PAGE, offset]).fetchall() # Calculate running balance for filtered results balance_query = "SELECT * FROM cash_book WHERE 1=1" balance_params = [] if search_query: balance_query += " AND (reason LIKE ? OR description LIKE ?)" balance_params.extend([like_query, like_query]) if filter_start_date: balance_query += " AND date >= ?" balance_params.append(filter_start_date) if filter_end_date: balance_query += " AND date <= ?" balance_params.append(filter_end_date) if filter_payment_method != "All": balance_query += " AND payment_method = ?" balance_params.append(filter_payment_method) if filter_payment_done_by != "All": balance_query += " AND payment_done_by = ?" balance_params.append(filter_payment_done_by) balance_query += " ORDER BY date ASC, id ASC" all_txns = db.execute(balance_query, balance_params).fetchall() balance = 0 balance_dict = {} for t in all_txns: amt = t["amount"] if t["type"] == "credit" else -t["amount"] balance += amt balance_dict[t["id"]] = balance txns_with_balance = [] for t in transactions: row = dict(t) row["running_balance"] = balance_dict.get(t["id"], 0) txns_with_balance.append(row) return render_template( "cash_book.html", transactions=txns_with_balance, edit_row=edit_row, types=types, payment_methods=payment_methods, payment_done_bys=payment_done_bys, filter_start_date=filter_start_date or "", filter_end_date=filter_end_date or "", filter_payment_method=filter_payment_method, filter_payment_done_by=filter_payment_done_by, search_query=search_query, total_pages=total_pages, current_page=page ) @app.route('/export_excel') @login_required def export_excel(): """Export orders to Excel. Pandas is imported only here so the app can start even if pandas is missing.""" start_date = request.args.get('start_date') end_date = request.args.get('end_date') if not start_date or not end_date: return "Please provide start_date and end_date", 400 try: import pandas as pd except Exception as exc: return f"Excel export needs pandas/xlsxwriter installed. Install requirements first. Details: {exc}", 500 with get_db() as db: rows = db.execute(""" SELECT * FROM orders WHERE order_date >= ? AND order_date <= ? ORDER BY order_date DESC """, (start_date, end_date)).fetchall() data = [dict(row) for row in rows] if not data: return "No orders found for the selected date range.", 404 df = pd.DataFrame(data) output = io.BytesIO() with pd.ExcelWriter(output, engine='xlsxwriter') as writer: df.to_excel(writer, index=False, sheet_name='Orders') output.seek(0) return send_file( output, mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', download_name=f'orders_{start_date}_to_{end_date}.xlsx', as_attachment=True ) # ============================================================================ # ADMIN ROUTES # ============================================================================ @app.route("/download_db") @login_required @admin_required def download_db(): """Download database file (admin only)""" return send_file( DB_PATH, as_attachment=True, download_name="orders.db", mimetype="application/octet-stream" ) # ============================================================================ # TASK MANAGEMENT ROUTES # ============================================================================ @app.route("/tasks", methods=["GET", "POST"]) @login_required def manage_tasks(): """Task management with search and filters""" with get_db() as db: priorities = ['Low', 'Medium', 'High', 'Urgent'] statuses = ['Todo', 'In Progress', 'Completed', 'On Hold'] # Get all users for assignment dropdown users_rows = db.execute("SELECT username, full_name FROM users WHERE is_active = 1").fetchall() all_users = [f"{u['full_name'] or u['username']}" for u in users_rows] # Get unique categories and tags categories_rows = db.execute("SELECT DISTINCT category FROM tasks WHERE category IS NOT NULL AND category != ''").fetchall() categories = [c['category'] for c in categories_rows] if request.method == "POST": action = request.form.get("action") task_id = request.form.get("task_id") if action == "add" or action == "edit": title = request.form.get("title") description = request.form.get("description", "") priority = request.form.get("priority", "Medium") status = request.form.get("status", "Todo") assigned_to = request.form.get("assigned_to") due_date = request.form.get("due_date") category = request.form.get("category") tags = request.form.get("tags", "") # Handle completion timestamp completed_at = None if status == "Completed": completed_at = datetime.now().strftime('%Y-%m-%d %H:%M:%S') if action == "edit" and task_id: db.execute(""" UPDATE tasks SET title=?, description=?, priority=?, status=?, assigned_to=?, due_date=?, category=?, tags=?, updated_at=?, completed_at=? WHERE id=? """, (title, description, priority, status, assigned_to, due_date, category, tags, datetime.now().strftime('%Y-%m-%d %H:%M:%S'), completed_at, task_id)) flash("✅ Task updated successfully.") else: db.execute(""" INSERT INTO tasks (title, description, priority, status, assigned_to, created_by, due_date, category, tags) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, (title, description, priority, status, assigned_to, session.get('full_name'), due_date, category, tags)) flash("✅ Task created successfully.") return redirect(url_for("manage_tasks")) elif action == "delete" and task_id: db.execute("DELETE FROM tasks WHERE id=?", (task_id,)) flash("🗑️ Task deleted.") return redirect(url_for("manage_tasks")) # Get filter parameters filter_status = request.args.get("status", "All") filter_priority = request.args.get("priority", "All") filter_assigned = request.args.get("assigned", "All") filter_category = request.args.get("category", "All") search_query = request.args.get("q", "").strip() # Build query query = "SELECT * FROM tasks WHERE 1=1" params = [] if filter_status != "All": query += " AND status = ?" params.append(filter_status) if filter_priority != "All": query += " AND priority = ?" params.append(filter_priority) if filter_assigned != "All": query += " AND assigned_to = ?" params.append(filter_assigned) if filter_category != "All": query += " AND category = ?" params.append(filter_category) if search_query: query += " AND (title LIKE ? OR description LIKE ? OR tags LIKE ?)" like_query = f"%{search_query}%" params.extend([like_query, like_query, like_query]) # Pagination page = get_page() count_query = query.replace("SELECT *", "SELECT COUNT(*)") total_tasks = db.execute(count_query, params).fetchone()[0] total_pages = max(1, math.ceil(total_tasks / PER_PAGE)) page = min(page, total_pages) offset = (page - 1) * PER_PAGE query += " ORDER BY CASE status WHEN 'In Progress' THEN 1 WHEN 'Todo' THEN 2 WHEN 'On Hold' THEN 3 WHEN 'Completed' THEN 4 END, " query += "CASE priority WHEN 'Urgent' THEN 1 WHEN 'High' THEN 2 WHEN 'Medium' THEN 3 WHEN 'Low' THEN 4 END, " query += "due_date ASC NULLS LAST, created_at DESC LIMIT ? OFFSET ?" tasks = db.execute(query, params + [PER_PAGE, offset]).fetchall() # Get task statistics stats = { 'total': db.execute("SELECT COUNT(*) FROM tasks").fetchone()[0], 'todo': db.execute("SELECT COUNT(*) FROM tasks WHERE status='Todo'").fetchone()[0], 'in_progress': db.execute("SELECT COUNT(*) FROM tasks WHERE status='In Progress'").fetchone()[0], 'completed': db.execute("SELECT COUNT(*) FROM tasks WHERE status='Completed'").fetchone()[0], 'on_hold': db.execute("SELECT COUNT(*) FROM tasks WHERE status='On Hold'").fetchone()[0], 'overdue': db.execute(""" SELECT COUNT(*) FROM tasks WHERE status != 'Completed' AND due_date < date('now') AND due_date IS NOT NULL """).fetchone()[0] } return render_template( "tasks.html", tasks=tasks, priorities=priorities, statuses=statuses, all_users=all_users, categories=categories, stats=stats, filter_status=filter_status, filter_priority=filter_priority, filter_assigned=filter_assigned, filter_category=filter_category, search_query=search_query, total_pages=total_pages, current_page=page ) @app.route("/tasks/") @login_required def view_task(task_id): """View single task details""" with get_db() as db: task = db.execute("SELECT * FROM tasks WHERE id=?", (task_id,)).fetchone() if not task: flash("❌ Task not found.") return redirect(url_for("manage_tasks")) return render_template("task_detail.html", task=task) # ============================================================================ # ASSETS MANAGEMENT ROUTES # ============================================================================ @app.route("/assets", methods=["GET", "POST"]) @login_required def manage_assets(): """Asset management with CRUD operations""" with get_db() as db: categories = ['Office Equipment', 'Furniture', 'Electronics', 'Vehicles', 'Software', 'Property', 'Tools', 'Other'] statuses = ['Active', 'In Repair', 'Retired', 'Sold'] # Get all users for assignment dropdown users_rows = db.execute("SELECT username, full_name FROM users WHERE is_active = 1").fetchall() all_users = [f"{u['full_name'] or u['username']}" for u in users_rows] if request.method == "POST": action = request.form.get("action") asset_id = request.form.get("asset_id") if action == "add" or action == "edit": asset_name = request.form.get("asset_name") category = request.form.get("category") purchase_date = request.form.get("purchase_date") purchase_price = to_float(request.form.get("purchase_price")) current_value = to_float(request.form.get("current_value")) depreciation_rate = to_float(request.form.get("depreciation_rate")) location = request.form.get("location") assigned_to = request.form.get("assigned_to") warranty_expiry = request.form.get("warranty_expiry") maintenance_date = request.form.get("maintenance_date") status = request.form.get("status", "Active") serial_number = request.form.get("serial_number") supplier = request.form.get("supplier") notes = request.form.get("notes") if action == "edit" and asset_id: db.execute(""" UPDATE assets SET asset_name=?, category=?, purchase_date=?, purchase_price=?, current_value=?, depreciation_rate=?, location=?, assigned_to=?, warranty_expiry=?, maintenance_date=?, status=?, serial_number=?, supplier=?, notes=?, updated_at=? WHERE id=? """, (asset_name, category, purchase_date, purchase_price, current_value, depreciation_rate, location, assigned_to, warranty_expiry, maintenance_date, status, serial_number, supplier, notes, datetime.now().strftime('%Y-%m-%d %H:%M:%S'), asset_id)) flash("✅ Asset updated successfully.") else: db.execute(""" INSERT INTO assets (asset_name, category, purchase_date, purchase_price, current_value, depreciation_rate, location, assigned_to, warranty_expiry, maintenance_date, status, serial_number, supplier, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (asset_name, category, purchase_date, purchase_price, current_value, depreciation_rate, location, assigned_to, warranty_expiry, maintenance_date, status, serial_number, supplier, notes)) flash("✅ Asset added successfully.") return redirect(url_for("manage_assets")) elif action == "delete" and asset_id: db.execute("DELETE FROM assets WHERE id=?", (asset_id,)) flash("🗑️ Asset deleted.") return redirect(url_for("manage_assets")) # Get filter parameters filter_category = request.args.get("category", "All") filter_status = request.args.get("status", "All") filter_assigned = request.args.get("assigned", "All") search_query = request.args.get("q", "").strip() # Build query query = "SELECT * FROM assets WHERE 1=1" params = [] if filter_category != "All": query += " AND category = ?" params.append(filter_category) if filter_status != "All": query += " AND status = ?" params.append(filter_status) if filter_assigned != "All": query += " AND assigned_to = ?" params.append(filter_assigned) if search_query: query += " AND (asset_name LIKE ? OR serial_number LIKE ? OR location LIKE ? OR notes LIKE ?)" like_query = f"%{search_query}%" params.extend([like_query, like_query, like_query, like_query]) # Pagination page = get_page() count_query = query.replace("SELECT *", "SELECT COUNT(*)") total_assets = db.execute(count_query, params).fetchone()[0] total_pages = max(1, math.ceil(total_assets / PER_PAGE)) page = min(page, total_pages) offset = (page - 1) * PER_PAGE query += " ORDER BY created_at DESC LIMIT ? OFFSET ?" assets = db.execute(query, params + [PER_PAGE, offset]).fetchall() # Calculate summary statistics summary_query = "SELECT * FROM assets WHERE 1=1" summary_params = [] if filter_category != "All": summary_query += " AND category = ?" summary_params.append(filter_category) if filter_status != "All": summary_query += " AND status = ?" summary_params.append(filter_status) if filter_assigned != "All": summary_query += " AND assigned_to = ?" summary_params.append(filter_assigned) if search_query: summary_query += " AND (asset_name LIKE ? OR serial_number LIKE ? OR location LIKE ? OR notes LIKE ?)" like_query = f"%{search_query}%" summary_params.extend([like_query, like_query, like_query, like_query]) all_filtered_assets = db.execute(summary_query, summary_params).fetchall() total_purchase_value = sum(asset['purchase_price'] for asset in all_filtered_assets) total_current_value = sum(asset['current_value'] for asset in all_filtered_assets) total_depreciation = total_purchase_value - total_current_value avg_depreciation_rate = sum(asset['depreciation_rate'] for asset in all_filtered_assets) / len(all_filtered_assets) if all_filtered_assets else 0 # Get asset statistics stats = { 'total': len(all_filtered_assets), 'active': sum(1 for a in all_filtered_assets if a['status'] == 'Active'), 'in_repair': sum(1 for a in all_filtered_assets if a['status'] == 'In Repair'), 'retired': sum(1 for a in all_filtered_assets if a['status'] == 'Retired'), 'sold': sum(1 for a in all_filtered_assets if a['status'] == 'Sold'), 'total_purchase_value': total_purchase_value, 'total_current_value': total_current_value, 'total_depreciation': total_depreciation, 'avg_depreciation_rate': avg_depreciation_rate } # Category breakdown category_breakdown = {} for category in categories: cat_assets = [a for a in all_filtered_assets if a['category'] == category] if cat_assets: category_breakdown[category] = { 'count': len(cat_assets), 'purchase_value': sum(a['purchase_price'] for a in cat_assets), 'current_value': sum(a['current_value'] for a in cat_assets) } return render_template( "assets.html", assets=assets, categories=categories, statuses=statuses, all_users=all_users, stats=stats, category_breakdown=category_breakdown, filter_category=filter_category, filter_status=filter_status, filter_assigned=filter_assigned, search_query=search_query, total_pages=total_pages, current_page=page ) # ============================================================================ # PHASE 1 BUSINESS ROUTES # ============================================================================ @app.route('/global-search') @login_required def global_search(): """Search customers/orders/devices by phone, IMEI, serial, invoice, or order ID.""" q = request.args.get('q', '').strip() orders = [] devices = [] returns = [] if q: like = f"%{q}%" with get_db() as db: orders = db.execute(""" SELECT * FROM orders WHERE invoice_no LIKE ? OR order_id LIKE ? OR serial LIKE ? OR imei LIKE ? OR contact1 LIKE ? OR contact2 LIKE ? OR item_name LIKE ? ORDER BY id DESC LIMIT 50 """, (like, like, like, like, like, like, like)).fetchall() devices = db.execute(""" SELECT * FROM imei_devices WHERE imei LIKE ? OR serial LIKE ? OR item_name LIKE ? OR trcsl_reference LIKE ? OR import_batch LIKE ? ORDER BY id DESC LIMIT 50 """, (like, like, like, like, like)).fetchall() returns = db.execute(""" SELECT r.*, o.invoice_no, o.imei, o.item_name FROM order_returns r LEFT JOIN orders o ON o.id = r.order_id WHERE o.invoice_no LIKE ? OR o.imei LIKE ? OR o.serial LIKE ? OR o.contact1 LIKE ? OR r.reason LIKE ? ORDER BY r.id DESC LIMIT 50 """, (like, like, like, like, like)).fetchall() return render_template('global_search.html', q=q, orders=orders, devices=devices, returns=returns) @app.route('/imei', methods=['GET', 'POST']) @login_required @admin_required def manage_imei(): """IMEI, serial, warranty and TRCSL tracking.""" statuses = ['in_stock', 'sold', 'returned', 'faulty'] trcsl_statuses = ['Not Required', 'Approved', 'Pending', 'Not Approved'] with get_db() as db: if request.method == 'POST': device_id = request.form.get('device_id') imei = request.form.get('imei', '').strip() if not imei: flash('❌ IMEI is required.') return redirect(url_for('manage_imei')) values = ( imei, request.form.get('serial', '').strip(), to_int(request.form.get('stock_id')) or None, request.form.get('item_name', '').strip(), request.form.get('status') or 'in_stock', request.form.get('product_warranty_expiry') or None, request.form.get('battery_warranty_expiry') or None, 1 if request.form.get('trcsl_required') == 'on' else 0, request.form.get('trcsl_status') or 'Not Required', request.form.get('trcsl_reference', '').strip(), request.form.get('import_batch', '').strip(), request.form.get('location', '').strip() or 'Shop', datetime.now().strftime('%Y-%m-%d %H:%M:%S') ) if device_id: db.execute(""" UPDATE imei_devices SET imei=?, serial=?, stock_id=?, item_name=?, status=?, product_warranty_expiry=?, battery_warranty_expiry=?, trcsl_required=?, trcsl_status=?, trcsl_reference=?, import_batch=?, location=?, updated_at=? WHERE id=? """, values + (device_id,)) record_activity(db, 'edit', 'imei_devices', to_int(device_id), f'Updated IMEI {imei}') flash('✅ Device updated.') else: try: cur = db.execute(""" INSERT INTO imei_devices (imei, serial, stock_id, item_name, status, product_warranty_expiry, battery_warranty_expiry, trcsl_required, trcsl_status, trcsl_reference, import_batch, location, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, values) record_activity(db, 'add', 'imei_devices', cur.lastrowid, f'Added IMEI {imei}') flash('✅ Device added.') except sqlite3.IntegrityError: flash('❌ This IMEI already exists.') return redirect(url_for('manage_imei')) delete_id = request.args.get('delete') if delete_id: db.execute('DELETE FROM imei_devices WHERE id=?', (delete_id,)) record_activity(db, 'delete', 'imei_devices', to_int(delete_id), 'Deleted IMEI/device record') flash('🗑️ Device deleted.') return redirect(url_for('manage_imei')) edit_id = request.args.get('edit') edit_row = db.execute('SELECT * FROM imei_devices WHERE id=?', (edit_id,)).fetchone() if edit_id else None q = request.args.get('q', '').strip() status_filter = request.args.get('status') or 'All' trcsl_filter = request.args.get('trcsl_status') or 'All' sellable_only = request.args.get('sellable_only') == 'on' query = 'SELECT * FROM imei_devices WHERE 1=1' params = [] if q: query += ' AND (imei LIKE ? OR serial LIKE ? OR item_name LIKE ? OR import_batch LIKE ? OR trcsl_reference LIKE ?)' like = f'%{q}%' params.extend([like, like, like, like, like]) if status_filter != 'All': query += ' AND status=?' params.append(status_filter) if trcsl_filter != 'All': query += ' AND trcsl_status=?' params.append(trcsl_filter) if sellable_only: query += " AND status='in_stock' AND (trcsl_required=0 OR trcsl_status='Approved')" devices = db.execute(query + ' ORDER BY id DESC LIMIT 200', params).fetchall() stocks = db.execute('SELECT id, stock_id, item_name FROM stock ORDER BY item_name').fetchall() return render_template('imei_devices.html', devices=devices, edit_row=edit_row, stocks=stocks, statuses=statuses, trcsl_statuses=trcsl_statuses, q=q, status_filter=status_filter, trcsl_filter=trcsl_filter, sellable_only=sellable_only) @app.route('/returns', methods=['GET', 'POST']) @login_required @admin_required def manage_returns(): """Order return, replacement and refund tracking.""" with get_db() as db: if request.method == 'POST': order_id = to_int(request.form.get('order_id')) order = db.execute('SELECT * FROM orders WHERE id=?', (order_id,)).fetchone() if not order: flash('❌ Order not found.') return redirect(url_for('manage_returns')) cur = db.execute(""" INSERT INTO order_returns (order_id, return_date, return_type, refund_amount, replacement_item, reason, created_by) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( order_id, request.form.get('return_date') or datetime.now().strftime('%Y-%m-%d'), request.form.get('return_type') or 'Return', to_float(request.form.get('refund_amount')), request.form.get('replacement_item', '').strip(), request.form.get('reason', '').strip(), current_user_name() )) if order['imei']: new_status = 'returned' if request.form.get('return_type') != 'Replacement' else 'faulty' db.execute('UPDATE imei_devices SET status=?, updated_at=? WHERE imei=?', (new_status, datetime.now().strftime('%Y-%m-%d %H:%M:%S'), order['imei'])) record_activity(db, 'add', 'order_returns', cur.lastrowid, f"Return/refund for order {order['invoice_no'] or order_id}") flash('✅ Return/refund recorded.') return redirect(url_for('manage_returns')) rows = db.execute(""" SELECT r.*, o.invoice_no, o.imei, o.item_name, o.contact1 FROM order_returns r LEFT JOIN orders o ON o.id = r.order_id ORDER BY r.id DESC LIMIT 200 """).fetchall() recent_orders = db.execute('SELECT id, invoice_no, order_id, item_name, imei, contact1 FROM orders ORDER BY id DESC LIMIT 200').fetchall() return render_template('returns.html', returns=rows, recent_orders=recent_orders) @app.route('/backups') @login_required @admin_required def backups(): """Backup management.""" if request.args.get('create') == '1': path = make_daily_backup(force=True) keep_recent_backups(30) add_activity('backup', 'orders.db', None, f'Created backup {path.name if path else ""}') flash('✅ Backup created.') return redirect(url_for('backups')) BACKUP_DIR.mkdir(exist_ok=True) files = sorted(BACKUP_DIR.glob('orders_*.db'), key=lambda x: x.stat().st_mtime, reverse=True) backup_rows = [{'name': f.name, 'size': f.stat().st_size, 'modified': datetime.fromtimestamp(f.stat().st_mtime).strftime('%Y-%m-%d %H:%M:%S')} for f in files] return render_template('backups.html', backups=backup_rows) @app.route('/backups/download/') @login_required @admin_required def download_backup(filename): path = BACKUP_DIR / Path(filename).name if not path.exists(): flash('❌ Backup not found.') return redirect(url_for('backups')) return send_file(path, as_attachment=True, download_name=path.name, mimetype='application/octet-stream') @app.route('/activity-log') @login_required @admin_required def activity_log(): with get_db() as db: rows = db.execute('SELECT * FROM activity_log ORDER BY id DESC LIMIT 300').fetchall() return render_template('activity_log.html', logs=rows) # ============================================================================ # ============================================================================ # PHASE 2 ROUTES # ============================================================================ @app.route('/delivery', methods=['GET', 'POST']) @login_required def delivery_tracking(): """Track courier, tracking number and delivery status for orders.""" statuses = ['Pending', 'Packed', 'Shipped', 'Delivered', 'Returned', 'Cancelled'] with get_db() as db: if request.method == 'POST': order_id = to_int(request.form.get('order_id')) order = db.execute('SELECT * FROM orders WHERE id=? AND (deleted_at IS NULL OR deleted_at="")', (order_id,)).fetchone() if not order: flash('❌ Order not found.') return redirect(url_for('delivery_tracking')) db.execute("UPDATE orders SET delivery_courier=?, tracking_no=?, delivery_status=? WHERE id=?", ( request.form.get('delivery_courier','').strip(), request.form.get('tracking_no','').strip(), request.form.get('delivery_status') or 'Pending', order_id )) record_activity(db, 'edit', 'orders', order_id, 'Updated delivery tracking') flash('✅ Delivery details updated.') return redirect(url_for('delivery_tracking')) status_filter = request.args.get('status') or 'All' q = request.args.get('q','').strip() sql = 'SELECT * FROM orders WHERE (deleted_at IS NULL OR deleted_at="")' params = [] if status_filter != 'All': sql += " AND COALESCE(delivery_status,'Pending')=?" params.append(status_filter) if q: like = f'%{q}%' sql += ' AND (invoice_no LIKE ? OR order_id LIKE ? OR contact1 LIKE ? OR tracking_no LIKE ? OR item_name LIKE ?)' params.extend([like, like, like, like, like]) orders = db.execute(sql + ' ORDER BY id DESC LIMIT 300', params).fetchall() return render_template('delivery.html', orders=orders, statuses=statuses, status_filter=status_filter, q=q) @app.route('/reports/platform') @login_required def platform_report(): """Platform wise sales report: Daraz/Tudo/Direct/etc.""" start_date = request.args.get('start_date') or get_current_period()['start_date'] end_date = request.args.get('end_date') or get_current_period()['end_date'] with get_db() as db: rows = db.execute(""" SELECT COALESCE(order_via, 'Direct') AS platform, COUNT(*) AS order_count, COALESCE(SUM(qty),0) AS total_qty, COALESCE(SUM(total_price),0) AS sales_total, COALESCE(SUM(cash_received),0) AS cash_total, COALESCE(SUM(cost),0) AS cost_total, COALESCE(SUM(cash_received - cost),0) AS gross_profit FROM orders WHERE (deleted_at IS NULL OR deleted_at='') AND order_date BETWEEN ? AND ? GROUP BY COALESCE(order_via, 'Direct') ORDER BY cash_total DESC """, (start_date, end_date)).fetchall() totals = db.execute(""" SELECT COALESCE(SUM(cash_received),0) AS revenue, COALESCE(SUM(cost),0) AS cost, COALESCE(SUM(cash_received - cost),0) AS gross_profit FROM orders WHERE (deleted_at IS NULL OR deleted_at='') AND order_date BETWEEN ? AND ? """, (start_date, end_date)).fetchone() return render_template('platform_report.html', rows=rows, totals=totals, start_date=start_date, end_date=end_date) @app.route('/cash-summary') @login_required def cash_summary(): """Cash, bank/card/other and total business balance summary.""" with get_db() as db: rows = db.execute(""" SELECT payment_method, COALESCE(SUM(CASE WHEN type='credit' THEN amount ELSE -amount END),0) AS balance, COALESCE(SUM(CASE WHEN type='credit' THEN amount ELSE 0 END),0) AS credits, COALESCE(SUM(CASE WHEN type='debit' THEN amount ELSE 0 END),0) AS debits FROM cash_book GROUP BY payment_method ORDER BY payment_method """).fetchall() total = db.execute("SELECT COALESCE(SUM(CASE WHEN type='credit' THEN amount ELSE -amount END),0) AS balance FROM cash_book").fetchone() return render_template('cash_summary.html', rows=rows, total=total) @app.route('/import-cost-calculator', methods=['GET', 'POST']) @login_required def import_cost_calculator(): """Calculate landed unit cost for imported stock.""" result = None if request.method == 'POST': qty = to_int(request.form.get('qty'), 1) or 1 product_cost = to_float(request.form.get('product_cost')) shipping = to_float(request.form.get('shipping_cost')) customs = to_float(request.form.get('customs_cost')) other = to_float(request.form.get('other_cost')) total = product_cost + shipping + customs + other result = {'qty': qty, 'total_cost': total, 'unit_cost': total / qty if qty else 0} return render_template('import_cost_calculator.html', result=result) @app.route('/suppliers', methods=['GET', 'POST']) @login_required @admin_required def suppliers(): """Supplier and purchase order management.""" with get_db() as db: if request.method == 'POST': action = request.form.get('action') if action == 'add_supplier': cur = db.execute(""" INSERT INTO suppliers (supplier_name, contact_person, phone, email, address, notes) VALUES (?, ?, ?, ?, ?, ?) """, ( request.form.get('supplier_name','').strip(), request.form.get('contact_person','').strip(), request.form.get('phone','').strip(), request.form.get('email','').strip(), request.form.get('address','').strip(), request.form.get('notes','').strip() )) record_activity(db, 'add', 'suppliers', cur.lastrowid, 'Added supplier') flash('✅ Supplier added.') elif action == 'add_purchase': qty = to_int(request.form.get('qty')) product_cost = to_float(request.form.get('product_cost')) shipping = to_float(request.form.get('shipping_cost')) customs = to_float(request.form.get('customs_cost')) other = to_float(request.form.get('other_cost')) total_cost = product_cost + shipping + customs + other unit_cost = total_cost / qty if qty else 0 paid = to_float(request.form.get('amount_paid')) status = request.form.get('status') or ('Paid' if paid >= total_cost and total_cost > 0 else 'Pending') cur = db.execute(""" INSERT INTO purchase_orders (supplier_id, po_date, item_name, qty, product_cost, shipping_cost, customs_cost, other_cost, total_cost, unit_cost, amount_paid, status, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (to_int(request.form.get('supplier_id')) or None, request.form.get('po_date') or datetime.now().strftime('%Y-%m-%d'), request.form.get('item_name','').strip(), qty, product_cost, shipping, customs, other, total_cost, unit_cost, paid, status, request.form.get('notes','').strip())) record_activity(db, 'add', 'purchase_orders', cur.lastrowid, 'Added purchase order') flash('✅ Purchase order added.') return redirect(url_for('suppliers')) suppliers_rows = db.execute('SELECT * FROM suppliers ORDER BY id DESC').fetchall() purchases = db.execute(""" SELECT p.*, s.supplier_name FROM purchase_orders p LEFT JOIN suppliers s ON s.id=p.supplier_id ORDER BY p.id DESC LIMIT 200 """).fetchall() return render_template('suppliers.html', suppliers=suppliers_rows, purchases=purchases) @app.route('/tax-report') @login_required @admin_required def tax_report(): """Simple monthly income vs expenses export page.""" year = to_int(request.args.get('year'), datetime.now().year) with get_db() as db: rows = db.execute(""" SELECT substr(order_date,1,7) AS month, COALESCE(SUM(cash_received),0) AS income, COALESCE(SUM(cost),0) AS direct_cost FROM orders WHERE (deleted_at IS NULL OR deleted_at='') AND substr(order_date,1,4)=? GROUP BY substr(order_date,1,7) """, (str(year),)).fetchall() expenses = db.execute(""" SELECT substr(date,1,7) AS month, COALESCE(SUM(amount),0) AS expenses FROM expenses WHERE substr(date,1,4)=? GROUP BY substr(date,1,7) """, (str(year),)).fetchall() exp_map = {r['month']: r['expenses'] for r in expenses} report = [] for r in rows: exp = exp_map.get(r['month'], 0) report.append({'month': r['month'], 'income': r['income'], 'direct_cost': r['direct_cost'], 'expenses': exp, 'net_profit': r['income'] - r['direct_cost'] - exp}) return render_template('tax_report.html', report=report, year=year) @app.route('/invoice-options/') @login_required def invoice_options(order_id): """Choose invoice print size.""" with get_db() as db: order = db.execute('SELECT * FROM orders WHERE id=? AND (deleted_at IS NULL OR deleted_at="")', (order_id,)).fetchone() if not order: flash('❌ Order not found.') return redirect(url_for('index')) return render_template('invoice_options.html', order=order) # ============================================================================ # PHASE 3 ROUTES - BULK IMPORT, LOCATIONS, REMINDERS, TRASH, ERROR LOGS # ============================================================================ @app.errorhandler(Exception) def handle_unexpected_error(error): """Log unexpected errors without exposing details to users.""" try: with get_db() as db: db.execute(""" INSERT INTO system_errors (error_type, message, path, method, username) VALUES (?, ?, ?, ?, ?) """, (type(error).__name__, str(error)[:1000], request.path, request.method, session.get('username'))) except Exception: pass raise error @app.route('/bulk-import', methods=['GET', 'POST']) @login_required @admin_required def bulk_import(): imported = 0 if request.method == 'POST': file = request.files.get('file') import_type = request.form.get('import_type', 'stock') if not file or not file.filename: flash('❌ Please choose a CSV/XLSX file.') return redirect(url_for('bulk_import')) filename = file.filename.lower() rows = [] try: if filename.endswith('.csv'): import csv data = file.read().decode('utf-8-sig').splitlines() rows = list(csv.DictReader(data)) elif filename.endswith('.xlsx'): try: from openpyxl import load_workbook except ImportError: flash('❌ openpyxl not installed. Install: pip install openpyxl') return redirect(url_for('bulk_import')) wb = load_workbook(file, data_only=True) ws = wb.active headers = [str(c.value or '').strip() for c in ws[1]] for row in ws.iter_rows(min_row=2, values_only=True): rows.append({headers[i]: row[i] for i in range(len(headers))}) else: flash('❌ Only CSV or XLSX files are supported.') return redirect(url_for('bulk_import')) with get_db() as db: for row in rows: if import_type == 'imei': imei = str(row.get('imei') or row.get('IMEI') or '').strip() serial = str(row.get('serial') or row.get('Serial') or '').strip() item_name = str(row.get('item_name') or row.get('Item Name') or '').strip() if not imei and not serial: continue db.execute(""" INSERT OR IGNORE INTO imei_devices (imei, serial, item_name, status, trcsl_required, trcsl_status, import_batch, location) VALUES (?, ?, ?, 'in_stock', ?, ?, ?, ?) """, (imei, serial, item_name, to_int(row.get('trcsl_required') or 0), row.get('trcsl_status') or 'Not Required', row.get('import_batch') or '', row.get('location') or 'Shop')) imported += 1 else: item_name = str(row.get('item_name') or row.get('Item Name') or '').strip() if not item_name: continue stock_id = str(row.get('stock_id') or row.get('Stock ID') or item_name[:10]).strip() db.execute(""" INSERT INTO stock (stock_id, item_name, item_category, date, order_price, logistic_charges, shipping_charges, customs_charges, unit_cost, qty, ordered_qty, location, reorder_level) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (stock_id, item_name, row.get('item_category') or row.get('Category') or '', row.get('date') or datetime.now().strftime('%Y-%m-%d'), to_float(row.get('order_price') or 0), to_float(row.get('logistic_charges') or 0), to_float(row.get('shipping_charges') or 0), to_float(row.get('customs_charges') or 0), to_float(row.get('unit_cost') or 0), to_int(row.get('qty') or 0), to_int(row.get('ordered_qty') or 0), row.get('location') or 'Shop', to_int(row.get('reorder_level') or 5))) imported += 1 add_activity('Bulk import', import_type, None, f'Imported {imported} rows') flash(f'✅ Imported {imported} rows successfully.') except Exception as exc: flash(f'❌ Import failed: {exc}') return render_template('bulk_import.html') @app.route('/locations', methods=['GET', 'POST']) @login_required def stock_locations(): if request.method == 'POST': name = request.form.get('location_name', '').strip() description = request.form.get('description', '').strip() if name: with get_db() as db: db.execute('INSERT OR IGNORE INTO stock_locations (location_name, description) VALUES (?, ?)', (name, description)) add_activity('Add location', 'stock_locations', None, name) flash('✅ Location saved.') return redirect(url_for('stock_locations')) with get_db() as db: for location_name in ['Shop', 'Home', 'Warehouse']: db.execute('INSERT OR IGNORE INTO stock_locations (location_name) VALUES (?)', (location_name,)) locations = db.execute('SELECT * FROM stock_locations ORDER BY location_name').fetchall() summary = db.execute(''' SELECT COALESCE(location, 'Shop') AS location, COUNT(*) AS item_count, COALESCE(SUM(qty),0) AS total_qty FROM stock GROUP BY COALESCE(location, 'Shop') ORDER BY location ''').fetchall() return render_template('locations.html', locations=locations, summary=summary) @app.route('/reminders', methods=['GET', 'POST']) @login_required def reminders_page(): if request.method == 'POST': with get_db() as db: db.execute('''INSERT INTO reminders (reminder_type, title, due_date, notes, status) VALUES (?, ?, ?, ?, 'Pending')''', (request.form.get('reminder_type') or 'General', request.form.get('title') or 'Reminder', request.form.get('due_date') or '', request.form.get('notes') or '')) add_activity('Add reminder', 'reminders', None, request.form.get('title') or '') flash('✅ Reminder added.') return redirect(url_for('reminders_page')) today = datetime.now().strftime('%Y-%m-%d') with get_db() as db: upcoming = db.execute(''' SELECT imei, serial, item_name, product_warranty_expiry, battery_warranty_expiry FROM imei_devices WHERE (product_warranty_expiry BETWEEN ? AND date(?, '+30 day')) OR (battery_warranty_expiry BETWEEN ? AND date(?, '+30 day')) ORDER BY product_warranty_expiry, battery_warranty_expiry ''', (today, today, today, today)).fetchall() reminders = db.execute('SELECT * FROM reminders ORDER BY status, due_date').fetchall() return render_template('reminders.html', reminders=reminders, upcoming=upcoming) @app.route('/reminders//done', methods=['POST']) @login_required def reminder_done(reminder_id): with get_db() as db: db.execute("UPDATE reminders SET status='Done' WHERE id=?", (reminder_id,)) add_activity('Complete reminder', 'reminders', reminder_id, '') flash('✅ Reminder marked done.') return redirect(url_for('reminders_page')) @app.route('/trash') @login_required @admin_required def trash_page(): with get_db() as db: deleted_orders = db.execute("SELECT * FROM orders WHERE deleted_at IS NOT NULL AND deleted_at != '' ORDER BY deleted_at DESC").fetchall() deleted_records = db.execute('SELECT * FROM soft_deleted_records ORDER BY deleted_at DESC LIMIT 100').fetchall() return render_template('trash.html', deleted_orders=deleted_orders, deleted_records=deleted_records) @app.route('/orders//restore', methods=['POST']) @login_required @admin_required def restore_order(order_id): with get_db() as db: db.execute("UPDATE orders SET deleted_at=NULL, deleted_by=NULL WHERE id=?", (order_id,)) add_activity('Restore order', 'orders', order_id, '') flash('✅ Order restored.') return redirect(url_for('trash_page')) @app.route('/error-logs') @login_required @admin_required def error_logs(): with get_db() as db: errors = db.execute('SELECT * FROM system_errors ORDER BY created_at DESC LIMIT 200').fetchall() log_text = '' if ERROR_LOG_FILE.exists(): log_text = ERROR_LOG_FILE.read_text(errors='ignore')[-10000:] return render_template('error_logs.html', errors=errors, log_text=log_text) @app.route('/sales-insights') @login_required def sales_insights(): with get_db() as db: top_items = db.execute(''' SELECT item_name, SUM(qty) AS sold_qty, SUM(total_price) AS revenue, SUM(cost) AS cost_total, (SUM(total_price)-SUM(cost)) AS gross_profit FROM orders WHERE deleted_at IS NULL OR deleted_at='' GROUP BY item_name ORDER BY sold_qty DESC LIMIT 20 ''').fetchall() low_margin = db.execute(''' SELECT id, item_name, total_price, cost, (total_price-cost) AS profit FROM orders WHERE (deleted_at IS NULL OR deleted_at='') AND total_price > 0 AND (total_price-cost) < (total_price*0.10) ORDER BY profit ASC LIMIT 20 ''').fetchall() return render_template('sales_insights.html', top_items=top_items, low_margin=low_margin) # ============================================================================ # CUSTOMER & SUPPLIER MANAGEMENT UPGRADE # ============================================================================ def ensure_customer_supplier_schema(): """Add customer and advanced supplier tables safely for existing databases.""" with get_db() as db: db.executescript(""" CREATE TABLE IF NOT EXISTS customers ( id INTEGER PRIMARY KEY AUTOINCREMENT, customer_name TEXT NOT NULL, phone1 TEXT, phone2 TEXT, email TEXT, address TEXT, city TEXT, customer_type TEXT CHECK(customer_type IN ('Retail','Wholesale','Service','Other')) DEFAULT 'Retail', notes TEXT, is_active INTEGER DEFAULT 1, created_at TEXT DEFAULT CURRENT_TIMESTAMP, updated_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS customer_transactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, customer_id INTEGER NOT NULL, trans_date TEXT NOT NULL, trans_type TEXT CHECK(trans_type IN ('Payment','Refund','Credit Note','Adjustment')) NOT NULL, amount REAL DEFAULT 0, order_id INTEGER, reason TEXT, notes TEXT, created_by TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS supplier_payments ( id INTEGER PRIMARY KEY AUTOINCREMENT, supplier_id INTEGER NOT NULL, payment_date TEXT NOT NULL, amount REAL DEFAULT 0, payment_method TEXT DEFAULT 'Cash', reference_no TEXT, notes TEXT, created_by TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); """) for sql in [ "ALTER TABLE orders ADD COLUMN customer_id INTEGER", "ALTER TABLE purchase_orders ADD COLUMN invoice_no TEXT", "ALTER TABLE purchase_orders ADD COLUMN due_date TEXT", "ALTER TABLE suppliers ADD COLUMN is_active INTEGER DEFAULT 1", "ALTER TABLE suppliers ADD COLUMN updated_at TEXT" ]: try: db.execute(sql) except sqlite3.OperationalError: pass def customer_order_where(customer): """Build SQL condition for a customer's directly linked or phone-matched orders.""" phones = [customer['phone1'], customer['phone2']] phones = [p for p in phones if p] where = ["customer_id=?"] params = [customer['id']] for phone in phones: where.append("contact1=? OR contact2=?") params.extend([phone, phone]) return "(" + " OR ".join(where) + ")", params @app.route('/customers', methods=['GET', 'POST']) @login_required def manage_customers(): """Customer management with quick add, search, balance and order matching.""" with get_db() as db: if request.method == 'POST': action = request.form.get('action') if action == 'add_customer': name = request.form.get('customer_name','').strip() phone1 = request.form.get('phone1','').strip() if not name: flash('❌ Customer name is required.') return redirect(url_for('manage_customers')) cur = db.execute(""" INSERT INTO customers (customer_name, phone1, phone2, email, address, city, customer_type, notes, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, (name, phone1, request.form.get('phone2','').strip(), request.form.get('email','').strip(), request.form.get('address','').strip(), request.form.get('city','').strip(), request.form.get('customer_type') or 'Retail', request.form.get('notes','').strip(), datetime.now().strftime('%Y-%m-%d %H:%M:%S'))) record_activity(db, 'add', 'customers', cur.lastrowid, f'Added customer {name}') flash('✅ Customer added.') return redirect(url_for('customer_detail', customer_id=cur.lastrowid)) if action == 'add_transaction': customer_id = to_int(request.form.get('customer_id')) amount = to_float(request.form.get('amount')) cur = db.execute(""" INSERT INTO customer_transactions (customer_id, trans_date, trans_type, amount, order_id, reason, notes, created_by) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, (customer_id, request.form.get('trans_date') or datetime.now().strftime('%Y-%m-%d'), request.form.get('trans_type') or 'Payment', amount, to_int(request.form.get('order_id')) or None, request.form.get('reason','').strip(), request.form.get('notes','').strip(), current_user_name())) record_activity(db, 'add', 'customer_transactions', cur.lastrowid, 'Added customer transaction') flash('✅ Customer transaction saved.') return redirect(url_for('customer_detail', customer_id=customer_id)) q = request.args.get('q','').strip() status = request.args.get('status','Active') sql = "SELECT * FROM customers WHERE 1=1" params = [] if status == 'Active': sql += " AND COALESCE(is_active,1)=1" elif status == 'Inactive': sql += " AND COALESCE(is_active,1)=0" if q: like = f'%{q}%' sql += " AND (customer_name LIKE ? OR phone1 LIKE ? OR phone2 LIKE ? OR email LIKE ? OR city LIKE ?)" params.extend([like, like, like, like, like]) customers = db.execute(sql + " ORDER BY id DESC LIMIT 300", params).fetchall() summaries = {} for c in customers: where, where_params = customer_order_where(c) order_sum = db.execute(f""" SELECT COUNT(*) AS order_count, COALESCE(SUM(total_price),0) AS total_sales, COALESCE(SUM(cash_received),0) AS received, COALESCE(SUM(total_price - cash_received),0) AS outstanding FROM orders WHERE (deleted_at IS NULL OR deleted_at='') AND {where} """, where_params).fetchone() trans_sum = db.execute(""" SELECT COALESCE(SUM(CASE WHEN trans_type IN ('Payment','Adjustment') THEN amount ELSE -amount END),0) AS net_transactions FROM customer_transactions WHERE customer_id=? """, (c['id'],)).fetchone() summaries[c['id']] = {'orders': order_sum, 'transactions': trans_sum} return render_template('customers.html', customers=customers, summaries=summaries, q=q, status=status) @app.route('/customers/', methods=['GET', 'POST']) @login_required def customer_detail(customer_id): """Detailed customer profile with orders, warranty devices, transactions and edit form.""" with get_db() as db: customer = db.execute('SELECT * FROM customers WHERE id=?', (customer_id,)).fetchone() if not customer: flash('❌ Customer not found.') return redirect(url_for('manage_customers')) if request.method == 'POST': action = request.form.get('action') if action == 'update_customer': db.execute(""" UPDATE customers SET customer_name=?, phone1=?, phone2=?, email=?, address=?, city=?, customer_type=?, notes=?, is_active=?, updated_at=? WHERE id=? """, (request.form.get('customer_name','').strip(), request.form.get('phone1','').strip(), request.form.get('phone2','').strip(), request.form.get('email','').strip(), request.form.get('address','').strip(), request.form.get('city','').strip(), request.form.get('customer_type') or 'Retail', request.form.get('notes','').strip(), 1 if request.form.get('is_active') == '1' else 0, datetime.now().strftime('%Y-%m-%d %H:%M:%S'), customer_id)) record_activity(db, 'edit', 'customers', customer_id, 'Updated customer profile') flash('✅ Customer updated.') return redirect(url_for('customer_detail', customer_id=customer_id)) if action == 'add_transaction': cur = db.execute(""" INSERT INTO customer_transactions (customer_id, trans_date, trans_type, amount, order_id, reason, notes, created_by) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, (customer_id, request.form.get('trans_date') or datetime.now().strftime('%Y-%m-%d'), request.form.get('trans_type') or 'Payment', to_float(request.form.get('amount')), to_int(request.form.get('order_id')) or None, request.form.get('reason','').strip(), request.form.get('notes','').strip(), current_user_name())) record_activity(db, 'add', 'customer_transactions', cur.lastrowid, 'Added customer transaction') flash('✅ Transaction added.') return redirect(url_for('customer_detail', customer_id=customer_id)) if action == 'link_order': order_id = to_int(request.form.get('order_id')) order = db.execute('SELECT id FROM orders WHERE id=?', (order_id,)).fetchone() if order: db.execute('UPDATE orders SET customer_id=? WHERE id=?', (customer_id, order_id)) record_activity(db, 'link', 'orders', order_id, f'Linked order to customer #{customer_id}') flash('✅ Order linked to customer.') else: flash('❌ Order not found.') return redirect(url_for('customer_detail', customer_id=customer_id)) where, params = customer_order_where(customer) orders = db.execute(f""" SELECT * FROM orders WHERE (deleted_at IS NULL OR deleted_at='') AND {where} ORDER BY id DESC LIMIT 200 """, params).fetchall() transactions = db.execute('SELECT * FROM customer_transactions WHERE customer_id=? ORDER BY id DESC LIMIT 200', (customer_id,)).fetchall() devices = db.execute(""" SELECT d.* FROM imei_devices d LEFT JOIN orders o ON o.id=d.order_id WHERE o.customer_id=? OR o.contact1 IN (?, ?) OR o.contact2 IN (?, ?) ORDER BY d.id DESC LIMIT 200 """, (customer_id, customer['phone1'] or '', customer['phone2'] or '', customer['phone1'] or '', customer['phone2'] or '')).fetchall() totals = db.execute(f""" SELECT COUNT(*) AS order_count, COALESCE(SUM(total_price),0) AS total_sales, COALESCE(SUM(cash_received),0) AS received, COALESCE(SUM(total_price-cash_received),0) AS outstanding FROM orders WHERE (deleted_at IS NULL OR deleted_at='') AND {where} """, params).fetchone() trans_total = db.execute(""" SELECT COALESCE(SUM(CASE WHEN trans_type IN ('Payment','Adjustment') THEN amount ELSE -amount END),0) AS total FROM customer_transactions WHERE customer_id=? """, (customer_id,)).fetchone() return render_template('customer_detail.html', customer=customer, orders=orders, transactions=transactions, devices=devices, totals=totals, trans_total=trans_total) @app.route('/customers//delete', methods=['POST']) @login_required @admin_required def delete_customer(customer_id): """Soft deactivate a customer.""" with get_db() as db: db.execute("UPDATE customers SET is_active=0, updated_at=? WHERE id=?", (datetime.now().strftime('%Y-%m-%d %H:%M:%S'), customer_id)) record_activity(db, 'deactivate', 'customers', customer_id, 'Customer deactivated') flash('✅ Customer deactivated.') return redirect(url_for('manage_customers')) @app.route('/suppliers/', methods=['GET', 'POST']) @login_required @admin_required def supplier_detail(supplier_id): """Detailed supplier profile with purchases, payments and outstanding balance.""" with get_db() as db: supplier = db.execute('SELECT * FROM suppliers WHERE id=?', (supplier_id,)).fetchone() if not supplier: flash('❌ Supplier not found.') return redirect(url_for('suppliers')) if request.method == 'POST': action = request.form.get('action') if action == 'update_supplier': db.execute(""" UPDATE suppliers SET supplier_name=?, contact_person=?, phone=?, email=?, address=?, notes=?, is_active=?, updated_at=? WHERE id=? """, (request.form.get('supplier_name','').strip(), request.form.get('contact_person','').strip(), request.form.get('phone','').strip(), request.form.get('email','').strip(), request.form.get('address','').strip(), request.form.get('notes','').strip(), 1 if request.form.get('is_active') == '1' else 0, datetime.now().strftime('%Y-%m-%d %H:%M:%S'), supplier_id)) record_activity(db, 'edit', 'suppliers', supplier_id, 'Updated supplier') flash('✅ Supplier updated.') return redirect(url_for('supplier_detail', supplier_id=supplier_id)) if action == 'add_payment': cur = db.execute(""" INSERT INTO supplier_payments (supplier_id, payment_date, amount, payment_method, reference_no, notes, created_by) VALUES (?, ?, ?, ?, ?, ?, ?) """, (supplier_id, request.form.get('payment_date') or datetime.now().strftime('%Y-%m-%d'), to_float(request.form.get('amount')), request.form.get('payment_method') or 'Cash', request.form.get('reference_no','').strip(), request.form.get('notes','').strip(), current_user_name())) record_activity(db, 'add', 'supplier_payments', cur.lastrowid, 'Added supplier payment') flash('✅ Supplier payment saved.') return redirect(url_for('supplier_detail', supplier_id=supplier_id)) purchases = db.execute('SELECT * FROM purchase_orders WHERE supplier_id=? ORDER BY id DESC LIMIT 300', (supplier_id,)).fetchall() payments = db.execute('SELECT * FROM supplier_payments WHERE supplier_id=? ORDER BY id DESC LIMIT 200', (supplier_id,)).fetchall() totals = db.execute(""" SELECT COALESCE(SUM(total_cost),0) AS purchase_total, COALESCE(SUM(amount_paid),0) AS po_paid, COALESCE(SUM(total_cost-amount_paid),0) AS po_outstanding FROM purchase_orders WHERE supplier_id=? """, (supplier_id,)).fetchone() pay_total = db.execute('SELECT COALESCE(SUM(amount),0) AS total FROM supplier_payments WHERE supplier_id=?', (supplier_id,)).fetchone() return render_template('supplier_detail.html', supplier=supplier, purchases=purchases, payments=payments, totals=totals, pay_total=pay_total) @app.route('/suppliers//delete', methods=['POST']) @login_required @admin_required def delete_supplier(supplier_id): with get_db() as db: db.execute("UPDATE suppliers SET is_active=0, updated_at=? WHERE id=?", (datetime.now().strftime('%Y-%m-%d %H:%M:%S'), supplier_id)) record_activity(db, 'deactivate', 'suppliers', supplier_id, 'Supplier deactivated') flash('✅ Supplier deactivated.') return redirect(url_for('suppliers')) @app.route('/manifest.json') def manifest(): return jsonify({'name':'STRATA Business System','short_name':'STRATA','start_url':'/','display':'standalone','background_color':'#ffffff','theme_color':'#667eea','icons':[]}) @app.route('/service-worker.js') def service_worker(): js = "self.addEventListener('install', event => { self.skipWaiting(); });\nself.addEventListener('fetch', event => { event.respondWith(fetch(event.request).catch(() => new Response('Offline. Please reconnect.'))); });" return app.response_class(js, mimetype='application/javascript') # APPLICATION INITIALIZATION # ============================================================================ # Initialize database on startup init_db() ensure_customer_supplier_schema() try: make_daily_backup() keep_recent_backups(30) except Exception: pass if __name__ == "__main__": app.run(debug=False, threaded=True)