"""

Stock Management System with MySQL Connectivity

Expanded version ~500 lines code
Features:
- MySQL DB connection & schema creation
- User authentication with hashed passwords
- Role-based access control (admin, staff)
- Stock item CRUD, purchase & sales management
- Multiple reports (low stock, monthly sales, revenue)
- Pagination support for large datasets
- Input validation and error handling
- Activity logging to file

Prerequisites:
- MySQL server operational
- Python package: mysql-connector-python
pip install mysql-connector-python
"""

import [Link]
from [Link] import Error
import datetime
import sys
import getpass
import hashlib
import os
import logging

# ============================= LOGGING SETUP ============================= #

LOG_FILENAME = "stock_management.log"
[Link](
filename=LOG_FILENAME,
level=[Link],
format='%(asctime)s [%(levelname)s]: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)

def log_info(message: str):

print(message)
[Link](message)

def log_error(message: str):

print(f"[ERROR] {message}")
[Link](message)

# ============================= DATABASE CONNECTION ============================= #

class Database:
def __init__(self, host="localhost", user="root", password="",
database="stocks_db"):
[Link] = host
[Link] = user
[Link] = password
[Link] = database
[Link] = None
[Link] = None
 def connect(self):
try:
[Link] = [Link](
host=[Link],
user=[Link],
password=[Link]
)
[Link] = [Link](buffered=True)
[Link](f"CREATE DATABASE IF NOT EXISTS {[Link]}")
[Link] = [Link]
log_info("[DB] Connected to MySQL successfully.")
except Error as e:
log_error(f"Database connection failed: {e}")
[Link](1)

def execute(self, query, params=None, commit=False):

try:
[Link](query, params or ())
if commit:
[Link]()
except Error as e:
log_error(f"SQL error: {e}")
raise

def fetchall(self):
return [Link]()

def fetchone(self):
return [Link]()

def close(self):
if [Link]:
[Link]()
[Link]()
log_info("[DB] Connection closed.")

# ============================= SCHEMA CREATION ============================= #

def create_tables(db: Database):
# Items Table - Stores the inventory items
[Link]("""
CREATE TABLE IF NOT EXISTS items (
item_id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
quantity INT NOT NULL DEFAULT 0,
price DECIMAL(10,2) NOT NULL DEFAULT 0.00,
description TEXT DEFAULT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP
)
""", commit=True)

# Transactions Table - Records purchase and sales transactions

[Link]("""
CREATE TABLE IF NOT EXISTS transactions (
trans_id INT AUTO_INCREMENT PRIMARY KEY,
item_id INT NOT NULL,
 quantity INT NOT NULL,
trans_type ENUM('PURCHASE','SALE') NOT NULL,
total_price DECIMAL(12,2) NOT NULL,
trans_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (item_id) REFERENCES items(item_id) ON DELETE CASCADE
)
""", commit=True)

# Users Table - Stores users with roles for authentication

[Link]("""
CREATE TABLE IF NOT EXISTS users (
user_id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(50) UNIQUE NOT NULL,
password_hash CHAR(64) NOT NULL,
role ENUM('admin','staff') DEFAULT 'staff',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""", commit=True)

log_info("[DB] Tables created or verified.")

# ============================= PASSWORD UTILITIES ============================= #

def hash_password(password: str) -> str:
"""Create SHA256 hash of a password"""
return hashlib.sha256([Link]('utf-8')).hexdigest()

def verify_password(password: str, hashed: str) -> bool:

return hash_password(password) == hashed

# ============================= USER AUTHENTICATION & MANAGEMENT

============================= #
class UserManager:
def __init__(self, db: Database):
[Link] = db
self.current_user = None

def create_user(self, username, password, role='staff'):

password_hash = hash_password(password)
try:
[Link](
"INSERT INTO users (username, password_hash, role) VALUES (%s, %s,
%s)",
(username, password_hash, role), commit=True
)
log_info(f"[USER] User '{username}' created with role '{role}'.")
except Error as e:
log_error(f"Failed to create user '{username}': {e}")

def login(self):
username = input("Username: ").strip()
password = [Link]("Password: ").strip()

[Link](
"SELECT user_id, password_hash, role FROM users WHERE username=%s",
(username,)
)
 row = [Link]()

if row and verify_password(password, row[1]):

