-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathAutoReplier.py
More file actions
166 lines (127 loc) · 5.14 KB
/
AutoReplier.py
File metadata and controls
166 lines (127 loc) · 5.14 KB
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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
import imaplib
import smtplib
import email
import time
import logging
from email.mime.text import MIMEText
from datetime import datetime, timedelta
IMAP_SERVER = 'imap.gmail.com'
SMTP_SERVER = 'smtp.gmail.com'
EMAIL_ACCOUNT = '' # your email here
PASSWORD = '' # your app password here
CHECK_INTERVAL = 5 # check emails every 5 seconds
AUTO_REPLY_MESSAGE = '''
Thank you for your email. I am currently away and will respond to your message when I return.
Best Regards,
'''
SUBJECT = 'Auto-Reply'
# logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('email_autoreplier.log'),
logging.StreamHandler()
]
)
replied_to = set()
def is_no_reply(sender_email):
no_reply_keywords = ['no-reply', 'noreply', 'donotreply', 'do-not-reply', 'automated', 'notification', 'alerts', 'system']
return any(keyword in sender_email.lower() for keyword in no_reply_keywords)
def is_auto_reply(message):
auto_reply_subjects = ['auto', 'automatic', 'reply', 'vacation', 'out of office', 'ooo', 'away']
subject = message.get('Subject', '').lower()
if any(keyword in subject for keyword in auto_reply_subjects):
return True
auto_reply_headers = [
'Auto-Submitted',
'X-Auto-Response-Suppress',
'X-Autoreply',
'X-Autorespond'
]
for header in auto_reply_headers:
if message.get(header):
return True
return False
def send_auto_reply(recipient, subject):
try:
server = smtplib.SMTP_SSL(SMTP_SERVER, 465)
server.login(EMAIL_ACCOUNT, PASSWORD)
msg = MIMEText(AUTO_REPLY_MESSAGE)
msg['Subject'] = SUBJECT
msg['From'] = EMAIL_ACCOUNT
msg['To'] = recipient
server.sendmail(EMAIL_ACCOUNT, recipient, msg.as_string())
server.quit()
logging.info(f'Auto-reply sent to: {recipient}')
return True
except Exception as e:
logging.error(f'Failed to send auto-reply to {recipient}: {str(e)}')
return False
def check_email():
try:
mail = imaplib.IMAP4_SSL(IMAP_SERVER)
mail.login(EMAIL_ACCOUNT, PASSWORD)
mail.select('inbox')
status, data = mail.search(None, 'UNSEEN')
mail_ids = data[0].split()
if not mail_ids:
logging.info('No new emails to process')
mail.logout()
return
logging.info(f'Found {len(mail_ids)} unread emails')
for num in mail_ids:
try:
status, data = mail.fetch(num, '(RFC822)')
raw_email = data[0][1]
message = email.message_from_bytes(raw_email)
# extract sender and message ID
sender_name, sender_email = email.utils.parseaddr(message['From'])
subject = message.get('Subject', '(No Subject)')
logging.info(f'Processing email from: {sender_email}, Subject: {subject}')
# skip if already replied to this sender recently
if sender_email in replied_to:
logging.info(f'Already replied to {sender_email} recently. Skipping.')
continue
# skip no-reply addresses
if is_no_reply(sender_email):
logging.info(f'Skipping no-reply address: {sender_email}')
continue
# skip auto-reply messages
if is_auto_reply(message):
logging.info(f'Skipping auto-reply message from: {sender_email}')
continue
# skip own email address
if sender_email == EMAIL_ACCOUNT:
logging.info('Skipping email from self')
continue
# send auto-reply
if send_auto_reply(sender_email, subject):
replied_to.add(sender_email)
except Exception as e:
logging.error(f'Error processing email {num}: {str(e)}')
mail.logout()
except Exception as e:
logging.error(f'Error checking emails: {str(e)}')
def clean_replied_to_cache():
# clear the replied_to set every 24 hours
global replied_to
replied_to = set()
logging.info('Cleared replied_to cache.')
def main():
logging.info('Starting Auto-Reply')
last_cache_clear = datetime.now()
try:
while True:
check_email()
if datetime.now() - last_cache_clear > timedelta(days=1):
clean_replied_to_cache()
last_cache_clear = datetime.now()
logging.info(f'Sleeping for {CHECK_INTERVAL} seconds.')
time.sleep(CHECK_INTERVAL)
except KeyboardInterrupt:
logging.info('Stopped')
except Exception as e:
logging.error(f'Service encountered an error: {str(e)}')
if __name__ == '__main__':
main()