Upload recipe_pipeline.py

recipe_pipeline.py

@@ -0,0 +1,130 @@
+import os
+import json
+import base64
+import io
+from PIL import Image
+from huggingface_hub import InferenceClient
+
+# ==========================================
+#  RECIPE DIGITALIZER PIPELINE (Image -> JSON)
+# ==========================================
+
+class RecipeDigitalizerPipeline:
+    def __init__(self):
+        print("Connecting to Hugging Face API (Qwen Mode)...")
+        # Ensure HF_TOKEN is set in your environment
+        self.token = os.getenv("HF_TOKEN")
+        
+        if not self.token:
+            print("⚠️ WARNING: HF_TOKEN is missing. The API call will likely fail.")
+
+        # --- WE ARE STICKING TO QWEN ---
+        # If 2.5 gives you trouble, you can try "Qwen/Qwen2-VL-7B-Instruct"
+        self.model_id = "Qwen/Qwen2.5-VL-7B-Instruct"
+
+        self.client = InferenceClient(token=self.token)
+
+    def compress_image(self, image_path):
+        """
+        Resizes the image so it doesn't crash the Free API.
+        """
+        with Image.open(image_path) as img:
+            if img.mode != 'RGB':
+                img = img.convert('RGB')
+
+            # Resize: Free API often rejects images larger than 1024x1024
+            max_size = 1024
+            if max(img.size) > max_size:
+                img.thumbnail((max_size, max_size))
+
+            # Save to memory as JPEG
+            buffer = io.BytesIO()
+            img.save(buffer, format="JPEG", quality=70) # Quality 70 is enough for text
+
+            # Convert to Base64
+            encoded_string = base64.b64encode(buffer.getvalue()).decode('utf-8')
+            return f"data:image/jpeg;base64,{encoded_string}"
+
+    def run_pipeline(self, image_path):
+        prompt = """Extract the recipe from this image.
+        Output strictly valid JSON with keys: title, ingredients (list), instructions (list), cuisine_type, difficulty.
+        Do not include markdown formatting like ```json, just the raw JSON."""
+
+        try:
+            # 1. Compress Image (Solves 400 Bad Request)
+            image_url = self.compress_image(image_path)
+
+            # 2. Call Qwen API
+            response = self.client.chat.completions.create(
+                model=self.model_id,
+                messages=[
+                    {
+                        "role": "user",
+                        "content": [
+                            {
+                                "type": "image_url",
+                                "image_url": {"url": image_url}
+                            },
+                            {"type": "text", "text": prompt}
+                        ]
+                    }
+                ],
+                max_tokens=1024
+            )
+
+            # 3. Clean Output
+            raw_text = response.choices[0].message.content
+            # Remove potential markdown fences
+            clean_json = raw_text.replace("```json", "").replace("```", "").strip()
+
+            # Extra safety: Find the first { and last }
+            start = clean_json.find('{')
+            end = clean_json.rfind('}') + 1
+            if start != -1 and end != -1:
+                clean_json = clean_json[start:end]
+
+            return json.loads(clean_json)
+
+        except Exception as e:
+            return {"error": f"Qwen API Error: {str(e)}"}
+
+# --- PART 4: EXECUTION EXAMPLE ---
+
+if __name__ == "__main__":
+    
+    # 1. AUTHENTICATION FIX (For Colab / Local Env)
+    try:
+        # If running in Colab, try to fetch from secrets
+        try:
+            from google.colab import userdata
+            hf1_secret = userdata.get('HF_TOKEN')
+            os.environ["HF_TOKEN"] = hf1_secret
+            print(f"✅ Successfully loaded token from secret HF_TOKEN")
+        except ImportError:
+            # If running locally, assume env var is already set or manually set here
+            pass 
+
+    except Exception as e:
+        print(f"⚠️ Warning: Could not load secret 'HF_TOKEN'. Make sure the name in the Key icon is exactly 'HF_TOKEN'.")
+        print(f"Error details: {e}")
+
+    # 2. INITIALIZE PIPELINE
+    try:
+        app = RecipeDigitalizerPipeline()
+
+        # 3. USER INPUT (Update this path to your actual image)
+        user_image = "Recipe.jfif" 
+
+        # 4. RUN PIPELINE
+        if os.path.exists(user_image):
+            print(f"Processing {user_image}...")
+            ai_output = app.run_pipeline(user_image)
+
+            # 5. AI OUTPUT
+            print("\n--- FINAL DIGITAL OUTPUT ---")
+            print(json.dumps(ai_output, indent=4))
+        else:
+            print(f"❌ Error: Image not found at {user_image}")
+
+    except Exception as e:
+        print(f"❌ Application Error: {e}")