Record TikTok Live Streams: Build Your Own Recorder

Recording approaches

I tried three different methods before settling on the final solution:

Setting up the recorder

Prerequisites

# Install dependencies
sudo apt update
sudo apt install -y python3 python3-pip ffmpeg

# Install Python packages
pip3 install yt-dlp selenium webdriver-manager requests

# If running headless, also install:
sudo apt install -y xvfb

Core recorder script

#!/usr/bin/env python3
import os
import time
import subprocess
from datetime import datetime
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

class TikTokLiveRecorder:
    def __init__(self, output_dir="recordings"):
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)

        # Setup Chrome in headless mode
        chrome_options = Options()
        chrome_options.add_argument('--headless=new')
        chrome_options.add_argument('--no-sandbox')
        chrome_options.add_argument('--disable-dev-shm-usage')
        chrome_options.add_argument('--disable-gpu')
        chrome_options.add_argument('--window-size=1920,1080')

        self.driver = webdriver.Chrome(options=chrome_options)

    def get_stream_url(self, username):
        """Extract live stream URL from TikTok profile"""
        url = f"https://www.tiktok.com/@{username}/live"

        try:
            self.driver.get(url)
            time.sleep(3)  # Wait for page load

            # Check if LIVE badge exists
            try:
                live_badge = WebDriverWait(self.driver, 5).until(
                    EC.presence_of_element_located((By.XPATH, "//span[contains(text(), 'LIVE')]"))
                )

                # Get the actual video element and extract stream URL
                video_elem = self.driver.find_element(By.TAG_NAME, 'video')
                stream_url = video_elem.get_attribute('src')

                if stream_url and 'live' in stream_url.lower():
                    return stream_url
            except:
                return None

        except Exception as e:
            print(f"Error checking stream: {e}")
            return None

    def record_stream(self, username, stream_url):
        """Record the stream using yt-dlp"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"{username}_{timestamp}.mp4"
        output_path = os.path.join(self.output_dir, filename)

        # yt-dlp command with proper options for live streams
        cmd = [
            'yt-dlp',
            '-f', 'best',  # Best quality
            '--live-from-start',  # Record from beginning if possible
            '--wait-for-video', '10',  # Wait 10s for stream to start
            '--retries', '10',  # Retry on network failures
            '--fragment-retries', '10',
            '-o', output_path,
            stream_url
        ]

        print(f"Starting recording: {filename}")
        process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

        return process, output_path

    def monitor_and_record(self, username, check_interval=60):
        """Monitor user and record when live"""
        print(f"Monitoring @{username}...")

        while True:
            stream_url = self.get_stream_url(username)

            if stream_url:
                print(f"@{username} is LIVE! Starting recording...")
                process, output_path = self.record_stream(username, stream_url)

                # Wait while recording
                while process.poll() is None:
                    # Check if still live every 30 seconds
                    time.sleep(30)
                    if not self.get_stream_url(username):
                        print("Stream ended. Stopping recording...")
                        process.terminate()
                        break

                print(f"Recording saved: {output_path}")

            else:
                print(f"@{username} is not live. Checking again in {check_interval}s...")

            time.sleep(check_interval)

    def cleanup(self):
        """Clean up resources"""
        self.driver.quit()


if __name__ == "__main__":
    recorder = TikTokLiveRecorder()

    try:
        # Monitor specific user
        recorder.monitor_and_record("username_here", check_interval=60)
    except KeyboardInterrupt:
        print("\nStopping recorder...")
        recorder.cleanup()

This script monitors a TikTok account every 60 seconds. When it detects a live stream, it automatically starts recording and continues until the stream ends.

Multi-user monitoring

Recording one user is fine, but I wanted to monitor multiple creators. Here's the multi-user version:

import threading
import queue

class MultiUserRecorder:
    def __init__(self, users, output_dir="recordings"):
        self.users = users
        self.output_dir = output_dir
        self.active_streams = set()
        self.task_queue = queue.Queue()

    def monitor_user(self, username):
        """Monitor a single user for live status"""
        recorder = TikTokLiveRecorder(self.output_dir)

        while True:
            if username in self.active_streams:
                time.sleep(30)
                continue

            stream_url = recorder.get_stream_url(username)
            if stream_url:
                print(f"[MONITOR] @{username} went live!")
                self.active_streams.add(username)
                self.task_queue.put((username, stream_url))

            time.sleep(45)  # Check every 45 seconds

    def recorder_worker(self):
        """Dedicated thread for recording"""
        recorder = TikTokLiveRecorder(self.output_dir)

        while True:
            username, stream_url = self.task_queue.get()

            try:
                print(f"[RECORDER] Starting recording for @{username}")
                process, output_path = recorder.record_stream(username, stream_url)

                # Monitor stream status
                while process.poll() is None:
                    time.sleep(30)
                    current_url = recorder.get_stream_url(username)

                    if not current_url or current_url != stream_url:
                        print(f"[RECORDER] Stream ended for @{username}")
                        process.terminate()
                        break

                print(f"[RECORDER] Saved: {output_path}")

            finally:
                self.active_streams.discard(username)
                self.task_queue.task_done()

    def start(self):
        """Start all monitoring threads"""
        # Start monitor threads for each user
        for user in self.users:
            t = threading.Thread(target=self.monitor_user, args=(user,), daemon=True)
            t.start()

        # Start recorder thread
        recorder_thread = threading.Thread(target=self.recorder_worker, daemon=True)
        recorder_thread.start()

        print(f"Monitoring {len(self.users)} users...")
        print(f"Users: {', '.join(self.users)}")

        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            print("\nShutting down...")


# Usage
if __name__ == "__main__":
    users_to_monitor = [
        "creator1",
        "creator2",
        "creator3"
    ]

    multi_rec = MultiUserRecorder(users_to_monitor)
    multi_rec.start()

This version can monitor unlimited users simultaneously. Each user gets their own monitoring thread, while a dedicated recorder thread handles the actual recording to avoid resource conflicts.

Adding notifications

What good is recording if you don't know it's happening? I added Telegram notifications:

import requests

class TelegramNotifier:
    def __init__(self, bot_token, chat_id):
        self.bot_token = bot_token
        self.chat_id = chat_id
        self.base_url = f"https://api.telegram.org/bot{bot_token}"

    def send_message(self, message):
        """Send message to Telegram"""
        url = f"{self.base_url}/sendMessage"
        data = {
            "chat_id": self.chat_id,
            "text": message,
            "parse_mode": "HTML"
        }
        requests.post(url, json=data)

    def notify_live_start(self, username):
        """Notify when a user goes live"""
        message = f"🔴 @{username} is LIVE!\n\nRecording started."
        self.send_message(message)

    def notify_recording_complete(self, username, filename):
        """Notify when recording is done"""
        message = f"✅ Recording complete!\n\n@{username}\nFile: {filename}"
        self.send_message(message)


# Integrate with recorder
class TikTokLiveRecorder:
    def __init__(self, output_dir="recordings", notifier=None):
        self.output_dir = output_dir
        self.notifier = notifier
        # ... rest of init

    def monitor_and_record(self, username, check_interval=60):
        while True:
            stream_url = self.get_stream_url(username)

            if stream_url:
                if self.notifier:
                    self.notifier.notify_live_start(username)

                process, output_path = self.record_stream(username, stream_url)

                while process.poll() is None:
                    time.sleep(30)
                    if not self.get_stream_url(username):
                        process.terminate()
                        break

                if self.notifier:
                    self.notifier.notify_recording_complete(username, os.path.basename(output_path))

            time.sleep(check_interval)

Now I get instant notifications on my phone when anyone I monitor goes live, plus another message when recording finishes.

Storage management

After 200+ recordings, storage became an issue. Here's how I manage it:

import shutil
from pathlib import Path

class StorageManager:
    def __init__(self, base_dir="recordings", max_size_gb=500):
        self.base_dir = Path(base_dir)
        self.max_size_bytes = max_size_gb * 1024**3

    def get_current_size(self):
        """Get total size of recordings directory"""
        total = sum(f.stat().st_size for f in self.base_dir.rglob('*') if f.is_file())
        return total

    def cleanup_old_recordings(self, keep_days=30):
        """Delete recordings older than keep_days"""
        cutoff = time.time() - (keep_days * 86400)

        for video_file in self.base_dir.glob("*.mp4"):
            if video_file.stat().st_mtime < cutoff:
                video_file.unlink()
                print(f"Deleted old recording: {video_file.name}")

    def auto_cleanup(self):
        """Automatically clean up if over size limit"""
        current_size = self.get_current_size()

        if current_size > self.max_size_bytes:
            print(f"Storage limit exceeded. Current: {current_size / 1024**3:.2f}GB")
            self.cleanup_old_recordings(keep_days=14)  # Aggressive cleanup

            # Still over limit? Delete more
            if self.get_current_size() > self.max_size_bytes:
                self.cleanup_old_recordings(keep_days=7)

Run the cleanup daily via cron, or integrate it into the recorder to check after each recording.

Running it 24/7

For reliable 24/7 operation, don't just run the script. Use proper process management:

Systemd service

# /etc/systemd/system/tiktok-recorder.service
[Unit]
Description=TikTok Live Recorder
After=network.target

[Service]
Type=simple
User=your-username
WorkingDirectory=/path/to/recorder
ExecStart=/usr/bin/python3 /path/to/recorder/multi_recorder.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

# Enable and start the service
sudo systemctl enable tiktok-recorder
sudo systemctl start tiktok-recorder

# Check logs
sudo journalctl -u tiktok-recorder -f

Docker alternative

# Dockerfile
FROM python:3.11-slim

RUN apt-get update && apt-get install -y \
    ffmpeg \
    xvfb \
    chromium \
    chromium-driver

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["python", "multi_recorder.py"]

# docker-compose.yml
version: "3.8"
services:
  tiktok-recorder:
    build: .
    volumes:
      - ./recordings:/app/recordings
    restart: always
    environment:
      - DISPLAY=:99
    command: Xvfb :99 -screen 0 1920x1080x24 && python multi_recorder.py

I prefer systemd for simplicity. Docker is overkill unless you're running multiple services on the same machine.