nngo/WebChecker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

04/23 Enhance app functionalities

Browse Source
This commit is contained in:
nngo
2026-04-23 16:54:24 -04:00
parent 64419a445a
commit f4eea48e4d
8 changed files with 427 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
models.py
+173 -19
View File
@@ -891,8 +891,14 @@ def get_shift_report(date_from=None, date_to=None, user_id=None, website_id=None
def get_unchecked_report(target_date=None, user_id=None):
"""
Return websites that were NOT checked on target_date (default: today)
for the given user (or all users if omitted).
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
@@ -902,36 +908,70 @@ def get_unchecked_report(target_date=None, user_id=None):
user_filter = "AND u.id = %s" if user_id else ""
# Use a bind param only when a specific date is provided.
# When no date is given, embed CURDATE() directly in SQL so MySQL
# evaluates it as a function — passing "CURDATE()" as a %s bind
# parameter treats it as a literal string, not a SQL function, and
# causes the NOT EXISTS filter to match nothing (returns 0 rows).
# 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 = []
params = date_params + ([user_id] if user_id else []) + date_params
# One user_id param slot for the user_filter inside the main query
user_filter_params = [user_id] if user_id else []
# The subquery needs: dow_params, (optional user_id for shift_users join)
# Outer query needs: date_params, user_filter_params, date_params
# EXISTS(shift) subquery: dow_params + (user_id if filtering by user)
# We build params carefully to match the f-string placeholders below.
params = (
date_params # {date_expr} in SELECT
+ user_filter_params # {user_filter} AND u.id = %s
+ dow_params # DAYOFWEEK(%s) in shift EXISTS
+ (user_id and [user_id] or []) # su.user_id=%s in shift EXISTS
+ date_params # DATE(sc.checked_at) = {date_expr}
)
cur.execute(
f"""
SELECT
{date_expr} AS check_date,
{date_expr} AS check_date,
u.username,
COALESCE(u.full_name, u.username) AS full_name,
w.name AS website_name,
COALESCE(u.full_name, u.username) AS full_name,
w.name AS website_name,
w.url,
'Not Checked' AS status
FROM websites w
CROSS JOIN users u
WHERE w.is_active = 1
AND u.is_active = 1
AND u.role = 'user'
'Not Checked' AS status
FROM users u
-- Only websites the user was expected to check on this date
JOIN (
SELECT DISTINCT sw.website_id
FROM shift_websites sw
JOIN shifts s ON s.id = sw.shift_id
JOIN shift_users su ON su.shift_id = s.id
JOIN websites w2 ON w2.id = sw.website_id
WHERE s.is_active = 1
AND w2.is_active = 1
AND LOCATE(CAST({dow_expr} AS CHAR), s.days_of_week) > 0
{"AND su.user_id = %s" if user_id else "AND su.user_id = u.id"}
AND (
w2.visibility = 'all'
OR EXISTS (
SELECT 1 FROM website_users wu
WHERE wu.website_id = w2.id AND wu.user_id = su.user_id
)
)
) expected ON 1=1
JOIN websites w ON w.id = expected.website_id
WHERE u.is_active = 1
AND u.role = 'user'
{user_filter}
-- 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
@@ -1461,7 +1501,12 @@ def create_criterion(admin_id: int, title: str, description: str,
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."""
"""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()
@@ -1476,8 +1521,13 @@ def update_criterion(admin_id: int, criterion_id: int, title: str,
)
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}' active={is_active}.")
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:
@@ -1502,3 +1552,107 @@ def delete_criterion(admin_id: int, criterion_id: int):
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()