🎉 start the project

.gitignore

@@ -0,0 +1,3 @@
+.idea
+.vscode
+.DS_Store

go.mod

@@ -0,0 +1,16 @@
+module github.com/loveuer/nfflow
+
+go 1.22.1
+
+require (
+	github.com/elastic/go-elasticsearch/v7 v7.17.10
+	github.com/loveuer/nf v0.1.3
+	github.com/sirupsen/logrus v1.9.3
+	github.com/spf13/cobra v1.8.0
+)
+
+require (
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
+	golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 // indirect
+)

go.sum

@@ -0,0 +1,28 @@
+github.com/cpuguy83/go-md2man/v2 v2.0.3/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+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/elastic/go-elasticsearch/v7 v7.17.10 h1:TCQ8i4PmIJuBunvBS6bwT2ybzVFxxUhhltAs3Gyu1yo=
+github.com/elastic/go-elasticsearch/v7 v7.17.10/go.mod h1:OJ4wdbtDNk5g503kvlHLyErCgQwwzmDtaFC4XyOxXA4=
+github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
+github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/loveuer/nf v0.1.3 h1:tZP+FtwhiU+VTfPwfaEQUmiw1z6U9XwfDzJV46h5vZw=
+github.com/loveuer/nf v0.1.3/go.mod h1:uKsKYym27ravyTXSBSnxU86V7osxx9cM6DJ+dVBfJ1Q=
+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/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
+github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+github.com/spf13/cobra v1.8.0 h1:7aJaZx1B85qltLMc546zn58BxxfZdR/W22ej9CFoEf0=
+github.com/spf13/cobra v1.8.0/go.mod h1:WXLWApfZ71AjXPya3WOlMsY9yMs7YeiHhFVlvLyhcho=
+github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
+github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

interfaces/input.go

@@ -0,0 +1,9 @@
+package interfaces
+
+import "context"
+
+type Input interface {
+	InitClient(ctx context.Context) error
+	Fetch(ctx context.Context, limit int)
+	Close(ctx context.Context) error
+}

interfaces/output.go

@@ -0,0 +1 @@
+package interfaces

interfaces/pipe.go

@@ -0,0 +1 @@
+package interfaces

internal/cmd/root.go

@@ -0,0 +1,34 @@
+package cmd
+
+import (
+	"fmt"
+	"github.com/loveuer/nfflow/internal/opt"
+	"github.com/loveuer/nfflow/internal/srv"
+	"github.com/spf13/cobra"
+)
+
+var (
+	version bool
+	root    = &cobra.Command{
+		Use: "nfflow",
+		Run: func(cmd *cobra.Command, args []string) {
+			if version {
+				fmt.Printf("nfflow \nVersion: %s\n", opt.Version)
+			}
+		},
+	}
+
+	run = &cobra.Command{
+		Use:     "run",
+		Example: "nfflow run --address ':80'",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			return srv.Run(cmd.Context())
+		},
+	}
+)
+
+func init() {
+	root.Flags().BoolVarP(&version, "version", "v", false, "print version")
+	run.Flags().StringVar(&opt.Cfg.Address, "address", ":80", "service listen address")
+	root.AddCommand(run)
+}

internal/cmd/run.go

@@ -0,0 +1,13 @@
+package cmd
+
+import (
+	"context"
+	"github.com/sirupsen/logrus"
+)
+
+func Run(ctx context.Context) {
+	if err := root.ExecuteContext(ctx); err != nil {
+		logrus.Debugf("cmd.Run: execute with err=%v", err)
+		return
+	}
+}

internal/handler/available.go

@@ -0,0 +1,12 @@
+package handler
+
+import (
+	"github.com/loveuer/nf"
+	"github.com/loveuer/nf/nft/resp"
+	"github.com/loveuer/nfflow/internal/opt"
+	"time"
+)
+
+func Available(c *nf.Ctx) error {
+	return resp.Resp200(c, nf.Map{"ok": true, "time": time.Now(), "version": opt.Version})
+}

