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]):
    # 非同期価格スクレイピング - 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):
    # メール送信のシミュレーション (ここではawaitを使用しない)
    print(f"Sending alert to {email}: Product {product_id} hit ${price}")

@app.post("/alerts")
async def create_alert(alert: PriceAlertRequest, bg: BackgroundTasks):
    # レスポンス送信後に実行するタスクを追加
    bg.add_task(send_price_alert_email, alert.email, alert.product_id, alert.target_price)
    return {"message": "Alert created", "product_id": alert.product_id}

# 関数定義成功

# 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,  # 実行後にACK, 実行前ではない
    worker_prefetch_multiplier=4,
    result_expires=3600,  # 結果は1時間後に期限切れ
)

# 実行成功

# 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]):
    """Scrape current prices for given products."""
    try:
        for pid in product_ids:
            # スクレイピングのシミュレーション (本番:価格ソースへのHTTPリクエスト)
            price = fetch_price_from_source(pid)
            # データベースに保存
            save_price_record(pid, price)
        return {"scraped": len(product_ids), "status": "success"}
    except Exception as exc:
        # 指数バックオフでリトライ
        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):
    """Scrape all products in batches - chord pattern."""
    batches = [
        list(range(i, min(i + batch_size, total_products)))
        for i in range(0, total_products, batch_size)
    ]
    # 個別のスクレイピングタスクにファンアウト
    for batch in batches:
        scrape_product_prices.delay(batch)
    return {"total_batches": len(batches), "status": "started"}

# 関数定義成功

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")),
):
    # Celeryタスクをトリガー - 即座にタスクIDを返す
    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"}

# 関数定義成功

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

# 関数定義成功

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

# Flowerモニタリングダッシュボードの起動
celery -A app.core.celery_app flower --port=5555

# http://localhost:5555 にアクセスしてモニタリングダッシュボードを表示

# コマンド実行成功

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):
        # 各プロダクトを処理
        price = fetch_price_from_source(pid)
        save_price_record(pid, price)
        
        # 進捗をレポート
        self.update_state(
            state="PROGRESS",
            meta={"current": i + 1, "total": total, "percent": (i + 1) / total * 100},
        )
    return {"scraped": total}

# 関数定義成功