path: root/addon.py

# Copyright © 2024 Jakob L. Kreuze <zerodaysfordays@sdf.org>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

"""
Video plugin for animations posted to Newgrounds.
"""

import concurrent.futures
from datetime import datetime
import dateutil
import os
import sys
from urllib.parse import urlencode, parse_qsl

from bs4 import BeautifulSoup
import requests
import xbmc
import xbmcgui
import xbmcplugin
from xbmcaddon import Addon
from xbmcvfs import translatePath

URL = sys.argv[0]
HANDLE = int(sys.argv[1])

ADDON_PATH = translatePath(Addon().getAddonInfo("path"))
ICONS_DIR = os.path.join(ADDON_PATH, "resources", "images", "icons")
FANART_DIR = os.path.join(ADDON_PATH, "resources", "images", "fanart")


def get_url(**kwargs):
    """
    Create a URL for calling the plugin recursively from the given set of keyword arguments.

    :param kwargs: "argument=value" pairs
    :return: plugin call URL
    :rtype: str
    """
    return "{}?{}".format(URL, urlencode(kwargs))


def list_top_level():
    """
    Create the list of entrypoints in the Kodi interface.
    """
    xbmcplugin.setPluginCategory(HANDLE, "Newgrounds")
    xbmcplugin.setContent(HANDLE, "movies")

    genres = ["Featured", "Latest", "Popular"]

    # Add a search dialog
    list_item = xbmcgui.ListItem(label="Search")
    info_tag = list_item.getVideoInfoTag()
    info_tag.setMediaType("video")
    info_tag.setTitle("Search")
    url = get_url(action="prompt_search")
    is_folder = False
    xbmcplugin.addDirectoryItem(HANDLE, url, list_item, is_folder)

    for name in genres:
        list_item = xbmcgui.ListItem(label=name)

        info_tag = list_item.getVideoInfoTag()
        info_tag.setMediaType("video")
        info_tag.setTitle(name)
        info_tag.setGenres([name])

        url = get_url(action="listing", genre=name)
        is_folder = True

        xbmcplugin.addDirectoryItem(HANDLE, url, list_item, is_folder)
    xbmcplugin.addSortMethod(HANDLE, xbmcplugin.SORT_METHOD_LABEL_IGNORE_THE)
    xbmcplugin.endOfDirectory(HANDLE)


def list_categories(genre="Featured"):
    """
    Create the list of movie categories in the Kodi interface.
    """
    xbmcplugin.setPluginCategory(HANDLE, "Newgrounds")
    xbmcplugin.setContent(HANDLE, "movies")

    categories = {
        "All": 0,
        "Action": 45,
        "Comedy - Original": 60,
        "Comedy - Parody": 61,
        "Drama": 47,
        "Experimental": 49,
        "Informative": 48,
        "Music Video": 50,
        "Other": 51,
        "Spam": 55,
    }

    for name in categories.keys():
        list_item = xbmcgui.ListItem(label=name)
        info_tag = list_item.getVideoInfoTag()
        info_tag.setMediaType("video")
        info_tag.setTitle(name)
        info_tag.setGenres([name])
        url = get_url(action="listing", genre=genre, category=categories[name])
        is_folder = True
        xbmcplugin.addDirectoryItem(HANDLE, url, list_item, is_folder)
    xbmcplugin.addSortMethod(HANDLE, xbmcplugin.SORT_METHOD_LABEL_IGNORE_THE)
    xbmcplugin.endOfDirectory(HANDLE)


def get_video_url(url):
    """
    Return the location of the stream for a movie.

    :param url: URL of Newgrounds movie page
    :return: stream URL
    :rtype: str
    """
    html = requests.get(url).text
    soup = BeautifulSoup(html, "html.parser")
    video_id = soup.find("meta", property="og:url")["content"].strip(
        "https://www.newgrounds.com/portal/view/"
    )
    response = requests.get(
        "https://www.newgrounds.com/portal/video/{}".format(video_id),
        headers={"X-Requested-With": "XMLHttpRequest"},
    ).json()
    preferences = ["1080p", "720p", "360p"]
    for preference in preferences:
        if preference in response["sources"]:
            source = next(
                filter(
                    lambda x: x["type"] == "video/mp4", response["sources"][preference]
                )
            )
            if source is not None:
                return source["src"]


