482 lines
16 KiB
Python
Executable File
482 lines
16 KiB
Python
Executable File
import sys
|
||
import json
|
||
import argparse
|
||
import logging
|
||
import os
|
||
import re
|
||
import time
|
||
from datetime import datetime
|
||
|
||
|
||
class ColoredFormatter(logging.Formatter):
|
||
# ANSI escape codes for colors
|
||
COLORS = {
|
||
"DEBUG": "\033[94m", # Blue
|
||
"INFO": "\033[92m", # Green
|
||
"WARNING": "\033[93m", # Yellow
|
||
"ERROR": "\033[91m", # Red
|
||
"CRITICAL": "\033[95m", # Magenta
|
||
"RESET": "\033[0m" # Reset
|
||
}
|
||
|
||
def format(self, record):
|
||
color = self.COLORS.get(record.levelname, self.COLORS["RESET"])
|
||
message = super().format(record)
|
||
return f"{color}{message}{self.COLORS['RESET']}"
|
||
|
||
def setup_logger(log_file=None, level=logging.DEBUG):
|
||
"""
|
||
Sets up a logger that logs messages to the console and optionally to a file.
|
||
|
||
Parameters:
|
||
log_file (str): Path to a log file. If None, only console logging is used.
|
||
level (int): Logging level (e.g., logging.INFO, logging.DEBUG, logging.ERROR).
|
||
|
||
Returns:
|
||
logging.Logger: Configured logger instance.
|
||
"""
|
||
logger = logging.getLogger(__name__)
|
||
logger.setLevel(level)
|
||
|
||
# Avoid adding multiple handlers if logger already exists
|
||
if not logger.handlers:
|
||
# Console handler (INFO and above)
|
||
ch = logging.StreamHandler()
|
||
ch.setLevel(logging.INFO)
|
||
ch_formatter = ColoredFormatter('%(asctime)s - %(levelname)s - %(message)s')
|
||
ch.setFormatter(ch_formatter)
|
||
logger.addHandler(ch)
|
||
|
||
# File handler (optional)
|
||
if log_file:
|
||
# Ensure the directory exists
|
||
os.makedirs(os.path.dirname(log_file), exist_ok=True)
|
||
fh = logging.FileHandler(log_file, mode='a')
|
||
fh.setLevel(level)
|
||
fh_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
|
||
fh.setFormatter(fh_formatter)
|
||
logger.addHandler(fh)
|
||
|
||
return logger
|
||
|
||
# -------- OpenAI --------
|
||
from openai import OpenAI
|
||
|
||
# -------- Gemini --------
|
||
from google import genai
|
||
|
||
# -------- Hugging Face --------
|
||
from huggingface_hub import InferenceClient
|
||
import requests
|
||
|
||
# -------------------------
|
||
# API Keys
|
||
# -------------------------
|
||
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") or "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"
|
||
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") or "AIzaSyDRO5QyNa44eUsbfE5eQbMpjagO3398Ihs"
|
||
|
||
# -------------------------
|
||
# Hugging Face
|
||
# -------------------------
|
||
HF_API_KEY = os.getenv("HF_API_KEY") or "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"
|
||
HF_MODEL = "Qwen/Qwen2.5-7B-Instruct"
|
||
HF_API_URL = f"https://api-inference.huggingface.co/models/{HF_MODEL}"
|
||
|
||
# -------------------------
|
||
# Groq
|
||
# -------------------------
|
||
# export GROQ_API_KEY= "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"
|
||
GROQ_API_KEY = os.getenv("GROQ_API_KEY") or "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"
|
||
GROQ_MODEL = "llama-3.1-8b-instant"
|
||
GROQ_MODEL = "openai/gpt-oss-120b"
|
||
# -------------------------
|
||
# Configuration
|
||
# -------------------------
|
||
OPENAI_MODEL = "gpt-4.1-mini"
|
||
GEMINI_MODEL = "gemini-2.0-flash"
|
||
REQUIRED_NUTRITION_FIELDS = [
|
||
"calories",
|
||
"proteinContent",
|
||
"fatContent",
|
||
"carbohydrateContent"]
|
||
|
||
REQUIRED_FIELDS = [
|
||
"keywords",
|
||
"dateCreated",
|
||
"recipeYield",
|
||
"recipeIngredient",
|
||
"recipeInstructions"
|
||
]
|
||
|
||
|
||
# -------------------------
|
||
# Prompt generation
|
||
# -------------------------
|
||
def generate_prompt(recipe_content: str) -> str:
|
||
return f"""
|
||
Du bist eine reine JSON-Transformationsmaschine.
|
||
|
||
Anweisungen:
|
||
|
||
1. Aktualisiere oder erstelle ein Objekt „nutrition“ für das eine Portion des Rezepts im JSON und berechne die folgenden Felder:
|
||
"nutrition": {{
|
||
"@type": "NutritionInformation",
|
||
"servingSize": "1 Portion",
|
||
"calories": "<numeric> kcal",
|
||
"proteinContent": "<numeric> g",
|
||
"fatContent": "<numeric> g",
|
||
"carbohydrateContent": "<numeric> g"
|
||
"cholesterolContent": "<numeric> g",
|
||
"fiberContent": "<numeric> g",
|
||
"saturatedFatContent": "<numeric> g",
|
||
"sodiumContent": "<numeric> g",
|
||
"sugarContent": "<numeric> g",
|
||
"transFatContent": "<numeric> g",
|
||
"unsaturatedFatContent": "<numeric> g"
|
||
}}
|
||
Berechne die Nährwerte anhand der Zutaten in „recipeIngredient“.
|
||
Die Zutaten sind auf Deutsch.
|
||
Füge die Einheiten zu den Feldern hinzu.
|
||
Falls du Annahmen bezüglich Mengen machst, schreibe diese zu den Zutaten, recipeIngredient.
|
||
2. Alle anderen Felder im ursprünglichen JSON unverändert lassen.
|
||
3. Gib das vollständige aktualisierte JSON-Objekt zurück, einschließlich der aktualisierten Nährwerte.
|
||
4. Gib ausschließlich gültiges JSON zurück.
|
||
5. Keine Markdown-Formatierung, keine Kommentare, keine Erklärungen und keinen zusätzlichen Text ausgeben.
|
||
6. Die Ausgabe muss direkt mit json.loads() parsbar sein.
|
||
|
||
Data:
|
||
{recipe_content}
|
||
"""
|
||
# -------------------------
|
||
# Hugging Face call (LEGACY – WORKING)
|
||
# -------------------------
|
||
def call_huggingface(prompt: str) -> str:
|
||
API_URL = "https://router.huggingface.co/v1/chat/completions"
|
||
headers = {
|
||
"Authorization": f"Bearer {HF_API_KEY}",
|
||
}
|
||
|
||
def query(payload):
|
||
response = requests.post(API_URL, headers=headers, json=payload)
|
||
return response.json()
|
||
|
||
response = query({
|
||
"messages": [
|
||
{
|
||
"role": "user",
|
||
"content": prompt
|
||
}
|
||
],
|
||
"model": "Qwen/Qwen2.5-72B:featherless-ai"
|
||
})
|
||
|
||
returnStr = response["choices"][0]["message"]["content"]
|
||
logger.debug(response)
|
||
logger.debug(returnStr)
|
||
return returnStr
|
||
|
||
# -------------------------
|
||
# Groq call
|
||
# -------------------------
|
||
def call_groq(prompt: str) -> str:
|
||
client = OpenAI(
|
||
api_key=GROQ_API_KEY,
|
||
base_url="https://api.groq.com/openai/v1"
|
||
)
|
||
|
||
response = client.chat.completions.create(
|
||
model=GROQ_MODEL,
|
||
messages=[
|
||
{"role": "system", "content": "You return valid JSON only."},
|
||
{"role": "user", "content": prompt},
|
||
],
|
||
temperature=0.3,
|
||
)
|
||
logger.debug(response)
|
||
returnStr = response.choices[0].message.content.strip()
|
||
return returnStr
|
||
|
||
# -------------------------
|
||
# OpenAI call
|
||
# -------------------------
|
||
def call_openai(prompt: str) -> str:
|
||
client = OpenAI(api_key=OPENAI_API_KEY)
|
||
|
||
response = client.chat.completions.create(
|
||
model=OPENAI_MODEL,
|
||
messages=[
|
||
{"role": "system", "content": "You return valid JSON only."},
|
||
{"role": "user", "content": prompt},
|
||
],
|
||
temperature=0.3,
|
||
)
|
||
|
||
return response.choices[0].message.content.strip()
|
||
|
||
|
||
# -------------------------
|
||
# Gemini call
|
||
# -------------------------
|
||
def call_gemini(prompt: str) -> str:
|
||
client = genai.Client(api_key=GEMINI_API_KEY)
|
||
|
||
response = client.models.generate_content(
|
||
model=GEMINI_MODEL,
|
||
contents=prompt,
|
||
)
|
||
|
||
text = response.text.strip()
|
||
|
||
# Defensive cleanup
|
||
if text.startswith("```"):
|
||
text = text.split("```")[1].replace("json", "").strip()
|
||
|
||
return text
|
||
# -------------------------
|
||
# Model router
|
||
# -------------------------
|
||
def call_model(provider: str, prompt: str) -> str:
|
||
if provider == "openai":
|
||
return call_openai(prompt)
|
||
elif provider == "gemini":
|
||
return call_gemini(prompt)
|
||
elif provider == "huggingface":
|
||
return call_huggingface(prompt)
|
||
elif provider == "groq":
|
||
return call_groq(prompt)
|
||
else:
|
||
raise ValueError(f"Unknown provider: {provider}")
|
||
# -------------------------
|
||
# Evaluation
|
||
# -------------------------
|
||
|
||
def list_dict_fields(d, parent_key=""):
|
||
"""
|
||
Recursively lists all keys and subkeys in a dict, returning full paths.
|
||
"""
|
||
fields = []
|
||
|
||
for k, v in d.items():
|
||
full_key = f"{parent_key}.{k}" if parent_key else k
|
||
|
||
fields.append(full_key)
|
||
|
||
if isinstance(v, dict):
|
||
fields.extend(list_dict_fields(v, full_key))
|
||
elif isinstance(v, list):
|
||
# If list of dicts, include their fields
|
||
for i, item in enumerate(v):
|
||
if isinstance(item, dict):
|
||
fields.extend(list_dict_fields(item, f"{full_key}[{i}]"))
|
||
|
||
return fields
|
||
|
||
def evaluate_response(data: dict) -> dict:
|
||
errors = []
|
||
all_fields = list_dict_fields(data)
|
||
for field in REQUIRED_FIELDS + REQUIRED_NUTRITION_FIELDS :
|
||
if not any(field in fields for fields in all_fields):
|
||
errors.append(f"Missing field: {field}")
|
||
|
||
# Check that recipeYield is numeric
|
||
recipe_yield = data.get("recipeYield")
|
||
if not isinstance(recipe_yield, (int, float)):
|
||
errors.append("recipeYield must be numeric")
|
||
|
||
return {
|
||
"valid": len(errors) == 0,
|
||
"errors": errors
|
||
}
|
||
|
||
|
||
# -------------------------
|
||
# Main processing for a single file
|
||
# -------------------------
|
||
|
||
def remove_keys(d: dict, keys_to_remove: list):
|
||
"""
|
||
Remove specified keys from a dictionary (top-level only).
|
||
"""
|
||
for key in keys_to_remove:
|
||
if key in d:
|
||
del d[key]
|
||
return d
|
||
|
||
|
||
def extract_and_remove_keys(d: dict, keys_to_remove: list):
|
||
"""
|
||
Remove specified keys from a dictionary (top-level only) and store their values.
|
||
|
||
Returns a tuple: (cleaned_dict, extracted_values)
|
||
"""
|
||
extracted = {}
|
||
for key in keys_to_remove:
|
||
if key in d:
|
||
extracted[key] = d.pop(key)
|
||
return d, extracted
|
||
|
||
|
||
def process_recipe(file_path: str, provider: str, update_nutrition: bool = True, delete_nutrition: bool = True):
|
||
if not os.path.exists(file_path):
|
||
logger.error(f"❌ File not found: {file_path}")
|
||
return
|
||
|
||
with open(file_path, "r", encoding="utf-8") as f:
|
||
try:
|
||
recipe_data = json.load(f)
|
||
except json.JSONDecodeError:
|
||
logger.error(f"❌ File is not valid JSON: {file_path}")
|
||
return
|
||
|
||
# -------------------------
|
||
# Nutrition check (optional)
|
||
# -------------------------
|
||
if "nutrition" in recipe_data and isinstance(recipe_data["nutrition"], dict):
|
||
all_fields = list_dict_fields(recipe_data["nutrition"])
|
||
for field in REQUIRED_NUTRITION_FIELDS:
|
||
if not any(field in fields for fields in all_fields):
|
||
update_nutrition = True
|
||
logger.warning(f"Missing field: {field}, Nutrition block will be updated.")
|
||
break
|
||
warning = f"⚠️ Nutrition block already exists in {os.path.basename(os.path.dirname(file_path))}/{os.path.basename(file_path)}. "
|
||
if update_nutrition:
|
||
logger.warning(f"{warning:<100} - It will be updated")
|
||
elif delete_nutrition:
|
||
logger.warning(f"{warning:<100} - It will be deleted")
|
||
recipe_data, _ = extract_and_remove_keys(recipe_data, ["nutrition"])
|
||
with open(file_path, "w", encoding="utf-8") as f:
|
||
json.dump(recipe_data, f, indent=2, ensure_ascii=False)
|
||
return True
|
||
else:
|
||
logger.warning(f"{warning:<100} - Skipping recipe")
|
||
return True
|
||
|
||
else:
|
||
logger.info(f"ℹ️ No nutrition block found in {file_path}. A new one will be added.")
|
||
|
||
recipe_data,_ = extract_and_remove_keys(recipe_data, ["video","aggregateRating","reviews"])
|
||
recipe_data, stored_keys = extract_and_remove_keys(recipe_data, ["valueInit"])
|
||
|
||
|
||
# Serialize JSON back to string for prompt
|
||
recipe_content = json.dumps(recipe_data, ensure_ascii=False)
|
||
prompt = generate_prompt(recipe_content)
|
||
|
||
logger.info(f"🚀 Using provider: {provider} → {file_path}")
|
||
|
||
logger.debug(prompt)
|
||
|
||
while(True):
|
||
try:
|
||
raw_response = call_model(provider, prompt)
|
||
start = raw_response.find("{")
|
||
end = raw_response.rfind("}")
|
||
raw_response = raw_response[start:end + 1]
|
||
updated_data = json.loads(raw_response)
|
||
|
||
evaluation = evaluate_response(updated_data)
|
||
if not evaluation["valid"]:
|
||
logger.error("❌ Evaluation failed. File NOT updated.")
|
||
for e in evaluation["errors"]:
|
||
logger.error(f"{e}")
|
||
return
|
||
|
||
combined = updated_data | stored_keys
|
||
|
||
with open(file_path, "w", encoding="utf-8") as f:
|
||
json.dump(combined, f, indent=2, ensure_ascii=False)
|
||
|
||
logger.info("✅ Recipe updated successfully.")
|
||
break
|
||
|
||
except json.JSONDecodeError:
|
||
logger.error("❌ Model did not return valid JSON.")
|
||
logger.error("RAW OUTPUT:\n", raw_response)
|
||
|
||
except Exception as e:
|
||
msg = str(e)
|
||
if "rate limit" in msg.lower():
|
||
# Extract wait time
|
||
match = re.search(r'Please try again in (\d+)m([\d\.]+)s', msg)
|
||
logger.warning(msg)
|
||
if match:
|
||
minutes = int(match.group(1))
|
||
seconds = float(match.group(2))
|
||
wait_time = (minutes * 60 + seconds) *1.5
|
||
logger.warning(f"Rate limit hit. Waiting for {wait_time:.2f} seconds...")
|
||
for remaining in range(int(wait_time), 0, -1):
|
||
print(f"\rTime left: {remaining}s", end="")
|
||
time.sleep(1)
|
||
logger.info("\nRetrying now...")
|
||
else:
|
||
logger.error("Rate limit hit, could not parse wait time. Waiting 200 seconds...")
|
||
return False
|
||
|
||
else:
|
||
logger.error("❌ Error:", e)
|
||
return False
|
||
return True
|
||
|
||
# -------------------------
|
||
# Recursively find all recipe.json files in a folder
|
||
# -------------------------
|
||
def process_folder(folder_path: str, provider: str, update_nutrition: bool = True , delete_nutrition: bool = True):
|
||
json_files = []
|
||
|
||
for root, dirs, files in os.walk(folder_path):
|
||
for file in files:
|
||
if file.lower() == "recipe.json":
|
||
full_path = os.path.join(root, file)
|
||
json_files.append(os.path.join(root, file))
|
||
i = 1
|
||
for file in json_files:
|
||
logger.info(f"Checking: {i}/{len(json_files)} ... {os.path.basename(os.path.dirname(file))}/{os.path.basename(file)}.")
|
||
i+=1
|
||
if not process_recipe(file, provider, update_nutrition, delete_nutrition):
|
||
logger.warning(f"File: {file} failed. Continuing")
|
||
|
||
# -------------------------
|
||
# CLI entry point
|
||
# -------------------------
|
||
if __name__ == "__main__":
|
||
script_name = os.path.splitext(os.path.basename(__file__))[0]
|
||
|
||
log_path = os.path.join(os.path.expanduser("~"), ".LOG",f"{script_name}",
|
||
f"{script_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
|
||
)
|
||
logger = setup_logger(log_file=log_path)
|
||
logger.info("Logger initialized.")
|
||
|
||
parser = argparse.ArgumentParser()
|
||
parser.add_argument("path", help="Recipe JSON file or folder containing recipe.json files")
|
||
parser.add_argument(
|
||
"--provider",
|
||
choices=["openai", "gemini", "huggingface", "groq"],
|
||
default="groq",
|
||
help="LLM provider to use"
|
||
)
|
||
parser.add_argument(
|
||
"-un", "--update_nutrition",
|
||
action="store_true",
|
||
help="Enable nutrition block check and update"
|
||
)
|
||
parser.add_argument(
|
||
"-dn", "--delete_nutrition",
|
||
action="store_true",
|
||
help="Enable nutrition block deletion"
|
||
)
|
||
|
||
args = parser.parse_args()
|
||
path = args.path
|
||
provider = args.provider
|
||
update_nutrition = args.update_nutrition
|
||
delete_nutrition = args.delete_nutrition
|
||
# path = "/data/Nextcloud/Documents/Dokumente/Shared/Diverses/Kochbuch/RecepiesLucas/Sauerteig herstellen und züchten"
|
||
# update_nutrition = False
|
||
# delete_nutrition = False
|
||
|
||
if os.path.isdir(path):
|
||
process_folder(path, provider, update_nutrition, delete_nutrition)
|
||
else:
|
||
process_recipe(path, provider, update_nutrition, delete_nutrition)
|