From fab3fb73e72fe8f80d78576c9731d50cf3c652f8 Mon Sep 17 00:00:00 2001 From: mubai <1609539827@qq.com> Date: Thu, 27 Feb 2025 00:17:34 +0800 Subject: [PATCH] add film recover crontab --- server/config/DataConfig.go | 2 ++ server/controller/CronController.go | 2 +- server/logic/CronLogic.go | 8 ++++++ server/main.go | 3 +-- server/model/system/CollectRecord.go | 13 ++++++++++ server/model/system/Crontab.go | 2 +- server/plugin/SystemInit/SpiderInit.go | 35 +++++++++++++++++++++++--- server/plugin/db/mysql.go | 3 ++- server/plugin/spider/Spider.go | 33 ++++++++++++++++++++++++ server/plugin/spider/SpiderCron.go | 13 ++++++++++ 10 files changed, 105 insertions(+), 9 deletions(-) diff --git a/server/config/DataConfig.go b/server/config/DataConfig.go index 767ff8e..f160896 100644 --- a/server/config/DataConfig.go +++ b/server/config/DataConfig.go @@ -69,6 +69,8 @@ const ( FilmCrontabKey = "Cron:Task:Film" // DefaultUpdateSpec 每20分钟执行一次 DefaultUpdateSpec = "0 */20 * * * ?" + // EveryWeekSpec 每周日凌晨4点更新一次 + EveryWeekSpec = "0 0 4 * * 7" // DefaultUpdateTime 每次采集最近 3 小时内更新的影片 DefaultUpdateTime = 3 ) diff --git a/server/controller/CronController.go b/server/controller/CronController.go index 4737ee7..b2d29d7 100644 --- a/server/controller/CronController.go +++ b/server/controller/CronController.go @@ -140,7 +140,7 @@ func validTaskInfo(t system.FilmCollectTask) error { // 任务添加参数校验 func validTaskAddVo(vo system.FilmCronVo) error { - if vo.Model != 0 && vo.Model != 1 { + if vo.Model != 0 && vo.Model != 1 && vo.Model != 2 { return errors.New("参数校验失败, 未定义的任务类型") } if vo.Time == 0 { diff --git a/server/logic/CronLogic.go b/server/logic/CronLogic.go index 1fc1168..504778f 100644 --- a/server/logic/CronLogic.go +++ b/server/logic/CronLogic.go @@ -40,6 +40,14 @@ func (cl *CronLogic) AddFilmCrontab(cv system.FilmCronVo) error { } // 将定时任务Id记录到Task中 task.Cid = cid + case 2: + cid, err := spider.AddFilmRecoverCron(task.Spec) + // 如果任务添加失败则直接返回错误信息 + if err != nil { + return errors.New(fmt.Sprint("影视更新定时任务添加失败: ", err.Error())) + } + // 将定时任务Id记录到Task中 + task.Cid = cid } // 如果没有异常则将当前定时任务信息记录到redis中 system.SaveFilmTask(task) diff --git a/server/main.go b/server/main.go index 149e450..259aa64 100644 --- a/server/main.go +++ b/server/main.go @@ -7,12 +7,11 @@ import ( "server/plugin/SystemInit" "server/plugin/db" "server/router" - "time" ) func init() { // 执行初始化前等待20s , 让mysql服务完成初始化指令 - time.Sleep(time.Second * 20) + //time.Sleep(time.Second * 20) //初始化redis客户端 err := db.InitRedisConn() if err != nil { diff --git a/server/model/system/CollectRecord.go b/server/model/system/CollectRecord.go index c3a572d..c230f16 100644 --- a/server/model/system/CollectRecord.go +++ b/server/model/system/CollectRecord.go @@ -80,6 +80,19 @@ func FindRecordById(id uint) *FailureRecord { return &fr } +// PendingRecord 查询所有待处理的记录信息 +func PendingRecord() []FailureRecord { + var list []FailureRecord + // 1. 获取 hour > 4320 || hour < 0 && status = 1 的影片信息 + db.Mdb.Where("(hour > 4320 OR hour < 0) AND status = 1").Find(&list) + // 2. 获取 hour > 0 && hour < 4320 && status = 1 的影片信息(只获取最早的一条记录) + var fr FailureRecord + db.Mdb.Where("hour > 0 AND hour < 4320 AND status = 1").Order("hour DESC, created_at ASC").First(&fr) + // 3. 将 fr 添加到 list中 + list = append(list, fr) + return list +} + // ChangeRecord 修改已完成二次采集的记录状态 func ChangeRecord(fr *FailureRecord, status int) { switch { diff --git a/server/model/system/Crontab.go b/server/model/system/Crontab.go index 9222336..7a1d31d 100644 --- a/server/model/system/Crontab.go +++ b/server/model/system/Crontab.go @@ -19,7 +19,7 @@ type FilmCollectTask struct { Cid cron.EntryID `json:"cid"` // 定时任务Id Time int `json:"time"` // 采集时长, 最新x小时更新的内容 Spec string `json:"spec"` // 执行周期 cron表达式 - Model int `json:"model"` // 任务类型, 0 - 自动更新已启用站点 || 1 - 更新Ids中的资源站数据 + Model int `json:"model"` // 任务类型, 0 - 自动更新已启用站点 || 1 - 更新Ids中的资源站数据 || 2 - 定期清理失败采集记录 State bool `json:"state"` // 状态 开启 | 禁用 Remark string `json:"remark"` // 任务备注信息 } diff --git a/server/plugin/SystemInit/SpiderInit.go b/server/plugin/SystemInit/SpiderInit.go index 8b7726c..ae298bd 100644 --- a/server/plugin/SystemInit/SpiderInit.go +++ b/server/plugin/SystemInit/SpiderInit.go @@ -65,16 +65,28 @@ func CollectCrontabInit() { } // 将定时任务Id记录到Task中 task.Cid = cid + case 2: + cid, err := spider.AddFilmRecoverCron(task.Spec) + // 如果任务添加失败则直接返回错误信息 + if err != nil { + log.Println("自动清理失败采集记录定时任务添加失败: ", err.Error()) + continue + } + // 将定时任务Id记录到Task中 + task.Cid = cid } system.UpdateFilmTask(task) } } else { - // 如果系统中不存在任何定时任务信息, 则添加默认的定时任务 - // 1. 添加一条默认任务, 定时更新所有已启用站点的影片信息 - // 生成任务信息 + /* + 如果系统中不存在任何定时任务信息, 则添加默认的定时任务 + 1. 添加一条默认任务, 定时更新所有已启用站点的影片信息 + 2. 添加一条默认任务, 定时处理采集失败的记录 + 3.生成任务信息 + */ task := system.FilmCollectTask{Id: util.GenerateSalt(), Time: config.DefaultUpdateTime, Spec: config.DefaultUpdateSpec, Model: 0, State: false, Remark: "每20分钟执行一次已启用站点数据的自动更新"} - // 添加一条定时任务 + // 添加一条定时任务-影片定时更新 cid, err := spider.AddAutoUpdateCron(task.Id, task.Spec) // 如果任务添加失败则直接返回错误信息 if err != nil { @@ -85,6 +97,21 @@ func CollectCrontabInit() { task.Cid = cid // 如果没有异常则将当前定时任务信息记录到redis中 system.SaveFilmTask(task) + + // 添加一条定时任务-定期处理失败请求 + recoverTask := system.FilmCollectTask{Id: util.GenerateSalt(), Time: 0, Spec: config.EveryWeekSpec, + Model: 2, State: false, Remark: "每周日凌晨4点清理一次采集失败的采集记录"} + // 添加一条定时任务-影片定时更新 + cid, err = spider.AddFilmRecoverCron(recoverTask.Spec) + // 如果任务添加失败则直接返回错误信息 + if err != nil { + log.Println("失败采集恢复定时任务添加失败: ", err.Error()) + return + } + // 将定时任务Id记录到Task中 + recoverTask.Cid = cid + // 如果没有异常则将当前定时任务信息记录到redis中 + system.SaveFilmTask(recoverTask) } // 完成初始化后启动 Cron diff --git a/server/plugin/db/mysql.go b/server/plugin/db/mysql.go index 582ca32..e522c66 100644 --- a/server/plugin/db/mysql.go +++ b/server/plugin/db/mysql.go @@ -3,6 +3,7 @@ package db import ( "gorm.io/driver/mysql" "gorm.io/gorm" + "gorm.io/gorm/logger" "gorm.io/gorm/schema" "server/config" ) @@ -24,7 +25,7 @@ func InitMysql() (err error) { SingularTable: true, //是否使用 结构体名称作为表名 (关闭自动变复数) //NameReplacer: strings.NewReplacer("spider_", ""), // 替表名和字段中的 Me 为 空 }, - //Logger: logger.Default.LogMode(logger.Info), //设置日志级别为Info + Logger: logger.Default.LogMode(logger.Info), //设置日志级别为Info }) return } diff --git a/server/plugin/spider/Spider.go b/server/plugin/spider/Spider.go index d42709e..5926d9d 100644 --- a/server/plugin/spider/Spider.go +++ b/server/plugin/spider/Spider.go @@ -319,6 +319,39 @@ func SingleRecoverSpider(fr *system.FailureRecord) { // 其余范围,暂不处理 break } +} + +// FullRecoverSpider 扫描记录表中的失败记录, 并进行处理 (用于定时任务定期处理失败采集) +func FullRecoverSpider() { + /* + 获取待处理的记录数据 + 1. 采集时长 > 168h (一周,7天) 状态-1 待处理, | 只获取满足条件的最早的待处理记录 + 2. 采集时长 > 4320h (半年,180天) 状态-1 待处理, | 获取满足条件的所有数据 + */ + list := system.PendingRecord() + + // 遍历记录信息切片, 针对不同时长进行不同处理 + for _, fr := range list { + switch { + case fr.Hour > 0 && fr.Hour < 4320: + // 将此记录之后的所有同类采集记录变更为已重试 + system.ChangeRecord(&fr, 0) + // 如果采集的内容是 7~15 天之内更新的内容,则采集此记录之后的所有更新内容 + // 获取采集参数h, 采集时长变更为 原采集时长 + 采集记录距现在的时长 + h := fr.Hour + int(math.Ceil(time.Since(fr.CreatedAt).Hours())) + // 对当前所有已启用的站点 更新最新 h 小时的内容 + AutoCollect(h) + case fr.Hour < 0, fr.Hour > 4320: + // 将此记录状态修改为已重试 + system.ChangeRecord(&fr, 0) + // 如果采集的是 最近180天内更新的内容 或全部内容, 则只对当前一条记录进行二次采集 + s := system.FindCollectSourceById(fr.OriginId) + collectFilm(s, fr.Hour, fr.PageNumber) + default: + // 其余范围,暂不处理 + break + } + } } diff --git a/server/plugin/spider/SpiderCron.go b/server/plugin/spider/SpiderCron.go index dda38ba..5408874 100644 --- a/server/plugin/spider/SpiderCron.go +++ b/server/plugin/spider/SpiderCron.go @@ -60,6 +60,19 @@ func AddAutoUpdateCron(id, spec string) (cron.EntryID, error) { }) } +// AddFilmRecoverCron 失败采集记录处理 +func AddFilmRecoverCron(spec string) (cron.EntryID, error) { + // 校验 spec 表达式的有效性 + if err := ValidSpec(spec); err != nil { + return -99, errors.New(fmt.Sprint("定时任务添加失败,Cron表达式校验失败: ", err.Error())) + } + return CronCollect.AddFunc(spec, func() { + // 执行失败采集记录恢复 + FullRecoverSpider() + log.Println("执行一次失败采集恢复任务") + }) +} + // RemoveCron 删除定时任务 func RemoveCron(id cron.EntryID) { // 通过定时任务EntryID移出对应的定时任务