dw-spider/spider/controller.go

package spider

import (
	"context"
	"fmt"
	"log/slog"
	"time"

	productv1 "gitea.timerzz.com/kedaya_haitao/common/model/product"
	dw_sdk "gitea.timerzz.com/kedaya_haitao/dw-sdk"
	"gitea.timerzz.com/kedaya_haitao/pusher/kitex_gen/push"
	"gitea.timerzz.com/kedaya_haitao/pusher/rpc/pusher"
	"github.com/golang/glog"
	"gorm.io/gorm"
)

type Controller struct {
	ctx        context.Context
	client     *dw_sdk.Client
	updateTime time.Time // 上次抓取时间
	db         *gorm.DB
}

func NewController(client *dw_sdk.Client, db *gorm.DB) *Controller {
	c := &Controller{
		client: client,
		db:     db,
	}
	c.AutoMigrate()
	return c
}

func (c *Controller) AutoMigrate() {
	if err := c.db.AutoMigrate(&productv1.Product{}); err != nil {
		panic(err)
	}
}

func (c *Controller) Run(ctx context.Context) {
	c.ctx = ctx
	ticker := time.NewTicker(24 * time.Hour)
	if err := c.Crawl(); err != nil {
		slog.Error(err.Error())
	} else {
		slog.Info("抓取信息成功")
		c.updateTime = time.Now()
	}
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if err := c.Crawl(); err != nil {
				slog.Error(err.Error())
			} else {
				slog.Info("抓取信息成功")
				c.updateTime = time.Now()
			}
		}

	}
}

func (c *Controller) Crawl() error {
	slog.Info("开始抓取信息")
	var products []productv1.Product
	c.db.Where("dw_sku_id!=(-1) or dw_sku_id is null").FindInBatches(&products, 50, func(tx *gorm.DB, batch int) error {
		for _, product := range products {
			if product.DWSkuId == 0 {
				p, err := c.fillDWSkuId(product)
				if err != nil {
					slog.Warn(err.Error())
				}
				if err = c.db.Select("dw_sku_id", "dw_spu_id").Updates(&p).Error; err != nil {
					slog.Warn(fmt.Sprintf("更新pid: %s sku失败：%v", p.Pid, err))
				}
				product = p
				if product.DWSkuId <= 0 {
					continue
				}
			}
			resp, err := c.client.ConsignBidClient().LowestPrice(product.DWSkuId)
			if err != nil {
				slog.Warn(fmt.Sprintf("请求sku:%d最低价失败:%v", product.DWSkuId, err))
				continue
			}
			if resp.Code != 200 {
				slog.Warn(fmt.Sprintf("请求sku: %d，得物返回错误：%s", product.DWSkuId, resp.Msg))
				continue
			}
			if len(resp.Data.Items) == 0 {
				slog.Warn(fmt.Sprintf("未获取到sku: %d 的最低价,数量为0", product.DWSkuId))
				continue
			}
			product.DWPrice = float64(resp.Data.Items[0].LowestPrice)/100*0.925 - 46 - 8.5
			product.CalRate()
			if err = c.db.Select("rate", "dw_price").Updates(&product).Error; err != nil {
				slog.Warn(fmt.Sprintf("更新pid:%s的数据库失败：%v", product.Pid, err))
			}
			if product.Rate >= 15 {
				c.Push(fmt.Sprintf("%s 的收益率达到%.2f%%", product.Pid, product.Rate),
					fmt.Sprintf("%s 的成本为%.2f, 得物收益为%.2f, 收益率为%.2f%%， 链接：%s", product.Pid, product.CNYPrice, product.DWPrice, product.Rate, product.Link))
			}
			continue
		}
		slog.Info("已处理50个数据")
		return nil
	})
	slog.Info("抓取完毕")
	return nil
}

// 填充得物
func (c *Controller) fillDWSkuId(product productv1.Product) (productv1.Product, error) {
	resp, err := c.client.ArticleService().BatchArticleNumber([]string{product.Pid})
	if err != nil {
		return product, fmt.Errorf("请求pid:%s出错：%v", product.Pid, err)
	}
	if resp.Code != 200 {
		product.DWSkuId = -1
		return product, fmt.Errorf("pid:%s 得物返回错误：%s", product.Pid, resp.Msg)
	}
	if len(resp.Data) == 0 {
		product.DWSkuId = -1
		return product, fmt.Errorf("pid:%s 得物返回spu为0", product.Pid)
	}
	var spu *dw_sdk.Spu
	for _, s := range resp.Data {
		if s.BrandId == dw_sdk.BrandId_Coach {
			spu = &s
		}
	}
	if spu == nil {
		product.DWSkuId = -1
		return product, fmt.Errorf("未获取到pid : %s 的coach spu信息:%v", product.Pid, resp.Data)
	}
	if len(spu.Skus) == 0 {
		product.DWSkuId = -1
		return product, fmt.Errorf("未获取到pid : %s 的coach sku信息:%v", product.Pid, spu)
	}
	product.DWSpuId, product.DWSkuId = spu.SpuId, spu.Skus[0].SkuId
	return product, nil
}

func (c *Controller) Push(title, content string) {
	resp, err := pusher.Push(c.ctx, &push.PushReq{
		Ids:     []int64{2},
		Title:   title,
		Content: content,
	})
	if err != nil {
		glog.Errorf("消息推送失败：%v", err)
	}
	if resp.Code != 0 {
		glog.Errorf("消息推送失败:%s", resp.Msg)
	}
}