# douyin_srs_ffplay.py import queue import unicodedata import traceback import subprocess import threading import re import time import datetime class StreamEnded(Exception): pass def sanitize_title(text: str, max_len: int = 120) -> str: if not text: return "" text = unicodedata.normalize("NFKC", text) text = ''.join(ch for ch in text if unicodedata.category(ch)[0] != 'C') text = ''.join(ch for ch in text if ord(ch) <= 0xFFFF) return text.strip()[:max_len] def _stderr_reader_thread(proc, q): try: while True: line = proc.stderr.readline() if not line: break q.put(line) except: pass finally: try: q.put(None) except: pass def cleanup_proc(proc, name="进程"): if proc and proc.poll() is None: try: proc.terminate() proc.wait(timeout=8) except subprocess.TimeoutExpired: proc.kill() except: proc.kill() print(f"[INFO] 已强制终止 {name}") def start_douyin_srs_ffplay( title, anchor_name, real_url, rec_info, record_name, recording, recording_time_list, running_list, monitoring, clear_record_info, color_obj ): # ====================== 配置参数 ====================== SRS_RTMP_URL = "rtmp://127.0.0.1/live/douyin" SRS_PLAY_URL = "http://127.0.0.1:8080/live/douyin.flv" MAX_RETRY_DELAY = 120 INITIAL_RETRY_DELAY = 3 NO_FRAME_TIMEOUT = 180 STALE_OUTPUT_SECONDS = 180 FFPLAY_RESTART_INTERVAL = 3600 * 6 END_KEYWORDS = [ "connection reset by peer", "failed to write trailer", "invalid data found", "server returned 403", "server returned 404", "ingestion error", "access denied", "packet too short", "client disconnected", "live has ended", "no route to host", "connection timed out", ] timestamp_str = time.strftime("%Y-%m-%d_%H-%M-%S", time.localtime()) _clean_title = sanitize_title(title) stream_title = sanitize_title(f"{anchor_name}{('_' + _clean_title) if _clean_title else ''}_{timestamp_str}") print(f"[INFO] {rec_info} 启动 SRS → ffplay HDMI 转播（低CPU原画版）: {stream_title}") douyin_headers = ( "User-Agent: Mozilla/5.0 (iPhone; CPU iPhone OS 18_1 like Mac OS X) AppleWebKit/605.1.15\r\n" "Referer: https://live.douyin.com/\r\n" "Origin: https://live.douyin.com\r\n" ) # ====================== FFmpeg 推流到 SRS（直抄原画，低CPU） ====================== ffmpeg_command = [ "ffmpeg", "-y", "-loglevel", "info", "-nostdin", "-headers", douyin_headers, "-rw_timeout", "30000000", "-reconnect", "1", "-reconnect_at_eof", "1", "-reconnect_streamed", "1", "-reconnect_delay_max", "7", "-fflags", "+genpts+discardcorrupt+igndts", "-err_detect", "ignore_err", "-i", real_url, "-c:v", "copy", "-c:a", "copy", "-f", "flv", SRS_RTMP_URL ] # ====================== ffplay 最佳参数（低CPU + 零漂移 + 全屏） ====================== ffplay_cmd = [ "ffplay", "-fflags", "nobuffer+genpts", "-flags", "low_delay", "-framedrop", "-sync", "ext", "-vf", "transpose=2", "-autoexit", "-fs", "-loglevel", "quiet", "-i", SRS_PLAY_URL ] ffmpeg_proc = None ffplay_proc = None stderr_q = queue.Queue() reader_t = None retry_delay = INITIAL_RETRY_DELAY last_ffplay_restart = time.time() try: while True: current_time = time.time() if ffplay_proc and (current_time - last_ffplay_restart > FFPLAY_RESTART_INTERVAL): print("[SCHEDULE] 6小时预防性重启 ffplay") cleanup_proc(ffplay_proc, "ffplay") ffplay_proc = None if ffplay_proc is None or ffplay_proc.poll() is not None: if ffplay_proc is not None: print("[WARN] ffplay 退出，重启中...") print(f"[INFO][{datetime.datetime.now():%H:%M:%S}] 启动 ffplay 播放 SRS 原画流") ffplay_proc = subprocess.Popen( ffplay_cmd, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True ) time.sleep(2) last_ffplay_restart = time.time() if retry_delay > INITIAL_RETRY_DELAY: print(f"[WARN] 等待 {retry_delay}s 后重连抖音源...") time.sleep(retry_delay) retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY) cleanup_proc(ffmpeg_proc, "FFmpeg") print(f"[INFO][{datetime.datetime.now():%H:%M:%S}] 启动 FFmpeg 直抄推流到 SRS") ffmpeg_proc = subprocess.Popen( ffmpeg_command, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True, bufsize=1, universal_newlines=True ) ffmpeg_proc.last_out_ts = time.time() stderr_q = queue.Queue() reader_t = threading.Thread(target=_stderr_reader_thread, args=(ffmpeg_proc, stderr_q)) reader_t.daemon = True reader_t.start() last_frame_time = time.time() while True: try: line = stderr_q.get(timeout=1.5) except queue.Empty: if time.time() - ffmpeg_proc.last_out_ts > STALE_OUTPUT_SECONDS: print("[WARN] FFmpeg 僵死，重启") break continue if line is None: break line = line.strip() if not line: continue ffmpeg_proc.last_out_ts = time.time() print(f"[FFmpeg] {line}") line_lower = line.lower() if any(kw in line_lower for kw in END_KEYWORDS): print("[INFO] 检测到下播关键字，停止转播") raise StreamEnded() if re.search(r'frame=\s*\d+', line_lower): last_frame_time = time.time() retry_delay = INITIAL_RETRY_DELAY if time.time() - last_frame_time > NO_FRAME_TIMEOUT: print(f"[INFO] {NO_FRAME_TIMEOUT}s 无新帧，判定下播") raise StreamEnded() except StreamEnded: print(f"[INFO] 主播已下播，停止 SRS → ffplay HDMI 转播 → {record_name}") except KeyboardInterrupt: print("[INFO] 用户手动中断") except Exception as e: print(f"[FATAL] 未知异常: {e}") traceback.print_exc() time.sleep(10) finally: cleanup_proc(ffmpeg_proc, "FFmpeg") cleanup_proc(ffplay_proc, "ffplay") try: recording.discard(record_name) recording_time_list.pop(record_name, None) if record_name in running_list: running_list.remove(record_name) # global monitoring # monitoring = max(0, monitoring - 1) monitoring[0] = max(0, monitoring[0] - 1) # 修改为列表元素 clear_record_info(record_name, real_url) except: pass color_obj.print_colored(f"[{record_name}] SRS → ffplay HDMI 转播已彻底停止并清理完毕\n", color_obj.GREEN)