internal/handler/new_input.go

@@ -0,0 +1,51 @@
+package handler
+
+import (
+	"github.com/loveuer/nf"
+	"github.com/loveuer/nf/nft/resp"
+	"github.com/loveuer/nfflow/internal/inputs/es7"
+	"github.com/loveuer/nfflow/internal/util"
+)
+
+func NewInput_ES7(c *nf.Ctx) error {
+	type Req struct {
+		Endpoints []string `json:"endpoints"`
+		Username  string   `json:"username"`
+		Password  string   `json:"password"`
+		CA        string   `json:"ca"`
+	}
+
+	var (
+		err error
+		req = new(Req)
+	)
+
+	if err = c.BodyParser(req); err != nil {
+		return resp.Resp400(c, err.Error())
+	}
+
+	if len(req.Endpoints) == 0 {
+		return resp.Resp400(c, req)
+	}
+
+	cli := &es7.Client{
+		Endpoints: req.Endpoints,
+		Username:  req.Username,
+		Password:  req.Password,
+		CA:        req.CA,
+	}
+
+	if err = cli.InitClient(util.Timeout(5)); err != nil {
+		return resp.Resp400(c, err.Error(), "连接失败, 请检查配置")
+	}
+
+	if err = cli.Ping(util.Timeout(3)); err != nil {
+		return resp.Resp400(c, err.Error(), "尝试连接失败, 请检查配置")
+	}
+
+	if cli.Save(util.Timeout(5)); err != nil {
+		return resp.Resp500(c, err.Error())
+	}
+
+	return resp.Resp200(c, nil, "保存 输入流 成功")
+}

internal/inputs/es7/client.go

@@ -0,0 +1,108 @@
+package es7
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	elastic "github.com/elastic/go-elasticsearch/v7"
+	"github.com/elastic/go-elasticsearch/v7/esapi"
+	"github.com/sirupsen/logrus"
+	"net/http"
+)
+
+type Client struct {
+	Endpoints []string `json:"endpoints"`
+	Username  string   `json:"username"`
+	Password  string   `json:"password"`
+	CA        string   `json:"ca"`
+
+	cli *elastic.Client
+}
+
+func (c *Client) InitClient(ctx context.Context) error {
+	var (
+		errCh    = make(chan error)
+		cliCh    = make(chan *elastic.Client)
+		hiddenCa = func(cs string) string {
+			if len(cs) > 0 {
+				return "******"
+			}
+
+			return "nil"
+		}
+	)
+
+	logrus.Debugf("es7.NewClient: endpoints=%v (username=%s password=%s ca=%s)", c.Endpoints, c.Username, c.Password, hiddenCa(c.CA))
+
+	ncFunc := func(endpoints []string, username, password, ca string) {
+		var (
+			err      error
+			cli      *elastic.Client
+			infoResp *esapi.Response
+		)
+
+		if cli, err = elastic.NewClient(
+			elastic.Config{
+				Addresses:     endpoints,
+				Username:      username,
+				Password:      password,
+				CACert:        []byte(c.CA),
+				RetryOnStatus: []int{429},
+				MaxRetries:    3,
+				RetryBackoff:  nil,
+				Transport: &http.Transport{
+					TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+				},
+			},
+		); err != nil {
+			logrus.Debugf("es7.NewClient: elastic new client with endponts=%v err=%v", endpoints, err)
+			errCh <- err
+			return
+		}
+
+		if infoResp, err = cli.Info(); err != nil {
+			logrus.Debugf("es7.NewClient: ping err=%v", err)
+			errCh <- err
+			return
+		}
+
+		if infoResp.StatusCode != 200 {
+			err = fmt.Errorf("info es status=%d", infoResp.StatusCode)
+			logrus.Debugf("es7.NewClient: status err=%v", err)
+			errCh <- err
+			return
+		}
+
+		cliCh <- cli
+	}
+
+	go ncFunc(c.Endpoints, c.Username, c.Password, c.CA)
+
+	select {
+	case <-ctx.Done():
+		return fmt.Errorf("dial es=%s err=%v", c.Endpoints, context.DeadlineExceeded)
+	case c.cli = <-cliCh:
+		return nil
+	case e := <-errCh:
+		return e
+	}
+}
+
+func (c *Client) Ping(ctx context.Context) error {
+	rr, err := c.cli.Info(
+		c.cli.Info.WithContext(ctx),
+	)
+	if err != nil {
+		return err
+	}
+
+	if rr.StatusCode != 200 {
+		return fmt.Errorf("ping status=%d msg=%s", rr.StatusCode, rr.String())
+	}
+
+	return nil
+}
+
+func (c *Client) Save(ctx context.Context) error {
+	return nil
+}

