如何在Laravel中通过Symfony Process调用Python函数（含视频流场景）

我有一个返回字符串类型数据的Python函数，运行正常，代码如下：

import mysql.connector

mydb = mysql.connector.connect(
    host="localhost",
    user="root",
    passwd="",
    database="db_absensi"
)

mycursor = mydb.cursor() 

def example():
    mycursor.execute("SELECT * FROM examples")
    data = mycursor.fetchall()
 
    return data

我在Laravel中使用Symfony Process调用的代码如下：

public function test()
{
    $process = new Process(['python ../../../app/data.py']);
    $process->setTimeout(3600);
    $process->run();

    if(!$process->isSuccessful())
    {
        throw new ProcessFailedException($process);
    }

    dd ($process->getOutput());

    return view("testView");
}

此外，我还有一个不返回数据而是生成视频流的Python函数（基于OpenCV的人脸识别流函数），代码如下：

def face_recognition():  # generate frame by frame from camera
    def draw_boundary(img, classifier, scaleFactor, minNeighbors, color, text, clf):
        gray_image = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
        features = classifier.detectMultiScale(gray_image, scaleFactor, minNeighbors)
 
        global justscanned
        global pause_cnt
 
        pause_cnt += 1
 
        coords = []
 
        for (x, y, w, h) in features:
            cv2.rectangle(img, (x, y), (x + w, y + h), color, 2)
            id, pred = clf.predict(gray_image[y:y + h, x:x + w])
            confidence = int(100 * (1 - pred / 300))
 
            if confidence > 70 and not justscanned:
                global cnt
                cnt += 1
 
                n = (100 / 30) * cnt
                w_filled = (cnt / 30) * w
 
                cv2.putText(img, str(int(n))+' %', (x + 20, y + h + 28), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (255,255,255), 2, cv2.LINE_AA)
 
                cv2.rectangle(img, (x, y + h + 40), (x + w, y + h + 50), color, 2)
                cv2.rectangle(img, (x, y + h + 40), (x + int(w_filled), y + h + 50), (255,255,255), cv2.FILLED)
 
                mycursor.execute("SELECT a.img_person, b.nama, b.kelas, b.tanggal_lahir "
                                 " FROM images a "
                                 " LEFT JOIN data_person b ON a.img_person = b.id_master "
                                 " WHERE img_id = " + str(id))
                row = mycursor.fetchone()
                pnbr = row[0]
                pname = row[1]
                pkelas = row[2]
 
                if int(cnt) == 30:
                    cnt = 0
 
                    mycursor.execute("INSERT INTO attendance_datamaster (attendance_date, attendance_person) VALUES('"+str(date.today())+"', '" + pnbr + "')")
                    mydb.commit()
 
                    cv2.putText(img, pname + ' | ' + pkelas, (x - 10, y - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (255,255,255), 2, cv2.LINE_AA)
                    time.sleep(4)
                    # speech.say(pname + "successfully processed")
                    # speech.runAndWait()

                    justscanned = True
                    pause_cnt = 0
 
            else:
                if not justscanned:
                    cv2.putText(img, 'UNKNOWN', (x, y - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 2, cv2.LINE_AA)
                else:
                    cv2.putText(img, ' ', (x, y - 5), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 0, 255), 2,cv2.LINE_AA)
 
                if pause_cnt > 80:
                    justscanned = False
 
            coords = [x, y, w, h]
        return coords
 
    def recognize(img, clf, faceCascade):
        coords = draw_boundary(img, faceCascade, 1.1, 10, (255, 255, 255), "Face", clf)
        return img
 
    faceCascade = cv2.CascadeClassifier("resources/haarcascade_frontalface_default.xml")
    clf = cv2.face.LBPHFaceRecognizer_create()
    clf.read("classifier.xml")
 
    wCam, hCam = 400, 400
 
    cap = cv2.VideoCapture(0)
    cap.set(3, wCam)
    cap.set(4, hCam)
 
    while True:
        ret, img = cap.read()
        img = recognize(img, clf, faceCascade)
 
        frame = cv2.imencode('.jpg', img)[1].tobytes()
        yield (b'--frame\r\n'
               b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n\r\n')
 
        key = cv2.waitKey(1)
        if key == 27:
            break

def video_feed(): 
    return Response(face_recognition(), mimetype='multipart/x-mixed-replace; boundary=frame')

在Flask中，我会将该函数关联到img标签的src属性，代码如下：

<div class="col-md-8 " style="margin-top: 10%;">
    <img src="{{ url_for('video_feed') }}" width="100%" class="img-thumbnail">
</div>

请问在Laravel环境中，是否有类似或可行的方式运行这类不返回字符串数据的Python函数？

在Laravel中可以通过两种方式实现类似Flask的视频流展示效果，核心是处理Python生成的multipart/x-mixed-replace格式流：

方案一：将Python脚本作为独立HTTP服务运行（推荐） #

1. 补充Flask启动代码，将视频流服务作为独立进程运行，监听指定端口：

from flask import Flask, Response
app = Flask(__name__)

# 此处放入你的face_recognition和video_feed函数

@app.route('/video_feed')
def serve_video():
    return video_feed()

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=False)

2. 在Laravel视图中，直接将img标签的src指向Python服务的地址：

<div class="col-md-8 " style="margin-top: 10%;">
    <img src="http://localhost:5000/video_feed" width="100%" class="img-thumbnail">
</div>

3. 生产环境可通过Nginx反向代理隐藏端口，或为Python服务配置专属域名。

方案二：通过Laravel响应流转发Python进程输出 #

若不想单独启动Flask服务，可结合Symfony Process与Laravel响应流实现：

1. 修改Python脚本，直接将视频流输出到标准输出：

import sys

# 此处放入你的face_recognition函数

def video_feed():
    for frame in face_recognition():
        sys.stdout.buffer.write(frame)
        sys.stdout.buffer.flush()

if __name__ == '__main__':
    video_feed()

2. 在Laravel控制器中创建路由方法，启动Python进程并流式返回输出：

use Symfony\Component\Process\Process;
use Illuminate\Http\Response;

public function videoStream()
{
    // 替换为你的Python脚本实际路径
    $process = new Process(['python', base_path('app/data_video.py')]);
    $process->start();

    return response()->stream(function () use ($process) {
        while ($process->isRunning()) {
            echo $process->getIncrementalOutput();
            flush();
            ob_flush();
            usleep(10000); // 控制流的刷新间隔
        }
    }, 200, [
        'Content-Type' => 'multipart/x-mixed-replace; boundary=frame',
        'Cache-Control' => 'no-cache',
        'Connection' => 'keep-alive',
    ]);
}

3. 注册Laravel路由：

Route::get('/video-feed', [YourController::class, 'videoStream']);

4. 在视图中使用该路由：

<div class="col-md-8 " style="margin-top: 10%;">
    <img src="{{ url('video-feed') }}" width="100%" class="img-thumbnail">
</div>

注意事项 #

• 方案二中每个请求会启动新的Python进程，资源占用较高，高并发场景优先选方案一。
• 确保服务器有权限访问摄像头（本地开发），或Python脚本可正确调用摄像头设备。
• 生产环境下，方案一的Python服务可通过gunicorn或uwsgi托管，配合supervisor实现进程守护。

内容的提问来源于stack exchange，提问作者Maulana A