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()