Compare commits

..

1 Commits

Author SHA1 Message Date
ba15e3cad3 moved from urllib to requests
Some checks failed
continuous-integration/drone/push Build is failing
2023-05-18 09:01:37 -04:00
3 changed files with 126 additions and 264 deletions

View File

@@ -88,13 +88,9 @@ docker-compose -p recipe-test up
running tests running tests
```sh ```sh
pytest --cov=src/recipe_graph --cov-report lcov --cov-report html pytest
``` ```
The html report is under `htmlcov/` and can be viewed through any browser.
The `lcov` file can be used for the [Coverage Gutters](https://marketplace.visualstudio.com/items?itemName=ryanluker.vscode-coverage-gutters)
plugin for VS Code to view coverage in your editor.
**WARNINING**: If you get `ERROR at setup of test_db_connection` and **WARNINING**: If you get `ERROR at setup of test_db_connection` and
`ERROR at setup of test_db_class_creation`, please check if testing database is `ERROR at setup of test_db_class_creation`, please check if testing database is
already initiated. Testing is destructive and should be done on a fresh database. already initiated. Testing is destructive and should be done on a fresh database.
@@ -109,6 +105,16 @@ docker-compose -p recipe-test down
Test are written in pytest framework. Currently focused on unittest and code Test are written in pytest framework. Currently focused on unittest and code
coverage. Integration tests to come. coverage. Integration tests to come.
To run test use:
```sh
pytest --cov=src/recipe_graph --cov-report lcov --cov-report html
```
The html report is under `htmlcov/` and can be viewed through any browser.
The `lcov` file can be used for the [Coverage Gutters](https://marketplace.visualstudio.com/items?itemName=ryanluker.vscode-coverage-gutters)
plugin for VS Code to view coverage in your editor.
## TODO ## TODO
> ☑ automate scraping\ > ☑ automate scraping\
> ☑ extracting quantity and name (via regex)\ > ☑ extracting quantity and name (via regex)\

View File

@@ -9,53 +9,40 @@ from urllib.parse import urljoin
import logging import logging
from argparse import ArgumentParser from argparse import ArgumentParser
def parse_ingredient(ingredient_text):
def ingredient_regex(units: list[str], instructions: list[str]) -> re.Pattern: units = ['teaspoon', 'tablespoon', 'gram', 'ounce', 'jar', 'cup', 'pinch',
number_regex = "((?:[\d\\./\\u00BC-\\u00BE\\u2150-\\u215E]*\s?(?:\(.+\))?)*)" 'container', 'slice', 'package', 'pound', 'can', 'dash', 'spear',
ingredient_regex = "([a-zA-Z '\-]+)" 'bunch', 'quart', 'cube', 'envelope', 'square', 'sprig', 'bag',
supplement_regex = ",?(.*)" 'box', 'drop', 'fluid ounce', 'gallon', 'head', 'link', 'loaf',
units_regex = "|".join( 'pint', 'pod', 'sheet', 'stalk', 'whole', 'bar', 'bottle', 'bulb',
[f"[{unit[0]}{unit[0].capitalize()}]{unit[1:]}" for unit in units] 'year', 'fillet', 'litter', 'packet', 'slices']
) instructions = ['and', 'or', 'chopped', 'diced', 'brewed', 'chilled',
'chunky', 'small', 'medium', 'large', 'couarse', 'cracked',
'crushed', 'ground', 'cooked', 'cubed', 'crumbled', 'cut',
'cold', 'hot', 'warm', 'day', 'old', 'drained', 'canned',
'dried', 'dry', 'fine', 'firm', 'fresh', 'frozen',
'grated', 'grilled', 'hard', 'hot', 'juliened?', 'leftover',
'light', 'lite', 'mashed', 'melted', 'minced', 'packed',
'peeled', 'pitted', 'sliced', 'prepared', 'refrigerated',
'rehydrated', 'seedless', 'shaved', 'shredded', 'sifted',
'sieved', 'shucked', 'slivered', 'thick', 'sliced', 'thin',
'toasted', 'trimmed', 'unbaked', 'uncooked', 'unpeeled',
'unopened', 'unseasoned']
number_regex = '((?:[\d\\./\\u00BC-\\u00BE\\u2150-\\u215E]*\s?(?:\(.+\))?)*)'
ingredient_regex = '([a-zA-Z \'\-]+)'
supplement_regex = ',?(.*)'
units_regex = "|".join([f'[{unit[0]}{unit[0].capitalize()}]{unit[1:]}'
for unit in units])
units_regex = f"((?:(?:{units_regex})e?s?)?)" units_regex = f"((?:(?:{units_regex})e?s?)?)"
instructions_regex = "|".join( instructions_regex = "|".join([f'[{inst[0]}{inst[0].capitalize()}]{inst[1:]}'
[f"[{inst[0]}{inst[0].capitalize()}]{inst[1:]}" for inst in instructions] for inst in instructions])
)
instructions_regex = f"((?:(?:(?:{instructions_regex})(?:ly)?)| )*)" instructions_regex = f"((?:(?:(?:{instructions_regex})(?:ly)?)| )*)"
return re.compile( regex = re.compile(number_regex +
number_regex units_regex +
+ units_regex instructions_regex +
+ instructions_regex ingredient_regex +
+ ingredient_regex supplement_regex)
+ supplement_regex
)
# TODO: load units and instructions from config.
# Moved data into optional parameters for the time being.
def parse_ingredient(
ingredient_text: str,
units: list[str] = [ "teaspoon", "tablespoon", "gram", "ounce", "jar",
"cup", "pinch", "container", "slice", "package",
"pound", "can", "dash", "spear", "bunch", "quart",
"cube", "envelope", "square", "sprig", "bag", "box",
"drop", "fluid ounce", "gallon", "head", "link",
"loaf", "pint", "pod", "sheet", "stalk", "whole",
"bar", "bottle", "bulb", "year", "fillet", "litter",
"packet", "slices"],
instructions: list[str] = [
"and", "or", "chopped", "diced", "brewed", "chilled", "chunky", "small",
"medium", "large", "couarse", "cracked", "crushed", "ground", "cooked",
"cubed", "crumbled", "cut", "cold", "hot", "warm", "day", "old",
"drained", "canned", "dried", "dry", "fine", "firm", "fresh", "frozen",
"grated", "grilled", "hard", "hot", "juliened?", "leftover", "light",
"lite", "mashed", "melted", "minced", "packed", "peeled", "pitted",
"sliced", "prepared", "refrigerated", "rehydrated", "seedless", "shaved",
"shredded", "sifted", "sieved", "shucked", "slivered", "thick", "sliced",
"thin", "toasted", "trimmed", "unbaked", "uncooked", "unpeeled",
"unopened", "unseasoned"],
):
regex = ingredient_regex(units, instructions)
m = regex.match(ingredient_text) m = regex.match(ingredient_text)
logging.info(f"Parsed {ingredient_text}, found: {m}") logging.info(f"Parsed {ingredient_text}, found: {m}")
@@ -64,120 +51,94 @@ def parse_ingredient(
return [text.strip() if text else None for text in m.groups()] return [text.strip() if text else None for text in m.groups()]
# this code is unused
# TODO: add tests when this is used
def missing_ingredients_query(session):
cte = (
except_(select(db.RecipeIngredient.id), select(db.RecipeIngredientParts.id))
).alias("missing")
missing = (
session.query(db.RecipeIngredient).where(db.RecipeIngredient.id.in_(cte)).all()
)
return missing
# this code is unused
# TODO: add tests when this is used def reparse_ingredients(session):
def parse_missing_ingredients(session): cte = (except_(select(db.RecipeIngredient.id),
missing = missing_ingredients_query(session) select(db.RecipeIngredientParts.id))).\
alias('missing')
missing = session.query(db.RecipeIngredient).where(db.RecipeIngredient.id.in_(cte)).all()
for ingredient in missing: for ingredient in missing:
parts = ingredient_to_parts(ingredient) parts = parse_ingredient(ingredient.text)
session.add(parts) if not parts:
continue
quantity, unit, instruction, name, supplement = parts
session.add(db.RecipeIngredientParts(id = ingredient.id,
quantity = quantity,
unit = unit,
instruction = instruction,
ingredient = name,
supplement = supplement))
def load_page(recipe_url: str) -> bs4.BeautifulSoup:
def load_page(recipe_url):
try: try:
logging.info(f"Loading Page: {recipe_url}") logging.info(f'Loading Page: {recipe_url}')
with req.get(recipe_url) as resp: with req.get(recipe_url) as f:
if resp.status_code == 404: if f.status_code == 404:
raise Exception(f"Page does not exist (404): {recipe_url}") raise Exception(f"Page does not exist (404): {recipe_url}")
return bs4.BeautifulSoup(resp.text, "html.parser") return bs4.BeautifulSoup(f.read().decode(), 'html.parser')
except Exception as e: except Exception as e:
logging.warning(f"Could not download or parse recipe: {recipe_url}") logging.warning(f"Could not download or parse recipe: {recipe_url}")
logging.warning(e) logging.warning(e)
def parse_recipe_name(
site: db.RecipeSite,
page: bs4.BeautifulSoup,
recipe: db.Recipe,
url: str = None,
) -> db.Recipe:
if not url:
url = {"site": site.base_url, "recipe": recipe.identifier}
name_candidates = page.find_all(class_=site.name_class)
if len(name_candidates) == 0:
raise Exception(f"Could not extract recipe name: {url}")
name_div = name_candidates[0]
recipe.name = name_div.text
logging.info(f"Adding Recipe {recipe.name} from {url}")
return recipe
def ingredient_to_parts(
ingredient: db.Ingredient
) -> db.RecipeIngredientParts:
parts = parse_ingredient(ingredient.text)
if parts:
quantity, unit, instruction, ingredient_name, supplement = parts
return db.RecipeIngredientParts(
id=ingredient.id,
quantity=quantity,
unit=unit,
instruction=instruction,
ingredient=ingredient_name,
supplement=supplement,
)
def parse_recipe(session, recipe, site): def parse_recipe(session, recipe, site):
recipe_url = urljoin(site.base_url, str(recipe.identifier)) recipe_url = urljoin(site.base_url, str(recipe.identifier))
recipe_page = load_page(recipe_url) recipe_page = load_page(recipe_url)
if not recipe_page: if not recipe_page:
return None return None
recipe = parse_recipe_name(site, recipe_page, recipe, recipe_url) name_candidates = recipe_page.find_all(class_=site.name_class)
if len(name_candidates) == 0:
raise Exception(f"Could not extract recipe name: {recipe_url}")
name_div = name_candidates[0]
recipe.name = name_div.text
logging.info(f"Adding Recipe {recipe.name} from {recipe_url}")
session.add(recipe) session.add(recipe)
session.flush() session.flush()
candidates = recipe_page.find_all(class_=site.ingredient_class) ingred_candidates = recipe_page.find_all(class_=site.ingredient_class)
for candidate in candidates: for candidate in ingred_candidates:
ingredient = db.RecipeIngredient(text=candidate.text, recipe_id=recipe.id) ingred = db.RecipeIngredient(text=candidate.text,
session.add(ingredient) recipe_id=recipe.id)
session.add(ingred)
session.flush() session.flush()
parts = ingredient_to_parts(ingredient) parts = parse_ingredient(ingred.text)
if parts: if parts:
session.add(parts) quantity, unit, instruction,ingredient, supplement = parts
ingred_parts = db.RecipeIngredientParts(id = ingred.id,
quantity = quantity,
unit = unit,
instruction = instruction,
ingredient = ingredient,
supplement = supplement)
session.add(ingred_parts)
logging.info(f"{len(candidates)} ingredients found. Inserting into DB") logging.info(f"{len(ingred_candidates)} ingredients found. Inserting into DB")
return recipe return recipe
def main(): # pragma: no cover def main(): # pragma: no cover
parser = ArgumentParser(description="Scrape a recipe site for recipies") parser = ArgumentParser(description="Scrape a recipe site for recipies")
parser.add_argument("site", help="Name of site") parser.add_argument('site',
parser.add_argument( help='Name of site')
"-id", parser.add_argument('-id', '--identifier', dest='id',
"--identifier", help='url of recipe(reletive to base url of site) or commma seperated list')
dest="id", parser.add_argument('-a', '--auto', action='store', dest='n',
help="url of recipe(reletive to base url of site) or commma seperated list", help='automaticaly generate identifier(must supply number of recipies to scrape)')
) parser.add_argument('-v', '--verbose', action='store_true')
parser.add_argument(
"-a",
"--auto",
action="store",
dest="n",
help="automaticaly generate identifier(must supply number of recipies to scrape)",
)
parser.add_argument("-v", "--verbose", action="store_true")
args = parser.parse_args(sys.argv) args = parser.parse_args(sys.argv)
if args.verbose: if args.verbose:
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.INFO)
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO) logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
eng = db.get_engine() eng = db.get_engine()
S = sessionmaker(eng) S = sessionmaker(eng)
@@ -190,23 +151,21 @@ def main(): # pragma: no cover
starting_id = 0 starting_id = 0
if args.id and not args.n: if args.id and not args.n:
recipe_ids.append(args.id) recipe_ids.append(args.id)
logging.info(f"Retreiving single recipe: {args.id}") logging.info(f'Retreiving single recipe: {args.id}')
elif args.n: elif args.n:
if not args.id: if not args.id:
last_recipe = ( last_recipe = sess.query(db.Recipe).\
sess.query(db.Recipe) where(db.Recipe.recipe_site_id == site.id).\
.where(db.Recipe.recipe_site_id == site.id) order_by(desc(db.Recipe.identifier)).\
.order_by(desc(db.Recipe.identifier)) limit(1).\
.limit(1) scalar()
.scalar()
)
starting_id = int(last_recipe.identifier) + 1 starting_id = int(last_recipe.identifier) + 1
else: else:
starting_id = int(args.id) starting_id = int(args.id)
recipe_ids = range(starting_id, starting_id+int(args.n)) recipe_ids = range(starting_id, starting_id+int(args.n))
logging.info( logging.info(f'Retreving {args.n} recipes from {site.base_url} starting at {starting_id}')
f"Retreving {args.n} recipes from {site.base_url} starting at {starting_id}"
)
for recipe_id in recipe_ids: for recipe_id in recipe_ids:
try: try:

View File

@@ -1,115 +1,12 @@
from recipe_graph import scrape from recipe_graph import scrape
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
from recipe_graph.db import RecipeSite, Recipe, RecipeIngredient, RecipeIngredientParts
from pytest import fixture import pytest
@fixture
def mock_site():
return RecipeSite(
name="mock-site",
ingredient_class="mock-ing",
name_class="mock-name",
base_url="example-site/mock-site",
)
# TODO: should probably load HTML from file
@fixture
def mock_page():
return BeautifulSoup(
"""
<header></header><body>
<div class="mock-name">test_recipe</div>
<div class="mock-ing">test_ingredient</div>
</body>
""",
"html.parser",
)
@fixture
def mock_blank_page():
return BeautifulSoup(""" <header></header><body> </body> """, "html.parser")
@fixture
def mock_recipe():
return Recipe(name="test_recipe", identifier="mock_1")
@fixture
def mock_ingredient():
return RecipeIngredient(text="1 ounce water")
@fixture
def mock_url():
return "example-site/mock-site"
def test_load_page(): def test_load_page():
page = scrape.load_page("https://www.google.com") page = scrape.load_page("https://hs.andreistoica.ca:4943")
assert type(page) == BeautifulSoup assert type(page) == BeautifulSoup
page = scrape.load_page("https://www.google.com/some-nonsense") page = scrape.load_page("https://hs.andreistoica.ca:4943/some-nonesense")
assert page == None assert page == None
def test_ingredient_regex():
regex = scrape.ingredient_regex(["cup"], ["crushed"])
assert (
regex.pattern
== "((?:[\\d\\./\\u00BC-\\u00BE\\u2150-\\u215E]*\\s?(?:\\(.+\\))?)*)((?:(?:[cC]up)e?s?)?)((?:(?:(?:[cC]rushed)(?:ly)?)| )*)([a-zA-Z '\\-]+),?(.*)"
)
regex = scrape.ingredient_regex(["cup", "ounce"], ["crushed", "ground"])
assert (
regex.pattern
== "((?:[\\d\\./\\u00BC-\\u00BE\\u2150-\\u215E]*\\s?(?:\\(.+\\))?)*)((?:(?:[cC]up|[oO]unce)e?s?)?)((?:(?:(?:[cC]rushed|[gG]round)(?:ly)?)| )*)([a-zA-Z '\\-]+),?(.*)"
)
def test_parse_ingredient(mock_ingredient):
parts = scrape.parse_ingredient(mock_ingredient.text)
assert len(parts) > 0
assert parts == ['1', 'ounce', '', 'water', None]
parts = scrape.parse_ingredient("Water")
assert len(parts) > 0
assert parts == [None, None, None, 'Water', None]
parts = scrape.parse_ingredient("")
assert parts == None
def test_parse_recipe_name(mock_site, mock_page, mock_recipe, mock_url, mock_blank_page,):
expected_name = mock_recipe.name
mock_recipe.name = None
mock_recipe = scrape.parse_recipe_name(
mock_site,
mock_page,
mock_recipe,
)
assert mock_recipe.name == expected_name
ex = None
try:
mock_recipe = scrape.parse_recipe_name(
mock_site,
mock_blank_page,
mock_recipe,
)
except Exception as e:
ex = e
url = {"site": mock_site.base_url, "recipe": mock_recipe.identifier}
assert str(e) == f"Could not extract recipe name: {url}"
assert ex
def test_ingredient_to_parts(mock_ingredient):
parts = scrape.ingredient_to_parts(mock_ingredient)
assert parts.quantity == "1"
assert parts.unit == "ounce"
assert parts.instruction == ""
assert parts.ingredient == "water"
assert parts.supplement == None