File size: 2,836 Bytes
7f88bdf | 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 | import psycopg
from psycopg.rows import dict_row
import re
from pathlib import Path
def load_dotenv(path: Path) -> dict[str, str]:
values: dict[str, str] = {}
if not path.exists():
return values
for raw_line in path.read_text(encoding="utf-8").splitlines():
line = raw_line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
values[key.strip()] = value
return values
def get_database_url():
repo_root = Path(__file__).resolve().parents[1]
env_path = repo_root / ".env.local"
loaded = load_dotenv(env_path)
value = loaded.get("DATABASE_URL", "").strip()
if value.startswith(":postgresql://"):
value = value[1:]
value = re.sub(
r"(postgres(?:ql)?://[^:/?#]+:)\[([^\]]+)\](@)",
r"\1\2\3",
value,
flags=re.IGNORECASE,
)
return value
def main():
db_url = get_database_url()
tender_id = '9dc7a1df-a8b3-48fc-8660-ace47506d613'
with psycopg.connect(db_url, row_factory=dict_row) as conn:
with conn.cursor() as cur:
print("=== ALL JOBS ===")
cur.execute("""
SELECT id, status, attempt_count, max_attempts, last_error, created_at, updated_at
FROM public.processing_jobs
WHERE tender_id = %s
ORDER BY created_at DESC
""", (tender_id,))
jobs = cur.fetchall()
for j in jobs:
print(f"{j['id']}: {j['status']} (attempt {j['attempt_count']}/{j['max_attempts']})")
if j['last_error']:
print(f" Error: {j['last_error'][:300]}")
print("\n=== PROVIDER LOGS ===")
cur.execute("""
SELECT created_at, provider, model, task_name, status, error
FROM public.provider_logs
WHERE tender_id = %s
ORDER BY created_at DESC
LIMIT 10
""", (tender_id,))
logs = cur.fetchall()
for l in logs:
print(f"{l['created_at']}: {l['provider']}/{l['task_name']} - {l['status']}")
if l['error']:
print(f" Error: {l['error'][:100]}")
print("\n=== TENDER STATUS ===")
cur.execute("""
SELECT id, status, outcome, storage_path, file_size_bytes
FROM public.tenders
WHERE id = %s
""", (tender_id,))
tender = cur.fetchone()
print(f"Status: {tender['status']}")
print(f"Outcome: {tender['outcome']}")
print(f"Storage: {tender['storage_path']}")
print(f"Size: {tender['file_size_bytes']}")
if __name__ == "__main__":
main()
|