Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .python-version
Original file line number Diff line number Diff line change
@@ -1 +1 @@
3.13.5
3.14.4
63 changes: 31 additions & 32 deletions get_new_media.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
Companies,
Keyword,
MovieGenre,
TmdbObj,
TvGenre,
get_mcu_movie_link,
get_mcu_show_link,
Expand Down Expand Up @@ -51,22 +50,22 @@ def get_safe_title(title: str) -> str:
return safe_title


def get_release_date(movie: TmdbObj, region: str) -> Optional[date]:
def get_release_date(movie: Dict[str, Any], region: str) -> Optional[date]:
"""
Gets the release date for the given region if the details exist, otherwise
returns the default release_date
"""
for region_releases in movie.release_dates.results:
if region_releases.iso_3166_1 == region:
for release_date in region_releases.release_dates:
for region_releases in movie["release_dates"]["results"]:
if region_releases["iso_3166_1"] == region:
for release_date in region_releases["release_dates"]:
# Type 3 is Theatrical release
if release_date.type == 3:
if release_date["type"] == 3:
try:
return date.fromisoformat(release_date.release_date[:10])
return date.fromisoformat(release_date["release_date"][:10])
except ValueError:
pass
try:
return date.fromisoformat(movie.release_date)
return date.fromisoformat(movie["release_date"])
except ValueError:
return None

Expand All @@ -81,25 +80,25 @@ def handle_stale_yaml_path(existing: GoogleMediaEvent, yaml_path: Path) -> None:
existing.file_path.rename(yaml_path)


def make_movie_yamls(dir_path: Path, movies: List[TmdbObj], release_date_gte: date) -> None:
def make_movie_yamls(dir_path: Path, movies: List[Dict[str, Any]], release_date_gte: date) -> None:
"""
Makes the movie yamls for each movie from the json data
"""
existing_movies = YamlCalendar.get_movies(dir_path)
untouched_movies = list(existing_movies)

