Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions astrbot/core/config/astrbot_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,14 @@ def __init__(
)
# 检查配置完整性,并插入
has_new = self.check_config_integrity(default_config, conf)

dashboard_conf = conf.get("dashboard")
if isinstance(dashboard_conf, dict):
host_val = dashboard_conf.get("host")
if isinstance(host_val, str) and host_val:
dashboard_conf["host"] = [host_val]
has_new = True
Comment thread
2278535805 marked this conversation as resolved.

reset_dashboard_password = self._consume_reset_dashboard_password_flag()
if reset_dashboard_password and "dashboard" in conf:
self._reset_generated_dashboard_password(conf)
Expand Down
2 changes: 1 addition & 1 deletion astrbot/core/config/default.py
Original file line number Diff line number Diff line change
Expand Up @@ -257,7 +257,7 @@
"password_storage_upgraded": False,
"password_change_required": False,
"jwt_secret": "",
"host": "0.0.0.0",
"host": ["0.0.0.0", "::"],
"port": 6185,
"disable_access_log": True,
"trust_proxy_headers": False,
Expand Down
285 changes: 285 additions & 0 deletions astrbot/core/utils/io.py
Original file line number Diff line number Diff line change
Expand Up @@ -361,5 +361,290 @@ def get_local_ip_addresses():
for addr in addrs:
if addr.family == socket.AF_INET: # 使用 socket.AF_INET 代替 psutil.AF_INET
network_ips.append(addr.address)
elif addr.family == socket.AF_INET6:
address = addr.address
scope_idx = address.find("%")
if scope_idx != -1:
address = address[:scope_idx]
network_ips.append(address)

return network_ips


def normalize_host_list(host_raw: str | list[str] | None) -> list[str]:
"""Normalize a host config value into a list of host strings.

Args:
host_raw: A string, list of strings, or None.

Returns:
A list of non-empty host strings.
"""
if host_raw is None:
return []
if isinstance(host_raw, list):
return [h for h in host_raw if h]
return [h.strip() for h in str(host_raw).split(",") if h.strip()]


def get_dashboard_dist_version(dist_dir: str | Path) -> str | None:
"""Read the WebUI version from a dashboard dist directory.

Args:
dist_dir: Dashboard dist directory path.

Returns:
The version string from assets/version, or None when unavailable.
"""

version_file = Path(dist_dir) / "assets" / "version"
try:
if version_file.exists():
return version_file.read_text(encoding="utf-8").strip()
except (OSError, UnicodeDecodeError) as exc:
logger.warning("Failed to read WebUI version from %s: %s", version_file, exc)
return None


def get_bundled_dashboard_dist_path() -> Path:
return Path(get_astrbot_path()) / "astrbot" / "dashboard" / "dist"


def _normalize_dashboard_version(version: str) -> str:
version = version.strip()
if version[:1].lower() == "v":
version = version[1:]
if not re.match(
r"^[0-9]+(?:\.[0-9]+)*"
r"(?:-[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?"
r"(?:\+.+)?$",
version,
):
raise ValueError(f"invalid dashboard version: {version!r}")
return version


def is_dashboard_version_compatible(
dashboard_version: str | None, current_version: str
) -> bool:
"""Check whether a WebUI version matches the current core version.

Args:
dashboard_version: Version read from the WebUI assets/version file.
current_version: Current AstrBot core version.

Returns:
True when both versions are valid SemVer values and compare equal.
"""

if dashboard_version is None:
return False
try:
return (
VersionComparator.compare_version(
_normalize_dashboard_version(dashboard_version),
_normalize_dashboard_version(current_version),
)
== 0
)
except (TypeError, ValueError):
return False


def is_dashboard_dist_compatible(dist_dir: str | Path, current_version: str) -> bool:
"""Check whether a WebUI dist is complete and matches the core version.

Args:
dist_dir: Dashboard dist directory path.
current_version: Current AstrBot core version.

Returns:
True when the dist has an index file and a compatible assets/version.
"""

dist_path = Path(dist_dir)
return (dist_path / "index.html").is_file() and is_dashboard_version_compatible(
get_dashboard_dist_version(dist_path),
current_version,
)


def should_use_bundled_dashboard_dist(
user_dist: str | Path, current_version: str
) -> bool:
"""Decide whether bundled WebUI should replace a user data dist.

Args:
user_dist: Runtime dashboard dist directory under data/.
current_version: Current AstrBot core version.

Returns:
True when user_dist exists but is missing or mismatched against the
current core version, and bundled WebUI matches the current core version.
"""

user_dist = Path(user_dist)
user_version = get_dashboard_dist_version(user_dist)
bundled_dist = get_bundled_dashboard_dist_path()
if not user_dist.exists() or not is_dashboard_dist_compatible(
bundled_dist,
current_version,
):
return False
if user_version is None or not (user_dist / "index.html").is_file():
return True
try:
return not is_dashboard_version_compatible(user_version, current_version)
except (TypeError, ValueError):
return False


