レストランメニューのデータパイプライン AI注文システム用

from datetime import datetime
from decimal import Decimal
from typing import Literal
from pydantic import BaseModel, Field, HttpUrl

class Money(BaseModel):
    currency: str = Field(min_length=3, max_length=3)
    amount: Decimal
    tax_included: bool | None = None

class Evidence(BaseModel):
    source_type: Literal[
        "OWNER_API",
        "OWNER_EXPORT",
        "BUSINESS_PROFILE",
        "JSON_LD",
        "AUTHORIZED_PAGE",
        "IMAGE_REVIEW",
    ]
    source_url: HttpUrl | None = None
    source_record_id: str | None = None
    observed_at: datetime
    source_modified_at: datetime | None = None
    content_hash: str
    language: str

class MenuOption(BaseModel):
    option_id: str
    name: str
    price_delta: Money | None = None
    available: bool | None = None

class MenuItem(BaseModel):
    item_id: str
    restaurant_id: str
    location_id: str
    menu_id: str
    section_id: str
    name: str
    description: str | None = None
    base_price: Money | None = None
    options: list[MenuOption] = []
    dietary_labels: list[str] = []
    allergens: list[str] = []
    ingredients: list[str] = []
    available: bool | None = None
    evidence: Evidence

class Restaurant(BaseModel):
    restaurant_id: str
    brand_name: str
    legal_name: str | None = None

class Location(BaseModel):
    location_id: str
    restaurant_id: str
    address_line: str
    city: str
    region: str | None = None
    postal_code: str | None = None
    country_code: str
    timezone: str

class Menu(BaseModel):
    menu_id: str
    location_id: str
    name: str
    service_modes: list[str]
    dayparts: list[str]
    valid_from: datetime | None = None
    valid_until: datetime | None = None
    language: str

from hashlib import sha256
import json


def canonical_hash(payload: dict) -> str:
    encoded = json.dumps(
        payload,
        sort_keys=True,
        ensure_ascii=False,
        separators=(",", ":"),
    ).encode("utf-8")
    return sha256(encoded).hexdigest()

async def ingest_owner_menu(api, location_id: str) -> list[MenuItem]:
    metadata = await api.get_menu_metadata(location_id)
    if metadata.not_modified:
        return []

    payload = await api.get_menu(location_id)
    observed_at = utc_now()
    records = []

    for menu in payload["menus"]:
        for section in menu.get("sections", []):
            for item in section.get("items", []):
                records.append(
                    normalize_owner_item(
                        location_id=location_id,
                        menu=menu,
                        section=section,
                        item=item,
                        observed_at=observed_at,
                        content_hash=canonical_hash(item),
                    )
                )
    return records

class NormalizedPrice(BaseModel):
    amount: Decimal
    currency: str
    original_text: str | None = None
    price_type: Literal[
        "FIXED",
        "FROM",
        "RANGE",
        "MARKET_PRICE",
        "INCLUDED",
        "UNKNOWN",
    ]
    upper_amount: Decimal | None = None


def normalize_price(raw: dict, currency: str) -> NormalizedPrice | None:
    if raw.get("market_price"):
        return NormalizedPrice(
            amount=Decimal("0"),
            currency=currency,
            original_text=raw.get("display"),
            price_type="MARKET_PRICE",
        )
    if raw.get("min") is not None and raw.get("max") is not None:
        return NormalizedPrice(
            amount=Decimal(str(raw["min"])),
            upper_amount=Decimal(str(raw["max"])),
            currency=currency,
            original_text=raw.get("display"),
            price_type="RANGE",
        )
    if raw.get("amount") is not None:
        return NormalizedPrice(
            amount=Decimal(str(raw["amount"])),
            currency=currency,
            original_text=raw.get("display"),
            price_type="FIXED",
        )
    return None

class ModifierChoice(BaseModel):
    choice_id: str
    name: str
    price_delta: Money | None = None
    available: bool | None = None

class ModifierGroup(BaseModel):
    group_id: str
    name: str
    minimum_selections: int
    maximum_selections: int
    choices: list[ModifierChoice]

class OrderableItem(MenuItem):
    modifier_groups: list[ModifierGroup] = []

class LocalizedText(BaseModel):
    language: str
    value: str
    source_value: str
    translated: bool = False
    translation_model: str | None = None

class LocalizedMenuItem(BaseModel):
    item_id: str
    names: list[LocalizedText]
    descriptions: list[LocalizedText]

class SafetyClaim(BaseModel):
    claim: str
    source_type: str
    explicit_source_text: str
    confidence: float
    reviewed: bool
    reviewer_id: str | None = None


def allow_safety_claim(claim: SafetyClaim) -> bool:
    return (
        claim.source_type in {"OWNER_API", "OWNER_EXPORT", "BUSINESS_PROFILE"}
        and bool(claim.explicit_source_text.strip())
        and claim.reviewed
    )

