diff --git a/README.md b/README.md
index f52b0f0..ff18ece 100644
--- a/README.md
+++ b/README.md
@@ -3,7 +3,7 @@
## 核心功能✨
- 💬 使用微信聊天记录微调LLM
- 🎙️ 使用微信语音消息➕0.5B大模型实现高质量声音克隆 👉[WeClone-audio](https://github.com/xming521/WeClone/tree/master/WeClone-audio)
-- 🔗 绑定到微信机器人，实现自己的数字分身
+- 🔗 绑定到微信、QQ、Telegram、企微、飞书机器人，实现自己的数字分身
## 特性与说明📋
@@ -197,4 +197,15 @@ python ./src/wechat_bot/main.py
+## ⭐ Star History
+> [!TIP]
+> 如果本项目对您有帮助，或者您关注本项目的未来发展，请给项目 Star，谢谢
+
+
+
+[](https://www.star-history.com/#xming521/WeClone&Date)
+
+
+
+
克隆我们，保留那灵魂的芬芳
diff --git a/WeClone-audio/src/server未完工/.env.example b/WeClone-audio/src/server未完工/.env.example
new file mode 100644
index 0000000..627d413
--- /dev/null
+++ b/WeClone-audio/src/server未完工/.env.example
@@ -0,0 +1,14 @@
+API_KEY=your_api_key_here
+PORT=5050
+
+DEFAULT_VOICE=en-US-AvaNeural
+DEFAULT_RESPONSE_FORMAT=mp3
+DEFAULT_SPEED=1.0
+
+DEFAULT_LANGUAGE=en-US
+
+REQUIRE_API_KEY=True
+
+REMOVE_FILTER=False
+
+EXPAND_API=True
\ No newline at end of file
diff --git a/WeClone-audio/src/server未完工/handle_text.py b/WeClone-audio/src/server未完工/handle_text.py
new file mode 100644
index 0000000..c4fba21
--- /dev/null
+++ b/WeClone-audio/src/server未完工/handle_text.py
@@ -0,0 +1,62 @@
+import re
+import emoji
+
+def prepare_tts_input_with_context(text: str) -> str:
+ """
+ Prepares text for a TTS API by cleaning Markdown and adding minimal contextual hints
+ for certain Markdown elements like headers. Preserves paragraph separation.
+
+ Args:
+ text (str): The raw text containing Markdown or other formatting.
+
+ Returns:
+ str: Cleaned text with contextual hints suitable for TTS input.
+ """
+
+ # Remove emojis
+ text = emoji.replace_emoji(text, replace='')
+
+ # Add context for headers
+ def header_replacer(match):
+ level = len(match.group(1)) # Number of '#' symbols
+ header_text = match.group(2).strip()
+ if level == 1:
+ return f"Title — {header_text}\n"
+ elif level == 2:
+ return f"Section — {header_text}\n"
+ else:
+ return f"Subsection — {header_text}\n"
+
+ text = re.sub(r"^(#{1,6})\s+(.*)", header_replacer, text, flags=re.MULTILINE)
+
+ # Announce links (currently commented out for potential future use)
+ # text = re.sub(r"\[([^\]]+)\]\((https?:\/\/[^\)]+)\)", r"\1 (link: \2)", text)
+
+ # Remove links while keeping the link text
+ text = re.sub(r"\[([^\]]+)\]\([^\)]+\)", r"\1", text)
+
+ # Describe inline code
+ text = re.sub(r"`([^`]+)`", r"code snippet: \1", text)
+
+ # Remove bold/italic symbols but keep the content
+ text = re.sub(r"(\*\*|__|\*|_)", '', text)
+
+ # Remove code blocks (multi-line) with a description
+ text = re.sub(r"```([\s\S]+?)```", r"(code block omitted)", text)
+
+ # Remove image syntax but add alt text if available
+ text = re.sub(r"!\[([^\]]*)\]\([^\)]+\)", r"Image: \1", text)
+
+ # Remove HTML tags
+ text = re.sub(r"]+(>|$)", '', text)
+
+ # Normalize line breaks
+ text = re.sub(r"\n{2,}", '\n\n', text) # Ensure consistent paragraph separation
+
+ # Replace multiple spaces within lines
+ text = re.sub(r" {2,}", ' ', text)
+
+ # Trim leading and trailing whitespace from the whole text
+ text = text.strip()
+
+ return text
diff --git a/WeClone-audio/src/server未完工/requirements.txt b/WeClone-audio/src/server未完工/requirements.txt
new file mode 100644
index 0000000..4019693
--- /dev/null
+++ b/WeClone-audio/src/server未完工/requirements.txt
@@ -0,0 +1,5 @@
+flask
+gevent
+python-dotenv
+edge-tts
+emoji
\ No newline at end of file
diff --git a/WeClone-audio/src/server未完工/server.py b/WeClone-audio/src/server未完工/server.py
new file mode 100644
index 0000000..486ffd3
--- /dev/null
+++ b/WeClone-audio/src/server未完工/server.py
@@ -0,0 +1,167 @@
+# server.py
+
+from flask import Flask, request, send_file, jsonify
+from gevent.pywsgi import WSGIServer
+from dotenv import load_dotenv
+import os
+
+from handle_text import prepare_tts_input_with_context
+from tts_handler import generate_speech, get_models, get_voices
+from utils import getenv_bool, require_api_key, AUDIO_FORMAT_MIME_TYPES
+
+app = Flask(__name__)
+load_dotenv()
+
+API_KEY = os.getenv('API_KEY', 'your_api_key_here')
+PORT = int(os.getenv('PORT', 5050))
+
+DEFAULT_VOICE = os.getenv('DEFAULT_VOICE', 'en-US-AvaNeural')
+DEFAULT_RESPONSE_FORMAT = os.getenv('DEFAULT_RESPONSE_FORMAT', 'mp3')
+DEFAULT_SPEED = float(os.getenv('DEFAULT_SPEED', 1.0))
+
+REMOVE_FILTER = getenv_bool('REMOVE_FILTER', False)
+EXPAND_API = getenv_bool('EXPAND_API', True)
+
+# DEFAULT_MODEL = os.getenv('DEFAULT_MODEL', 'tts-1')
+
+@app.route('/v1/audio/speech', methods=['POST'])
+@app.route('/audio/speech', methods=['POST']) # Add this line for the alias
+@require_api_key
+def text_to_speech():
+ data = request.json
+ if not data or 'input' not in data:
+ return jsonify({"error": "Missing 'input' in request body"}), 400
+
+ text = data.get('input')
+
+ if not REMOVE_FILTER:
+ text = prepare_tts_input_with_context(text)
+
+ # model = data.get('model', DEFAULT_MODEL)
+ voice = data.get('voice', DEFAULT_VOICE)
+
+ response_format = data.get('response_format', DEFAULT_RESPONSE_FORMAT)
+ speed = float(data.get('speed', DEFAULT_SPEED))
+
+ mime_type = AUDIO_FORMAT_MIME_TYPES.get(response_format, "audio/mpeg")
+
+ # Generate the audio file in the specified format with speed adjustment
+ output_file_path = generate_speech(text, voice, response_format, speed)
+
+ # Return the file with the correct MIME type
+ return send_file(output_file_path, mimetype=mime_type, as_attachment=True, download_name=f"speech.{response_format}")
+
+@app.route('/v1/models', methods=['GET', 'POST'])
+@app.route('/models', methods=['GET', 'POST'])
+@require_api_key
+def list_models():
+ return jsonify({"data": get_models()})
+
+@app.route('/v1/voices', methods=['GET', 'POST'])
+@app.route('/voices', methods=['GET', 'POST'])
+@require_api_key
+def list_voices():
+ specific_language = None
+
+ data = request.args if request.method == 'GET' else request.json
+ if data and ('language' in data or 'locale' in data):
+ specific_language = data.get('language') if 'language' in data else data.get('locale')
+
+ return jsonify({"voices": get_voices(specific_language)})
+
+@app.route('/v1/voices/all', methods=['GET', 'POST'])
+@app.route('/voices/all', methods=['GET', 'POST'])
+@require_api_key
+def list_all_voices():
+ return jsonify({"voices": get_voices('all')})
+
+"""
+Support for ElevenLabs and Azure AI Speech
+ (currently in beta)
+"""
+
+# http://localhost:5050/elevenlabs/v1/text-to-speech
+# http://localhost:5050/elevenlabs/v1/text-to-speech/en-US-AndrewNeural
+@app.route('/elevenlabs/v1/text-to-speech/', methods=['POST'])
+@require_api_key
+def elevenlabs_tts(voice_id):
+ if not EXPAND_API:
+ return jsonify({"error": f"Endpoint not allowed"}), 500
+
+ # Parse the incoming JSON payload
+ try:
+ payload = request.json
+ if not payload or 'text' not in payload:
+ return jsonify({"error": "Missing 'text' in request body"}), 400
+ except Exception as e:
+ return jsonify({"error": f"Invalid JSON payload: {str(e)}"}), 400
+
+ text = payload['text']
+
+ if not REMOVE_FILTER:
+ text = prepare_tts_input_with_context(text)
+
+ voice = voice_id # ElevenLabs uses the voice_id in the URL
+
+ # Use default settings for edge-tts
+ response_format = 'mp3'
+ speed = DEFAULT_SPEED # Optional customization via payload.get('speed', DEFAULT_SPEED)
+
+ # Generate speech using edge-tts
+ try:
+ output_file_path = generate_speech(text, voice, response_format, speed)
+ except Exception as e:
+ return jsonify({"error": f"TTS generation failed: {str(e)}"}), 500
+
+ # Return the generated audio file
+ return send_file(output_file_path, mimetype="audio/mpeg", as_attachment=True, download_name="speech.mp3")
+
+# tts.speech.microsoft.com/cognitiveservices/v1
+# https://{region}.tts.speech.microsoft.com/cognitiveservices/v1
+# http://localhost:5050/azure/cognitiveservices/v1
+@app.route('/azure/cognitiveservices/v1', methods=['POST'])
+@require_api_key
+def azure_tts():
+ if not EXPAND_API:
+ return jsonify({"error": f"Endpoint not allowed"}), 500
+
+ # Parse the SSML payload
+ try:
+ ssml_data = request.data.decode('utf-8')
+ if not ssml_data:
+ return jsonify({"error": "Missing SSML payload"}), 400
+
+ # Extract the text and voice from SSML
+ from xml.etree import ElementTree as ET
+ root = ET.fromstring(ssml_data)
+ text = root.find('.//{http://www.w3.org/2001/10/synthesis}voice').text
+ voice = root.find('.//{http://www.w3.org/2001/10/synthesis}voice').get('name')
+ except Exception as e:
+ return jsonify({"error": f"Invalid SSML payload: {str(e)}"}), 400
+
+ # Use default settings for edge-tts
+ response_format = 'mp3'
+ speed = DEFAULT_SPEED
+
+ if not REMOVE_FILTER:
+ text = prepare_tts_input_with_context(text)
+
+ # Generate speech using edge-tts
+ try:
+ output_file_path = generate_speech(text, voice, response_format, speed)
+ except Exception as e:
+ return jsonify({"error": f"TTS generation failed: {str(e)}"}), 500
+
+ # Return the generated audio file
+ return send_file(output_file_path, mimetype="audio/mpeg", as_attachment=True, download_name="speech.mp3")
+
+print(f" Edge TTS (Free Azure TTS) Replacement for OpenAI's TTS API")
+print(f" ")
+print(f" * Serving OpenAI Edge TTS")
+print(f" * Server running on http://localhost:{PORT}")
+print(f" * TTS Endpoint: http://localhost:{PORT}/v1/audio/speech")
+print(f" ")
+
+if __name__ == '__main__':
+ http_server = WSGIServer(('0.0.0.0', PORT), app)
+ http_server.serve_forever()
diff --git a/WeClone-audio/src/server未完工/tts_handler.py b/WeClone-audio/src/server未完工/tts_handler.py
new file mode 100644
index 0000000..f243acc
--- /dev/null
+++ b/WeClone-audio/src/server未完工/tts_handler.py
@@ -0,0 +1,133 @@
+import edge_tts
+import asyncio
+import tempfile
+import subprocess
+import os
+from pathlib import Path
+
+# Language default (environment variable)
+DEFAULT_LANGUAGE = os.getenv('DEFAULT_LANGUAGE', 'en-US')
+
+# OpenAI voice names mapped to edge-tts equivalents
+voice_mapping = {
+ 'alloy': 'en-US-AvaNeural',
+ 'echo': 'en-US-AndrewNeural',
+ 'fable': 'en-GB-SoniaNeural',
+ 'onyx': 'en-US-EricNeural',
+ 'nova': 'en-US-SteffanNeural',
+ 'shimmer': 'en-US-EmmaNeural'
+}
+
+def is_ffmpeg_installed():
+ """Check if FFmpeg is installed and accessible."""
+ try:
+ subprocess.run(['ffmpeg', '-version'], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ return True
+ except (subprocess.CalledProcessError, FileNotFoundError):
+ return False
+
+async def _generate_audio(text, voice, response_format, speed):
+ """Generate TTS audio and optionally convert to a different format."""
+ # Determine if the voice is an OpenAI-compatible voice or a direct edge-tts voice
+ edge_tts_voice = voice_mapping.get(voice, voice) # Use mapping if in OpenAI names, otherwise use as-is
+
+ # Generate the TTS output in mp3 format first
+ temp_output_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp3")
+
+ # Convert speed to SSML rate format
+ try:
+ speed_rate = speed_to_rate(speed) # Convert speed value to "+X%" or "-X%"
+ except Exception as e:
+ print(f"Error converting speed: {e}. Defaulting to +0%.")
+ speed_rate = "+0%"
+
+ # Generate the MP3 file
+ communicator = edge_tts.Communicate(text=text, voice=edge_tts_voice, rate=speed_rate)
+ await communicator.save(temp_output_file.name)
+
+ # If the requested format is mp3, return the generated file directly
+ if response_format == "mp3":
+ return temp_output_file.name
+
+ # Check if FFmpeg is installed
+ if not is_ffmpeg_installed():
+ print("FFmpeg is not available. Returning unmodified mp3 file.")
+ return temp_output_file.name
+
+ # Create a new temporary file for the converted output
+ converted_output_file = tempfile.NamedTemporaryFile(delete=False, suffix=f".{response_format}")
+
+ # Build the FFmpeg command
+ ffmpeg_command = [
+ "ffmpeg",
+ "-i", temp_output_file.name, # Input file
+ "-c:a", {
+ "aac": "aac",
+ "mp3": "libmp3lame",
+ "wav": "pcm_s16le",
+ "opus": "libopus",
+ "flac": "flac"
+ }.get(response_format, "aac"), # Default to AAC if unknown
+ "-b:a", "192k" if response_format != "wav" else None, # Bitrate not needed for WAV
+ "-f", {
+ "aac": "mp4", # AAC in MP4 container
+ "mp3": "mp3",
+ "wav": "wav",
+ "opus": "ogg",
+ "flac": "flac"
+ }.get(response_format, response_format), # Default to matching format
+ "-y", # Overwrite without prompt
+ converted_output_file.name # Output file
+ ]
+
+ try:
+ # Run FFmpeg command and ensure no errors occur
+ subprocess.run(ffmpeg_command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ except subprocess.CalledProcessError as e:
+ raise RuntimeError(f"FFmpeg error during audio conversion: {e}")
+
+ # Clean up the original temporary file
+ Path(temp_output_file.name).unlink(missing_ok=True)
+
+ return converted_output_file.name
+
+def generate_speech(text, voice, response_format, speed=1.0):
+ return asyncio.run(_generate_audio(text, voice, response_format, speed))
+
+def get_models():
+ return [
+ {"id": "tts-1", "name": "Text-to-speech v1"},
+ {"id": "tts-1-hd", "name": "Text-to-speech v1 HD"}
+ ]
+
+async def _get_voices(language=None):
+ # List all voices, filter by language if specified
+ all_voices = await edge_tts.list_voices()
+ language = language or DEFAULT_LANGUAGE # Use default if no language specified
+ filtered_voices = [
+ {"name": v['ShortName'], "gender": v['Gender'], "language": v['Locale']}
+ for v in all_voices if language == 'all' or language is None or v['Locale'] == language
+ ]
+ return filtered_voices
+
+def get_voices(language=None):
+ return asyncio.run(_get_voices(language))
+
+def speed_to_rate(speed: float) -> str:
+ """
+ Converts a multiplicative speed value to the edge-tts "rate" format.
+
+ Args:
+ speed (float): The multiplicative speed value (e.g., 1.5 for +50%, 0.5 for -50%).
+
+ Returns:
+ str: The formatted "rate" string (e.g., "+50%" or "-50%").
+ """
+ if speed < 0 or speed > 2:
+ raise ValueError("Speed must be between 0 and 2 (inclusive).")
+
+ # Convert speed to percentage change
+ percentage_change = (speed - 1) * 100
+
+ # Format with a leading "+" or "-" as required
+ return f"{percentage_change:+.0f}%"
diff --git a/WeClone-audio/src/server未完工/utils.py b/WeClone-audio/src/server未完工/utils.py
new file mode 100644
index 0000000..7fe9845
--- /dev/null
+++ b/WeClone-audio/src/server未完工/utils.py
@@ -0,0 +1,38 @@
+# utils.py
+
+from flask import request, jsonify
+from functools import wraps
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+def getenv_bool(name: str, default: bool = False) -> bool:
+ return os.getenv(name, str(default)).lower() in ("yes", "y", "true", "1", "t")
+
+API_KEY = os.getenv('API_KEY', 'your_api_key_here')
+REQUIRE_API_KEY = getenv_bool('REQUIRE_API_KEY', True)
+
+def require_api_key(f):
+ @wraps(f)
+ def decorated_function(*args, **kwargs):
+ if not REQUIRE_API_KEY:
+ return f(*args, **kwargs)
+ auth_header = request.headers.get('Authorization')
+ if not auth_header or not auth_header.startswith('Bearer '):
+ return jsonify({"error": "Missing or invalid API key"}), 401
+ token = auth_header.split('Bearer ')[1]
+ if token != API_KEY:
+ return jsonify({"error": "Invalid API key"}), 401
+ return f(*args, **kwargs)
+ return decorated_function
+
+# Mapping of audio format to MIME type
+AUDIO_FORMAT_MIME_TYPES = {
+ "mp3": "audio/mpeg",
+ "opus": "audio/ogg",
+ "aac": "audio/aac",
+ "flac": "audio/flac",
+ "wav": "audio/wav",
+ "pcm": "audio/L16"
+}
diff --git a/WeClone-audio/src/text_to_speech.py b/WeClone-audio/src/text_to_speech.py
deleted file mode 100644
index 2bb468b..0000000
--- a/WeClone-audio/src/text_to_speech.py
+++ /dev/null
@@ -1,131 +0,0 @@
-import os
-from transformers import AutoTokenizer, AutoModelForCausalLM
-import torch
-import soundfile as sf
-from xcodec2.modeling_xcodec2 import XCodec2Model
-import torchaudio
-
-
-class TextToSpeech:
- def __init__(self, sample_audio_path, sample_audio_text):
- self.sample_audio_text = sample_audio_text
- # 初始化模型
- llasa_3b = "HKUSTAudio/Llasa-3B"
- xcodec2 = "HKUSTAudio/xcodec2"
-
- self.tokenizer = AutoTokenizer.from_pretrained(llasa_3b)
- self.llasa_3b_model = AutoModelForCausalLM.from_pretrained(
- llasa_3b,
- trust_remote_code=True,
- device_map="auto",
- )
- self.llasa_3b_model.eval()
-
- self.xcodec_model = XCodec2Model.from_pretrained(xcodec2)
- self.xcodec_model.eval().cuda()
-
- # 处理音频
- waveform, sample_rate = torchaudio.load(sample_audio_path)
- if len(waveform[0]) / sample_rate > 15:
- print("已将音频裁剪至前15秒。")
- waveform = waveform[:, : sample_rate * 15]
-
- # 检查音频是否为立体声
- if waveform.size(0) > 1:
- waveform_mono = torch.mean(waveform, dim=0, keepdim=True)
- else:
- waveform_mono = waveform
-
- self.prompt_wav = torchaudio.transforms.Resample(
- orig_freq=sample_rate, new_freq=16000
- )(waveform_mono)
-
- # Encode the prompt wav
- vq_code_prompt = self.xcodec_model.encode_code(input_waveform=self.prompt_wav)
- vq_code_prompt = vq_code_prompt[0, 0, :]
- self.speech_ids_prefix = self.ids_to_speech_tokens(vq_code_prompt)
- self.speech_end_id = self.tokenizer.convert_tokens_to_ids("<|SPEECH_GENERATION_END|>")
-
- def ids_to_speech_tokens(self, speech_ids):
- speech_tokens_str = []
- for speech_id in speech_ids:
- speech_tokens_str.append(f"<|s_{speech_id}|>")
- return speech_tokens_str
-
- def extract_speech_ids(self, speech_tokens_str):
- speech_ids = []
- for token_str in speech_tokens_str:
- if token_str.startswith("<|s_") and token_str.endswith("|>"):
- num_str = token_str[4:-2]
- num = int(num_str)
- speech_ids.append(num)
- else:
- print(f"Unexpected token: {token_str}")
- return speech_ids
-
- @torch.inference_mode()
- def infer(self, target_text):
- if len(target_text) == 0:
- return None
- elif len(target_text) > 300:
- print("文本过长，请保持在300字符以内。")
- target_text = target_text[:300]
-
- input_text = self.sample_audio_text + " " + target_text
-
- formatted_text = (
- f"<|TEXT_UNDERSTANDING_START|>{input_text}<|TEXT_UNDERSTANDING_END|>"
- )
-
- chat = [
- {
- "role": "user",
- "content": "Convert the text to speech:" + formatted_text,
- },
- {
- "role": "assistant",
- "content": "<|SPEECH_GENERATION_START|>"
- + "".join(self.speech_ids_prefix),
- },
- ]
-
- input_ids = self.tokenizer.apply_chat_template(
- chat, tokenize=True, return_tensors="pt", continue_final_message=True
- )
- input_ids = input_ids.to("cuda")
-
- outputs = self.llasa_3b_model.generate(
- input_ids,
- max_length=2048,
- eos_token_id=self.speech_end_id,
- do_sample=True,
- top_p=1,
- temperature=0.8,
- )
- generated_ids = outputs[0][input_ids.shape[1] - len(self.speech_ids_prefix): -1]
-
- speech_tokens = self.tokenizer.batch_decode(
- generated_ids, skip_special_tokens=True
- )
-
- speech_tokens = self.extract_speech_ids(speech_tokens)
- speech_tokens = torch.tensor(speech_tokens).cuda().unsqueeze(0).unsqueeze(0)
-
- gen_wav = self.xcodec_model.decode_code(speech_tokens)
- gen_wav = gen_wav[:, :, self.prompt_wav.shape[1]:]
-
- return (16000, gen_wav[0, 0, :].cpu().numpy())
-
-
-if __name__ == "__main__":
- # 如果遇到问题，请尝试将参考音频转换为WAV或MP3格式，将其裁剪至15秒以内，并缩短提示文本。
- sample_audio_text = "对，这就是我万人敬仰的太乙真人，虽然有点婴儿肥，但也掩不住我逼人的帅气。"
- sample_audio_path = os.path.join(os.path.dirname(__file__), "sample.wav")
-
- tts = TextToSpeech(sample_audio_path, sample_audio_text)
- target_text = "晚上好啊，吃了吗您"
- result = tts.infer(target_text)
- sf.write(os.path.join(os.path.dirname(__file__), "output.wav"), result[1], result[0])
- target_text = "我是老北京正黄旗！"
- result = tts.infer(target_text)
- sf.write(os.path.join(os.path.dirname(__file__), "output1.wav"), result[1], result[0])