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

    10

import log_line, utils, config

    11

    12

class LogSearchError (Exception) :

    13

    """

    14

        General search error

    15

    """

    16

    17

    pass

    18

    19

class SearchIndexError (LogSearchError) :

    20

    """

    21

        Error manipulating the index

    22

    """

    23

    24

    def __init__ (self, msg, db) :

    25

        """

    26

            Build the error from the given message + HyperEstraier.Database

    27

        """

    28

    29

        super(SearchIndexError, self).__init__("%s: %s" % (msg, db.err_msg(db.error())))

    30

    31

class NoResultsFound (LogSearchError) :

    32

    """

    33

        No results found

    34

    """

    35

    36

    pass

    37

    38

class LogSearchIndex (object) :

    39

    """

    40

        An index on the logs for a group of channels.

    41

    42

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

    43

    44

        These log documents have the following attributes:

    45

            @uri                - channel/date/line

    46

            channel             - channel code

    47

            type                - the LogType id

    48

            timestamp           - UTC timestamp

    49

            source_nickname     - source nickname

    50

            source_username     - source username

    51

            source_hostname     - source hostname

    52

            source_chanflags    - source channel flags

    53

            target_nickname     - target nickname

    54

    55

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

    56

    """

    57

    58

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

    59

        """

    60

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

    61

                First char:

    62

                    r       - read, error if not exists

    63

                    w       - write, create if not exists

    64

                    a       - write, error if not exists

    65

                    c       - create, error if exists

    66

    67

                Additional chars:

    68

                    trunc   - truncate if exists

    69

                    +       - read as well as write

    70

                    ?       - non-blocking lock open, i.e. it fails if already open

    71

    72

            Channels is the ChannelList.

    73

        """

    74

    75

        # store

    76

        self.channels = channels

    77

        self.path = path

    78

        self.mode = mode

    79

    80

        # check it does not already exist?

    81

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

    82

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

    83

    84

        # mapping of { mode -> flags }

    85

        mode_to_flag = {

    86

            'r':    hype.Database.DBREADER,

    87

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

    88

            'a':    hype.Database.DBWRITER,

    89

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

    90

        }

    91

    92

        # flags to use, standard modes

    93

        flags = mode_to_flag[mode[0]]

    94

    95

        # mode-flags

    96

        if '?' in mode :

    97

            # non-blocking locking

    98

            flags |= hype.Database.DBLCKNB

    99

   100

        elif '+' in mode :

   101

            # read

   102

            flags |= hype.Database.DBREADER

   103

   104

        elif 'trunc' in mode :

   105

            # truncate. Dangerous!

   106

            flags |= hype.Database.DBTRUNC

   107

   108

        # make instance

   109

        self.db = hype.Database()

   110

   111

        # open

   112

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

   113

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

   114

   115

    def close (self) :

   116

        """

   117

            Explicitly close the index, this is done automatically on del

   118

        """

   119

   120

        if not self.db.close() :

   121

            raise SearchIndexError("Index close failed", self.db)

   122

   123

    def insert (self, channel, lines) :

   124

        """

   125

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

   126

        """

   127

   128

        # count from zero

   129

        count = 0

   130

   131

        # iterate

   132

        for line in lines :

   133

            # insert

   134

            self.insert_line(channel, line)

   135

   136

            # count

   137

            count += 1

   138

   139

        # return

   140

        return count

   141

   142

    def insert_line (self, channel, line) :

   143

        """

   144

            Adds a single LogLine for the given LogChannel to the index

   145

        """

   146

   147

        # validate the LogChannel

   148

        assert channel.id

   149

   150

        # validate the LogLine

   151

        assert line.offset

   152

        assert line.timestamp

   153

   154

        # create new document

   155

        doc = hype.Document()

   156

   157

        # line date

   158

        date = line.timestamp.date()

   159

   160

        # ensure that it's not 1900

   161

        assert date.year != 1900

   162

   163

        # add URI

   164

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

   165

   166

        # add channel id

   167

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

   168

   169

        # add type

   170

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

   171

   172

        # add UTC timestamp

   173

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

   174

   175

        # add source attribute?

   176

        if line.source :

   177

            source_nickname, source_username, source_hostname, source_chanflags = line.source

   178

   179

            if source_nickname :

   180

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

   181

   182

            if source_username :

   183

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

   184

   185

            if source_hostname :

   186

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

   187

   188

            if source_chanflags :

   189

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

   190

   191

        # add target attributes?

   192

        if line.target :

   193

            target_nickname = line.target

   194

   195

            if target_nickname :

   196

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

   197

   198

        # add data

   199

        if line.data :

   200

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

   201

   202

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

   203

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

   204

            raise SearchIndexError("put_doc", self.db)

   205

   206

    def search_cond (self, cond) :

   207

        """

   208

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

   209

        """

   210

   211

        # execute search, unused 'flags' arg stays zero

   212

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

   213

   214

        # no results?

   215

        if not results :

   216

            raise NoResultsFound()

   217

   218

        # iterate over the document IDs

   219

        for doc_id in results :

   220

            # load document, this throws an exception...

   221

            # option constants are hype.Database.GDNOATTR/GDNOTEXT

   222

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

   223

   224

            # load the attributes/text

   225

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

   226

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

   227

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

   228

   229

            # source

   230

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

   231

   232

            # target

   233

            target = doc.attr('target_nickname')

   234

   235

            # message text

   236

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

   237

   238

            # build+yield to as LogLine

   239

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

   240

   241

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

   242

        """

   243

            Search with flexible parameters

   244

   245

                options     - bitmask of hype.Condition.*

   246

                channel     - LogChannel object

   247

                attrs       - raw attribute expressions

   248

                phrase      - the search query phrase

   249

                order       - order attribute expression

   250

                max         - number of results to return

   251

                skip        - number of results to skip

   252

        """

   253

   254

        # build condition

   255

        cond = hype.Condition()

   256

   257

        if options :

   258

            # set options

   259

            cond.set_options(options)

   260

   261

        if channel :

   262

            # add channel attribute

   263

            cond.add_attr(("channel STREQ %s" % channel.id).encode('utf8'))

   264

   265

        if attrs :

   266

            # add attributes

   267

            for attr in attrs :

   268

                cond.add_attr(attr.encode('utf8'))

   269

   270

        if phrase :

   271

            # add phrase

   272

            cond.set_phrase(phrase.encode('utf8'))

   273

   274

        if order :

   275

            # set order

   276

            cond.set_order(order)

   277

   278

        if max :

   279

            # set max

   280

            cond.set_max(max)

   281

   282

        if skip :

   283

            # set skip

   284

            cond.set_skip(skip)

   285

   286

        # execute

   287

        return self.search_cond(cond)

   288

   289

    def search_simple (self, channel, query, count=None, offset=None, search_msg=True, search_nick=False) :

   290

        """

   291

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

   292

   293

            The search_* params define which attributes to search for (using fulltext search for the message, STROR for

   294

            attributes).

   295

        """

   296

   297

        # search attributes

   298

        attrs = []

   299

   300

        # nickname target query

   301

        if search_nick :

   302

            attrs.append("source_nickname STRINC %s" % query)

   303

