0

I have currently set up a process for R files where it detects if .rout file contains certain string. However, Python files do not output .rout files. I have a process that writes a csv on a daily basis to an os directory. The logic I want to use is that if that file doesn't update on a certain day, this means, the script failed and I want to get an email alert. So lets say a file in a path

path = 'C:/Python'
file = Data.csv

I want to receive an email whenever the file timestamp is not updated every 24 hours using my below code logic.

My current code for Rout files-

import pandas as pd
import smtplib
from email.message import EmailMessage
import glob
import os
import shutil
df = pd.read_fwf(r'Service-Now-Data.Rout', header=None)
end_str = '- END -'
cols_to_check = ["0"]
def email_alert(subject,body,to):
    msg = EmailMessage()
    msg.set_content(body)
    msg['subject'] = subject
    msg['to'] = to    
    user = "DataScienceScriptAlerts@chxx.com"
    msg['from'] = user
    server = smtplib.SMTP("smtprelay.corp.chxx.com", 25)
    server.starttls()
    #server.login(user,password)
    server.send_message(msg)     
    server.quit()
src = r'C:/R'
dest = r'C:/R/Failed Scripts'
if __name__ == '__main__':
    for col in cols_to_check:
        if not df[0].str.contains(end_str).any():
            body = "The Service-Now-Data.R script in PMIV312 had errors on the last execution" + col + "."
            print(body)
            email_alert("Service-Now-Data failure alert",body,"htvldba@chxx.com")
        if not df[0].str.contains(end_str).any():
                for file_path in glob.glob(os.path.join(src,'*.Rout'), recursive=True):
                    new_path = os.path.join(dest, os.path.basename(file_path))
                    shutil.copy(file_path, new_path)

3 Answers3

2

If you only want to check the files modification time to not be older than a day, this code should do the trick:

import os
import smtplib
from email.message import EmailMessage
from datetime import datetime as dt

def email_alert(subject,body,to):
    msg = EmailMessage()
    msg.set_content(body)
    msg['subject'] = subject
    msg['to'] = to    
    user = "DataScienceScriptAlerts@chxx.com"
    msg['from'] = user
    server = smtplib.SMTP("smtprelay.corp.chxx.com", 25)
    server.starttls()
    #server.login(user,password)
    server.send_message(msg)     
    server.quit()
    
def check_file(file_name):
    alert = None
    file_stat = None
    try:
        file_stat = os.stat(file_name)
    except FileNotFoundError:
        alert = f'File {file_name} does not exist'
    except Exception as err:
        alert = f'Can not get file stats for file {file_name} because of error {str(err)}'
    else:
        file_age=(dt.utcnow()-dt.utcfromtimestamp(file_stat.st_mtime)).total_seconds()

        if file_age>24*60*60:
            days  = int(file_age/(24*60*60))
            hours = int((file_age%(24*60*60))/3600)
            mins  = int((file_age%3600)/60)
            secs  = int(file_age%60)
            alert = ('File {file:s} is {days:d} days and {hours:02d}:{mins:02d}:{secs:02d} hours old'
                        .format(file=file_name,days=days,hours=hours,mins=mins,secs=secs))
    if alert is not None:
        print(f'eMail alert: {alert}')
        # email_alert('Service-Data-Now failure alert', alert, 'htvldba@chxx.com')
    else:
        print(f'Everything is fine with file {file_name}. No email alert!')
        
if __name__ == '__main__':
    path = 'C:/Python'
    file = 'Data.csv'
    check_file(os.path.join(path,file))

See it in action here with additional test cases: https://onlinegdb.com/EK_0XBn8B (you can easily fork the project and then play around changing the code as well!)

Frank
  • 884
  • 2
  • 10
0

Monitoring the file's modification time is one approach if the modification time updating but the contents staying the same shouldn't trigger an email. (For that, see Frank's answer).

Alternatively, if the file updating but the contents staying the same should trigger an email, you could hash the contents and check whether that hash changes. For this approach, see Hashing a file in Python

jonathanfuchs
  • 69
  • 6
0

Try this. Create a python file with a send email script, then call that script in your other notebook:

import smtplib
from datetime import datetime
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText


def send_error_email(exception, message):
    me = "your_email@email.com"
    you = "your_email@email.com"
    msg = MIMEMultipart('alternative')
    msg['Subject'] = "Error in Dealer File Processing"
    msg['From'] = me
    msg['To'] = you
    
    now = datetime.now()
    date = now.strftime("%m/%d/%Y")
    time = now.strftime("%H:%M:%S")

    text = "Exception mail"
    html = f"""
        <html>
          <body>
    
            <b> <b> Exception: {exception} on {date} at {time}.
            <b>
                {message}
            </b>
    
          </body>
        </html>
        """#.format(exception=exception, message=message)

    part1 = MIMEText(text, 'plain')
    part2 = MIMEText(html, 'html')

    msg.attach(part1)
    msg.attach(part2)
    mail = smtplib.SMTP_SSL('smtp.gmail.com', 465)
    mail.ehlo()

    #mail.starttls()
    #mail.ehlo()
   
    mail.login('your_email@email.com', 'password')
    mail.sendmail(me, you, msg.as_string())
    mail.quit()

Then in the other notebook or file (making sure the two are in the same directory):

from send_mail import send_error_email
from datetime import datetime
import logging
import logging.handlers

now = datetime.now()
date = now.strftime("%m_%d_%Y") 
time = now.strftime("%H_%M_%S")
log_filename = "log_" + date + "_" + time + ".txt"

#Initialize log file
logging.basicConfig(filename=log_filename, level=logging.DEBUG, format='%(asctime)s %(levelname)s %(name)s %(message)s', force = True)
logger=logging.getLogger(__name__)
logging.getLogger('PIL').setLevel(logging.WARNING)

'''
The below function is a decorator function that checks every function in the file at runtime for exceptions,
and updates the log file and sends an email with details of exception
'''

def catch_exception_and_send_email(func):
    def exception_catcher(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            print(e)
            send_error_email(exception=e, message="An exception {} occurred in {}".format(e, func.__name__))
            logging.exception('Oh no! {} On {} at {} in: {} function.'.format(e, date, time, func.__name__), exc_info = False)
            sys.exit(1)
    return exception_catcher

This also generates a log file with the error message. Then what you would do is use this function as a decorator on other functions. Example:

@catch_exception_and_send_email
def is_empty(df):
    '''
    This function checks if sheets within a file are empty.
    
    args:
        df : dataframe for each sheet of dealer file
    
    '''
    if(df.empty):
        raise Exception("The sheet: {} is empty".format(sheet_name))
    else:
        pass
    for col in df.columns:
        miss = df[col].isnull().sum()
        if miss>0:
            raise Exception("{} has {} missing value(s)".format(col,miss))
    return None

And if is_empty() fails, it will update the log file and send an email with the error message and time of error (or exception).

Note: The email function is set up for sending emails through gmail. If it is through another provider I would uncomment the mail.starttls() and mail.ehlo(), and remove _SSL and change the port number as shown here:mail = smtplib.SMTP_SSL('smtp.gmail.com', 465) to mail = smtplib.SMTP('email server', change port #). Also this is more of a skeleton that you could try to use to implement your 24-hr cycle checking--I did not include that solution here.

Hefe
  • 421
  • 3
  • 23