添加webapi, 优化模块启动

zhangzengfei

2024-05-17 9ee887fce2f87f7a79d0b94640cf1d341a254319

添加webapi, 优化模块启动

 client/client.go

@@ -20,7 +20,6 @@
    go registerLoop(ctx)
    go keepaliveLoop(ctx)
    go syncTimeLoop(ctx)

}

func registerLoop(ctx context.Context) {

 client/notify.go

@@ -29,3 +29,27 @@

    return vo.StatusSuccess
}

func Subscribe(url string, msg []byte) int {
    //if clientStatus != vo.StatusSuccess {
    //    return clientStatus
    //}

    rsp, err := util.HttpPost(url, headers, msg)
    if err != nil {
        logger.Warn("Post subscribe failed, %s", err.Error())
        return vo.StatusOtherError
    }

    var stat vo.ResponseStatusList
    err = json.Unmarshal(rsp, &stat)
    if err != nil {
        logger.Warn("Post subscribe response unmarshal failed, %s", err.Error())
        logger.Warn("response, %s", string(rsp))
        return vo.StatusOtherError
    }

    logger.Debug("Post notification success.")

    return vo.StatusSuccess
}

 config/config.go

@@ -20,6 +20,7 @@
}

type client struct {
    ServerId           string `mapstructure:"server-id"`
    DeviceID           string `mapstructure:"device-id"`
    Username           string `mapstructure:"username"`
    Password           string `mapstructure:"password"`
@@ -49,6 +50,12 @@
    Enable         bool   `mapstructure:"enable"`
}

// 梯控设备
type nvcs struct {
    Model string // 型号
    Port  string // 端口
}

type rateLimitConfig struct {
    FillInterval int64 `mapstructure:"fill-interval" json:"fillInterval"`
    Capacity     int64 `mapstructure:"capacity" json:"capacity"`
@@ -59,6 +66,7 @@
var ForwardConf = &forward{}
var RateLimitConf = &rateLimitConfig{}
var ClientConf = &client{}
var NVCSConf = &nvcs{}

// Init is an exported method that takes the environment starts the viper
// (external lib) and returns the configuration struct.
@@ -87,6 +95,7 @@
    v.UnmarshalKey("forward", ForwardConf)
    v.UnmarshalKey("rate-limit", RateLimitConf)
    v.UnmarshalKey("client", ClientConf)
    v.UnmarshalKey("nvcs", NVCSConf)

    if LogConf.Level == "" {
        LogConf.Level = "info"

 controller/captureCtl.go

@@ -38,15 +38,17 @@
        return
    }

    // 如果开启了下级, 身份应该是消息代理, 不再转发到服务器
    face := req.FaceListObject.FaceObject[0]
    logger.Debug("Receive new message, Id:%s Ip:%s faceId:%s, LeftTopX:%d, appearTime:%s", c.RemoteIP(), face.DeviceID, face.FaceID, face.LeftTopX, face.FaceAppearTime)

    // 如果开启了下级, 身份应该是消息代理, 不再转发到服务器
    if config.ClientConf.Enable && config.ServeConf.Role == "agent" {
        go a.Repository.VIIDMsgForward(&req)
    } else if config.ServeConf.Role == "cascade" {
        logger.Debug("Receive new message, Id:%s Ip:%s faceId:%s, LeftTopX:%d, appearTime:%s", c.RemoteIP(), face.DeviceID, face.FaceID, face.LeftTopX, face.FaceAppearTime)
        go service.AddFaceCapture(&face)
    } else {
        logger.Debug("Receive new message, Id:%s Ip:%s faceId:%s, LeftTopX:%d, appearTime:%s", c.RemoteIP(), face.DeviceID, face.FaceID, face.LeftTopX, face.FaceAppearTime)
        go service.AddFaceNotification(&face)
    }

    if config.ForwardConf.SyncServer != "" {
        go a.Repository.FaceForward(req.FaceListObject.FaceObject)
    }


 controller/subPlatform.go

New file
@@ -0,0 +1,71 @@
package controller

import (
    "net/http"

    "gat1400Exchange/repository"
    "gat1400Exchange/vo"
    "github.com/gin-gonic/gin"
)

type SubPlatformController struct {
    Repository repository.SubPlatformRepository
}

// 构造函数
func NewSubPlatformController() SubPlatformController {
    svr := repository.NewSubPlatformRepository()
    controller := SubPlatformController{svr}

    return controller
}

func (s SubPlatformController) List(c *gin.Context) {
    subList, _ := s.Repository.List()

    c.JSON(http.StatusOK, gin.H{"data": subList})
}

func (s SubPlatformController) Create(c *gin.Context) {
    var req vo.RequestSubPlatform
    if err := c.BindJSON(&req); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{"msg": err.Error()})
        return
    }

    if err := s.Repository.Create(&req); err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"msg": err.Error()})
        return
    }

    c.JSON(http.StatusOK, gin.H{"msg": "ok"})
}

