process_counter.py

import json

from system.plugins import PluginBase

PLUGIN_API_VERSION = "2.0.0"


class ProcessCounterPlugin(PluginBase):
    name = "process_counter"
    version = "1.0.0"
    description = "Shows process count and detected work languages"
    author = "vlalikoffc"
    repo_url = "https://github.com/telegrambotrepos/plugins"

    _CONFIG_PATH = "self:/config.json"

    def on_load(self, ctx) -> None:
        if not ctx.fs.exists(self._CONFIG_PATH):
            ctx.fs.write_text(self._CONFIG_PATH, json.dumps(self._default_config(), ensure_ascii=False, indent=2))

    def on_render(self, render_ctx, ctx) -> None:
        default_status = getattr(render_ctx, "default_status", None)
        if default_status is None:
            return
        config = self._load_config(ctx)
        if not config.get("enabled", True):
            return

        count = default_status.process_count
        if count is None:
            return
        line = config.get("line", "?? ?????????: {count}")
        line = self._safe_format(line, {"count": count})
        if line:
            ctx.status.add_line(line)
        languages = list(default_status.work_languages or [])
        if languages:
            lang_line = config.get("languages_line", "?? ?????: {languages}")
            lang_line = self._safe_format(lang_line, {"languages": ", ".join(languages)})
            if lang_line:
                ctx.status.add_line(lang_line)

    def _load_config(self, ctx) -> dict:
        if not ctx.fs.exists(self._CONFIG_PATH):
            cfg = self._default_config()
            ctx.fs.write_text(self._CONFIG_PATH, json.dumps(cfg, ensure_ascii=False, indent=2))
            return cfg
        try:
            raw = ctx.fs.read_text(self._CONFIG_PATH)
            return json.loads(raw)
        except (OSError, json.JSONDecodeError):
            cfg = self._default_config()
            ctx.fs.write_text(self._CONFIG_PATH, json.dumps(cfg, ensure_ascii=False, indent=2))
            return cfg

    @staticmethod
    def _default_config() -> dict:
        return {
            "enabled": False,
            "line": "?? ?????????: {count}",
            "languages_line": "?? ?????: {languages}",
        }

    @staticmethod
    def _safe_format(template: str, values: dict) -> str:
        class SafeDict(dict):
            def __missing__(self, key):
                return "{" + key + "}"

        return str(template).format_map(SafeDict(values)).strip()