424 lines
10 KiB
Go
424 lines
10 KiB
Go
package main
|
||
|
||
import (
|
||
"bufio"
|
||
"context"
|
||
"dypid-client/api"
|
||
"dypid-client/config"
|
||
"dypid-client/utils/folder"
|
||
"fmt"
|
||
"net/http"
|
||
"os"
|
||
"path/filepath"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"fyne.io/fyne/v2"
|
||
"fyne.io/fyne/v2/app"
|
||
"fyne.io/fyne/v2/container"
|
||
"fyne.io/fyne/v2/layout"
|
||
"fyne.io/fyne/v2/widget"
|
||
"golang.org/x/sync/errgroup"
|
||
)
|
||
|
||
var (
|
||
version = "dev"
|
||
isRun = false
|
||
logText = widget.NewMultiLineEntry()
|
||
ctx, cancel = context.WithCancel(context.Background())
|
||
)
|
||
|
||
func main() {
|
||
//初始化配置
|
||
config.InitConfig()
|
||
|
||
a := app.New()
|
||
newWindow := a.NewWindow("抖音数据上传工具 - 版本" + version)
|
||
newWindow.Resize(fyne.NewSize(930, 600))
|
||
|
||
logText.Scroll = container.ScrollVerticalOnly
|
||
|
||
// URL输入组件
|
||
urlEntry := widget.NewEntry()
|
||
urlEntry.SetPlaceHolder("http://127.0.0.1:8080")
|
||
urlEntry.Text = config.APPConfig.Url
|
||
urlEntry.OnChanged = func(s string) {
|
||
config.WriteConfig("url", urlEntry.Text)
|
||
}
|
||
//Token输入组件
|
||
tokenEntry := widget.NewEntry()
|
||
tokenEntry.SetPlaceHolder("请输入Token")
|
||
tokenEntry.Text = config.APPConfig.Token
|
||
tokenEntry.OnChanged = func(s string) {
|
||
config.WriteConfig("token", tokenEntry.Text)
|
||
}
|
||
//目录选择输入组件
|
||
selectedDirLabel := widget.NewEntry()
|
||
selectedDirLabel.SetPlaceHolder("未选择目录（默认为程序运行目录）")
|
||
selectedDirLabel.Text = config.APPConfig.LookingPath
|
||
selectedDirLabel.OnChanged = func(s string) {
|
||
config.WriteConfig("looking-path", selectedDirLabel.Text)
|
||
}
|
||
//目录选择按钮
|
||
selectDirBtn := widget.NewButton("选择检测目录", func() {
|
||
// 调用CGO实现的Windows原生对话框
|
||
selectedPath := folder.OpenFolderDialog()
|
||
if selectedPath == "" {
|
||
return
|
||
}
|
||
selectedDirLabel.SetText(selectedPath)
|
||
config.WriteConfig("looking-path", selectedPath)
|
||
})
|
||
//上传线程数输入组件
|
||
threadCountLabel := widget.NewEntry()
|
||
threadCountLabel.SetPlaceHolder("10")
|
||
threadCountLabel.Text = strconv.Itoa(config.APPConfig.ThreadCount)
|
||
threadCountLabel.OnChanged = func(s string) {
|
||
i, err := strconv.Atoi(threadCountLabel.Text)
|
||
if err != nil {
|
||
AddLog("输入 上传线程 错误")
|
||
}
|
||
config.WriteConfig("thread-count", i)
|
||
}
|
||
//同时处理文件数输入组件
|
||
handleFileCountLabel := widget.NewEntry()
|
||
handleFileCountLabel.SetPlaceHolder("50")
|
||
handleFileCountLabel.Text = strconv.Itoa(config.APPConfig.HandleFileCount)
|
||
handleFileCountLabel.OnChanged = func(s string) {
|
||
i, err := strconv.Atoi(handleFileCountLabel.Text)
|
||
if err != nil {
|
||
AddLog("输入 同时处理文件数 错误")
|
||
}
|
||
config.WriteConfig("handle-file-count", i)
|
||
}
|
||
|
||
//是否启动程序时启动上传程序组件
|
||
isRunOnStartWidget := widget.NewCheck("启动程序时启动上传程序", func(b bool) {
|
||
config.WriteConfig("is-run-on-start", b)
|
||
})
|
||
isRunOnStartWidget.Checked = config.APPConfig.IsRunOnStart
|
||
//开始运行按钮
|
||
startRun := func() {
|
||
s := "==============================="
|
||
|
||
AddLog(s)
|
||
if strings.TrimSpace(tokenEntry.Text) == "" {
|
||
AddLog("错误：请输入Token")
|
||
return
|
||
}
|
||
AddLog(fmt.Sprintf("服务器地址：%s", config.APPConfig.Url))
|
||
AddLog(fmt.Sprintf("Token：%s", config.APPConfig.Token))
|
||
AddLog(fmt.Sprintf("检测目录：%s", config.APPConfig.LookingPath))
|
||
AddLog(fmt.Sprintf("同时处理文件数：%v", config.APPConfig.HandleFileCount))
|
||
AddLog(fmt.Sprintf("单文件上传线程：%v", config.APPConfig.ThreadCount))
|
||
AddLog(s)
|
||
|
||
isRun = true
|
||
go StartLooking(ctx, config.APPConfig.LookingPath)
|
||
}
|
||
startBtn := widget.NewButton("开始运行", startRun)
|
||
//停止运行按钮
|
||
stopBtn := widget.NewButton("停止运行", func() {
|
||
cancel()
|
||
ctx, cancel = context.WithCancel(context.Background())
|
||
isRun = false
|
||
})
|
||
// 清除日志按钮
|
||
clearLogBtn := widget.NewButton("清除日志", func() {
|
||
logText.SetText("")
|
||
AddLog("日志已清除")
|
||
})
|
||
|
||
// 组装左侧面板
|
||
leftPanel := container.NewBorder(
|
||
nil,
|
||
// 底部 - 放置按钮
|
||
container.NewVBox(
|
||
isRunOnStartWidget,
|
||
startBtn,
|
||
stopBtn,
|
||
clearLogBtn,
|
||
),
|
||
nil,
|
||
nil,
|
||
// 中间内容
|
||
container.NewVBox(
|
||
widget.NewLabelWithStyle("服务器地址：", fyne.TextAlignLeading, fyne.TextStyle{Bold: true}),
|
||
urlEntry,
|
||
widget.NewSeparator(),
|
||
widget.NewLabelWithStyle("Token：", fyne.TextAlignLeading, fyne.TextStyle{Bold: true}),
|
||
tokenEntry,
|
||
widget.NewSeparator(),
|
||
widget.NewLabelWithStyle("检测目录：", fyne.TextAlignLeading, fyne.TextStyle{Bold: true}),
|
||
selectedDirLabel,
|
||
selectDirBtn,
|
||
widget.NewSeparator(),
|
||
container.NewHBox(
|
||
widget.NewLabelWithStyle("同时处理文件数：", fyne.TextAlignLeading, fyne.TextStyle{Bold: true}),
|
||
handleFileCountLabel,
|
||
),
|
||
container.NewHBox(
|
||
widget.NewLabelWithStyle("单文件上传线程：", fyne.TextAlignLeading, fyne.TextStyle{Bold: true}),
|
||
threadCountLabel,
|
||
),
|
||
widget.NewSeparator(),
|
||
layout.NewSpacer(), // 添加一个弹性空间，将内容向上推
|
||
),
|
||
)
|
||
// 组装右侧面板（日志显示）
|
||
rightPanel := container.NewBorder(
|
||
widget.NewLabelWithStyle("运行日志", fyne.TextAlignCenter, fyne.TextStyle{Bold: true}),
|
||
nil, nil, nil,
|
||
container.NewScroll(logText),
|
||
)
|
||
// 使用HSplit容器创建可调整大小的左右分割布局
|
||
splitContainer := container.NewHSplit(leftPanel, rightPanel)
|
||
splitContainer.SetOffset(0.35) // 左侧占35%宽度
|
||
|
||
// 按钮状态同步
|
||
go func() {
|
||
for {
|
||
if isRun && (!startBtn.Disabled() || stopBtn.Disabled()) {
|
||
fyne.Do(func() {
|
||
startBtn.Disable()
|
||
stopBtn.Enable()
|
||
})
|
||
} else if !isRun && (startBtn.Disabled() || !stopBtn.Disabled()) {
|
||
fyne.Do(func() {
|
||
startBtn.Enable()
|
||
stopBtn.Disable()
|
||
})
|
||
}
|
||
time.Sleep(100 * time.Microsecond)
|
||
}
|
||
}()
|
||
|
||
//在程序启动时运行上传程序
|
||
go func() {
|
||
if !config.APPConfig.IsRunOnStart {
|
||
return
|
||
}
|
||
time.Sleep(time.Second)
|
||
startRun()
|
||
}()
|
||
|
||
newWindow.SetContent(splitContainer)
|
||
newWindow.ShowAndRun()
|
||
}
|
||
|
||
func AddLog(message string) {
|
||
fyne.Do(func() {
|
||
logText.Append(message + "\n")
|
||
// 自动滚动到底部
|
||
logText.CursorRow = len(logText.Text)
|
||
})
|
||
}
|
||
|
||
// 上传数据代码
|
||
var httpClient = &http.Client{
|
||
Transport: &http.Transport{
|
||
MaxIdleConns: 200,
|
||
MaxIdleConnsPerHost: 100,
|
||
IdleConnTimeout: 30 * time.Second,
|
||
},
|
||
Timeout: 30 * time.Second,
|
||
}
|
||
|
||
type Task struct {
|
||
FilePath string
|
||
FileLines int
|
||
}
|
||
|
||
func StartLooking(ctx context.Context, lookingPath string) {
|
||
AddLog("正在运行上传程序")
|
||
t := time.NewTicker(time.Minute)
|
||
defer t.Stop()
|
||
|
||
f := func() {
|
||
var path = "./"
|
||
if lookingPath != "" {
|
||
path = lookingPath
|
||
}
|
||
files, err := getTxtFiles(path)
|
||
if err != nil {
|
||
AddLog(err.Error())
|
||
return
|
||
}
|
||
if files == nil {
|
||
return
|
||
}
|
||
|
||
//检测到文件
|
||
start := time.Now()
|
||
//统计文件行数
|
||
fileLines := make(map[string]int)
|
||
AddLog(fmt.Sprintf("正在统计 %v 个文件行数", len(files)))
|
||
isAllEmpty := true
|
||
for _, filePath := range files {
|
||
select {
|
||
case <-ctx.Done():
|
||
AddLog("上传程序已退出")
|
||
return
|
||
default:
|
||
file, err := os.Open(filePath)
|
||
if err != nil {
|
||
AddLog("打开文件失败：" + err.Error())
|
||
}
|
||
// 使用 bufio.Scanner 逐行读取
|
||
scanner := bufio.NewScanner(file)
|
||
lineCount := 0
|
||
for scanner.Scan() {
|
||
lineCount++
|
||
}
|
||
file.Close()
|
||
if lineCount == 0 {
|
||
continue
|
||
}
|
||
fileLines[filepath.Base(filePath)] = lineCount
|
||
if lineCount != 0 {
|
||
isAllEmpty = false
|
||
}
|
||
AddLog(fmt.Sprintf("%s 文件行数：%v", filepath.Base(filePath), lineCount))
|
||
}
|
||
}
|
||
if isAllEmpty {
|
||
AddLog("所有文件都为空，不进行上传")
|
||
return
|
||
}
|
||
|
||
//添加任务
|
||
var tasks []Task
|
||
for k, v := range fileLines {
|
||
tasks = append(tasks, Task{FilePath: k, FileLines: v})
|
||
}
|
||
|
||
// 使用errgroup控制并发
|
||
g, ctx := errgroup.WithContext(ctx)
|
||
g.SetLimit(config.APPConfig.HandleFileCount) // 设置最大同时处理文件数为50
|
||
// 执行所有任务
|
||
for _, task := range tasks {
|
||
task := task // 创建局部变量
|
||
g.Go(func() error {
|
||
select {
|
||
case <-ctx.Done():
|
||
return ctx.Err()
|
||
default:
|
||
AddLog("正在上传文件：" + filepath.Base(task.FilePath))
|
||
processFile(ctx, task.FilePath, task.FileLines)
|
||
err := os.Truncate(task.FilePath, 0)
|
||
if err != nil {
|
||
AddLog("清空文件失败：" + err.Error())
|
||
}
|
||
return nil
|
||
}
|
||
})
|
||
}
|
||
|
||
// 等待所有任务完成
|
||
if err := g.Wait(); err != nil {
|
||
AddLog(fmt.Sprintf("任务执行出错: %v", err))
|
||
} else {
|
||
AddLog("所有任务执行完成！")
|
||
}
|
||
|
||
AddLog(fmt.Sprintf("上传完成，耗时：%s", time.Since(start).String()))
|
||
}
|
||
|
||
for {
|
||
f()
|
||
select {
|
||
case <-ctx.Done():
|
||
AddLog("上传程序已退出")
|
||
return
|
||
case <-t.C:
|
||
f()
|
||
}
|
||
}
|
||
}
|
||
|
||
// 获取目录中的所有txt文件
|
||
func getTxtFiles(dir string) (txtFiles []string, err error) {
|
||
err = filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// 只处理普通文件，跳过目录
|
||
if !info.Mode().IsRegular() {
|
||
return nil
|
||
}
|
||
|
||
// 检查文件扩展名是否为.txt
|
||
if strings.ToLower(filepath.Ext(path)) == ".txt" {
|
||
txtFiles = append(txtFiles, path)
|
||
}
|
||
|
||
return nil
|
||
})
|
||
|
||
return txtFiles, err
|
||
}
|
||
|
||
func processFile(ctx context.Context, filePath string, fileLines int) {
|
||
var wg sync.WaitGroup
|
||
// 打开文件
|
||
file, err := os.Open(filePath)
|
||
if err != nil {
|
||
AddLog(fmt.Sprintf("无法打开文件 %s: %v", filePath, err))
|
||
return
|
||
}
|
||
defer file.Close()
|
||
|
||
// 创建行通道
|
||
lines := make(chan string, 100)
|
||
// 创建10个worker处理文件上传
|
||
for i := 0; i < config.APPConfig.ThreadCount; i++ {
|
||
wg.Add(1)
|
||
go func(workerID int) {
|
||
processLines(ctx, lines, workerID, filePath)
|
||
wg.Done()
|
||
}(i)
|
||
}
|
||
|
||
// 读取文件并发送到通道
|
||
scanner := bufio.NewScanner(file)
|
||
lineCount := 0
|
||
for scanner.Scan() {
|
||
lines <- scanner.Text()
|
||
lineCount++
|
||
if lineCount%10000 == 0 {
|
||
AddLog(fmt.Sprintf("文件【%s】处理进度：%.2f%%", filePath, float64(lineCount)/float64(fileLines)*100))
|
||
}
|
||
}
|
||
|
||
close(lines)
|
||
wg.Wait()
|
||
|
||
if err := scanner.Err(); err != nil {
|
||
AddLog(fmt.Sprintf("读取文件 %s 错误: %v", filePath, err))
|
||
return
|
||
}
|
||
|
||
AddLog(fmt.Sprintf("文件【%s】处理完成，共处理 %d 行数据", filePath, lineCount))
|
||
}
|
||
|
||
func processLines(ctx context.Context, lines <-chan string, workerID int, filePath string) {
|
||
for line := range lines {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
default:
|
||
// 跳过空行
|
||
if strings.TrimSpace(line) == "" {
|
||
continue
|
||
}
|
||
// 上传数据
|
||
if err := api.UploadDataToServer(httpClient, line); err != nil {
|
||
AddLog(fmt.Sprintf("Worker %d (文件 %s): 上传失败: %v", workerID, filePath, err))
|
||
}
|
||
}
|
||
}
|
||
}
|