Skip to content

杨树贤 / liexin_scm_wms_sync

Go to a project
Commit ad0068b2 by mushishixian
Options

精简优化代码

parent c07ab32e
Inline Side-by-side
Showing with 114 additions and 127 deletions
cmd/queue/main.go
......@@ -6,6 +6,7 @@ import (
"scm_server/cmd/queue/sync_goods"
"scm_server/cmd/queue/sync_in_store"
"scm_server/cmd/queue/sync_out_store"
"scm_server/cmd/queue/sync_status"
"scm_server/cmd/queue/sync_supplier"
"scm_server/configs"
"time"
......@@ -55,6 +56,7 @@ func main() {
go sync_goods.Sync()
go sync_in_store.Sync()
go sync_out_store.Sync()
go sync_status.Sync()
//定时请求open falcon证明自己没挂
for {
pushAlive()
......
cmd/queue/queue 0 → 100644
The file could not be displayed because it is too large.
cmd/queue/queue.exe 0 → 100644
The file could not be displayed because it is too large.
cmd/queue/sync_customer/main.go
......@@ -13,7 +13,6 @@ import (
"scm_server/internal/logic"
"scm_server/internal/model"
"scm_server/internal/service"
"time"
)
type RecvPro struct {
......@@ -38,7 +37,7 @@ func (t *RecvPro) Consumer(dataByte []byte) error {
err error
customer model.Customer
operateType string
syncLog model.SyncLog
)
//先去转换队列消息的json,如果失败,记录起来
if err = json.Unmarshal(dataByte, &message); err != nil {
......@@ -88,26 +87,7 @@ func (t *RecvPro) Consumer(dataByte []byte) error {
}
return nil
ERR:
//还要存到一个统一错误表
syncLog = model.SyncLog{
AddTime: time.Now().Unix(),
SyncTime: time.Now().Unix(),
QueueMessage: string(dataByte),
UniqueId: customer.ErpId,
SyncError: err.Error(),
SyncName: "customer",
}
if err = logic.InsertSyncLog(syncLog); err != nil {
service.SendDingTalkRobotToApi(err.Error())
}
//发送钉钉错误消息
msg, _ := json.Marshal(syncLog)
service.SendDingTalkRobotToApi(string(msg))
//保存日志
log.Error("%s", string(msg))
return nil
return logic.DealSyncError(dataByte, customer.ErpId, "customer", err)
}
//同步数据
......
cmd/queue/sync_goods/main.go
......@@ -13,7 +13,6 @@ import (
"scm_server/internal/logic"
"scm_server/internal/model"
"scm_server/internal/service"
"time"
)
type RecvPro struct {
......@@ -39,7 +38,7 @@ func (t *RecvPro) Consumer(dataByte []byte) error {
err error
goods model.Goods
operateType string
syncLog model.SyncLog
)
//先去转换队列消息的json,如果失败,记录起来
if err = json.Unmarshal(dataByte, &message); err != nil {
......@@ -91,26 +90,7 @@ func (t *RecvPro) Consumer(dataByte []byte) error {
}
return nil
ERR:
//还要存到一个统一错误表
syncLog = model.SyncLog{
AddTime: time.Now().Unix(),
SyncTime: time.Now().Unix(),
QueueMessage: string(dataByte),
UniqueId: goods.ErpId, //有可能为
SyncError: err.Error(),
SyncName: "goods",
}
//错误日志插入错误,需要告警
if err = logic.InsertSyncLog(syncLog); err != nil {
service.SendDingTalkRobotToApi(err.Error())
}
//发送钉钉错误消息
msg, _ := json.Marshal(syncLog)
service.SendDingTalkRobotToApi(string(msg))
//保存日志
log.Error("%s", string(msg))
return nil
return logic.DealSyncError(dataByte, goods.ErpId, "goods", err)
}
//同步数据
......
cmd/queue/sync_in_store/main.go
......@@ -13,7 +13,6 @@ import (
"scm_server/internal/logic"
"scm_server/internal/model"
"scm_server/internal/service"
"time"
)
type RecvPro struct {
......@@ -60,8 +59,7 @@ func (t *RecvPro) Consumer(dataByte []byte) error {
err error
inStore model.InStore
operateType string
syncLog model.SyncLog
msg, messageData []byte
messageData []byte
)
//先去转换队列消息的json,如果失败,记录起来
if err = json.Unmarshal(dataByte, &message); err != nil {
......@@ -112,25 +110,7 @@ func (t *RecvPro) Consumer(dataByte []byte) error {
}
return nil
ERR:
//还要存到一个统一错误表
syncLog = model.SyncLog{
AddTime: time.Now().Unix(),
SyncTime: time.Now().Unix(),
QueueMessage: string(dataByte),
UniqueId: inStore.BillId, //有可能为
SyncError: err.Error(),
SyncName: "inStore",
}
//错误日志插入错误,需要告警
if err = logic.InsertSyncLog(syncLog); err != nil {
service.SendDingTalkRobotToApi(err.Error())
}
//发送钉钉错误消息
msg, _ = json.Marshal(syncLog)
service.SendDingTalkRobotToApi(string(msg))
//保存日志
log.Error("%s", string(msg))
return nil
return logic.DealSyncError(dataByte, inStore.BillId, "inStore", err)
}
//同步数据
......@@ -250,6 +230,11 @@ func (t *RecvPro) FailAction(dataByte []byte) error {
}
func Sync() {
defer func() {
if err := recover(); err != nil {
service.SendDingTalkRobotToApi(fmt.Sprintf("%s", err))
}
}()
t := &RecvPro{}
rabbitmq.Recv(rabbitmq.QueueExchange{
"store_in",
......
cmd/queue/sync_out_store/main.go
......@@ -13,7 +13,6 @@ import (
"scm_server/internal/logic"
"scm_server/internal/model"
"scm_server/internal/service"
"time"
)
type RecvPro struct {
......@@ -60,8 +59,7 @@ func (t *RecvPro) Consumer(dataByte []byte) error {
err error
outStore model.OutStore
operateType string
syncLog model.SyncLog
msg, messageData []byte
messageData []byte
)
fmt.Println(string(dataByte))
//先去转换队列消息的json,如果失败,记录起来
......@@ -113,26 +111,7 @@ func (t *RecvPro) Consumer(dataByte []byte) error {
}
return nil
ERR:
//还要存到一个统一错误表
syncLog = model.SyncLog{
AddTime: time.Now().Unix(),
SyncTime: time.Now().Unix(),
QueueMessage: string(dataByte),
UniqueId: outStore.BillId, //有可能为
SyncError: err.Error(),
SyncName: "outStore",
}
//错误日志插入错误,需要告警
if err = logic.InsertSyncLog(syncLog); err != nil {
service.SendDingTalkRobotToApi(err.Error())
}
//发送钉钉错误消息
msg, _ = json.Marshal(syncLog)
service.SendDingTalkRobotToApi(string(msg))
//保存日志
log.Error("%s", string(msg))
return nil
return logic.DealSyncError(dataByte, outStore.BillId, "outStore", err)
}
//同步数据
......@@ -224,6 +203,11 @@ func (t *RecvPro) FailAction(dataByte []byte) error {
}
func Sync() {
defer func() {
if err := recover(); err != nil {
service.SendDingTalkRobotToApi(fmt.Sprintf("%s", err))
}
}()
t := &RecvPro{}
rabbitmq.Recv(rabbitmq.QueueExchange{
"store_out",
......
cmd/queue/sync_status/main.go
package main
package sync_status
import (
"encoding/json"
......@@ -29,7 +29,6 @@ func init() {
}
func (t *RecvPro) Consumer(dataByte []byte) (err error) {
fmt.Println(string(dataByte))
var (
message QueueStatus
syncStatusLog model.SyncStatusLog
......@@ -135,15 +134,15 @@ func SyncErpDataStatus(syncName, uniqueId string) (err error) {
return
}
if apiMethod == "synRewriteErpBaseDataStatus" {
return CheckBaseDataRequest(res)
return CheckBaseDataRequest(syncName, uniqueId, res)
} else if apiMethod == "synErpInventoryStatus" {
return CheckBillDataRequest(res)
return CheckBillDataRequest(syncName, uniqueId, res)
}
return
}
//坚持基础物料的修改状态情况
func CheckBaseDataRequest(res *gosoap.Response) (err error) {
func CheckBaseDataRequest(syncName, uniqueId string, res *gosoap.Response) (err error) {
doc := etree.NewDocument()
if err = doc.ReadFromBytes(res.Body); err != nil {
return
......@@ -161,7 +160,14 @@ func CheckBaseDataRequest(res *gosoap.Response) (err error) {
//金蝶判断成功的标志
if key == "0000" {
//成功还要去修改记录表
fmt.Println("OK")
switch syncName {
case "supplier":
return logic.SyncSupplierErpSuccess(uniqueId)
case "customer":
return logic.SyncCustomerErpSuccess(uniqueId)
case "material":
return logic.SyncGoodsErpSuccess(uniqueId)
}
return
}
//金额判断失败的标志
......@@ -174,8 +180,7 @@ func CheckBaseDataRequest(res *gosoap.Response) (err error) {
}
//检查出入库单的情况
func CheckBillDataRequest(res *gosoap.Response) (err error) {
fmt.Println("checkBillStatus")
func CheckBillDataRequest(syncName, uniqueId string, res *gosoap.Response) (err error) {
doc := etree.NewDocument()
if err = doc.ReadFromBytes(res.Body); err != nil {
return
......@@ -192,6 +197,13 @@ func CheckBillDataRequest(res *gosoap.Response) (err error) {
for key, value := range responseData {
//金蝶判断成功的标志
if key == "0000" {
//成功还要去修改记录表
switch syncName {
case "inStore":
return logic.SyncInStoreErpSuccess(uniqueId)
case "outStore":
return logic.SyncOutStoreErpSuccess(uniqueId)
}
return
}
//金额判断失败的标志
......@@ -208,7 +220,12 @@ func (t *RecvPro) FailAction(dataByte []byte) error {
return nil
}
func main() {
func Sync() {
defer func() {
if err := recover(); err != nil {
service.SendDingTalkRobotToApi(fmt.Sprintf("%s", err))
}
}()
t := &RecvPro{}
rabbitmq.Recv(rabbitmq.QueueExchange{
"store_sync_status",
......
cmd/queue/sync_supplier/main.go
......@@ -13,7 +13,6 @@ import (
"scm_server/internal/logic"
"scm_server/internal/model"
"scm_server/internal/service"
"time"
)
type RecvPro struct {
......@@ -38,7 +37,7 @@ func (t *RecvPro) Consumer(dataByte []byte) error {
err error
supplier model.Supplier
operateType string
syncLog model.SyncLog
)
//先去转换队列消息的json,如果失败,记录起来
if err = json.Unmarshal(dataByte, &message); err != nil {
......@@ -90,25 +89,7 @@ func (t *RecvPro) Consumer(dataByte []byte) error {
}
return nil
ERR:
//还要存到一个统一错误表
syncLog = model.SyncLog{
AddTime: time.Now().Unix(),
SyncTime: time.Now().Unix(),
QueueMessage: string(dataByte),
UniqueId: supplier.ErpId, //有可能为
SyncError: err.Error(),
SyncName: "supplier",
}
if err = logic.InsertSyncLog(syncLog); err != nil {
service.SendDingTalkRobotToApi(err.Error())
}
//发送钉钉错误消息
msg, _ := json.Marshal(syncLog)
service.SendDingTalkRobotToApi(string(msg))
//保存日志
log.Error("%s", string(msg))
return nil
return logic.DealSyncError(dataByte, supplier.ErpId, "supplier", err)
}
//同步数据
......
internal/logic/customer.go
......@@ -18,6 +18,14 @@ const DeleteCustomerSql = "UPDATE `lie_customer` SET `status` = ? , `update_time
const SyncCustomerSuccessSql = "UPDATE `lie_customer` SET sync_status = 1 , sync_time = ? WHERE erp_id = ?"
const SyncCustomerErpSuccessSql = "UPDATE `lie_customer` SET sync_erp_status = 1 , sync_time = ? WHERE erp_id = ?"
//修改同步到erp成功的状态
func SyncCustomerErpSuccess(erpId string) (err error) {
_, err = dao.GetDb().Exec(SyncCustomerErpSuccessSql, time.Now().Unix(), erpId)
return
}
//检查用户是否存在
func CheckCustomerExist(erpId string) (exist bool) {
var customerId int
......
internal/logic/goods.go
......@@ -20,6 +20,14 @@ const SyncGoodsSuccessSql = "UPDATE `lie_goods` SET sync_status = 1 , sync_time
const SyncGoodsSuccessWithStatusSql = "UPDATE `lie_goods` SET sync_status = 1 , sync_time = ? , status = ? WHERE erp_id = ?"
const SyncGoodsErpSuccessSql = "UPDATE `lie_goods` SET sync_erp_status = 1 , sync_time = ? WHERE erp_id = ?"
//修改同步到erp成功的状态
func SyncGoodsErpSuccess(erpId string) (err error) {
_, err = dao.GetDb().Exec(SyncGoodsErpSuccessSql, time.Now().Unix(), erpId)
return
}
//检查物料是否存在
func CheckGoodsExist(erpId string) (exist bool) {
var goodsId int
......
internal/logic/in_store.go
......@@ -14,12 +14,18 @@ const InsertInStoreSql = "INSERT INTO `lie_in_store` (`queue_message`,`bill_id`,
const DeleteInStoreSql = "UPDATE `lie_in_store` SET `status` = ? , `update_time` = ? WHERE `bill_id` = ?"
const WriteInStoreSyncErrorSql = "UPDATE `lie_in_store` SET sync_error = ? WHERE bill_id = ?"
const SyncInStoreSuccessSql = "UPDATE `lie_in_store` SET sync_status = 1 , sync_time = ? WHERE bill_id = ?"
const SyncInStoreSuccessSqlWithStatus = "UPDATE `lie_in_store` SET sync_status = 1 , sync_time = ? , status = ? WHERE bill_id = ?"
const SyncInStoreErpSuccessSql = "UPDATE `lie_in_store` SET sync_erp_status = 1 , sync_time = ? WHERE bill_id = ?"
//修改同步到erp成功的状态
func SyncInStoreErpSuccess(billId string) (err error) {
_, err = dao.GetDb().Exec(SyncInStoreErpSuccessSql, time.Now().Unix(), billId)
return
}
//检查出库是否存在
func CheckInStoreExist(billId string) (exist bool) {
var inStoreId int
......@@ -48,12 +54,6 @@ func DeleteInStore(inStore model.InStore) (err error) {
return
}
//写入同步失败的原因
func WriteInStoreSyncError(billId string, syncError string) (err error) {
_, err = dao.GetDb().Exec(WriteInStoreSyncErrorSql, syncError, billId)
return
}
//修改成功的同步状态
func SyncInStoreSuccess(billId string, isDelete bool) (err error) {
//不是删除,不需要修改状态
......
internal/logic/out_store.go
......@@ -20,6 +20,13 @@ const SyncOutStoreSuccessSql = "UPDATE `lie_out_store` SET sync_status = 1 , syn
const SyncOutStoreSuccessSqlWithStatus = "UPDATE `lie_out_store` SET sync_status = 1 , sync_time = ? , status = ? WHERE bill_id = ?"
const SyncOutStoreErpSuccessSql = "UPDATE `lie_out_store` SET sync_erp_status = 1 , sync_time = ? WHERE bill_id = ?"
//修改同步到erp成功的状态
func SyncOutStoreErpSuccess(billId string) (err error) {
_, err = dao.GetDb().Exec(SyncOutStoreErpSuccessSql, time.Now().Unix(), billId)
return
}
//检查出库是否存在
func CheckOutStoreExist(erpId string) (exist bool) {
var goodsId int
......
internal/logic/supplier.go
......@@ -20,6 +20,14 @@ const WriteSupplierSyncErrorSql = "UPDATE `lie_supplier` SET sync_time = ? , syn
const SyncSupplierSuccessSql = "UPDATE `lie_supplier` SET sync_status = 1 , sync_time = ? WHERE erp_id = ?"
const SyncSupplierErpSuccessSql = "UPDATE `lie_supplier` SET sync_erp_status = 1 , sync_time = ? WHERE erp_id = ?"
//修改同步到erp成功的状态
func SyncSupplierErpSuccess(erpId string) (err error) {
_, err = dao.GetDb().Exec(SyncSupplierErpSuccessSql, time.Now().Unix(), erpId)
return
}
//检查供应商是否存在
func CheckSupplierExist(erpId string) (exist bool) {
var supplierId int = 0
......
internal/logic/sync_log.go
package logic
import (
"encoding/json"
"github.com/go-kratos/kratos/pkg/log"
"scm_server/internal/dao"
"scm_server/internal/model"
"scm_server/internal/service"
"time"
)
const InsertSyncLogSql = "INSERT INTO `lie_sync_log` (`sync_name`,`sync_time`,`sync_error`,`unique_id`,`queue_message`,`add_time`) VALUES (?,?,?,?,?,?)"
......@@ -12,3 +16,26 @@ func InsertSyncLog(syncLog model.SyncLog) (err error) {
_, err = dao.GetDb().Exec(InsertSyncLogSql, syncLog.SyncName, syncLog.SyncTime, syncLog.SyncError, syncLog.UniqueId, syncLog.QueueMessage, syncLog.AddTime)
return err
}
func DealSyncError(dataByte []byte, uniqueId, syncName string, err error) error {
//还要存到一个统一错误表
syncLog := model.SyncLog{
AddTime: time.Now().Unix(),
SyncTime: time.Now().Unix(),
QueueMessage: string(dataByte),
UniqueId: uniqueId,
SyncError: err.Error(),
SyncName: syncName,
}
if err = InsertSyncLog(syncLog); err != nil {
service.SendDingTalkRobotToApi(err.Error())
}
//发送钉钉错误消息
msg, _ := json.Marshal(syncLog)
service.SendDingTalkRobotToApi(string(msg))
//保存日志
log.Error("%s", string(msg))
return nil
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment