#!/usr/bin/env python
#
# Copyright 2007 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""Stub implementation for Log Service that uses sqlite."""

import atexit
import codecs
import logging
import time

import sqlite3

from google.appengine.api import apiproxy_stub
from google.appengine.api import appinfo
from google.appengine.api.logservice import log_service_pb
from google.appengine.runtime import apiproxy_errors


_REQUEST_LOG_CREATE = """
CREATE TABLE IF NOT EXISTS RequestLogs (
  id INTEGER NOT NULL PRIMARY KEY,
  user_request_id TEXT NOT NULL,
  app_id TEXT NOT NULL,
  version_id TEXT NOT NULL,
  module TEXT NOT NULL,
  ip TEXT NOT NULL,
  nickname TEXT NOT NULL,
  start_time INTEGER NOT NULL,
  end_time INTEGER DEFAULT 0 NOT NULL,
  method TEXT NOT NULL,
  resource TEXT NOT NULL,
  http_version TEXT NOT NULL,
  status INTEGER DEFAULT 0 NOT NULL,
  response_size INTEGER DEFAULT 0 NOT NULL,
  user_agent TEXT NOT NULL,
  url_map_entry TEXT DEFAULT '' NOT NULL,
  host TEXT NOT NULL,
  task_queue_name TEXT DEFAULT '' NOT NULL,
  task_name TEXT DEFAULT '' NOT NULL,
  latency INTEGER DEFAULT 0 NOT NULL,
  mcycles INTEGER DEFAULT 0 NOT NULL,
  finished INTEGER DEFAULT 0 NOT NULL
);
"""

_REQUEST_LOG_ADD_MODULE_COLUMN = """
ALTER TABLE RequestLogs
  ADD COLUMN module TEXT DEFAULT '%s' NOT NULL;
""" % appinfo.DEFAULT_MODULE

_APP_LOG_CREATE = """
CREATE TABLE IF NOT EXISTS AppLogs (
  id INTEGER NOT NULL PRIMARY KEY,
  request_id INTEGER NOT NULL,
  timestamp INTEGER NOT NULL,
  level INTEGER NOT NULL,
  message TEXT NOT NULL,
  FOREIGN KEY(request_id) REFERENCES RequestLogs(id)
);
"""