func (s SubPlatformController) Update(c *gin.Context) {
    var req vo.RequestSubPlatform
    if err := c.BindJSON(&req); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{"msg": err.Error()})
        return
    }

    if err := s.Repository.Update(&req); err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"msg": err.Error()})
        return
    }

    c.JSON(http.StatusOK, gin.H{"msg": "ok"})
}

func (s SubPlatformController) Delete(c *gin.Context) {
    if c.Param("id") == "" {
        c.JSON(http.StatusBadRequest, gin.H{"msg": "请求的id为空"})
        return
    }

    if err := s.Repository.Delete(c.Param("id")); err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"msg": err.Error()})
        return
    }

    c.JSON(http.StatusOK, gin.H{"msg": "ok"})
}

 controller/subscribeCtl.go

@@ -1,6 +1,7 @@
package controller

import (
    "gat1400Exchange/config"
    "gat1400Exchange/pkg/logger"
    "gat1400Exchange/repository"
    "gat1400Exchange/vo"
@@ -25,14 +26,15 @@
    return controller
}

func (s SubscribeController) SubscribesList(c *gin.Context) {
    subList, _ := s.Repository.List()
func (s SubscribeController) VIIDSubscribesList(c *gin.Context) {
    fromId := c.GetHeader("User-Identify")
    subList, _ := s.Repository.ListByFromId(fromId)

    c.Header("Content-Type", "application/VIID+json;charset=UTF-8")
    c.JSON(http.StatusOK, gin.H{"SubscribeList": subList})
}

func (s SubscribeController) Subscribes(c *gin.Context) {
func (s SubscribeController) VIIDSubscribes(c *gin.Context) {
    var req vo.RequestSubscribe
    if err := c.BindJSON(&req); err != nil {
        c.AbortWithStatus(http.StatusBadRequest)
@@ -43,7 +45,7 @@

    var rsp vo.ResponseStatusList
    for idx, sub := range req.SubscribeListObject.SubscribeObject {
        if err := s.Repository.CreateSubscribe(fromId, &req.SubscribeListObject.SubscribeObject[idx]); err == nil {
        if err := s.Repository.SaveReceiveSubscribe(fromId, &req.SubscribeListObject.SubscribeObject[idx]); err == nil {
            rsp.ResponseStatusObject = append(rsp.ResponseStatusObject, vo.ResponseStatus{
                RequestURL:   c.FullPath(),
                StatusCode:   vo.StatusSuccess,
@@ -58,7 +60,7 @@
    c.JSON(http.StatusOK, gin.H{"ResponseStatusListObject": rsp})
}

func (s SubscribeController) UpdateSubscribes(c *gin.Context) {
func (s SubscribeController) VIIDUpdateSubscribes(c *gin.Context) {
    var req vo.RequestSubscribe
    if err := c.BindJSON(&req); err != nil {
        c.AbortWithStatus(http.StatusBadRequest)
@@ -67,7 +69,7 @@

    var rsp vo.ResponseStatusList
    for idx, sub := range req.SubscribeListObject.SubscribeObject {
        if err := s.Repository.UpdateSubscribe(&req.SubscribeListObject.SubscribeObject[idx]); err == nil {
        if err := s.Repository.UpdateReceiveSubscribe(&req.SubscribeListObject.SubscribeObject[idx]); err == nil {
            rsp.ResponseStatusObject = append(rsp.ResponseStatusObject, vo.ResponseStatus{
                RequestURL:   c.FullPath(),
                StatusCode:   vo.StatusSuccess,
@@ -82,12 +84,12 @@
    c.JSON(http.StatusOK, gin.H{"ResponseStatusListObject": rsp})
}

func (s SubscribeController) DeleteSubscribe(c *gin.Context) {
func (s SubscribeController) VIIDDeleteSubscribe(c *gin.Context) {
    idList := c.Query("IDList")
    var rsp vo.ResponseStatusList

    for _, id := range strings.Split(idList, ",") {
        if err := s.Repository.DeleteSubscribe(id); err == nil {
        if err := s.Repository.DeleteReceiveSubscribe(id); err == nil {
            rsp.ResponseStatusObject = append(rsp.ResponseStatusObject, vo.ResponseStatus{
                RequestURL:   c.FullPath(),
                StatusCode:   vo.StatusSuccess,
@@ -122,9 +124,31 @@
        })

        // 转发
        go s.Srv.FaceForward(msg.FaceObjectList.FaceObject)
        if config.ForwardConf.SyncServer != "" {
            go s.Srv.FaceForward(msg.FaceObjectList.FaceObject)
        }
    }

    c.Header("Content-Type", "application/VIID+json;charset=UTF-8")
    c.JSON(http.StatusOK, gin.H{"ResponseStatusListObject": rsp})
}

// CreateSubscribes 添加下级平台的订阅消息
func (s SubscribeController) CreateSubscribes(c *gin.Context) {
    var req vo.Subscribe
    if err := c.BindJSON(&req); err != nil {
        c.JSON(http.StatusBadRequest, gin.H{"msg": err.Error()})
        return
    }

    if c.Param("id") == "" {
        c.JSON(http.StatusBadRequest, gin.H{"msg": "下级id为空"})
    }

    if err := s.Repository.CreateSubscribe(c.Param("id"), &req); err != nil {
        c.JSON(http.StatusInternalServerError, gin.H{"msg": err.Error()})
        return
    }

    c.JSON(http.StatusOK, gin.H{"msg": "ok"})
}

 cron/cron.go

@@ -16,6 +16,7 @@
func Init() {
    s.Every(config.ForwardConf.RetryInterval).Minutes().StartImmediately().Do(service.ResendImageData)
    s.Every(config.ForwardConf.ReportInterval).Second().StartImmediately().Do(service.DeviceInfoReportTask)
    s.Every(1).Day().StartImmediately().Do(service.CleanExpireData)

    s.StartAsync()
}

 main.go

@@ -42,7 +42,7 @@
    go client.Init1400Client(ctx)

    // 启动网络视频字符叠加器服务
    go service.NVCSServer()
    go service.StartNVCSServer()

    go service.InitSubscribeTask()


 models/positions.go

@@ -1,5 +1,7 @@
package models

import "time"

type Positions struct {
    Id         uint   `gorm:"column:id;primary_key;auto_increment;unique;not null;"`
    DeviceId   string `gorm:"column:device_id;" json:"device_id"`
@@ -19,3 +21,8 @@
func (d *Positions) FindPositionByTime(timestamp int64) error {
    return db.Table(d.TableName()).Where("create_time <= ?", timestamp).Order("create_time desc").First(&d).Error
}

func (d *Positions) Clean() error {
    timestamp := time.Now().Unix() - 3600
    return db.Table(d.TableName()).Where("create_time <= ?", timestamp).Delete(&d).Error
}

 models/subplatform.go

@@ -18,3 +18,24 @@
func (s *SubPlatform) TableName() string {
    return "sub_platforms"
}

func (s *SubPlatform) FindById(id string) error {
    return db.Table(s.TableName()).First(&s, "id = ?", id).Error
}

func (s *SubPlatform) Save() error {
    return db.Table(s.TableName()).Save(s).Error
}

func (s *SubPlatform) DeleteById(id string) error {
    return db.Table(s.TableName()).Where("id = ?", id).Delete(s).Error
}

func (s *SubPlatform) FindAll() ([]SubPlatform, error) {
    var list []SubPlatform
    if err := db.Table(s.TableName()).Find(&list).Error; err != nil {
        return nil, err
    }

    return list, nil
}

 models/subscribe.go

@@ -38,3 +38,12 @@

    return subs, nil
}

func (s *Subscribe) FindByFromId(fromId string) ([]Subscribe, error) {
    var subs []Subscribe
    if err := db.Table(s.TableName()).Where("from_id = ?", fromId).Find(&subs).Error; err != nil {
        return nil, err
    }

    return subs, nil
}

 repository/subPlatformRepo.go

New file
@@ -0,0 +1,68 @@
package repository

import (
    "errors"
    "gat1400Exchange/models"
    "gat1400Exchange/vo"
)

type SubPlatformRepository struct {
}

func NewSubPlatformRepository() SubPlatformRepository {
    return SubPlatformRepository{}
}

func (s *SubPlatformRepository) Create(req *vo.RequestSubPlatform) error {
    var plat models.SubPlatform

    // 设备存在
    if err := plat.FindById(req.Id); err == nil {
        return errors.New("记录已存在")
    }

    plat = models.SubPlatform{
        Id:            req.Id,
        HeartbeatTime: "",
        Name:          req.Name,
        UserName:      req.UserName,
        Realm:         req.Realm,
        Password:      req.Password,
        Description:   req.Description,
        RemoteIP:      req.RemoteIP,
        RemotePort:    req.RemotePort,
    }

    return plat.Save()
}

func (s *SubPlatformRepository) Delete(id string) error {
    var plat models.SubPlatform

    return plat.DeleteById(id)
}

func (s *SubPlatformRepository) List() ([]models.SubPlatform, error) {
    var plat models.SubPlatform

    return plat.FindAll()
}

func (s *SubPlatformRepository) Update(req *vo.RequestSubPlatform) error {
    var plat models.SubPlatform

    err := plat.FindById(req.Id)
    if err != nil {
        return err
    }

    plat.Name = req.Name
    plat.UserName = req.UserName
    plat.Realm = req.Realm
    plat.Password = req.Password
    plat.Description = req.Description
    plat.RemoteIP = req.RemoteIP
    plat.RemotePort = req.RemotePort

    return plat.Save()
}

 repository/subscribeRepo.go

@@ -1,9 +1,15 @@
package repository

import (
    "encoding/json"
    "errors"
    "fmt"
    "gat1400Exchange/client"
    "gat1400Exchange/models"
    "gat1400Exchange/pkg/snowflake"
    "gat1400Exchange/service"
    "gat1400Exchange/vo"
    "time"
)

type SubscribeRepository struct {
@@ -13,7 +19,42 @@
    return SubscribeRepository{}
}

func (s *SubscribeRepository) CreateSubscribe(fromId string, subscribe *vo.Subscribe) error {
func (s *SubscribeRepository) CreateSubscribe(sid string, req *vo.Subscribe) error {
    triggerTime := time.Now().Format("20060102150405")
    req.SubscribeID = triggerTime + snowflake.GenerateIdStr()

    var subscribeMsg vo.RequestSubscribe
    subscribeMsg.SubscribeListObject.SubscribeObject = []vo.Subscribe{*req}

    // 查找下级
    var platform models.SubPlatform
    err := platform.FindById(sid)
    if err != nil {
        return err
    }

    uri := fmt.Sprintf("http://%s:%d/VIID/Subscribes", platform.RemoteIP, platform.RemotePort)
    body, _ := json.Marshal(subscribeMsg)
    if client.Subscribe(uri, body) != vo.StatusSuccess {
        return errors.New("发送订阅消息失败")
    }

    var sub = models.Subscribe{
        Id:     req.SubscribeID,
        Status: req.SubscribeStatus,
        FromId: sid,
        Ext:    *req,
    }

    err = sub.Save()
    if err != nil {
        return err
    }

    return err
}

func (s *SubscribeRepository) SaveReceiveSubscribe(fromId string, subscribe *vo.Subscribe) error {
    var sub = models.Subscribe{
        Id:     subscribe.SubscribeID,
        Status: subscribe.SubscribeStatus,
@@ -31,7 +72,7 @@
    return err
}

func (s *SubscribeRepository) UpdateSubscribe(subscribe *vo.Subscribe) error {
func (s *SubscribeRepository) UpdateReceiveSubscribe(subscribe *vo.Subscribe) error {
    var sub = models.Subscribe{}
    err := sub.FindById(subscribe.SubscribeID)
    if err != nil {
@@ -46,7 +87,7 @@
    return sub.Save()
}

func (s *SubscribeRepository) DeleteSubscribe(id string) error {
func (s *SubscribeRepository) DeleteReceiveSubscribe(id string) error {
    var sub = models.Subscribe{}
    err := sub.DeleteById(id)
    if err != nil {
@@ -58,8 +99,8 @@
    return err
}

func (s *SubscribeRepository) List() ([]models.Subscribe, error) {
func (s *SubscribeRepository) ListByFromId(id string) ([]models.Subscribe, error) {
    var sub models.Subscribe

    return sub.FindAll()
    return sub.FindByFromId(id)
}

 routes/routes.go

@@ -26,16 +26,18 @@
    // 启用全局跨域中间件
    r.Use(middleware.CORSMiddleware())

    // 路由分组
    apiGroup := r.Group("/VIID")

    // VIID路由分组, 视图库标准接口
    viidGroup := r.Group("/VIID")
    // 注册公共功能路由
    InitSystemRouters(apiGroup)

    InitSystemRouters(viidGroup)
    // 注册采集接口路由
    InitCaptureRouters(apiGroup)
    InitCaptureRouters(viidGroup)
    // 注册订阅通知接口路由
    InitSubscribesRouters(viidGroup)

    InitSubscribesRouters(apiGroup)
    // web api 路由分组
    apiGroup := r.Group("/api")
    InitWebApiRouters(apiGroup)

    logger.Info("初始化路由完成！")


 routes/subscribe.go

@@ -8,10 +8,10 @@
func InitSubscribesRouters(r *gin.RouterGroup) gin.IRoutes {
    subCtl := controller.NewSubscribeController()

    r.GET("/Subscribes", subCtl.SubscribesList)
    r.POST("/Subscribes", subCtl.Subscribes)
    r.PUT("/Subscribes", subCtl.UpdateSubscribes)
    r.DELETE("/Subscribes", subCtl.DeleteSubscribe)
    r.GET("/Subscribes", subCtl.VIIDSubscribesList)
    r.POST("/Subscribes", subCtl.VIIDSubscribes)
    r.PUT("/Subscribes", subCtl.VIIDUpdateSubscribes)
    r.DELETE("/Subscribes", subCtl.VIIDDeleteSubscribe)
    r.POST("/SubscribeNotifications", subCtl.Notifications)

    return r

 routes/webApi.go

New file
@@ -0,0 +1,29 @@
package routes

import (
    "gat1400Exchange/controller"
    "github.com/gin-gonic/gin"
)

func InitWebApiRouters(r *gin.RouterGroup) gin.IRoutes {
    subCtl := controller.NewSubscribeController()

    router := r.Group("/subscribe")
    {
        router.POST("/:id", subCtl.CreateSubscribes)
        //router.GET("/:id", subCtl.Faces)
        //router.PUT("/:id", subCtl.Faces)
        //router.DELETE("/:id", subCtl.Faces)
    }

    platCtl := controller.NewSubPlatformController()
    router = r.Group("/sub_platform")
    {
        router.GET("/list", platCtl.List)
        router.POST("/create", platCtl.Create)
        router.PUT("/:id", platCtl.Update)
        router.DELETE("/:id", platCtl.Delete)
    }

    return r
}

 service/clean.go

New file
@@ -0,0 +1,8 @@
package service

import "gat1400Exchange/models"

func CleanExpireData() {
    var pos models.Positions
    pos.Clean()
}

 service/device.go

@@ -16,7 +16,7 @@
var deviceAliveCache *expirable.LRU[string, bool]

func init() {
    deviceAliveCache = expirable.NewLRU[string, bool](100, nil, time.Second*60) //过滤指定时间内的重复人物
    deviceAliveCache = expirable.NewLRU[string, bool](100, nil, time.Second*60)
}

type DevReportData struct {
@@ -42,18 +42,17 @@
        return nil
    }

    logger.Info("Start device info report task, server:%s.", config.ForwardConf.ReportServer)

    if config.ForwardConf.ReportServer == "" {
        return errors.New("Server addr is empty!")
        return nil
    }

    logger.Info("Start device info report task, server:%s.", config.ForwardConf.ReportServer)

    var d models.Device
    devices, err := d.FindAll()
    if err != nil {
        return err
    }


    for _, dev := range devices{
        if _, exists := deviceAliveCache.Get(dev.Id); !exists {

 service/nvcs.go

@@ -36,7 +36,7 @@
}

// 对接网络视频字符叠加器,接收udp发送的楼层信息, 更新device地址
func NVCSServer() {
func NVCSA1UDPServer() {
    // 指定监听的端口
    port := config.ServeConf.Port

@@ -93,7 +93,7 @@
        elevator := data.Elevator[0]

        // 程序部署在设备端, 字符叠加器上报的名称允许为空. 在云端, 名称必须与摄像机相同
        if !config.ClientConf.Enable {
        if config.ServeConf.Role != "agent" {
            elevator.Name = strings.Trim(elevator.Name, " ")
            if elevator.Name == "" {
                continue
@@ -116,3 +116,9 @@
        logger.Debug("Received %d bytes from %s, %+v", numBytes, clientAddr, data)
    }
}

func StartNVCSServer() {
    if config.NVCSConf.Model == "A1" {
        go NVCSA1UDPServer()
    }
}

 service/resend.go

@@ -10,11 +10,6 @@
)

func ResendImageData() {
    if err := util.HttpGet(config.ForwardConf.SyncServer); err != nil {
        logger.Debug("The server cannot be reached. %s", err.Error())
        return
    }

    var cacheMod models.Cache
    cacheItems, _ := cacheMod.FindAll()
    logger.Debug("Start resend task. cache len:%d", len(cacheItems))
@@ -24,13 +19,13 @@
            if client.FaceCapture([]byte(c.Data)) != vo.StatusSuccess {
                c.UpdateRetryCount()
                logger.Warn("The data resend failed. retry count %d", c.Retry+1)
                continue
                return
            }
        } else {
            if !util.SendData([]byte(c.Data), config.ForwardConf.SyncServer) {
                c.UpdateRetryCount()
                logger.Warn("The data resend failed. retry count %d", c.Retry+1)
                continue
                return
            }
        }


 service/subscribe.go

@@ -3,6 +3,7 @@
import (
    "context"
    "encoding/json"
    "gat1400Exchange/config"
    "strings"
    "sync"
    "time"
@@ -52,12 +53,12 @@
    }
}

func InitSubscribeTask() error {
func InitSubscribeTask() {
    var s models.Subscribe
    subList, err := s.FindAll()
    subList, err := s.FindByFromId(config.ClientConf.ServerId)
    if err != nil {
        logger.Error("Find account by channel error:%v", err)
        return err
        logger.Error("Find subscribe error, server id %s, %v", config.ClientConf.ServerId, err)
        return
    }

    for idx, _ := range subList {
@@ -68,17 +69,16 @@
        CreateTask(&subList[idx])
    }

    return nil
    return
}

func AddFaceCapture(face *vo.FaceObject) {
func AddFaceNotification(face *vo.FaceObject) {
    TaskProcMap.Range(func(key, value interface{}) bool {
        value.(TaskProcInfo).task.AddFace(face)
        return true
    })

    logger.Debug("添加人脸.")
    //TaskWaitGroup.Wait()
    logger.Debug("Add Face Notification.")
}

func StopSubscribeTask() {

 vo/subPlatform.go

New file
@@ -0,0 +1,12 @@
package vo

type RequestSubPlatform struct {
    Id          string `json:"Id" binding:"required"`
    Name        string `json:"Name" binding:"required"`
    UserName    string `json:"Username" binding:"required"`
    Realm       string `json:"Realm"`
    Password    string `json:"Password" binding:"required"`
    Description string `json:"Description" binding:"required"`
    RemoteIP    string `json:"RemoteIp" binding:"required"`
    RemotePort  int    `json:"RemotePort" binding:"required"`
}

 vo/subscribe.go

@@ -8,22 +8,22 @@

type Subscribe struct {
    SubscribeID           string `json:"SubscribeID"`
    Title                 string `json:"Title"`
    SubscribeDetail       string `json:"SubscribeDetail"`
    ResourceURI           string `json:"ResourceURI"`
    ApplicantName         string `json:"ApplicantName"`
    ApplicantOrg          string `json:"ApplicantOrg"`
    BeginTime             string `json:"BeginTime"` // Kept as string for direct compatibility
    EndTime               string `json:"EndTime"`   // Kept as string for direct compatibility
    ReceiveAddr           string `json:"ReceiveAddr"`
    ReportInterval        int    `json:"ReportInterval"`
    Reason                string `json:"Reason"`
    OperateType           int    `json:"OperateType"`
    SubscribeStatus       int    `json:"SubscribeStatus"`
    SubscribeCancelOrg    string `json:"SubscribeCancelOrg"`
    SubscribeCancelPerson string `json:"SubscribeCancelPerson"`
    CancelTime            string `json:"CancelTime"` // Kept as string for direct compatibility
    CancelReason          string `json:"CancelReason"`
    Title                 string `json:"Title" binding:"required"`
    SubscribeDetail       string `json:"SubscribeDetail" binding:"required"`
    ResourceURI           string `json:"ResourceURI" binding:"required"`
    ApplicantName         string `json:"ApplicantName" binding:"required"`
    ApplicantOrg          string `json:"ApplicantOrg" binding:"required"`
    BeginTime             string `json:"BeginTime" binding:"required"`
    EndTime               string `json:"EndTime" binding:"required"`
    ReceiveAddr           string `json:"ReceiveAddr" binding:"required"`
    ReportInterval        int    `json:"ReportInterval" binding:"required"`
    Reason                string `json:"Reason" binding:"required"`
    OperateType           int    `json:"OperateType" `          // 0：订阅；1：取消订阅
    SubscribeStatus       int    `json:"SubscribeStatus" `      // 0：订阅中 1：已取消订阅 2：订阅到期 9：未订阅
    SubscribeCancelOrg    string `json:"SubscribeCancelOrg"`    // 仅在取消订阅时使用
    SubscribeCancelPerson string `json:"SubscribeCancelPerson"` // 仅在取消订阅时使用
    CancelTime            string `json:"CancelTime"`            // 仅在取消订阅时使用
    CancelReason          string `json:"CancelReason"`          // 仅在取消订阅时使用
}

func (s *Subscribe) Scan(value interface{}) error {