322 lines
12 KiB
Python
322 lines
12 KiB
Python
"""Build portable ZIP bundles for generated HTML artifacts.
|
|
|
|
Generated pages intentionally reference the Gateway's vendored browser runtime
|
|
instead of public CDNs. A plain ``.html`` download therefore stops working
|
|
after it is moved away from the DeerFlow deployment. This module rewrites
|
|
those trusted Gateway URLs to relative paths and copies the corresponding
|
|
packaged assets into a ZIP. It never downloads arbitrary third-party URLs.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import zipfile
|
|
from dataclasses import dataclass
|
|
from importlib.resources import files
|
|
from pathlib import Path, PurePosixPath
|
|
|
|
from deerflow.html_runtime.public_images import resolve_html_image_path, verify_html_image_token
|
|
|
|
_STATIC_ASSETS = {
|
|
"alpine.min.js",
|
|
"bootstrap.bundle.min.js",
|
|
"bootstrap.min.css",
|
|
"brand-logo-dark.png",
|
|
"echarts.min.js",
|
|
"g6.min.js",
|
|
"leaflet.css",
|
|
"leaflet.js",
|
|
"topbar-dark.svg",
|
|
"topbar-light.svg",
|
|
}
|
|
_GEO_MAP_ASSETS = {
|
|
"china-provinces.geojson",
|
|
"japan-prefectures.geojson",
|
|
"taiwan-counties.geojson",
|
|
"us-states.geojson",
|
|
}
|
|
_COGNITIVE_ASSET_SUFFIXES = {
|
|
".css",
|
|
".js",
|
|
".json",
|
|
".jpeg",
|
|
".jpg",
|
|
".mjs",
|
|
".png",
|
|
".svg",
|
|
".woff2",
|
|
}
|
|
_LICENSE_BY_ASSET = {
|
|
"alpine.min.js": "LICENSE.alpine.txt",
|
|
"bootstrap.bundle.min.js": "LICENSE.bootstrap.txt",
|
|
"bootstrap.min.css": "LICENSE.bootstrap.txt",
|
|
"echarts.min.js": "LICENSE.echarts.txt",
|
|
"g6.min.js": "LICENSE.g6.txt",
|
|
"leaflet.css": "LICENSE.leaflet.txt",
|
|
"leaflet.js": "LICENSE.leaflet.txt",
|
|
}
|
|
_RUNTIME_ARCHIVE_ROOT = PurePosixPath("assets/html-runtime")
|
|
_HTML_IMAGES_ARCHIVE_ROOT = PurePosixPath("assets/html-images")
|
|
_STANDARD_RUNTIME_BASE_RE = (
|
|
r"(?:https?://[^/\"'\s<>()]+)?"
|
|
r"(?:/[A-Za-z0-9._~!$&()*+,;=:@%+-]+)*/api/public/html-runtime"
|
|
)
|
|
_HTML_IMAGE_URL_RE = re.compile(
|
|
r"(?P<url>(?:https?://[^/\"'\s<>()]+)?"
|
|
r"(?:/[A-Za-z0-9._~!$&()*+,;=:@%+-]+)*/api/public/html-images/"
|
|
r"(?P<token>[^\"'\s<>()?#]+)(?:\?[^\"'\s<>()#]*)?)",
|
|
re.IGNORECASE,
|
|
)
|
|
_EXTERNAL_URL_RE = re.compile(r"https?://[^\s\"'<>]+", re.IGNORECASE)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class OfflineBundleResult:
|
|
"""Summary of one completed offline bundle."""
|
|
|
|
bundled_assets: tuple[str, ...]
|
|
external_urls: tuple[str, ...]
|
|
|
|
|
|
def _runtime_base_pattern(runtime_base_url: str) -> str:
|
|
bases = [_STANDARD_RUNTIME_BASE_RE]
|
|
configured = str(runtime_base_url or "").strip().rstrip("/")
|
|
if configured:
|
|
bases.insert(0, re.escape(configured))
|
|
return "(?:" + "|".join(bases) + ")"
|
|
|
|
|
|
def _safe_runtime_source(runtime_dir: Path, relative_path: str) -> Path | None:
|
|
"""Resolve only assets that the public HTML-runtime router exposes."""
|
|
|
|
normalized = relative_path.strip().lstrip("/").replace("\\", "/")
|
|
pure_path = PurePosixPath(normalized)
|
|
if not normalized or pure_path.is_absolute() or ".." in pure_path.parts:
|
|
return None
|
|
|
|
allowed = False
|
|
if len(pure_path.parts) == 1 and normalized in _STATIC_ASSETS:
|
|
allowed = True
|
|
elif len(pure_path.parts) == 2 and pure_path.parts[0] == "maps" and pure_path.name in _GEO_MAP_ASSETS:
|
|
allowed = True
|
|
elif pure_path.parts and pure_path.parts[0] == "cognitive-situation-map":
|
|
allowed = pure_path.suffix.lower() in _COGNITIVE_ASSET_SUFFIXES
|
|
if not allowed:
|
|
return None
|
|
|
|
root = runtime_dir.resolve()
|
|
source = (root / Path(*pure_path.parts)).resolve()
|
|
try:
|
|
source.relative_to(root)
|
|
except ValueError:
|
|
return None
|
|
return source if source.is_file() else None
|
|
|
|
|
|
def _add_file(
|
|
archive_files: dict[str, Path],
|
|
archive_path: PurePosixPath,
|
|
source: Path,
|
|
) -> None:
|
|
archive_files[archive_path.as_posix()] = source
|
|
|
|
|
|
def _add_cognitive_runtime(archive_files: dict[str, Path], runtime_dir: Path) -> None:
|
|
source_root = runtime_dir / "cognitive-situation-map"
|
|
if not source_root.is_dir():
|
|
return
|
|
for source in source_root.rglob("*"):
|
|
if not source.is_file():
|
|
continue
|
|
is_license = source.name.lower().startswith("license")
|
|
if source.suffix.lower() not in _COGNITIVE_ASSET_SUFFIXES and not is_license:
|
|
continue
|
|
relative = source.relative_to(runtime_dir)
|
|
_add_file(archive_files, _RUNTIME_ARCHIVE_ROOT / PurePosixPath(relative.as_posix()), source)
|
|
|
|
|
|
def _generated_logo_destination(generated_logo_path: Path | None) -> PurePosixPath:
|
|
suffix = generated_logo_path.suffix.lower() if generated_logo_path is not None else ".png"
|
|
if suffix not in {".jpg", ".jpeg", ".png", ".webp"}:
|
|
suffix = ".png"
|
|
return _RUNTIME_ARCHIVE_ROOT / f"generated-page-logo{suffix}"
|
|
|
|
|
|
def _runtime_script(
|
|
runtime_dir: Path,
|
|
*,
|
|
map_tile_url: str,
|
|
include_geo_maps: bool,
|
|
) -> str:
|
|
config: dict[str, object] = {"geoMaps": {}, "mapTileUrl": map_tile_url}
|
|
if include_geo_maps:
|
|
config["geoMaps"] = {
|
|
Path(filename).stem: (_RUNTIME_ARCHIVE_ROOT / "maps" / filename).as_posix()
|
|
for filename in sorted(_GEO_MAP_ASSETS)
|
|
}
|
|
prefix = "window.DEERFLOW_HTML_RUNTIME_CONFIG=" + json.dumps(
|
|
config,
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
) + ";\n"
|
|
return prefix + (runtime_dir / "deerflow-runtime.js").read_text(encoding="utf-8")
|
|
|
|
|
|
def _readme(original_name: str, external_urls: list[str]) -> str:
|
|
external_note = (
|
|
"未打包的外部地址已记录在 manifest.json 的 external_urls 中；例如在线地图瓦片、"
|
|
"用户主动引用的第三方图片或接口仍需要网络。"
|
|
if external_urls
|
|
else "本页没有检测到仍需联网的 http(s) 地址。"
|
|
)
|
|
return (
|
|
"DeerFlow 网页离线包\n"
|
|
"====================\n\n"
|
|
f"原始文件：{original_name}\n"
|
|
"入口文件：index.html\n"
|
|
"依赖目录：assets/\n\n"
|
|
"部署方式：请保持目录结构不变，将解压后的全部文件一起放到任意静态网站目录。\n"
|
|
"若页面含 GeoJSON 地图等 fetch 请求，不建议直接双击 file:// 打开；可使用 Nginx、"
|
|
"Apache、IIS 或其他静态 HTTP 服务访问 index.html。\n\n"
|
|
f"{external_note}\n"
|
|
)
|
|
|
|
|
|
def build_html_offline_bundle(
|
|
html_path: Path,
|
|
destination: Path,
|
|
*,
|
|
runtime_base_url: str,
|
|
map_tile_url: str,
|
|
generated_logo_path: Path | None = None,
|
|
runtime_dir: Path | None = None,
|
|
) -> OfflineBundleResult:
|
|
"""Write a portable HTML ZIP to *destination* and return its manifest summary."""
|
|
|
|
html_path = Path(html_path)
|
|
destination = Path(destination)
|
|
runtime_dir = runtime_dir or Path(str(files("deerflow.html_runtime.assets")))
|
|
runtime_dir = Path(runtime_dir)
|
|
html = html_path.read_text(encoding="utf-8", errors="replace")
|
|
archive_files: dict[str, Path] = {}
|
|
generated_files: dict[str, str] = {}
|
|
bundled_assets: set[str] = set()
|
|
licenses: set[str] = set()
|
|
include_geo_maps = bool(re.search(r"data-df-geo\s*=", html, re.IGNORECASE))
|
|
uses_leaflet_map = bool(re.search(r"data-df-map\s*=", html, re.IGNORECASE))
|
|
|
|
base_pattern = _runtime_base_pattern(runtime_base_url)
|
|
cognitive_base_re = re.compile(
|
|
rf"(?P<url>{base_pattern}/cognitive-situation-map)(?=$|[\"'\s<>()/])",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
def replace_cognitive_base(match: re.Match[str]) -> str:
|
|
_add_cognitive_runtime(archive_files, runtime_dir)
|
|
bundled_assets.add((_RUNTIME_ARCHIVE_ROOT / "cognitive-situation-map").as_posix() + "/")
|
|
return (_RUNTIME_ARCHIVE_ROOT / "cognitive-situation-map").as_posix()
|
|
|
|
html = cognitive_base_re.sub(replace_cognitive_base, html)
|
|
|
|
runtime_url_re = re.compile(
|
|
rf"(?P<url>{base_pattern}/(?P<asset>[A-Za-z0-9._/-]+)"
|
|
r"(?:\?[^\"'\s<>()#]*)?)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
def replace_runtime_url(match: re.Match[str]) -> str:
|
|
asset = match.group("asset").strip("/")
|
|
if asset == "generated-page-logo":
|
|
source = generated_logo_path if generated_logo_path and generated_logo_path.is_file() else runtime_dir / "brand-logo-dark.png"
|
|
if not source.is_file():
|
|
return match.group("url")
|
|
archive_path = _generated_logo_destination(source)
|
|
_add_file(archive_files, archive_path, source)
|
|
bundled_assets.add(archive_path.as_posix())
|
|
return archive_path.as_posix()
|
|
if asset == "deerflow-runtime.js":
|
|
source = runtime_dir / asset
|
|
if not source.is_file():
|
|
return match.group("url")
|
|
archive_path = _RUNTIME_ARCHIVE_ROOT / asset
|
|
generated_files[archive_path.as_posix()] = _runtime_script(
|
|
runtime_dir,
|
|
map_tile_url=map_tile_url,
|
|
include_geo_maps=include_geo_maps,
|
|
)
|
|
bundled_assets.add(archive_path.as_posix())
|
|
if include_geo_maps:
|
|
for filename in sorted(_GEO_MAP_ASSETS):
|
|
map_source = _safe_runtime_source(runtime_dir, f"maps/{filename}")
|
|
if map_source is not None:
|
|
map_archive_path = _RUNTIME_ARCHIVE_ROOT / "maps" / filename
|
|
_add_file(archive_files, map_archive_path, map_source)
|
|
bundled_assets.add(map_archive_path.as_posix())
|
|
return archive_path.as_posix()
|
|
|
|
source = _safe_runtime_source(runtime_dir, asset)
|
|
if source is None:
|
|
return match.group("url")
|
|
archive_path = _RUNTIME_ARCHIVE_ROOT / PurePosixPath(asset)
|
|
_add_file(archive_files, archive_path, source)
|
|
bundled_assets.add(archive_path.as_posix())
|
|
license_name = _LICENSE_BY_ASSET.get(asset)
|
|
if license_name and (runtime_dir / license_name).is_file():
|
|
licenses.add(license_name)
|
|
return archive_path.as_posix()
|
|
|
|
html = runtime_url_re.sub(replace_runtime_url, html)
|
|
|
|
def replace_html_image(match: re.Match[str]) -> str:
|
|
token = match.group("token")
|
|
try:
|
|
_thread_id, _kind, filename = verify_html_image_token(token)
|
|
source = resolve_html_image_path(_thread_id, _kind, filename)
|
|
except (FileNotFoundError, ValueError):
|
|
return match.group("url")
|
|
digest = hashlib.sha256(token.encode("utf-8")).hexdigest()[:12]
|
|
safe_name = Path(filename).name
|
|
archive_path = _HTML_IMAGES_ARCHIVE_ROOT / f"{digest}-{safe_name}"
|
|
_add_file(archive_files, archive_path, source)
|
|
bundled_assets.add(archive_path.as_posix())
|
|
return archive_path.as_posix()
|
|
|
|
html = _HTML_IMAGE_URL_RE.sub(replace_html_image, html)
|
|
|
|
external_url_set = {
|
|
match.group(0).rstrip("),.;")
|
|
for match in _EXTERNAL_URL_RE.finditer(html)
|
|
}
|
|
if uses_leaflet_map and str(map_tile_url).lower().startswith(("http://", "https://")):
|
|
external_url_set.add(str(map_tile_url))
|
|
external_urls = sorted(external_url_set)
|
|
manifest = {
|
|
"format": "deerflow-html-offline-bundle-v1",
|
|
"entry": "index.html",
|
|
"source_name": html_path.name,
|
|
"bundled_assets": sorted(bundled_assets),
|
|
"external_urls": external_urls,
|
|
}
|
|
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
with zipfile.ZipFile(destination, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=6) as archive:
|
|
archive.writestr("index.html", html.encode("utf-8"))
|
|
archive.writestr(
|
|
"manifest.json",
|
|
json.dumps(manifest, ensure_ascii=False, indent=2).encode("utf-8"),
|
|
)
|
|
archive.writestr("README.txt", _readme(html_path.name, external_urls).encode("utf-8"))
|
|
for archive_path, source in sorted(archive_files.items()):
|
|
archive.write(source, archive_path)
|
|
for archive_path, content in sorted(generated_files.items()):
|
|
archive.writestr(archive_path, content.encode("utf-8"))
|
|
for license_name in sorted(licenses):
|
|
archive.write(runtime_dir / license_name, f"THIRD_PARTY_LICENSES/{license_name}")
|
|
|
|
return OfflineBundleResult(
|
|
bundled_assets=tuple(sorted(bundled_assets)),
|
|
external_urls=tuple(external_urls),
|
|
)
|