lbry-sdk/lbry/file/file_manager.py

import asyncio
import logging
import typing
from typing import Optional
from aiohttp.web import Request
from lbry.error import ResolveError, DownloadSDTimeoutError, InsufficientFundsError
from lbry.error import ResolveTimeoutError, DownloadDataTimeoutError, KeyFeeAboveMaxAllowedError
from lbry.stream.managed_stream import ManagedStream
from lbry.torrent.torrent_manager import TorrentSource
from lbry.utils import cache_concurrent
from lbry.schema.url import URL
from lbry.wallet.dewies import dewies_to_lbc
from lbry.file.source_manager import SourceManager
from lbry.file.source import ManagedDownloadSource
if typing.TYPE_CHECKING:
    from lbry.conf import Config
    from lbry.extras.daemon.analytics import AnalyticsManager
    from lbry.extras.daemon.storage import SQLiteStorage
    from lbry.wallet import WalletManager, Output
    from lbry.extras.daemon.exchange_rate_manager import ExchangeRateManager

log = logging.getLogger(__name__)


class FileManager:
    def __init__(self, loop: asyncio.AbstractEventLoop, config: 'Config', wallet_manager: 'WalletManager',
                 storage: 'SQLiteStorage', analytics_manager: Optional['AnalyticsManager'] = None):
        self.loop = loop
        self.config = config
        self.wallet_manager = wallet_manager
        self.storage = storage
        self.analytics_manager = analytics_manager
        self.source_managers: typing.Dict[str, SourceManager] = {}
        self.started = asyncio.Event()

    @property
    def streams(self):
        return self.source_managers['stream']._sources

    async def create_stream(self, file_path: str, key: Optional[bytes] = None, **kwargs) -> ManagedDownloadSource:
        if 'stream' in self.source_managers:
            return await self.source_managers['stream'].create(file_path, key, **kwargs)
        raise NotImplementedError

    async def start(self):
        await asyncio.gather(*(source_manager.start() for source_manager in self.source_managers.values()))
        for manager in self.source_managers.values():
            await manager.started.wait()
        self.started.set()

    def stop(self):
        for manager in self.source_managers.values():
            # fixme: pop or not?
            manager.stop()
        self.started.clear()

    @cache_concurrent
    async def download_from_uri(self, uri, exchange_rate_manager: 'ExchangeRateManager',
                                timeout: Optional[float] = None, file_name: Optional[str] = None,
                                download_directory: Optional[str] = None,
                                save_file: Optional[bool] = None, resolve_timeout: float = 3.0,
                                wallet: Optional['Wallet'] = None) -> ManagedDownloadSource:

        wallet = wallet or self.wallet_manager.default_wallet
        timeout = timeout or self.config.download_timeout
        start_time = self.loop.time()
        resolved_time = None
        stream = None
        claim = None
        error = None
        outpoint = None
        if save_file is None:
            save_file = self.config.save_files
        if file_name and not save_file:
            save_file = True
        if save_file:
            download_directory = download_directory or self.config.download_dir
        else:
            download_directory = None

        payment = None
        try:
            # resolve the claim
            if not URL.parse(uri).has_stream:
                raise ResolveError("cannot download a channel claim, specify a /path")
            try:
                resolved_result = await asyncio.wait_for(
                    self.wallet_manager.ledger.resolve(
                        wallet.accounts, [uri],
                        include_purchase_receipt=True,
                        include_is_my_output=True
                    ), resolve_timeout
                )
            except asyncio.TimeoutError:
                raise ResolveTimeoutError(uri)
            except Exception as err:
                if isinstance(err, asyncio.CancelledError):
                    raise
                log.exception("Unexpected error resolving stream:")
                raise ResolveError(f"Unexpected error resolving stream: {str(err)}")
            if 'error' in resolved_result:
                raise ResolveError(f"Unexpected error resolving uri for download: {resolved_result['error']}")
            if not resolved_result or uri not in resolved_result:
                raise ResolveError(f"Failed to resolve stream at '{uri}'")
            txo = resolved_result[uri]
            if isinstance(txo, dict):
                raise ResolveError(f"Failed to resolve stream at '{uri}': {txo}")
            claim = txo.claim
            outpoint = f"{txo.tx_ref.id}:{txo.position}"
            resolved_time = self.loop.time() - start_time
            await self.storage.save_claim_from_output(self.wallet_manager.ledger, txo)

            ####################
            # update or replace
            ####################

            if claim.stream.source.bt_infohash:
                source_manager = self.source_managers['torrent']
                existing = source_manager.get_filtered(bt_infohash=claim.stream.source.bt_infohash)
            else:
                source_manager = self.source_managers['stream']
                existing = source_manager.get_filtered(sd_hash=claim.stream.source.sd_hash)

            # resume or update an existing stream, if the stream changed: download it and delete the old one after
            to_replace, updated_stream = None, None
            if existing and existing[0].claim_id != txo.claim_id:
                raise ResolveError(f"stream for {existing[0].claim_id} collides with existing download {txo.claim_id}")
            if existing:
                log.info("claim contains a metadata only update to a stream we have")
                if claim.stream.source.bt_infohash:
                    await self.storage.save_torrent_content_claim(
                        existing[0].identifier, outpoint, existing[0].torrent_length, existing[0].torrent_name
                    )
                    claim_info = await self.storage.get_content_claim_for_torrent(existing[0].identifier)
                    existing[0].set_claim(claim_info, claim)
                else:
                    await self.storage.save_content_claim(
                        existing[0].stream_hash, outpoint
                    )
                    await source_manager._update_content_claim(existing[0])
                updated_stream = existing[0]
            else:
                existing_for_claim_id = self.get_filtered(claim_id=txo.claim_id)
                if existing_for_claim_id:
                    log.info("claim contains an update to a stream we have, downloading it")
                    if save_file and existing_for_claim_id[0].output_file_exists:
                        save_file = False
                    if not claim.stream.source.bt_infohash:
                        existing_for_claim_id[0].downloader.node = source_manager.node
                    await existing_for_claim_id[0].start(timeout=timeout, save_now=save_file)
                    if not existing_for_claim_id[0].output_file_exists and (
                            save_file or file_name or download_directory):
                        await existing_for_claim_id[0].save_file(
                            file_name=file_name, download_directory=download_directory
                        )
                    to_replace = existing_for_claim_id[0]

            # resume or update an existing stream, if the stream changed: download it and delete the old one after
            if updated_stream:
                log.info("already have stream for %s", uri)
                if save_file and updated_stream.output_file_exists:
                    save_file = False
                if not claim.stream.source.bt_infohash:
                    updated_stream.downloader.node = source_manager.node
                await updated_stream.start(timeout=timeout, save_now=save_file)
                if not updated_stream.output_file_exists and (save_file or file_name or download_directory):
                    await updated_stream.save_file(
                        file_name=file_name, download_directory=download_directory
                    )
                return updated_stream

            ####################
            # pay fee
            ####################

            needs_purchasing = (
                not to_replace and
                not txo.is_my_output and
                txo.has_price and
                not txo.purchase_receipt
            )

            if needs_purchasing:
                payment = await self.wallet_manager.create_purchase_transaction(
                    wallet.accounts, txo, exchange_rate_manager
                )

            ####################
            # make downloader and wait for start
            ####################

            if not claim.stream.source.bt_infohash:
                # fixme: this shouldnt be here
                stream = ManagedStream(
                    self.loop, self.config, source_manager.blob_manager, claim.stream.source.sd_hash,
                    download_directory, file_name, ManagedStream.STATUS_RUNNING, content_fee=payment,
                    analytics_manager=self.analytics_manager
                )
                stream.downloader.node = source_manager.node
            else:
                stream = TorrentSource(
                    self.loop, self.config, self.storage, identifier=claim.stream.source.bt_infohash,
                    file_name=file_name, download_directory=download_directory or self.config.download_dir,
                    status=ManagedStream.STATUS_RUNNING,
                    analytics_manager=self.analytics_manager,
                    torrent_session=source_manager.torrent_session
                )
            log.info("starting download for %s", uri)

            before_download = self.loop.time()
            await stream.start(timeout, save_file)

            ####################
            # success case: delete to_replace if applicable, broadcast fee payment
            ####################

            if to_replace:  # delete old stream now that the replacement has started downloading
                await source_manager.delete(to_replace)

            if payment is not None:
                await self.wallet_manager.broadcast_or_release(payment)
                payment = None  # to avoid releasing in `finally` later
                log.info("paid fee of %s for %s", dewies_to_lbc(stream.content_fee.outputs[0].amount), uri)
                await self.storage.save_content_fee(stream.stream_hash, stream.content_fee)

            source_manager.add(stream)

            if not claim.stream.source.bt_infohash:
                await self.storage.save_content_claim(stream.stream_hash, outpoint)
            else:
                await self.storage.save_torrent_content_claim(
                    stream.identifier, outpoint, stream.torrent_length, stream.torrent_name
                )
                claim_info = await self.storage.get_content_claim_for_torrent(stream.identifier)
                stream.set_claim(claim_info, claim)
            if save_file:
                await asyncio.wait_for(stream.save_file(), timeout - (self.loop.time() - before_download),
                                       loop=self.loop)
            return stream
        except asyncio.TimeoutError:
            error = DownloadDataTimeoutError(stream.sd_hash)
            raise error
        except Exception as err:  # forgive data timeout, don't delete stream
            expected = (DownloadSDTimeoutError, DownloadDataTimeoutError, InsufficientFundsError,
                        KeyFeeAboveMaxAllowedError)
            if isinstance(err, expected):
                log.warning("Failed to download %s: %s", uri, str(err))
            elif isinstance(err, asyncio.CancelledError):
                pass
            else:
                log.exception("Unexpected error downloading stream:")
            error = err
            raise
        finally:
            if payment is not None:
                # payment is set to None after broadcasting, if we're here an exception probably happened
                await self.wallet_manager.ledger.release_tx(payment)
            if self.analytics_manager and claim and claim.stream.source.bt_infohash:
                # TODO: analytics for torrents
                pass
            elif self.analytics_manager and (error or (stream and (stream.downloader.time_to_descriptor or
                                                                   stream.downloader.time_to_first_bytes))):
                server = self.wallet_manager.ledger.network.client.server
                self.loop.create_task(
                    self.analytics_manager.send_time_to_first_bytes(
                        resolved_time, self.loop.time() - start_time, None if not stream else stream.download_id,
                        uri, outpoint,
                        None if not stream else len(stream.downloader.blob_downloader.active_connections),
                        None if not stream else len(stream.downloader.blob_downloader.scores),
                        None if not stream else len(stream.downloader.blob_downloader.connection_failures),
                        False if not stream else stream.downloader.added_fixed_peers,
                        self.config.fixed_peer_delay if not stream else stream.downloader.fixed_peers_delay,
                        None if not stream else stream.sd_hash,
                        None if not stream else stream.downloader.time_to_descriptor,
                        None if not (stream and stream.descriptor) else stream.descriptor.blobs[0].blob_hash,
                        None if not (stream and stream.descriptor) else stream.descriptor.blobs[0].length,
                        None if not stream else stream.downloader.time_to_first_bytes,
                        None if not error else error.__class__.__name__,
                        None if not error else str(error),
                        None if not server else f"{server[0]}:{server[1]}"
                    )
                )

    async def stream_partial_content(self, request: Request, sd_hash: str):
        return await self.source_managers['stream'].stream_partial_content(request, sd_hash)

    def get_filtered(self, *args, **kwargs) -> typing.List[ManagedDownloadSource]:
        """
        Get a list of filtered and sorted ManagedStream objects

        :param sort_by: field to sort by
        :param reverse: reverse sorting
        :param comparison: comparison operator used for filtering
        :param search_by: fields and values to filter by
        """
        return sum((manager.get_filtered(*args, **kwargs) for manager in self.source_managers.values()), [])

    async def delete(self, source: ManagedDownloadSource, delete_file=False):
        for manager in self.source_managers.values():
            await manager.delete(source, delete_file)