scripts.py

import os
import re
import shutil
import subprocess
import sys
from pathlib import Path


def check_exit_code(command):
    process = subprocess.Popen(
        command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT
    )

    if not process.stdout:
        raise RuntimeError("Process has no STDOUT")

    while True:
        output = process.stdout.readline()
        if output == b"" and process.poll() is not None:
            break
        if output:
            print(str(output.strip(), "utf-8"))
    exit_code = process.poll()
    if exit_code != 0:
        sys.exit(exit_code)


def run_tests():
    check_exit_code("uv run --env-file .env pytest")


def run_formatter():
    check_exit_code("ruff format .")


def run_format_check():
    check_exit_code("ruff format . --check")


def run_linter():
    check_exit_code("ruff check .")


def run_linter_fix():
    check_exit_code("ruff check . --fix")


def generate_docs():
    here = Path(__file__).parent
    input = here / "doc"

    if input.exists():
        shutil.rmtree(input)

    check_exit_code(
        "pdoc \
    --include-undocumented \
    --favicon https://logo.swmansion.com/membrane/\\?width\\=100\\&variant\\=signetDark\
    --logo https://logo.swmansion.com/membrane/\\?width\\=70\\&variant\\=signetDark\
    -t templates/doc \
    -o doc \
    fishjam"
    )
    out = here / "docs" / "api"

    if out.exists():
        shutil.rmtree(out)

    shutil.copytree(input, out)

    # ...and rename the .html files to .md so that mkdocs picks them up!
    for f in out.glob("**/*.html"):
        f.rename(f.with_suffix(".md"))


def clean_mdx_content(content: str) -> str:
    parts = re.split(r"((?:```[\s\S]*?```|`[^`\n]+`))", content)

    # example: convert `fishjam._fishjam_openapi_client.models.peer.Peer` into `Peer`
    internal_path_pattern = r"fishjam\.(?:[\w.]+\.)?_[\w.]+\."

    cleaned_parts = []
    for part in parts:
        if part.startswith("`"):
            text = (
                part
                .replace("&lt;", "<")
                .replace("&gt;", ">")
                .replace("&#39;", "'")
                .replace("builtins.", "")
            )

            text = re.sub(internal_path_pattern, "", text)

            cleaned_parts.append(text)
        else:
            text = part.replace("{", "\\{").replace("}", "\\}").replace("<", "&lt;")

            cleaned_parts.append(text)

    return "".join(cleaned_parts)


def generate_docusaurus():
    here = Path(__file__).parent
    input = here / "doc"
    out = here / "docusaurus"

    if input.exists():
        shutil.rmtree(input)
    if out.exists():
        shutil.rmtree(out)

    check_exit_code(
        "pdoc \
    --include-undocumented \
    -t templates/docusaurus \
    -o doc \
    fishjam"
    )
    try:
        os.remove(input / "index.html")
    except FileNotFoundError:
        pass

    out.mkdir(parents=True, exist_ok=True)

    for f in input.glob("**/*.html"):
        content = f.read_text(encoding="utf-8")
        safe_content = clean_mdx_content(content)
        rel_path = f.relative_to(input)
        dest_path = out / rel_path.parent / rel_path.stem / "index.md"
        dest_path.parent.mkdir(parents=True, exist_ok=True)

        dest_path.write_text(safe_content, encoding="utf-8")


def _spec_argument(name: str) -> str:
    if len(sys.argv) < 2:
        raise RuntimeError(f"Missing {name} url or path positional argument")

    url_or_path = sys.argv[1]
    is_url = url_or_path.startswith("http://") or url_or_path.startswith("https://")
    return f"--url {url_or_path}" if is_url else f"--path {url_or_path}"


def _generate_client(file_arg: str, config: str, output_path: str):
    check_exit_code(
        f"openapi-python-client generate \
            {file_arg} \
            --config {config} \
            --meta=none \
            --overwrite \
            --output-path={output_path} \
            --custom-template-path=templates/openapi"
    )
    _fix_multi_content_type_endpoints(output_path)


def _fix_multi_content_type_endpoints(output_path: str):
    """Work around openapi-python-client 0.28 on endpoints taking several content types.

    An endpoint that accepts more than one request content type is generated with a body
    annotated `Unset` that the module never imports, and with a `multipart/form-data`
    content type set without the boundary httpx computes for the parts it uploads.

    Args:
        output_path: Directory the client was generated into.
    """
    unset_import = re.compile(
        r"^from \.\.\.types import UNSET, Response$", re.MULTILINE
    )
    multipart_header = '        headers["Content-Type"] = "multipart/form-data"\n'
    fixed = []

    for path in Path(output_path).rglob("*.py"):
        source = path.read_text()
        patched = source.replace(f"{multipart_header}\n", "").replace(
            multipart_header, ""
        )

        if "| Unset = UNSET" in patched:
            patched = unset_import.sub(
                "from ...types import UNSET, Response, Unset", patched
            )

        if patched != source:
            path.write_text(patched)
            fixed.append(str(path))

    for path in fixed:
        print(f"Patched {path}")


def update_client():
    _generate_client(
        _spec_argument("fishjam openapi.yaml"),
        "openapi-python-client-config.yaml",
        "fishjam/_fishjam_openapi_client/",
    )


def update_composition_client():
    _generate_client(
        _spec_argument("Composition API openapi.json"),
        "openapi-python-client-composition-config.yaml",
        "fishjam/_composition_openapi_client/",
    )


def start_room_manager():
    current_path = os.getcwd()
    current_folder = os.path.basename(current_path)

    if current_folder != "python-server-sdk":
        raise RuntimeError(
            "Room Manager has to be started from the `python-server-sdk` directory."
        )

    subprocess.run(
        ["python", "-m", "examples.room_manager.main"] + sys.argv[1:], check=False
    )