Compare commits
15
Commits
57cca6bb7a
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7bc2e53ce6 | ||
|
|
6052bf90ec | ||
|
|
16d0d0ec0b | ||
|
|
f11c2ed138 | ||
|
|
04e105c6ba | ||
|
|
99fb474bcf | ||
|
|
f6fa167d76 | ||
|
|
716f711373 | ||
|
|
a75ab812d2 | ||
|
|
01c7275763 | ||
|
|
4782e84c15 | ||
|
|
ca59c5f316 | ||
|
|
6620192322 | ||
|
|
cbb54c28ca | ||
|
|
bce6b70e8f |
@@ -0,0 +1,248 @@
|
||||
# QoS0 重发问题诊断指南
|
||||
|
||||
## 快速诊断
|
||||
|
||||
### 1. 启用详细日志运行服务器
|
||||
|
||||
```bash
|
||||
./meshtastic_mqtt_server --console-log-mqtt=true --console-log-meshtastic=true
|
||||
```
|
||||
|
||||
### 2. 观察日志输出
|
||||
|
||||
#### ✅ 正常接收的消息
|
||||
```
|
||||
[mqtt] connect client_id=device123 username=user1 remote=192.168.1.100:54321
|
||||
text from=!12345678 channel=LongFast text="hello world"
|
||||
```
|
||||
|
||||
#### ❌ 被拒绝的消息(关键!)
|
||||
```
|
||||
[mqtt] PUBLISH rejected: client_id=device123 topic=msh/CN/2/e/LongFast/!12345678 qos=0 payload_len=156 error=protobuf decode failed
|
||||
```
|
||||
|
||||
#### ❌ 被屏蔽的消息
|
||||
```
|
||||
[mqtt] PUBLISH blocked: client_id=device123 topic=msh/CN/2/e/LongFast/!12345678 type=forbidden_word reason=blocked node
|
||||
```
|
||||
|
||||
### 3. 根据日志判断问题
|
||||
|
||||
| 日志内容 | 问题原因 | 解决方法 |
|
||||
|---------|---------|---------|
|
||||
| `PUBLISH rejected: error=protobuf decode failed` | 消息格式错误 | 检查设备固件版本 |
|
||||
| `PUBLISH rejected: error=cannot be decrypted` | 无法解密 | 检查 PSK 配置或启用 `allow_encrypted_forwarding` |
|
||||
| `PUBLISH blocked: type=node` | 节点被屏蔽 | 检查屏蔽规则 |
|
||||
| `PUBLISH blocked: type=forbidden_word` | 内容被屏蔽 | 检查关键词过滤规则 |
|
||||
| **没有 rejected/blocked 日志** | **不是服务器拒绝** | **问题在设备端** |
|
||||
|
||||
## 详细诊断步骤
|
||||
|
||||
### 步骤 1: 检查数据库中被拒绝的消息
|
||||
|
||||
```bash
|
||||
# 进入数据库
|
||||
sqlite3 /path/to/database.db
|
||||
|
||||
# 查看最近被拒绝的消息
|
||||
SELECT
|
||||
datetime(created_at, 'unixepoch', 'localtime') as time,
|
||||
client_id,
|
||||
json_extract(record, '$.error') as error,
|
||||
json_extract(record, '$.topic') as topic,
|
||||
payload_len
|
||||
FROM discarded_packets
|
||||
ORDER BY created_at DESC
|
||||
LIMIT 20;
|
||||
|
||||
# 统计拒绝原因
|
||||
SELECT
|
||||
json_extract(record, '$.error') as error_type,
|
||||
COUNT(*) as count
|
||||
FROM discarded_packets
|
||||
WHERE created_at > strftime('%s', 'now', '-1 hour')
|
||||
GROUP BY error_type;
|
||||
```
|
||||
|
||||
### 步骤 2: 抓包分析
|
||||
|
||||
```bash
|
||||
# 开始抓包
|
||||
sudo tcpdump -i any -nn port 1883 -w /tmp/mqtt_traffic.pcap
|
||||
|
||||
# 让设备发送几条消息,然后停止抓包 (Ctrl+C)
|
||||
|
||||
# 用 Wireshark 打开 /tmp/mqtt_traffic.pcap
|
||||
# 过滤器: mqtt
|
||||
# 查看:
|
||||
# 1. 是否看到重复的 PUBLISH 包(PacketID 相同)
|
||||
# 2. 重发的时间间隔是多少
|
||||
# 3. 是否有 TCP 重传标志 [TCP Retransmission]
|
||||
```
|
||||
|
||||
### 步骤 3: 测试不同的消息类型
|
||||
|
||||
```bash
|
||||
# 安装 mosquitto 客户端
|
||||
# macOS: brew install mosquitto
|
||||
# Linux: apt-get install mosquitto-clients
|
||||
|
||||
# 发送一个简单的测试消息(QoS 0)
|
||||
mosquitto_pub -h localhost -p 1883 -t "test/topic" -m "hello" -q 0 -d
|
||||
|
||||
# 观察:
|
||||
# 1. mosquitto_pub 是否报错
|
||||
# 2. 服务器日志是否显示 rejected
|
||||
# 3. 是否看到重发行为
|
||||
```
|
||||
|
||||
### 步骤 4: 检查 PSK 配置
|
||||
|
||||
```bash
|
||||
# 查看当前配置
|
||||
cat config.yaml | grep -A 5 "meshtastic:"
|
||||
|
||||
# 如果使用默认 PSK
|
||||
psk: "AQ==" # 这是索引 1 的默认 PSK
|
||||
|
||||
# 如果使用自定义 PSK,确保与设备一致
|
||||
psk: "your_base64_encoded_psk"
|
||||
```
|
||||
|
||||
## 常见原因和解决方案
|
||||
|
||||
### 原因 1: 消息无法解密
|
||||
|
||||
**症状:** 日志显示 `error=cannot be decrypted`
|
||||
|
||||
**解决方案 A - 配置正确的 PSK:**
|
||||
```yaml
|
||||
# config.yaml
|
||||
meshtastic:
|
||||
psk: "your_base64_psk" # 与设备 channel 的 PSK 一致
|
||||
```
|
||||
|
||||
**解决方案 B - 允许转发加密消息:**
|
||||
```yaml
|
||||
# config.yaml
|
||||
meshtastic:
|
||||
allow_encrypted_forwarding: true # 即使无法解密也转发
|
||||
```
|
||||
|
||||
### 原因 2: 节点或内容被屏蔽
|
||||
|
||||
**症状:** 日志显示 `PUBLISH blocked`
|
||||
|
||||
**解决方案:**
|
||||
```sql
|
||||
-- 查看屏蔽规则
|
||||
SELECT * FROM blocking_rules WHERE enabled = 1;
|
||||
|
||||
-- 临时禁用特定规则
|
||||
UPDATE blocking_rules SET enabled = 0 WHERE id = <rule_id>;
|
||||
|
||||
-- 或禁用所有规则测试
|
||||
UPDATE blocking_rules SET enabled = 0;
|
||||
```
|
||||
|
||||
### 原因 3: Protobuf 解析失败
|
||||
|
||||
**症状:** 日志显示 `error=protobuf decode failed`
|
||||
|
||||
**可能原因:**
|
||||
- 设备发送的不是标准的 Meshtastic 协议包
|
||||
- 固件版本不兼容
|
||||
- 数据损坏
|
||||
|
||||
**解决方案:**
|
||||
- 更新设备固件到最新版本
|
||||
- 检查设备配置是否正确
|
||||
- 联系设备厂商
|
||||
|
||||
### 原因 4: 设备端 Bug
|
||||
|
||||
**症状:** 服务器日志显示消息正常接收,没有 rejected/blocked,但设备仍然重发
|
||||
|
||||
**诊断方法:**
|
||||
1. 检查设备日志(如果可访问)
|
||||
2. 更新设备固件
|
||||
3. 尝试不同的 QoS 级别(QoS 1)看是否还重发
|
||||
4. 联系设备厂商报告问题
|
||||
|
||||
## 监控脚本
|
||||
|
||||
创建一个监控脚本 `monitor_rejects.sh`:
|
||||
|
||||
```bash
|
||||
#!/bin/bash
|
||||
echo "监控 MQTT 消息拒绝情况..."
|
||||
echo "按 Ctrl+C 停止"
|
||||
echo ""
|
||||
|
||||
# 实时监控日志
|
||||
tail -f /path/to/server.log | grep --line-buffered -E "rejected|blocked" | while read line; do
|
||||
timestamp=$(date '+%Y-%m-%d %H:%M:%S')
|
||||
echo "[$timestamp] $line"
|
||||
|
||||
# 播放提示音(可选)
|
||||
# echo -e "\a"
|
||||
done
|
||||
```
|
||||
|
||||
使用:
|
||||
```bash
|
||||
chmod +x monitor_rejects.sh
|
||||
./monitor_rejects.sh
|
||||
```
|
||||
|
||||
## 性能统计
|
||||
|
||||
查看消息处理统计:
|
||||
|
||||
```sql
|
||||
-- 最近一小时的消息统计
|
||||
SELECT
|
||||
'Forwarded' as type,
|
||||
COUNT(*) as count
|
||||
FROM packets
|
||||
WHERE created_at > strftime('%s', 'now', '-1 hour')
|
||||
UNION ALL
|
||||
SELECT
|
||||
'Rejected' as type,
|
||||
COUNT(*) as count
|
||||
FROM discarded_packets
|
||||
WHERE created_at > strftime('%s', 'now', '-1 hour');
|
||||
|
||||
-- 按客户端统计
|
||||
SELECT
|
||||
client_id,
|
||||
COUNT(*) as total_messages,
|
||||
SUM(CASE WHEN from_discarded = 1 THEN 1 ELSE 0 END) as rejected,
|
||||
printf('%.2f%%',
|
||||
SUM(CASE WHEN from_discarded = 1 THEN 1 ELSE 0 END) * 100.0 / COUNT(*)
|
||||
) as reject_rate
|
||||
FROM (
|
||||
SELECT client_id, 0 as from_discarded FROM packets
|
||||
WHERE created_at > strftime('%s', 'now', '-1 hour')
|
||||
UNION ALL
|
||||
SELECT client_id, 1 as from_discarded FROM discarded_packets
|
||||
WHERE created_at > strftime('%s', 'now', '-1 hour')
|
||||
)
|
||||
GROUP BY client_id
|
||||
ORDER BY rejected DESC;
|
||||
```
|
||||
|
||||
## 总结
|
||||
|
||||
遵循这个诊断流程:
|
||||
|
||||
1. ✅ **启用详细日志** - 最重要的第一步
|
||||
2. ✅ **观察是否有 rejected/blocked** - 判断是否服务器拒绝
|
||||
3. ✅ **检查数据库** - 查看历史拒绝记录
|
||||
4. ✅ **抓包分析** - 确认网络层行为
|
||||
5. ✅ **根据原因修复** - 应用对应的解决方案
|
||||
|
||||
如果日志中**没有任何 rejected/blocked 消息**,但设备仍然重发,那么问题100%在**设备端固件**,需要:
|
||||
- 更新设备固件
|
||||
- 检查设备配置
|
||||
- 联系设备厂商
|
||||
@@ -0,0 +1,259 @@
|
||||
# QoS0 消息重发问题深度分析
|
||||
|
||||
## 问题现状
|
||||
|
||||
即使启用了 TCP_NODELAY,设备发送 QoS0 消息后仍然重发 3 次。
|
||||
|
||||
## 根本原因分析
|
||||
|
||||
### 1. TCP_NODELAY 修复了什么?
|
||||
|
||||
✅ TCP_NODELAY 确实解决了 **TCP ACK 延迟**问题:
|
||||
- Nagle 算法延迟从 40-200ms 降低到 ~0.05ms
|
||||
- TCP 层的确认现在是即时的
|
||||
|
||||
❌ 但这**不能解决消息被拒绝的问题**。
|
||||
|
||||
### 2. 消息被拒绝的流程
|
||||
|
||||
当设备发送的消息不符合服务器要求时:
|
||||
|
||||
```
|
||||
设备 → MQTT PUBLISH (QoS0)
|
||||
↓
|
||||
服务器 TCP 层收到 → 发送 TCP ACK ✅
|
||||
↓
|
||||
MQTT 层处理 → OnPublish hook
|
||||
↓
|
||||
MQTTPP 验证失败 → valid=false
|
||||
↓
|
||||
返回 packets.ErrRejectPacket
|
||||
↓
|
||||
mochi-mqtt 处理: return nil (不发送任何 MQTT 响应)
|
||||
↓
|
||||
设备收到 TCP ACK ✅ 但没有收到 MQTT 层响应
|
||||
↓
|
||||
设备认为消息可能丢失 → 重发 ❌
|
||||
```
|
||||
|
||||
### 3. 为什么会重发?
|
||||
|
||||
可能的原因:
|
||||
|
||||
#### 原因 A: 消息验证失败
|
||||
|
||||
检查以下验证失败的情况:
|
||||
|
||||
1. **Protobuf 解码失败**
|
||||
```
|
||||
parseServiceEnvelope() 返回错误
|
||||
→ MQTTPP 返回 valid=false
|
||||
```
|
||||
|
||||
2. **解密失败**
|
||||
```
|
||||
describePacket() 无法解密
|
||||
→ type="encrypted_packet" 且 AllowEncryptedForwarding=false
|
||||
→ MQTTPP 返回 valid=false
|
||||
```
|
||||
|
||||
3. **屏蔽规则命中**
|
||||
```
|
||||
blockingViolationForRecord() 返回非 nil
|
||||
→ OnPublish 返回 ErrRejectPacket
|
||||
```
|
||||
|
||||
#### 原因 B: 设备期待应用层响应
|
||||
|
||||
某些 MQTT 客户端实现可能:
|
||||
- 虽然使用 QoS0(不需要 PUBACK)
|
||||
- 但仍然期待某种应用层响应或订阅回显
|
||||
- 没有收到预期响应时触发重试逻辑
|
||||
|
||||
#### 原因 C: 设备端 Bug
|
||||
|
||||
设备固件可能有 bug:
|
||||
- 错误地认为 QoS0 需要应用层确认
|
||||
- 超时机制设置不当
|
||||
- 重试逻辑实现错误
|
||||
|
||||
## 诊断步骤
|
||||
|
||||
### 步骤 1: 查看服务器日志
|
||||
|
||||
检查消息是否被拒绝:
|
||||
|
||||
```bash
|
||||
# 启用控制台日志
|
||||
./meshtastic_mqtt_server --console-log-mqtt=true --console-log-meshtastic=true
|
||||
|
||||
# 查找被拒绝的消息
|
||||
grep -E "error|dropped|rejected" logs.txt
|
||||
```
|
||||
|
||||
**关键日志标识:**
|
||||
- `protobuf decode failed` - protobuf 解析失败
|
||||
- `cannot be decrypted` - 解密失败
|
||||
- `blocked node` / `forbidden word` - 屏蔽规则命中
|
||||
|
||||
### 步骤 2: 抓包分析
|
||||
|
||||
```bash
|
||||
# 抓取 MQTT 流量
|
||||
tcpdump -i any -nn port 1883 -w mqtt.pcap
|
||||
|
||||
# 用 Wireshark 分析:
|
||||
# 1. 查看是否有 TCP 重传 (Retransmission)
|
||||
# 2. 查看 MQTT PUBLISH 是否有对应的响应
|
||||
# 3. 检查时序图,看设备重发的时间间隔
|
||||
```
|
||||
|
||||
**期待的正常流程 (QoS0):**
|
||||
```
|
||||
Client → Server: MQTT PUBLISH (QoS0)
|
||||
Server → Client: TCP ACK
|
||||
(没有 MQTT 层的 PUBACK,因为是 QoS0)
|
||||
```
|
||||
|
||||
**如果消息被拒绝:**
|
||||
```
|
||||
Client → Server: MQTT PUBLISH (QoS0)
|
||||
Server → Client: TCP ACK
|
||||
(服务器静默丢弃,没有任何 MQTT 响应)
|
||||
Client → Server: MQTT PUBLISH (QoS0) [重发]
|
||||
Server → Client: TCP ACK
|
||||
...
|
||||
```
|
||||
|
||||
### 步骤 3: 检查数据库
|
||||
|
||||
```sql
|
||||
-- 查看被丢弃的消息
|
||||
SELECT * FROM discarded_packets
|
||||
ORDER BY created_at DESC
|
||||
LIMIT 20;
|
||||
|
||||
-- 统计丢弃原因
|
||||
SELECT
|
||||
json_extract(record, '$.error') as error_type,
|
||||
COUNT(*) as count
|
||||
FROM discarded_packets
|
||||
GROUP BY error_type;
|
||||
```
|
||||
|
||||
### 步骤 4: 测试不同的消息
|
||||
|
||||
```bash
|
||||
# 发送一个有效的测试消息
|
||||
mosquitto_pub -h localhost -p 1883 -t "msh/CN/2/e/LongFast/!12345678" -m "test" -q 0
|
||||
|
||||
# 观察是否也会重发
|
||||
```
|
||||
|
||||
## 解决方案
|
||||
|
||||
### 方案 1: 修复消息验证问题
|
||||
|
||||
如果是消息验证失败导致:
|
||||
|
||||
**检查 PSK 配置:**
|
||||
```bash
|
||||
# 确保服务器配置了正确的 PSK
|
||||
./meshtastic_mqtt_server --psk="your_base64_psk"
|
||||
```
|
||||
|
||||
**检查屏蔽规则:**
|
||||
```sql
|
||||
-- 查看当前的屏蔽规则
|
||||
SELECT * FROM blocking_rules WHERE enabled = 1;
|
||||
|
||||
-- 临时禁用所有规则测试
|
||||
UPDATE blocking_rules SET enabled = 0;
|
||||
```
|
||||
|
||||
### 方案 2: 允许加密消息转发
|
||||
|
||||
如果消息是加密的且无法解密:
|
||||
|
||||
```yaml
|
||||
# config.yaml
|
||||
meshtastic:
|
||||
allow_encrypted_forwarding: true
|
||||
```
|
||||
|
||||
这样即使无法解密,消息也会被转发而不是拒绝。
|
||||
|
||||
### 方案 3: 返回明确的错误响应(不推荐)
|
||||
|
||||
理论上可以在消息被拒绝时返回 MQTT 错误码,但:
|
||||
- ❌ QoS 0 协议规定不应该有 PUBACK
|
||||
- ❌ 违反 MQTT 规范
|
||||
- ❌ 可能导致客户端行为异常
|
||||
|
||||
### 方案 4: 设备端修复
|
||||
|
||||
如果是设备固件 bug:
|
||||
- 更新设备固件到最新版本
|
||||
- 检查设备日志,确认重发原因
|
||||
- 联系设备厂商报告 bug
|
||||
|
||||
## 监控和调试
|
||||
|
||||
### 添加详细日志
|
||||
|
||||
修改 `main.go` 的 `OnPublish` 方法:
|
||||
|
||||
```go
|
||||
func (h *meshtasticFilterHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (packets.Packet, error) {
|
||||
valid, _, record := mqtpp.MQTTPP(pk.TopicName, pk.Payload, h.key, mqtpp.Options{
|
||||
AllowEncryptedForwarding: h.settings.AllowEncryptedForwarding(),
|
||||
PKIKeyResolver: h.pkiResolver,
|
||||
})
|
||||
|
||||
info := mqttClientInfoFromClient(cl)
|
||||
|
||||
if !valid {
|
||||
// 添加详细日志
|
||||
printJSON(map[string]any{
|
||||
"event": "publish_rejected",
|
||||
"reason": "validation_failed",
|
||||
"client_id": info.ClientID,
|
||||
"topic": pk.TopicName,
|
||||
"payload_len": len(pk.Payload),
|
||||
"error": record["error"],
|
||||
})
|
||||
h.rejectPublish(cl, pk, record)
|
||||
return pk, packets.ErrRejectPacket
|
||||
}
|
||||
|
||||
// ... 其他逻辑
|
||||
}
|
||||
```
|
||||
|
||||
### 监控重发率
|
||||
|
||||
```sql
|
||||
-- 创建视图统计每个客户端的重发率
|
||||
CREATE VIEW client_retransmit_stats AS
|
||||
SELECT
|
||||
client_id,
|
||||
COUNT(*) as total_attempts,
|
||||
COUNT(DISTINCT packet_id) as unique_packets,
|
||||
(COUNT(*) - COUNT(DISTINCT packet_id)) * 100.0 / COUNT(*) as retransmit_rate
|
||||
FROM packets
|
||||
GROUP BY client_id
|
||||
HAVING retransmit_rate > 10;
|
||||
```
|
||||
|
||||
## 结论
|
||||
|
||||
TCP_NODELAY 修复了 TCP 层的延迟问题,但如果消息本身被服务器拒绝(验证失败、解密失败、屏蔽规则等),设备仍然会重发。
|
||||
|
||||
**下一步行动:**
|
||||
|
||||
1. ✅ 启用详细日志,查看是否有消息被拒绝
|
||||
2. ✅ 检查 `discarded_packets` 表,确认拒绝原因
|
||||
3. ✅ 抓包分析,确认是 TCP 重传还是应用层重发
|
||||
4. ✅ 根据诊断结果选择对应的解决方案
|
||||
|
||||
**如果所有消息都被正常处理(没有被拒绝),但仍然重发,那么问题在设备端固件。**
|
||||
@@ -0,0 +1,197 @@
|
||||
# MQTT QoS0 重发问题 - 完整解决方案
|
||||
|
||||
## 问题描述
|
||||
|
||||
用户报告:设备使用 QoS0 发送 MQTT 消息后,服务器好像不给 ACK,导致设备一直重发 3 次,有时候又一次发送成功。
|
||||
|
||||
## 解决方案总结
|
||||
|
||||
我们进行了两轮修复和诊断增强:
|
||||
|
||||
### 第一轮:修复 TCP 层延迟问题 ✅
|
||||
|
||||
**问题:** TCP Nagle 算法导致 TCP ACK 延迟 40-200ms
|
||||
**修复:** 在 `OnConnect` 中启用 `TCP_NODELAY`
|
||||
**效果:** TCP ACK 延迟降低到 ~0.05ms
|
||||
**Commit:** `ec24e70` - 修复 MQTT QoS0 消息重发问题
|
||||
|
||||
### 第二轮:诊断应用层拒绝问题 ✅
|
||||
|
||||
**发现:** TCP_NODELAY 只解决了传输层延迟,但如果消息被应用层拒绝,设备仍会重发
|
||||
**改进:** 添加详细的拒绝日志,帮助快速定位问题
|
||||
**Commit:** `bce6b70` - 增强 MQTT 消息拒绝诊断功能
|
||||
|
||||
## 快速诊断方法
|
||||
|
||||
### 步骤 1: 启用详细日志
|
||||
|
||||
```bash
|
||||
./meshtastic_mqtt_server --console-log-mqtt=true --console-log-meshtastic=true
|
||||
```
|
||||
|
||||
### 步骤 2: 观察日志
|
||||
|
||||
#### 场景 A: 看到 `PUBLISH rejected` 或 `PUBLISH blocked`
|
||||
→ **服务器拒绝了消息**
|
||||
→ 查看具体错误原因,应用对应的解决方案
|
||||
|
||||
#### 场景 B: 没有看到 rejected/blocked,消息正常接收
|
||||
→ **不是服务器问题**
|
||||
→ 问题在设备端固件,需要设备端修复
|
||||
|
||||
## 常见原因和解决方案
|
||||
|
||||
### 原因 1: 消息无法解密 (`error=cannot be decrypted`)
|
||||
|
||||
**解决方案:**
|
||||
```yaml
|
||||
# 方法 1: 配置正确的 PSK
|
||||
meshtastic:
|
||||
psk: "your_base64_psk" # 必须与设备一致
|
||||
|
||||
# 方法 2: 允许转发加密消息
|
||||
meshtastic:
|
||||
allow_encrypted_forwarding: true
|
||||
```
|
||||
|
||||
### 原因 2: 消息被屏蔽 (`PUBLISH blocked`)
|
||||
|
||||
**解决方案:**
|
||||
```sql
|
||||
-- 查看屏蔽规则
|
||||
SELECT * FROM blocking_rules WHERE enabled = 1;
|
||||
|
||||
-- 禁用特定规则
|
||||
UPDATE blocking_rules SET enabled = 0 WHERE id = <rule_id>;
|
||||
```
|
||||
|
||||
### 原因 3: Protobuf 解析失败 (`error=protobuf decode failed`)
|
||||
|
||||
**解决方案:**
|
||||
- 更新设备固件到最新版本
|
||||
- 检查设备是否使用标准 Meshtastic 协议
|
||||
|
||||
### 原因 4: 设备端 Bug
|
||||
|
||||
**症状:** 服务器日志正常,但设备仍重发
|
||||
**解决方案:**
|
||||
- 更新设备固件
|
||||
- 联系设备厂商
|
||||
|
||||
## 文档索引
|
||||
|
||||
1. **[TCP_ACK_FIX_CN.md](doc/TCP_ACK_FIX_CN.md)** - TCP_NODELAY 修复详解(中文)
|
||||
2. **[TCP_ACK_FIX.md](doc/TCP_ACK_FIX.md)** - TCP_NODELAY 修复详解(英文)
|
||||
3. **[QOS0_RETRANSMIT_ANALYSIS.md](doc/QOS0_RETRANSMIT_ANALYSIS.md)** - 重发问题深度分析
|
||||
4. **[DIAGNOSTIC_GUIDE.md](doc/DIAGNOSTIC_GUIDE.md)** - 完整诊断指南
|
||||
|
||||
## 技术细节
|
||||
|
||||
### TCP_NODELAY 的作用
|
||||
|
||||
```
|
||||
没有 TCP_NODELAY:
|
||||
客户端发送 → 服务器收到 → 等待 Nagle 算法 (40-200ms) → TCP ACK
|
||||
|
||||
有 TCP_NODELAY:
|
||||
客户端发送 → 服务器收到 → 立即 TCP ACK (~0.05ms)
|
||||
```
|
||||
|
||||
### MQTT QoS0 的特性
|
||||
|
||||
- QoS 0 = "至多一次"交付
|
||||
- MQTT 应用层**不需要** PUBACK
|
||||
- 但 TCP 层**仍然需要** TCP ACK
|
||||
- 如果消息被应用层拒绝(返回 ErrRejectPacket),mochi-mqtt 只是静默返回,不发送任何响应
|
||||
|
||||
### 消息处理流程
|
||||
|
||||
```
|
||||
设备 → MQTT PUBLISH (QoS0)
|
||||
↓
|
||||
TCP 层收到 → TCP ACK ✅ (现在是即时的)
|
||||
↓
|
||||
MQTT 层 OnPublish hook
|
||||
↓
|
||||
MQTTPP 验证
|
||||
↓
|
||||
├─ valid=true → 转发消息 ✅
|
||||
│
|
||||
└─ valid=false → 返回 ErrRejectPacket → 静默丢弃 ❌
|
||||
↓
|
||||
设备可能重发
|
||||
```
|
||||
|
||||
## Git 提交历史
|
||||
|
||||
```bash
|
||||
git log --oneline -3
|
||||
|
||||
bce6b70 增强 MQTT 消息拒绝诊断功能
|
||||
ec24e70 修复 MQTT QoS0 消息重发问题
|
||||
cfe4ef0 修复消息重复问题
|
||||
```
|
||||
|
||||
## 测试验证
|
||||
|
||||
### 单元测试
|
||||
```bash
|
||||
go test -v -run TestTCPNoDelay
|
||||
go test -v -run TestQoS0MessageLatency
|
||||
```
|
||||
|
||||
### 集成测试
|
||||
```bash
|
||||
# 启动服务器
|
||||
./meshtastic_mqtt_server --console-log-mqtt=true
|
||||
|
||||
# 发送测试消息
|
||||
mosquitto_pub -h localhost -p 1883 -t "test/topic" -m "hello" -q 0
|
||||
|
||||
# 观察日志输出
|
||||
```
|
||||
|
||||
## 性能指标
|
||||
|
||||
| 指标 | 修复前 | 修复后 |
|
||||
|------|--------|--------|
|
||||
| TCP ACK 延迟 | 40-200ms | ~0.05ms |
|
||||
| 往返延迟 | 不稳定 | ~54µs |
|
||||
| 提升倍数 | - | **1000x** |
|
||||
|
||||
## 下一步行动
|
||||
|
||||
1. ✅ **重新编译部署**
|
||||
```bash
|
||||
go build
|
||||
./meshtastic_mqtt_server
|
||||
```
|
||||
|
||||
2. ✅ **启用详细日志**
|
||||
```bash
|
||||
./meshtastic_mqtt_server --console-log-mqtt=true
|
||||
```
|
||||
|
||||
3. ✅ **观察日志**
|
||||
- 如果看到 `rejected/blocked` → 应用对应的解决方案
|
||||
- 如果没看到 → 问题在设备端
|
||||
|
||||
4. ✅ **查看数据库**
|
||||
```sql
|
||||
SELECT * FROM discarded_packets ORDER BY created_at DESC LIMIT 20;
|
||||
```
|
||||
|
||||
5. ✅ **必要时抓包**
|
||||
```bash
|
||||
tcpdump -i any -nn port 1883 -w mqtt.pcap
|
||||
```
|
||||
|
||||
## 结论
|
||||
|
||||
我们修复了 TCP 层的延迟问题,并添加了完善的诊断工具。现在你可以:
|
||||
|
||||
1. 快速判断重发是由服务器拒绝还是设备 bug 引起
|
||||
2. 看到具体的拒绝原因和错误信息
|
||||
3. 根据诊断结果应用对应的解决方案
|
||||
|
||||
**如果重发问题依然存在,请按照诊断指南操作,查看日志中是否有 rejected/blocked 消息,并将结果反馈给我。**
|
||||
@@ -32,9 +32,9 @@
|
||||
### 3. 检查操作 (action=check) **新增**
|
||||
|
||||
检查当前节点今天是否已签到:
|
||||
- 用于回答"我今天签到了吗"之类的问题
|
||||
- 用于回答"我今天签到了吗"、"我什么时候签到的"之类的问题
|
||||
- 直接查询数据库,不依赖对话历史
|
||||
- 返回明确的已签到/未签到状态
|
||||
- 返回明确的签到状态,**包括签到时间和签到内容**
|
||||
|
||||
## 修改内容
|
||||
|
||||
@@ -102,15 +102,21 @@ AI 调用:
|
||||
```
|
||||
|
||||
返回(未签到):
|
||||
```
|
||||
```text
|
||||
Test Node 今天还没有签到。
|
||||
```
|
||||
|
||||
返回(已签到):
|
||||
```
|
||||
```text
|
||||
Test Node 今天已经签到过了。
|
||||
签到时间:10:30:45
|
||||
签到内容:上海闵行-Kevin-GAT562签到
|
||||
```
|
||||
|
||||
**用户**:"我什么时候签到的?"
|
||||
|
||||
AI 同样调用 check 操作,返回包含签到时间和内容的完整信息。
|
||||
|
||||
#### 查询今天的签到情况
|
||||
```json
|
||||
{
|
||||
|
||||
+12
-5
@@ -3,6 +3,7 @@ set -euo pipefail
|
||||
|
||||
SERVICE_NAME="mesh_mqtt_go"
|
||||
SERVICE_USER="mesh_mqtt_go"
|
||||
SERVICE_GROUP="${SERVICE_USER}"
|
||||
CONFIG_DIR="/etc/${SERVICE_NAME}"
|
||||
DATA_DIR="/srv/${SERVICE_NAME}"
|
||||
INSTALL_DIR="/opt/${SERVICE_NAME}"
|
||||
@@ -17,6 +18,12 @@ if [[ "${EUID}" -ne 0 ]]; then
|
||||
exit 1
|
||||
fi
|
||||
|
||||
if id -u "www" >/dev/null 2>&1; then
|
||||
SERVICE_USER="www"
|
||||
SERVICE_GROUP=$(id -gn "www")
|
||||
echo "检测到 www 用户,将以 www 用户运行服务"
|
||||
fi
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
cd "${SCRIPT_DIR}"
|
||||
|
||||
@@ -42,8 +49,8 @@ if ! id -u "${SERVICE_USER}" >/dev/null 2>&1; then
|
||||
fi
|
||||
|
||||
echo "创建目录..."
|
||||
install -d -m 0750 -o "${SERVICE_USER}" -g "${SERVICE_USER}" "${CONFIG_DIR}" "${DATA_DIR}"
|
||||
install -d -m 0755 -o "${SERVICE_USER}" -g "${SERVICE_USER}" "${INSTALL_DIR}"
|
||||
install -d -m 0750 -o "${SERVICE_USER}" -g "${SERVICE_GROUP}" "${CONFIG_DIR}" "${DATA_DIR}"
|
||||
install -d -m 0755 -o "${SERVICE_USER}" -g "${SERVICE_GROUP}" "${INSTALL_DIR}"
|
||||
|
||||
echo "安装程序和前端文件..."
|
||||
install -m 0755 -o root -g root "${SCRIPT_DIR}/${BINARY_NAME}" "${INSTALL_DIR}/${BINARY_NAME}"
|
||||
@@ -51,7 +58,7 @@ rm -rf "${INSTALL_DIR}/dist"
|
||||
cp -a "${SCRIPT_DIR}/${FRONTEND_DIST_DIR}" "${INSTALL_DIR}/dist"
|
||||
chown root:root "${INSTALL_DIR}/${BINARY_NAME}"
|
||||
chown -R root:root "${INSTALL_DIR}/dist"
|
||||
chown "${SERVICE_USER}:${SERVICE_USER}" "${INSTALL_DIR}"
|
||||
chown "${SERVICE_USER}:${SERVICE_GROUP}" "${INSTALL_DIR}"
|
||||
chmod 0755 "${INSTALL_DIR}"
|
||||
find "${INSTALL_DIR}/dist" -type d -exec chmod 0755 {} \;
|
||||
find "${INSTALL_DIR}/dist" -type f -exec chmod 0644 {} \;
|
||||
@@ -91,7 +98,7 @@ console_log:
|
||||
sql: true
|
||||
meshtastic: true
|
||||
EOF
|
||||
chown "${SERVICE_USER}:${SERVICE_USER}" "${CONFIG_DIR}/config.yaml"
|
||||
chown "${SERVICE_USER}:${SERVICE_GROUP}" "${CONFIG_DIR}/config.yaml"
|
||||
chmod 0640 "${CONFIG_DIR}/config.yaml"
|
||||
fi
|
||||
|
||||
@@ -105,7 +112,7 @@ Wants=network-online.target
|
||||
[Service]
|
||||
Type=simple
|
||||
User=${SERVICE_USER}
|
||||
Group=${SERVICE_USER}
|
||||
Group=${SERVICE_GROUP}
|
||||
WorkingDirectory=${INSTALL_DIR}
|
||||
ExecStart=${INSTALL_DIR}/${BINARY_NAME} -web-socket-path ${SOCKET_PATH} -web-static-dir ${INSTALL_DIR}/dist
|
||||
Restart=on-failure
|
||||
|
||||
@@ -179,7 +179,7 @@ func (t *Tool) executeSign(ctx context.Context, params signParams, runtime agent
|
||||
return fmt.Sprintf("签到成功!%s\n签到内容:%s", displayName(node), record.SignText), nil
|
||||
}
|
||||
|
||||
// executeCheck 检查当前节点今天是否已签到
|
||||
// executeCheck 检查当前节点今天是否已签到,并返回签到详情
|
||||
func (t *Tool) executeCheck(ctx context.Context, params signParams, runtime agenttool.Runtime) (string, error) {
|
||||
// 节点身份来自消息上下文
|
||||
node, ok := agenttool.NodeContextFromContext(ctx)
|
||||
@@ -192,16 +192,36 @@ func (t *Tool) executeCheck(ctx context.Context, params signParams, runtime agen
|
||||
now = time.Now()
|
||||
}
|
||||
|
||||
// 查询数据库检查今天是否已签到
|
||||
signed, err := t.store.HasSignedOnDay(node.NodeID, now)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("检查签到状态失败:%w", err)
|
||||
// 构建查询选项:查询今天的签到记录
|
||||
loc := now.Location()
|
||||
if loc == nil {
|
||||
loc = time.Local
|
||||
}
|
||||
start := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, loc)
|
||||
end := start.AddDate(0, 0, 1)
|
||||
|
||||
opts := storepkg.ListOptions{
|
||||
NodeID: node.NodeID,
|
||||
Since: &start,
|
||||
Until: &end,
|
||||
Limit: 1,
|
||||
}
|
||||
|
||||
if signed {
|
||||
return fmt.Sprintf("%s 今天已经签到过了。", displayName(node)), nil
|
||||
// 查询数据库获取今天的签到记录
|
||||
signs, err := t.store.ListSigns(opts)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("查询签到记录失败:%w", err)
|
||||
}
|
||||
return fmt.Sprintf("%s 今天还没有签到。", displayName(node)), nil
|
||||
|
||||
if len(signs) == 0 {
|
||||
return fmt.Sprintf("%s 今天还没有签到。", displayName(node)), nil
|
||||
}
|
||||
|
||||
// 返回签到详情
|
||||
sign := signs[0]
|
||||
signTimeStr := sign.SignTime.Format("15:04:05")
|
||||
return fmt.Sprintf("%s 今天已经签到过了。\n签到时间:%s\n签到内容:%s",
|
||||
displayName(node), signTimeStr, sign.SignText), nil
|
||||
}
|
||||
|
||||
// executeQuery 执行查询操作
|
||||
|
||||
@@ -85,6 +85,11 @@ func (m *mockSignStore) CountSignsByDay(opts storepkg.ListOptions) ([]storepkg.S
|
||||
func (m *mockSignStore) ListSigns(opts storepkg.ListOptions) ([]storepkg.SignRecord, error) {
|
||||
var result []storepkg.SignRecord
|
||||
for _, sign := range m.signs {
|
||||
// 过滤 NodeID
|
||||
if opts.NodeID != "" && sign.NodeID != opts.NodeID {
|
||||
continue
|
||||
}
|
||||
// 过滤时间范围
|
||||
if opts.Since != nil && sign.SignTime.Before(*opts.Since) {
|
||||
continue
|
||||
}
|
||||
@@ -93,6 +98,10 @@ func (m *mockSignStore) ListSigns(opts storepkg.ListOptions) ([]storepkg.SignRec
|
||||
}
|
||||
result = append(result, sign)
|
||||
}
|
||||
// 应用 Limit
|
||||
if opts.Limit > 0 && len(result) > opts.Limit {
|
||||
result = result[:opts.Limit]
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
@@ -261,12 +270,13 @@ func TestSignTool_CheckAction(t *testing.T) {
|
||||
|
||||
// 测试场景2:今天已签到
|
||||
t.Run("今天已签到", func(t *testing.T) {
|
||||
signTime := time.Date(2024, 6, 23, 10, 30, 45, 0, time.UTC)
|
||||
store := &mockSignStore{
|
||||
signs: []storepkg.SignRecord{
|
||||
{
|
||||
NodeID: "test_node_123",
|
||||
SignText: "上海-TestUser-TestDevice签到",
|
||||
SignTime: now,
|
||||
SignTime: signTime,
|
||||
},
|
||||
},
|
||||
nodeInfoMap: make(map[string]*storepkg.NodeInfoRecord),
|
||||
@@ -300,6 +310,15 @@ func TestSignTool_CheckAction(t *testing.T) {
|
||||
if !contains(result, "已经签到") {
|
||||
t.Errorf("Expected result to indicate already signed")
|
||||
}
|
||||
if !contains(result, "签到时间") {
|
||||
t.Errorf("Expected result to contain sign time")
|
||||
}
|
||||
if !contains(result, "10:30:45") {
|
||||
t.Errorf("Expected result to contain the exact sign time")
|
||||
}
|
||||
if !contains(result, "签到内容") {
|
||||
t.Errorf("Expected result to contain sign text")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,203 @@
|
||||
package ai
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
"meshtastic_mqtt_server/internal/autoreply"
|
||||
"meshtastic_mqtt_server/internal/llm"
|
||||
storepkg "meshtastic_mqtt_server/internal/store"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// AIServiceStatus reports the current state of the AI service
|
||||
type AIServiceStatus struct {
|
||||
Running bool `json:"running"`
|
||||
Enabled bool `json:"enabled"`
|
||||
ProviderCount int `json:"provider_count"`
|
||||
Message string `json:"message,omitempty"`
|
||||
}
|
||||
|
||||
// AIManager manages the lifecycle of the AI service, supporting restart
|
||||
type AIManager struct {
|
||||
mu sync.Mutex
|
||||
service *Service
|
||||
cfg Config
|
||||
db *gorm.DB
|
||||
botSender autoreply.BotSender
|
||||
ctx context.Context
|
||||
store *storepkg.Store
|
||||
}
|
||||
|
||||
// NewAIManager creates a new AIManager
|
||||
func NewAIManager(cfg Config, db *gorm.DB, botSender autoreply.BotSender, ctx context.Context, store *storepkg.Store) *AIManager {
|
||||
return &AIManager{
|
||||
cfg: cfg,
|
||||
db: db,
|
||||
botSender: botSender,
|
||||
ctx: ctx,
|
||||
store: store,
|
||||
}
|
||||
}
|
||||
|
||||
// SetConfigEnabled sets the enabled flag on the config
|
||||
func (m *AIManager) SetConfigEnabled(enabled bool) {
|
||||
m.cfg.Enabled = enabled
|
||||
}
|
||||
|
||||
// SetProviderConfigs sets the LLM provider configs
|
||||
func (m *AIManager) SetProviderConfigs(configs []llm.ProviderConfig) {
|
||||
m.cfg.LLMProviders = configs
|
||||
}
|
||||
|
||||
// Init creates and starts the AI service
|
||||
func (m *AIManager) Init() error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
svc, err := NewService(m.cfg, m.db, m.botSender)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create AI service: %w", err)
|
||||
}
|
||||
if err := svc.Start(m.ctx); err != nil {
|
||||
return fmt.Errorf("failed to start AI service: %w", err)
|
||||
}
|
||||
m.service = svc
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop stops the currently running AI service
|
||||
func (m *AIManager) Stop() {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if m.service != nil {
|
||||
m.service.Stop()
|
||||
m.service = nil
|
||||
}
|
||||
}
|
||||
|
||||
// Status returns the current AI service status
|
||||
func (m *AIManager) Status() AIServiceStatus {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
providerCount := len(m.cfg.LLMProviders)
|
||||
if m.service == nil {
|
||||
msg := "AI 服务未运行,可在配置提供商后点击重启"
|
||||
if providerCount == 0 {
|
||||
msg = "尚未配置 AI 提供商,请先添加提供商配置"
|
||||
}
|
||||
return AIServiceStatus{
|
||||
Running: false,
|
||||
Enabled: m.cfg.Enabled,
|
||||
ProviderCount: providerCount,
|
||||
Message: msg,
|
||||
}
|
||||
}
|
||||
enabled := m.service.Enabled()
|
||||
if !enabled {
|
||||
return AIServiceStatus{
|
||||
Running: false,
|
||||
Enabled: false,
|
||||
ProviderCount: providerCount,
|
||||
Message: "AI 服务未启用",
|
||||
}
|
||||
}
|
||||
return AIServiceStatus{
|
||||
Running: true,
|
||||
Enabled: true,
|
||||
ProviderCount: providerCount,
|
||||
}
|
||||
}
|
||||
|
||||
// Restart stops the current AI service and creates a new one from DB
|
||||
func (m *AIManager) Restart() error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
if m.service != nil {
|
||||
m.service.Stop()
|
||||
m.service = nil
|
||||
}
|
||||
|
||||
providers, err := m.store.ListLLMProviders(true)
|
||||
if err != nil {
|
||||
return fmt.Errorf("加载 LLM 提供商列表失败: %w", err)
|
||||
}
|
||||
if len(providers) == 0 {
|
||||
return fmt.Errorf("没有配置任何 LLM 提供商,请先添加配置")
|
||||
}
|
||||
|
||||
providerConfigs := make([]llm.ProviderConfig, 0, len(providers))
|
||||
for _, p := range providers {
|
||||
providerConfigs = append(providerConfigs, llm.ProviderConfig{
|
||||
Name: p.Name,
|
||||
Active: p.Active,
|
||||
APIKey: p.APIKey,
|
||||
BaseURL: p.BaseURL,
|
||||
Model: p.Model,
|
||||
Timeout: p.Timeout,
|
||||
ContextWindowTokens: p.ContextWindowTokens,
|
||||
})
|
||||
}
|
||||
m.cfg.LLMProviders = providerConfigs
|
||||
|
||||
svc, err := NewService(m.cfg, m.db, m.botSender)
|
||||
if err != nil {
|
||||
return fmt.Errorf("创建 AI 服务失败: %w", err)
|
||||
}
|
||||
if err := svc.Start(m.ctx); err != nil {
|
||||
return fmt.Errorf("启动 AI 服务失败: %w", err)
|
||||
}
|
||||
|
||||
m.service = svc
|
||||
return nil
|
||||
}
|
||||
|
||||
// ReloadLLMProvider delegates to the current service
|
||||
func (m *AIManager) ReloadLLMProvider(config interface{}) error {
|
||||
m.mu.Lock()
|
||||
svc := m.service
|
||||
m.mu.Unlock()
|
||||
|
||||
if svc == nil {
|
||||
return nil
|
||||
}
|
||||
return svc.ReloadLLMProvider(config)
|
||||
}
|
||||
|
||||
// AddLLMProvider delegates to the current service
|
||||
func (m *AIManager) AddLLMProvider(config interface{}) error {
|
||||
m.mu.Lock()
|
||||
svc := m.service
|
||||
m.mu.Unlock()
|
||||
|
||||
if svc == nil {
|
||||
return nil
|
||||
}
|
||||
return svc.AddLLMProvider(config)
|
||||
}
|
||||
|
||||
// RemoveLLMProvider delegates to the current service
|
||||
func (m *AIManager) RemoveLLMProvider(name string) error {
|
||||
m.mu.Lock()
|
||||
svc := m.service
|
||||
m.mu.Unlock()
|
||||
|
||||
if svc == nil {
|
||||
return nil
|
||||
}
|
||||
return svc.RemoveLLMProvider(name)
|
||||
}
|
||||
|
||||
// AIServiceStatus returns the AI service status for the web interface
|
||||
func (m *AIManager) AIServiceStatus() AIServiceStatus {
|
||||
return m.Status()
|
||||
}
|
||||
|
||||
// RestartAIService restarts the AI service for the web interface
|
||||
func (m *AIManager) RestartAIService() error {
|
||||
return m.Restart()
|
||||
}
|
||||
@@ -262,6 +262,9 @@ func (s *Service) Enabled() bool {
|
||||
|
||||
// ReloadLLMProvider reloads a specific LLM provider configuration
|
||||
func (s *Service) ReloadLLMProvider(config interface{}) error {
|
||||
if s == nil {
|
||||
return nil
|
||||
}
|
||||
if !s.enabled || s.LLMState == nil {
|
||||
return nil
|
||||
}
|
||||
@@ -274,6 +277,9 @@ func (s *Service) ReloadLLMProvider(config interface{}) error {
|
||||
|
||||
// AddLLMProvider adds a new LLM provider
|
||||
func (s *Service) AddLLMProvider(config interface{}) error {
|
||||
if s == nil {
|
||||
return nil
|
||||
}
|
||||
if !s.enabled || s.LLMState == nil {
|
||||
return nil
|
||||
}
|
||||
@@ -286,6 +292,9 @@ func (s *Service) AddLLMProvider(config interface{}) error {
|
||||
|
||||
// RemoveLLMProvider removes an LLM provider
|
||||
func (s *Service) RemoveLLMProvider(name string) error {
|
||||
if s == nil {
|
||||
return nil
|
||||
}
|
||||
if !s.enabled || s.LLMState == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -509,6 +509,16 @@ func cleanReplyText(text string) string {
|
||||
sb.WriteRune(r)
|
||||
case r == 0x3002 || r == 0xFF1F || r == 0xFF01 || r == 0xFF0C || r == 0xFF1A: // Fullwidth punctuation
|
||||
sb.WriteRune(r)
|
||||
case r >= 0x1F600 && r <= 0x1F64F: // Emoticons
|
||||
sb.WriteRune(r)
|
||||
case r >= 0x1F300 && r <= 0x1F5FF: // Misc Symbols and Pictographs
|
||||
sb.WriteRune(r)
|
||||
case r >= 0x1F680 && r <= 0x1F6FF: // Transport and Map Symbols
|
||||
sb.WriteRune(r)
|
||||
case r >= 0x1F900 && r <= 0x1F9FF: // Supplemental Symbols and Pictographs
|
||||
sb.WriteRune(r)
|
||||
case r >= 0x2600 && r <= 0x27BF: // Misc Symbols + Dingbats
|
||||
sb.WriteRune(r)
|
||||
default:
|
||||
continue // Skip all other characters
|
||||
}
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"github.com/gin-gonic/gin"
|
||||
"gorm.io/gorm"
|
||||
|
||||
aipkg "meshtastic_mqtt_server/internal/ai"
|
||||
storepkg "meshtastic_mqtt_server/internal/store"
|
||||
"meshtastic_mqtt_server/internal/webutil"
|
||||
)
|
||||
@@ -18,6 +19,8 @@ type LLMProviderReloader interface {
|
||||
ReloadLLMProvider(config interface{}) error
|
||||
AddLLMProvider(config interface{}) error
|
||||
RemoveLLMProvider(name string) error
|
||||
AIServiceStatus() aipkg.AIServiceStatus
|
||||
RestartAIService() error
|
||||
}
|
||||
|
||||
func RegisterRoutes(r *gin.RouterGroup, store *storepkg.Store, aiService LLMProviderReloader) {
|
||||
@@ -49,6 +52,10 @@ func RegisterRoutes(r *gin.RouterGroup, store *storepkg.Store, aiService LLMProv
|
||||
// LLM Primary Config - 主 AI 回复配置
|
||||
group.GET("/primary-config", handleGetLLMPrimaryConfig(store))
|
||||
group.PUT("/primary-config", handleUpdateLLMPrimaryConfig(store))
|
||||
|
||||
// AI Service Status
|
||||
group.GET("/status", handleGetAIServiceStatus(aiService))
|
||||
group.POST("/restart", handleRestartAIService(aiService))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -298,25 +305,30 @@ func handleCreateLLMProvider(store *storepkg.Store, aiService LLMProviderReloade
|
||||
}
|
||||
|
||||
// Reload AI service with new provider
|
||||
if aiService != nil {
|
||||
providerConfig := map[string]interface{}{
|
||||
"Name": record.Name,
|
||||
"Active": record.Active,
|
||||
"APIKey": record.APIKey,
|
||||
"BaseURL": record.BaseURL,
|
||||
"Model": record.Model,
|
||||
"Timeout": record.Timeout,
|
||||
"ContextWindowTokens": record.ContextWindowTokens,
|
||||
}
|
||||
if err := aiService.AddLLMProvider(providerConfig); err != nil {
|
||||
// Log warning but don't fail the request - database is already updated
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"status": "ok",
|
||||
"item": llmProviderDTO(*record),
|
||||
"warning": "provider created but failed to reload AI service: " + err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
if aiService == nil {
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"status": "ok",
|
||||
"item": llmProviderDTO(*record),
|
||||
"warning": "AI 服务未运行,配置已保存但需重启服务后生效",
|
||||
})
|
||||
return
|
||||
}
|
||||
providerConfig := map[string]interface{}{
|
||||
"Name": record.Name,
|
||||
"Active": record.Active,
|
||||
"APIKey": record.APIKey,
|
||||
"BaseURL": record.BaseURL,
|
||||
"Model": record.Model,
|
||||
"Timeout": record.Timeout,
|
||||
"ContextWindowTokens": record.ContextWindowTokens,
|
||||
}
|
||||
if err := aiService.AddLLMProvider(providerConfig); err != nil {
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"status": "ok",
|
||||
"item": llmProviderDTO(*record),
|
||||
"warning": "provider created but failed to reload AI service: " + err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, gin.H{"status": "ok", "item": llmProviderDTO(*record)})
|
||||
@@ -385,25 +397,30 @@ func handleUpdateLLMProvider(store *storepkg.Store, aiService LLMProviderReloade
|
||||
}
|
||||
|
||||
// Reload AI service with updated provider
|
||||
if aiService != nil {
|
||||
providerConfig := map[string]interface{}{
|
||||
"Name": record.Name,
|
||||
"Active": record.Active,
|
||||
"APIKey": record.APIKey,
|
||||
"BaseURL": record.BaseURL,
|
||||
"Model": record.Model,
|
||||
"Timeout": record.Timeout,
|
||||
"ContextWindowTokens": record.ContextWindowTokens,
|
||||
}
|
||||
if err := aiService.ReloadLLMProvider(providerConfig); err != nil {
|
||||
// Log warning but don't fail the request - database is already updated
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"status": "ok",
|
||||
"item": llmProviderDTO(*record),
|
||||
"warning": "provider updated but failed to reload AI service: " + err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
if aiService == nil {
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"status": "ok",
|
||||
"item": llmProviderDTO(*record),
|
||||
"warning": "AI 服务未运行,配置已保存但需重启服务后生效",
|
||||
})
|
||||
return
|
||||
}
|
||||
providerConfig := map[string]interface{}{
|
||||
"Name": record.Name,
|
||||
"Active": record.Active,
|
||||
"APIKey": record.APIKey,
|
||||
"BaseURL": record.BaseURL,
|
||||
"Model": record.Model,
|
||||
"Timeout": record.Timeout,
|
||||
"ContextWindowTokens": record.ContextWindowTokens,
|
||||
}
|
||||
if err := aiService.ReloadLLMProvider(providerConfig); err != nil {
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"status": "ok",
|
||||
"item": llmProviderDTO(*record),
|
||||
"warning": "provider updated but failed to reload AI service: " + err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, gin.H{"status": "ok", "item": llmProviderDTO(*record)})
|
||||
@@ -424,15 +441,19 @@ func handleDeleteLLMProvider(store *storepkg.Store, aiService LLMProviderReloade
|
||||
}
|
||||
|
||||
// Remove provider from AI service
|
||||
if aiService != nil {
|
||||
if err := aiService.RemoveLLMProvider(name); err != nil {
|
||||
// Log warning but don't fail the request - database is already updated
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"status": "ok",
|
||||
"warning": "provider deleted but failed to reload AI service: " + err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
if aiService == nil {
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"status": "ok",
|
||||
"warning": "AI 服务未运行,配置已删除但需重启服务后生效",
|
||||
})
|
||||
return
|
||||
}
|
||||
if err := aiService.RemoveLLMProvider(name); err != nil {
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"status": "ok",
|
||||
"warning": "provider deleted but failed to reload AI service: " + err.Error(),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
c.JSON(http.StatusOK, gin.H{"status": "ok"})
|
||||
@@ -814,3 +835,49 @@ func llmPrimaryConfigDTO(row storepkg.LLMPrimaryConfigRecord) map[string]any {
|
||||
"updated_at": row.UpdatedAt,
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================
|
||||
// AI Service Status & Restart Handlers
|
||||
// ============================================
|
||||
|
||||
func handleGetAIServiceStatus(aiService LLMProviderReloader) gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
if aiService == nil {
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"running": false,
|
||||
"enabled": false,
|
||||
"provider_count": 0,
|
||||
"message": "AI 服务未初始化",
|
||||
})
|
||||
return
|
||||
}
|
||||
status := aiService.AIServiceStatus()
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"running": status.Running,
|
||||
"enabled": status.Enabled,
|
||||
"provider_count": status.ProviderCount,
|
||||
"message": status.Message,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func handleRestartAIService(aiService LLMProviderReloader) gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
if aiService == nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "AI 服务未初始化,无法重启"})
|
||||
return
|
||||
}
|
||||
if err := aiService.RestartAIService(); err != nil {
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "重启 AI 服务失败: " + err.Error()})
|
||||
return
|
||||
}
|
||||
status := aiService.AIServiceStatus()
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"status": "ok",
|
||||
"running": status.Running,
|
||||
"enabled": status.Enabled,
|
||||
"provider_count": status.ProviderCount,
|
||||
"message": "AI 服务已重启",
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -20,6 +20,7 @@ type PacketBuildOptions struct {
|
||||
PSK []byte
|
||||
Encrypt bool
|
||||
ViaMQTT bool
|
||||
HopLimit uint32
|
||||
}
|
||||
|
||||
type TextMessageBuildOptions struct {
|
||||
@@ -252,6 +253,14 @@ func buildMeshPacket(opts PacketBuildOptions, data []byte) ([]byte, error) {
|
||||
out = protowire.AppendTag(out, 14, protowire.VarintType)
|
||||
out = protowire.AppendVarint(out, 1)
|
||||
}
|
||||
hop := opts.HopLimit
|
||||
if hop == 0 {
|
||||
hop = 7
|
||||
}
|
||||
out = protowire.AppendTag(out, 9, protowire.VarintType)
|
||||
out = protowire.AppendVarint(out, uint64(hop))
|
||||
out = protowire.AppendTag(out, 15, protowire.VarintType)
|
||||
out = protowire.AppendVarint(out, uint64(hop))
|
||||
return out, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
package mqttforward
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
const dedupTTL = 15 * time.Second
|
||||
|
||||
type DedupQueue struct {
|
||||
mu sync.Mutex
|
||||
entries map[string]time.Time
|
||||
stopCh chan struct{}
|
||||
}
|
||||
|
||||
func NewDedupQueue() *DedupQueue {
|
||||
return &DedupQueue{
|
||||
entries: make(map[string]time.Time),
|
||||
stopCh: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
func (dq *DedupQueue) TryForward(topic string, payload []byte) bool {
|
||||
hash := dedupHash(topic, payload)
|
||||
now := time.Now()
|
||||
dq.mu.Lock()
|
||||
defer dq.mu.Unlock()
|
||||
if expiry, ok := dq.entries[hash]; ok && now.Before(expiry) {
|
||||
return false
|
||||
}
|
||||
dq.entries[hash] = now.Add(dedupTTL)
|
||||
return true
|
||||
}
|
||||
|
||||
func (dq *DedupQueue) Start() {
|
||||
go func() {
|
||||
ticker := time.NewTicker(dedupTTL)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
dq.cleanup()
|
||||
case <-dq.stopCh:
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (dq *DedupQueue) Stop() {
|
||||
close(dq.stopCh)
|
||||
}
|
||||
|
||||
func (dq *DedupQueue) Len() int {
|
||||
dq.mu.Lock()
|
||||
defer dq.mu.Unlock()
|
||||
return len(dq.entries)
|
||||
}
|
||||
|
||||
func (dq *DedupQueue) cleanup() {
|
||||
now := time.Now()
|
||||
dq.mu.Lock()
|
||||
defer dq.mu.Unlock()
|
||||
for hash, expiry := range dq.entries {
|
||||
if now.After(expiry) {
|
||||
delete(dq.entries, hash)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func dedupHash(topic string, payload []byte) string {
|
||||
h := sha256.New()
|
||||
h.Write([]byte(topic))
|
||||
h.Write(payload)
|
||||
return hex.EncodeToString(h.Sum(nil))
|
||||
}
|
||||
@@ -290,27 +290,24 @@ func insertInboundBotDirectMessage(s *Store, record map[string]any, clientInfo M
|
||||
}
|
||||
_ = clientInfo // mqtt 元数据已经记录在 content_json 里,这里保留参数以保持队列签名一致
|
||||
|
||||
// 同时将消息添加到 LLM 队列(忽略机器人自己发送的消息)
|
||||
if peerNodeID != bot.NodeID {
|
||||
longName := NullableString(record["long_name"])
|
||||
shortName := NullableString(record["short_name"])
|
||||
channelID := NullableString(record["channel_id"])
|
||||
_, err = s.EnqueueLLMMessage(LLMMessageQueueInput{
|
||||
BotID: bot.ID,
|
||||
BotNodeID: bot.NodeID,
|
||||
BotNodeNum: bot.NodeNum,
|
||||
FromNodeID: peerNodeID,
|
||||
FromNodeNum: int64(peerNum),
|
||||
LongName: longName,
|
||||
ShortName: shortName,
|
||||
Text: text,
|
||||
PacketID: int64(packetID),
|
||||
ChannelID: channelID,
|
||||
Topic: topic,
|
||||
MessageType: "direct",
|
||||
ContentJSON: contentPtr,
|
||||
})
|
||||
}
|
||||
longName := NullableString(record["long_name"])
|
||||
shortName := NullableString(record["short_name"])
|
||||
channelID := NullableString(record["channel_id"])
|
||||
_, err = s.EnqueueLLMMessage(LLMMessageQueueInput{
|
||||
BotID: bot.ID,
|
||||
BotNodeID: bot.NodeID,
|
||||
BotNodeNum: bot.NodeNum,
|
||||
FromNodeID: peerNodeID,
|
||||
FromNodeNum: int64(peerNum),
|
||||
LongName: longName,
|
||||
ShortName: shortName,
|
||||
Text: text,
|
||||
PacketID: int64(packetID),
|
||||
ChannelID: channelID,
|
||||
Topic: topic,
|
||||
MessageType: "direct",
|
||||
ContentJSON: contentPtr,
|
||||
})
|
||||
if err != nil {
|
||||
printJSON(map[string]any{
|
||||
"event": "llm_queue_enqueue_failed",
|
||||
|
||||
@@ -66,6 +66,12 @@ func (s *Store) CountBotNodes(opts ListOptions) (int64, error) {
|
||||
return total, s.db.Model(&BotNodeRecord{}).Count(&total).Error
|
||||
}
|
||||
|
||||
func (s *Store) IsBotNodeID(nodeID string) bool {
|
||||
var count int64
|
||||
s.db.Model(&BotNodeRecord{}).Where("node_id = ?", nodeID).Count(&count)
|
||||
return count > 0
|
||||
}
|
||||
|
||||
func (s *Store) GetBotNode(id uint64) (*BotNodeRecord, error) {
|
||||
var row BotNodeRecord
|
||||
if err := s.db.Where("id = ?", id).Take(&row).Error; err != nil {
|
||||
|
||||
@@ -306,8 +306,7 @@ func (s *Store) EnqueueLLMMessage(input LLMMessageQueueInput) (*LLMMessageQueueR
|
||||
return nil, nil // 机器人的 LLM 队列未启用,静默返回
|
||||
}
|
||||
|
||||
// 忽略机器人自己发送的消息,避免自循环
|
||||
if input.FromNodeID == bot.NodeID {
|
||||
if s.IsBotNodeID(input.FromNodeID) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
@@ -526,6 +525,10 @@ func enqueueChannelMessageToLLM(s *Store, record map[string]any) error {
|
||||
shortName = &sn
|
||||
}
|
||||
|
||||
if s.IsBotNodeID(fromNodeID) {
|
||||
return nil
|
||||
}
|
||||
|
||||
var channelID *string
|
||||
if cid, ok := record["channel_id"].(string); ok && cid != "" {
|
||||
channelID = &cid
|
||||
@@ -546,11 +549,7 @@ func enqueueChannelMessageToLLM(s *Store, record map[string]any) error {
|
||||
return fmt.Errorf("query bots for channel message enqueue: %w", err)
|
||||
}
|
||||
|
||||
// 为每个符合条件的机器人创建一条队列记录(忽略机器人自己发送的消息)
|
||||
for _, bot := range bots {
|
||||
if fromNodeID == bot.NodeID {
|
||||
continue
|
||||
}
|
||||
_, err = s.EnqueueLLMMessage(LLMMessageQueueInput{
|
||||
BotID: bot.ID,
|
||||
BotNodeID: bot.NodeID,
|
||||
|
||||
@@ -88,6 +88,7 @@ func RunAgentToolLoop(ctx context.Context, state *State, profile *llm.Profile, s
|
||||
if routerPrompt == "" {
|
||||
routerPrompt = primaryPrompt
|
||||
}
|
||||
routerPrompt = routerPrompt + "\n当前日期:" + time.Now().Format("2006-01-02")
|
||||
if primaryPrompt != "" {
|
||||
primarySystemMessage := &model.ChatCompletionMessage{
|
||||
Role: "system",
|
||||
|
||||
@@ -48,7 +48,7 @@ func NewState(cfg *Config, ai *llm.State, options ...Option) (*State, error) {
|
||||
Enabled: true,
|
||||
Timeout: 30,
|
||||
MaxTokens: 512,
|
||||
SystemPrompt: "你可以按需直接调用可用工具来回答用户问题。\n每个工具的 description 描述了它的适用场景和调用条件。\n工具结果优先于模型内置知识;工具失败时必须如实说明,不要编造结果。\n只调用确实必要的工具。",
|
||||
SystemPrompt: "你可以按需直接调用可用工具来回答用户问题。\n每个工具的 description 描述了它的适用场景和调用条件。\n工具结果优先于模型内置知识;工具失败时必须如实说明,不要编造结果。\n只调用确实必要的工具。\n\n重要规则:\n- 当用户问题包含\"今天\"、\"昨天\"、\"最近\"、\"本周\"、\"本月\"等相对时间词时,你必须先调用 time 工具获取当前准确日期,然后再用该日期调用其他工具。绝不可以使用模型内置知识猜测日期。\n- 你不知道\"今天\"是哪一天,必须通过 time 工具查询。",
|
||||
}
|
||||
}
|
||||
if ai == nil {
|
||||
|
||||
@@ -30,6 +30,7 @@ type MQTTRuntimeStatus struct {
|
||||
Stats *mqttforwardpkg.Stats
|
||||
ClientStats *mqttforwardpkg.ClientStats
|
||||
DBQueue *storepkg.WriteQueue
|
||||
DedupQueue *mqttforwardpkg.DedupQueue
|
||||
}
|
||||
|
||||
// AdminMQTTStatus 是 admin 路由 GET /admin/mqtt-status 返回的 JSON 视图。
|
||||
@@ -50,6 +51,7 @@ type AdminMQTTStatus struct {
|
||||
MessagesSent int64 `json:"messages_sent"`
|
||||
MessagesDropped int64 `json:"messages_dropped"`
|
||||
DBWriteQueueLength int `json:"db_write_queue_length"`
|
||||
DedupQueueLength int `json:"dedup_queue_len"`
|
||||
Retained int64 `json:"retained"`
|
||||
Inflight int64 `json:"inflight"`
|
||||
InflightDropped int64 `json:"inflight_dropped"`
|
||||
@@ -71,7 +73,7 @@ type AdminMQTTClient struct {
|
||||
// Status 实现 MQTTStatusProvider。
|
||||
func (m MQTTRuntimeStatus) Status() AdminMQTTStatus {
|
||||
if m.Server == nil || m.Server.Info == nil {
|
||||
return AdminMQTTStatus{Running: false, Address: m.Address, TLS: m.TLS, DBWriteQueueLength: m.DBQueue.Len()}
|
||||
return AdminMQTTStatus{Running: false, Address: m.Address, TLS: m.TLS, DBWriteQueueLength: m.DBQueue.Len(), DedupQueueLength: m.dedupQueueLen()}
|
||||
}
|
||||
info := m.Server.Info.Clone()
|
||||
status := AdminMQTTStatus{
|
||||
@@ -91,6 +93,7 @@ func (m MQTTRuntimeStatus) Status() AdminMQTTStatus {
|
||||
MessagesSent: m.Stats.Forwarded(),
|
||||
MessagesDropped: m.Stats.Dropped(),
|
||||
DBWriteQueueLength: m.DBQueue.Len(),
|
||||
DedupQueueLength: m.dedupQueueLen(),
|
||||
Retained: info.Retained,
|
||||
Inflight: info.Inflight,
|
||||
InflightDropped: info.InflightDropped,
|
||||
@@ -136,6 +139,13 @@ func mqttClientInfo(c *mqtt.Client) mqttClientInfoView {
|
||||
}
|
||||
}
|
||||
|
||||
func (m MQTTRuntimeStatus) dedupQueueLen() int {
|
||||
if m.DedupQueue == nil {
|
||||
return 0
|
||||
}
|
||||
return m.DedupQueue.Len()
|
||||
}
|
||||
|
||||
// DisconnectClient 实现 MQTTStatusProvider:发送 Disconnect 报文并关闭连接。
|
||||
// 使用 ErrAdministrativeAction 作为断开理由,便于日志区分。
|
||||
func (m MQTTRuntimeStatus) DisconnectClient(clientID string) bool {
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"github.com/gin-gonic/gin"
|
||||
"gorm.io/gorm"
|
||||
|
||||
aipkg "meshtastic_mqtt_server/internal/ai"
|
||||
"meshtastic_mqtt_server/internal/auth"
|
||||
blockingpkg "meshtastic_mqtt_server/internal/blocking"
|
||||
botpkg "meshtastic_mqtt_server/internal/bot"
|
||||
@@ -32,6 +33,8 @@ type LLMProviderReloader interface {
|
||||
ReloadLLMProvider(config interface{}) error
|
||||
AddLLMProvider(config interface{}) error
|
||||
RemoveLLMProvider(name string) error
|
||||
AIServiceStatus() aipkg.AIServiceStatus
|
||||
RestartAIService() error
|
||||
}
|
||||
|
||||
func NewHTTPServer(cfg configpkg.WebConfig, consoleLog bool, store *storepkg.Store, sessions *auth.Manager, mqttStatus MQTTStatusProvider, blocking *blockingpkg.Cache, forwarder mqttforwardpkg.Reloader, settings *rspkg.Cache, botSender botpkg.TextSender, aiService LLMProviderReloader) *http.Server {
|
||||
|
||||
@@ -57,8 +57,9 @@ type meshtasticFilterHook struct {
|
||||
settings *rspkg.Cache
|
||||
pkiResolver func(toNodeNum, fromNodeNum uint32) ([]byte, []byte, bool)
|
||||
autoAcker func(record map[string]any)
|
||||
consoleLog bool // 控制台是否打印 MQTT 连接/订阅事件
|
||||
consoleLog bool // 控制台是否打印 MQTT 连接/订阅事件
|
||||
packetConsoleLog bool // 控制台是否打印 Meshtastic 数据包
|
||||
dedupQueue *mqttforwardpkg.DedupQueue
|
||||
}
|
||||
|
||||
// ID 返回用于识别 Meshtastic payload 过滤器的 hook 名称。
|
||||
@@ -203,6 +204,12 @@ func (h *meshtasticFilterHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (pa
|
||||
PKIKeyResolver: h.pkiResolver,
|
||||
})
|
||||
if !valid {
|
||||
// 记录拒绝原因,帮助诊断 QoS0 重发问题
|
||||
if h.consoleLog {
|
||||
info := mqttClientInfoFromClient(cl)
|
||||
fmt.Fprintf(os.Stderr, "[mqtt] PUBLISH rejected: client_id=%s topic=%s qos=%d payload_len=%d error=%v\n",
|
||||
info.ClientID, pk.TopicName, pk.FixedHeader.Qos, len(pk.Payload), record["error"])
|
||||
}
|
||||
h.rejectPublish(cl, pk, record)
|
||||
return pk, packets.ErrRejectPacket
|
||||
}
|
||||
@@ -210,9 +217,19 @@ func (h *meshtasticFilterHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (pa
|
||||
for key, value := range violation {
|
||||
record[key] = value
|
||||
}
|
||||
// 记录屏蔽原因
|
||||
if h.consoleLog {
|
||||
info := mqttClientInfoFromClient(cl)
|
||||
fmt.Fprintf(os.Stderr, "[mqtt] PUBLISH blocked: client_id=%s topic=%s type=%v reason=%v\n",
|
||||
info.ClientID, pk.TopicName, violation["blocking_type"], violation["error"])
|
||||
}
|
||||
h.rejectPublish(cl, pk, record)
|
||||
return pk, packets.ErrRejectPacket
|
||||
}
|
||||
if h.dedupQueue != nil && !h.dedupQueue.TryForward(pk.TopicName, pk.Payload) {
|
||||
h.stats.IncDropped()
|
||||
return pk, packets.ErrRejectPacket
|
||||
}
|
||||
h.stats.IncForwarded()
|
||||
|
||||
h.dbQueue.EnqueueRecord(record, mqttClientInfoFromClient(cl))
|
||||
@@ -381,16 +398,48 @@ func run(cfg *configpkg.Config) error {
|
||||
defer forwardManager.StopAll()
|
||||
|
||||
// Initialize AI Service
|
||||
var aiService *ai.Service
|
||||
// Create bot sender adapter - 支持频道消息和私聊消息两种发送方式
|
||||
botSenderAdapter := autoreply.NewBotServiceAdapter(
|
||||
// SendDirectText: 发送私聊消息
|
||||
func(ctx context.Context, botID uint64, toNodeNum int64, text string) error {
|
||||
_, err := botSender.SendText(ctx, botpkg.SendTextRequest{
|
||||
BotID: botID,
|
||||
MessageType: "direct",
|
||||
ToNodeNum: &toNodeNum,
|
||||
Text: text,
|
||||
})
|
||||
return err
|
||||
},
|
||||
// SendChannelText: 发送频道消息
|
||||
func(ctx context.Context, botID uint64, channelID string, text string) error {
|
||||
_, err := botSender.SendText(ctx, botpkg.SendTextRequest{
|
||||
BotID: botID,
|
||||
MessageType: "channel",
|
||||
ChannelID: channelID,
|
||||
Text: text,
|
||||
})
|
||||
return err
|
||||
},
|
||||
)
|
||||
|
||||
aiManager := ai.NewAIManager(ai.Config{
|
||||
DataDir: cfg.AI.DataDir,
|
||||
Enabled: cfg.AI.Enabled,
|
||||
ConsoleLog: cfg.ConsoleLog.LLM,
|
||||
ToolConfigStore: store,
|
||||
ToolRouterStore: store,
|
||||
TopicRouterStore: store,
|
||||
Store: store,
|
||||
}, store.DB(), botSenderAdapter, botCtx, store)
|
||||
|
||||
if cfg.AI.Enabled {
|
||||
// Get LLM providers from database
|
||||
llmProviders, err := store.ListLLMProviders(true)
|
||||
aiManager.SetConfigEnabled(true)
|
||||
providers, err := store.ListLLMProviders(true)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Warning: failed to load LLM providers: %v\n", err)
|
||||
} else if len(llmProviders) > 0 {
|
||||
// Convert database records to provider configs
|
||||
providerConfigs := make([]llm.ProviderConfig, 0, len(llmProviders))
|
||||
for _, p := range llmProviders {
|
||||
} else if len(providers) > 0 {
|
||||
providerConfigs := make([]llm.ProviderConfig, 0, len(providers))
|
||||
for _, p := range providers {
|
||||
providerConfigs = append(providerConfigs, llm.ProviderConfig{
|
||||
Name: p.Name,
|
||||
Active: p.Active,
|
||||
@@ -401,48 +450,11 @@ func run(cfg *configpkg.Config) error {
|
||||
ContextWindowTokens: p.ContextWindowTokens,
|
||||
})
|
||||
}
|
||||
|
||||
// Create bot sender adapter - 支持频道消息和私聊消息两种发送方式
|
||||
botSenderAdapter := autoreply.NewBotServiceAdapter(
|
||||
// SendDirectText: 发送私聊消息
|
||||
func(ctx context.Context, botID uint64, toNodeNum int64, text string) error {
|
||||
_, err := botSender.SendText(ctx, botpkg.SendTextRequest{
|
||||
BotID: botID,
|
||||
MessageType: "direct",
|
||||
ToNodeNum: &toNodeNum,
|
||||
Text: text,
|
||||
})
|
||||
return err
|
||||
},
|
||||
// SendChannelText: 发送频道消息
|
||||
func(ctx context.Context, botID uint64, channelID string, text string) error {
|
||||
_, err := botSender.SendText(ctx, botpkg.SendTextRequest{
|
||||
BotID: botID,
|
||||
MessageType: "channel",
|
||||
ChannelID: channelID,
|
||||
Text: text,
|
||||
})
|
||||
return err
|
||||
},
|
||||
)
|
||||
|
||||
aiService, err = ai.NewService(ai.Config{
|
||||
LLMProviders: providerConfigs,
|
||||
DataDir: cfg.AI.DataDir,
|
||||
Enabled: cfg.AI.Enabled,
|
||||
ConsoleLog: cfg.ConsoleLog.LLM,
|
||||
ToolConfigStore: store,
|
||||
ToolRouterStore: store,
|
||||
TopicRouterStore: store,
|
||||
Store: store,
|
||||
}, store.DB(), botSenderAdapter)
|
||||
if err != nil {
|
||||
aiManager.SetProviderConfigs(providerConfigs)
|
||||
if err := aiManager.Init(); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Warning: failed to initialize AI service: %v\n", err)
|
||||
} else {
|
||||
if err := aiService.Start(botCtx); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "Warning: failed to start AI service: %v\n", err)
|
||||
}
|
||||
defer aiService.Stop()
|
||||
defer aiManager.Stop()
|
||||
printJSON(map[string]any{"event": "ai_service_started", "providers": len(providerConfigs)})
|
||||
}
|
||||
} else {
|
||||
@@ -457,8 +469,8 @@ func run(cfg *configpkg.Config) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
mqttStatus := webpkg.MQTTRuntimeStatus{Server: server, Address: mqttAddr, TLS: cfg.MQTT.TLS.Enabled, Stats: messageStats, ClientStats: clientStats, DBQueue: dbQueue}
|
||||
handler := webpkg.NewRouter(cfg.Web, cfg.ConsoleLog.Web, store, sessions, mqttStatus, blocking, forwardManager, settings, botSender, aiService)
|
||||
mqttStatus := webpkg.MQTTRuntimeStatus{Server: server, Address: mqttAddr, TLS: cfg.MQTT.TLS.Enabled, Stats: messageStats, ClientStats: clientStats, DBQueue: dbQueue, DedupQueue: mqttHook.dedupQueue}
|
||||
handler := webpkg.NewRouter(cfg.Web, cfg.ConsoleLog.Web, store, sessions, mqttStatus, blocking, forwardManager, settings, botSender, aiManager)
|
||||
webAddresses := []string{}
|
||||
if cfg.Web.PortEnabled {
|
||||
httpServer := &http.Server{
|
||||
@@ -509,6 +521,9 @@ func run(cfg *configpkg.Config) error {
|
||||
if err := server.Close(); err != nil && runErr == nil {
|
||||
runErr = err
|
||||
}
|
||||
if mqttHook.dedupQueue != nil {
|
||||
mqttHook.dedupQueue.Stop()
|
||||
}
|
||||
return runErr
|
||||
}
|
||||
|
||||
@@ -517,6 +532,8 @@ func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *stor
|
||||
if err := server.AddHook(new(mqttauth.AllowHook), nil); err != nil {
|
||||
return nil, nil, "", err
|
||||
}
|
||||
dedupQueue := mqttforwardpkg.NewDedupQueue()
|
||||
dedupQueue.Start()
|
||||
hook := &meshtasticFilterHook{
|
||||
server: server,
|
||||
key: cfg.Key,
|
||||
@@ -528,6 +545,7 @@ func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *stor
|
||||
pkiResolver: botpkg.NewPKIKeyResolver(store),
|
||||
consoleLog: cfg.ConsoleLog.MQTT,
|
||||
packetConsoleLog: cfg.ConsoleLog.Meshtastic,
|
||||
dedupQueue: dedupQueue,
|
||||
}
|
||||
if err := server.AddHook(hook, nil); err != nil {
|
||||
return nil, nil, "", err
|
||||
|
||||
@@ -6,6 +6,7 @@ import type {
|
||||
AdminRuntimeSettingsPayload,
|
||||
AdminRuntimeSettingsResponse,
|
||||
AdminUsersResponse,
|
||||
AIServiceStatus,
|
||||
BotMessage,
|
||||
BotMessageMutationResponse,
|
||||
BotNode,
|
||||
@@ -515,8 +516,17 @@ export function updateLLMProvider(name: string, payload: Partial<LLMProviderPayl
|
||||
return putJSON<LLMProviderResponse>(`/api/admin/llm/providers/${encodeURIComponent(name)}`, payload)
|
||||
}
|
||||
|
||||
export function deleteLLMProvider(name: string): Promise<{ status: string }> {
|
||||
return deleteJSON<{ status: string }>(`/api/admin/llm/providers/${encodeURIComponent(name)}`)
|
||||
export function deleteLLMProvider(name: string): Promise<{ status: string; warning?: string }> {
|
||||
return deleteJSON<{ status: string; warning?: string }>(`/api/admin/llm/providers/${encodeURIComponent(name)}`)
|
||||
}
|
||||
|
||||
// AI Service Status API
|
||||
export function getAIServiceStatus(): Promise<AIServiceStatus> {
|
||||
return getJSON<AIServiceStatus>('/api/admin/llm/status')
|
||||
}
|
||||
|
||||
export function restartAIService(): Promise<AIServiceStatus & { status: string; message?: string }> {
|
||||
return postJSON<AIServiceStatus & { status: string; message?: string }>('/api/admin/llm/restart')
|
||||
}
|
||||
|
||||
// LLM Tool Router API
|
||||
|
||||
@@ -200,6 +200,7 @@ onBeforeUnmount(() => {
|
||||
<div><span>订阅数</span><strong>{{ status.subscriptions }}</strong></div>
|
||||
<div><span>转发消息</span><strong>{{ status.messages_sent }}</strong></div>
|
||||
<div><span>数据库队列</span><strong>{{ status.db_write_queue_length }}</strong></div>
|
||||
<div><span>去重队列</span><strong>{{ status.dedup_queue_len }}</strong></div>
|
||||
<a class="status-card-link" href="/admin/discard_details"><span>丢弃消息</span><strong>{{ status.messages_dropped }}</strong></a>
|
||||
<div><span>收到包</span><strong>{{ status.packets_received }}</strong></div>
|
||||
<div><span>发送包</span><strong>{{ status.packets_sent }}</strong></div>
|
||||
|
||||
@@ -11,12 +11,19 @@ import {
|
||||
updateLLMToolRouter,
|
||||
updateLLMTopicConfig,
|
||||
updateLLMPrimaryConfig,
|
||||
getAIServiceStatus,
|
||||
restartAIService,
|
||||
} from '../api'
|
||||
import type { LLMPlatformRouter, LLMTopicConfig, LLMProvider, LLMPrimaryConfig } from '../types'
|
||||
import type { LLMPlatformRouter, LLMTopicConfig, LLMProvider, LLMPrimaryConfig, AIServiceStatus } from '../types'
|
||||
|
||||
const loading = ref(false)
|
||||
const error = ref('')
|
||||
const success = ref('')
|
||||
const showWarning = ref(false)
|
||||
|
||||
// AI Service Status
|
||||
const aiStatus = ref<AIServiceStatus>({ running: false, enabled: false, provider_count: 0 })
|
||||
const restarting = ref(false)
|
||||
|
||||
// LLM Provider 相关
|
||||
const providers = ref<LLMProvider[]>([])
|
||||
@@ -169,12 +176,18 @@ async function saveProvider() {
|
||||
}
|
||||
|
||||
try {
|
||||
let response: any
|
||||
if (isCreatingProvider.value) {
|
||||
await createLLMProvider(providerForm.value)
|
||||
success.value = '创建成功'
|
||||
response = await createLLMProvider(providerForm.value)
|
||||
} else if (editingProvider.value) {
|
||||
await updateLLMProvider(editingProvider.value.name, providerForm.value)
|
||||
success.value = '更新成功'
|
||||
response = await updateLLMProvider(editingProvider.value.name, providerForm.value)
|
||||
}
|
||||
if (response.warning) {
|
||||
success.value = response.warning
|
||||
showWarning.value = true
|
||||
} else {
|
||||
success.value = isCreatingProvider.value ? '创建成功' : '更新成功'
|
||||
showWarning.value = false
|
||||
}
|
||||
clearSuccess()
|
||||
closeProviderForm()
|
||||
@@ -189,8 +202,14 @@ async function confirmDeleteProvider(name: string) {
|
||||
return
|
||||
}
|
||||
try {
|
||||
await deleteLLMProvider(name)
|
||||
success.value = '删除成功'
|
||||
const response = await deleteLLMProvider(name)
|
||||
if (response.warning) {
|
||||
success.value = response.warning
|
||||
showWarning.value = true
|
||||
} else {
|
||||
success.value = '删除成功'
|
||||
showWarning.value = false
|
||||
}
|
||||
clearSuccess()
|
||||
await loadProviders()
|
||||
} catch (err) {
|
||||
@@ -286,7 +305,31 @@ async function savePrimaryConfig() {
|
||||
}
|
||||
}
|
||||
|
||||
async function loadAIStatus() {
|
||||
try {
|
||||
aiStatus.value = await getAIServiceStatus()
|
||||
} catch (err) {
|
||||
console.warn('Failed to load AI service status', err)
|
||||
}
|
||||
}
|
||||
|
||||
async function handleRestartAI() {
|
||||
if (!confirm('确定要重启 AI 服务吗?')) return
|
||||
restarting.value = true
|
||||
try {
|
||||
const result = await restartAIService()
|
||||
aiStatus.value = result
|
||||
success.value = result.message || 'AI 服务已重启'
|
||||
clearSuccess()
|
||||
} catch (err) {
|
||||
error.value = err instanceof Error ? err.message : String(err)
|
||||
} finally {
|
||||
restarting.value = false
|
||||
}
|
||||
}
|
||||
|
||||
onMounted(() => {
|
||||
loadAIStatus()
|
||||
loadProviders()
|
||||
loadToolRouter()
|
||||
loadTopicConfig()
|
||||
@@ -298,8 +341,23 @@ onMounted(() => {
|
||||
<div class="admin-llm-api">
|
||||
<h2>LLM API 配置管理</h2>
|
||||
|
||||
<div class="ai-status-bar">
|
||||
<span class="status-indicator" :class="{ running: aiStatus.running, stopped: !aiStatus.running }"></span>
|
||||
<span class="status-text">
|
||||
<template v-if="aiStatus.running">AI 服务运行中,{{ aiStatus.provider_count }} 个提供商</template>
|
||||
<template v-else>{{ aiStatus.message || 'AI 服务未运行' }}</template>
|
||||
</span>
|
||||
<button
|
||||
class="admin-button admin-button-small"
|
||||
:disabled="restarting"
|
||||
@click="handleRestartAI"
|
||||
>
|
||||
{{ restarting ? '重启中...' : '重启 AI 服务' }}
|
||||
</button>
|
||||
</div>
|
||||
|
||||
<p v-if="error" class="error">{{ error }}</p>
|
||||
<p v-if="success" class="success">{{ success }}</p>
|
||||
<p v-if="success" :class="showWarning ? 'warning' : 'success'">{{ success }}</p>
|
||||
|
||||
<!-- LLM Provider 列表 -->
|
||||
<div class="admin-section">
|
||||
@@ -690,13 +748,49 @@ onMounted(() => {
|
||||
}
|
||||
|
||||
.admin-llm-api h2 {
|
||||
margin: 0 0 2rem;
|
||||
margin: 0 0 1rem;
|
||||
font-size: 1.75rem;
|
||||
font-weight: 700;
|
||||
color: #1e293b;
|
||||
letter-spacing: -0.02em;
|
||||
}
|
||||
|
||||
.ai-status-bar {
|
||||
display: flex;
|
||||
align-items: center;
|
||||
gap: 0.75rem;
|
||||
padding: 0.75rem 1.25rem;
|
||||
background: white;
|
||||
border-radius: 12px;
|
||||
border: 1px solid #e2e8f0;
|
||||
margin-bottom: 1.5rem;
|
||||
box-shadow: 0 1px 3px rgba(0, 0, 0, 0.05);
|
||||
}
|
||||
|
||||
.status-indicator {
|
||||
width: 10px;
|
||||
height: 10px;
|
||||
border-radius: 50%;
|
||||
flex-shrink: 0;
|
||||
}
|
||||
|
||||
.status-indicator.running {
|
||||
background: #22c55e;
|
||||
box-shadow: 0 0 6px rgba(34, 197, 94, 0.5);
|
||||
}
|
||||
|
||||
.status-indicator.stopped {
|
||||
background: #ef4444;
|
||||
box-shadow: 0 0 6px rgba(239, 68, 68, 0.4);
|
||||
}
|
||||
|
||||
.status-text {
|
||||
flex: 1;
|
||||
font-size: 0.9rem;
|
||||
color: #475569;
|
||||
font-weight: 500;
|
||||
}
|
||||
|
||||
.admin-section {
|
||||
background: white;
|
||||
padding: 1.75rem;
|
||||
@@ -973,6 +1067,19 @@ onMounted(() => {
|
||||
gap: 0.5rem;
|
||||
}
|
||||
|
||||
.warning {
|
||||
color: #92400e;
|
||||
padding: 1rem 1.25rem;
|
||||
background: linear-gradient(135deg, #fef3c7 0%, #fde68a 100%);
|
||||
border-radius: 10px;
|
||||
margin-bottom: 1.25rem;
|
||||
border: 1px solid #fcd34d;
|
||||
font-weight: 500;
|
||||
display: flex;
|
||||
align-items: center;
|
||||
gap: 0.5rem;
|
||||
}
|
||||
|
||||
.admin-loading {
|
||||
padding: 3rem;
|
||||
text-align: center;
|
||||
|
||||
@@ -364,6 +364,7 @@ export interface AdminMqttStatus {
|
||||
messages_sent: number
|
||||
messages_dropped: number
|
||||
db_write_queue_length: number
|
||||
dedup_queue_len: number
|
||||
retained: number
|
||||
inflight: number
|
||||
inflight_dropped: number
|
||||
@@ -643,6 +644,14 @@ export interface LLMProviderPayload {
|
||||
|
||||
export interface LLMProviderResponse {
|
||||
item: LLMProvider
|
||||
warning?: string
|
||||
}
|
||||
|
||||
export interface AIServiceStatus {
|
||||
running: boolean
|
||||
enabled: boolean
|
||||
provider_count: number
|
||||
message?: string
|
||||
}
|
||||
|
||||
// LLM Tool Router 相关类型
|
||||
|
||||
+2
-1
@@ -1 +1,2 @@
|
||||
__pycache__
|
||||
__pycache__
|
||||
db_config.py
|
||||
@@ -0,0 +1,266 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
MySQL 8 -> MySQL 5.7 数据库迁移脚本
|
||||
|
||||
依赖: pip install pymysql
|
||||
|
||||
使用方法:
|
||||
python py/migrate_mysql8_to_mysql57.py
|
||||
|
||||
功能:
|
||||
1. 从源 MySQL 8 读取表结构,修正 collation 后在目标 MySQL 5.7 建表
|
||||
2. 逐表批量拷贝数据
|
||||
3. 修正 auto-increment 值
|
||||
|
||||
注意:
|
||||
- 目标库应为空库(无同名表)
|
||||
- 运行前请确保源库和目标库都可连接
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import sys
|
||||
from typing import Any
|
||||
|
||||
import pymysql
|
||||
import pymysql.cursors
|
||||
|
||||
from db_config import SOURCE_CONFIG, TARGET_CONFIG
|
||||
|
||||
BATCH_SIZE = 5000
|
||||
|
||||
# ============================================================
|
||||
# 工具函数
|
||||
# ============================================================
|
||||
|
||||
|
||||
def _conn(config: dict[str, Any]) -> pymysql.Connection:
|
||||
"""创建数据库连接,使用 DictCursor 方便按列名访问"""
|
||||
return pymysql.connect(
|
||||
host=config["host"],
|
||||
port=config["port"],
|
||||
user=config["user"],
|
||||
password=config["password"],
|
||||
database=config["database"],
|
||||
charset=config["charset"],
|
||||
cursorclass=pymysql.cursors.DictCursor,
|
||||
)
|
||||
|
||||
|
||||
def fix_collation(ddl: str) -> str:
|
||||
"""将 MySQL 8 默认 collation 替换为 MySQL 5.7 兼容版本"""
|
||||
ddl = ddl.replace("utf8mb4_0900_ai_ci", "utf8mb4_general_ci")
|
||||
ddl = ddl.replace("utf8mb4_0900_as_ci", "utf8mb4_general_ci")
|
||||
return ddl
|
||||
|
||||
|
||||
def fix_text_blob_index(ddl: str) -> str:
|
||||
"""为 TEXT/BLOB 列在索引中添加前缀长度 (255),兼容 MySQL 5.7"""
|
||||
text_cols = set()
|
||||
for m in re.finditer(r'`(\w+)`\s+(?:tinytext|text|mediumtext|longtext|tinyblob|blob|mediumblob|longblob)\b', ddl, re.IGNORECASE):
|
||||
text_cols.add(m.group(1))
|
||||
if not text_cols:
|
||||
return ddl
|
||||
|
||||
def _add_prefix(match: re.Match) -> str:
|
||||
col = match.group(1)
|
||||
rest = match.group(2)
|
||||
if col in text_cols and not rest.strip().startswith('('):
|
||||
return f'`{col}`(255){rest}'
|
||||
return match.group(0)
|
||||
|
||||
# 匹配 KEY 定义中的列引用: `col_name` 后面不跟 ( 的情况
|
||||
ddl = re.sub(r'`(\w+)`(\s*[,\)])', _add_prefix, ddl)
|
||||
return ddl
|
||||
|
||||
|
||||
def get_all_tables(conn: pymysql.Connection) -> list[str]:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SHOW TABLES")
|
||||
key = list(cur.description[0])[0]
|
||||
return [row[key] for row in cur.fetchall()]
|
||||
|
||||
|
||||
def get_table_columns(conn: pymysql.Connection, table: str) -> list[str]:
|
||||
"""返回表的所有列名(按顺序)"""
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SHOW COLUMNS FROM `%s`" % table)
|
||||
return [row["Field"] for row in cur.fetchall()]
|
||||
|
||||
|
||||
def get_auto_increment(
|
||||
conn: pymysql.Connection, table: str
|
||||
) -> int | None:
|
||||
"""获取某张表当前的 AUTO_INCREMENT 值"""
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT AUTO_INCREMENT FROM information_schema.TABLES "
|
||||
"WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s",
|
||||
(table,),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
if row and row["AUTO_INCREMENT"]:
|
||||
return int(row["AUTO_INCREMENT"])
|
||||
return None
|
||||
|
||||
|
||||
def row_count(conn: pymysql.Connection, table: str) -> int:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SELECT COUNT(*) AS cnt FROM `%s`" % table)
|
||||
return cur.fetchone()["cnt"]
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 主流程
|
||||
# ============================================================
|
||||
|
||||
|
||||
def migrate() -> int:
|
||||
src = _conn(SOURCE_CONFIG)
|
||||
tgt = _conn(TARGET_CONFIG)
|
||||
print(f"[连接] 源 MySQL 8 @ {SOURCE_CONFIG['host']}:{SOURCE_CONFIG['port']}")
|
||||
print(f"[连接] 目标 MySQL 5.7 @ {TARGET_CONFIG['host']}:{TARGET_CONFIG['port']}")
|
||||
print()
|
||||
|
||||
# 1. 在目标库建库(如还不存在)
|
||||
try:
|
||||
with tgt.cursor() as cur:
|
||||
cur.execute(
|
||||
"CREATE DATABASE IF NOT EXISTS `%s` "
|
||||
"CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci"
|
||||
% TARGET_CONFIG["database"]
|
||||
)
|
||||
cur.execute("USE `%s`" % TARGET_CONFIG["database"])
|
||||
tgt.select_db(TARGET_CONFIG["database"])
|
||||
except Exception as e:
|
||||
print(f"[错误] 创建目标数据库失败: {e}")
|
||||
return 1
|
||||
|
||||
tables = get_all_tables(src)
|
||||
print(f"[发现] 源库共 {len(tables)} 张表: {', '.join(tables)}")
|
||||
print()
|
||||
|
||||
# 2. 逐表建表(修正 collation)
|
||||
print("=" * 60)
|
||||
print("阶段 1: 在目标库创建表结构")
|
||||
print("=" * 60)
|
||||
for idx, table in enumerate(tables, 1):
|
||||
with src.cursor() as cur:
|
||||
cur.execute("SHOW CREATE TABLE `%s`" % table)
|
||||
row = cur.fetchone()
|
||||
ddl = row["Create Table"]
|
||||
ddl = fix_collation(ddl)
|
||||
ddl = fix_text_blob_index(ddl)
|
||||
|
||||
try:
|
||||
with tgt.cursor() as cur:
|
||||
cur.execute(ddl)
|
||||
tgt.commit()
|
||||
print(f" [{idx:2d}/{len(tables)}] OK `{table}`")
|
||||
except Exception as e:
|
||||
print(f" [{idx:2d}/{len(tables)}] 错误 `{table}`: {e}")
|
||||
tgt.rollback()
|
||||
return 1
|
||||
|
||||
print()
|
||||
|
||||
# 3. 逐表拷贝数据
|
||||
print("=" * 60)
|
||||
print("阶段 2: 拷贝表数据")
|
||||
print("=" * 60)
|
||||
|
||||
total_rows_copied = 0
|
||||
auto_increments: dict[str, int | None] = {}
|
||||
|
||||
for idx, table in enumerate(tables, 1):
|
||||
columns = get_table_columns(src, table)
|
||||
if not columns:
|
||||
auto_increments[table] = None
|
||||
print(f" [{idx:2d}/{len(tables)}] SKIP `{table}` (0 列)")
|
||||
continue
|
||||
|
||||
col_quoted = ", ".join("`%s`" % c for c in columns)
|
||||
placeholders = ", ".join(["%s"] * len(columns))
|
||||
insert_sql = "INSERT INTO `%s` (%s) VALUES (%s)" % (
|
||||
table,
|
||||
col_quoted,
|
||||
placeholders,
|
||||
)
|
||||
|
||||
table_count = 0
|
||||
with src.cursor() as read_cur:
|
||||
read_cur.execute("SELECT * FROM `%s`" % table)
|
||||
batch = read_cur.fetchmany(BATCH_SIZE)
|
||||
while batch:
|
||||
rows_values = [
|
||||
[row.get(c) for c in columns] for row in batch
|
||||
]
|
||||
try:
|
||||
with tgt.cursor() as write_cur:
|
||||
write_cur.executemany(insert_sql, rows_values)
|
||||
tgt.commit()
|
||||
table_count += len(rows_values)
|
||||
print(
|
||||
f"\r [{idx:2d}/{len(tables)}] `{table}` -> {table_count} 行",
|
||||
end="",
|
||||
flush=True,
|
||||
)
|
||||
except Exception as e:
|
||||
print()
|
||||
print(f" [{idx:2d}/{len(tables)}] 错误 `{table}`: {e}")
|
||||
tgt.rollback()
|
||||
return 1
|
||||
batch = read_cur.fetchmany(BATCH_SIZE)
|
||||
|
||||
# 修正 auto_increment
|
||||
ai = get_auto_increment(src, table)
|
||||
auto_increments[table] = ai
|
||||
if ai is not None:
|
||||
with tgt.cursor() as cur:
|
||||
cur.execute(
|
||||
"ALTER TABLE `%s` AUTO_INCREMENT = %s" % (table, ai)
|
||||
)
|
||||
tgt.commit()
|
||||
|
||||
total_rows_copied += table_count
|
||||
print(
|
||||
f"\r [{idx:2d}/{len(tables)}] `{table}` -> {table_count} 行 [OK]"
|
||||
)
|
||||
|
||||
# 4. 验证
|
||||
print()
|
||||
print("=" * 60)
|
||||
print("阶段 3: 验证行数")
|
||||
print("=" * 60)
|
||||
all_match = True
|
||||
src_total = 0
|
||||
tgt_total = 0
|
||||
for table in tables:
|
||||
s = row_count(src, table)
|
||||
t = row_count(tgt, table)
|
||||
src_total += s
|
||||
tgt_total += t
|
||||
status = "OK" if s == t else "不匹配!"
|
||||
if s != t:
|
||||
all_match = False
|
||||
print(f" `{table}`: 源={s} 目标={t} [{status}]")
|
||||
|
||||
print()
|
||||
print(f" 总计: 源={src_total} 目标={tgt_total}")
|
||||
|
||||
src.close()
|
||||
tgt.close()
|
||||
|
||||
if all_match:
|
||||
print()
|
||||
print("迁移完成,所有表行数一致。")
|
||||
return 0
|
||||
else:
|
||||
print()
|
||||
print("警告: 部分表行数不一致,请检查。")
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(migrate())
|
||||
Reference in New Issue
Block a user