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, embedding_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), } # Client self-scrape — if the PM confirmed/swapped a URL for the # tenant's own trip page (UrlConfirmation.vue's "Your trip page" row), # scrape and extract it once here so every competitor's build_prompt() # call below gets real ground-truth data instead of only typed # destination/duration/style strings. A side-effect failure (site # blocked, extraction error) must never fail the whole batch — logged # and left as None, which build_prompt() treats identically to a # tenant who never confirmed an own-trip URL for this run. own_trip_url = data.get('own_trip_url') context['own_trip'] = None if own_trip_url: try: context['own_trip'] = embedding_service.scrape_own_trip(tenant_id, own_trip_url, context) except Exception as e: logger.error(f'scrape_own_trip failed for tenant {tenant_id}: {str(e)}') 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, # own_trip surfaces the tenant's own extracted trip data (or None # if no website was confirmed this run) to the dashboard — before # this it was only ever consumed internally by build_prompt(), # with no way for the PM to see it was actually scraped and used. 'results': {'success': success, 'failed': failed, 'own_trip': context.get('own_trip')}, '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 )