openrvdas.logger.writers.logfile_writer

  1#!/usr/bin/env python3
  2
  3import os
  4import json
  5import logging
  6import re
  7import math
  8from datetime import datetime, timedelta, timezone
  9
 10from typing import Union
 11from logger.utils.das_record import DASRecord  # noqa: E402
 12from logger.utils import timestamp  # noqa: E402
 13from logger.writers.writer import Writer  # noqa: E402
 14from logger.writers.file_writer import FileWriter  # noqa: E402
 15
 16DEFAULT_DATETIME_STR = '-' + timestamp.DATE_FORMAT
 17
 18
 19class LogfileWriter(Writer):
 20    """Write to the specified filebase, with datestamp appended. If filebase
 21    is a <regex>:<filebase> dict, write records to every filebase whose
 22    regex appears in the record.
 23    """
 24    def __init__(self,
 25                 filebase=None,
 26                 delimiter='\n',
 27                 flush=True,
 28                 split_interval='24H',
 29                 header=None,
 30                 header_file=None,
 31                 time_format=timestamp.TIME_FORMAT,
 32                 date_format=DEFAULT_DATETIME_STR,
 33                 time_zone=timezone.utc,
 34                 suffix=None,
 35                 split_char=' ',
 36                 **kwargs):
 37        """Write timestamped records to a filebase. The filebase will
 38        have the current date appended, in keeping with R2R format
 39        recommendations (http://www.rvdata.us/operators/directory). When the
 40        timestamped date on records rolls over to next day, create a new file
 41        with the new date suffix.
 42
 43        If filebase is a dict of <string>:<filebase> pairs, The writer will
 44        attempt to match a <string> in the dict to each record it receives.
 45        It will write the record to the filebase corresponding to the first
 46        string it matches (Note that the order of comparison is not
 47        guaranteed!). If no strings match, the record will be written to the
 48        standalone filebase provided.
 49
 50        Four formats of records can be written by a LogfileWriter:
 51            1. A string prefixed by a timestamp
 52            2. A DASRecord
 53            3. A dict that has a 'timestamp' key
 54            4. A list of any of the above
 55
 56        ```
 57        filebase        A filebase string to write to or a dict mapping
 58                        <string>:<filebase>.
 59
 60        delimiter       A character to trucate each incoming record.
 61
 62        flush           If True (default), flush after every write() call
 63
 64        split_interval  If set the file will trucate at the specified interval.
 65                        The value must be a string containing an integer
 66                        followed by a 'H' (hours) or 'M' (minutes). Default
 67                        value is '24H' (daily).
 68
 69        header          A string to add to the beginning of a new log file
 70                        or a dict mapping <string>:<header> to select the
 71                        header string based on the record contents.
 72
 73        header_file     A string containing the path to file containing a
 74                        header string to add to the beginning of a new log
 75                        file or a dict mapping <string>:<header_file> to select
 76                        the header filepath based on the record contents.
 77
 78        time_format     The format of the record's timestamp. Defaults to
 79                        whatever's defined in utils.timestamp.TIME_FORMAT.
 80
 81        date_fomat      A strftime-compatible string, such as '%Y-%m-%d';
 82                        defaults to '-' plus whatever's defined in
 83                        utils.timestamps.DATE_FORMAT.  If the value starts with
 84                        a '^' character, the string will prepend the file
 85                        name portion of the filebase
 86
 87        time_zone       Timezone to use when constructing the date_format
 88                        portion of the filenames.
 89
 90        suffix          A suffix string to add to the log filename or a dict
 91                        mapping <string>:<suffix> to select the suffix to
 92                        add to a filename based on the record contents.
 93
 94        split_char      Delimiter between timestamp and rest of message
 95
 96        quiet           If True, don't complain if a record doesn't match
 97                        any mapped prefix
 98        ```
 99        """
