| """Logger class for IPython's logging facilities. |
| """ |
|
|
| |
| |
| |
| |
| |
| |
| |
|
|
| |
| |
|
|
| |
| import glob |
| import io |
| import logging |
| import os |
| import time |
|
|
|
|
| |
| logging.getLogger("parso").setLevel(logging.WARNING) |
|
|
| |
| |
| |
|
|
| class Logger: |
| """A Logfile class with different policies for file creation""" |
|
|
| def __init__(self, home_dir, logfname='Logger.log', loghead=u'', |
| logmode='over'): |
|
|
| |
| |
| self.home_dir = home_dir |
|
|
| self.logfname = logfname |
| self.loghead = loghead |
| self.logmode = logmode |
| self.logfile = None |
|
|
| |
| self.log_raw_input = False |
|
|
| |
| self.log_output = False |
|
|
| |
| self.timestamp = False |
|
|
| |
| self.log_active = False |
|
|
| |
| def _set_mode(self,mode): |
| if mode not in ['append','backup','global','over','rotate']: |
| raise ValueError('invalid log mode %s given' % mode) |
| self._logmode = mode |
|
|
| def _get_mode(self): |
| return self._logmode |
|
|
| logmode = property(_get_mode,_set_mode) |
|
|
| def logstart(self, logfname=None, loghead=None, logmode=None, |
| log_output=False, timestamp=False, log_raw_input=False): |
| """Generate a new log-file with a default header. |
| |
| Raises RuntimeError if the log has already been started""" |
|
|
| if self.logfile is not None: |
| raise RuntimeError('Log file is already active: %s' % |
| self.logfname) |
|
|
| |
| if logfname is not None: self.logfname = logfname |
| if loghead is not None: self.loghead = loghead |
| if logmode is not None: self.logmode = logmode |
|
|
| |
| self.timestamp = timestamp |
| self.log_output = log_output |
| self.log_raw_input = log_raw_input |
|
|
| |
| isfile = os.path.isfile |
| logmode = self.logmode |
|
|
| if logmode == 'append': |
| self.logfile = io.open(self.logfname, 'a', encoding='utf-8') |
|
|
| elif logmode == 'backup': |
| if isfile(self.logfname): |
| backup_logname = self.logfname+'~' |
| |
| |
| if isfile(backup_logname): |
| os.remove(backup_logname) |
| os.rename(self.logfname,backup_logname) |
| self.logfile = io.open(self.logfname, 'w', encoding='utf-8') |
|
|
| elif logmode == 'global': |
| self.logfname = os.path.join(self.home_dir,self.logfname) |
| self.logfile = io.open(self.logfname, 'a', encoding='utf-8') |
|
|
| elif logmode == 'over': |
| if isfile(self.logfname): |
| os.remove(self.logfname) |
| self.logfile = io.open(self.logfname,'w', encoding='utf-8') |
|
|
| elif logmode == 'rotate': |
| if isfile(self.logfname): |
| if isfile(self.logfname+'.001~'): |
| old = glob.glob(self.logfname+'.*~') |
| old.sort() |
| old.reverse() |
| for f in old: |
| root, ext = os.path.splitext(f) |
| num = int(ext[1:-1])+1 |
| os.rename(f, root+'.'+repr(num).zfill(3)+'~') |
| os.rename(self.logfname, self.logfname+'.001~') |
| self.logfile = io.open(self.logfname, 'w', encoding='utf-8') |
|
|
| if logmode != 'append': |
| self.logfile.write(self.loghead) |
|
|
| self.logfile.flush() |
| self.log_active = True |
|
|
| def switch_log(self,val): |
| """Switch logging on/off. val should be ONLY a boolean.""" |
|
|
| if val not in [False,True,0,1]: |
| raise ValueError('Call switch_log ONLY with a boolean argument, ' |
| 'not with: %s' % val) |
|
|
| label = {0:'OFF',1:'ON',False:'OFF',True:'ON'} |
|
|
| if self.logfile is None: |
| print(""" |
| Logging hasn't been started yet (use logstart for that). |
| |
| %logon/%logoff are for temporarily starting and stopping logging for a logfile |
| which already exists. But you must first start the logging process with |
| %logstart (optionally giving a logfile name).""") |
|
|
| else: |
| if self.log_active == val: |
| print('Logging is already',label[val]) |
| else: |
| print('Switching logging',label[val]) |
| self.log_active = not self.log_active |
| self.log_active_out = self.log_active |
|
|
| def logstate(self): |
| """Print a status message about the logger.""" |
| if self.logfile is None: |
| print('Logging has not been activated.') |
| else: |
| state = self.log_active and 'active' or 'temporarily suspended' |
| print('Filename :', self.logfname) |
| print('Mode :', self.logmode) |
| print('Output logging :', self.log_output) |
| print('Raw input log :', self.log_raw_input) |
| print('Timestamping :', self.timestamp) |
| print('State :', state) |
|
|
| def log(self, line_mod, line_ori): |
| """Write the sources to a log. |
| |
| Inputs: |
| |
| - line_mod: possibly modified input, such as the transformations made |
| by input prefilters or input handlers of various kinds. This should |
| always be valid Python. |
| |
| - line_ori: unmodified input line from the user. This is not |
| necessarily valid Python. |
| """ |
|
|
| |
| |
| if self.log_raw_input: |
| self.log_write(line_ori) |
| else: |
| self.log_write(line_mod) |
|
|
| def log_write(self, data, kind='input'): |
| """Write data to the log file, if active""" |
|
|
| |
| if self.log_active and data: |
| write = self.logfile.write |
| if kind=='input': |
| if self.timestamp: |
| write(time.strftime('# %a, %d %b %Y %H:%M:%S\n', time.localtime())) |
| write(data) |
| elif kind=='output' and self.log_output: |
| odata = u'\n'.join([u'#[Out]# %s' % s |
| for s in data.splitlines()]) |
| write(u'%s\n' % odata) |
| try: |
| self.logfile.flush() |
| except OSError: |
| print("Failed to flush the log file.") |
| print( |
| f"Please check that {self.logfname} exists and have the right permissions." |
| ) |
| print( |
| "Also consider turning off the log with `%logstop` to avoid this warning." |
| ) |
|
|
| def logstop(self): |
| """Fully stop logging and close log file. |
| |
| In order to start logging again, a new logstart() call needs to be |
| made, possibly (though not necessarily) with a new filename, mode and |
| other options.""" |
|
|
| if self.logfile is not None: |
| self.logfile.close() |
| self.logfile = None |
| else: |
| print("Logging hadn't been started.") |
| self.log_active = False |
|
|
| |
| close_log = logstop |
|
|
