ljydlv
/
videoplayer
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'refs/heads/feat-user-sync'

This commit is contained in:
junleea 2024-12-15 14:55:01 +08:00
parent f60328be98 62d336e1c0
commit eb7a57c215
8 changed files with 520 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
dao/user.go
View File

@ -35,7 +35,10 @@ func CreateUser(name, password, email, gender string, age int) uint {
}
func DeleteUserByID(id int) int {
DB.Delete(&User{}, id)
res := DB.Delete(&User{}, id)
if res.Error != nil {
return 0
}
return id
}
@ -98,6 +101,44 @@ func UpdateUserByID2(id int, req proto.UpdateUserInfoReq) {
}
// 用户修改自己的信息
func UpdateUserByID3(id int, req proto.UpdateUserInfoReq) {
DB.Model(&User{}).Where("id = ?", id).Updates(User{Name: req.Username, Age: req.Age, Avatar: req.Avatar, Gender: req.Gender})
func UpdateUserByID3(id int, req proto.UpdateUserInfoReq) error {
res := DB.Model(&User{}).Where("id = ?", id).Updates(User{Name: req.Username, Age: req.Age, Avatar: req.Avatar, Gender: req.Gender})
if res.Error != nil {
return res.Error
}
return nil
}
// 用户数据同步-添加
func AddUserSync(req proto.UserAddOrUpdate) uint {
res := DB.Exec("insert into users (id, created_at, updated_at, deleted_at, name, age, email, password,gender,role,redis,run,upload,video_func,device_func,cid_func,avatar,create_time,update_time) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", req.ID, req.CreatedAt, req.UpdatedAt, req.DeletedAt, req.Name, req.Age, req.Email, req.Password, req.Gender, req.Role, req.Redis, req.Run, req.Upload, req.VideoFunc, req.DeviceFunc, req.CIDFunc, req.Avatar, req.CreateTime, req.UpdateTime)
if res.Error != nil {
return 0
}
return req.ID
}
// 用户数据同步-更新
func UpdateUserSync(req proto.UserAddOrUpdate) uint {
res := DB.Exec("update users set created_at=?, updated_at=?, deleted_at=?, name=?, age=?, email=?, password=?,gender=?,role=?,redis=?,run=?,upload=?,video_func=?,device_func=?,cid_func=?,avatar=?,create_time=?,update_time=? where id=?", req.CreatedAt, req.UpdatedAt, req.DeletedAt, req.Name, req.Age, req.Email, req.Password, req.Gender, req.Role, req.Redis, req.Run, req.Upload, req.VideoFunc, req.DeviceFunc, req.CIDFunc, req.Avatar, req.CreateTime, req.UpdateTime, req.ID)
if res.Error != nil {
return 0
}
return req.ID
}
// 用户数据同步-删除
func DeleteUserSync(req proto.UserDelID) uint {
res := DB.Delete(&User{}, req.ID)
if res.Error != nil {
return 0
}
return req.ID
}
// 获取所有用户
func GetAllUser() []User {
var users []User
DB.Find(&users)
return users
}

72
handler/user.go
View File

@ -26,6 +26,8 @@ func SetUpUserGroup(router *gin.Engine) {
userGroup.POST("/search", SearchHandler)
userGroup.POST("/info", GetUserInfo)
userGroup.POST("/update", UpdateUserInfo)
userGroup.POST("/sync", GetSyncUserInfo)
userGroup.POST("/delete", DeleteUser)
}
type RLReq struct {
@ -77,6 +79,23 @@ func GetUserInfo(c *gin.Context) {
}
}
func DeleteUser(c *gin.Context) {
var req GetUserInfoReq
id, _ := c.Get("id")
user_id := int(id.(float64))
if err := c.ShouldBind(&req); err == nil {
res := service.DeleteUserService(req.ID, user_id)
if res != 0 {
c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": res})
} else {
c.JSON(200, gin.H{"code": proto.OperationFailed, "message": "failed", "data": res})
}
} else {
c.JSON(200, gin.H{"code": proto.ParameterError, "message": err, "data": "2"})
return
}
}
func UpdateUserInfo(c *gin.Context) {
var req_data proto.UpdateUserInfoReq
id, _ := c.Get("id")
@ -350,3 +369,56 @@ func registerHandler(c *gin.Context) {
c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": data})
return
}
type SyncUserReq struct {
Token string `json:"token" form:"token"`
Types int `json:"type" form:"type"` // 1为全量同步 2为增量同步
Device string `json:"device" form:"device"`
Confirm proto.UserSyncConfirm `json:"confirm" form:"confirm"`
}
func GetSyncUserInfo(c *gin.Context) {
var req_data SyncUserReq
if err := c.ShouldBind(&req_data); err == nil {
if req_data.Token == "" {
c.JSON(200, gin.H{"code": proto.ParameterError, "message": "error", "data": "token is empty"})
return
} else {
if worker.IsContainSet("super_permission_tokens", req_data.Token) {
if proto.Config.SERVER_USER_TYPE == "master" {
if req_data.Types == 1 { //1为全量同步
add_users := dao.GetAllUser()
resp := proto.UserSyncResp{}
resp.Add = add_users
c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": resp})
} else if req_data.Types == 2 { //2为增量同步
if req_data.Device == "" || worker.IsContainSet("sync_devices_ids", req_data.Device) == false {
c.JSON(200, gin.H{"code": proto.ParameterError, "message": "error", "data": "device is empty or not exist"})
return
}
res := service.GetUserSyncData(req_data.Device)
c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": res})
} else if req_data.Types == 3 { //3为确认同步数据
res := service.ConfirmSyncUserData(req_data.Device, req_data.Confirm)
if res == nil {
c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": "success"})
} else {
c.JSON(200, gin.H{"code": proto.OperationFailed, "message": "failed", "data": "failed"})
}
} else {
c.JSON(200, gin.H{"code": proto.ParameterError, "message": "type is error", "data": proto.UserSyncResp{}})
}
} else {
c.JSON(200, gin.H{"code": proto.NoPermission, "message": "no permission,server is not master", "data": proto.UserSync{}})
}
} else {
c.JSON(200, gin.H{"code": proto.NoPermission, "message": "error", "data": "no permission"})
return
}
}
} else {
c.JSON(200, gin.H{"error": err.Error(), "code": proto.ParameterError, "message": "error"})
return
}
}

50
main.go
View File

@ -14,6 +14,7 @@ import (
"videoplayer/dao"
"videoplayer/handler"
"videoplayer/proto"
"videoplayer/service"
"videoplayer/worker"
)
@ -223,8 +224,34 @@ func ReadConfigToSetSystem() {
logClean.Every = 86400
cron_infos = append(cron_infos, logClean)
}
}
var is_exist bool
user_sync_id := -1
for i, v := range cron_infos {
if v.Type == 2 {
is_exist = true
if proto.Config.USER_SYNC_TIME != v.Every {
v.Every = proto.Config.USER_SYNC_TIME
v.Curr = proto.Config.USER_SYNC_TIME
}
user_sync_id = i
break
}
}
if proto.Config.SERVER_USER_TYPE == "slave" {
if proto.Config.USER_SYNC_TIME > 0 && !is_exist {
var userSync proto.CronInfo
userSync.Type = 2
userSync.Info = "user"
userSync.Curr = proto.Config.USER_SYNC_TIME
userSync.Every = proto.Config.USER_SYNC_TIME
cron_infos = append(cron_infos, userSync)
} else if user_sync_id != -1 {
cron_infos = append(cron_infos[:user_sync_id], cron_infos[user_sync_id+1:]...) //删除
}
}
} else {
if proto.Config.LOG_SAVE_DAYS > 0 {
var logClean proto.CronInfo
@ -234,6 +261,14 @@ func ReadConfigToSetSystem() {
logClean.Every = 86400
cron_infos = append(cron_infos, logClean)
}
if proto.Config.SERVER_USER_TYPE == "slave" && proto.Config.USER_SYNC_TIME > 0 {
var userSync proto.CronInfo
userSync.Type = 2
userSync.Info = "user"
userSync.Curr = proto.Config.USER_SYNC_TIME
userSync.Every = proto.Config.USER_SYNC_TIME
cron_infos = append(cron_infos, userSync)
}
}
//存入redis
json_data, err := json.Marshal(cron_infos)
@ -267,6 +302,18 @@ func RunGeneralCron() {
v.Curr -= 10
}
}
//2 从服务器同步数据
if v.Type == 2 {
if v.Curr <= 0 {
//执行从服务器同步数据
if proto.Config.SERVER_USER_TYPE == "slave" && v.Info == "user" {
go service.UserSyncDataFromMaster()
}
v.Curr = v.Every
} else {
v.Curr -= 10
}
}
}
//存入redis
json_data, err := json.Marshal(cron_infos)
@ -275,7 +322,6 @@ func RunGeneralCron() {
} else {
worker.SetRedis(key, string(json_data))
}
}
}

