summary refs log blame commit diff stats
path: root/PodWeb.py
blob: a48bcf53248857fa494033f75ef3b7e3dfa0a722 (plain) (tree)























































































































































































                                                                                                               
#!/usr/bin/env python3
import os
import click
import yaml
import xml.etree.ElementTree as xmlet

import podcastparser
import urllib
import urllib.request
import pprint

global options
options = {"DEBUG": False, 
           "serverlist": os.path.normpath(os.path.join(os.path.expanduser("~/.config/podweb"), "serverlist")), 
           "podcastpath": ""}

class PodWeb():
    
    def __init__(self, 
                 debug : bool = False, 
                 config : None | str = None, 
                 server_list : None | str = None, 
                 download_location : None | str = None) -> None:
        self.options = options
        self.options.update({"DEBUG": debug})
        self.servers = []
        self.DEFAULT_SERVERLIST_HEADING = '''## You can add podcast xml feeds here.  
## You can also optionally add categories, website url, image urls, and names for the podcasts.
## The order of category, name, and url does not matter.
## Here are some example entries:
##  - category: example category
##    name: example podcast 1
##    url: https://example.com/feed.xml
##    img: https://example.com/image.jpg
##    site: https://example.com
##  - name: example podcast 2
##    url: example.com/feed2.xml
'''

        if self.options["DEBUG"]:
            self.config_path = os.path.abspath(os.path.curdir)
            self.config_filepath = "debug_config.yaml"
            self.options["serverlist"] = os.path.join(self.config_path, "debug_serverlist")
        else:
            self.config_path = os.path.normpath(os.path.expanduser("~/.config/podweb"))
            self.config_filepath = os.path.join(self.config_path, "config.yaml")
        if config:
            self.config_filepath = os.path.normpath(os.path.expanduser(config))
        if server_list:
            self.options["serverlist"] = os.path.normpath(os.path.expanduser(server_list))
        if download_location:
            self.options["podcastpath"] = os.path.normpath(os.path.expanduser(download_location))
        self._load_config()
        self._load_serverlist()

    def _load_config(self) -> None:

        if not os.path.exists(self.config_path): os.makedirs(self.config_path)

        if not os.path.isfile(self.config_filepath):
            with open(self.config_filepath, "w+") as f:
                yaml.dump(self.options, f)
        
        else:
            with open(self.config_filepath, "r+") as f:
                self.options.update(yaml.full_load(f))

    def _update_config(self, changed_option : dict) -> None:
        '''Makes a change to the config file'''
        with open(self.options["serverlist"], "w+") as f:
            config_options = yaml.full_load(f)
            config_options.update(changed_option)
            f.write(config_options)

    def _load_serverlist(self) -> list:
        '''Loads the contents of the serverlist'''
        self._create_serverlist()
        with open(self.options["serverlist"], "r+") as f:
            content = yaml.full_load(f)
        if content:
            self.servers = content

    def _create_serverlist(self) -> None:
        '''Checks if the serverlist does not exist and creates it if not'''
        if not os.path.isfile(self.options["serverlist"]):
            with open(self.options["serverlist"], "w+") as f:
                f.write(self.DEFAULT_SERVERLIST_HEADING)

    def _update_serverlist(self) -> None:
        '''This is destructive and overwrites the current serverlist with the stored serverlist'''
        with open(self.options["serverlist"], "w+") as f:
            f.write(self.DEFAULT_SERVERLIST_HEADING)
        if len(self.servers):
            with open(self.options["serverlist"], "a") as f:
                yaml.dump(self.servers, f)
    
    def add_podcast(self, feedurl : str, name = None, category = None, site = None, img = None) -> None:
        feedparse = urllib.parse.urlparse(feedurl)
        for i in self.servers:
            iparse = urllib.parse.urlparse(i["url"])
            if iparse.hostname == feedparse.hostname and iparse.path == feedparse.path:
                return None
        new_feed = {"url": feedurl}
        if not name or not img or not site:
            parsed = podcastparser.parse(feedurl, urllib.request.urlopen(feedurl))
            if not name:    name = parsed.get("title")
            if not img:     img =  parsed.get("cover_url")
            if not site:    site = parsed.get("link")
        if name:        new_feed.update({"name": name})
        if site:        new_feed.update({"site": site})
        if img:         new_feed.update({"img": img})
        if category:    new_feed.update({"category": category})
        self.servers.append(new_feed)
        self._update_serverlist()

    def import_opml(self, opml_path : str) -> None:
        body = xmlet.parse(source=opml_path).getroot().find("body")
        for child in body:
            i = child.attrib
            if i["type"] == "rss":
                self.add_podcast(feedurl = i["xmlUrl"], 
                                 name = i.get("text"), 
                                 site = i.get("htmlUrl"), 
                                 img = i.get("imageUrl"))
    
    def _parse_rss(self, url : str) -> dict:
        parsed = podcastparser.parse(url, urllib.request.urlopen(url))
        return parsed
    
    def _parse_local_rss(self, file : str) -> dict:
        with open(file, "rb") as f:
            parsed = podcastparser.parse(file, f)
        return parsed

@click.group()
@click.pass_context
@click.option('-d', '--debug', is_flag=True)
@click.option('--config', default=None)
@click.option('--server-list', default=None)
@click.option('--download-location', default=None)
def cli(ctx, 
        debug : bool, 
        config : None | str, 
        server_list : None | str, 
        download_location : None | str):
    """a simple podfetcher for the CLI."""
    ctx.obj = PodWeb(debug = debug, 
                     config = config,
                     server_list = server_list,
                     download_location = download_location)
    ctx.show_default = True

@cli.command()
@click.argument("setting",
                type=click.Choice(['configlocation', 'serverlistlocation', 'downloadlocation', 'servers'], 
                case_sensitive=False))
@click.pass_obj
def get_setting(obj, setting):
    if setting == "configlocation":
        click.echo(obj.config_filepath)
    if setting == 'serverlistlocation':
        click.echo(obj.options["serverlist"])
    if setting == "downloadlocation":
        click.echo(obj.options["podcastpath"])
    if setting == "servers":
        for i in obj.servers:
            name = ""
            if i.get("name"): name = F"{i["name"]} - "
            click.echo(F"{name}{i["url"]}")

@cli.command()
@click.argument("url")
@click.pass_obj
def parse(obj, url):
    click.echo(pprint.pformat(obj._parse_rss(url)))

@cli.command()
@click.argument("filepath", type=click.Path(exists=True))
@click.pass_obj
def parse_file(obj, filepath):
    click.echo(pprint.pformat(obj._parse_local_rss(filepath)))

if __name__ == "__main__":
    cli()