03月23日
一、今日完成情况
- 睡觉的时候想到要完成一个AI保姆，是不是给我发消息，和我同步我学习的最新近况，让我抓住主要矛盾，频繁的在我耳边吹吹风。 –正在谋划中
- 完成API接口服务器转发，这样我就可以实现使用一个API就可以集大成者了。[[#五、new API 服务器中继]]
- 工作内容对接，我需要梳理一下刚才费老师实验室项目对接中，我需要完成的情况[[#四、工作对接情况]]
- 企业微信机器人消息，连接服务器打通流程，实现通讯。 –未来实现AI自动化，并且可以把消息推送到我的微信上。
- 配置Obsidian内置AI，使用claudian https://cloud.tencent.com/developer/article/2627758
二、今日感悟
- 核心业务数据：
- AI的API服务器公共IP转发
- 企业微信和服务器通讯打通
- 费老师PPT内容项目梳理
- 今日工作总结：
- 非常好，对于我未来长远的工作自动化，和API统一化有奠基的作用。在腾讯云服务器配置好我的clash verge内核代理，可以实现科学上网之后，我的服务器安装配置各种环境，就和我自己电脑的终端一样了，非常方便。∂
- 明日工作计划：
- 明天需要完成PPT的内容和流程图的制作，尝试使用Obsidian的插件。
- 今日学习成长：
- 无
三、备注
- 无
四、工作对接情况
1、我的PPT任务
S：完成廖师兄代码对应PPT制作
- 项目分解
- 1、通用的流程图设计方法，快速上手
- 2、项目PPT模板确定
- 3、项目文本确定
- 4、项目演示图片确定
M：
- 数量：三页左右
- 流程图：清晰明朗，可复用可修改，灵活性强
A：
- 轻轻松松可完成
R：
- 资料：老师的PPT模板参考
- 人员：无
T：
- 本周周五之前完成
五、new API 服务器中继
1、背景
我有许多API，可是我无法统一为同一个格式，因此我调用的时候非常麻烦， 我理想当中的情况，就是我可以把所有不同公司的API进行整合，变成同一个公司的固定格式，这样我python包在使用的时候就非常节约精力了，这是我找到的方案，使用new API进行公共服务器的中转。
2、Docker 配置
具体的最大众的Docker配置，可以看我这篇知识库的文章，我已经导出为公共笔记了：
[[2、Docker常用指令]]
# 创建存储数据的文件夹
mkdir -p /home/caprina/new-api
# 运行容器
docker run -d --restart always --name new-api \
-p 3000:3000 \
-v /home/ubuntu/new-api:/data \
-e TZ=Asia/Shanghai \
calciumion/new-api:latest
然后需要再防火墙上开放3000端口，用于TCP连接，在启动Docker之后，就可以公共网址访问我的API了：
现在离开黑窗口，打开电脑上的浏览器，访问：http://你的服务器IP:3000，这里我的IP是162.14.77.140,访问窗口如下：
这里配置好登录账号之后，就可以在窗口页面配置API的密钥和分发了，具体网页版的配置不做介绍，非常简单，最后只需要创建密钥就可以了，统一使用openai的格式进行访问：
python代码测试访问：
import openai
# --- 配置区 ---
# 1. 填入你在 New API "令牌管理" 里生成的 sk-xxxx
MY_API_KEY = "sk-uy6EJkbLhBsqZ11SdbDDTJ1MKcZeC17NT1mcLyRZa8bLQBoK"
# 2. 填入你的服务器公网 IP，注意必须带 /v1
MY_BASE_URL = "http://162.14.77.140:3000/v1"
# 3. 你想测试的模型列表（必须是你已经在渠道管理里添加过的模型名）
test_models = [
"deepseek-chat", # DeepSeek V3
"gemini-2.5-flash", # Gemini (走代理测试)
"grok-3", # xAI Grok
"qwen-max" # 阿里通义千问
]
# --------------
client = openai.OpenAI(
api_key=MY_API_KEY,
base_url=MY_BASE_URL
)
def test_model(model_name):
print(f"正在测试模型: {model_name} ...", end="", flush=True)
try:
response = client.chat.completions.create(
model=model_name,
messages=[{"role": "user", "content": "你好，请回复'连接成功'"}]
)
content = response.choices[0].message.content
print(f" ✅ 成功！回复内容: {content}")
except Exception as e:
print(f" ❌ 失败！报错信息: {e}")
if __name__ == "__main__":
print(f"--- 开始聚合 API 连通性测试 ---")
print(f"Base URL: {MY_BASE_URL}\n")
for model in test_models:
test_model(model)
print(f"\n--- 测试结束 ---")
对了，关于Gemini和xAI因为服务器在国外，所以我需要打通服务器代理的环节，这个不难，只需要在Docker启动的指令上面，显示确认端口即可，具体语法参考这个部分 [[2、Docker常用指令#^f2855c]]
六、企业微信 服务器 消息流程打通
1、创建机器人
机器人ID和密码记录：
- ID： 1000002
- 密码：TE12XMQp0eR3Yee_MP87YKuKvw472rlKkhrX2J3f_t8
- Company ID：wwda5759879600d079
2、开放IP白名单
Set the server URL for receiving messages，再此页面需要提前配置，才能在之后配置IP白名单。
Token：neSowIRGvBv
EncodingAESKey：BKebatbf9ADquRP2xmo8huQL9Hjfyx4cA5XuapaWvsX
在服务器端编写代码运行服务，监听8000端口，作为可以访问的域名：
from flask import Flask, request
from wechatpy.enterprise.crypto import WeChatCrypto
from wechatpy.exceptions import InvalidSignatureException
app = Flask(__name__)
# --- 换成你刚才在网页上生成的参数 ---
TOKEN = 'neSowIRGvBv'
AES_KEY = 'BKebatbf9ADquRP2xmo8huQL9Hjfyx4cA5XuapaWvsX'
CORP_ID = 'wwda5759879600d079'
crypto = WeChatCrypto(TOKEN, AES_KEY, CORP_ID)
@app.route('/wechat', methods=['GET'])
def verify():
msg_signature = request.args.get('msg_signature')
timestamp = request.args.get('timestamp')
nonce = request.args.get('nonce')
echostr = request.args.get('echostr')
try:
# 解密并返回原文字符串以通过验证
echo_str = crypto.check_signature(msg_signature, timestamp, nonce, echostr)
except InvalidSignatureException:
return 'error', 403
return echo_str
if __name__ == '__main__':
app.run(host='0.0.0.0', port=8000) # 腾讯云记得要在安全组开启8000端口
点击确定，进行授权，可以看到授权成功！
3、消息接受配置
需要点击机器人配置的消息接受设置，需要记录鉴权码：
url：http://162.14.77.140:8000/wechat
Token：neSowIRGvBv
EncodingAESKey：BKebatbf9ADquRP2xmo8huQL9Hjfyx4cA5XuapaWvsX
4、填入IP白名单/通讯测试
只要页面提示“保存成功”，那个灰色的 Company’s Trusted IP（企业可信IP）就会自动解封，变成可输入状态！
这时候把 162.14.77.140 填进去，测试第一个 mama_push.py 脚本是否可以跑通。
import requests
import json
import time
class MamaAI:
def __init__(self):
self.CORPID = "wwda5759879600d079"
self.SECRET = "TE12XMQp0eR3Yee_MP87YKuKvw472rlKkhrX2J3f_t8"
self.AGENTID = 1000002
self._token = None
self._token_expires = 0
def get_token(self):
# 简单缓存 token，避免频繁请求
if time.time() < self._token_expires:
return self._token
url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={self.CORPID}&corpsecret={self.SECRET}"
res = requests.get(url).json()
self._token = res.get("access_token")
self._token_expires = time.time() + 7000 # Token有效期通常为2小时
return self._token
def say(self, message):
token = self.get_token()
url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
data = {
"touser": "@all", # 发送给你可见范围内的所有人（也就是你自己）
"msgtype": "text",
"agentid": self.AGENTID,
"text": {"content": message},
"safe": 0
}
r = requests.post(url, data=json.dumps(data))
return r.json()
if __name__ == "__main__":
# 第一次测试
bot = MamaAI()
response = bot.say("Caprina，隧道打通了。接下来秋招的每一步，我都会在你耳边‘吹吹风’。")
print("发送结果：", response)
运行代码，代码的error_code 字段为0，可以知道，是成功了，现在打开手机，查看消息推送，可以看到，收到了我服务器这边发过来的微信消息，未来长远来讲，我可以让AI在服务器端实时运行，帮我手机端推送消息，我可以通信，和服务端的AI进行交互。
5、通讯 发送、接受 打通
import os
import struct
import base64
import hashlib
import xml.etree.ElementTree as ET
import requests
from flask import Flask, request
from Crypto.Cipher import AES
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
app = Flask(__name__)
# ====================== 配置信息 ======================
CORPID = os.getenv("CORPID")
SECRET = os.getenv("CORPSECRET")
AGENTID = int(os.getenv("AGENTID"))
TOKEN = os.getenv("WECHAT_TOKEN")
ENCODING_AES_KEY = os.getenv("ENCODING_AES_KEY")
# AI 客户端 (适配 Gemini 或 OpenAI 格式)
ai_client = OpenAI(
base_url=os.getenv("NEWAPI_URL"),
api_key=os.getenv("NEWAPI_KEY")
)
# ====================== 核心加解密类（修正版） ======================
class WXBizMsgCrypt:
def __init__(self, token, aes_key, corpid):
self.key = base64.b64decode(aes_key + "=")
self.token = token
self.corpid = corpid
def decrypt(self, encrypted):
"""解密并提取纯净的 XML 内容"""
# 1. AES-CBC 解密
cipher = AES.new(self.key, AES.MODE_CBC, self.key[:16])
raw = cipher.decrypt(base64.b64decode(encrypted))
# 2. 去掉 PKCS7 填充
pad = raw[-1]
content = raw[:-pad]
# 3. 剥离消息结构
# content[0:16] 是随机字符串
# content[16:20] 是 4 字节的消息长度 (网络字节序/大端序)
xml_len = struct.unpack('>I', content[16:20])[0]
# 4. 截取真正的 XML 文本内容
xml_content = content[20 : 20 + xml_len].decode('utf-8')
return xml_content
crypt = WXBizMsgCrypt(TOKEN, ENCODING_AES_KEY, CORPID)
# ====================== 工具函数 ======================
def get_token():
"""获取企业微信 Access Token"""
url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={CORPID}&corpsecret={SECRET}"
resp = requests.get(url).json()
return resp.get("access_token")
def ai_answer(content):
"""请求 AI 生成回复"""
try:
resp = ai_client.chat.completions.create(
model="gemini-2.5-flash", # 根据你的实际模型名修改
messages=[{"role": "user", "content": content}]
)
return resp.choices[0].message.content
except Exception as e:
print(f"AI 调用失败: {e}")
return "🤖 哎呀，我的大脑断网了..."
def send_msg(user, content):
"""主动推送消息给用户"""
token = get_token()
url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={token}"
data = {
"touser": user,
"msgtype": "text",
"agentid": AGENTID,
"text": {"content": content}
}
requests.post(url, json=data)
# ====================== 路由接口 ======================
@app.route("/wechat", methods=["GET"])
def verify():
"""验证 URL 有效性 (企业微信后台设置时触发)"""
echostr = request.args.get("echostr")
if not echostr:
return "Hello WeChat Bot"
try:
# GET 里的 echostr 也需要解密并原样返回明文
return crypt.decrypt(echostr)
except Exception as e:
print(f"验证失败: {e}")
return "verify error"
@app.route("/wechat", methods=["POST"])
def receive():
"""接收并处理消息"""
try:
# 1. 提取 XML 中的加密数据
xml_tree = ET.fromstring(request.data)
encrypt_content = xml_tree.find("Encrypt").text
# 2. 解密得到纯净的明文 XML
plain_xml = crypt.decrypt(encrypt_content)
msg_tree = ET.fromstring(plain_xml)
# 3. 解析用户 ID 和 消息内容
user_id = msg_tree.find("FromUserName").text
user_msg = msg_tree.find("Content").text
print(f"📩 收到消息: {user_msg} (来自: {user_id})")
# 4. 调用 AI (注意：如果 AI 响应超过 5 秒，微信会断开连接并重试)
reply = ai_answer(user_msg)
# 5. 发送回复
send_msg(user_id, reply)
return "success" # 必须给微信服务器返回 success 或空字符串
except Exception as e:
print(f"❌ 运行错误：{e}")
return "success"
if __name__ == "__main__":
# 企业微信要求必须是 80 端口（或其他已备案映射端口）
print("🚀 企业微信 AI 机器人已启动！")
app.run(host="0.0.0.0", port=8000)
6、后台启动关闭
启动：
nohup python3 wechat_agent.py > bot.log 2>&1 &
关闭：
ps -ef | grep wechat_agent.py
kill PID
七、claudian 内嵌AI终于实现
转载请注明来源，欢迎对文章中的引用来源进行考证，欢迎指出任何有错误或不够清晰的表达。可以在下面评论区评论，也可以邮件至 kipleyarch@gmail.com
