Skip to content
This repository was archived by the owner on Sep 19, 2024. It is now read-only.

Commit 02a0e20

Browse files
optim-kazuhiro-seidaoptim-kazuhiro-seida
committed
update #7 datasore download
1 parent 3b63795 commit 02a0e20

4 files changed

Lines changed: 171 additions & 67 deletions

File tree

‎cli/publishsubscribe/datastore.go‎

Lines changed: 164 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,8 @@ import (
77
"strings"
88
"time"
99

10+
"github.com/optim-kazuhiro-seida/go-advance-type/wrapper"
11+
1012
. "github.com/optim-corp/cios-cli/cli"
1113
"github.com/optim-corp/cios-cli/models"
1214
"github.com/optim-corp/cios-cli/utils"
@@ -27,6 +29,7 @@ func GetDataStoreCommand() *cli.Command {
2729
deleteDataStore(),
2830
listDataStore(),
2931
embezzleDateStore(),
32+
saveDataStore(),
3033
},
3134
}
3235
}
@@ -94,84 +97,68 @@ func listDataStore() *cli.Command {
9497
Aliases: models.ALIAS_LIST,
9598
Flags: []cli.Flag{
9699
&cli.StringFlag{Name: "channel_id", Aliases: []string{"c"}},
97-
&cli.StringFlag{Name: "packer_format", Aliases: []string{"pf"}},
98-
&cli.StringFlag{Name: "timestamp_range", Aliases: []string{"tr"}},
99-
&cli.StringFlag{Name: "save_dir", Aliases: []string{"out"}},
100+
&cli.StringFlag{Name: "packer_format", Aliases: []string{"pf"}, Value: "payload_only"},
101+
&cli.StringFlag{Name: "timestamp_range", Aliases: []string{"tr"}, DefaultText: "Now Time", Value: ":" + convert.MustStr(time.Now().UnixNano())},
100102
&cli.StringFlag{Name: "label", Aliases: []string{"lb"}},
101103
&cli.BoolFlag{Name: "data", Aliases: []string{"d"}},
102-
&cli.BoolFlag{Name: "save", Aliases: []string{"s"}},
103104
&cli.Int64Flag{Name: "limit", Aliases: []string{"l"}, DefaultText: "30", Value: 30},
104105
&cli.Int64Flag{Name: "offset", Aliases: []string{"o"}, DefaultText: "0", Value: 0},
105106
&cli.StringFlag{Name: "resource_owner_id", Aliases: []string{"r", "ro"}},
106-
&cli.StringFlag{Name: "change_channel_id", Aliases: []string{"chg-c"}},
107-
&cli.StringFlag{Name: "data_replace", Aliases: []string{"replace", "rep"}, Usage: ""},
108107
},
109108
Action: func(c *cli.Context) error {
110109
var (
111110
channelID = c.String("channel_id")
112111
packerFormat = c.String("packer_format")
113-
outputDir = c.String("save_dir")
114112
limit = c.Int64("limit")
115113
offset = c.Int64("offset")
116-
timestampRange = is(c.String("timestamp_range") != "").T(c.String("timestamp_range")).F(":" + convert.MustStr(time.Now().UnixNano())).Value.(string)
114+
timestampRange = c.String("timestamp_range")
117115
resourceOwnerID = c.String("resource_owner_id")
118-
dataFlag = c.Bool("data")
119-
saveFlag = c.Bool("save")
120116
label = c.String("label")
121-
replace = c.String("data_replace")
117+
dataFlag = c.Bool("data")
122118
channelsMap, _, _ = Client.PubSub.GetChannelsMapByID(ciossdk.MakeGetChannelsOpts(), context.Background())
123119
)
124120

125-
printObj := func(channel cios.Channel, limit int64) {
126-
if dataFlag {
127-
stageDSDir := datastoreDir + "/" + models.GetStage()
128-
channelDir := datastoreDir + "/" + models.GetStage() + "/" + channelsMap[channel.Id].Name + "___" + channel.Id
129-
data, err := Client.PubSub.GetStreamAll(channel.Id, ciossdk.MakeGetStreamOpts().Limit(limit).PackerFormat(packerFormat).TimestampRange(timestampRange).Label(label).Offset(offset), context.Background())
130-
assert(err).Log()
131-
fPrintf("\n|Channel ID| : %s \n|Channel Name|: %s\n\n", channel.Id, channelsMap[channel.Id].Name)
132-
if saveFlag {
133-
path(datastoreDir).CreateDir()
134-
path(stageDSDir).CreateDir()
135-
path(channelDir).CreateDir()
136-
}
137-
138-
for count, val := range data {
139-
if replace != "" {
140-
splitInComma := strings.Split(replace, ",")
141-
for _, _val := range splitInComma {
142-
splitVal := strings.Split(_val, ":")
143-
if len(splitVal) >= 2 {
144-
val = strings.Replace(val, splitVal[0], splitVal[1], -1)
145-
} else {
146-
log.Warn("Replace Missing format string")
147-
}
148-
}
149-
}
150-
if saveFlag {
151-
if outputDir != "" {
152-
channelDir = outputDir
153-
}
154-
filePrefix := strings.Repeat("0", len(strconv.Itoa(len(data)))-len(strconv.Itoa(count)))
155-
_path := channelDir + "/" + filePrefix + strconv.Itoa(count) + ".txt"
156-
assert(path(_path).WriteFileAsString(val)).Log()
157-
fPrintln(val)
121+
printObject := func(channelId string) {
122+
objects, _, err := Client.PubSub.GetObjectsAll(channelId, ciossdk.MakeGetObjectsOpts().
123+
Limit(limit).
124+
TimestampRange(timestampRange).
125+
Label(label).
126+
Offset(offset),
127+
context.Background())
128+
if len(objects) == 0 || resourceOwnerID != "" && channelsMap[channelId].ResourceOwnerId != resourceOwnerID {
129+
return
130+
}
131+
fPrintf("\n|Channel ID| : %s \n|Channel Name|: %s\n\n", channelId, channelsMap[channelId].Name)
132+
fPrintln("\t|ID|\t\t|Timestamp|\t |Mime Type|")
133+
assert(err).
134+
Log().
135+
NoneErr(func() {
136+
for _, obj := range objects {
137+
fPrintf("%s %s %s\n", obj.Id, obj.Timestamp, obj.MimeType)
158138
}
159-
fPrintln(val)
160-
}
161-
} else {
162-
objects, _, err := Client.PubSub.GetObjectsAll(channel.Id, ciossdk.MakeGetObjectsOpts().Limit(limit).TimestampRange(timestampRange).Label(label).Offset(offset), context.Background())
163-
if len(objects) == 0 || resourceOwnerID != "" && channelsMap[channel.Id].ResourceOwnerId != resourceOwnerID {
164-
return
165-
}
166-
fPrintf("\n|Channel ID| : %s \n|Channel Name|: %s\n\n", channel.Id, channelsMap[channel.Id].Name)
167-
fPrintln("\t|ID|\t\t|Timestamp|\t |Mime Type|")
168-
assert(err).
169-
Log().
170-
NoneErr(func() {
171-
for _, obj := range objects {
172-
fPrintf("%s %s %s\n", obj.Id, obj.Timestamp, obj.MimeType)
173-
}
174-
})
139+
})
140+
141+
}
142+
printData := func(channelId string) []string {
143+
data, err := Client.PubSub.GetStreamAll(channelId, ciossdk.MakeGetStreamOpts().
144+
Limit(limit).Offset(offset).
145+
PackerFormat(packerFormat).
146+
TimestampRange(timestampRange).
147+
Label(label),
148+
context.Background())
149+
assert(err).Log()
150+
fPrintf("\n|Channel ID| : %s \n|Channel Name|: %s\n\n", channelId, channelsMap[channelId].Name)
151+
for _, val := range data {
152+
fPrintln(val)
153+
}
154+
return data
155+
}
156+
printJob := func(channel cios.Channel, limit int64) {
157+
switch {
158+
case dataFlag:
159+
printData(channel.Id)
160+
default:
161+
printObject(channel.Id)
175162
}
176163
assert(out.Flush()).Log()
177164
}
@@ -181,14 +168,14 @@ func listDataStore() *cli.Command {
181168
channel, _, err := Client.PubSub.GetChannel(channelID, nil, nil, context.Background())
182169
assert(err).
183170
Log().
184-
NoneErr(func() { printObj(channel, limit) })
171+
NoneErr(func() { printJob(channel, limit) })
185172
} else {
186173
channels, _, err := Client.PubSub.GetChannelsAll(ciossdk.MakeGetChannelsOpts().ResourceOwnerId(resourceOwnerID), context.Background())
187174
assert(err).
188175
Log().
189176
NoneErr(func() {
190177
for _, channel := range channels {
191-
printObj(channel, limit)
178+
printJob(channel, limit)
192179
}
193180
})
194181
}
@@ -197,6 +184,121 @@ func listDataStore() *cli.Command {
197184
},
198185
}
199186
}
187+
func saveDataStore() *cli.Command {
188+
return &cli.Command{
189+
Name: "save",
190+
Aliases: []string{"download", "get", "sv", "dl"},
191+
Flags: []cli.Flag{
192+
&cli.StringFlag{Name: "channel_id", Aliases: []string{"c"}},
193+
&cli.StringFlag{Name: "packer_format", Aliases: []string{"pf"}, Value: "payload_only"},
194+
&cli.StringFlag{Name: "timestamp_range", Aliases: []string{"tr"}, DefaultText: "Now Time", Value: ":" + convert.MustStr(time.Now().UnixNano())},
195+
&cli.StringFlag{Name: "save_dir", Aliases: []string{"out"}},
196+
&cli.StringFlag{Name: "label", Aliases: []string{"lb"}},
197+
&cli.BoolFlag{Name: "data", Aliases: []string{"d"}},
198+
&cli.BoolFlag{Name: "indent", Aliases: []string{"idt", "idnt", "i"}},
199+
&cli.BoolFlag{Name: "collective", Aliases: []string{"compact", "coll", "collect"}},
200+
&cli.Int64Flag{Name: "limit", Aliases: []string{"l"}, DefaultText: "30", Value: 30},
201+
&cli.Int64Flag{Name: "offset", Aliases: []string{"o"}, DefaultText: "0", Value: 0},
202+
&cli.StringFlag{Name: "resource_owner_id", Aliases: []string{"r", "ro"}},
203+
&cli.StringFlag{Name: "replace_save_data_channel", Aliases: []string{"replace", "rep"}, Usage: "-replace <channel id>"},
204+
},
205+
Action: func(c *cli.Context) error {
206+
var (
207+
channelID = c.String("channel_id")
208+
packerFormat = c.String("packer_format")
209+
limit = c.Int64("limit")
210+
offset = c.Int64("offset")
211+
timestampRange = c.String("timestamp_range")
212+
resourceOwnerID = c.String("resource_owner_id")
213+
label = c.String("label")
214+
outputDir = wrapper.AsString(c.String("save_dir"))
215+
replaced = wrapper.AsString(c.String("replace_save_data_channel"))
216+
indent = c.Bool("indent")
217+
collective = c.Bool("collective")
218+
channelsMap, _, _ = Client.PubSub.GetChannelsMapByID(ciossdk.MakeGetChannelsOpts(), context.Background())
219+
stageDSDir = fmt.Sprintf("%s/%s", datastoreDir, models.GetStage())
220+
)
221+
replaceChannelId := func(data string) string {
222+
var jsonFormat cios.PackerFormatJson
223+
assert(convert.UnMarshalJson([]byte(data), &jsonFormat)).Log()
224+
jsonFormat.Header.ChannelId = replaced.Str()
225+
return convert.MustCompactJson(jsonFormat)
226+
}
227+
indentJson := func(data string) string {
228+
var jsonFormat cios.PackerFormatJson
229+
assert(convert.UnMarshalJson([]byte(data), &jsonFormat)).Log()
230+
return convert.MustIndentJson(jsonFormat)
231+
}
232+
job := func(channel cios.Channel, limit int64) {
233+
switch {
234+
case replaced != "":
235+
packerFormat = "json"
236+
fallthrough
237+
case outputDir == "":
238+
outputDir = wrapper.String(fmt.Sprintf("%s/%s/%s___%s", datastoreDir, models.GetStage(), channelsMap[channel.Id].Name, channel.Id))
239+
fallthrough
240+
default:
241+
data, err := Client.PubSub.GetStreamAll(channel.Id, ciossdk.MakeGetStreamOpts().
242+
Limit(limit).Offset(offset).
243+
PackerFormat(packerFormat).
244+
TimestampRange(timestampRange).
245+
Label(label),
246+
context.Background())
247+
assert(err).Log()
248+
for idx, val := range data {
249+
switch {
250+
case replaced.IsPreset():
251+
data[idx] = replaceChannelId(val)
252+
fallthrough
253+
case indent:
254+
data[idx] = indentJson(val)
255+
}
256+
}
257+
path(datastoreDir).CreateDir()
258+
path(stageDSDir).CreateDir()
259+
path(outputDir.Str()).CreateDir()
260+
switch {
261+
case collective:
262+
fileName := fmt.Sprintf("%s/%s_%s.txt",
263+
outputDir,
264+
packerFormat,
265+
strings.Replace(timestampRange, ":", "-", -1))
266+
assert(path(fileName).WriteFileAsString("[\n"+strings.Join(data, ",")+"]\n")).
267+
Log().
268+
NoneErrPrintln("Completed ", fileName)
269+
default:
270+
for idx, val := range data {
271+
allLength := len(str(len(data)))
272+
currentLength := len(str(idx))
273+
filePrefixZero := strings.Repeat("0", allLength-currentLength)
274+
fileName := fmt.Sprintf("%s/%s%d.txt", outputDir, filePrefixZero, idx)
275+
assert(path(fileName).WriteFileAsString(val)).
276+
Log().
277+
NoneErrPrintln("Completed ", fileName)
278+
}
279+
}
280+
}
281+
}
282+
283+
if channelID != "" {
284+
channel, _, err := Client.PubSub.GetChannel(channelID, nil, nil, context.Background())
285+
assert(err).
286+
Log().
287+
NoneErr(func() { job(channel, limit) })
288+
} else {
289+
channels, _, err := Client.PubSub.GetChannelsAll(ciossdk.MakeGetChannelsOpts().ResourceOwnerId(resourceOwnerID), context.Background())
290+
assert(err).
291+
Log().
292+
NoneErr(func() {
293+
for _, channel := range channels {
294+
job(channel, limit)
295+
}
296+
})
297+
}
298+
return nil
299+
},
300+
}
301+
}
200302
func embezzleDateStore() *cli.Command {
201303
return &cli.Command{
202304
Name: "embezzle",

‎cli/publishsubscribe/init.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"github.com/optim-corp/cios-cli/models"
55
"github.com/optim-corp/cios-cli/utils"
66
"github.com/optim-kazuhiro-seida/ftil"
7+
"github.com/optim-kazuhiro-seida/go-advance-type/convert"
78
)
89

910
var (
@@ -18,4 +19,5 @@ var (
1819
datastoreDir = models.DatastoreDir
1920
path = ftil.Path
2021
assert = utils.EAssert
22+
str = convert.MustStr
2123
)

‎go.mod‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,7 @@ require (
77
github.com/k0kubun/colorstring v0.0.0-20150214042306-9440f1994b88 // indirect
88
github.com/k0kubun/pp v3.0.1+incompatible
99
github.com/mitchellh/go-homedir v1.1.0
10-
github.com/mitchellh/mapstructure v1.4.0
11-
github.com/optim-corp/cios-golang-sdk v0.0.6
10+
github.com/optim-corp/cios-golang-sdk v0.1.0
1211
github.com/optim-kazuhiro-seida/ftil v1.0.0
1312
github.com/optim-kazuhiro-seida/go-advance-type v1.3.0
1413
github.com/optim-kazuhiro-seida/loglog v0.5.0
@@ -17,5 +16,5 @@ require (
1716
github.com/urfave/cli/v2 v2.2.0
1817
golang.org/x/sync v0.0.0-20201207232520-09787c993a3a
1918
gopkg.in/AlecAivazis/survey.v1 v1.8.8
20-
gopkg.in/yaml.v2 v2.3.0
19+
gopkg.in/yaml.v2 v2.3.0 // indirect
2120
)

‎go.sum‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,7 @@ github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ
119119
github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
120120
github.com/hinshun/vt10x v0.0.0-20180616224451-1954e6464174 h1:WlZsjVhE8Af9IcZDGgJGQpNflI3+MJSBhsgT5PCtzBQ=
121121
github.com/hinshun/vt10x v0.0.0-20180616224451-1954e6464174/go.mod h1:DqJ97dSdRW1W22yXSB90986pcOyQ7r45iio1KN2ez1A=
122+
github.com/iancoleman/strcase v0.1.2 h1:gnomlvw9tnV3ITTAxzKSgTF+8kFWcU/f+TgttpXGz1U=
122123
github.com/iancoleman/strcase v0.1.2/go.mod h1:SK73tn/9oHe+/Y0h39VT4UCxmurVJkR5NA7kMEAOgSE=
123124
github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
124125
github.com/json-iterator/go v1.1.10 h1:Kz6Cvnvv2wGdaG/V8yMvfkmNiXq9Ya2KUv4rouJJr68=
@@ -158,14 +159,14 @@ github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9
158159
github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
159160
github.com/optim-corp/cios-golang-sdk v0.0.6 h1:yPXciFu45QG308Ocqd9v+xbtRhuAMBYSSdrH8HbtNdI=
160161
github.com/optim-corp/cios-golang-sdk v0.0.6/go.mod h1:L2RQeKwAavA8J8YYpkwwLdFy3cjcIpppQOAPqzxkNXI=
162+
github.com/optim-corp/cios-golang-sdk v0.1.0 h1:AcRJVZMkoxmyWGU6qycgRBgXGnso4v7Zppcf1juaOkQ=
163+
github.com/optim-corp/cios-golang-sdk v0.1.0/go.mod h1:L2RQeKwAavA8J8YYpkwwLdFy3cjcIpppQOAPqzxkNXI=
161164
github.com/optim-kazuhiro-seida/ftil v1.0.0 h1:f+dj6KIPzRLTrq1oHY18eKDnXIqEhfNN0uRt2ticp+s=
162165
github.com/optim-kazuhiro-seida/ftil v1.0.0/go.mod h1:dTGehrWE16Kye+Ww40G8wSLjKKXOfWqgHSwISWrHpmU=
163166
github.com/optim-kazuhiro-seida/go-advance-type v1.2.4 h1:LUk8wNjGfO7/SYhW8ejmcrl3A4187jB2aW9VQaLPneI=
164167
github.com/optim-kazuhiro-seida/go-advance-type v1.2.4/go.mod h1:t8CQQ6DsvogVqWuXuTIaCQsbfSYqDJ1YOXm+RtZd2H0=
165168
github.com/optim-kazuhiro-seida/go-advance-type v1.3.0 h1:YNe0LH1d+hMiYRdg8g37VomGH97sDFvGXMese2HC2Ns=
166169
github.com/optim-kazuhiro-seida/go-advance-type v1.3.0/go.mod h1:t8CQQ6DsvogVqWuXuTIaCQsbfSYqDJ1YOXm+RtZd2H0=
167-
github.com/optim-kazuhiro-seida/loglog v0.4.0 h1:utoA/slmkTkVOrzDS5sWI2rsN0FyhI5SXiobssY10EI=
168-
github.com/optim-kazuhiro-seida/loglog v0.4.0/go.mod h1:49632KJ6jnHKLb+eA+knAR57nZ7yFpAfe8JziL2EEIo=
169170
github.com/optim-kazuhiro-seida/loglog v0.5.0 h1:cjE7K/UVDfbs+JGnamLElFHVRg04xOxA1lQSNCp57FM=
170171
github.com/optim-kazuhiro-seida/loglog v0.5.0/go.mod h1:ng87+rDVhUYVvJzd0c7T3paBwXNXhxlNGPmo7usFtkM=
171172
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=

0 commit comments

Comments
 (0)