Ma/GoFilm
1
0
Fork 0
mirror of https://github.com/ProudMuBai/GoFilm.git synced 2026-06-12 21:33:10 +08:00
Code Issues Packages Projects Releases Wiki Activity

collect optimize

Browse Source
This commit is contained in:
mubai
2026-04-01 00:21:29 +08:00
parent 8a6bc33cad
commit 8243d2171f
47 changed files with 2774 additions and 1770 deletions
Download Patch File Download Diff File Expand all files Collapse all files

629
film/server/plugin/spider/Spider.go
View File

@@ -13,6 +13,7 @@ import (
"server/model/system"
"server/plugin/common/conver"
"server/plugin/common/util"
"strconv"
"time"
)
@@ -25,93 +26,6 @@ var spiderCore = &JsonCollect{}
// ======================================================= 通用采集方法 =======================================================
// HandleCollect 影视采集 id-采集站ID h-时长/h
func HandleCollect(id string, h int) error {
// 1. 首先通过ID获取对应采集站信息
s := system.FindCollectSourceById(id)
if s == nil {
log.Println("Cannot Find Collect Source Site")
return errors.New(" Cannot Find Collect Source Site ")
} else if !s.State {
log.Println(" The acquisition site was disabled ")
return errors.New(" The acquisition site was disabled ")
}
// 如果是主站点且状态为启用则先获取分类tree信息
if s.Grade == system.MasterCollect && s.State {
// 是否存在分类树信息, 不存在则获取
if !system.ExistsCategoryTree() {
CollectCategory(s)
}
}
// 生成 RequestInfo
r := util.RequestInfo{Uri: s.Uri, Params: url.Values{}}
// 如果 h == 0 则直接返回错误信息
if h == 0 {
log.Println(" Collect time cannot be zero ")
return errors.New(" Collect time cannot be zer ")
}
// 如果 h = -1 则进行全量采集
if h > 0 {
r.Params.Set("h", fmt.Sprint(h))
}
// 2. 首先获取分页采集的页数
pageCount, err := spiderCore.GetPageCount(r)
// 分页页数失败 则再进行一次尝试
if err != nil {
// 如果第二次获取分页页数依旧获取失败则关闭当前采集任务
pageCount, err = spiderCore.GetPageCount(r)
if err != nil {
return err
}
}
// 通过采集类型分别执行不同的采集方法
switch s.CollectType {
case system.CollectVideo:
// 采集视频资源
// 如果采集源参数中采集间隔参数大于500ms,则使用单线程采集
if s.Interval > 500 {
// 少量数据不开启协程
for i := 1; i <= pageCount; i++ {
collectFilm(s, h, i)
// 执行一次采集后休眠指定时长
time.Sleep(time.Duration(s.Interval) * time.Millisecond)
}
} else if pageCount <= config.MAXGoroutine*2 {
// 少量数据不开启协程
for i := 1; i <= pageCount; i++ {
collectFilm(s, h, i)
}
} else {
// 如果分页数量较大则开启协程
ConcurrentPageSpider(pageCount, s, h, collectFilm)
}
// 视频数据采集完成后同步相关信息到mysql
if s.Grade == system.MasterCollect {
// 执行影片信息更新操作
if h > 0 {
// 执行数据更新操作
system.SyncSearchInfo(1)
} else {
// 清空searchInfo中的数据并重新添加, 否则执行
system.SyncSearchInfo(0)
}
// 开启图片同步
if s.SyncPictures {
system.SyncFilmPicture()
}
// 每次成功执行完都清理redis中的相关API接口数据缓存
ClearCache()
}
case system.CollectArticle, system.CollectActor, system.CollectRole, system.CollectWebSite:
log.Println("暂未开放此采集功能!!!")
return errors.New("暂未开放此采集功能")
}
log.Println("Spider Task Exercise Success")
return nil
}
// CollectCategory 影视分类采集
func CollectCategory(s *system.FilmSource) {
// 获取分类树形数据
@@ -127,32 +41,197 @@ func CollectCategory(s *system.FilmSource) {
}
}
// collectFilm 影视详情采集 (单一源分页全采集)
func collectFilm(s *system.FilmSource, h, pg int) {
// 生成请求参数
r := util.RequestInfo{Uri: s.Uri, Params: url.Values{}}
// 设置分页页数
r.Params.Set("pg", fmt.Sprint(pg))
// 如果 h = -1 则进行全量采集
if h > 0 {
r.Params.Set("h", fmt.Sprint(h))
// HandleCollectRefine 采集处理(改版, 新增分类筛选优化)
func HandleCollectRefine(id string, h int) error {
// 1. 首先通过ID获取对应采集站信息
s := system.FindCollectSourceById(id)
if s == nil {
log.Println("Cannot Find Collect Source Site")
return errors.New(" Cannot Find Collect Source Site ")
} else if !s.State {
log.Println(" The acquisition site was disabled ")
return errors.New(" The acquisition site was disabled ")
}
// 如果是主站点且状态为启用则先获取分类tree信息
if s.Grade == system.MasterCollect && s.State {
// 是否存在分类树信息, 不存在则获取
if !system.ExistsCategoryTree() {
CollectCategory(s)
}
}
// 生成 RequestInfo
r := util.RequestInfo{Uri: s.Uri, Params: url.Values{}}
// 通过 采集时长 h 的不同来执行不同前置出处理方式
switch {
case h < 0:
// 采集时长为负数则先执行对应数据表的重置
if s.Grade == system.MasterCollect {
// 重置主站信息表
system.FilmZero()
// 主站点触发重置时, 清除次级站点数据表
system.ResetSlaveMovieInfoTable()
} else {
// 如果存在则删除表中对应的当前站点所有数据
system.DelSlaveInfoBySid(s.Id)
}
case h > 0:
// 如果采集时长是正常数值, 则设置参数 h
r.Params.Set("h", fmt.Sprint(h))
default:
log.Println("Params Collect time Exception !!!")
return errors.New(" Params Collect time Exception !!! ")
}
// 通过采集类型分别执行不同的采集方法
switch s.CollectType {
case system.CollectVideo:
// 采集视频资源 根据采集站类型进行不同逻辑
switch s.Grade {
case system.MasterCollect:
// 获取展示的分类切片信息
cl := system.GetRevealCategoryList()
for _, c := range cl {
// 获取分类采集页数
r.Params.Set("t", fmt.Sprint(c.Id))
pageCount, err := spiderCore.GetPageCount(r)
if err != nil {
// 如果第二次获取分页页数依旧获取失败则关闭当前采集任务
pageCount, err = spiderCore.GetPageCount(r)
if err != nil {
return err
}
}
// 如果采集源参数中采集间隔参数大于500ms,则使用单线程采集
if s.Interval > 500 {
// 少量数据不开启协程
for i := 1; i <= pageCount; i++ {
// 设置采集参数pg
r.Params.Set("pg", fmt.Sprint(i))
collectFilmRefine(s, r)
// 执行一次采集后休眠指定时长
time.Sleep(time.Duration(s.Interval) * time.Millisecond)
}
} else if pageCount <= config.MAXGoroutine*5 {
// 少量数据不开启协程
for i := 1; i <= pageCount; i++ {
r.Params.Set("pg", fmt.Sprint(i))
collectFilmRefine(s, r)
}
} else {
// 如果分页数量较大则开启协程
collectFilmMT(pageCount, s, r, collectFilmRefine)
}
}
case system.SlaveCollect:
pageCount, err := spiderCore.GetPageCount(r)
if err != nil {
// 如果第二次获取分页页数依旧获取失败则关闭当前采集任务
pageCount, err = spiderCore.GetPageCount(r)
if err != nil {
return err
}
}
// 如果采集源参数中采集间隔参数大于500ms,则使用单线程采集
if s.Interval > 500 {
// 少量数据不开启协程
for i := 1; i <= pageCount; i++ {
// 设置采集参数pg
r.Params.Set("pg", fmt.Sprint(i))
collectFilmRefine(s, r)
// 执行一次采集后休眠指定时长
time.Sleep(time.Duration(s.Interval) * time.Millisecond)
}
} else if pageCount <= config.MAXGoroutine*5 {
// 少量数据不开启协程
for i := 1; i <= pageCount; i++ {
r.Params.Set("pg", fmt.Sprint(i))
collectFilmRefine(s, r)
}
} else {
// 如果分页数量较大则开启协程
collectFilmMT(pageCount, s, r, collectFilmRefine)
}
}
// 视频数据采集完成后 对暂存数据进行处理和优化
if s.Grade == system.MasterCollect {
// 如果采集时长为负, (全量采集), 则在数据采集完成后为search表添加索引
switch {
case h < 0:
// 全量采集时进行数据同步以及添加索引(仅保存)
system.SyncMovieDetail(s.Id, s.Grade, 0)
system.AddSearchIndex()
system.AddMovieDetailIndex()
case h < 169:
// 一周内的数据等待定时任务同步, search表需要进行数据同步
break
case h > 720 && h < 8760:
// 一月 < h < 一年, 执行redis同步更新
system.SyncMovieDetail(s.Id, s.Grade, 1)
default:
log.Println("采集内容过多,建议重新进行全量采集或减少采集时长的数值 !!!")
}
// 开启图片同步
if s.SyncPictures {
system.SyncFilmPicture()
}
// 每次成功执行完都清理redis中的相关API接口数据缓存
ClearCache()
} else if s.Grade == system.SlaveCollect {
switch {
case h < 0:
// 如果采集时长为负, (全量采集), 则在数据采集完成后为search表添加索引
system.SyncMovieDetail(s.Id, s.Grade, 0)
system.AddSlaveMovieInfoIndex()
case h < 169:
// 只更新 search 表的内容,从属站点无需处理
break
case h > 720 && h < 8760:
// 一月 < h < 一年, 执行redis同步更新
system.SyncMovieDetail(s.Id, s.Grade, 1)
default:
log.Println("采集内容过多,建议重新进行全量采集或减少采集时长的数值 !!!")
}
}
case system.CollectArticle, system.CollectActor, system.CollectRole, system.CollectWebSite:
log.Println("暂未开放此采集功能!!!")
return errors.New("暂未开放此采集功能")
}
log.Println("Spider Task Exercise Success")
return nil
}
// 影片信息采集, 改进版
func collectFilmRefine(s *system.FilmSource, r util.RequestInfo) {
// 执行采集方法 获取影片详情list
//log.Printf("%s?%s", r.Uri, r.Params.Encode())
list, err := spiderCore.GetFilmDetail(r)
// 获取采集时长信息
h, _ := strconv.Atoi(r.Params.Get("h"))
if err != nil || len(list) <= 0 {
// 添加采集失败记录
fr := system.FailureRecord{OriginId: s.Id, OriginName: s.Name, Uri: s.Uri, CollectType: system.CollectVideo, PageNumber: pg, Hour: h, Cause: fmt.Sprintln(err), Status: 1}
system.SaveFailureRecord(fr)
log.Println("GetMovieDetail Error: ", err)
//uri := fmt.Sprintf("%s?%s", r.Uri, r.Params.Encode())
// h < 2 时会被定时任务覆盖,因此不做处理0
if h > 2 || h < 0 {
fr := system.FailureRecord{OriginId: s.Id, OriginName: s.Name, Uri: s.Uri, Params: r.Params.Encode(), Hour: h, Cause: fmt.Sprintln(err), Status: 1}
system.SaveFailureRecord(fr)
log.Println("GetMovieDetail Error: ", err)
}
return
}
// 通过采集站 Grade 类型, 执行不同的存储逻辑
switch s.Grade {
case system.MasterCollect:
// 主站点 保存完整影片详情信息到 redis
if err = system.SaveDetails(list); err != nil {
// 将数据缓存到redis中
if err = system.MovieDetailCache(list); err != nil {
log.Println("SaveDetails Error: ", err)
}
// 如果采集数据为主站点, 且采集时长小于168 (定时任务更新时需要同步更新search检索信息)
if h > 0 && h <= 168 {
if err := system.BatchSaveOrUpdate(list); err != nil {
log.Println("BatchSaveOrUpdate Search Error: ", err)
}
}
// 如果主站点开启了图片同步, 则将图片url以及对应的mid存入ZSet集合中
if s.SyncPictures {
if err = system.SaveVirtualPic(conver.ConvertVirtualPicture(list)); err != nil {
@@ -160,62 +239,31 @@ func collectFilm(s *system.FilmSource, h, pg int) {
}
}
case system.SlaveCollect:
// 附属站点 仅保存影片播放信息到redis
if err = system.SaveSitePlayList(s.Id, list); err != nil {
// 将采集数据缓存到redis中
if err = system.SlaveDetailCache(s.Id, list); err != nil {
log.Println("SaveDetails Error: ", err)
}
}
}
// collectFilmById 采集指定ID的影片信息
func collectFilmById(ids string, s *system.FilmSource) {
// 生成请求参数
r := util.RequestInfo{Uri: s.Uri, Params: url.Values{}}
// 设置分页页数
r.Params.Set("pg", "1")
// 设置影片IDS参数信息
r.Params.Set("ids", ids)
// 执行采集方法 获取影片详情list
list, err := spiderCore.GetFilmDetail(r)
if err != nil || len(list) <= 0 {
log.Println("GetMovieDetail Error: ", err)
return
}
// 通过采集站 Grade 类型, 执行不同的存储逻辑
switch s.Grade {
case system.MasterCollect:
// 主站点 保存完整影片详情信息到 redis 和 mysql 中
if err = system.SaveDetail(list[0]); err != nil {
log.Println("SaveDetails Error: ", err)
}
// 如果主站点开启了图片同步, 则将图片url以及对应的mid存入ZSet集合中
if s.SyncPictures {
if err = system.SaveVirtualPic(conver.ConvertVirtualPicture(list)); err != nil {
log.Println("SaveVirtualPic Error: ", err)
}
}
case system.SlaveCollect:
// 附属站点 仅保存影片播放信息到redis
if err = system.SaveSitePlayList(s.Id, list); err != nil {
log.Println("SaveDetails Error: ", err)
}
}
}
// ConcurrentPageSpider 并发分页采集, 不限类型
func ConcurrentPageSpider(capacity int, s *system.FilmSource, h int, collectFunc func(s *system.FilmSource, hour, pageNumber int)) {
// 开启协程并发执行
// collectFilmMT 并发采集影片信息
func collectFilmMT(capacity int, s *system.FilmSource, r util.RequestInfo, collectFunc func(s *system.FilmSource, r util.RequestInfo)) {
// 初始化 channel, 容量为 capacity
ch := make(chan int, capacity)
// 收集结束标识
waitCh := make(chan int)
// 循环将所有需采集的页码写入 ch
for i := 1; i <= capacity; i++ {
ch <- i
}
close(ch)
// 开启 MAXGoroutine 数量的协程, 如果分页页数小于协程数则将协程数限制为分页页数
// 开启 MAXGoroutine 数量的协程, 如果分页页数小于设定的最大线程数, 则将线程数设置为1
var GoroutineNum = config.MAXGoroutine
if capacity < GoroutineNum {
GoroutineNum = capacity
if capacity < GoroutineNum*5 {
GoroutineNum = 1
}
// 如果满足开启并发的条件, 则开启GoroutineNum数量的协程进行并发采集
for i := 0; i < GoroutineNum; i++ {
go func() {
defer func() { waitCh <- 0 }()
@@ -225,11 +273,14 @@ func ConcurrentPageSpider(capacity int, s *system.FilmSource, h int, collectFunc
if !ok {
break
}
// 执行对应的采集方法
collectFunc(s, h, pg)
// 执行对应的采集方法, 并发时不同使用同一个requestInfo
requestInfo := util.CopyRequestInfo(r)
requestInfo.Params.Set("pg", fmt.Sprint(pg))
collectFunc(s, requestInfo)
}
}()
}
// 等待所有协程执行完毕
for i := 0; i < GoroutineNum; i++ {
<-waitCh
}
@@ -242,7 +293,7 @@ func BatchCollect(h int, ids ...string) {
if fs := system.FindCollectSourceById(id); fs != nil && fs.State {
// 采用协程并发执行, 每个站点单独开启一个协程执行
go func() {
err := HandleCollect(fs.Id, h)
err := HandleCollectRefine(fs.Id, h)
if err != nil {
log.Println(err)
}
@@ -261,7 +312,7 @@ func AutoCollect(h int) {
for _, s := range system.GetCollectSourceList() {
// 如果当前站点为启用状态 则执行 HandleCollect 进行数据采集
if s.State {
if err := HandleCollect(s.Id, h); err != nil {
if err := HandleCollectRefine(s.Id, h); err != nil {
log.Println(err)
}
}
@@ -295,33 +346,278 @@ func CollectSingleFilm(ids string) {
}
}
// collectFilmById 采集指定ID的影片信息
func collectFilmById(ids string, s *system.FilmSource) {
// 生成请求参数
r := util.RequestInfo{Uri: s.Uri, Params: url.Values{}}
// 设置分页页数
r.Params.Set("pg", "1")
// 设置影片IDS参数信息
r.Params.Set("ids", ids)
// 执行采集方法 获取影片详情list
list, err := spiderCore.GetFilmDetail(r)
if err != nil || len(list) <= 0 {
log.Println("GetMovieDetail Error: ", err)
return
}
// 通过采集站 Grade 类型, 执行不同的存储逻辑
switch s.Grade {
case system.MasterCollect:
// 主站点 保存完整影片详情信息到 redis 和 mysql 中
if err = system.SaveDetail(list[0]); err != nil {
log.Println("SaveDetails Error: ", err)
}
// 如果主站点开启了图片同步, 则将图片url以及对应的mid存入ZSet集合中
if s.SyncPictures {
if err = system.SaveVirtualPic(conver.ConvertVirtualPicture(list)); err != nil {
log.Println("SaveVirtualPic Error: ", err)
}
}
case system.SlaveCollect:
// 附属站点 仅保存影片播放信息到redis
if err = system.UpdateSitePlayList(s.Id, list); err != nil {
log.Println("SaveDetails Error: ", err)
}
}
}
// ======================================================= 待废弃采集方法 =======================================================
//
//// HandleCollect 影视采集 id-采集站ID h-时长/h
//func HandleCollect(id string, h int) error {
// // 1. 首先通过ID获取对应采集站信息
// s := system.FindCollectSourceById(id)
// if s == nil {
// log.Println("Cannot Find Collect Source Site")
// return errors.New(" Cannot Find Collect Source Site ")
// } else if !s.State {
// log.Println(" The acquisition site was disabled ")
// return errors.New(" The acquisition site was disabled ")
// }
// // 如果是主站点且状态为启用则先获取分类tree信息
// if s.Grade == system.MasterCollect && s.State {
// // 是否存在分类树信息, 不存在则获取
// if !system.ExistsCategoryTree() {
// CollectCategory(s)
// }
// }
//
// // 生成 RequestInfo
// r := util.RequestInfo{Uri: s.Uri, Params: url.Values{}}
// // 通过 采集时长 h 的不同来执行不同出处理方式
// switch {
// case h < 0:
// // 采集时长为负数则先执行对应数据表的重置
// if s.Grade == system.MasterCollect {
// system.FilmZero()
// } else {
// // 如果所处站点是次级站点, 则删除对应站点在表中的数据
// system.ResetSlaveMovieInfoTable()
// }
// case h > 0:
// // 如果采集时长是正常数值, 则设置参数 h
// r.Params.Set("h", fmt.Sprint(h))
// default:
// log.Println("Params Collect time Exception !!!")
// return errors.New(" Params Collect time Exception !!! ")
// }
// // 2. 首先获取分页采集的页数
// pageCount, err := spiderCore.GetPageCount(r)
// // 分页页数失败 则再进行一次尝试
// if err != nil {
// // 如果第二次获取分页页数依旧获取失败则关闭当前采集任务
// pageCount, err = spiderCore.GetPageCount(r)
// if err != nil {
// return err
// }
// }
// // 通过采集类型分别执行不同的采集方法
// switch s.CollectType {
// case system.CollectVideo:
// // 采集视频资源
// // 如果采集源参数中采集间隔参数大于500ms,则使用单线程采集
// if s.Interval > 500 {
// // 少量数据不开启协程
// for i := 1; i <= pageCount; i++ {
// collectFilm(s, h, i)
// // 执行一次采集后休眠指定时长
// time.Sleep(time.Duration(s.Interval) * time.Millisecond)
// }
// } else if pageCount <= config.MAXGoroutine*2 {
// // 少量数据不开启协程
// for i := 1; i <= pageCount; i++ {
// collectFilm(s, h, i)
// }
// } else {
// // 如果分页数量较大则开启协程
// ConcurrentPageSpider(pageCount, s, h, collectFilm)
// }
// // 视频数据采集完成后 对暂存数据进行处理和优化
// if s.Grade == system.MasterCollect {
// // 如果采集时长为负, (全量采集), 则在数据采集完成后为search表添加索引
// if h < 0 {
// // 全量采集时进行数据同步(仅保存)
// system.SyncMovieDetail(s.Id, s.Grade)
// system.AddSearchIndex()
// }
// // 采集时长在一定阈值内时执行redis数据同步 (存在则更新, 不存在则新增)
//
// // 开启图片同步
// if s.SyncPictures {
// system.SyncFilmPicture()
// }
// // 每次成功执行完都清理redis中的相关API接口数据缓存
// ClearCache()
// } else if s.Grade == system.SlaveCollect {
// // 如果采集时长为负, (全量采集), 则在数据采集完成后为search表添加索引
// if h < 0 {
// // 全量采集时进行数据同步
// system.SyncMovieDetail(s.Id, s.Grade)
// }
// }
//
// case system.CollectArticle, system.CollectActor, system.CollectRole, system.CollectWebSite:
// log.Println("暂未开放此采集功能!!!")
// return errors.New("暂未开放此采集功能")
// }
// log.Println("Spider Task Exercise Success")
// return nil
//}
//
//// collectFilm 影视详情采集 (单一源分页全采集)
//func collectFilm(s *system.FilmSource, h, pg int) {
// // 生成请求参数
// r := util.RequestInfo{Uri: s.Uri, Params: url.Values{}}
// // 设置分页页数
// r.Params.Set("pg", fmt.Sprint(pg))
// // 如果 h = -1 则进行全量采集
// if h > 0 {
// r.Params.Set("h", fmt.Sprint(h))
// }
// // 执行采集方法 获取影片详情list
// list, err := spiderCore.GetFilmDetail(r)
// if err != nil || len(list) <= 0 {
// // 添加采集失败记录
// fr := system.FailureRecord{OriginId: s.Id, OriginName: s.Name, Uri: s.Uri, CollectType: system.CollectVideo, PageNumber: pg, Hour: h, Cause: fmt.Sprintln(err), Status: 1}
// system.SaveFailureRecord(fr)
// log.Println("GetMovieDetail Error: ", err)
// return
// }
// // 通过采集站 Grade 类型, 执行不同的存储逻辑
// switch s.Grade {
// case system.MasterCollect:
// // 将数据缓存到redis中
// if err = system.MovieDetailCache(list); err != nil {
// log.Println("SaveDetails Error: ", err)
// }
// //break
// // 如果 采集时长 h 小于阈值, 则将主体数据缓存到redis
// //if h > 0 && h < config.FilmSaveCacheThreshold {
// // // 主站点 执行保存或更新
// // if err = system.BatchUpdateDetails(list); err != nil {
// // log.Println("SaveDetails Error: ", err)
// // }
// //} else {
// // // 主站点 从零开始只执行保存逻辑
// // if err = system.MovieDetailCache(list); err != nil {
// // log.Println("SaveDetails Error: ", err)
// // }
// //}
// // 如果主站点开启了图片同步, 则将图片url以及对应的mid存入ZSet集合中
// if s.SyncPictures {
// if err = system.SaveVirtualPic(conver.ConvertVirtualPicture(list)); err != nil {
// log.Println("SaveVirtualPic Error: ", err)
// }
// }
// case system.SlaveCollect:
// // 将采集数据缓存到redis中
// if err = system.SlaveDetailCache(s.Id, list); err != nil {
// log.Println("SaveDetails Error: ", err)
// }
// //if h > 0 && h < config.FilmSaveCacheThreshold {
// // // 附属站点 仅保存影片播放信息到mysql
// // if err = system.UpdateSitePlayList(s.Id, list); err != nil {
// // log.Println("SaveDetails Error: ", err)
// // }
// //} else {
// // // 附属站点 仅保存影片播放信息到mysql
// // if err = system.SlaveDetailCache(s.Id, list); err != nil {
// // log.Println("SaveDetails Error: ", err)
// // }
// //}
//
// }
//}
//
//// ConcurrentPageSpider 并发分页采集, 不限类型
//func ConcurrentPageSpider(capacity int, s *system.FilmSource, h int, collectFunc func(s *system.FilmSource, hour, pageNumber int)) {
// // 开启协程并发执行
// ch := make(chan int, capacity)
// waitCh := make(chan int)
// for i := 1; i <= capacity; i++ {
// ch <- i
// }
// close(ch)
// // 开启 MAXGoroutine 数量的协程, 如果分页页数小于协程数则将协程数限制为分页页数
// var GoroutineNum = config.MAXGoroutine
// if capacity < GoroutineNum {
// GoroutineNum = capacity
// }
// for i := 0; i < GoroutineNum; i++ {
// go func() {
// defer func() { waitCh <- 0 }()
// for {
// // 从channel中获取 pageNumber
// pg, ok := <-ch
// if !ok {
// break
// }
// // 执行对应的采集方法
// collectFunc(s, h, pg)
// }
// }()
// }
// for i := 0; i < GoroutineNum; i++ {
// <-waitCh
// }
//}
// ======================================================= 采集拓展内容 =======================================================
// SingleRecoverSpider 二次采集
func SingleRecoverSpider(fr *system.FailureRecord) {
// 获取采集参数中的h时间参数
h := fr.Hour
// 通过采集时长范围执行不同的采集方式
switch {
case fr.Hour > 168 && fr.Hour < 360:
case h > 2 && h < 360:
// 将此记录之后的所有同类采集记录变更为已重试
system.ChangeRecord(fr, 0)
// 如果采集的内容是 7~15 天之内更新的内容,则采集此记录之后的所有更新内容
// 获取采集参数h, 采集时长变更为 原采集时长 + 采集记录距现在的时长
h := fr.Hour + int(math.Ceil(time.Since(fr.CreatedAt).Hours()))
h := h + int(math.Ceil(time.Since(fr.CreatedAt).Hours()))
// 对当前所有已启用的站点 更新最新 h 小时的内容
AutoCollect(h)
case fr.Hour < 0, fr.Hour > 4320:
case h < 0, h > 4320:
// 将此记录状态修改为已重试
system.ChangeRecord(fr, 0)
// 如果采集的是 最近180天内更新的内容 或全部内容, 则只对当前一条记录进行二次采集
// 如果采集的是 最近180天以上更新的内容 或全部内容, 则只对当前一条记录进行二次采集
s := system.FindCollectSourceById(fr.OriginId)
collectFilm(s, fr.Hour, fr.PageNumber)
// 初始化请求参数
params, err := url.ParseQuery(fr.Params)
if err != nil {
log.Println("SingleRecoverSpider Error: ", err)
return
}
r := util.RequestInfo{Uri: fr.Uri, Params: params}
collectFilmRefine(s, r)
default:
// 其余范围,暂不处理
break
}
}
// FullRecoverSpider 扫描记录表中的失败记录, 并进行处理 (用于定时任务定期处理失败采集)
// FullRecoverSpider 扫描记录表中的失败记录, 并进行处理 (用于定时任务定期处理失败采集) 每天凌晨4点处理一次
func FullRecoverSpider() {
/*
获取待处理的记录数据
@@ -332,13 +628,14 @@ func FullRecoverSpider() {
// 遍历记录信息切片, 针对不同时长进行不同处理
for _, fr := range list {
// 处理时间参数, 计算偏移量
h := fr.Hour + int(math.Ceil(time.Since(fr.CreatedAt).Hours()))
switch {
case fr.Hour > 0 && fr.Hour < 4320:
// 将此记录之后的所有同类采集记录变更为已重试
system.ChangeRecord(&fr, 0)
// 如果采集的内容是 0~180 天之内更新的内容,则采集此记录之后的所有更新内容
// 获取采集参数h, 采集时长变更为 原采集时长 + 采集记录距现在的时长
h := fr.Hour + int(math.Ceil(time.Since(fr.CreatedAt).Hours()))
// 对当前所有已启用的站点 更新最新 h 小时的内容
AutoCollect(h)
case fr.Hour < 0, fr.Hour > 4320:
@@ -346,7 +643,16 @@ func FullRecoverSpider() {
system.ChangeRecord(&fr, 0)
// 如果采集的是 180天之前更新的内容 或全部内容, 则只对当前一条记录进行二次采集
s := system.FindCollectSourceById(fr.OriginId)
collectFilm(s, fr.Hour, fr.PageNumber)
// 初始化请求参数
params, err := url.ParseQuery(fr.Params)
if err != nil {
log.Println("SingleRecoverSpider Error: ", err)
break
}
// 对Hour参数进行偏移处理
r := util.RequestInfo{Uri: fr.Uri, Params: params}
r.Params.Set("h", strconv.Itoa(h))
collectFilmRefine(s, r)
default:
// 其余范围,暂不处理
break
@@ -355,6 +661,17 @@ func FullRecoverSpider() {
}
// FullSyncMovieDetail 同步全部redis中的影片信息
func FullSyncMovieDetail() {
// 获取采集站信息列表
for _, s := range system.GetCollectSourceList() {
// 如果当前站点为启用状态 则执行当前站点的影片信息同步
if s.State {
system.SyncMovieDetail(s.Id, s.Grade, 1)
}
}
}
// ======================================================= 公共方法 =======================================================
// CollectApiTest 测试采集接口是否可用

2
film/server/plugin/spider/SpiderCore.go
View File

@@ -89,7 +89,7 @@ func (jc *JsonCollect) GetFilmDetail(r util.RequestInfo) (list []system.MovieDet
r.Params.Set(`ac`, `detail`)
util.ApiGet(&r)
// 影视详情信息
detailPage := collect.FilmDetailLPage{}
var detailPage collect.FilmDetailLPage
//details := system.DetailListInfo{}
// 如果返回数据为空则直接结束本次循环
if len(r.Resp) <= 0 {

60
film/server/plugin/spider/SpiderCron.go
View File

@@ -3,10 +3,11 @@ package spider
import (
"errors"
"fmt"
"github.com/robfig/cron/v3"
"log"
"server/config"
"server/model/system"
"github.com/robfig/cron/v3"
)
var (
@@ -18,30 +19,8 @@ func CreateCron() *cron.Cron {
return cron.New(cron.WithSeconds())
}
// AddFilmUpdateCron 添加 指定站点的影片更新定时任务
func AddFilmUpdateCron(id, 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() {
// 通过创建任务时生成的 Id 获取任务相关数据
ft, err := system.GetFilmTaskById(id)
if err != nil {
log.Println("FilmCollectCron Exec Failed: ", err)
}
// 如果当前定时任务状态为开启则执行对应的采集任务
if ft.State && ft.Model == 1 {
// 对指定ids的资源站数据进行更新操作
BatchCollect(ft.Time, ft.Ids...)
}
// 任务执行完毕
log.Printf("执行一次定时任务: Task[%s]\n", ft.Id)
})
}
// AddAutoUpdateCron 添加 所有已启用站点的影片更新定时任务
func AddAutoUpdateCron(id, spec string) (cron.EntryID, error) {
// AddCron 添加定时任务
func AddCron(id, spec string) (cron.EntryID, error) {
// 校验 spec 表达式的有效性
if err := ValidSpec(spec); err != nil {
return -99, errors.New(fmt.Sprint("定时任务添加失败,Cron表达式校验失败: ", err.Error()))
@@ -53,24 +32,25 @@ func AddAutoUpdateCron(id, spec string) (cron.EntryID, error) {
log.Println("FilmCollectCron Exec Failed: ", err)
}
// 开启对系统中已启用站点的自动更新
if ft.State && ft.Model == 0 {
AutoCollect(ft.Time)
log.Println("执行一次自动更新任务")
if ft.State {
switch ft.Model {
case 0:
AutoCollect(ft.Time)
log.Println("执行一次已启用站点的影片自动更新任务")
case 1:
// 对指定ids的资源站数据进行更新操作
BatchCollect(ft.Time, ft.Ids...)
log.Println("执行一次指定站点的影片自动更新任务")
case 2:
FullRecoverSpider()
log.Println("执行一次采集失败的记录处理任务")
case 3:
FullSyncMovieDetail()
log.Println("执行一次所有站点的影片信息同步任务")
}
}
})
}
// 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 删除定时任务