Add requirements for OCR, barcode detection, database management, and caching

---

For more details, open the [Copilot Workspace session](https://copilot-workspace.githubnext.com/prathameshks/FoodAnalyzer-API?shareId=XXXX-XXXX-XXXX-XXXX).

database.py

@@ -0,0 +1,19 @@
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
+
+engine = create_engine(
+    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
+)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

main.py

@@ -1,10 +1,16 @@
 from fastapi import FastAPI
 from routes.extract_product_info_from_barcode import router as extract_product_info_router
 from routes.fetch_product_data import router as fetch_product_data_router
+from routes.auth import router as auth_router
+from routes.analysis import router as analysis_router
+from routes.history import router as history_router
 
 app = FastAPI()
 
 app.include_router(extract_product_info_router, prefix="/api")
 app.include_router(fetch_product_data_router, prefix="/api")
+app.include_router(auth_router, prefix="/api")
+app.include_router(analysis_router, prefix="/api")
+app.include_router(history_router, prefix="/api")
 
-# To run the FastAPI app, use the command: uvicorn main:app --reload
+# To run the FastAPI app, use the command: uvicorn main:app --reload

models/ingredient.py

@@ -0,0 +1,9 @@
+from sqlalchemy import Column, Integer, String, JSON
+from database import Base
+
+class Ingredient(Base):
+    __tablename__ = "ingredients"
+
+    id = Column(Integer, primary_key=True, index=True)
+    name = Column(String, unique=True, index=True)
+    nutritional_info = Column(JSON)

models/scan_history.py

@@ -0,0 +1,13 @@
+from sqlalchemy import Column, Integer, ForeignKey, DateTime
+from sqlalchemy.orm import relationship
+from database import Base
+
+class ScanHistory(Base):
+    __tablename__ = "scan_history"
+
+    id = Column(Integer, primary_key=True, index=True)
+    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
+    product_id = Column(Integer, nullable=False)
+    scan_date = Column(DateTime, nullable=False)
+
+    user = relationship("User", back_populates="scan_history")

models/user.py

@@ -0,0 +1,10 @@
+from sqlalchemy import Column, Integer, String
+from database import Base
+
+class User(Base):
+    __tablename__ = "users"
+
+    id = Column(Integer, primary_key=True, index=True)
+    username = Column(String, unique=True, index=True)
+    email = Column(String, unique=True, index=True)
+    hashed_password = Column(String)

models/user_preferences.py

@@ -0,0 +1,15 @@
+from sqlalchemy import Column, Integer, String, ForeignKey
+from sqlalchemy.orm import relationship
+from database import Base
+
+class UserPreferences(Base):
+    __tablename__ = "user_preferences"
+
+    id = Column(Integer, primary_key=True, index=True)
+    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
+    dietary_restrictions = Column(String, nullable=True)
+    allergens = Column(String, nullable=True)
+    preferred_ingredients = Column(String, nullable=True)
+    disliked_ingredients = Column(String, nullable=True)
+
+    user = relationship("User", back_populates="preferences")

requirements.txt

@@ -1,4 +1,9 @@
 fastapi
 uvicorn
 pydantic
-requests
+requests
+pytesseract
+opencv-python
+sqlalchemy
+alembic
+redis

routers/analysis.py

@@ -0,0 +1,26 @@
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+from typing import List, Dict, Any
+from database import get_db
+from models.user import User
+from models.ingredient import Ingredient
+from services.analysis_agent import analyze_ingredients, provide_personalized_recommendations
+from services.auth_service import get_current_user
+
+router = APIRouter()
+
+@router.post("/analyze_ingredients")
+def analyze_ingredients_endpoint(ingredients: List[Dict[str, Any]], db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
+    try:
+        analysis_results = analyze_ingredients(db, ingredients, current_user.id)
+        return analysis_results
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@router.get("/personalized_recommendations")
+def personalized_recommendations_endpoint(db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
+    try:
+        recommendations = provide_personalized_recommendations(db, current_user.id)
+        return recommendations
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))

routers/auth.py

@@ -0,0 +1,46 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+from fastapi.security import OAuth2PasswordRequestForm
+from pydantic import BaseModel
+from database import get_db
+from services.auth_service import authenticate_user, create_access_token, create_user, get_current_active_user
+from datetime import timedelta
+
+router = APIRouter()
+
+class UserCreate(BaseModel):
+    username: str
+    email: str
+    password: str
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+@router.post("/register", response_model=Token)
+def register(user: UserCreate, db: Session = Depends(get_db)):
+    db_user = create_user(db, user.username, user.email, user.password)
+    access_token_expires = timedelta(minutes=30)
+    access_token = create_access_token(
+        data={"sub": db_user.username}, expires_delta=access_token_expires
+    )
+    return {"access_token": access_token, "token_type": "bearer"}
+
+@router.post("/login", response_model=Token)
+def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
+    user = authenticate_user(db, form_data.username, form_data.password)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    access_token_expires = timedelta(minutes=30)
+    access_token = create_access_token(
+        data={"sub": user.username}, expires_delta=access_token_expires
+    )
+    return {"access_token": access_token, "token_type": "bearer"}
+
+@router.get("/users/me", response_model=UserCreate)
+def read_users_me(current_user: UserCreate = Depends(get_current_active_user)):
+    return current_user

routers/history.py

@@ -0,0 +1,30 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+from pydantic import BaseModel
+from database import get_db
+from services.scan_history import record_scan, get_scan_history
+from models.scan_history import ScanHistory
+
+router = APIRouter()
+
+class ScanHistoryCreate(BaseModel):
+    user_id: int
+    product_id: int
+
+class ScanHistoryResponse(BaseModel):
+    id: int
+    user_id: int
+    product_id: int
+    scan_date: str
+
+@router.post("/scan", response_model=ScanHistoryResponse)
+def create_scan(scan: ScanHistoryCreate, db: Session = Depends(get_db)):
+    scan_entry = record_scan(db, scan.user_id, scan.product_id)
+    return scan_entry
+
+@router.get("/history/{user_id}", response_model=list[ScanHistoryResponse])
+def read_scan_history(user_id: int, db: Session = Depends(get_db)):
+    scan_history = get_scan_history(db, user_id)
+    if not scan_history:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Scan history not found")
+    return scan_history

services/ai_agent.py

@@ -0,0 +1,88 @@
+from sqlalchemy.orm import Session
+from fastapi import HTTPException
+from utils import fetch_product_data_from_api, save_json_file
+from models.ingredient import Ingredient
+from services.ingredients import get_ingredient_by_name, save_ingredient_data
+from typing import Dict, Any
+import json
+
+def preprocess_data(barcode: str) -> Dict[str, Any]:
+    data = fetch_product_data_from_api(barcode)
+    product = data.get('product', {})
+
+    product_info = {
+        "product_name": product.get('product_name_en', product.get('product_name', 'N/A')),
+        "generic_name": product.get('generic_name_en', product.get('generic_name', 'N/A')),
+        "brands": product.get('brands', 'N/A'),
+        "ingredients": [],
+        "ingredients_text": product.get('ingredients_text_en', product.get('ingredients_text', 'N/A')),
+        "ingredients_analysis": product.get('ingredients_analysis', {}),
+        "nutriscore": product.get('nutriscore', {}),
+        "nutrient_levels": product.get('nutrient_levels', {}),
+        "nutriments": product.get('nutriments', {}),
+        "data_quality_warnings": product.get('data_quality_warnings_tags', [])
+    }
+
+    ingredients_list = product.get('ingredients', [])
+    for ingredient in ingredients_list:
+        ingredient_info = {
+            "text": ingredient.get('text', 'N/A'),
+            "percent": ingredient.get('percent', ingredient.get('percent_estimate', 'N/A')),
+            "vegan": ingredient.get('vegan', 'N/A'),
+            "vegetarian": ingredient.get('vegetarian', 'N/A'),
+            "sub_ingredients": []
+        }
+        sub_ingredients = ingredient.get('ingredients', [])
+        for sub_ingredient in sub_ingredients:
+            sub_ingredient_info = {
+                "text": sub_ingredient.get('text', 'N/A'),
+                "percent": sub_ingredient.get('percent', sub_ingredient.get('percent_estimate', 'N/A')),
+                "vegan": sub_ingredient.get('vegan', 'N/A'),
+                "vegetarian": sub_ingredient.get('vegetarian', 'N/A')
+            }
+            ingredient_info["sub_ingredients"].append(sub_ingredient_info)
+        product_info["ingredients"].append(ingredient_info)
+
+    return product_info
+
+def validate_data(data: Dict[str, Any]) -> bool:
+    required_fields = ["product_name", "generic_name", "brands", "ingredients", "nutriscore", "nutrient_levels", "nutriments"]
+    for field in required_fields:
+        if field not in data or not data[field]:
+            return False
+    return True
+
+def clean_data(data: Dict[str, Any]) -> Dict[str, Any]:
+    for ingredient in data["ingredients"]:
+        if "percent" in ingredient and ingredient["percent"] == "N/A":
+            ingredient["percent"] = 0
+        for sub_ingredient in ingredient["sub_ingredients"]:
+            if "percent" in sub_ingredient and sub_ingredient["percent"] == "N/A":
+                sub_ingredient["percent"] = 0
+    return data
+
+def standardize_data(data: Dict[str, Any]) -> Dict[str, Any]:
+    for ingredient in data["ingredients"]:
+        ingredient["text"] = ingredient["text"].lower()
+        for sub_ingredient in ingredient["sub_ingredients"]:
+            sub_ingredient["text"] = sub_ingredient["text"].lower()
+    return data
+
+def enrich_data(db: Session, data: Dict[str, Any]) -> Dict[str, Any]:
+    for ingredient in data["ingredients"]:
+        ingredient_data = get_ingredient_by_name(db, ingredient["text"])
+        if not ingredient_data:
+            ingredient_data = fetch_product_data_from_api(ingredient["text"])
+            save_ingredient_data(db, ingredient["text"], ingredient_data)
+        ingredient["nutritional_info"] = ingredient_data
+    return data
+
+def process_data(db: Session, barcode: str) -> Dict[str, Any]:
+    data = preprocess_data(barcode)
+    if not validate_data(data):
+        raise HTTPException(status_code=400, detail="Invalid data")
+    data = clean_data(data)
+    data = standardize_data(data)
+    data = enrich_data(db, data)
+    save_json_file(barcode, data)
+    return data

services/analysis_agent.py

@@ -0,0 +1,48 @@
+from sqlalchemy.orm import Session
+from fastapi import HTTPException
+from typing import Dict, Any, List
+from models.user_preferences import UserPreferences
+from services.ingredients import get_ingredient_data, filter_ingredients_by_preferences
+
+def analyze_ingredients(db: Session, ingredients: List[Dict[str, Any]], user_id: int) -> Dict[str, Any]:
+    preferences = db.query(UserPreferences).filter(UserPreferences.user_id == user_id).first()
+    if not preferences:
+        raise HTTPException(status_code=404, detail="User preferences not found")
+
+    filtered_ingredients = filter_ingredients_by_preferences(ingredients, preferences)
+    analysis_results = {
+        "safe_ingredients": [],
+        "unsafe_ingredients": [],
+        "additional_facts": []
+    }
+
+    for ingredient in filtered_ingredients:
+        ingredient_data = get_ingredient_data(db, ingredient["text"])
+        if ingredient_data:
+            analysis_results["safe_ingredients"].append({
+                "name": ingredient["text"],
+                "nutritional_info": ingredient_data
+            })
+        else:
+            analysis_results["unsafe_ingredients"].append({
+                "name": ingredient["text"],
+                "reason": "Information not found"
+            })
+
+    return analysis_results
+
+def provide_personalized_recommendations(db: Session, user_id: int) -> Dict[str, Any]:
+    preferences = db.query(UserPreferences).filter(UserPreferences.user_id == user_id).first()
+    if not preferences:
+        raise HTTPException(status_code=404, detail="User preferences not found")
+
+    recommended_ingredients = []
+    all_ingredients = db.query(Ingredient).all()
+    for ingredient in all_ingredients:
+        if ingredient.name not in preferences.disliked_ingredients:
+            recommended_ingredients.append({
+                "name": ingredient.name,
+                "nutritional_info": ingredient.nutritional_info
+            })
+
+    return {"recommended_ingredients": recommended_ingredients}

services/auth_service.py

@@ -0,0 +1,87 @@
+from passlib.context import CryptContext
+from jose import JWTError, jwt
+from datetime import datetime, timedelta
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from sqlalchemy.orm import Session
+from database import get_db
+from models.user import User
+from pydantic import BaseModel
+
+# to get a string like this run:
+# openssl rand -hex 32
+SECRET_KEY = "09d8f7a6b5c4e3d2f1a0b9c8d7e6f5a4"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+class TokenData(BaseModel):
+    username: str | None = None
+
+class UserInDB(User):
+    hashed_password: str
+
+def verify_password(plain_password, hashed_password):
+    return pwd_context.verify(plain_password, hashed_password)
+
+def get_password_hash(password):
+    return pwd_context.hash(password)
+
+def get_user(db, username: str):
+    return db.query(User).filter(User.username == username).first()
+
+def authenticate_user(db, username: str, password: str):
+    user = get_user(db, username)
+    if not user:
+        return False
+    if not verify_password(password, user.hashed_password):
+        return False
+    return user
+
+def create_access_token(data: dict, expires_delta: timedelta | None = None):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+async def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username: str = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+        token_data = TokenData(username=username)
+    except JWTError:
+        raise credentials_exception
+    user = get_user(db, username=token_data.username)
+    if user is None:
+        raise credentials_exception
+    return user
+
+async def get_current_active_user(current_user: User = Depends(get_current_user)):
+    if not current_user.is_active:
+        raise HTTPException(status_code=400, detail="Inactive user")
+    return current_user
+
+def create_user(db: Session, username: str, email: str, password: str):
+    hashed_password = get_password_hash(password)
+    db_user = User(username=username, email=email, hashed_password=hashed_password)
+    db.add(db_user)
+    db.commit()
+    db.refresh(db_user)
+    return db_user

services/ingredients.py

@@ -0,0 +1,45 @@
+from sqlalchemy.orm import Session
+from models.ingredient import Ingredient
+from fastapi import HTTPException
+from cachetools import cached, TTLCache
+from typing import List, Dict, Any
+import requests
+
+cache = TTLCache(maxsize=100, ttl=300)
+
+def get_ingredient_by_name(db: Session, name: str) -> Ingredient:
+    return db.query(Ingredient).filter(Ingredient.name == name).first()
+
+@cached(cache)
+def fetch_ingredient_data_from_api(name: str) -> Dict[str, Any]:
+    url = f"https://api.example.com/ingredients/{name}"
+    response = requests.get(url)
+    if response.status_code != 200:
+        raise HTTPException(status_code=response.status_code, detail=f"Failed to fetch data for ingredient {name}")
+    return response.json()
+
+def get_ingredient_data(db: Session, name: str) -> Dict[str, Any]:
+    ingredient = get_ingredient_by_name(db, name)
+    if ingredient:
+        return ingredient.nutritional_info
+    data = fetch_ingredient_data_from_api(name)
+    save_ingredient_data(db, name, data)
+    return data
+
+def save_ingredient_data(db: Session, name: str, data: Dict[str, Any]):
+    ingredient = Ingredient(name=name, nutritional_info=data)
+    db.add(ingredient)
+    db.commit()
+    db.refresh(ingredient)
+
+def filter_ingredients_by_preferences(ingredients: List[Dict[str, Any]], preferences: Dict[str, Any]) -> List[Dict[str, Any]]:
+    filtered_ingredients = []
+    for ingredient in ingredients:
+        if preferences.get("low_sugar") and ingredient.get("sugar", 0) > 5:
+            continue
+        if preferences.get("low_fat") and ingredient.get("fat", 0) > 5:
+            continue
+        if preferences.get("allergens") and any(allergen in ingredient.get("allergens", []) for allergen in preferences["allergens"]):
+            continue
+        filtered_ingredients.append(ingredient)
+    return filtered_ingredients

services/logging_service.py

@@ -0,0 +1,38 @@
+import logging
+from logging.handlers import RotatingFileHandler
+
+# Configure logging
+logger = logging.getLogger("food_analyzer")
+logger.setLevel(logging.DEBUG)
+
+# Create a file handler that logs debug and higher level messages
+file_handler = RotatingFileHandler("food_analyzer.log", maxBytes=5*1024*1024, backupCount=3)
+file_handler.setLevel(logging.DEBUG)
+
+# Create a console handler that logs error and higher level messages
+console_handler = logging.StreamHandler()
+console_handler.setLevel(logging.ERROR)
+
+# Create a formatter and set it for both handlers
+formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+file_handler.setFormatter(formatter)
+console_handler.setFormatter(formatter)
+
+# Add the handlers to the logger
+logger.addHandler(file_handler)
+logger.addHandler(console_handler)
+
+def log_debug(message: str):
+    logger.debug(message)
+
+def log_info(message: str):
+    logger.info(message)
+
+def log_warning(message: str):
+    logger.warning(message)
+
+def log_error(message: str):
+    logger.error(message)
+
+def log_critical(message: str):
+    logger.critical(message)

services/scan_history.py

@@ -0,0 +1,13 @@
+from sqlalchemy.orm import Session
+from models.scan_history import ScanHistory
+from datetime import datetime
+
+def record_scan(db: Session, user_id: int, product_id: int):
+    scan_entry = ScanHistory(user_id=user_id, product_id=product_id, scan_date=datetime.utcnow())
+    db.add(scan_entry)
+    db.commit()
+    db.refresh(scan_entry)
+    return scan_entry
+
+def get_scan_history(db: Session, user_id: int):
+    return db.query(ScanHistory).filter(ScanHistory.user_id == user_id).all()

tests/test_ai_agent_service.py

@@ -0,0 +1,141 @@
+import unittest
+from unittest.mock import patch, MagicMock
+from sqlalchemy.orm import Session
+from services.ai_agent import preprocess_data, validate_data, clean_data, standardize_data, enrich_data, process_data
+
+class TestAIAgentService(unittest.TestCase):
+
+    @patch('services.ai_agent.fetch_product_data_from_api')
+    def test_preprocess_data(self, mock_fetch_product_data_from_api):
+        mock_fetch_product_data_from_api.return_value = {
+            'product': {
+                'product_name_en': 'Test Product',
+                'generic_name_en': 'Test Generic',
+                'brands': 'Test Brand',
+                'ingredients': [
+                    {
+                        'text': 'Test Ingredient',
+                        'percent': 50,
+                        'vegan': 'yes',
+                        'vegetarian': 'yes',
+                        'ingredients': [
+                            {
+                                'text': 'Sub Ingredient',
+                                'percent': 25,
+                                'vegan': 'yes',
+                                'vegetarian': 'yes'
+                            }
+                        ]
+                    }
+                ],
+                'ingredients_text_en': 'Test Ingredients Text',
+                'ingredients_analysis': {},
+                'nutriscore': {},
+                'nutrient_levels': {},
+                'nutriments': {},
+                'data_quality_warnings_tags': []
+            }
+        }
+        result = preprocess_data('test_barcode')
+        self.assertEqual(result['product_name'], 'Test Product')
+        self.assertEqual(result['generic_name'], 'Test Generic')
+        self.assertEqual(result['brands'], 'Test Brand')
+        self.assertEqual(result['ingredients'][0]['text'], 'Test Ingredient')
+        self.assertEqual(result['ingredients'][0]['sub_ingredients'][0]['text'], 'Sub Ingredient')
+
+    def test_validate_data(self):
+        valid_data = {
+            'product_name': 'Test Product',
+            'generic_name': 'Test Generic',
+            'brands': 'Test Brand',
+            'ingredients': [],
+            'nutriscore': {},
+            'nutrient_levels': {},
+            'nutriments': {}
+        }
+        invalid_data = {
+            'product_name': '',
+            'generic_name': '',
+            'brands': '',
+            'ingredients': [],
+            'nutriscore': {},
+            'nutrient_levels': {},
+            'nutriments': {}
+        }
+        self.assertTrue(validate_data(valid_data))
+        self.assertFalse(validate_data(invalid_data))
+
+    def test_clean_data(self):
+        data = {
+            'ingredients': [
+                {
+                    'text': 'Test Ingredient',
+                    'percent': 'N/A',
+                    'sub_ingredients': [
+                        {
+                            'text': 'Sub Ingredient',
+                            'percent': 'N/A'
+                        }
+                    ]
+                }
+            ]
+        }
+        cleaned_data = clean_data(data)
+        self.assertEqual(cleaned_data['ingredients'][0]['percent'], 0)
+        self.assertEqual(cleaned_data['ingredients'][0]['sub_ingredients'][0]['percent'], 0)
+
+    def test_standardize_data(self):
+        data = {
+            'ingredients': [
+                {
+                    'text': 'Test Ingredient',
+                    'sub_ingredients': [
+                        {
+                            'text': 'Sub Ingredient'
+                        }
+                    ]
+                }
+            ]
+        }
+        standardized_data = standardize_data(data)
+        self.assertEqual(standardized_data['ingredients'][0]['text'], 'test ingredient')
+        self.assertEqual(standardized_data['ingredients'][0]['sub_ingredients'][0]['text'], 'sub ingredient')
+
+    @patch('services.ai_agent.get_ingredient_by_name')
+    @patch('services.ai_agent.fetch_product_data_from_api')
+    @patch('services.ai_agent.save_ingredient_data')
+    def test_enrich_data(self, mock_save_ingredient_data, mock_fetch_product_data_from_api, mock_get_ingredient_by_name):
+        db = MagicMock(spec=Session)
+        data = {
+            'ingredients': [
+                {
+                    'text': 'Test Ingredient',
+                    'sub_ingredients': []
+                }
+            ]
+        }
+        mock_get_ingredient_by_name.return_value = None
+        mock_fetch_product_data_from_api.return_value = {'nutritional_info': 'Test Info'}
+        enriched_data = enrich_data(db, data)
+        self.assertEqual(enriched_data['ingredients'][0]['nutritional_info'], {'nutritional_info': 'Test Info'})
+        mock_save_ingredient_data.assert_called_once_with(db, 'test ingredient', {'nutritional_info': 'Test Info'})
+
+    @patch('services.ai_agent.preprocess_data')
+    @patch('services.ai_agent.validate_data')
+    @patch('services.ai_agent.clean_data')
+    @patch('services.ai_agent.standardize_data')
+    @patch('services.ai_agent.enrich_data')
+    @patch('services.ai_agent.save_json_file')
+    def test_process_data(self, mock_save_json_file, mock_enrich_data, mock_standardize_data, mock_clean_data, mock_validate_data, mock_preprocess_data):
+        db = MagicMock(spec=Session)
+        mock_preprocess_data.return_value = {'product_name': 'Test Product'}
+        mock_validate_data.return_value = True
+        mock_clean_data.return_value = {'product_name': 'Test Product'}
+        mock_standardize_data.return_value = {'product_name': 'Test Product'}
+        mock_enrich_data.return_value = {'product_name': 'Test Product'}
+        result = process_data(db, 'test_barcode')
+        self.assertEqual(result['product_name'], 'Test Product')
+        mock_save_json_file.assert_called_once_with('test_barcode', {'product_name': 'Test Product'})
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_analysis_agent_service.py

@@ -0,0 +1,56 @@
+import unittest
+from unittest.mock import MagicMock
+from sqlalchemy.orm import Session
+from services.analysis_agent import analyze_ingredients, provide_personalized_recommendations
+from models.user_preferences import UserPreferences
+from models.ingredient import Ingredient
+
+class TestAnalysisAgentService(unittest.TestCase):
+
+    def setUp(self):
+        self.db = MagicMock(spec=Session)
+        self.user_id = 1
+        self.ingredients = [
+            {"text": "sugar"},
+            {"text": "salt"},
+            {"text": "flour"}
+        ]
+        self.preferences = UserPreferences(
+            user_id=self.user_id,
+            dietary_restrictions="low sugar",
+            allergens="",
+            preferred_ingredients="",
+            disliked_ingredients=""
+        )
+        self.db.query.return_value.filter.return_value.first.return_value = self.preferences
+
+    def test_analyze_ingredients(self):
+        self.db.query.return_value.filter.return_value.first.return_value = self.preferences
+        self.db.query.return_value.all.return_value = [
+            Ingredient(name="sugar", nutritional_info={"calories": 100}),
+            Ingredient(name="salt", nutritional_info={"sodium": 200}),
+            Ingredient(name="flour", nutritional_info={"carbs": 300})
+        ]
+
+        result = analyze_ingredients(self.db, self.ingredients, self.user_id)
+
+        self.assertIn("safe_ingredients", result)
+        self.assertIn("unsafe_ingredients", result)
+        self.assertIn("additional_facts", result)
+        self.assertEqual(len(result["safe_ingredients"]), 2)
+        self.assertEqual(len(result["unsafe_ingredients"]), 1)
+
+    def test_provide_personalized_recommendations(self):
+        self.db.query.return_value.all.return_value = [
+            Ingredient(name="sugar", nutritional_info={"calories": 100}),
+            Ingredient(name="salt", nutritional_info={"sodium": 200}),
+            Ingredient(name="flour", nutritional_info={"carbs": 300})
+        ]
+
+        result = provide_personalized_recommendations(self.db, self.user_id)
+
+        self.assertIn("recommended_ingredients", result)
+        self.assertEqual(len(result["recommended_ingredients"]), 3)
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_auth_service.py

@@ -0,0 +1,55 @@
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from database import Base, get_db
+from main import app
+from services.auth_service import create_user, get_password_hash
+from models.user import User
+
+SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
+engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
+TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base.metadata.create_all(bind=engine)
+
+def override_get_db():
+    try:
+        db = TestingSessionLocal()
+        yield db
+    finally:
+        db.close()
+
+app.dependency_overrides[get_db] = override_get_db
+
+client = TestClient(app)
+
+@pytest.fixture(scope="module")
+def test_db():
+    Base.metadata.create_all(bind=engine)
+    db = TestingSessionLocal()
+    yield db
+    db.close()
+    Base.metadata.drop_all(bind=engine)
+
+def test_register_user(test_db):
+    response = client.post("/auth/register", json={"username": "testuser", "email": "testuser@example.com", "password": "testpassword"})
+    assert response.status_code == 200
+    assert "access_token" in response.json()
+    assert response.json()["token_type"] == "bearer"
+
+def test_login_user(test_db):
+    create_user(test_db, "testuser", "testuser@example.com", get_password_hash("testpassword"))
+    response = client.post("/auth/login", data={"username": "testuser", "password": "testpassword"})
+    assert response.status_code == 200
+    assert "access_token" in response.json()
+    assert response.json()["token_type"] == "bearer"
+
+def test_get_current_user(test_db):
+    create_user(test_db, "testuser", "testuser@example.com", get_password_hash("testpassword"))
+    response = client.post("/auth/login", data={"username": "testuser", "password": "testpassword"})
+    token = response.json()["access_token"]
+    response = client.get("/auth/users/me", headers={"Authorization": f"Bearer {token}"})
+    assert response.status_code == 200
+    assert response.json()["username"] == "testuser"
+    assert response.json()["email"] == "testuser@example.com"

tests/test_ingredients_service.py

@@ -0,0 +1,52 @@
+import unittest
+from unittest.mock import patch, MagicMock
+from sqlalchemy.orm import Session
+from services.ingredients import get_ingredient_data, save_ingredient_data, filter_ingredients_by_preferences
+
+class TestIngredientService(unittest.TestCase):
+
+    @patch('services.ingredients.get_ingredient_by_name')
+    @patch('services.ingredients.fetch_ingredient_data_from_api')
+    @patch('services.ingredients.save_ingredient_data')
+    def test_get_ingredient_data(self, mock_save_ingredient_data, mock_fetch_ingredient_data_from_api, mock_get_ingredient_by_name):
+        db = MagicMock(spec=Session)
+        mock_get_ingredient_by_name.return_value = None
+        mock_fetch_ingredient_data_from_api.return_value = {"name": "test_ingredient", "nutritional_info": "test_info"}
+
+        result = get_ingredient_data(db, "test_ingredient")
+
+        mock_get_ingredient_by_name.assert_called_once_with(db, "test_ingredient")
+        mock_fetch_ingredient_data_from_api.assert_called_once_with("test_ingredient")
+        mock_save_ingredient_data.assert_called_once_with(db, "test_ingredient", {"name": "test_ingredient", "nutritional_info": "test_info"})
+        self.assertEqual(result, {"name": "test_ingredient", "nutritional_info": "test_info"})
+
+    def test_save_ingredient_data(self):
+        db = MagicMock(spec=Session)
+        name = "test_ingredient"
+        data = {"name": "test_ingredient", "nutritional_info": "test_info"}
+
+        save_ingredient_data(db, name, data)
+
+        db.add.assert_called_once()
+        db.commit.assert_called_once()
+        db.refresh.assert_called_once()
+
+    def test_filter_ingredients_by_preferences(self):
+        ingredients = [
+            {"text": "ingredient1", "sugar": 10, "fat": 5, "allergens": ["allergen1"]},
+            {"text": "ingredient2", "sugar": 3, "fat": 2, "allergens": []},
+            {"text": "ingredient3", "sugar": 6, "fat": 1, "allergens": ["allergen2"]}
+        ]
+        preferences = {
+            "low_sugar": True,
+            "low_fat": True,
+            "allergens": ["allergen1"]
+        }
+
+        result = filter_ingredients_by_preferences(ingredients, preferences)
+
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0]["text"], "ingredient2")
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_scan_history_service.py

