Add cleaned data and rewrite the code

Author: Omnia-Agabani

smart_pantry_manager/data/clean_data

@@ -0,0 +1,128 @@
+# spell-checker: disable
+"""
+Clean recipe CSV and create SQLite DB with diet type.
+"""
+
+import re
+import sqlite3
+
+import pandas as pd
+
+# Load CSV
+df = pd.read_csv("smart_pantry_manager/data/Recipe_Dataset.csv")
+
+# Forbidden (haram) ingredients
+haram_keywords = [
+    "pork",
+    "ham",
+    "bacon",
+    "prosciutto",
+    "pancetta",
+    "sausage",
+    "wine",
+    "beer",
+    "bourbon",
+    "rum",
+    "whisky",
+    "vodka",
+    "tequila",
+    "cognac",
+    "brandy",
+    "liqueur",
+    "alcohol",
+    "champagne",
+    "sake",
+    "sherry",
+    "gin",
+]
+
+# Meat ingredients
+meat_keywords = [
+    "chicken",
+    "beef",
+    "lamb",
+    "turkey",
+    "fish",
+    "shrimp",
+    "salmon",
+    "tuna",
+    "meat",
+    "steak",
+    "duck",
+    "anchovy",
+    "crab",
+    "lobster",
+    "clam",
+    "oyster",
+    "scallop",
+    "mussel",
+    "squid",
+    "sausage",
+]
+
+# Animal products (vegetarian)
+animal_product_keywords = [
+    "egg",
+    "milk",
+    "cheese",
+    "butter",
+    "cream",
+    "yogurt",
+    "ghee",
+    "honey",
+    "mayonnaise",
+    "whey",
+    "casein",
+    "gelatin",
+]
+
+
+# Check haram
+def contains_haram(ingredient_text):
+    if pd.isna(ingredient_text):
+        return False
+    text = ingredient_text.lower()
+    for kw in haram_keywords:
+        if re.search(rf"\b{kw}\b", text):
+            return True
+    return False
+
+
+# Classify diet type
+def classify_diet_type(ingredient_text):
+    if pd.isna(ingredient_text):
+        return "Unknown"
+    text = ingredient_text.lower()
+    for kw in meat_keywords:
+        if re.search(rf"\b{kw}\b", text):
+            return "Non-Vegetarian"
+    for kw in animal_product_keywords:
+        if re.search(rf"\b{kw}\b", text):
+            return "Vegetarian"
+    return "Vegan"
+
+
+# Filter haram
+df["contains_haram"] = df["Ingredients"].apply(contains_haram)
+df_clean = df[~df["contains_haram"]].copy()
+df_clean["Diet_Type"] = df_clean["Ingredients"].apply(classify_diet_type)
+df_clean.drop("contains_haram", axis=1, inplace=True)
+
+# SQLite DB
+conn = sqlite3.connect("smart_pantry_manager/data/cleaned_data.sqlite")
+
+# Tables for each diet type
+for diet in ["Vegan", "Vegetarian", "Non-Vegetarian"]:
+    diet_df = df_clean[df_clean["Diet_Type"] == diet].copy()
+    diet_df.to_sql(
+        diet.lower().replace("-", "_") + "_recipes",
+        conn,
+        if_exists="replace",
+        index=False,
+    )
+
+# Combined table
+df_clean.to_sql("all_recipes", conn, if_exists="replace", index=False)
+
+conn.close()
+print("✅ SQLite DB created with all_recipes and diet tables.")

smart_pantry_manager/data/cleaned_data.sqlite

@@ -1,6 +1,7 @@
 # spell-checker: disable
 """
-All Recipes Page for Smart Pantry Application (SQLite version)
+All Recipes Page for Smart Pantry Manager (SQLite version)
+Shows all recipes with diet type info.
 """
 
 import ast
@@ -11,96 +12,86 @@
 import streamlit as st
 
 st.set_page_config(page_title="All Recipes", page_icon="📜", layout="wide")
-
 st.title("📜 All Recipes")
-st.caption("Browse all available recipes in the Smart Pantry system.")
+st.caption("Browse all recipes and see diet type (Vegan/Vegetarian/Non-Vegetarian)")
 
 
-# ---------- Load recipes from SQLite ----------
+# ---------- Load recipes ----------
 @st.cache_data
 def load_recipes():
-    """
-    Load recipes from the SQLite database and normalize column names.
-    Returns a DataFrame with columns: Recipe, Ingredients, Instructions
-    """
-    db_path = "smart_pantry_manager/data/Recipe_Dataset.sqlite"
-
-    # Check if database file exists
+    db_path = "smart_pantry_manager/data/cleaned_data.sqlite"
     if not os.path.exists(db_path):