for movie in movies:
if movie.imdb_id is None:
if movie["imdb_id"] is None:
continue
release_date = get_release_date(movie, "US")
if release_date is None:
continue
yaml_path = dir_path / (get_safe_title(movie.title) + ".yaml")
yaml_path = dir_path / (get_safe_title(movie["title"]) + ".yaml")
movie_data = {
"title": movie.title,
"title": movie["title"],
"release_date": release_date,
"imdb_id": movie.imdb_id.strip(),
"description": f"https://www.imdb.com/title/{movie.imdb_id}\n",
"imdb_id": movie["imdb_id"].strip(),
"description": f"https://www.imdb.com/title/{movie['imdb_id']}\n",
}
if dir_path.stem == "mcu-movies":
official_link = get_mcu_movie_link(movie)
Expand All @@ -114,49 +113,49 @@ def make_movie_yamls(dir_path: Path, movies: List[TmdbObj], release_date_gte: da
with open(yaml_path, "w", encoding="UTF-8") as yaml_file:
yaml.safe_dump(movie_data, yaml_file, sort_keys=False)

for movie in untouched_movies:
if movie.release_date >= release_date_gte:
for untouched_movie in untouched_movies:
if untouched_movie.release_date >= release_date_gte:
# If our query didn't find a movie that was already in the yaml, it probably was canceled
movie.file_path.unlink()
untouched_movie.file_path.unlink()


def get_season_release_dates(season: TmdbObj) -> List[date]:
def get_season_release_dates(season: Dict[str, Any]) -> List[date]:
"""
Gets the number of distinct weeks in a given season
"""
air_dates = set()
for episode in season.episodes:
if episode.air_date:
air_dates.add(date.fromisoformat(episode.air_date))
for episode in season["episodes"]:
if episode["air_date"]:
air_dates.add(date.fromisoformat(episode["air_date"]))
air_dates_list = list(air_dates)
air_dates_list.sort()
return air_dates_list


def make_show_yamls(dir_path: Path, shows: List[TmdbObj]) -> None:
def make_show_yamls(dir_path: Path, shows: List[Dict[str, Any]]) -> None:
"""
Makes the show yamls for each season from the show json data
"""
for show in shows:
# print(show.name)
safe_title = get_safe_title(show.name)
for season in show.seasons:
safe_title = get_safe_title(show["name"])
for season in show["seasons"]:
show_data = {
"title": show.name,
"title": show["name"],
"release_dates": get_season_release_dates(season),
"imdb_id": show.external_ids.imdb_id,
"description": f"https://www.imdb.com/title/{show.external_ids.imdb_id}\n",
"imdb_id": show["external_ids"]["imdb_id"],
"description": f"https://www.imdb.com/title/{show['external_ids']['imdb_id']}\n",
}
if dir_path.stem == "mcu-shows":
official_link = get_mcu_show_link(show, season)
if official_link is not None:
show_data["description"] += f"{official_link}\n"

if season.season_number == 1:
if season["season_number"] == 1:
yaml_path = dir_path / (safe_title + ".yaml")
else:
show_data["title"] += f" ({season.name})"
yaml_path = dir_path / f"{safe_title}_{season.season_number}.yaml"
show_data["title"] += f" ({season['name']})"
yaml_path = dir_path / f"{safe_title}_{season['season_number']}.yaml"

with open(yaml_path, "w", encoding="UTF-8") as yaml_file:
yaml.safe_dump(show_data, yaml_file, sort_keys=False)
Expand Down Expand Up @@ -203,13 +202,13 @@ def get_new_media(release_date_gte: date) -> None:

for folder, payload in movie_queries.items():
movies = get_movies(payload)
print(folder, [s.title for s in movies])
print(folder, [s["title"] for s in movies])
make_movie_yamls(data_dir / folder, movies, release_date_gte)
progress.update(task, advance=1)

for folder, payload in show_queries.items():
shows = get_shows(payload)
print(folder, [s.name for s in shows])
print(folder, [s["name"] for s in shows])
make_show_yamls(data_dir / folder, shows)
progress.update(task, advance=1)

Expand Down
69 changes: 34 additions & 35 deletions mcu_calendar/webscraping.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,7 @@
from urllib.parse import quote_plus as url_encode

import requests
from tmdbv3api import TV, Discover, Movie, Season
from tmdbv3api.as_obj import AsObj as TmdbObj
import tmdbsimple as TMDB


class Companies(Enum):
Expand Down Expand Up @@ -101,31 +100,32 @@ class TvGenre(Enum):


def query_all_pages(
func: Callable[[Discover, int, Dict[str, Any]], List[TmdbObj]]
) -> Callable[[Dict[str, Any]], List[TmdbObj]]:
func: Callable[[TMDB.Discover, int, Dict[str, Any]], Dict[str, Any]]
) -> Callable[[Dict[str, Any]], List[Dict[str, Any]]]:
"""
Function decorator that aggregates the results of func over multiple pages
"""

@wraps(func)
def wrapper(payload: Dict[str, Any] = {}) -> List[TmdbObj]:
discoverer = Discover()
def wrapper(payload: Dict[str, Any] = {}) -> List[Dict[str, Any]]:
discoverer = TMDB.Discover()
page = 0
data = []
while True:
page += 1
data += func(discoverer, page, payload)
if discoverer.total_pages is None:
response = func(discoverer, page, payload)
data += response["results"]
if response["total_pages"] is None:
break
if page >= int(discoverer.total_pages):
if page >= int(response["total_pages"]):
break
return data

return wrapper


@query_all_pages
def _discover_movies(discoverer: Discover, page: int, payload: Dict[str, Any]) -> List[TmdbObj]:
def _discover_movies(discoverer: TMDB.Discover, page: int, payload: Dict[str, Any]) -> Dict[str, Any]:
"""
Discovers movies from TMDB on all pages
"""
Expand All @@ -134,11 +134,12 @@ def _discover_movies(discoverer: Discover, page: int, payload: Dict[str, Any]) -
"sort_by": "release_date.asc",
"page": page,
}
return discoverer.discover_movies({**base_payload, **payload})

return discoverer.movie(**{**base_payload, **payload})


