275 lines
11 KiB
Python
275 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
import xml.etree.ElementTree as ET
|
|
from typing import Any
|
|
from urllib import parse, request
|
|
|
|
from app.config import settings
|
|
from app.services.recipe_importer import normalize_recipe_from_url
|
|
from app.services.repository import repo
|
|
from app.services.scoring import extract_ingredients, extract_tags, score_recipe
|
|
from app.services.store import JsonStore
|
|
|
|
|
|
SEARCH_URL = "https://www.bing.com/search?format=rss&q="
|
|
ENGLISH_HINT = " site:allrecipes.com OR site:budgetbytes.com OR site:seriouseats.com OR site:recipetineats.com OR site:spendwithpennies.com OR site:delish.com OR site:food.com"
|
|
ALLOWED_DOMAINS = (
|
|
"allrecipes.com",
|
|
"budgetbytes.com",
|
|
"seriouseats.com",
|
|
"recipetineats.com",
|
|
"spendwithpennies.com",
|
|
"delish.com",
|
|
"food.com",
|
|
"feelgoodfoodie.net",
|
|
"simplyrecipes.com",
|
|
"themediterraneandish.com",
|
|
)
|
|
FALLBACK_SEED_URLS = [
|
|
"https://www.budgetbytes.com/curry-beef-with-peas/",
|
|
"https://www.recipetineats.com/asian-beef-bowls/",
|
|
"https://www.spendwithpennies.com/ground-beef-stroganoff/",
|
|
"https://www.budgetbytes.com/chicken-and-rice-skillet/",
|
|
]
|
|
QUERY_URL_FALLBACKS = {
|
|
"colcannon recipe": [
|
|
"https://www.allrecipes.com/recipe/58642/dianes-colcannon/",
|
|
"https://www.spendwithpennies.com/colcannon-recipe-cabbage-and-potatoes/",
|
|
],
|
|
"boxty recipe": [
|
|
"https://www.delish.com/cooking/recipe-ideas/a42452906/boxty-recipe/",
|
|
],
|
|
"coddle recipe": [
|
|
"https://www.foodnetwork.com/recipes/food-network-kitchen/irish-coddle-12348168",
|
|
],
|
|
"corned beef and cabbage recipe": [
|
|
"https://www.spendwithpennies.com/corned-beef-and-cabbage/",
|
|
"https://www.thepioneerwoman.com/food-cooking/recipes/a11462/corned-beef-and-cabbage/",
|
|
],
|
|
"shepherds pie recipe": [
|
|
"https://natashaskitchen.com/shepherds-pie-recipe/",
|
|
],
|
|
"kofta recipe ground beef": [
|
|
"https://www.allrecipes.com/recipe/106030/kofta-kebabs/",
|
|
],
|
|
"kafta recipe ground beef": [
|
|
"https://feelgoodfoodie.net/recipe/beef-kafta/",
|
|
],
|
|
"arayes recipe": [
|
|
"https://www.recipetineats.com/arayes-lebanese-meat-stuffed-pita/",
|
|
"https://www.allrecipes.com/arayes-lebanese-crispy-meat-stuffed-pita-recipe-8635793",
|
|
],
|
|
"hashweh recipe": [
|
|
"https://www.themediterraneandish.com/lebanese-rice-hashweh-recipe/",
|
|
],
|
|
"beef shawarma recipe": [
|
|
"https://www.themediterraneandish.com/grilled-beef-shawarma/",
|
|
],
|
|
"kofta kebabs recipe": [
|
|
"https://www.allrecipes.com/recipe/106030/kofta-kebabs/",
|
|
],
|
|
"kibbeh recipe": [
|
|
"https://feelgoodfoodie.net/recipe/baked-kibbeh/",
|
|
],
|
|
}
|
|
QUERY_STOPWORDS = {
|
|
"a", "an", "and", "for", "from", "in", "inspired", "meal", "meals",
|
|
"recipe", "recipes", "style", "using", "with",
|
|
}
|
|
CUE_WORDS = {
|
|
"asian", "irish", "italian", "mexican", "thai", "indian", "korean",
|
|
"japanese", "greek", "mediterranean", "southern", "cajun", "middle", "eastern",
|
|
}
|
|
TITLE_STOPWORDS = {
|
|
"a", "an", "and", "best", "classic", "easy", "for", "how", "make",
|
|
"recipe", "recipes", "style", "the", "to", "traditional", "slow", "cooker", "s",
|
|
}
|
|
|
|
|
|
class RecipeSearchService:
|
|
def __init__(self) -> None:
|
|
self.store = JsonStore()
|
|
|
|
def _dish_key(self, title: str, payload: dict[str, Any]) -> str:
|
|
base = str(title or payload.get("name") or "").lower()
|
|
base = re.sub(r"\([^)]*\)", " ", base)
|
|
base = re.sub(r"[^a-z0-9\s-]", " ", base)
|
|
tokens = [token for token in re.split(r"\s+", base) if token and token not in TITLE_STOPWORDS]
|
|
if len(tokens) > 1 and tokens[0].endswith("s"):
|
|
tokens = tokens[1:]
|
|
canonical_tokens = sorted(dict.fromkeys(tokens))
|
|
return " ".join(canonical_tokens[:8]).strip()
|
|
|
|
def _candidate_queries(self, query: str) -> list[str]:
|
|
normalized = re.sub(r"[^a-z0-9\s-]", " ", query.lower())
|
|
normalized = re.sub(r"\s+", " ", normalized).strip()
|
|
tokens = [token for token in normalized.split() if token]
|
|
filtered = [token for token in tokens if token not in QUERY_STOPWORDS]
|
|
cues = [token for token in filtered if token in CUE_WORDS]
|
|
ingredients = [token for token in filtered if token not in CUE_WORDS]
|
|
expansion_queries: list[str] = []
|
|
filtered_set = set(filtered)
|
|
if {"irish", "cabbage", "potatoes"} <= filtered_set:
|
|
expansion_queries.extend([
|
|
"colcannon recipe",
|
|
"boxty recipe",
|
|
"coddle recipe",
|
|
"corned beef and cabbage recipe",
|
|
"shepherds pie recipe",
|
|
])
|
|
if {"middle", "eastern", "ground", "beef"} <= filtered_set:
|
|
expansion_queries.extend([
|
|
"kofta recipe ground beef",
|
|
"kafta recipe ground beef",
|
|
"middle eastern beef kofta",
|
|
"arayes recipe",
|
|
"hashweh recipe",
|
|
"beef shawarma recipe",
|
|
"kofta kebabs recipe",
|
|
"kibbeh recipe",
|
|
])
|
|
variants = [
|
|
query.strip(),
|
|
" ".join(filtered),
|
|
" ".join(cues + ingredients + ["recipe"]).strip(),
|
|
" ".join(ingredients + ["recipe"]).strip(),
|
|
] + expansion_queries
|
|
deduped: list[str] = []
|
|
seen: set[str] = set()
|
|
for variant in variants:
|
|
cleaned = variant.strip()
|
|
if not cleaned:
|
|
continue
|
|
key = cleaned.lower()
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
deduped.append(cleaned)
|
|
return deduped
|
|
|
|
def _search_page(self, query: str) -> list[dict[str, str]]:
|
|
if query in QUERY_URL_FALLBACKS:
|
|
return [{"url": url, "title": url, "snippet": ""} for url in QUERY_URL_FALLBACKS[query]]
|
|
req = request.Request(
|
|
SEARCH_URL + parse.quote(query + ENGLISH_HINT),
|
|
headers={
|
|
"User-Agent": "Mozilla/5.0 (Doris Kitchen)",
|
|
"Accept-Language": "en-US,en;q=0.9",
|
|
},
|
|
)
|
|
with request.urlopen(req, timeout=30) as resp:
|
|
rss_text = resp.read().decode("utf-8", "replace")
|
|
root = ET.fromstring(rss_text)
|
|
results: list[dict[str, str]] = []
|
|
for item in root.findall("./channel/item"):
|
|
href = str(item.findtext("link") or "").strip()
|
|
title = str(item.findtext("title") or "").strip()
|
|
snippet = str(item.findtext("description") or "").strip()
|
|
host = parse.urlparse(href).netloc.lower()
|
|
if not href.startswith("http"):
|
|
continue
|
|
if not any(host.endswith(domain) or host == domain for domain in ALLOWED_DOMAINS):
|
|
continue
|
|
results.append({"url": href, "title": title, "snippet": snippet})
|
|
deduped: list[dict[str, str]] = []
|
|
seen: set[str] = set()
|
|
for item in results:
|
|
if item["url"] in seen:
|
|
continue
|
|
seen.add(item["url"])
|
|
deduped.append(item)
|
|
return deduped
|
|
|
|
def search(self, query: str, limit: int | None = None) -> list[dict[str, Any]]:
|
|
limit = limit or settings.search_result_limit
|
|
profile = repo.get_profile()
|
|
results: list[dict[str, Any]] = []
|
|
seen_urls: set[str] = set()
|
|
seen_dishes: set[str] = set()
|
|
for candidate_query in self._candidate_queries(query):
|
|
before_count = len(results)
|
|
for hit in self._search_page(candidate_query):
|
|
if len(results) >= limit:
|
|
break
|
|
if hit["url"] in seen_urls:
|
|
continue
|
|
try:
|
|
payload = normalize_recipe_from_url(hit["url"], timeout=12)
|
|
except Exception:
|
|
continue
|
|
dish_key = self._dish_key(str(payload.get("name") or hit["title"] or ""), payload)
|
|
if dish_key and dish_key in seen_dishes:
|
|
continue
|
|
score, reasons = score_recipe(query, payload, profile)
|
|
suggestion = {
|
|
"id": self.store.new_id("suggestion"),
|
|
"title": payload.get("name") or hit["title"],
|
|
"url": hit["url"],
|
|
"source_query": query,
|
|
"source_title": hit["title"],
|
|
"source_snippet": hit["snippet"],
|
|
"payload": payload,
|
|
"ingredients": extract_ingredients(payload),
|
|
"tags": extract_tags(payload),
|
|
"score": score,
|
|
"score_reasons": reasons,
|
|
"status": "suggested",
|
|
"created_at": self.store.now(),
|
|
"updated_at": self.store.now(),
|
|
"kitchenowl_status": "not-imported",
|
|
"ready_for_auto_import": bool(profile.get("auto_import_enabled") and score >= int(profile.get("auto_import_threshold", settings.auto_import_threshold))),
|
|
}
|
|
saved = repo.upsert_suggestion(suggestion)
|
|
results.append(saved)
|
|
seen_urls.add(hit["url"])
|
|
if dish_key:
|
|
seen_dishes.add(dish_key)
|
|
if len(results) >= limit:
|
|
break
|
|
return results
|
|
|
|
def _fallback_seed(self) -> list[dict[str, Any]]:
|
|
profile = repo.get_profile()
|
|
results: list[dict[str, Any]] = []
|
|
for url in FALLBACK_SEED_URLS:
|
|
try:
|
|
payload = normalize_recipe_from_url(url, timeout=12)
|
|
except Exception:
|
|
continue
|
|
score, reasons = score_recipe("seeded household dinner", payload, profile)
|
|
suggestion = {
|
|
"id": self.store.new_id("suggestion"),
|
|
"title": payload.get("name") or url,
|
|
"url": url,
|
|
"source_query": "seeded household dinner",
|
|
"source_title": payload.get("name") or url,
|
|
"source_snippet": "Curated fallback suggestion for initial household testing.",
|
|
"payload": payload,
|
|
"ingredients": extract_ingredients(payload),
|
|
"tags": extract_tags(payload),
|
|
"score": score,
|
|
"score_reasons": reasons,
|
|
"status": "suggested",
|
|
"created_at": self.store.now(),
|
|
"updated_at": self.store.now(),
|
|
"kitchenowl_status": "not-imported",
|
|
"ready_for_auto_import": False,
|
|
}
|
|
saved = repo.upsert_suggestion(suggestion)
|
|
results.append(saved)
|
|
return results
|
|
|
|
def seed_queries(self) -> list[dict[str, Any]]:
|
|
profile = repo.get_profile()
|
|
queries = profile.get("seed_queries") or []
|
|
seeded: list[dict[str, Any]] = []
|
|
for query in queries:
|
|
seeded.extend(self.search(query, limit=2))
|
|
if not seeded:
|
|
seeded = self._fallback_seed()
|
|
return seeded
|
|
|
|
|
|
recipe_search = RecipeSearchService()
|