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

Commit 7809a26

Browse files
optim-kazuhiro-seidaoptim-kazuhiro-seida
authored
Merge pull request #8 from optim-corp/issue/7
update #7 datasore download
2 parents 3b63795 + 5e36f40 commit 7809a26

5 files changed

Lines changed: 172 additions & 69 deletions

File tree

‎cli/publishsubscribe/datastore.go‎

Lines changed: 163 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,120 @@ 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: "indent", Aliases: []string{"idt", "idnt", "i"}},
198+
&cli.BoolFlag{Name: "collective", Aliases: []string{"compact", "coll", "collect"}},
199+
&cli.Int64Flag{Name: "limit", Aliases: []string{"l"}, DefaultText: "30", Value: 30},
200+
&cli.Int64Flag{Name: "offset", Aliases: []string{"o"}, DefaultText: "0", Value: 0},
201+
&cli.StringFlag{Name: "resource_owner_id", Aliases: []string{"r", "ro"}},
202+
&cli.StringFlag{Name: "replace_save_data_channel", Aliases: []string{"replace", "rep"}, Usage: "-replace <channel id>"},
203+
},
204+
Action: func(c *cli.Context) error {
205+
var (
206+
channelID = c.String("channel_id")
207+
packerFormat = c.String("packer_format")
208+
limit = c.Int64("limit")
209+
offset = c.Int64("offset")
210+
timestampRange = c.String("timestamp_range")
211+
resourceOwnerID = c.String("resource_owner_id")
212+
label = c.String("label")
213+
outputDir = wrapper.AsString(c.String("save_dir"))
214+
replaced = wrapper.AsString(c.String("replace_save_data_channel"))
215+
indent = c.Bool("indent")
216+
collective = c.Bool("collective")
217+
channelsMap, _, _ = Client.PubSub.GetChannelsMapByID(ciossdk.MakeGetChannelsOpts(), context.Background())
218+
stageDSDir = fmt.Sprintf("%s/%s", datastoreDir, models.GetStage())
219+
)
220+
replaceChannelId := func(data string) string {
221+
var jsonFormat cios.PackerFormatJson
222+
assert(convert.UnMarshalJson([]byte(data), &jsonFormat)).Log()
223+
jsonFormat.Header.ChannelId = replaced.Str()
224+
return convert.MustCompactJson(jsonFormat)
225+
}
226+
indentJson := func(data string) string {
227+
var jsonFormat interface{}
228+
assert(convert.UnMarshalJson([]byte(data), &jsonFormat)).Log()
229+
return convert.MustIndentJson(jsonFormat)
230+
}
231+
job := func(channel cios.Channel, limit int64) {
232+
switch {
233+
case replaced != "":
234+
packerFormat = "json"
235+
fallthrough
236+
case outputDir == "":
237+
outputDir = wrapper.String(fmt.Sprintf("%s/%s/%s___%s", datastoreDir, models.GetStage(), channelsMap[channel.Id].Name, channel.Id))
238+
fallthrough
239+
default:
240+
data, err := Client.PubSub.GetStreamAll(channel.Id, ciossdk.MakeGetStreamOpts().
241+
Limit(limit).Offset(offset).
242+
PackerFormat(packerFormat).
243+
TimestampRange(timestampRange).
244+
Label(label),
245+
context.Background())
246+
assert(err).Log()
247+
for idx, val := range data {
248+
switch {
249+
case replaced.IsPreset():
250+
data[idx] = replaceChannelId(val)
251+
fallthrough
252+
case indent:
253+
data[idx] = indentJson(val)
254+
}
255+
}
256+
path(datastoreDir).CreateDir()
257+
path(stageDSDir).CreateDir()
258+
path(outputDir.Str()).CreateDir()
259+
switch {
260+
case collective:
261+
fileName := fmt.Sprintf("%s/%s_%s.txt",
262+
outputDir,
263+
packerFormat,
264+
strings.Replace(timestampRange, ":", "-", -1))
265+
assert(path(fileName).WriteFileAsString("[\n"+strings.Join(data, ",")+"]\n")).
266+
Log().
267+
NoneErrPrintln("Completed ", fileName)
268+
default:
269+
for idx, val := range data {
270+
allLength := len(str(len(data)))
271+
currentLength := len(str(idx))
272+
filePrefixZero := strings.Repeat("0", allLength-currentLength)
273+
fileName := fmt.Sprintf("%s/%s%d.txt", outputDir, filePrefixZero, idx)
274+
assert(path(fileName).WriteFileAsString(val)).
275+
Log().
276+
NoneErrPrintln("Completed ", fileName)
277+
}
278+
}
279+
}
280+
}
281+
282+
if channelID != "" {
283+
channel, _, err := Client.PubSub.GetChannel(channelID, nil, nil, context.Background())
284+
assert(err).
285+
Log().
286+
NoneErr(func() { job(channel, limit) })
287+
} else {
288+
channels, _, err := Client.PubSub.GetChannelsAll(ciossdk.MakeGetChannelsOpts().ResourceOwnerId(resourceOwnerID), context.Background())
289+
assert(err).
290+
Log().
291+
NoneErr(func() {
292+
for _, channel := range channels {
293+
job(channel, limit)
294+
}
295+
})
296+
}
297+
return nil
298+
},
299+
}
300+
}
200301
func embezzleDateStore() *cli.Command {
201302
return &cli.Command{
202303
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=

‎main.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,8 @@ import (
2525
)
2626

2727
const (
28-
NAME = "Could CLI"
29-
VERSION = "0.2.4"
28+
NAME = "Could IoT OS CLI"
29+
VERSION = "0.3.0"
3030
)
3131

3232
var (

0 commit comments

Comments
 (0)