1973 lines
70 KiB
Python
1973 lines
70 KiB
Python
"""
|
||
models.py — Data-access layer for all entities.
|
||
Ported 1-for-1 from the desktop version; all function signatures preserved.
|
||
"""
|
||
|
||
import datetime
|
||
import hashlib
|
||
import logging
|
||
import bcrypt
|
||
from config import get_connection
|
||
from utils.crypto import encrypt as _enc, decrypt as _dec
|
||
|
||
logger = logging.getLogger("models")
|
||
|
||
# ─── Password Helpers ─────────────────────────────────────────────────────────
|
||
|
||
def _hash_password(password: str) -> str:
|
||
return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt(rounds=12)).decode("utf-8")
|
||
|
||
|
||
def _verify_password(password: str, stored: str) -> bool:
|
||
if len(stored) == 64 and stored.isalnum():
|
||
return hashlib.sha256(password.encode("utf-8")).hexdigest() == stored
|
||
try:
|
||
return bcrypt.checkpw(password.encode("utf-8"), stored.encode("utf-8"))
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def _needs_rehash(stored: str) -> bool:
|
||
return len(stored) == 64 and stored.isalnum()
|
||
|
||
|
||
# ─── Activity Log ─────────────────────────────────────────────────────────────
|
||
|
||
def log_action(user_id, action, entity=None, entity_id=None, detail=None):
|
||
"""Insert a record into activity_log."""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"INSERT INTO activity_log (user_id, action, entity, entity_id, detail) "
|
||
"VALUES (%s,%s,%s,%s,%s)",
|
||
(user_id, action, entity, entity_id, detail),
|
||
)
|
||
conn.commit()
|
||
cur.close()
|
||
logger.info(f"[LOG] user_id={user_id} action={action} entity={entity} id={entity_id} - {detail}")
|
||
except Exception as e:
|
||
logger.error(f"Failed to write activity log: {e}")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Authentication ────────────────────────────────────────────────────────────
|
||
|
||
def authenticate(username: str, password: str, ip_address: str = None):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute("SELECT * FROM users WHERE username=%s AND is_active=1", (username,))
|
||
user = cur.fetchone()
|
||
if not user or not _verify_password(password, user["password"]):
|
||
logger.warning(f"Failed login attempt for username='{username}'.")
|
||
cur.close()
|
||
record_failed_attempt(username, ip_address)
|
||
return None
|
||
if _needs_rehash(user["password"]):
|
||
new_hash = _hash_password(password)
|
||
cur.execute("UPDATE users SET password=%s WHERE id=%s", (new_hash, user["id"]))
|
||
conn.commit()
|
||
logger.info(f"Password rehashed to bcrypt for user '{username}'.")
|
||
cur.close()
|
||
clear_failed_attempts(user["id"])
|
||
log_action(user["id"], "LOGIN", "users", user["id"], f"User '{username}' logged in.")
|
||
return user
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Login Rate Limiting ───────────────────────────────────────────────────────
|
||
|
||
MAX_FAILED_ATTEMPTS = 5
|
||
LOCKOUT_MINUTES = 15
|
||
|
||
|
||
def check_login_allowed(username: str) -> tuple:
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute("SELECT failed_attempts, locked_until FROM users WHERE username=%s", (username,))
|
||
row = cur.fetchone()
|
||
cur.close()
|
||
if not row:
|
||
return True, 0
|
||
locked_until = row.get("locked_until")
|
||
if locked_until:
|
||
now = datetime.datetime.now()
|
||
if now < locked_until:
|
||
remaining = int((locked_until - now).total_seconds())
|
||
return False, remaining
|
||
return True, 0
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def record_failed_attempt(username: str, ip_address: str = None):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute("SELECT id, failed_attempts FROM users WHERE username=%s", (username,))
|
||
row = cur.fetchone()
|
||
cur.execute(
|
||
"INSERT INTO login_attempts (username, ip_address) VALUES (%s,%s)",
|
||
(username, ip_address),
|
||
)
|
||
if row:
|
||
new_count = (row["failed_attempts"] or 0) + 1
|
||
locked_until = None
|
||
if new_count >= MAX_FAILED_ATTEMPTS:
|
||
locked_until = datetime.datetime.now() + datetime.timedelta(minutes=LOCKOUT_MINUTES)
|
||
logger.warning(f"Account '{username}' locked until {locked_until}.")
|
||
log_action(row["id"], "ACCOUNT_LOCKED", "users", row["id"],
|
||
f"Account locked after {new_count} failed login attempts.")
|
||
cur.execute(
|
||
"UPDATE users SET failed_attempts=%s, locked_until=%s WHERE id=%s",
|
||
(new_count, locked_until, row["id"]),
|
||
)
|
||
conn.commit()
|
||
cur.close()
|
||
except Exception as e:
|
||
logger.error(f"record_failed_attempt error: {e}")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def clear_failed_attempts(user_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute("UPDATE users SET failed_attempts=0, locked_until=NULL WHERE id=%s", (user_id,))
|
||
conn.commit()
|
||
cur.close()
|
||
except Exception as e:
|
||
logger.error(f"clear_failed_attempts error: {e}")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Change Password ───────────────────────────────────────────────────────────
|
||
|
||
PW_MIN_LENGTH = 8
|
||
PW_REQUIRE_UPPER = True
|
||
PW_REQUIRE_DIGIT = True
|
||
PW_REQUIRE_SPECIAL = True
|
||
_SPECIAL_CHARS = set("!@#$%^&*()_+-=[]{}|;':\",./<>?")
|
||
|
||
|
||
def validate_password_strength(password: str) -> list:
|
||
errors = []
|
||
if len(password) < PW_MIN_LENGTH:
|
||
errors.append(f"At least {PW_MIN_LENGTH} characters")
|
||
if PW_REQUIRE_UPPER and not any(c.isupper() for c in password):
|
||
errors.append("At least one uppercase letter")
|
||
if PW_REQUIRE_DIGIT and not any(c.isdigit() for c in password):
|
||
errors.append("At least one number")
|
||
if PW_REQUIRE_SPECIAL and not any(c in _SPECIAL_CHARS for c in password):
|
||
errors.append("At least one special character (!@#$%^&* etc.)")
|
||
return errors
|
||
|
||
|
||
def change_password(user_id: int, old_password: str, new_password: str) -> tuple:
|
||
issues = validate_password_strength(new_password)
|
||
if issues:
|
||
return False, "New password does not meet requirements: " + "; ".join(issues)
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute("SELECT password FROM users WHERE id=%s AND is_active=1", (user_id,))
|
||
row = cur.fetchone()
|
||
if not row:
|
||
cur.close()
|
||
return False, "User account not found."
|
||
if not _verify_password(old_password, row["password"]):
|
||
cur.close()
|
||
log_action(user_id, "CHANGE_PASSWORD_FAIL", "users", user_id,
|
||
"Incorrect current password provided.")
|
||
return False, "Current password is incorrect."
|
||
if old_password == new_password:
|
||
cur.close()
|
||
return False, "New password must differ from the current password."
|
||
cur.execute("UPDATE users SET password=%s WHERE id=%s", (_hash_password(new_password), user_id))
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(user_id, "CHANGE_PASSWORD", "users", user_id, "Password changed successfully.")
|
||
logger.info(f"Password changed for user_id={user_id}.")
|
||
return True, "Password changed successfully."
|
||
except Exception as e:
|
||
logger.error(f"change_password error: {e}")
|
||
return False, f"An error occurred: {e}"
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── User CRUD ────────────────────────────────────────────────────────────────
|
||
|
||
def get_all_users():
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"SELECT id, username, role, full_name, email, is_active, created_at "
|
||
"FROM users ORDER BY username"
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_user_by_id(user_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"SELECT id, username, role, full_name, email, is_active FROM users WHERE id=%s",
|
||
(user_id,),
|
||
)
|
||
row = cur.fetchone()
|
||
cur.close()
|
||
return row
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def update_user_profile(user_id: int, full_name: str, email: str):
|
||
"""Allow a user to update their own full name and email address."""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"UPDATE users SET full_name=%s, email=%s WHERE id=%s",
|
||
(full_name.strip() or None, email.strip().lower() or None, user_id),
|
||
)
|
||
conn.commit()
|
||
cur.close()
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def register_user(username: str, password: str, full_name: str = "", email: str = None):
|
||
"""Self-service registration. Role is always 'user'. Raises ValueError on duplicates."""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute("SELECT id FROM users WHERE username=%s", (username,))
|
||
if cur.fetchone():
|
||
raise ValueError("Username is already taken.")
|
||
if email:
|
||
cur.execute("SELECT id FROM users WHERE email=%s", (email,))
|
||
if cur.fetchone():
|
||
raise ValueError("An account with that email address already exists.")
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"INSERT INTO users (username, password, role, full_name, email) VALUES (%s,%s,'user',%s,%s)",
|
||
(username, _hash_password(password), full_name or username, email or None),
|
||
)
|
||
conn.commit()
|
||
new_id = cur.lastrowid
|
||
cur.close()
|
||
log_action(None, "REGISTER", "users", new_id,
|
||
f"New user self-registered: '{username}'.")
|
||
try:
|
||
assign_default_shift(new_id)
|
||
except Exception as e:
|
||
logger.warning(f"register_user: could not assign default shift to user_id={new_id}: {e}")
|
||
return new_id
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# Name that identifies the auto-created default shift for new registrations.
|
||
# Admins can rename it from the Shifts page without breaking anything —
|
||
# once the shift exists it is looked up by id (stored nowhere persistent here),
|
||
# but a rename just means the next registration creates a second default shift.
|
||
_DEFAULT_SHIFT_NAME = "Default (Mon–Fri)"
|
||
_DEFAULT_SHIFT_DAYS = "23456" # MySQL DAYOFWEEK: Mon=2 Tue=3 Wed=4 Thu=5 Fri=6
|
||
_DEFAULT_SHIFT_START = "09:00:00"
|
||
_DEFAULT_SHIFT_END = "17:30:00"
|
||
|
||
|
||
def assign_default_shift(user_id: int) -> None:
|
||
"""
|
||
Find or create the Default shift and enrol a newly registered user in it.
|
||
Also ensures every currently-active daily website is present in the shift.
|
||
Safe to call multiple times — INSERT IGNORE throughout, nothing is overwritten.
|
||
"""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
|
||
# Find the default shift (by name)
|
||
cur.execute(
|
||
"SELECT id FROM shifts WHERE name=%s AND is_active=1 LIMIT 1",
|
||
(_DEFAULT_SHIFT_NAME,),
|
||
)
|
||
row = cur.fetchone()
|
||
|
||
if row:
|
||
shift_id = row["id"]
|
||
else:
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"INSERT INTO shifts "
|
||
"(name, days_of_week, start_time, end_time, note, is_active, created_by) "
|
||
"VALUES (%s, %s, %s, %s, %s, 1, NULL)",
|
||
(
|
||
_DEFAULT_SHIFT_NAME, _DEFAULT_SHIFT_DAYS,
|
||
_DEFAULT_SHIFT_START, _DEFAULT_SHIFT_END,
|
||
"Auto-created default shift for self-registered users.",
|
||
),
|
||
)
|
||
conn.commit()
|
||
shift_id = cur.lastrowid
|
||
log_action(None, "CREATE_SHIFT", "shifts", shift_id,
|
||
f"Auto-created default shift '{_DEFAULT_SHIFT_NAME}'.")
|
||
|
||
# Sync all active daily websites into the shift (INSERT IGNORE = no duplicates)
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"SELECT id FROM websites WHERE check_type='daily' AND is_active=1 ORDER BY name"
|
||
)
|
||
daily_sites = cur.fetchall()
|
||
|
||
cur = conn.cursor()
|
||
for idx, w in enumerate(daily_sites):
|
||
cur.execute(
|
||
"INSERT IGNORE INTO shift_websites (shift_id, website_id, sort_order) "
|
||
"VALUES (%s, %s, %s)",
|
||
(shift_id, w["id"], idx),
|
||
)
|
||
|
||
# Enrol the user (INSERT IGNORE — idempotent)
|
||
cur.execute(
|
||
"INSERT IGNORE INTO shift_users (shift_id, user_id) VALUES (%s, %s)",
|
||
(shift_id, user_id),
|
||
)
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(None, "ASSIGN_DEFAULT_SHIFT", "shifts", shift_id,
|
||
f"User id={user_id} enrolled in '{_DEFAULT_SHIFT_NAME}' "
|
||
f"with {len(daily_sites)} daily website(s).")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def create_user(admin_id, username, password, role, full_name, email=None):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"INSERT INTO users (username, password, role, full_name, email) VALUES (%s,%s,%s,%s,%s)",
|
||
(username, _hash_password(password), role, full_name, email or None),
|
||
)
|
||
conn.commit()
|
||
new_id = cur.lastrowid
|
||
cur.close()
|
||
log_action(admin_id, "CREATE_USER", "users", new_id,
|
||
f"Created user '{username}' role='{role}' email='{email or ''}'.")
|
||
return new_id
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def update_user(admin_id, user_id, username, role, full_name, is_active, password=None, email=None):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
|
||
# Guard: prevent demoting or deactivating the last active admin.
|
||
cur.execute("SELECT role FROM users WHERE id=%s", (user_id,))
|
||
target = cur.fetchone()
|
||
if target and target["role"] == "admin":
|
||
removing_admin = (role != "admin") or (not is_active)
|
||
if removing_admin:
|
||
cur.execute(
|
||
"SELECT COUNT(*) AS n FROM users WHERE role='admin' AND is_active=1"
|
||
)
|
||
if cur.fetchone()["n"] <= 1:
|
||
cur.close()
|
||
raise ValueError(
|
||
"Cannot demote or deactivate the last active administrator. "
|
||
"Promote another user to admin first."
|
||
)
|
||
|
||
cur = conn.cursor() # switch back to plain cursor for the UPDATE
|
||
if password:
|
||
cur.execute(
|
||
"UPDATE users SET username=%s, role=%s, full_name=%s, is_active=%s, "
|
||
"email=%s, password=%s WHERE id=%s",
|
||
(username, role, full_name, is_active, email or None, _hash_password(password), user_id),
|
||
)
|
||
else:
|
||
cur.execute(
|
||
"UPDATE users SET username=%s, role=%s, full_name=%s, is_active=%s, "
|
||
"email=%s WHERE id=%s",
|
||
(username, role, full_name, is_active, email or None, user_id),
|
||
)
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(admin_id, "UPDATE_USER", "users", user_id, f"Updated user id={user_id}.")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def delete_user(admin_id, user_id):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
if admin_id == user_id:
|
||
cur.close()
|
||
raise ValueError("You cannot delete your own account.")
|
||
cur.execute("SELECT role FROM users WHERE id=%s", (user_id,))
|
||
target = cur.fetchone()
|
||
if target and target["role"] == "admin":
|
||
cur.execute("SELECT COUNT(*) AS n FROM users WHERE role='admin' AND is_active=1")
|
||
if cur.fetchone()["n"] <= 1:
|
||
cur.close()
|
||
raise ValueError(
|
||
"Cannot delete the last active administrator account. "
|
||
"Promote another user to admin first."
|
||
)
|
||
cur.execute("DELETE FROM users WHERE id=%s", (user_id,))
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(admin_id, "DELETE_USER", "users", user_id, f"Deleted user id={user_id}.")
|
||
logger.info(f"User id={user_id} deleted by admin_id={admin_id}.")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Website CRUD ─────────────────────────────────────────────────────────────
|
||
|
||
def get_all_websites():
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"""
|
||
SELECT w.*, u.username AS creator
|
||
FROM websites w
|
||
LEFT JOIN users u ON u.id = w.created_by
|
||
WHERE w.is_active = 1
|
||
ORDER BY w.name
|
||
"""
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_existing_website_urls() -> set:
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT url FROM websites")
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return {(r[0] or "").strip().lower() for r in rows}
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_existing_website_names() -> set:
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT name FROM websites")
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return {(r[0] or "").strip().lower() for r in rows}
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_website_by_id(website_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute("SELECT * FROM websites WHERE id=%s", (website_id,))
|
||
row = cur.fetchone()
|
||
cur.close()
|
||
return row
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def create_website(admin_id, name, url, check_type, note, credentials: list,
|
||
visibility: str = "all", assigned_user_ids: list = None):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"INSERT INTO websites (name, url, check_type, visibility, note, created_by) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s)",
|
||
(name, url, check_type, visibility, note, admin_id),
|
||
)
|
||
conn.commit()
|
||
new_id = cur.lastrowid
|
||
for cred in credentials:
|
||
cur.execute(
|
||
"INSERT INTO website_credentials (website_id, username, password, label) "
|
||
"VALUES (%s,%s,%s,%s)",
|
||
(new_id, cred["username"], _enc(cred["password"]), cred.get("label", "")),
|
||
)
|
||
log_action(admin_id, "ADD_CREDENTIAL", "website_credentials", new_id,
|
||
f"Added credential label='{cred.get('label','')}' user='{cred['username']}' for website '{name}'.")
|
||
if visibility == "assigned" and assigned_user_ids:
|
||
for uid in assigned_user_ids:
|
||
cur.execute(
|
||
"INSERT IGNORE INTO website_users (website_id, user_id) VALUES (%s,%s)",
|
||
(new_id, uid),
|
||
)
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(admin_id, "CREATE_WEBSITE", "websites", new_id,
|
||
f"Created website '{name}' check_type='{check_type}' visibility='{visibility}'.")
|
||
return new_id
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def update_website(admin_id, website_id, name, url, check_type, note,
|
||
credentials: list, visibility: str = "all",
|
||
assigned_user_ids: list = None):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"SELECT username, label FROM website_credentials WHERE website_id=%s",
|
||
(website_id,),
|
||
)
|
||
old_creds = {(r["username"], r["label"] or "") for r in cur.fetchall()}
|
||
cur.execute(
|
||
"UPDATE websites SET name=%s, url=%s, check_type=%s, visibility=%s, note=%s WHERE id=%s",
|
||
(name, url, check_type, visibility, note, website_id),
|
||
)
|
||
cur.execute("DELETE FROM website_credentials WHERE website_id=%s", (website_id,))
|
||
new_creds = set()
|
||
for cred in credentials:
|
||
cur.execute(
|
||
"INSERT INTO website_credentials (website_id, username, password, label) "
|
||
"VALUES (%s,%s,%s,%s)",
|
||
(website_id, cred["username"], _enc(cred["password"]), cred.get("label", "")),
|
||
)
|
||
new_creds.add((cred["username"], cred.get("label", "") or ""))
|
||
cur.execute("DELETE FROM website_users WHERE website_id=%s", (website_id,))
|
||
if visibility == "assigned" and assigned_user_ids:
|
||
for uid in assigned_user_ids:
|
||
cur.execute(
|
||
"INSERT IGNORE INTO website_users (website_id, user_id) VALUES (%s,%s)",
|
||
(website_id, uid),
|
||
)
|
||
conn.commit()
|
||
cur.close()
|
||
for user, label in (new_creds - old_creds):
|
||
log_action(admin_id, "ADD_CREDENTIAL", "website_credentials", website_id,
|
||
f"Added credential label='{label}' user='{user}' for website id={website_id}.")
|
||
for user, label in (old_creds - new_creds):
|
||
log_action(admin_id, "REMOVE_CREDENTIAL", "website_credentials", website_id,
|
||
f"Removed credential label='{label}' user='{user}' from website id={website_id}.")
|
||
log_action(admin_id, "UPDATE_WEBSITE", "websites", website_id,
|
||
f"Updated website id={website_id} check_type='{check_type}' visibility='{visibility}'.")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def delete_website(admin_id, website_id):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute("UPDATE websites SET is_active=0 WHERE id=%s", (website_id,))
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(admin_id, "DELETE_WEBSITE", "websites", website_id,
|
||
f"Soft-deleted website id={website_id}.")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_website_credentials(website_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute("SELECT * FROM website_credentials WHERE website_id=%s", (website_id,))
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
for row in rows:
|
||
row["password"] = _dec(row["password"])
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_website_assigned_users(website_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"""
|
||
SELECT u.id, u.username, u.full_name
|
||
FROM website_users wu
|
||
JOIN users u ON u.id = wu.user_id
|
||
WHERE wu.website_id = %s
|
||
ORDER BY u.username
|
||
""",
|
||
(website_id,),
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Shift Check CRUD ─────────────────────────────────────────────────────────
|
||
|
||
def get_today_checks(user_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"""
|
||
SELECT COUNT(*) AS cnt
|
||
FROM shifts s
|
||
JOIN shift_users su ON su.shift_id = s.id
|
||
WHERE su.user_id = %s
|
||
AND s.is_active = 1
|
||
AND LOCATE(CAST(DAYOFWEEK(CURDATE()) AS CHAR), s.days_of_week) > 0
|
||
""",
|
||
(user_id,),
|
||
)
|
||
has_shifts = cur.fetchone()["cnt"] > 0
|
||
|
||
if has_shifts:
|
||
cur.execute(
|
||
"""
|
||
SELECT
|
||
w.id, w.name, w.url, w.check_type,
|
||
w.note AS site_note,
|
||
MIN(sw.sort_order) AS sort_order,
|
||
GROUP_CONCAT(DISTINCT s.name ORDER BY s.name SEPARATOR ', ') AS shift_names,
|
||
sc.id AS check_id, sc.checked_at, sc.user_note,
|
||
sc.user_id AS checked_by_id
|
||
FROM shift_websites sw
|
||
JOIN shifts s ON s.id = sw.shift_id
|
||
JOIN websites w ON w.id = sw.website_id
|
||
JOIN shift_users su ON su.shift_id = s.id AND su.user_id = %s
|
||
LEFT JOIN shift_checks sc
|
||
ON sc.id = (
|
||
SELECT sc2.id FROM shift_checks sc2
|
||
WHERE sc2.website_id = w.id AND sc2.user_id = %s
|
||
AND IF(w.check_type = 'weekly',
|
||
YEARWEEK(sc2.checked_at, 1) = YEARWEEK(CURDATE(), 1),
|
||
DATE(sc2.checked_at) = CURDATE())
|
||
ORDER BY sc2.checked_at DESC
|
||
LIMIT 1
|
||
)
|
||
WHERE s.is_active = 1 AND w.is_active = 1
|
||
AND LOCATE(CAST(DAYOFWEEK(CURDATE()) AS CHAR), s.days_of_week) > 0
|
||
AND (w.visibility = 'all'
|
||
OR EXISTS (SELECT 1 FROM website_users wu WHERE wu.website_id=w.id AND wu.user_id=%s))
|
||
GROUP BY w.id, sc.id, sc.checked_at, sc.user_note, sc.user_id
|
||
ORDER BY sort_order, w.name
|
||
""",
|
||
(user_id, user_id, user_id),
|
||
)
|
||
else:
|
||
cur.execute(
|
||
"""
|
||
SELECT w.id, w.name, w.url, w.check_type,
|
||
w.note AS site_note, 0 AS sort_order, NULL AS shift_names,
|
||
sc.id AS check_id, sc.checked_at, sc.user_note, sc.user_id AS checked_by_id
|
||
FROM websites w
|
||
LEFT JOIN shift_checks sc
|
||
ON sc.id = (
|
||
SELECT sc2.id FROM shift_checks sc2
|
||
WHERE sc2.website_id = w.id AND sc2.user_id = %s
|
||
AND IF(w.check_type = 'weekly',
|
||
YEARWEEK(sc2.checked_at, 1) = YEARWEEK(CURDATE(), 1),
|
||
DATE(sc2.checked_at) = CURDATE())
|
||
ORDER BY sc2.checked_at DESC
|
||
LIMIT 1
|
||
)
|
||
WHERE w.is_active=1
|
||
AND (w.visibility='all'
|
||
OR EXISTS (SELECT 1 FROM website_users wu WHERE wu.website_id=w.id AND wu.user_id=%s))
|
||
ORDER BY w.name
|
||
""",
|
||
(user_id, user_id),
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def mark_website_checked(user_id, website_id, user_note=""):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"DELETE FROM shift_checks WHERE website_id=%s AND user_id=%s AND DATE(checked_at)=CURDATE()",
|
||
(website_id, user_id),
|
||
)
|
||
cur.execute(
|
||
"INSERT INTO shift_checks (website_id, user_id, user_note) VALUES (%s,%s,%s)",
|
||
(website_id, user_id, user_note),
|
||
)
|
||
conn.commit()
|
||
new_id = cur.lastrowid
|
||
cur.close()
|
||
log_action(user_id, "CHECK_WEBSITE", "websites", website_id,
|
||
f"User {user_id} checked website {website_id}. Note: {user_note}")
|
||
return new_id
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def unmark_website_checked(user_id, website_id):
|
||
"""Remove today's check record — web-only feature for undo support."""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"DELETE FROM shift_checks WHERE website_id=%s AND user_id=%s AND DATE(checked_at)=CURDATE()",
|
||
(website_id, user_id),
|
||
)
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(user_id, "UNCHECK_WEBSITE", "websites", website_id,
|
||
f"User {user_id} unchecked website {website_id}.")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def update_check_note(user_id, website_id, user_note):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"UPDATE shift_checks SET user_note=%s "
|
||
"WHERE website_id=%s AND user_id=%s AND DATE(checked_at)=CURDATE()",
|
||
(user_note, website_id, user_id),
|
||
)
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(user_id, "UPDATE_NOTE", "shift_checks", website_id,
|
||
f"Updated note for website {website_id}.")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_activity_log(limit=200, search: str = ""):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
params = []
|
||
sql = (
|
||
"SELECT al.id, al.user_id, al.action, al.entity, al.entity_id,"
|
||
" al.detail, al.logged_at AS created_at, u.username"
|
||
" FROM activity_log al"
|
||
" LEFT JOIN users u ON u.id = al.user_id"
|
||
)
|
||
if search:
|
||
sql += (
|
||
" WHERE al.action LIKE %s OR u.username LIKE %s"
|
||
" OR al.entity LIKE %s OR al.detail LIKE %s"
|
||
)
|
||
like = f"%{search}%"
|
||
params.extend([like, like, like, like])
|
||
sql += " ORDER BY al.logged_at DESC LIMIT %s"
|
||
params.append(limit)
|
||
cur.execute(sql, params)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Reports ──────────────────────────────────────────────────────────────────
|
||
|
||
def get_shift_report(date_from=None, date_to=None, user_id=None, website_id=None):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
conditions, params = [], []
|
||
if date_from:
|
||
conditions.append("DATE(sc.checked_at) >= %s"); params.append(str(date_from))
|
||
if date_to:
|
||
conditions.append("DATE(sc.checked_at) <= %s"); params.append(str(date_to))
|
||
if user_id:
|
||
conditions.append("sc.user_id = %s"); params.append(user_id)
|
||
if website_id:
|
||
conditions.append("sc.website_id = %s"); params.append(website_id)
|
||
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
|
||
cur.execute(
|
||
f"""
|
||
SELECT DATE(sc.checked_at) AS check_date, sc.checked_at,
|
||
u.username, COALESCE(u.full_name, u.username) AS full_name,
|
||
w.name AS website_name, w.url,
|
||
COALESCE(sc.user_note,'') AS user_note, 'Checked' AS status
|
||
FROM shift_checks sc
|
||
JOIN users u ON u.id=sc.user_id
|
||
JOIN websites w ON w.id=sc.website_id
|
||
{where}
|
||
ORDER BY sc.checked_at DESC, u.username, w.name
|
||
""",
|
||
params,
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
logger.info(f"Shift report queried - {len(rows)} rows returned.")
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_unchecked_report(target_date=None, user_id=None):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
user_filter = "AND u.id = %s" if user_id else ""
|
||
if target_date:
|
||
date_val, date_expr, date_params = str(target_date), "%s", [str(target_date)]
|
||
dow_expr, dow_params = "DAYOFWEEK(%s)", [str(target_date)]
|
||
else:
|
||
date_expr, date_params = "CURDATE()", []
|
||
dow_expr, dow_params = "DAYOFWEEK(CURDATE())", []
|
||
user_filter_params = [user_id] if user_id else []
|
||
params = date_params + user_filter_params + dow_params + date_params
|
||
cur.execute(
|
||
f"""
|
||
SELECT {date_expr} AS check_date,
|
||
u.username, COALESCE(u.full_name, u.username) AS full_name,
|
||
w.name AS website_name, w.url, 'Not Checked' AS status
|
||
FROM users u
|
||
JOIN websites w ON w.is_active=1
|
||
WHERE u.is_active=1 AND u.role='user'
|
||
{user_filter}
|
||
AND EXISTS (
|
||
SELECT 1 FROM shift_websites sw
|
||
JOIN shifts s ON s.id=sw.shift_id
|
||
JOIN shift_users su ON su.shift_id=s.id AND su.user_id=u.id
|
||
WHERE sw.website_id=w.id AND s.is_active=1
|
||
AND LOCATE(CAST({dow_expr} AS CHAR), s.days_of_week)>0
|
||
AND (w.visibility='all'
|
||
OR EXISTS (SELECT 1 FROM website_users wu WHERE wu.website_id=w.id AND wu.user_id=u.id))
|
||
)
|
||
AND NOT EXISTS (
|
||
SELECT 1 FROM shift_checks sc
|
||
WHERE sc.website_id=w.id AND sc.user_id=u.id AND DATE(sc.checked_at)={date_expr}
|
||
)
|
||
ORDER BY u.username, w.name
|
||
""",
|
||
params,
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
logger.info(f"Unchecked report queried - {len(rows)} rows returned.")
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_summary_report(date_from=None, date_to=None):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
conditions, params = [], []
|
||
if date_from:
|
||
conditions.append("DATE(sc.checked_at) >= %s"); params.append(str(date_from))
|
||
if date_to:
|
||
conditions.append("DATE(sc.checked_at) <= %s"); params.append(str(date_to))
|
||
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
|
||
cur.execute(
|
||
f"""
|
||
WITH agg AS (
|
||
SELECT DATE(sc.checked_at) AS check_date, sc.user_id,
|
||
COUNT(DISTINCT sc.website_id) AS checked_count
|
||
FROM shift_checks sc {where}
|
||
GROUP BY DATE(sc.checked_at), sc.user_id
|
||
)
|
||
SELECT agg.check_date, u.username,
|
||
COALESCE(u.full_name, u.username) AS full_name,
|
||
agg.checked_count,
|
||
(SELECT COUNT(DISTINCT sw2.website_id) FROM shift_websites sw2
|
||
JOIN shifts s2 ON s2.id=sw2.shift_id
|
||
JOIN shift_users su2 ON su2.shift_id=s2.id AND su2.user_id=u.id
|
||
WHERE s2.is_active=1
|
||
AND LOCATE(CAST(DAYOFWEEK(agg.check_date) AS CHAR),s2.days_of_week)>0
|
||
) AS total_sites,
|
||
ROUND(agg.checked_count*100.0/NULLIF((
|
||
SELECT COUNT(DISTINCT sw2.website_id) FROM shift_websites sw2
|
||
JOIN shifts s2 ON s2.id=sw2.shift_id
|
||
JOIN shift_users su2 ON su2.shift_id=s2.id AND su2.user_id=u.id
|
||
WHERE s2.is_active=1
|
||
AND LOCATE(CAST(DAYOFWEEK(agg.check_date) AS CHAR),s2.days_of_week)>0
|
||
),0),1) AS pct_complete
|
||
FROM agg
|
||
JOIN users u ON u.id=agg.user_id
|
||
ORDER BY agg.check_date DESC, u.username
|
||
""",
|
||
params,
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
logger.info(f"Summary report queried - {len(rows)} rows returned.")
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_report_filter_options():
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute("SELECT id, username, full_name FROM users WHERE is_active=1 ORDER BY username")
|
||
users = cur.fetchall()
|
||
cur.execute("SELECT id, name FROM websites WHERE is_active=1 ORDER BY name")
|
||
websites = cur.fetchall()
|
||
cur.close()
|
||
return users, websites
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Shift Management ─────────────────────────────────────────────────────────
|
||
|
||
def get_all_shifts():
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"""
|
||
SELECT s.*, u.username AS creator,
|
||
COUNT(DISTINCT su.user_id) AS user_count,
|
||
COUNT(DISTINCT sw.website_id) AS website_count
|
||
FROM shifts s
|
||
LEFT JOIN users u ON u.id=s.created_by
|
||
LEFT JOIN shift_users su ON su.shift_id=s.id
|
||
LEFT JOIN shift_websites sw ON sw.shift_id=s.id
|
||
GROUP BY s.id ORDER BY s.name
|
||
"""
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_shift_by_id(shift_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute("SELECT * FROM shifts WHERE id=%s", (shift_id,))
|
||
row = cur.fetchone()
|
||
cur.close()
|
||
return row
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_shift_assigned_users(shift_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"""
|
||
SELECT u.id, u.username, u.full_name
|
||
FROM shift_users su JOIN users u ON u.id=su.user_id
|
||
WHERE su.shift_id=%s ORDER BY u.username
|
||
""",
|
||
(shift_id,),
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_shift_assigned_websites(shift_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"""
|
||
SELECT w.id, w.name, w.url, sw.sort_order
|
||
FROM shift_websites sw JOIN websites w ON w.id=sw.website_id
|
||
WHERE sw.shift_id=%s ORDER BY sw.sort_order, w.name
|
||
""",
|
||
(shift_id,),
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def create_shift(admin_id, name, days_of_week, start_time, end_time,
|
||
note, user_ids: list, website_ids: list):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"INSERT INTO shifts (name, days_of_week, start_time, end_time, note, created_by) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s)",
|
||
(name, days_of_week, start_time, end_time, note, admin_id),
|
||
)
|
||
conn.commit()
|
||
shift_id = cur.lastrowid
|
||
for uid in user_ids:
|
||
cur.execute("INSERT IGNORE INTO shift_users (shift_id, user_id) VALUES (%s,%s)", (shift_id, uid))
|
||
for idx, wid in enumerate(website_ids):
|
||
cur.execute(
|
||
"INSERT IGNORE INTO shift_websites (shift_id, website_id, sort_order) VALUES (%s,%s,%s)",
|
||
(shift_id, wid, idx),
|
||
)
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(admin_id, "CREATE_SHIFT", "shifts", shift_id,
|
||
f"Created shift '{name}' days={days_of_week} users={user_ids} websites={website_ids}.")
|
||
return shift_id
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def update_shift(admin_id, shift_id, name, days_of_week, start_time, end_time,
|
||
note, is_active, user_ids: list, website_ids: list):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"UPDATE shifts SET name=%s, days_of_week=%s, start_time=%s, end_time=%s, "
|
||
"note=%s, is_active=%s WHERE id=%s",
|
||
(name, days_of_week, start_time, end_time, note, is_active, shift_id),
|
||
)
|
||
cur.execute("DELETE FROM shift_users WHERE shift_id=%s", (shift_id,))
|
||
for uid in user_ids:
|
||
cur.execute("INSERT IGNORE INTO shift_users (shift_id, user_id) VALUES (%s,%s)", (shift_id, uid))
|
||
cur.execute("DELETE FROM shift_websites WHERE shift_id=%s", (shift_id,))
|
||
for idx, wid in enumerate(website_ids):
|
||
cur.execute(
|
||
"INSERT IGNORE INTO shift_websites (shift_id, website_id, sort_order) VALUES (%s,%s,%s)",
|
||
(shift_id, wid, idx),
|
||
)
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(admin_id, "UPDATE_SHIFT", "shifts", shift_id,
|
||
f"Updated shift id={shift_id} name='{name}' active={is_active}.")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def delete_shift(admin_id, shift_id):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute("UPDATE shifts SET is_active=0 WHERE id=%s", (shift_id,))
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(admin_id, "DELETE_SHIFT", "shifts", shift_id, f"Soft-deleted shift id={shift_id}.")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_user_active_shifts(user_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"""
|
||
SELECT s.* FROM shifts s
|
||
JOIN shift_users su ON su.shift_id=s.id
|
||
WHERE su.user_id=%s AND s.is_active=1
|
||
AND LOCATE(CAST(DAYOFWEEK(CURDATE()) AS CHAR), s.days_of_week)>0
|
||
ORDER BY s.start_time
|
||
""",
|
||
(user_id,),
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Admin Dashboard ──────────────────────────────────────────────────────────
|
||
|
||
_admin_stats_cache: dict | None = None
|
||
_admin_stats_cached_at: datetime.datetime | None = None
|
||
_ADMIN_STATS_TTL_S = 60 # seconds
|
||
|
||
|
||
def invalidate_admin_stats_cache():
|
||
global _admin_stats_cache, _admin_stats_cached_at
|
||
_admin_stats_cache = None
|
||
_admin_stats_cached_at = None
|
||
|
||
|
||
def get_admin_dashboard_stats():
|
||
global _admin_stats_cache, _admin_stats_cached_at
|
||
now = datetime.datetime.now()
|
||
if (_admin_stats_cache is not None and _admin_stats_cached_at is not None
|
||
and (now - _admin_stats_cached_at).total_seconds() < _ADMIN_STATS_TTL_S):
|
||
return _admin_stats_cache
|
||
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"""
|
||
SELECT u.id AS user_id, u.username,
|
||
COALESCE(u.full_name, u.username) AS full_name,
|
||
COUNT(DISTINCT sc.website_id) AS checked_count,
|
||
(SELECT COUNT(DISTINCT sw2.website_id) FROM shift_websites sw2
|
||
JOIN shifts s2 ON s2.id=sw2.shift_id
|
||
JOIN shift_users su2 ON su2.shift_id=s2.id AND su2.user_id=u.id
|
||
WHERE s2.is_active=1
|
||
AND LOCATE(CAST(DAYOFWEEK(CURDATE()) AS CHAR),s2.days_of_week)>0
|
||
) AS total_sites,
|
||
ROUND(COUNT(DISTINCT sc.website_id)*100.0/NULLIF((
|
||
SELECT COUNT(DISTINCT sw2.website_id) FROM shift_websites sw2
|
||
JOIN shifts s2 ON s2.id=sw2.shift_id
|
||
JOIN shift_users su2 ON su2.shift_id=s2.id AND su2.user_id=u.id
|
||
WHERE s2.is_active=1
|
||
AND LOCATE(CAST(DAYOFWEEK(CURDATE()) AS CHAR),s2.days_of_week)>0
|
||
),0),1) AS pct_complete
|
||
FROM users u
|
||
LEFT JOIN shift_checks sc ON sc.user_id=u.id AND DATE(sc.checked_at)=CURDATE()
|
||
WHERE u.is_active=1 AND u.role='user'
|
||
GROUP BY u.id ORDER BY pct_complete DESC, u.username
|
||
"""
|
||
)
|
||
user_stats = cur.fetchall()
|
||
cur.execute("SELECT COUNT(*) AS n FROM websites WHERE is_active=1")
|
||
total_sites = cur.fetchone()["n"]
|
||
cur.execute("SELECT COUNT(*) AS n FROM users WHERE is_active=1 AND role='user'")
|
||
total_users = cur.fetchone()["n"]
|
||
cur.execute("SELECT COUNT(DISTINCT user_id) AS n FROM shift_checks WHERE DATE(checked_at)=CURDATE()")
|
||
active_today = cur.fetchone()["n"]
|
||
cur.execute("SELECT COUNT(*) AS n FROM bid_tracker WHERE status IN ('open','monitoring')")
|
||
open_bids = cur.fetchone()["n"]
|
||
cur.execute(
|
||
"SELECT COUNT(*) AS n FROM bid_tracker"
|
||
" WHERE status IN ('open','monitoring')"
|
||
" AND due_date BETWEEN CURDATE() AND DATE_ADD(CURDATE(), INTERVAL 7 DAY)"
|
||
)
|
||
bids_due_soon = cur.fetchone()["n"]
|
||
cur.execute(
|
||
"SELECT COUNT(*) AS n FROM ai_analysis_log"
|
||
" WHERE analyzed_at >= DATE_SUB(NOW(), INTERVAL 30 DAY)"
|
||
)
|
||
ai_analyses_30d = cur.fetchone()["n"]
|
||
cur.close()
|
||
result = {
|
||
"user_stats": user_stats,
|
||
"total_sites": total_sites,
|
||
"total_users": total_users,
|
||
"active_today": active_today,
|
||
"open_bids": open_bids,
|
||
"bids_due_soon": bids_due_soon,
|
||
"ai_analyses_30d": ai_analyses_30d,
|
||
}
|
||
_admin_stats_cache = result
|
||
_admin_stats_cached_at = now
|
||
return result
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── AI Criteria CRUD ─────────────────────────────────────────────────────────
|
||
|
||
def get_all_criteria():
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"SELECT id, title, description, is_active, sort_order, created_by, created_at, updated_at "
|
||
"FROM ai_criteria ORDER BY sort_order, id"
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_active_criteria():
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"SELECT id, title, description FROM ai_criteria WHERE is_active=1 ORDER BY sort_order, id"
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def create_criterion(admin_id: int, title: str, description: str, is_active: bool, sort_order: int) -> int:
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"INSERT INTO ai_criteria (title, description, is_active, sort_order, created_by) "
|
||
"VALUES (%s,%s,%s,%s,%s)",
|
||
(title, description, int(is_active), sort_order, admin_id),
|
||
)
|
||
conn.commit()
|
||
new_id = cur.lastrowid
|
||
cur.close()
|
||
log_action(admin_id, "CREATE_AI_CRITERION", "ai_criteria", new_id,
|
||
f"Created criterion '{title}' active={is_active} order={sort_order}.")
|
||
logger.info(f"AI criterion id={new_id} '{title}' created by admin_id={admin_id}.")
|
||
return new_id
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def update_criterion(admin_id: int, criterion_id: int, title: str,
|
||
description: str, is_active: bool, sort_order: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"UPDATE ai_criteria SET title=%s, description=%s, is_active=%s, sort_order=%s WHERE id=%s",
|
||
(title, description, int(is_active), sort_order, criterion_id),
|
||
)
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(admin_id, "UPDATE_AI_CRITERION", "ai_criteria", criterion_id,
|
||
f"Updated criterion id={criterion_id} '{title}' active={is_active} order={sort_order}. "
|
||
f"Description: {description[:500]}")
|
||
logger.info(f"AI criterion id={criterion_id} updated by admin_id={admin_id}.")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def delete_criterion(admin_id: int, criterion_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute("SELECT title FROM ai_criteria WHERE id=%s", (criterion_id,))
|
||
row = cur.fetchone()
|
||
title = row["title"] if row else str(criterion_id)
|
||
cur.execute("DELETE FROM ai_criteria WHERE id=%s", (criterion_id,))
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(admin_id, "DELETE_AI_CRITERION", "ai_criteria", criterion_id,
|
||
f"Deleted criterion id={criterion_id} '{title}'.")
|
||
logger.info(f"AI criterion id={criterion_id} '{title}' deleted by admin_id={admin_id}.")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def save_ai_analysis(user_id: int, file_names: str, model: str,
|
||
verdict, criteria_snapshot, summary_text: str) -> int:
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"INSERT INTO ai_analysis_log (user_id, file_names, model, verdict, criteria_snapshot, summary_text) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s)",
|
||
(user_id, file_names, model, verdict, criteria_snapshot, summary_text),
|
||
)
|
||
conn.commit()
|
||
new_id = cur.lastrowid
|
||
cur.close()
|
||
logger.info(f"AI analysis saved: id={new_id} user_id={user_id} verdict={verdict}.")
|
||
return new_id
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_ai_analysis_history(user_id=None, limit: int = 100):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
if user_id:
|
||
cur.execute(
|
||
"SELECT al.id, u.username, al.file_names, al.model, al.verdict, al.analyzed_at "
|
||
"FROM ai_analysis_log al LEFT JOIN users u ON u.id=al.user_id "
|
||
"WHERE al.user_id=%s ORDER BY al.analyzed_at DESC LIMIT %s",
|
||
(user_id, limit),
|
||
)
|
||
else:
|
||
cur.execute(
|
||
"SELECT al.id, u.username, al.file_names, al.model, al.verdict, al.analyzed_at "
|
||
"FROM ai_analysis_log al LEFT JOIN users u ON u.id=al.user_id "
|
||
"ORDER BY al.analyzed_at DESC LIMIT %s",
|
||
(limit,),
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_ai_analysis_detail(analysis_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"SELECT al.*, u.username FROM ai_analysis_log al "
|
||
"LEFT JOIN users u ON u.id=al.user_id WHERE al.id=%s",
|
||
(analysis_id,),
|
||
)
|
||
row = cur.fetchone()
|
||
cur.close()
|
||
return row
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Bid Tracker ──────────────────────────────────────────────────────────────
|
||
|
||
BID_STATUSES = ("open", "monitoring", "awarded", "no_bid", "cancelled")
|
||
|
||
|
||
def get_all_bids(status_filter: str = "", limit: int = 50, offset: int = 0) -> list:
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
parts = [
|
||
"SELECT b.*, u.username AS added_by_username, COUNT(bu.id) AS update_count"
|
||
" FROM bid_tracker b"
|
||
" LEFT JOIN users u ON u.id=b.added_by"
|
||
" LEFT JOIN bid_updates bu ON bu.bid_id=b.id"
|
||
]
|
||
params = []
|
||
if status_filter:
|
||
parts.append("WHERE b.status=%s")
|
||
params.append(status_filter)
|
||
parts.append(
|
||
"GROUP BY b.id ORDER BY b.due_date IS NULL, b.due_date ASC, b.created_at DESC"
|
||
" LIMIT %s OFFSET %s"
|
||
)
|
||
params.extend([limit, offset])
|
||
cur.execute(" ".join(parts), params)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_bid(bid_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"SELECT b.*, u.username AS added_by_username FROM bid_tracker b "
|
||
"LEFT JOIN users u ON u.id=b.added_by WHERE b.id=%s",
|
||
(bid_id,),
|
||
)
|
||
row = cur.fetchone()
|
||
cur.close()
|
||
return row
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def create_bid(user_id: int, title: str, url: str, source: str,
|
||
solicitation_number: str, status: str, due_date, notes: str) -> int:
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"INSERT INTO bid_tracker (title, url, source, solicitation_number, status, due_date, notes, added_by) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s)",
|
||
(title, url, source or None, solicitation_number or None,
|
||
status, due_date or None, notes or None, user_id),
|
||
)
|
||
conn.commit()
|
||
new_id = cur.lastrowid
|
||
cur.close()
|
||
log_action(user_id, "CREATE_BID", "bid_tracker", new_id, f"Created bid '{title}' status={status}.")
|
||
logger.info(f"Bid id={new_id} '{title}' created by user_id={user_id}.")
|
||
return new_id
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def update_bid(user_id: int, bid_id: int, title: str, url: str,
|
||
source: str, solicitation_number: str, status: str, due_date, notes: str):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"UPDATE bid_tracker SET title=%s, url=%s, source=%s, solicitation_number=%s, "
|
||
"status=%s, due_date=%s, notes=%s WHERE id=%s",
|
||
(title, url, source or None, solicitation_number or None,
|
||
status, due_date or None, notes or None, bid_id),
|
||
)
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(user_id, "UPDATE_BID", "bid_tracker", bid_id, f"Updated bid '{title}' status={status}.")
|
||
logger.info(f"Bid id={bid_id} updated by user_id={user_id}.")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def delete_bid(user_id: int, bid_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute("SELECT title FROM bid_tracker WHERE id=%s", (bid_id,))
|
||
row = cur.fetchone()
|
||
title = row["title"] if row else str(bid_id)
|
||
cur.execute("DELETE FROM bid_tracker WHERE id=%s", (bid_id,))
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(user_id, "DELETE_BID", "bid_tracker", bid_id, f"Deleted bid '{title}'.")
|
||
logger.info(f"Bid id={bid_id} '{title}' deleted by user_id={user_id}.")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_bid_updates(bid_id: int) -> list:
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"SELECT bu.*, u.username AS posted_by_username, "
|
||
"COALESCE(u.full_name, u.username) AS posted_by_full_name "
|
||
"FROM bid_updates bu LEFT JOIN users u ON u.id=bu.user_id "
|
||
"WHERE bu.bid_id=%s ORDER BY bu.created_at DESC",
|
||
(bid_id,),
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def add_bid_update(user_id: int, bid_id: int, content: str) -> int:
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"INSERT INTO bid_updates (bid_id, user_id, content) VALUES (%s,%s,%s)",
|
||
(bid_id, user_id, content),
|
||
)
|
||
conn.commit()
|
||
new_id = cur.lastrowid
|
||
cur.close()
|
||
log_action(user_id, "ADD_BID_UPDATE", "bid_updates", new_id, f"Posted update on bid_id={bid_id}.")
|
||
logger.info(f"Bid update id={new_id} posted on bid_id={bid_id} by user_id={user_id}.")
|
||
return new_id
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def delete_bid_update(user_id: int, update_id: int):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute("DELETE FROM bid_updates WHERE id=%s", (update_id,))
|
||
conn.commit()
|
||
cur.close()
|
||
log_action(user_id, "DELETE_BID_UPDATE", "bid_updates", update_id,
|
||
f"Deleted bid update id={update_id}.")
|
||
logger.info(f"Bid update id={update_id} deleted by user_id={user_id}.")
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Application Log ──────────────────────────────────────────────────────────
|
||
|
||
def get_app_log(limit: int = 500, level_filter: str = "", search: str = "") -> list:
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
conditions, params = [], []
|
||
if level_filter:
|
||
conditions.append("level=%s"); params.append(level_filter)
|
||
if search:
|
||
conditions.append("(logger_name LIKE %s OR message LIKE %s)")
|
||
like = f"%{search}%"
|
||
params.extend([like, like])
|
||
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
|
||
params.append(limit)
|
||
cur.execute(
|
||
f"SELECT id, level, logger_name, message, logged_at FROM app_log "
|
||
f"{where} ORDER BY logged_at DESC LIMIT %s",
|
||
params,
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def purge_app_log(older_than_days: int = 30):
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute("DELETE FROM app_log WHERE logged_at < NOW() - INTERVAL %s DAY", (older_than_days,))
|
||
deleted = cur.rowcount
|
||
conn.commit()
|
||
cur.close()
|
||
logger.info(f"App log purged: {deleted} records older than {older_than_days} days deleted.")
|
||
return deleted
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Missed Shifts ────────────────────────────────────────────────────────────
|
||
|
||
def get_missed_shifts_today() -> list:
|
||
"""Return shift+user pairs scheduled today where the user has checked 0 sites."""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"""
|
||
SELECT s.name AS shift_name,
|
||
u.id AS user_id, u.username,
|
||
COALESCE(u.full_name, u.username) AS full_name,
|
||
COUNT(DISTINCT sw.website_id) AS total_sites,
|
||
COUNT(DISTINCT sc.website_id) AS checked_sites
|
||
FROM shifts s
|
||
JOIN shift_users su ON su.shift_id = s.id
|
||
JOIN users u ON u.id = su.user_id AND u.is_active = 1
|
||
JOIN shift_websites sw ON sw.shift_id = s.id
|
||
LEFT JOIN shift_checks sc
|
||
ON sc.user_id = su.user_id
|
||
AND sc.website_id = sw.website_id
|
||
AND DATE(sc.checked_at) = CURDATE()
|
||
WHERE s.is_active = 1
|
||
AND LOCATE(CAST(DAYOFWEEK(CURDATE()) AS CHAR), s.days_of_week) > 0
|
||
GROUP BY s.id, u.id
|
||
HAVING total_sites > 0 AND checked_sites = 0
|
||
ORDER BY s.name, u.username
|
||
"""
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Shift Incomplete Reminders ───────────────────────────────────────────────
|
||
|
||
def get_incomplete_shift_users_near_end(window_minutes: int = 40) -> list:
|
||
"""
|
||
Return users in today's active shifts whose end_time falls within the next
|
||
`window_minutes` minutes and who still have at least one unchecked website.
|
||
|
||
Each dict has: user_id, username, full_name, email, shift_id, shift_name,
|
||
end_time, total_count, checked_count, unchecked_count,
|
||
unchecked_sites (list of {id, name, url}).
|
||
"""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"""
|
||
SELECT
|
||
u.id AS user_id,
|
||
u.username,
|
||
COALESCE(u.full_name, u.username) AS full_name,
|
||
u.email,
|
||
s.id AS shift_id,
|
||
s.name AS shift_name,
|
||
s.end_time,
|
||
COUNT(DISTINCT sw.website_id) AS total_count,
|
||
COUNT(DISTINCT sc.website_id) AS checked_count,
|
||
COUNT(DISTINCT sw.website_id) - COUNT(DISTINCT sc.website_id) AS unchecked_count
|
||
FROM shifts s
|
||
JOIN shift_users su ON su.shift_id = s.id
|
||
JOIN users u ON u.id = su.user_id AND u.is_active = 1
|
||
JOIN shift_websites sw ON sw.shift_id = s.id
|
||
JOIN websites w ON w.id = sw.website_id AND w.is_active = 1
|
||
LEFT JOIN shift_checks sc
|
||
ON sc.website_id = sw.website_id
|
||
AND sc.user_id = su.user_id
|
||
AND DATE(sc.checked_at) = CURDATE()
|
||
WHERE s.is_active = 1
|
||
AND LOCATE(CAST(DAYOFWEEK(CURDATE()) AS CHAR), s.days_of_week) > 0
|
||
AND s.end_time > CURTIME()
|
||
AND s.end_time <= ADDTIME(CURTIME(), SEC_TO_TIME(%s * 60))
|
||
AND NOT EXISTS (
|
||
SELECT 1 FROM shift_reminder_log srl
|
||
WHERE srl.user_id = su.user_id
|
||
AND srl.shift_id = s.id
|
||
AND srl.sent_date = CURDATE()
|
||
)
|
||
GROUP BY u.id, s.id
|
||
HAVING unchecked_count > 0
|
||
ORDER BY s.end_time, u.username
|
||
""",
|
||
(window_minutes,)
|
||
)
|
||
rows = cur.fetchall()
|
||
if not rows:
|
||
cur.close()
|
||
return []
|
||
|
||
# Attach the list of unchecked site names/URLs for each user+shift pair.
|
||
result = []
|
||
for row in rows:
|
||
cur.execute(
|
||
"""
|
||
SELECT w.id, w.name, w.url
|
||
FROM shift_websites sw
|
||
JOIN websites w ON w.id = sw.website_id AND w.is_active = 1
|
||
WHERE sw.shift_id = %s
|
||
AND NOT EXISTS (
|
||
SELECT 1 FROM shift_checks sc
|
||
WHERE sc.website_id = sw.website_id
|
||
AND sc.user_id = %s
|
||
AND DATE(sc.checked_at) = CURDATE()
|
||
)
|
||
ORDER BY sw.sort_order, w.name
|
||
""",
|
||
(row["shift_id"], row["user_id"])
|
||
)
|
||
result.append({**row, "unchecked_sites": cur.fetchall()})
|
||
|
||
cur.close()
|
||
return result
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def record_shift_reminder(user_id: int, shift_id: int) -> int:
|
||
"""
|
||
Atomically claim a reminder slot for this user+shift today via INSERT IGNORE.
|
||
Returns 1 if the slot was claimed (caller should send), 0 if already claimed.
|
||
"""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"INSERT IGNORE INTO shift_reminder_log (user_id, shift_id, sent_date) "
|
||
"VALUES (%s, %s, CURDATE())",
|
||
(user_id, shift_id),
|
||
)
|
||
conn.commit()
|
||
rows = cur.rowcount
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Bid Reminders ────────────────────────────────────────────────────────────
|
||
|
||
def get_bids_due_soon(days: int = 7) -> list:
|
||
"""Return open/monitoring bids with due_date within the next `days` days."""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"SELECT b.*, u.username AS added_by_username, u.full_name AS added_by_full_name"
|
||
" FROM bid_tracker b"
|
||
" LEFT JOIN users u ON u.id = b.added_by"
|
||
" WHERE b.status IN ('open','monitoring')"
|
||
" AND b.due_date IS NOT NULL"
|
||
" AND b.due_date BETWEEN CURDATE() AND DATE_ADD(CURDATE(), INTERVAL %s DAY)"
|
||
" ORDER BY b.due_date ASC",
|
||
(days,),
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return rows
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_admin_emails() -> list:
|
||
"""Return email addresses of all active admin users that have one set."""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"SELECT email FROM users"
|
||
" WHERE role='admin' AND is_active=1 AND email IS NOT NULL AND email != ''"
|
||
)
|
||
rows = cur.fetchall()
|
||
cur.close()
|
||
return [r["email"] for r in rows]
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Server-side Health Check ─────────────────────────────────────────────────
|
||
|
||
def get_website_url(website_id: int):
|
||
"""Return the URL of an active website by ID, or None if not found."""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT url FROM websites WHERE id=%s AND is_active=1", (website_id,))
|
||
row = cur.fetchone()
|
||
cur.close()
|
||
return row[0] if row else None
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
# ─── Password Reset ───────────────────────────────────────────────────────────
|
||
|
||
def get_user_by_email(email: str):
|
||
"""Return the active user matching the given email (case-insensitive), or None."""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"SELECT * FROM users WHERE LOWER(email)=%s AND is_active=1 LIMIT 1",
|
||
(email.lower(),),
|
||
)
|
||
row = cur.fetchone()
|
||
cur.close()
|
||
return row
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def create_password_reset_token(user_id: int) -> str:
|
||
"""Generate a secure 1-hour reset token; invalidates any prior token for the user."""
|
||
import secrets
|
||
token = secrets.token_urlsafe(48)
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute("DELETE FROM password_reset_tokens WHERE user_id=%s", (user_id,))
|
||
cur.execute(
|
||
"INSERT INTO password_reset_tokens (user_id, token, expires_at)"
|
||
" VALUES (%s, %s, DATE_ADD(NOW(), INTERVAL 1 HOUR))",
|
||
(user_id, token),
|
||
)
|
||
conn.commit()
|
||
cur.close()
|
||
return token
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def get_password_reset_user(token: str):
|
||
"""Return the user row for a valid, unexpired reset token, or None."""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"SELECT u.* FROM password_reset_tokens t"
|
||
" JOIN users u ON u.id = t.user_id"
|
||
" WHERE t.token=%s AND t.expires_at > NOW()",
|
||
(token,),
|
||
)
|
||
row = cur.fetchone()
|
||
cur.close()
|
||
return row
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def consume_password_reset_token(token: str, new_password: str) -> bool:
|
||
"""Validate token, update the user's password, and delete the token. Returns True on success."""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor(dictionary=True)
|
||
cur.execute(
|
||
"SELECT user_id FROM password_reset_tokens WHERE token=%s AND expires_at > NOW()",
|
||
(token,),
|
||
)
|
||
row = cur.fetchone()
|
||
if not row:
|
||
cur.close()
|
||
return False
|
||
user_id = row["user_id"]
|
||
new_hash = _hash_password(new_password)
|
||
cur = conn.cursor()
|
||
cur.execute("UPDATE users SET password=%s WHERE id=%s", (new_hash, user_id))
|
||
cur.execute("DELETE FROM password_reset_tokens WHERE token=%s", (token,))
|
||
conn.commit()
|
||
cur.close()
|
||
return True
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def purge_expired_reset_tokens():
|
||
"""Delete all expired password reset tokens."""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute("DELETE FROM password_reset_tokens WHERE expires_at < NOW()")
|
||
conn.commit()
|
||
cur.close()
|
||
finally:
|
||
if conn:
|
||
conn.close()
|
||
|
||
|
||
def delete_ai_analysis(analysis_id: int):
|
||
"""Permanently delete an AI analysis record."""
|
||
conn = None
|
||
try:
|
||
conn = get_connection()
|
||
cur = conn.cursor()
|
||
cur.execute("DELETE FROM ai_analysis_log WHERE id=%s", (analysis_id,))
|
||
conn.commit()
|
||
cur.close()
|
||
finally:
|
||
if conn:
|
||
conn.close()
|