social/dev.py

#!/usr/bin/env python3
import asyncio
import platform
import shlex
import subprocess
import sys
import time
import tomllib
import traceback

from collections.abc import Callable
from datetime import datetime, timedelta
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import TypedDict

try:
	import watchfiles

	from click import Context, echo, group, option, pass_context
	from tomlkit.items import Array, String, StringType, Trivia
	from tomlkit.toml_file import TOMLFile

except ImportError:
	print("Installing missing dependencies...")
	deps = " ".join(["build", "click", "tomlkit", "watchfiles"])
	subprocess.run(shlex.split(f"{sys.executable} -m pip install {deps}"))

	print("Restarting script...")
	subprocess.run([sys.executable, *sys.argv])
	sys.exit()


REPO = Path(__file__).resolve().parent


class WatchfilesOptions(TypedDict):
	watch_filter: Callable[[watchfiles.Change, str], bool]
	recursive: bool
	ignore_permission_denied: bool
	rust_timeout: int


@group("cli")
def cli() -> None:
	...


@cli.command("install")
def cli_install_deps() -> None:
	with open("pyproject.toml", "rb") as fd:
		pyproject = tomllib.load(fd)

	dependencies = pyproject["project"]["dependencies"]
	dependencies.extend(pyproject["project"]["optional-dependencies"]["dev"])
	dependencies.extend(pyproject["project"]["optional-dependencies"]["docs"])
	dependencies = list(dep.replace(" ", "") for dep in dependencies)

	run_python("-m", "pip", "install", "-U", "pip", "setuptools", "wheel")
	run_python("-m", "pip", "install", *dependencies)
	echo("Installed dependencies :3")


@cli.command("lint")
@option("--path", "-p", type = Path, default = REPO.joinpath("barkshark_social"))
@option("--watch", "-w", is_flag = True, help = "Watch for changes to the source")
def cli_lint(path: Path, watch: bool) -> None:
	path = path.expanduser().resolve()

	if watch:
		script = str(Path(__file__).resolve())
		handle_run_watcher(script, "lint", "--path", str(path))
		return

	echo("----- flake8 -----")
	run_python("-m", "flake8", str(path))

	echo("\n----- mypy -----")
	run_python("-m", "mypy", str(path))


@cli.command("build")
def cli_build() -> None:
	cli_update_files.callback() # type: ignore

	with open("pyproject.toml", "rb") as fd:
		pyproject = tomllib.load(fd)
		__version__ = pyproject["project"]["version"]

	with TemporaryDirectory() as tmp:
		arch = "amd64" if sys.maxsize >= 2**32 else "i386"
		run_python(
			"-m", "PyInstaller",
			"--collect-data", "barkshark_social",
			"--hidden-import", "pg8000",
			"--hidden-import", "sqlite3",
			"--name", f"barkshark-social-{__version__}-{platform.system().lower()}-{arch}",
			"--workpath", tmp,
			"--onefile", "barkshark_social/__main__.py",
			"--specpath", tmp,
			"--strip"
		)


@cli.command("build-package")
def cli_build_package() -> None:
	cli_update_files.callback() # type: ignore
	run_python("-m", " build", "--outdir", "dist-pypi")


@cli.command("update-files")
def cli_update_files() -> None:
	project_file = TOMLFile(REPO.joinpath("pyproject.toml"))
	project = project_file.read()
	paths = []

	for path in REPO.joinpath("barkshark_social").rglob("*"):
		if path.is_file() and not path.suffix == ".py":
			strpath = str(path.parent).split("barkshark_social")[1][1:] + "/*"

			if strpath in paths:
				continue

			paths.append(strpath)

	parsed_paths = [String(StringType.SLB, path, path, Trivia()) for path in paths]
	files = Array(parsed_paths, multiline = True, trivia = Trivia(indent = "\t")) # type: ignore
	project["tool"]["setuptools"]["package-data"]["barkshark_social"] = files # type: ignore
	project_file.write(project)


@cli.command("exec", context_settings = {"allow_extra_args": True})
@pass_context
def cli_run(ctx: Context, watch: bool = False) -> None:
	run_python("-m", "barkshark_social", *ctx.args)


def run_python(*arguments: str) -> subprocess.CompletedProcess[bytes]:
	return subprocess.run([sys.executable, *arguments])


def handle_run_watcher(*command: str) -> None:
	asyncio.run(_handle_run_watcher(*command))


async def _handle_run_watcher(*command: str) -> None:
	proc: subprocess.Popen[bytes] = subprocess.Popen([sys.executable, *command])
	last_restart: datetime = datetime.now()

	options: WatchfilesOptions = {
		"watch_filter": lambda _, path: path.endswith(".py"),
		"recursive": True,
		"ignore_permission_denied": True,
		"rust_timeout": 1000
	}

	async for changes in watchfiles.awatch(REPO.joinpath("barkshark_social"), **options):
		if datetime.now() - timedelta(seconds = 3) < last_restart:
			continue

		if proc.poll() is None:
			echo(f"Terminating process {proc.pid}")
			proc.terminate()

			sec = 0.0

			while proc.poll() is None:
				time.sleep(0.1)
				sec += 0.1

				if sec < 5.0:
					continue

				echo("Failed to terminate. Killing process...")
				proc.kill()
				break

			echo("Process terminated")

		proc = subprocess.Popen([sys.executable, *command])
		last_restart = datetime.now()
		echo(f"Started processes with PID: {proc.pid}")


if __name__ == "__main__":
	cli()