@@ -0,0 +1,43 @@
+import unittest
+from unittest.mock import MagicMock
+from datetime import datetime
+from sqlalchemy.orm import Session
+from models.scan_history import ScanHistory
+from services.scan_history import record_scan, get_scan_history
+
+class TestScanHistoryService(unittest.TestCase):
+
+    def setUp(self):
+        self.db = MagicMock(spec=Session)
+        self.user_id = 1
+        self.product_id = 123
+        self.scan_date = datetime.utcnow()
+
+    def test_record_scan(self):
+        scan_entry = ScanHistory(user_id=self.user_id, product_id=self.product_id, scan_date=self.scan_date)
+        self.db.add.return_value = None
+        self.db.commit.return_value = None
+        self.db.refresh.return_value = None
+
+        result = record_scan(self.db, self.user_id, self.product_id)
+
+        self.db.add.assert_called_once_with(scan_entry)
+        self.db.commit.assert_called_once()
+        self.db.refresh.assert_called_once_with(scan_entry)
+        self.assertEqual(result.user_id, self.user_id)
+        self.assertEqual(result.product_id, self.product_id)
+        self.assertEqual(result.scan_date, self.scan_date)
+
+    def test_get_scan_history(self):
+        scan_entry = ScanHistory(user_id=self.user_id, product_id=self.product_id, scan_date=self.scan_date)
+        self.db.query.return_value.filter.return_value.all.return_value = [scan_entry]
+
+        result = get_scan_history(self.db, self.user_id)
+
+        self.db.query.assert_called_once_with(ScanHistory)
+        self.db.query.return_value.filter.assert_called_once_with(ScanHistory.user_id == self.user_id)
+        self.db.query.return_value.filter.return_value.all.assert_called_once()
+        self.assertEqual(result, [scan_entry])
+
+if __name__ == '__main__':
+    unittest.main()

utils.py

@@ -2,6 +2,8 @@ import json
 import os
 import requests
 from fastapi import HTTPException
+import cv2
+import pytesseract
 
 def fetch_product_data_from_api(barcode):
     url = f"https://india.openfoodfacts.org/api/v2/product/{barcode}.json"
@@ -14,4 +16,21 @@ def save_json_file(item, data):
     os.makedirs("v2", exist_ok=True)
     with open(f"v2/{item}.json", "w") as file:
         json.dump(data, file)
-    print(f"Saved {item}.json")
+    print(f"Saved {item}.json")
+
+def extract_text_from_image(image_path):
+    image = cv2.imread(image_path)
+    gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+    gray = cv2.medianBlur(gray, 3)
+    text = pytesseract.image_to_string(gray)
+    return text
+
+def detect_barcode_from_image(image_path):
+    image = cv2.imread(image_path)
+    gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+    gray = cv2.medianBlur(gray, 3)
+    detector = cv2.QRCodeDetector()
+    data, bbox, _ = detector.detectAndDecode(gray)
+    if bbox is not None:
+        return data
+    return None