Skip to content

Instantly share code, notes, and snippets.

@r3
Last active August 29, 2015 14:10
Show Gist options
  • Save r3/d43c5175134b029f7ff3 to your computer and use it in GitHub Desktop.
Save r3/d43c5175134b029f7ff3 to your computer and use it in GitHub Desktop.
import datetime
import itertools
import os
import pathlib
import smtplib
FIFTEEN_MINUTES = datetime.timedelta(minutes=15)
class EmailedLog():
"""Manages a buffer for logging that is flushed via SMTP"""
template = "To: {}\r\nFrom: {}\r\nSubject: {}\r\n\r\n{}"
def __init__(self,
destination,
credentials,
server,
port=587,
subject="Old Files",
line_limit=20):
"""Define the details of the buffer and configuration of email
Required Arguments:
destination - A `str` containing an email address of the recipient
credentials - A `dict` containing 'user' and 'pass' key/values for
logging into the server
server - A `str` with the URL of the SMTP server to be used.
TLS will be used when connecting.
Optional Arguments:
port - An `int` with the port to be used. 587 by default.
subject - A `str` with the subject line. "Old Files" by default.
line_limit - An `int` limiting the number of lines to be buffered
before sending an email and clearing the buffer.
"""
self.line_limit = line_limit
self.destination = destination
self.credentials = credentials
self.subject = subject
self.server = server
self.port = port
self._buffer = []
def __call__(self, message):
"""Accepts a message to log
After logging the message, a check is made to determine if the buffer
is full. If the buffer is full, it will be flushed.
"""
self._buffer.append(str(message))
if len(self._buffer) >= self.line_limit:
self.flush_buffer()
def flush_buffer(self):
"""Flushes the buffer to an email"""
with smtplib.SMTP(self.server, self.port) as server:
message = self.template.format(self.destination,
self.credentials['user'],
self.subject,
'\r\n'.join(self._buffer))
server.ehlo()
server.starttls()
server.login(self.credentials['user'],
self.credentials['pass'])
server.sendmail(self.credentials['user'],
[self.destination],
message)
self._buffer = []
log = EmailedLog(destination='[email protected]',
credentials={'user': '[email protected]',
'pass': 'some_password'},
server='smtp.some_address.com')
def time_since_modified(path):
"""Determines the time that has passed since a file was last modified
Accepts a path and returns a `datetime.datetime` and will log any permission
problems.
"""
try:
timestamp = os.path.getmtime(str(path))
last_modified = datetime.datetime.fromtimestamp(timestamp)
return datetime.datetime.now() - last_modified
except PermissionError as err:
log("PermissionError opening file: {}".format(repr(path)))
return datetime.datetime.min
def text_files_older_than_limit(path, limit):
"""Returns a generator of the files older than a given time limit
The limit should be a `datetime.timedelta` object representing the
minimum age of modification for a file to be included in the result.
Only files with a ".txt" extension will be examined.
"""
wrapped_path = pathlib.Path(path)
text_files = wrapped_path.glob('*.txt')
return (x for x in text_files if time_since_modified(x) > limit)
if __name__ == '__main__':
paths = ('first/path/to/check',
'second/path/to/check',
'third/path/to/check')
old_files = itertools.chain.from_iterable(
text_files_older_than_limit(x, limit=FIFTEEN_MINUTES) for x in paths
)
for file in old_files:
log(file)
log.flush_buffer()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment