Feature/channel info cli

setup.cfg

@@ -24,6 +24,10 @@ packages = find:
 python_requires = >=3.7
 install_requires = file: requirements/base.txt
 
+[options.entry_points]
+console_scripts =
+    youtool = youtool.cli:main
+
 [options.extras_require]
 cli = file: requirements/cli.txt
 dev = file: requirements/dev.txt

youtool/__init__.py

@@ -11,7 +11,7 @@
 import isodate  # TODO: implement duration parser to remove dependency?
 import requests
 
-REGEXP_CHANNEL_ID = re.compile('"externalId":"([^"]+)"')
+REGEXP_CHANNEL_ID = re.compile('"channelId":"([^"]+)"')
 REGEXP_LOCATION_RADIUS = re.compile(r"^[0-9.]+(?:m|km|ft|mi)$")
 REGEXP_NAIVE_DATETIME = re.compile(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}[T ][0-9]{2}:[0-9]{2}:[0-9]{2}$")
 REGEXP_DATETIME_MILLIS = re.compile(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}[T ][0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]+")

youtool/cli.py

@@ -0,0 +1,46 @@
+import argparse
+import os
+
+from youtool.commands import COMMANDS
+
+
+def main():
+    """Main function for the YouTube CLI Tool.
+
+    This function sets up the argument parser for the CLI tool, including options for the YouTube API key and
+    command-specific subparsers. It then parses the command-line arguments, retrieving the YouTube API key
+    from either the command-line argument '--api-key' or the environment variable 'YOUTUBE_API_KEY'. If the API
+    key is not provided through any means, it raises an argparse.ArgumentError.
+
+    Finally, the function executes the appropriate command based on the parsed arguments. If an exception occurs
+    during the execution of the command, it is caught and raised as an argparse error for proper handling.
+
+    Raises:
+        argparse.ArgumentError: If the YouTube API key is not provided.
+        argparse.ArgumentError: If there is an error during the execution of the command.
+    """
+    parser = argparse.ArgumentParser(description="CLI Tool for managing YouTube videos add playlists")
+    parser.add_argument("--api-key", type=str, help="YouTube API Key", dest="api_key")
+    parser.add_argument("--debug", help="Debug mode", dest="debug", default=False, action="store_true")
+
+    subparsers = parser.add_subparsers(required=True, dest="command", title="Command", help="Command to be executed")
+
+    for command in COMMANDS:
+        command.parse_arguments(subparsers)
+
+    args = parser.parse_args()
+    args.api_key = args.api_key or os.environ.get("YOUTUBE_API_KEY")
+
+    if not args.api_key:
+        parser.error("YouTube API Key is required")
+
+    try:
+        print(args.func(**args.__dict__))
+    except Exception as error:
+        if args.debug:
+            raise error
+        parser.error(error)
+
+
+if __name__ == "__main__":
+    main()

youtool/commands/__init__.py

@@ -0,0 +1,12 @@
+from .base import Command
+from .channel_id import ChannelId
+from .channel_info import ChannelInfo
+
+COMMANDS = [ChannelId, ChannelInfo]
+
+__all__ = [
+    "Command",
+    "COMMANDS",
+    "ChannelId",
+    "ChannelInfo",
+]

youtool/commands/base.py

@@ -0,0 +1,125 @@
+import argparse
+import csv
+from datetime import datetime
+from io import StringIO
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+
+class Command:
+    """A base class for commands to inherit from, following a specific structure.
+
+    Attributes:
+        name (str): The name of the command.
+        arguments (List[Dict[str, Any]]): A list of dictionaries, each representing an argument for the command.
+    """
+
+    name: str
+    arguments: List[Dict[str, Any]]
+
+    @classmethod
+    def generate_parser(cls, subparsers: argparse._SubParsersAction):
+        """Creates a parser for the command and adds it to the subparsers.
+
+        Args:
+            subparsers (argparse._SubParsersAction): The subparsers action to add the parser to.
+
+        Returns:
+            argparse.ArgumentParser: The parser for the command.
+        """
+        return subparsers.add_parser(cls.name, help=cls.__doc__)
+
+    @classmethod
+    def parse_arguments(cls, subparsers: argparse._SubParsersAction) -> None:
+        """Parses the arguments for the command and sets the command's execute method as the default function to call.
+
+        Args:
+            subparsers (argparse._SubParsersAction): The subparsers action to add the parser to.
+        """
+        parser = cls.generate_parser(subparsers)
+        groups = {}
+
+        for argument in cls.arguments:
+            argument_copy = {**argument}
+            argument_name = argument_copy.pop("name")
+
+            group_name = argument_copy.pop("mutually_exclusive_group", None)
+            if group_name:
+                if group_name not in groups:
+                    groups[group_name] = parser.add_argument_group(group_name)
+                groups[group_name].add_argument(argument_name, **argument_copy)
+            else:
+                parser.add_argument(argument_name, **argument_copy)
+        parser.set_defaults(func=cls.execute)
+
+    @classmethod
+    def execute(cls, **kwargs) -> str:  # noqa: D417
+        """Executes the command.
+
+        This method should be overridden by subclasses to define the command's behavior.
+
+        Args:
+            arguments (argparse.Namespace): The parsed arguments for the command.
+        """
+        raise NotImplementedError()
+
+    @staticmethod
+    def data_from_csv(file_path: Path, data_column_name: Optional[str] = None) -> List[str]:
+        """Extracts a list of URLs from a specified CSV file.
+
+        Args:
+            file_path: The path to the CSV file containing the URLs.
+            data_column_name: The name of the column in the CSV file that contains the URLs.
+                                If not provided, it defaults to `ChannelId.URL_COLUMN_NAME`.
+
+        Returns:
+            A list of URLs extracted from the specified CSV file.
+
+        Raises:
+            Exception: If the file path is invalid or the file cannot be found.
+        """
+        data = []
+
+        if not file_path.is_file():
+            raise FileNotFoundError(f"Invalid file path: {file_path}")
+
+        with file_path.open("r", newline="") as csv_file:
+            reader = csv.DictReader(csv_file)
+            fieldnames = reader.fieldnames
+
+            if fieldnames is None:
+                raise ValueError("Fieldnames is None")
+
+            if data_column_name not in fieldnames:
+                raise Exception(f"Column {data_column_name} not found on {file_path}")
+            for row in reader:
+                value = row.get(data_column_name)
+                if value is not None:
+                    data.append(str(value))
+        return data
+
+    @classmethod
+    def data_to_csv(cls, data: List[Dict], output_file_path: Optional[str] = None) -> str:
+        """Converts a list of channel IDs into a CSV file.
+
+        Parameters:
+        channels_ids (List[str]): List of channel IDs to be written to the CSV.
+        output_file_path (str, optional): Path to the file where the CSV will be saved. If not provided, the CSV will be returned as a string.
+        channel_id_column_name (str, optional): Name of the column in the CSV that will contain the channel IDs.
+                                                If not provided, the default value defined in ChannelId.CHANNEL_ID_COLUMN_NAME will be used.
+
+        Returns:
+        str: The path of the created CSV file or, if no path is provided, the contents of the CSV as a string.
+        """
+        if output_file_path:
+            output_path = Path(output_file_path)
+            if output_path.is_dir():
+                command_name = cls.name.replace("-", "_")
+                timestamp = datetime.now().strftime("%M%S%f")
+                output_file_path = output_path / f"{command_name}_{timestamp}.csv"
+
+        with Path(output_file_path).open("w", newline="") if output_file_path else StringIO() as csv_file:
+            writer = csv.DictWriter(csv_file, fieldnames=list(data[0].keys()) if data else [])
+            writer.writeheader()
+            writer.writerows(data)
+            return str(output_file_path) if output_file_path else csv_file.getvalue()

