MIL-004: top 100 movies scraper #38
@@ -1,6 +1,7 @@
|
||||
"""@brief The ports (interfaces) the use cases need from the outside world."""
|
||||
|
||||
from collections.abc import Callable
|
||||
from collections.abc import Callable, Sequence
|
||||
from pathlib import Path
|
||||
from typing import TypeAlias
|
||||
|
||||
from scrape_movies.domain.story import Story
|
||||
@@ -9,3 +10,7 @@ from scrape_movies.domain.story import Story
|
||||
HtmlFetcher: TypeAlias = Callable[[str], str]
|
||||
## Turns Hacker News HTML into stories.
|
||||
StoryParser: TypeAlias = Callable[[str], list[Story]]
|
||||
## Turns the Empire page HTML into the movie lines found on it, in page order.
|
||||
MovieParser: TypeAlias = Callable[[str], list[str]]
|
||||
## Writes lines to a file, one per line.
|
||||
MovieWriter: TypeAlias = Callable[[Sequence[str], Path], None]
|
||||
|
||||
@@ -1,26 +1,67 @@
|
||||
"""@brief Use cases of the scraping exercises."""
|
||||
|
||||
from scrape_movies.application.ports import HtmlFetcher, StoryParser
|
||||
from scrape_movies.domain.story import Story, top_story
|
||||
|
||||
|
||||
def list_stories(fetch: HtmlFetcher, parse: StoryParser, url: str) -> list[Story]:
|
||||
"""@brief Fetch a Hacker News page and read its stories.
|
||||
|
||||
@param fetch Fetches the HTML of the page.
|
||||
@param parse Reads the stories from the HTML.
|
||||
@param url The page address.
|
||||
@return The stories in page order.
|
||||
"""
|
||||
return parse(fetch(url))
|
||||
|
||||
|
||||
def find_top_story(fetch: HtmlFetcher, parse: StoryParser, url: str) -> Story | None:
|
||||
"""@brief Find the story with the most upvotes on a Hacker News page.
|
||||
|
||||
@param fetch Fetches the HTML of the page.
|
||||
@param parse Reads the stories from the HTML.
|
||||
@param url The page address.
|
||||
@return The top story, or None when the page has no stories.
|
||||
"""
|
||||
return top_story(list_stories(fetch, parse, url))
|
||||
"""@brief Use cases of the scraping exercises."""
|
||||
|
||||
from collections.abc import Sequence
|
||||
from pathlib import Path
|
||||
|
||||
from scrape_movies.application.ports import (
|
||||
HtmlFetcher,
|
||||
MovieParser,
|
||||
MovieWriter,
|
||||
StoryParser,
|
||||
)
|
||||
from scrape_movies.domain.movie import format_titles
|
||||
from scrape_movies.domain.story import Story, top_story
|
||||
|
||||
|
||||
def list_stories(fetch: HtmlFetcher, parse: StoryParser, url: str) -> list[Story]:
|
||||
"""@brief Fetch a Hacker News page and read its stories.
|
||||
|
||||
@param fetch Fetches the HTML of the page.
|
||||
@param parse Reads the stories from the HTML.
|
||||
@param url The page address.
|
||||
@return The stories in page order.
|
||||
"""
|
||||
return parse(fetch(url))
|
||||
|
||||
|
||||
def find_top_story(fetch: HtmlFetcher, parse: StoryParser, url: str) -> Story | None:
|
||||
"""@brief Find the story with the most upvotes on a Hacker News page.
|
||||
|
||||
@param fetch Fetches the HTML of the page.
|
||||
@param parse Reads the stories from the HTML.
|
||||
@param url The page address.
|
||||
@return The top story, or None when the page has no stories.
|
||||
"""
|
||||
return top_story(list_stories(fetch, parse, url))
|
||||
|
||||
|
||||
def scrape_movie_titles(fetch: HtmlFetcher, parse: MovieParser, url: str) -> list[str]:
|
||||
"""@brief Fetch the Empire page and read the titles in page order.
|
||||
|
||||
@param fetch Fetches the HTML of the page.
|
||||
@param parse Reads the titles from the HTML.
|
||||
@param url The page address.
|
||||
@return The titles in page order (100 down to 1).
|
||||
"""
|
||||
return parse(fetch(url))
|
||||
|
||||
|
||||
def save_movie_list(
|
||||
fetch: HtmlFetcher,
|
||||
parse: MovieParser,
|
||||
write: MovieWriter,
|
||||
url: str,
|
||||
path: Path,
|
||||
) -> Sequence[str]:
|
||||
"""@brief Scrape the list and write it, ascending and numbered, to a file.
|
||||
|
||||
@param fetch Fetches the HTML of the page.
|
||||
@param parse Reads the titles from the HTML.
|
||||
@param write Writes the lines to the file.
|
||||
@param url The page address.
|
||||
@param path The file to write, for example `movies.txt`.
|
||||
@return The lines that were written.
|
||||
"""
|
||||
lines = format_titles(scrape_movie_titles(fetch, parse, url))
|
||||
write(lines, path)
|
||||
return lines
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
"""@brief The movie list rule: ascending order, every line starting with its rank."""
|
||||
|
||||
import re
|
||||
from collections.abc import Sequence
|
||||
|
||||
_RANK_PREFIX = re.compile(r"^\d+\) ")
|
||||
|
||||
|
||||
def format_titles(titles: Sequence[str]) -> list[str]:
|
||||
"""@brief Turn a descending page list into an ascending, numbered list.
|
||||
|
||||
Empire lists the movies from 100 down to 1, so the page order is reversed.
|
||||
A title that already starts with its rank, such as `3) Jaws`, is kept as
|
||||
it is; any other title gets `<rank>) ` in front of it.
|
||||
|
||||
@param titles The titles in page order.
|
||||
@return The lines `1) ...`, `2) ...` in ascending order.
|
||||
"""
|
||||
lines: list[str] = []
|
||||
for rank, title in enumerate(reversed(titles), 1):
|
||||
lines.append(title if _RANK_PREFIX.match(title) else f"{rank}) {title}")
|
||||
return lines
|
||||
@@ -0,0 +1,15 @@
|
||||
"""@brief Writing the movie list to a text file."""
|
||||
|
||||
from collections.abc import Sequence
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def write_movies(lines: Sequence[str], path: Path) -> None:
|
||||
"""@brief Write one line per movie, UTF-8, with a final newline.
|
||||
|
||||
@param lines The lines to write.
|
||||
@param path The file to create or overwrite.
|
||||
"""
|
||||
with path.open("w", encoding="utf-8", newline="\n") as file:
|
||||
for line in lines:
|
||||
file.write(f"{line}\n")
|
||||
@@ -2,8 +2,9 @@
|
||||
|
||||
import argparse
|
||||
from collections.abc import Sequence
|
||||
from pathlib import Path
|
||||
|
||||
from scrape_movies.presentation import hacker_news_steps, soup_steps
|
||||
from scrape_movies.presentation import hacker_news_steps, movies_steps, soup_steps
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
@@ -26,6 +27,17 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
default="mirror",
|
||||
help="the course mirror (default), the live site or the saved copy",
|
||||
)
|
||||
movies = commands.add_parser("movies", help="scrape Empire's 100 greatest movies")
|
||||
movies.add_argument("step", choices=movies_steps.STEPS)
|
||||
movies.add_argument(
|
||||
"--source",
|
||||
choices=movies_steps.SOURCES,
|
||||
default="wayback",
|
||||
help="the Wayback snapshot (default), the live site or the saved copy",
|
||||
)
|
||||
movies.add_argument(
|
||||
"--output", type=Path, help="file written by the write step (movies.txt)"
|
||||
)
|
||||
return parser
|
||||
|
||||
|
||||
@@ -43,5 +55,8 @@ def main(argv: Sequence[str] | None = None) -> int:
|
||||
if args.command == "hacker-news":
|
||||
print(hacker_news_steps.render_step(args.step, args.source))
|
||||
return 0
|
||||
if args.command == "movies":
|
||||
print(movies_steps.render_step(args.step, args.source, args.output))
|
||||
return 0
|
||||
parser.print_help()
|
||||
return 0
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
"""@brief Text output of each "100 Movies to Watch" lecture step."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from scrape_movies.application import use_cases
|
||||
from scrape_movies.application.ports import HtmlFetcher
|
||||
from scrape_movies.constants import (
|
||||
EMPIRE_FIXTURE_PATH,
|
||||
EMPIRE_LIVE_URL,
|
||||
EMPIRE_WAYBACK_URL,
|
||||
MOVIES_OUTPUT_FILE,
|
||||
)
|
||||
from scrape_movies.infrastructure import empire_movies, http_client, soup_basics
|
||||
from scrape_movies.infrastructure.movie_file import write_movies
|
||||
|
||||
## Where a step reads the page: the Wayback snapshot, the live site or the saved copy.
|
||||
SOURCES: tuple[str, ...] = ("wayback", "live", "fixture")
|
||||
_URLS: dict[str, str] = {
|
||||
"wayback": EMPIRE_WAYBACK_URL,
|
||||
"live": EMPIRE_LIVE_URL,
|
||||
"fixture": str(EMPIRE_FIXTURE_PATH),
|
||||
}
|
||||
## The lecture steps by command-line name, in lecture order.
|
||||
STEPS: tuple[str, ...] = ("markup", "titles", "reversed", "write")
|
||||
_SHOWN_LINES = 12
|
||||
|
||||
|
||||
def _fetcher(source: str) -> HtmlFetcher:
|
||||
if source == "fixture":
|
||||
return lambda path: soup_basics.load_html(EMPIRE_FIXTURE_PATH)
|
||||
return http_client.fetch_html
|
||||
|
||||
|
||||
def _head(lines: list[str]) -> str:
|
||||
shown = lines[:_SHOWN_LINES]
|
||||
more = len(lines) - len(shown)
|
||||
return "\n".join(shown + ([f"... and {more} more"] if more > 0 else []))
|
||||
|
||||
|
||||
def render_step(step: str, source: str, output: Path | None = None) -> str:
|
||||
"""@brief Run one lecture step.
|
||||
|
||||
@param step One of STEPS.
|
||||
@param source One of SOURCES.
|
||||
@param output The file the `write` step creates; `movies.txt` when None.
|
||||
@return The text the step prints.
|
||||
@throws KeyError When the step or the source is unknown.
|
||||
"""
|
||||
if step not in STEPS:
|
||||
raise KeyError(step)
|
||||
fetch, url = _fetcher(source), _URLS[source]
|
||||
if step == "markup":
|
||||
return empire_movies.first_title_markup(fetch(url))
|
||||
if step == "titles":
|
||||
titles = use_cases.scrape_movie_titles(fetch, empire_movies.parse_titles, url)
|
||||
return _head(titles)
|
||||
target = output or Path(MOVIES_OUTPUT_FILE)
|
||||
lines = use_cases.save_movie_list(
|
||||
fetch, empire_movies.parse_titles, write_movies, url, target
|
||||
)
|
||||
if step == "reversed":
|
||||
return _head(list(lines))
|
||||
return f"{_head(list(lines))}\nWrote {len(lines)} lines to {target}"
|
||||
Reference in New Issue
Block a user