eta_api/services/report_pdf_service.go

package services

import (
	"encoding/json"
	"eta/eta_api/models"
	"eta/eta_api/models/report"
	"eta/eta_api/models/smart_report"
	"eta/eta_api/utils"
	"fmt"
	"github.com/panjf2000/ants"
	"io"
	"net/http"
	"strings"
	"sync"
	"time"
)

const (
	taskInterval      = 10 * time.Second
	httpTimeout       = 5 * time.Minute
	maxPoolSize       = 4
	BlockingQueueSize = 10
)

var (
	reportOnce sync.Once

	TaskPlanMap = new(sync.Map)
)

type PdfRequest struct {
	ReportUrl  string `json:"ReportUrl"`
	ReportId   int    `json:"ReportId"`
	FreeLayout bool   `json:"FreeLayout"`
}
type PdfResponse struct {
	PdfUrl       string `json:"PdfUrl"`
	JpgUrl       string `json:"JpgUrl"`
	MobilePdfUrl string `json:"MobilePdfUrl"`
	MobileJpgUrl string `json:"MobileJpgUrl"`
}
type BaseResponse struct {
	Ret         int
	Msg         string
	ErrMsg      string
	ErrCode     string
	Data        json.RawMessage
	Success     bool
	IsSendEmail bool
}
type ReportPDFService interface {
	//RejectCallback 任务拒绝策略
	RejectCallback(p interface{})
	// SubmitTask 提交任务
	SubmitTask(tasks *report.ReportPdfTask)

	RegisterWorkPool(pool *ants.Pool)

	ReleasePool()

	TryTimer() *time.Timer
}
type ReportPDFHandler struct {
	Timer      *time.Timer
	WorkerPool *ants.Pool
	HttpClient *http.Client
}

func (d *ReportPDFHandler) ReleasePool() {
	d.WorkerPool.Release()
}

func (d *ReportPDFHandler) RegisterWorkPool(pool *ants.Pool) {
	d.WorkerPool = pool
}

func (d *ReportPDFHandler) TryTimer() *time.Timer {
	return d.Timer
}
func (d *ReportPDFHandler) SubmitTask(task *report.ReportPdfTask) {
	_ = d.WorkerPool.Submit(func() {
		defer func() {
			if r := recover(); r != nil {
				utils.FileLog.Error("任务执行异常: %v", r)
				d.RejectCallback(task)
			}
		}()
		d.processTask(task)
	})
}

// processTask 处理单个任务：发送HTTP请求、更新数据库
func (d *ReportPDFHandler) processTask(task *report.ReportPdfTask) {
	var err error
	if task.ReportType == 1 {
		err = models.UpdatePdfUrlReportById(task.ReportId)
		if err != nil {
			utils.FileLog.Info("清空pdf长图字段失败， Err: \n" + err.Error())
		}
	} else if task.ReportType == 2 {
		err = models.UpdatePdfUrlEnglishReportById(task.ReportId)
		if err != nil {
			utils.FileLog.Info("清空pdf长图字段失败， Err: \n" + err.Error())
		}
	} else if task.ReportType == 3 {
		err = smart_report.UpdatePdfUrlSmartReportById(task.ReportId)
		if err != nil {
			utils.FileLog.Info("清空pdf长图字段失败， Err: \n" + err.Error())
		}
	}
	if err != nil {
		task.LastFinishTime = time.Now()
		task.Message = err.Error()
		_ = report.FailTask(task)
		TaskPlanMap.Delete(task.ReportId)
		return
	}
	// 模拟调用 HTTP 接口获取 MinIO URL
	pdfMinioURL, err := d.fetchPDFMinioURL(task)
	if err != nil {
		utils.FileLog.Error("获取  URL 失败: %v", err)
		task.LastFinishTime = time.Now()
		task.Message = err.Error()
		_ = report.FailTask(task)
		TaskPlanMap.Delete(task.ReportId)
		return
	}
	if task.ReportType == 3 {
		// 更新pdf url
		ob := new(smart_report.SmartReport)
		ob.SmartReportId = task.ReportId
		ob.DetailPdfUrl = pdfMinioURL.PdfUrl
		ob.DetailImgUrl = pdfMinioURL.JpgUrl
		ob.DetailImgUrlMobile = pdfMinioURL.MobileJpgUrl
		ob.DetailPdfUrlMobile = pdfMinioURL.MobilePdfUrl
		if err = ob.Update([]string{"DetailPdfUrl", "DetailImgUrl", "DetailImgUrlMobile", "DetailPdfUrlMobile"}); err != nil {
			utils.FileLog.Info("更新研报失败， Err: \n" + err.Error())
		}
	} else if task.ReportType == 2 {
		err = models.ModifyEnglishReportPdfAndJPGUrl(task.ReportId, pdfMinioURL.PdfUrl, pdfMinioURL.JpgUrl, pdfMinioURL.MobilePdfUrl, pdfMinioURL.MobileJpgUrl)
		if err != nil {
			utils.FileLog.Info("更新研报失败， Err: \n" + err.Error())
		}
	} else if task.ReportType == 1 {
		err = models.ModifyReportPdfAndJpgUrl(task.ReportId, pdfMinioURL.PdfUrl, pdfMinioURL.JpgUrl, pdfMinioURL.MobilePdfUrl, pdfMinioURL.MobileJpgUrl)
		if err != nil {
			utils.FileLog.Info("更新研报失败， Err: \n" + err.Error())
		}
	}
	task.LastFinishTime = time.Now()
	if err != nil {
		task.Message = err.Error()
		_ = report.FailTask(task)
		return
	}
	_ = report.DoneTask(task)

	TaskPlanMap.Delete(task.ReportId)
	utils.FileLog.Info("任务 %d 成功更新 pdf &jpeg URL: %s", task.ReportId, pdfMinioURL)
}

