File size: 8,127 Bytes
2010013 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 | #!/usr/bin/env python3
"""
Script to queue verification jobs for tenders that have completed primary analysis.
Run this script after the primary worker completes analysis to trigger verification.
"""
import os
import sys
import json
import argparse
from datetime import datetime, timezone
from typing import List, Dict, Any
import psycopg
from psycopg.rows import dict_row
def connect_database() -> psycopg.Connection:
"""Connect to PostgreSQL database"""
database_url = os.getenv("DATABASE_URL")
if not database_url:
raise RuntimeError("DATABASE_URL environment variable is required")
return psycopg.connect(
database_url,
autocommit=False,
prepare_threshold=None,
)
def get_tenders_for_verification(conn: psycopg.Connection, limit: int = 50) -> List[Dict[str, Any]]:
"""Get tenders that are ready for verification"""
with conn.cursor(row_factory=dict_row) as cur:
cur.execute("""
SELECT
t.id,
t.title,
t.organization_id,
t.status,
t.verification_status,
e.structured_output is not null as has_analysis,
COUNT(vj.id) as pending_verification_jobs
FROM public.tenders t
LEFT JOIN public.extractions e ON e.tender_id = t.id
LEFT JOIN public.processing_jobs vj ON vj.tender_id = t.id AND vj.job_type = 'VERIFY' AND vj.status = 'QUEUED'
WHERE t.status = 'ANALYSIS_READY'
AND e.structured_output IS NOT NULL
AND (t.verification_status = 'PENDING' OR t.verification_status IS NULL)
AND vj.id IS NULL
GROUP BY t.id, t.title, t.organization_id, t.status, t.verification_status, e.structured_output is not null
ORDER BY t.updated_at DESC
LIMIT %s
""", (limit,))
return cur.fetchall()
def queue_verification_job(conn: psycopg.Connection, tender_id: str) -> bool:
"""Queue a verification job for a tender"""
try:
with conn.transaction():
with conn.cursor() as cur:
# Check if verification job already exists
cur.execute("""
SELECT id FROM public.processing_jobs
WHERE tender_id = %s AND job_type = 'VERIFY' AND status IN ('QUEUED', 'RUNNING')
""", (tender_id,))
if cur.fetchone():
print(f"Verification job already exists for tender {tender_id}")
return False
# Insert verification job
cur.execute("""
INSERT INTO public.processing_jobs
(tender_id, job_type, payload, status, max_attempts, available_at, created_at, updated_at)
VALUES (%s, 'VERIFY', %s::jsonb, 'QUEUED', 3, now(), now(), now())
RETURNING id
""", (tender_id, "{}"))
job_id = cur.fetchone()[0]
# Update tender status
cur.execute("""
UPDATE public.tenders
SET verification_status = 'PROCESSING',
updated_at = now()
WHERE id = %s
""", (tender_id,))
print(f"Queued verification job {job_id} for tender {tender_id}")
return True
except Exception as e:
print(f"Failed to queue verification job for tender {tender_id}: {e}")
return False
def queue_all_verifications(limit: int = 50, organization_id: str = None) -> int:
"""Queue verification jobs for all eligible tenders"""
queued_count = 0
try:
with connect_database() as conn:
tenders = get_tenders_for_verification(conn, limit)
if organization_id:
tenders = [t for t in tenders if t["organization_id"] == organization_id]
print(f"Found {len(tenders)} tenders ready for verification")
for tender in tenders:
if queue_verification_job(conn, tender["id"]):
queued_count += 1
print(f"Successfully queued {queued_count} verification jobs")
except Exception as e:
print(f"Error queueing verification jobs: {e}")
return 0
return queued_count
def get_verification_status(conn: psycopg.Connection, tender_id: str = None) -> List[Dict[str, Any]]:
"""Get verification status for tenders"""
query = """
SELECT
t.id as tender_id,
t.title as tender_title,
t.organization_id,
t.status as tender_status,
t.verification_status,
t.verification_score,
t.last_verified_at,
j.id as job_id,
j.job_type,
j.status as job_status,
j.created_at as job_created_at,
j.locked_at as job_started_at,
j.last_error,
v.analysis is not null as has_verification,
v.comparison->>'agreement_score' as agreement_score
FROM public.tenders t
LEFT JOIN public.processing_jobs j ON j.tender_id = t.id AND j.job_type = 'VERIFY'
LEFT JOIN public.webai_verifications v ON v.tender_id = t.id
WHERE t.status = 'ANALYSIS_READY'
"""
params = []
if tender_id:
query += " AND t.id = %s"
params.append(tender_id)
query += " ORDER BY t.updated_at DESC"
with conn.cursor(row_factory=dict_row) as cur:
cur.execute(query, params)
return cur.fetchall()
def print_verification_status(tender_id: str = None, organization_id: str = None):
"""Print verification status report"""
try:
with connect_database() as conn:
results = get_verification_status(conn, tender_id)
if organization_id:
results = [r for r in results if r["organization_id"] == organization_id]
if not results:
print("No verification data found")
return
print(f"\n{'Tender ID':<36} {'Title':<40} {'Status':<12} {'Agreement':<10} {'Job Status':<12}")
print("-" * 120)
for result in results:
agreement = result["agreement_score"] or "N/A"
if agreement != "N/A":
agreement = f"{float(agreement):.2f}"
title = (result["tender_title"] or "Untitled")[:37]
print(f"{result['tender_id']:<36} {title:<40} {result['verification_status'] or 'PENDING':<12} {agreement:<10} {result['job_status'] or 'NONE':<12}")
except Exception as e:
print(f"Error getting verification status: {e}")
def main():
parser = argparse.ArgumentParser(description="Manage WebAI verification jobs")
subparsers = parser.add_subparsers(dest="command", help="Available commands")
# Queue command
queue_parser = subparsers.add_parser("queue", help="Queue verification jobs")
queue_parser.add_argument("--limit", type=int, default=50, help="Maximum number of jobs to queue")
queue_parser.add_argument("--org", type=str, help="Filter by organization ID")
# Status command
status_parser = subparsers.add_parser("status", help="Show verification status")
status_parser.add_argument("--tender", type=str, help="Specific tender ID")
status_parser.add_argument("--org", type=str, help="Filter by organization ID")
args = parser.parse_args()
if not args.command:
parser.print_help()
return
if args.command == "queue":
count = queue_all_verifications(args.limit, args.org)
print(f"\nQueued {count} verification jobs")
elif args.command == "status":
print_verification_status(args.tender, args.org)
if __name__ == "__main__":
main()
|