-
Notifications
You must be signed in to change notification settings - Fork 0
/
fetch_mails.py
99 lines (80 loc) · 3.45 KB
/
fetch_mails.py
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
from gmail_auth import gmail_authenticate
from bs4 import BeautifulSoup
import base64
import re
from datetime import datetime
import email.utils
def search_messages(service, user_id='me'):
query = '(from:noreply@wise.com ("You spent" OR "is now in")) OR (from:service@paypal.de "Von Ihnen gezahlt")'
try:
response = service.users().messages().list(userId=user_id, q=query).execute()
messages = response.get('messages', [])
return messages
except Exception as error:
print(f'An error occurred: {error}')
return None
def get_email_body(service, user_id, msg_id):
try:
message = service.users().messages().get(userId=user_id, id=msg_id, format='full').execute()
parts = [message['payload']]
while parts:
part = parts.pop()
if part.get('parts'):
parts.extend(part['parts'])
if part.get('mimeType') == 'text/html':
data = part['body']['data']
html_content = base64.urlsafe_b64decode(data).decode('utf-8')
return html_content
return ""
except Exception as error:
print(f'An error occurred: {error}')
return None
def parse_email(service, user_id, msg_id):
html_content = get_email_body(service, user_id, msg_id)
soup = BeautifulSoup(html_content, 'html.parser')
if soup.title:
soup.title.decompose()
text_content = soup.get_text(separator=" ", strip=True)
transaction_details = {
'date': None,
'info': None,
'account': None
}
message = service.users().messages().get(userId=user_id, id=msg_id, format='metadata').execute()
headers = message['payload']['headers']
from_header = next((header for header in headers if header['name'] == 'From'), None)
if from_header:
if 'wise.com' in from_header['value']:
transaction_details['account'] = 'Wise'
elif 'paypal.de' in from_header['value']:
transaction_details['account'] = 'PayPal'
date_header = next((header for header in headers if header['name'] == 'Date'), None)
if date_header:
date_tuple = email.utils.parsedate_tz(date_header['value'])
if date_tuple:
transaction_details['date'] = datetime.fromtimestamp(email.utils.mktime_tz(date_tuple)).strftime('%d-%m-%Y %H:%M %p')
wise_pattern = re.compile(r'You spent ([\d,\.]+) ([A-Z]{3}) at ([^.]+)')
paypal_pattern = re.compile(r'Sie haben .* gesendet(?= Ihre)')
wise_match = wise_pattern.search(text_content)
paypal_match = paypal_pattern.search(text_content)
if wise_match:
amount, currency, merchant = wise_match.groups()
transaction_details['info'] = f"You spent {amount} {currency} at {merchant}."
elif paypal_match:
transaction_details['info'] = paypal_match.group(0).strip()
return transaction_details
# Integration in the main workflow
if __name__ == '__main__':
service = gmail_authenticate()
messages = search_messages(service)
if messages:
for msg in messages:
transaction_info = parse_email(service, 'me', msg['id'])
if transaction_info['info']:
print(transaction_info)
#print('Date:', transaction_info['date'])
#print('AI Model Prompt:', transaction_info['info'])
else:
print('Transaction details not found.')
else:
print('No messages found.')