Last active
August 29, 2015 14:10
-
-
Save r3/d43c5175134b029f7ff3 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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