task.go

package main

import (
	"code.wuban.net.cn/odysseus/twitter_syncer/swarm"
	"fmt"
	"log/slog"
	"sync"
	"time"
)

type Work struct {
	Lock sync.Mutex

	Task map[string]chan<- interface{}
	// userClient map[string]*Client
}

var Worker Work

func init() {
	Worker = Work{
		Task: make(map[string]chan<- interface{}),
		//userClient: make(map[string]*Client),
	}
}

func (w *Work) StopJob(req StopTaskReq) error {
	w.Lock.Lock()
	defer w.Lock.Unlock()

	key := req.User + "-" + req.TaskId + "-" + req.TaskType

	if v, ok := w.Task[key]; ok {
		close(v)
		delete(w.Task, key)

	} else {
		return fmt.Errorf("%s do not run", key)
	}

	return nil
}

func (w *Work) AddJob(t TaskJob) error {
	w.Lock.Lock()
	defer w.Lock.Unlock()

	key := t.UserId + "-" + t.TaskId + "-" + t.TaskType

	if _, ok := w.Task[key]; ok {
		return fmt.Errorf("%s has run", key)
	}

	done := w.RunJob(t)

	w.Task[key] = done

	return nil
}

type TimeAndFollowCount struct {
	Date        time.Time
	FollowCount int
}

func (w *Work) RunJob(t TaskJob) chan<- interface{} {

	slog.Info("exec job", "userid", t.UserId, "task type", t.TaskType, "task id", t.TaskId, "t.FollowerCount", t.FollowerCount)

	done := make(chan interface{})
	go func() {
		var cli *Client

		if t.TaskType == RetweetType {
			cli = NewRetweeterClient(t.Config)
		} else if t.TaskType == TweetLikingUsersType {
			cli = NewLikeClient(t.Config)
		}

		page := NewPageUsers(NewIdx(t.Idx))

		if t.TaskType == FollowType {

			cli := swarm.GetSwarm()

			secondTicker := time.NewTicker(time.Second * 3)
			fiveMinutesTicker := time.NewTicker(time.Minute * 1)
			halfHourTicker := time.NewTicker(time.Minute * 30)

			recordFc := make([]TimeAndFollowCount, 0, 100)

			for {
				select {
				case _, ok := <-done: // task is stopped by api.
					if !ok {
						return
					}
				case <-fiveMinutesTicker.C:
					slog.Info("case <-fiveMinutesTicker.C:")

					maybeFound := false // trigger followers request when maybeFound is true.

					for k, v := range recordFc {
						fmt.Println(k, v)
						if v.FollowCount != t.FollowerCount {
							maybeFound = true
						}
						if k == len(recordFc)-1 {
							t.FollowerCount = v.FollowCount
						}
					}

					fmt.Println(" t.FollowerCount", t.FollowerCount, "maybeFound", maybeFound)

					if maybeFound {
						fiveMinutesTicker.Reset(time.Minute * 3)
						halfHourTicker.Reset(time.Minute * 30)
						if err := Request(cli.GetFollowerList, page, t); err != nil {
							slog.Error(" page.Request", "task id", t.TaskId, "t.TaskType", t.TaskType, "err", err.Error())
							continue
						}

						if err := UpdateFollowerTaskCount(t.TaskId, t.TaskType, t.UserId, t.FollowerCount); err != nil {
							slog.Error("UpdateFollowerTaskCount", "err", err.Error(), "t.TaskId,", t.TaskId, "t.TaskType", t.TaskType, "t.UserId", t.UserId)
						}
					}

					recordFc = make([]TimeAndFollowCount, 0, 100)

					//recordFc = make(map[string]int)

				case <-secondTicker.C:
					fc, err := cli.GetFollowerCount(t.TaskId)

					if err != nil {
						slog.Error("TryProfileFollowerCount", "err", err.Error())
						continue
					}

					recordFc = append(recordFc, TimeAndFollowCount{
						Date:        time.Now(),
						FollowCount: fc,
					})
					secondTicker.Reset(time.Minute * 1)

					// 先用和like retweet一样的周期模式；

				case <-halfHourTicker.C:
					slog.Info("case <-halfHourTicker.C:")
					for k, v := range recordFc {

						fmt.Println(k, v)
						if k == len(recordFc)-1 {
							t.FollowerCount = v.FollowCount
						}
					}

					fmt.Println(" t.FollowerCount", t.FollowerCount)

					if err := Request(cli.GetFollowerList, page, t); err != nil {
						slog.Error(" page.Request", "task id", t.TaskId, "t.TaskType", t.TaskType, "err", err.Error())
						continue
					}

					if err := UpdateFollowerTaskCount(t.TaskId, t.TaskType, t.UserId, t.FollowerCount); err != nil {
						slog.Error("UpdateFollowerTaskCount", "err", err.Error(), "t.TaskId,", t.TaskId, "t.TaskType", t.TaskType, "t.UserId", t.UserId)
					}
					//}

					recordFc = make([]TimeAndFollowCount, 0, 100)

					// if err := Request(Follower, page, t); err != nil {
					// 	slog.Error(" page.Request", "task id", t.TaskId, "t.TaskType", t.TaskType, "err", err.Error())
					// 	continue
					// }
				}
			}
		}

		for {

			select {
			case _, ok := <-done:
				if !ok {
					return
				}
			default:

				var f req

				if t.TaskType == RetweetType {
					f = cli.Retweeters
				} else {
					f = cli.TweetLikingUsers
				}

				if err := Request(f, page, t); err != nil {
					slog.Error(" page.Request", "task id", t.TaskId, "t.TaskType", t.TaskType, "err", err.Error())
					continue
				}

			}
		}

	}()

	return done

}

func Request(f req, page *PageUsers, t TaskJob) error {

	users, err := page.Request(t.TaskId, "", f)

	if err != nil {
		return err
	}

	if err := InsertTaskRes(users, t.TaskType, t.TaskId); err != nil {

		for k, v := range users {
			fmt.Println(k, v.UserId, v.UserName)
		}

		slog.Error("InsertTaskRes", "task id", t.TaskId, "t.TaskType", t.TaskType, "len(users)", len(users), "err", err.Error())
	}

	slog.Info("InsertTaskRes", "task id", t.TaskId, "t.TaskType", t.TaskType, "len(users)", len(users))

	return nil

}