82 lines
2.9 KiB
Python
82 lines
2.9 KiB
Python
# mjpeg_streamer.pynano
|
||
import threading
|
||
import time
|
||
from flask import Flask, Response, render_template, request, redirect, session, url_for
|
||
import cv2
|
||
|
||
# ====== 新增：登录配置 ======
|
||
AUTO_LOGIN = None # 👈 设置为 True 可跳过登录
|
||
VALID_USER = {"username": "admin", "password": "admin"}
|
||
|
||
class MJPEGServer:
|
||
def __init__(self, frame_buffer, host="0.0.0.0", port=8080):
|
||
self.frame_buffer = frame_buffer
|
||
self.host = host
|
||
self.port = port
|
||
self.app = Flask(__name__)
|
||
self.app.secret_key = 'your-secret-key-change-in-prod' # 用于 session
|
||
|
||
# 路由
|
||
self.app.add_url_rule('/', 'index', self.index)
|
||
self.app.add_url_rule('/login', 'login', self.login, methods=['GET', 'POST'])
|
||
self.app.add_url_rule('/logout', 'logout', self.logout)
|
||
self.app.add_url_rule('/video_feed', 'video_feed', self.video_feed)
|
||
|
||
# 静态文件自动托管（Layui）
|
||
self.app.static_folder = 'static'
|
||
|
||
def is_logged_in(self):
|
||
return session.get('logged_in', False)
|
||
|
||
def check_auth(self):
|
||
if AUTO_LOGIN:
|
||
session['logged_in'] = True
|
||
return True
|
||
return self.is_logged_in()
|
||
|
||
def index(self):
|
||
if not self.check_auth():
|
||
return redirect(url_for('login'))
|
||
return render_template('index.html')
|
||
|
||
def login(self):
|
||
if request.method == 'POST':
|
||
username = request.form['username']
|
||
password = request.form['password']
|
||
if username == VALID_USER['username'] and password == VALID_USER['password']:
|
||
session['logged_in'] = True
|
||
return redirect(url_for('index'))
|
||
else:
|
||
return '<script>alert("用户名或密码错误");window.history.back();</script>'
|
||
return render_template('login.html')
|
||
|
||
def logout(self):
|
||
session.pop('logged_in', None)
|
||
return redirect(url_for('login'))
|
||
|
||
def video_feed(self):
|
||
if not self.check_auth():
|
||
return '', 403
|
||
return Response(self._gen(),
|
||
mimetype='multipart/x-mixed-replace; boundary=frame')
|
||
|
||
def _gen(self):
|
||
while True:
|
||
success, frame = self.frame_buffer.get_frame()
|
||
if not success or frame is None:
|
||
time.sleep(0.1)
|
||
continue
|
||
ret, buffer = cv2.imencode('.jpg', frame, [int(cv2.IMWRITE_JPEG_QUALITY), 70])
|
||
if not ret:
|
||
continue
|
||
yield (b'--frame\r\n'
|
||
b'Content-Type: image/jpeg\r\n\r\n' + buffer.tobytes() + b'\r\n')
|
||
|
||
def start(self):
|
||
thread = threading.Thread(
|
||
target=self.app.run,
|
||
kwargs={'host': self.host, 'port': self.port, 'debug': False, 'use_reloader': False},
|
||
daemon=True
|
||
)
|
||
thread.start()
|
||
print(f"[MJPEG] Web 系统已启动，访问 http://{self.host}:{self.port}") |