@query_all_pages
def _discover_shows(discoverer: Discover, page: int, payload: Dict[str, Any]) -> List[TmdbObj]:
def _discover_shows(discoverer: TMDB.Discover, page: int, payload: Dict[str, Any]) -> Dict[str, Any]:
"""
Discovers shows from TMDB on all pages
"""
Expand All @@ -149,55 +150,53 @@ def _discover_shows(discoverer: Discover, page: int, payload: Dict[str, Any]) ->
"sort_by": "first_air_date.asc",
"page": page,
}
return discoverer.discover_tv_shows({**base_payload, **payload})

return discoverer.tv(**{**base_payload, **payload})


def get_movies(payload: Dict[str, Any]) -> List[TmdbObj]:
def get_movies(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
"""
Gets movies from themoviedb.org with the given keyword
"""
movies = _discover_movies(payload)
movies = [m for m in movies if "release_date" in m and m.release_date != ""]
movie_api = Movie()
movies = [m for m in movies if "release_date" in m and m["release_date"] != ""]
movie_details = []
for movie in movies:
movie_details.append(movie_api.details(movie["id"], append_to_response="release_dates"))
movie_details.append(TMDB.Movies(movie["id"]).info(append_to_response="release_dates"))

return movie_details


def should_skip(season: TmdbObj, payload: Dict[str, Any]) -> bool:
def should_skip(season: Dict[str, Any], payload: Dict[str, Any]) -> bool:
"""
Checks if the given season should be skipped based on the payload filter criteria
"""
if season.air_date is None:
if season["air_date"] is None:
return True
if "air_date.gte" in payload:
return season.air_date < payload["air_date.gte"]
return season["air_date"] < payload["air_date.gte"]
if "air_date.lte" in payload:
return season.air_date > payload["air_date.lte"]
return season["air_date"] > payload["air_date.lte"]
return False


def get_shows(payload: Dict[str, Any]) -> List[TmdbObj]:
def get_shows(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
"""
Gets tv shwos from themoviedb.org with the given keyword
"""
shows = _discover_shows(payload)
shows = [s for s in shows if "first_air_date" in s and s.first_air_date != ""]
shows = [s for s in shows if "first_air_date" in s and s["first_air_date"] != ""]
# The discover api doesn't return season information, so we
# still need to get the details
tv_api = TV()
season_api = Season()
show_details = []
for show in shows:
show_detail = tv_api.details(show["id"], append_to_response="external_ids")
show_detail = TMDB.TV(show["id"]).info(append_to_response="external_ids")
season_details = []
for season in show_detail.seasons:
for season in show_detail["seasons"]:
if should_skip(season, payload):
continue
season_details.append(season_api.details(show.id, season.season_number))
show_detail.seasons = season_details
season_details.append(TMDB.TV_Seasons(show["id"], season["season_number"]).info())
show_detail["seasons"] = season_details
show_details.append(show_detail)

return show_details
Expand All @@ -208,25 +207,25 @@ def get_shows(payload: Dict[str, Any]) -> List[TmdbObj]:
GOOGLE_SEARCH_FOMRAT = "https://www.googleapis.com/customsearch/v1?key={api_key}&cx={cx}&q={query}"


def get_mcu_movie_link(movie: TmdbObj) -> Optional[str]:
def get_mcu_movie_link(movie: Dict[str, Any]) -> Optional[str]:
"""
Searches google to try to find the official webpage for the given mcu movie
"""
return __get_search_link(
movie.title,
movie["title"],
MARVEL_MOVIES_CX,
url_encode(movie.title),
url_encode(movie["title"]),
)


def get_mcu_show_link(show: TmdbObj, season: TmdbObj) -> Optional[str]:
def get_mcu_show_link(show: Dict[str, Any], season: Dict[str, Any]) -> Optional[str]:
"""
Searches google to try to find the official webpage for the given mcu show/season
"""
return __get_search_link(
show.name,
show["name"],
MARVEL_SHOWS_CX,
url_encode(f"{show.name} {season.name}"),
url_encode(f"{show['name']} {season['name']}"),
)


Expand Down
2 changes: 1 addition & 1 deletion requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,6 @@ pylint==3.0.3
pytest==8.0.1
PyYAML==6.0.1
rich==13.7.0
tmdbv3api==1.9.0
tmdbsimple==2.9.6
types-PyYAML==6.0.12.12
types-requests==2.31.0.20240218
Loading