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 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

                First char:

    50

                    r       - read, error if not exists

    51

                    w       - write, create if not exists

    52

                    a       - write, error if not exists

    53

                    c       - create, error if exists

    54

    55

                Additional chars:

    56

                    trunc   - truncate if exists

    57

                    +       - read as well as write

    58

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

    59

    60

            Channels is the ChannelList.

    61

        """

    62

    63

        # store

    64

        self.channels = channels

    65

        self.path = path

    66

        self.mode = mode

    67

    68

        # check it does not already exist?

    69

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

    70

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

    71

    72

        # mapping of { mode -> flags }

    73

        mode_to_flag = {

    74

            'r':    hype.Database.DBREADER,

    75

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

    76

            'a':    hype.Database.DBWRITER,

    77

            'c':    hype.Database.DBCREAT,

    78

        }

    79

    80

        # flags to use, standard modes

    81

        flags = mode_to_flag[mode[0]]

    82

    83

        # mode-flags

    84

        if '?' in mode :

    85

            # non-blocking locking

    86

            flags |= hype.Database.DBLCKNB

    87

    88

        elif '+' in mode :

    89

            # read

    90

            flags |= hype.Database.DBREADER

    91

    92

        elif 'trunc' in mode :

    93

            # truncate. Dangerous!

    94

            flags |= hype.Database.DBTRUNC

    95

    96

        # make instance

    97

        self.db = hype.Database()

    98

    99

        # open

   100

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

   101

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

   102

   103

    def insert (self, channel, lines) :

   104

        """

   105

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

   106

        """

   107

   108

        # count from zero

   109

        count = 0

   110

   111

        # iterate

   112

        for line in lines :

   113

            # insert

   114

            self.insert_line(channel, line)

   115

   116

            # count

   117

            count += 1

   118

   119

        # return

   120

        return count

   121

   122

   123

   124

    def insert_line (self, channel, line) :

   125

        """

   126

            Adds a single LogLine for the given LogChannel to the index

   127

        """

   128

   129

        # validate the LogChannel

   130

        assert channel.id

   131

   132

        # validate the LogLine

   133

        assert line.offset

   134

        assert line.timestamp

   135

   136

        # create new document

   137

        doc = hype.Document()

   138

   139

        # line date

   140

        date = line.timestamp.date()

   141

   142

        # ensure that it's not 1900

   143

        assert date.year != 1900

   144

   145

        # add URI

   146

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

   147

   148

        # add channel id

   149

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

   150

   151

        # add type

   152

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

   153

   154

        # add UTC timestamp

   155

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

   156

   157

        # add source attribute?

   158

        if line.source :

   159

            source_nickname, source_username, source_hostname, source_chanflags = line.source

   160

   161

            if source_nickname :

   162

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

   163

   164

            if source_username :

   165

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

   166

   167

            if source_hostname :

   168

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

   169

   170

            if source_chanflags :

   171

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

   172

   173

        # add target attributes?

   174

        if line.target :

   175

            target_nickname = line.target

   176

   177

            if target_nickname :

   178

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

   179

   180

        # add data

   181

        if line.data :

   182

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

   183

   184

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

   185

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

   186

            raise Exeception("Index put_doc failed")

   187

   188

    def search_cond (self, cond) :

   189

        """

   190

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

   191

        """

   192

   193

        # execute search, unused 'flags' arg stays zero

   194

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

   195

   196

        # no results?

   197

        if not results :

   198

            raise NoResultsFound()

   199

   200

        # iterate over the document IDs

   201

        for doc_id in results :

   202

            # load document, this throws an exception...

   203

            # option constants are hype.Database.GDNOATTR/GDNOTEXT

   204

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

   205

   206

            # load the attributes/text

   207

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

   208

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

   209

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

   210

   211

            # source

   212

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

   213

   214

            # target

   215

            target = doc.attr('target_nickname')

   216

   217

            # message text

   218

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

   219

   220

            # build+yield to as LogLine

   221

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

   222

   223

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

   224

        """

   225

            Search with flexible parameters

   226

   227

                options     - bitmask of hype.Condition.*

   228

                channel     - LogChannel object

   229

                attrs       - raw attribute expressions

   230

                phrase      - the search query phrase

   231

                order       - order attribute expression

   232

                max         - number of results to return

   233

                skip        - number of results to skip

   234

        """

   235

   236

        # build condition

   237

        cond = hype.Condition()

   238

   239

        if options :

   240

            # set options

   241

            cond.set_options(options)

   242

   243

        if channel :

   244

            # add channel attribute

   245

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

   246

   247

        if attrs :

   248

            # add attributes

   249

            for attr in attrs :

   250

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

   251

   252

        if phrase :

   253

            # add phrase

   254

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

   255

   256

        if order :

   257

            # set order

   258

            cond.set_order(order)

   259

   260

        if max :

   261

            # set max

   262

            cond.set_max(max)

   263

   264

        if skip :

   265

            # set skip

   266

            cond.set_skip(skip)

   267

   268

        # execute

   269

        return self.search_cond(cond)

   270

   271

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

   272

        """

   273

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

   274

   275

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

   276

            attributes).

   277

        """

   278

   279

        # search attributes

   280

        attrs = []

   281

   282

        # nickname target query

   283

        if search_nick :

   284

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

   285

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

   286

   287

        # use search(), backwards

   288

        results = list(self.search(

   289

            # simplified phrase

   290

            options     = hype.Condition.SIMPLE,

   291

   292

            # specific channel

   293

            channel     = channel,

   294

   295

            # given phrase

   296

            phrase      = query if search_msg else None,

   297

   298

            # attributes defined above

   299

            attrs       = attrs,

   300

   301

            # order by timestamp, descending (backwards)

   302

            order       = "timestamp NUMD",

   303

   304

            # count/offset

   305

            max         = count,

   306

            skip        = offset,

   307

        ))

   308

   309

        # reverse

   310

        return reversed(results)

   311

   312

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

   313

        """

   314

            List all indexed log items for the given UTC date

   315

        """

   316

   317

        # start/end dates

   318

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

   319

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

   320

   321

        # search

   322

        return self.search(

   323

            # specific channel

   324

            channel     = channel,

   325

   326

            # specific date range

   327

            attrs       = [

   328

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

   329

            ],

   330

   331

            # order correctly

   332

            order       = "timestamp NUMA",

   333

   334

            # max count/offset

   335

            max         = count,

   336

            skip        = skip

   337

        )

   338

   339

# global read-only index

   340

_index = None

   341

   342

def get_index () :

   343

    """

   344

        Returns the default read-only index, suitable for searching

   345

    """

   346

   347

    global _index

   348

   349

    # open?

   350

    if not _index :

   351

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

   352

   353

    # return

   354

    return _index

   355