Add offline archive site with media thumbnails and player

This commit is contained in:
thePR0M3TH3AN committed 2026-04-02 11:48:59 -04:00
1 parent 5a8a36b060
commit cbbb7c78f6
11 files changed
+1410 -7

No files matched your search

+2 -1
View File
@@ -27,6 +27,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
- **Download limit** - Limit downloads to most recent N items per channel (default: 10) - **Download limit** - Limit downloads to most recent N items per channel (default: 10)
- **Interactive setup script** - Automated installation with user prompts - **Interactive setup script** - Automated installation with user prompts
- **Uninstall script** - Easy removal of the application - **Uninstall script** - Easy removal of the application
- **Offline archive site builder** - Generates a static local website from the channel/claim/version folder tree
- **Metadata/image backfill for skipped items** - Existing downloads can fill in missing JSON sidecars and thumbnails without re-downloading the file
### Technical Details ### Technical Details
- JSON-RPC client for lbrynet daemon - JSON-RPC client for lbrynet daemon
@@ -49,6 +51,5 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
### Under Consideration ### Under Consideration
- GUI interface - GUI interface
- Docker containerization - Docker containerization
- Automatic thumbnail/metadata download
- Video format selection - Video format selection
- Bandwidth limiting - Bandwidth limiting
+7
View File
@@ -23,6 +23,8 @@ For most users:
- **Flexible Input**: Accepts Odysee URLs, LBRY URIs, or claim IDs - **Flexible Input**: Accepts Odysee URLs, LBRY URIs, or claim IDs
- **Audit Trail**: Maintains run history in JSONL format - **Audit Trail**: Maintains run history in JSONL format
- **Dry-Run Mode**: Test what would happen without downloading - **Dry-Run Mode**: Test what would happen without downloading
- **Offline Archive Site**: Builds a static browseable website from the download tree, including channel/claim metadata and local file links
- **Metadata Backfill**: Skipped existing downloads still get missing `channel.json`, `claim.json`, `metadata.json`, and image assets when available
## Installation ## Installation
@@ -116,6 +118,9 @@ general:
base_dir: "~/Documents/lbry-downloads" base_dir: "~/Documents/lbry-downloads"
max_workers: 2 max_workers: 2
dry_run: false dry_run: false
build_offline_site: true
offline_site_dir: "~/Documents/lbry-downloads/site"
fetch_missing_metadata_assets: true
include_reposts: false include_reposts: false
direct_base_urls: direct_base_urls:
- "https://odysee.com" - "https://odysee.com"
@@ -132,6 +137,8 @@ channels:
content_mode: "non_video_only" content_mode: "non_video_only"
``` ```
When `build_offline_site` is enabled, the downloader writes a generated static site under `offline_site_dir` after each non-dry run. The site reflects the existing archive structure and links back to the real downloaded files under each channel/claim/version folder.
### Finding Channel URLs ### Finding Channel URLs
**Yes, you can get channel info from Odysee!** Here's how: **Yes, you can get channel info from Odysee!** Here's how:
+6
View File
@@ -45,6 +45,12 @@ general:
# If true, direct-mode runs will try the local lbrynet daemon as a fallback # If true, direct-mode runs will try the local lbrynet daemon as a fallback
# after repeated rate-limit failures. Requires a working local daemon. # after repeated rate-limit failures. Requires a working local daemon.
direct_auto_fallback_to_p2p: false direct_auto_fallback_to_p2p: false
# Build a static browseable offline site from the archive after each run.
build_offline_site: true
# Output directory for the generated static site.
offline_site_dir: "~/Documents/lbry-downloads/site"
# Fetch missing channel/claim images and sidecar metadata on reruns.
fetch_missing_metadata_assets: true
# Download limit: Maximum number of downloads per channel per run # Download limit: Maximum number of downloads per channel per run
# #
+25
View File
@@ -120,11 +120,27 @@ def _parse_config(raw: Dict[str, Any]) -> Config:
"direct_auto_fallback_to_p2p", "direct_auto_fallback_to_p2p",
config.general.direct_auto_fallback_to_p2p, config.general.direct_auto_fallback_to_p2p,
), ),
build_offline_site=general_raw.get(
"build_offline_site", config.general.build_offline_site
),
offline_site_dir=general_raw.get(
"offline_site_dir", config.general.offline_site_dir
),
fetch_missing_metadata_assets=general_raw.get(
"fetch_missing_metadata_assets",
config.general.fetch_missing_metadata_assets,
),
) )
# Validate and expand paths # Validate and expand paths
config.general.base_dir = str(expand_path(config.general.base_dir)) config.general.base_dir = str(expand_path(config.general.base_dir))
config.general.state_file = str(expand_path(config.general.state_file)) config.general.state_file = str(expand_path(config.general.state_file))
if config.general.offline_site_dir:
config.general.offline_site_dir = str(expand_path(config.general.offline_site_dir))
else:
config.general.offline_site_dir = str(
expand_path(f"{config.general.base_dir}/site")
)
# Validate max_workers # Validate max_workers
if config.general.max_workers < 1: if config.general.max_workers < 1:
@@ -271,6 +287,12 @@ general:
direct_retry_backoff_seconds: 2.0 direct_retry_backoff_seconds: 2.0
# If true, direct-mode runs will try the local node after repeated 429s # If true, direct-mode runs will try the local node after repeated 429s
direct_auto_fallback_to_p2p: false direct_auto_fallback_to_p2p: false
# Build a static offline browseable site from the archive after each run
build_offline_site: true
# Where to write the generated offline site
offline_site_dir: "~/Documents/lbry-downloads/site"
# Fetch missing channel/claim images and metadata assets on reruns
fetch_missing_metadata_assets: true
# Download limit: Number of most recent matching downloads per channel (0 = all) # Download limit: Number of most recent matching downloads per channel (0 = all)
download_limit: 10 download_limit: 10
@@ -304,3 +326,6 @@ def ensure_directories(config: Config) -> None:
# Create channels directory # Create channels directory
channels_path = base_path / "channels" channels_path = base_path / "channels"
channels_path.mkdir(exist_ok=True) channels_path.mkdir(exist_ok=True)
if config.general.build_offline_site:
Path(config.general.offline_site_dir).mkdir(parents=True, exist_ok=True)
+32 -6
View File
@@ -17,10 +17,12 @@ from downloader import DownloadError, Downloader
from direct_downloader import DirectDownloadError from direct_downloader import DirectDownloadError
from lbry_client import LbryClient, LbryClientError, check_daemon_health from lbry_client import LbryClient, LbryClientError, check_daemon_health
from models import Channel, Config, DownloadAction from models import Channel, Config, DownloadAction
from offline_site import ArchiveSiteManager
from planner import Planner from planner import Planner
from state_db import StateDb from state_db import StateDb
from utils import ( from utils import (
create_channel_folder_name, create_channel_folder_name,
extract_channel_info,
expand_path, expand_path,
format_summary, format_summary,
normalize_odysee_url, normalize_odysee_url,
@@ -124,6 +126,7 @@ def resolve_channel(
folder = create_channel_folder_name(channel_name, channel_claim_id) folder = create_channel_folder_name(channel_name, channel_claim_id)
value = claim.get("value", {}) value = claim.get("value", {})
channel_info = extract_channel_info(claim)
channel = Channel( channel = Channel(
input=channel_input, input=channel_input,
@@ -131,9 +134,18 @@ def resolve_channel(
channel_claim_id=channel_claim_id, channel_claim_id=channel_claim_id,
channel_name=channel_name, channel_name=channel_name,
folder=folder, folder=folder,
display_name=value.get("title"), display_name=channel_info.get("title"),
permanent_url=claim.get("permanent_url"), permanent_url=channel_info.get("permanent_url"),
short_url=claim.get("short_url"), short_url=channel_info.get("short_url"),
canonical_url=channel_info.get("canonical_url"),
description=channel_info.get("description"),
thumbnail_url=channel_info.get("thumbnail_url"),
cover_url=channel_info.get("cover_url"),
website_url=channel_info.get("website_url"),
email=channel_info.get("email"),
tags=channel_info.get("tags", []),
languages=channel_info.get("languages", []),
links=channel_info.get("links", []),
) )
logger.info(f"Resolved channel: {channel_name} ({channel_claim_id})") logger.info(f"Resolved channel: {channel_name} ({channel_claim_id})")
@@ -309,6 +321,7 @@ def run_sync(
# Initialize planner and downloader # Initialize planner and downloader
base_path = Path(config.general.base_dir) base_path = Path(config.general.base_dir)
planner = Planner(config, state_db.data, str(base_path)) planner = Planner(config, state_db.data, str(base_path))
site_manager = ArchiveSiteManager(config, base_path)
if direct: if direct:
logger.info("Using DIRECT mode (Odysee proxy for metadata, Odysee CDN for downloads)") logger.info("Using DIRECT mode (Odysee proxy for metadata, Odysee CDN for downloads)")
@@ -359,6 +372,9 @@ def run_sync(
if existing.download_path: if existing.download_path:
channel.download_path = existing.download_path channel.download_path = existing.download_path
if not dry_run:
site_manager.ensure_channel_artifacts(channel)
# Enumerate claims # Enumerate claims
claims = enumerate_channel_claims(client, channel, config) claims = enumerate_channel_claims(client, channel, config)
@@ -432,9 +448,12 @@ def run_sync(
try: try:
success = downloader.download(action, dry_run=dry_run) success = downloader.download(action, dry_run=dry_run)
if success and not dry_run and action.action not in ("skip_existing",): if success and not dry_run:
# Update state site_manager.ensure_claim_artifacts(channel, action)
planner.update_state_from_action(action, success=True)
if action.action not in ("skip_existing",):
# Update state only after metadata/artifacts are written
planner.update_state_from_action(action, success=True)
# Log to history # Log to history
history_file = base_path / "state" / "run-history.jsonl" history_file = base_path / "state" / "run-history.jsonl"
@@ -522,11 +541,18 @@ def run_sync(
channel.last_scan = datetime.now(timezone.utc).isoformat() channel.last_scan = datetime.now(timezone.utc).isoformat()
state_db.set_channel(channel) state_db.set_channel(channel)
if not dry_run:
site_manager.ensure_channel_artifacts(channel)
# Save state # Save state
if not dry_run: if not dry_run:
state_db.update_last_run() state_db.update_last_run()
state_db.save() state_db.save()
if config.general.build_offline_site:
try:
site_manager.build_site(channel_index=state_db.data.channels)
except Exception as exc:
logger.exception("Offline site build failed: %s", exc)
logger.info("State saved") logger.info("State saved")
return stats return stats
+30
View File
@@ -41,6 +41,9 @@ class GeneralConfig:
direct_max_retries_per_url: int = 2 direct_max_retries_per_url: int = 2
direct_retry_backoff_seconds: float = 2.0 direct_retry_backoff_seconds: float = 2.0
direct_auto_fallback_to_p2p: bool = False direct_auto_fallback_to_p2p: bool = False
build_offline_site: bool = True
offline_site_dir: Optional[str] = None
fetch_missing_metadata_assets: bool = True
@dataclass @dataclass
@@ -73,6 +76,15 @@ class Channel:
display_name: Optional[str] = None display_name: Optional[str] = None
permanent_url: Optional[str] = None permanent_url: Optional[str] = None
short_url: Optional[str] = None short_url: Optional[str] = None
canonical_url: Optional[str] = None
description: Optional[str] = None
thumbnail_url: Optional[str] = None
cover_url: Optional[str] = None
website_url: Optional[str] = None
email: Optional[str] = None
tags: List[str] = field(default_factory=list)
languages: List[str] = field(default_factory=list)
links: List[Dict[str, Any]] = field(default_factory=list)
last_scan: Optional[str] = None last_scan: Optional[str] = None
def to_dict(self) -> Dict[str, Any]: def to_dict(self) -> Dict[str, Any]:
@@ -86,6 +98,15 @@ class Channel:
"display_name": self.display_name, "display_name": self.display_name,
"permanent_url": self.permanent_url, "permanent_url": self.permanent_url,
"short_url": self.short_url, "short_url": self.short_url,
"canonical_url": self.canonical_url,
"description": self.description,
"thumbnail_url": self.thumbnail_url,
"cover_url": self.cover_url,
"website_url": self.website_url,
"email": self.email,
"tags": self.tags,
"languages": self.languages,
"links": self.links,
"last_scan": self.last_scan, "last_scan": self.last_scan,
} }
@@ -101,6 +122,15 @@ class Channel:
display_name=data.get("display_name"), display_name=data.get("display_name"),
permanent_url=data.get("permanent_url"), permanent_url=data.get("permanent_url"),
short_url=data.get("short_url"), short_url=data.get("short_url"),
canonical_url=data.get("canonical_url"),
description=data.get("description"),
thumbnail_url=data.get("thumbnail_url"),
cover_url=data.get("cover_url"),
website_url=data.get("website_url"),
email=data.get("email"),
tags=data.get("tags", []),
languages=data.get("languages", []),
links=data.get("links", []),
last_scan=data.get("last_scan"), last_scan=data.get("last_scan"),
) )
+894
View File
@@ -0,0 +1,894 @@
"""Metadata backfill and offline site generation for the archive."""
import html
import json
import logging
import mimetypes
import os
import shutil
import subprocess
import tempfile
from pathlib import Path
from typing import Any, Dict, List, Optional
from urllib.parse import urlparse
try:
import requests
except ImportError:
requests = None
from models import Channel, Config, DownloadAction
logger = logging.getLogger(__name__)
class ArchiveSiteManager:
"""Maintains archive metadata sidecars and builds a static offline site."""
def __init__(self, config: Config, base_dir: Path):
self.config = config
self.base_dir = base_dir
self.channels_root = self.base_dir / "channels"
self.site_root = Path(config.general.offline_site_dir)
self.session = requests.Session() if requests else None
if self.session:
self.session.headers.update(
{
"User-Agent": (
"Mozilla/5.0 (X11; Linux x86_64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/123.0.0.0 Safari/537.36"
)
}
)
def ensure_channel_artifacts(self, channel: Channel) -> None:
"""Write channel metadata and fetch any missing channel assets."""
channel_dir = self._channel_dir(channel)
channel_dir.mkdir(parents=True, exist_ok=True)
channel_payload = {
"channel_claim_id": channel.channel_claim_id,
"channel_name": channel.channel_name,
"display_name": channel.display_name,
"input": channel.input,
"normalized_uri": channel.normalized_uri,
"permanent_url": channel.permanent_url,
"short_url": channel.short_url,
"canonical_url": channel.canonical_url,
"description": channel.description,
"thumbnail_url": channel.thumbnail_url,
"cover_url": channel.cover_url,
"website_url": channel.website_url,
"email": channel.email,
"tags": channel.tags,
"languages": channel.languages,
"links": channel.links,
"folder": channel.folder,
"download_path": channel.download_path,
"last_scan": channel.last_scan,
}
self._write_json(channel_dir / "channel.json", channel_payload)
if self.config.general.fetch_missing_metadata_assets:
if channel.thumbnail_url:
self._ensure_remote_asset(
channel.thumbnail_url,
channel_dir,
basename="channel-profile",
)
if channel.cover_url:
self._ensure_remote_asset(
channel.cover_url,
channel_dir,
basename="channel-cover",
)
def ensure_claim_artifacts(self, channel: Channel, action: DownloadAction) -> None:
"""Write/update claim- and version-level metadata for a claim."""
version_dir = Path(action.target_dir)
claim_dir = version_dir.parent.parent
claim_dir.mkdir(parents=True, exist_ok=True)
version_dir.mkdir(parents=True, exist_ok=True)
file_path = self._resolve_action_file_path(action, version_dir)
file_size = file_path.stat().st_size if file_path and file_path.exists() else None
claim_payload = {
"claim_id": action.claim_id,
"channel_claim_id": action.channel_claim_id,
"name": action.claim_name,
"title": action.metadata.get("title"),
"permanent_url": action.uri,
"short_url": action.metadata.get("short_url"),
"canonical_url": action.metadata.get("canonical_url"),
"description": action.metadata.get("description"),
"thumbnail_url": action.metadata.get("thumbnail_url"),
"tags": action.metadata.get("tags", []),
"languages": action.metadata.get("languages", []),
"links": action.metadata.get("links", []),
"release_time": action.metadata.get("release_time"),
"timestamp": action.metadata.get("timestamp"),
"stream_type": action.metadata.get("stream_type"),
"source_media_type": action.metadata.get("source_media_type"),
"source_name": action.metadata.get("source_name"),
"claim_folder": self._path_label(claim_dir),
"latest_version_token": action.version_token,
"channel_folder": channel.folder,
}
self._write_json(claim_dir / "claim.json", claim_payload)
version_payload = {
"claim_id": action.claim_id,
"channel_claim_id": action.channel_claim_id,
"name": action.claim_name,
"title": action.metadata.get("title"),
"description": action.metadata.get("description"),
"version_token": action.version_token,
"permanent_url": action.uri,
"short_url": action.metadata.get("short_url"),
"canonical_url": action.metadata.get("canonical_url"),
"download_source": self._download_source_label(action),
"file_name": file_path.name if file_path else action.metadata.get("file_name"),
"file_size": file_size,
"stream_type": action.metadata.get("stream_type"),
"source_media_type": action.metadata.get("source_media_type"),
"thumbnail_url": action.metadata.get("thumbnail_url"),
"tags": action.metadata.get("tags", []),
"languages": action.metadata.get("languages", []),
"links": action.metadata.get("links", []),
"release_time": action.metadata.get("release_time"),
"timestamp": action.metadata.get("timestamp"),
}
self._write_json(version_dir / "metadata.json", version_payload)
if action.metadata.get("thumbnail_url") and self.config.general.fetch_missing_metadata_assets:
self._ensure_remote_asset(
action.metadata["thumbnail_url"],
claim_dir,
basename="claim-thumbnail",
)
def build_site(self, channel_index: Optional[Dict[str, Channel]] = None) -> None:
"""Build a static offline site from the archive metadata."""
self.site_root.mkdir(parents=True, exist_ok=True)
self._write_text(self.site_root / "assets" / "style.css", self._site_css())
channels = self._collect_channel_records(channel_index=channel_index)
self._write_text(
self.site_root / "index.html",
self._render_index_page(channels),
)
channels_site_root = self.site_root / "channels"
for channel in channels:
channel_page_dir = channels_site_root / channel["folder"]
channel_page_dir.mkdir(parents=True, exist_ok=True)
self._write_text(
channel_page_dir / "index.html",
self._render_channel_page(channel, channel_page_dir),
)
for claim in channel["claims"]:
claim_page_dir = channel_page_dir / "claims" / claim["folder"]
claim_page_dir.mkdir(parents=True, exist_ok=True)
self._write_text(
claim_page_dir / "index.html",
self._render_claim_page(channel, claim, claim_page_dir),
)
def _channel_dir(self, channel: Channel) -> Path:
if channel.download_path:
return Path(channel.download_path) / channel.folder
return self.channels_root / channel.folder
def _resolve_action_file_path(
self, action: DownloadAction, version_dir: Path
) -> Optional[Path]:
local_file_path = action.metadata.get("local_file_path")
if local_file_path:
path = Path(local_file_path)
if not path.is_absolute():
path = self.base_dir / path
if path.exists():
return path
existing_relpath = action.metadata.get("existing_file_relpath")
if existing_relpath:
path = self.base_dir / existing_relpath
if path.exists():
return path
source_name = action.metadata.get("source_name") or action.metadata.get("file_name")
if source_name:
path = version_dir / source_name
if path.exists():
return path
candidates = [
item
for item in version_dir.iterdir()
if item.is_file() and item.name not in {"metadata.json", "download.json", "checksums.txt"}
] if version_dir.exists() else []
return candidates[0] if candidates else None
def _download_source_label(self, action: DownloadAction) -> str:
media_type = (action.metadata.get("source_media_type") or "").lower()
if media_type.startswith("video/") or media_type.startswith("audio/"):
return "odysee_stream"
return "odysee_download"
def _ensure_remote_asset(self, url: str, target_dir: Path, basename: str) -> Optional[Path]:
if not self.session:
return None
existing = self._existing_asset(target_dir, basename)
if existing:
return existing[0]
try:
response = self.session.get(url, timeout=60, stream=True)
response.raise_for_status()
with tempfile.NamedTemporaryFile(dir=target_dir, delete=False) as handle:
shutil.copyfileobj(response.raw, handle)
temp_path = Path(handle.name)
suffix = self._suffix_for_response(url, response.headers.get("Content-Type"), temp_path)
target_path = target_dir / f"{basename}{suffix}"
temp_path.replace(target_path)
return target_path
except Exception as exc:
logger.warning("Failed to fetch metadata asset %s: %s", url, exc)
temp_path = locals().get("temp_path")
if temp_path and Path(temp_path).exists():
Path(temp_path).unlink()
return None
def _suffix_for_url(self, url: str) -> str:
parsed = urlparse(url)
suffix = Path(parsed.path).suffix
if suffix:
return suffix
mime_guess, _ = mimetypes.guess_type(url)
return mimetypes.guess_extension(mime_guess or "") or ".bin"
def _suffix_for_response(
self, url: str, content_type: Optional[str], downloaded_path: Path
) -> str:
suffix = self._suffix_for_content_type(content_type)
if suffix:
return suffix
suffix = self._suffix_for_image_bytes(downloaded_path)
if suffix:
return suffix
return self._suffix_for_url(url)
def _suffix_for_content_type(self, content_type: Optional[str]) -> Optional[str]:
if not content_type:
return None
mime = content_type.split(";", 1)[0].strip().lower()
if mime == "image/jpeg":
return ".jpg"
if mime == "image/png":
return ".png"
if mime == "image/webp":
return ".webp"
guessed = mimetypes.guess_extension(mime)
return guessed
def _suffix_for_image_bytes(self, path: Path) -> Optional[str]:
try:
header = path.read_bytes()[:16]
except OSError:
return None
if header.startswith(b"\x89PNG\r\n\x1a\n"):
return ".png"
if header.startswith(b"\xff\xd8\xff"):
return ".jpg"
if header.startswith(b"RIFF") and header[8:12] == b"WEBP":
return ".webp"
return None
def _existing_asset(self, directory: Path, basename: str) -> List[Path]:
normalized = []
for asset_path in sorted(directory.glob(f"{basename}.*")):
normalized.append(self._normalize_asset_extension(asset_path))
return normalized
def _normalize_asset_extension(self, path: Path) -> Path:
actual_suffix = self._suffix_for_image_bytes(path)
if not actual_suffix or path.suffix.lower() == actual_suffix:
return path
corrected_path = path.with_suffix(actual_suffix)
if corrected_path.exists():
path.unlink()
return corrected_path
path.rename(corrected_path)
return corrected_path
def _collect_channel_records(
self, channel_index: Optional[Dict[str, Channel]] = None
) -> List[Dict[str, Any]]:
channel_dirs: Dict[Path, Path] = {}
if self.channels_root.exists():
for channel_dir in self.channels_root.iterdir():
if channel_dir.is_dir():
channel_dirs[channel_dir.resolve()] = channel_dir
for channel in (channel_index or {}).values():
channel_dir = self._channel_dir(channel)
if channel_dir.exists():
channel_dirs[channel_dir.resolve()] = channel_dir
channels = []
for channel_dir in sorted(channel_dirs.values()):
channel_data = self._read_json(channel_dir / "channel.json") or {
"folder": channel_dir.name,
"channel_name": channel_dir.name,
"display_name": channel_dir.name,
"description": None,
"links": [],
}
channel_data["folder"] = channel_dir.name
channel_data["path"] = channel_dir
channel_data["profile_asset"] = self._find_asset(channel_dir, "channel-profile")
channel_data["cover_asset"] = self._find_asset(channel_dir, "channel-cover")
claims = self._collect_claim_records(channel_dir)
channel_data["claims"] = claims
channel_data["claim_count"] = len(claims)
channel_data["version_count"] = sum(len(claim["versions"]) for claim in claims)
channels.append(channel_data)
channels.sort(
key=lambda item: (item.get("display_name") or item.get("channel_name") or "").lower()
)
channels.sort(key=lambda item: item.get("last_scan") or "", reverse=True)
return channels
def _collect_claim_records(self, channel_dir: Path) -> List[Dict[str, Any]]:
claims_root = channel_dir / "claims"
claims = []
for claim_dir in sorted(claims_root.iterdir()) if claims_root.exists() else []:
if not claim_dir.is_dir():
continue
claim_data = self._read_json(claim_dir / "claim.json") or {
"name": claim_dir.name,
"title": claim_dir.name,
"description": None,
"links": [],
}
versions = self._collect_versions(claim_dir)
latest_version = versions[0] if versions else {}
claim_data["folder"] = claim_dir.name
claim_data["path"] = claim_dir
claim_data["thumbnail_asset"] = self._find_asset(claim_dir, "claim-thumbnail")
if not claim_data["thumbnail_asset"]:
claim_data["thumbnail_asset"] = self._extract_video_thumbnail(claim_dir, versions)
claim_data["thumbnail_url"] = claim_data.get("thumbnail_url") or latest_version.get(
"thumbnail_url"
)
claim_data["title"] = claim_data.get("title") or latest_version.get("title")
claim_data["description"] = claim_data.get("description") or latest_version.get(
"description"
)
claim_data["release_time"] = claim_data.get("release_time") or latest_version.get(
"release_time"
)
claim_data["versions"] = versions
claims.append(claim_data)
claims.sort(
key=lambda item: (
-self._numeric_sort_value(item.get("release_time")),
(item.get("title") or item.get("name") or "").lower(),
)
)
return claims
def _collect_versions(self, claim_dir: Path) -> List[Dict[str, Any]]:
versions_root = claim_dir / "versions"
versions = []
for version_dir in sorted(versions_root.iterdir()) if versions_root.exists() else []:
if not version_dir.is_dir():
continue
version_data = self._read_json(version_dir / "metadata.json") or {
"version_token": version_dir.name
}
version_data["path"] = version_dir
version_data["version_token"] = version_data.get("version_token") or version_dir.name
file_path = self._find_primary_file(version_dir)
version_data["file_path"] = file_path
versions.append(version_data)
versions.sort(
key=lambda item: (
-self._numeric_sort_value(item.get("release_time")),
item["version_token"],
)
)
return versions
def _extract_video_thumbnail(
self, claim_dir: Path, versions: List[Dict[str, Any]]
) -> Optional[Path]:
for version in versions:
file_path = version.get("file_path")
if not file_path or not self._is_video_version(version, file_path):
continue
target_path = claim_dir / "claim-thumbnail.jpg"
if self._run_ffmpeg_thumbnail(file_path, target_path):
return target_path
return None
def _is_video_version(self, version: Dict[str, Any], file_path: Path) -> bool:
media_type = self._version_media_type(version, file_path).lower()
return media_type.startswith("video/")
def _run_ffmpeg_thumbnail(self, input_path: Path, output_path: Path) -> bool:
cmd = [
"ffmpeg",
"-y",
"-ss",
"1",
"-i",
str(input_path),
"-frames:v",
"1",
"-vf",
"scale='min(960,iw)':-2",
str(output_path),
]
try:
subprocess.run(
cmd,
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
return output_path.exists() and output_path.stat().st_size > 0
except (OSError, subprocess.CalledProcessError):
if output_path.exists():
output_path.unlink()
return False
def _find_primary_file(self, version_dir: Path) -> Optional[Path]:
metadata_name = {"metadata.json", "download.json", "checksums.txt"}
files = [item for item in version_dir.iterdir() if item.is_file() and item.name not in metadata_name]
return files[0] if files else None
def _find_asset(self, directory: Path, basename: str) -> Optional[Path]:
matches = self._existing_asset(directory, basename)
return matches[0] if matches else None
def _numeric_sort_value(self, value: Any) -> int:
"""Convert mixed metadata values into a safe descending sort key."""
if value is None or value == "":
return 0
if isinstance(value, bool):
return int(value)
if isinstance(value, (int, float)):
return int(value)
if isinstance(value, str):
stripped = value.strip()
if not stripped:
return 0
try:
return int(stripped)
except ValueError:
try:
return int(float(stripped))
except ValueError:
logger.debug("Non-numeric release_time value in archive metadata: %r", value)
return 0
return 0
def _render_index_page(self, channels: List[Dict[str, Any]]) -> str:
cards = []
for channel in channels:
title = channel.get("display_name") or channel.get("channel_name") or channel["folder"]
description = channel.get("description") or ""
profile = self._image_tag(
channel.get("profile_asset"),
self.site_root,
css_class="avatar",
mirror_key=f"channels/{channel['folder']}/channel-profile",
)
link = f"channels/{channel['folder']}/index.html"
cards.append(
f"""
<article class="card channel-card">
<a class="card-link" href="{html.escape(link)}">
{profile}
<div class="card-body">
<h2>{html.escape(title)}</h2>
<p class="muted">{html.escape(channel.get('channel_name') or '')}</p>
<p>{html.escape(self._truncate(description, 220))}</p>
<p class="meta">{channel['claim_count']} claims | {channel['version_count']} versions</p>
</div>
</a>
</article>
"""
)
body = "\n".join(cards) or "<p>No channels found.</p>"
return self._page(
title="LBRY Archive",
body=f"""
<header class="hero">
<div>
<p class="eyebrow">Offline Archive</p>
<h1>LBRY / Odysee Archive</h1>
<p class="lede">Browse downloaded channels, claims, metadata, and local files without needing the live site.</p>
</div>
</header>
<section class="grid">{body}</section>
""",
root=self.site_root,
)
def _render_channel_page(self, channel: Dict[str, Any], page_dir: Path) -> str:
title = channel.get("display_name") or channel.get("channel_name") or channel["folder"]
cover = self._image_tag(
channel.get("cover_asset"),
page_dir,
css_class="cover",
mirror_key=f"channels/{channel['folder']}/channel-cover",
)
profile = self._image_tag(
channel.get("profile_asset"),
page_dir,
css_class="avatar large",
mirror_key=f"channels/{channel['folder']}/channel-profile",
)
claims_html = []
for claim in channel["claims"]:
claim_title = claim.get("title") or claim.get("name") or claim["folder"]
thumb = self._image_tag(
claim.get("thumbnail_asset"),
page_dir,
css_class="thumb",
fallback_url=claim.get("thumbnail_url"),
mirror_key=f"channels/{channel['folder']}/claims/{claim['folder']}/claim-thumbnail",
)
link = f"claims/{claim['folder']}/index.html"
claims_html.append(
f"""
<article class="card claim-card">
<a class="card-link" href="{html.escape(link)}">
{thumb}
<div class="card-body">
<h2>{html.escape(claim_title)}</h2>
<p>{html.escape(self._truncate(claim.get('description') or '', 180))}</p>
<p class="meta">{len(claim['versions'])} version(s)</p>
</div>
</a>
</article>
"""
)
links = self._render_links(channel.get("links", []))
return self._page(
title=title,
body=f"""
<nav class="breadcrumbs"><a href="../../index.html">Archive</a></nav>
{cover}
<header class="channel-header">
{profile}
<div>
<p class="eyebrow">Channel</p>
<h1>{html.escape(title)}</h1>
<p class="muted">{html.escape(channel.get('channel_name') or '')}</p>
<p>{html.escape(channel.get('description') or '')}</p>
{links}
</div>
</header>
<section class="grid">{''.join(claims_html) or '<p>No claims found.</p>'}</section>
""",
root=page_dir,
)
def _render_claim_page(self, channel: Dict[str, Any], claim: Dict[str, Any], page_dir: Path) -> str:
title = claim.get("title") or claim.get("name") or claim["folder"]
thumb = self._image_tag(
claim.get("thumbnail_asset"),
page_dir,
css_class="hero-thumb",
fallback_url=claim.get("thumbnail_url"),
mirror_key=f"channels/{channel['folder']}/claims/{claim['folder']}/claim-thumbnail",
)
versions = []
for version in claim["versions"]:
file_link = ""
media_preview = ""
if version.get("file_path"):
href = self._rel_href(page_dir, version["file_path"])
file_name = version["file_path"].name
file_link = f'<p><a href="{html.escape(href)}">Open local file: {html.escape(file_name)}</a></p>'
media_preview = self._render_version_preview(version, page_dir, href)
versions.append(
f"""
<article class="version">
<h2>{html.escape(version.get('version_token') or 'version')}</h2>
<p class="meta">{html.escape(version.get('source_media_type') or version.get('stream_type') or '')}</p>
<p>{html.escape(version.get('description') or '')}</p>
{media_preview}
{file_link}
</article>
"""
)
links = self._render_links(claim.get("links", []))
return self._page(
title=title,
body=f"""
<nav class="breadcrumbs">
<a href="../../../../index.html">Archive</a>
<a href="../../index.html">{html.escape(channel.get('display_name') or channel.get('channel_name') or channel['folder'])}</a>
</nav>
<header class="claim-header">
{thumb}
<div>
<p class="eyebrow">Claim</p>
<h1>{html.escape(title)}</h1>
<p>{html.escape(claim.get('description') or '')}</p>
{links}
</div>
</header>
<section class="versions">{''.join(versions) or '<p>No versions found.</p>'}</section>
""",
root=page_dir,
)
def _image_tag(
self,
image_path: Optional[Path],
page_dir: Path,
css_class: str,
fallback_url: Optional[str] = None,
mirror_key: Optional[str] = None,
) -> str:
if image_path:
site_path = self._mirror_site_asset(image_path, mirror_key)
href = self._rel_href(page_dir, site_path)
elif fallback_url:
href = fallback_url
else:
return ""
return f'<img class="{html.escape(css_class)}" src="{html.escape(href)}" alt="" loading="lazy">'
def _render_version_preview(self, version: Dict[str, Any], page_dir: Path, href: str) -> str:
file_path = version.get("file_path")
if not file_path:
return ""
media_type = self._version_media_type(version, file_path)
if media_type.startswith("video/"):
poster = self._poster_attribute(version, page_dir)
return (
f'<video class="media-player" controls preload="metadata"{poster}>'
f'<source src="{html.escape(href)}" type="{html.escape(media_type)}">'
"Your browser does not support embedded video playback."
"</video>"
)
if media_type.startswith("audio/"):
return (
f'<audio class="audio-player" controls preload="metadata">'
f'<source src="{html.escape(href)}" type="{html.escape(media_type)}">'
"Your browser does not support embedded audio playback."
"</audio>"
)
return ""
def _version_media_type(self, version: Dict[str, Any], file_path: Path) -> str:
media_type = version.get("source_media_type")
if media_type:
return str(media_type)
guessed, _ = mimetypes.guess_type(file_path.name)
return guessed or ""
def _poster_attribute(self, version: Dict[str, Any], page_dir: Path) -> str:
claim_dir = Path(version["path"]).parent.parent
thumbnail_asset = self._find_asset(claim_dir, "claim-thumbnail")
if thumbnail_asset:
claim_folder = claim_dir.name
channel_folder = claim_dir.parent.parent.name
site_path = self._mirror_site_asset(
thumbnail_asset,
f"channels/{channel_folder}/claims/{claim_folder}/claim-thumbnail",
)
return f' poster="{html.escape(self._rel_href(page_dir, site_path))}"'
thumbnail_url = version.get("thumbnail_url")
if thumbnail_url:
return f' poster="{html.escape(thumbnail_url)}"'
return ""
def _mirror_site_asset(self, source_path: Path, mirror_key: Optional[str] = None) -> Path:
source = Path(source_path)
if mirror_key:
target = self.site_root / "_assets" / f"{mirror_key}{source.suffix.lower()}"
else:
target = self.site_root / "_assets" / source.name
target.parent.mkdir(parents=True, exist_ok=True)
if not target.exists():
shutil.copy2(source, target)
return target
def _render_links(self, links: List[Dict[str, str]]) -> str:
if not links:
return ""
rendered = []
for link in links:
label = link.get("label") or link.get("url") or "Link"
url = link.get("url")
if not url:
continue
rendered.append(
f'<a class="pill" href="{html.escape(url)}">{html.escape(label)}</a>'
)
if not rendered:
return ""
return f'<div class="pill-row">{"".join(rendered)}</div>'
def _page(self, *, title: str, body: str, root: Path) -> str:
stylesheet = self._rel_href(root, self.site_root / "assets" / "style.css")
return f"""<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{html.escape(title)}</title>
<link rel="stylesheet" href="{html.escape(stylesheet)}">
</head>
<body>
<main class="shell">
{body}
</main>
</body>
</html>
"""
def _rel_href(self, page_dir: Path, target: Path) -> str:
return os.path.relpath(Path(target).resolve(), start=Path(page_dir).resolve())
def _path_label(self, path: Path) -> str:
try:
return str(path.relative_to(self.base_dir))
except ValueError:
return str(path)
def _write_json(self, path: Path, payload: Dict[str, Any]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w", encoding="utf-8") as handle:
json.dump(payload, handle, indent=2, sort_keys=True)
def _read_json(self, path: Path) -> Optional[Dict[str, Any]]:
if not path.exists():
return None
with open(path, "r", encoding="utf-8") as handle:
return json.load(handle)
def _write_text(self, path: Path, content: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "w", encoding="utf-8") as handle:
handle.write(content)
def _truncate(self, text: str, limit: int) -> str:
if len(text) <= limit:
return text
return text[: limit - 3].rstrip() + "..."
def _site_css(self) -> str:
return """
:root {
--bg: #f5f1e8;
--ink: #1d1b19;
--muted: #6b6257;
--card: #fffaf1;
--line: #d9cfbf;
--accent: #9c4a1a;
}
* { box-sizing: border-box; }
body {
margin: 0;
background:
radial-gradient(circle at top, rgba(156,74,26,.14), transparent 35%),
linear-gradient(180deg, #f8f4ec 0%, var(--bg) 100%);
color: var(--ink);
font: 16px/1.5 Georgia, "Times New Roman", serif;
}
a { color: var(--accent); text-decoration: none; }
a:hover { text-decoration: underline; }
.shell { max-width: 1180px; margin: 0 auto; padding: 2rem 1.25rem 4rem; }
.hero, .channel-header, .claim-header {
display: grid;
gap: 1rem;
align-items: start;
margin-bottom: 2rem;
}
.hero {
min-height: 12rem;
border: 1px solid var(--line);
background: rgba(255,250,241,.8);
padding: 1.5rem;
}
.eyebrow, .meta, .muted, .breadcrumbs { color: var(--muted); }
.breadcrumbs { margin-bottom: 1rem; display: flex; gap: .75rem; flex-wrap: wrap; }
.grid {
display: grid;
gap: 1rem;
grid-template-columns: repeat(auto-fit, minmax(260px, 1fr));
}
.card, .version {
border: 1px solid var(--line);
background: var(--card);
}
.card-link { display: block; color: inherit; height: 100%; }
.card-body { padding: 1rem; }
.avatar {
width: 72px;
height: 72px;
object-fit: cover;
border-radius: 999px;
border: 1px solid var(--line);
margin: 1rem 1rem 0;
}
.avatar.large { width: 112px; height: 112px; margin: 0; }
.cover {
width: 100%;
max-height: 240px;
object-fit: cover;
border: 1px solid var(--line);
margin-bottom: 1rem;
}
.thumb, .hero-thumb {
width: 100%;
aspect-ratio: 16 / 9;
object-fit: cover;
border-bottom: 1px solid var(--line);
}
.hero-thumb { max-width: 420px; border: 1px solid var(--line); }
.pill-row {
display: flex;
gap: .5rem;
flex-wrap: wrap;
margin-top: .75rem;
}
.pill {
display: inline-block;
border: 1px solid var(--line);
padding: .35rem .65rem;
border-radius: 999px;
background: #fff;
}
.versions {
display: grid;
gap: 1rem;
}
.version { padding: 1rem; }
.media-player, .audio-player {
width: 100%;
margin: .75rem 0;
}
.media-player {
max-width: 100%;
aspect-ratio: 16 / 9;
background: #000;
}
@media (min-width: 860px) {
.channel-header, .claim-header {
grid-template-columns: auto 1fr;
}
}
"""
+1
View File
@@ -55,4 +55,5 @@ py-modules = [
"state_db", "state_db",
"planner", "planner",
"downloader", "downloader",
"offline_site",
] ]
+17
View File
@@ -149,6 +149,23 @@ channels:
assert config.general.direct_retry_backoff_seconds == 1.5 assert config.general.direct_retry_backoff_seconds == 1.5
assert config.general.direct_auto_fallback_to_p2p is True assert config.general.direct_auto_fallback_to_p2p is True
def test_offline_site_defaults_to_base_dir_site(self, tmp_path):
config_file = tmp_path / "config.yaml"
config_file.write_text("""
general:
base_dir: "~/test-archive"
channels:
- input: "https://odysee.com/@Test:1"
enabled: true
""")
config = load_config(str(config_file))
assert config.general.build_offline_site is True
assert config.general.fetch_missing_metadata_assets is True
assert config.general.offline_site_dir.endswith("/test-archive/site")
def test_invalid_direct_retry_backoff(self, tmp_path): def test_invalid_direct_retry_backoff(self, tmp_path):
config_file = tmp_path / "config.yaml" config_file = tmp_path / "config.yaml"
config_file.write_text(""" config_file.write_text("""
+327
View File
@@ -0,0 +1,327 @@
"""Tests for offline site generation and metadata backfill."""
import json
from models import Channel, Config, DownloadAction
from offline_site import ArchiveSiteManager
WEBP_STUB = b"RIFF\x1a\x00\x00\x00WEBPVP8 " + b"\x00" * 10
def test_offline_site_builds_from_archive_metadata(tmp_path):
config = Config()
config.general.base_dir = str(tmp_path)
config.general.offline_site_dir = str(tmp_path / "site")
config.general.fetch_missing_metadata_assets = False
manager = ArchiveSiteManager(config, tmp_path)
channel = Channel(
input="https://odysee.com/@example:1",
normalized_uri="lbry://@example#1",
channel_claim_id="chan123",
channel_name="@example",
folder="example__chan123",
display_name="Example Channel",
description="Channel description",
links=[{"label": "Website", "url": "https://example.com"}],
)
manager.ensure_channel_artifacts(channel)
version_dir = (
tmp_path
/ "channels"
/ channel.folder
/ "claims"
/ "my-claim__claim123"
/ "versions"
/ "sd_abc123"
)
version_dir.mkdir(parents=True)
file_path = version_dir / "video.mp4"
file_path.write_bytes(b"video-data")
thumbnail_path = version_dir.parent.parent / "claim-thumbnail.jpg"
thumbnail_path.write_bytes(b"thumb")
action = DownloadAction(
claim_id="claim123",
claim_name="my-claim",
channel_claim_id=channel.channel_claim_id,
version_token="sd_abc123",
action="skip_existing",
uri="lbry://@example#1/my-claim#claim123",
target_dir=str(version_dir),
metadata={
"title": "My Claim",
"description": "Claim description",
"source_media_type": "video/mp4",
"stream_type": "video",
"source_name": "video.mp4",
"local_file_path": str(file_path.relative_to(tmp_path)),
"thumbnail_url": "https://example.com/thumb.jpg",
"release_time": 1234567890,
"links": [
{
"label": "External",
"url": "https://odysee.com/@example:1/my-claim",
}
],
},
)
manager.ensure_claim_artifacts(channel, action)
manager.build_site(channel_index={channel.channel_claim_id: channel})
channel_json = tmp_path / "channels" / channel.folder / "channel.json"
claim_json = version_dir.parent.parent / "claim.json"
metadata_json = version_dir / "metadata.json"
site_index = tmp_path / "site" / "index.html"
channel_page = tmp_path / "site" / "channels" / channel.folder / "index.html"
claim_page = (
tmp_path
/ "site"
/ "channels"
/ channel.folder
/ "claims"
/ "my-claim__claim123"
/ "index.html"
)
assert channel_json.exists()
assert claim_json.exists()
assert metadata_json.exists()
assert site_index.exists()
assert channel_page.exists()
assert claim_page.exists()
channel_payload = json.loads(channel_json.read_text())
claim_payload = json.loads(claim_json.read_text())
version_payload = json.loads(metadata_json.read_text())
assert channel_payload["display_name"] == "Example Channel"
assert claim_payload["title"] == "My Claim"
assert version_payload["file_name"] == "video.mp4"
assert "Example Channel" in site_index.read_text()
channel_page_html = channel_page.read_text()
claim_page_html = claim_page.read_text()
assert "My Claim" in channel_page_html
assert 'class="thumb"' in channel_page_html
assert "Open local file: video.mp4" in claim_page_html
assert '<video class="media-player" controls preload="metadata"' in claim_page_html
assert 'poster="../../../../_assets/channels/example__chan123/claims/my-claim__claim123/claim-thumbnail.jpg"' in claim_page_html
assert '<source src="../../../../../channels/example__chan123/claims/my-claim__claim123/versions/sd_abc123/video.mp4" type="video/mp4">' in claim_page_html
def test_offline_site_handles_string_release_time(tmp_path):
config = Config()
config.general.base_dir = str(tmp_path)
config.general.offline_site_dir = str(tmp_path / "site")
config.general.fetch_missing_metadata_assets = False
manager = ArchiveSiteManager(config, tmp_path)
channel_dir = tmp_path / "channels" / "example__chan123"
claim_dir = channel_dir / "claims" / "string-time__claim123"
version_dir = claim_dir / "versions" / "sd_abc123"
version_dir.mkdir(parents=True)
(channel_dir / "channel.json").write_text(
json.dumps({"display_name": "Example", "channel_name": "@example"})
)
(claim_dir / "claim.json").write_text(
json.dumps({"title": "String Time Claim", "release_time": "1712100000"})
)
(version_dir / "metadata.json").write_text(
json.dumps({"version_token": "sd_abc123", "release_time": "1712100000"})
)
(version_dir / "file.bin").write_bytes(b"x")
manager.build_site()
assert (tmp_path / "site" / "index.html").exists()
def test_channel_page_uses_thumbnail_from_latest_version_when_claim_json_lacks_it(tmp_path):
config = Config()
config.general.base_dir = str(tmp_path)
config.general.offline_site_dir = str(tmp_path / "site")
config.general.fetch_missing_metadata_assets = False
manager = ArchiveSiteManager(config, tmp_path)
channel_dir = tmp_path / "channels" / "example__chan123"
claim_dir = channel_dir / "claims" / "video-claim__claim123"
version_dir = claim_dir / "versions" / "sd_abc123"
version_dir.mkdir(parents=True)
(channel_dir / "channel.json").write_text(
json.dumps({"display_name": "Example", "channel_name": "@example"})
)
(claim_dir / "claim.json").write_text(
json.dumps({"title": "Video Claim", "description": "Claim-level description"})
)
(version_dir / "metadata.json").write_text(
json.dumps(
{
"version_token": "sd_abc123",
"title": "Video Claim",
"description": "Version description",
"thumbnail_url": "https://example.com/version-thumb.jpg",
"release_time": 1712100000,
"source_media_type": "video/mp4",
}
)
)
(version_dir / "video.mp4").write_bytes(b"x")
manager.build_site()
channel_page = tmp_path / "site" / "channels" / "example__chan123" / "index.html"
claim_page = (
tmp_path
/ "site"
/ "channels"
/ "example__chan123"
/ "claims"
/ "video-claim__claim123"
/ "index.html"
)
channel_page_html = channel_page.read_text()
claim_page_html = claim_page.read_text()
assert 'class="thumb" src="https://example.com/version-thumb.jpg"' in channel_page_html
assert 'class="hero-thumb" src="https://example.com/version-thumb.jpg"' in claim_page_html
assert 'poster="https://example.com/version-thumb.jpg"' in claim_page_html
def test_build_site_normalizes_mislabeled_thumbnail_extensions(tmp_path):
config = Config()
config.general.base_dir = str(tmp_path)
config.general.offline_site_dir = str(tmp_path / "site")
config.general.fetch_missing_metadata_assets = False
manager = ArchiveSiteManager(config, tmp_path)
channel_dir = tmp_path / "channels" / "example__chan123"
claim_dir = channel_dir / "claims" / "video-claim__claim123"
version_dir = claim_dir / "versions" / "sd_abc123"
version_dir.mkdir(parents=True)
(channel_dir / "channel.json").write_text(
json.dumps({"display_name": "Example", "channel_name": "@example"})
)
(claim_dir / "claim.json").write_text(
json.dumps({"title": "Video Claim", "thumbnail_url": "https://example.com/thumb.jpg"})
)
(version_dir / "metadata.json").write_text(json.dumps({"version_token": "sd_abc123"}))
(version_dir / "video.mp4").write_bytes(b"x")
mislabeled = claim_dir / "claim-thumbnail.jpg"
mislabeled.write_bytes(WEBP_STUB)
manager.build_site()
corrected = claim_dir / "claim-thumbnail.webp"
channel_page = tmp_path / "site" / "channels" / "example__chan123" / "index.html"
claim_page = (
tmp_path
/ "site"
/ "channels"
/ "example__chan123"
/ "claims"
/ "video-claim__claim123"
/ "index.html"
)
assert not mislabeled.exists()
assert corrected.exists()
assert 'claim-thumbnail.webp' in channel_page.read_text()
assert 'claim-thumbnail.webp' in claim_page.read_text()
def test_build_site_mirrors_thumbnail_assets_into_site_tree(tmp_path):
config = Config()
config.general.base_dir = str(tmp_path)
config.general.offline_site_dir = str(tmp_path / "site")
config.general.fetch_missing_metadata_assets = False
manager = ArchiveSiteManager(config, tmp_path)
channel_dir = tmp_path / "channels" / "example__chan123"
claim_dir = channel_dir / "claims" / "video-claim__claim123"
version_dir = claim_dir / "versions" / "sd_abc123"
version_dir.mkdir(parents=True)
(channel_dir / "channel.json").write_text(
json.dumps({"display_name": "Example", "channel_name": "@example"})
)
(claim_dir / "claim.json").write_text(json.dumps({"title": "Video Claim"}))
(version_dir / "metadata.json").write_text(json.dumps({"version_token": "sd_abc123"}))
(version_dir / "video.mp4").write_bytes(b"x")
(claim_dir / "claim-thumbnail.webp").write_bytes(WEBP_STUB)
manager.build_site()
mirrored = (
tmp_path
/ "site"
/ "_assets"
/ "channels"
/ "example__chan123"
/ "claims"
/ "video-claim__claim123"
/ "claim-thumbnail.webp"
)
channel_page = tmp_path / "site" / "channels" / "example__chan123" / "index.html"
assert mirrored.exists()
assert '_assets/channels/example__chan123/claims/video-claim__claim123/claim-thumbnail.webp' in channel_page.read_text()
def test_build_site_extracts_video_thumbnail_when_metadata_thumbnail_is_missing(
tmp_path, monkeypatch
):
config = Config()
config.general.base_dir = str(tmp_path)
config.general.offline_site_dir = str(tmp_path / "site")
config.general.fetch_missing_metadata_assets = False
manager = ArchiveSiteManager(config, tmp_path)
channel_dir = tmp_path / "channels" / "example__chan123"
claim_dir = channel_dir / "claims" / "video-claim__claim123"
version_dir = claim_dir / "versions" / "sd_abc123"
version_dir.mkdir(parents=True)
(channel_dir / "channel.json").write_text(
json.dumps({"display_name": "Example", "channel_name": "@example"})
)
(version_dir / "metadata.json").write_text(
json.dumps(
{
"version_token": "sd_abc123",
"title": "Video Claim",
"description": "Video with no upstream thumbnail metadata",
"source_media_type": "video/mp4",
"release_time": 1712100000,
}
)
)
(version_dir / "video.mp4").write_bytes(b"x")
def fake_extract(input_path, output_path):
output_path.write_bytes(b"jpeg-data")
return True
monkeypatch.setattr(manager, "_run_ffmpeg_thumbnail", fake_extract)
manager.build_site()
channel_page = tmp_path / "site" / "channels" / "example__chan123" / "index.html"
extracted_thumb = claim_dir / "claim-thumbnail.jpg"
assert extracted_thumb.exists()
assert '_assets/channels/example__chan123/claims/video-claim__claim123/claim-thumbnail.jpg' in channel_page.read_text()
+69
View File
@@ -3,6 +3,7 @@
import hashlib import hashlib
import json import json
import re import re
from urllib.parse import urlparse
from datetime import datetime, timezone from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from typing import Any, Dict, Optional from typing import Any, Dict, Optional
@@ -223,6 +224,13 @@ def extract_channel_info(resolved_data: Dict[str, Any]) -> Dict[str, Any]:
"value_type": claim.get("value_type"), "value_type": claim.get("value_type"),
"title": value.get("title"), "title": value.get("title"),
"description": value.get("description"), "description": value.get("description"),
"thumbnail_url": extract_asset_url(value.get("thumbnail")),
"cover_url": extract_asset_url(value.get("cover")),
"website_url": value.get("website_url"),
"email": value.get("email"),
"tags": value.get("tags", []),
"languages": value.get("languages", []),
"links": extract_links(value),
} }
@@ -264,12 +272,14 @@ def extract_claim_metadata(claim_data: Dict[str, Any]) -> Dict[str, Any]:
"release_time": value.get("release_time"), "release_time": value.get("release_time"),
"title": value.get("title"), "title": value.get("title"),
"description": value.get("description"), "description": value.get("description"),
"thumbnail_url": extract_asset_url(value.get("thumbnail")),
"sd_hash": source.get("sd_hash"), "sd_hash": source.get("sd_hash"),
"stream_hash": claim.get("stream_hash"), "stream_hash": claim.get("stream_hash"),
"txid": claim.get("txid"), "txid": claim.get("txid"),
"nout": claim.get("nout"), "nout": claim.get("nout"),
"tags": value.get("tags", []), "tags": value.get("tags", []),
"languages": value.get("languages", []), "languages": value.get("languages", []),
"links": extract_links(value),
"fee": fee, "fee": fee,
"source_name": source.get("name"), "source_name": source.get("name"),
"source_media_type": source_media_type, "source_media_type": source_media_type,
@@ -277,6 +287,65 @@ def extract_claim_metadata(claim_data: Dict[str, Any]) -> Dict[str, Any]:
} }
def extract_asset_url(asset_data: Any) -> Optional[str]:
"""Extract a URL from a thumbnail/cover field when present."""
if isinstance(asset_data, str):
return asset_data
if isinstance(asset_data, dict):
for key in ("url", "uri", "src", "href"):
value = asset_data.get(key)
if isinstance(value, str) and value:
return value
return None
def extract_links(value: Dict[str, Any]) -> list[Dict[str, str]]:
"""Extract any presentation links from a claim value."""
links = []
website_url = value.get("website_url")
if isinstance(website_url, str) and website_url:
links.append({"label": "Website", "url": website_url})
for key in ("locations", "links"):
raw_items = value.get(key)
if isinstance(raw_items, dict):
raw_items = [raw_items]
if not isinstance(raw_items, list):
continue
for item in raw_items:
if not isinstance(item, dict):
continue
url = None
for field in ("url", "uri", "href"):
candidate = item.get(field)
if isinstance(candidate, str) and candidate:
url = candidate
break
if not url:
continue
label = item.get("label") or item.get("title") or item.get("name") or _host_label(url)
links.append({"label": str(label), "url": url})
deduped = []
seen = set()
for link in links:
key = (link["label"], link["url"])
if key in seen:
continue
seen.add(key)
deduped.append(link)
return deduped
def _host_label(url: str) -> str:
parsed = urlparse(url)
if parsed.netloc:
return parsed.netloc
return "Link"
def format_summary( def format_summary(
channels_scanned: int, channels_scanned: int,
claims_examined: int, claims_examined: int,