feat(server-go): 微信订阅消息骨架(#11,绑定/订阅/巡检触发,配置占位)

This commit is contained in:
weijuesen
2026-08-12 17:28:07 +08:00
parent 79e524b1d4
commit 921b185da2
9 changed files with 517 additions and 4 deletions
+3 -1
View File
@@ -73,6 +73,7 @@ func main() {
mediaSvc := service.NewMediaService(cfg)
transcodeSvc := service.NewTranscodeService(db, s3Svc)
aiSvc := service.NewAIClient(cfg.AIServiceBase)
wechatSvc := service.NewWechatService(cfg.WechatAppID, cfg.WechatSecret)
// 9. 创建 Gin 引擎
gin.SetMode(gin.ReleaseMode)
@@ -108,8 +109,9 @@ func main() {
handler.RegisterVideoRecordRoutes(api, db, mediaSvc, cfg)
handler.RegisterStorageRoutes(api, db)
handler.RegisterKnowledgeRoutes(api, db, s3Svc, cfg.S3BucketImages)
handler.RegisterInspectionRoutes(api, db, s3Svc, aiSvc, cfg.S3BucketImages)
handler.RegisterInspectionRoutes(api, db, s3Svc, aiSvc, cfg.S3BucketImages, wechatSvc, cfg.WechatTemplateInspection)
handler.RegisterTrayBatchRoutes(api, db)
handler.RegisterWechatRoutes(api, db, wechatSvc)
// 启动后台设备状态同步(每 30 秒查询 WVP 设备在线状态)
go startDeviceStatusSync(db, mediaSvc)
+4
View File
@@ -25,6 +25,10 @@ type Config struct {
ZLMSecret string `env:"ZLM_SECRET" envDefault:"su6TiedN2rVAmBbIDX0aa0QTiBJLBdcf"`
RecorderAPIBase string `env:"RECORDER_API_BASE" envDefault:"http://localhost:9090"`
AIServiceBase string `env:"AI_SERVICE_BASE" envDefault:"http://localhost:8000"`
WechatAppID string `env:"WECHAT_APPID" envDefault:""`
WechatSecret string `env:"WECHAT_SECRET" envDefault:""`
WechatTemplateAlarm string `env:"WECHAT_TEMPLATE_ALARM" envDefault:""`
WechatTemplateInspection string `env:"WECHAT_TEMPLATE_INSPECTION" envDefault:""`
InternalAPIKey string `env:"INTERNAL_API_KEY" envDefault:"silk-internal-2026"`
Port int `env:"PORT" envDefault:"3000"`
DefaultAdminUsername string `env:"DEFAULT_ADMIN_USERNAME" envDefault:"admin"`
+1
View File
@@ -30,6 +30,7 @@ func Init(cfg *config.Config) error {
&model.Disease{}, &model.KnowledgeArticle{},
&model.InspectionRecord{},
&model.Tray{}, &model.Batch{}, &model.RearingRecord{},
&model.WechatBinding{},
); err != nil {
slog.Warn("自动迁移有警告(可忽略)", "err", err)
}
+31 -3
View File
@@ -2,6 +2,7 @@ package handler
import (
"bytes"
"context"
"encoding/json"
"io"
"net/http"
@@ -25,8 +26,8 @@ func isUUID(s string) bool {
}
// RegisterInspectionRoutes 注册 AI 巡检路由
func RegisterInspectionRoutes(rg *gin.RouterGroup, db *gorm.DB, s3 *service.S3Service, ai *service.AIClient, imageBucket string) {
rg.POST("/inspections", middleware.RequirePermission(db, "inspection:create"), createInspection(db, s3, ai, imageBucket))
func RegisterInspectionRoutes(rg *gin.RouterGroup, db *gorm.DB, s3 *service.S3Service, ai *service.AIClient, imageBucket string, wechat *service.WechatService, inspectionTemplateID string) {
rg.POST("/inspections", middleware.RequirePermission(db, "inspection:create"), createInspection(db, s3, ai, imageBucket, wechat, inspectionTemplateID))
rg.GET("/inspections", middleware.RequirePermission(db, "inspection:read"), listInspections(db))
}
@@ -44,7 +45,7 @@ func currentUserID(c *gin.Context) *string {
// createInspection 拍照巡检:图片存 S3 → 调 AI /detect → 写记录。
// 幂等:客户端传 Idempotency-Key 头时,重复请求返回已有记录。
func createInspection(db *gorm.DB, s3 *service.S3Service, ai *service.AIClient, bucket string) gin.HandlerFunc {
func createInspection(db *gorm.DB, s3 *service.S3Service, ai *service.AIClient, bucket string, wechat *service.WechatService, inspectionTemplateID string) gin.HandlerFunc {
return func(c *gin.Context) {
idemKey := strings.TrimSpace(c.GetHeader("Idempotency-Key"))
roomID := strings.TrimSpace(c.PostForm("roomId"))
@@ -129,6 +130,33 @@ func createInspection(db *gorm.DB, s3 *service.S3Service, ai *service.AIClient,
rec.RiskScore = &score
level := service.RiskLevel(score)
rec.RiskLevel = &level
// 微信订阅消息(#11 骨架):风险非绿且用户已授权时异步推送
if key := service.WechatTemplateKey(level); key != "" {
go func(uid *string, lv string, sc float64) {
if uid == nil || !wechat.Configured() || inspectionTemplateID == "" {
return
}
var binding model.WechatBinding
if db.Where("user_id = ?", *uid).First(&binding).Error != nil {
return
}
var authorized []string
if len(binding.AuthorizedTemplates) > 0 {
_ = json.Unmarshal(binding.AuthorizedTemplates, &authorized)
}
if !service.IsAuthorized(authorized, key) {
return
}
_ = wechat.SendSubscribe(
context.Background(),
binding.OpenID,
inspectionTemplateID,
service.BuildSubscribeData(lv, sc),
"pages/inspection/index",
)
}(rec.UserID, level, score)
}
}
if err := db.Create(&rec).Error; err != nil {
+129
View File
@@ -0,0 +1,129 @@
package handler
import (
"encoding/json"
"net/http"
"silk-server-go/internal/middleware"
"silk-server-go/internal/model"
"silk-server-go/internal/service"
"github.com/gin-gonic/gin"
"gorm.io/gorm"
)
// RegisterWechatRoutes 注册微信订阅消息路由(骨架,配置占位)
func RegisterWechatRoutes(rg *gin.RouterGroup, db *gorm.DB, wechat *service.WechatService) {
read := middleware.RequirePermission(db, "notification:read")
write := middleware.RequirePermission(db, "notification:write")
rg.GET("/wechat/binding", read, getWechatBinding(db))
rg.POST("/wechat/bind", write, bindWechat(db, wechat))
rg.POST("/wechat/subscribe", write, updateWechatSubscribe(db))
}
// getWechatBinding 当前用户的微信绑定与授权模板
func getWechatBinding(db *gorm.DB) gin.HandlerFunc {
return func(c *gin.Context) {
uid := currentUserID(c)
if uid == nil {
c.JSON(http.StatusUnauthorized, gin.H{"error": "未登录"})
return
}
var b model.WechatBinding
if db.Where("user_id = ?", *uid).First(&b).Error != nil {
c.JSON(http.StatusOK, gin.H{"bound": false})
return
}
c.JSON(http.StatusOK, gin.H{
"bound": true,
"openId": b.OpenID,
"authorized": b.AuthorizedTemplates,
})
}
}
// bindWechat 用 wx.login 的 code 换 openid 并绑定当前用户
func bindWechat(db *gorm.DB, wechat *service.WechatService) gin.HandlerFunc {
return func(c *gin.Context) {
uid := currentUserID(c)
if uid == nil {
c.JSON(http.StatusUnauthorized, gin.H{"error": "未登录"})
return
}
var body struct {
Code string `json:"code"`
}
if err := c.ShouldBindJSON(&body); err != nil || body.Code == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "缺少 code"})
return
}
openid, err := wechat.Code2Session(c.Request.Context(), body.Code)
if err != nil {
c.JSON(http.StatusBadGateway, gin.H{"error": err.Error()})
return
}
var b model.WechatBinding
if db.Where("user_id = ?", *uid).First(&b).Error != nil {
b = model.WechatBinding{UserID: *uid, OpenID: openid}
if err := db.Create(&b).Error; err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": "绑定失败"})
return
}
} else {
db.Model(&model.WechatBinding{}).Where("id = ?", b.ID).Update("open_id", openid)
}
c.JSON(http.StatusOK, gin.H{"bound": true, "openId": openid})
}
}
// updateWechatSubscribe 更新用户对某场景的订阅授权(alarm/inspection
func updateWechatSubscribe(db *gorm.DB) gin.HandlerFunc {
return func(c *gin.Context) {
uid := currentUserID(c)
if uid == nil {
c.JSON(http.StatusUnauthorized, gin.H{"error": "未登录"})
return
}
var body struct {
Template string `json:"template"`
Authorized bool `json:"authorized"`
}
if err := c.ShouldBindJSON(&body); err != nil || body.Template == "" {
c.JSON(http.StatusBadRequest, gin.H{"error": "缺少 template"})
return
}
var b model.WechatBinding
if db.Where("user_id = ?", *uid).First(&b).Error != nil {
c.JSON(http.StatusNotFound, gin.H{"error": "请先绑定微信"})
return
}
var templates []string
if len(b.AuthorizedTemplates) > 0 {
_ = json.Unmarshal(b.AuthorizedTemplates, &templates)
}
templates = toggleTemplate(templates, body.Template, body.Authorized)
raw, _ := json.Marshal(templates)
db.Model(&model.WechatBinding{}).Where("id = ?", b.ID).Update("authorized_templates", raw)
c.JSON(http.StatusOK, gin.H{"authorized": templates})
}
}
// toggleTemplate 增删授权模板键
func toggleTemplate(list []string, key string, add bool) []string {
out := make([]string, 0, len(list))
exists := false
for _, k := range list {
if k == key {
exists = true
if add {
out = append(out, k)
}
continue
}
out = append(out, k)
}
if add && !exists {
out = append(out, key)
}
return out
}
@@ -32,6 +32,8 @@ var AllPermissions = []PermissionDef{
{"batch:write", "批次管理", "新增、编辑、删除批次"},
{"rearing:read", "饲养记录查看", "查看饲养记录"},
{"rearing:write", "饲养记录管理", "新增、编辑、删除饲养记录"},
{"notification:read", "订阅查看", "查看微信订阅绑定状态"},
{"notification:write", "订阅管理", "绑定微信并管理订阅授权"},
{"user:manage", "用户管理", "管理用户、角色和权限"},
{"audit:read", "审计查看", "查看审计日志"},
}
@@ -45,6 +47,7 @@ var RolePermissionMap = map[string][]string{
"knowledge:read", "knowledge:write",
"inspection:create", "inspection:read",
"tray:read", "tray:write", "batch:read", "batch:write", "rearing:read", "rearing:write",
"notification:read", "notification:write",
"user:manage", "audit:read",
},
RoleOperator: {
@@ -54,6 +57,7 @@ var RolePermissionMap = map[string][]string{
"knowledge:read", "knowledge:write",
"inspection:create", "inspection:read",
"tray:read", "tray:write", "batch:read", "batch:write", "rearing:read", "rearing:write",
"notification:read", "notification:write",
},
RoleViewer: {
"dashboard:view", "room:read", "device:read",
@@ -61,6 +65,7 @@ var RolePermissionMap = map[string][]string{
"knowledge:read",
"inspection:read",
"tray:read", "batch:read", "rearing:read",
"notification:read",
},
RoleFarmer: {
"dashboard:view", "room:read", "device:read", "device:control",
@@ -68,5 +73,6 @@ var RolePermissionMap = map[string][]string{
"knowledge:read",
"inspection:create", "inspection:read",
"tray:read", "batch:read", "rearing:read",
"notification:read", "notification:write",
},
}
+18
View File
@@ -0,0 +1,18 @@
package model
import (
"encoding/json"
"time"
)
// WechatBinding 用户微信订阅绑定
type WechatBinding struct {
ID string `gorm:"type:uuid;primaryKey;default:gen_random_uuid()" json:"id"`
UserID string `gorm:"column:user_id;type:uuid;uniqueIndex" json:"userId"`
OpenID string `gorm:"column:open_id;size:128;uniqueIndex" json:"openId"`
AuthorizedTemplates json.RawMessage `gorm:"column:authorized_templates;type:jsonb" json:"authorizedTemplates,omitempty"`
CreatedAt time.Time `gorm:"type:timestamptz" json:"createdAt"`
UpdatedAt time.Time `gorm:"type:timestamptz" json:"updatedAt"`
}
func (WechatBinding) TableName() string { return "wechat_bindings" }
+213
View File
@@ -0,0 +1,213 @@
package service
import (
"bytes"
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"sync"
"time"
)
const wechatAPIBase = "https://api.weixin.qq.com"
// WechatTemplateKey 风险等级 → 订阅消息场景键(green 不推送)
func WechatTemplateKey(level string) string {
switch level {
case "yellow", "orange", "red":
return "inspection"
default:
return ""
}
}
// IsAuthorized 判断授权模板列表是否包含场景键
func IsAuthorized(authorized []string, key string) bool {
for _, k := range authorized {
if k == key {
return true
}
}
return false
}
// levelZh 风险等级中文
func levelZh(level string) string {
switch level {
case "green":
return "绿"
case "yellow":
return "黄"
case "orange":
return "橙"
case "red":
return "红"
default:
return level
}
}
// BuildSubscribeData 构造订阅消息 datathing1=风险等级 thing2=评分;字段名需与微信模板字段一致)
func BuildSubscribeData(level string, score float64) map[string]map[string]string {
return map[string]map[string]string{
"thing1": {"value": levelZh(level)},
"thing2": {"value": fmt.Sprintf("%.0f分", score)},
}
}
// WechatService 微信小程序订阅消息服务(骨架;未配置时 Configured()=false,调用返回明确错误)
type WechatService struct {
appID string
secret string
baseURL string
accessToken string
tokenExpire time.Time
mu sync.Mutex
httpClient *http.Client
}
// NewWechatService 创建微信服务
func NewWechatService(appID, secret string) *WechatService {
return &WechatService{
appID: appID,
secret: secret,
baseURL: wechatAPIBase,
httpClient: &http.Client{Timeout: 15 * time.Second},
}
}
// Configured 是否已配置 AppID/Secret
func (s *WechatService) Configured() bool {
return s.appID != "" && s.secret != ""
}
// Code2Session 用 wx.login 的 code 换取 openid
func (s *WechatService) Code2Session(ctx context.Context, code string) (string, error) {
if !s.Configured() {
return "", fmt.Errorf("微信未配置(WECHAT_APPID/WECHAT_SECRET")
}
u := s.baseURL + "/sns/jscode2session?appid=" + url.QueryEscape(s.appID) +
"&secret=" + url.QueryEscape(s.secret) +
"&js_code=" + url.QueryEscape(code) + "&grant_type=authorization_code"
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
if err != nil {
return "", err
}
resp, err := s.httpClient.Do(req)
if err != nil {
return "", err
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return "", err
}
var out struct {
OpenID string `json:"openid"`
ErrCode int `json:"errcode"`
ErrMsg string `json:"errmsg"`
}
if err := json.Unmarshal(body, &out); err != nil {
return "", err
}
if out.ErrCode != 0 {
return "", fmt.Errorf("微信 code2session 失败 (%d): %s", out.ErrCode, out.ErrMsg)
}
return out.OpenID, nil
}
// getAccessToken 获取并缓存 access_token(提前 60 秒过期)
func (s *WechatService) getAccessToken(ctx context.Context) (string, error) {
s.mu.Lock()
if s.accessToken != "" && time.Now().Before(s.tokenExpire) {
tok := s.accessToken
s.mu.Unlock()
return tok, nil
}
s.mu.Unlock()
if !s.Configured() {
return "", fmt.Errorf("微信未配置(WECHAT_APPID/WECHAT_SECRET")
}
u := s.baseURL + "/cgi-bin/token?grant_type=client_credential&appid=" +
url.QueryEscape(s.appID) + "&secret=" + url.QueryEscape(s.secret)
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
if err != nil {
return "", err
}
resp, err := s.httpClient.Do(req)
if err != nil {
return "", err
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return "", err
}
var out struct {
AccessToken string `json:"access_token"`
ExpiresIn int `json:"expires_in"`
ErrCode int `json:"errcode"`
ErrMsg string `json:"errmsg"`
}
if err := json.Unmarshal(body, &out); err != nil {
return "", err
}
if out.AccessToken == "" {
return "", fmt.Errorf("微信 access_token 获取失败 (%d): %s", out.ErrCode, out.ErrMsg)
}
s.mu.Lock()
s.accessToken = out.AccessToken
s.tokenExpire = time.Now().Add(time.Duration(out.ExpiresIn-60) * time.Second)
s.mu.Unlock()
return out.AccessToken, nil
}
// SendSubscribe 发送订阅消息
func (s *WechatService) SendSubscribe(ctx context.Context, openid, templateID string, data map[string]map[string]string, page string) error {
token, err := s.getAccessToken(ctx)
if err != nil {
return err
}
payload := map[string]any{
"touser": openid,
"template_id": templateID,
"data": data,
}
if page != "" {
payload["page"] = page
}
raw, err := json.Marshal(payload)
if err != nil {
return err
}
u := s.baseURL + "/cgi-bin/message/subscribe/send?access_token=" + url.QueryEscape(token)
req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, bytes.NewReader(raw))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
resp, err := s.httpClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return err
}
var out struct {
ErrCode int `json:"errcode"`
ErrMsg string `json:"errmsg"`
}
if err := json.Unmarshal(body, &out); err != nil {
return err
}
if out.ErrCode != 0 {
return fmt.Errorf("微信订阅消息发送失败 (%d): %s", out.ErrCode, out.ErrMsg)
}
return nil
}
+112
View File
@@ -0,0 +1,112 @@
package service
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
)
func TestWechatTemplateKey(t *testing.T) {
cases := map[string]string{
"green": "",
"yellow": "inspection",
"orange": "inspection",
"red": "inspection",
"": "",
"unknown": "",
}
for level, want := range cases {
if got := WechatTemplateKey(level); got != want {
t.Errorf("WechatTemplateKey(%q) = %q, want %q", level, got, want)
}
}
}
func TestIsAuthorized(t *testing.T) {
auth := []string{"alarm", "inspection"}
if !IsAuthorized(auth, "inspection") {
t.Error("inspection 应在授权列表内")
}
if IsAuthorized(auth, "push") {
t.Error("push 不应在授权列表内")
}
if IsAuthorized(nil, "inspection") {
t.Error("空授权列表应返回 false")
}
}
func TestBuildSubscribeData(t *testing.T) {
data := BuildSubscribeData("orange", 75)
if data["thing1"]["value"] != "橙" {
t.Errorf("thing1 应为橙色等级,实际 %v", data["thing1"])
}
if data["thing2"]["value"] != "75分" {
t.Errorf("thing2 应为 75分,实际 %v", data["thing2"])
}
}
func TestWechatCode2Session(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/sns/jscode2session" {
t.Errorf("path = %s", r.URL.Path)
}
q := r.URL.Query()
if q.Get("appid") != "app1" || q.Get("secret") != "sec1" || q.Get("js_code") != "code123" {
t.Errorf("参数不正确: %v", q)
}
_ = json.NewEncoder(w).Encode(map[string]string{"openid": "oAbC123", "session_key": "sk"})
}))
defer srv.Close()
s := NewWechatService("app1", "sec1")
s.baseURL = srv.URL
openid, err := s.Code2Session(context.Background(), "code123")
if err != nil {
t.Fatalf("Code2Session 错误: %v", err)
}
if openid != "oAbC123" {
t.Errorf("openid = %s", openid)
}
}
func TestWechatSendSubscribe(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/cgi-bin/message/subscribe/send" {
t.Errorf("path = %s", r.URL.Path)
}
var body map[string]any
_ = json.NewDecoder(r.Body).Decode(&body)
if body["touser"] != "oAbC123" || body["template_id"] != "tmpl1" {
t.Errorf("body 不正确: %v", body)
}
_, _ = w.Write([]byte(`{"errcode":0,"errmsg":"ok"}`))
}))
defer srv.Close()
s := NewWechatService("app1", "sec1")
s.baseURL = srv.URL
s.accessToken = "fake-token"
s.tokenExpire = time.Now().Add(time.Hour)
err := s.SendSubscribe(context.Background(), "oAbC123", "tmpl1", BuildSubscribeData("red", 88), "")
if err != nil {
t.Fatalf("SendSubscribe 错误: %v", err)
}
}
func TestWechatSendSubscribeErrorCode(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_, _ = w.Write([]byte(`{"errcode":40003,"errmsg":"invalid openid"}`))
}))
defer srv.Close()
s := NewWechatService("app1", "sec1")
s.baseURL = srv.URL
s.accessToken = "fake-token"
s.tokenExpire = time.Now().Add(time.Hour)
if err := s.SendSubscribe(context.Background(), "bad", "tmpl1", nil, ""); err == nil {
t.Error("errcode!=0 应返回错误")
}
}