diff --git a/.gitea/workflows/release.yml b/.gitea/workflows/release.yml new file mode 100644 index 0000000..196ccb5 --- /dev/null +++ b/.gitea/workflows/release.yml @@ -0,0 +1,66 @@ +name: Gitea Go Release Actions +run-name: ${{ gitea.actor }} go🚀 +on: + push: + tags: + - '*' + +jobs: + releases-matrix: + name: Release Go Binary + runs-on: ubuntu-latest + strategy: + matrix: + goos: [linux, windows, darwin] + goarch: [amd64, arm64] + include: + - goarch: arm + goos: linux + # 不整活了,做个正常人 + # # 俺路由器是 mips + # - goarch: mips + # goos: linux + # # 谁创 riscv ? + # - goarch: riscv64 + # goos: linux + # # woc,🐲 + # - goarch: loong64 + # goos: linux + exclude: + - goarch: arm64 + goos: windows + steps: + - uses: actions/checkout@v4 + - uses: wangyoucao577/go-release-action@v1 + id: go-release-action + with: + github_token: ${{ secrets.GITHUB_TOKEN }} + goos: ${{ matrix.goos }} + goarch: ${{ matrix.goarch }} + multi_binaries: true + project_path: ./cmd/... + upload: false + # output is release_asset_dir + - name: rename artifact + # append os and arch to the artifact name (handle windows .exe) + run: | + cd ${{ steps.go-release-action.outputs.release_asset_dir }} + for f in *; do + if [[ $f == *.exe ]]; then + noextname=$(basename "$f" .exe) + mv "$f" "${noextname}-${{ matrix.goos }}-${{ matrix.goarch }}.exe" + else + mv "$f" "${f}-${{ matrix.goos }}-${{ matrix.goarch }}" + fi + done + cd - + - name: Compress binaries + continue-on-error: true + uses: svenstaro/upx-action@v2 + with: + files: | + ${{ steps.go-release-action.outputs.release_asset_dir }}/** + - uses: akkuman/gitea-release-action@v1 + with: + files: |- + ${{ steps.go-release-action.outputs.release_asset_dir }}/** diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a844afb --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +acdanmaku +*.conf \ No newline at end of file diff --git a/cmd/acdanmaku/acdanmaku.go b/cmd/acdanmaku/acdanmaku.go new file mode 100644 index 0000000..73f8b2d --- /dev/null +++ b/cmd/acdanmaku/acdanmaku.go @@ -0,0 +1,192 @@ +package main + +import ( + "encoding/json" + "fmt" + "log" + "net/http" + "os" + "os/signal" + "sync" + "time" + + "strconv" + + acfun_api "git.saveweb.org/saveweb/acfunction_go/pkg" + savewebtracker "git.saveweb.org/saveweb/saveweb_tracker/src/saveweb_tracker" + "github.com/hashicorp/go-retryablehttp" + "github.com/tidwall/gjson" +) + +var BASE_CONCURRENCY = 10 +var WITH_DELAY = true + +var tasks_chan = make(chan savewebtracker.Task, BASE_CONCURRENCY) +var Interrupted = false +var WaitClaimWorker sync.WaitGroup +var WaitProcesserWorker sync.WaitGroup + +var Logger = log.New(os.Stdout, "[acdanmaku] ", log.Ldate|log.Ltime|log.Lmsgprefix) +var DEBUG = false + +func init() { + if os.Getenv("BASE_CONCURRENCY") != "" { + fmt.Println("BASE_CONCURRENCY:", os.Getenv("BASE_CONCURRENCY")) + BASE_CONCURRENCY, _ = strconv.Atoi(os.Getenv("BASE_CONCURRENCY")) + } + if os.Getenv("NO_WITH_DELAY") != "" { + fmt.Println("NO_WITH_DELAY:", os.Getenv("NO_WITH_DELAY")) + WITH_DELAY = false + } + if os.Getenv("DEBUG") != "" { + DEBUG = true + } +} + +// ClaimTask 并把任务放入 task_chan +func claimWorker(i int, tracker *savewebtracker.Tracker) { + workerName := fmt.Sprintf("[ClaimWorker(%d)]", i) + Logger.Println("[START]", workerName) + defer Logger.Println("[STOP]", workerName, " exited...") + defer WaitClaimWorker.Done() + for { + if Interrupted { + return + } + task := tracker.ClaimTask(WITH_DELAY) + if task == nil { + notask_sleep := max( + time.Duration(tracker.Project().Client.ClaimTaskDelay)*10*time.Second, + time.Duration(10)*time.Second, + ) + Logger.Println(workerName, "No task to claim, sleep", notask_sleep) + time.Sleep(notask_sleep) + continue + } + Logger.Println(workerName, "Claimed task", task.Id) + tasks_chan <- *task + } +} + +func ProcesserWorker(i int, tracker *savewebtracker.Tracker) { + workerName := fmt.Sprintf("[ProcesserWorker(%d)]", i) + Logger.Println("[START]", workerName) + defer Logger.Println("[STOP]", workerName, " exited...") + defer WaitProcesserWorker.Done() + for task := range tasks_chan { + Logger.Println(workerName, "Processing task", task.Id) + + // 在这儿处理任务 + danmakus, err := acfun_api.GetDanmaku(tracker.HTTP_client, task.Id) + if err != nil { + tracker.UpdateTask(task.Id, task.Id_type, savewebtracker.StatusFAIL) + Logger.Println(workerName, "Failed to get danmaku", task.Id, err) + Interrupted = true + continue + } + items := []savewebtracker.Item{} + for _, danmaku := range danmakus { + r_danmakuId := gjson.Get(danmaku, "danmakuId") + + if !r_danmakuId.Exists() || r_danmakuId.Type != gjson.Number { + Logger.Println(workerName, "danmakuId not found or not a number") + tracker.UpdateTask(task.Id, task.Id_type, savewebtracker.StatusFAIL) + Interrupted = true + continue + } + + danmakuId := r_danmakuId.Int() + items = append(items, savewebtracker.Item{ + Item_id: fmt.Sprintf("%d", danmakuId), + Item_id_type: "int", + Item_status: "None", + Item_status_type: "None", + Payload: danmaku, + }) + } + if len(items) != 0 { + resp := tracker.InsertMany(items) + Logger.Println(workerName, "InsertMany", "task", task.Id, "->", len(items), "items", resp) + } + tracker.UpdateTask(task.Id, task.Id_type, savewebtracker.StatusDONE) + Logger.Println(workerName, "UpdateTask", task.Id) + } +} + +func InterruptHandler() { + fmt.Println("Press Ctrl+C to exit") + interrupt_c := make(chan os.Signal, 1) + signal.Notify(interrupt_c, os.Interrupt) + for { + s := <-interrupt_c + Logger.Println("Interrupted by", s, "signal (Press Ctrl+C again to force exit)") + if Interrupted { + Logger.Println("Force exit") + os.Exit(1) + return + } + Interrupted = true + } +} + +func GetRetryableHttpClient(timeout time.Duration, debug bool) *http.Client { + retryClient := retryablehttp.NewClient() + retryClient.RetryMax = 3 + retryClient.RetryWaitMin = 1 * time.Second + retryClient.RetryWaitMax = 10 * time.Second + retryClient.HTTPClient.Timeout = timeout + if !debug { + retryClient.Logger = nil + } + standardClient := retryClient.StandardClient() // *http.Client + Logger.Println("standardClient.Timeout:", standardClient.Timeout) + return standardClient +} + +func ShowStatus(t *savewebtracker.Tracker) { + for { + project_json, err := json.Marshal(t.Project()) + if err != nil { + panic(err) + } + Logger.Println("Project:", string(project_json)) + time.Sleep(60 * time.Second) + } +} + +func main() { + tracker := savewebtracker.GetTracker("acdanmaku", "0.1", savewebtracker.Archivist()) + tracker.PING_client = GetRetryableHttpClient(10*time.Second, DEBUG) + tracker.HTTP_client = GetRetryableHttpClient(60*time.Second, DEBUG) + tracker.SelectBestTracker() + _, err := tracker.FetchProject(10 * time.Second) + if err != nil { + panic(err) + } + tracker.StartSelectTrackerBackground().StartFetchProjectBackground() + + go InterruptHandler() + go ShowStatus(tracker) + + Logger.Println("-- Start --") + + for i := 0; i < BASE_CONCURRENCY; i++ { + go claimWorker(i, tracker) + WaitClaimWorker.Add(1) + go ProcesserWorker(i, tracker) + WaitProcesserWorker.Add(1) + } + + // wait for all claimWorker to finish + WaitClaimWorker.Wait() + Logger.Println("[STOP] All claimWorker done") + // close task_chan + close(tasks_chan) + Logger.Println("[STOP] task_chan closed") + // wait for all task_chan to finish + WaitProcesserWorker.Wait() + Logger.Println("[STOP] All ProcesserWorker done") + + Logger.Println("-- All done --") + +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..660cd9b --- /dev/null +++ b/go.mod @@ -0,0 +1,15 @@ +module git.saveweb.org/saveweb/acfunction_go + +go 1.22.4 + +require ( + git.saveweb.org/saveweb/saveweb_tracker v0.1.11 + github.com/hashicorp/go-retryablehttp v0.7.7 + github.com/tidwall/gjson v1.17.1 +) + +require ( + github.com/hashicorp/go-cleanhttp v0.5.2 // indirect + github.com/tidwall/match v1.1.1 // indirect + github.com/tidwall/pretty v1.2.1 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..0ba2648 --- /dev/null +++ b/go.sum @@ -0,0 +1,31 @@ +git.saveweb.org/saveweb/saveweb_tracker v0.1.11 h1:re/ohuSRmcpDzJ2fQjmm5U6EI4lO4nQBG0SXqz+OeJY= +git.saveweb.org/saveweb/saveweb_tracker v0.1.11/go.mod h1:p891f4fshoA/Wiwmey23f2xJ9sKNEZwd5kmzG6lobik= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/fatih/color v1.16.0 h1:zmkK9Ngbjj+K0yRhTVONQh1p/HknKYSlNT+vZCzyokM= +github.com/fatih/color v1.16.0/go.mod h1:fL2Sau1YI5c0pdGEVCbKQbLXB6edEj1ZgiY4NijnWvE= +github.com/hashicorp/go-cleanhttp v0.5.2 h1:035FKYIWjmULyFRBKPs8TBQoi0x6d9G4xc9neXJWAZQ= +github.com/hashicorp/go-cleanhttp v0.5.2/go.mod h1:kO/YDlP8L1346E6Sodw+PrpBSV4/SoxCXGY6BqNFT48= +github.com/hashicorp/go-hclog v1.6.3 h1:Qr2kF+eVWjTiYmU7Y31tYlP1h0q/X3Nl3tPGdaB11/k= +github.com/hashicorp/go-hclog v1.6.3/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M= +github.com/hashicorp/go-retryablehttp v0.7.7 h1:C8hUCYzor8PIfXHa4UrZkU4VvK8o9ISHxT2Q8+VepXU= +github.com/hashicorp/go-retryablehttp v0.7.7/go.mod h1:pkQpWZeYWskR+D1tR2O5OcBFOxfA7DoAO6xtkuQnHTk= +github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= +github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= +github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/tidwall/gjson v1.17.1 h1:wlYEnwqAHgzmhNUFfw7Xalt2JzQvsMx2Se4PcoFCT/U= +github.com/tidwall/gjson v1.17.1/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk= +github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA= +github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM= +github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU= +github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4= +github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU= +golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y= +golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/pkg/acfun_api.go b/pkg/acfun_api.go new file mode 100644 index 0000000..c4720c7 --- /dev/null +++ b/pkg/acfun_api.go @@ -0,0 +1,76 @@ +package acfun_api + +import ( + "fmt" + "io" + "log" + "net/http" + "net/url" + "os" + "strings" + "time" + + "github.com/tidwall/gjson" +) + +var Logger = log.New(os.Stdout, "[acfun_api] ", log.Ldate|log.Ltime|log.Lmsgprefix) + +// var json = jsoniter.ConfigCompatibleWithStandardLibrary + +func GetDanmaku(client *http.Client, resource_id string) ([]string, error) { + danmakus := []string{} + pcursor := "1" + for { + + data := url.Values{ + "resourceId": {resource_id}, + "resourceType": {"9"}, + "enableAdvanced": {"true"}, + "pcursor": {pcursor}, + "count": {"200"}, + "sortType": {"1"}, + "asc": {"false"}, + } + dataEncoded := data.Encode() + + req, err := http.NewRequest("POST", "https://www.acfun.cn/rest/pc-direct/new-danmaku/list", strings.NewReader(dataEncoded)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + req.Header.Set("Accept", "application/json, text/plain, */*") + req.Header.Set("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36 Viewer/99.9.8782.87") + + response, err := client.Do(req) + if err != nil { + return nil, err + } + + if response.StatusCode != 200 { + panic("StatusCode != 200") + } + + body, err := io.ReadAll(response.Body) + if err != nil { + return nil, err + } + + result := gjson.GetBytes(body, "result").Int() + if result != 0 { + // panic(gjson.GetBytes(body, "error_msg").String()) + return nil, fmt.Errorf("result != 0: %s", gjson.GetBytes(body, "error_msg").String()) + } + pcursor = gjson.GetBytes(body, "pcursor").String() + + for _, danmaku := range gjson.GetBytes(body, "danmakus").Array() { + danmakus = append(danmakus, danmaku.Raw) + } + + Logger.Println(resource_id, "pcursor", pcursor, len(danmakus), "danmakus") + if pcursor == "no_more" { + break + } + time.Sleep(1 * time.Second) + } + return danmakus, nil +} diff --git a/pkg/acfun_api_test.go b/pkg/acfun_api_test.go new file mode 100644 index 0000000..04a215e --- /dev/null +++ b/pkg/acfun_api_test.go @@ -0,0 +1,23 @@ +package acfun_api + +import ( + "fmt" + "net/http" + "testing" +) + +func TestGetDanmaku(t *testing.T) { + client := &http.Client{} + danmakus, err := GetDanmaku(client, "12528563") + if err != nil { + t.Fatal(err) + } + t.Log(len(danmakus)) + // fmt.Println(danmakus) + for _, danmaku := range danmakus { + fmt.Println(danmaku) + } + if len(danmakus) == 0 { + t.Fatal("danmakus is empty") + } +}