100        super().__init__(**kwargs)  # processes 'quiet' and type hints
101
102        self.filebase = filebase
103        self.flush = flush
104        self.delimiter = delimiter
105        self.split_interval = self._validate_split_interval(split_interval)
106        self.split_interval_in_seconds = self._get_split_interval_in_seconds()
107        self.time_format = time_format
108        self.date_format = self._validate_date_format(date_format)
109        self.time_zone = time_zone
110        self.split_char = split_char
111        self.suffix = suffix or ''
112
113        self.header = self._load_header(header, header_file)
114
115        # If our filebase is a dict, we're going to be doing our
116        # fancy pattern->filebase mapping.
117        self.do_filebase_mapping = isinstance(self.filebase, dict)
118
119        if self.do_filebase_mapping:
120            # Do our matches faster by precompiling
121            self.compiled_filebase_map = {
122                pattern: re.compile(pattern) for pattern in self.filebase
123            }
124
125        # If our suffix is a dict, we're going to be doing our
126        # fancy pattern->suffix mapping.
127        self.do_suffix_mapping = isinstance(self.suffix, dict)
128
129        if self.do_suffix_mapping:
130            # Do our matches faster by precompiling
131            self.compiled_suffix_map = {
132                pattern: re.compile(pattern) for pattern in self.suffix
133            }
134
135        # If our header is a dict, we're going to be doing our
136        # fancy pattern->header mapping.
137        self.do_header_mapping = isinstance(self.header, dict)
138
139        if self.do_header_mapping:
140            # Do our matches faster by precompiling
141            self.compiled_header_map = {
142                pattern: re.compile(pattern) for pattern in self.header
143            }
144
145        self.current_filename = {}
146        self.writer = {}
147
148    ############################
149    def _validate_split_interval(self, split_interval):
150        """
151        Helper function to validate split_interval
152        """
153        if split_interval is None:
154            return None
155        if not isinstance(split_interval, str):
156            raise ValueError("split_interval must be a string like '1H' or '30M'")
157        if not split_interval.endswith(("H", "M")):
158            raise ValueError("must be an integer followed by 'H' or 'M'")
159        try:
160            return (int(split_interval[:-1]), split_interval[-1])
161        except ValueError:
162            raise ValueError("must be an integer followed by 'H' or 'M'")
163        return None
164
165    ############################
166    def _validate_date_format(self, date_format):
167        if not self.split_interval:
168            return date_format or ""
169
170        unit = self.split_interval[1]
171        value = self.split_interval[0]
172
173        # --- Decide requirements based on interval ---
174        if unit == "H":
175            even_days = value % 24 == 0
176            needs_hour = not even_days
177            needs_minute = False
178        elif unit == "M":
179            even_hours = value % 60 == 0
180            needs_hour = True
181            needs_minute = not even_hours
182        else:
183            return DEFAULT_DATETIME_STR  # fallback
184
185        # --- Default formats if user didn’t supply one ---
186        if not date_format:
187            if unit == "H":
188                return DEFAULT_DATETIME_STR if even_days else DEFAULT_DATETIME_STR + "T%H00"
189            if unit == "M":
190                return f"{DEFAULT_DATETIME_STR}T%H{'%M' if needs_minute else '00'}"
191        # --- Extract directives ---
192        found = set(re.findall(r"%[a-zA-Z]", date_format))
193
194        # Must always have year
195        if "%Y" not in found:
196            raise ValueError("date_format must include %Y (year).")
197
198        # Must have either month+day or julian day
199        if not ({"%m", "%d"} <= found or "%j" in found):
200            raise ValueError("date_format must include %m, %d (month, day) or %j (day-of-year).")
201
202        # Hours?
203        if needs_hour and "%H" not in found:
204            raise ValueError("date_format must include %H (hour).")
205
206        # Minutes?
207        if needs_minute and "%M" not in found:
208            raise ValueError("date_format must include %M (minute).")
209
210        return date_format
211
212    ############################
213    def _load_header(self, header, header_file):
214        """
215        Helper function to verify the header. If a header_file is specified the
216        files are read into a local str or dict depending on the data type of
217        the header_file argument
218        """
219
220        if header and header_file:
221            raise ValueError("Cannot specify both `header` and `header_file`")
222
223        # Case 1: simple string header
224        if header:
225            return header
226
227        # Case 2: header is a dict {key: filepath}
228        if isinstance(header, dict):
229            result = {}
230            for key, header_str in header.items():
231                if not isinstance(header_str, str):
232                    raise ValueError(f"Invalid string for header key {key}: {header_str!r}")
233                result[key] = header_str
234
235            return result
236
237        # Case 3: header_file is a single path
238        if isinstance(header_file, str):
239            try:
240                with open(header_file, "r", encoding="utf-8") as hf:
241                    return hf.read().strip()
242            except OSError as e:
243                raise ValueError(f"Error reading header_file {header_file}: {e}")
244
245        # Case 4: header_file is a dict {key: filepath}
246        if isinstance(header_file, dict):
247            result = {}
248            for key, path in header_file.items():
249                if not isinstance(path, str):
250                    raise ValueError(f"Invalid path for header key {key}: {path!r}")
251                try:
252                    with open(path, "r", encoding="utf-8") as hf:
253                        result[key] = hf.read().strip()
254                except OSError as e:
255                    raise ValueError(f"Error reading header_file {path} for key {key}: {e}")
256            return result
257
258        return None
259
260    ############################
261    def _get_split_interval_in_seconds(self):
262        """
263        Helper function to calculate the value of the split_interval in seconds
264        """
265
266        if not self.split_interval:
267            return 0
268
269        if self.split_interval[1] == 'H':
270            return self.split_interval[0] * 3600
271
272        if self.split_interval[1] == 'M':
273            return self.split_interval[0] * 60
274
275        return 0
276
277    ############################
278    def _get_file_date_format(self, ts):
279        """
280        Helper function to return build the date_format portion of the
281        filename.
282        """
283
284        # if the data is being split by N hours
285        if self.split_interval[1] == 'H':  # hour
286            timestamp_raw = datetime.fromtimestamp(ts, tz=self.time_zone)
287            timestamp_hour = (self.split_interval[0] *
288                              math.floor(timestamp_raw.hour/self.split_interval[0]))
289            timestamp_proc = timestamp_raw.replace(hour=timestamp_hour, minute=0, second=0)
290            self.next_file_split = (timestamp_proc +
291                                    timedelta(seconds=self.split_interval_in_seconds))
292
293            return timestamp.time_str(timestamp=timestamp_proc.timestamp(),
294                                      time_zone=self.time_zone,
295                                      time_format=self.date_format)
296
297        # if the data is being split by N minutes
298        elif self.split_interval[1] == 'M':  # minute
299            timestamp_raw = datetime.fromtimestamp(ts, tz=self.time_zone)
300            timestamp_minute = (self.split_interval[0] *
301                                math.floor(timestamp_raw.minute/self.split_interval[0]))
302            timestamp_proc = timestamp_raw.replace(minute=timestamp_minute, second=0)
303            self.next_file_split = (timestamp_proc +
304                                    timedelta(seconds=self.split_interval_in_seconds))
305
306            return timestamp.time_str(timestamp=timestamp_proc.timestamp(),
307                                      time_zone=self.time_zone,
308                                      time_format=self.date_format)
309
310        return ""
311
312    ############################
313    def fetch_suffix(self, record: str, filename_pattern: str = 'fixed'):
314        """
315        Return the suffix for the given record.  If filename_pattern os defined
316        (because filebase has already matched to a pattern) then that pattern
317        is used over what pattern would normally match.
318        """
319
320        if not self.do_suffix_mapping:
321            return self.suffix
322
323        if filename_pattern != "fixed":
324            return_suffix = self.suffix.get(filename_pattern)
325
326            if return_suffix:
327                return return_suffix
328
329            if not self.quiet:
330                logging.warning('LogfileWriter.fetch_suffix() - no suffix match: "%s"!', record)
331            return
332
333        for pattern, regex in self.compiled_suffix_map.items():
334            if regex and regex.search(record):
335                return self.suffix.get(pattern)
336
337        logging.warning('LogfileWriter.fetch_suffix() - no suffix match: "%s"!', record)
338
339    ############################
340    def fetch_header(self, record: str, filename_pattern: str = 'fixed'):
341        """
342        Return the header for the given record.  If filename_pattern os defined
343        (because filebase has already matched to a pattern) then that pattern
344        is used over what pattern would normally match.
345        """
346
347        if not self.do_header_mapping:
348            return self.header
349
350        if filename_pattern != "fixed":
351            return_header = self.header.get(filename_pattern)
352
353            if return_header:
354                return return_header
355
356            if not self.quiet:
357                logging.warning('LogfileWriter.fetch_header() - no header match: "%s"', record)
358            return ''
359
360        for pattern, regex in self.compiled_header_map.items():
361            if regex and regex.search(record):
362                return self.header.get(pattern, '')
363
364        logging.warning('LogfileWriter.fetch_header() - no header match: "%s"', record)
365        return ''
366
367    ############################
368    def write(self, record: Union[str, DASRecord, dict]):
369        if record == '':
370            return
371
372        # See if it's something we can process, and if not, try digesting
373        if not self.can_process_record(record):  # inherited from BaseModule()
374            self.digest_record(record)  # inherited from BaseModule()
375            return
376
377        # Look for the timestamp
378        if isinstance(record, DASRecord):  # If DASRecord or structured dict,
379            ts = record.timestamp          # convert to JSON before writing
380            record = record.as_json()
381
382        elif isinstance(record, dict):
383            ts = record.get('timestamp')
384            if ts is None:
385                if not self.quiet:
386                    logging.error('LogfileWriter.write() - bad timestamp: "%s"', record)
387                return
388            record = json.dumps(record)
389
390        elif isinstance(record, str):  # If str, it better begin with time string
391            try:  # Try to extract timestamp from record
392                time_str = record.split(self.split_char)[0]
393                ts = timestamp.timestamp(time_str, time_format=self.time_format)
394            except ValueError:
395                if not self.quiet:
396                    logging.error('LogfileWriter.write() - bad timestamp: "%s"', record)
397                    return
398        else:
399            if not self.quiet:
400                logging.error(f'LogfileWriter received badly formatted record. Must be DASRecord, '
401                              f'dict, or timestamp-prefixed string. Received: "{record}"')
402            return
403
404        # Now parse ts into hour and date strings
405        datetime_str = self._get_file_date_format(ts)
406
407        # Figure out where we're going to write
408        if self.do_filebase_mapping:
409            matched_patterns = [self.write_if_match(record, pattern, datetime_str)
410                                for pattern in self.filebase]
411            if True not in matched_patterns:
412                if not self.quiet:
413                    logging.warning(f'No patterns matched in LogfileWriter '
414                                    f'options for record "{record}"')
415        else:
416            pattern = 'fixed'  # just an arbitrary fixed pattern
417
418            suffix = self.fetch_suffix(record, pattern)
419            if suffix is None:
420                logging.error(f'System error: found no suffix matching record: "{record}"!')
421                return None
422
423            if datetime_str.startswith('^'):
424                filename = (
425                    os.path.dirname(self.filebase)
426                    + datetime_str[1:]
427                    + os.path.basename(self.filebase)
428                    + suffix
429                )
430            else:
431                filename = self.filebase + datetime_str + suffix
432
433            self.write_filename(record, pattern, filename)
434
435    ############################
436    def write_if_match(self, record, pattern, datetime_str):
437        """
438        If the record matches the pattern, write to the matching filebase.
439        """
440
441        # Find the compiled regex matching the pattern
442        regex = self.compiled_filebase_map.get(pattern)
443        if not regex:
444            logging.error(f'System error: found no regex matching pattern: "{pattern}"!')
445            return None
446
447        # If the pattern isn't in this record, go home quietly
448        if regex.search(record) is None:
449            return None
450
451        # Otherwise, we write.
452        filebase = self.filebase.get(pattern)
453        if filebase is None:
454            logging.error(f'System error: found no filebase matching pattern "{pattern}"!')
455            return None
456
457        suffix = self.fetch_suffix(record, pattern)
458        if suffix is None:
459            logging.error(f'System error: found no suffix matching pattern: "{pattern}"!')
460            return None
461
462        if datetime_str.startswith('^'):
463            filename = (
464                os.path.dirname(filebase)
465                + datetime_str[1:]
466                + os.path.basename(filebase)
467                + suffix
468            )
469        else:
470            filename = filebase + datetime_str + suffix
471
472        self.write_filename(record, pattern, filename)
473        return True
474
475    ############################
476    def write_filename(self, record, pattern, filename):
477        """
478        Write record to filename. If it's the first time we're writing to
479        this filename, create the appropriate FileWriter and insert it into
480        the map for the relevant pattern.
481        """
482
483        # Are we currently writing to this file? If not, open/create it.
484        if not filename == self.current_filename.get(pattern):
485
486            # calculate header/header_file and suffix
487            header = self.fetch_header(record, pattern) if self.do_header_mapping else self.header
488
489            self.current_filename[pattern] = filename
490            self.writer[pattern] = FileWriter(filename=filename,
491                                              delimiter=self.delimiter,
492                                              header=header,
493                                              flush=self.flush)
494        # Now, if our logic is correct, should *always* have a matching_writer
495        matching_writer = self.writer.get(pattern)
496        matching_writer.write(record)