"""Plan analysis job tracking for async processing.
Tracks every plan analysis request (sync and async) with status,
metadata, and results. Large PDFs are processed asynchronously
via background threads; this module handles the CRUD operations
for the job queue.
Job lifecycle:
pending → processing → completed | failed | stale
"""
import json
import logging
import secrets
from datetime import datetime
from src.db import BACKEND, execute_write, get_connection, query, query_one
logger = logging.getLogger(__name__)
def create_job(
*,
user_id: int | None = None,
filename: str,
file_size_mb: float,
pdf_data: bytes | None = None,
property_address: str | None = None,
permit_number: str | None = None,
project_description: str | None = None,
permit_type: str | None = None,
is_addendum: bool = False,
quick_check: bool = False,
is_async: bool = False,
analysis_mode: str = "sample",
submission_stage: str | None = None,
) -> str:
"""Create a new plan analysis job.
Args:
user_id: Logged-in user (None for anonymous)
filename: Original PDF filename
file_size_mb: File size in MB
pdf_data: Raw PDF bytes (stored for async processing, cleared after)
property_address: Manual property address from user
permit_number: Manual permit number from user
project_description: User-provided project description
permit_type: User-selected permit type
is_addendum: Whether this is a site permit addendum
quick_check: Whether this is metadata-only (no vision)
is_async: Whether this job runs in background
analysis_mode: 'sample' (free tier) or 'full' (pro tier, all pages)
Returns:
job_id (str): Unique job identifier
"""
job_id = secrets.token_urlsafe(16)
# Phase D2 — compute SHA-256 hash of PDF bytes at upload time (Layer 1)
pdf_hash: str | None = None
pdf_hash_failed: bool = False
if pdf_data:
from web.plan_fingerprint import compute_pdf_hash
pdf_hash = compute_pdf_hash(pdf_data)
if pdf_hash is None:
pdf_hash_failed = True
if BACKEND == "postgres":
import psycopg2
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(
"INSERT INTO plan_analysis_jobs "
"(job_id, user_id, filename, file_size_mb, pdf_data, "
"property_address, permit_number, address_source, permit_source, "
"project_description, permit_type, is_addendum, quick_check, is_async, "
"analysis_mode, submission_stage, pdf_hash, pdf_hash_failed) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
(
job_id,
user_id,
filename,
file_size_mb,
psycopg2.Binary(pdf_data) if pdf_data else None,
property_address,
permit_number,
"manual" if property_address else None,
"manual" if permit_number else None,
project_description,
permit_type,
is_addendum,
quick_check,
is_async,
analysis_mode,
submission_stage,
pdf_hash,
pdf_hash_failed,
),
)
conn.commit()
finally:
conn.close()
else:
execute_write(
"INSERT INTO plan_analysis_jobs "
"(job_id, user_id, filename, file_size_mb, pdf_data, "
"property_address, permit_number, address_source, permit_source, "
"project_description, permit_type, is_addendum, quick_check, is_async, "
"analysis_mode, submission_stage, pdf_hash, pdf_hash_failed) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
job_id,
user_id,
filename,
file_size_mb,
pdf_data,
property_address,
permit_number,
"manual" if property_address else None,
"manual" if permit_number else None,
project_description,
permit_type,
is_addendum,
quick_check,
is_async,
analysis_mode,
submission_stage,
pdf_hash,
pdf_hash_failed,
),
)
logger.info(
f"Created plan job {job_id}: {filename} ({file_size_mb:.1f} MB, "
f"async={is_async}, user={user_id}, pdf_hash={'ok' if pdf_hash else ('failed' if pdf_hash_failed else 'none')})"
)
return job_id
def get_job(job_id: str) -> dict | None:
"""Get job metadata (without pdf_data to avoid loading large blobs).
Returns:
dict with job fields, or None if not found
"""
row = query_one(
"SELECT job_id, user_id, session_id, filename, file_size_mb, "
"status, is_async, project_description, permit_type, "
"is_addendum, quick_check, report_md, error_message, "
"property_address, permit_number, address_source, permit_source, "
"created_at, started_at, completed_at, email_sent, "
"progress_stage, progress_detail, "
"vision_usage_json, gallery_duration_ms, "
"analysis_mode, pages_analyzed, "
"submission_stage, "
"structural_fingerprint, version_group, version_number, parent_job_id "
"FROM plan_analysis_jobs WHERE job_id = %s",
(job_id,),
)
if not row:
return None
return {
"job_id": row[0],
"user_id": row[1],
"session_id": row[2],
"filename": row[3],
"file_size_mb": row[4],
"status": row[5],
"is_async": row[6],
"project_description": row[7],
"permit_type": row[8],
"is_addendum": row[9],
"quick_check": row[10],
"report_md": row[11],
"error_message": row[12],
"property_address": row[13],
"permit_number": row[14],
"address_source": row[15],
"permit_source": row[16],
"created_at": row[17],
"started_at": row[18],
"completed_at": row[19],
"email_sent": row[20],
"progress_stage": row[21],
"progress_detail": row[22],
"vision_usage_json": row[23],
"gallery_duration_ms": row[24],
"analysis_mode": row[25],
"pages_analyzed": row[26],
"submission_stage": row[27] if len(row) > 27 else None,
"structural_fingerprint": row[28] if len(row) > 28 else None,
"version_group": row[29] if len(row) > 29 else None,
"version_number": row[30] if len(row) > 30 else None,
"parent_job_id": row[31] if len(row) > 31 else None,
}
def get_job_pdf(job_id: str) -> bytes | None:
"""Get the stored PDF bytes for a job (used by background worker).
Returns:
Raw PDF bytes, or None if not found or already cleared
"""
row = query_one(
"SELECT pdf_data FROM plan_analysis_jobs WHERE job_id = %s",
(job_id,),
)
if not row or not row[0]:
return None
# PostgreSQL returns memoryview for BYTEA; convert to bytes
data = row[0]
return bytes(data) if isinstance(data, memoryview) else data
def update_job_status(job_id: str, status: str, **fields) -> None:
"""Update job status and optional additional fields.
Args:
job_id: Job identifier
status: New status (pending, processing, completed, failed, stale)
**fields: Additional columns to update (e.g., session_id, report_md,
error_message, started_at, completed_at, email_sent,
property_address, permit_number, address_source, permit_source)
"""
set_parts = ["status = %s"]
params = [status]
for col, val in fields.items():
set_parts.append(f"{col} = %s")
params.append(val)
params.append(job_id)
sql = f"UPDATE plan_analysis_jobs SET {', '.join(set_parts)} WHERE job_id = %s"
execute_write(sql, tuple(params))
def get_user_jobs(
user_id: int,
limit: int = 20,
order_by: str = "newest",
include_archived: bool = False,
) -> list[dict]:
"""Get recent jobs for a user with configurable sort order.
Args:
user_id: User identifier
limit: Max results
order_by: Sort order — one of 'newest', 'oldest', 'address',
'filename', 'status'. Defaults to 'newest'.
include_archived: If True, include archived/closed jobs. Default False.
Returns:
List of job dicts (without pdf_data or report_md for efficiency)
"""
# Validate order_by against allowed values to prevent SQL injection
ORDER_CLAUSES = {
"newest": "ORDER BY created_at DESC",
"oldest": "ORDER BY created_at ASC",
"address": "ORDER BY property_address ASC NULLS LAST, created_at DESC",
"filename": "ORDER BY filename ASC, created_at DESC",
"status": (
"ORDER BY CASE WHEN status='failed' THEN 0 "
"WHEN status='stale' THEN 1 "
"WHEN status='processing' THEN 2 "
"WHEN status='pending' THEN 3 "
"WHEN status='completed' THEN 4 "
"ELSE 5 END, created_at DESC"
),
}
order_clause = ORDER_CLAUSES.get(order_by, ORDER_CLAUSES["newest"])
archive_filter = "" if include_archived else "AND is_archived = FALSE "
try:
rows = query(
"SELECT job_id, session_id, filename, file_size_mb, status, "
"is_async, quick_check, property_address, permit_number, "
"created_at, completed_at, error_message, "
"analysis_mode, pages_analyzed, started_at, is_archived, "
"parent_job_id, version_group "
"FROM plan_analysis_jobs "
f"WHERE user_id = %s {archive_filter}"
f"{order_clause} "
"LIMIT %s",
(user_id, limit),
)
except Exception:
# is_archived column not yet migrated — fall back to pre-D1 query
logger.warning("is_archived column missing — falling back to unfiltered query")
rows_raw = query(
"SELECT job_id, session_id, filename, file_size_mb, status, "
"is_async, quick_check, property_address, permit_number, "
"created_at, completed_at, error_message, "
"analysis_mode, pages_analyzed, started_at "
"FROM plan_analysis_jobs "
f"WHERE user_id = %s "
f"{order_clause} "
"LIMIT %s",
(user_id, limit),
)
return [
{
"job_id": r[0],
"session_id": r[1],
"filename": r[2],
"file_size_mb": r[3],
"status": r[4],
"is_async": r[5],
"quick_check": r[6],
"property_address": r[7],
"permit_number": r[8],
"created_at": r[9],
"completed_at": r[10],
"error_message": r[11],
"analysis_mode": r[12],
"pages_analyzed": r[13],
"started_at": r[14],
"is_archived": False,
"parent_job_id": None,
}
for r in rows_raw
]
return [
{
"job_id": r[0],
"session_id": r[1],
"filename": r[2],
"file_size_mb": r[3],
"status": r[4],
"is_async": r[5],
"quick_check": r[6],
"property_address": r[7],
"permit_number": r[8],
"created_at": r[9],
"completed_at": r[10],
"error_message": r[11],
"analysis_mode": r[12],
"pages_analyzed": r[13],
"started_at": r[14],
"is_archived": r[15],
"parent_job_id": r[16] if len(r) > 16 else None,
}
for r in rows
]
def search_jobs(user_id: int, query_text: str, limit: int = 20) -> list[dict]:
"""Search user's jobs by address, permit number, or filename.
Args:
user_id: User identifier
query_text: Search text (matched against address, permit, filename)
limit: Max results
Returns:
List of matching job dicts
"""
pattern = f"%{query_text}%"
rows = query(
"SELECT job_id, session_id, filename, file_size_mb, status, "
"is_async, quick_check, property_address, permit_number, "
"created_at, completed_at, error_message, "
"analysis_mode, pages_analyzed, started_at, "
"is_archived, parent_job_id "
"FROM plan_analysis_jobs "
"WHERE user_id = %s AND is_archived = FALSE "
"AND (property_address ILIKE %s OR permit_number ILIKE %s OR filename ILIKE %s) "
"ORDER BY created_at DESC "
"LIMIT %s",
(user_id, pattern, pattern, pattern, limit),
)
return [
{
"job_id": r[0],
"session_id": r[1],
"filename": r[2],
"file_size_mb": r[3],
"status": r[4],
"is_async": r[5],
"quick_check": r[6],
"property_address": r[7],
"permit_number": r[8],
"created_at": r[9],
"completed_at": r[10],
"error_message": r[11],
"analysis_mode": r[12],
"pages_analyzed": r[13],
"started_at": r[14],
"is_archived": r[15] if len(r) > 15 else False,
"parent_job_id": r[16] if len(r) > 16 else None,
}
for r in rows
]
def find_previous_analyses(
*,
job_id: str | None = None,
property_address: str | None = None,
permit_number: str | None = None,
user_id: int | None = None,
limit: int = 5,
) -> list[dict]:
"""Find previous completed analyses for the same address or permit.
Used for revision tracking — when a user re-uploads plans for the same
property, show links to their prior analyses.
Matches on:
1. Same permit_number (strongest match)
2. Same property_address (weaker match, addresses may vary in format)
Excludes the current job_id if provided.
Returns list of job dicts ordered by completed_at DESC.
"""
if not property_address and not permit_number:
return []
conditions = []
params: list = []
if permit_number:
conditions.append("permit_number = %s")
params.append(permit_number)
elif property_address:
conditions.append("UPPER(property_address) = UPPER(%s)")
params.append(property_address)
# Only completed analyses with results
conditions.append("status = 'completed'")
conditions.append("session_id IS NOT NULL")
# Exclude current job
if job_id:
conditions.append("job_id != %s")
params.append(job_id)
# Optionally scope to same user
if user_id:
conditions.append("user_id = %s")
params.append(user_id)
where = " AND ".join(conditions)
params.append(limit)
try:
rows = query(
f"SELECT job_id, session_id, filename, file_size_mb, "
f"property_address, permit_number, analysis_mode, pages_analyzed, "
f"created_at, completed_at "
f"FROM plan_analysis_jobs "
f"WHERE {where} "
f"ORDER BY completed_at DESC "
f"LIMIT %s",
tuple(params),
)
except Exception:
logger.debug("find_previous_analyses failed", exc_info=True)
return []
return [
{
"job_id": r[0],
"session_id": r[1],
"filename": r[2],
"file_size_mb": r[3],
"property_address": r[4],
"permit_number": r[5],
"analysis_mode": r[6],
"pages_analyzed": r[7],
"created_at": r[8],
"completed_at": r[9],
}
for r in rows
]
def mark_stale_jobs(max_age_minutes: int = 15) -> int:
"""Mark jobs stuck in 'processing' as 'stale'.
Called on startup to handle jobs interrupted by worker restart.
Args:
max_age_minutes: How long a job can be in 'processing' before marked stale
Returns:
Number of jobs marked stale
"""
if BACKEND == "postgres":
row = query_one(
"SELECT COUNT(*) FROM plan_analysis_jobs "
"WHERE status = 'processing' "
"AND started_at < NOW() - INTERVAL '%s minutes'",
(max_age_minutes,),
)
else:
row = query_one(
"SELECT COUNT(*) FROM plan_analysis_jobs "
"WHERE status = 'processing' "
"AND started_at < CURRENT_TIMESTAMP - INTERVAL '"
+ str(max_age_minutes)
+ " minutes'"
)
count = row[0] if row else 0
if count > 0:
if BACKEND == "postgres":
execute_write(
"UPDATE plan_analysis_jobs SET status = 'stale' "
"WHERE status = 'processing' "
"AND started_at < NOW() - INTERVAL '%s minutes'",
(max_age_minutes,),
)
else:
execute_write(
"UPDATE plan_analysis_jobs SET status = 'stale' "
"WHERE status = 'processing' "
"AND started_at < CURRENT_TIMESTAMP - INTERVAL '"
+ str(max_age_minutes)
+ " minutes'"
)
logger.info(f"Marked {count} stale plan analysis jobs (>{max_age_minutes}m old)")
return count
def close_project(job_ids: list[str], user_id: int) -> int:
"""Archive (close) one or more jobs belonging to a user.
Sets is_archived=TRUE. Idempotent — already-closed jobs are a no-op.
Args:
job_ids: List of job IDs to close
user_id: Owner user ID (ensures users can only close their own jobs)
Returns:
Number of rows updated
"""
if not job_ids:
return 0
placeholders = ", ".join(["%s"] * len(job_ids))
execute_write(
f"UPDATE plan_analysis_jobs SET is_archived = TRUE "
f"WHERE user_id = %s AND job_id IN ({placeholders})",
(user_id, *job_ids),
)
logger.info("Closed %d plan job(s) for user %d", len(job_ids), user_id)
return len(job_ids)
def reopen_project(job_ids: list[str], user_id: int) -> int:
"""Re-open (unarchive) one or more jobs belonging to a user.
Sets is_archived=FALSE. Idempotent — already-open jobs are a no-op.
Args:
job_ids: List of job IDs to reopen
user_id: Owner user ID (ensures users can only reopen their own jobs)
Returns:
Number of rows updated
"""
if not job_ids:
return 0
placeholders = ", ".join(["%s"] * len(job_ids))
execute_write(
f"UPDATE plan_analysis_jobs SET is_archived = FALSE "
f"WHERE user_id = %s AND job_id IN ({placeholders})",
(user_id, *job_ids),
)
logger.info("Reopened %d plan job(s) for user %d", len(job_ids), user_id)
return len(job_ids)
def _get_protected_version_groups(days: int) -> set[str]:
"""Return version_group values that contain any job created within `days` days.
These groups must not be cleaned up even if some members are older than the
threshold. Returns an empty set if the version_group column does not yet
exist (pre-D2 databases).
"""
try:
if BACKEND == "postgres":
rows = query(
"SELECT DISTINCT version_group FROM plan_analysis_jobs "
"WHERE version_group IS NOT NULL "
"AND created_at >= NOW() - INTERVAL '%s days'",
(days,),
)
else:
rows = query(
"SELECT DISTINCT version_group FROM plan_analysis_jobs "
"WHERE version_group IS NOT NULL "
"AND created_at >= CURRENT_TIMESTAMP - INTERVAL '"
+ str(days)
+ " days'"
)
return {r[0] for r in rows if r[0]}
except Exception:
# Column doesn't exist yet — safe to skip the guard
return set()
def cleanup_old_jobs(days: int = 30) -> int:
"""Delete jobs older than N days. Returns count deleted.
Jobs that belong to a version_group where any member was created within
`days` days are skipped entirely — closing a project shouldn't prematurely
delete shared-group history.
Args:
days: Age threshold in days (default 30)
Returns:
Number of jobs deleted
"""
protected_groups = _get_protected_version_groups(days)
if BACKEND == "postgres":
if protected_groups:
placeholders = ", ".join(["%s"] * len(protected_groups))
row = query_one(
"SELECT COUNT(*) FROM plan_analysis_jobs "
f"WHERE created_at < NOW() - INTERVAL '%s days' "
f"AND (version_group IS NULL OR version_group NOT IN ({placeholders}))",
(days, *protected_groups),
)
count = row[0] if row else 0
if count > 0:
execute_write(
"DELETE FROM plan_analysis_jobs "
f"WHERE created_at < NOW() - INTERVAL '%s days' "
f"AND (version_group IS NULL OR version_group NOT IN ({placeholders}))",
(days, *protected_groups),
)
else:
row = query_one(
"SELECT COUNT(*) FROM plan_analysis_jobs "
"WHERE created_at < NOW() - INTERVAL '%s days'",
(days,),
)
count = row[0] if row else 0
if count > 0:
execute_write(
"DELETE FROM plan_analysis_jobs "
"WHERE created_at < NOW() - INTERVAL '%s days'",
(days,),
)
else:
age_expr = f"CURRENT_TIMESTAMP - INTERVAL '{days} days'"
if protected_groups:
placeholders = ", ".join(["%s" for _ in protected_groups])
row = query_one(
f"SELECT COUNT(*) FROM plan_analysis_jobs "
f"WHERE created_at < {age_expr} "
f"AND (version_group IS NULL OR version_group NOT IN ({placeholders}))",
tuple(protected_groups),
)
count = row[0] if row else 0
if count > 0:
execute_write(
f"DELETE FROM plan_analysis_jobs "
f"WHERE created_at < {age_expr} "
f"AND (version_group IS NULL OR version_group NOT IN ({placeholders}))",
tuple(protected_groups),
)
else:
row = query_one(
"SELECT COUNT(*) FROM plan_analysis_jobs "
f"WHERE created_at < {age_expr}"
)
count = row[0] if row else 0
if count > 0:
execute_write(
"DELETE FROM plan_analysis_jobs "
f"WHERE created_at < {age_expr}"
)
if count > 0:
logger.info(f"Cleaned up {count} old plan analysis jobs (>{days}d old)")
return count
def delete_job(job_id: str, user_id: int) -> bool:
"""Soft-delete a job belonging to a specific user (sets is_archived=TRUE).
The job can be restored within 30 seconds via restore_job().
Permanently deleted later by cleanup_old_jobs().
Args:
job_id: Job identifier
user_id: Owner user ID (ensures users can only delete their own jobs)
Returns:
True if a row was archived, False otherwise
"""
# Check ownership first
row = query_one(
"SELECT user_id FROM plan_analysis_jobs WHERE job_id = %s",
(job_id,),
)
if not row or row[0] != user_id:
return False
execute_write(
"UPDATE plan_analysis_jobs SET is_archived = TRUE WHERE job_id = %s AND user_id = %s",
(job_id, user_id),
)
logger.info(f"Soft-deleted (archived) plan job {job_id} for user {user_id}")
return True
def restore_job(job_id: str, user_id: int) -> bool:
"""Restore a soft-deleted job (undo within grace period).
Args:
job_id: Job identifier
user_id: Owner user ID
Returns:
True if restored, False otherwise
"""
row = query_one(
"SELECT user_id, is_archived FROM plan_analysis_jobs WHERE job_id = %s",
(job_id,),
)
if not row or row[0] != user_id:
return False
execute_write(
"UPDATE plan_analysis_jobs SET is_archived = FALSE WHERE job_id = %s AND user_id = %s",
(job_id, user_id),
)
logger.info(f"Restored plan job {job_id} for user {user_id}")
return True
def bulk_delete_jobs(job_ids: list[str], user_id: int) -> int:
"""Soft-delete multiple jobs owned by user. Returns count archived."""
if not job_ids:
return 0
# Build parameterized IN clause
placeholders = ", ".join(["%s"] * len(job_ids))
execute_write(
f"UPDATE plan_analysis_jobs SET is_archived = TRUE WHERE user_id = %s AND job_id IN ({placeholders})",
(user_id, *job_ids),
)
logger.info("Soft-deleted %d jobs for user %d: %s", len(job_ids), user_id, job_ids)
return len(job_ids)
def cancel_job(job_id: str, user_id: int) -> bool:
"""Cancel a running job belonging to a specific user.
Sets status to 'cancelled'. The background worker thread continues
but results will be discarded (the status check in plan_job_status
won't show completed results for cancelled jobs).
Args:
job_id: Job identifier
user_id: Owner user ID (ensures users can only cancel their own jobs)
Returns:
True if the job was cancelled, False otherwise
"""
row = query_one(
"SELECT user_id, status FROM plan_analysis_jobs WHERE job_id = %s",
(job_id,),
)
if not row:
return False
# Must be owned by user (or anonymous job with None user_id)
if row[0] is not None and row[0] != user_id:
return False
# Only cancel jobs that are still active
if row[1] not in ("pending", "processing"):
return False
execute_write(
"UPDATE plan_analysis_jobs SET status = 'cancelled' WHERE job_id = %s",
(job_id,),
)
logger.info(f"Cancelled plan job {job_id} for user {user_id}")
return True
def assign_version_group(job_id: str, group_id: str) -> None:
"""Assign a job to a version group and set its version_number.
The version_number is auto-incremented within the group: it equals
1 + the current max version_number for existing group members.
If the group has no existing members (this job is the first), version_number = 1.
Also sets parent_job_id to the most recent prior version in the group
(the job with the highest version_number before this assignment).
Args:
job_id: Job to assign
group_id: Shared version group identifier (UUID or job_id of first member)
"""
# Find the current max version number and the job_id with that version
row = query_one(
"SELECT MAX(version_number), MAX(job_id) " # placeholder for parent lookup
"FROM plan_analysis_jobs WHERE version_group = %s AND job_id != %s",
(group_id, job_id),
)
# Get the actual parent — the job with the max version_number in this group
parent_row = query_one(
"SELECT job_id FROM plan_analysis_jobs "
"WHERE version_group = %s AND job_id != %s "
"ORDER BY version_number DESC NULLS LAST "
"LIMIT 1",
(group_id, job_id),
)
max_version = row[0] if (row and row[0] is not None) else 0
next_version = int(max_version) + 1
parent_job_id = parent_row[0] if parent_row else None
execute_write(
"UPDATE plan_analysis_jobs "
"SET version_group = %s, version_number = %s, parent_job_id = %s "
"WHERE job_id = %s",
(group_id, next_version, parent_job_id, job_id),
)
logger.info(
"Assigned job %s to version_group=%s as v%d (parent=%s)",
job_id,
group_id,
next_version,
parent_job_id,
)
def get_version_chain(version_group: str) -> list[dict]:
"""Return all jobs in a version group ordered by version_number ascending.
Args:
version_group: Shared group identifier
Returns:
List of job dicts (job_id, version_number, parent_job_id, filename,
status, created_at, completed_at, property_address, permit_number)
ordered by version_number ASC.
"""
try:
rows = query(
"SELECT job_id, version_number, parent_job_id, filename, status, "
"created_at, completed_at, property_address, permit_number, "
"analysis_mode, pages_analyzed "
"FROM plan_analysis_jobs "
"WHERE version_group = %s "
"ORDER BY version_number ASC NULLS LAST",
(version_group,),
)
except Exception:
logger.debug("get_version_chain failed for group %s", version_group, exc_info=True)
return []
return [
{
"job_id": r[0],
"version_number": r[1],
"parent_job_id": r[2],
"filename": r[3],
"status": r[4],
"created_at": r[5],
"completed_at": r[6],
"property_address": r[7],
"permit_number": r[8],
"analysis_mode": r[9],
"pages_analyzed": r[10],
}
for r in rows
]
# ---------------------------------------------------------------------------
# Phase F1: Analysis Stats Banner
# ---------------------------------------------------------------------------
def get_analysis_stats(user_id: int) -> dict:
"""Compute actionable stats for the analysis history stats banner.
Returns:
{
"awaiting_resubmittal": int, # projects with unresolved issues
"new_issues": int, # new annotations found in latest scans
"last_scan_at": datetime | None, # most recent completed job
}
"""
try:
# Awaiting resubmittal: completed jobs that have comparison_json
# with summary.new > 0 or summary.resolved < total non-unchanged
# Simplified: count version groups where the latest comparison has new > 0
if BACKEND == "postgres":
row_resub = query_one(
"SELECT COUNT(DISTINCT COALESCE(version_group, job_id)) "
"FROM plan_analysis_jobs "
"WHERE user_id = %s AND is_archived = FALSE "
"AND comparison_json IS NOT NULL "
"AND comparison_json::text LIKE %s",
(user_id, '%"new": %'),
)
# Count total new issues across all latest comparisons
row_new = query_one(
"SELECT comparison_json FROM plan_analysis_jobs "
"WHERE user_id = %s AND is_archived = FALSE "
"AND comparison_json IS NOT NULL "
"ORDER BY completed_at DESC LIMIT 20",
(user_id,),
)
# Last scan
row_last = query_one(
"SELECT completed_at FROM plan_analysis_jobs "
"WHERE user_id = %s AND status = 'completed' AND is_archived = FALSE "
"ORDER BY completed_at DESC LIMIT 1",
(user_id,),
)
else:
row_resub = query_one(
"SELECT COUNT(DISTINCT COALESCE(version_group, job_id)) "
"FROM plan_analysis_jobs "
"WHERE user_id = %s AND is_archived = FALSE "
"AND comparison_json IS NOT NULL",
(user_id,),
)
row_new = None
row_last = query_one(
"SELECT completed_at FROM plan_analysis_jobs "
"WHERE user_id = %s AND status = 'completed' AND is_archived = FALSE "
"ORDER BY completed_at DESC LIMIT 1",
(user_id,),
)
# Count new issues from comparison JSONs
new_issues = 0
awaiting = 0
try:
if BACKEND == "postgres":
rows_cmp = query(
"SELECT comparison_json FROM plan_analysis_jobs "
"WHERE user_id = %s AND is_archived = FALSE "
"AND comparison_json IS NOT NULL "
"AND status = 'completed' "
"ORDER BY completed_at DESC LIMIT 50",
(user_id,),
)
seen_groups = set()
for r in rows_cmp:
raw = r[0]
try:
cmp = json.loads(raw) if isinstance(raw, str) else raw
summary = cmp.get("summary", {})
n = summary.get("new", 0)
if n > 0:
new_issues += n
# Count unique projects awaiting resubmittal
job_b_id = cmp.get("job_b_id", "")
if job_b_id not in seen_groups:
seen_groups.add(job_b_id)
awaiting += 1
except Exception:
pass
else:
awaiting = int(row_resub[0]) if (row_resub and row_resub[0]) else 0
except Exception:
awaiting = int(row_resub[0]) if (row_resub and row_resub[0]) else 0
last_scan = row_last[0] if (row_last and row_last[0]) else None
return {
"awaiting_resubmittal": awaiting,
"new_issues": new_issues,
"last_scan_at": last_scan,
}
except Exception:
logger.debug("get_analysis_stats failed for user %d", user_id, exc_info=True)
return {
"awaiting_resubmittal": 0,
"new_issues": 0,
"last_scan_at": None,
}
def clear_pdf_data(job_id: str) -> None:
"""Clear stored PDF bytes after processing to free storage.
Args:
job_id: Job identifier
"""
execute_write(
"UPDATE plan_analysis_jobs SET pdf_data = NULL WHERE job_id = %s",
(job_id,),
)
logger.debug(f"Cleared PDF data for job {job_id}")
