From 25be2d5b1236825ebbdf9bf6cb96a865f7aa74d5 Mon Sep 17 00:00:00 2001 From: zhengsl Date: Sun, 6 Sep 2026 21:05:03 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E4=BA=86=E6=88=91=E7=9A=84?= =?UTF-8?q?=E6=91=84=E5=83=8F=E5=A4=B4=E6=8E=A5=E5=85=A5=E8=A7=86=E9=A2=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 1 + scripts/local_webcam.py | 226 ++++++++++++++++++------------------- tests/test_local_webcam.py | 53 +-------- 3 files changed, 112 insertions(+), 168 deletions(-) diff --git a/.gitignore b/.gitignore index da193eb..a632c2b 100644 --- a/.gitignore +++ b/.gitignore @@ -59,3 +59,4 @@ _apply_windows.py report_assets/ report_render_runtime/ 系统部署性能分析评估报告.docx +monitor_runtime/local-webcam/ diff --git a/scripts/local_webcam.py b/scripts/local_webcam.py index 6f355a8..8a4d56d 100644 --- a/scripts/local_webcam.py +++ b/scripts/local_webcam.py @@ -1,36 +1,32 @@ -"""Open the local webcam and register it without starting ZLMediaKit or FFmpeg.""" +"""Open the local webcam, register it, and push it into ZLMediaKit via FFmpeg.""" import argparse -import configparser from datetime import datetime -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import os from pathlib import Path -import secrets import socket import sqlite3 import subprocess import sys -import threading import time import psutil -import requests ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from scripts.local_server import PYTHON, local_environment, recorded_process as web_process -WORK = ROOT / '.runtime' / 'local-webcam' +# The repository's historical .runtime tree may be owned by an elevated +# services process. Keep this camera-only helper's state in a user-writable +# runtime directory so it can start alongside that process. +WORK = ROOT / 'monitor_runtime' / 'local-webcam' SCRIPT = Path(__file__).resolve() PID_FILE = WORK / 'process.json' STATUS = WORK / 'status.json' STOP = WORK / 'stop.request' STREAM = 'laptop_cam' DEVICE = 'Integrated Camera' -MARKER = 'Managed by scripts/local_webcam.py; camera registration only; no ZLM' -PORTS = {'http': 10002, 'rtsp': 10554, 'rtmp': 10935} -PUSH_URL = f'rtmp://127.0.0.1:{PORTS["rtmp"]}/live/{STREAM}' +MARKER = 'Managed by scripts/local_webcam.py;' CAMERA_INDEX = int(os.environ.get('MONITOR_WEBCAM_INDEX', '0')) @@ -40,100 +36,52 @@ def write_json(path, data): os.replace(temp, path) -def make_config(source, api_secret, hook_url): - config = configparser.ConfigParser(interpolation=None, strict=False) - config.optionxform = str - config.read_string(source) - changes = { - 'general': {'listen_ip': '127.0.0.1', 'check_nvidia_dev': '0', 'enable_ffmpeg_log': '0', - 'mediaServerId': 'local-webcam', 'wait_add_track_ms': '100', 'maxStreamWaitMS': '3000'}, - 'api': {'secret': api_secret, 'apiDebug': '0', 'downloadRoot': str(WORK / 'www'), - 'snapRoot': str(WORK / 'www'), 'defaultSnap': ''}, - 'cluster': {'origin_url': ''}, - 'http': {'port': str(PORTS['http']), 'sslport': '0', 'rootPath': str(WORK / 'www'), - 'virtualPath': '', 'dirMenu': '0', 'allow_ip_range': '127.0.0.1', 'allow_cross_domains': '0'}, - 'rtmp': {'port': str(PORTS['rtmp']), 'sslport': '0'}, - 'rtsp': {'port': str(PORTS['rtsp']), 'sslport': '0', 'rtpTransportType': '0'}, - 'rtc': {'port': '0', 'tcpPort': '0'}, 'rtp_proxy': {'port': '0', 'dumpDir': ''}, - 'srt': {'port': '0'}, 'shell': {'port': '0'}, 'onvif': {'port': '0'}, - 'protocol': {'enable_audio': '0', 'add_mute_audio': '0', 'enable_mp4': '0', - 'enable_hls': '0', 'enable_hls_fmp4': '0', 'enable_ts': '0', - 'enable_rtsp': '1', 'enable_rtmp': '1', 'enable_fmp4': '1', - 'continue_push_ms': '0', 'auto_close': '0', - 'mp4_save_path': str(WORK / 'www'), 'hls_save_path': str(WORK / 'www')}, - 'record': {'enableFmp4': '0'}, 'hls': {'broadcastRecordTs': '0', 'segKeep': '0'}, - # ZLM must never launch another ffmpeg process or save a snapshot here. - 'ffmpeg': {'bin': '', 'cmd': '', 'snap': '', 'log': ''}, - } - for section, values in changes.items(): - if not config.has_section(section): - config.add_section(section) - config[section].update(values) - if config.has_section('hook'): - config.remove_section('hook') - config['hook'] = {'enable': '1', 'on_publish': hook_url, 'timeoutSec': '5', 'retry': '0'} - return config +def zlm_rtmp_port(): + try: + cfg = json.loads((ROOT / 'config.json').read_text(encoding='utf-8')) + return int(cfg.get('mediaRtmpPort') or 10935) + except Exception: + return 10935 -def capture_command(ffmpeg): - return [ffmpeg, '-hide_banner', '-loglevel', 'warning', '-f', 'dshow', - '-rtbufsize', '32M', '-video_size', '640x480', '-framerate', '15', - '-i', 'video=' + DEVICE, '-map', '0:v:0', '-an', '-sn', '-dn', - '-c:v', 'libx264', '-preset', 'ultrafast', '-tune', 'zerolatency', - '-pix_fmt', 'yuv420p', '-g', '30', '-threads', '2', '-f', 'flv', PUSH_URL] +def zlm_ready(port): + try: + with socket.create_connection(('127.0.0.1', port), timeout=1.0): + return True + except OSError: + return False -def allowed_publish(data): - return (data.get('app') == 'live' and data.get('stream') == STREAM - and data.get('schema') == 'rtmp' and data.get('ip') == '127.0.0.1') +def start_push(width, height, fps, port): + """Pipe OpenCV BGR frames into FFmpeg and push H264 to ZLM over RTMP.""" + ffmpeg = os.environ.get('MONITOR_FFMPEG') or 'ffmpeg' + fps = max(1, min(int(round(fps)) or 15, 60)) + cmd = [ffmpeg, '-hide_banner', '-loglevel', 'warning', + '-f', 'rawvideo', '-pix_fmt', 'bgr24', '-s', '%dx%d' % (width, height), + '-r', str(fps), '-i', '-', + '-an', '-c:v', 'libx264', '-preset', 'veryfast', '-tune', 'zerolatency', + '-pix_fmt', 'yuv420p', '-g', str(fps), + '-f', 'flv', 'rtmp://127.0.0.1:%d/live/%s' % (port, STREAM)] + # stdout/stderr inherit the supervisor log opened by start(). + return subprocess.Popen(cmd, stdin=subprocess.PIPE, + creationflags=subprocess.CREATE_NO_WINDOW) -def hook_server(): - route = '/publish/' + secrets.token_urlsafe(32) - - class Handler(BaseHTTPRequestHandler): - def log_message(self, *args): - pass # Never log the per-run hook token or payload. - - def do_POST(self): - accepted = False - try: - size = int(self.headers.get('Content-Length', '0')) - if self.path == route and self.client_address[0] == '127.0.0.1' and 0 < size <= 16384: - self.connection.settimeout(3) - accepted = allowed_publish(json.loads(self.rfile.read(size))) - except (ValueError, OSError, TypeError): - pass - data = {'code': 0 if accepted else -1, 'msg': 'video-only local publisher', - 'enable_audio': False, 'add_mute_audio': False, 'enable_mp4': False, - 'enable_hls': False, 'enable_hls_fmp4': False} - body = json.dumps(data).encode() - self.send_response(200) - self.send_header('Content-Type', 'application/json') - self.send_header('Content-Length', str(len(body))) - self.end_headers() - self.wfile.write(body) - - server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) - threading.Thread(target=server.serve_forever, daemon=True).start() - return server, f'http://127.0.0.1:{server.server_port}{route}' - - -def check_connections(processes, allowed_ports): - listeners = set() - for proc in processes: - for conn in proc.connections(kind='inet'): - if conn.type == socket.SOCK_DGRAM: - raise RuntimeError('Unexpected UDP socket; stopping webcam') - if conn.laddr and conn.laddr.ip not in ('127.0.0.1', '::1'): - raise RuntimeError(f'Non-loopback binding {conn.laddr} ({conn.status}); stopping webcam') - if conn.raddr and conn.raddr.ip not in ('127.0.0.1', '::1'): - raise RuntimeError('Non-loopback peer; stopping webcam') - if conn.status == psutil.CONN_LISTEN: - if conn.laddr.port not in allowed_ports: - raise RuntimeError('Unexpected listener; stopping webcam') - listeners.add(conn.laddr.port) - return listeners +def stop_push(push): + if push is None or push.poll() is not None: + return + try: + # Closing stdin sends EOF so FFmpeg finalizes the FLV stream and exits. + push.stdin.close() + push.wait(timeout=5) + except Exception: + pass + if push.poll() is None: + push.kill() + try: + push.wait(timeout=5) + except subprocess.TimeoutExpired: + pass def owned_process(): @@ -164,10 +112,12 @@ def terminate(child): def run(): - """Keep the physical camera open and register an inventory-only stream.""" + """Keep the physical camera open, register it, and push it into ZLM.""" os.environ.update(local_environment()) camera = row = None StreamModel = None + push = None + rtmp_port = zlm_rtmp_port() try: web = web_process() if not web: @@ -203,35 +153,58 @@ def run(): if existing and not (existing.remark or '').startswith('Managed by scripts/local_webcam.py;'): raise RuntimeError('laptop_cam belongs to another stream; refusing to overwrite') camera_url = f'camera://index/{CAMERA_INDEX}' + # pull_stream_type=32 (passive RTMP) lets the ZLM on_publish hook accept + # this stream and write back forward_state for the play button. row, _ = StreamModel.objects.get_or_create(code=STREAM, defaults={ 'user_id': 0, 'sort': 0, 'app': 'live', 'name': STREAM, 'nickname': '笔记本内置摄像头', - 'remark': MARKER, 'pull_stream_type': 0, 'pull_stream_transfer_mode': 0, + 'remark': MARKER, 'pull_stream_type': 32, 'pull_stream_transfer_mode': 0, 'pull_stream_url': camera_url, 'pull_stream_ip': '127.0.0.1', 'pull_stream_port': 0, - 'forward_state': 0, 'is_audio': 0, 'record_enable': 0, 'state': 0, - 'camera_name': DEVICE, 'camera_device_id': 'local-webcam'}) - if not (row.remark or '').startswith('Managed by scripts/local_webcam.py;'): + 'forward_state': 0, 'is_audio': 0, 'pull_stream_username': '', 'pull_stream_password': '', + 'snap_filepath': '', 'record_enable': 0, 'state': 0, 'camera_name': DEVICE, + 'camera_device_id': 'local-webcam', 'camera_manufacturer': 'Microsoft', + 'camera_owner': '', 'camera_model': 'Integrated Camera', 'camera_parent_id': '', + 'camera_civilcode': ''}) + if not (row.remark or '').startswith(MARKER): raise RuntimeError('Stream ownership changed') StreamModel.objects.filter(pk=row.pk).update( - remark=MARKER, pull_stream_type=0, pull_stream_url=camera_url, - pull_stream_ip='127.0.0.1', pull_stream_port=0, forward_state=0, - record_enable=0, is_audio=0, camera_name=DEVICE, camera_device_id='local-webcam') - print(f'Ready: laptop_cam ({DEVICE}, {width}x{height}/{fps:.0f}fps, camera only; ZLM not started)', flush=True) + remark=MARKER, pull_stream_type=32, pull_stream_ip='127.0.0.1', pull_stream_port=0, + forward_state=0, record_enable=0, is_audio=0, camera_name=DEVICE, camera_device_id='local-webcam') + print(f'Ready: laptop_cam ({DEVICE}, {width}x{height}/{fps:.0f}fps; ' + f'RTMP push -> 127.0.0.1:{rtmp_port}/live/{STREAM} once ZLM is up)', flush=True) + last_retry = 0.0 + last_status = 0.0 while not STOP.exists(): if not web.is_running(): raise RuntimeError('Local web server exited; stopping camera') ok, frame = camera.read() if not ok or frame is None: raise RuntimeError('Camera frame read failed') - write_json(STATUS, {'ready': True, 'stream_id': row.pk, 'stream': STREAM, 'device': DEVICE, - 'camera_index': CAMERA_INDEX, 'width': width, 'height': height, 'fps': fps, - 'media_started': False, 'zlm_started': False, 'audio': False, 'recording': False, - 'sip': False, 'checked_at': datetime.now().isoformat(timespec='seconds')}) - time.sleep(0.2) + now = time.monotonic() + if push is not None and push.poll() is not None: + push = None # FFmpeg exited (ZLM down or connection lost); retry below + if push is None and now - last_retry >= 2.0: + last_retry = now + if zlm_ready(rtmp_port): + print('ZLM detected; starting RTMP push', flush=True) + push = start_push(width, height, fps, rtmp_port) + if push is not None and push.poll() is None: + try: + push.stdin.write(frame.tobytes()) + except (BrokenPipeError, OSError): + push = None # FFmpeg died mid-frame; retry on the next cooldown + if now - last_status >= 1.0: + last_status = now + write_json(STATUS, {'ready': True, 'stream_id': row.pk, 'stream': STREAM, 'device': DEVICE, + 'camera_index': CAMERA_INDEX, 'width': width, 'height': height, 'fps': fps, + 'media_started': bool(push is not None and push.poll() is None), + 'zlm_started': zlm_ready(rtmp_port), 'audio': False, 'recording': False, + 'sip': False, 'checked_at': datetime.now().isoformat(timespec='seconds')}) finally: + stop_push(push) if camera is not None: camera.release() if row is not None and StreamModel is not None: - StreamModel.objects.filter(pk=row.pk, remark=MARKER).update(forward_state=0) + StreamModel.objects.filter(pk=row.pk, remark__startswith=MARKER).update(forward_state=0) write_json(STATUS, {'ready': False, 'stopped_at': datetime.now().isoformat(timespec='seconds')}) @@ -243,7 +216,9 @@ def start(): return STOP.unlink(missing_ok=True) STATUS.unlink(missing_ok=True) - with (WORK / 'supervisor.log').open('ab') as log: + # Use a fresh per-run log so a previous elevated run cannot lock the file. + log_path = WORK / ('supervisor-' + datetime.now().strftime('%Y%m%d-%H%M%S-%f') + '.log') + with log_path.open('ab') as log: child = subprocess.Popen([str(PYTHON), str(SCRIPT), 'run'], cwd=ROOT, env=local_environment(), stdin=subprocess.DEVNULL, stdout=log, stderr=log, creationflags=subprocess.CREATE_NO_WINDOW) @@ -251,9 +226,9 @@ def start(): write_json(PID_FILE, {'pid': proc.pid, 'created': proc.create_time(), 'root': str(ROOT)}) for _ in range(60): if child.poll() is not None: - raise RuntimeError('Webcam start failed; inspect .runtime/local-webcam/supervisor.log') + raise RuntimeError(f'Webcam start failed; inspect {log_path}') if STATUS.exists() and json.loads(STATUS.read_text(encoding='utf-8')).get('ready'): - print('Ready: http://127.0.0.1:10001/stream/index | laptop_cam | camera only; ZLM not started') + print('Ready: http://127.0.0.1:10001/stream/index | laptop_cam | RTMP push starts when ZLM is reachable') return time.sleep(1) STOP.touch() @@ -269,9 +244,24 @@ def stop(): try: proc.wait(timeout=20) except psutil.TimeoutExpired: - raise RuntimeError('Graceful stop timed out; inspect owned processes before retrying') + # The finally block normally stops the FFmpeg push child; force-stop + # both when the graceful path hangs so no push process leaks. + try: + children = proc.children(recursive=True) + except psutil.Error: + children = [] + for child in children: + try: + child.kill() + except psutil.Error: + pass + try: + proc.kill() + proc.wait(timeout=5) + except psutil.Error: + pass PID_FILE.unlink(missing_ok=True) - print('Webcam stopped; web server and registered camera record retained') + print('Webcam stopped; RTMP push ended, web server and camera record retained') if __name__ == '__main__': diff --git a/tests/test_local_webcam.py b/tests/test_local_webcam.py index 25a1f9f..f38f00d 100644 --- a/tests/test_local_webcam.py +++ b/tests/test_local_webcam.py @@ -1,59 +1,12 @@ -import configparser -import socket import unittest -from types import SimpleNamespace -from unittest import mock - -import psutil from scripts import local_webcam as webcam class LocalWebcamTests(unittest.TestCase): - def test_config_disables_audio_recording_and_other_protocols(self): - config = webcam.make_config('[hook]\non_server_started=http://remote/\n[protocol]\nenable_mp4=1\n', - 'test-secret', 'http://127.0.0.1:12345/publish/test') - self.assertEqual(config['general']['listen_ip'], '127.0.0.1') - for name in ('enable_audio', 'add_mute_audio', 'enable_mp4', 'enable_hls', 'enable_hls_fmp4'): - self.assertEqual(config['protocol'][name], '0') - for section in ('rtc', 'rtp_proxy', 'srt', 'shell', 'onvif'): - self.assertEqual(config[section]['port'], '0') - self.assertNotIn('on_server_started', config['hook']) - self.assertEqual(config['rtsp']['rtpTransportType'], '0') - self.assertEqual(config['ffmpeg']['bin'], '') - - def test_command_is_video_only_and_fixed_loopback_destination(self): - command = webcam.capture_command('ffmpeg.exe') - self.assertIn('-an', command) - self.assertEqual(command[command.index('-i') + 1], 'video=Integrated Camera') - self.assertEqual(command[command.index('-map') + 1], '0:v:0') - self.assertEqual(command[-1], 'rtmp://127.0.0.1:10935/live/laptop_cam') - self.assertNotIn('audio=', ' '.join(command)) - - def test_publish_gate_rejects_other_streams_and_remote_publishers(self): - data = {'app': 'live', 'stream': 'laptop_cam', 'schema': 'rtmp', 'ip': '127.0.0.1'} - self.assertTrue(webcam.allowed_publish(data)) - for key, value in [('stream', 'other'), ('ip', '192.168.1.2'), ('app', 'rtp'), ('schema', 'rtsp')]: - self.assertFalse(webcam.allowed_publish({**data, key: value})) - - def test_network_guard_rejects_exposure_udp_and_outbound_connections(self): - proc = mock.Mock() - local = SimpleNamespace(ip='127.0.0.1', port=10935) - good = SimpleNamespace(type=socket.SOCK_STREAM, laddr=local, raddr=(), status=psutil.CONN_LISTEN) - proc.connections.return_value = [good] - self.assertEqual(webcam.check_connections([proc], {10935}), {10935}) - proc.connections.return_value = [SimpleNamespace( - type=socket.SOCK_STREAM, laddr=SimpleNamespace(ip='::1', port=60001), - raddr=SimpleNamespace(ip='::1', port=60002), status=psutil.CONN_ESTABLISHED)] - self.assertEqual(webcam.check_connections([proc], {10935}), set()) - for bad in [dict(type=socket.SOCK_DGRAM), - dict(laddr=SimpleNamespace(ip='0.0.0.0', port=10935)), - dict(raddr=SimpleNamespace(ip='8.8.8.8', port=443))]: - proc.connections.return_value = [SimpleNamespace(**{**vars(good), **bad})] - with self.assertRaises(RuntimeError): - webcam.check_connections([proc], {10935}) - def test_webcam_does_not_bootstrap_full_service_manager(self): source = webcam.SCRIPT.read_text(encoding='utf-8') self.assertNotIn('get_service_manager().start()', source) self.assertNotIn('shell=True', source) - self.assertIn('ServiceLeaderLock()', source) + self.assertNotIn('ServiceLeaderLock()', source) + self.assertNotIn("mediaStartPath", source) + self.assertNotIn("subprocess.Popen([str(exe)", source)