Spaces:
Runtime error
Runtime error
| """Celery task: execute one outbox job (review inference or product aggregation). | |
| Delivery model (see :mod:`app.models.job`): Redis messages are only wake-ups | |
| and may be duplicated or lost; the ``jobs`` row is the source of truth. Every | |
| run locks its job row, skips finished jobs, and records retries durably with | |
| exponential back-off. :mod:`app.worker.outbox` re-sends anything unfinished. | |
| ML models load lazily through :func:`get_pipeline`, so importing this module | |
| (e.g. from tests) never touches model artifacts. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| from datetime import UTC, datetime, timedelta | |
| from functools import lru_cache | |
| from typing import TYPE_CHECKING, Any, Final | |
| from sqlalchemy import text | |
| from sqlalchemy.orm import Session | |
| from app.config import settings | |
| from app.database import SessionLocal | |
| from app.models.job import Job, JobKind | |
| from app.models.product import Product | |
| from app.models.review import Review, ReviewStatus | |
| from app.worker.celery_app import celery_app | |
| if TYPE_CHECKING: | |
| from app.ml.pipeline import VERAPipeline | |
| logger = logging.getLogger(__name__) | |
| MAX_REVIEW_ATTEMPTS: Final = 4 | |
| MAX_AGGREGATE_ATTEMPTS: Final = 10 | |
| MAX_BACKOFF_SECONDS: Final = 300 | |
| #: Arbitrary constant key for the Postgres advisory lock around aggregation. | |
| AGGREGATION_LOCK_KEY: Final = 19842026 | |
| FAILED_REVIEW_MESSAGE: Final = "Review analysis failed. Please try again later." | |
| def get_pipeline() -> VERAPipeline: | |
| """Load production artifacts once per worker process. Never falls back.""" | |
| from app.ml.pipeline import VERAPipeline | |
| return VERAPipeline(settings.MODEL_DIR, settings.HF_MODEL_REPO, settings.HF_TOKEN) | |
| def _as_utc(value: datetime) -> datetime: | |
| """Normalise DB datetimes: SQLite returns naive values, Postgres aware ones.""" | |
| return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC) | |
| def backoff_seconds(attempts: int) -> int: | |
| """5 s, 10 s, 20 s, … capped at :data:`MAX_BACKOFF_SECONDS`.""" | |
| return min(MAX_BACKOFF_SECONDS, 5 * 2 ** min(attempts, 6)) | |
| # ---------------------------------------------------------------------------- | |
| # Job handlers — run inside the caller's transaction; the caller commits. | |
| # ---------------------------------------------------------------------------- | |
| def process_review(db: Session, review_id: int) -> None: | |
| """Run the pipeline for one review and store its result. | |
| The review row stays locked for the whole inference, so a crash rolls back | |
| cleanly and a duplicate delivery cannot write concurrently. The status is not | |
| committed as "processing" separately; clients see "pending" until "done". | |
| """ | |
| review = db.query(Review).filter(Review.id == review_id).with_for_update().one_or_none() | |
| if review is None or review.status in (ReviewStatus.DONE.value, ReviewStatus.FAILED.value): | |
| return | |
| result = get_pipeline().process( | |
| text=review.text, | |
| star_rating=review.star_rating, | |
| helpful_votes=review.helpful_votes, | |
| verified_purchase=review.verified_purchase, | |
| product_title=review.product_title, | |
| ) | |
| review.corrected_score = result["corrected_score"] | |
| review.interval_lower, review.interval_upper = result["interval"] | |
| review.confidence_value = result["confidence"] | |
| review.star_text_gap = result["star_text_gap"] | |
| review.review_quality = result["review_quality"] | |
| review.aspects = result["aspects"] | |
| review.flags = result["flags"] | |
| review.explanation = result["explanation"] | |
| review.error_message = None | |
| review.status = ReviewStatus.DONE.value | |
| # Reviews without an ASIN never create aggregation work (PRD weakness #3). | |
| if review.product_asin: | |
| db.add(Job(kind=JobKind.AGGREGATE.value)) | |
| def aggregate_products(db: Session, current_job_id: str | None = None) -> int: | |
| """Recompute EB aggregates and ranks for ALL products. Returns product count. | |
| The EB prior is a population quantity, so every product is refreshed under | |
| one lock (Postgres advisory transaction lock — released on commit/rollback, | |
| no TTL to expire mid-run). Other unfinished aggregate jobs visible *before* | |
| the review snapshot is read are marked finished: their reviews are already | |
| committed and therefore included, so re-running them would be pure waste. | |
| """ | |
| from app.ml.calibration.empirical_bayes import ( | |
| EmpiricalBayesAggregator, | |
| ReviewSignal, | |
| rank_products, | |
| ) | |
| if db.get_bind().dialect.name == "postgresql": | |
| db.execute(text("SELECT pg_advisory_xact_lock(:key)"), {"key": AGGREGATION_LOCK_KEY}) | |
| superseded = ( | |
| db.query(Job) | |
| .filter( | |
| Job.kind == JobKind.AGGREGATE.value, | |
| Job.finished_at.is_(None), | |
| Job.id != (current_job_id or ""), | |
| ) | |
| .with_for_update(skip_locked=True) | |
| .all() | |
| ) | |
| rows = ( | |
| db.query( | |
| Review.product_asin, | |
| Review.corrected_score, | |
| Review.review_quality, | |
| Review.star_rating, | |
| Review.aspects, | |
| ) | |
| .filter( | |
| Review.status == ReviewStatus.DONE.value, | |
| Review.product_asin.is_not(None), | |
| Review.corrected_score.is_not(None), | |
| ) | |
| .all() | |
| ) | |
| now = datetime.now(UTC) | |
| for job in superseded: | |
| job.finished_at = now | |
| signals = [ | |
| ReviewSignal( | |
| product_asin=row.product_asin, | |
| corrected_score=float(row.corrected_score), | |
| quality_weight=float(row.review_quality) if row.review_quality is not None else 0.5, | |
| star_rating=float(row.star_rating), | |
| aspects=row.aspects or {}, | |
| ) | |
| for row in rows | |
| ] | |
| if not signals: | |
| return 0 | |
| _, aggregates = EmpiricalBayesAggregator().aggregate_all(signals) | |
| for rank, asin in enumerate(rank_products(aggregates), start=1): | |
| aggregate = aggregates[asin] | |
| product = db.get(Product, asin) | |
| if product is None: # FK makes this impossible unless data was edited by hand | |
| raise RuntimeError(f"Completed review references missing product {asin!r}") | |
| product.previous_rank_position = product.rank_position | |
| product.rank_position = rank | |
| product.corrected_score = aggregate.corrected_score | |
| product.uncertainty = aggregate.uncertainty | |
| product.trust_weighted_score = aggregate.trust_weighted_score | |
| product.star_average = aggregate.star_average | |
| product.eb_shrinkage_factor = aggregate.eb_shrinkage_factor | |
| product.review_count = aggregate.review_count | |
| product.aspect_summary = aggregate.aspect_summary | |
| product.last_aggregated_at = now | |
| return len(aggregates) | |
| # ---------------------------------------------------------------------------- | |
| # Celery entry point | |
| # ---------------------------------------------------------------------------- | |
| def _record_failure(db: Session, job_id: str) -> None: | |
| """Increment attempts; schedule a retry or mark the job (and review) failed.""" | |
| job = db.query(Job).filter(Job.id == job_id).with_for_update().one_or_none() | |
| if job is None or job.finished_at is not None: | |
| return # a duplicate delivery finished it meanwhile | |
| job.attempts += 1 | |
| limit = MAX_REVIEW_ATTEMPTS if job.kind == JobKind.REVIEW.value else MAX_AGGREGATE_ATTEMPTS | |
| now = datetime.now(UTC) | |
| if job.attempts >= limit: | |
| job.finished_at = now | |
| if job.kind == JobKind.REVIEW.value and job.review_id is not None: | |
| review = db.get(Review, job.review_id) | |
| if review is not None and review.status != ReviewStatus.DONE.value: | |
| review.status = ReviewStatus.FAILED.value | |
| review.error_message = FAILED_REVIEW_MESSAGE | |
| logger.error("Job %s (%s) failed permanently after %d attempts", job.id, job.kind, limit) | |
| else: | |
| job.available_at = now + timedelta(seconds=backoff_seconds(job.attempts)) | |
| job.last_dispatched_at = None # let the outbox re-send once due | |
| db.commit() | |
| def execute_job(job_id: str) -> str: | |
| """Run one job to completion or record its failure. Returns an outcome label.""" | |
| with SessionLocal() as db: | |
| try: | |
| job = ( | |
| db.query(Job) | |
| .filter(Job.id == job_id) | |
| .with_for_update(skip_locked=True) | |
| .one_or_none() | |
| ) | |
| if job is None or job.finished_at is not None: | |
| return "ignored" # unknown, finished, or locked by a concurrent delivery | |
| if _as_utc(job.available_at) > datetime.now(UTC): | |
| return "deferred" | |
| if job.kind == JobKind.REVIEW.value and job.review_id is not None: | |
| process_review(db, job.review_id) | |
| elif job.kind == JobKind.AGGREGATE.value: | |
| aggregate_products(db, current_job_id=job.id) | |
| else: | |
| raise ValueError(f"Unknown job kind {job.kind!r}") | |
| job.finished_at = datetime.now(UTC) | |
| db.commit() | |
| return "done" | |
| except Exception: | |
| db.rollback() | |
| logger.exception("Job %s failed; recording durable retry", job_id) | |
| _record_failure(db, job_id) | |
| return "retry_or_failed" | |
| def run_job(job_id: str) -> dict[str, Any]: | |
| """Celery wrapper around :func:`execute_job`.""" | |
| return {"job_id": job_id, "status": execute_job(job_id)} | |