themes v0.1.1 [extra]

"""themes.engine — файловая логика менеджера тем (без ctx).

Правила безопасности (как в modsafe.installer):
  id темы — ^[a-z0-9_-]{1,32}$; имена файлов — без путей и ..;
  архив — .zip/.tar.gz до 1MB, theme.json в корне обязателен;
  дефолтную тему удалить нельзя; установка — только новый id.
Черновик: <id>-draft (валидный id, виден в списке с бейджем).
"""
import io
import json
import logging
import os
import re
import shutil
import tarfile
import zipfile

logger = logging.getLogger(__name__)

_ID_RX = re.compile(r"^[a-z0-9_-]{1,32}$")
_FILE_RX = re.compile(r"^[A-Za-z0-9._-]{1,64}$")
_VAR_RX = re.compile(r"(--[A-Za-z0-9_-]{1,64})\s*:\s*([^;{}]{1,500});")
_MAX_ARCHIVE = 1024 * 1024


def _check_id(name):
    if not _ID_RX.match(name or ""):
        raise ValueError("недопустимый id темы: %s" % name)
    return name


def _theme_dir(themes_dir, name):
    _check_id(name)
    return os.path.join(themes_dir, name)


def _manifest(themes_dir, name):
    fp = os.path.join(_theme_dir(themes_dir, name), "theme.json")
    try:
        with open(fp, "r", encoding="utf-8") as f:
            m = json.load(f)
    except (OSError, ValueError):
        raise ValueError("темы %s нет или битый theme.json" % name)
    if not isinstance(m, dict) or m.get("id") != name:
        raise ValueError("theme.json %s не соответствует id" % name)
    return m


def _default_id(themes_dir):
    try:
        with open(os.path.join(themes_dir, "default.txt"),
                  "r", encoding="utf-8") as f:
            tid = (f.read() or "").strip().split()[0]
        if _ID_RX.match(tid) and os.path.isfile(
                os.path.join(themes_dir, tid, "theme.json")):
            return tid
    except (OSError, IndexError):
        pass
    return ""


def _is_draft(name):
    return name.endswith("-draft")


def list_themes(themes_dir):
    """[{id, name, description, version, draft, is_default, files}]."""
    out = []
    try:
        entries = sorted(os.listdir(themes_dir))
    except OSError:
        return out
    default = _default_id(themes_dir)
    for tid in entries:
        if not _ID_RX.match(tid):
            continue
        try:
            m = _manifest(themes_dir, tid)
        except ValueError:
            continue
        try:
            files = sorted(os.listdir(os.path.join(themes_dir, tid)))
        except OSError:
            files = []
        out.append({"id": tid, "name": str(m.get("name") or tid),
                    "description": str(m.get("description") or ""),
                    "version": str(m.get("version") or "1"),
                    "draft": _is_draft(tid),
                    "is_default": tid == default,
                    "files": files})
    return out


def set_default(themes_dir, theme_id):
    """Записать default.txt (черновик — нельзя)."""
    _check_id(theme_id)
    if _is_draft(theme_id):
        raise ValueError("черновик нельзя сделать дефолтом")
    _manifest(themes_dir, theme_id)
    fp = os.path.join(themes_dir, "default.txt")
    tmp = fp + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        f.write(theme_id + "\n")
    os.replace(tmp, fp)
    return {"ok": True}


def _read_stage(stage_dir):
    items = []
    try:
        entries = sorted(os.listdir(stage_dir))
    except OSError:
        return items
    for name in entries:
        fp = os.path.join(stage_dir, name)
        if os.path.isfile(fp + ".json"):
            try:
                with open(fp + ".json", "r", encoding="utf-8") as f:
                    items.append(json.load(f))
            except (OSError, ValueError):
                pass
    return items


def staged(stage_dir):
    """Что лежит в стейджинге."""
    return _read_stage(stage_dir)


def _safe_members(names):
    """Проверка имён в архиве: без абсолютных, .. и мусора."""
    clean = []
    for n in names:
        n = (n or "").replace("\\", "/").lstrip("/")
        if not n or n.startswith(".") or "/../" in n or n.startswith("../"):
            raise ValueError("подозрительный путь в архиве: %s" % n)
        parts = [p for p in n.split("/") if p not in ("", ".")]
        if not parts or any(p == ".." for p in parts):
            raise ValueError("подозрительный путь в архиве: %s" % n)
        clean.append("/".join(parts))
    return clean


