With the below way I am streaming and it is working.
You can see live stream on my channel : Twitch
Now the thing is I want to set alerts as well
However I don’t know how can I set them while streaming via API
Is that possible? I am ok with default alerts if it is possible from twitch desktop interface
My alert box browser url : [Preformatted text](https://dashboard.twitch.tv/widgets/alertbox#eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhbGVydF9zZXRfaWQiOiJjOThmNTMwMC1lMDIyLTQzYWUtOTE2OS04NDg1ODYwY2E3ZDQiLCJ1c2VyX2lkIjoiOTA2MjMzMDgxIn0.noTmyCMRWD9ftF7hvoG5nF17nsleCB92uZDIyqApbss)
My stream code how do I need to edit it?
async def stream_video(self, video, metadata):
if not await self.verify_twitch_token():
logging.error("Invalid Twitch token. Please update your token.")
return
video_path = os.path.abspath(metadata['file_path'])
logging.info(f"Attempting to stream video from: {video_path}")
if not os.path.exists(video_path):
logging.error(f"Error: Video file not found at {video_path}")
logging.info(f"Current working directory: {os.getcwd()}")
logging.info(f"Files in downloads directory: {os.listdir('downloads')}")
return
ingest_servers = await self.get_ingest_servers()
if not ingest_servers:
logging.error("No ingest servers available. Cannot stream.")
return
stream_key = await self.get_stream_key()
if not stream_key:
logging.error("Failed to get stream key. Cannot stream.")
return
await self.update_stream_info(metadata['title'], metadata['description'], metadata['thumbnail_url'])
retry_count = 0
while retry_count < MAX_RETRY_ATTEMPTS:
for server in ingest_servers:
rtmp_url = f"{server['url_template'].replace('{stream_key}', stream_key)}"
duration = self.get_video_duration(video_path)
ffmpeg_command = [
FFMPEG_PATH,
'-re',
'-i', video_path,
'-c:v', 'libx264',
'-preset', 'veryfast',
'-b:v', '6000k',
'-maxrate', '6000k',
'-bufsize', '12000k',
'-pix_fmt', 'yuv420p',
'-g', '50',
'-c:a', 'aac',
'-b:a', '160k',
'-ac', '2',
'-ar', '44100',
'-vf', 'scale=1920:1080',
'-f', 'flv',
'-progress', 'pipe:1',
rtmp_url
]
process = None
try:
logging.info(f"Attempting to stream to: {rtmp_url}")
process = subprocess.Popen(ffmpeg_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, universal_newlines=True)
def log_output(stream, prefix):
for line in stream:
logging.info(f"{prefix}: {line.strip()}")
threading.Thread(target=log_output, args=(process.stdout, "FFmpeg stdout")).start()
threading.Thread(target=log_output, args=(process.stderr, "FFmpeg stderr")).start()
# Wait for the stream to potentially go live
await asyncio.sleep(30)
is_live = await self.check_stream_status()
if not is_live:
logging.error("Stream failed to go live on Twitch")
process.terminate()
continue
await self.stream_with_timeout(process, duration)
logging.info("Sending SIGTERM to FFmpeg process")
process.terminate()
try:
process.wait(timeout=10)
except subprocess.TimeoutExpired:
logging.warning("FFmpeg process did not terminate in time. Sending SIGKILL.")
process.kill()
if process.returncode == 0:
logging.info(f"Successfully streamed to {rtmp_url}")
return True # Successful stream, exit the function
else:
logging.error(f"Failed to stream to {rtmp_url}. Trying next server.")
except Exception as e:
logging.error(f"An error occurred while streaming: {e}")
logging.error(traceback.format_exc())
if process:
stderr_output = process.stderr.read()
logging.error(f"FFmpeg stderr output: {stderr_output}")
finally:
if process:
try:
process.kill() # Ensure the process is terminated
except Exception as e:
logging.error(f"Error while terminating FFmpeg process: {e}")
retry_count += 1
if retry_count < MAX_RETRY_ATTEMPTS:
logging.info(f"Retrying stream in {RETRY_DELAY} seconds... (Attempt {retry_count + 1}/{MAX_RETRY_ATTEMPTS})")
await asyncio.sleep(RETRY_DELAY)
else:
logging.error(f"Failed to stream video after {MAX_RETRY_ATTEMPTS} attempts. Skipping this video.")
logging.info(f"Finished attempting to stream video: {metadata['title']}")
await self.update_post_stream_info(metadata)
return False
