dnd/deadsands/site_tools/cli.py

import asyncio
import os
import shlex
import shutil
import subprocess
import sys
import termios
import webbrowser
from collections import defaultdict
from enum import Enum
from pathlib import Path
from time import sleep

import click
import typer
from livereload import Server
from livereload.watcher import INotifyWatcher
from pelican import main as pelican_main
from rolltable.tables import RollTable
from typing_extensions import Annotated

import site_tools as st
from site_tools.content_manager import create
from site_tools.shell.interactive_shell import DMShell

CONFIG = defaultdict(dict)
CONFIG.update(
    {
        "settings_base": st.DEV_SETTINGS_FILE_BASE,
        "settings_publish": st.PUB_SETTINGS_FILE_BASE,
        # Output path. Can be absolute or relative to tasks.py. Default: 'output'
        "deploy_path": st.SETTINGS["OUTPUT_PATH"],
        # Remote server configuration
        "ssh_user": "greg",
        "ssh_host": "froghat.club",
        "ssh_port": "22",
        "ssh_path": "/usr/local/deploy/deadsands/",
        # Host and port for `serve`
        "host": "localhost",
        "port": 8000,
        # content manager config
        "templates_path": "markdown-templates",
        # directory to watch for new assets
        "import_path": "imports",
        # where new asseets will be made available
        "production_host": "deadsands.froghat.club",
        # where to find roll table sources
        "table_sources_path": "sources",
    }
)

app = typer.Typer()

class ContentType(str, Enum):
    post = "post"
    lore = "lore"
    monster = "monster"
    region = "region"
    location = "location"
    page = "page"

class Die(str, Enum):
    d100 = "100"
    d20 = "20"
    d12 = "12"
    d10 = "10"
    d6 = "6"
    d4 = "4"

def pelican_run(cmd: list = [], publish=False) -> None:
    settings = CONFIG["settings_publish" if publish else "settings_base"]
    pelican_main(["-s", settings] + cmd)

@app.command()
def clean() -> None:
    if os.path.isdir(CONFIG["deploy_path"]):
        shutil.rmtree(CONFIG["deploy_path"])
        os.makedirs(CONFIG["deploy_path"])

@app.command()
def build() -> None:
    subprocess.run(shlex.split("git submodule update --remote --merge"))
    pelican_run()

@app.command()
def watch() -> None:
    import_path = Path(CONFIG["import_path"])
    content_path = Path(st.SETTINGS["PATH"])

    def do_import():
        assets = []
        for src in import_path.rglob("*"):
            relpath = src.relative_to(import_path)
            target = content_path / relpath
            if src.is_dir():
                target.mkdir(parents=True, exist_ok=True)
                continue
            if target.exists():
                print(f"{target}: exists; skipping.")
                continue
            print(f"{target}: importing...")
            src.link_to(target)
            subprocess.run(shlex.split(f"git add {target}"))
            uri = target.relative_to("content")
            assets.append(f"https://{CONFIG['production_host']}/{uri}")
            src.unlink()
        if assets:
            publish()
            print("\n\t".join(["\nImported Asset URLS:"] + assets))
            print("\n")
    watcher = INotifyWatcher()
    watcher.watch(import_path, do_import)
    watcher.start(do_import)
    print(f"Watching {import_path}. CTRL+C to exit.")
    while True:
        watcher.examine()
        sleep(5)

@app.command()
def serve() -> None:
    url = "http://{host}:{port}/".format(**CONFIG)

    def cached_build():
        pelican_run(["-ve", "CACHE_CONTENT=true", "LOAD_CONTENT_CACHE=true", "SHOW_DRAFTS=true", f'SITEURL="{url}"'])
    clean()
    cached_build()
    server = Server()
    theme_path = st.SETTINGS["THEME"]
    watched_globs = [
        CONFIG["settings_base"],
        "{}/templates/**/*.html".format(theme_path),
    ]

    content_file_extensions = [".md", ".rst"]
    for extension in content_file_extensions:
        content_glob = "{0}/**/*{1}".format(st.SETTINGS["PATH"], extension)
        watched_globs.append(content_glob)

    static_file_extensions = [".css", ".js"]
    for extension in static_file_extensions:
        static_file_glob = "{0}/static/**/*{1}".format(theme_path, extension)
        watched_globs.append(static_file_glob)

    for glob in watched_globs:
        server.watch(glob, cached_build)

    if st.OPEN_BROWSER_ON_SERVE:
        webbrowser.open(url)

    server.serve(host=CONFIG["host"], port=CONFIG["port"], root=CONFIG["deploy_path"])

@app.command()
def publish() -> None:
    clean()
    pelican_run(publish=True)
    subprocess.run(
        shlex.split(
            'rsync --delete --exclude ".DS_Store" -pthrvz -c '
            '-e "ssh -p {ssh_port}" '
            "{} {ssh_user}@{ssh_host}:{ssh_path}".format(CONFIG["deploy_path"].rstrip("/") + "/", **CONFIG)
        )
    )

@app.command()
def restock(
    source: str = typer.Argument(..., help="The source file for the store."),
    frequency: str = Annotated[str, typer.Option("default", help="use the specified frequency from the source file")],
    die: Die = typer.Option(20, help="The size of the die for which to create a table"),
    template_dir: str = Annotated[
        str,
        typer.Argument(
            CONFIG["templates_path"],
            help="Override the default location for markdown templates.",
        ),
    ],
) -> None:
    rt = RollTable([Path(source).read_text()], frequency=frequency, die=die, hide_rolls=True)
    store = rt.datasources[0].metadata["store"]

    click.edit(
        filename=create(
            content_type="post",
            title=store["title"],
            template_dir=template_dir,
            category="stores",
            template="store",
            extra_context=dict(inventory=rt.as_markdown, **store),
        )
    )

@app.command()
def new(
    content_type: ContentType = typer.Argument(
        ...,
        help="The type of content to create.",
    ),
    title: str = typer.Argument(
        ...,
        help="The title of the content.",
    ),
    category: str = typer.Argument(
        None,
        help='Override the default category; required for "post" content.',
    ),
    template: str = typer.Argument(
        None,
        help="Override the default template for the content_type.",
    ),
    template_dir: str = typer.Argument(
        CONFIG["templates_path"],
        help="Override the default location for markdown templates.",
    ),
) -> None:
    if not category:
        match content_type:
            case "post":
                print("You must specify a category for 'post' content.")
                sys.exit()
            case "monster":
                category = "bestiary"
            case "region":
                category = "locations"
            case "location":
                category = "locations"
            case "page":
                category = "pages"
            case _:
                category = content_type.value
    click.edit(filename=create(content_type.value, title, template_dir, category, template or content_type.value))

def dmsh():
    old_attrs = termios.tcgetattr(sys.stdin)
    try:
        asyncio.run(DMShell(CONFIG).start())
    finally:
        termios.tcsetattr(sys.stdin, termios.TCSANOW, old_attrs)

if __name__ == "__main__":
    app()