// fetchPDFMinioURL 发起 HTTP 请求获取 MinIO 地址（根据实际接口修改）
func (d *ReportPDFHandler) fetchPDFMinioURL(task *report.ReportPdfTask) (pdfResponse PdfResponse, err error) {

	postData, _ := json.Marshal(PdfRequest{
		ReportUrl:  task.ReportURL,
		ReportId:   task.ReportId,
		FreeLayout: task.FreeLayout,
	})
	body := io.NopCloser(strings.NewReader(string(postData)))
	req, err := http.NewRequest("POST", task.PostUrl, body)
	if err != nil {
		utils.FileLog.Error("PDF generate http.NewRequest Err:"+err.Error(), 1)
		return
	}
	contentType := "application/json;charset=utf-8"
	req.Header.Set("Content-Type", contentType)
	resp, err := d.HttpClient.Do(req)
	if err != nil {
		fmt.Println("PDF generate http client.Do Err:" + err.Error())
		return
	}
	defer resp.Body.Close()
	b, err := io.ReadAll(resp.Body)
	if err != nil {
		return
	}
	result := new(BaseResponse)
	err = json.Unmarshal(b, &result)
	if err != nil {
		utils.FileLog.Error("解析应答失败" + err.Error())
	}
	if result.Success {
		err = json.Unmarshal(result.Data, &pdfResponse)
	} else {
		err = fmt.Errorf("PDF generate http client.Do Err:" + result.Msg)
	}
	return
}

func GetTaskHandlerInstance() (reportHandler ReportPDFService, err error) {
	if reportHandler == nil {
		reportOnce.Do(func() {
			reportHandler = &ReportPDFHandler{
				Timer: time.NewTimer(taskInterval),
				HttpClient: &http.Client{
					Timeout: httpTimeout,
				},
			}
			pool, initPoolErr := ants.NewPool(maxPoolSize, ants.WithMaxBlockingTasks(BlockingQueueSize), ants.WithNonblocking(false), ants.WithPanicHandler(reportHandler.RejectCallback))
			if initPoolErr != nil {
				err = fmt.Errorf("初始化协程池失败: %v", initPoolErr)
			}
			reportHandler.RegisterWorkPool(pool)
		})
	}
	return reportHandler, nil
}
func (d *ReportPDFHandler) RejectCallback(p interface{}) {
	utils.FileLog.Error("任务被拒绝: %+v", p)
}
func StartGenerateReportPDF() {
	reportHandler, err := GetTaskHandlerInstance()
	if err != nil {
		utils.FileLog.Warn("初始化协程池失败: %v", err)
	}
	defer reportHandler.ReleasePool()
	conf, err := models.GetBusinessConfByKey("RemoteGeneratePdfUrl")
	if err != nil {
		utils.FileLog.Error("获取RemoteGeneratePdfUrl失败: %v", err)
		return
	}
	for {
		select {
		case <-reportHandler.TryTimer().C:
			// 从数据库中获取待处理的任务
			tasks, getErr := report.GetPendingTasks()
			if getErr != nil {
				utils.FileLog.Error("获取待处理任务失败: %v", getErr)
				for len(reportHandler.TryTimer().C) > 0 {
					<-reportHandler.TryTimer().C
				}
				reportHandler.TryTimer().Reset(taskInterval)
				continue
			}
			if len(tasks) == 0 {
				utils.FileLog.Info("暂无待处理任务")
				for len(reportHandler.TryTimer().C) > 0 {
					<-reportHandler.TryTimer().C
				}
				reportHandler.TryTimer().Reset(taskInterval)
				continue
			}
			// 提交任务到协程池
			for _, task := range tasks {
				if _, replaced := TaskPlanMap.Load(task.ReportId); !replaced {
					TaskPlanMap.Store(task.ReportId, true)
					task.PostUrl = conf.ConfVal
					reportHandler.SubmitTask(task)
				} else {
					utils.FileLog.Info("任务正在执行中，忽略当前任务,reportId:%d", task.ReportId)
				}
			}
			for len(reportHandler.TryTimer().C) > 0 {
				<-reportHandler.TryTimer().C
			}
			// 重置定时器，等待下一次执行
			reportHandler.TryTimer().Reset(taskInterval)
			fmt.Println("任务执行完成")
		}
	}
}