Video download failed after few time later

[masumr]

I start to download video in tiktok live after few time later this downloading off automatic but event data fetch no problem. Error log is given below-

After this tiktok live disconnect and try to new tiktok live connect same user at this time we get this issue-

Package Version

My version of TikTokLive is 6.0.3.

[isaackogan]

Hi,

I can't really help with this. It seems to be to do with ffmpeg and/or TikTok. The library provides the URL to fetch streams directly from TikTok. If that URL doesn't work, there isn't much to be done on my end, unfortunately.

Especially without specifics of what you are doing, I can't begin to guess.

[masumr]

import json
import re
from re import Match

from TikTokLive.client.client import TikTokLiveClient
from TikTokLive.client.web.routes.fetch_video import VideoFetchQuality
from TikTokLive.events import ConnectEvent, DisconnectEvent, LiveEndEvent
from TikTokLive.events import proto_events, custom_events

HANDLED_MESSAGE_EVENTS = [
    proto_events.JoinEvent, proto_events.LikeEvent, proto_events.GiftEvent,
    proto_events.EnvelopeEvent, proto_events.SubscribeEvent, custom_events.FollowEvent,
    custom_events.ShareEvent, proto_events.EmoteChatEvent
]

LIVE_URL_REGEX = re.compile('^https?://(www\\.)?tiktok\\.com/@([^/]+)/live$')
LIVESTREAM_STOP_OR_FAILED_RESPONSE = -1

def extract_unique_id(stream_url: str):
    match: Match = LIVE_URL_REGEX.match(stream_url)
    if match:
        return match.groups()[-1]
    return None

class TikTokLiveStreamIterator(TikTokLiveClient):
    def __init__(self, stream_url: str, **options):
        TikTokLiveClient.__init__(self, unique_id=stream_url)
        self.add_listener(ConnectEvent, self._process_connect_event)
        self.add_listener(DisconnectEvent, self._process_disconnect_event)
        self.add_listener(LiveEndEvent, self._process_live_end_event)
        # self.on("error", self._process_error_event)

        # for message_event in HANDLED_MESSAGE_EVENTS:
        #     self.add_listener(message_event, self._process_message_event)

    @property
    def streaming_url(self):
        _streaming_url: str | None = None
        if self.web.fetch_video.is_recording:
            record_data: dict = json.loads(
                self.room_info['stream_url']['live_core_sdk_data']['pull_data']['stream_data']
            )
            record_url_data: dict = record_data['data'][VideoFetchQuality.ORIGIN.value]['main']
            _streaming_url: str = record_url_data.get('mp4') or record_url_data['flv']
        return _streaming_url

    async def _process_connect_event(self, event: ConnectEvent):
        from datetime import datetime
        # await self.is_live()
        output_path = f"{event.unique_id}-{datetime.utcnow()}.mp4"
        self.web.fetch_video.start(
            output_fp=output_path,
            room_info=self.room_info,
            output_format="mp4",
            quality=VideoFetchQuality.ORIGIN
        )
        print(f"Client connected.")

    async def _process_disconnect_event(self, event: DisconnectEvent):
        if self.web.fetch_video.is_recording:
            self.web.fetch_video.stop()

        print(f"Client disconnected!")

    async def _process_live_end_event(self, event: LiveEndEvent):
        print(f"Stream ended.")

    async def _process_error_event(self, error):
        print("_process_error_event is invoked.")
        if error is not None:
            print(f"Error: {error}")

    async def _process_message_event(self, event):
        print(f"\n\n**********************--- {event.get_type().lower()}  ---******************************")
        try:
            print(f"{event.user.unique_id}")
        except Exception as e:
            print(e)
        print(event.__dict__)
        print("********************************************************************************\n\n")

    async def start(self) -> None:
        print(f"Starting iterator for unique id: {self.unique_id}.")
        await TikTokLiveClient.start(self, fetch_room_info=True)

    async def stop(self) -> None:
        print(f"Stopping iterator for unique id: {self.unique_id}.")
        TikTokLiveClient.remove_all_listeners(self)
        await TikTokLiveClient.disconnect(self)
        print(f"Stopped iterator for unique id: {self.unique_id}.")

    @classmethod
    def can_monitor_stream_url(cls, stream_url: str):
        return LIVE_URL_REGEX.match(stream_url) is not None

I run this code in ec2 instance and start in fastapi route call. First time package connect to downloading start and get event data smoothly. After some time later event data is coming properly but downloading stop automatically generate this log -

[isaackogan]

Once again, this seems to be an ffmpeg-related question, not TikTokLive. All TikTokLive does is open ffmpeg with the wrapper. You can easily retrieve the URL as you are doing with streaming_url and use your own external library to download streams if you find an issue with the python-ffmpeg library. This is not TikTokLive related.