added API for vuforia to get analysis testing pending

routers/analysis.py

@@ -18,6 +18,7 @@ from langsmith import traceable
 from services.ingredientFinderAgent import IngredientInfoAgentLangGraph
 from services.productAnalyzerAgent import analyze_product_ingredients
 from utils.db_utils import ingredient_db_to_pydantic
+from services.analysis_service import get_product_data_by_marker_id as get_analysis_service_data
 from utils.ingredient_utils import process_single_ingredient
 
 # Load environment variables
@@ -120,3 +121,23 @@ async def process_ingredients_endpoint(product_ingredient: ProductIngredientsReq
     except Exception as e:
         log_error(f"Error in process_ingredients_endpoint: {str(e)}",e)
         raise HTTPException(status_code=500, detail="Internal Server Error")
+
+
+@router.get("/get_by_marker_id/{target_id}", response_model=ProductAnalysisResponse)
+async def get_analysis_by_marker_id(target_id: str, db: Session = Depends(get_db)):
+    """
+    Retrieves product analysis and ingredient information by marker ID.
+    """
+    log_info(f"Received request for analysis by marker ID: {target_id}")
+    try:
+        product_data = get_analysis_service_data(db, target_id)
+
+        if not product_data:
+            raise HTTPException(status_code=404, detail=f"Product not found for marker ID: {target_id}")
+
+        log_info(f"Successfully retrieved product data for marker ID: {target_id}")
+        return product_data
+
+    except Exception as e:
+        log_error(f"Error in get_analysis_by_marker_id: {str(e)}", e)
+        raise HTTPException(status_code=500, detail="Internal Server Error")

services/analysis_service.py

@@ -0,0 +1,44 @@
+from sqlalchemy.orm import Session
+from db.models import Marker, Product
+from utils.analysis_utils import format_product_analysis_response
+from utils.logger_manager import log_info, log_error
+
+def get_product_data_by_marker_id(db: Session, target_id: str):
+    """
+    Retrieves product analysis and ingredient information by marker ID.
+
+    Args:
+        db: The database session.
+        target_id: The target ID from the marker table.
+
+    Returns:
+        A dictionary containing product analysis and ingredient information,
+        or None if no product is found.
+    """
+    log_info(f"Attempting to retrieve product data for marker ID: {target_id}")
+    try:
+        # Find the marker with the given target_id
+        marker = db.query(Marker).filter(Marker.target_id == target_id).first()
+
+        if not marker:
+            log_info(f"No marker found for target ID: {target_id}")
+            return {"found": False, "message": f"No product found for marker ID: {target_id}"}
+
+        # Get the product associated with the marker
+        product = db.query(Product).filter(Product.id == marker.product_id).first()
+
+        if not product:
+            log_info(f"No product found for product_id: {marker.product_id} linked to marker ID: {target_id}")
+            return {"found": False, "message": f"No product found for marker ID: {target_id}"}
+
+        log_info(f"Product found for marker ID {target_id}: {product.name}")
+
+        # Format the response using the utility function
+        response_data = format_product_analysis_response(product)
+
+        return response_data
+
+    except Exception as e:
+        log_error(f"Error retrieving product data for marker ID {target_id}: {str(e)}", e)
+        # Return a structured error response
+        return {"found": False, "message": "An error occurred while retrieving product data."}

utils/analysis_utils.py

@@ -0,0 +1,63 @@
+from typing import Dict, Any, Optional
+
+def format_analysis_response(product_data: Optional[Dict[str, Any]]) -> Dict[str, Any]:
+    """
+    Formats the retrieved product analysis data into a consistent response structure.
+
+    Args:
+        product_data: A dictionary containing product data, or None if not found.
+
+    Returns:
+        A dictionary representing the formatted response.
+    """
+    if product_data is None:
+        return {
+            "found": False,
+            "safety_score": {"isPresent": False, "value": None},
+            "product_info": None,
+            "ingredient_info": None,
+            "allergen_info": None,
+            "diet_info": None,
+            "nutritional_info": None,
+            "manufacturing_info": None,
+        }
+
+    # Assuming product_data is a dictionary representing the joined data from product, ingredients, etc.
+    # Adjust field names based on your actual database schema and query results.
+    safety_score_value = product_data.get("safety_score")
+    safety_score_isPresent = safety_score_value is not None
+
+    return {
+        "found": True,
+        "safety_score": {
+            "calculated": safety_score_calculated,
+            "value": safety_score_value,
+        },
+        "product_info": {
+            "id": product_data.get("product_id"),
+            "name": product_data.get("product_name"),
+            "barcode": product_data.get("product_barcode"),
+            "image_url": product_data.get("product_image_url"),
+            "brand": product_data.get("product_brand"),
+            "manufacturing_places": product_data.get("product_manufacturing_places"),
+            "stores": product_data.get("product_stores"),
+            "countries": product_data.get("product_countries"),
+        } if product_data.get("product_id") is not None else None,
+        "ingredient_info": {
+            "ingredients_text": product_data.get("ingredients_text"),
+            "ingredients_analysis": product_data.get("ingredients_analysis"),
+            "additives": product_data.get("additives"),
+        } if product_data.get("ingredients_text") is not None else None,
+        "allergen_info": {
+            "allergens": product_data.get("allergens"),
+            "traces": product_data.get("traces"),
+        } if product_data.get("allergens") is not None or product_data.get("traces") is not None else None,
+        "diet_info": {
+            "vegan": product_data.get("vegan"),
+            "vegetarian": product_data.get("vegetarian"),
+        } if product_data.get("vegan") is not None or product_data.get("vegetarian") is not None else None,
+        "nutritional_info": product_data.get("nutritional_info"),  # Assuming this is already a dict
+        "manufacturing_info": product_data.get("manufacturing_info"),  # Assuming this is already a dict
+    }
+
+# Add other helper functions as needed for analysis