import json
from bs4 import BeautifulSoup


def extract_json_ld_menu(html: str) -> list[dict]:
    soup = BeautifulSoup(html, "html.parser")
    menus = []
    for script in soup.select('script[type="application/ld+json"]'):
        try:
            payload = json.loads(script.string or "")
        except json.JSONDecodeError:
            continue

        nodes = payload if isinstance(payload, list) else [payload]
        for node in nodes:
            if not isinstance(node, dict):
                continue
            if node.get("@type") == "Menu":
                menus.append(node)
            graph = node.get("@graph", [])
            menus.extend(
                entry
                for entry in graph
                if isinstance(entry, dict) and entry.get("@type") == "Menu"
            )
    return menus

class CollectionDecision(BaseModel):
    source_type: str
    authorized: bool
    public_fields_only: bool
    challenge_type: str | None = None
    rate_limit_ok: bool


def may_use_challenge_service(decision: CollectionDecision) -> bool:
    return (
        decision.source_type == "AUTHORIZED_PAGE"
        and decision.authorized
        and decision.public_fields_only
        and decision.rate_limit_ok
        and decision.challenge_type in {
            "RECAPTCHA_V2",
            "RECAPTCHA_V3",
            "CLOUDFLARE_TURNSTILE",
            "CLOUDFLARE_CHALLENGE",
        }
    )

class SourceLedgerEntry(BaseModel):
    run_id: str
    restaurant_id: str
    location_id: str
    source_type: str
    source_url: str | None
    permission_basis: str
    observed_at: datetime
    source_modified_at: datetime | None
    record_count: int
    content_hash: str
    challenge_observed: bool
    human_review_required: bool

from datetime import timedelta

FRESHNESS = {
    "availability": timedelta(minutes=15),
    "price": timedelta(hours=6),
    "description": timedelta(days=7),
    "dietary_labels": timedelta(days=7),
    "allergens": timedelta(days=1),
    "media": timedelta(days=30),
}


def is_fresh(field: str, observed_at: datetime, now: datetime) -> bool:
    maximum_age = FRESHNESS[field]
    return now - observed_at <= maximum_age

class MenuChange(BaseModel):
    restaurant_id: str
    location_id: str
    item_id: str
    field: str
    before: object
after: object
    source_before: str
    source_after: str
    requires_review: bool

HIGH_RISK_FIELDS = {"allergens", "dietary_labels", "availability", "price"}


def compare_items(previous: MenuItem, current: MenuItem) -> list[MenuChange]:
    changes = []
    for field in [
        "name",
        "description",
        "base_price",
        "options",
        "dietary_labels",
        "allergens",
        "available",
    ]:
        before = getattr(previous, field)
        after = getattr(current, field)
        if before != after:
            changes.append(
                MenuChange(
                    restaurant_id=current.restaurant_id,
                    location_id=current.location_id,
                    item_id=current.item_id,
                    field=field,
                    before=before,
                    after=after,
                    source_before=previous.evidence.content_hash,
                    source_after=current.evidence.content_hash,
                    requires_review=field in HIGH_RISK_FIELDS,
                )
            )
    return changes

class OrderProposal(BaseModel):
    location_id: str
    item_id: str
    option_ids: list[str]
    quoted_total: Money
    menu_observed_at: datetime
    source_record_id: str
    safety_claims_reviewed: bool


def may_present_for_confirmation(proposal: OrderProposal, now: datetime) -> bool:
    return (
        is_fresh("price", proposal.menu_observed_at, now)
        and proposal.safety_claims_reviewed
        and proposal.quoted_total.amount >= 0
    )

class QualityResult(BaseModel):
    accepted: bool
    reasons: list[str]


def validate_item(item: MenuItem) -> QualityResult:
    reasons = []
    if not item.name.strip():
        reasons.append("missing_name")
    if item.base_price and item.base_price.amount < 0:
        reasons.append("negative_price")
    if item.evidence.source_type == "IMAGE_REVIEW":
        reasons.append("image_source_requires_review")
    if item.allergens and item.evidence.source_type not in {
        "OWNER_API",
        "OWNER_EXPORT",
        "BUSINESS_PROFILE",
    }:
        reasons.append("allergen_source_requires_review")
    return QualityResult(accepted=not reasons, reasons=reasons)

import pytest


def test_market_price_is_not_zero_price():
    value = normalize_price(
        {"market_price": True, "display": "Market price"},
        "USD",
    )
    assert value.price_type == "MARKET_PRICE"
    assert value.original_text == "Market price"


def test_unreviewed_allergen_claim_is_blocked():
    claim = SafetyClaim(
        claim="contains peanuts",
        source_type="JSON_LD",
        explicit_source_text="contains peanuts",
        confidence=1.0,
        reviewed=False,
        reviewer_id=None,
    )
    assert allow_safety_claim(claim) is False