mirror of
https://github.com/PR0M3TH3AN/LBRY-Downloader.git
synced 2026-10-27 08:22:24 +00:00
672 lines
22 KiB
Python
672 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
LBRY Downloader - Incrementally sync LBRY/Odysee channel content.
|
|
|
|
This tool downloads all downloadable file claims from configured channels,
|
|
maintaining persistent state to avoid re-downloading unchanged content.
|
|
"""
|
|
|
|
import argparse
|
|
import logging
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import List, Optional
|
|
|
|
from config_loader import ConfigError, ensure_directories, load_config
|
|
from downloader import DownloadError, Downloader
|
|
from direct_downloader import DirectDownloadError
|
|
from lbry_client import LbryClient, LbryClientError, check_daemon_health
|
|
from models import Channel, Config, DownloadAction
|
|
from offline_site import ArchiveSiteManager
|
|
from planner import Planner
|
|
from state_db import StateDb
|
|
from utils import (
|
|
create_channel_folder_name,
|
|
extract_channel_info,
|
|
expand_path,
|
|
format_summary,
|
|
normalize_odysee_url,
|
|
)
|
|
|
|
|
|
VIDEO_EXTENSIONS = {
|
|
".mp4",
|
|
".webm",
|
|
".mkv",
|
|
".mov",
|
|
".avi",
|
|
".m4v",
|
|
".mpeg",
|
|
".mpg",
|
|
".wmv",
|
|
".flv",
|
|
}
|
|
|
|
|
|
def setup_logging(log_level: str) -> None:
|
|
"""Configure logging."""
|
|
logging.basicConfig(
|
|
level=getattr(logging, log_level),
|
|
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
|
|
|
|
def resolve_channel(
|
|
client: LbryClient,
|
|
channel_input: str,
|
|
) -> Optional[Channel]:
|
|
"""
|
|
Resolve a channel input to a Channel object.
|
|
|
|
Args:
|
|
client: Lbry client instance.
|
|
channel_input: URL, URI, or claim ID.
|
|
|
|
Returns:
|
|
Resolved Channel or None if resolution failed.
|
|
"""
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Normalize input
|
|
if channel_input.startswith("http"):
|
|
uri = normalize_odysee_url(channel_input)
|
|
elif not channel_input.startswith("lbry://"):
|
|
uri = f"lbry://{channel_input}"
|
|
else:
|
|
uri = channel_input
|
|
|
|
try:
|
|
logger.debug(f"Resolving: {uri}")
|
|
result = client.resolve([uri])
|
|
|
|
if not result or uri not in result:
|
|
logger.error(f"Could not resolve channel: {channel_input}")
|
|
return None
|
|
|
|
resolved = result[uri]
|
|
|
|
# Check if it's a channel
|
|
if "claim" in resolved:
|
|
claim = resolved["claim"]
|
|
else:
|
|
claim = resolved
|
|
|
|
value_type = claim.get("value_type", "").lower()
|
|
if value_type != "channel":
|
|
logger.error(f"Resolved claim is not a channel: {channel_input}")
|
|
return None
|
|
|
|
channel_claim_id = claim.get("claim_id")
|
|
channel_name = claim.get("name")
|
|
|
|
# Odysee's public proxy can resolve vanity channel URLs to older
|
|
# non-controlling channel claims. Enumerating streams from those claim IDs
|
|
# returns zero items, so normalize to the controlling claim before scanning.
|
|
if not client.supports_file_ops and channel_name:
|
|
controlling_claim = client.find_controlling_channel(channel_name)
|
|
if controlling_claim and controlling_claim.get("claim_id"):
|
|
controlling_claim_id = controlling_claim.get("claim_id")
|
|
if controlling_claim_id != channel_claim_id:
|
|
logger.debug(
|
|
"Normalized proxy-resolved channel %s from %s to controlling claim %s",
|
|
channel_name,
|
|
channel_claim_id,
|
|
controlling_claim_id,
|
|
)
|
|
claim = controlling_claim
|
|
channel_claim_id = controlling_claim_id
|
|
channel_name = claim.get("name")
|
|
|
|
if not channel_claim_id:
|
|
logger.error(f"No claim_id in resolved channel: {channel_input}")
|
|
return None
|
|
|
|
# Create folder name
|
|
folder = create_channel_folder_name(channel_name, channel_claim_id)
|
|
|
|
value = claim.get("value", {})
|
|
channel_info = extract_channel_info(claim)
|
|
|
|
channel = Channel(
|
|
input=channel_input,
|
|
normalized_uri=uri,
|
|
channel_claim_id=channel_claim_id,
|
|
channel_name=channel_name,
|
|
folder=folder,
|
|
display_name=channel_info.get("title"),
|
|
permanent_url=channel_info.get("permanent_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})")
|
|
return channel
|
|
|
|
except LbryClientError as e:
|
|
logger.error(f"Failed to resolve channel {channel_input}: {e}")
|
|
return None
|
|
|
|
|
|
def enumerate_channel_claims(
|
|
client: LbryClient,
|
|
channel: Channel,
|
|
config: Config,
|
|
) -> List[dict]:
|
|
"""
|
|
Enumerate all downloadable claims from a channel.
|
|
|
|
Args:
|
|
client: Lbry client.
|
|
channel: Channel to enumerate.
|
|
config: Application config.
|
|
|
|
Returns:
|
|
List of claim metadata.
|
|
"""
|
|
logger = logging.getLogger(__name__)
|
|
all_claims = []
|
|
page = 1
|
|
|
|
logger.info(f"Scanning channel: {channel.channel_name}")
|
|
|
|
while True:
|
|
try:
|
|
result = client.get_channel_claims(
|
|
channel_claim_id=channel.channel_claim_id,
|
|
page=page,
|
|
page_size=config.general.channel_page_size,
|
|
include_reposts=config.general.include_reposts,
|
|
)
|
|
|
|
items = result.get("items", [])
|
|
if not items:
|
|
break
|
|
|
|
all_claims.extend(items)
|
|
logger.debug(f" Page {page}: {len(items)} claims")
|
|
|
|
# Check if we've reached the end
|
|
total_pages = result.get("total_pages", 1)
|
|
if page >= total_pages:
|
|
break
|
|
|
|
page += 1
|
|
|
|
except LbryClientError as e:
|
|
logger.error(f"Failed to enumerate page {page}: {e}")
|
|
break
|
|
|
|
logger.info(f" Total claims found: {len(all_claims)}")
|
|
return all_claims
|
|
|
|
|
|
def _action_is_video(action: DownloadAction) -> bool:
|
|
"""Return True when a planned action looks like a video download."""
|
|
media_type = (action.metadata.get("source_media_type") or "").lower()
|
|
if media_type.startswith("video/"):
|
|
return True
|
|
|
|
stream_type = (action.metadata.get("stream_type") or "").lower()
|
|
if stream_type == "video":
|
|
return True
|
|
|
|
source_name = action.metadata.get("source_name") or action.metadata.get("file_name")
|
|
if source_name and Path(source_name).suffix.lower() in VIDEO_EXTENSIONS:
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def filter_actions_by_content_type(
|
|
actions: List[DownloadAction],
|
|
*,
|
|
video_only: bool = False,
|
|
non_video_only: bool = False,
|
|
) -> List[DownloadAction]:
|
|
"""Filter planned actions by content type before limits and execution."""
|
|
if not video_only and not non_video_only:
|
|
return actions
|
|
|
|
filtered_actions = []
|
|
for action in actions:
|
|
is_video = _action_is_video(action)
|
|
if video_only and is_video:
|
|
filtered_actions.append(action)
|
|
elif non_video_only and not is_video:
|
|
filtered_actions.append(action)
|
|
|
|
return filtered_actions
|
|
|
|
|
|
def resolve_content_filter_mode(
|
|
channel_config,
|
|
*,
|
|
video_only: bool = False,
|
|
non_video_only: bool = False,
|
|
) -> tuple[bool, bool, str]:
|
|
"""Resolve the effective content filter for a channel."""
|
|
if video_only:
|
|
return True, False, "video-only (CLI override)"
|
|
if non_video_only:
|
|
return False, True, "non-video-only (CLI override)"
|
|
|
|
content_mode = getattr(channel_config, "content_mode", "all")
|
|
if content_mode == "video_only":
|
|
return True, False, "video-only (channel config)"
|
|
if content_mode == "non_video_only":
|
|
return False, True, "non-video-only (channel config)"
|
|
|
|
return False, False, "all content"
|
|
|
|
|
|
def resolve_transport_mode(
|
|
*,
|
|
direct: bool = False,
|
|
p2p: bool = False,
|
|
) -> tuple[bool, str]:
|
|
"""Resolve the effective transport mode."""
|
|
if p2p:
|
|
return False, "p2p"
|
|
return True, "direct"
|
|
|
|
|
|
def _is_rate_limit_error(exc: Exception) -> bool:
|
|
"""Return True when the error indicates direct-mode rate limiting."""
|
|
return "rate-limited" in str(exc).lower()
|
|
|
|
|
|
def run_sync(
|
|
config: Config,
|
|
dry_run: bool = False,
|
|
direct: bool = True,
|
|
*,
|
|
video_only: bool = False,
|
|
non_video_only: bool = False,
|
|
) -> dict:
|
|
"""
|
|
Run the main synchronization process.
|
|
|
|
Args:
|
|
config: Application configuration.
|
|
dry_run: If True, don't actually download.
|
|
direct: If True, use Odysee proxy/CDN instead of P2P.
|
|
|
|
Returns:
|
|
Statistics dictionary.
|
|
"""
|
|
logger = logging.getLogger(__name__)
|
|
stats = {
|
|
"channels_scanned": 0,
|
|
"claims_examined": 0,
|
|
"new_downloads": 0,
|
|
"new_versions": 0,
|
|
"skipped_existing": 0,
|
|
"redownloaded_missing": 0,
|
|
"failures": 0,
|
|
}
|
|
|
|
# Load state
|
|
state_db = StateDb(config.general.state_file)
|
|
state_db.load()
|
|
|
|
# Initialize planner and downloader
|
|
base_path = Path(config.general.base_dir)
|
|
planner = Planner(config, state_db.data, str(base_path))
|
|
site_manager = ArchiveSiteManager(config, base_path)
|
|
|
|
if direct:
|
|
logger.info("Using DIRECT mode (Odysee proxy for metadata, Odysee CDN for downloads)")
|
|
client = LbryClient.create_odysee_proxy(timeout=config.lbrynet.timeout_seconds)
|
|
from direct_downloader import DirectDownloader
|
|
|
|
downloader = DirectDownloader(config, base_path)
|
|
p2p_fallback_client = None
|
|
p2p_fallback_downloader = None
|
|
p2p_fallback_available = None
|
|
else:
|
|
client = LbryClient(
|
|
api_url=config.lbrynet.api_url,
|
|
timeout=config.lbrynet.timeout_seconds,
|
|
)
|
|
logger.info(f"Checking daemon at {config.lbrynet.api_url}")
|
|
try:
|
|
check_daemon_health(client)
|
|
logger.info("Daemon is healthy")
|
|
except LbryClientError as e:
|
|
logger.error(str(e))
|
|
return stats
|
|
|
|
logger.info("Using P2P mode (LBRY network)")
|
|
downloader = Downloader(client, config, base_path)
|
|
|
|
# Process each enabled channel
|
|
enabled_channels = [c for c in config.channels if c.enabled]
|
|
|
|
for channel_config in enabled_channels:
|
|
stats["channels_scanned"] += 1
|
|
|
|
# Resolve channel
|
|
channel = resolve_channel(client, channel_config.input)
|
|
if not channel:
|
|
stats["failures"] += 1
|
|
continue
|
|
|
|
# Set custom download path if provided in config
|
|
if channel_config.download_path:
|
|
channel.download_path = channel_config.download_path
|
|
|
|
# Check if we already have this channel in state
|
|
existing = state_db.get_channel(channel.channel_claim_id)
|
|
if existing:
|
|
# Update with new info but keep folder and download_path
|
|
channel.folder = existing.folder
|
|
if existing.download_path:
|
|
channel.download_path = existing.download_path
|
|
|
|
if not dry_run:
|
|
site_manager.ensure_channel_artifacts(channel)
|
|
|
|
# Enumerate claims
|
|
claims = enumerate_channel_claims(client, channel, config)
|
|
|
|
# Plan downloads
|
|
actions = planner.process_channel_claims(channel, claims)
|
|
actions_before_filter = len(actions)
|
|
channel_video_only, channel_non_video_only, mode_label = (
|
|
resolve_content_filter_mode(
|
|
channel_config,
|
|
video_only=video_only,
|
|
non_video_only=non_video_only,
|
|
)
|
|
)
|
|
actions = filter_actions_by_content_type(
|
|
actions,
|
|
video_only=channel_video_only,
|
|
non_video_only=channel_non_video_only,
|
|
)
|
|
|
|
if len(actions) != actions_before_filter:
|
|
logger.info(
|
|
"Content filter (%s) skipped %s planned claim(s).",
|
|
mode_label,
|
|
actions_before_filter - len(actions),
|
|
)
|
|
|
|
# Apply download limit if configured
|
|
# Separate skip actions from download actions
|
|
skip_actions = [a for a in actions if a.action == "skip_existing"]
|
|
download_actions = [a for a in actions if a.action != "skip_existing"]
|
|
|
|
# Sort download actions by release_time (most recent first)
|
|
# Claims without release_time will be at the end
|
|
download_actions.sort(
|
|
key=lambda a: a.metadata.get("release_time") or 0, reverse=True
|
|
)
|
|
|
|
# Apply limit: 0 means "all" (no limit), otherwise limit to download_limit
|
|
if (
|
|
config.general.download_limit > 0
|
|
and len(download_actions) > config.general.download_limit
|
|
):
|
|
skipped_due_to_limit = download_actions[config.general.download_limit :]
|
|
download_actions = download_actions[: config.general.download_limit]
|
|
|
|
logger.info(
|
|
f"Download limit ({config.general.download_limit}) reached. "
|
|
f"Skipping {len(skipped_due_to_limit)} older items."
|
|
)
|
|
|
|
# Mark skipped items in stats
|
|
for action in skipped_due_to_limit:
|
|
stats["skipped_existing"] += 1
|
|
|
|
# Recombine actions
|
|
actions = skip_actions + download_actions
|
|
stats["claims_examined"] += len(actions)
|
|
|
|
# Execute downloads
|
|
for action in actions:
|
|
if action.action == "skip_existing":
|
|
stats["skipped_existing"] += 1
|
|
elif action.action == "redownload_missing":
|
|
stats["redownloaded_missing"] += 1
|
|
elif action.action in ("download_new", "download_new_version"):
|
|
if action.action == "download_new":
|
|
stats["new_downloads"] += 1
|
|
else:
|
|
stats["new_versions"] += 1
|
|
|
|
try:
|
|
success = downloader.download(action, dry_run=dry_run)
|
|
|
|
if success and not dry_run:
|
|
site_manager.ensure_claim_artifacts(channel, action)
|
|
|
|
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
|
|
history_file = base_path / "state" / "run-history.jsonl"
|
|
state_db.log_run_history(
|
|
str(history_file),
|
|
channel.channel_claim_id,
|
|
action.claim_id,
|
|
action.action,
|
|
action.version_token,
|
|
"success" if success else "failure",
|
|
)
|
|
|
|
except (DownloadError, DirectDownloadError) as e:
|
|
if (
|
|
direct
|
|
and config.general.direct_auto_fallback_to_p2p
|
|
and isinstance(e, DirectDownloadError)
|
|
and _is_rate_limit_error(e)
|
|
and not dry_run
|
|
):
|
|
if p2p_fallback_available is None:
|
|
try:
|
|
p2p_fallback_client = LbryClient(
|
|
api_url=config.lbrynet.api_url,
|
|
timeout=config.lbrynet.timeout_seconds,
|
|
)
|
|
check_daemon_health(p2p_fallback_client)
|
|
p2p_fallback_downloader = Downloader(
|
|
p2p_fallback_client, config, base_path
|
|
)
|
|
p2p_fallback_available = True
|
|
logger.warning(
|
|
"Direct mode hit rate limiting. Local P2P fallback is available."
|
|
)
|
|
except LbryClientError as fallback_init_error:
|
|
p2p_fallback_available = False
|
|
logger.warning(
|
|
"Direct mode hit rate limiting, but P2P fallback is unavailable: %s",
|
|
fallback_init_error,
|
|
)
|
|
|
|
if p2p_fallback_available and p2p_fallback_downloader is not None:
|
|
try:
|
|
logger.warning(
|
|
"Retrying %s via P2P fallback after direct-mode rate limiting.",
|
|
action.claim_name,
|
|
)
|
|
success = p2p_fallback_downloader.download(
|
|
action, dry_run=False
|
|
)
|
|
if success:
|
|
planner.update_state_from_action(action, success=True)
|
|
history_file = base_path / "state" / "run-history.jsonl"
|
|
state_db.log_run_history(
|
|
str(history_file),
|
|
channel.channel_claim_id,
|
|
action.claim_id,
|
|
action.action,
|
|
action.version_token,
|
|
"success",
|
|
)
|
|
continue
|
|
except DownloadError as fallback_error:
|
|
e = DirectDownloadError(
|
|
f"{e}; P2P fallback also failed: {fallback_error}"
|
|
)
|
|
|
|
logger.error(f"Download failed: {e}")
|
|
stats["failures"] += 1
|
|
|
|
# Log failure
|
|
history_file = base_path / "state" / "run-history.jsonl"
|
|
state_db.log_run_history(
|
|
str(history_file),
|
|
channel.channel_claim_id,
|
|
action.claim_id,
|
|
action.action,
|
|
action.version_token,
|
|
"failure",
|
|
str(e),
|
|
)
|
|
|
|
# Update channel last scan time
|
|
from datetime import datetime, timezone
|
|
|
|
channel.last_scan = datetime.now(timezone.utc).isoformat()
|
|
state_db.set_channel(channel)
|
|
if not dry_run:
|
|
site_manager.ensure_channel_artifacts(channel)
|
|
|
|
# Save state
|
|
if not dry_run:
|
|
state_db.update_last_run()
|
|
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")
|
|
|
|
return stats
|
|
|
|
|
|
def main():
|
|
"""Main entry point."""
|
|
parser = argparse.ArgumentParser(
|
|
description="LBRY/Odysee Channel Downloader",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
python main.py # Run with default config (DIRECT mode)
|
|
python main.py --config ./config.yaml # Use custom config
|
|
python main.py --dry-run # Plan actions but don't download
|
|
python main.py --p2p # Use the local LBRY node instead
|
|
""",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--config",
|
|
"-c",
|
|
help="Path to configuration file (default: ~/Documents/lbry-downloads/config.yaml)",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--dry-run",
|
|
"-n",
|
|
action="store_true",
|
|
help="Show what would be downloaded without actually downloading",
|
|
)
|
|
|
|
transport_group = parser.add_mutually_exclusive_group()
|
|
transport_group.add_argument(
|
|
"--direct",
|
|
"-d",
|
|
action="store_true",
|
|
help="Use direct mode explicitly (default: Odysee public proxy + CDN)",
|
|
)
|
|
transport_group.add_argument(
|
|
"--p2p",
|
|
action="store_true",
|
|
help="Use the local LBRY daemon and peer-to-peer network instead of direct mode",
|
|
)
|
|
|
|
content_filter_group = parser.add_mutually_exclusive_group()
|
|
content_filter_group.add_argument(
|
|
"--video-only",
|
|
action="store_true",
|
|
help="Only include video downloads such as mp4/webm/mkv",
|
|
)
|
|
content_filter_group.add_argument(
|
|
"--non-video-only",
|
|
action="store_true",
|
|
help="Only include non-video downloads such as zip files",
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
direct_mode, transport_label = resolve_transport_mode(
|
|
direct=args.direct,
|
|
p2p=args.p2p,
|
|
)
|
|
|
|
# Load config
|
|
config = load_config(args.config)
|
|
|
|
# Setup logging
|
|
setup_logging(config.general.log_level)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
logger.info("LBRY Downloader starting")
|
|
logger.info("Effective transport mode: %s", transport_label)
|
|
|
|
# Ensure directories exist
|
|
ensure_directories(config)
|
|
|
|
# Run sync
|
|
stats = run_sync(
|
|
config,
|
|
dry_run=args.dry_run,
|
|
direct=direct_mode,
|
|
video_only=args.video_only,
|
|
non_video_only=args.non_video_only,
|
|
)
|
|
|
|
# Print summary
|
|
summary = format_summary(
|
|
channels_scanned=stats["channels_scanned"],
|
|
claims_examined=stats["claims_examined"],
|
|
new_downloads=stats["new_downloads"],
|
|
new_versions=stats["new_versions"],
|
|
skipped_existing=stats["skipped_existing"],
|
|
redownloaded_missing=stats["redownloaded_missing"],
|
|
failures=stats["failures"],
|
|
)
|
|
print("\n" + summary)
|
|
|
|
# Exit with error code if there were failures
|
|
if stats["failures"] > 0:
|
|
sys.exit(1)
|
|
|
|
except ConfigError as e:
|
|
print(f"Configuration error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
except KeyboardInterrupt:
|
|
print("\nInterrupted by user", file=sys.stderr)
|
|
sys.exit(130)
|
|
except Exception as e:
|
|
print(f"Unexpected error: {e}", file=sys.stderr)
|
|
raise
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|