import os import subprocess import threading import asyncio import uuid import time import signal import sys import logging import platform import aiohttp import spider import stream # ----------------- Logging Setup ----------------- logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S" ) IS_WINDOWS = platform.system() == "Windows" # ----------------- Douyin Recorder ----------------- class DouyinRecorder: def __init__(self, audio_bitrate=128, max_minutes=8, loop_record=False): self.record_proc = None self.stop_flag = threading.Event() self.audio_bitrate = audio_bitrate self.start_time = None self.room_name = None self.out_file = None self.temp_file = None self.max_minutes = max_minutes self.loop_record = loop_record # <-- 新增循环开关 os.makedirs("records", exist_ok=True) # ---- 构建 FFmpeg 命令 ---- def _build_record_cmd(self, real_url): ts_name = f"{self.room_name}_{int(time.time())}.ts" self.temp_file = os.path.join("records", ts_name).replace("\\", "/") # 竖屏裁 16:9 + 模糊背景 vf_filter = ( "split [a][b];" "[a] scale=-1:720 [main];" "[b] scale=1280:720, boxblur=luma_radius=20:luma_power=1:chroma_radius=20:chroma_power=1 [bg];" "[bg][main] overlay=(W-w)/2:(H-h)/2" ) cmd = [ "ffmpeg", "-y", "-i", real_url, "-c:v", "libx264", "-preset", "veryfast", "-tune", "zerolatency", "-vf", vf_filter, "-c:a", "aac", "-b:a", f"{self.audio_bitrate}k", "-ar", "44100", "-ac", "2", "-f", "mpegts", self.temp_file ] return cmd # ---- 开始录制 ---- def start_record(self, real_url, room_name=None): self.room_name = room_name or f"抖音_{str(uuid.uuid4())[:8]}" logging.info(f"[{self.room_name}] 启动本地录制") threading.Thread(target=self._run_record_loop, args=(real_url,), daemon=True).start() threading.Thread(target=self._monitor_thread, daemon=True).start() def _start_new_record(self, real_url): record_cmd = self._build_record_cmd(real_url) self.record_proc = subprocess.Popen( record_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) logging.info(f"[{self.room_name}] 本地录制开始: {self.temp_file}") def monitor_proc(proc): try: for line in iter(proc.stderr.readline, b""): if self.stop_flag.is_set(): break if line: decoded = line.decode("utf-8", errors="ignore").strip() logging.debug(f"[{self.room_name} record]: {decoded}") except Exception as e: logging.error(f"[{self.room_name} record stderr异常: {e}") threading.Thread(target=monitor_proc, args=(self.record_proc,), daemon=True).start() # ---- 录制循环 ---- def _run_record_loop(self, real_url): first_run = True while not self.stop_flag.is_set(): try: self.start_time = time.time() self._start_new_record(real_url) while not self.stop_flag.is_set(): elapsed = time.time() - self.start_time # 如果循环开关关闭，则只录一次，达到 max_minutes 就停止 if not self.loop_record and elapsed >= self.max_minutes * 60: logging.info(f"[{self.room_name}] 达到 {self.max_minutes} 分钟，录制完成（循环关闭）") self.stop_flag.set() # break self.stop() # ⚠️ 这里自动退出整个程序 os._exit(0) # 如果循环开启，达到 max_minutes 就切分录制 if self.loop_record and elapsed >= self.max_minutes * 60: logging.info(f"[{self.room_name}] 达到 {self.max_minutes} 分钟，切分录制段") self._finalize_recording() self.start_time = time.time() self._start_new_record(real_url) if self.record_proc.poll() is not None: logging.warning(f"[{self.room_name}] FFmpeg 进程退出，1秒后重启") time.sleep(1) self._start_new_record(real_url) time.sleep(2) except Exception as e: logging.error(f"[{self.room_name}] FFmpeg异常: {e}") time.sleep(1) # ---- 停止 FFmpeg ---- def _terminate_proc(self): if self.record_proc and self.record_proc.poll() is None: self.record_proc.terminate() try: self.record_proc.wait(timeout=2) except subprocess.TimeoutExpired: self.record_proc.kill() self.record_proc = None # ---- 监控线程显示 ---- def _monitor_thread(self): while not self.stop_flag.is_set(): uptime = int(time.time() - self.start_time) if self.start_time else 0 m, s = divmod(uptime, 60) status = "录制中" if self.record_proc and self.record_proc.poll() is None else "启动中" print(f"\r[{self.room_name}] {status} | 运行时间: {m:02d}:{s:02d} | 音频码率: {self.audio_bitrate}k", end="") time.sleep(2) # ---- 停止 ---- def stop(self): logging.info(f"[{self.room_name}] 停止录制...") self.stop_flag.set() self._terminate_proc() self._finalize_recording() # ---- TS → MP4 ---- def _finalize_recording(self): if self.temp_file and os.path.exists(self.temp_file): mp4_name = self.temp_file.replace(".ts", f".mp4") logging.info(f"[{self.room_name}] 转 MP4: {mp4_name}") cmd = [ "ffmpeg", "-y", "-i", self.temp_file, "-c", "copy", "-movflags", "+faststart", mp4_name ] try: subprocess.run(cmd, check=True) size_mb = os.path.getsize(mp4_name) / (1024 * 1024) logging.info(f"[{self.room_name}] MP4 转换完成 | 大小: {size_mb:.2f} MB") except Exception as e: logging.error(f"[{self.room_name}] MP4 转换失败: {e}") finally: try: os.remove(self.temp_file) except Exception: pass # ----------------- Douyin 流解析 ----------------- async def get_douyin_real_url(record_url, record_quality="原画", proxy_addr=None, cookies=''): headers = { "User-Agent": ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36"), "Referer": "https://www.douyin.com/", "Origin": "https://www.douyin.com", "Accept": "application/json, text/javascript, */*; q=0.01", } session_args = {"headers": headers} if proxy_addr: session_args["proxy"] = proxy_addr async with aiohttp.ClientSession(**session_args) as session: for attempt in range(3): try: async with session.get(record_url, cookies=cookies) as resp: text = await resp.text() if not text or "captcha" in text.lower(): logging.warning(f"第{attempt+1}次尝试获取流失败，返回空或验证码，重试...") await asyncio.sleep(1) continue if 'v.douyin.com' not in record_url and '/user/' not in record_url: json_data = await spider.get_douyin_stream_data( url=record_url, proxy_addr=proxy_addr, cookies=cookies) else: json_data = await spider.get_douyin_app_stream_data( url=record_url, proxy_addr=proxy_addr, cookies=cookies) port_info = await stream.get_douyin_stream_url(json_data, record_quality, proxy_addr) if port_info and port_info.get("record_url"): return port_info.get("record_url") logging.warning(f"第{attempt+1}次尝试解析失败，port_info为空") await asyncio.sleep(1) except Exception as e: logging.warning(f"第{attempt+1}次尝试抓取抖音流异常: {e}") await asyncio.sleep(1) logging.error("多次尝试后仍获取抖音真实流失败") return None def start_douyin_record(record_url, room_name=None, proxy_addr=None, cookies='', audio_bitrate=128): real_url = asyncio.run(get_douyin_real_url(record_url, proxy_addr=proxy_addr, cookies=cookies)) if not real_url: logging.error("获取真实直播流失败") return None logging.info(f"实际直播流: {real_url}") recorder = DouyinRecorder(audio_bitrate=audio_bitrate) recorder.start_record(real_url, room_name) return recorder # ----------------- Main ----------------- if __name__ == "__main__": DOUYIN_URL = "https://live.douyin.com/163813589919" recorder_instance = start_douyin_record(DOUYIN_URL) if not recorder_instance: logging.error("录制实例创建失败，程序退出") sys.exit(1) def signal_handler(sig, frame): if recorder_instance: recorder_instance.stop() sys.exit(0) signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) logging.info("程序已启动，本地录制中，按 Ctrl+C 停止...") while True: time.sleep(60)