#            attrs.append("target_nickname STRINC %s" % query)

   304

   305

        # use search(), backwards

   306

        results = list(self.search(

   307

            # simplified phrase

   308

            options     = hype.Condition.SIMPLE,

   309

   310

            # specific channel

   311

            channel     = channel,

   312

   313

            # given phrase

   314

            phrase      = query if search_msg else None,

   315

   316

            # attributes defined above

   317

            attrs       = attrs,

   318

   319

            # order by timestamp, descending (backwards)

   320

            order       = "timestamp NUMD",

   321

   322

            # count/offset

   323

            max         = count,

   324

            skip        = offset,

   325

        ))

   326

   327

        # reverse

   328

        return reversed(results)

   329

   330

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

   331

        """

   332

            List all indexed log items for the given UTC date

   333

        """

   334

   335

        # start/end dates

   336

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

   337

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

   338

   339

        # search

   340

        return self.search(

   341

            # specific channel

   342

            channel     = channel,

   343

   344

            # specific date range

   345

            attrs       = [

   346

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

   347

            ],

   348

   349

            # order correctly

   350

            order       = "timestamp NUMA",

   351

   352

            # max count/offset

   353

            max         = count,

   354

            skip        = skip

   355

        )

   356

   357

def get_index () :

   358

    """

   359

        Returns the default read-only index, suitable for searching

   360

    """

   361

   362

    # XXX: no caching, just open it every time

   363

    _index = LogSearchIndex(config.LOG_CHANNELS, config.SEARCH_INDEX_PATH, 'r')

   364

   365

    # return

   366

    return _index

   367