3
proto/conf.go
View File

@ -77,6 +77,9 @@ type ConfigStruct struct {
MONITOR bool `json:"monitor"` // 状态监控及邮件通知
SERVER_PORT string `json:"server_port"` // 服务端口
LOG_SAVE_DAYS int `json:"log_save_days"` // 日志保存天数,-1表示不保存，0表示永久保存
SERVER_USER_TYPE string `json:"user_type"` // 服务器用户类型，master: 主服务器，slave: 从服务器，从服务器会定时同步数据
MASTER_SERVER_DOMAIN string `json:"master_server_domain"` // 主服务器域名
USER_SYNC_TIME int `json:"user_sync_time"` // 用户数据同步时间，单位秒
}
// 读取配置文件

58
proto/user_req.go
View File

@ -1,5 +1,10 @@
package proto
import (
"time"
"videoplayer/dao"
)
type UpdateUserInfoReq struct {
ID int `json:"id" form:"id"` //用户id
Username string `json:"name" form:"name"` //用户名
@ -23,8 +28,59 @@ type CIDRUN struct {
// 用于执行函数，方法
type CronInfo struct {
Type int `json:"type" form:"type"` //类型编码,1日志清理（且只会有一个），其他待定
Type int `json:"type" form:"type"` //类型编码,1日志清理（且只会有一个），其他待定，2从服务器同步数据
Info string `json:"info" form:"info"` //信息
Curr int `json:"curr" form:"curr"` //当前剩余时间，每次执行减10s小于等于0则执行
Every int `json:"every" form:"every"` //每隔多少秒执行一次,小于等于0表示不执行，时间粒度为10s
}
// 用户数据同步
type UserSyncResp struct {
Update []dao.User `json:"update" form:"update"` //更新用户
Add []dao.User `json:"add" form:"add"` //添加用户
Delete []UserDelID `json:"delete" form:"delete"` //删除用户
}
// 用户数据同步
type UserSync struct {
Update []UserAddOrUpdate `json:"update" form:"update"` //更新用户
Add []UserAddOrUpdate `json:"add" form:"add"` //添加用户
Delete []UserDelID `json:"delete" form:"delete"` //删除用户
}
// 用户数据同步确认
type UserSyncConfirm struct {
Add []UserConfirmID `json:"add" form:"add"` //添加用户
Update []UserConfirmID `json:"update" form:"update"` //更新用户
Delete []UserConfirmID `json:"delete" form:"delete"` //删除用户
}
type UserConfirmID struct {
ID uint `json:"id" form:"id"` //用户id
}
type UserDelID struct {
ID uint `json:"ID" form:"ID"` //用户id
}
type UserAddOrUpdate struct {
ID uint `json:"ID" form:"ID"` //用户id
CreatedAt time.Time `json:"CreatedAt" form:"CreatedAt"` //创建时间
UpdatedAt time.Time `json:"UpdatedAt" form:"UpdatedAt"` //更新时间
DeletedAt time.Time `json:"DeletedAt" form:"DeletedAt"` //删除时间
Name string `json:"Name" form:"Name"` //用户名
Age int `json:"Age" form:"Age"` //年龄
Email string `json:"Email" form:"Email"` //邮箱
Password string `json:"Password" form:"Password"` //密码
Gender string `json:"Gender" form:"Gender"` //性别
Role string `json:"Role" form:"Role"` //角色
Redis bool `json:"Redis" form:"Redis"` //是否刷新redis
Run bool `json:"Run" form:"Run"` //是否运行
Upload bool `json:"Upload" form:"Upload"` //是否上传头像
VideoFunc bool `json:"VideoFunc" form:"VideoFunc"` //视频功能
DeviceFunc bool `json:"DeviceFunc" form:"DeviceFunc"`
CIDFunc bool `json:"CIDFunc" form:"CIDFunc"`
Avatar string `json:"Avatar" form:"Avatar"` //头像
CreateTime string `json:"CreateTime" form:"CreateTime"`
UpdateTime string `json:"UpdateTime" form:"UpdateTime"`
}

184
service/userService.go
View File

@ -1,13 +1,25 @@
package service
import (
"errors"
"regexp"
"strconv"
"time"
"videoplayer/dao"
"videoplayer/proto"
"videoplayer/worker"
)
func CreateUser(name, password, email, gender string, age int) uint {
return dao.CreateUser(name, password, email, gender, age)
id := dao.CreateUser(name, password, email, gender, age)
if id != 0 {
//添加用户信息到同步列表
err := setSyncUserDataSet("add", int(id))
if err != nil {
return id
}
}
return id
}
func GetUser(name, email, password string) dao.User {
@ -46,8 +58,15 @@ func UpdateUser(user_id int, req proto.UpdateUserInfoReq) (int, error) {
cur_user := dao.FindUserByID2(user_id)
//fmt.Println("cur_user:", cur_user, "req:", req)
if user_id == req.ID && cur_user.Role != "admin" {
dao.UpdateUserByID3(user_id, req) //用户修改自己的信息，不能修改权限信息
err := dao.UpdateUserByID3(user_id, req) //用户修改自己的信息，不能修改权限信息
//添加修改用户信息到同步列表
if err == nil {
err2 := setSyncUserDataSet("update", user_id)
if err2 != nil {
return user_id, nil
}
}
return user_id, err
} else if cur_user.Role == "admin" {
dao.UpdateUserByID2(req.ID, req)
return req.ID, nil
@ -55,3 +74,164 @@ func UpdateUser(user_id int, req proto.UpdateUserInfoReq) (int, error) {
return 0, nil
}
}
func DeleteUserService(id, user_id int) int {
res := 0
if user_id == id {
res = dao.DeleteUserByID(id)
} else {
user := dao.FindUserByID2(user_id)
if user.Role == "admin" {
res = dao.DeleteUserByID(id)
}
}
if res != 0 {
//添加删除用户信息到同步列表
err := setSyncUserDataSet("delete", id)
if err != nil {
return res
}
}
return res
}
func UserSyncDataFromMaster() {
//从接口获取数据
url := "http://" + proto.Config.MASTER_SERVER_DOMAIN + "/user/sync"
tokens := worker.GetRedisSetMembers("super_permission_tokens")
user_sync_data := worker.SyncDataFromMasterReq(url, tokens[0])
add_users := user_sync_data.Add
update_users := user_sync_data.Update
delete_users := user_sync_data.Delete
//未成功操作的id
var fail_ids []uint
//添加用户
for _, v := range add_users {
res := dao.AddUserSync(v)
if res == 0 {
fail_ids = append(fail_ids, v.ID)
}
}
//更新用户
for _, v := range update_users {
res := dao.UpdateUserSync(v)
if res == 0 {
fail_ids = append(fail_ids, v.ID)
}
}
//删除用户
for _, v := range delete_users {
res := dao.DeleteUserSync(v)
if res == 0 {
fail_ids = append(fail_ids, v.ID)
}
}
}
// 同步数据到主服务器-增删改数据
func GetUserSyncData(device string) proto.UserSyncResp {
key := device + "_sync_user_ids"
add_temp_key := device + "_sync_user_ids_add_confirm_temp"
update_temp_key := device + "_sync_user_ids_update_confirm_temp"
delete_temp_key := device + "_sync_user_ids_delete_confirm_temp"
//需要获取暂存集合的并集，清空暂存集合，存入待确认集合
add_user_ids := worker.GetRedisSetUnion(key+"_add", add_temp_key)
update_user_ids := worker.GetRedisSetUnion(key+"_update", update_temp_key)
delete_user_ids := worker.GetRedisSetUnion(key+"_delete", delete_temp_key)
add_users := []dao.User{}
update_users := []dao.User{}
delete_users := []proto.UserDelID{}
for _, v := range add_user_ids {
id, _ := strconv.Atoi(v)
user := dao.FindUserByUserID(id)
add_users = append(add_users, user)
}
for _, v := range update_user_ids {
id, _ := strconv.Atoi(v)
user := dao.FindUserByUserID(id)
update_users = append(update_users, user)
}
for _, v := range delete_user_ids {
id, _ := strconv.Atoi(v)
delete_users = append(delete_users, proto.UserDelID{ID: uint(id)})
}
//将id存入暂存集合，清空原集合，存入待确认集合主要保证在确认时，有新的数据加入不会在确认时漏掉
worker.SetRedisSetUnionAndStore(add_temp_key, key+"_add")
worker.ClearRedisSet(key + "_add")
worker.SetRedisSetUnionAndStore(update_temp_key, key+"_update")
worker.ClearRedisSet(key + "_update")
worker.SetRedisSetUnionAndStore(delete_temp_key, key+"_delete")
worker.ClearRedisSet(key + "_delete")
return proto.UserSyncResp{Add: add_users, Update: update_users, Delete: delete_users}
}
func setSyncUserDataSet(t string, id int) error {
devices := worker.GetRedisSetMembers("sync_devices_ids") //主服务器查看从服务器的设备列表
var err error
for _, device := range devices {
key := device + "_sync_user_ids"
if t == "add" {
key_ := key + "_add"
worker.SetRedisSetAdd(key_, strconv.Itoa(id))
} else if t == "update" {
key_ := key + "_update"
worker.SetRedisSetAdd(key_, strconv.Itoa(id))
} else if t == "delete" {
key_ := key + "_delete"
worker.SetRedisSetAdd(key_, strconv.Itoa(id))
} else {
err = errors.New("error")
}
}
return err
}
// 确认同步数据
func ConfirmSyncUserData(device string, data proto.UserSyncConfirm) error {
var ids_add []string
var err error
for _, v := range data.Add {
ids_add = append(ids_add, strconv.Itoa(int(v.ID)))
}
add_key := device + "_sync_user_ids_add_confirm"
isSuccess := worker.SetRedisSetAddBatchWithExpire(add_key, ids_add, time.Second*30)
if !isSuccess {
err = errors.New("set confirm error")
}
var ids_update []string
for _, v := range data.Update {
ids_update = append(ids_update, strconv.Itoa(int(v.ID)))
}
update_key := device + "_sync_user_ids_update_confirm"
isSuccess = worker.SetRedisSetAddBatchWithExpire(update_key, ids_update, time.Second*30)
if !isSuccess {
err = errors.New("set confirm error")
}
var ids_delete []string
for _, v := range data.Delete {
ids_delete = append(ids_delete, strconv.Itoa(int(v.ID)))
}
del_key := device + "_sync_user_ids_delete_confirm"
isSuccess = worker.SetRedisSetAddBatchWithExpire(del_key, ids_delete, time.Second*30)
if !isSuccess {
err = errors.New("set confirm error")
}
//待确认集合暂存
ids_add_confirm_temp := device + "_sync_user_ids_add_confirm_temp"
ids_update_confirm_temp := device + "_sync_user_ids_update_confirm_temp"
ids_delete_confirm_temp := device + "_sync_user_ids_delete_confirm_temp"
//取差集
add_diff := worker.SetRedisSetDiffAndStore(ids_add_confirm_temp, add_key)
update_diff := worker.SetRedisSetDiffAndStore(ids_update_confirm_temp, update_key)
delete_diff := worker.SetRedisSetDiffAndStore(ids_delete_confirm_temp, del_key)
if add_diff != true || update_diff != true || delete_diff != true {
err = errors.New("confirm error")
}
return err
}

61
worker/redis.go
View File

@ -292,6 +292,23 @@ func SetRedisSetAdd(key string, value string) bool {
return true
}
// 批量添加元素
func SetRedisSetAddBatchWithExpire(key string, values []string, expire time.Duration) bool {
ctx := context.Background()
err := RedisClient.SAdd(ctx, key, values).Err()
if err != nil {
fmt.Println("Error setting key: %v", err)
return false
}
err = RedisClient.Expire(ctx, key, expire).Err()
if err != nil {
fmt.Println("Error setting key: %v", err)
return false
}
return true
}
// 设置set,添加元素
func SetRedisSetAddWithExpire(key string, value string, expire time.Duration) bool {
ctx := context.Background()
@ -399,3 +416,47 @@ func Subscribe(channel string) []string {
}
return messages
}
// redis两个set求差集存入第一个set
func SetRedisSetDiffAndStore(key1 string, key2 string) bool {
ctx := context.Background()
err := RedisClient.SDiffStore(ctx, key1, key1, key2).Err() //将key1和key2的差集存入key1
if err != nil {
fmt.Println("Error setting key: %v", err)
return false
}
return true
}
// redis将第二个set存入第一个set
func SetRedisSetUnionAndStore(key1 string, key2 string) bool {
ctx := context.Background()
err := RedisClient.SUnionStore(ctx, key1, key1, key2).Err() //将key1和key2的并集存入key1
if err != nil {
fmt.Println("Error setting key: %v", err)
return false
}
return true
}
// redis 清空set
func ClearRedisSet(key string) bool {
ctx := context.Background()
err := RedisClient.Del(ctx, key).Err()
if err != nil {
fmt.Println("Error setting key: %v", err)
return false
}
return true
}
// 获取两个集合的并集
func GetRedisSetUnion(key1 string, key2 string) []string {
ctx := context.Background()
val, err := RedisClient.SUnion(ctx, key1, key2).Result()
if err != nil {
fmt.Println("Error getting key: %v", err)
return nil
}
return val
}

38
worker/req.go
View File

@ -3,10 +3,12 @@ package worker
import (
"bytes"
"encoding/json"
"fmt"
"io"
"io/ioutil"
"net/http"
"strings"
"videoplayer/proto"
)
var client *http.Client
@ -80,3 +82,39 @@ func GenerateCompletion(url, prompt string, model string) (map[string]interface{
return result, nil
}
// 获取同步数据通用方法
func SyncDataFromMasterReq(url string, token string) proto.UserSync {
//从接口获取数据
req, err := http.NewRequest("POST", url, nil)
if err != nil {
return proto.UserSync{}
}
req.Header.Set("token", token)
client := &http.Client{}
//获取数据
resp, err := client.Do(req)
if err != nil {
return proto.UserSync{}
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
if err != nil {
return proto.UserSync{}
}
var result map[string]interface{}
err = json.Unmarshal(body, &result)
if err != nil {
return proto.UserSync{}
}
fmt.Println("SyncDataFromMasterReq result:", result)
if result["code"].(float64) != 0 {
return proto.UserSync{}
}
var userSync proto.UserSync
err = json.Unmarshal([]byte(result["data"].(string)), &userSync)
if err != nil {
return proto.UserSync{}
}
return userSync
}