internal/opt/opt.go

@@ -0,0 +1,9 @@
+package opt
+
+type config struct {
+	Address string `json:"address"`
+}
+
+var (
+	Cfg = &config{}
+)

internal/opt/var.go

@@ -0,0 +1,3 @@
+package opt
+
+const Version = "v0.0.1"

internal/srv/api.go

@@ -0,0 +1,22 @@
+package srv
+
+import (
+	"github.com/loveuer/nf"
+	"github.com/loveuer/nfflow/internal/handler"
+)
+
+func initApp() *nf.App {
+	engine := nf.New()
+
+	app := engine.Group("/api")
+	app.Get("/available", handler.Available)
+
+	{
+		api := app.Group("/new")
+
+		inputApi := api.Group("/input")
+		inputApi.Post("/es7", handler.NewInput_ES7)
+	}
+
+	return engine
+}

internal/srv/run.go

@@ -0,0 +1,39 @@
+package srv
+
+import (
+	"context"
+	"github.com/loveuer/nfflow/internal/opt"
+	"github.com/loveuer/nfflow/internal/util"
+	"github.com/sirupsen/logrus"
+	"net"
+)
+
+func Run(ctx context.Context) error {
+	app := initApp()
+
+	ln, err := net.Listen("tcp", opt.Cfg.Address)
+	if err != nil {
+		logrus.Errorf("srv.Run: err=%v", err)
+		return err
+	}
+
+	ready := make(chan bool)
+
+	go func() {
+		ready <- true
+		<-ctx.Done()
+
+		if err = app.Shutdown(util.Timeout(2)); err != nil {
+			logrus.Infof("srv.Run: nf app shutdown err=%v", err)
+		}
+	}()
+
+	<-ready
+
+	if err = app.RunListener(ln); err != nil {
+		logrus.Errorf("srv.Run: nf app run err=%v", err)
+		return err
+	}
+
+	return nil
+}

internal/util/ctx.go

@@ -0,0 +1,28 @@
+package util
+
+import (
+	"context"
+	"time"
+)
+
+func Timeout(seconds ...int) context.Context {
+	second := 30
+	if len(seconds) > 0 && seconds[0] > 0 {
+		second = seconds[0]
+	}
+
+	ctx, _ := context.WithTimeout(context.Background(), time.Duration(second)*time.Second)
+
+	return ctx
+}
+
+func TimeoutCtx(ctx context.Context, seconds ...int) context.Context {
+	second := 30
+	if len(seconds) > 0 && seconds[0] > 0 {
+		second = seconds[0]
+	}
+
+	timeout, _ := context.WithTimeout(ctx, time.Duration(second)*time.Second)
+
+	return timeout
+}

main.go

@@ -0,0 +1,15 @@
+package main
+
+import (
+	"context"
+	"github.com/loveuer/nfflow/internal/cmd"
+	"os/signal"
+	"syscall"
+)
+
+func main() {
+	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
+	defer cancel()
+
+	cmd.Run(ctx)
+}

xtest/map.js

@@ -0,0 +1,11 @@
+const raw = `{"_id":"db8dfbc2fb6bec9eb8ed5ae4b806c81d","_index":"sonar_post","_source":{"area":"{\\"country\\":[{\\"country_code\\":\\"CA\\",\\"num\\":2},{\\"country_code\\":\\"CN\\",\\"num\\":2}],\\"address\\":[{\\"src\\":\\"Canada\\",\\"num\\":2,\\"cn\\":\\"加拿大\\"},{\\"src\\":\\"China\\",\\"num\\":1,\\"cn\\":\\"中国\\"},{\\"src\\":\\"Shanghai\\",\\"num\\":1,\\"cn\\":\\"上海\\"}],\\"name\\":[{\\"src\\":\\"Jennifer Lynn Lalonde\\",\\"num\\":1,\\"cn\\":\\"詹妮弗·林恩·拉隆德\\"}]}","area_country":[{"country_code":"CA","num":2},{"country_code":"CN","num":2}],"area_name":[{"cn":"詹妮弗·林恩·拉隆德","num":1,"src":"Jennifer Lynn Lalonde"}],"category_tags":[31],"collect_date":1686602410,"comments_count":1,"content":"To clarify: Hours after #Canada’s expulsion of a Chinese diplomat on allegations of foreign interference, #China responds by declaring Jennifer Lynn Lalonde, consul at Canada’s Consulate General in Shanghai, ‘persona non grata’ with 4 days to leave, per MoFA statement. @NBCNews https://t.co/dIqN31d74U","create_date":1683589320,"data_type":"user_post","emotion":"Negative","event_tag":[{"tag_date":"2023-06-14","tag_id":"3cbd333dda674709b4b79a5bdcd05376-TOPIC","tag_type":"TOPIC"},{"tag_date":"2023-06-14","tag_id":"f712334d77964f0bac93dcc9c16df4ff-TOPIC","tag_type":"TOPIC"}],"group_create_date":0,"hash_tags":["Canada","China"],"id":"db8dfbc2fb6bec9eb8ed5ae4b806c81d","image":["{\\"origin_url\\":\\"https://pbs.twimg.com/media/Fvq6LNSWAAEIrql.jpg\\",\\"file_hash\\":\\"0689e2ad7b6e69040650f8e8c10bdc0b\\",\\"file_type\\":\\"image\\",\\"s3_url\\":\\"https://data-tmp-1306241199.cos.ap-hongkong.myqcloud.com/0689e2ad7b6e69040650f8e8c10bdc0b\\",\\"format\\":\\"jpg\\",\\"description\\":\\"{\\\\\\"tweet_url\\\\\\": \\\\\\"https://t.co/dIqN31d74U\\\\\\"}\\",\\"file_size\\":117402}"],"input_date":1687169193,"involve_user_id":["NBCNews"],"involve_user_id2":["14173315"],"key_phrase":["上海总领事馆领事-1","中国外交官-1","詹妮弗·林恩·拉隆德-1"],"language":"en","likes_count":1,"nationality":["US"],"new_id":1,"platform":"twitter","political_more_cls":[8],"repost_create_date":0,"reposts_count":0,"standpoint_analysis":"Negative","tokenlized":["加拿大-2","外交官-1","总领事馆-1","上海-1","声明-1","领事-1","中国-1","外国-1","小时-1","詹妮弗·林恩·拉隆德-1"],"translate_content":"澄清一下:根据外交部的声明，在加拿大以外国干涉指控驱逐一名中国外交官几小时后，中国的回应是宣布加拿大驻上海总领事馆领事詹妮弗·林恩·拉隆德为“不受欢迎的人”，并将离开4天。 @nbcnews https://t.co/diqn31d74u","url":"https://twitter.com/janisfrayer/status/1655840529530208257","user_id":"janisfrayer","user_id2":"137223549","user_name":"Janis Mackey Frayer"}}`
+
+let item = JSON.parse(raw)
+
+let result = item['_source']['image']
+
+let data = JSON.parse(result)
+
+let res = {hash: data['file_hash']}
+
+console.log(res)