2238 lines
76 KiB
Python
2238 lines
76 KiB
Python
"""
|
|
models.py — Data-access layer for all entities.
|
|
Each public function logs its action via the activity_log table.
|
|
"""
|
|
|
|
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:
|
|
"""Hash a plaintext password with bcrypt (rounds=12). Returns a str."""
|
|
return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt(rounds=12)).decode("utf-8")
|
|
|
|
|
|
def _verify_password(password: str, stored: str) -> bool:
|
|
"""
|
|
Verify a plaintext password against a stored hash.
|
|
Supports both bcrypt hashes (current) and legacy SHA-256 hex strings
|
|
(64-char hex, no '$' prefix) so existing accounts keep working after upgrade.
|
|
Returns True on match.
|
|
"""
|
|
# Legacy SHA-256 detection: 64 hex chars, no bcrypt prefix
|
|
if len(stored) == 64 and stored.isalnum():
|
|
return hashlib.sha256(password.encode("utf-8")).hexdigest() == stored
|
|
# bcrypt
|
|
try:
|
|
return bcrypt.checkpw(password.encode("utf-8"), stored.encode("utf-8"))
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _needs_rehash(stored: str) -> bool:
|
|
"""Return True if the stored hash is a legacy SHA-256 string."""
|
|
return len(stored) == 64 and stored.isalnum()
|
|
|
|
|
|
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):
|
|
"""
|
|
Return user dict on success, None on failure.
|
|
If the stored hash is a legacy SHA-256 string, transparently rehashes it
|
|
with bcrypt on successful login so the account is silently upgraded.
|
|
"""
|
|
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)
|
|
return None
|
|
|
|
# Transparent bcrypt upgrade for legacy SHA-256 accounts
|
|
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 # lock after this many consecutive failures
|
|
LOCKOUT_MINUTES = 15 # lock duration in minutes
|
|
|
|
|
|
def check_login_allowed(username: str) -> tuple[bool, int]:
|
|
"""
|
|
Check whether the given username is permitted to attempt a login.
|
|
|
|
Returns:
|
|
(allowed: bool, seconds_remaining: int)
|
|
- allowed=True, seconds_remaining=0 => may proceed
|
|
- allowed=False, seconds_remaining>0 => account locked; wait N seconds
|
|
"""
|
|
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 # unknown username — let authenticate() handle it
|
|
|
|
locked_until = row.get("locked_until")
|
|
if locked_until:
|
|
import datetime
|
|
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):
|
|
"""
|
|
Increment failed_attempts for the user.
|
|
If MAX_FAILED_ATTEMPTS is reached, set locked_until to now + LOCKOUT_MINUTES.
|
|
Also inserts a row in login_attempts for the audit trail.
|
|
"""
|
|
conn = None
|
|
try:
|
|
import datetime
|
|
conn = get_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
cur.execute(
|
|
"SELECT id, failed_attempts FROM users WHERE username=%s",
|
|
(username,)
|
|
)
|
|
row = cur.fetchone()
|
|
|
|
# Always log the attempt regardless of whether the user exists
|
|
cur.execute(
|
|
"INSERT INTO login_attempts (username) VALUES (%s)",
|
|
(username,)
|
|
)
|
|
|
|
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} "
|
|
f"after {new_count} failed attempts."
|
|
)
|
|
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):
|
|
"""Reset failed_attempts and locked_until on successful login."""
|
|
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 ───────────────────────────────────────────────────────────
|
|
|
|
# Password strength rules
|
|
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[str]:
|
|
"""
|
|
Return a list of unmet requirement strings.
|
|
Empty list means the password passes all rules.
|
|
"""
|
|
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[bool, str]:
|
|
"""
|
|
Change a user's password after verifying the current one.
|
|
|
|
Returns:
|
|
(success: bool, message: str)
|
|
"""
|
|
# Strength check first — no DB round-trip needed
|
|
issues = validate_password_strength(new_password)
|
|
if issues:
|
|
return False, "New password does not meet requirements:\n• " + "\n• ".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 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()
|
|
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):
|
|
"""
|
|
Hard-delete a user record.
|
|
Guards:
|
|
- An admin cannot delete their own account.
|
|
- The last active admin account cannot be deleted.
|
|
Raises ValueError with a descriptive message when either guard fires.
|
|
"""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
|
|
# Guard 1: self-delete
|
|
if admin_id == user_id:
|
|
cur.close()
|
|
raise ValueError("You cannot delete your own account.")
|
|
|
|
# Guard 2: prevent removing the last active admin
|
|
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"
|
|
)
|
|
admin_count = cur.fetchone()["n"]
|
|
if admin_count <= 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_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):
|
|
"""
|
|
credentials: list of dicts with keys: username, password, label
|
|
check_type: 'daily' or 'weekly'
|
|
visibility: 'all' (all users) or 'assigned' (only website_users)
|
|
"""
|
|
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','')}' "
|
|
f"user='{cred['username']}' for website '{name}'.")
|
|
|
|
# Assign specific users if visibility='assigned'
|
|
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}' "
|
|
f"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)
|
|
|
|
# Fetch existing credentials for diff/audit
|
|
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)
|
|
)
|
|
# Replace credentials
|
|
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 ""))
|
|
|
|
# Replace assigned users
|
|
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()
|
|
|
|
# Audit: log added and removed credentials
|
|
added = new_creds - old_creds
|
|
removed = old_creds - new_creds
|
|
for user, label in added:
|
|
log_action(admin_id, "ADD_CREDENTIAL", "website_credentials", website_id,
|
|
f"Added credential label='{label}' user='{user}' "
|
|
f"for website id={website_id}.")
|
|
for user, label in removed:
|
|
log_action(admin_id, "REMOVE_CREDENTIAL", "website_credentials", website_id,
|
|
f"Removed credential label='{label}' user='{user}' "
|
|
f"from website id={website_id}.")
|
|
|
|
log_action(admin_id, "UPDATE_WEBSITE", "websites", website_id,
|
|
f"Updated website id={website_id} check_type='{check_type}' "
|
|
f"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()
|
|
# Decrypt passwords transparently (legacy plaintext passes through unchanged)
|
|
for row in rows:
|
|
row["password"] = _dec(row["password"])
|
|
return rows
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
|
|
def get_website_assigned_users(website_id: int):
|
|
"""Return users explicitly assigned to a website (visibility='assigned')."""
|
|
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):
|
|
"""
|
|
Return all websites the user must check today, with today's check status.
|
|
- Daily sites: shown every day.
|
|
- Weekly sites: shown only once per ISO week (hidden once checked this week).
|
|
Falls back to all active websites when no shifts are assigned.
|
|
"""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
|
|
# Determine if user has any active shifts today
|
|
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.website_id = w.id
|
|
AND sc.user_id = %s
|
|
AND DATE(sc.checked_at) = CURDATE()
|
|
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
|
|
)
|
|
)
|
|
AND (
|
|
w.check_type = 'daily'
|
|
OR (
|
|
w.check_type = 'weekly'
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM shift_checks sc2
|
|
WHERE sc2.website_id = w.id
|
|
AND sc2.user_id = %s
|
|
AND YEARWEEK(sc2.checked_at, 1) = YEARWEEK(CURDATE(), 1)
|
|
)
|
|
)
|
|
)
|
|
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, user_id)
|
|
)
|
|
else:
|
|
# Legacy fallback: show active websites, filtered by visibility
|
|
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.website_id = w.id
|
|
AND sc.user_id = %s
|
|
AND DATE(sc.checked_at) = CURDATE()
|
|
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
|
|
)
|
|
)
|
|
AND (
|
|
w.check_type = 'daily'
|
|
OR (
|
|
w.check_type = 'weekly'
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM shift_checks sc2
|
|
WHERE sc2.website_id = w.id
|
|
AND sc2.user_id = %s
|
|
AND YEARWEEK(sc2.checked_at, 1) = YEARWEEK(CURDATE(), 1)
|
|
)
|
|
)
|
|
)
|
|
ORDER BY w.name
|
|
""",
|
|
(user_id, 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()
|
|
# Allow only one check per site per day per user; upsert via delete+insert
|
|
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 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 = ""):
|
|
"""
|
|
Return recent activity_log entries, newest first.
|
|
The DB column is created_at; we alias it to logged_at for a consistent
|
|
key across both log tables so the view layer never needs to care.
|
|
"""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
|
|
where = ""
|
|
params: list = []
|
|
if search:
|
|
where = """
|
|
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])
|
|
params.append(limit)
|
|
|
|
cur.execute(
|
|
f"""
|
|
SELECT al.id,
|
|
al.user_id,
|
|
al.action,
|
|
al.entity,
|
|
al.entity_id,
|
|
al.detail,
|
|
al.logged_at,
|
|
u.username
|
|
FROM activity_log al
|
|
LEFT JOIN users u ON u.id = al.user_id
|
|
{where}
|
|
ORDER BY al.logged_at DESC
|
|
LIMIT %s
|
|
""",
|
|
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):
|
|
"""
|
|
Flexible shift-check report query.
|
|
All parameters are optional; omitting them returns the full dataset.
|
|
Returns list of dicts with columns:
|
|
check_date, checked_at, username, full_name,
|
|
website_name, url, user_note, status
|
|
"""
|
|
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_clause = ("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_clause}
|
|
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):
|
|
"""
|
|
Return websites that a user was EXPECTED to check on target_date but did not.
|
|
|
|
"Expected" is defined by shift membership on that day-of-week:
|
|
- Only websites assigned to a shift the user belongs to are included.
|
|
- Visibility rules (all / assigned) are respected.
|
|
- When a user has no shifts, falls back to all active visible websites.
|
|
When no target_date is supplied, defaults to today via CURDATE().
|
|
|
|
Columns: check_date, username, full_name, website_name, url, status
|
|
"""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
|
|
user_filter = "AND u.id = %s" if user_id else ""
|
|
|
|
# Embed CURDATE() directly when no date given — passing it as a %s
|
|
# bind param would treat it as a literal string, not a SQL function.
|
|
if target_date:
|
|
date_val = str(target_date)
|
|
date_expr = "%s"
|
|
date_params = [date_val]
|
|
# DAYOFWEEK for a specific date
|
|
dow_expr = "DAYOFWEEK(%s)"
|
|
dow_params = [date_val]
|
|
else:
|
|
date_expr = "CURDATE()"
|
|
date_params = []
|
|
dow_expr = "DAYOFWEEK(CURDATE())"
|
|
dow_params = []
|
|
|
|
# One user_id param slot for the user_filter inside the main query
|
|
user_filter_params = [user_id] if user_id else []
|
|
|
|
# Fix for Bug #1: MySQL does not allow a derived table (subquery in FROM/JOIN)
|
|
# to reference outer-query aliases (e.g. u.id). The previous approach used
|
|
# "AND su.user_id = u.id" inside a derived table when user_id=None, which
|
|
# MySQL rejects with "Unknown column 'u.id' in 'where clause'".
|
|
#
|
|
# Solution: replace the derived-table JOIN with an EXISTS correlated subquery
|
|
# in the WHERE clause. Correlated subqueries CAN reference outer aliases,
|
|
# so u.id is always in scope. This works identically for both the
|
|
# single-user and all-users cases.
|
|
#
|
|
# Param order:
|
|
# date_params → {date_expr} in SELECT
|
|
# user_filter_params → {user_filter} AND u.id = %s
|
|
# dow_params → DAYOFWEEK(%s) in EXISTS
|
|
# user_filter_params → su.user_id = u.id or %s in EXISTS (always u.id now)
|
|
# date_params → DATE(sc.checked_at) = {date_expr}
|
|
params = (
|
|
date_params # {date_expr} in SELECT
|
|
+ user_filter_params # {user_filter} AND u.id = %s
|
|
+ dow_params # DAYOFWEEK(%s) in shift EXISTS
|
|
+ date_params # DATE(sc.checked_at) = {date_expr}
|
|
)
|
|
|
|
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}
|
|
-- Only include websites the user was expected to check via their shifts
|
|
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
|
|
)
|
|
)
|
|
)
|
|
-- Exclude sites the user DID check on the target date
|
|
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):
|
|
"""
|
|
Per-user per-day summary: sites checked vs the sites that user was
|
|
expected to check on that specific day (shift-scoped total).
|
|
|
|
The previous implementation used a global COUNT(*) of all active websites
|
|
as the denominator, producing misleading percentages — a user in a 3-site
|
|
shift who checked all 3 would show 15% against 20 global sites.
|
|
|
|
The corrected subquery counts the distinct websites in the shifts the user
|
|
was assigned to that ran on the check_date's day-of-week. For historical
|
|
dates this still uses DAYOFWEEK(check_date) to match the shift schedule.
|
|
|
|
Columns: check_date, username, full_name, checked_count, total_sites, pct_complete
|
|
"""
|
|
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_clause = ("WHERE " + " AND ".join(conditions)) if conditions else ""
|
|
|
|
# Fix for Bug #2: MySQL only_full_group_by rejects referencing sc.checked_at
|
|
# (the full timestamp) inside correlated subqueries when only DATE(sc.checked_at)
|
|
# appears in the GROUP BY clause. Even though sc.checked_at is functionally
|
|
# determined by DATE(sc.checked_at) in intent, MySQL strict mode does not
|
|
# infer that relationship automatically.
|
|
#
|
|
# Solution: pre-aggregate in a CTE (agg) that produces a single, unambiguous
|
|
# check_date (DATE) and user_id per group. The outer SELECT then references
|
|
# agg.check_date — a fully grouped column — inside the correlated subqueries,
|
|
# satisfying only_full_group_by completely.
|
|
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_clause}
|
|
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():
|
|
"""Return (users_list, websites_list) for populating filter dropdowns."""
|
|
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():
|
|
"""Return all shifts with assigned user count and website count."""
|
|
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} "
|
|
f"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)
|
|
)
|
|
# Replace user assignments
|
|
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)
|
|
)
|
|
# Replace website assignments
|
|
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):
|
|
"""Soft-delete: mark inactive. Preserves shift_checks history."""
|
|
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):
|
|
"""
|
|
Return all active shifts assigned to a user that are scheduled for today
|
|
(day-of-week match only; time window is informational for now).
|
|
MySQL DAYOFWEEK: 1=Sunday … 7=Saturday.
|
|
We store days as a string of digits e.g. '1234567' or '23456'.
|
|
"""
|
|
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()
|
|
|
|
|
|
def get_today_checks_for_shift(user_id: int, shift_id: int):
|
|
"""
|
|
Return websites assigned to a specific shift with today's check status
|
|
for the given user.
|
|
"""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
w.id,
|
|
w.name,
|
|
w.url,
|
|
w.note AS site_note,
|
|
sw.sort_order,
|
|
sc.id AS check_id,
|
|
sc.checked_at,
|
|
sc.user_note,
|
|
sc.user_id AS checked_by_id
|
|
FROM shift_websites sw
|
|
JOIN websites w ON w.id = sw.website_id
|
|
LEFT JOIN shift_checks sc
|
|
ON sc.website_id = w.id
|
|
AND sc.user_id = %s
|
|
AND DATE(sc.checked_at) = CURDATE()
|
|
WHERE sw.shift_id = %s
|
|
AND w.is_active = 1
|
|
ORDER BY sw.sort_order, w.name
|
|
""",
|
|
(user_id, shift_id)
|
|
)
|
|
rows = cur.fetchall()
|
|
cur.close()
|
|
return rows
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
# ─── Admin Dashboard ──────────────────────────────────────────────────────────
|
|
|
|
def get_admin_dashboard_stats():
|
|
"""
|
|
Return today's completion stats for ALL active regular users.
|
|
Columns: user_id, username, full_name, checked_count, total_sites, pct_complete
|
|
Also returns total_sites and total_users as separate scalars.
|
|
"""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
|
|
# Per-user completion for today
|
|
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()
|
|
|
|
# Overall totals
|
|
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.close()
|
|
return {
|
|
"user_stats": user_stats,
|
|
"total_sites": total_sites,
|
|
"total_users": total_users,
|
|
"active_today": active_today,
|
|
}
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
|
|
def get_unchecked_sites_for_user(user_id: int):
|
|
"""Return unchecked sites for a user today (for notification system)."""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
cur.execute(
|
|
"""
|
|
SELECT w.id, w.name, s.end_time
|
|
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
|
|
WHERE s.is_active = 1
|
|
AND w.is_active = 1
|
|
AND LOCATE(CAST(DAYOFWEEK(CURDATE()) AS CHAR), s.days_of_week) > 0
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM shift_checks sc
|
|
WHERE sc.website_id = w.id
|
|
AND sc.user_id = %s
|
|
AND DATE(sc.checked_at) = CURDATE()
|
|
)
|
|
GROUP BY w.id, s.end_time
|
|
ORDER BY s.end_time, w.name
|
|
""",
|
|
(user_id, user_id)
|
|
)
|
|
rows = cur.fetchall()
|
|
cur.close()
|
|
return rows
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
# ─── AI Criteria CRUD ─────────────────────────────────────────────────────────
|
|
|
|
def get_all_criteria():
|
|
"""Return all AI evaluation criteria ordered by sort_order, then id.
|
|
|
|
The creator username is intentionally omitted — the criteria treeview does
|
|
not display it and the LEFT JOIN was adding a needless per-call cost.
|
|
If a creator column is ever added to the UI, restore the JOIN here.
|
|
"""
|
|
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():
|
|
"""Return only active criteria for use in AI prompt construction."""
|
|
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:
|
|
"""Insert a new AI evaluation criterion. Returns the new row id."""
|
|
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):
|
|
"""Update an existing AI evaluation criterion.
|
|
|
|
The full description text is included in the activity_log detail field so
|
|
there is a complete audit trail of exactly what criteria wording the AI was
|
|
evaluating against at any point in time.
|
|
"""
|
|
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()
|
|
# Include full description in detail so audit log captures wording at
|
|
# time of change — essential for reconstructing what criteria were
|
|
# active during any historical AI analysis.
|
|
log_action(admin_id, "UPDATE_AI_CRITERION", "ai_criteria", criterion_id,
|
|
f"Updated criterion id={criterion_id} '{title}' "
|
|
f"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):
|
|
"""Hard-delete an AI evaluation criterion."""
|
|
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()
|
|
|
|
|
|
# ─── AI Analysis History ──────────────────────────────────────────────────────
|
|
|
|
def save_ai_analysis(user_id: int, file_names: str, model: str,
|
|
verdict: str | None, criteria_snapshot: str | None,
|
|
summary_text: str) -> int:
|
|
"""
|
|
Persist an AI analysis result to ai_analysis_log.
|
|
verdict : 'PURSUE' | 'PASS' | 'UNCLEAR' | None
|
|
criteria_snapshot: JSON/text of active criteria at time of analysis
|
|
Returns the new row id.
|
|
"""
|
|
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} "
|
|
f"verdict={verdict} files='{file_names[:80]}'."
|
|
)
|
|
return new_id
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
|
|
def get_ai_analysis_history(user_id: int | None = None, limit: int = 100):
|
|
"""
|
|
Return recent AI analysis log entries.
|
|
When user_id is provided, filters to that user's own analyses.
|
|
Admins pass user_id=None to see all users' analyses.
|
|
Columns: id, username, file_names, model, verdict, analyzed_at
|
|
(summary_text excluded for list view — fetch by id for detail).
|
|
"""
|
|
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):
|
|
"""Return a single ai_analysis_log row including summary_text and criteria_snapshot."""
|
|
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 ──────────────────────────────────────────────────────────────
|
|
|
|
def get_all_bids(status_filter=None):
|
|
"""
|
|
Return all bids ordered by updated_at DESC.
|
|
status_filter: optional string to filter by status, e.g. 'open'.
|
|
Includes creator username and latest update timestamp.
|
|
"""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
where = ""
|
|
params = []
|
|
if status_filter and status_filter != "all":
|
|
where = "WHERE b.status = %s"
|
|
params.append(status_filter)
|
|
cur.execute(
|
|
f"""
|
|
SELECT b.*,
|
|
COALESCE(u.full_name, u.username) AS creator_name,
|
|
u.username AS creator_username,
|
|
(SELECT COUNT(*) FROM bid_updates bu WHERE bu.bid_id = b.id)
|
|
AS update_count,
|
|
(SELECT MAX(bu2.created_at) FROM bid_updates bu2
|
|
WHERE bu2.bid_id = b.id) AS last_update_at
|
|
FROM bid_tracker b
|
|
LEFT JOIN users u ON u.id = b.created_by
|
|
{where}
|
|
ORDER BY b.updated_at DESC
|
|
""",
|
|
params
|
|
)
|
|
rows = cur.fetchall()
|
|
cur.close()
|
|
return rows
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
|
|
def get_bid_by_id(bid_id: int):
|
|
"""Return a single bid row with creator info."""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
cur.execute(
|
|
"""
|
|
SELECT b.*,
|
|
COALESCE(u.full_name, u.username) AS creator_name,
|
|
u.username AS creator_username
|
|
FROM bid_tracker b
|
|
LEFT JOIN users u ON u.id = b.created_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, notes: str, status: str) -> int:
|
|
"""Insert a new bid. Returns the new row id."""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO bid_tracker (title, url, source, notes, status, created_by)
|
|
VALUES (%s, %s, %s, %s, %s, %s)
|
|
""",
|
|
(title, url, source or None, notes or None, status, 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} url={url[:80]}")
|
|
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, notes: str, status: str):
|
|
"""Update an existing bid record."""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor()
|
|
cur.execute(
|
|
"""
|
|
UPDATE bid_tracker
|
|
SET title=%s, url=%s, source=%s, notes=%s, status=%s
|
|
WHERE id=%s
|
|
""",
|
|
(title, url, source or None, notes or None, status, bid_id)
|
|
)
|
|
conn.commit()
|
|
cur.close()
|
|
log_action(user_id, "UPDATE_BID", "bid_tracker", bid_id,
|
|
f"Updated bid id={bid_id} '{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):
|
|
"""Hard-delete a bid and all its updates (CASCADE)."""
|
|
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 id={bid_id} '{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):
|
|
"""Return all updates for a bid, newest first."""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
cur.execute(
|
|
"""
|
|
SELECT bu.*,
|
|
COALESCE(u.full_name, u.username) AS author_name,
|
|
u.username AS author_username
|
|
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:
|
|
"""Add an update entry to a bid. Also touches bid_tracker.updated_at."""
|
|
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)
|
|
)
|
|
# Touch bid updated_at so it sorts to top of list
|
|
cur.execute(
|
|
"UPDATE bid_tracker SET updated_at=NOW() WHERE id=%s", (bid_id,))
|
|
conn.commit()
|
|
new_id = cur.lastrowid
|
|
cur.close()
|
|
log_action(user_id, "ADD_BID_UPDATE", "bid_updates", new_id,
|
|
f"Added update to bid id={bid_id}: {content[:100]}")
|
|
logger.info(f"Bid update id={new_id} added to bid id={bid_id} "
|
|
f"by user_id={user_id}.")
|
|
return new_id
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
|
|
def delete_bid_update(user_id: int, update_id: int):
|
|
"""Delete a single bid update entry."""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
cur.execute("SELECT bid_id, content FROM bid_updates WHERE id=%s",
|
|
(update_id,))
|
|
row = cur.fetchone()
|
|
bid_id = row["bid_id"] if row else 0
|
|
snippet = (row["content"][:60] if row else "") if row else ""
|
|
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 update id={update_id} from bid id={bid_id}: {snippet}")
|
|
logger.info(f"Bid update id={update_id} deleted by user_id={user_id}.")
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
|
|
# ─── Bid Tracker CRUD ─────────────────────────────────────────────────────────
|
|
|
|
BID_STATUSES = ("open", "monitoring", "awarded", "no_bid", "cancelled")
|
|
|
|
|
|
def get_all_bids(status_filter: str = "") -> list:
|
|
"""
|
|
Return all bids ordered by due_date (nulls last), then created_at desc.
|
|
When status_filter is given, only bids with that status are returned.
|
|
Includes the adder's username and the count of updates per bid.
|
|
"""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
where = "WHERE b.status = %s" if status_filter else ""
|
|
params = (status_filter,) if status_filter else ()
|
|
cur.execute(
|
|
f"""
|
|
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
|
|
{where}
|
|
GROUP BY b.id
|
|
ORDER BY b.due_date IS NULL, b.due_date ASC, b.created_at DESC
|
|
""",
|
|
params,
|
|
)
|
|
rows = cur.fetchall()
|
|
cur.close()
|
|
return rows
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
|
|
def get_bid(bid_id: int) -> dict | None:
|
|
"""Return a single bid row with adder username."""
|
|
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:
|
|
"""Insert a new bid. Returns new row id."""
|
|
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):
|
|
"""Update an existing bid."""
|
|
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):
|
|
"""Hard-delete a bid and all its updates (CASCADE)."""
|
|
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()
|
|
|
|
|
|
# ─── Bid Updates CRUD ─────────────────────────────────────────────────────────
|
|
|
|
def get_bid_updates(bid_id: int) -> list:
|
|
"""Return all updates for a bid, newest first."""
|
|
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:
|
|
"""Post a new update on a bid. Returns new row id."""
|
|
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):
|
|
"""Delete a single bid update. Any user can delete their own; admin can delete any."""
|
|
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 (DB-backed logging) ─────────────────────────────────────
|
|
|
|
def get_app_log(limit: int = 500, level_filter: str = "",
|
|
search: str = "") -> list:
|
|
"""
|
|
Return recent application log entries from app_log.
|
|
level_filter : one of DEBUG/INFO/WARNING/ERROR/CRITICAL — empty = all
|
|
search : substring match against logger_name or message
|
|
"""
|
|
conn = None
|
|
try:
|
|
conn = get_connection()
|
|
cur = conn.cursor(dictionary=True)
|
|
|
|
conditions = []
|
|
params: list = []
|
|
|
|
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
|
|
{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):
|
|
"""
|
|
Delete app_log entries older than older_than_days days.
|
|
Called from the admin log view's Purge button.
|
|
"""
|
|
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 "
|
|
f"{older_than_days} days deleted.")
|
|
return deleted
|
|
finally:
|
|
if conn:
|
|
conn.close() |