async def get_dashboard_version():
"""Return the effective WebUI version for the current runtime.

Returns:
The matching data/dist version, matching bundled version, or the raw
data/dist version when no compatible bundled WebUI is available.
"""

from astrbot.core.config.default import VERSION

# First check user data directory (manually updated / downloaded dashboard).
dist_dir = os.path.join(get_astrbot_data_path(), "dist")
if os.path.exists(dist_dir):
user_version = get_dashboard_dist_version(dist_dir)
if is_dashboard_dist_compatible(dist_dir, VERSION):
return user_version

bundled = get_bundled_dashboard_dist_path()
if is_dashboard_dist_compatible(bundled, VERSION):
return get_dashboard_dist_version(bundled)
return user_version

bundled = get_bundled_dashboard_dist_path()
if is_dashboard_dist_compatible(bundled, VERSION):
return get_dashboard_dist_version(bundled)
return None


async def download_dashboard(
path: str | None = None,
extract_path: str = "data",
latest: bool = True,
version: str | None = None,
proxy: str | None = None,
progress_callback=None,
extract: bool = True,
allow_insecure_ssl_fallback: bool = True,
) -> None:
"""Download dashboard assets and optionally extract them.

Args:
path: Destination zip path. Defaults to the AstrBot data directory.
extract_path: Directory where assets should be extracted.
latest: Whether to download the latest dashboard build.
version: Specific release tag or commit hash to download.
proxy: Optional download proxy prefix.
progress_callback: Optional callback for download progress payloads.
extract: Whether to extract the archive after download.
allow_insecure_ssl_fallback: Whether certificate failures may retry with
TLS certificate verification disabled.

Returns:
None.
"""
if path is None:
zip_path = Path(get_astrbot_data_path()).absolute() / "dashboard.zip"
else:
zip_path = Path(path).absolute()
ensure_dir(zip_path.parent)

if latest or len(str(version)) != 40:
ver_name = "latest" if latest else version
dashboard_release_url = f"https://astrbot-registry.soulter.top/download/astrbot-dashboard/{ver_name}/dist.zip"
logger.info(
f"Downloading AstrBot WebUI from {dashboard_release_url}",
)
try:
await download_file(
dashboard_release_url,
str(zip_path),
show_progress=True,
progress_callback=progress_callback,
allow_insecure_ssl_fallback=allow_insecure_ssl_fallback,
)
if not zipfile.is_zipfile(zip_path):
raise RuntimeError(
"Downloaded dashboard package is not a valid ZIP file"
)
except BaseException as _:
if latest:
# Resolve latest release tag from GitHub API to construct correct asset URL
ssl_context = ssl.create_default_context(cafile=certifi.where())
async with aiohttp.ClientSession(
connector=aiohttp.TCPConnector(ssl=ssl_context),
trust_env=True,
) as session:
async with session.get(
"https://api.github.com/repos/AstrBotDevs/AstrBot/releases/latest",
timeout=30,
headers={"Accept": "application/vnd.github+json"},
) as api_resp:
api_resp.raise_for_status()
release_data = await api_resp.json()
tag = release_data["tag_name"]
else:
tag = version
dashboard_release_url = f"https://github.com/AstrBotDevs/AstrBot/releases/download/{tag}/AstrBot-{tag}-dashboard.zip"
if proxy:
dashboard_release_url = f"{proxy}/{dashboard_release_url}"
await download_file(
dashboard_release_url,
str(zip_path),
show_progress=True,
progress_callback=progress_callback,
allow_insecure_ssl_fallback=allow_insecure_ssl_fallback,
)
if not zipfile.is_zipfile(zip_path):
raise RuntimeError(
"Downloaded dashboard package is not a valid ZIP file"
)
else:
url = f"https://github.com/AstrBotDevs/astrbot-release-harbour/releases/download/release-{version}/dist.zip"
logger.info(f"Downloading AstrBot WebUI from {url}")
if proxy:
url = f"{proxy}/{url}"
await download_file(
url,
str(zip_path),
show_progress=True,
progress_callback=progress_callback,
allow_insecure_ssl_fallback=allow_insecure_ssl_fallback,
)
if not zipfile.is_zipfile(zip_path):
raise RuntimeError("Downloaded dashboard package is not a valid ZIP file")
if extract:
extract_dashboard(zip_path, extract_path)


def extract_dashboard(zip_path: str | Path, extract_path: str | Path = "data") -> None:
"""Extract a downloaded dashboard archive.

Args:
zip_path: Dashboard zip archive path.
extract_path: Directory where the archive contents should be extracted.

Returns:
None.
"""

extract_root = Path(extract_path).resolve()
ensure_dir(extract_root)
with zipfile.ZipFile(zip_path, "r") as z:
for member in z.infolist():
target_path = (extract_root / member.filename).resolve()
if not target_path.is_relative_to(extract_root):
raise ValueError(
f"Unsafe dashboard archive path: {member.filename}",
)
z.extract(member, extract_root)
Loading