237 lines
11 KiB
Python
237 lines
11 KiB
Python
import os
|
|
import time
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from redis import Redis
|
|
from rq import Queue
|
|
from rq.job import Job
|
|
from rq.exceptions import NoSuchJobError
|
|
|
|
from repositories.scrape_repository import scrape_repository
|
|
from repositories.competitor_repository import competitor_repository
|
|
from repositories.tenant_repository import tenant_repository
|
|
from services import analysis_service, alert_service, battlecard_service
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
REDIS_URL = os.getenv('REDIS_URL', 'redis://bettersight-redis:6379')
|
|
redis_conn = Redis.from_url(REDIS_URL)
|
|
|
|
# Three priority queues — Intelligence tier (not yet built) will use
|
|
# `high`; Analyse-tier on-demand and n8n cron jobs use `normal`;
|
|
# background/cleanup work uses `low`. Worker command:
|
|
# rq worker high normal low -u $REDIS_URL
|
|
high_q = Queue('high', connection=redis_conn)
|
|
normal_q = Queue('normal', connection=redis_conn)
|
|
low_q = Queue('low', connection=redis_conn)
|
|
|
|
JOB_SLOW_SECONDS = 300 # 5 minutes — §29 THRESHOLDS['job_slow_mins']
|
|
JOB_STUCK_SECONDS = 600 # 10 minutes — §29 THRESHOLDS['job_stuck_mins']
|
|
|
|
|
|
def enqueue_research_job(job_id, data):
|
|
"""
|
|
Enqueues a research job to the `normal` priority queue. Never runs
|
|
synchronously — every /research-style route calls this instead of
|
|
invoking run_research_job() directly.
|
|
|
|
Data flow:
|
|
job_id (the scrape_runs record id, reused as the RQ job id) + data →
|
|
normal_q.enqueue(run_research_job, job_id, data) →
|
|
RQ job id returned (== job_id, since we pass it explicitly)
|
|
"""
|
|
normal_q.enqueue(run_research_job, job_id, data, job_id=job_id, job_timeout=900)
|
|
return job_id
|
|
|
|
|
|
def cancel_research_job(job_id):
|
|
"""
|
|
PM-initiated cancellation — distinct from a job merely surviving a
|
|
frontend disconnect (CLAUDE.md §18/§21: run_research_job() runs to
|
|
completion regardless of dashboard polling state, by design). This
|
|
is the opposite: the PM explicitly asked the job to stop.
|
|
|
|
RQ's own Job.cancel() only removes a job still sitting in the
|
|
queue — it has no way to interrupt code already executing inside a
|
|
worker process. So this handles both cases:
|
|
|
|
Data flow:
|
|
job_id → fetch the RQ Job record →
|
|
if still queued (never started): job.cancel() removes it from
|
|
the queue outright, and scrape_runs.status is set to
|
|
'cancelled' directly since run_research_job() will now never
|
|
run at all →
|
|
if already running (or RQ has no record — job finished, or was
|
|
never an RQ job, e.g. cancelled twice): set
|
|
scrape_runs.cancel_requested = True regardless — a running
|
|
job checks this flag between competitors and self-terminates
|
|
with status='cancelled' at the next checkpoint (see
|
|
run_research_job()'s loop) →
|
|
returns True if the scrape_runs record exists, False if job_id
|
|
is unknown entirely
|
|
"""
|
|
run = scrape_repository.get_by_id(job_id)
|
|
if not run:
|
|
return False
|
|
|
|
try:
|
|
job = Job.fetch(job_id, connection=redis_conn)
|
|
if job.get_status() in ('queued', 'deferred', 'scheduled'):
|
|
job.cancel()
|
|
scrape_repository.update(job_id, {'status': 'cancelled', 'cancel_requested': True})
|
|
return True
|
|
except NoSuchJobError:
|
|
pass
|
|
|
|
# Already running (or RQ has no record of it any more) — set the
|
|
# cooperative flag so an in-flight run_research_job() stops itself
|
|
# at its next between-competitor checkpoint.
|
|
scrape_repository.update(job_id, {'cancel_requested': True})
|
|
return True
|
|
|
|
|
|
def run_research_job(job_id, data):
|
|
"""
|
|
The RQ job function. Loops over every requested competitor, running
|
|
the fast/refresh path decision per CLAUDE.md §7. One competitor's
|
|
failure is logged and skipped — it never aborts the rest of the
|
|
batch (§18 resilience rule). Records duration and fires Gotify
|
|
alerts on slow/stuck/failed completion (§29 Component 2). Checks
|
|
for PM-initiated cancellation (cancel_research_job()) once per
|
|
competitor and stops early with status='cancelled' if requested.
|
|
|
|
Data flow:
|
|
job_id + { tenant_id, competitors, destination, duration,
|
|
travel_style, tab_type } →
|
|
scrape_runs status → running →
|
|
per competitor: check cancel_requested → stop early if set →
|
|
analysis_service.run_competitor_analysis() →
|
|
success: append to results, competitors_done += 1 →
|
|
failure: append to results.failed, continue →
|
|
scrape_runs status → complete / failed / cancelled →
|
|
duration computed → scrape_runs.duration_seconds updated →
|
|
slow/stuck duration → Gotify alert fired (skipped if cancelled)
|
|
"""
|
|
start = time.time()
|
|
scrape_repository.update(job_id, {'status': 'running'})
|
|
|
|
tenant_id = data.get('tenant_id')
|
|
competitors_input = data.get('competitors', [])
|
|
tenant = tenant_repository.get_by_id(tenant_id)
|
|
context = {
|
|
'destination': data.get('destination'),
|
|
'duration': data.get('duration'),
|
|
'travel_style': data.get('travel_style'),
|
|
'tab_type': data.get('tab_type'),
|
|
'product_name': data.get('productName') or data.get('product_name'),
|
|
# The tenant's own company name — build_prompt() uses this instead
|
|
# of a hardcoded brand so the prompt (and its "differences from
|
|
# {company}" comments) is correct for whichever client is actually
|
|
# running the analysis. Fetched once per job, not per competitor,
|
|
# since it's the same tenant throughout the batch.
|
|
'company_name': (tenant or {}).get('name') or 'the client',
|
|
# PM-initiated override (TripIntentForm.vue's collapsed "Advanced
|
|
# options" toggle) — bypasses both cache-shortcut points in
|
|
# analysis_service.run_competitor_analysis() for this run only.
|
|
'force_refresh': data.get('force_refresh', False),
|
|
}
|
|
|
|
success, failed = [], []
|
|
done = 0
|
|
cancelled = False
|
|
|
|
for entry in competitors_input:
|
|
# Cancellation checkpoint — cancel_research_job() sets this flag
|
|
# for a job already running (it can't interrupt a scrape/extraction
|
|
# already in flight, only stop the next one from starting). Checked
|
|
# once per competitor rather than mid-competitor, since that's the
|
|
# natural unit of work here (scrape + AI extraction as one step).
|
|
if scrape_repository.get_by_id(job_id).get('cancel_requested'):
|
|
cancelled = True
|
|
break
|
|
|
|
competitor = competitor_repository.get_by_id(entry['id']) if isinstance(entry, dict) else None
|
|
if not competitor:
|
|
failed.append({'name': entry.get('name', 'Unknown') if isinstance(entry, dict) else str(entry),
|
|
'error': 'Competitor not found'})
|
|
continue
|
|
|
|
# The URL confirmed at UrlConfirmation.vue (Firecrawl /map match or
|
|
# a manual swap) travels per-competitor, not in the shared context —
|
|
# each competitor in the same batch can have a different confirmed
|
|
# page. Previously only entry['id'] was ever read here, so the
|
|
# confirmed/matched URL was silently discarded and every scrape fell
|
|
# back to the competitor's stored root domain regardless of what was
|
|
# matched or swapped — see analysis_service.run_competitor_analysis()'s
|
|
# target_url resolution for the other half of this fix.
|
|
confirmed_url = (entry.get('url') or '').strip() or None if isinstance(entry, dict) else None
|
|
competitor_context = {**context, 'confirmed_url': confirmed_url}
|
|
|
|
try:
|
|
outcome = analysis_service.run_competitor_analysis(tenant_id, competitor, competitor_context)
|
|
success.append({'competitor_id': competitor['id'], 'name': competitor['name'], **outcome})
|
|
|
|
# Battlecard regeneration fires on genuinely fresh data only —
|
|
# the fast path just re-narrates data already reflected in the
|
|
# current battlecard, and an 'unchanged' refresh (content-hash
|
|
# diff found no change — see analysis_service.detect_change())
|
|
# extracted nothing new either, so neither has anything to
|
|
# regenerate from. CLAUDE.md §15 describes this as n8n
|
|
# reacting to a scrape_runs-complete PocketBase webhook; wired
|
|
# directly here instead since jobs.py already knows the exact
|
|
# moment a competitor's data actually changed, which is more
|
|
# reliable than a webhook round trip this environment has no
|
|
# way to verify actually fires. A regeneration failure is a
|
|
# side-effect failure — never aborts the job.
|
|
if outcome.get('path') == 'refresh' and not outcome.get('unchanged'):
|
|
try:
|
|
battlecard_service.generate_battlecard(competitor['id'], tenant_id)
|
|
except Exception as e:
|
|
logger.error(f'Battlecard regeneration failed for {competitor["name"]}: {str(e)}')
|
|
except Exception as e:
|
|
logger.error(f'{competitor.get("name", entry)} failed in job {job_id}: {str(e)}')
|
|
failed.append({'competitor_id': competitor['id'], 'name': competitor['name'], 'error': str(e)})
|
|
finally:
|
|
done += 1
|
|
scrape_repository.update(job_id, {'competitors_done': done})
|
|
|
|
status = 'cancelled' if cancelled else ('complete' if success or not competitors_input else 'failed')
|
|
scrape_repository.update(job_id, {
|
|
'status': status,
|
|
'results': {'success': success, 'failed': failed},
|
|
'error_log': '; '.join(f['error'] for f in failed) if failed else '',
|
|
'completed_at': datetime.now(timezone.utc).isoformat(),
|
|
})
|
|
|
|
if status == 'failed':
|
|
alert_service.send_gotify(
|
|
title='🔴 Analysis job failed',
|
|
message=f'Job failed for tenant {tenant_id}.\nErrors: {"; ".join(f["error"] for f in failed)[:200]}',
|
|
priority=7
|
|
)
|
|
|
|
duration = round(time.time() - start)
|
|
scrape_repository.update(job_id, {'duration_seconds': duration})
|
|
|
|
# A cancellation is intentional, not an operational problem — skip
|
|
# the slow/stuck alerts, which exist to flag jobs that are running
|
|
# long for reasons nobody chose (proxy/Playwright issues, model
|
|
# rate-limiting, etc).
|
|
if cancelled:
|
|
return
|
|
|
|
if duration > JOB_STUCK_SECONDS:
|
|
alert_service.send_gotify(
|
|
title='🔴 Job likely stuck',
|
|
message=f'Job took {round(duration / 60, 1)} mins. Tenant: {tenant_id}. '
|
|
f'Possible proxy or Playwright issue.',
|
|
priority=10
|
|
)
|
|
elif duration > JOB_SLOW_SECONDS:
|
|
alert_service.send_gotify(
|
|
title='🟡 Slow job detected',
|
|
message=f'Job took {round(duration / 60, 1)} mins. Tenant: {tenant_id}.',
|
|
priority=5
|
|
)
|