homepage v0.6.0 [extra]

"""homepage.engine — хранение блоков домашней страницы (JSON-файл).

Видимость (как в мониторинге): "public" (все, даже без входа) |
"auth" (вошедшие в веб) | "right:<право>[;<право>...]" (любое из них).
Блок задаёт дефолт, ссылка может переопределить (пусто = наследовать).
Дефолт своих блоков/ссылок — "public" (совместимость: GET был публичным).
"""
import json
import logging
import os
import threading

logger = logging.getLogger(__name__)

_lock = threading.Lock()
_file = ""


def configure(path: str) -> None:
    global _file
    _file = path


def load():
    try:
        with open(_file, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return {"blocks": []}


def save(data) -> bool:
    with _lock:
        try:
            os.makedirs(os.path.dirname(_file) or ".", exist_ok=True)
            tmp = _file + ".tmp"
            with open(tmp, "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            os.replace(tmp, _file)
            return True
        except OSError as e:
            logger.warning("Не удалось сохранить homepage: %s", e)
            return False


DEFAULT_VISIBILITY = "public"

MAX_FOLDERS_PER_BLOCK = 50
MAX_EXT_STATE_KEYS = 200
MAX_EXT_STATE_KEY_LEN = 200
EXTERNAL_MAX_LINKS = 200


def ext_key(module, contract) -> str:
    """Стабильный ключ внешнего блока: модуль + контрактное (не
    переименованное) название. Переименование меняет только показ."""
    return "%s / %s" % (module or "?", contract or "?")


def is_folder(item) -> bool:
    """Папка ссылок: элемент links с type=folder и своим links[]."""
    return isinstance(item, dict) and item.get("type") == "folder"


def valid_visibility(value) -> bool:
    """public | auth | right:имя[;имя2] (≤5, как в monitoring)."""
    if value in ("public", "auth"):
        return True
    if isinstance(value, str) and value.startswith("right:"):
        import re as _re
        names = [n.strip() for n in value[6:].split(";")]
        if not 1 <= len(names) <= 5:
            return False
        return all(_re.match(r"^[A-Za-z0-9_.]+$", n) for n in names)
    return False


def normalize_visibility(value, default=DEFAULT_VISIBILITY):
    """Пусто/мусор -> default; корректное -> как есть (для POST)."""
    if not value:
        return default
    if isinstance(value, str) and valid_visibility(value.strip()):
        return value.strip()
    return default


def visible(visibility, sess, has_right) -> bool:
    """Виден ли элемент запросившему (sess — None для анонима)."""
    vis = visibility or DEFAULT_VISIBILITY
    if vis == "public":
        return True
    if not sess:
        return False
    if vis == "auth":
        return True
    if isinstance(vis, str) and vis.startswith("right:"):
        for name in vis[6:].split(";"):
            name = (name or "").strip()
            if not name:
                continue
            try:
                if has_right(sess, name):
                    return True
            except Exception:  # noqa: BLE001
                continue
        return False
    return False


def _visible_links(links, inherited, sess, has_right):
    """Ссылки, видимые запросившему (папки пропускаются — их разбирает
    filter_blocks отдельно)."""
    out = []
    for l in links or []:
        if not isinstance(l, dict) or is_folder(l):
            continue
        if visible(normalize_visibility(l.get("visibility"), inherited),
                   sess, has_right):
            out.append(l)
    return out


def filter_blocks(blocks, sess, has_right):
    """Блоки+ссылки+папки, видимые запросившему. Ссылка наследует блок,
    папка — тоже (своя видимость перекрывает), ссылки в папке наследуют
    папку. Пустая после фильтра папка остаётся (как блок с пустыми
    ссылками) — иначе редактор без прав потёр бы её при сохранении."""
    out = []
    for b in blocks or []:
        if not isinstance(b, dict):
            continue
        bvis = normalize_visibility(b.get("visibility"))
        if not visible(bvis, sess, has_right):
            continue
        nb = dict(b)
        nb["links"] = []
        for item in b.get("links") or []:
            if not isinstance(item, dict):
                continue
            if not is_folder(item):
                if visible(normalize_visibility(
                        item.get("visibility"), bvis), sess, has_right):
                    nb["links"].append(item)
                continue
            fvis = normalize_visibility(item.get("visibility"), bvis)
            if not visible(fvis, sess, has_right):
                continue
            nf = dict(item)
            nf["links"] = _visible_links(item.get("links"), fvis,
                                         sess, has_right)
            nb["links"].append(nf)
        out.append(nb)
    return out


def check_ext_state(state):
    """Проверка POST ext_state (скрытие/переименование/позиция/размер
    внешних блоков). Вернуть текст ошибки или ''."""
    if not isinstance(state, dict):
        return "ext_state должен быть объектом"
    if len(state) > MAX_EXT_STATE_KEYS:
        return "ext_state: больше %d записей" % MAX_EXT_STATE_KEYS
    for k, v in state.items():
        if (not isinstance(k, str) or not k.strip()
                or len(k) > MAX_EXT_STATE_KEY_LEN):
            return "ext_state: плохой ключ"
        if not isinstance(v, dict):
            return "ext_state: запись должна быть объектом"
        if "hidden" in v and not isinstance(v["hidden"], bool):
            return "ext_state: hidden — true/false"
        t = v.get("title", "")
        if t and (not isinstance(t, str) or len(t) > 80):
            return "ext_state: title — строка ≤80"
        hl = v.get("hidden_links", [])
        if (not isinstance(hl, list) or len(hl) > EXTERNAL_MAX_LINKS
                or any(not isinstance(u, str) or not u or len(u) > 500
                       for u in hl)):
            return "ext_state: hidden_links — список URL ≤%d" % (
                EXTERNAL_MAX_LINKS)
        if "pos" in v and (isinstance(v["pos"], bool)
                           or not isinstance(v["pos"], int)
                           or not 0 <= v["pos"] <= 10000):
            return "ext_state: pos — целое 0..10000"
        if "colSpan" in v and (not isinstance(v["colSpan"], int)
                               or isinstance(v["colSpan"], bool)
                               or not 1 <= v["colSpan"] <= 3):
            return "ext_state: colSpan — целое 1..3"
        if "linkCols" in v and (not isinstance(v["linkCols"], int)
                                or isinstance(v["linkCols"], bool)
                                or not 1 <= v["linkCols"] <= 4):
            return "ext_state: linkCols — целое 1..4"
    return ""


def apply_ext_state(blocks, state, include_hidden=False):
    """Наложить ext_state на внешние блоки (до фильтра видимости).
    hidden-блок: выкинуть (или пометить ext_hidden для редактора);
    title: показать своё название; hidden_links: выкинуть ссылки по URL
    (или пометить ext_link_hidden для редактора); pos/colSpan/linkCols:
    прикрепить как _pos/colSpan/linkCols (позицию применяет
    order_combined, размер — рендер). Копии в своих блоках
    (from="модуль / контракт") не трогаем — они живут сами."""
    out = []
    state = state if isinstance(state, dict) else {}
    for b in blocks or []:
        if not isinstance(b, dict):
            continue
        st = state.get(ext_key(b.get("module"), b.get("contract")
                               or b.get("title"))) or {}
        if not isinstance(st, dict):
            st = {}
        nb = dict(b)
        t = (st.get("title") or "")
        if isinstance(t, str) and t.strip():
            nb["title"] = t.strip()[:80]
        if isinstance(st.get("pos"), int) and not isinstance(
                st.get("pos"), bool) and 0 <= st["pos"] <= 10000:
            nb["_pos"] = st["pos"]
        if isinstance(st.get("colSpan"), int) and not isinstance(
                st.get("colSpan"), bool) and 1 <= st["colSpan"] <= 3:
            nb["colSpan"] = st["colSpan"]
        if isinstance(st.get("linkCols"), int) and not isinstance(
                st.get("linkCols"), bool) and 1 <= st["linkCols"] <= 4:
            nb["linkCols"] = st["linkCols"]
        hidden_urls = set(
            u for u in (st.get("hidden_links") or [])
            if isinstance(u, str))
        if hidden_urls:
            kept = []
            for l in nb.get("links") or []:
                if (isinstance(l, dict) and l.get("url") in hidden_urls
                        and not is_folder(l)):
                    if include_hidden:
                        nl = dict(l)
                        nl["ext_link_hidden"] = True
                        kept.append(nl)
                    continue
                kept.append(l)
            nb["links"] = kept
        if st.get("hidden"):
            if not include_hidden:
                continue
            nb["ext_hidden"] = True
        out.append(nb)
    return out


def order_combined(own, ext):
    """Собрать итоговый порядок: свои блоки в своём порядке, внешние —
    вставкой по _pos (индекс в итоговом списке, clamp). Без _pos —
    в конец в порядке поступления. Детерминировано, без ничьих."""
    out = [b for b in (own or []) if isinstance(b, dict)]
    ordered = sorted(
        [b for b in (ext or []) if isinstance(b, dict)],
        key=lambda b: (b.get("_pos")
                       if isinstance(b.get("_pos"), int) else 10 ** 9))
    for b in ordered:
        pos = b.get("_pos")
        if not isinstance(pos, int) or isinstance(pos, bool):
            out.append(b)
            continue
        out.insert(max(0, min(len(out), pos)), b)
    return out


def check_blocks(blocks):
    """Проверка POST-конфига. Вернуть текст ошибки или ''."""
    if not isinstance(blocks, list) or len(blocks) > 200:
        return "blocks должен быть списком ≤200"
    for b in blocks:
        if not isinstance(b, dict):
            return "блок должен быть объектом"
        if len(b.get("links") or []) > 200:
            return "в блоке больше 200 ссылок"
        if b.get("visibility") and not valid_visibility(
                str(b["visibility"]).strip()):
            return ("видимость блока %r: public|auth|right:право[;право2]"
                    % (b.get("title") or "?"))
        folders = 0
        for item in (b.get("links") or []):
            if not isinstance(item, dict):
                return "ссылка должна быть объектом"
            if not is_folder(item):
                if item.get("visibility") and not valid_visibility(
                        str(item.get("visibility")).strip()):
                    return ("видимость ссылки %r: public|auth|right:право / пусто"
                            % (item.get("title") or "?"))
                continue
            folders += 1
            if folders > MAX_FOLDERS_PER_BLOCK:
                return ("в блоке %r больше %d папок"
                        % (b.get("title") or "?",
                           MAX_FOLDERS_PER_BLOCK))
            if item.get("visibility") and not valid_visibility(
                    str(item["visibility"]).strip()):
                return ("видимость папки %r: public|auth|right:право / пусто"
                        % (item.get("title") or "?"))
            if len(item.get("links") or []) > 200:
                return ("в папке %r больше 200 ссылок"
                        % (item.get("title") or "?"))
            for l in (item.get("links") or []):
                if not isinstance(l, dict):
                    return "ссылка должна быть объектом"
                if is_folder(l):
                    return ("вложенные папки запрещены (папка %r)"
                            % (item.get("title") or "?"))
                if l.get("visibility") and not valid_visibility(
                        str(l.get("visibility")).strip()):
                    return ("видимость ссылки %r: public|auth|right:право / пусто"
                            % (l.get("title") or "?"))
    return ""

"""homepage.favicon — иконка сайта по URL (кеш data-URL для поля «Иконка»).

Зачем: в поле иконки ссылки можно вписать URL картинки вручную, но удобнее
поставить галочку «подтянуть с сайта» — сервер сам скачает favicon страницы
(через <link rel="icon"> или /favicon.ico), проверит что это картинка
(сигнатура, не Content-Type) и вернёт data:-URL: поле иконки его уже умеет
показывать (<img>), в homepage.json он хранится как обычный текст.

Ограничения (анти-абьюз, роут только с правом homepage_edit):
- только http/https, без userinfo в URL;
- страница ≤512 КБ, иконка ≤128 КБ, таймауты 10 с;
- кеш data/cache/homepage/icons/<sha256>.<ext> + index.json, TTL 7 суток
  (повторный запрос по тому же URL сеть не трогает).
"""
import base64
import hashlib
import json
import logging
import os
import re
import threading
import time
import urllib.parse
import urllib.request

logger = logging.getLogger(__name__)

_lock = threading.Lock()
_cache_dir = ""

UA = "botmod-homepage/1.0"
PAGE_TIMEOUT = 10
ICON_TIMEOUT = 10
MAX_PAGE_BYTES = 512 * 1024
MAX_ICON_BYTES = 128 * 1024
CACHE_TTL = 7 * 24 * 3600

_LINK_RE = re.compile(r"<link\b[^>]{0,2000}>", re.I)
_REL_RE = re.compile(
    r"rel\s*=\s*(?:\"([^\"]+)\"|'([^']+)'|([^\s>]+))", re.I)
_HREF_RE = re.compile(
    r"href\s*=\s*(?:\"([^\"]{1,2000})\"|'([^']{1,2000})'|([^\s>]{1,2000}))",
    re.I)

_EXT_MIME = {"png": "image/png", "jpg": "image/jpeg",
             "gif": "image/gif", "bmp": "image/bmp",
             "webp": "image/webp", "ico": "image/x-icon",
             "tif": "image/tiff", "svg": "image/svg+xml"}


def configure(cache_dir: str) -> None:
    global _cache_dir
    _cache_dir = cache_dir or ""


def clean_url(url) -> str:
    """Проверка+нормализация URL страницы. Мусор -> ValueError."""
    s = (url or "").strip()
    try:
        p = urllib.parse.urlsplit(s)
    except Exception:
        raise ValueError("плохой URL")
    if p.scheme.lower() not in ("http", "https"):
        raise ValueError("разрешён только http(s)")
    if not p.hostname or p.username or p.password:
        raise ValueError("плохой URL (хост/userinfo)")
    # Без фрагмента; дефолтные порты схлопываем для стабильности кеша.
    netloc = p.hostname.lower()
    try:
        port = p.port
    except ValueError:
        raise ValueError("плохой порт в URL")
    if port and not ((p.scheme.lower() == "http" and port == 80)
                     or (p.scheme.lower() == "https" and port == 443)):
        netloc += ":%d" % port
    path = p.path or "/"
    q = ("?" + p.query) if p.query else ""
    return "%s://%s%s%s" % (p.scheme.lower(), netloc, path, q)


def _http_get(url: str, timeout: int, limit: int):
    """Тело (обрезанное лимитом) + content-type. Превышение/ошибка -> ValueError."""
    req = urllib.request.Request(url, headers={"User-Agent": UA})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            body = r.read(limit + 1)
            ctype = (r.headers.get("Content-Type") or "").split(";")[0].strip()
    except Exception as e:  # noqa: BLE001
        raise ValueError("запрос не удался: %s" % str(e)[:120])
    if len(body) > limit:
        raise ValueError("ответ больше %d КБ" % (limit // 1024))
    if not body:
        raise ValueError("пустой ответ")
    return body, ctype


def find_icon_href(html: bytes, page_url: str):
    """href иконки из <link rel=*icon*> (абсолютный) или None."""
    try:
        text = html.decode("utf-8", errors="replace")
    except Exception:  # noqa: BLE001
        return None
    fallback = None
    for tag in _LINK_RE.findall(text):
        rel = _REL_RE.search(tag)
        href = _HREF_RE.search(tag)
        if not rel or not href:
            continue
        rel_val = rel.group(1) or rel.group(2) or rel.group(3) or ""
        tokens = rel_val.lower().split()
        if not any("icon" in t for t in tokens):
            continue
        href_val = href.group(1) or href.group(2) or href.group(3) or ""
        try:
            abs_url = urllib.parse.urljoin(page_url, href_val.strip())
            p = urllib.parse.urlsplit(abs_url)
        except Exception:  # noqa: BLE001
            continue
        if p.scheme.lower() not in ("http", "https"):
            continue
        if "apple-touch" in tokens or "touch" in " ".join(tokens):
            fallback = fallback or abs_url
        else:
            return abs_url
    return fallback


def sniff(data: bytes):
    """Сигнатура картинки -> (ext, mime) или None."""
    if data[:8] == b"\x89PNG\r\n\x1a\n":
        return "png", "image/png"
    if len(data) >= 3 and data[:3] == b"\xff\xd8\xff":
        return "jpg", "image/jpeg"
    if len(data) >= 6 and data[:6] in (b"GIF87a", b"GIF89a"):
        return "gif", "image/gif"
    if len(data) >= 2 and data[:2] == b"BM":
        return "bmp", "image/bmp"
    if len(data) >= 12 and data[:4] == b"RIFF" and data[8:12] == b"WEBP":
        return "webp", "image/webp"
    if len(data) >= 4 and data[:4] == b"\x00\x00\x01\x00":
        return "ico", "image/x-icon"
    if len(data) >= 4 and data[:4] in (b"II*\x00", b"MM\x00*"):
        return "tif", "image/tiff"
    head = data[:4096].lstrip()[:200].lower()
    if head.startswith(b"<svg") or (head.startswith(b"<?xml")
                                    and b"<svg" in data[:4096].lower()):
        return "svg", "image/svg+xml"
    return None


def _index_path() -> str:
    return os.path.join(_cache_dir, "index.json")


def _index_load() -> dict:
    try:
        with open(_index_path(), "r", encoding="utf-8") as f:
            data = json.load(f)
        return data if isinstance(data, dict) else {}
    except Exception:  # noqa: BLE001
        return {}


def _index_save(index: dict) -> None:
    try:
        os.makedirs(_cache_dir, exist_ok=True)
        tmp = _index_path() + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump(index, f, ensure_ascii=False)
        os.replace(tmp, _index_path())
    except OSError as e:
        logger.warning("favicon: индекс не записался: %s", e)


def _cache_get(key: str):
    """Свежая запись кеша -> data-URL или None."""
    with _lock:
        rec = _index_load().get(key) or {}
        name = rec.get("file") or ""
        at = rec.get("at") or 0
        if not name or time.time() - at > CACHE_TTL:
            return None
        path = os.path.join(_cache_dir, os.path.basename(name))
        if not os.path.isfile(path):
            return None
        try:
            with open(path, "rb") as f:
                blob = f.read(MAX_ICON_BYTES + 1)
        except OSError:
            return None
    ext = (os.path.splitext(name)[1] or "").lstrip(".").lower()
    mime = _EXT_MIME.get(ext)
    if not mime or not blob:
        return None
    return "data:%s;base64,%s" % (
        mime, base64.b64encode(blob).decode("ascii"))


def _cache_put(key: str, ext: str, blob: bytes, source: str) -> None:
    with _lock:
        try:
            os.makedirs(_cache_dir, exist_ok=True)
            name = "%s.%s" % (key, ext)
            with open(os.path.join(_cache_dir, name), "wb") as f:
                f.write(blob)
        except OSError as e:
            logger.warning("favicon: кеш не записался: %s", e)
            return
        index = _index_load()
        index[key] = {"file": name, "at": int(time.time()),
                      "source": (source or "")[:500]}
        # Чистка протухшего (дешёвая: по одному проходу на запись).
        now = int(time.time())
        for k in [k for k, v in index.items()
                  if now - (v.get("at") or 0) > CACHE_TTL]:
            try:
                os.remove(os.path.join(
                    _cache_dir, os.path.basename(index[k].get("file") or "")))
            except OSError:
                pass
            del index[k]
        _index_save(index)


def fetch_icon(page_url: str):
    """Скачать иконку страницы. Вернуть (ok, icon|error, cached)."""
    try:
        url = clean_url(page_url)
    except ValueError as e:
        return False, str(e), False
    key = hashlib.sha256(url.encode()).hexdigest()[:32]
    hit = _cache_get(key)
    if hit:
        return True, hit, True
    candidates = []
    try:
        page, _ = _http_get(url, PAGE_TIMEOUT, MAX_PAGE_BYTES)
    except ValueError as e:
        # Страница не скачалась (толстая/лежит) — шанс остаётся:
        # прямой /favicon.ico пробуем всё равно.
        page, page_err = None, str(e)
    else:
        page_err = ""
        href = find_icon_href(page, url)
        if href:
            candidates.append(href)
    try:
        p = urllib.parse.urlsplit(url)
        candidates.append("%s://%s/favicon.ico" % (p.scheme, p.netloc))
    except Exception:  # noqa: BLE001
        pass
    if not candidates:
        return False, page_err or "иконка не найдена", False
    last_err = page_err or "иконка не найдена"
    for icon_url in candidates:
        try:
            blob, _ = _http_get(icon_url, ICON_TIMEOUT, MAX_ICON_BYTES)
        except ValueError as e:
            last_err = str(e)
            continue
        sniffed = sniff(blob)
        if not sniffed:
            last_err = "по ссылке не картинка"
            continue
        ext, mime = sniffed
        _cache_put(key, ext, blob, icon_url)
        return True, "data:%s;base64,%s" % (
            mime, base64.b64encode(blob).decode("ascii")), False
    return False, last_err, False

"""homepage.Module — блоки домашней страницы веб-панели (/api/homepage)."""
import logging
import os

from core.base_module import BaseModule

from . import engine as hp

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("homepage", mod_dir)
        self.ctx = ctx

        def _f(key, default):
            return (self.config.get(key, "") or "").strip() or default

        def _fp(key, default):
            from core.paths import abs_data
            return abs_data(ctx.data_dir,
                            (self.config.get(key, "") or "").strip(),
                            default)

        hp.configure(_fp("HOMEPAGE_FILE",
                         os.path.join(ctx.data_dir, "homepage.json")))
        ctx.api.register("homepage", "load", hp.load)
        ctx.api.register("homepage", "save", hp.save)
        ctx.api.register("homepage", "config",
                         lambda: {"HOMEPAGE_TITLE": _f("HOMEPAGE_TITLE", "")})

    def health(self) -> dict:
        try:
            blocks = len((hp.load().get("blocks") or []))
        except Exception:  # noqa: BLE001
            blocks = -1
        title = ""
        try:
            title = (self.config.get("HOMEPAGE_TITLE", "") or "").strip()
        except Exception:  # noqa: BLE001
            pass
        return {"ok": blocks >= 0, "module": self.name,
                "blocks": blocks, "title": title or "default"}

"""homepage.ui_web — GET/POST /api/homepage (блоки домашней страницы).

GET публичный, но отдаёт только видимое запросившему (фильтр по сессии:
public/auth/right — как панели мониторинга). POST — право homepage_edit.
Внешние ссылки: любой модуль может заявить в своём ui_web опциональный
HOMEPAGE_LINKS = [{"block", "title", "url", "desc", "icon", "newTab",
"visibility"}] — homepage соберёт их в read-only блоки после своих
(block — группа, по умолчанию имя модуля; visibility по умолчанию "auth",
т.к. чужие ссылки часто ведут во внутренние сервисы). remotedesk и прочие
модули трогать не надо — контракт опциональный, нет списка — нет блоков.
"""

ROUTES = [
    ("GET", "/api/homepage", "homepage", {"right": None}),
    ("POST", "/api/homepage", "homepage", {"right": "homepage_edit"}),
    ("GET", "/api/homepage/icon", "homepage_icon",
     {"right": "homepage_edit"}),
]

PANELS = []

NAV = [{"id": "homepage", "title": "Домашняя", "view": "homepage",
        "icon": "", "right": None}]

DEFAULT_TITLE = "Домашний помощник"
EXTERNAL_DEFAULT_VISIBILITY = "auth"
EXTERNAL_MAX_LINKS = 200


def _has_right(ctx, sess, name) -> bool:
    try:
        return bool(ctx.api.call("transport_web.check", sess, name))
    except Exception:  # noqa: BLE001
        return False


def _title(config) -> str:
    try:
        t = ((config or {}).get("HOMEPAGE_TITLE", "") or "").strip()
    except Exception:  # noqa: BLE001
        t = ""
    return t or DEFAULT_TITLE


def collect_external(ctx):
    """Ссылки других модулей: [{block, links[], module}] (группы по block).

    Читает опциональный HOMEPAGE_LINKS из botmod_<name>.ui_web загруженных
    модулей. Битые записи пропускаются (best-effort, в лог — ничего: тихий
    контракт, модуль мог опечататься — его проблема, страница живёт).
    """
    from . import engine as hp

    groups = {}
    order = []
    mods = []
    try:
        mods = list(getattr(ctx, "loaded_modules", None) or [])
    except Exception:  # noqa: BLE001
        pass
    import importlib as _il
    for m in mods:
        name = getattr(m, "name", "") or ""
        if not name or name == "homepage":
            continue
        try:
            ui = _il.import_module("botmod_%s.ui_web" % name)
        except Exception:  # noqa: BLE001
            continue
        raw = getattr(ui, "HOMEPAGE_LINKS", None)
        if not raw:
            continue
        items = raw if isinstance(raw, list) else [raw]
        for e in items:
            if not isinstance(e, dict):
                continue
            url = str(e.get("url") or "").strip()
            title = str(e.get("title") or "").strip()
            if not url or not title:
                continue
            if not (url.startswith(("http://", "https://", "/"))
                    and not url.startswith("//")):
                continue
            block = str(e.get("block") or name).strip()[:80] or name
            vis = hp.normalize_visibility(
                e.get("visibility"), EXTERNAL_DEFAULT_VISIBILITY)
            link = {"title": title[:120], "url": url[:500],
                    "visibility": vis}
            for opt in ("desc", "icon"):
                v = str(e.get(opt) or "").strip()
                if v:
                    link[opt] = v[:300]
            if e.get("newTab"):
                link["newTab"] = True
            if block not in groups:
                groups[block] = {"module": name, "links": []}
                order.append(block)
            if len(groups[block]["links"]) < EXTERNAL_MAX_LINKS:
                groups[block]["links"].append(link)
    out = []
    for b in order:
        links = groups[b]["links"]
        # Видимость блока = самая широкая из ссылок (public > auth >
        # объединение right через ; = «хоть одно», ≤5 имён, иначе auth).
        bvis = EXTERNAL_DEFAULT_VISIBILITY
        rights = []
        for l in links:
            v = l.get("visibility") or EXTERNAL_DEFAULT_VISIBILITY
            if v == "public":
                bvis = "public"
                break
            if v == "auth":
                bvis = "auth"
            elif v.startswith("right:"):
                for n in v[6:].split(";"):
                    n = (n or "").strip()
                    if n and n not in rights:
                        rights.append(n)
        if bvis != "public":
            bvis = "auth" if any(
                (l.get("visibility") or "") == "auth"
                for l in links) else (
                    "right:" + ";".join(rights[:5]) if rights
                    else EXTERNAL_DEFAULT_VISIBILITY)
        out.append({"title": b, "contract": b, "module": groups[b]["module"],
                    "external": True, "visibility": bvis, "links": links})
    return out


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

    from . import engine as hp

    if method == "homepage_icon":
        # Иконка сайта по URL (чекбокс в форме ссылки): сервер качает
        # favicon, проверяет сигнатуру и отдаёт data:-URL для поля иконки.
        # Только с правом homepage_edit (иначе — SSRF-абьюз чужими руками).
        from . import favicon as fav
        import os as _os

        url = (((req.get("query") or {}).get("url")) or "").strip()
        if not url:
            return 400, {"ok": False, "error": "нужен ?url="}
        fav.configure(_os.path.join(ctx.data_dir, "cache",
                                    "homepage", "icons"))
        ok, payload, cached = fav.fetch_icon(url)
        if not ok:
            return 400, {"ok": False, "error": payload}
        return {"ok": True, "icon": payload, "cached": cached}

    if method == "homepage":
        if req.get("http_method") == "GET":
            data = hp.load()
            if not isinstance(data, dict):
                data = {}
            blocks = data.get("blocks", [])
            if not isinstance(blocks, list):
                blocks = []
            ext_state = data.get("ext_state", {})
            if not isinstance(ext_state, dict):
                ext_state = {}
            sess = req.get("session")
            has_right = lambda s, n: _has_right(ctx, s, n)  # noqa: E731
            can_edit = bool(sess) and _has_right(ctx, sess, "homepage_edit")
            # Редактор получает СВОИ блоки сырыми (без фильтра): иначе
            # папка/ссылка с видимостью, которой у него нет, пропадает из
            # его GET — смена видимости «не применяется», а следующий сейв
            # молча трёт её из файла. Прецедент — сырой ext_state ниже.
            own = (blocks if can_edit
                   else hp.filter_blocks(blocks, sess, has_right))
            ext = hp.apply_ext_state(collect_external(ctx), ext_state,
                                     include_hidden=can_edit)
            ext = [b for b in hp.filter_blocks(ext, sess, has_right)
                   if b.get("links") or (can_edit and b.get("ext_hidden"))]
            out = {"ok": True, "title": _title(config),
                   "blocks": hp.order_combined(own, ext)}
            if can_edit:
                # Сырое состояние — только редактору (нужно чтобы снять
                # скрытие: скрытые блоки/ссылки иначе неоткуда взять).
                out["ext_state"] = ext_state
            return out
        body = req.get("body") or {}
        blocks = body.get("blocks", [])
        # Внешние блоки — read-only с сервера: в файл не пишем никогда
        # (пересоберутся из HOMEPAGE_LINKS при следующем GET).
        blocks = [b for b in blocks
                  if isinstance(b, dict) and not b.get("external")]
        err = hp.check_blocks(blocks)
        if err:
            return 400, {"ok": False, "error": err}
        ext_state = body.get("ext_state", {})
        err = hp.check_ext_state(ext_state)
        if err:
            return 400, {"ok": False, "error": err}
        if not hp.save({"blocks": blocks, "ext_state": ext_state}):
            return 500, {"ok": False, "error": "Ошибка записи"}
        return {"ok": True}

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