美工模板网站,余姚生活网,南昌网站建设700起,北京新机场建设指挥部网站Gewe 个微框架
GeWe#xff08;个微框架#xff09;是一个创新性的软件开发框架#xff0c;专注于IPAD协议#xff0c;为个人微信号以及企业信息安全提供了强大的功能和保障。GeWe的设计旨在简化开发过程#xff0c;使开发者能够高效、灵活地构建和定制通信协议#xff…Gewe 个微框架
GeWe个微框架是一个创新性的软件开发框架专注于IPAD协议为个人微信号以及企业信息安全提供了强大的功能和保障。GeWe的设计旨在简化开发过程使开发者能够高效、灵活地构建和定制通信协议以满足不同应用场景的需求。
灵活可扩展GeWe框架采用灵活可扩展的设计理念为开发者提供了丰富的选项和配置确保协议能够适应不同的应用场景和需求。
高效数据传输通过优化传输算法和数据压缩机制GeWe框架构建的IPAD协议能够更快地传输数据提高用户体验并节省带宽资源。
安全性与隐私保护GeWe框架重视安全性和隐私保护提供了一系列的加密和身份验证机制确保数据在传输过程中得到保护防止未经授权的访问。
丰富的功能模块GeWe框架提供了丰富的功能模块包括消息处理、好友管理、群组操作等使个人微信号具备更多功能和扩展性。
友好的开发接口GeWe框架提供简洁友好的开发接口使得开发者能够轻松地扩展个人微信号的功能如开发自动回复机器人、定时任务等。GeWe框架是一个功能强大、灵活可扩展的软件开发框架为个人和企业提供了丰富的功能和保障。通过利用GeWe框架开发者可以高效地构建和定制通信协议满足不同的应用需求。同时GeWe框架也重视安全性和隐私保护确保数据在传输过程中的安全性。
官方文档网站http://doc.geweapi.com/ 注意Gewe 详细接口调用请自行测试 登陆微信 Pad 协议
注册登陆 gewe 后台http://manager.geweapi.com/#/login?redirect%2Faccount%2Findex 登陆微信 Pad 协议http://manager.geweapi.com/#/account/wechat 项目结构预览
基于百度智能云、Gewe Pad 端、本地电脑端内网穿透实现。
详细流程图预览 源码结构 源码结构说明
.env 环境变量设置
audio_api 百度智能云合成语音 / 识别语音
bot.py 监听 gewe 消息回调主函数
download_msg.py 请求 gewe 下载语音消息
ffmpeg.exe 音频处理工具
mp3_to_silk mp3格式 - silk格式
send_msg 请求 gewe 发送语音消息
silk_to_pcm silk格式 - pcm格式
silk_v3_decoder.exe silk 解码操作
silk_v3_encoder.exe silk 编码操作
util.py 工具类配置环境变量
.env 环境变量配置
APP_ID百度云语音合成、识别 APP_ID
API_KEY百度云语音合成、识别 API_KEY
SECRET_KEY百度云语音合成、识别 SECRET_KEY
QIANFAN_API_KEY百度云千帆大模型 API_KEY
QIANFAN_SECRET_KEY百度云千帆大模型 SECRET_KEY
X_GEWE_TOKENGEWE Token
GEWE_APP_IDGEWE 设备号百度智能云千帆大模型
百度智能云使用参考博客https://wrist.blog.csdn.net/article/details/134628415
一站式企业级大模型平台提供先进的生成式AI生产及应用全流程开发工具链。直接调用ERNIE-Bot 4.0及其他主流大模型并提供可视化开发工具链支持数据闭环管理、专属大模型定制、大模型训练调优、插件编排等功能。 千帆大模型链接地址https://cloud.baidu.com/product/wenxinworkshop
chat_api.py
import json
import osimport requestsAPI_KEY os.getenv(QIANFAN_API_KEY)
SECRET_KEY os.getenv(QIANFAN_SECRET_KEY)payload {user_id: python,messages: [],system: 每次回答不允许换行并且不超过50个字。,disable_search: False,enable_citation: False
}def get_access_token(): # 这里是获取token一般不用管使用 AKSK 生成鉴权签名Access Token:return: access_token或是None(如果错误)url https://aip.baidubce.com/oauth/2.0/tokenparams {grant_type: client_credentials, client_id: API_KEY, client_secret: SECRET_KEY}return str(requests.post(url, paramsparams).json().get(access_token))def chat_answer(text):url https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/completions?access_token get_access_token()headers {Content-Type: application/json}payload[messages] [{role: user,content: text}]json_payload json.dumps(payload)# 发送请求response requests.request(POST, url, headersheaders, datajson_payload)result response.json().get(result) # 提取并打印AI的回复return resultffmpeg 下载安装
FFmpeg是一套开源的计算机程序用于处理数字音频、视频以及将其转化为流。它以其强大的音视频处理能力、广泛的格式支持和跨平台特性成为了多媒体处理领域的重要工具。
ffmpeg 基本功能
音视频编解码FFmpeg支持各种音频和视频编解码器包括常见的H.264、H.265、AAC、MP3等使其能够处理各种媒体格式。
格式转换可以将不同格式的音频和视频文件相互转换例如将MP4转换为AVI、将WAV转换为MP3等。
流媒体处理支持从网络摄像头、文件、或其他来源捕获音视频流也能将处理后的流推送到服务器或其他设备。
图像处理支持图像处理如提取视频帧合成图片和视频等。
剪辑和编辑可以对音视频进行剪辑和编辑包括裁剪、剪切、拼接等操作。
字幕处理支持字幕的添加、移除和编辑可以将字幕嵌入到视频中或者从视频中提取字幕。
音频处理提供了丰富的音频处理功能包括音频的剪辑、混音、音量调节等。
实时视频处理能够处理实时的音视频流适用于直播、视频会议等场景。下载安装 FFmpeg 下载地址https://github.com/BtbN/FFmpeg-Builds/releases
ffmpeg 工具 bin 目录 配置 ffmpeg/bin 系统环境变量 silk 编译器下载安装
Windows 用户可以直接使用提供的 silk_v3_decoder.exe 和 silk_v3_encoder.exe 工具进行转换。对于 macOS 和 Linux 用户需要下载项目代码并运行 converter.sh 脚本进行转换。转换过程需要依赖 gcc 和 ffmpeg 库。
下载地址https://github.com/kn007/silk-v3-decoder silk_v3_encoder.exe 和 silk_v3_decoder.exe
silk_v3_encoder.exe
功能用于将音频编码为Silk v3格式。可能用于需要将音频数据转换为Silk v3编码以进行存储、传输或与其他使用此编码格式的系统进行交互的场景。作为Silk-V3-Decoder项目的一部分silk_v3_encoder.exe如果存在并可用可能支持多种操作系统但具体兼容性信息在参考文章中未明确提及。
silk_v3_decoder.exe
功能解码 Silk v3 音频文件如微信amr、aud文件和QQ slk文件。将这些文件转换为其他音频格式如mp3。支持批量转换。
高性能针对Silk v3编码进行了优化提供快速的解码速度。
跨平台支持适用于多种操作系统包括Windows、macOS和Linux。但图形界面可能仅限于Windows版本。
简单易用提供了清晰简洁的接口和API降低了学习和使用的门槛。
开源与持续维护该项目遵循MIT许可证并在GitCode或其他代码托管平台上进行托管接受社区贡献和反馈。微信 silk 音频处理
mp3_to_silk.py 实现将 mp3 文件转换成 silk 文件
import math
import subprocess
import time
from pydub import AudioSegmentdef get_mp3_duration(mp3_file):# 使用 pydub 读取 mp3 并获取时长audio AudioSegment.from_mp3(mp3_file)duration_seconds len(audio) / 1000.0return duration_secondsdef convert_mp3_to_pcm(mp3_file, pcm_file):# 使用 ffmpeg 将 mp3 转换为 pcmcommand fffmpeg -y -i {mp3_file} -f s16le -ar 24000 -ac 1 -acodec pcm_s16le {pcm_file}try:subprocess.run(command, shellTrue, checkTrue)print(fConverted {mp3_file} to {pcm_file})except subprocess.CalledProcessError as e:print(fFailed to convert {mp3_file} to {pcm_file}: {e})def convert_pcm_to_silk(pcm_file, silk_file):# 使用 silk_v3_encoder.exe 将 PCM 转换为 SILKcommand fsilk_v3_encoder.exe {pcm_file} {silk_file} -tencentresult subprocess.run(command, shellTrue)if result.returncode 0:print(fConverted {pcm_file} to {silk_file})else:print(fFailed to convert {pcm_file} to {silk_file}: {result.stderr})def transfrom_silk(file_name):# 获取 mp3 文件时长duration get_mp3_duration(fmp3/{file_name}.mp3)print(fDuration of {file_name}.mp3: {duration} seconds)# 记录开始时间start_time time.time()# 转换流程convert_mp3_to_pcm(fmp3/{file_name}.mp3, f./pcm/{file_name}.pcm)convert_pcm_to_silk(f./pcm/{file_name}.pcm, f./silk/{file_name}.silk)# 记录结束时间end_time time.time()# 计算并打印转换所花的总时间total_time end_time - start_timeprint(fTotal time taken for conversion: {total_time} seconds)return {fileName: file_name, voiceDuration: math.ceil(duration) * 1000}silk_to_pcm.py 实现将 silk 文件转换成 mp3 文件
import subprocessdef convert_silk_to_pcm(silk_file, pcm_file):# 使用 silk_v3_decoder.exe 将 silk 转换为 pcmcommand fsilk_v3_decoder.exe {silk_file} {pcm_file}try:subprocess.run(command, shellTrue, checkTrue)print(fConverted {silk_file} to {pcm_file})except subprocess.CalledProcessError as e:print(fFailed to convert {silk_file} to {pcm_file}: {e})def transfrom_pcm(file_name):# 转换流程convert_silk_to_pcm(fsilk/{file_name}.silk, fpcm/{file_name}.pcm)保存语音消息
保存语音消息 download_msg.py
import json
import osimport dotenv
dotenv.load_dotenv(.env)
import requestsheaders {X-GEWE-TOKEN: os.getenv(X_GEWE_TOKEN),User-Agent: Apifox/1.0.0 (https://apifox.com),Content-Type: application/json
}def download_audio_msg(msgId: int, xml: str):data {appId: os.getenv(GEWE_APP_ID),msgId: msgId,xml: xml}print(json.dumps(data))response requests.post(http://api.geweapi.com/gewe/v2/api/message/downloadVoice, jsondata, headersheaders)if response.ok:data response.json()if data[ret] 200:print(Gewe download audio msg successfully.)print(data[data][fileUrl])return data[data][fileUrl]else:print(Gewe download audio msg in error.)return Falseelse:return Falsedef download_image_msg(xml: str):data {appId: wx_HtqjtTglwIKjQHSsSUYDi,type: 2,xml: xml}print(json.dumps(data))response requests.post(http://api.geweapi.com/gewe/v2/api/message/downloadImage, jsondata, headersheaders)if response.ok:data response.json()if data[ret] 200:print(Gewe download audio msg successfully.)print(data[data][fileUrl])return data[data][fileUrl]else:print(Gewe download audio msg in error.)return Falseelse:return Falsedef download_audio_file(fileUrl: str, file_name: str):# 定义保存文件的本地路径和文件名local_filename f./silk/{file_name}.silk# 使用requests库的get方法获取文件内容response requests.get(fileUrl, streamTrue)# 检查请求是否成功if response.status_code 200:# 打开文件以二进制写入模式with open(local_filename, wb) as f:# 逐块写入文件通常使用1024字节的块大小for chunk in response.iter_content(1024):f.write(chunk)print(f文件已成功下载到 {local_filename})else:print(f请求失败状态码: {response.status_code})百度云语音合成、识别
百度云语音合成、识别参考博客https://wrist.blog.csdn.net/article/details/139551206
百度智能云语音识别采用国际领先的流式端到端语音语言一体化建模算法将语音快速准确识别为文字支持手机应用语音交互、语音内容分析、机器人对话等场景。百度短语音识别可以将 60 秒以下的音频识别为文字。适用于语音对话、语音控制、语音输入等场景。 百度云语音合成、识别 audio_api
import os
import timeimport requests
from aip import AipSpeech
import dotenvdotenv.load_dotenv(.env)APP_ID os.getenv(APP_ID)
API_KEY os.getenv(API_KEY)
SECRET_KEY os.getenv(SECRET_KEY)client AipSpeech(APP_ID, API_KEY, SECRET_KEY)def create_audio(text: str, file_name: str):url https://tsn.baidu.com/text2audio# pcm# payload ftex{text}tok get_access_token() cuidnaAUL3Evzwj8A79l1kFHvRz1oUWZfOmoctp1lanzhspd5pit5vol5per0aue4# mp3payload ftex{text}tok get_access_token() cuidnaAUL3Evzwj8A79l1kFHvRz1oUWZfOmoctp1lanzhspd5pit5vol5per5003aue3response requests.get(urlf{url}?{payload}, datapayload, streamTrue)# 检查请求是否成功if response.status_code 200:# 打开文件以二进制写入模式with open(fmp3/{file_name}.mp3, wb) as file:# 分块读取并写入文件for chunk in response.iter_content(1024):if chunk:file.write(chunk)else:print(f请求失败状态码{response.status_code})def get_access_token():使用 AKSK 生成鉴权签名Access Token:return: access_token或是None(如果错误)url https://aip.baidubce.com/oauth/2.0/tokenparams {grant_type: client_credentials, client_id: API_KEY, client_secret: SECRET_KEY}return str(requests.post(url, paramsparams).json().get(access_token))# 读取文件
def get_file_content(file_name):read pcm file:param file_name::return:with open(f./pcm/{file_name}.pcm, rb) as fp:return fp.read()def recognize_audio(file_name):recognize pcm by baidu speech-api:param file_name::return:# 识别本地文件time.sleep(1)result client.asr(get_file_content(file_name), pcm, 16000, {dev_pid: 1537,})return result[result][0]
natapp 内网穿透
内网穿透NAT穿透是一种网络技术主要用于实现内部网络通常指局域网或私有网络与外部网络如互联网之间的通信。内网穿透指的是通过一定技术手段将内部网络中的计算机或服务暴露给公网实现内网与公网之间的互通。它的核心目标是使原本只能在内网内部访问的资源能够被外部网络的用户所访问。
内网穿透原理
内网穿透技术通过将公网地址映射到内网中的某个计算机或服务的地址实现了内外网之间的通信。在实现过程中可能涉及NAT网络地址转换技术该技术可以将私有保留地址转化为合法的IP地址。
UDP内网穿透技术是其中一种实现方式它利用路由器上的NAT系统使一个普通的内网节点在需要时将自己的服务器端口自动呈现在公网上并且能够让系统内其他节点正确获取这个地址。
内网穿透实现方法
反向代理是内网穿透的一种常见实现方法。通过反向代理服务器外部网络用户可以访问内部网络资源而无需直接访问内部网络。端口映射也称为端口转发是另一种实现方式它将外部网络的请求转发到内部网络的特定端口上。
内网穿透工具 natapp 官方地址https://natapp.cn/ 选择合适版本下载 购买隧道配置信息 本地创建 config.ini 配置文件
[default]
authtoken # 对应一条隧道的authtoken
clienttoken # 对应客户端的clienttoken,将会忽略authtoken,若无请留空,
lognone # log 日志文件,可指定本地文件, none不做记录,stdout直接屏幕输出 ,默认为none
loglevelERROR # 日志等级 DEBUG, INFO, WARNING, ERROR 默认为 DEBUG
http_proxy # 代理设置 如 http://10.123.10.10:3128 非代理上网用户请务必留空本地 natapp.exe 和 config.ini 放在同目录 启动内网穿透 发送语音消息
消息模块发送语音消息必须要是 https 协议的文件不能是本地 file 协议的文件所以我们需要通过内网穿透让我们的图片能够在公网访问这样 gewe 才能进行发送。
发送语音消息 send_msg.py
import osimport requests
import dotenv
dotenv.load_dotenv(.env)headers {X-GEWE-TOKEN: os.getenv(X_GEWE_TOKEN),User-Agent: Apifox/1.0.0 (https://apifox.com),Content-Type: application/json
}def send_audio_msg(toWxid: str, voice_file: str, voiceDuration: int):data {appId: wx_XqQOb78UPUFngK2ni2R9u,toWxid: toWxid,voiceUrl: fhttp://edgehacker.natapp1.cc/silk/{voice_file}.silk,voiceDuration: voiceDuration}response requests.post(http://api.geweapi.com/gewe/v2/api/message/postVoice, jsondata, headersheaders)if response.ok:data response.json()if data[ret] 200:print(Gewe sends audio msg successfully.)return Trueelse:print(Gewe sends audio msg in error.)return Falseelse:return FalseGewe 配置回调监听
我们需要登录API后台系统 点击“访问控制-填写回调地址” 将您的消息接收地址设置成功后续此Toekn下登录的所有微信机器人收到消息都会自动转发到您设置的回调接口地址。
Gewe 配置回调监听http://manager.geweapi.com/#/account/access 开发者需要提供的回调接口要求
微信消息是POST application/json请求您设置的回调接口地址。您可以直接打印请求body体可详细看到所有消息参数消息回调接口必须在3秒内响应否则平台将放弃本次请求结果回调接口可直接返回空字符串机器人通过接口发送的消息不会有回调因为回调是接收消息发送不属于接收但是手机微信发送的消息也会有因为这属于消息多端同步IM的原理配置成功后会接收一条包含文字“验证回调地址是否可用”的JSON回调注意回调接口必须发布到公网也就是第三方可以访问的临时测试可用内网穿透一类的但是网速不稳定可能丢消息若是显示回调验证失败您可以用Apifox访问您的回调接口验证是否可访问。Flask 服务接收微信消息
工具类 util.py
from datetime import datetimedef generate_timestamp():# 获取当前时间now datetime.now()# 格式化时间字符串为 yyyyMMddHHmmssSStimestamp now.strftime(%Y%m%d%H%M%S%f)[:-4]return timestampdef at_extract_content(text):# 找到最后一个空格的索引last_space_index text.rfind( )if last_space_index ! -1:# 返回空格后面的内容return text[last_space_index 1:]return def audio_extract_content(text):result text.split(\n, 1)[1]return result监听消息服务器 bot.py
import os
import time
from xml import etreefrom flask import Flask, send_from_directory, requestimport audio_api
import audio_to_text
import baidu_audio_to_text
import chat_api
import download_msg
import send_msg
import mp3_to_silk
import silk_to_pcm
import utilapp Flask(__name__)app.route(/silk/filename)
def serve_mp3(filename):# 指定 audios 目录的路径directory silk# 检查文件是否存在if not filename.endswith(.silk) or not os.path.isfile(os.path.join(directory, filename)):return File not found, 404# 使用 send_from_directory 发送文件return send_from_directory(directory, filename, as_attachmentFalse)app.route(/messages, methods[POST])
def gewe_message():msg request.get_json()PushContent msg[Data][PushContent]if PushContent.endswith(在群聊中发了一段语音):FromUserName msg[Data][FromUserName][string]file_name util.generate_timestamp()MsgId msg[Data][MsgId]Content util.audio_extract_content(msg[Data][Content][string])fileUrl download_msg.download_audio_msg(MsgId, Content)download_msg.download_audio_file(fileUrl, file_name)silk_to_pcm.transfrom_pcm(file_name)result baidu_audio_to_text.recognize_audio(file_name)print(result)result chat_api.chat_answer(result)print(result)audio_api.create_audio(nova, result, file_name)silk_data mp3_to_silk.transfrom_silk(file_name)send_msg.send_audio_msg(FromUserName, file_name, silk_data[voiceDuration])if PushContent.endswith([语音]):FromUserName msg[Data][FromUserName][string]file_name util.generate_timestamp()MsgId msg[Data][MsgId]Content msg[Data][Content][string]fileUrl download_msg.download_audio_msg(MsgId, Content)download_msg.download_audio_file(fileUrl, file_name)silk_to_pcm.transfrom_pcm(file_name)result baidu_audio_to_text.recognize_audio(file_name)result chat_api.chat_answer(result)audio_api.create_audio(nova, result, file_name)silk_data mp3_to_silk.transfrom_silk(file_name)send_msg.send_audio_msg(FromUserName, file_name, silk_data[voiceDuration])return successif __name__ __main__:app.run(debugTrue, port80)程序运行演示
控制台执行运行
python bot.py执行效果预览
发送语音泡自动智能回复消息