self.current_user = {
"user_id": row[0],
"username": username,
"role": row[2]
}
log_info(f"[LOGIN] User '{username}' logged in successfully as
{row[2]}.")
return True
else:
log_error("Login failed: Invalid username or password.")
return False

def require_role(self, roles):

if self.current_user and self.current_user['role'] in roles:
return True
else:
log_error(f"Permission denied. Required roles: {roles}")
return False

def is_admin(self):
return self.current_user and self.current_user['role'] == 'admin'

# ============================= VALIDATION UTILS ============================= #

def input_int(prompt, minimum=None, maximum=None):
while True:
try:
val = int(input(prompt))
if (minimum is not None and val < minimum) or (maximum is not None and
val > maximum):
log_error(f"Input must be between {minimum} and {maximum}.")
continue
return val
except ValueError:
log_error("Invalid integer. Try again.")

def input_float(prompt, minimum=None, maximum=None):

while True:
try:
val = float(input(prompt))
if (minimum is not None and val < minimum) or (maximum is not None and
val > maximum):
log_error(f"Input must be between {minimum} and {maximum}.")
continue
return val
except ValueError:
log_error("Invalid number. Try again.")

def input_string(prompt, allow_empty=False):

while True:
val = input(prompt).strip()
if not val and not allow_empty:
log_error("Input cannot be empty.")
continue
 return val

# ============================= PAGINATION UTILS ============================= #

def paginate(data_list, page_size=10):
total_items = len(data_list)
total_pages = (total_items + page_size - 1) // page_size
current_page = 1

while True:
start = (current_page - 1) * page_size
end = start + page_size
yield data_list[start:end], current_page, total_pages
if current_page >= total_pages:
break
current_page += 1

# ============================= STOCK MANAGEMENT ============================= #

class StockManager:
def __init__(self, db: Database, user_manager: UserManager):
[Link] = db
self.user_manager = user_manager

def add_item(self):
if not self.user_manager.require_role(['admin', 'staff']):
return
name = input_string("Item name: ")
quantity = input_int("Quantity (>=0): ", 0)
price = input_float("Price per unit (>=0): ", 0.0)
description = input_string("Description (optional): ", allow_empty=True)
try:
[Link](
"INSERT INTO items (name, quantity, price, description) VALUES (%s,
%s, %s, %s)",
(name, quantity, price, description), commit=True
)
log_info(f"Item '{name}' added with quantity {quantity} at price
{price}.")
except Error as e:
log_error(f"Failed to add item: {e}")

def update_item(self):
if not self.user_manager.require_role(['admin', 'staff']):
return
item_id = input_int("Enter item ID to update: ", 1)
# Check if exists
[Link]("SELECT * FROM items WHERE item_id=%s", (item_id,))
item = [Link]()
if not item:
log_error("Item not found.")
return
print(f"Current item details: Name='{item[1]}', Qty={item[2]},
Price={item[3]}, Desc={item[4]}")
name = input_string("New name (blank to skip): ", allow_empty=True)
quantity_inp = input("New quantity (blank to skip): ").strip()
price_inp = input("New price (blank to skip): ").strip()
description = input("New description (blank to skip): ").strip()
 updates = []
params = []

if name != "":
[Link]("name=%s")
[Link](name)
if quantity_inp != "":
try:
quantity = int(quantity_inp)
if quantity < 0:
log_error("Quantity must be >= 0.")
return
[Link]("quantity=%s")
[Link](quantity)
except ValueError:
log_error("Invalid quantity input.")
return
if price_inp != "":
try:
price = float(price_inp)
if price < 0:
log_error("Price must be >= 0.")
return
[Link]("price=%s")
[Link](price)
except ValueError:
log_error("Invalid price input.")
return
if description != "":
[Link]("description=%s")
[Link](description)

if not updates:
log_info("No changes provided; item not updated.")
return

query = "UPDATE items SET " + ", ".join(updates) + " WHERE item_id=%s"
[Link](item_id)
try:
[Link](query, tuple(params), commit=True)
log_info("Item updated successfully.")
except Error as e:
log_error(f"Failed to update item: {e}")

def delete_item(self):
if not self.user_manager.require_role(['admin']):
log_error("Only admin can delete items.")
return
item_id = input_int("Enter item ID to delete: ", 1)
[Link]("SELECT * FROM items WHERE item_id=%s", (item_id,))
if not [Link]():
log_error("Item not found.")
return
confirm = input_string(f"Confirm delete item {item_id}? Type 'YES' to
confirm: ")
if confirm == "YES":
try:
[Link]("DELETE FROM items WHERE item_id=%s", (item_id,),
commit=True)
 log_info(f"Item {item_id} deleted.")
except Error as e:
log_error(f"Deletion failed: {e}")
else:
log_info("Deletion aborted.")

def view_items(self):
[Link]("SELECT item_id, name, quantity, price, description FROM
items ORDER BY item_id")
items = [Link]()
if not items:
log_info("No items in inventory.")
return

page_gen = paginate(items, page_size=10)

for page_data, current_page, total_pages in page_gen:

print("\n==== Stock Inventory (Page {}/{} ) ====".format(current_page,
total_pages))
print("{:<6} {:<25} {:<10} {:<10} {:<30}".format(
"ID", "Name", "Quantity", "Price", "Description"))
for i in page_data:
print("{:<6} {:<25} {:<10} {:<10.2f} {:<30}".format(i[0], i[1],
i[2], float(i[3]), (i[4] or '')[:30]))
if current_page != total_pages:
input("Press ENTER for next page...")
print("="*40)

def search_item(self):
keyword = input_string("Enter item name keyword to search: ")
[Link](
"SELECT item_id, name, quantity, price, description FROM items WHERE
name LIKE %s ORDER BY name",
("%" + keyword + "%",)
)
results = [Link]()
if not results:
log_info("No items found matching the keyword.")
return

print("\nSearch Results:")
print("{:<6} {:<25} {:<10} {:<10} {:<30}".format(
"ID", "Name", "Quantity", "Price", "Description"))
for r in results:
print("{:<6} {:<25} {:<10} {:<10.2f} {:<30}".format(r[0], r[1], r[2],
float(r[3]), (r[4] or '')[:30]))

def manage_transaction(self, trans_type):

if trans_type not in ['PURCHASE', 'SALE']:
log_error("Invalid transaction type.")
return
if not self.user_manager.require_role(['admin', 'staff']):
return
item_id = input_int(f"Enter item ID for {trans_type.lower()}: ", 1)
[Link]("SELECT quantity, price, name FROM items WHERE item_id =
%s", (item_id,))
item = [Link]()
if not item:
log_error("Item not found.")
 return
qty = input_int(f"Enter quantity to {trans_type.lower()}: ", 1)
current_qty, price_per_unit, item_name = item
if trans_type == 'SALE' and qty > current_qty:
log_error(f"Insufficient stock. Current quantity: {current_qty}")
return
total_price = round(qty * price_per_unit, 2)

try:
if trans_type == 'PURCHASE':
[Link](
"UPDATE items SET quantity = quantity + %s WHERE item_id = %s",
(qty, item_id), commit=True
)
else: # SALE
[Link](
"UPDATE items SET quantity = quantity - %s WHERE item_id = %s",
(qty, item_id), commit=True
)

[Link](
"INSERT INTO transactions (item_id, quantity, trans_type,
total_price) VALUES (%s, %s, %s, %s)",
(item_id, qty, trans_type, total_price), commit=True
)
log_info(f"{trans_type.title()} recorded: {qty} units of '{item_name}'
for total price ${total_price:.2f}.")
except Error as e:
log_error(f"Transaction failed: {e}")

def view_transactions(self):
page = 1
per_page = 10
[Link]("""
SELECT t.trans_id, [Link], [Link], t.trans_type, t.total_price,
t.trans_date
FROM transactions t
JOIN items i ON t.item_id = i.item_id
ORDER BY t.trans_date DESC
""")
transactions = [Link]()
if not transactions:
log_info("No transactions recorded.")
return

total_pages = (len(transactions) + per_page - 1) // per_page

while True:
start_idx = (page - 1) * per_page
end_idx = start_idx + per_page
current_page_data = transactions[start_idx:end_idx]

print(f"\n--- Transactions List (Page {page}/{total_pages}) ---")

print("{:<6} {:<25} {:<8} {:<9} {:<12} {:<20}".format(
"ID", "Item", "Qty", "Type", "Total Price", "Date"))
for tr in current_page_data:
print("{:<6} {:<25} {:<8} {:<9} ${:<11.2f} {:<20}".format(
tr[0], tr[1], tr[2], tr[3], float(tr[4]), tr[5].strftime("%Y-
%m-%d %H:%M:%S")))
if page >= total_pages:
 break
next_pg = input("Press ENTER for next page, or Q to quit:
").strip().lower()
if next_pg == 'q':
break
page += 1

def low_stock_report(self):
threshold = input_int("Enter low stock threshold (default 5): ", 0)
[Link]("SELECT item_id, name, quantity FROM items WHERE quantity
<= %s ORDER BY quantity ASC", (threshold,))
low_stock_items = [Link]()
if not low_stock_items:
log_info("No items below threshold.")
return
print("\n===== LOW STOCK REPORT =====")
for item in low_stock_items:
print(f"ID: {item[0]}, Name: {item[1]}, Qty: {item[2]}")
print("============================")

def monthly_sales_report(self):
year = input_int("Enter year (YYYY): ", 2000, [Link]().year)
month = input_int("Enter month (1-12): ", 1, 12)
start_date = [Link](year, month, 1)
if month == 12:
end_date = [Link](year + 1, 1, 1)
else:
end_date = [Link](year, month + 1, 1)
[Link]("""
SELECT [Link], SUM([Link]) as total_qty, SUM(t.total_price) as
total_revenue
FROM transactions t
JOIN items i ON t.item_id = i.item_id
WHERE t.trans_type = 'SALE' AND t.trans_date >= %s AND t.trans_date <
%s
GROUP BY [Link]
ORDER BY total_revenue DESC
""", (start_date, end_date))
results = [Link]()
if not results:
log_info("No sales in this period.")
return
print(f"\n--- Monthly Sales Report for {year}-{month:02d} ---")
print("{:<30} {:<15} {:<15}".format("Item Name", "Quantity Sold",
"Revenue"))
for r in results:
print("{:<30} {:<15} ${:<14.2f}".format(r[0], r[1], float(r[2])))
print("-"*50)

def revenue_report(self):
[Link]("""
SELECT DATE(trans_date) as date, SUM(total_price) as revenue
FROM transactions
WHERE trans_type = 'SALE'
GROUP BY DATE(trans_date)
ORDER BY date DESC
LIMIT 30
""")
data = [Link]()
 if not data:
log_info("No sales revenue data available.")
return
print("\nLast 30 Days Revenue Report:")
print("{:<12} {:<12}".format("Date", "Revenue"))
for d in data:
print(f"{d[0]} ${float(d[1]):.2f}")

# ============================= CLI SYSTEM ============================= #

def print_welcome():
print("\n========== STOCK MANAGEMENT SYSTEM ==========")
print("Manage stock, transactions, users and reports")
print("=============================================\n")

def admin_menu(manager: StockManager, user_manager: UserManager):

while True:
print("""
----- ADMIN MENU -----
1. Add New Item
2. Update Item
3. Delete Item
4. View All Items
5. Search Item
6. Purchase Stock
7. Sell Item
8. View Transactions
9. Low Stock Report
10. Monthly Sales Report
11. Revenue Report
12. Create New User
0. Logout
----------------------
""")
choice = input("Enter choice > ").strip()
if choice == "1":
manager.add_item()
elif choice == "2":
manager.update_item()
elif choice == "3":
manager.delete_item()
elif choice == "4":
manager.view_items()
elif choice == "5":
manager.search_item()
elif choice == "6":
manager.manage_transaction("PURCHASE")
elif choice == "7":
manager.manage_transaction("SALE")
elif choice == "8":
manager.view_transactions()
elif choice == "9":
manager.low_stock_report()
elif choice == "10":
manager.monthly_sales_report()
elif choice == "11":
manager.revenue_report()
elif choice == "12":
 if user_manager.is_admin():
username = input_string("New username: ")
password = [Link]("New password: ")
role = input_string("Role (admin/staff, default staff): ",
allow_empty=True) or "staff"
if role not in ['admin', 'staff']:
log_error("Invalid role given.")
else:
user_manager.create_user(username, password, role)
else:
log_error("Only admin can create users.")
elif choice == "0":
log_info("Logging out...")
break
else:
log_error("Invalid choice. Try again.")

def staff_menu(manager: StockManager):

while True:
print("""
----- STAFF MENU -----
1. View All Items
2. Search Item
3. Purchase Stock
4. Sell Item
5. View Transactions
6. Low Stock Report
0. Logout
----------------------
""")
choice = input("Enter choice > ").strip()
if choice == "1":
manager.view_items()
elif choice == "2":
manager.search_item()
elif choice == "3":
manager.manage_transaction("PURCHASE")
elif choice == "4":
manager.manage_transaction("SALE")
elif choice == "5":
manager.view_transactions()
elif choice == "6":
manager.low_stock_report()
elif choice == "0":
log_info("Logging out...")
break
else:
log_error("Invalid choice. Try again.")

# ============================= INITIAL ADMIN USER CREATION

============================= #
def init_admin_user(user_manager: UserManager):
user_manager.[Link]("SELECT COUNT(*) FROM users")
count = user_manager.[Link]()[0]
if count == 0:
print("No users found. Let's create an initial admin user.")
while True:
 username = input_string("Admin username: ")
password = [Link]("Admin password: ")
password2 = [Link]("Confirm password: ")
if password != password2:
log_error("Passwords do not match. Try again.")
continue
user_manager.create_user(username, password, role='admin')
log_info("Initial admin user created.")
break

# ============================= MAIN ENTRY POINT ============================= #

def main():
print_welcome()

user = input("MySQL user [root]: ") or "root"

pwd = [Link]("MySQL password (leave blank if none): ") or ""

db = Database(user=user, password=pwd)
[Link]()
create_tables(db)

user_manager = UserManager(db)
init_admin_user(user_manager)

for _ in range(3):
if user_manager.login():
break
else:
log_error("Login failed. Try again.")
else:
log_error("Max login attempts exceeded. Exiting.")
[Link]()
[Link](1)

stock_manager = StockManager(db, user_manager)

if user_manager.is_admin():
admin_menu(stock_manager, user_manager)
else:
staff_menu(stock_manager)

[Link]()
log_info("Goodbye!")

if __name__ == "__main__":
main()