Public Access
974 lines
38 KiB
Python
974 lines
38 KiB
Python
"""Bounded synchronous acquisition and staging for Wikibooks cookbook pages."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import hashlib
|
||
import json
|
||
import math
|
||
import re
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from email.utils import parsedate_to_datetime
|
||
from typing import Any, Callable, Mapping
|
||
from urllib.parse import quote
|
||
|
||
import requests
|
||
from bs4 import BeautifulSoup, Tag
|
||
from sqlalchemy import select
|
||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import RecipeImportStaging, RecipeImportStatus
|
||
|
||
|
||
WIKIBOOKS_API_URL = "https://en.wikibooks.org/w/api.php"
|
||
WIKIBOOKS_SOURCE = "wikibooks"
|
||
CC_BY_SA_4_LICENSE_NAME = "CC BY-SA 4.0"
|
||
CC_BY_SA_4_LICENSE_URL = "https://creativecommons.org/licenses/by-sa/4.0/"
|
||
|
||
MAX_DISCOVERY_LIMIT = 100
|
||
MAX_DISCOVERY_REQUESTS = 10
|
||
MAX_TITLE_LENGTH = 200
|
||
MAX_PAYLOAD_BYTES = 1_000_000
|
||
MAX_HTTP_ATTEMPTS = 3
|
||
MAX_POSTGRES_INTEGER = 2_147_483_647
|
||
RETRYABLE_HTTP_STATUSES = frozenset({429, 500, 502, 503, 504})
|
||
RETRYABLE_MEDIAWIKI_ERRORS = frozenset({"maxlag", "ratelimited"})
|
||
|
||
_UNICODE_FRACTIONS = {
|
||
"¼": 0.25,
|
||
"½": 0.5,
|
||
"¾": 0.75,
|
||
"⅓": 1 / 3,
|
||
"⅔": 2 / 3,
|
||
"⅛": 0.125,
|
||
"⅜": 0.375,
|
||
"⅝": 0.625,
|
||
"⅞": 0.875,
|
||
}
|
||
_UNICODE_FRACTION_CHARS = "".join(_UNICODE_FRACTIONS)
|
||
_UNIT_PATTERN = re.compile(
|
||
r"^(?P<unit>(?:cups?|tablespoons?|tbsp\.?|teaspoons?|tsp\.?|ounces?|oz\.?|"
|
||
r"pounds?|lbs?\.?|grams?|g|kilograms?|kgs?|kg|millilit(?:er|re)s?|ml|"
|
||
r"lit(?:er|re)s?|l|cloves?|cans?|slices?|packages?|packs?|pinches?))\b\.?",
|
||
re.IGNORECASE,
|
||
)
|
||
_ALTERNATE_METRIC_QUANTITY = re.compile(
|
||
r"\s*\(\s*(?:\d+(?:\.\d+)?(?:\s*/\s*\d+)?|[" + _UNICODE_FRACTION_CHARS + r"]|"
|
||
r"\d+[" + _UNICODE_FRACTION_CHARS + r"])\s*(?:ml|millilit(?:er|re)s?|l|lit(?:er|re)s?|"
|
||
r"g|grams?|kg|kilograms?|oz|ounces?|lb|lbs?|pounds?)\b[^)]*\)",
|
||
re.IGNORECASE,
|
||
)
|
||
_TITLE_GENERIC_WORDS = frozenset(
|
||
{
|
||
"a",
|
||
"an",
|
||
"and",
|
||
"american",
|
||
"baked",
|
||
"baking",
|
||
"bread",
|
||
"boiled",
|
||
"braised",
|
||
"classic",
|
||
"cookbook",
|
||
"cooked",
|
||
"cookie",
|
||
"dessert",
|
||
"dish",
|
||
"dressing",
|
||
"for",
|
||
"fried",
|
||
"glaze",
|
||
"glazed",
|
||
"grilled",
|
||
"in",
|
||
"meal",
|
||
"minute",
|
||
"of",
|
||
"on",
|
||
"or",
|
||
"out",
|
||
"pan",
|
||
"pie",
|
||
"recipe",
|
||
"roasted",
|
||
"salad",
|
||
"sauce",
|
||
"seared",
|
||
"steamed",
|
||
"stroganoff",
|
||
"style",
|
||
"the",
|
||
"with",
|
||
}
|
||
)
|
||
_TITLE_VARIANT_TOKENS = frozenset({"I", "II", "III", "IV", "V", "VI", "VII", "VIII", "IX", "X"})
|
||
_PLACEHOLDER_INSTRUCTIONS = frozenset(
|
||
{
|
||
"coming soon",
|
||
"directions",
|
||
"instructions",
|
||
"method",
|
||
"n a",
|
||
"tbd",
|
||
"todo",
|
||
}
|
||
)
|
||
|
||
|
||
class WikibooksImportError(RuntimeError):
|
||
"""Raised when a Wikibooks response cannot safely be used."""
|
||
|
||
|
||
class WikibooksResponseError(WikibooksImportError):
|
||
"""Raised for malformed or unexpected API response data."""
|
||
|
||
|
||
def canonical_json_sha256(value: Mapping[str, Any]) -> str:
|
||
"""Return the stable SHA-256 checksum for a JSON-compatible mapping."""
|
||
canonical = json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
|
||
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
|
||
|
||
|
||
class WikibooksImportService:
|
||
"""Acquire and stage text-only recipe candidates from the fixed Wikibooks API."""
|
||
|
||
def __init__(
|
||
self,
|
||
user_agent: str,
|
||
*,
|
||
session: requests.Session | Any | None = None,
|
||
timeout: tuple[float, float] = (3.05, 20.0),
|
||
min_request_interval: float = 1.0,
|
||
max_payload_bytes: int = MAX_PAYLOAD_BYTES,
|
||
max_attempts: int = MAX_HTTP_ATTEMPTS,
|
||
backoff_seconds: float = 1.0,
|
||
max_retry_delay: float = 60.0,
|
||
sleep: Callable[[float], None] = time.sleep,
|
||
monotonic: Callable[[], float] = time.monotonic,
|
||
) -> None:
|
||
cleaned_user_agent = user_agent.strip() if isinstance(user_agent, str) else ""
|
||
if len(cleaned_user_agent) < 12 or not re.search(r"[A-Za-z]", cleaned_user_agent):
|
||
raise ValueError("user_agent must be a meaningful caller-supplied identifier")
|
||
if isinstance(max_payload_bytes, bool) or not isinstance(max_payload_bytes, int) or max_payload_bytes <= 0:
|
||
raise ValueError("max_payload_bytes must be a positive integer")
|
||
if isinstance(max_attempts, bool) or not isinstance(max_attempts, int) or max_attempts < 1:
|
||
raise ValueError("max_attempts must be a positive integer")
|
||
|
||
self._user_agent = cleaned_user_agent
|
||
self._session = session or requests.Session()
|
||
self._timeout = _timeout_pair(timeout)
|
||
self._min_request_interval = _finite_nonnegative(
|
||
min_request_interval, "min_request_interval"
|
||
)
|
||
self._max_payload_bytes = max_payload_bytes
|
||
self._max_attempts = max_attempts
|
||
self._backoff_seconds = _finite_positive(backoff_seconds, "backoff_seconds")
|
||
self._max_retry_delay = _finite_positive(max_retry_delay, "max_retry_delay")
|
||
self._sleep = sleep
|
||
self._monotonic = monotonic
|
||
self._last_request_at: float | None = None
|
||
|
||
def discover_recipe_titles(self, limit: int = MAX_DISCOVERY_LIMIT) -> list[str]:
|
||
"""Return bounded Cookbook page titles from ``Category:Recipes``."""
|
||
if isinstance(limit, bool) or not isinstance(limit, int) or limit < 1:
|
||
raise ValueError("limit must be a positive integer")
|
||
|
||
return self._discover_category_recipe_titles(
|
||
requested=min(limit, MAX_DISCOVERY_LIMIT),
|
||
max_requests=MAX_DISCOVERY_REQUESTS,
|
||
)
|
||
|
||
def _discover_category_recipe_titles(
|
||
self, *, requested: int, max_requests: int
|
||
) -> list[str]:
|
||
titles: list[str] = []
|
||
seen_titles: set[str] = set()
|
||
continuation: str | None = None
|
||
|
||
for _ in range(max_requests):
|
||
params: dict[str, str | int] = {
|
||
"action": "query",
|
||
"list": "categorymembers",
|
||
"cmtitle": "Category:Recipes",
|
||
"cmtype": "page",
|
||
"cmlimit": min(requested - len(titles), MAX_DISCOVERY_LIMIT),
|
||
"format": "json",
|
||
"formatversion": 2,
|
||
"maxlag": 5,
|
||
}
|
||
if continuation:
|
||
params["cmcontinue"] = continuation
|
||
|
||
payload = self._request(params)
|
||
query = _mapping(payload.get("query"), "query")
|
||
members = query.get("categorymembers")
|
||
if not isinstance(members, list):
|
||
raise WikibooksResponseError("query.categorymembers must be a list")
|
||
|
||
for member in members:
|
||
member_data = _mapping(member, "query.categorymembers item")
|
||
title = member_data.get("title")
|
||
if not isinstance(title, str):
|
||
raise WikibooksResponseError("query.categorymembers item title must be a string")
|
||
if title.startswith("Cookbook:") and title not in seen_titles:
|
||
_validate_title(title)
|
||
seen_titles.add(title)
|
||
titles.append(title)
|
||
if len(titles) == requested:
|
||
return titles
|
||
|
||
next_data = payload.get("continue")
|
||
if next_data is None:
|
||
return titles
|
||
next_mapping = _mapping(next_data, "continue")
|
||
next_continuation = next_mapping.get("cmcontinue")
|
||
if not isinstance(next_continuation, str) or not next_continuation:
|
||
raise WikibooksResponseError("continue.cmcontinue must be a non-empty string")
|
||
if next_continuation == continuation:
|
||
raise WikibooksResponseError("repeated category continuation token")
|
||
continuation = next_continuation
|
||
|
||
return titles
|
||
|
||
def search_recipe_titles(
|
||
self, queries: list[str], limit: int = MAX_DISCOVERY_LIMIT
|
||
) -> list[str]:
|
||
"""Return bounded Cookbook titles relevant to feedback search queries."""
|
||
if isinstance(limit, bool) or not isinstance(limit, int) or limit < 1:
|
||
raise ValueError("limit must be a positive integer")
|
||
|
||
requested = min(limit, MAX_DISCOVERY_LIMIT)
|
||
cleaned_queries: list[str] = []
|
||
for query in queries:
|
||
cleaned = " ".join(query.split()) if isinstance(query, str) else ""
|
||
if cleaned and cleaned not in cleaned_queries:
|
||
cleaned_queries.append(cleaned[:MAX_TITLE_LENGTH])
|
||
if len(cleaned_queries) == MAX_DISCOVERY_REQUESTS:
|
||
break
|
||
|
||
titles: list[str] = []
|
||
seen_titles: set[str] = set()
|
||
for query in cleaned_queries:
|
||
payload = self._request(
|
||
{
|
||
"action": "query",
|
||
"list": "search",
|
||
"srsearch": query,
|
||
"srnamespace": 102,
|
||
"srlimit": requested - len(titles),
|
||
"format": "json",
|
||
"formatversion": 2,
|
||
"maxlag": 5,
|
||
}
|
||
)
|
||
search = _mapping(payload.get("query"), "query").get("search")
|
||
if not isinstance(search, list):
|
||
raise WikibooksResponseError("query.search must be a list")
|
||
for result in search:
|
||
title = _mapping(result, "query.search item").get("title")
|
||
if not isinstance(title, str):
|
||
raise WikibooksResponseError(
|
||
"query.search item title must be a string"
|
||
)
|
||
if title.startswith("Cookbook:") and title not in seen_titles:
|
||
_validate_title(title)
|
||
seen_titles.add(title)
|
||
titles.append(title)
|
||
if len(titles) == requested:
|
||
return titles
|
||
return titles
|
||
|
||
def fetch_page(self, title: str) -> dict[str, Any]:
|
||
"""Fetch one validated ``Cookbook:`` parse object without augmentation."""
|
||
_validate_title(title)
|
||
payload = self._request(
|
||
{
|
||
"action": "parse",
|
||
"page": title,
|
||
"redirects": 1,
|
||
"prop": "text|wikitext|sections|revid|displaytitle",
|
||
"format": "json",
|
||
"formatversion": 2,
|
||
"maxlag": 5,
|
||
}
|
||
)
|
||
parse_payload = _mapping(payload.get("parse"), "parse")
|
||
_validate_parse_payload(parse_payload)
|
||
return parse_payload
|
||
|
||
def normalize_page(self, parse_payload: Mapping[str, Any]) -> dict[str, Any]:
|
||
"""Convert a validated rendered cookbook page into text-only recipe data."""
|
||
_validate_parse_payload(parse_payload)
|
||
title = str(parse_payload["title"])
|
||
soup = BeautifulSoup(str(parse_payload["text"]), "html.parser")
|
||
root = soup.select_one(".mw-parser-output") or soup
|
||
elements = root.find_all(["h2", "h3", "h4", "h5", "h6", "p", "ul", "ol"])
|
||
|
||
ingredients_heading = _find_heading(elements, {"ingredients"})
|
||
instruction_heading = _find_heading(
|
||
elements, {"procedure", "instructions", "method", "directions"}
|
||
)
|
||
description = _description_before_heading(elements, ingredients_heading)
|
||
infobox = _infobox_values(root)
|
||
|
||
ingredients = [
|
||
_normalize_ingredient(item)
|
||
for item in _section_list_items(ingredients_heading, {"ul", "ol"})
|
||
]
|
||
instructions = _section_list_items(instruction_heading, {"ol"})
|
||
|
||
return {
|
||
"name": title.removeprefix("Cookbook:").replace("_", " ").strip(),
|
||
"description": description,
|
||
"servings": _servings_from_infobox(infobox),
|
||
"prep_time_minutes": _time_from_infobox(infobox, ("prep", "preparation")),
|
||
"cook_time_minutes": _time_from_infobox(infobox, ("cook", "cooking")),
|
||
"baking_time_minutes": _time_from_infobox(infobox, ("bake", "baking")),
|
||
"ingredients": ingredients,
|
||
"instructions": instructions,
|
||
}
|
||
|
||
def validate_recipe(self, candidate: Mapping[str, Any]) -> dict[str, Any]:
|
||
"""Return deterministic structural and cross-field validation results."""
|
||
errors: list[dict[str, Any]] = []
|
||
warnings: list[dict[str, Any]] = []
|
||
servings = candidate.get("servings")
|
||
if (
|
||
isinstance(servings, bool)
|
||
or not isinstance(servings, int)
|
||
or servings <= 0
|
||
):
|
||
errors.append(_issue("invalid_servings", "servings", "Servings must be a positive integer."))
|
||
elif servings > MAX_POSTGRES_INTEGER:
|
||
errors.append(
|
||
_issue(
|
||
"servings_out_of_range",
|
||
"servings",
|
||
"Servings exceed the supported database range.",
|
||
)
|
||
)
|
||
|
||
for field in ("prep_time_minutes", "cook_time_minutes", "baking_time_minutes"):
|
||
duration = candidate.get(field)
|
||
if isinstance(duration, int) and not isinstance(duration, bool) and duration > MAX_POSTGRES_INTEGER:
|
||
errors.append(
|
||
_issue(
|
||
"time_out_of_range",
|
||
field,
|
||
"Duration exceeds the supported database range.",
|
||
)
|
||
)
|
||
combined_cook_time = (candidate.get("cook_time_minutes") or 0) + (
|
||
candidate.get("baking_time_minutes") or 0
|
||
)
|
||
if (
|
||
isinstance(combined_cook_time, int)
|
||
and not isinstance(combined_cook_time, bool)
|
||
and combined_cook_time > MAX_POSTGRES_INTEGER
|
||
):
|
||
errors.append(
|
||
_issue(
|
||
"combined_cook_time_out_of_range",
|
||
"times",
|
||
"Combined cook and baking time exceed the supported database range.",
|
||
)
|
||
)
|
||
|
||
ingredients = candidate.get("ingredients")
|
||
if not isinstance(ingredients, list) or len(ingredients) < 2:
|
||
errors.append(
|
||
_issue("too_few_ingredients", "ingredients", "At least two ingredients are required.")
|
||
)
|
||
ingredient_items: list[Any] = []
|
||
else:
|
||
ingredient_items = ingredients
|
||
|
||
instructions = candidate.get("instructions")
|
||
if not isinstance(instructions, list) or len(instructions) < 2:
|
||
errors.append(
|
||
_issue("too_few_instructions", "instructions", "At least two instructions are required.")
|
||
)
|
||
instruction_items: list[Any] = []
|
||
else:
|
||
instruction_items = instructions
|
||
|
||
for index, ingredient in enumerate(ingredient_items):
|
||
quantity = ingredient.get("quantity") if isinstance(ingredient, Mapping) else None
|
||
if (
|
||
isinstance(quantity, bool)
|
||
or not isinstance(quantity, (int, float))
|
||
or not math.isfinite(quantity)
|
||
or quantity <= 0
|
||
):
|
||
errors.append(
|
||
_issue(
|
||
"invalid_ingredient_quantity",
|
||
f"ingredients[{index}].quantity",
|
||
"Ingredient quantity must be positive.",
|
||
)
|
||
)
|
||
|
||
for index, instruction in enumerate(instruction_items):
|
||
if not isinstance(instruction, str) or _is_placeholder_instruction(instruction):
|
||
errors.append(
|
||
_issue(
|
||
"placeholder_instruction",
|
||
f"instructions[{index}]",
|
||
"Instruction is missing or a placeholder.",
|
||
)
|
||
)
|
||
|
||
title_name = candidate.get("name")
|
||
if isinstance(title_name, str) and ingredient_items:
|
||
missing_tokens = _missing_title_ingredient_tokens(title_name, ingredient_items)
|
||
if missing_tokens:
|
||
errors.append(
|
||
_issue(
|
||
"title_ingredient_mismatch",
|
||
"name",
|
||
"Distinctive title food tokens are absent from ingredients.",
|
||
missing_tokens=missing_tokens,
|
||
)
|
||
)
|
||
|
||
if not isinstance(candidate.get("description"), str) or not candidate["description"].strip():
|
||
warnings.append(_issue("missing_description", "description", "Intro description is unavailable."))
|
||
if not any(
|
||
isinstance(candidate.get(field), int) and candidate[field] > 0
|
||
for field in ("prep_time_minutes", "cook_time_minutes", "baking_time_minutes")
|
||
):
|
||
warnings.append(_issue("missing_time", "times", "No preparation, cook, or baking time is available."))
|
||
|
||
return {"valid": not errors, "errors": errors, "warnings": warnings}
|
||
|
||
def stage_page(self, db: Session, title: str) -> RecipeImportStaging:
|
||
"""Fetch, normalize, validate, and flush one candidate without committing."""
|
||
return self.stage_parse_payload(db, self.fetch_page(title))
|
||
|
||
def stage_discovered_pages(
|
||
self,
|
||
db: Session,
|
||
limit: int = 5,
|
||
queries: list[str] | None = None,
|
||
) -> list[RecipeImportStaging]:
|
||
"""Discover and stage a bounded relevant or category batch without committing."""
|
||
titles = (
|
||
self.search_recipe_titles(queries, limit)
|
||
if queries
|
||
else self.discover_recipe_titles(limit)
|
||
)
|
||
return [self.stage_page(db, title) for title in titles]
|
||
|
||
def stage_parse_payload(
|
||
self, db: Session, parse_payload: Mapping[str, Any]
|
||
) -> RecipeImportStaging:
|
||
"""Stage a validated parse object, reusing an existing page revision."""
|
||
snapshot = copy.deepcopy(dict(parse_payload))
|
||
_validate_parse_payload(snapshot)
|
||
external_id = str(snapshot["pageid"])
|
||
source_revision = str(snapshot["revid"])
|
||
normalized = self.normalize_page(snapshot)
|
||
validation_report = self.validate_recipe(normalized)
|
||
status = (
|
||
RecipeImportStatus.PENDING
|
||
if validation_report["valid"]
|
||
else RecipeImportStatus.QUARANTINED
|
||
)
|
||
title = str(snapshot["title"])
|
||
values = {
|
||
"source": WIKIBOOKS_SOURCE,
|
||
"external_id": external_id,
|
||
"source_url": _page_url(title),
|
||
"source_revision": source_revision,
|
||
"license_name": CC_BY_SA_4_LICENSE_NAME,
|
||
"license_url": CC_BY_SA_4_LICENSE_URL,
|
||
"attribution": f"Wikibooks contributors; revision history: {_history_url(title)}",
|
||
"raw_payload": snapshot,
|
||
"payload_sha256": canonical_json_sha256(snapshot),
|
||
"normalized_recipe": normalized,
|
||
"validation_report": validation_report,
|
||
"status": status,
|
||
"rejection_reason": (
|
||
None
|
||
if validation_report["valid"]
|
||
else "; ".join(error["code"] for error in validation_report["errors"])
|
||
),
|
||
}
|
||
identity = {
|
||
"source": WIKIBOOKS_SOURCE,
|
||
"external_id": external_id,
|
||
"source_revision": source_revision,
|
||
}
|
||
with db.no_autoflush:
|
||
db.execute(
|
||
pg_insert(RecipeImportStaging.__table__)
|
||
.values(**values)
|
||
.on_conflict_do_nothing(
|
||
constraint="uq_recipe_import_staging_source_revision"
|
||
)
|
||
.returning(RecipeImportStaging.id)
|
||
)
|
||
return db.execute(
|
||
select(RecipeImportStaging).filter_by(**identity)
|
||
).scalar_one()
|
||
|
||
def revalidate_staged_imports(
|
||
self, db: Session, *, limit: int = 100
|
||
) -> dict[str, int]:
|
||
"""Recompute unreviewed Wikibooks candidates without committing."""
|
||
if isinstance(limit, bool) or not isinstance(limit, int) or not 1 <= limit <= 1000:
|
||
raise ValueError("limit must be between 1 and 1000")
|
||
|
||
candidates = (
|
||
db.execute(
|
||
select(RecipeImportStaging)
|
||
.where(
|
||
RecipeImportStaging.source == WIKIBOOKS_SOURCE,
|
||
RecipeImportStaging.status.in_(
|
||
[RecipeImportStatus.PENDING, RecipeImportStatus.QUARANTINED]
|
||
),
|
||
RecipeImportStaging.reviewed_at.is_(None),
|
||
RecipeImportStaging.reviewed_by.is_(None),
|
||
)
|
||
.order_by(RecipeImportStaging.acquired_at, RecipeImportStaging.id)
|
||
.limit(limit)
|
||
.with_for_update(skip_locked=True)
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
counts = {"processed": len(candidates), "pending": 0, "quarantined": 0}
|
||
for staged in candidates:
|
||
snapshot = copy.deepcopy(staged.raw_payload)
|
||
if canonical_json_sha256(snapshot) != staged.payload_sha256:
|
||
staged.validation_report = {
|
||
"valid": False,
|
||
"errors": [
|
||
_issue(
|
||
"payload_checksum_mismatch",
|
||
"raw_payload",
|
||
"Raw payload checksum does not match immutable provenance.",
|
||
)
|
||
],
|
||
"warnings": [],
|
||
}
|
||
staged.status = RecipeImportStatus.QUARANTINED
|
||
staged.rejection_reason = "payload_checksum_mismatch"
|
||
counts[RecipeImportStatus.QUARANTINED.value] += 1
|
||
continue
|
||
_validate_parse_payload(snapshot)
|
||
normalized = self.normalize_page(snapshot)
|
||
validation_report = self.validate_recipe(normalized)
|
||
status = (
|
||
RecipeImportStatus.PENDING
|
||
if validation_report["valid"]
|
||
else RecipeImportStatus.QUARANTINED
|
||
)
|
||
staged.normalized_recipe = normalized
|
||
staged.validation_report = validation_report
|
||
staged.status = status
|
||
staged.rejection_reason = (
|
||
None
|
||
if validation_report["valid"]
|
||
else "; ".join(error["code"] for error in validation_report["errors"])
|
||
)
|
||
counts[status.value] += 1
|
||
db.flush()
|
||
return counts
|
||
|
||
def _request(self, params: Mapping[str, str | int]) -> dict[str, Any]:
|
||
for attempt in range(self._max_attempts):
|
||
response: Any | None = None
|
||
retry_error: WikibooksImportError | None = None
|
||
retry_delay: float | None = None
|
||
try:
|
||
self._wait_for_request_slot()
|
||
response = self._session.get(
|
||
WIKIBOOKS_API_URL,
|
||
params=dict(params),
|
||
headers={"Accept": "application/json", "User-Agent": self._user_agent},
|
||
timeout=self._timeout,
|
||
stream=True,
|
||
allow_redirects=False,
|
||
)
|
||
status_code = getattr(response, "status_code", None)
|
||
if isinstance(status_code, bool) or not isinstance(status_code, int):
|
||
raise WikibooksResponseError("HTTP response status code is invalid")
|
||
if status_code in RETRYABLE_HTTP_STATUSES:
|
||
retry_error = WikibooksImportError(f"Wikibooks returned retryable HTTP {status_code}")
|
||
retry_delay = _retry_after_seconds(getattr(response, "headers", {}))
|
||
elif 300 <= status_code < 400:
|
||
raise WikibooksImportError(f"Wikibooks redirect HTTP {status_code} rejected")
|
||
elif status_code < 200 or status_code >= 300:
|
||
raise WikibooksImportError(f"Wikibooks returned HTTP {status_code}")
|
||
else:
|
||
payload = self._read_json_response(response)
|
||
api_error = payload.get("error")
|
||
if api_error is None:
|
||
return payload
|
||
error_data = _mapping(api_error, "error")
|
||
error_code = error_data.get("code")
|
||
if not isinstance(error_code, str) or not error_code:
|
||
raise WikibooksResponseError("error.code must be a non-empty string")
|
||
if error_code.lower() in RETRYABLE_MEDIAWIKI_ERRORS:
|
||
retry_error = WikibooksImportError(
|
||
f"Wikibooks returned retryable MediaWiki error {error_code}"
|
||
)
|
||
else:
|
||
raise WikibooksImportError(
|
||
f"Wikibooks returned MediaWiki error {error_code}"
|
||
)
|
||
except requests.RequestException as exc:
|
||
retry_error = WikibooksImportError(f"Wikibooks request failed: {exc}")
|
||
finally:
|
||
if response is not None:
|
||
response.close()
|
||
|
||
if retry_error is None:
|
||
raise AssertionError("request did not return or produce an error")
|
||
if attempt == self._max_attempts - 1:
|
||
raise retry_error
|
||
delay = retry_delay if retry_delay is not None else self._backoff_seconds * (2**attempt)
|
||
self._sleep(min(delay, self._max_retry_delay))
|
||
|
||
raise AssertionError("unreachable")
|
||
|
||
def _wait_for_request_slot(self) -> None:
|
||
if self._last_request_at is not None:
|
||
wait_seconds = self._min_request_interval - (
|
||
self._monotonic() - self._last_request_at
|
||
)
|
||
if wait_seconds > 0:
|
||
self._sleep(wait_seconds)
|
||
self._last_request_at = self._monotonic()
|
||
|
||
def _read_json_response(self, response: Any) -> dict[str, Any]:
|
||
headers = getattr(response, "headers", {})
|
||
if not isinstance(headers, Mapping):
|
||
raise WikibooksResponseError("HTTP response headers are invalid")
|
||
content_length = headers.get("Content-Length")
|
||
if content_length is not None:
|
||
try:
|
||
declared_size = int(content_length)
|
||
except (TypeError, ValueError) as exc:
|
||
raise WikibooksResponseError("HTTP Content-Length is invalid") from exc
|
||
if declared_size < 0 or declared_size > self._max_payload_bytes:
|
||
raise WikibooksResponseError("Wikibooks payload exceeds configured size limit")
|
||
|
||
body = bytearray()
|
||
for chunk in response.iter_content(chunk_size=64 * 1024):
|
||
if not isinstance(chunk, (bytes, bytearray)):
|
||
raise WikibooksResponseError("HTTP response body chunk is invalid")
|
||
body.extend(chunk)
|
||
if len(body) > self._max_payload_bytes:
|
||
raise WikibooksResponseError("Wikibooks payload exceeds configured size limit")
|
||
try:
|
||
decoded = json.loads(bytes(body).decode("utf-8"))
|
||
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||
raise WikibooksResponseError("Wikibooks response is not valid UTF-8 JSON") from exc
|
||
return _mapping(decoded, "response")
|
||
|
||
|
||
def _mapping(value: Any, name: str) -> dict[str, Any]:
|
||
if not isinstance(value, dict):
|
||
raise WikibooksResponseError(f"{name} must be an object")
|
||
return value
|
||
|
||
|
||
def _validate_title(title: str) -> None:
|
||
if (
|
||
not isinstance(title, str)
|
||
or not title.startswith("Cookbook:")
|
||
or len(title) <= len("Cookbook:")
|
||
or len(title) > MAX_TITLE_LENGTH
|
||
or any(character in title for character in "\r\n\x00")
|
||
):
|
||
raise WikibooksResponseError("title must be a sane Cookbook: page title")
|
||
|
||
|
||
def _validate_parse_payload(parse_payload: Mapping[str, Any]) -> None:
|
||
if not isinstance(parse_payload, Mapping):
|
||
raise WikibooksResponseError("parse must be an object")
|
||
_positive_integer(parse_payload.get("pageid"), "parse.pageid")
|
||
_positive_integer(parse_payload.get("revid"), "parse.revid")
|
||
title = parse_payload.get("title")
|
||
_validate_title(title)
|
||
for field in ("text", "wikitext", "displaytitle"):
|
||
if not isinstance(parse_payload.get(field), str):
|
||
raise WikibooksResponseError(f"parse.{field} must be a string")
|
||
sections = parse_payload.get("sections")
|
||
if not isinstance(sections, list):
|
||
raise WikibooksResponseError("parse.sections must be a list")
|
||
for section in sections:
|
||
section_data = _mapping(section, "parse.sections item")
|
||
if not isinstance(section_data.get("line"), str):
|
||
raise WikibooksResponseError("parse.sections item line must be a string")
|
||
|
||
|
||
def _positive_integer(value: Any, name: str) -> int:
|
||
if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
|
||
raise WikibooksResponseError(f"{name} must be a positive integer")
|
||
return value
|
||
|
||
|
||
def _finite_positive(value: Any, name: str) -> float:
|
||
if (
|
||
isinstance(value, bool)
|
||
or not isinstance(value, (int, float))
|
||
or not math.isfinite(value)
|
||
or value <= 0
|
||
):
|
||
raise ValueError(f"{name} must be finite and positive")
|
||
return float(value)
|
||
|
||
|
||
def _finite_nonnegative(value: Any, name: str) -> float:
|
||
if (
|
||
isinstance(value, bool)
|
||
or not isinstance(value, (int, float))
|
||
or not math.isfinite(value)
|
||
or value < 0
|
||
):
|
||
raise ValueError(f"{name} must be finite and non-negative")
|
||
return float(value)
|
||
|
||
|
||
def _timeout_pair(timeout: Any) -> tuple[float, float]:
|
||
if not isinstance(timeout, (tuple, list)) or len(timeout) != 2:
|
||
raise ValueError("timeout must be a (connect, read) pair")
|
||
return (
|
||
_finite_positive(timeout[0], "timeout connect value"),
|
||
_finite_positive(timeout[1], "timeout read value"),
|
||
)
|
||
|
||
|
||
def _retry_after_seconds(headers: Mapping[str, Any]) -> float | None:
|
||
value = headers.get("Retry-After")
|
||
if value is None:
|
||
return None
|
||
try:
|
||
seconds = float(value)
|
||
return max(0.0, seconds) if math.isfinite(seconds) else None
|
||
except (TypeError, ValueError):
|
||
try:
|
||
retry_at = parsedate_to_datetime(str(value))
|
||
except (TypeError, ValueError, IndexError):
|
||
return None
|
||
if retry_at.tzinfo is None:
|
||
retry_at = retry_at.replace(tzinfo=timezone.utc)
|
||
seconds = (retry_at - datetime.now(timezone.utc)).total_seconds()
|
||
return max(0.0, seconds) if math.isfinite(seconds) else None
|
||
|
||
|
||
def _page_url(title: str) -> str:
|
||
return "https://en.wikibooks.org/wiki/" + quote(title.replace(" ", "_"), safe=":")
|
||
|
||
|
||
def _history_url(title: str) -> str:
|
||
return "https://en.wikibooks.org/w/index.php?title=" + quote(title, safe="") + "&action=history"
|
||
|
||
|
||
def _text(tag: Tag) -> str:
|
||
return re.sub(r"\s+", " ", " ".join(tag.stripped_strings)).strip()
|
||
|
||
|
||
def _heading_text(tag: Tag) -> str:
|
||
return re.sub(r"\s*\[edit\]\s*$", "", _text(tag), flags=re.IGNORECASE).strip().lower()
|
||
|
||
|
||
def _find_heading(elements: list[Tag], names: set[str]) -> Tag | None:
|
||
for element in elements:
|
||
if element.name in {"h2", "h3", "h4", "h5", "h6"} and _heading_text(element) in names:
|
||
return element
|
||
return None
|
||
|
||
|
||
def _description_before_heading(elements: list[Tag], heading: Tag | None) -> str | None:
|
||
if heading is None:
|
||
return None
|
||
descriptions: list[str] = []
|
||
for element in elements:
|
||
if element is heading:
|
||
break
|
||
if element.name == "p" and element.find_parent(["li", "td", "th"]) is None:
|
||
text = _text(element)
|
||
if text:
|
||
descriptions.append(text)
|
||
return " ".join(descriptions) or None
|
||
|
||
|
||
def _infobox_values(root: Tag | BeautifulSoup) -> dict[str, str]:
|
||
values: dict[str, str] = {}
|
||
for table in root.find_all("table"):
|
||
classes = " ".join(table.get("class", []))
|
||
if "infobox" not in classes.lower():
|
||
continue
|
||
for row in table.find_all("tr"):
|
||
cells = row.find_all(["th", "td"], recursive=False)
|
||
if len(cells) < 2:
|
||
continue
|
||
label = _text(cells[0]).lower()
|
||
value = _text(cells[1])
|
||
if label and value and label not in values:
|
||
values[label] = value
|
||
return values
|
||
|
||
|
||
def _servings_from_infobox(values: Mapping[str, str]) -> int | None:
|
||
for label, value in values.items():
|
||
if any(term in label for term in ("servings", "serves", "yield", "makes")):
|
||
match = re.search(r"\b(\d+)\b", value)
|
||
if match:
|
||
return int(match.group(1))
|
||
return None
|
||
|
||
|
||
def _time_from_infobox(values: Mapping[str, str], labels: tuple[str, ...]) -> int | None:
|
||
for label, value in values.items():
|
||
if any(expected in label for expected in labels):
|
||
return _parse_duration(value)
|
||
return None
|
||
|
||
|
||
def _parse_duration(value: str) -> int | None:
|
||
clock = re.search(r"\b(\d{1,2}):(\d{2})\b", value)
|
||
if clock:
|
||
hours, minutes = (int(part) for part in clock.groups())
|
||
total = hours * 60 + minutes
|
||
return total if total > 0 else None
|
||
hours = re.search(r"\b(\d+)\s*(?:hours?|hrs?|hr|h)\b", value, re.IGNORECASE)
|
||
minutes = re.search(r"\b(\d+)\s*(?:minutes?|mins?|min|m)\b", value, re.IGNORECASE)
|
||
total = (int(hours.group(1)) * 60 if hours else 0) + (int(minutes.group(1)) if minutes else 0)
|
||
return total if total > 0 else None
|
||
|
||
|
||
def _section_list_items(heading: Tag | None, list_names: set[str]) -> list[str]:
|
||
if heading is None:
|
||
return []
|
||
heading_level = int(heading.name[1])
|
||
items: list[str] = []
|
||
for element in heading.find_all_next(["h2", "h3", "h4", "h5", "h6", "ul", "ol"]):
|
||
if element.name and element.name.startswith("h") and int(element.name[1]) <= heading_level:
|
||
break
|
||
if element.name not in list_names:
|
||
continue
|
||
for item in element.find_all("li", recursive=False):
|
||
text = _list_item_text(item)
|
||
if text:
|
||
items.append(text)
|
||
return items
|
||
|
||
|
||
def _list_item_text(item: Tag) -> str:
|
||
clone = copy.copy(item)
|
||
for nested_list in clone.find_all(["ul", "ol"]):
|
||
nested_list.decompose()
|
||
return _text(clone)
|
||
|
||
|
||
def _normalize_ingredient(source_text: str) -> dict[str, Any]:
|
||
quantity, quantity_text, remainder = _parse_leading_quantity(source_text)
|
||
unit: str | None = None
|
||
name = remainder if quantity_text is not None else source_text
|
||
if quantity_text is not None:
|
||
unit_match = _UNIT_PATTERN.match(name.lstrip())
|
||
if unit_match:
|
||
unit = unit_match.group("unit").lower().rstrip(".")
|
||
name = name.lstrip()[unit_match.end() :]
|
||
name = _ALTERNATE_METRIC_QUANTITY.sub("", name)
|
||
name = name.strip(" \t,;:-") or source_text
|
||
return {
|
||
"quantity": quantity,
|
||
"quantity_text": quantity_text,
|
||
"unit": unit,
|
||
"name": name,
|
||
"source_text": source_text,
|
||
}
|
||
|
||
|
||
def _parse_leading_quantity(value: str) -> tuple[float | None, str | None, str]:
|
||
text = value.strip()
|
||
if re.match(r"^\d+(?:\.\d+)?\s*(?:-|–|—)\s*\d", text):
|
||
return None, None, text
|
||
fraction_prefix = re.match(r"^\d+(?:\.\d+)?(?:\s+\d+)?\s*/\s*(\d*)", text)
|
||
if fraction_prefix and (not fraction_prefix.group(1) or int(fraction_prefix.group(1)) == 0):
|
||
return None, None, text
|
||
mixed = re.match(r"^(\d+(?:\.\d+)?)\s+(\d+\s*/\s*\d+|[" + _UNICODE_FRACTION_CHARS + r"])", text)
|
||
if mixed:
|
||
fraction = _fraction_value(mixed.group(2))
|
||
if fraction is not None:
|
||
return float(mixed.group(1)) + fraction, mixed.group(0), text[mixed.end() :]
|
||
compact = re.match(r"^(\d+)([" + _UNICODE_FRACTION_CHARS + r"])", text)
|
||
if compact:
|
||
return int(compact.group(1)) + _UNICODE_FRACTIONS[compact.group(2)], compact.group(0), text[compact.end() :]
|
||
ascii_fraction = re.match(r"^(\d+)\s*/\s*(\d+)", text)
|
||
if ascii_fraction:
|
||
denominator = int(ascii_fraction.group(2))
|
||
if denominator:
|
||
return (
|
||
int(ascii_fraction.group(1)) / denominator,
|
||
ascii_fraction.group(0),
|
||
text[ascii_fraction.end() :],
|
||
)
|
||
number = re.match(r"^\d+(?:\.\d+)?", text)
|
||
if number:
|
||
return float(number.group(0)), number.group(0), text[number.end() :]
|
||
unicode_fraction = re.match(r"^(?:[" + _UNICODE_FRACTION_CHARS + r"])", text)
|
||
if unicode_fraction:
|
||
token = unicode_fraction.group(0)
|
||
return _UNICODE_FRACTIONS[token], token, text[unicode_fraction.end() :]
|
||
return None, None, text
|
||
|
||
|
||
def _fraction_value(value: str) -> float | None:
|
||
if value in _UNICODE_FRACTIONS:
|
||
return _UNICODE_FRACTIONS[value]
|
||
numerator, denominator = (int(part.strip()) for part in value.split("/", 1))
|
||
return numerator / denominator if denominator else None
|
||
|
||
|
||
def _issue(code: str, field: str, message: str, **extra: Any) -> dict[str, Any]:
|
||
return {"code": code, "field": field, "message": message, **extra}
|
||
|
||
|
||
def _is_placeholder_instruction(instruction: str) -> bool:
|
||
normalized = re.sub(r"[^a-z0-9]+", " ", instruction.lower()).strip()
|
||
return normalized in _PLACEHOLDER_INSTRUCTIONS or bool(
|
||
re.fullmatch(r"(?:step )?\d+", normalized)
|
||
)
|
||
|
||
|
||
def _missing_title_ingredient_tokens(title: str, ingredients: list[Any]) -> list[str]:
|
||
title_tokens: list[str] = []
|
||
for raw_token in re.findall(r"[A-Za-z]+", title):
|
||
if raw_token in _TITLE_VARIANT_TOKENS:
|
||
continue
|
||
token = _singular_token(raw_token.lower())
|
||
if len(token) > 1 and token not in _TITLE_GENERIC_WORDS:
|
||
title_tokens.append(token)
|
||
ingredient_tokens: set[str] = set()
|
||
for ingredient in ingredients:
|
||
if isinstance(ingredient, Mapping) and isinstance(ingredient.get("name"), str):
|
||
ingredient_tokens.update(
|
||
_singular_token(token)
|
||
for token in re.findall(r"[a-z]+", ingredient["name"].lower())
|
||
)
|
||
return [token for token in dict.fromkeys(title_tokens) if token not in ingredient_tokens]
|
||
|
||
|
||
def _singular_token(token: str) -> str:
|
||
if token == "pies":
|
||
return "pie"
|
||
if token == "cookies":
|
||
return "cookie"
|
||
if token.endswith("ies") and len(token) > 3:
|
||
return token[:-3] + "y"
|
||
if token.endswith(("ches", "shes", "sses", "xes", "zes")) and len(token) > 3:
|
||
return token[:-2]
|
||
if token.endswith("s") and not token.endswith("ss") and len(token) > 2:
|
||
return token[:-1]
|
||
return token
|