meilisearch 搜索
大约 4 分钟
package movie_cloud
import (
"errors"
"fmt"
"goravel/app/models"
"log"
"sync"
"github.com/goravel/framework/facades"
"github.com/meilisearch/meilisearch-go"
SpiderFacades "github.com/orangbus/spider/facades"
"github.com/orangbus/spider/pkg/movie_spider"
"github.com/spf13/cast"
)
type MovieCloud struct {
client meilisearch.ServiceManager
indexName string
taskCh chan models.MovieApis // 通知消息队列
msgs chan Message // 通知消息队列
scoutCh chan []movie_spider.MovieItem // 通知消息队列
mu sync.Mutex
Status bool
spiderLimit int
}
var taskList []models.MovieApis
type Message struct {
Title string
Content string
Type int
}
func NewMovieCloud(host, key string, index_name string) *MovieCloud {
client := meilisearch.New(host, meilisearch.WithAPIKey(key))
return &MovieCloud{
client: client,
indexName: index_name,
msgs: make(chan Message, 100),
taskCh: make(chan models.MovieApis, 100),
scoutCh: make(chan []movie_spider.MovieItem, 100),
mu: sync.Mutex{},
Status: false,
spiderLimit: 10,
}
}
func (s *MovieCloud) Start() {
go func(s *MovieCloud) {
for {
select {
case msg, ok := <-s.msgs:
if !ok {
log.Print("消息通知已退出")
return
}
log.Printf("%s:%s", msg.Title, msg.Content)
}
}
}(s)
//s.CheckTask()
log.Print("后台采集功能已开启")
for {
select {
case task, ok := <-s.taskCh:
if !ok {
log.Print("采集任务已经退出")
return
}
s.SpiderToScout(task)
s.RemoveTask(task)
}
}
}
func (s *MovieCloud) Check() error {
app := s.client
if app == nil {
return errors.New("scout实例化失败")
}
return nil
}
func (s *MovieCloud) AddDocument(index_name, primaryKey string, data map[string]any) (*meilisearch.TaskInfo, error) {
var document []map[string]interface{}
document = append(document, data)
return s.client.Index(s.getIndexName(index_name)).AddDocuments(document, &primaryKey)
}
func (s *MovieCloud) AddDocuments(index_name, primaryKey string, dataList []map[string]interface{}) (*meilisearch.TaskInfo, error) {
return s.client.Index(s.getIndexName(index_name)).AddDocuments(dataList, &primaryKey)
}
func (s *MovieCloud) Search(api_id string, keywords string, page, limit int64, Filter ...[]string) (*meilisearch.SearchResponse, error) {
options := meilisearch.SearchRequest{
Page: page,
Limit: limit,
//Sort: []string{"vod_time:desc"},
AttributesToSearchOn: []string{
"vod_title", "type_name", "vod_content", "vod_tag", "vod_class", "vod_actor", "vod_author", "vod_director", "vod_time",
},
}
if len(Filter) > 0 {
where := [][]string{}
where = append(where, Filter[0])
options.Filter = where
}
return s.client.Index(s.getIndexName(api_id)).Search(keywords, &options)
}
func (s *MovieCloud) getIndexName(tag string) string {
return s.indexName + "_" + tag
}
func (s *MovieCloud) CreateIndex(index_id, primaryKey string) error {
_, err := s.client.CreateIndex(&meilisearch.IndexConfig{
index_id, primaryKey,
})
return err
}
func (s *MovieCloud) DeleteIndex(index_id string) error {
_, err := s.client.DeleteIndex(s.getIndexName(index_id))
return err
}
func (s *MovieCloud) ListMessage() {
}
// 采集 -> 解析数据 -> 同步到 meilisearch
func (s *MovieCloud) SpiderToScout(api models.MovieApis) {
if api.Status != 1 {
s.notice("采集任务", fmt.Sprintf("%s 状态已关闭", api.Name), 1)
return
}
hour := 0
if api.Day > 0 {
hour = api.Day * 24
} else if api.IndexName != "" {
hour = 24
}
response, err := SpiderFacades.Spider().BaseUrl(api.URL).SetHour(hour).GetList(1, 10)
if err != nil {
s.notice("采集任务", fmt.Sprintf("%s请求错误：%s", api.Name, err.Error()), 1)
return
}
s.scoutCh <- response.List
if response.PageCount <= 1 {
return
}
var wg sync.WaitGroup
spiderCh := make(chan []movie_spider.MovieItem, 50)
wg.Add(2)
go func() {
defer wg.Done()
var wg2 sync.WaitGroup
semaphore := make(chan struct{}, s.spiderLimit)
for start := 2; start <= response.PageCount; start++ {
wg2.Add(1)
go func() {
defer wg2.Done()
semaphore <- struct{}{}
defer func() { <-semaphore }() // 释放信号量
response, err = SpiderFacades.Spider().BaseUrl(api.URL).SetHour(hour).SetAcVideoList().GetList(start, 10)
if err != nil {
s.notice("采集任务", fmt.Sprintf("%s请求错误：%s", api.Name, err.Error()), 1)
return
}
s.notice("采集任务", fmt.Sprintf("%s正在采集：%d/%d ,累计条数：%d", api.Name, start, response.PageCount, response.Total), 1)
spiderCh <- response.List
}()
}
wg2.Wait()
close(spiderCh)
}()
// 同步
go func() {
defer wg.Done()
list := []map[string]any{}
indexName := cast.ToString(api.ID) // movies_1
for {
msgs, ok := <-spiderCh
if !ok {
break
}
for _, v := range msgs {
doc := map[string]interface{}{
"id": v.VodID,
"vod_id": v.VodID,
"vod_name": v.VodName,
"type_id": v.TypeID,
"type_id_1": v.TypeID1,
"group_id": v.GroupID,
"vod_sub": v.VodSub,
"vod_en": v.VodEn,
"vod_status": v.VodStatus,
"vod_letter": v.VodLetter,
"vod_color": v.VodColor,
"vod_tag": v.VodTag,
"vod_class": v.VodClass,
"vod_pic": v.VodPic,
"vod_pic_thumb": v.VodPicThumb,
"vod_pic_slide": v.VodPicSlide,
"vod_actor": v.VodActor,
"vod_director": v.VodDirector,
"vod_remarks": v.VodRemarks,
"vod_pubdate": v.VodPubdate,
"vod_total": v.VodTotal,
"vod_weekday": v.VodWeekday,
"vod_area": v.VodArea,
"vod_lang": v.VodLang,
"vod_year": v.VodYear,
"vod_version": v.VodVersion,
"vod_state": v.VodState,
"vod_author": v.VodAuthor,
"vod_hits": v.VodHits,
"vod_hits_day": v.VodHitsDay,
"vod_hits_week": v.VodHitsWeek,
"vod_hits_month": v.VodHitsMonth,
"vod_duration": v.VodDuration,
"vod_score": v.VodScore,
"vod_score_all": v.VodScoreNum,
"vod_time": v.VodTime,
"vod_douban_id": v.VodDoubanID,
"vod_douban_score": v.VodDoubanScore,
"vod_content": v.VodContent,
"vod_play_from": v.VodPlayFrom,
"vod_play_server": v.VodPlayServer,
"vod_play_note": v.VodPlayNote,
"vod_play_url": v.VodPlayURL,
"type_name": v.TypeName,
}
list = append(list, doc)
}
if len(list) >= 100 {
_, err = s.AddDocuments(indexName, "vod_id", list)
if err != nil {
s.notice("采集同步任务", "采集同步错误："+err.Error(), 1)
continue
}
// 清空列表
list = list[0:0]
}
}
if len(list) > 0 {
_, err = s.AddDocuments(indexName, "vod_id", list)
if err != nil {
s.notice("采集同步任务", "采集同步错误："+err.Error(), 1)
}
}
s.notice("同步任务", fmt.Sprintf("%s 同步完成", api.Name), 1)
}()
wg.Wait()
// 更新索引
if api.IndexName == "" {
indexName := s.getIndexName(cast.ToString(api.ID))
_, err = s.client.Index(indexName).UpdateSettings(&meilisearch.Settings{
RankingRules: nil,
DistinctAttribute: nil,
Dictionary: nil, // 自定义分词
SearchableAttributes: []string{
"vod_name", "type_name", "vod_content", "vod_tag", "vod_class",
"vod_actor", "vod_author", "vod_director", "vod_time",
}, // 可搜索的字段
FilterableAttributes: []string{"type_id", "vod_id", "vod_name"}, // 查询条件
SortableAttributes: []string{"vod_time", "vod_score", "id", "vod_id"}, // 可排序字段
})
if err != nil {
s.notice("索引更新失败", api.Name)
}
// 更新数据库索引名称
api.IndexName = indexName
}
// 获取的文章总数
res, err := SpiderFacades.Spider().Get(api.URL)
if err != nil {
api.MovieCount = response.Total
} else {
api.MovieCount = res.Total
}
api.Day = 0 // 重置采集时长
if err := facades.Orm().Query().Save(&api); err != nil {
s.notice("索引设置失败", api.Name)
}
// 第一次更新分类
if api.IndexName == "" && len(res.Class) > 0 {
cates := []models.MovieCate{}
for _, v := range res.Class {
cates = append(cates, models.MovieCate{
ApiId: api.ID,
TypeID: v.TypeID,
TypePid: v.TypePid,
TypeName: v.TypeName,
})
}
if err2 := facades.Orm().Query().Create(&cates); err2 != nil {
s.notice("保存数据失败", fmt.Sprintf("%s 分类添加失败：%s", api.Name, err2.Error()))
}
}
s.notice("采集任务", fmt.Sprintf("%s 采集完成", api.Name))
}
func (s *MovieCloud) AddTask(api models.MovieApis) {
s.mu.Lock()
defer s.mu.Unlock()
if len(taskList) == 0 {
taskList = append(taskList, api)
} else {
exist := true
for _, item := range taskList {
if item.ID == api.ID {
exist = false
break
}
}
if exist {
taskList = append(taskList, api)
}
}
s.CheckTask()
}
func (s *MovieCloud) RemoveTask(api models.MovieApis) {
s.mu.Lock()
defer s.mu.Unlock()
for i, item := range taskList {
if item.ID == api.ID {
taskList = append(taskList[:i], taskList[i+1:]...)
break
}
}
s.CheckTask()
}
func (s *MovieCloud) CheckTask() {
if len(taskList) == 0 {
return
}
s.taskCh <- taskList[0]
s.notice("任务", fmt.Sprintf("开始采集任务：%s", taskList[0].Name))
}
func (s *MovieCloud) notice(title, content string, msgType ...int) {
Type := 0
if len(msgType) > 0 {
Type = msgType[0]
}
s.msgs <- Message{
Title: title,
Content: content,
Type: Type,
}
}