def get_addon_video_info(url):
    """
    Scrape metadata from `url`.

    :param url: URL of Newgrounds movie page
    :return: dictionary of video metadata
    :rtype: dict
    """
    html = requests.get(url).text
    soup = BeautifulSoup(html, "html.parser")

    title = soup.find("div", id="embed_header").h2.string
    thumbnail = soup.find("meta", property="og:image")
    if thumbnail is not None:
        thumbnail = thumbnail["content"]

    metadata = {
        "title": title,
        "thumbnail": thumbnail,
        "url": url,
        "views": 0,
        "faves": 0,
        "votes": 0,
        "score": 0.0,
        "upload_date": datetime.fromtimestamp(0),
        "genre": "Unknown",
        "tags": [],
        "description": "",
        "authors": [],
        "rating": "e",
    }

    sidestats_popularity = soup.find_all("dl", "sidestats")[0].find_all("dd")
    if len(sidestats_popularity) == 4:
        metadata["views"] = int(sidestats_popularity[0].string.replace(",", ""))
        faves = sidestats_popularity[1].a
        if faves is not None:
            metadata["faves"] = int(faves.string.replace(",", ""))
        metadata["votes"] = int(sidestats_popularity[2].string)
        score = sidestats_popularity[3].span
        if score is not None:
            metadata["score"] = float(score.string)

    sidestats_date_and_genre = soup.find_all("dl", "sidestats")[1].find_all("dd")
    if len(sidestats_date_and_genre) == 3:
        metadata["upload_date"] = dateutil.parser.parse(
            sidestats_date_and_genre[0].string
            + " "
            + sidestats_date_and_genre[1].string
        )
        genre = sidestats_date_and_genre[2].a
        if genre is not None:
            metadata["genre"] = genre.string

    tags = soup.find("dd", "tags")
    if tags is not None:
        metadata["tags"] = [tag.string for tag in tags.find_all("a")]

    author_comments = soup.find("div", id="author_comments")
    metadata["description"] = " ".join([s for s in author_comments.strings]).strip()

    authors = soup.find("ul", "authorlinks")
    metadata["authors"] = [
        author.string for author in authors.find_all("a") if author.get("class") is None
    ]

    if len(soup.find_all("h2", "rated-a")) != 0:
        rating = "a"
    elif len(soup.find_all("h2", "rated-m")) != 0:
        rating = "m"
    elif len(soup.find_all("h2", "rated-t")) != 0:
        rating = "t"
    else:
        rating = "e"

    return metadata


def get_frontpage_video_urls(
    grouping="Featured", interval="today", sort_by="date", category=0, offset=0
):
    """
    Return a list of movies from the Newgrounds frontpage.

    :param grouping: one of "Featured", "Latest", or "Popular"
    :param interval: one of "today", "yesterday", "week", "month", "year", "all"
    :param sort_by: one of "date", score", "views"
    :param category: numeric identifier for video category
    :param offset: pagination offset
    :return: list of video URLs
    :rtype: list
    """
    URLS = {
        "Featured": "https://www.newgrounds.com/movies/featured",
        "Latest": "https://www.newgrounds.com/movies/browse",
        "Popular": "https://www.newgrounds.com/movies/popular",
    }

    assert interval in ["today", "yesterday", "week", "month", "year", "all"]
    assert sort_by in ["date", "score", "views"]
    assert isinstance(category, int)
    assert isinstance(offset, int)

    url = URLS[grouping]
    response = requests.get(
        url,
        params={
            "interval": interval,
            "sort": sort_by,
            "genre": category,
            "isAjaxRequest": True,
            "offset": offset,
            "inner": 1,
        },
        headers={
            "X-Requested-With": "XMLHttpRequest",
        },
    ).json()

    soup = BeautifulSoup(response["content"], "html.parser")
    return [tag["href"] for tag in soup.find_all("a", "inline-card-portalsubmission")]


def search(term, sort_by="relevance", offset=1):
    """
    Return a list of movies matching a given search term.

    :param term: Term to search for
    :param sort_by: one of "relevance", "date-asc", "date-desc", "score-asc", "score-desc", "views-asc", "views-desc"
    :param offset: pagination offset
    :return: list of video URLs
    :rtype: list
    """
    assert isinstance(term, str)
    assert sort_by in [
        "relevance",
        "date-asc",
        "date-desc",
        "score-asc",
        "score-desc",
        "views-asc",
        "views-desc",
    ]
    assert isinstance(offset, int)

    response = requests.get(
        "https://www.newgrounds.com/search/conduct/movies",
        params={
            "suitabilities": "etm",
            "sort": sort_by,
            "terms": term,
            "page": offset,
            "inner": 1,
        },
        headers={
            "X-Requested-With": "XMLHttpRequest",
        },
    ).json()

    soup = BeautifulSoup(response["content"], "html.parser")
    return [tag["href"] for tag in soup.find_all("a", "item-portalsubmission")]


def list_videos(video_urls, title=None, next_page=None):
    """
    Create the list of playable videos in the Kodi interface.

    :param genre_index: the index of genre in the list of movie genres
    :type genre: str
    :type offset: int
    """
    videos = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
        # Start the load operations and mark each future with its URL
        future_to_url = {
            executor.submit(get_addon_video_info, url): url for url in video_urls
        }
        for future in concurrent.futures.as_completed(future_to_url):
            url = future_to_url[future]
            videos.append(future.result())
    xbmcplugin.setPluginCategory(HANDLE, title)
    xbmcplugin.setContent(HANDLE, "movies")

    for video in videos:
        list_item = xbmcgui.ListItem(label=video["title"])
        if video["thumbnail"] is not None:
            list_item.setArt({"thumb": video["thumbnail"]})

        info_tag = list_item.getVideoInfoTag()
        info_tag.setMediaType("movie")
        info_tag.setTitle(video["title"])
        info_tag.setYear(video["upload_date"].year)
        info_tag.setRating(video["score"], video["votes"])
        info_tag.setPlaycount(video["views"])
        info_tag.setPlot(video["description"])
        info_tag.setGenres([video["genre"]])
        info_tag.setDirectors(video["authors"])
        info_tag.setPremiered(video["upload_date"].strftime("%Y-%m-%d"))
        info_tag.setTags(video["tags"])

        if video["rating"] == "a":
            info_tag.setMpaa("NC-17")
        elif video["rating"] == "m":
            info_tag.setMpaa("R")
        elif video["rating"] == "t":
            info_tag.setMpaa("PG-13")
        else:
            info_tag.setMpaa("PG")

        list_item.setProperty("IsPlayable", "true")

        url = get_url(action="play", video=video["url"])
        is_folder = False
        xbmcplugin.addDirectoryItem(HANDLE, url, list_item, is_folder)

    # Add a "continue" folder.
    if next_page is not None and len(video_urls) > 0:
        list_item = xbmcgui.ListItem(label="Next Page")
        info_tag = list_item.getVideoInfoTag()
        info_tag.setMediaType("video")
        info_tag.setTitle("Next Page")
        is_folder = True
        xbmcplugin.addDirectoryItem(HANDLE, next_page, list_item, is_folder)

    xbmcplugin.addSortMethod(HANDLE, xbmcplugin.SORT_METHOD_LABEL_IGNORE_THE)
    xbmcplugin.addSortMethod(HANDLE, xbmcplugin.SORT_METHOD_VIDEO_YEAR)
    xbmcplugin.endOfDirectory(HANDLE)


def play_video(path):
    play_item = xbmcgui.ListItem(path=get_video_url(path))
    play_item.setProperty("IsPlayable", "true")
    xbmcplugin.setResolvedUrl(HANDLE, True, play_item)


def router(paramstring):
    """
    Router function that calls other functions
    depending on the provided paramstring

    :param paramstring: URL encoded plugin paramstring
    :type paramstring: str
    """
    params = dict(parse_qsl(paramstring))
    if not params:
        list_top_level()
    elif params["action"] == "prompt_search":
        dialog = xbmcgui.Dialog()
        term = dialog.input(heading="Search")
        url = get_url(action="search", term=term)
        xbmc.executebuiltin("Container.Update(%s)" % url)
    elif params["action"] == "listing" and "category" in params:
        offset = int(params.get("offset", 0))
        video_urls = get_frontpage_video_urls(
            grouping=params["genre"],
            interval="all",
            category=params["category"],
            offset=offset,
        )
        next_page = get_url(
            action="listing",
            genre=params["genre"],
            category=params["category"],
            offset=offset + 20,
        )
        list_videos(video_urls, params["genre"], next_page)
    elif params["action"] == "listing":
        list_categories(params["genre"])
    elif params["action"] == "search":
        term = params["term"]
        offset = int(params.get("offset", 1))
        video_urls = search(term, offset=offset)
        next_page = get_url(action="search", term=term, offset=offset + 1)
        list_videos(video_urls, "Search Results: {}".format(term), next_page)
    elif params["action"] == "play":
        play_video(params["video"])
    else:
        raise ValueError(f"Invalid paramstring: {paramstring}!")


if __name__ == "__main__":
    router(sys.argv[2][1:])