import json import os import subprocess import pydantic import requests from pydantic import BaseModel from config import mp4_path, post_record_list_url, post_record_url, info_topic class BaseResponse: code: int msg: str # 统一返回 def publish_payload(code, msg): return json.dumps(BaseResponse(code=code, msg=msg), ensure_ascii=False) def exception_handler(func): def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: print(f"函数{func.__name__}中发生了异常:{e}") return wrapper def push_stream(client): p = subprocess.Popen(['/bin/bash /home/pi/lot_manager/bash/start_push_stream.sh'], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) out, err = p.communicate() output = out.decode('utf-8').strip() client.publish(info_topic, payload=publish_payload(200, json.dumps(output, ensure_ascii=False)), qos=0) def close_stream(client): p = subprocess.Popen(['/bin/bash /home/pi/lot_manager/bash/stop_push_stream.sh'], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) out, err = p.communicate() output = out.decode('utf-8').strip() client.publish(info_topic, payload=publish_payload(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0) def exec_sh(msg, client): origin_data = json.loads(msg.payload.decode('utf-8')) if 'data' not in origin_data: client.publish(info_topic, payload=publish_payload(code=404, msg='data must be supplied'), qos=0) return cmd = origin_data["data"] if cmd in ["supervisorctl stop __mqtt__", "supervisorctl restart __mqtt__", "supervisorctl stop all"]: return if cmd == "supervisorctl reload": client.publish(info_topic, payload=publish_payload(code=200, msg='reloading'), qos=0) subprocess.Popen([cmd], shell=True) return p = subprocess.Popen([cmd], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) out, err = p.communicate() output = out.decode('utf-8').strip() client.publish(info_topic, payload=publish_payload(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0) def get_status(client): p = subprocess.Popen(['supervisorctl status'], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) out, err = p.communicate() output = out.decode('utf-8').strip() client.publish(info_topic, payload=publish_payload(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0) def update(client): p = subprocess.Popen(['/bin/bash /home/pi/lot_manager/git_update.sh'], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) out, err = p.communicate() output = out.decode('utf-8').strip() client.publish(info_topic, payload=publish_payload(code=200, msg=json.dumps(output, ensure_ascii=False)), qos=0) def reload(client): client.publish(info_topic, payload=publish_payload(200, "reloading"), qos=0) subprocess.Popen(['supervisorctl reload'], shell=True) def get_list_record(client): data = { "data": os.listdir(mp4_path) } r = requests.post(post_record_list_url, json=data) client.publish(info_topic, payload=publish_payload(code=200, msg=str(r)), qos=0) def get_record(msg, client): filename = json.loads(msg.payload.decode('utf-8'))["data"] if filename is None or filename == '': client.publish(info_topic, payload=publish_payload(code=404, msg='没有该文件'), qos=0) return files = {filename: open(os.path.join(mp4_path, filename), 'rb'), "Content-Type": "application/octet-stream"} r = requests.post(post_record_url, files=files) client.publish(info_topic, payload=publish_payload(code=200, msg=str(r)), qos=0)