1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201

#!/usr/bin/env python
#
#   Library:    pyflowctrl
#   Module:     textprocessing2
#   Dependency:    core3
#   Examples:   
#

import re
from core3 import EmptyStream, Stream, Process
from core3 import WAITING, PROCESSING

# ------------------------------------------
class TextSplitter(Process):
    ''' TextSplitter '''
    def __init__(self, separator='\n'):
        ''' __init__  '''
        super(TextSplitter, self).__init__()
        self.io = { 
            'input': Stream(),
            'output': Stream(),
        }
        self.separator = separator
        
    def main(self):
        while True:
            try:
                (path, data) = self.io['input'].get()
            except EmptyStream:
                yield WAITING
                continue    
                
            for part in data.split(self.separator):
                self.io['output'].put((path,part))            
                yield PROCESSING

# ------------------------------------------
class SelectorsNotADict(Exception): pass    

class TextSelector(Process):
    ''' TextSelector '''
    def __init__(self, selectors={}):
        ''' __init__  
        
        selectors are regular expressions for selecting text block'''
        super(TextSelector, self).__init__()
        self.io = { 
            'input': Stream(),
        }
        if isinstance(selectors, dict):
            self.selectors = selectors
            # for each selectors create own output stream
            for s in self.selectors:
                self.io[s] = Stream()
        else:
            raise SelectorsNotADict(selectors)
        
    def main(self):
        while True:
            try:
                (path, data) = self.io['input'].get()
            except EmptyStream:
                yield WAITING
                continue    
                
            for s in self.selectors:
                for results in re.finditer(self.selectors[s],data):
                    for result in results.groups():
                        self.io[s].put((path, result))
                        yield PROCESSING
            
# ------------------------------------------
class Text2Dict(Process):
    ''' Text2Dict '''
    def __init__(self, selectors={}):
        ''' __init__  
        
        selectors are regular expressions for conversion text block to dictionary '''
        super(Text2Dict, self).__init__()
        self.io = { 
            'input': Stream(),
            'output': Stream(),
        }
        if isinstance(selectors, dict):
            self.selectors = selectors
        else:
            raise UnknownSelectorsType(selectors)
    
    def prepare_string(self, s):
        res = s.rstrip().lstrip()
        if res:
            return res
        else:
            return None
    
    def prepare_list(self, lst):
        f_lst = []
        for s in lst:
            s = s.rstrip().lstrip()
            if s:
                f_lst.append(s)
        if len(f_lst) <> 0:
            return f_lst
        else:
            return None

    def prepare_results(self, results):
        ''' prepare results:
        - remove empty fields '''
        
        if isinstance(results, (list,tuple)):
            if len(results) == 1:
                return self.prepare_string(results[0])
            return self.prepare_list(results)        
            
        elif isinstance(results, str):
            return self.prepare_string(results)
            
        else:
            return None
        
    def main(self):
        while True:
            try:
                (path, data) = self.io['input'].get()
            except EmptyStream:
                yield WAITING
                continue

            res_dict = {}
            for s in self.selectors:
                results = re.findall(self.selectors[s],data)
                if s in res_dict:
                    f_results = self.prepare_results(results)
                    if f_results:
                        if isinstance(res_dict[s], (list,tuple)):
                            res_dict[s] = res_dict[s].extend(f_results)
                        elif isinstance(res_dict[s], str):
                            t = res_dict[s]
                            res_dict[s] = [t]
                            res_dict[s].extend(f_results)                            
                else:
                    f_results = self.prepare_results(results)
                    if f_results:
                        if isinstance(f_results, str):
                            res_dict[s] = f_results
                        elif isinstance(f_results, (tuple,list)):
                            res_dict[s] = []
                            res_dict[s].extend(f_results)
            if len(res_dict) <> 0:
                self.io['output'].put((path, res_dict))
            yield PROCESSING

# ------------------------------------------
class DictProcessing(Process):
    ''' DictProcessing '''
    def __init__(self, rules=()):
        ''' __init__  
        
        - rules are used for dictionary fields 
        processing and creation new fields if defined:
        
        (source_field, target_field, func)
        '''
        super(DictProcessing, self).__init__()
        self.io = { 
            'input': Stream(),
            'output': Stream(),
        }
        if isinstance(rules, (list,tuple)):
            self.rules = rules
        else:
            raise UnknownSelectorsType(selectors)
    
    def main(self):
        while True:
            try:
                (path, data) = self.io['input'].get()
            except EmptyStream:
                yield WAITING
                continue
                
            res_dict = {}
            for source_field, target_field, func in self.rules:
            
                if source_field not in data:
                    continue
                if target_field is None:
                    del(data[source_field])
                    continue
                if isinstance(data[source_field], (list, tuple)):
                    targets = []
                    for i, v in enumerate(data[source_field]):
                        targets.append(func(data[source_field][i]))
                    data[target_field] = targets
                else:
                    data[target_field] = func(data[source_field])
            if len(data) <> 0:
                self.io['output'].put((path, data))
            yield PROCESSING