mirror of https://github.com/craigerl/aprsd.git
243 lines
7.7 KiB
Python
243 lines
7.7 KiB
Python
import fnmatch
|
|
import importlib
|
|
import inspect
|
|
import logging
|
|
import os
|
|
import pkgutil
|
|
import re
|
|
import sys
|
|
from traceback import print_tb
|
|
from urllib.parse import urljoin
|
|
|
|
from bs4 import BeautifulSoup
|
|
import click
|
|
import requests
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
from rich.text import Text
|
|
from thesmuggler import smuggle
|
|
|
|
from aprsd import cli_helper
|
|
from aprsd import plugin as aprsd_plugin
|
|
from aprsd.main import cli
|
|
from aprsd.plugins import (
|
|
email, fortune, location, notify, ping, query, time, version, weather,
|
|
)
|
|
|
|
|
|
LOG = logging.getLogger("APRSD")
|
|
|
|
|
|
def onerror(name):
|
|
print(f"Error importing module {name}")
|
|
type, value, traceback = sys.exc_info()
|
|
print_tb(traceback)
|
|
|
|
|
|
def is_plugin(obj):
|
|
for c in inspect.getmro(obj):
|
|
if issubclass(c, aprsd_plugin.APRSDPluginBase):
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def plugin_type(obj):
|
|
for c in inspect.getmro(obj):
|
|
if issubclass(c, aprsd_plugin.APRSDRegexCommandPluginBase):
|
|
return "RegexCommand"
|
|
if issubclass(c, aprsd_plugin.APRSDWatchListPluginBase):
|
|
return "WatchList"
|
|
if issubclass(c, aprsd_plugin.APRSDPluginBase):
|
|
return "APRSDPluginBase"
|
|
|
|
return "Unknown"
|
|
|
|
|
|
def walk_package(package):
|
|
return pkgutil.walk_packages(
|
|
package.__path__,
|
|
package.__name__ + ".",
|
|
onerror=onerror,
|
|
)
|
|
|
|
|
|
def get_module_info(package_name, module_name, module_path):
|
|
if not os.path.exists(module_path):
|
|
return None
|
|
|
|
dir_path = os.path.realpath(module_path)
|
|
pattern = "*.py"
|
|
|
|
obj_list = []
|
|
|
|
for path, _subdirs, files in os.walk(dir_path):
|
|
for name in files:
|
|
if fnmatch.fnmatch(name, pattern):
|
|
module = smuggle(f"{path}/{name}")
|
|
for mem_name, obj in inspect.getmembers(module):
|
|
if inspect.isclass(obj) and is_plugin(obj):
|
|
obj_list.append(
|
|
{
|
|
"package": package_name,
|
|
"name": mem_name, "obj": obj,
|
|
"version": obj.version,
|
|
"path": f"{'.'.join([module_name, obj.__name__])}",
|
|
},
|
|
)
|
|
|
|
return obj_list
|
|
|
|
|
|
def get_installed_plugins():
|
|
# installed plugins
|
|
ip = {}
|
|
for finder, name, ispkg in pkgutil.iter_modules():
|
|
if name.startswith("aprsd_"):
|
|
if ispkg:
|
|
module = importlib.import_module(name)
|
|
pkgs = walk_package(module)
|
|
for pkg in pkgs:
|
|
pkg_info = get_module_info(module.__name__, pkg.name, module.__path__[0])
|
|
ip[name] = pkg_info
|
|
return ip
|
|
|
|
|
|
def show_built_in_plugins(console):
|
|
modules = [email, fortune, location, notify, ping, query, time, version, weather]
|
|
plugins = []
|
|
|
|
for module in modules:
|
|
entries = inspect.getmembers(module, inspect.isclass)
|
|
for entry in entries:
|
|
cls = entry[1]
|
|
if issubclass(cls, aprsd_plugin.APRSDPluginBase):
|
|
info = {
|
|
"name": cls.__qualname__,
|
|
"path": f"{cls.__module__}.{cls.__qualname__}",
|
|
"version": cls.version,
|
|
"docstring": cls.__doc__,
|
|
"short_desc": cls.short_description,
|
|
}
|
|
|
|
if issubclass(cls, aprsd_plugin.APRSDRegexCommandPluginBase):
|
|
info["command_regex"] = cls.command_regex
|
|
info["type"] = "RegexCommand"
|
|
|
|
if issubclass(cls, aprsd_plugin.APRSDWatchListPluginBase):
|
|
info["type"] = "WatchList"
|
|
|
|
plugins.append(info)
|
|
|
|
plugins = sorted(plugins, key=lambda i: i["name"])
|
|
|
|
table = Table(
|
|
title="[not italic]:snake:[/] [bold][magenta]APRSD Built-in Plugins [not italic]:snake:[/]",
|
|
)
|
|
table.add_column("Plugin Name", style="cyan", no_wrap=True)
|
|
table.add_column("Info", style="bold yellow")
|
|
table.add_column("Type", style="bold green")
|
|
table.add_column("Plugin Path", style="bold blue")
|
|
for entry in plugins:
|
|
table.add_row(entry["name"], entry["short_desc"], entry["type"], entry["path"])
|
|
|
|
console.print(table)
|
|
|
|
|
|
def show_pypi_plugins(installed_plugins, console):
|
|
query = "aprsd"
|
|
api_url = "https://pypi.org/search/"
|
|
snippets = []
|
|
s = requests.Session()
|
|
for page in range(1, 3):
|
|
params = {"q": query, "page": page}
|
|
r = s.get(api_url, params=params)
|
|
soup = BeautifulSoup(r.text, "html.parser")
|
|
snippets += soup.select('a[class*="snippet"]')
|
|
if not hasattr(s, "start_url"):
|
|
s.start_url = r.url.rsplit("&page", maxsplit=1).pop(0)
|
|
|
|
title = Text.assemble(
|
|
("Pypi.org APRSD Installable Plugin Packages\n\n", "bold magenta"),
|
|
("Install any of the following plugins with ", "bold yellow"),
|
|
("'pip install ", "bold white"),
|
|
("<Plugin Package Name>'", "cyan"),
|
|
)
|
|
|
|
table = Table(title=title)
|
|
table.add_column("Plugin Package Name", style="cyan", no_wrap=True)
|
|
table.add_column("Description", style="yellow")
|
|
table.add_column("Version", style="yellow", justify="center")
|
|
table.add_column("Released", style="bold green", justify="center")
|
|
table.add_column("Installed?", style="red", justify="center")
|
|
for snippet in snippets:
|
|
link = urljoin(api_url, snippet.get("href"))
|
|
package = re.sub(r"\s+", " ", snippet.select_one('span[class*="name"]').text.strip())
|
|
version = re.sub(r"\s+", " ", snippet.select_one('span[class*="version"]').text.strip())
|
|
created = re.sub(r"\s+", " ", snippet.select_one('span[class*="created"]').text.strip())
|
|
description = re.sub(r"\s+", " ", snippet.select_one('p[class*="description"]').text.strip())
|
|
emoji = ":open_file_folder:"
|
|
|
|
if "aprsd-" not in package or "-plugin" not in package:
|
|
continue
|
|
|
|
under = package.replace("-", "_")
|
|
if under in installed_plugins:
|
|
installed = "Yes"
|
|
else:
|
|
installed = "No"
|
|
|
|
table.add_row(
|
|
f"[link={link}]{emoji}[/link] {package}",
|
|
description, version, created, installed,
|
|
)
|
|
|
|
console.print("\n")
|
|
console.print(table)
|
|
return
|
|
|
|
|
|
def show_installed_plugins(installed_plugins, console):
|
|
if not installed_plugins:
|
|
return
|
|
|
|
table = Table(
|
|
title="[not italic]:snake:[/] [bold][magenta]APRSD Installed 3rd party Plugins [not italic]:snake:[/]",
|
|
)
|
|
table.add_column("Package Name", style=" bold white", no_wrap=True)
|
|
table.add_column("Plugin Name", style="cyan", no_wrap=True)
|
|
table.add_column("Version", style="yellow", justify="center")
|
|
table.add_column("Type", style="bold green")
|
|
table.add_column("Plugin Path", style="bold blue")
|
|
for name in installed_plugins:
|
|
for plugin in installed_plugins[name]:
|
|
table.add_row(
|
|
name.replace("_", "-"),
|
|
plugin["name"],
|
|
plugin["version"],
|
|
plugin_type(plugin["obj"]),
|
|
plugin["path"],
|
|
)
|
|
|
|
console.print("\n")
|
|
console.print(table)
|
|
|
|
|
|
@cli.command()
|
|
@cli_helper.add_options(cli_helper.common_options)
|
|
@click.pass_context
|
|
@cli_helper.process_standard_options_no_config
|
|
def list_plugins(ctx):
|
|
"""List the built in plugins available to APRSD."""
|
|
console = Console()
|
|
|
|
with console.status("Show Built-in Plugins") as status:
|
|
show_built_in_plugins(console)
|
|
|
|
status.update("Fetching pypi.org plugins")
|
|
installed_plugins = get_installed_plugins()
|
|
show_pypi_plugins(installed_plugins, console)
|
|
|
|
status.update("Looking for installed APRSD plugins")
|
|
show_installed_plugins(installed_plugins, console)
|