openrvdas.logger.writers.sealog_writer

  1#!/usr/bin/env python3
  2
  3import pprint
  4import logging
  5import urllib
  6from typing import Union
  7from logger.utils.das_record import DASRecord  # noqa:E402
  8from logger.utils.read_config import read_config  # noqa:E402
  9from logger.utils.sealog_event import SealogEvent, to_event  # noqa:E402
 10from logger.writers.writer import Writer  # noqa: E402
 11
 12
 13################################################################################
 14class SealogWriter(Writer):
 15    """Submit Sealog event dicts to a Sealog Server API.
 16
 17    url    - URL to Sealog Server, i.e. http://<ip_addr>:8000/sealog-server
 18
 19    token  - Java Web Token (JWT) authorized to submit new events to the
 20             Sealog Server API
 21
 22    config_file  - Path to a YAML configuration file specifying:
 23                   `data_id` to determine which set of rules to apply
 24                   `event_value` (optional): default event value for the record.
 25                   `event_author` (optional): author string to include in the event.
 26                   `event_free_text` (optional): free text string for the event.
 27                   `field_map` (optional): mapping of input record field names to event
 28                                           option names.
 29                   `_default` (optional): fallback configuration applied if the record's
 30                                          `data_id` is not found.
 31
 32    Sample configuration file:
 33    ---
 34    qinsy:
 35        event_value: LOGGING_STATUS
 36        event_author: "qinsy"
 37        event_free_text: ""
 38        event_options:
 39            system: EM124
 40        field_map:
 41            status: status
 42            filename: filename
 43
 44    _default:
 45        event_value: UNKNOWN
 46        event_free_text: ""
 47        field_map: null
 48
 49
 50    quiet   - suppress warning messages
 51
 52    """
 53
 54    ############################
 55    def __init__(self, url: str, token: str, config_file: str, **kwargs):
 56        super().__init__(**kwargs)  # processes 'quiet' and type hints
 57
 58        self.url = url
 59        self.token = token
 60
 61        try:
 62            self.configs = read_config(config_file)
 63            logging.info('Loaded sealog config file: %s', pprint.pformat(self.configs))
 64        except Exception as err:
 65            logging.error("Could not find or could not process config file."
 66                          " All records will be ignored.")
 67            logging.info(str(err))
 68            pass
 69
 70        self.test_api_connectivity()
 71
 72    ############################
 73    def test_api_connectivity(self) -> bool:
 74        """
 75        Test connectivity to a restricted API route using JWT authentication
 76
 77        Returns:
 78            bool: True if request succeeds (HTTP 200), False otherwise.
 79        """
 80        req = urllib.request.Request(self.url + '/restricted')
 81        req.add_header("Authorization", f"Bearer {self.token}")
 82
 83        try:
 84            with urllib.request.urlopen(req, timeout=5) as response:
 85                if response.status == 200:
 86                    logging.info("Connection to Sealog Server successful")
 87                    return True
 88                else:
 89                    logging.error(f"Connection to Sealog Server failed: {response.status}")
 90                    return False
 91        except urllib.error.HTTPError as e:
 92            logging.error(f"HTTP error: {e.code} {e.reason}")
 93            return False
 94        except urllib.error.URLError as e:
 95            logging.error(f"Connection error: {e.reason}")
 96            return False
 97
 98    ############################
 99    def write(self, record: Union[DASRecord, SealogEvent, list]):
100        """Submit dicts to Sealog Server
101        Note: Assume record is a dict or list of dict.
102        """
103
104        if isinstance(record, list):
105            for single_record in record:
106                self.write(single_record)
107            return
108
109        event = record if isinstance(record, SealogEvent) else to_event(record, self.configs)
110
111        json_data = event.as_json().encode("utf-8")
112
113        req = urllib.request.Request(self.url + '/api/v1/events', data=json_data, method="POST")
114        req.add_header("Authorization", f"Bearer {self.token}")
115        req.add_header("Content-Type", "application/json")
116
117        try:
118            with urllib.request.urlopen(req, timeout=5) as response:
119                if response.status == 201:
120                    logging.debug("POST successful")
121                else:
122                    logging.error(f"POST failed: {response.status}")
123        except urllib.error.HTTPError as e:
124            logging.error(f"HTTP error: {e.code} {e.reason}")
125        except urllib.error.URLError as e:
126            logging.error(f"Connection error: {e.reason}")