09-12-2022
This commit is contained in:
Jonathan Jenne
@@ -1,5 +1,6 @@
|
||||
Imports DigitalData.Modules.Database
|
||||
Imports DigitalData.Modules.Logging
|
||||
Imports DigitalData.Modules.Windream
|
||||
Imports ECM.JobRunner.Common
|
||||
Imports Quartz
|
||||
|
||||
@@ -8,11 +9,14 @@ Namespace Scheduler.Jobs
|
||||
Friend LogConfig As LogConfig
|
||||
Friend Logger As Logger
|
||||
Friend Database As MSSQLServer
|
||||
Friend Windream As Windream
|
||||
Friend State As State
|
||||
|
||||
Friend Id As Integer
|
||||
Friend Name As String
|
||||
|
||||
Friend ResultItems As New List(Of HistoryItem.HistoryStep)
|
||||
|
||||
Private ctx As IJobExecutionContext
|
||||
|
||||
Public Function InitializeJob(context As IJobExecutionContext) As Dictionary(Of String, String)
|
||||
@@ -23,6 +27,7 @@ Namespace Scheduler.Jobs
|
||||
LogConfig = oJobData.Item(Constants.Scheduler.JOB_CONFIG_LOGCONFIG)
|
||||
Database = oJobData.Item(Constants.Scheduler.JOB_CONFIG_DATABASE)
|
||||
State = oJobData.Item(Constants.Scheduler.JOB_CONFIG_STATE)
|
||||
Windream = oJobData.Item(Constants.Scheduler.JOB_CONFIG_WINDREAM)
|
||||
Logger = LogConfig.GetLogger()
|
||||
|
||||
Id = Integer.Parse(oArgs.Item("Id"))
|
||||
@@ -36,9 +41,45 @@ Namespace Scheduler.Jobs
|
||||
State.JobStatus.Update(ctx, pCurrentValue, pTotalValue)
|
||||
End Sub
|
||||
|
||||
Public Sub CompleteJob()
|
||||
State.JobStatus.Complete(ctx)
|
||||
Public Sub LogStep(pLevel As HistoryItem.StepLevel, pMessage As String, ParamArray pArgs As String())
|
||||
ResultItems.Add(New HistoryItem.HistoryStep With {
|
||||
.Message = String.Format(pMessage, pArgs),
|
||||
.Level = pLevel
|
||||
})
|
||||
End Sub
|
||||
|
||||
Public Function CompleteJob() As Task(Of Boolean)
|
||||
ctx.Result = New JobResult With {
|
||||
.Successful = True,
|
||||
.Steps = ResultItems,
|
||||
.Description = "Job completed."
|
||||
}
|
||||
State.JobStatus.Complete(ctx)
|
||||
|
||||
Return Task.FromResult(True)
|
||||
End Function
|
||||
|
||||
Public Function CompleteJob(pCompletionMessage As String) As Task(Of Boolean)
|
||||
ctx.Result = New JobResult With {
|
||||
.Successful = True,
|
||||
.Steps = ResultItems,
|
||||
.Description = pCompletionMessage
|
||||
}
|
||||
State.JobStatus.Complete(ctx)
|
||||
|
||||
Return Task.FromResult(True)
|
||||
End Function
|
||||
|
||||
Public Function CompleteJob(pException As Exception) As Task(Of Boolean)
|
||||
ctx.Result = New JobResult With {
|
||||
.Successful = False,
|
||||
.Steps = ResultItems,
|
||||
.Description = pException.Message
|
||||
}
|
||||
State.JobStatus.Complete(ctx)
|
||||
|
||||
Return Task.FromResult(False)
|
||||
End Function
|
||||
End Class
|
||||
|
||||
End Namespace
|
||||
@@ -18,7 +18,7 @@ Namespace Scheduler.Jobs
|
||||
|
||||
context.Result = oResult
|
||||
|
||||
CompleteJob()
|
||||
CompleteJob("Done!")
|
||||
Return Task.FromResult(True)
|
||||
End Function
|
||||
End Class
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
Imports ECM.JobRunner.Windows.Scheduler.Jobs
|
||||
Imports Quartz
|
||||
|
||||
|
||||
Imports System.Text.RegularExpressions
|
||||
Imports DigitalData.Modules.Filesystem
|
||||
|
||||
Namespace Scheduler.Jobs
|
||||
|
||||
@@ -50,47 +50,271 @@ Namespace Scheduler.Jobs
|
||||
Inherits BaseJob
|
||||
Implements IJob
|
||||
|
||||
Private FileEx As File
|
||||
|
||||
Public Function Execute(context As IJobExecutionContext) As Task Implements IJob.Execute
|
||||
Dim oArgs = MyBase.InitializeJob(context)
|
||||
|
||||
Try
|
||||
Logger.Info("Running File Import [{0}]", Name)
|
||||
FileEx = New File(LogConfig)
|
||||
|
||||
Dim oProfile = State.ProfileDefintions.ImportProfiles.Where(Function(p) p.JobId = Id).SingleOrDefault()
|
||||
|
||||
Dim oSourceDirectory As String = oProfile.SourceFolder
|
||||
Dim oRecursive As Boolean = oProfile.IncludeSubfolders
|
||||
Dim oFiles = GetFiles(oSourceDirectory, oRecursive)
|
||||
Logger.Info("Running File Import [{0}]", Name)
|
||||
Logger.Info("Source directory: [{0}]", oProfile.SourceFolder)
|
||||
Logger.Info("Target directory: [{0}]", oProfile.TargetFolder)
|
||||
Logger.Info("Backup directory: [{0}]", oProfile.BackupFolder)
|
||||
|
||||
If oFiles.Count = 0 Then
|
||||
Logger.Info("No Files for Profile [{0}]", Name)
|
||||
Return Task.FromResult(True)
|
||||
Dim oObjectType = State.ObjectTypes.Where(Function(o) o.Name = oProfile.ObjectTypeName).SingleOrDefault()
|
||||
|
||||
If IO.Directory.Exists(oProfile.SourceFolder) = False Then
|
||||
LogStep(HistoryItem.StepLevel.Error, "Source directory [{0}] does not exist!", oProfile.SourceFolder)
|
||||
Return Task.FromResult(False)
|
||||
End If
|
||||
|
||||
Logger.Info("[0] files found in source directory [{1}]", oFiles.Count, oSourceDirectory)
|
||||
Dim oRecursive As Boolean = oProfile.IncludeSubfolders
|
||||
Dim oFileNames = GetFiles(oProfile.SourceFolder, oRecursive)
|
||||
Dim oRegexList As List(Of Regex) = GetRegexList(oProfile.FileExcludeRegex)
|
||||
|
||||
'Dim oMax = 100
|
||||
'For index = 1 To oMax
|
||||
' UpdateProgress(index, oMax)
|
||||
' Threading.Thread.Sleep(100)
|
||||
'Next
|
||||
Logger.Debug("[{0}] Regexes loaded", oRegexList.Count)
|
||||
|
||||
Dim oResult = New JobResult() With {
|
||||
.Description = $"File Import Job [{Name}] completed!"
|
||||
}
|
||||
If oFileNames.Count = 0 Then
|
||||
Logger.Info("No Files for Profile [{0}]", Name)
|
||||
Return CompleteJob("No files for profile")
|
||||
End If
|
||||
|
||||
context.Result = oResult
|
||||
Logger.Info("[{0}] files found in source directory [{1}]", oFileNames.Count, oProfile.SourceFolder)
|
||||
LogStep(HistoryItem.StepLevel.Info, "{0} files found in source directory {1}", oFileNames.Count, oProfile.SourceFolder)
|
||||
|
||||
Return Task.FromResult(True)
|
||||
' - [ ] Process Rules, build list of files and indexes
|
||||
' - [x] Process Regex to filter out files
|
||||
' - [x] Check time to filter out files
|
||||
' - [ ] (Check if files can be accessed)
|
||||
' - [ ] Check if backup is needed and backup files
|
||||
' - [ ] Import into windream
|
||||
' - [ ] Create original subfolder structure
|
||||
' - [ ] Create DateTime Subfolders
|
||||
' - [x] Check if file exists and version
|
||||
' - [x] Do import
|
||||
' - [ ] Check for filesize 0
|
||||
' - [ ] Write indexes (using data from getimportfile)
|
||||
' - [ ] Check if orig file should be deleted
|
||||
' - [ ] (delete subdirectories in source path)
|
||||
|
||||
Dim oFiles = oFileNames.
|
||||
Select(Function(f) New ImportFile(f)).
|
||||
ToList()
|
||||
Dim oFilteredFiles = oFiles
|
||||
|
||||
' Check time to filter out files
|
||||
Dim oDateFilteredFiles = oFilteredFiles.Where(Function(f) FileIsOlderThan(f, pMinutes:=1)).ToList()
|
||||
Dim oDateFilteredCount = oFilteredFiles.Except(oDateFilteredFiles).Count()
|
||||
Logger.Debug("[{0}] Files filtered out for being too new.", oDateFilteredCount)
|
||||
LogStep(HistoryItem.StepLevel.Debug, "{0} Files filtered out for being too new.", oDateFilteredCount)
|
||||
oFilteredFiles = oDateFilteredFiles
|
||||
|
||||
' Process Regex to filter out files
|
||||
Dim oRegexFilteredFiles = oFilteredFiles.Where(Function(f) Not FileMatchesRegex(f, oRegexList))
|
||||
Dim oRegexFilteredCount = oFilteredFiles.Except(oRegexFilteredFiles).Count()
|
||||
Logger.Debug("[{0}] Files filtered out for matching exclusion Regex.", oRegexFilteredCount)
|
||||
LogStep(HistoryItem.StepLevel.Debug, "{0} Files filtered out for matching exclusion Regex.", oRegexFilteredCount)
|
||||
oFilteredFiles = oDateFilteredFiles
|
||||
|
||||
Logger.Info("Importing files..")
|
||||
Dim oImportedFiles As New List(Of ImportFile)
|
||||
For Each oFile In oFilteredFiles
|
||||
|
||||
Dim oImportedPath = ImportFile(oFile, oProfile)
|
||||
If oImportedPath Is Nothing Then
|
||||
Logger.Warn("File [{0}] could not be imported!", oFile.FilePath)
|
||||
Continue For
|
||||
End If
|
||||
oFile.FilePathWindream = oImportedPath
|
||||
|
||||
oImportedFiles.Add(oFile)
|
||||
Next
|
||||
|
||||
Logger.Info("[{0}] files successfully Imported!", oImportedFiles.Count)
|
||||
LogStep(HistoryItem.StepLevel.Info, "{0} files successfully Imported!", oImportedFiles.Count)
|
||||
|
||||
Dim oIndexedFiles As New List(Of ImportFile)
|
||||
For Each oFile In oImportedFiles
|
||||
Logger.Info("Indexing file [{0}]", oFile.FilePathWindream)
|
||||
Dim oIndexResult = IndexFile(oFile, oProfile)
|
||||
If oIndexResult = True Then
|
||||
oIndexedFiles.Add(oFile)
|
||||
End If
|
||||
Logger.Info("Indexing of file [{0}] done!", oFile.FilePathWindream)
|
||||
Next
|
||||
|
||||
Logger.Info("[{0}] files successfully Indexed!", oIndexedFiles.Count)
|
||||
LogStep(HistoryItem.StepLevel.Info, "{0} files successfully Indexed!", oIndexedFiles.Count)
|
||||
|
||||
Return CompleteJob($"{oImportedFiles.Count} files successfully Processed!")
|
||||
Catch ex As Exception
|
||||
Logger.Error(ex)
|
||||
|
||||
LogStep(HistoryItem.StepLevel.Error, "Unexpected Error: [{0}]", ex.Message)
|
||||
|
||||
Return Task.FromResult(False)
|
||||
Finally
|
||||
CompleteJob()
|
||||
Return CompleteJob(ex)
|
||||
End Try
|
||||
End Function
|
||||
|
||||
Private Function FileIsOlderThan(pFile As ImportFile, pMinutes As Integer)
|
||||
Return pFile.FileInfo.CreationTime.AddMinutes(pMinutes) < Now
|
||||
End Function
|
||||
|
||||
Private Function FileMatchesRegex(pFile As ImportFile, pRegexList As List(Of Regex))
|
||||
Return pRegexList.Any(Function(r) r.IsMatch(pFile.FilePath))
|
||||
End Function
|
||||
|
||||
Private Function GetRegexList(pRegexString As String) As List(Of Regex)
|
||||
Return pRegexString.
|
||||
Split(vbNewLine).
|
||||
ToList().
|
||||
Where(Function(r) String.IsNullOrWhiteSpace(r) = False).
|
||||
Select(Function(s)
|
||||
Logger.Debug("Regex loaded: [{0}]", s)
|
||||
Return New Regex(s)
|
||||
End Function).ToList()
|
||||
|
||||
End Function
|
||||
|
||||
Private Function ImportFile(pFile As ImportFile, pProfile As ImportProfile) As String
|
||||
'Check if target folder exists
|
||||
If Windream.TestFolderExists(pProfile.TargetFolder) = False Then
|
||||
If Windream.NewFolder(pProfile.TargetFolder) = False Then
|
||||
Logger.Warn("Folder [{0}] could not be created!", pProfile.TargetFolder)
|
||||
Return Nothing
|
||||
End If
|
||||
End If
|
||||
|
||||
' Generate new filepath and stream file
|
||||
Dim oFileName = IO.Path.GetFileName(pFile.FilePath)
|
||||
Dim oNewFilePath As String = IO.Path.Combine(pProfile.TargetFolder, oFileName)
|
||||
Dim oVersionedFilePath = GetVersionedWindreamPath(oNewFilePath)
|
||||
|
||||
If Windream.NewFileStream(pFile.FilePathOriginal, oVersionedFilePath, pProfile.ObjectTypeName) = False Then
|
||||
Logger.Warn("File [{0}] could not be streamed to path [{1}]!", pFile.FilePathOriginal, oNewFilePath)
|
||||
Return Nothing
|
||||
End If
|
||||
|
||||
Return oVersionedFilePath
|
||||
End Function
|
||||
|
||||
Private Function GetVersionedWindreamPath(pWindreamFilePath As String) As String
|
||||
Dim oAbsolutePath = Windream.GetAbsolutePath(pWindreamFilePath)
|
||||
|
||||
' This versions the filename but does not rely on access though the filesystem.
|
||||
' Instead the check for file existence is made through the windream sdk.
|
||||
Dim oVersionedPath = FileEx.GetVersionedFilenameWithFilecheck(oAbsolutePath, Function(pPath As String) Windream.TestFileExists(pPath))
|
||||
Dim oVersionedAndNormalizedPath = Windream.GetNormalizedPath(oVersionedPath, False)
|
||||
|
||||
Return oVersionedAndNormalizedPath
|
||||
End Function
|
||||
|
||||
Private Function GetImportFile(pFilename As String, pProfile As ImportProfile) As ImportFile
|
||||
Dim oImportFile = New ImportFile(pFilename)
|
||||
|
||||
Return oImportFile
|
||||
End Function
|
||||
|
||||
Private Function ProcessSteps(pFile As ImportFile, pProfile As ImportProfile) As List(Of ImportFile.IndexItem)
|
||||
Dim oIndexItems = New List(Of ImportFile.IndexItem)
|
||||
|
||||
' Process Steps on Filename
|
||||
For Each oStep In pProfile.Steps
|
||||
Dim oValue As String
|
||||
|
||||
' Get the base value by checking scope
|
||||
Select Case oStep.Scope
|
||||
Case ImportProfileStep.StepScope.FILE
|
||||
oValue = pFile.FileInfo.Name
|
||||
Case ImportProfileStep.StepScope.FOLDER
|
||||
oValue = pFile.FileInfo.DirectoryName
|
||||
Case Else
|
||||
oValue = pFile.FilePath
|
||||
End Select
|
||||
|
||||
|
||||
' TODO: Error handling!
|
||||
Select Case oStep.Method
|
||||
Case ImportProfileStep.StepMethod.SUBSTRING
|
||||
Try
|
||||
Dim oIndex1 = Integer.Parse(oStep.Argument1)
|
||||
Dim oLength = Integer.Parse(oStep.Argument2)
|
||||
oValue = oValue.Substring(oIndex1, oLength)
|
||||
Catch ex As Exception
|
||||
LogStep(HistoryItem.StepLevel.Error, "Method SUBSTRING could not be applied to Index '{0}'. Error: '{1}'", oStep.IndexName, ex.Message)
|
||||
Logger.Error(ex)
|
||||
End Try
|
||||
|
||||
Case ImportProfileStep.StepMethod.SPLIT
|
||||
Try
|
||||
Dim oSeparator = oStep.Argument1.Substring(0, 1)
|
||||
Dim oIndex = Integer.Parse(oStep.Argument2)
|
||||
Dim oSplit = oValue.Split(oSeparator)
|
||||
oValue = oSplit(oIndex)
|
||||
Catch ex As Exception
|
||||
LogStep(HistoryItem.StepLevel.Error, "Method SPLIT could not be applied to Index '{0}'. Error: '{1}'", oStep.IndexName, ex.Message)
|
||||
Logger.Error(ex)
|
||||
End Try
|
||||
|
||||
Case ImportProfileStep.StepMethod.REGEX
|
||||
Try
|
||||
Dim oRegex = New Regex(oStep.Argument1)
|
||||
Dim oTrueValue = oStep.Argument2
|
||||
Dim oFalseValue = oStep.Argument3
|
||||
If oRegex.IsMatch(oValue) Then
|
||||
oValue = oTrueValue
|
||||
Else
|
||||
oValue = oFalseValue
|
||||
End If
|
||||
Catch ex As Exception
|
||||
LogStep(HistoryItem.StepLevel.Error, "Method REGEX could not be applied to Index '{0}'. Error: '{1}'", oStep.IndexName, ex.Message)
|
||||
Logger.Error(ex)
|
||||
End Try
|
||||
|
||||
Case ImportProfileStep.StepMethod.VALUE
|
||||
oValue = oStep.Argument1
|
||||
|
||||
Case ImportProfileStep.StepMethod.ALL
|
||||
'noop
|
||||
Case Else
|
||||
'noop
|
||||
End Select
|
||||
|
||||
oIndexItems.Add(New ImportFile.IndexItem With {
|
||||
.IndexName = oStep.IndexName,
|
||||
.Value = oValue
|
||||
})
|
||||
Next
|
||||
|
||||
Return oIndexItems
|
||||
End Function
|
||||
|
||||
Private Function IndexFile(pFile As ImportFile, pProfile As ImportProfile) As Boolean
|
||||
Logger.Debug("Writing [{0}] indexes for File [{1}]", pFile.IndexValues.Count, pFile.FilePathWindream)
|
||||
|
||||
Dim oIndexItems = ProcessSteps(pFile, pProfile)
|
||||
|
||||
Dim oResults = oIndexItems.
|
||||
Select(Function(v)
|
||||
Dim oIndexResult = False
|
||||
Logger.Info("Writing Index [{0}] with value [{1}]", v.IndexName, v.Value)
|
||||
|
||||
If Windream.TestIndexNameIsVectorIndex(v.IndexName) Then
|
||||
oIndexResult = Windream.SetFileIndex(pFile.FilePathWindream, v.IndexName, New List(Of String) From {v.Value}, pProfile.ObjectTypeName)
|
||||
Else
|
||||
oIndexResult = Windream.SetFileIndex(pFile.FilePathWindream, v.IndexName, v.Value, pProfile.ObjectTypeName)
|
||||
End If
|
||||
|
||||
Return oIndexResult
|
||||
End Function).
|
||||
ToList()
|
||||
|
||||
' Return True if all Indexes were set correctly
|
||||
Return oResults.All(Function(r) r = True)
|
||||
End Function
|
||||
|
||||
Private Function GetFiles(pDirectory As String, pRecursive As Boolean) As String()
|
||||
|
||||
@@ -17,7 +17,7 @@ Namespace Scheduler.Jobs
|
||||
|
||||
context.Result = oResult
|
||||
|
||||
CompleteJob()
|
||||
CompleteJob("Done!")
|
||||
Return Task.FromResult(True)
|
||||
End Function
|
||||
End Class
|
||||
|
||||
Reference in New Issue
Block a user