Ciallo～(∠・ω< )⌒★

import os import ssl import time import logging import socket import queue import base64 from threading import Thread from config import Global from utils.other import terminate_thread from src.asr import speech_recognition from flask import Flask, send_from_directory from flask_socketio import SocketIO, emit port = Global.weblive2d["port"] cert_file = 'assets\cert.pem' key_file = 'assets\key.pem' app = Flask(__name__, static_folder='.') app.config['SECRET_KEY'] = 'live2dchat' socketio = SocketIO(app, cors_allowed_origins="*") logging.getLogger('werkzeug').setLevel(logging.ERROR) class AudioStream: def __init__(self): self.buffer = queue.Queue(10) def write(self, data): self.buffer.put(data) def read(self, chunk_size=None): chunk_size *= 2 data = b'' while len(data) != chunk_size: data += self.buffer.get() if len(data) > chunk_size: remaining = data[chunk_size:] self.buffer.put(remaining) data = data[:chunk_size] return data def generate_html(): """动态生成HTML内容""" # 嘴型同步参数配置 mouth_sync_config = { 'smoothing_factor': 0.3, # 平滑系数 (0-1) 'silence_threshold': 0, # 静音阈值 'max_rms_scale': Global.character["max_rms_scale"]*Global.weblive2d["max_rms_scale"], # 最大RMS缩放 'mouth_scale': 1.0, # 整体嘴巴开合缩放 'power_exponent': 2.0, # 非线性变换指数 'update_interval': 16, # 更新间隔(ms)，16约等于60fps 'close_speed': 0.85 # 嘴巴关闭速度 } return f""" Ciallo～(∠・ω< )⌒★

""" @app.route('/') def index(): return generate_html() @app.route('/Character/') def serve_models(path): return send_from_directory('Character', path) @app.route('/assets/') def serve_static(path): return send_from_directory('assets', path) @socketio.on('connect') def handle_connect(): print('web客户端连接成功') emit('status', {'message': '服务器连接成功'}) if Global.web_asr_thread: terminate_thread(Global.web_asr_thread) if Global.asr_thread: terminate_thread(Global.asr_thread) Global.web_asr_thread = Thread(target=speech_recognition, args=(True,), daemon=True) Global.web_asr_thread.start() @socketio.on('disconnect') def handle_disconnect(): print('web客户端断开连接') if Global.web_asr_thread: terminate_thread(Global.web_asr_thread) if Global.asr_thread: terminate_thread(Global.asr_thread) Global.asr_thread = Thread(target=speech_recognition, daemon=True) Global.asr_thread.start() @socketio.on('audio_chunk') def handle_audio_chunk(data): try: Global.web_audio_stream.write(data) except Exception as e: print(f"处理音频块失败: {e}") emit('error', {'message': f'音频处理错误: {str(e)}'}) @socketio.on('photo_captured') def handle_photo_captured(data): """处理照片数据""" try: image_data = data.get('image_data') format_type = data.get('format', 'jpeg') width = data.get('width', 0) height = data.get('height', 0) facing_mode = data.get('facing_mode', 'environment') print(f"收到照片: {width}x{height}, 格式: {format_type}, 摄像头: {facing_mode}") image_bytes = base64.b64decode(image_data) filepath = f'temp/img.{format_type}' with open(filepath, 'wb') as f: f.write(image_bytes) Global.received_photo.put(filepath) emit('status', {'message': '照片接收成功'}) except Exception as e: print(f"处理照片失败: {e}") emit('error', {'message': f'照片处理错误: {str(e)}'}) def play_audio(audio_data, format='wav', subtitle=None, subtitle_duration=3000): """播放音频""" try: if isinstance(audio_data, (bytes, bytearray)): audio_base64 = base64.b64encode(audio_data).decode('utf-8') else: audio_base64 = base64.b64encode(bytes(audio_data)).decode('utf-8') emit_data = { 'audio_data': audio_base64, 'format': format, 'enable_mouth_sync': True } if subtitle: emit_data['subtitle'] = subtitle emit_data['subtitle_duration'] = subtitle_duration socketio.emit('play_audio', emit_data) return True except Exception as e: print(f"发送音频数据失败: {e}") return False def play_audio_file(file_path, format=None, subtitle=None, subtitle_duration=3000): """播放音频文件""" try: if not os.path.exists(file_path): print(f"音频文件不存在: {file_path}") return False if format is None: ext = os.path.splitext(file_path)[1].lower().lstrip('.') format = ext if ext in ['wav', 'mp3', 'ogg', 'm4a'] else 'wav' with open(file_path, 'rb') as f: audio_data = f.read() return play_audio(audio_data, format, subtitle, subtitle_duration) except Exception as e: print(f"播放音频文件失败: {e}") return False def clear_audio_queue(): """清空音频播放队列""" try: socketio.emit('clear_audio') print("已发送清空音频队列指令") return True except Exception as e: print(f"发送清空音频队列指令失败: {e}") return False def request_photo(): """请求客户端拍照""" try: socketio.emit('request_photo', { 'timestamp': int(time.time() * 1000), 'message': '服务器请求拍照' }) print("已发送拍照请求到客户端") return True except Exception as e: print(f"发送拍照请求失败: {e}") return False def create_signed_cert(): if not os.path.exists(cert_file) or not os.path.exists(key_file): print("正在生成自签名证书...") try: import subprocess subprocess.run([ 'openssl', 'req', '-x509', '-newkey', 'rsa:4096', '-nodes', '-out', cert_file, '-keyout', key_file, '-days', '365', '-subj', '/C=CN/ST=State/L=City/O=Organization/CN=localhost' ], check=True) print("证书生成成功！") except (subprocess.CalledProcessError, FileNotFoundError): print("无法自动生成证书，请手动运行:") print("openssl req -x509 -newkey rsa:4096 -nodes -out cert.pem -keyout key.pem -days 365") return False return True def get_local_ip(): try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] s.close() return ip except: return "127.0.0.1" def run_web(): with open('assets\live2d.js', 'r', encoding='utf-8') as f: js = f.readlines() path = Global.character["live2d_model"] path = path.replace('\\', '/') js[1581] = f'xb = "{path}",\n' js[1588] = f'_i.x = {Global.character["web_x"]};\n' js[1589] = f'_i.y = {Global.character["web_y"]};\n' js[1592] = f'_i.scale.set({Global.character["web_scale"]});\n' if 'watermark' in Global.character: js[1595] = f'_i.internalModel.coreModel.setParameterValueById("{Global.character["watermark"]}", 1);\n' else: js[1595] = '\n' with open('assets\live2d.js', 'w', encoding='utf-8') as f: f.writelines(js) local_ip = get_local_ip() if os.path.exists(cert_file) and os.path.exists(key_file): try: context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) context.load_cert_chain(cert_file, key_file) print(f"Web对话接口: https://{local_ip}:{port}") socketio.run(app, port=port, host="0.0.0.0", ssl_context=context, debug=False, use_reloader=False) except Exception as e: print(f"HTTPS启动失败: {e}") print("回退到HTTP模式...") socketio.run(app, port=port, host="0.0.0.0", debug=False, use_reloader=False) else: print("未找到SSL证书文件") if create_signed_cert(): run_web() else: print(f"Web对话接口: http://{local_ip}:{port}") print("注意: 手机浏览器可能需要HTTPS才能使用麦克风功能") socketio.run(app, port=port, host="0.0.0.0", debug=False) Global.web_clear_audio = clear_audio_queue Global.web_audio_stream = AudioStream() Global.web_request_photo = request_photo