def _validate_tree(files):
    """files: {relpath: bytes} — проверка набора темы."""
    if "theme.json" not in files:
        raise ValueError("в архиве нет theme.json в корне")
    try:
        m = json.loads(files["theme.json"].decode("utf-8"))
    except (ValueError, UnicodeDecodeError):
        raise ValueError("битый theme.json")
    tid = m.get("id") or ""
    if not _ID_RX.match(tid):
        raise ValueError("плохой id в theme.json: %s" % tid)
    if _is_draft(tid):
        raise ValueError("id не должен кончаться на -draft")
    if "style.css" not in files:
        raise ValueError("в архиве нет style.css")
    for rel, blob in files.items():
        if len(blob) > _MAX_ARCHIVE:
            raise ValueError("файл %s больше 1MB" % rel)
        if not _FILE_RX.match(os.path.basename(rel)):
            raise ValueError("плохое имя файла: %s" % rel)
    return tid, m


def stage(stage_dir, filename, content_type, body):
    """Принять архив в стейджинг. Возвращает {ok, name, version, files}."""
    fn = (filename or "").lower()
    if len(body or b"") > _MAX_ARCHIVE * 4:
        raise ValueError("архив больше 4MB")
    if len(body or b"") == 0:
        raise ValueError("пустой файл")
    files = {}
    try:
        if fn.endswith(".zip"):
            with zipfile.ZipFile(io.BytesIO(body)) as z:
                for info in z.infolist():
                    if info.is_dir():
                        continue
                    n = _safe_members([info.filename])[0]
                    files[n] = z.read(info.filename)
        elif fn.endswith((".tar.gz", ".tgz", ".tar")):
            with tarfile.open(fileobj=io.BytesIO(body), mode="r:*") as t:
                for m in t.getmembers():
                    if not m.isfile():
                        continue
                    n = _safe_members([m.name])[0]
                    f = t.extractfile(m)
                    files[n] = f.read() if f else b""
        else:
            raise ValueError("только .zip / .tar.gz")
    except (zipfile.BadZipFile, tarfile.TarError) as e:
        raise ValueError("битый архив: %s" % e)
    tid, m = _validate_tree(files)
    os.makedirs(stage_dir, mode=0o700, exist_ok=True)
    fp = os.path.join(stage_dir, tid + ".json")
    with open(fp, "w", encoding="utf-8") as f:
        json.dump({"name": tid, "version": str(m.get("version") or "1"),
                   "files": sorted(files.keys()),
                   "payload": {k: v.decode("latin1") for k, v in files.items()}},
                  f, ensure_ascii=False)
    os.chmod(fp, 0o600)
    return {"ok": True, "name": tid,
            "version": str(m.get("version") or "1"),
            "files": sorted(files.keys())}


def install(themes_dir, stage_dir, name):
    """Установить из стейджинга (только новый id)."""
    _check_id(name)
    fp = os.path.join(stage_dir, name + ".json")
    if not os.path.isfile(fp):
        raise ValueError("в стейджинге нет %s" % name)
    with open(fp, "r", encoding="utf-8") as f:
        item = json.load(f)
    dest = os.path.join(themes_dir, name)
    if os.path.exists(dest):
        raise ValueError("тема %s уже есть (правьте через черновик)" % name)
    os.makedirs(dest, mode=0o755)
    try:
        for rel, txt in (item.get("payload") or {}).items():
            p = os.path.realpath(os.path.join(dest, rel))
            if not p.startswith(os.path.realpath(dest) + os.sep):
                raise ValueError("выход за каталог: %s" % rel)
            os.makedirs(os.path.dirname(p), exist_ok=True)
            with open(p, "wb") as f:
                f.write(txt.encode("latin1"))
    except Exception:
        shutil.rmtree(dest, ignore_errors=True)
        raise
    os.remove(fp)
    logger.info("themes: установлена %s", name)
    return {"ok": True, "name": name}


def pack(themes_dir, name):
    """Скачать тему zip (b64)."""
    d = _theme_dir(themes_dir, name)
    _manifest(themes_dir, name)
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as z:
        for root, _dirs, files in os.walk(d):
            for fn in sorted(files):
                p = os.path.join(root, fn)
                z.write(p, os.path.relpath(p, d))
    import base64 as _b64
    return {"ok": True, "filename": "%s.zip" % name,
            "b64": _b64.b64encode(buf.getvalue()).decode("ascii")}


def delete(themes_dir, name):
    """Удалить тему (кроме дефолтной)."""
    _check_id(name)
    if name == _default_id(themes_dir):
        raise ValueError("дефолтную тему удалять нельзя")
    d = _theme_dir(themes_dir, name)
    if not os.path.isdir(d):
        raise ValueError("темы %s нет" % name)
    shutil.rmtree(d)
    logger.info("themes: удалена %s", name)
    return {"ok": True}


