Spaces:
Sleeping
Sleeping
File size: 4,368 Bytes
7fa4837 a0ab790 7fa4837 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 |
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)...")
self.token = os.getenv("HF_TOKEN")
# This targets the larger model which uses a DIFFERENT server
self.model_id = "Qwen/Qwen2.5-VL-72B-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}")
|