Skip to content

T
twitter_syncer
Commit 287cebb3 authored by vicotor's avatar vicotor
Browse files
Options

remove unused file

parent 9c239c18
Hide whitespace changes
Inline Side-by-side
Showing with 0 additions and 1417 deletions
_stream.go_ deleted 100644 → 0
View file @ 9c239c18
package main
import (
"container/list"
b64 "encoding/base64"
"encoding/json"
"fmt"
"log/slog"
"strings"
"time"
twitterscraper "github.com/imperatrona/twitter-scraper"
)
func newSync(done <-chan interface{}) (chan<- taskInterface, error) {
//done := make(<-chan interface{})
connStream := make(chan taskInterface, 1000)
tasks, err := GetTasksIdx()
if err != nil {
return nil, err
}
for _, task := range tasks {
connStream <- task
}
idxStream, listStream := TaskIdx(done, connStream)
usersStream := BackListToQueue(done, listStream)
if err := InsertOrUpdateUsers(done, usersStream); err != nil {
return nil, err
}
resStream, err := InitResource()
if err != nil {
return nil, err
}
availRes := AvailableResource(done, resStream)
taskFetchStream, resTailStream := TaskImplement(done, idxStream, availRes)
ConnectTailTask(done, connStream, taskFetchStream)
ConnectTailResource(done, resStream, resTailStream)
//select {}
return connStream, nil
}
func TaskIdx(done <-chan interface{}, inStream <-chan taskInterface) (<-chan taskInterface, <-chan TaskIdAndList) {
outStream := make(chan taskInterface, 1000)
outListStream := make(chan TaskIdAndList, 1)
go func() {
for {
select {
case <-done:
return
case task := <-inStream:
slog.Info("TaskIdx", "task.TaskType()", task.TaskType(), "task.ID()", task.ID(), "task.UserName()", task.UserName(), "task.InitIdx()", task.InitIdx())
if !task.InitIdx() {
list, ok := task.UpdateIdx()
if ok {
select {
case <-done:
return
case outListStream <- TaskIdAndList{
TaskId: task.ID(),
List: list,
TaskType: task.TaskType(),
}:
}
}
}
select {
case <-done:
return
case outStream <- task:
}
}
}
}()
return outStream, outListStream
}
type TaskIdAndList struct {
TaskType string
TaskId string
List *list.List
}
type TaskIdAndProfiles struct {
TaskType string
TaskId string
Profiles []Profile
}
func BackListToQueue(done <-chan interface{}, inStream <-chan TaskIdAndList) <-chan TaskIdAndProfiles {
outStream := make(chan TaskIdAndProfiles, 1)
go func() {
for {
select {
case <-done:
return
case users, ok := <-inStream:
if ok == false {
return
}
c := 0
// if c < 100 {
// c = c + 1
res := make([]Profile, 0, users.List.Len())
for e := users.List.Front(); e != nil; e = e.Next() {
if user, ok := e.Value.(Profile); ok {
//fmt.Printf("The data is a string: %s\n", str)
res = append(res, user)
c = c + 1
if c%100 == 0 {
fmt.Println("BackListToQueue", "len(inStream)", len(inStream), "len(outStream)", len(outStream), "len(res)", len(res))
select {
case <-done:
return
case outStream <- TaskIdAndProfiles{
Profiles: res,
TaskId: users.TaskId,
TaskType: users.TaskType,
}:
res = make([]Profile, 0, users.List.Len())
}
}
}
}
fmt.Println("BackListToQueue", "len(inStream)", len(inStream), "len(outStream)", len(outStream), "len(res)", len(res))
select {
case <-done:
return
case outStream <- TaskIdAndProfiles{
Profiles: res,
TaskId: users.TaskId,
TaskType: users.TaskType,
}:
}
// } else {
// c = 0
// }
}
}
}()
return outStream
}
func InsertOrUpdateUsers(done <-chan interface{}, inStream <-chan TaskIdAndProfiles) error {
// client, err := supabase.NewClient(API_URL, API_KEY, nil)
// if err != nil {
// return fmt.Errorf("cannot initalize client: %v .", err)
// }
go func() {
for {
select {
case <-done:
return
case users, ok := <-inStream:
if ok == false {
return
}
//rows := make([]map[string]string, 0, len(users.Profiles))
//res, c, err := client.From("followers").Insert(rows, true, "", "representation", "").Execute()
var res []byte
var err error
if users.TaskType == FollowType {
rows := make([]Follower, 0, len(users.Profiles))
for _, user := range users.Profiles {
sDec, _ := b64.StdEncoding.DecodeString(user.UserID)
userId, _ := strings.CutPrefix(string(sDec), "User:")
row := Follower{
Follower: userId,
UserName: user.Username,
UserId: users.TaskId,
}
rows = append(rows, row)
}
res, _, err = client.From("followers").Insert(rows, true, "", "representation", "").Execute()
}
if users.TaskType == RetweetType {
rows := make([]Retweeter, 0, len(users.Profiles))
for _, user := range users.Profiles {
sDec, _ := b64.StdEncoding.DecodeString(user.UserID)
userId, _ := strings.CutPrefix(string(sDec), "User:")
row := Retweeter{
TweetId: users.TaskId, //string `json:"tweet_id"`
RetweeterId: userId, //string `json:"retweeter_id"`
RetweeterUserName: user.Username, //string `json:"retweeter_username"`
// Follower: userId,
// UserName: user.Username,
// UserId: users.TaskId,
}
rows = append(rows, row)
}
res, _, err = client.From("retweeters").Insert(rows, true, "", "representation", "").Execute()
}
if err != nil {
slog.Error("insert into followers or retweeters ", err)
for _, user := range users.Profiles {
usersAsJson, err := json.Marshal(user)
if err != nil {
slog.Error("insert into followers or retweeters json.Marshal", err)
continue
}
sDec, _ := b64.StdEncoding.DecodeString(user.UserID)
userId, _ := strings.CutPrefix(string(sDec), "User:")
slog.Error("insert into followers or retweeters error", string(usersAsJson), userId)
}
} else {
slog.Info("insert into followers or retweeters", string(res), err)
}
fmt.Println("InsertOrUpdateUsers", "len(inStream)", len(inStream))
}
}
}()
return nil
}
type ScraperTimer struct {
Scraper *twitterscraper.Scraper
Timer time.Timer
}
func InitResource() (chan ScraperTimer, error) {
outStream := make(chan ScraperTimer, 1000)
accounts, err := TwitterAccountFromDB()
if err != nil {
return nil, err
}
for _, v := range accounts {
fmt.Println(v.User, v.PassWd)
scraper, err := InitScraper(v.User, v.PassWd)
if err != nil {
return nil, err
}
newScraperTimer := ScraperTimer{
Scraper: scraper,
Timer: *time.NewTimer(0),
}
//
outStream <- newScraperTimer
}
fmt.Println("twitter init ok")
return outStream, nil
}
func AvailableResource(done <-chan interface{}, inStream <-chan ScraperTimer) <-chan ScraperTimer {
outStream := make(chan ScraperTimer, 1000)
go func() {
for {
select {
case <-done:
return
case scraper := <-inStream:
//TODO
// 1. 携带web服务器的rate limit 做判断；
// 2. 校验 是否可用
slog.Info("AvailableResource------------------------------")
<-scraper.Timer.C
scraper.Timer = *time.NewTimer(60 * time.Second)
//scraper.Timer.Reset(30 * time.Second)
outStream <- scraper
}
}
}()
return outStream
}
func TaskImplement(done <-chan interface{}, inTaskStream <-chan taskInterface, inResourceStream <-chan ScraperTimer) (<-chan taskInterface, <-chan ScraperTimer) {
taskOutStream := make(chan taskInterface, 1000)
scraperOutStream := make(chan ScraperTimer, 1000)
go func() {
for {
select {
case <-done:
return
case task := <-inTaskStream:
select {
case <-done:
return
case res := <-inResourceStream:
slog.Info("TaskImplement", "task.TaskType()", task.TaskType(), "task.ID()", task.ID(), "task.UserName()", task.UserName(), "len(inResourceStream)", len(inResourceStream))
if err := task.Fetch(res.Scraper); err != nil {
slog.Error("task.Fetch", "err", err.Error())
}
select {
case <-done:
return
case taskOutStream <- task:
}
select {
case <-done:
return
case scraperOutStream <- res:
}
}
}
}
}()
return taskOutStream, scraperOutStream
}
func ConnectTailTask(done <-chan interface{}, beginStream chan<- taskInterface, tailStream <-chan taskInterface) {
go func() {
for {
select {
case <-done:
return
case beginStream <- <-tailStream:
}
}
}()
}
func ConnectTailResource(done <-chan interface{}, beginStream chan<- ScraperTimer, tailStream <-chan ScraperTimer) {
go func() {
for {
select {
case <-done:
return
case beginStream <- <-tailStream:
}
}
}()
}
_stream_api_test.go_ deleted 100644 → 0
View file @ 9c239c18
package main
import (
"encoding/json"
"testing"
)
func TestMe(t *testing.T) {
//cli := NewOAuth2Client()
cli := NewOAuth2Client0816()
me, err := cli.Me()
if err != nil {
t.Error(err)
}
t.Log("Me")
meAsJson, err := json.Marshal(me)
if err != nil {
t.Error(err)
}
t.Log(string(meAsJson))
}
// func TestCreateTweet(t *testing.T) {
// cli := NewOAuth2Client()
// //cli := NewOAuth2ClientSelf()
// res, err := cli.CreateTweet("Hello world!")
// if err != nil {
// t.Error(err)
// }
// t.Log("Me")
// meAsJson, err := json.Marshal(res)
// if err != nil {
// t.Error(err)
// }
// t.Log(string(meAsJson))
// }
// func TestFollowers(t *testing.T) {
// //cli := NewOAuth2Client()
// //cli = NewOAuth2ClientSelf()
// cli := NewOAuth2Client0816()
// users, rateLinmit, next, err := cli.Followers("1783145144700874752", "")
// if err != nil {
// t.Error(err)
// }
// t.Log("next", next)
// meAsJson, err := json.Marshal(users)
// if err != nil {
// t.Error(err)
// }
// t.Log(string(meAsJson))
// rAsJson, err := json.Marshal(rateLinmit)
// if err != nil {
// t.Error(err)
// }
// t.Log(string(rAsJson))
// }
//1800805503066661056
func TestLike(t *testing.T) {
//cli := NewOAuth2Client()
cli := NewOAuth2Client0817()
users, next, rateLimit, err := cli.TweetLikedUsers("1800805503066661056", "")
if err != nil {
t.Error(err)
}
t.Log("next", next)
for k, v := range users {
t.Logf("k %v v %v \n", k, v)
}
rAsJson, err := json.Marshal(rateLimit)
if err != nil {
t.Error(err)
}
t.Log(string(rAsJson))
}
func TestRetweet(t *testing.T) {
//cli := NewOAuth2Client()
cli := NewOAuth2Client0816()
users, next, rateLimit, err := cli.Retweeters("1800805503066661056", "")
if err != nil {
t.Error(err)
}
t.Log("next", next)
for k, v := range users {
t.Logf("k %v v %v \n", k, v)
}
rAsJson, err := json.Marshal(rateLimit)
if err != nil {
t.Error(err)
}
t.Log(string(rAsJson))
}
// TweetSearchAll
func TestTweetSearchAll(t *testing.T) {
cli := NewOAuth2Client()
//cli := NewOAuth2Client2()
users, err := cli.TweetTweetLookup("1800805503066661056")
if err != nil {
t.Error(err)
}
rAsJson, err := json.Marshal(users)
if err != nil {
t.Error(err)
}
t.Log(string(rAsJson))
}
_stream_basic.go_ deleted 100644 → 0
View file @ 9c239c18
package main
func Controller(done <-chan interface{}, inStream <-chan taskInterface) (<-chan taskInterface, <-chan taskInterface) {
return nil, nil
}
// func Idx(done <-chan interface{}, inStream <-chan taskInterface) (<-chan taskInterface, <-chan taskInterface) {
// return nil, nil
// }
func Scheduler(done <-chan interface{}, inStream <-chan taskInterface) (<-chan taskInterface, <-chan taskInterface) {
return nil, nil
}
func Request(done <-chan interface{}, inStream <-chan taskInterface) (<-chan taskInterface, <-chan taskInterface) {
return nil, nil
}
func ConnectTask(done <-chan interface{}, beginStream chan<- taskInterface, tailStream <-chan taskInterface) {
go func() {
for {
select {
case <-done:
return
case beginStream <- <-tailStream:
}
}
}()
}
_stream_project.go_ deleted 100644 → 0
View file @ 9c239c18
package main
import (
"time"
"github.com/g8rswimmer/go-twitter/v2"
)
type Token struct {
Bearer string
ConsumerKey string
ConsumerSecret string
AccessToken string
AccessSecret string
//MonthRateLimit
}
// const FollowRateLimit
type Fee struct {
RateLimits map[string]twitter.RateLimit
Apps int
}
// type RateLimit struct {
// Duration time.Duration
// Num int
// }
type Project struct {
name string
token Token
fee Fee
}
var rateLimit map[string]twitter.RateLimit
const ProjectMonthKey = "ProjectMonth"
// const RetweetPerApp = "RetweetPerApp"
// const RetweetPerUser = "RetweetPerUser"
func getMonthEnd(t time.Time) time.Time {
//func getMonthStartAndEnd(t time.Time) (time.Time, time.Time) {
year, month, _ := t.Date()
start := time.Date(year, month, 1, 0, 0, 0, 0, t.Location())
end := start.AddDate(0, 1, -1).Add(23 * time.Hour).Add(59 * time.Minute).Add(59 * time.Second).Add(999 * time.Nanosecond)
//return start, end
return end
}
func init() {
rateLimit = make(map[string]twitter.RateLimit)
rateLimit[ProjectMonthKey] = twitter.RateLimit{
Limit: 10000,
Reset: twitter.Epoch(getMonthEnd(time.Now()).Unix()),
}
}
_stream_test.go_ deleted 100644 → 0
View file @ 9c239c18
package main
import (
"fmt"
"testing"
"time"
)
func add1[T int | float32 | float64](a, b T) T {
c := a + b
return c
}
func TestChan(t *testing.T) {
// 创建一个可以容纳任何类型的通道
ch := make(chan any, 10)
// 使用通道
ch <- 123 // 可以存放int类型
number := <-ch // 读取通道，number类型为any
t.Logf("type %T %v \n", number, number)
ch <- "hello" // 可以存放string类型
number = <-ch // 读取通道，number类型为any
t.Logf("type %T %v \n", number, number)
}
/*
// 定义
type Struct1 [T string|int|float64] struct {
Title string
Content T
}
// 实例化
test：= Struct1[string]{
Title:"aaa"
Content :"bbb"
}
// 定义
type MyMap[K string | int, V string | int] map[K]V
// 实例化
test := MyMap[string, int]{
"1": 1,
"2": 2,
}
fmt.Println(test["1"])
type gchan[T any] chan T
实例化
strchan := make(gchan[string])
*/
//type gchan[T FollowTask | RetweetTask] chan T
//NewTask
//func InitTask(done <-chan interface{}, instream <-chan any) {
func TestTimer(t *testing.T) {
timer1 := time.NewTimer(2 * time.Second)
time.Sleep(4 * time.Second)
<-timer1.C
fmt.Println("Timer 1 fired")
timer1.Reset(2 * time.Second)
time.Sleep(4 * time.Second)
<-timer1.C
fmt.Println("Timer 1 fired")
}
_stream_type.go_ deleted 100644 → 0
View file @ 9c239c18
package main
import "container/list"
type TaskJob struct {
//URL string
TaskTypeStr string
TaskId string
UserNameStr string
Next string
Idx []UserTaskIdAndTime
//[]RetweeterId
// backPushPop = list.New()
backPushPop *list.List
NewIdx []UserTaskIdAndTime
Res Users
}
// func (r *RetweetTask) URl() string {
// return r.URL
// }
func NewTaskJob(Id, userName, taskType string) *TaskJob {
return &TaskJob{
TaskTypeStr: taskType,
TaskId: Id,
UserNameStr: userName,
backPushPop: list.New(),
Idx: make([]UserTaskIdAndTime, 0),
}
}
_task.go_ deleted 100644 → 0
View file @ 9c239c18
package main
import (
"strings"
"time"
b64 "encoding/base64"
twitterscraper "github.com/imperatrona/twitter-scraper"
)
// task
// func sync() error {
// idxs, err := GetIndexList()
// if err != nil {
// return err
// }
// done := make(chan interface{})
// defer close(done)
// backListStream, err := FollowersToBackList(done, idxs)
// if err != nil {
// return err
// }
// usersStream := BackListToQueue(done, backListStream)
// if err := InsertOrUpdateUsers(done, usersStream); err != nil {
// return err
// }
// select {}
// }
// func sync() error {
// idxs, err := GetTasksFollowIdx()
// //idxs, err := GetIndexList()
// if err != nil {
// return err
// }
// done := make(chan interface{})
// defer close(done)
// backListStream, err := FollowersToBackList(done, idxs)
// if err != nil {
// return err
// }
// _ = backListStream
// return nil
// }
func tasks(done <-chan interface{}, inStream <-chan TaskParam) (<-chan TaskParam, <-chan TaskParam) {
outResource := make(chan TaskParam)
outDb := make(chan TaskParam)
go func() {
}()
return outResource, outDb
}
type TwitterUserPasswd struct {
User string
PassWd string
}
var twitterUsers []TwitterUserPasswd
func Resource() ([]*twitterscraper.Scraper, error) {
users := make([]struct {
User string
PassWd string
}, 0)
users = append(users, struct {
User string
PassWd string
}{
User: "Wade_Leeeee",
PassWd: "923881393time",
})
res := make([]*twitterscraper.Scraper, 0, len(users))
for _, v := range users {
scraper, err := InitScraper(v.User, v.PassWd)
if err != nil {
return nil, err
}
res = append(res, scraper)
}
return res, nil
}
func InitScraper(user, password string) (*twitterscraper.Scraper, error) {
scraper3 := twitterscraper.New()
//err := scraper3.Login("Wade_Leeeee", "923881393time")
if err := scraper3.Login(user, password); err != nil {
return nil, err
}
// Deserialize from JSON
// var cookies []*http.Cookie
// f, _ := os.Open("cookies.json")
// json.NewDecoder(f).Decode(&cookies)
// scraper3.SetCookies(cookies)
// if !scraper3.IsLoggedIn() {
// panic("Invalid cookies")
// }
return scraper3, nil
}
type TaskParam struct {
Follow Follow
}
type Retweet struct {
Enable bool
User string
Next string
Res chan interface{}
}
type Follow struct {
Enable bool
User string
Next string
Res chan Users
}
func ImplementTask(done <-chan interface{}, scraper3 *twitterscraper.Scraper, tickerD time.Duration, inStream <-chan TaskParam) {
go func() {
ticker := time.NewTicker(tickerD)
for {
select {
case <-done:
return
case <-ticker.C:
select {
case <-done:
return
case req := <-inStream:
if req.Follow.Enable {
followers, err := FetchFollowers(scraper3, req.Follow.User, req.Follow.Next)
followers.Error = err
select {
case <-done:
return
case req.Follow.Res <- followers:
}
}
}
}
}
}()
}
func FetchFollowers(scraper *twitterscraper.Scraper, user, next string) (Users, error) {
users, newNext, err := scraper.FetchFollowers(user, 20, next)
if err != nil {
return Users{}, err
}
usersWithUserNumber := make([]Profile, 0, len(users))
for _, v := range users {
sDec, _ := b64.StdEncoding.DecodeString(v.UserID)
userId, _ := strings.CutPrefix(string(sDec), "User:")
item := Profile{
UserIdAsNumber: userId,
Profile: v,
}
usersWithUserNumber = append(usersWithUserNumber, item)
}
res := Users{
Profiles: usersWithUserNumber,
Current: next,
Next: newNext,
}
return res, nil
}
_type.go_ deleted 100644 → 0
View file @ 9c239c18
package main
import (
"container/list"
b64 "encoding/base64"
"log/slog"
"strings"
twitterscraper "github.com/imperatrona/twitter-scraper"
)
/*
users {"Avatar":"https://abs.twimg.com/sticky/default_profile_images/default_profile_normal.png","Banner":"","Biography":"","Birthday":"","FollowersCount":61,"FollowingCount":7,"FriendsCount":77,"IsPrivate":false,"IsVerified":false,"Joined":"2022-06-11T15:16:41Z","LikesCount":7,"ListedCount":0,"Location":"","Name":"Wade","PinnedTweetIDs":[],"TweetsCount":55,"URL":"https://twitter.com/Wade_Leeeee","UserID":"VXNlcjoxNTM1NjQyMTUyNTY2MjU5NzEy","Username":"Wade_Leeeee","Website":"","Sensitive":false,"Following":false,"FollowedBy":false}
k 0 Wade_Leeeee VXNlcjoxNTM1NjQyMTUyNTY2MjU5NzEy
User:1535642152566259712
users {"Avatar":"https://pbs.twimg.com/profile_images/1787913259074691072/XioiWSMF_normal.jpg","Banner":"https://pbs.twimg.com/profile_banners/1568295462486016008/1715106782","Biography":"Positive mind, positive vibes, positive life ✨","Birthday":"","FollowersCount":47,"FollowingCount":0,"FriendsCount":548,"IsPrivate":false,"IsVerified":false,"Joined":"2022-09-09T17:49:17Z","LikesCount":0,"ListedCount":0,"Location":"South Kensington, UK","Name":"Aurora","PinnedTweetIDs":[],"TweetsCount":21,"URL":"https://twitter.com/Shae85083106","UserID":"VXNlcjoxNTY4Mjk1NDYyNDg2MDE2MDA4","Username":"Shae85083106","Website":"","Sensitive":false,"Following":false,"FollowedBy":false}
k 1 Shae85083106 VXNlcjoxNTY4Mjk1NDYyNDg2MDE2MDA4
User:1568295462486016008
*/
type Users struct {
Profiles []Profile
Current string
Next string
Error error
}
type Profile struct {
UserIdAsNumber string
*twitterscraper.Profile
}
// type NewTask[T FollowTask | RetweetTask] struct {
// Task T
// Init bool
// }
type FollowTask struct {
// URL string
//Init bool
TaskTypeStr string
UserIdStr string
UserNameStr string
Next string
Idx []FollowerId
NewIdx []FollowerId
backPushPop *list.List
Res Users
//Scraper *twitterscraper.Scraper
}
func NewFollowTask(userId, userName, taskType string) *FollowTask {
return &FollowTask{
backPushPop: list.New(),
Idx: make([]FollowerId, 0),
UserIdStr: userId,
UserNameStr: userName,
TaskTypeStr: taskType,
}
}
type taskInterface interface {
TaskType() string
ID() string
UserName() string
Fetch(scraper *twitterscraper.Scraper) error
InitIdx() bool
UpdateIdx() (*list.List, bool)
}
// func (f *FollowTask) URl() string {
// return f.URL
// }
func (f *FollowTask) InitIdx() bool {
if len(f.Res.Profiles) > 0 {
return false
}
return true
}
func (f *FollowTask) TaskType() string {
return f.TaskTypeStr
}
func (f *FollowTask) ID() string {
return f.UserIdStr
}
func (f *FollowTask) UserName() string {
return f.UserNameStr
}
func (f *FollowTask) UpdateIdx() (*list.List, bool) {
//newIdxs := make([]FollowerId, 0, len(f.Idx))
if len(f.NewIdx) == 0 {
for k, v := range f.Res.Profiles {
f.NewIdx = append(f.NewIdx, FollowerId{
Follower: Follower{
Follower: v.UserIdAsNumber,
UserName: v.Username,
},
})
if k > 5 {
break
}
}
}
profiles, ok := MatchIdx(f.Res, f.Idx)
for _, v := range profiles {
f.backPushPop.PushFront(v)
}
slog.Info("MatchIdx", "f.UserIdStr", f.UserIdStr, "f.UserNameStr", f.UserNameStr, "match", ok, "len(profiles)", len(profiles))
if !ok {
f.Next = f.Res.Next
return nil, false
}
// 这个想一想
if f.backPushPop.Len() == 0 {
f.Next = ""
//time.Sleep(time.Second * 20)
return nil, false
}
f.Idx = f.NewIdx
f.NewIdx = make([]FollowerId, 0, len(f.Idx))
resList := f.backPushPop
f.backPushPop = list.New()
return resList, true
}
func MatchIdx(data Users, idxs []FollowerId) ([]Profile, bool) {
if len(idxs) == 0 {
return data.Profiles, true
}
for k, v := range data.Profiles {
for _, idx := range idxs {
if v.UserIdAsNumber == idx.Follower.Follower {
return data.Profiles[:k], true
}
}
}
return data.Profiles, false
}
func MatchIdxRetweet(data Users, idxs []RetweeterId) ([]Profile, bool) {
if len(idxs) == 0 {
return data.Profiles, true
}
for k, v := range data.Profiles {
for _, idx := range idxs {
if v.UserIdAsNumber == idx.Retweeter.RetweeterId {
//if v.UserIdAsNumber == idx.Follower.Follower {
return data.Profiles[:k], true
}
}
}
return data.Profiles, false
}
func (f *FollowTask) Fetch(scraper *twitterscraper.Scraper) error {
// f.UserId = "OnlyDD_D"
users, newNext, err := scraper.FetchFollowersByUserID(f.UserIdStr, 20, f.Next)
//users, newNext, err := scraper.FetchFollowers(f.UserId, 20, f.Next)
if err != nil {
return err
}
usersWithUserNumber := make([]Profile, 0, len(users))
for _, v := range users {
sDec, _ := b64.StdEncoding.DecodeString(v.UserID)
userId, _ := strings.CutPrefix(string(sDec), "User:")
item := Profile{
UserIdAsNumber: userId,
Profile: v,
}
//slog.Info("fetch-----------------", item.UserID, item.Profile.Username, item.Profile.Joined)
usersWithUserNumber = append(usersWithUserNumber, item)
}
res := Users{
Profiles: usersWithUserNumber,
Current: f.Next,
Next: newNext,
}
slog.Info("follow fetch", "f.UserIdStr", f.UserIdStr, "f.UserNameStr", f.UserNameStr, "res.Current", res.Current, "res.Next", res.Next, "len(res.Profiles)", len(res.Profiles))
f.Res = res
return nil
}
type RetweetTask struct {
//URL string
TaskTypeStr string
TweetId string
UserNameStr string
Next string
Idx []RetweeterId
//[]RetweeterId
// backPushPop = list.New()
backPushPop *list.List
NewIdx []RetweeterId
Res Users
}
// func (r *RetweetTask) URl() string {
// return r.URL
// }
func NewRetweetTask(Id, userName, taskType string) *RetweetTask {
return &RetweetTask{
TaskTypeStr: taskType,
TweetId: Id,
UserNameStr: userName,
backPushPop: list.New(),
Idx: make([]RetweeterId, 0),
}
}
func (f *RetweetTask) TaskType() string {
return f.TaskTypeStr
}
func (f *RetweetTask) UserName() string {
return f.UserNameStr
}
func (f *RetweetTask) ID() string {
return f.TweetId
}
// func (f *RetweetTask) SetScraper(scraper *twitterscraper.Scraper) {
// f.Scraper = scraper
// }
func (f *RetweetTask) Fetch(scraper *twitterscraper.Scraper) error {
slog.Info("retweet fetch", "f.TweetId", f.TweetId, "f.UserNameStr", f.UserNameStr, "f.Next", f.Next)
//return scraper.RetweetsUsers(f.TweetId)
users, newNext, err := scraper.RetweetsUsers(f.TweetId, 20, f.Next)
//users, newNext, err := scraper.FetchFollowers(f.UserId, 20, f.Next)
if err != nil {
return err
}
usersWithUserNumber := make([]Profile, 0, len(users))
for _, v := range users {
sDec, _ := b64.StdEncoding.DecodeString(v.UserID)
userId, _ := strings.CutPrefix(string(sDec), "User:")
item := Profile{
UserIdAsNumber: userId,
Profile: v,
}
//slog.Info("fetch-----------------", item.UserID, item.Profile.Username, item.Profile.Joined)
usersWithUserNumber = append(usersWithUserNumber, item)
}
res := Users{
Profiles: usersWithUserNumber,
Current: f.Next,
Next: newNext,
}
slog.Info("retweet fetch", "f.TweetId", f.TweetId, "f.UserNameStr", f.UserNameStr, "res.Current", res.Current, "res.Next", res.Next, "len(res.Profiles)", len(res.Profiles))
f.Res = res
return nil
}
// func (f *RetweetTask) SetScraperNil() {
// f.Scraper = nil
// }
func (f *RetweetTask) InitIdx() bool {
if len(f.Res.Profiles) > 0 {
return false
}
return true
}
func (f *RetweetTask) UpdateIdx() (*list.List, bool) {
if len(f.NewIdx) == 0 {
for k, v := range f.Res.Profiles {
f.NewIdx = append(f.NewIdx, RetweeterId{
Retweeter: Retweeter{
RetweeterId: v.UserIdAsNumber,
RetweeterUserName: v.Username,
},
})
if k > 5 {
break
}
}
}
profiles, ok := MatchIdxRetweet(f.Res, f.Idx)
for _, v := range profiles {
f.backPushPop.PushFront(v)
}
slog.Info("MatchIdx", "f.TweetId", f.TweetId, "f.UserNameStr", f.UserNameStr, "match", ok, "len(profiles)", len(profiles))
if !ok {
f.Next = f.Res.Next
return nil, false
}
// 这个想一想
if f.backPushPop.Len() == 0 {
f.Next = ""
//time.Sleep(time.Second * 20)
return nil, false
}
f.Idx = f.NewIdx
f.NewIdx = make([]RetweeterId, 0, len(f.Idx))
resList := f.backPushPop
f.backPushPop = list.New()
return resList, true
}
type TwitterAccount struct {
User string
PassWd string
}
// var users []TwitterAccount = []TwitterAccount{
// TwitterAccount{
// User: "Wade_Leeeee",
// PassWd: "923881393time",
// },
// }
syncer deleted 100755 → 0
View file @ 9c239c18
File deleted
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment