from fastapi import FastAPI, APIRouter, Request, HTTPException from fastapi.responses import PlainTextResponse from dotenv import load_dotenv from starlette.middleware.cors import CORSMiddleware from starlette.middleware.base import BaseHTTPMiddleware import re from motor.motor_asyncio import AsyncIOMotorClient import os import logging from pathlib import Path from pydantic import BaseModel, Field, ConfigDict from typing import List, Dict, Optional import uuid from datetime import datetime, timezone ROOT_DIR = Path(__file__).parent load_dotenv(ROOT_DIR / '.env') # MongoDB connection mongo_url = os.environ['MONGO_URL'] client = AsyncIOMotorClient(mongo_url) db = client[os.environ['DB_NAME']] # Create the main app without a prefix app = FastAPI() # Liste des bots connus à bloquer BOT_USER_AGENTS = [ 'googlebot', 'bingbot', 'slurp', 'duckduckbot', 'baiduspider', 'yandexbot', 'sogou', 'exabot', 'facebot', 'ia_archiver', 'mj12bot', 'ahrefsbot', 'semrushbot', 'dotbot', 'rogerbot', 'seznambot', 'ccbot', 'applebot', 'crawler', 'spider', 'scraper', 'wget', 'curl', 'python-requests', 'python-urllib', 'java', 'perl', 'ruby', 'httpclient', 'libwww', 'go-http-client', 'phantomjs', 'headlesschrome' ] # Middleware anti-bot class BotBlockerMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): # Permettre les routes d'API analytics (pour le tracking) if request.url.path.startswith('/api/analytics'): return await call_next(request) # Permettre robots.txt pour que les bots sachent qu'ils ne doivent pas crawler if request.url.path == '/api/robots.txt' or request.url.path == '/robots.txt': return await call_next(request) # Vérifier le User-Agent user_agent = request.headers.get('user-agent', '').lower() for bot in BOT_USER_AGENTS: if bot in user_agent: logger.warning(f"Bot blocked: {user_agent}") return PlainTextResponse( content="Access denied", status_code=403 ) return await call_next(request) # Create a router with the /api prefix api_router = APIRouter(prefix="/api") # ============== ANALYTICS MODELS ============== class TrackEventRequest(BaseModel): event_type: str # page_view, security_passed, product_click product_id: Optional[str] = None product_title: Optional[str] = None class ProductClickData(BaseModel): id: str title: str clicks: int last_click: Optional[str] = None class AnalyticsResponse(BaseModel): page_views: int security_passed: int total_clicks: int product_clicks: Dict[str, ProductClickData] start_date: str active_visitors: int class VisitorHeartbeat(BaseModel): visitor_id: str # ============== ANALYTICS ROUTES ============== @api_router.post("/analytics/track") async def track_event(request: TrackEventRequest): """Track an analytics event""" timestamp = datetime.now(timezone.utc).isoformat() # Get or create analytics document analytics = await db.analytics.find_one({"_id": "main"}) if not analytics: analytics = { "_id": "main", "page_views": 0, "security_passed": 0, "total_clicks": 0, "product_clicks": {}, "start_date": timestamp } await db.analytics.insert_one(analytics) # Update based on event type update = {} if request.event_type == "page_view": update = {"$inc": {"page_views": 1}} elif request.event_type == "security_passed": update = {"$inc": {"security_passed": 1}} elif request.event_type == "product_click" and request.product_id: # Increment product clicks product_key = f"product_clicks.{request.product_id}" # Check if product exists existing = await db.analytics.find_one({"_id": "main"}) if existing and request.product_id in existing.get("product_clicks", {}): update = { "$inc": { f"{product_key}.clicks": 1, "total_clicks": 1 }, "$set": { f"{product_key}.last_click": timestamp } } else: update = { "$set": { product_key: { "id": request.product_id, "title": request.product_title or request.product_id, "clicks": 1, "last_click": timestamp } }, "$inc": {"total_clicks": 1} } if update: await db.analytics.update_one({"_id": "main"}, update) return {"success": True, "event": request.event_type, "timestamp": timestamp} @api_router.get("/analytics", response_model=AnalyticsResponse) async def get_analytics(): """Get all analytics data""" analytics = await db.analytics.find_one({"_id": "main"}) if not analytics: analytics = { "page_views": 0, "security_passed": 0, "total_clicks": 0, "product_clicks": {}, "start_date": datetime.now(timezone.utc).isoformat() } # Count active visitors (active in last 2 minutes) two_minutes_ago = datetime.now(timezone.utc).timestamp() - 120 active_count = await db.visitors.count_documents({ "last_seen": {"$gte": two_minutes_ago} }) return AnalyticsResponse( page_views=analytics.get("page_views", 0), security_passed=analytics.get("security_passed", 0), total_clicks=analytics.get("total_clicks", 0), product_clicks={ k: ProductClickData(**v) for k, v in analytics.get("product_clicks", {}).items() }, start_date=analytics.get("start_date", datetime.now(timezone.utc).isoformat()), active_visitors=active_count ) @api_router.post("/analytics/heartbeat") async def visitor_heartbeat(request: VisitorHeartbeat): """Register or update visitor activity""" timestamp = datetime.now(timezone.utc).timestamp() await db.visitors.update_one( {"_id": request.visitor_id}, { "$set": {"last_seen": timestamp}, "$setOnInsert": {"first_seen": timestamp} }, upsert=True ) # Clean up old visitors (older than 5 minutes) five_minutes_ago = timestamp - 300 await db.visitors.delete_many({"last_seen": {"$lt": five_minutes_ago}}) return {"success": True} @api_router.post("/analytics/reset") async def reset_analytics(): """Reset all analytics data""" await db.analytics.delete_one({"_id": "main"}) await db.visitors.delete_many({}) new_analytics = { "_id": "main", "page_views": 0, "security_passed": 0, "total_clicks": 0, "product_clicks": {}, "start_date": datetime.now(timezone.utc).isoformat() } await db.analytics.insert_one(new_analytics) return {"success": True, "message": "Analytics reset"} # ============== ORIGINAL ROUTES ============== class StatusCheck(BaseModel): model_config = ConfigDict(extra="ignore") id: str = Field(default_factory=lambda: str(uuid.uuid4())) client_name: str timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) class StatusCheckCreate(BaseModel): client_name: str @api_router.get("/") async def root(): return {"message": "Hello World"} @api_router.get("/robots.txt") async def robots_txt(): """Servir robots.txt qui bloque tous les crawlers""" content = """User-agent: * Disallow: / User-agent: Googlebot Disallow: / User-agent: Bingbot Disallow: / """ return PlainTextResponse(content=content, media_type="text/plain") @api_router.post("/status", response_model=StatusCheck) async def create_status_check(input: StatusCheckCreate): status_dict = input.model_dump() status_obj = StatusCheck(**status_dict) doc = status_obj.model_dump() doc['timestamp'] = doc['timestamp'].isoformat() _ = await db.status_checks.insert_one(doc) return status_obj @api_router.get("/status", response_model=List[StatusCheck]) async def get_status_checks(): status_checks = await db.status_checks.find({}, {"_id": 0}).to_list(1000) for check in status_checks: if isinstance(check['timestamp'], str): check['timestamp'] = datetime.fromisoformat(check['timestamp']) return status_checks # Include the router in the main app app.include_router(api_router) # Ajouter le middleware anti-bot AVANT CORS app.add_middleware(BotBlockerMiddleware) app.add_middleware( CORSMiddleware, allow_credentials=True, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) @app.on_event("shutdown") async def shutdown_db_client(): client.close()