def _css_path(themes_dir, name):
    _manifest(themes_dir, name)
    for cand in ("style.css",):
        p = os.path.join(_theme_dir(themes_dir, name), cand)
        if os.path.isfile(p):
            return p
    raise ValueError("style.css не найден")


def get_vars(themes_dir, name):
    """CSS-переменные style.css: [{name, value}]."""
    with open(_css_path(themes_dir, name), "r", encoding="utf-8",
              errors="replace") as f:
        css = f.read()
    out, seen = [], set()
    for m in _VAR_RX.finditer(css):
        if m.group(1) not in seen:
            seen.add(m.group(1))
            out.append({"name": m.group(1), "value": m.group(2).strip()})
    return out


def set_vars(themes_dir, name, values):
    """Сохранить CSS-переменные (имена строгие, значения без ;{})."""
    values = values or {}
    clean = {}
    for k, v in values.items():
        if not re.fullmatch(r"--[A-Za-z0-9_-]{1,64}", k or ""):
            raise ValueError("плохое имя: %s" % k)
        v = str(v or "").strip()
        if not v or len(v) > 500 or any(c in v for c in ";{}"):
            raise ValueError("плохое значение %s" % k)
        clean[k] = v
    if not clean:
        raise ValueError("нечего сохранять")
    p = _css_path(themes_dir, name)
    with open(p, "r", encoding="utf-8", errors="replace") as f:
        css = f.read()

    def _sub(m):
        if m.group(1) in clean:
            return "%s: %s;" % (m.group(1), clean[m.group(1)])
        return m.group(0)

    new, n = _VAR_RX.subn(_sub, css)
    if not n:
        raise ValueError("переменные не найдены в style.css")
    with open(p, "w", encoding="utf-8") as f:
        f.write(new)
    logger.info("themes: %s vars %d", name, n)
    return {"ok": True, "changed": sorted(clean)}


def draft_create(themes_dir, theme_id):
    """Копия <id>-draft для смелых правок."""
    _manifest(themes_dir, theme_id)
    if _is_draft(theme_id):
        raise ValueError("это уже черновик")
    draft = theme_id + "-draft"
    if len(draft) > 32:
        raise ValueError("id слишком длинный для черновика")
    dest = os.path.join(themes_dir, draft)
    if os.path.exists(dest):
        raise ValueError("черновик уже есть")
    shutil.copytree(os.path.join(themes_dir, theme_id), dest)
    mfp = os.path.join(dest, "theme.json")
    try:
        with open(mfp, "r", encoding="utf-8") as f:
            m = json.load(f)
        m["id"] = draft
        m["name"] = str(m.get("name") or theme_id) + " (черновик)"
        with open(mfp, "w", encoding="utf-8") as f:
            json.dump(m, f, ensure_ascii=False, indent=2)
    except (OSError, ValueError) as e:
        shutil.rmtree(dest, ignore_errors=True)
        raise ValueError("черновик не создан: %s" % e)
    return {"ok": True, "draft": draft}


def draft_publish(themes_dir, theme_id):
    """Файлы черновика -> основная тема (theme.json id не трогаем)."""
    if _is_draft(theme_id):
        raise ValueError("публикуют НЕ черновик, а основу")
    src = os.path.join(themes_dir, theme_id + "-draft")
    dest = _theme_dir(themes_dir, theme_id)
    if not os.path.isdir(src):
        raise ValueError("черновика нет")
    _manifest(themes_dir, theme_id)
    for root, _dirs, files in os.walk(src):
        for fn in files:
            if fn == "theme.json":
                continue
            s = os.path.join(root, fn)
            rel = os.path.relpath(s, src)
            if not _FILE_RX.match(os.path.basename(rel)):
                continue
            t = os.path.realpath(os.path.join(dest, rel))
            if not t.startswith(os.path.realpath(dest) + os.sep):
                continue
            shutil.copy2(s, t)
    # версию бампаем minor, чтобы сбросить кеш
    mfp = os.path.join(dest, "theme.json")
    try:
        with open(mfp, "r", encoding="utf-8") as f:
            m = json.load(f)
        ver = str(m.get("version") or "1")
        parts = ver.split(".")
        try:
            parts[-1] = str(int(parts[-1]) + 1)
        except ValueError:
            parts.append("1")
        m["version"] = ".".join(parts)
        with open(mfp, "w", encoding="utf-8") as f:
            json.dump(m, f, ensure_ascii=False, indent=2)
    except (OSError, ValueError):
        pass
    logger.info("themes: %s опубликован из черновика", theme_id)
    return {"ok": True}


