feat: Initialize core application structure with item tracking, price history, and notification channel management.

app/alembic/versions/add_notification_channels.py

@@ -0,0 +1,44 @@
+"""add notification channels
+
+Revision ID: add_notification_channels
+Revises: merge_heads
+Create Date: 2025-11-23
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'add_notification_channels'
+down_revision = 'merge_heads'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # Create notification_channels table
+    op.create_table('notification_channels',
+        sa.Column('id', sa.Integer(), nullable=False),
+        sa.Column('name', sa.String(), nullable=False),
+        sa.Column('type', sa.String(), nullable=False),
+        sa.Column('configuration', sa.Text(), nullable=False),
+        sa.Column('is_active', sa.Boolean(), nullable=True),
+        sa.Column('created_at', sa.DateTime(), nullable=True),
+        sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_notification_channels_id'), 'notification_channels', ['id'], unique=False)
+
+    # Add notification_channel_id to items table
+    op.add_column('items', sa.Column('notification_channel_id', sa.Integer(), nullable=True))
+    op.create_foreign_key(None, 'items', 'notification_channels', ['notification_channel_id'], ['id'])
+
+
+def downgrade() -> None:
+    # Remove foreign key and column from items table
+    op.drop_constraint(None, 'items', type_='foreignkey')
+    op.drop_column('items', 'notification_channel_id')
+
+    # Drop notification_channels table
+    op.drop_index(op.f('ix_notification_channels_id'), table_name='notification_channels')
+    op.drop_table('notification_channels')

app/main.py

@@ -13,7 +13,7 @@ from sqlalchemy import text
 
 from app.database import SessionLocal, engine
 from app.limiter import limiter
-from app.routers import auth, items, jobs, openrouter, search, search_sites, settings
+from app.routers import auth, items, jobs, notifications, openrouter, search, search_sites, settings
 from app.services.scheduler_service import scheduled_refresh, scheduler
 from app.services import auth_service, search_service
 
@@ -163,7 +163,7 @@ os.makedirs("screenshots", exist_ok=True)
 app.mount("/screenshots", StaticFiles(directory="screenshots"), name="screenshots")
 
 # Routers
-for router in [items.router, settings.router, jobs.router, openrouter.router]:
+for router in [notifications.router, items.router, settings.router, jobs.router, openrouter.router]:
     app.include_router(router, prefix="/api")
 
 # Search routers (already have /api prefix)

app/models.py

@@ -51,6 +51,9 @@ class Item(Base):
 
     price_history = relationship("PriceHistory", back_populates="item", cascade="all, delete-orphan")
 
+    notification_channel_id: int | None = Column(Integer, ForeignKey("notification_channels.id"), nullable=True)  # type: ignore
+    notification_channel = relationship("NotificationChannel", back_populates="items")
+
 
 class PriceHistory(Base):
     __tablename__ = "price_history"
@@ -96,3 +99,16 @@ class SearchSite(Base):
     product_link_selector: str | None = Column(String, nullable=True)  # type: ignore  # Sélecteur CSS pour les liens produits
     created_at: datetime = Column(DateTime, default=lambda: datetime.now(UTC))  # type: ignore
     updated_at: datetime = Column(DateTime, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))  # type: ignore
+
+
+class NotificationChannel(Base):
+    __tablename__ = "notification_channels"
+
+    id: int = Column(Integer, primary_key=True, index=True)  # type: ignore
+    name: str = Column(String, nullable=False)  # type: ignore
+    type: str = Column(String, nullable=False)  # type: ignore  # "email", "discord", "mattermost"
+    configuration: str = Column(Text, nullable=False)  # type: ignore  # JSON or encrypted string
+    is_active: bool = Column(Boolean, default=True)  # type: ignore
+    created_at: datetime = Column(DateTime, default=lambda: datetime.now(UTC))  # type: ignore
+
+    items = relationship("Item", back_populates="notification_channel")

app/routers/notifications.py

@@ -0,0 +1,55 @@
+from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks
+from sqlalchemy.orm import Session
+from app import database, models, schemas
+from app.services.notification_service import NotificationService
+
+router = APIRouter(
+    prefix="/notifications",
+    tags=["notifications"],
+)
+
+@router.get("/channels", response_model=list[schemas.NotificationChannelResponse])
+def get_channels(db: Session = Depends(database.get_db)):
+    return db.query(models.NotificationChannel).all()
+
+@router.post("/channels", response_model=schemas.NotificationChannelResponse)
+def create_channel(channel: schemas.NotificationChannelCreate, db: Session = Depends(database.get_db)):
+    db_channel = models.NotificationChannel(**channel.model_dump())
+    db.add(db_channel)
+    db.commit()
+    db.refresh(db_channel)
+    return db_channel
+
+@router.put("/channels/{channel_id}", response_model=schemas.NotificationChannelResponse)
+def update_channel(channel_id: int, channel_update: schemas.NotificationChannelUpdate, db: Session = Depends(database.get_db)):
+    db_channel = db.query(models.NotificationChannel).filter(models.NotificationChannel.id == channel_id).first()
+    if not db_channel:
+        raise HTTPException(status_code=404, detail="Channel not found")
+    
+    for key, value in channel_update.model_dump(exclude_unset=True).items():
+        setattr(db_channel, key, value)
+    
+    db.commit()
+    db.refresh(db_channel)
+    return db_channel
+
+@router.delete("/channels/{channel_id}")
+def delete_channel(channel_id: int, db: Session = Depends(database.get_db)):
+    db_channel = db.query(models.NotificationChannel).filter(models.NotificationChannel.id == channel_id).first()
+    if not db_channel:
+        raise HTTPException(status_code=404, detail="Channel not found")
+    
+    db.delete(db_channel)
+    db.commit()
+    return {"ok": True}
+
+@router.post("/channels/{channel_id}/test")
+async def test_channel(channel_id: int, background_tasks: BackgroundTasks, db: Session = Depends(database.get_db)):
+    db_channel = db.query(models.NotificationChannel).filter(models.NotificationChannel.id == channel_id).first()
+    if not db_channel:
+        raise HTTPException(status_code=404, detail="Channel not found")
+    
+    # Send test notification in background
+    background_tasks.add_task(NotificationService.send_test_notification, db_channel)
+    
+    return {"message": "Test notification queued"}

app/schemas.py

@@ -12,6 +12,7 @@ class ItemCreate(BaseModel):
     category: str | None = None
     tags: str | None = None
     description: str | None = None
+    notification_channel_id: int | None = None
 
 
 class ItemResponse(ItemCreate):
@@ -116,4 +117,27 @@ class SearchProgress(BaseModel):
     completed: int
     current_site: str | None = None
     results: list[SearchResultItem] = []
+    results: list[SearchResultItem] = []
     message: str | None = None
+
+
+# === Notification Schemas ===
+
+class NotificationChannelCreate(BaseModel):
+    name: str
+    type: str  # "email", "discord", "mattermost"
+    configuration: str  # JSON string for now
+    is_active: bool = True
+
+
+class NotificationChannelUpdate(BaseModel):
+    name: str | None = None
+    type: str | None = None
+    configuration: str | None = None
+    is_active: bool | None = None
+
+
+class NotificationChannelResponse(NotificationChannelCreate):
+    id: int
+    created_at: datetime
+    model_config = ConfigDict(from_attributes=True)

app/services/item_service.py

@@ -80,6 +80,7 @@ class ItemService:
             return None, None
 
         settings = {s.key: s.value for s in db.query(models.Settings).all()}
+        channel = item.notification_channel
 
         item_data = {
             "id": item.id,
@@ -89,6 +90,7 @@ class ItemService:
             "current_price": item.current_price,
             "in_stock": item.in_stock,
             "target_price": item.target_price,
+            "notification_channel": channel,
         }
 
         config = {

app/services/notification_service.py

@@ -0,0 +1,83 @@
+import logging
+import json
+import apprise
+from app import models
+
+logger = logging.getLogger(__name__)
+
+class NotificationService:
+    @staticmethod
+    async def send_notification(channel: models.NotificationChannel, title: str, body: str):
+        """Send a notification via the specified channel using Apprise."""
+        if not channel.is_active:
+            logger.info(f"Channel {channel.name} is inactive, skipping notification.")
+            return False
+
+        apobj = apprise.Apprise()
+        
+        # Construct Apprise URL based on channel type and configuration
+        config_url = NotificationService._get_config_url(channel)
+        
+        if not config_url:
+            logger.error(f"Failed to construct Apprise URL for channel {channel.name}")
+            return False
+
+        apobj.add(config_url)
+        
+        try:
+            await apobj.async_notify(
+                body=body,
+                title=title,
+            )
+            logger.info(f"Notification sent to {channel.name}")
+            return True
+        except Exception as e:
+            logger.error(f"Failed to send notification to {channel.name}: {e}")
+            return False
+
+    @staticmethod
+    def _get_config_url(channel: models.NotificationChannel) -> str | None:
+        """Construct Apprise URL from channel configuration."""
+        try:
+            # If configuration is already a URL (e.g. for simple webhooks), use it
+            if channel.configuration.startswith("http") or "://" in channel.configuration:
+                return channel.configuration
+            
+            # If it's JSON, parse it
+            config = json.loads(channel.configuration)
+            
+            if channel.type == "email":
+                # mailto://user:password@smtp.example.com:2525
+                user = config.get("user", "")
+                password = config.get("password", "")
+                host = config.get("host", "")
+                port = config.get("port", "")
+                
+                auth = f"{user}:{password}@" if user and password else ""
+                port_str = f":{port}" if port else ""
+                
+                return f"mailto://{auth}{host}{port_str}"
+            
+            elif channel.type == "discord":
+                # discord://webhook_id/webhook_token
+                # Or just the full webhook URL which Apprise supports
+                return config.get("webhook_url")
+                
+            elif channel.type == "mattermost":
+                # mattermost://hostname/token
+                # Or the full webhook URL
+                return config.get("webhook_url")
+                
+            return None
+            
+        except Exception as e:
+            logger.error(f"Error parsing configuration for channel {channel.name}: {e}")
+            return None
+
+    @staticmethod
+    async def send_test_notification(channel: models.NotificationChannel):
+        return await NotificationService.send_notification(
+            channel,
+            title="Test Priceflow",
+            body="Ceci est une notification de test de Priceflow ! 🚀"
+        )

app/services/scheduler_service.py

@@ -8,6 +8,7 @@ from app import database, models
 from app.ai_schema import AIExtractionMetadata, AIExtractionResponse
 from app.services.ai_service import AIService
 from app.services.item_service import ItemService
+from app.services.notification_service import NotificationService
 from app.services.scraper_service import ScraperService
 
 logger = logging.getLogger(__name__)
@@ -133,6 +134,32 @@ async def process_item_check(item_id: int):
             None, _update_db_result, item_id, extraction, metadata, thresholds, screenshot_path
         )
 
+        # Check for notifications
+        if item_data["notification_channel"] and extraction.price:
+            channel = item_data["notification_channel"]
+            
+            # 1. Target Price Reached
+            if item_data["target_price"] and extraction.price <= item_data["target_price"]:
+                # Only notify if we haven't already notified for this price (or if price dropped further)
+                # For simplicity, we notify if current price <= target. 
+                # Ideally we should check if we already notified recently, but let's keep it simple.
+                if not old_price or old_price > item_data["target_price"]:
+                    await NotificationService.send_notification(
+                        channel,
+                        title=f"🎯 Prix cible atteint : {item_data['name']}",
+                        body=f"Le prix de {item_data['name']} est passé à {extraction.price}€ (Cible: {item_data['target_price']}€)\n{item_data['url']}"
+                    )
+            
+            # 2. Price Drop
+            elif old_price and extraction.price < old_price:
+                drop_percent = (old_price - extraction.price) / old_price * 100
+                if drop_percent >= 5:  # Notify only for significant drops (> 5%)
+                    await NotificationService.send_notification(
+                        channel,
+                        title=f"📉 Baisse de prix : {item_data['name']}",
+                        body=f"Le prix de {item_data['name']} a baissé de {drop_percent:.1f}% !\nNouveau prix : {extraction.price}€ (Ancien: {old_price}€)\n{item_data['url']}"
+                    )
+
 
 
     except Exception as e: