zhangmenu/menu/services/parser_service.py

from urllib.parse import urlparse

import requests
from bs4 import BeautifulSoup
from django.conf import settings
import re
import json


class ParserService:
    """URL解析业务逻辑"""

    @staticmethod
    def parse_url(url):
        """解析外部URL获取菜品信息"""
        if not url:
            raise ValueError('请提供URL')

        parsed_url = urlparse(url)
        hostname = parsed_url.hostname
        allowed_hosts = getattr(settings, 'ALLOWED_RECIPE_HOSTS', set())

        if parsed_url.scheme not in {'http', 'https'}:
            raise ValueError('URL协议不受支持')
        if not hostname:
            raise ValueError('URL缺少主机名')
        if hostname not in allowed_hosts:
            raise ValueError('该站点不在允许的解析范围内')

        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
        }
        response = requests.get(url, headers=headers, timeout=10)
        response.encoding = 'utf-8'
        soup = BeautifulSoup(response.text, 'html.parser')

        result = {
            'name': '',
            'cover_image': '',
            'description': '',
            'ingredients': []
        }

        # 根据不同网站选择解析方法
        if hostname in {'xiachufang.com', 'www.xiachufang.com'}:
            result = ParserService._parse_xiachufang(soup, url)
        elif hostname in {'meishij.net', 'www.meishij.net'}:
            result = ParserService._parse_meishij(soup)
        elif hostname in {'caipuw.com', 'www.caipuw.com'}:
            result = ParserService._parse_caipuw(soup)
        else:
            result = ParserService._parse_generic(soup)

        return result

    @staticmethod
    def _parse_xiachufang(soup, url):
        """解析下厨房网站"""
        result = {
            'name': '',
            'cover_image': '',
            'description': '',
            'ingredients': []
        }

        # 尝试从JSON-LD获取数据
        json_ld = soup.find_all('script', type='application/ld+json')
        for script in json_ld:
            try:
                data = json.loads(script.string)
                if data.get('@type') == 'Recipe':
                    result['name'] = data.get('name', '')
                    result['cover_image'] = data.get('image', '')
                    if isinstance(result['cover_image'], list):
                        result['cover_image'] = result['cover_image'][0]

                    # 解析材料
                    for ing in data.get('recipeIngredient', []):
                        match = re.match(r'^([\d.]+\s*[克g个勺片块大中小]+)?\s*(.+)$', ing)
                        if match and match.group(2):
                            amount, name = match.groups()
                            result['ingredients'].append({
                                'name': name.strip(),
                                'amount': (amount.strip() + '克') if amount and not ('克' in amount or '个' in amount or '勺' in amount or '片' in amount or '块' in amount) else (amount.strip() if amount else '')
                            })
                        else:
                            result['ingredients'].append({'name': ing, 'amount': ''})

                    # 解析做法
                    steps = []
                    recipe_steps = data.get('recipeInstructions', [])
                    if isinstance(recipe_steps, list):
                        for step in recipe_steps:
                            text = step.get('text', '') if isinstance(step, dict) else str(step)
                            if text:
                                steps.append(text)
                    elif isinstance(recipe_steps, str):
                        parts = re.split(r'(?=\d+\.)', recipe_steps)
                        for part in parts:
                            part = re.sub(r'^\d+\.\s*', '', part.strip())
                            if part:
                                steps.append(part)

                    if steps:
                        result['description'] = '\n\n'.join([f'{i+1}. {s}' for i, s in enumerate(steps)])
                    break
            except (TypeError, ValueError, json.JSONDecodeError):
                continue

        # 回退到HTML解析
        if not result['name']:
            title = soup.select_one('h1.title')
            if title:
                result['name'] = title.get_text(strip=True)

        if not result['cover_image']:
            cover = soup.select_one('div.cover img') or soup.select_one('div.image img')
            if cover:
                result['cover_image'] = cover.get('src', '')

        return result

    @staticmethod
    def _parse_meishij(soup):
        """解析美食杰网站"""
        result = {
            'name': '',
            'cover_image': '',
            'description': '',
            'ingredients': []
        }

        title = soup.select_one('h1.recipe_title')
        if title:
            result['name'] = title.get_text(strip=True)

        cover = soup.select_one('div.recipe_img img')
        if cover:
            result['cover_image'] = cover.get('src', '')

        # 材料
        ing_list = soup.select('div.yl li')
        for ing in ing_list:
            text = ing.get_text(strip=True)
            if text:
                parts = text.split()
                if len(parts) >= 2:
                    result['ingredients'].append({'name': parts[0], 'amount': ' '.join(parts[1:])})
                elif len(parts) == 1:
                    result['ingredients'].append({'name': parts[0], 'amount': ''})

        # 做法
        steps = []
        step_list = soup.select('div.step_content')
        for step in step_list:
            text = step.get_text(strip=True)
            if text:
                steps.append(text)
        result['description'] = '\n\n'.join([f'{i+1}. {s}' for i, s in enumerate(steps)])

        return result

    @staticmethod
    def _parse_caipuw(soup):
        """解析菜谱大全网站"""
        result = {
            'name': '',
            'cover_image': '',
            'description': '',
            'ingredients': []
        }

        title = soup.select_one('h1.title')
        if title:
            result['name'] = title.get_text(strip=True)

        cover = soup.select_one('div.image img')
        if cover:
            result['cover_image'] = cover.get('src', '')

        # 材料
        ing_list = soup.select('div.ingredients li') or soup.select('ul.ingredients li')
        for ing in ing_list:
            text = ing.get_text(strip=True)
            if text:
                result['ingredients'].append({'name': text, 'amount': ''})

        # 做法
        steps = []
        step_list = soup.select('div.steps li') or soup.select('div.step')
        for step in step_list:
            text = step.get_text(strip=True)
            if text:
                steps.append(text)
        result['description'] = '\n\n'.join([f'{i+1}. {s}' for i, s in enumerate(steps)])

        return result

    @staticmethod
    def _parse_generic(soup):
        """通用解析"""
        result = {
            'name': '',
            'cover_image': '',
            'description': '',
            'ingredients': []
        }

        title = soup.select_one('h1') or soup.select_one('title')
        if title:
            result['name'] = title.get_text(strip=True)[:100]

        img = soup.select_one('article img') or soup.select_one('div.content img') or soup.select_one('img')
        if img:
            result['cover_image'] = img.get('src', '')

        return result