class LogServiceStub(apiproxy_stub.APIProxyStub):
  """Python stub for Log Service service."""

  THREADSAFE = True

  _ACCEPTS_REQUEST_ID = True


  _DEFAULT_READ_COUNT = 20


  _MIN_COMMIT_INTERVAL = 5

  def __init__(self, persist=False, logs_path=None, request_data=None):
    """Initializer.

    Args:
      persist: For backwards compatability. Has no effect.
      logs_path: A str containing the filename to use for logs storage. Defaults
        to in-memory if unset.
      request_data: A apiproxy_stub.RequestData instance used to look up state
        associated with the request that generated an API call.
    """

    super(LogServiceStub, self).__init__('logservice',
                                         request_data=request_data)
    self._request_id_to_request_row_id = {}
    if logs_path is None:
      logs_path = ':memory:'
    self._conn = sqlite3.connect(logs_path, check_same_thread=False)
    self._conn.row_factory = sqlite3.Row
    self._conn.execute(_REQUEST_LOG_CREATE)
    self._conn.execute(_APP_LOG_CREATE)

    column_names = set(c['name'] for c in
                       self._conn.execute('PRAGMA table_info(RequestLogs)'))
    if 'module' not in column_names:
      self._conn.execute(_REQUEST_LOG_ADD_MODULE_COLUMN)

    self._last_commit = time.time()
    atexit.register(self._conn.commit)

  @staticmethod
  def _get_time_usec():
    return int(time.time() * 1e6)

  def _maybe_commit(self):
    now = time.time()
    if (now - self._last_commit) > self._MIN_COMMIT_INTERVAL:
      self._conn.commit()
      self._last_commit = now

  @apiproxy_stub.Synchronized
  def start_request(self, request_id, user_request_id, ip, app_id, version_id,
                    nickname, user_agent, host, method, resource, http_version,
                    start_time=None, module=None):
    """Starts logging for a request.

    Each start_request call must be followed by a corresponding end_request call
    to cleanup resources allocated in start_request.

    Args:
      request_id: A unique string identifying the request associated with the
        API call.
      user_request_id: A user-visible unique string for retrieving the request
        log at a later time.
      ip: The user's IP address.
      app_id: A string representing the application ID that this request
        corresponds to.
      version_id: A string representing the version ID that this request
        corresponds to.
      nickname: A string representing the user that has made this request (that
        is, the user's nickname, e.g., 'foobar' for a user logged in as
        'foobar@gmail.com').
      user_agent: A string representing the agent used to make this request.
      host: A string representing the host that received this request.
      method: A string containing the HTTP method of this request.
      resource: A string containing the path and query string of this request.
      http_version: A string containing the HTTP version of this request.
      start_time: An int containing the start time in micro-seconds. If unset,
        the current time is used.
      module: The string name of the module handling this request.
    """
    if module is None:
      module = appinfo.DEFAULT_MODULE
    if version_id is None:
      version_id = 'NO-VERSION'
    major_version_id = version_id.split('.', 1)[0]
    if start_time is None:
      start_time = self._get_time_usec()
    cursor = self._conn.execute(
        'INSERT INTO RequestLogs (user_request_id, ip, app_id, version_id, '
        'nickname, user_agent, host, start_time, method, resource, '
        'http_version, module)'
        ' VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)', (
            user_request_id, ip, app_id, major_version_id, nickname, user_agent,
            host, start_time, method, resource, http_version, module))
    self._request_id_to_request_row_id[request_id] = cursor.lastrowid
    self._maybe_commit()

  @apiproxy_stub.Synchronized
  def end_request(self, request_id, status, response_size, end_time=None):
    """Ends logging for a request.

    Args:
      request_id: A unique string identifying the request associated with the
        API call.
      status: An int containing the HTTP status code for this request.
      response_size: An int containing the content length of the response.
      end_time: An int containing the end time in micro-seconds. If unset, the
        current time is used.
    """
    row_id = self._request_id_to_request_row_id.pop(request_id, None)
    if not row_id:
      return
    if end_time is None:
      end_time = self._get_time_usec()
    self._conn.execute(
        'UPDATE RequestLogs SET '
        'status = ?, response_size = ?, end_time = ?, finished = 1 '
        'WHERE id = ?', (
            status, response_size, end_time, row_id))
    self._maybe_commit()

  def _Dynamic_Flush(self, request, unused_response, request_id):
    """Writes application-level log messages for a request."""
    group = log_service_pb.UserAppLogGroup(request.logs())
    self._insert_app_logs(request_id, group.log_line_list())

  @apiproxy_stub.Synchronized
  def _insert_app_logs(self, request_id, log_lines):
    row_id = self._request_id_to_request_row_id.get(request_id)
    if row_id is None:
      return
    new_app_logs = (self._tuple_from_log_line(row_id, log_line)
                    for log_line in log_lines)
    self._conn.executemany(
        'INSERT INTO AppLogs (request_id, timestamp, level, message) VALUES '
        '(?, ?, ?, ?)', new_app_logs)
    self._maybe_commit()

  @staticmethod
  def _tuple_from_log_line(row_id, log_line):
    message = log_line.message()
    if isinstance(message, str):
      message = codecs.decode(message, 'utf-8', 'replace')
    return (row_id, log_line.timestamp_usec(), log_line.level(), message)

  @apiproxy_stub.Synchronized
  def _Dynamic_Read(self, request, response, request_id):
    if (request.module_version_size() < 1 and
        request.version_id_size() < 1 and
        request.request_id_size() < 1):
      raise apiproxy_errors.ApplicationError(
          log_service_pb.LogServiceError.INVALID_REQUEST)

    if request.module_version_size() > 0 and request.version_id_size() > 0:
      raise apiproxy_errors.ApplicationError(
          log_service_pb.LogServiceError.INVALID_REQUEST)

    if (request.request_id_size() and
        (request.has_start_time() or request.has_end_time() or
         request.has_offset())):
      raise apiproxy_errors.ApplicationError(
          log_service_pb.LogServiceError.INVALID_REQUEST)

    if request.request_id_size():
      for request_id in request.request_id_list():
        log_row = self._conn.execute(
            'SELECT * FROM RequestLogs WHERE user_request_id = ?',
            (request_id,)).fetchone()
        if log_row:
          log = response.add_log()
          self._fill_request_log(log_row, log, request.include_app_logs())
      return

    if request.has_count():
      count = request.count()
    else:
      count = self._DEFAULT_READ_COUNT
    filters, values = self._extract_read_filters(request)
    filter_string = ' WHERE %s' % ' and '.join(filters)

    if request.has_minimum_log_level():
      query = ('SELECT * FROM RequestLogs INNER JOIN AppLogs ON '
               'RequestLogs.id = AppLogs.request_id%s GROUP BY '
               'RequestLogs.id ORDER BY id DESC')
    else:
      query = 'SELECT * FROM RequestLogs%s ORDER BY id DESC'
    logs = self._conn.execute(query % filter_string,
                              values).fetchmany(count + 1)
    if logging.getLogger(__name__).isEnabledFor(logging.DEBUG):
      self._debug_query(filter_string, values, len(logs))
    for log_row in logs[:count]:
      log = response.add_log()
      self._fill_request_log(log_row, log, request.include_app_logs())
    if len(logs) > count:
      response.mutable_offset().set_request_id(str(logs[-2]['id']))

  def _debug_query(self, filter_string, values, result_count):
    logging.debug('\n\n')
    logging.debug(filter_string)
    logging.debug(values)
    logging.debug('%d results.', result_count)
    logging.debug('DB dump:')
    for l in self._conn.execute('SELECT * FROM RequestLogs'):
      logging.debug('%r %r %d %d %s', l['module'], l['version_id'],
                    l['start_time'], l['end_time'],
                    l['finished'] and 'COMPLETE' or 'INCOMPLETE')

  def _fill_request_log(self, log_row, log, include_app_logs):
    log.set_request_id(str(log_row['user_request_id']))
    log.set_app_id(log_row['app_id'])
    log.set_version_id(log_row['version_id'])
    log.set_module_id(log_row['module'])
    log.set_ip(log_row['ip'])
    log.set_nickname(log_row['nickname'])
    log.set_start_time(log_row['start_time'])
    log.set_host(log_row['host'])
    log.set_end_time(log_row['end_time'])
    log.set_method(log_row['method'])
    log.set_resource(log_row['resource'])
    log.set_status(log_row['status'])
    log.set_response_size(log_row['response_size'])
    log.set_http_version(log_row['http_version'])
    log.set_user_agent(log_row['user_agent'])
    log.set_url_map_entry(log_row['url_map_entry'])
    log.set_latency(log_row['latency'])
    log.set_mcycles(log_row['mcycles'])
    log.set_finished(log_row['finished'])
    log.mutable_offset().set_request_id(str(log_row['id']))
    time_seconds = (log_row['end_time'] or log_row['start_time']) / 10**6
    date_string = time.strftime('%d/%b/%Y:%H:%M:%S %z',
                                time.localtime(time_seconds))
    log.set_combined('%s - %s [%s] "%s %s %s" %d %d - "%s"' %
                     (log_row['ip'], log_row['nickname'], date_string,
                      log_row['method'], log_row['resource'],
                      log_row['http_version'], log_row['status'] or 0,
                      log_row['response_size'] or 0, log_row['user_agent']))
    if include_app_logs:
      log_messages = self._conn.execute(
          'SELECT timestamp, level, message FROM AppLogs '
          'WHERE request_id = ?',
          (log_row['id'],)).fetchall()
      for message in log_messages:
        line = log.add_line()
        line.set_time(message['timestamp'])
        line.set_level(message['level'])
        line.set_log_message(message['message'])

  @staticmethod
  def _extract_read_filters(request):
    """Extracts SQL filters from the LogReadRequest.

    Args:
      request: the incoming LogReadRequest.
    Returns:
      a pair of (filters, values); filters is a list of SQL filter expressions,
      to be joined by AND operators; values is a list of values to be
      interpolated into the filter expressions by the db library.
    """
    filters = []
    values = []

    module_filters = []
    module_values = []
    for module_version in request.module_version_list():
      module_filters.append('(version_id = ? AND module = ?)')
      module_values.append(module_version.version_id())
      module = appinfo.DEFAULT_MODULE
      if module_version.has_module_id():
        module = module_version.module_id()
      module_values.append(module)
    if module_filters:
      filters.append('(' + ' or '.join(module_filters) + ')')
      values += module_values

    if request.has_offset():
      filters.append('RequestLogs.id < ?')
      values.append(int(request.offset().request_id()))
    if request.has_minimum_log_level():
      filters.append('AppLogs.level >= ?')
      values.append(request.minimum_log_level())






    finished_filter = 'finished = 1 '
    finished_filter_values = []
    unfinished_filter = 'finished = 0'
    unfinished_filter_values = []

    if request.has_start_time():
      finished_filter += ' and end_time >= ? '
      finished_filter_values.append(request.start_time())
      unfinished_filter += ' and start_time >= ? '
      unfinished_filter_values.append(request.start_time())
    if request.has_end_time():
      finished_filter += ' and end_time < ? '
      finished_filter_values.append(request.end_time())
      unfinished_filter += ' and start_time < ? '
      unfinished_filter_values.append(request.end_time())

    if request.include_incomplete():
      filters.append(
          '((' + finished_filter + ') or (' + unfinished_filter + '))')
      values += finished_filter_values + unfinished_filter_values
    else:
      filters.append(finished_filter)
      values += finished_filter_values

    return filters, values

  def _Dynamic_SetStatus(self, unused_request, unused_response,
                         unused_request_id):
    raise NotImplementedError

  def _Dynamic_Usage(self, unused_request, unused_response, unused_request_id):
    raise apiproxy_errors.CapabilityDisabledError('Usage not allowed in tests.')
