-
-
Notifications
You must be signed in to change notification settings - Fork 47
/
log.py
122 lines (97 loc) · 4.01 KB
/
log.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
import logging
import os
import sys
from datetime import datetime
from logging.handlers import RotatingFileHandler
from typing import List, Optional
from . import dirs
from .decorators import deprecated
# NOTE: Will be removed in a future version since it's not compatible
# with running a multi-service process.
# TODO: prefix with `_`
log_file_path = None
@deprecated
def get_log_file_path() -> Optional[str]: # pragma: no cover
"""DEPRECATED: Use get_latest_log_file instead."""
return log_file_path
def setup_logging(
name: str,
testing=False,
verbose=False,
log_stderr=True,
log_file=False,
): # pragma: no cover
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG if verbose else logging.INFO)
root_logger.handlers = []
# run with LOG_LEVEL=DEBUG to customize log level across all AW components
log_level = os.environ.get("LOG_LEVEL")
if log_level:
if hasattr(logging, log_level.upper()):
root_logger.setLevel(getattr(logging, log_level.upper()))
else:
root_logger.warning(
f"No logging level called {log_level} (as specified in env var)"
)
if log_stderr:
root_logger.addHandler(_create_stderr_handler())
if log_file:
root_logger.addHandler(_create_file_handler(name, testing=testing))
def excepthook(type_, value, traceback):
root_logger.exception("Unhandled exception", exc_info=(type_, value, traceback))
# call the default excepthook if log_stderr isn't true
# (otherwise it'll just get duplicated)
if not log_stderr:
sys.__excepthook__(type_, value, traceback)
sys.excepthook = excepthook
def _get_latest_log_files(name, testing=False) -> List[str]: # pragma: no cover
"""
Returns a list with the paths of all available logfiles for `name`,
sorted by latest first.
"""
log_dir = dirs.get_log_dir(name)
files = filter(lambda filename: name in filename, os.listdir(log_dir))
files = filter(
lambda filename: "testing" in filename
if testing
else "testing" not in filename,
files,
)
return [os.path.join(log_dir, filename) for filename in sorted(files, reverse=True)]
def get_latest_log_file(name, testing=False) -> Optional[str]: # pragma: no cover
"""
Returns the filename of the last logfile with ``name``.
Useful when you want to read the logfile of another ActivityWatch service.
"""
last_logs = _get_latest_log_files(name, testing=testing)
return last_logs[0] if last_logs else None
def _create_stderr_handler() -> logging.Handler: # pragma: no cover
stderr_handler = logging.StreamHandler(stream=sys.stderr)
stderr_handler.setFormatter(_create_human_formatter())
return stderr_handler
def _create_file_handler(
name, testing=False, log_json=False
) -> logging.Handler: # pragma: no cover
log_dir = dirs.get_log_dir(name)
# Set logfile path and name
global log_file_path
# Should result in something like:
# $LOG_DIR/aw-server_testing_2017-01-05T00:21:39.log
file_ext = ".log.json" if log_json else ".log"
now_str = str(datetime.now().replace(microsecond=0).isoformat()).replace(":", "-")
log_name = name + "_" + ("testing_" if testing else "") + now_str + file_ext
log_file_path = os.path.join(log_dir, log_name)
# Create rotating logfile handler, max 10MB per file, 3 files max
# Prevents logfile from growing too large, like in:
# - https://github.com/ActivityWatch/activitywatch/issues/815#issue-1423555466
# - https://github.com/ActivityWatch/activitywatch/issues/756#issuecomment-1266662861
fh = RotatingFileHandler(
log_file_path, mode="a", maxBytes=10 * 1024 * 1024, backupCount=3
)
fh.setFormatter(_create_human_formatter())
return fh
def _create_human_formatter() -> logging.Formatter: # pragma: no cover
return logging.Formatter(
"%(asctime)s [%(levelname)-5s]: %(message)s (%(name)s:%(lineno)s)",
"%Y-%m-%d %H:%M:%S",
)