128 lines
2.7 KiB
Python
Executable file
128 lines
2.7 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
from collections.abc import Callable
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
try:
|
|
import click
|
|
|
|
from watchdog.observers import Observer
|
|
from watchdog.events import PatternMatchingEventHandler
|
|
|
|
except ImportError:
|
|
CMD = f"{sys.executable} -m pip install click watchdog"
|
|
PROC = subprocess.run(shlex.split(CMD), check = False)
|
|
|
|
if PROC.returncode != 0:
|
|
sys.exit()
|
|
|
|
print("Successfully installed click")
|
|
|
|
import click
|
|
|
|
from watchdog.observers import Observer
|
|
from watchdog.events import PatternMatchingEventHandler
|
|
|
|
|
|
REPO = Path(__file__).resolve().parent
|
|
|
|
|
|
@click.group("cli")
|
|
def cli() -> None:
|
|
pass
|
|
|
|
|
|
@cli.command("clean")
|
|
def cli_clean() -> None:
|
|
for directory in {"build", "dist", "dist-pypi", "activitypub_utils.egg-info"}:
|
|
try:
|
|
shutil.rmtree(directory)
|
|
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
@cli.command("lint")
|
|
@click.option("--directory", "-d", default = "aputils")
|
|
@click.option("--watch", "-w", is_flag = True)
|
|
def cli_lint(directory: str, watch: bool) -> None:
|
|
if not watch:
|
|
click.echo("--- flake8 ---")
|
|
subprocess.run(shlex.split(f"{sys.executable} -m flake8 {directory}"))
|
|
|
|
click.echo("\n--- mypy ---")
|
|
subprocess.run(shlex.split(f"{sys.executable} -m mypy {directory}"))
|
|
|
|
return
|
|
|
|
print('Starting process watcher')
|
|
|
|
handler = WatchHandler(cli_lint.callback, directory, False)
|
|
handler.run_callback()
|
|
|
|
watcher = Observer()
|
|
watcher.schedule(handler, str(REPO.joinpath("aputils")), recursive=True)
|
|
watcher.start()
|
|
|
|
try:
|
|
while True:
|
|
time.sleep(0.1)
|
|
|
|
except KeyboardInterrupt:
|
|
pass
|
|
|
|
watcher.stop()
|
|
watcher.join()
|
|
|
|
|
|
@cli.command("test")
|
|
def cli_text() -> None:
|
|
subprocess.run([sys.executable, "-m", "tests"])
|
|
|
|
|
|
@cli.command("install")
|
|
def cli_install() -> None:
|
|
subprocess.run(shlex.split(f"{sys.executable} -m pip install -e .[dev,doc]"), check = False)
|
|
|
|
|
|
class WatchHandler(PatternMatchingEventHandler):
|
|
patterns = ['*.py']
|
|
|
|
|
|
def __init__(self, callback: Callable, *args: Any, **kwargs: Any):
|
|
PatternMatchingEventHandler.__init__(self)
|
|
|
|
self.callback: Callable = callback
|
|
self.args: tuple[Any] = args
|
|
self.kwargs: dict[str, Any] = kwargs
|
|
self.last_restart: datetime | None = None
|
|
|
|
|
|
def run_callback(self, restart=False):
|
|
timestamp = datetime.timestamp(datetime.now())
|
|
self.last_restart = timestamp if not self.last_restart else 0
|
|
|
|
if restart:
|
|
if timestamp - 3 < self.last_restart:
|
|
return
|
|
|
|
self.callback(*self.args, **self.kwargs)
|
|
|
|
|
|
def on_any_event(self, event):
|
|
if event.event_type not in ['modified', 'created', 'deleted']:
|
|
return
|
|
|
|
self.run_callback(restart = True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cli()
|