-        st.error(f"❌ Database file not found at: {db_path}")
-        return pd.DataFrame(columns=["Recipe", "Ingredients", "Instructions"])
-
-    # Connect to SQLite database
+        st.error("⚠️ Recipes database not found.")
+        return pd.DataFrame(
+            columns=["Recipe", "Ingredients", "Instructions", "Diet_type"]
+        )
     conn = sqlite3.connect(db_path)
-
     try:
-        # Load the table "recipes"
-        df = pd.read_sql_query("SELECT * FROM recipes", conn)
+        df = pd.read_sql_query("SELECT * FROM all_recipes", conn)
     except Exception as e:
-        st.error(f"Error reading database: {e}")
+        st.error(f"Error reading recipes: {e}")
         conn.close()
-        return pd.DataFrame(columns=["Recipe", "Ingredients", "Instructions"])
-
+        return pd.DataFrame(
+            columns=["Recipe", "Ingredients", "Instructions", "Diet_type"]
+        )
     conn.close()
-
-    # Normalize column names
+    # Normalize columns
     df.columns = [c.strip().lower() for c in df.columns]
-
-    # Rename columns if they exist
-    rename_map = {
-        "title": "Recipe",
-        "cleaned_ingredients": "Ingredients",
-        "instruction": "Instructions",
-        "instructions": "Instructions",
-    }
     df.rename(
-        columns={k: v for k, v in rename_map.items() if k in df.columns}, inplace=True
+        columns={
+            "title": "Recipe",
+            "instruction": "Instructions",
+            "instructions": "Instructions",
+            "diet_type": "Diet_type",
+            "ingredient": "Ingredients",
+            "cleaned_ingredients": "Ingredients",
+        },
+        inplace=True,
     )
+    for col in ["Recipe", "Ingredients", "Instructions", "Diet_type"]:
+        if col not in df.columns:
+            df[col] = ""
+    return df[["Recipe", "Ingredients", "Instructions", "Diet_type"]]
 
-    # Keep only required columns
-    required_cols = ["Recipe", "Ingredients", "Instructions"]
-    df = df[[col for col in required_cols if col in df.columns]]
-
-    return df
 
-
-def format_ingredients(ingredients_str):
-    """
-    Convert ingredients from string representation of list to a Python list.
-    """
+def parse_ingredients(ingredients_str):
+    if not ingredients_str:
+        return []
+    s = str(ingredients_str)
     try:
-        # Try to parse as a Python list
-        if ingredients_str.startswith("["):
-            return ast.literal_eval(ingredients_str)
-        # If it's comma-separated
-        elif "," in ingredients_str:
-            return [item.strip() for item in ingredients_str.split(",")]
-        else:
-            return [ingredients_str]
+        if s.startswith("[") and s.endswith("]"):
+            parsed = ast.literal_eval(s)
+            return [str(x).strip() for x in parsed if str(x).strip()]
+        if "," in s:
+            return [x.strip() for x in s.split(",") if x.strip()]
+        return [s]
     except:
-        # If parsing fails, return as single item list
-        return [ingredients_str]
+        return [s]
 
 
 recipes = load_recipes()
-
-# ---------- Display recipes ----------
 if recipes.empty:
     st.info("No recipes found.")
-else:
-    search = st.text_input("🔍 Search for a recipe:")
-    filtered = (
-        recipes[recipes["Recipe"].str.contains(search, case=False, na=False)]
-        if search
-        else recipes
-    )
-
-    for _, row in filtered.iterrows():
-        with st.expander(f"📖 {row['Recipe']}"):
-            st.markdown("**🧂 Ingredients:**")
-            ingredients_list = format_ingredients(row["Ingredients"])
-            for ingredient in ingredients_list:
-                st.write(f"• {ingredient}")
-
-            st.markdown("**👩‍🍳 Instructions:**")
-            st.write(row["Instructions"])
+    st.stop()
+
+search = st.text_input("🔍 Search for a recipe:")
+filtered = (
+    recipes[recipes["Recipe"].str.contains(search, case=False, na=False)]
+    if search
+    else recipes
+)
+
+for _, row in filtered.iterrows():
+    diet = row.get("Diet_type", "Unknown").capitalize()
+    recipe_name = row.get("Recipe", "Unnamed Recipe")
+    with st.expander(f"📖 {recipe_name} — {diet}"):
+        st.markdown(f"**Diet Type:** {diet}")
+        st.markdown("**🧂 Ingredients:**")
+        ing_list = parse_ingredients(row["Ingredients"])
+        if ing_list:
+            for ing in ing_list:
+                st.write(f"• {ing}")
+        else:
+            st.write("No ingredient data available.")
+        st.markdown("**👩‍🍳 Instructions:**")
+        st.write(row["Instructions"] or "No instructions available.")