flask实现直播功能

import pyaudio import config from flask import Flask, session, g, Response, render_template from blueprints import qa_bp, user_bp from decorators import login_required from exts import mail, db from flask_migrate import Migrate from models import UserModel import socket from io import BytesIO from pyautogui import screenshot import cv2 app = Flask(__name__) app.config.from_object(config) app.secret_key = 'keduoli' db.init_app(app) mail.init_app(app) migrate = Migrate(app, db) app.register_blueprint(qa_bp) app.register_blueprint(user_bp) @app.before_request def before_request(): user_id = session.get('user_id') if user_id: try: user = UserModel.query.get(user_id) g.user = user except: g.user = None @app.context_processor def context_processor(): if hasattr(g, 'user'): return {'user': g.user} else: return {} def screencap(): while True: img = screenshot() buffer = BytesIO() img.save(buffer, 'jpeg') yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + buffer.getvalue() + b'\r\n\r\n') def getIp(): ip = '127.0.0.1' try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(('8.8.8.8', 80)) ip = s.getsockname()[0] s.close() except Exception as e: print(str(e)[0:0] + '127.0.0.1') return ip @app.route('/zhibo') @login_required def zhibo(): users = UserModel.query.all() return render_template('zhibo.html',users=users) @app.route('/video') def video(): return Response(screencap(), mimetype='multipart/x-mixed-replace; boundary=frame') @app.route('/cameraVideo') @login_required def cameraVideo(): return render_template('cameraVideo.html') def gen(): vid = cv2.VideoCapture(0) while True: return_value, frame = vid.read() image = cv2.imencode('.jpg', frame)[1].tobytes() yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + image + b'\r\n') @app.route('/video_feed') def video_feed(): return Response(gen(), mimetype='multipart/x-mixed-replace; boundary=frame') FORMAT = pyaudio.paInt16 CHANNELS = 2 RATE = 44100 CHUNK = 1024 RECORD_SECONDS = 5 audio1 = pyaudio.PyAudio() def genHeader(sampleRate, bitsPerSample, channels): datasize = 2000 * 10 ** 6 o = bytes("RIFF", 'ascii') # (4byte) Marks file as RIFF o += (datasize + 36).to_bytes(4, 'little') # (4byte) File size in bytes excluding this and RIFF marker o += bytes("WAVE", 'ascii') # (4byte) File type o += bytes("fmt ", 'ascii') # (4byte) Format Chunk Marker o += (16).to_bytes(4, 'little') # (4byte) Length of above format data o += (1).to_bytes(2, 'little') # (2byte) Format type (1 - PCM) o += (channels).to_bytes(2, 'little') # (2byte) o += (sampleRate).to_bytes(4, 'little') # (4byte) o += (sampleRate * channels * bitsPerSample // 8).to_bytes(4, 'little') # (4byte) o += (channels * bitsPerSample // 8).to_bytes(2, 'little') # (2byte) o += (bitsPerSample).to_bytes(2, 'little') # (2byte) o += bytes("data", 'ascii') # (4byte) Data Chunk Marker o += (datasize).to_bytes(4, 'little') # (4byte) Data size in bytes return o @app.route('/audio') def audio(): # start Recording def sound(): CHUNK = 1024 sampleRate = 44100 bitsPerSample = 16 channels = 2 wav_header = genHeader(sampleRate, bitsPerSample, channels) stream = audio1.open(format=FORMAT, channels=CHANNELS, rate=RATE, input=True, input_device_index=1, frames_per_buffer=CHUNK) print("recording...") # frames = [] while True: data = wav_header + stream.read(CHUNK) yield (data) return Response(sound()) if __name__ == "__main__": app.run(debug=True, host='0.0.0.0', port=9527)