def draft_drop(themes_dir, theme_id):
    """Удалить черновик."""
    if _is_draft(theme_id):
        raise ValueError("укажите основу, не черновик")
    dest = os.path.join(themes_dir, theme_id + "-draft")
    if not os.path.isdir(dest):
        raise ValueError("черновика нет")
    shutil.rmtree(dest)
    return {"ok": True}

"""themes.Module — менеджер тем оформления (static/themes/<id>/).

Полный цикл как у modsafe, но для CSS: список, загрузка в стейджинг
с валидацией, установка, упаковка, удаление, редактор CSS-переменных
(безопасный уровень), черновики (<id>-draft) для смелых правок,
тема по умолчанию (default.txt). После мутаций — rescan modsafe.
"""
import logging
import os

from core.base_module import BaseModule

logger = logging.getLogger(__name__)


class Module(BaseModule):
    def setup(self, ctx) -> None:
        mod_dir = os.path.dirname(os.path.abspath(__file__))
        self.config = ctx.module_config("themes", mod_dir)
        self.ctx = ctx
        from . import engine as _eng
        self._eng = _eng
        api = ctx.api.register
        api("themes", "list", self.api_list)
        api("themes", "set_default", self.api_set_default)
        api("themes", "stage", self.api_stage)
        api("themes", "staged", self.api_staged)
        api("themes", "install", self.api_install)
        api("themes", "pack", self.api_pack)
        api("themes", "delete", self.api_delete)
        api("themes", "vars", self.api_vars)
        api("themes", "set_vars", self.api_set_vars)
        api("themes", "draft_create", self.api_draft_create)
        api("themes", "draft_publish", self.api_draft_publish)
        api("themes", "draft_drop", self.api_draft_drop)

    def _dirs(self):
        tdir = os.path.join(
            getattr(self.ctx, "transports_dir", ""), "transport_web",
            "static", "themes")
        sdir = os.path.join(self.ctx.data_dir, "themes_stage")
        return tdir, sdir

    def _audit_rec(self, action, target, actor='', detail=''):
        try:
            self.ctx.api.call('audit.record', actor or '?', action,
                              target, detail, True)
        except Exception:  # noqa: BLE001
            pass

    def _rescan(self):
        try:
            self.ctx.api.call("modsafe.rescan", "transport_web")
        except Exception:  # noqa: BLE001
            pass

    def api_list(self):
        """Все темы с флагами (право themes_manage)."""
        tdir, _sdir = self._dirs()
        return self._eng.list_themes(tdir)

    def api_set_default(self, theme_id, actor=''):
        """Тема по умолчанию (право themes_manage)."""
        tdir, _sdir = self._dirs()
        self._eng.set_default(tdir, theme_id)
        self._audit_rec('themes.default', theme_id, actor)
        return {"ok": True}

    def api_stage(self, filename, content_type, body):
        """Загрузка архива в стейджинг (право themes_manage)."""
        _tdir, sdir = self._dirs()
        return self._eng.stage(sdir, filename, content_type, body)

    def api_staged(self):
        """Стейджинг (право themes_manage)."""
        _tdir, sdir = self._dirs()
        return self._eng.staged(sdir)

    def api_install(self, name, actor=''):
        """Установка из стейджинга (право themes_manage)."""
        tdir, sdir = self._dirs()
        out = self._eng.install(tdir, sdir, name)
        self._audit_rec('themes.install', name, actor)
        self._rescan()
        return out

    def api_pack(self, name):
        """Скачать тему zip (право themes_manage)."""
        tdir, _sdir = self._dirs()
        return self._eng.pack(tdir, name)

    def api_delete(self, name, actor=''):
        """Удалить тему (право themes_manage)."""
        tdir, _sdir = self._dirs()
        out = self._eng.delete(tdir, name)
        self._audit_rec('themes.delete', name, actor)
        self._rescan()
        return out

    def api_vars(self, name):
        """CSS-переменные темы (право themes_manage)."""
        tdir, _sdir = self._dirs()
        return self._eng.get_vars(tdir, name)

    def api_set_vars(self, name, values, actor=''):
        """Сохранить CSS-переменные (право themes_manage)."""
        tdir, _sdir = self._dirs()
        out = self._eng.set_vars(tdir, name, values or {})
        self._audit_rec('themes.vars', name, actor)
        self._rescan()
        return out

    def api_draft_create(self, theme_id, actor=''):
        """Черновик-копия (право themes_manage)."""
        tdir, _sdir = self._dirs()
        out = self._eng.draft_create(tdir, theme_id)
        self._audit_rec('themes.draft', theme_id, actor)
        self._rescan()
        return out

    def api_draft_publish(self, theme_id, actor=''):
        """Опубликовать черновик (право themes_manage)."""
        tdir, _sdir = self._dirs()
        out = self._eng.draft_publish(tdir, theme_id)
        self._audit_rec('themes.publish', theme_id, actor)
        self._rescan()
        return out

    def api_draft_drop(self, theme_id, actor=''):
        """Удалить черновик (право themes_manage)."""
        tdir, _sdir = self._dirs()
        out = self._eng.draft_drop(tdir, theme_id)
        self._audit_rec('themes.draft_drop', theme_id, actor)
        self._rescan()
        return out

    def health(self) -> dict:
        try:
            tdir, sdir = self._dirs()
            n = len(self._eng.list_themes(tdir))
            staged = len(self._eng.staged(sdir))
            return {"ok": True, "module": self.name,
                    "themes": n, "staged": staged}
        except Exception as e:  # noqa: BLE001
            return {"ok": False, "module": self.name, "error": str(e)}

