68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
|
from google_auth_oauthlib.flow import InstalledAppFlow
|
||
|
|
from googleapiclient.discovery import build
|
||
|
|
import nlp_summary
|
||
|
|
|
||
|
|
SCOPES = ["https://www.googleapis.com/auth/gmail.modify"]
|
||
|
|
nlp = nlp_summary.load("en_core_web_sm")
|
||
|
|
|
||
|
|
# Define keyword-based categories
|
||
|
|
CATEGORIES = {
|
||
|
|
"Work": ["meeting", "deadline", "project", "report"],
|
||
|
|
"Finance": ["invoice", "bill", "receipt", "payment", "tax"],
|
||
|
|
"Security": ["verification", "sign in attempt", "password"],
|
||
|
|
"Promotions": ["sale", "deal", "offer", "discount", "promotion"],
|
||
|
|
"Events": ["webinar", "conference", "event", "invitation"]
|
||
|
|
}
|
||
|
|
|
||
|
|
def authenticate_gmail():
|
||
|
|
flow = InstalledAppFlow.from_client_secrets_file("credentials.json", SCOPES)
|
||
|
|
creds = flow.run_local_server(port=0)
|
||
|
|
return build("gmail", "v1", credentials=creds)
|
||
|
|
|
||
|
|
def categorize_email(subject):
|
||
|
|
doc = nlp(subject.lower())
|
||
|
|
for category, keywords in CATEGORIES.items():
|
||
|
|
if any(word in doc.text for word in keywords):
|
||
|
|
return category
|
||
|
|
return "Uncategorized"
|
||
|
|
|
||
|
|
def list_and_categorize_emails(service):
|
||
|
|
results = service.users().messages().list(userId="me", maxResults=10).execute()
|
||
|
|
messages = results.get("messages", [])
|
||
|
|
|
||
|
|
for msg in messages:
|
||
|
|
msg_data = service.users().messages().get(userId="me", id=msg["id"]).execute()
|
||
|
|
subject = msg_data.get("snippet", "No Subject")
|
||
|
|
category = categorize_email(subject)
|
||
|
|
|
||
|
|
print(f"📩 Subject: {subject}")
|
||
|
|
print(f" 🏷️ Category: {category}\n")
|
||
|
|
|
||
|
|
# Apply the category label in Gmail
|
||
|
|
label_email(service, msg["id"], category)
|
||
|
|
|
||
|
|
def label_email(service, message_id, category):
|
||
|
|
label_id = get_or_create_label(service, category)
|
||
|
|
service.users().messages().modify(
|
||
|
|
userId="me",
|
||
|
|
id=message_id,
|
||
|
|
body={"addLabelIds": [label_id]}
|
||
|
|
).execute()
|
||
|
|
|
||
|
|
def get_or_create_label(service, label_name):
|
||
|
|
labels = service.users().labels().list(userId="me").execute().get("labels", [])
|
||
|
|
for label in labels:
|
||
|
|
if label["name"].lower() == label_name.lower():
|
||
|
|
return label["id"]
|
||
|
|
|
||
|
|
# Create a new label if not found
|
||
|
|
label = service.users().labels().create(
|
||
|
|
userId="me",
|
||
|
|
body={"name": label_name, "labelListVisibility": "labelShow"}
|
||
|
|
).execute()
|
||
|
|
return label["id"]
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
gmail_service = authenticate_gmail()
|
||
|
|
list_and_categorize_emails(gmail_service)
|