1

"""

     2

    Full-text searching of logs

     3

"""

     4

     5

import datetime, calendar, pytz

     6

import os.path

     7

     8

import HyperEstraier as hype

     9

    11

    12

class LogSearchError (Exception) :

    13

    """

    14

        General search error

    15

    """

    16

    17

    pass

    18

    19

class NoResultsFound (LogSearchError) :

    20

    """

    21

        No results found

    22

    """

    23

    24

    pass

    25

    26

class LogSearchIndex (object) :

    27

    """

    28

        An index on the logs for a group of channels.

    29

    30

        This uses Hyper Estraier to handle searching, whereby each log line is a document (yes, I have a powerful server).

    31

    32

        These log documents have the following attributes:

    33

            @uri                - channel/date/line

    34

            channel             - channel code

    35

            type                - the LogType id

    36

            timestamp           - UTC timestamp

    37

            source_nickname     - source nickname

    38

            source_username     - source username

    39

            source_hostname     - source hostname

    40

            source_chanflags    - source channel flags

    41

            target_nickname     - target nickname

    42

    43

        Each document then has a single line of data, which is the log data message

    44

    """

    45

    46

    def __init__ (self, channels, path, mode='r') :

    47

        """

    48

            Open the database at the given path, with the given mode:

    49

                r       - read, error if not exists

    50

                w       - write, create if not exists

    51

                a       - write, error if not exists

    52

                c       - write, create, error if exists

    53

                *       - write, create, truncate if exists

    54

    55

            Channels is the ChannelList.

    56

        """

    57

    58

        # store

    59

        self.channels = channels

    60

        self.path = path

    61

        self.mode = mode

    62

    63

        # check it does not already exist?

    64

        if mode in 'c' and os.path.exists(path) :

    65

            raise LogSearchError("Index already exists: %s" % (path, ))

    66

    67

        # mapping of { mode -> flags }

    68

        mode_to_flag = {

    69

            'r':    hype.Database.DBREADER,

    70

            'w':    hype.Database.DBWRITER | hype.Database.DBCREAT,

    71

            'a':    hype.Database.DBWRITER,

    72

            'c':    hype.Database.DBWRITER | hype.Database.DBCREAT,

    73

            '*':    hype.Database.DBWRITER | hype.Database.DBCREAT | hype.Database.DBTRUNC,

    74

        }

    75

    76

        # look up flags

    77

        flags = mode_to_flag[mode]

    78

    79

        # make instance

    80

        self.db = hype.Database()

    81

    82

        # open

    83

        if not self.db.open(path, flags) :

    84

            raise Exception("Index open failed: %s, mode=%s, flags=%#06x: %s" % (path, mode, flags, self.db.err_msg(self.db.error())))

    85

    86

    def insert (self, channel, lines) :

    87

        """

    88

            Adds a sequence of LogLines from the given LogChannel to the index, and return the number of added items

    89

        """

    90

    91

        # count from zero

    92

        count = 0

    93

    94

        # iterate

    95

        for line in lines :

    96

            # insert

    97

            self.insert_line(channel, line)

    98

    99

            # count

   100

            count += 1

   101

   102

        # return

   103

        return count

   104

   105

   106

   107

    def insert_line (self, channel, line) :

   108

        """

   109

            Adds a single LogLine for the given LogChannel to the index

   110

        """

   111

   112

        # validate the LogChannel

   113

        assert channel.id

   114

   115

        # validate the LogLine

   116

        assert line.offset

   117

        assert line.timestamp

   118

   119

        # create new document

   120

        doc = hype.Document()

   121

   122

        # line date

   123

        date = line.timestamp.date()

   124

   125

        # ensure that it's not 1900

   126

        assert date.year != 1900

   127

   128

        # add URI

   129

        doc.add_attr('@uri',        "%s/%s/%d" % (channel.id, date.strftime('%Y-%m-%d'), line.offset))

   130

   131

        # add channel id

   132

        doc.add_attr('channel',     channel.id)

   133

   134

        # add type

   135

        doc.add_attr('type',        str(line.type))

   136

   137

        # add UTC timestamp

   138

        doc.add_attr('timestamp',   str(utils.to_utc_timestamp(line.timestamp)))

   139

   140

        # add source attribute?

   141

        if line.source :

   142

            source_nickname, source_username, source_hostname, source_chanflags = line.source

   143

   144

            if source_nickname :

   145

                doc.add_attr('source_nickname', source_nickname.encode('utf8'))

   146

   147

            if source_username :

   148

                doc.add_attr('source_username', source_username.encode('utf8'))

   149

   150

            if source_hostname :

   151

                doc.add_attr('source_hostname', source_hostname.encode('utf8'))

   152

   153

            if source_chanflags :

   154

                doc.add_attr('source_chanflags', source_chanflags.encode('utf8'))

   155

   156

        # add target attributes?

   157

        if line.target :

   158

            target_nickname = line.target

   159

   160

            if target_nickname :

   161

                doc.add_attr('target_nickname', target_nickname.encode('utf8'))

   162

   163

        # add data

   164

        if line.data :

   165

            doc.add_text(line.data.encode('utf8'))

   166

   167

        # put, "clean up dispensable regions of the overwritten document"

   168

        if not self.db.put_doc(doc, hype.Database.PDCLEAN) :

   169

            raise Exeception("Index put_doc failed")

   170

   171

    def search_cond (self, cond) :

   172

        """

   173

            Search using a raw hype.Condition. Raises NoResultsFound if there aren't any results

   174

        """

   175

   176

        # execute search, unused 'flags' arg stays zero

   177

        results = self.db.search(cond, 0)

   178

   179

        # no results?

   180

        if not results :

   181

            raise NoResultsFound()

   182

   183

        # iterate over the document IDs

   184

        for doc_id in results :

   185

            # load document, this throws an exception...

   186

            # option constants are hype.Database.GDNOATTR/GDNOTEXT

   187

            doc = self.db.get_doc(doc_id, 0)

   188

   189

            # load the attributes/text

   190

            channel         = self.channels.lookup(doc.attr('channel'))

   191

            type            = int(doc.attr('type'))

   192

            timestamp       = utils.from_utc_timestamp(int(doc.attr('timestamp')))

   193

   194

            # source

   195

            source = (doc.attr('source_nickname'), doc.attr('source_username'), doc.attr('source_hostname'), doc.attr('source_chanflags'))

   196

   197

            # target

   198

            target = doc.attr('target_nickname')

   199

   200

            # message text

   201

            message         = doc.cat_texts().decode('utf8')

   202

   203

            # build+yield to as LogLine

   204

            yield log_line.LogLine(channel, None, type, timestamp, source, target, message)

   205

   206

    def search (self, options=None, channel=None, attrs=None, phrase=None, order=None, max=None, skip=None) :

   207

        """

   208

            Search with flexible parameters

   209

   210

                options     - bitmask of hype.Condition.*

   211

                channel     - LogChannel object

   212

                attrs       - raw attribute expressions

   213

                phrase      - the search query phrase

   214

                order       - order attribute expression

   215

                max         - number of results to return

   216

                skip        - number of results to skip

   217

        """

   218

   219

        # build condition

   220

        cond = hype.Condition()

   221

   222

        if options :

   223

            # set options

   224

            cond.set_options(options)

   225

   226

        if channel :

   227

            # add channel attribute

   228

            cond.add_attr("channel STREQ %s" % (channel.id, ))

   229

   230

        if attrs :

   231

            # add attributes

   232

            for attr in attrs :

   233

                cond.add_attr(attr)

   234

   235

        if phrase :

   236

            # add phrase

   237

            cond.set_phrase(phrase)

   238

   239

        if order :

   240

            # set order

   241

            cond.set_order(order)

   242

   243

        if max :

   244

            # set max

   245

            cond.set_max(max)

   246

   247

        if skip :

   248

            # set skip

   249

            cond.set_skip(skip)

   250

   251

        # execute

   252

        return self.search_cond(cond)

   253

   254

    def search_simple (self, channel, query, count=None, offset=None) :

   255

        """

   256

            Search for lines from the given channel for the given simple query

   257

        """

   258

   259

        # use search(), backwards

   260

        results = list(self.search(

   261

            # simplified phrase

   262

            options     = hype.Condition.SIMPLE,

   263

   264

            # specific channel

   265

            channel     = channel,

   266

   267

            # given phrase

   268

            phrase      = query,

   269

   270

            # order by timestamp, descending (backwards)

   271

            order       = "timestamp NUMD",

   272

   273

            # count/offset

   274

            max         = count,

   275

            skip        = offset,

   276

        ))

   277

   278

        # reverse

   279

        return reversed(results)

   280

   281

    def list (self, channel, date, count=None, skip=None) :

   282

        """

   283

            List all indexed log items for the given UTC date

   284

        """

   285

   286

        # start/end dates

   287

        dt_start = datetime.datetime(date.year, date.month, date.day, 0, 0, 0, 0)

   288

        dt_end   = datetime.datetime(date.year, date.month, date.day, 23, 23, 59, 999999)

   289

   290

        # search

   291

        return self.search(

   292

            # specific channel

   293

            channel     = channel,

   294

   295

            # specific date range

   296

            attrs       = [

   297

                "timestamp NUMBT %d %d" % (utils.to_utc_timestamp(dt_start), utils.to_utc_timestamp(dt_end))

   298

            ],

   299

   300

            # order correctly

   301

            order       = "timestamp NUMA",

   302

   303

            # max count/offset

   304

            max         = count,

   305

            skip        = skip

   306

        )