diff --git a/apier/v1/cdrs.go b/apier/v1/cdrs.go index 7879f64d5..f0ad83f5f 100644 --- a/apier/v1/cdrs.go +++ b/apier/v1/cdrs.go @@ -25,28 +25,23 @@ import ( "os" "path" "time" + "strings" ) -type AttrExpCsvCdrs struct { - TimeStart string // If provided, will represent the starting of the CDRs interval (>=) - TimeEnd string // If provided, will represent the end of the CDRs interval (<) -} - -type ExportedCsvCdrs struct { - ExportedFilePath string // Full path to the newly generated export file - NumberOfCdrs int // Number of CDRs in the export file -} - -func (self *ApierV1) ExportCsvCdrs(attr *AttrExpCsvCdrs, reply *ExportedCsvCdrs) error { +func (self *ApierV1) ExportCdrsToFile(attr utils.AttrExpFileCdrs, reply *utils.ExportedFileCdrs) error { var tStart, tEnd time.Time var err error + cdrFormat := strings.ToLower(attr.CdrFormat) + if !utils.IsSliceMember(utils.CdreCdrFormats, cdrFormat) { + return fmt.Errorf("%s:%s", utils.ERR_MANDATORY_IE_MISSING, "CdrFormat") + } if len(attr.TimeStart) != 0 { - if tStart, err = utils.ParseDate(attr.TimeStart); err != nil { + if tStart, err = utils.ParseTimeDetectLayout(attr.TimeStart); err != nil { return err } } if len(attr.TimeEnd) != 0 { - if tEnd, err = utils.ParseDate(attr.TimeEnd); err != nil { + if tEnd, err = utils.ParseTimeDetectLayout(attr.TimeEnd); err != nil { return err } } @@ -54,21 +49,33 @@ func (self *ApierV1) ExportCsvCdrs(attr *AttrExpCsvCdrs, reply *ExportedCsvCdrs) if err != nil { return err } - fileName := path.Join(self.Config.CdreDir, "cgr", "csv", fmt.Sprintf("cdrs_%d.csv", time.Now().Unix())) - fileOut, err := os.Create(fileName) - if err != nil { - return err - } else { - defer fileOut.Close() - } - csvWriter := cdrexporter.NewCsvCdrWriter(fileOut, self.Config.RoundingDecimals, self.Config.CdreExtraFields) - for _, cdr := range cdrs { - if err := csvWriter.Write(cdr); err != nil { - os.Remove(fileName) + var fileName string + if cdrFormat == utils.CDRE_CSV && len(cdrs) != 0 { + fileName = path.Join(self.Config.CdreDir, fmt.Sprintf("cdrs_%d.csv", time.Now().Unix())) + fileOut, err := os.Create(fileName) + if err != nil { return err + } else { + defer fileOut.Close() + } + csvWriter := cdrexporter.NewCsvCdrWriter(fileOut, self.Config.RoundingDecimals, self.Config.CdreExtraFields) + for _, cdr := range cdrs { + if err := csvWriter.Write(cdr); err != nil { + os.Remove(fileName) + return err + } + } + csvWriter.Close() + if attr.RemoveFromDb { + cgrIds := make([]string, len(cdrs)) + for idx, cdr := range cdrs { + cgrIds[idx] = cdr.CgrId + } + if err := self.CdrDb.RemRatedCdrs(cgrIds); err != nil { + return err + } } } - csvWriter.Close() - *reply = ExportedCsvCdrs{fileName, len(cdrs)} + *reply = utils.ExportedFileCdrs{fileName, len(cdrs)} return nil } diff --git a/cdrexporter/csv.go b/cdrexporter/csv.go index 2406b2b29..dde1535b9 100644 --- a/cdrexporter/csv.go +++ b/cdrexporter/csv.go @@ -37,7 +37,7 @@ func NewCsvCdrWriter(writer io.Writer, roundDecimals int, extraFields []string) } func (dcw *CsvCdrWriter) Write(cdr *utils.RatedCDR) error { - primaryFields := []string{cdr.CgrId, cdr.AccId, cdr.CdrHost, cdr.ReqType, cdr.Direction, cdr.Tenant, cdr.TOR, cdr.Account, cdr.Subject, + primaryFields := []string{cdr.CgrId, cdr.MediationRunId, cdr.AccId, cdr.CdrHost, cdr.ReqType, cdr.Direction, cdr.Tenant, cdr.TOR, cdr.Account, cdr.Subject, cdr.Destination, cdr.AnswerTime.String(), strconv.Itoa(int(cdr.Duration)), strconv.FormatFloat(cdr.Cost, 'f', dcw.roundDecimals, 64)} if len(dcw.extraFields) == 0 { dcw.extraFields = utils.MapKeys(cdr.ExtraFields) diff --git a/cdrexporter/csv_test.go b/cdrexporter/csv_test.go index 6b9a4a2a8..0b518b0c1 100644 --- a/cdrexporter/csv_test.go +++ b/cdrexporter/csv_test.go @@ -30,12 +30,12 @@ func TestCsvCdrWriter(t *testing.T) { writer := &bytes.Buffer{} csvCdrWriter := NewCsvCdrWriter(writer, 4, []string{"extra3", "extra1"}) ratedCdr := &utils.RatedCDR{CgrId: utils.FSCgrId("dsafdsaf"), AccId: "dsafdsaf", CdrHost: "192.168.1.1", ReqType: "rated", Direction: "*out", Tenant: "cgrates.org", - TOR: "call", Account: "1001", Subject: "1001", Destination: "1002", AnswerTime: time.Unix(1383813746, 0).UTC(), Duration: 10, + TOR: "call", Account: "1001", Subject: "1001", Destination: "1002", AnswerTime: time.Unix(1383813746, 0).UTC(), Duration: 10, MediationRunId: utils.DEFAULT_RUNID, ExtraFields: map[string]string{"extra1": "val_extra1", "extra2": "val_extra2", "extra3": "val_extra3"}, Cost: 1.01, } csvCdrWriter.Write(ratedCdr) csvCdrWriter.Close() - expected := "b18944ef4dc618569f24c27b9872827a242bad0c,dsafdsaf,192.168.1.1,rated,*out,cgrates.org,call,1001,1001,1002,2013-11-07 08:42:26 +0000 UTC,10,1.0100,val_extra3,val_extra1" + expected := "b18944ef4dc618569f24c27b9872827a242bad0c,default,dsafdsaf,192.168.1.1,rated,*out,cgrates.org,call,1001,1001,1002,2013-11-07 08:42:26 +0000 UTC,10,1.0100,val_extra3,val_extra1" result := strings.TrimSpace(writer.String()) if result != expected { t.Errorf("Expected %s received %s.", expected, result) diff --git a/console/export_cdrs.go b/console/export_cdrs.go new file mode 100644 index 000000000..b864662ab --- /dev/null +++ b/console/export_cdrs.go @@ -0,0 +1,95 @@ +/* +Rating system designed to be used in VoIP Carriers World +Copyright (C) 2013 ITsysCOM + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program. If not, see +*/ + +package console + +import ( + "fmt" + "github.com/cgrates/cgrates/utils" + "time" +) + +func init() { + commands["export_cdrs"] = &CmdExportCdrs{} +} + +// Commander implementation +type CmdExportCdrs struct { + rpcMethod string + rpcParams *utils.AttrExpFileCdrs + rpcResult utils.ExportedFileCdrs +} + +// name should be exec's name +func (self *CmdExportCdrs) Usage(name string) string { + return fmt.Sprintf("\n\tUsage: cgr-console [cfg_opts...{-h}] export_cdrs [ [ [remove_from_db]]]") +} + +// set param defaults +func (self *CmdExportCdrs) defaults() error { + self.rpcMethod = "ApierV1.ExportCdrsToFile" + self.rpcParams = &utils.AttrExpFileCdrs{CdrFormat:"csv"} + return nil +} + +func (self *CmdExportCdrs) FromArgs(args []string) error { + self.defaults() + var timeStart, timeEnd string + if len(args) < 3 { + return fmt.Errorf(self.Usage("")) + } + if !utils.IsSliceMember(utils.CdreCdrFormats, args[2]) { + return fmt.Errorf(self.Usage("")) + } + self.rpcParams.CdrFormat = args[2] + switch len(args) { + case 4: + timeStart = args[3] + + case 5: + timeStart = args[3] + timeEnd = args[4] + case 6: + timeStart = args[3] + timeEnd = args[4] + if args[5] == "remove_from_db" { + self.rpcParams.RemoveFromDb = true + } + } + if timeStart == "*one_month" { + now := time.Now() + self.rpcParams.TimeStart = now.AddDate(0,-1,0).String() + self.rpcParams.TimeEnd = now.String() + } else { + self.rpcParams.TimeStart = timeStart + self.rpcParams.TimeEnd = timeEnd + } + return nil +} + +func (self *CmdExportCdrs) RpcMethod() string { + return self.rpcMethod +} + +func (self *CmdExportCdrs) RpcParams() interface{} { + return self.rpcParams +} + +func (self *CmdExportCdrs) RpcResult() interface{} { + return &self.rpcResult +} diff --git a/engine/storage_interface.go b/engine/storage_interface.go index 4ba43e621..ca07d11cf 100644 --- a/engine/storage_interface.go +++ b/engine/storage_interface.go @@ -96,6 +96,7 @@ type CdrStorage interface { SetCdr(utils.RawCDR) error SetRatedCdr(*utils.RatedCDR, string) error GetRatedCdrs(time.Time, time.Time) ([]*utils.RatedCDR, error) + RemRatedCdrs([]string) error } type LogStorage interface { diff --git a/engine/storage_sql.go b/engine/storage_sql.go index 4c053801d..ae03e59af 100644 --- a/engine/storage_sql.go +++ b/engine/storage_sql.go @@ -760,14 +760,14 @@ func (self *SQLStorage) GetRatedCdrs(timeStart, timeEnd time.Time) ([]*utils.Rat for rows.Next() { var cgrid, accid, cdrhost, cdrsrc, reqtype, direction, tenant, tor, account, subject, destination, runid string var extraFields []byte - var answerTimestamp, duration int64 + var answerTime time.Time + var duration int64 var cost float64 var extraFieldsMp map[string]string - if err := rows.Scan(&cgrid, &accid, &cdrhost, &cdrsrc, &reqtype, &direction, &tenant, &tor, &account, &subject, &destination, &answerTimestamp, &duration, + if err := rows.Scan(&cgrid, &accid, &cdrhost, &cdrsrc, &reqtype, &direction, &tenant, &tor, &account, &subject, &destination, &answerTime, &duration, &extraFields, &runid, &cost); err != nil { return nil, err } - answerTime := time.Unix(answerTimestamp, 0) if err := json.Unmarshal(extraFields, &extraFieldsMp); err != nil { return nil, err } @@ -781,6 +781,32 @@ func (self *SQLStorage) GetRatedCdrs(timeStart, timeEnd time.Time) ([]*utils.Rat return cdrs, nil } +// Remove CDR data out of all CDR tables based on their cgrid +func (self *SQLStorage) RemRatedCdrs(cgrIds []string) error { + if len(cgrIds) == 0 { + return nil + } + buffRated := bytes.NewBufferString(fmt.Sprintf("DELETE FROM %s WHERE", utils.TBL_RATED_CDRS)) + buffCosts := bytes.NewBufferString(fmt.Sprintf("DELETE FROM %s WHERE", utils.TBL_COST_DETAILS)) + buffCdrExtra := bytes.NewBufferString(fmt.Sprintf("DELETE FROM %s WHERE",utils.TBL_CDRS_EXTRA)) + buffCdrPrimary := bytes.NewBufferString(fmt.Sprintf("DELETE FROM %s WHERE",utils.TBL_CDRS_PRIMARY)) + qryBuffers := []*bytes.Buffer{buffRated, buffCosts, buffCdrExtra, buffCdrPrimary} + for idx, cgrId := range cgrIds { + for _, buffer := range qryBuffers { + if idx != 0 { + buffer.WriteString(" OR") + } + buffer.WriteString(fmt.Sprintf(" cgrid='%s'",cgrId)) + } + } + for _, buffer := range qryBuffers { + if _, err := self.Db.Exec(buffer.String()); err != nil { + return err + } + } + return nil +} + func (self *SQLStorage) GetTpDestinations(tpid, tag string) ([]*Destination, error) { var dests []*Destination q := fmt.Sprintf("SELECT * FROM %s WHERE tpid='%s'", utils.TBL_TP_DESTINATIONS, tpid) diff --git a/utils/apitpdata.go b/utils/apitpdata.go index 0d021870a..aba11f578 100644 --- a/utils/apitpdata.go +++ b/utils/apitpdata.go @@ -304,3 +304,16 @@ type CachedItemAge struct { RatingProfile time.Duration Action time.Duration } + +type AttrExpFileCdrs struct { + CdrFormat string // Cdr output file format + TimeStart string // If provided, will represent the starting of the CDRs interval (>=) + TimeEnd string // If provided, will represent the end of the CDRs interval (<) + RemoveFromDb bool // If true the CDRs will be also deleted after export + +} + +type ExportedFileCdrs struct { + ExportedFilePath string // Full path to the newly generated export file + NumberOfRecords int // Number of CDRs in the export file +} diff --git a/utils/consts.go b/utils/consts.go index 0d83fad81..5f17455f2 100644 --- a/utils/consts.go +++ b/utils/consts.go @@ -78,4 +78,10 @@ const ( DURATION = "duration" DEFAULT_RUNID = "default" STATIC_VALUE_PREFIX = "^" + CDRE_CSV = "csv" + CDRE_DRYRUN = "dry_run" +) + +var ( + CdreCdrFormats = []string{CDRE_CSV, CDRE_DRYRUN} ) diff --git a/utils/coreutils.go b/utils/coreutils.go index 082a333d1..eaf44b603 100644 --- a/utils/coreutils.go +++ b/utils/coreutils.go @@ -101,12 +101,15 @@ func Round(x float64, prec int, method string) float64 { func ParseTimeDetectLayout(tmStr string) (time.Time, error) { var nilTime time.Time - rfc3339Rule := regexp.MustCompile(`^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.+`) - sqlRule := regexp.MustCompile(`^\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}`) - unixTimestampRule := regexp.MustCompile(`^\d{10}`) + rfc3339Rule := regexp.MustCompile(`^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.+$`) + sqlRule := regexp.MustCompile(`^\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}$`) + gotimeRule := regexp.MustCompile(`^\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}\.?\d*\s[+,-]\d+\s\w+$`) + unixTimestampRule := regexp.MustCompile(`^\d{10}$`) switch { case rfc3339Rule.MatchString(tmStr): return time.Parse(time.RFC3339, tmStr) + case gotimeRule.MatchString(tmStr): + return time.Parse("2006-01-02 15:04:05.999999999 -0700 MST", tmStr) case sqlRule.MatchString(tmStr): return time.Parse("2006-01-02 15:04:05", tmStr) case unixTimestampRule.MatchString(tmStr): diff --git a/utils/utils_test.go b/utils/utils_test.go index 51089fe4d..cbc0924dd 100644 --- a/utils/utils_test.go +++ b/utils/utils_test.go @@ -156,6 +156,28 @@ func TestParseTimeDetectLayout(t *testing.T) { if err == nil { t.Errorf("Expecting error") } + goTmStr := "2013-12-30 15:00:01 +0000 UTC" + goTm, err := ParseTimeDetectLayout(goTmStr) + if err != nil { + t.Error(err) + } else if !goTm.Equal(expectedTime) { + t.Errorf("Unexpected time parsed: %v, expecting: %v", goTm, expectedTime) + } + _, err = ParseTimeDetectLayout(goTmStr[1:]) + if err == nil { + t.Errorf("Expecting error") + } + goTmStr = "2013-12-30 15:00:01.000000000 +0000 UTC" + goTm, err = ParseTimeDetectLayout(goTmStr) + if err != nil { + t.Error(err) + } else if !goTm.Equal(expectedTime) { + t.Errorf("Unexpected time parsed: %v, expecting: %v", goTm, expectedTime) + } + _, err = ParseTimeDetectLayout(goTmStr[1:]) + if err == nil { + t.Errorf("Expecting error") + } } func TestParseDateUnix(t *testing.T) {