import asyncio import threading import time import tkinter as tk from tkinter import ttk, scrolledtext, messagebox import logging import json import os import requests from io import BytesIO from PIL import Image, ImageTk, ImageDraw, ImageFont from http.server import HTTPServer, SimpleHTTPRequestHandler import webbrowser from TikTokLive.client.client import TikTokLiveClient from TikTokLive.client.logger import LogLevel from TikTokLive.events import ConnectEvent, DisconnectEvent, GiftEvent, CommentEvent, LiveEndEvent from tkinter import font import websockets import subprocess # Logger für diese Datei setzen logger = logging.getLogger(__name__) logger.setLevel(LogLevel.INFO.value) # Stream-Handler stream_handler = logging.StreamHandler() stream_handler.setLevel(LogLevel.INFO.value) # Formatter formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') stream_handler.setFormatter(formatter) # Logging zu der Datei logger.addHandler(stream_handler) class CustomFilter(logging.Filter): def filter(self, record): if "ConnectionClosedOK" in record.msg: return False return True logger.addFilter(CustomFilter()) CONFIG_FILE = "config.json" CHAT_HTML = "chat.html" FAVICON_ICO = "favicon.ico" CHAT_VIDEO = "chat.mp4" FONT_FILE = "arial.ttf" class TikTokRecorderGUI: def __init__(self, root): self.root = root self.root.title("TikTok Live Recorder") self.root.geometry("800x600") self.recording = False self.start_time = None self.bot_start_time = time.time() self.video_start_time = None self.stop_recording_event = threading.Event() self.unique_id = tk.StringVar(value="") self.session_id = tk.StringVar(value="") self.client = None self.connection_status = tk.StringVar(value="Nicht verbunden") self.recording_status = tk.StringVar(value="Nicht am aufnehmen") self.bot_duration_label = None self.video_duration_label = None self.update_duration_id = None self.live_status_label = None self.live_status_text = tk.StringVar(value="Nicht Live") self.room_id_label = None self.user_id_label = None self.room_id_text = tk.StringVar(value="") self.user_id_text = tk.StringVar(value="") self.is_manually_stopped = False self.chat_area = None self.main_task_future = None self.video_signal = True self.main_task = None self.images = {} self.chat_messages = [] self.server = None self.likes = 0 self.viewers = 0 self.diamonds = 0 self.chat_video = None self.font = None self.load_config() self.create_widgets() self.start_server() if os.path.exists(FONT_FILE): self.font = ImageFont.truetype(FONT_FILE, size=12) else: self.font = ImageFont.load_default() def load_config(self): if os.path.exists(CONFIG_FILE): with open(CONFIG_FILE, 'r') as f: try: config = json.load(f) self.session_id.set(config.get("session_id", "")) except json.JSONDecodeError: logger.warning("Fehler beim Laden der Konfigurationsdatei") def save_config(self): config = { "session_id": self.session_id.get() } with open(CONFIG_FILE, 'w') as f: json.dump(config, f) def start_server(self): threading.Thread(target=self.run_server, daemon=True).start() def run_server(self): class MyRequestHandler(SimpleHTTPRequestHandler): def do_GET(self): if self.path == "/": self.path = CHAT_HTML elif self.path == f"/{FAVICON_ICO}": self.path = FAVICON_ICO return SimpleHTTPRequestHandler.do_GET(self) self.server = HTTPServer(("localhost", 8000), MyRequestHandler) self.update_html() self.server.serve_forever() def create_widgets(self): # Label & Input für TikTok User ttk.Label(self.root, text="TikTok Benutzername (ohne @):").grid(row=0, column=0, padx=5, pady=5, sticky="w") ttk.Entry(self.root, textvariable=self.unique_id).grid(row=0, column=1, padx=5, pady=5, sticky="ew") ttk.Label(self.root, text="TikTok Session ID (optional):").grid(row=1, column=0, padx=5, pady=5, sticky="w") ttk.Entry(self.root, textvariable=self.session_id).grid(row=1, column=1, padx=5, pady=5, sticky="ew") # Status Labels ttk.Label(self.root, text="Verbindungsstatus:").grid(row=2, column=0, padx=5, pady=5, sticky="w") ttk.Label(self.root, textvariable=self.connection_status).grid(row=2, column=1, padx=5, pady=5, sticky="w") ttk.Label(self.root, text="Aufnahmestatus:").grid(row=3, column=0, padx=5, pady=5, sticky="w") ttk.Label(self.root, textvariable=self.recording_status).grid(row=3, column=1, padx=5, pady=5, sticky="w") ttk.Label(self.root, text="Bot Laufzeit:").grid(row=4, column=0, padx=5, pady=5, sticky="w") self.bot_duration_label = ttk.Label(self.root, text="00:00:00") self.bot_duration_label.grid(row=4, column=1, padx=5, pady=5, sticky="w") ttk.Label(self.root, text="Video Aufnahme Dauer:").grid(row=5, column=0, padx=5, pady=5, sticky="w") self.video_duration_label = ttk.Label(self.root, text="00:00:00") self.video_duration_label.grid(row=5, column=1, padx=5, pady=5, sticky="w") # Live Status, Room ID & User ID ttk.Label(self.root, text="Live Status:").grid(row=6, column=0, padx=5, pady=5, sticky="w") self.live_status_label = ttk.Label(self.root, textvariable=self.live_status_text, font=font.Font(weight="bold")) self.live_status_label.grid(row=6, column=1, padx=5, pady=5, sticky="w") ttk.Label(self.root, text="Room ID:").grid(row=7, column=0, padx=5, pady=5, sticky="w") self.room_id_label = ttk.Label(self.root, textvariable=self.room_id_text) self.room_id_label.grid(row=7, column=1, padx=5, pady=5, sticky="w") ttk.Label(self.root, text="User ID:").grid(row=8, column=0, padx=5, pady=5, sticky="w") self.user_id_label = ttk.Label(self.root, textvariable=self.user_id_text) self.user_id_label.grid(row=8, column=1, padx=5, pady=5, sticky="w") # Start/Stop/Resume Buttons self.start_stop_button = ttk.Button(self.root, text="Start Aufnahme", command=self.start_recording) self.start_stop_button.grid(row=9, column=0, columnspan=2, padx=10, pady=5) self.resume_button = ttk.Button(self.root, text="Wiederaufnahme", command=self.resume_recording) self.resume_button.grid(row=10, column=0, columnspan=2, padx=10, pady=5) self.resume_button.config(state=tk.DISABLED) self.quit_button = ttk.Button(self.root, text="Beenden", command=self.quit_program) self.quit_button.grid(row=11, column=0, columnspan=2, padx=10, pady=10) # Chat und Geschenke Anzeige ttk.Label(self.root, text="Chat und Geschenke:").grid(row=0, column=2, padx=5, pady=5, sticky="w") self.chat_area = scrolledtext.ScrolledText(self.root, width=60, height=25, bg="black", fg="white") self.chat_area.grid(row=1, column=2, rowspan=10, padx=5, pady=5, sticky="nsew") self.chat_area.config(state=tk.DISABLED) self.chat_area.tag_configure("user", foreground="lightblue") self.chat_area.tag_configure("giftname", foreground="yellow") self.chat_area.tag_configure("gift", foreground="green") def start_recording(self): if self.recording: self.stop_recording() self.start_stop_button.config(text="Start Aufnahme") self.recording = False self.recording_status.set("Nicht am aufnehmen") self.resume_button.config(state=tk.NORMAL) else: self.start_stop_button.config(text="Stop Aufnahme") self.recording = True self.recording_status.set("Am aufnehmen") self.video_start_time = None self.is_manually_stopped = False self.resume_button.config(state=tk.DISABLED) self.update_duration() self.chat_messages = [] self.main_task_future = threading.Thread(target=self.run_tiktok_bot, daemon=True) self.main_task_future.start() self.update_html() webbrowser.open("http://localhost:8000") def resume_recording(self): self.start_stop_button.config(text="Stop Aufnahme") self.recording = True self.recording_status.set("Am aufnehmen") self.video_start_time = None self.is_manually_stopped = False self.resume_button.config(state=tk.DISABLED) self.update_duration() self.main_task_future = threading.Thread(target=self.run_tiktok_bot, daemon=True) self.main_task_future.start() self.update_html() webbrowser.open("http://localhost:8000") def stop_recording(self): if self.client: asyncio.run(self.disconnect_from_live()) self.stop_recording_event.set() if self.update_duration_id: self.root.after_cancel(self.update_duration_id) self.is_manually_stopped = True self.resume_button.config(state=tk.NORMAL) if self.main_task: self.main_task.cancel() def update_duration(self): if self.recording: # Update Bot Duration bot_elapsed_time = time.time() - self.bot_start_time bot_hours = int(bot_elapsed_time // 3600) bot_minutes = int((bot_elapsed_time % 3600) // 60) bot_seconds = int(bot_elapsed_time % 60) self.bot_duration_label.config(text=f"{bot_hours:02d}:{bot_minutes:02d}:{bot_seconds:02d}") # Update Video Duration if self.video_start_time: video_elapsed_time = time.time() - self.video_start_time video_hours = int(video_elapsed_time // 3600) video_minutes = int((video_elapsed_time % 3600) // 60) video_seconds = int(video_elapsed_time % 60) self.video_duration_label.config(text=f"{video_hours:02d}:{video_minutes:02d}:{video_seconds:02d}") else: self.video_duration_label.config(text="00:00:00") self.update_duration_id = self.root.after(1000, self.update_duration) def run_tiktok_bot(self): asyncio.run(self.main_task_function()) async def main_task_function(self): self.main_task = asyncio.create_task(self._main_task()) await self.main_task async def _main_task(self): unique_id = self.unique_id.get() session_id = self.session_id.get() self.client: TikTokLiveClient = TikTokLiveClient( unique_id=f"@{unique_id}" ) if session_id: self.client.web.set_session_id(session_id) @self.client.on(ConnectEvent) async def on_connect(event: ConnectEvent): logger.info(f"Verbunden mit @{event.unique_id}!") self.connection_status.set("Verbunden") self.video_start_time = time.time() self.live_status_text.set("Live") self.live_status_label.config(foreground="green") if hasattr(self.client, 'room_info') and isinstance(self.client.room_info, dict): self.room_id_text.set(str(self.client.room_info.get('room_id', ''))) self.user_id_text.set(str(self.client.room_info.get('host_user_id', ''))) elif hasattr(self.client, 'room_info'): self.room_id_text.set(str(self.client.room_info.room_id)) self.user_id_text.set(str(self.client.room_info.host_user_id)) else: self.room_id_text.set("Keine Room ID") self.user_id_text.set("Keine User ID") try: self.likes = getattr(self.client.room_info,"likeCount",0) self.viewers = getattr(self.client.room_info,"viewerCount",0) self.diamonds = getattr(self.client.room_info,"diamondCount",0) except: logger.warning("Die Live Informationen konnten nicht geladen werden!") self.update_html() # Start a recording if hasattr(self.client, 'room_info'): self.client.web.fetch_video.start( output_fp=f"{event.unique_id}.mp4", room_info=self.client.room_info, output_format="mp4" ) self.video_signal = True @self.client.on(CommentEvent) async def on_comment(event: CommentEvent): formatted_message = f"{event.user.unique_id}: {event.comment}" self.display_message(formatted_message, "chat", user=event.user) @self.client.on(GiftEvent) async def on_gift(event: GiftEvent): logger.info("Ein Geschenk wurde empfangen!") gift_value = getattr(event.gift, "diamond_count", 0) gift_name = event.gift.name gift_image = getattr(event.gift.image, "url", None) # Can have a streak and streak is over if event.gift.streakable and not event.streaking: formatted_message = f"{event.user.unique_id} hat {event.repeat_count}x \"{gift_name}\" gesendet im Wert von: {gift_value}" # Cannot have a streak elif not event.gift.streakable: formatted_message = f"{event.user.unique_id} hat \"{gift_name}\" gesendet im Wert von: {gift_value}" self.display_message(formatted_message, "gift", user=event.user, gift_image=gift_image, gift_name=gift_name) @self.client.on(LiveEndEvent) async def on_live_end(event: LiveEndEvent): if hasattr(self.client, 'room_info'): self.likes = getattr(self.client.room_info,"likeCount",0) self.viewers = getattr(self.client.room_info,"viewerCount",0) self.diamonds = getattr(self.client.room_info,"diamondCount",0) self.update_html() @self.client.on(DisconnectEvent) async def on_disconnect(event: DisconnectEvent): """Stop the download when we disconnect""" logger.info("Verbindung getrennt!") self.connection_status.set("Verbindung getrennt") self.live_status_text.set("Nicht Live") self.live_status_label.config(foreground="red") self.room_id_text.set("") self.user_id_text.set("") self.video_signal = False self.likes = 0 self.viewers = 0 self.diamonds = 0 self.update_html() if hasattr(self.client, 'web') and hasattr(self.client.web, 'fetch_video') and self.client.web.fetch_video.is_recording: self.client.web.fetch_video.stop() async def check_loop(): # Run 24/7 while not self.stop_recording_event.is_set(): # Check if they're live if not self.is_manually_stopped: if session_id: logger.info("Session ID wurde gesetzt.") while True: try: if not await self.client.is_live(): logger.info("Client ist gerade nicht live. Überprüfe in 60 Sekunden erneut.") self.connection_status.set("Nicht Verbunden") self.live_status_text.set("Nicht Live") self.live_status_label.config(foreground="red") self.room_id_text.set("") self.user_id_text.set("") self.video_start_time = None self.video_signal = False self.likes = 0 self.viewers = 0 self.diamonds = 0 self.update_html() await asyncio.sleep(60) # Spamming the endpoint will get you blocked else: break except TikTokLive.client.errors.UserNotFoundError as e: logger.warning(f"Benutzer nicht gefunden, versuche erneut in 60 Sekunden. Fehler: {e}") self.connection_status.set("Nicht Verbunden") self.live_status_text.set("Nicht Live") self.live_status_label.config(foreground="red") self.room_id_text.set("") self.user_id_text.set("") self.video_start_time = None self.video_signal = False self.likes = 0 self.viewers = 0 self.diamonds = 0 self.update_html() await asyncio.sleep(60) # Connect once they become live logger.info("Gewünschter Client ist live!") await self.client.connect(fetch_room_info=True) while self.video_signal == True and not self.stop_recording_event.is_set() and await self.client.is_live(): # Video Check hier (Placeholder) # if (Video Signal check == false) # print("Kein Videosignal") # self.video_signal = False # break # else # print("Videosignal da") await asyncio.sleep(3) else: await asyncio.sleep(1) await check_loop() async def disconnect_from_live(self): if self.client: if hasattr(self.client, 'web') and hasattr(self.client.web, 'fetch_video') and self.client.web.fetch_video.is_recording: self.client.web.fetch_video.stop() if self.client._event_loop_task: await self.client.disconnect() def display_message(self, message, type="", user=None, gift_image=None, gift_name=None): formatted_message = "" if type == "gift": formatted_message = f"
" formatted_message+=f" {getattr(user, 'unique_id', 'Unbekannt')}" formatted_message+=f" hat " if gift_image: formatted_message+=f"" formatted_message+=f" {gift_name} {message}" formatted_message+="
" elif type == "chat": formatted_message = f"
" if user: formatted_message+=f" {user.unique_id}:" formatted_message+=f" {message}
" else: formatted_message = f"
{message}
" self.chat_messages.append(formatted_message) self.update_html() def update_html(self): with open(CHAT_HTML, "w") as f: f.write(""" TikTok Chat
Zuschauer: """+ str(self.viewers) + """ Likes: """+ str(self.likes) + """ Diamanten: """ + str(self.diamonds) + """
""") for msg in self.chat_messages: f.write(msg) f.write("""
""") with open(FAVICON_ICO, "wb") as f: response = requests.get("https://www.tiktok.com/favicon.ico", stream=True) response.raise_for_status() for block in response.iter_content(1024): f.write(block) async def quit_program(self): self.stop_recording() self.save_config() if self.server: self.server.shutdown() if self.client: await self.disconnect_from_live() self.root.destroy() if __name__ == "__main__": root = tk.Tk() gui = TikTokRecorderGUI(root) root.mainloop()