config/config.yaml

@@ -55,6 +55,9 @@ email:
   from_address: "info@99tales.net"  # Sender address
   from_name: "AlpineBits Monitor"  # Sender display name
 
+  api_tokens:
+    - "tLTI8wXF1OVEvUX7kdZRhSW3Qr5feBCz0mHo-kbnEp0"
+
   # Monitoring and alerting
   monitoring:
     # Daily report configuration

src/alpine_bits_python/api.py

@@ -15,11 +15,13 @@ import httpx
 from fastapi import APIRouter, Depends, FastAPI, HTTPException, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import HTMLResponse, Response
-from fastapi.security import HTTPBasic, HTTPBasicCredentials
+from fastapi.security import HTTPBasic, HTTPBasicCredentials, HTTPBearer, HTTPAuthorizationCredentials
 from slowapi.errors import RateLimitExceeded
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 
 from alpine_bits_python.schemas import ReservationData
+from fast_langdetect import detect
+from pydantic import BaseModel
 
 from .alpinebits_server import (
     AlpineBitsActionName,
@@ -52,6 +54,18 @@ _LOGGER = get_logger(__name__)
 
 # HTTP Basic auth for AlpineBits
 security_basic = HTTPBasic()
+# HTTP Bearer auth for API endpoints
+security_bearer = HTTPBearer()
+
+
+# Pydantic models for language detection
+class LanguageDetectionRequest(BaseModel):
+    text: str
+
+
+class LanguageDetectionResponse(BaseModel):
+    language_code: str
+    score: float
 
 
 # --- Enhanced event dispatcher with hotel-specific routing ---
@@ -375,6 +389,76 @@ async def health_check(request: Request):
     }
 
 
+@api_router.post("/detect-language", response_model=LanguageDetectionResponse)
+@limiter.limit(DEFAULT_RATE_LIMIT)
+async def detect_language(
+    request: Request,
+    data: LanguageDetectionRequest,
+    credentials: HTTPAuthorizationCredentials = Depends(security_bearer),
+):
+    """Detect language of text, restricted to Italian or German.
+
+    Requires Bearer token authentication.
+    Returns the most likely language (it or de) with confidence score.
+    """
+    # Validate bearer token
+    token = credentials.credentials
+    config = request.app.state.config
+
+    # Check if token is valid (you may want to implement proper token validation)
+    valid_tokens = config.get("api_tokens", [])
+    if not valid_tokens or token not in valid_tokens:
+        raise HTTPException(
+            status_code=401,
+            detail="Invalid authentication token",
+        )
+
+    try:
+        # Detect language with k=2 to get top 2 candidates
+        results = detect(data.text, k=2)
+
+        _LOGGER.info("Language detection results: %s", results)
+
+        # Filter for Italian (it) or German (de)
+        italian_german_results = [r for r in results if r.get('lang') in ['it', 'de']]
+
+        if italian_german_results:
+            # Return the best match between Italian and German
+            best_match = italian_german_results[0]
+            return LanguageDetectionResponse(
+                language_code=best_match['lang'],
+                score=best_match.get('score', 0.0)
+            )
+        else:
+            # If neither Italian nor German detected in top 2, check all results
+            all_results = detect(data.text, k=10)
+            italian_german_all = [r for r in all_results if r.get('lang') in ['it', 'de']]
+
+            if italian_german_all:
+                best_match = italian_german_all[0]
+                return LanguageDetectionResponse(
+                    language_code=best_match['lang'],
+                    score=best_match.get('score', 0.0)
+                )
+
+            # Default to German if no clear detection
+            _LOGGER.warning(
+                "Could not detect Italian or German in text: %s, defaulting to 'de'",
+                data.text[:100]
+            )
+            return LanguageDetectionResponse(
+                language_code='de',
+                score=0.0
+            )
+
+    except Exception as e:
+        _LOGGER.exception("Error detecting language: %s", e)
+        raise HTTPException(
+            status_code=500,
+            detail=f"Error detecting language: {str(e)}"
+        )
+
+
 # Extracted business logic for handling Wix form submissions
 async def process_wix_form_submission(request: Request, data: dict[str, Any], db):
     """Shared business logic for handling Wix form submissions (test and production)."""