youtube-golang/main_program/moveData/moveDataService.go

package move_data

import (
	"container/list"
	"database/sql"
	"fmt"
	"main_program/config"
	"main_program/entity"

	_ "github.com/mattn/go-sqlite3"
	"github.com/tealeg/xlsx"
	"gorm.io/driver/mysql"
	"gorm.io/gorm"
)

type moveDataService struct{}

var MoveDataService moveDataService

func (m *moveDataService) Start(table string, sqlitePath string, mysqlConfig config.MysqlEntity) {
	config.Logger.Info("开始迁移sqlite3到mysql")
	sqliteDB, err := sql.Open("sqlite3", sqlitePath)
	if err != nil {
		config.Logger.Fatal(err)
	}
	config.Logger.Info("连接成功sqlite3...")
	dsn := fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8mb4&parseTime=True&loc=Local", mysqlConfig.User, mysqlConfig.Password, mysqlConfig.Host, mysqlConfig.Database)
	mysqlDB, err := gorm.Open(mysql.Open(dsn), &gorm.Config{})
	if err != nil {
		config.Logger.Fatal("数据库连接失败")
	}
	config.Logger.Info("连接成功mysql...")
	if table == "Channel" {
		m.moveChannel(sqliteDB, mysqlDB)
	}
}

func (m *moveDataService) moveChannel(sqliteDB *sql.DB, mysqlDB *gorm.DB) {
	config.Logger.Info("读取xlsx获取region")
	file, err := xlsx.OpenFile("D:/Work/Code/youtube_dev/youtube-golang/main_program/moveData/channel_region.xlsx")
	if err != nil {
		config.Logger.Fatalf("Error opening file: %s", err)
	}
	config.Logger.Info("开始迁移Channel表")
	// 从sqlite3获取前50个Channel
	continueFlag := true
	for continueFlag {
		rows, err := sqliteDB.Query("SELECT * FROM Channel_copy WHERE is_copy = 0 limit 50")
		if err != nil {
			config.Logger.Fatal(err)
		}
		ChannelList := list.New()
		for rows.Next() {
			channelCopy := new(entity.ChannelCopy)
			if err := rows.Scan(&channelCopy.Id, &channelCopy.ChannelId, &channelCopy.ChannelTitle, &channelCopy.ChannelLanguage, &channelCopy.ChannelReptileTime, &channelCopy.Is_Copy); err != nil {
				config.Logger.Fatal(err)
			}
			ChannelList.PushBack(channelCopy)
		}
		rows.Close()
		if ChannelList.Len() <= 0 {
			break
		}
		// 放入mysql里
		for e := ChannelList.Front(); e != nil; e = e.Next() {
			channelCopy := e.Value.(*entity.ChannelCopy)
			config.Logger.Info(channelCopy)
			channel := new(entity.Channel)
			channel.ChannelId = channelCopy.ChannelId
			channel.ChannelTitle = channelCopy.ChannelTitle
			channel.ChannelLanguage = channelCopy.ChannelLanguage
			channel.ChannelReptileTime = channelCopy.ChannelReptileTime
			channel.Region = m.getRegionByChannelId(channel.ChannelId, file)
			result := mysqlDB.Create(&channel)
			if result.Error != nil {
				config.Logger.Fatal(result.Error)
			}
			// 修改sqlite里状态
			sqlStr, err := sqliteDB.Prepare("UPDATE Channel_copy SET is_copy = 1 WHERE id = ?")
			if err != nil {
				config.Logger.Error(err)
			}
			_, err = sqlStr.Exec(channelCopy.Id)
			if err != nil {
				config.Logger.Error(err)
			}
			sqlStr.Close()
		}
		tmpRows, _ := sqliteDB.Query("SELECT * FROM Channel_copy WHERE is_copy = 0 limit 50")
		continueFlag = tmpRows.Next()
		tmpRows.Close()
	}

}

func (m *moveDataService) getRegionByChannelId(channnelId string, file *xlsx.File) string {
	for _, sheet := range file.Sheets {
		for _, row := range sheet.Rows {
			if row.Cells[3].Value == channnelId {
				return row.Cells[2].Value
			}
		}
	}
	return ""
}