from celery import Celery

celery_app = Celery("pricetracker", broker="redis://localhost:6379/0")

@celery_app.task(bind=True, max_retries=3)
def scrape_prices(self, product_ids: list[int]):
    # Extração assíncrona de preços - roda no Celery Worker
    ...

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel

app = FastAPI()

class PriceAlertRequest(BaseModel):
    product_id: int
    target_price: float
    email: str

def send_price_alert_email(email: str, product_id: int, price: float):
    # Simular envio de email (NÃO use await aqui)
    print(f"Enviando alerta para {email}: Produto {product_id} atingiu ${price}")

@app.post("/alerts")
async def create_alert(alert: PriceAlertRequest, bg: BackgroundTasks):
    # Adicionar tarefa para rodar após a resposta ser enviada
    bg.add_task(send_price_alert_email, alert.email, alert.product_id, alert.target_price)
    return {"message": "Alerta criado", "product_id": alert.product_id}

# Função definida com sucesso

# app/core/celery_app.py
from celery import Celery

celery_app = Celery(
    "pricetracker",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

celery_app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_track_started=True,
    task_acks_late=True,  # Confirmar após execução, não antes
    worker_prefetch_multiplier=4,
    result_expires=3600,  # Resultados expiram após 1 hora
)

# Execução Bem-sucedida

# app/tasks/price_scraping.py
from app.core.celery_app import celery_app
import asyncio
from sqlalchemy import select

@celery_app.task(bind=True, max_retries=3, default_retry_delay=60)
def scrape_product_prices(self, product_ids: list[int]):
    """Extrair preços atuais dos produtos dados."""
    try:
        for pid in product_ids:
            # Simular extração (em produção: requisições HTTP para fontes de preços)
            price = fetch_price_from_source(pid)
            # Armazenar no banco de dados
            save_price_record(pid, price)
        return {"scraped": len(product_ids), "status": "success"}
    except Exception as exc:
        # Retentativa com backoff exponencial
        raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))

@celery_app.task(bind=True)
def bulk_scrape_all(self, total_products: int = 1000000, batch_size: int = 1000):
    """Extrair todos os produtos em lotes - padrão chord."""
    batches = [
        list(range(i, min(i + batch_size, total_products)))
        for i in range(0, total_products, batch_size)
    ]
    # Distribuir para tarefas de extração individuais
    for batch in batches:
        scrape_product_prices.delay(batch)
    return {"total_batches": len(batches), "status": "started"}

# Função definida com sucesso

from fastapi import FastAPI, Depends
from app.core.celery_app import celery_app
from app.tasks.price_scraping import scrape_product_prices, bulk_scrape_all

app = FastAPI()

@app.post("/api/v1/scrape/prices")
async def trigger_scrape(
    product_ids: list[int],
    user=Depends(require_subscription("pro")),
):
    # Disparar tarefa Celery - retorna ID da tarefa imediatamente
    task = scrape_product_prices.delay(product_ids)
    return {"task_id": task.id, "status": "pending"}

@app.post("/api/v1/scrape/bulk")
async def trigger_bulk_scrape(
    user=Depends(require_subscription("enterprise")),
):
    task = bulk_scrape_all.delay(total_products=1000000)
    return {"task_id": task.id, "status": "pending"}

# Função definida com sucesso

from celery.result import AsyncResult

@app.get("/api/v1/tasks/{task_id}")
async def get_task_status(task_id: str):
    result = AsyncResult(task_id, app=celery_app)
    
    response = {
        "task_id": task_id,
        "status": result.status,
    }
    
    if result.ready():
        if result.successful():
            response["result"] = result.result
        else:
            response["error"] = str(result.result)
    elif result.state == "PROGRESS":
        response["progress"] = result.info
    
    return response

# Função definida com sucesso

# Iniciar Celery Worker
celery -A app.core.celery_app worker --loglevel=info --concurrency=4

# Iniciar dashboard de monitoramento Flower
celery -A app.core.celery_app flower --port=5555

# Visite http://localhost:5555 para o dashboard de monitoramento

# Comando executado com sucesso

from celery import current_task

@celery_app.task(bind=True)
def scrape_with_progress(self, product_ids: list[int]):
    total = len(product_ids)
    for i, pid in enumerate(product_ids):
        # Processar cada produto
        price = fetch_price_from_source(pid)
        save_price_record(pid, price)
        
        # Relatar progresso
        self.update_state(
            state="PROGRESS",
            meta={"current": i + 1, "total": total, "percent": (i + 1) / total * 100},
        )
    return {"scraped": total}

# Função definida com sucesso