JWeBot/tools/sync-local-extensions.py

#!/usr/bin/env python3
"""Generate or verify all reviewed local extension adapter packages."""

from __future__ import annotations

import argparse
import json
import shutil
import sys
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
OFFICIAL_SRC = (
    ROOT
    / "packages"
    / "jwebot-official-extensions"
    / "src"
)
SDK_SRC = ROOT / "packages" / "jwebot-sdk" / "src"
sys.path[:0] = [str(SDK_SRC), str(OFFICIAL_SRC)]

from jwebot_official_extensions import LOCAL_EXTENSIONS  # noqa: E402


LEGACY_EXTENSION_DIRS = (
    ROOT / "extensions" / "official" / "core-help",
    ROOT / "extensions" / "official" / "group-insights",
    ROOT / "extensions" / "official" / "local-operations",
    ROOT / "extensions" / "official" / "random-games",
    ROOT / "extensions" / "official" / "session-games",
    ROOT / "extensions" / "official" / "steam-presence",
)


def module_name(extension_id: str) -> str:
    return extension_id.replace("-", "_")


def manifest_text(descriptor) -> str:
    command_values = ", ".join(
        json.dumps(command, ensure_ascii=False)
        for command in descriptor.commands
    )
    help_values = ", ".join(
        json.dumps(entry, ensure_ascii=False)
        for entry in descriptor.help_entries
    )
    extension_id = descriptor.extension_id
    return f"""id = "{extension_id}"
author = "{descriptor.author}"
description = {json.dumps(descriptor.description, ensure_ascii=False)}
help = [{help_values}]
version = "1.0.0"
api_version = "1"
entrypoint = "{module_name(extension_id)}:plugin"
type = "command"

[capabilities]
commands = [{command_values}]
storage = false
network_profiles = []
service_profiles = ["{extension_id}"]
"""


def module_text(descriptor) -> str:
    extension_id = descriptor.extension_id
    return f'''"""Generated reviewed adapter for {extension_id}."""

from jwebot_sdk import ServiceCommandAdapter, SingleGroupContext


PROFILE = {extension_id!r}
AUTHOR = {descriptor.author!r}
DESCRIPTION = {descriptor.description!r}
HELP_ENTRIES = {descriptor.help_entries!r}
COMMANDS = {descriptor.commands!r}
ADAPTER = ServiceCommandAdapter(
    PROFILE,
    metadata={{"local_extension": PROFILE, "author": AUTHOR}},
)


def plugin(context: SingleGroupContext) -> None:
    ADAPTER.register(context, COMMANDS)
'''


def readme_text(descriptor) -> str:
    command_list = "\n".join(
        f"- `{command}`"
        for command in descriptor.commands
    )
    return f"""# {descriptor.extension_id}

Author: `{descriptor.author}`

{descriptor.description}

Help:

{chr(10).join(f"- `{entry}`" for entry in descriptor.help_entries)}

Registered commands:

{command_list}

The adapter has no direct Ledger, GUI, environment, shell, or network access.
It delegates to the reviewed `{descriptor.extension_id}` Kernel service profile
and returns a structured `ReplyPlan`.
"""


def test_text(descriptor) -> str:
    module = module_name(descriptor.extension_id)
    first_command = descriptor.commands[0]
    return f'''import sys
import unittest
from pathlib import Path
from unittest import mock


SOURCE = Path(__file__).resolve().parents[1] / "src"
sys.path.insert(0, str(SOURCE))

from {module} import AUTHOR, DESCRIPTION, HELP_ENTRIES, plugin


class PluginTests(unittest.TestCase):
    def test_registers_declared_commands_and_author(self):
        context = mock.Mock()
        plugin(context)
        registered = [
            call.args[0]
            for call in context.commands.register.call_args_list
        ]
        self.assertEqual(AUTHOR, {descriptor.author!r})
        self.assertEqual(DESCRIPTION, {descriptor.description!r})
        self.assertEqual(HELP_ENTRIES, {descriptor.help_entries!r})
        self.assertIn({first_command!r}, registered)
        self.assertEqual(len(registered), {len(descriptor.commands)})


if __name__ == "__main__":
    unittest.main()
'''


def collection_readme(author: str) -> str:
    if author == "official":
        description = (
            "Official-owned reviewed extensions. Person memory is the only "
            "official command plugin in the current inventory."
        )
    else:
        description = (
            "Reviewed local extensions maintained by the JWeBot project. Each "
            "independently "
            "understandable function is packaged separately; dice and dice "
            "challenge intentionally remain one complete feature collection."
        )
    return f"""# {author}

{description}

Every child directory is generated from the authoritative local extension
ownership inventory.
"""


def expected_files() -> dict[Path, str]:
    values = {
        ROOT / "extensions" / "official" / "README.md": collection_readme(
            "official"
        ),
        ROOT / "extensions" / "jwebot" / "README.md": collection_readme(
            "jwebot"
        ),
    }
    for descriptor in LOCAL_EXTENSIONS:
        root = (
            ROOT
            / "extensions"
            / descriptor.directory
            / descriptor.extension_id
        )
        module = module_name(descriptor.extension_id)
        values.update(
            {
                root / "plugin.toml": manifest_text(descriptor),
                root / "src" / f"{module}.py": module_text(descriptor),
                (
                    OFFICIAL_SRC
                    / "jwebot_official_extensions"
                    / "adapters"
                    / f"{module}.py"
                ): module_text(descriptor),
                root / "tests" / "test_plugin.py": test_text(descriptor),
                root / "README.md": readme_text(descriptor),
            }
        )
    return values


def stale_paths(expected: set[Path]) -> list[Path]:
    stale = [
        path
        for path in LEGACY_EXTENSION_DIRS
        if path.exists() and path not in expected
    ]
    adapters = (
        OFFICIAL_SRC
        / "jwebot_official_extensions"
        / "adapters"
    )
    stale.extend(
        path
        for path in adapters.glob("*.py")
        if path.name != "__init__.py" and path not in expected
    )
    return sorted(stale)


def synchronize(*, check: bool) -> list[Path]:
    expected = expected_files()
    changed: list[Path] = []
    for path, content in expected.items():
        current = (
            path.read_text(encoding="utf-8")
            if path.is_file()
            else None
        )
        if current == content:
            continue
        changed.append(path)
        if not check:
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text(content, encoding="utf-8")

    stale = stale_paths(set(expected))
    changed.extend(stale)
    if not check:
        for path in stale:
            if path.is_dir():
                shutil.rmtree(path)
            else:
                path.unlink()
    return changed


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--check", action="store_true")
    args = parser.parse_args()
    changed = synchronize(check=args.check)
    if args.check and changed:
        for path in changed:
            print(path.relative_to(ROOT), file=sys.stderr)
        return 1
    for path in changed:
        print(path.relative_to(ROOT))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())