bahmcloud/bahmcloud_store
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 8 Wiki Activity

custom_components/bahmcloud_store/core.py aktualisiert

Browse Source
This commit is contained in:
René Bachmann
2026-01-15 16:21:14 +00:00
parent c4361cc8bd
commit ac5bc8a6f4
1 changed files with 52 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
custom_components/bahmcloud_store/core.py
View File

@@ -7,7 +7,7 @@ import time
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from urllib.parse import urlparse from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.aiohttp_client import async_get_clientsession
@@ -156,27 +156,62 @@ class BCSCore:
await asyncio.gather(*(process_one(r) for r in merged.values()), return_exceptions=True) await asyncio.gather(*(process_one(r) for r in merged.values()), return_exceptions=True)
async def _load_index_repos(self) -> tuple[list[RepoItem], int]: def _add_cache_buster(self, url: str) -> str:
"""Append a timestamp query parameter safely (works with existing query params)."""
parts = urlsplit(url)
q = dict(parse_qsl(parts.query, keep_blank_values=True))
q["t"] = str(int(time.time()))
new_query = urlencode(q)
return urlunsplit((parts.scheme, parts.netloc, parts.path, new_query, parts.fragment))
def _gitea_src_to_raw(self, url: str) -> str:
"""If the URL points to a Gitea 'src' page, convert it to a raw file URL."""
# Example:
# /owner/repo/src/branch/main/store.yaml -> /owner/repo/raw/branch/main/store.yaml
parts = urlsplit(url)
path = parts.path
path2 = path.replace("/src/branch/", "/raw/branch/")
if path2 == path:
return url
return urlunsplit((parts.scheme, parts.netloc, path2, parts.query, parts.fragment))
async def _fetch_store_text(self, url: str) -> str:
session = async_get_clientsession(self.hass) session = async_get_clientsession(self.hass)
# ---- CACHE BUSTING FIX (Phase B) ----
# Force store.yaml to always be reloaded from remote:
# - add timestamp query parameter
# - disable HTTP cache via headers
url = f"{self.config.store_url}?t={int(time.time())}"
headers = { headers = {
"Cache-Control": "no-cache", "User-Agent": "BahmcloudStore (Home Assistant)",
"Cache-Control": "no-cache, no-store, max-age=0",
"Pragma": "no-cache", "Pragma": "no-cache",
"Expires": "0",
} }
async with session.get(url, timeout=20, headers=headers) as resp:
if resp.status != 200:
raise BCSError(f"store_url returned {resp.status}")
return await resp.text()
async def _load_index_repos(self) -> tuple[list[RepoItem], int]:
# ---- Phase B: force an actual remote reload of store.yaml ----
store_url = (self.config.store_url or "").strip()
if not store_url:
raise BCSError("store_url is empty")
url = self._add_cache_buster(store_url)
try: try:
async with session.get(url, timeout=20, headers=headers) as resp: raw = await self._fetch_store_text(url)
if resp.status != 200:
raise BCSError(f"store_url returned {resp.status}") # If we accidentally fetched a HTML "src" page, try converting to raw.
raw = await resp.text() # This makes the system robust even if someone configures a /src/branch/ URL.
if "<html" in raw.lower() or "<!doctype html" in raw.lower():
fallback = self._add_cache_buster(self._gitea_src_to_raw(store_url))
if fallback != url:
_LOGGER.debug("BCS store index looked like HTML, retrying raw URL")
raw = await self._fetch_store_text(fallback)
except Exception as e: except Exception as e:
raise BCSError(f"Failed fetching store index: {e}") from e raise BCSError(f"Failed fetching store index: {e}") from e
# ---- END CACHE BUSTING FIX ---- # ---- end Phase B fix ----
try: try:
data = ha_yaml.parse_yaml(raw) data = ha_yaml.parse_yaml(raw)
@@ -192,16 +227,16 @@ class BCSCore:
for i, r in enumerate(repos): for i, r in enumerate(repos):
if not isinstance(r, dict): if not isinstance(r, dict):
continue continue
url = str(r.get("url", "")).strip() repo_url = str(r.get("url", "")).strip()
if not url: if not repo_url:
continue continue
name = str(r.get("name") or url).strip() name = str(r.get("name") or repo_url).strip()
items.append( items.append(
RepoItem( RepoItem(
id=f"index:{i}", id=f"index:{i}",
name=name, name=name,
url=url, url=repo_url,
source="index", source="index",
) )
) )