make_full_history.go

package main

import (
	"fmt"
	"log"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"encoding/json"

	"github.com/go-git/go-git/v5"
	"github.com/go-git/go-git/v5/plumbing"
)

// gitLogCommand runs a git log command to fetch file commits since a specific date.
func gitLogCommand() []string {
	cmd := "git log --full-history --reverse --name-only --pretty=format:\"Commit: %H\" -- \"coinbase/24h_stats/*.json\" | awk '/^Commit:/ {commit=$2} !/^Commit:/ && NF {print commit, $0}'"
	// cmd := "git log --full-history --reverse --since=2024-03-01 --name-only --pretty=format:\"Commit: %H\" -- \"coinbase/24h_stats/*.json\" | awk '/^Commit:/ {commit=$2} !/^Commit:/ && NF {print commit, $0}'"
	out, err := exec.Command("bash", "-c", cmd).Output()
	if err != nil {
		fmt.Println("Error executing git log command:", err)
		return nil
	}
	lines := strings.Split(string(out), "\n")
	return lines
}

type CommitFile struct {
	Hash     string
	Filename string
}

func main() {
	var counter int64
	var counter_last int64
	counter_last = 0

	cmdChan := make(chan CommitFile, 100000000) // Buffer may need adjustment
	monitor := time.NewTicker(1 * time.Second)

	go func() {
		for range monitor.C {
			finalCount := atomic.LoadInt64(&counter)
			log.Println("chan : q:", len(cmdChan), "done:", finalCount, "rate:", finalCount-counter_last)
			counter_last = finalCount
		}
	}()

	lines := gitLogCommand()
	fileCommits := make(map[string][]string)

	for _, line := range lines {
		parts := strings.SplitN(line, " ", 2)
		if len(parts) < 2 {
			continue
		}
		commit, file := parts[0], parts[1]
		fileCommits[file] = append(fileCommits[file], commit)
	}

	var wg_add_commits sync.WaitGroup
	for filename, commits := range fileCommits {
		wg_add_commits.Add(1)
		go func(filename string, commits []string) {
			defer wg_add_commits.Done()

			for _, commit := range commits {
				cmdChan <- CommitFile{Hash: commit, Filename: filename}
			}

		}(filename, commits)
	}

	go func() {
		wg_add_commits.Wait()
		close(cmdChan)
	}()

	var wg sync.WaitGroup
	for i := 0; i < 24; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()

			r, _ := git.PlainOpen(".")

			for command := range cmdChan {
				atomic.AddInt64(&counter, 1)
				outputFilename := filepath.Base(command.Filename) + "l"

				hash := plumbing.NewHash(command.Hash)
				commit, err := r.CommitObject(hash)
				if err != nil {
					log.Printf("Failed to get commit object for hash %s: %s", command.Filename, command.Hash, err)
					continue
				}

				// Get the file content from the commit
				file, err := commit.File(command.Filename)
				if err != nil {
					log.Println("Error getting file from commit:", err)
					continue
				}

				fr, err := file.Reader()
				if err != nil {
					log.Println("Error getting reader from file:", err)
					continue
				}

				// Assuming fileContent is a JSON string, unmarshal it, add your fields, and marshal it again
				var jsonData map[string]interface{}
				err = json.NewDecoder(fr).Decode(&jsonData)
				if err != nil {
					log.Printf("Failed to unmarshal JSON for %s: %s", command.Filename, err)
					continue
				}
				fr.Close()

				jsonData["commit_time"] = commit.Committer.When.Format(time.RFC3339)
				jsonData["hash"] = command.Hash

				outputFile, err := os.OpenFile(fmt.Sprintf("coinbase/jsonl/%s", outputFilename), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
				if err != nil {
					log.Printf("Failed to open output file: %s", err)
					continue
				}

				err = json.NewEncoder(outputFile).Encode(jsonData)
				if err != nil {
					log.Printf("Failed to write the json: %s", err)
					continue
				}
				outputFile.Close()
			}
		}()
	}

	wg.Wait()
	monitor.Stop()
}