youtool/commands/channel_id.py

@@ -0,0 +1,92 @@
+from pathlib import Path
+
+from youtool import YouTube
+
+from .base import Command
+
+
+class ChannelId(Command):
+    """Get channel IDs from a list of URLs (or CSV filename with URLs inside), generate CSV output (just the IDs)."""
+
+    name = "channel-id"
+    arguments = [
+        {
+            "name": "--urls",
+            "type": str,
+            "help": "Channels urls",
+            "nargs": "*",
+            "mutually_exclusive_group": "input_source",
+        },
+        {
+            "name": "--urls-file-path",
+            "type": Path,
+            "help": "Channels urls csv file path",
+            "mutually_exclusive_group": "input_source",
+        },
+        {"name": "--output-file-path", "type": Path, "help": "Output csv file path"},
+        {"name": "--url-column-name", "type": str, "help": "URL column name on csv input files"},
+        {"name": "--id-column-name", "type": str, "help": "Channel ID column name on csv output files"},
+    ]
+
+    URL_COLUMN_NAME: str = "channel_url"
+    CHANNEL_ID_COLUMN_NAME: str = "channel_id"
+
+    @classmethod
+    def execute(cls, **kwargs) -> str:
+        """Execute the channel-id command to fetch YouTube channel IDs from URLs and save them to a CSV file.
+
+        This command retrieves YouTube channel IDs from one of two possible inputs:
+            - a list of YouTube channel URLs (`--urls`), or
+            - a CSV file containing those URLs (`--urls-file-path`).
+
+            Args:
+                urls (list[str]): List of YouTube channel URLs.
+                    Mutually exclusive with `urls_file_path`.
+                urls_file_path (Path): Path to a CSV file containing YouTube channel URLs.
+                    Mutually exclusive with `urls`.
+                    Requires url_column_name to specify the column with URLs.
+                output_file_path (Path, optional): Path to the output CSV file where channel IDs will be saved.
+                    If not provided, the result will be returned as a string.
+                api_key (str): The API key to authenticate with the YouTube Data API.
+                url_column_name (str, optional): The name of the column in the urls_file_path CSV file that contains the URLs.
+                    Default is "url".
+                id_column_name (str, optional): The name of the column for channel IDs in the output CSV file.
+                    Default is "channel_id".
+
+        Returns:
+            str: A message indicating the result of the command. If output_file_path is specified, the message will
+                 include the path to the generated CSV file. Otherwise, it will return the result as a string.
+
+        Raises:
+            ValueError: If neither `urls` nor `urls_file_path` is provided, or if both are provided at the same time.
+        """
+        urls = kwargs.get("urls") or []
+        urls_file_path = kwargs.get("urls_file_path")
+        output_file_path = kwargs.get("output_file_path")
+        api_key = kwargs.get("api_key")
+
+        url_column_name = kwargs.get("url_column_name")
+        id_column_name = kwargs.get("id_column_name")
+
+        urls = cls.resolve_urls(urls, urls_file_path, url_column_name)
+
+        youtube = YouTube([api_key], disable_ipv6=True)
+
+        channels_ids = [youtube.channel_id_from_url(url) for url in urls if url]
+
+        result = cls.data_to_csv(
+            data=[{(id_column_name or cls.CHANNEL_ID_COLUMN_NAME): channel_id} for channel_id in channels_ids],
+            output_file_path=output_file_path,
+        )
+
+        return result
+
+    @classmethod
+    def resolve_urls(cls, urls, urls_file_path, url_column_name):
+        if urls_file_path:
+            urls += cls.data_from_csv(
+                file_path=Path(urls_file_path), data_column_name=url_column_name or cls.URL_COLUMN_NAME
+            )
+        if not urls:
+            raise Exception("Either 'username' or 'url' must be provided for the channel-id command")
+        return urls