vrobbler/vrobbler/apps/foods/sources/rscraper.py

# services/recipe_scraper.py
from recipe_scrapers import scrape_html
from recipe_scrapers._exceptions import (
    NoSchemaFoundInWildMode,
    SchemaOrgException,
)
import requests
from typing import Dict, Optional, List
from django.core.files.base import ContentFile


class RecipeScraperService:
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update(
            {"User-Agent": "Mozilla/5.0 (compatible; Vrobbler/0.3)"}
        )

    def scrape(self, html: str, org_url: str) -> Dict:
        """Scrape recipe data from HTML.

        Args:
            html: The HTML content of the page
            org_url: The original URL the HTML was fetched from
        """
        try:
            scraper = scrape_html(html, org_url=org_url, supported_only=False)
        except Exception as e:
            raise ValueError(f"Failed to scrape {org_url}: {str(e)}")

        data_dict = {
            "title": scraper.title(),
            "ingredients": self._safe_call(scraper, "ingredients", []),
            "instructions": self._safe_call(scraper, "instructions_list", []),
            "yields": self._safe_call(scraper, "yields"),
            "nutrition": self._safe_call(scraper, "nutrition"),
            "course": self._safe_call(scraper, "course"),
            "dietary": self._safe_call(scraper, "dietary", []),
            "prep_time": self._safe_call(scraper, "prep_time"),
            "total_time": self._safe_call(scraper, "total_time"),
            "image": self._safe_call(scraper, "image"),
            "cook_time": self._safe_call(scraper, "cook_time"),
            "cuisine": self._safe_call(scraper, "cuisine"),
            "site": self._safe_call(scraper, "host"),
            "url": org_url,
            "category": self._safe_call(scraper, "category"),
            "keywords": self._safe_call(scraper, "keywords"),
        }

        return data_dict

    def scrape_url(self, url: str):
        response = self.session.get(url)
        if response.status_code != 200:
            raise Exception("Recipe website returned non 200 response", e)
        return self.scrape(response.text, org_url=url)

    @classmethod
    def is_recipe(self, html: str, org_url: str) -> bool:
        """Check if HTML contains a recipe.

        Args:
            html: The HTML content to check
            org_url: The original URL the HTML was fetched from
        """
        try:
            scraper = scrape_html(html, org_url=org_url, supported_only=False)
            return scraper.schema is not None
        except NoSchemaFoundInWildMode:
            return False
        except Exception:
            return False

    @classmethod
    def is_recipe_url(cls, url: str) -> bool:
        """Check if a URL points to a recipe by fetching and checking the HTML."""
        try:
            response = requests.get(
                url,
                timeout=10,
                headers={
                    "User-Agent": "Mozilla/5.0 (compatible; Vrobbler/0.3)"
                },
            )
            if response.status_code != 200:
                logger.debug("Recipe website returned non 200 response")
                return False
            return cls.is_recipe(response.text, url)
        except Exception as e:
            print(e)
            return False

    def _safe_call(self, scraper, method, default=None):
        """
        Safely call a scraper method, returning default on error.
        Handles both missing methods (AttributeError) and methods that return None.
        """
        try:
            # Check if method exists before calling
            if not hasattr(scraper, method):
                return default

            result = getattr(scraper, method)()
            return result if result else default
        except (
            AttributeError,
            TypeError,
            KeyError,
            ValueError,
            SchemaOrgException,
        ):
            # AttributeError: method doesn't exist on this scraper class
            # TypeError: method exists but can't be called
            # KeyError: method exists but data structure is unexpected
            return default

    def download_image(self, image_url: str) -> Optional[ContentFile]:
        """Download recipe image and return as Django ContentFile."""
        if not image_url:
            return None

        try:
            response = self.session.get(image_url, timeout=10)
            response.raise_for_status()

            filename = image_url.split("/")[-1].split("?")[0]
            if not filename.endswith((".jpg", ".jpeg", ".png", ".webp")):
                filename = "recipe_image.jpg"

            return ContentFile(response.content, name=filename)
        except Exception:
            return None

    def parse_servings(self, yield_text: str) -> Optional[int]:
        """Extract number of servings from yield text."""
        import re

        if not yield_text:
            return None

        match = re.search(r"(\d+)", yield_text)
        return int(match.group(1)) if match else None

    def extract_tags(self, recipe_data: Dict) -> List[str]:
        """Extract all tags from recipe metadata."""
        tags = set()

        # Add cuisine
        if recipe_data.get("cuisine"):
            cuisine = recipe_data["cuisine"]
            if isinstance(cuisine, str):
                tags.add(cuisine.strip())
            elif isinstance(cuisine, list):
                tags.update([c.strip() for c in cuisine if c])

        # Add dietary
        if recipe_data.get("dietary"):
            dietary = recipe_data["dietary"]
            if isinstance(dietary, str):
                tags.add(dietary.strip())
            elif isinstance(dietary, list):
                tags.update([d.strip() for d in dietary if d])

        # Add course
        if recipe_data.get("course"):
            course = recipe_data["course"]
            if isinstance(course, str):
                tags.add(course.strip())
            elif isinstance(course, list):
                tags.update([c.strip() for c in course if c])

        # Add keywords
        keywords = recipe_data.get("keywords")
        if keywords:
            if isinstance(keywords, str):
                tags.update(
                    [k.strip() for k in keywords.split(",") if k.strip()]
                )
            elif isinstance(keywords, list):
                tags.update([k.strip() for k in keywords if k.strip()])

        return list(tags)