"""themes.ui_web — кабинет «Темы»."""

ROUTES = [
    ("GET", "/api/themes/manage", "manage", {"right": "themes_manage",
     "desc": "Темы: список (?staged=1 — стейджинг)"}),
    ("POST", "/api/themes/manage", "manage", {"right": "themes_manage",
     "desc": "Темы: set_default{id}|install{name}|delete{name}|vars{name}|set_vars{name, values}|pack{name}|draft_create{id}|..."}),
    ("POST", "/api/themes/upload", "upload", {"right": "themes_manage",
     "desc": "Загрузить архив темы (multipart file=*.zip)"}),
]

SERVICE = [
    {"key": "themes_manage", "icon": "🎨", "title": "Темы оформления", "right": "themes_manage",
     "phase": 1, "desc": "Установка, переменные, черновики, дефолт"},
]

PANELS = []


def handle_api(ctx, config, method, req):
    from core.errors import UserError

    if method == "manage":
        if req.get("http_method") == "GET":
            query = req.get("query") or {}
            if (query.get("staged") or "") == "1":
                return {"ok": True, "staged": ctx.api.call("themes.staged")}
            return {"ok": True, "themes": ctx.api.call("themes.list")}
        body = req.get("body") or {}
        action = (body.get("action") or "").strip()
        mapping = {
            "set_default": ("themes.set_default", ["id"]),
            "install": ("themes.install", ["name"]),
            "delete": ("themes.delete", ["name"]),
            "vars": ("themes.vars", ["name"]),
            "set_vars": ("themes.set_vars", ["name", "values"]),
            "pack": ("themes.pack", ["name"]),
            "draft_create": ("themes.draft_create", ["id"]),
            "draft_publish": ("themes.draft_publish", ["id"]),
            "draft_drop": ("themes.draft_drop", ["id"]),
        }
        if action not in mapping:
            raise UserError(
                "action: set_default|install|delete|vars|set_vars|pack|"
                "draft_create|draft_publish|draft_drop")
        api_name, keys = mapping[action]
        try:
            sess = req.get('session') or {}
            actor = 'web:%s' % sess.get('uid') if sess.get('uid') else '?'
            args = [body.get(k) for k in keys]
            if action in ('set_default', 'install', 'delete', 'set_vars',
                          'draft_create', 'draft_publish', 'draft_drop'):
                args.append(actor)
            if action == "set_vars" and not isinstance(args[1], dict):
                raise UserError("values должен быть объектом")
            out = ctx.api.call(api_name, *args)
            if action == "vars":
                return {"ok": True, "vars": out}
            return out
        except (ValueError, RuntimeError) as e:
            return 400, {"ok": False, "error": str(e)}
    if method == "upload":
        files = req.get("files") or {}
        f = files.get("file")
        if isinstance(f, list):
            f = f[0] if f else None
        if not f or not isinstance(f, dict):
            return 400, {"ok": False, "error": "нужен multipart file=*.zip"}
        try:
            return ctx.api.call("themes.stage", f.get("filename") or "",
                                f.get("content_type") or "",
                                f.get("body") or b"")
        except (ValueError, RuntimeError) as e:
            return 400, {"ok": False, "error": str(e)}

    raise UserError("неизвестный метод: %s" % method)