cli/ytrssil/cli.py

from collections.abc import Iterator
from os import execv, fork
from subprocess import PIPE, Popen
from sys import argv, stderr

from inject import autoparams

from ytrssil.bindings import setup_dependencies
from ytrssil.config import Configuration
from ytrssil.datatypes import Video
from ytrssil.protocols import Client


def user_query(videos: list[Video], reverse: bool = False) -> list[str]:
    p = Popen(
        ['fzf', '-m'],
        stdout=PIPE,
        stdin=PIPE,
    )
    video_list: Iterator[Video]
    if reverse:
        video_list = reversed(videos)
    else:
        video_list = iter(videos)

    input_bytes = '\n'.join(map(str, video_list)).encode('UTF-8')
    stdout, _ = p.communicate(input=input_bytes)
    videos_str: list[str] = stdout.decode('UTF-8').strip().split('\n')
    ret: list[str] = []
    for video_str in videos_str:
        *_, video_id = video_str.split(' - ')

        try:
            ret.append(video_id)
        except KeyError:
            pass

    return ret


@autoparams()
def fetch_new_videos(client: Client) -> int:
    client.fetch()
    return 0


@autoparams()
def register(client: Client) -> int:
    client.register()
    return 0


@autoparams()
def watch_videos(config: Configuration, client: Client) -> int:
    videos = client.get_new_videos()
    if not videos:
        print('No new videos', file=stderr)
        return 1

    selected_videos = user_query(videos)
    if not selected_videos:
        print('No video selected', file=stderr)
        return 2

    video_urls = [
        f'https://www.youtube.com/watch?v={video_id}'
        for video_id in selected_videos
    ]
    cmd = ['/usr/bin/mpv', *config.mpv_options, *video_urls]
    if (fork() == 0):
        execv(cmd[0], cmd)

    for video_id in selected_videos:
        client.mark_video_as_watched(video_id)

    return 0


@autoparams()
def print_url(client: Client) -> int:
    videos = client.get_new_videos()
    if not videos:
        print('No new videos', file=stderr)
        return 1

    selected_videos = user_query(videos)
    if not selected_videos:
        print('No video selected', file=stderr)
        return 2

    for video_id in selected_videos:
        client.mark_video_as_watched(video_id)
        print(f'https://www.youtube.com/watch?v={video_id}')

    return 0


@autoparams()
def mark_as_watched(client: Client) -> int:
    videos = client.get_new_videos()
    if not videos:
        print('No new videos', file=stderr)
        return 1

    selected_videos = user_query(videos)
    if not selected_videos:
        print('No video selected', file=stderr)
        return 2

    for video_id in selected_videos:
        client.mark_video_as_watched(video_id)

    return 0


@autoparams()
def watch_history(config: Configuration, client: Client) -> int:
    videos = client.get_watched_videos()
    if not videos:
        print('No new videos', file=stderr)
        return 1

    selected_videos = user_query(videos)
    if not selected_videos:
        print('No video selected', file=stderr)
        return 2

    video_urls = [
        f'https://www.youtube.com/watch?v={video_id}'
        for video_id in selected_videos
    ]
    cmd = ['/usr/bin/mpv', *config.mpv_options, *video_urls]
    if (fork() == 0):
        execv(cmd[0], cmd)

    return 0


@autoparams()
def mark_as_unwatched(client: Client) -> int:
    videos = client.get_watched_videos()
    if not videos:
        print('No new videos', file=stderr)
        return 1

    selected_videos = user_query(videos)
    if not selected_videos:
        print('No video selected', file=stderr)
        return 2

    for video_id in selected_videos:
        client.mark_video_as_unwatched(video_id)

    return 0


def main(args: list[str] = argv) -> int:
    setup_dependencies()
    command: str
    try:
        command = args[1]
    except IndexError:
        command = 'watch'

    if command == 'fetch':
        return fetch_new_videos()
    elif command == 'register':
        return register()
    elif command == 'watch':
        return watch_videos()
    elif command == 'print':
        return print_url()
    elif command == 'history':
        return watch_history()
    elif command == 'mark':
        return mark_as_watched()
    elif command == 'unmark':
        return mark_as_unwatched()
    else:
        print(f'Unknown command "{command}"', file=stderr)
        print(
            'Available commands: fetch, watch, print, history, mark, unmark',
            file=stderr,
        )
        return 1

    return 0