152 lines
4.9 KiB
Python
152 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import asdict
|
|
from pathlib import Path
|
|
from typing import Any, TYPE_CHECKING
|
|
|
|
from aiohttp import web
|
|
from homeassistant.components.http import HomeAssistantView
|
|
|
|
if TYPE_CHECKING:
|
|
from .core import BCSCore # typing only, avoids runtime circular import
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
class StaticAssetsView(HomeAssistantView):
|
|
url = "/api/bahmcloud_store_static/{path:.*}"
|
|
name = "api:bahmcloud_store_static"
|
|
|
|
# RESTORE previous behavior: require HA auth (this worked for you before)
|
|
requires_auth = True
|
|
|
|
async def get(self, request: web.Request, path: str) -> web.Response:
|
|
base = Path(__file__).resolve().parent / "panel"
|
|
base_resolved = base.resolve()
|
|
|
|
req_path = (path or "").lstrip("/")
|
|
if req_path == "":
|
|
req_path = "index.html"
|
|
|
|
target = (base / req_path).resolve()
|
|
|
|
# Prevent path traversal
|
|
if not str(target).startswith(str(base_resolved)):
|
|
return web.Response(status=404)
|
|
|
|
if target.is_dir():
|
|
target = (target / "index.html").resolve()
|
|
|
|
if not target.exists():
|
|
return web.Response(status=404)
|
|
|
|
# aiohttp: charset must NOT be included in content_type string
|
|
content_type = "text/plain"
|
|
charset = None
|
|
|
|
if target.suffix == ".js":
|
|
content_type = "application/javascript"
|
|
charset = "utf-8"
|
|
elif target.suffix == ".html":
|
|
content_type = "text/html"
|
|
charset = "utf-8"
|
|
elif target.suffix == ".css":
|
|
content_type = "text/css"
|
|
charset = "utf-8"
|
|
elif target.suffix == ".svg":
|
|
content_type = "image/svg+xml"
|
|
elif target.suffix == ".png":
|
|
content_type = "image/png"
|
|
|
|
return web.Response(body=target.read_bytes(), content_type=content_type, charset=charset)
|
|
|
|
|
|
class BCSApiView(HomeAssistantView):
|
|
url = "/api/bcs"
|
|
name = "api:bcs"
|
|
requires_auth = True
|
|
|
|
def __init__(self, core: Any) -> None:
|
|
self.core = core
|
|
|
|
async def get(self, request: web.Request) -> web.Response:
|
|
return web.json_response(
|
|
{
|
|
"ok": True,
|
|
"version": self.core.version,
|
|
"repos": self.core.list_repos_public(),
|
|
}
|
|
)
|
|
|
|
async def post(self, request: web.Request) -> web.Response:
|
|
data = await request.json()
|
|
op = data.get("op")
|
|
|
|
if op == "add_custom_repo":
|
|
url = str(data.get("url") or "").strip()
|
|
name = data.get("name")
|
|
name = str(name).strip() if name else None
|
|
if not url:
|
|
return web.json_response({"ok": False, "message": "Missing url"}, status=400)
|
|
repo = await self.core.add_custom_repo(url=url, name=name)
|
|
return web.json_response({"ok": True, "repo": asdict(repo)})
|
|
|
|
return web.json_response({"ok": False, "message": "Unknown operation"}, status=400)
|
|
|
|
|
|
class BCSCustomRepoView(HomeAssistantView):
|
|
url = "/api/bcs/custom_repo"
|
|
name = "api:bcs_custom_repo"
|
|
requires_auth = True
|
|
|
|
def __init__(self, core: Any) -> None:
|
|
self.core = core
|
|
|
|
async def delete(self, request: web.Request) -> web.Response:
|
|
repo_id = request.query.get("id")
|
|
if not repo_id:
|
|
return web.json_response({"ok": False, "message": "Missing id"}, status=400)
|
|
await self.core.remove_custom_repo(repo_id)
|
|
return web.json_response({"ok": True})
|
|
|
|
|
|
class BCSReadmeView(HomeAssistantView):
|
|
url = "/api/bcs/readme"
|
|
name = "api:bcs_readme"
|
|
requires_auth = True
|
|
|
|
def __init__(self, core: Any) -> None:
|
|
self.core = core
|
|
|
|
async def get(self, request: web.Request) -> web.Response:
|
|
repo_id = request.query.get("repo_id")
|
|
if not repo_id:
|
|
return web.json_response({"ok": False, "message": "Missing repo_id"}, status=400)
|
|
|
|
md = await self.core.fetch_readme_markdown(repo_id)
|
|
if not md:
|
|
return web.json_response({"ok": False, "message": "README not found."}, status=404)
|
|
|
|
html = None
|
|
|
|
# Best-effort: if HA provides a markdown renderer util, we use it.
|
|
try:
|
|
from homeassistant.util.markdown import async_render_markdown # type: ignore
|
|
|
|
html = await async_render_markdown(self.core.hass, md)
|
|
except Exception as e:
|
|
_LOGGER.debug("Markdown render failed: %s", e)
|
|
html = None
|
|
|
|
# Best-effort sanitization if available
|
|
if html:
|
|
try:
|
|
from homeassistant.util.sanitize_html import async_sanitize_html # type: ignore
|
|
|
|
html = await async_sanitize_html(self.core.hass, html)
|
|
except Exception as e:
|
|
_LOGGER.debug("HTML sanitize not available/failed: %s", e)
|
|
|
|
return web.json_response({"ok": True, "readme": md, "html": html})
|