logger.isolog() - Code Metrics - isomeric/isomer - Measure and Improve Code Quality continuously with Scrutinizer

logger.isolog() F
last analyzed 2021-02-24 12:38 UTC

↳ Parent: logger

Complexity

Conditions

Size

Total Lines	165
Code Lines	102

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	31
eloc	102
nop	2
dl	0
loc	165
rs	0
c	0
b	0
f	0

How to fix Long Method Complexity

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

# Isomer - The distributed application framework
# ==============================================
# Copyright (C) 2011-2020 Heiko 'riot' Weinen <[email protected]> and others.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""
Module: Logger
==============

Isomer's own logger to avoid namespace clashes etc. Comes with some fancy
functions.

Log Levels
----------

verbose = 5
debug = 10
info = 20
warn = 30
error = 40
critical = 50
off = 100


"""

# from circuits.core import Event
import random
import string
import pprint
import tempfile
from traceback import format_exception

# from circuits import Component, handler
# from uuid import uuid4
# import json


import time
import datetime
import sys
import inspect

import os

root = None

temp = 1
events = 2
network = 4
verbose = 5
debug = 10
info = 20
warn = 30
error = 40
critical = 50
hilight = 60
version = 99
off = 100

# https://en.wikipedia.org/wiki/ANSI_escape_code#Colors
level_data = {
    temp: ["TEMP", "\033[1;30m"],
    events: ["EVENT", "\033[1:36m"],
    verbose: ["VERB", "\033[1;30m"],
    network: ["NET", "\033[1;34m"],
    debug: ["DEBUG", "\033[1;97m"],
    info: ["INFO", "\033[1;92m"],
    warn: ["WARN", "\033[1;93m"],
    error: ["ERROR", "\033[1;31;43m"],
    critical: ["CRIT", "\033[1;33;41m"],
    hilight: ["HILIGHT", "\033[1;4;34;106m"],
    version: ["VER", "\033[1;96;44m"],
}

terminator = "\033[0m"

process_identifier = ''.join(
    [random.choice(string.ascii_letters + string.digits) for i in range(3)])

count = 0

logfile = "/var/log/isomer/service.log"
log_emergency = False

console = verbose
live = False

verbosity = {"global": console, "file": off, "system": info, "console": console}

uncut = True
color = False

mute = []
solo = []
mark = []

LiveLog = []

start = time.time()


def set_color(value: bool = True):
    """Activate colorful logging"""
    global color
    color = value


def set_verbosity(global_level: int, console_level: int = None, file_level: int = None):
    """Adjust logging verbosity"""
    global verbosity

    if console_level is None:
        console_level = verbosity["console"]
    if file_level is None:
        file_level = verbosity["file"]

    verbosity["global"] = global_level
    verbosity["console"] = console_level
    verbosity["file"] = file_level


def get_verbosity():
    """Returns logging verbosity"""
    global verbosity

    return verbosity


def set_logfile(path: str, instance: str, filename: str = None):
    """
    Specify logfile path

    :param path: Path to the logfile
    :param instance: Name of the instance
    :param filename: Exact name of logfile
    """

    global logfile

    if path is None:
        path = "."

    if filename is not None:
        logfile = os.path.join(os.path.normpath(path), filename)
    else:
        logfile = os.path.join(os.path.normpath(path), "isomer." + instance + ".log")


def get_logfile() -> str:
    """Return the whole filename of the logfile"""
    return logfile


def clear():
    """Clear the live log"""
    global LiveLog

    LiveLog = []


def is_muted(what) -> bool:
    """
    Checks if a logged event is to be muted for debugging purposes.

    Also goes through the solo list - only items in there will be logged!

    :param what:
    """

    state = False

    for item in solo:
        if item not in what:
            state = True
        else:
            state = False
            break

    for item in mute:
        if item in what:
            state = True
            break

    return state


def is_marked(what) -> bool:
    """Check if log line qualifies for highlighting"""

    for item in mark:
        if item in what:
            return True

    return False


def setup_root(new_root: "isomer.components.Component"):
    """
    Sets up the root component, so the logger knows where to send logging
    signals.

    :param isomer.components.Component new_root:
    """
    global root

    root = new_root


# noinspection PyUnboundLocalVariable,PyIncorrectDocstring
def isolog(*what, **kwargs):
    """Logs all non keyword arguments.

    :param tuple/str what: Loggable objects (i.e. they have a string
        representation)
    :param int lvl: Debug message level
    :param str emitter: Optional log source, where this can't be determined
        automatically
    :param str sourceloc: Give specific source code location hints, used
        internally
    :param int frameref: Specify a non default frame for tracebacks
    :param bool tb: Include a traceback
    :param bool nc: Do not use color
    :param bool exc: Switch to better handle exceptions, use if logging in an
        except clause

    """

    global count
    global verbosity

    lvl = kwargs.get("lvl", info)

    if lvl < verbosity["global"]:
        return

    def assemble_things(things) -> str:
        result = ""

        for thing in things:
            result += " "
            if kwargs.get("pretty", False) and not isinstance(thing, str):
                result += "\n" + pprint.pformat(thing)
            else:
                result += str(thing)

        return result

    def write_to_log(message: str):
        global logfile

        while True:
            try:
                f = open(logfile, "a")
                f.write(message + "\n")
                f.flush()
                f.close()

                break
            except IOError:
                global log_emergency
                print("Cannot open logfile '%s' for writing" % logfile)

                if log_emergency is True:
                    print(
                        "Safe temporary logging is not working either, "
                        "giving up on file logging"
                    )
                    break

                logfile = os.path.join(
                    tempfile.mkdtemp(prefix="isomer_"),
                    "emergency_log"
                )
                print("Logging to temporary logfile '%s'" % logfile)
                log_emergency = True

                continue

    def write_to_console(message: str):
        try:
            print(message)
        except UnicodeEncodeError as e:
            print(message.encode("utf-8"))
            isolog("Bad encoding encountered on previous message:", e, lvl=error)
        except BlockingIOError:
            isolog("Too long log line encountered:", message[:20], lvl=warn)

    # Count all messages (missing numbers give a hint at too high log level)
    count += 1

    emitter = kwargs.get("emitter", "UNKNOWN")
    traceback = kwargs.get("tb", False)
    frame_ref = kwargs.get("frame_ref", 0)
    no_color = kwargs.get("nc", False)
    exception = kwargs.get("exc", False)

    timestamp = time.time()
    runtime = timestamp - start
    callee = None

    if exception:
        exc_type, exc_obj, exc_tb = sys.exc_info()  # NOQA

    if verbosity["global"] <= debug or traceback:
        # Automatically log the current function details.

        if "sourceloc" not in kwargs:
            frame = kwargs.get("frame", frame_ref)

            # Get the previous frame in the stack, otherwise it would
            # be this function
            current_frame = inspect.currentframe()
            while frame > 0:
                frame -= 1
                current_frame = current_frame.f_back

            func = current_frame.f_code
            # Dump the message + the name of this function to the log.

            if exception:
                # noinspection PyUnboundLocalVariable
                line_no = exc_tb.tb_lineno

                if lvl <= error:
                    lvl = error
            else:
                line_no = func.co_firstlineno

            callee = "[%.10s@%s:%i]" % (func.co_name, func.co_filename, line_no)
        else:
            callee = kwargs["sourceloc"]

    now = datetime.datetime.now().isoformat()
    msg = "[%s]:[%s]:%s:%.5f:%3i: [%5s]" % (
        now,
        process_identifier,
        level_data[lvl][0],
        runtime,
        count,
        emitter,
    )

    if callee:
        if not uncut and lvl > 10:
            msg += "%-60s" % callee
        else:
            msg += "%s" % callee

    content = assemble_things(what)
    msg += content

    if exception:
        msg += "\n" + "".join(format_exception(exc_type, exc_obj, exc_tb))


    if is_muted(msg):
        return

    if not uncut and lvl > 10 and len(msg) > 1000:
        msg = msg[:1000]

    if lvl >= verbosity["file"]:
        write_to_log(msg)

    if is_marked(msg):
        lvl = hilight

    if lvl >= verbosity["console"]:
        output = str(msg)
        if color and not no_color:
            output = level_data[lvl][1] + output + terminator
        write_to_console(output)

    if live:
        item = [now, lvl, runtime, count, emitter, str(content)]
        LiveLog.append(item)


1			#!/usr/bin/env python
2			# -- coding: UTF-8 --
3
4			# Isomer - The distributed application framework
5			# ==============================================
6			# Copyright (C) 2011-2020 Heiko 'riot' Weinen <[email protected]> and others.
7			#
8			# This program is free software: you can redistribute it and/or modify
9			# it under the terms of the GNU Affero General Public License as published by
10			# the Free Software Foundation, either version 3 of the License, or
11			# (at your option) any later version.
12			#
13			# This program is distributed in the hope that it will be useful,
14			# but WITHOUT ANY WARRANTY; without even the implied warranty of
15			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16			# GNU Affero General Public License for more details.
17			#
18			# You should have received a copy of the GNU Affero General Public License
19			# along with this program. If not, see <http://www.gnu.org/licenses/>.
20
21			"""
22			Module: Logger
23			==============
24
25			Isomer's own logger to avoid namespace clashes etc. Comes with some fancy
26			functions.
27
28			Log Levels
29			----------
30
31			verbose = 5
32			debug = 10
33			info = 20
34			warn = 30
35			error = 40
36			critical = 50
37			off = 100
38
39
40			"""
41
42			# from circuits.core import Event
43			import random
44			import string
45			import pprint
46			import tempfile
47			from traceback import format_exception
48
49			# from circuits import Component, handler
50			# from uuid import uuid4
51			# import json
52
53
54			import time
55			import datetime
56			import sys
57			import inspect
58
59			import os
60
61			root = None
62
63			temp = 1
64			events = 2
65			network = 4
66			verbose = 5
67			debug = 10
68			info = 20
69			warn = 30
70			error = 40
71			critical = 50
72			hilight = 60
73			version = 99
74			off = 100
75
76			# https://en.wikipedia.org/wiki/ANSI_escape_code#Colors
77			level_data = {
78			temp: ["TEMP", "\033[1;30m"],
79			events: ["EVENT", "\033[1:36m"],
80			verbose: ["VERB", "\033[1;30m"],
81			network: ["NET", "\033[1;34m"],
82			debug: ["DEBUG", "\033[1;97m"],
83			info: ["INFO", "\033[1;92m"],
84			warn: ["WARN", "\033[1;93m"],
85			error: ["ERROR", "\033[1;31;43m"],
86			critical: ["CRIT", "\033[1;33;41m"],
87			hilight: ["HILIGHT", "\033[1;4;34;106m"],
88			version: ["VER", "\033[1;96;44m"],
89			}
90
91			terminator = "\033[0m"
92
93			process_identifier = ''.join(
94			[random.choice(string.ascii_letters + string.digits) for i in range(3)])
95
96			count = 0
97
98			logfile = "/var/log/isomer/service.log"
99			log_emergency = False
100
101			console = verbose
102			live = False
103
104			verbosity = {"global": console, "file": off, "system": info, "console": console}
105
106			uncut = True
107			color = False
108
109			mute = []
110			solo = []
111			mark = []
112
113			LiveLog = []
114
115			start = time.time()
116
117
118			def set_color(value: bool = True):
119			"""Activate colorful logging"""
120			global color
121			color = value
122
123
124			def set_verbosity(global_level: int, console_level: int = None, file_level: int = None):
125			"""Adjust logging verbosity"""
126			global verbosity
127
128			if console_level is None:
129			console_level = verbosity["console"]
130			if file_level is None:
131			file_level = verbosity["file"]
132
133			verbosity["global"] = global_level
134			verbosity["console"] = console_level
135			verbosity["file"] = file_level
136
137
138			def get_verbosity():
139			"""Returns logging verbosity"""
140			global verbosity
141
142			return verbosity
143
144
145			def set_logfile(path: str, instance: str, filename: str = None):
146			"""
147			Specify logfile path
148
149			:param path: Path to the logfile
150			:param instance: Name of the instance
151			:param filename: Exact name of logfile
152			"""
153
154			global logfile
155
156			if path is None:
157			path = "."
158
159			if filename is not None:
160			logfile = os.path.join(os.path.normpath(path), filename)
161			else:
162			logfile = os.path.join(os.path.normpath(path), "isomer." + instance + ".log")
163
164
165			def get_logfile() -> str:
166			"""Return the whole filename of the logfile"""
167			return logfile
168
169
170			def clear():
171			"""Clear the live log"""
172			global LiveLog
173
174			LiveLog = []
175
176
177			def is_muted(what) -> bool:
178			"""
179			Checks if a logged event is to be muted for debugging purposes.
180
181			Also goes through the solo list - only items in there will be logged!
182
183			:param what:
184			"""
185
186			state = False
187
188			for item in solo:
189			if item not in what:
190			state = True
191			else:
192			state = False
193			break
194
195			for item in mute:
196			if item in what:
197			state = True
198			break
199
200			return state
201
202
203			def is_marked(what) -> bool:
204			"""Check if log line qualifies for highlighting"""
205
206			for item in mark:
207			if item in what:
208			return True
209
210			return False
211
212
213			def setup_root(new_root: "isomer.components.Component"):
214			"""
215			Sets up the root component, so the logger knows where to send logging
216			signals.
217
218			:param isomer.components.Component new_root:
219			"""
220			global root
221
222			root = new_root
223
224
225			# noinspection PyUnboundLocalVariable,PyIncorrectDocstring
226			def isolog(what, *kwargs):
227			"""Logs all non keyword arguments.
228
229			:param tuple/str what: Loggable objects (i.e. they have a string
230			representation)
231			:param int lvl: Debug message level
232			:param str emitter: Optional log source, where this can't be determined
233			automatically
234			:param str sourceloc: Give specific source code location hints, used
235			internally
236			:param int frameref: Specify a non default frame for tracebacks
237			:param bool tb: Include a traceback
238			:param bool nc: Do not use color
239			:param bool exc: Switch to better handle exceptions, use if logging in an
240			except clause
241
242			"""
243
244			global count
245			global verbosity
246
247			lvl = kwargs.get("lvl", info)
248
249			if lvl < verbosity["global"]:
250			return
251
252			def assemble_things(things) -> str:
253			result = ""
254
255			for thing in things:
256			result += " "
257			if kwargs.get("pretty", False) and not isinstance(thing, str):
258			result += "\n" + pprint.pformat(thing)
259			else:
260			result += str(thing)
261
262			return result
263
264			def write_to_log(message: str):
265			global logfile
266
267			while True:
268			try:
269			f = open(logfile, "a")
270			f.write(message + "\n")
271			f.flush()
272			f.close()
273
274			break
275			except IOError:
276			global log_emergency
277			print("Cannot open logfile '%s' for writing" % logfile)
278
279			if log_emergency is True:
280			print(
281			"Safe temporary logging is not working either, "
282			"giving up on file logging"
283			)
284			break
285
286			logfile = os.path.join(
287			tempfile.mkdtemp(prefix="isomer_"),
288			"emergency_log"
289			)
290			print("Logging to temporary logfile '%s'" % logfile)
291			log_emergency = True
292
293			continue
294
295			def write_to_console(message: str):
296			try:
297			print(message)
298			except UnicodeEncodeError as e:
299			print(message.encode("utf-8"))
300			isolog("Bad encoding encountered on previous message:", e, lvl=error)
301			except BlockingIOError:
302			isolog("Too long log line encountered:", message[:20], lvl=warn)
303
304			# Count all messages (missing numbers give a hint at too high log level)
305			count += 1
306
307			emitter = kwargs.get("emitter", "UNKNOWN")
308			traceback = kwargs.get("tb", False)
309			frame_ref = kwargs.get("frame_ref", 0)
310			no_color = kwargs.get("nc", False)
311			exception = kwargs.get("exc", False)
312
313			timestamp = time.time()
314			runtime = timestamp - start
315			callee = None
316
317			if exception:
318			exc_type, exc_obj, exc_tb = sys.exc_info() # NOQA
319
320			if verbosity["global"] <= debug or traceback:
321			# Automatically log the current function details.
322
323			if "sourceloc" not in kwargs:
324			frame = kwargs.get("frame", frame_ref)
325
326			# Get the previous frame in the stack, otherwise it would
327			# be this function
328			current_frame = inspect.currentframe()
329			while frame > 0:
330			frame -= 1
331			current_frame = current_frame.f_back
332
333			func = current_frame.f_code
334			# Dump the message + the name of this function to the log.
335
336			if exception:
337			# noinspection PyUnboundLocalVariable
338			line_no = exc_tb.tb_lineno
			0 ignored issues – show introduced 2020-02-27 17:41 UTC by Report Bug Copy Issue Report The variable `exc_tb` does not seem to be defined in case `exception` on line `317` is `False`. Are you sure this can never be the case? Loading history...
339			if lvl <= error:
340			lvl = error
341			else:
342			line_no = func.co_firstlineno
343
344			callee = "[%.10s@%s:%i]" % (func.co_name, func.co_filename, line_no)
345			else:
346			callee = kwargs["sourceloc"]
347
348			now = datetime.datetime.now().isoformat()
349			msg = "[%s]:[%s]:%s:%.5f:%3i: [%5s]" % (
350			now,
351			process_identifier,
352			level_data[lvl][0],
353			runtime,
354			count,
355			emitter,
356			)
357
358			if callee:
359			if not uncut and lvl > 10:
360			msg += "%-60s" % callee
361			else:
362			msg += "%s" % callee
363
364			content = assemble_things(what)
365			msg += content
366
367			if exception:
368			msg += "\n" + "".join(format_exception(exc_type, exc_obj, exc_tb))
			0 ignored issues – show introduced 2020-02-27 17:41 UTC by Report Bug Copy Issue Report The variable `exc_obj` does not seem to be defined in case `exception` on line `317` is `False`. Are you sure this can never be the case? Loading history... introduced 2020-02-27 17:41 UTC by Report Bug Copy Issue Report The variable `exc_type` does not seem to be defined in case `exception` on line `317` is `False`. Are you sure this can never be the case? Loading history...
369
370			if is_muted(msg):
371			return
372
373			if not uncut and lvl > 10 and len(msg) > 1000:
374			msg = msg[:1000]
375
376			if lvl >= verbosity["file"]:
377			write_to_log(msg)
378
379			if is_marked(msg):
380			lvl = hilight
381
382			if lvl >= verbosity["console"]:
383			output = str(msg)
384			if color and not no_color:
385			output = level_data[lvl][1] + output + terminator
386			write_to_console(output)
387
388			if live:
389			item = [now, lvl, runtime, count, emitter, str(content)]
390			LiveLog.append(item)
391

isomeric / isomer

logger.isolog() F last analyzed 2021-02-24 12:38 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

logger.isolog() F
last analyzed 2021-02-24 12:38 UTC