pip install transformers torch google-auth google-auth-oauthlib google-auth-httplib2 google-api-python-client

from transformers import pipeline

import torch



class EmailClassifier:

    def __init__(self):

        # Use a lightweight model optimized for text classification

        self.classifier = pipeline(

            "zero-shot-classification",

            model="facebook/bart-large-mnli",

            device=0 if torch.cuda.is_available() else -1

        )



        # Define our business categories

        self.categories = [

            "urgent customer issue",

            "spam or promotional", 

            "customer support request",

            "sales inquiry",

            "internal communication",

            "newsletter or update"

        ]



    def classify_email(self, subject, body):

        # Combine subject and first 500 chars of body for classification

        text = f"{subject} {body[:500]}"



        result = self.classifier(text, self.categories)



        # Return the top prediction with confidence score

        return {

            'category': result['labels'][0],

            'confidence': result['scores'][0],

            'all_scores': dict(zip(result['labels'], result['scores']))

        }



# Test our classifier

classifier = EmailClassifier()



# Example email

test_subject = "URGENT: Website is down, customers can't checkout"

test_body = "Hi team, we're getting reports that our e-commerce site is completely inaccessible. This is affecting sales immediately."



result = classifier.classify_email(test_subject, test_body)

print(f"Category: {result['category']}")

print(f"Confidence: {result['confidence']:.2f}")

import pickle

import os

from google.auth.transport.requests import Request

from google_auth_oauthlib.flow import InstalledAppFlow

from googleapiclient.discovery import build

import base64

import email



class GmailConnector:

    SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']



    def __init__(self, credentials_file='credentials.json'):

        self.service = self._authenticate(credentials_file)



    def _authenticate(self, credentials_file):

        creds = None

        if os.path.exists('token.pickle'):

            with open('token.pickle', 'rb') as token:

                creds = pickle.load(token)



        if not creds or not creds.valid:

            if creds and creds.expired and creds.refresh_token:

                creds.refresh(Request())

            else:

                flow = InstalledAppFlow.from_client_secrets_file(

                    credentials_file, self.SCOPES)

                creds = flow.run_local_server(port=0)



            with open('token.pickle', 'wb') as token:

                pickle.dump(creds, token)



        return build('gmail', 'v1', credentials=creds)



    def get_recent_emails(self, max_results=10):

        """Fetch recent unread emails"""

        try:

            results = self.service.users().messages().list(

                userId='me', 

                q='is:unread',

                maxResults=max_results

            ).execute()



            messages = results.get('messages', [])

            emails = []



            for message in messages:

                msg = self.service.users().messages().get(

                    userId='me', 

                    id=message['id']

                ).execute()



                email_data = self._parse_email(msg)

                emails.append(email_data)



            return emails



        except Exception as error:

            print(f'An error occurred: {error}')

            return []



    def _parse_email(self, message):

        """Extract subject, sender, and body from Gmail message"""

        headers = message['payload'].get('headers', [])



        subject = next((h['value'] for h in headers if h['name'] == 'Subject'), 'No Subject')

        sender = next((h['value'] for h in headers if h['name'] == 'From'), 'Unknown Sender')



        # Extract body text

        body = self._get_email_body(message['payload'])



        return {

            'id': message['id'],

            'subject': subject,

            'sender': sender,

            'body': body

        }



    def _get_email_body(self, payload):

        """Recursively extract email body text"""

        body = ""



        if 'parts' in payload:

            for part in payload['parts']:

                body += self._get_email_body(part)

        else:

            if payload.get('mimeType') == 'text/plain':

                data = payload.get('body', {}).get('data')

                if data:

                    body = base64.urlsafe_b64decode(data).decode('utf-8')



        return body

import time

import json

from datetime import datetime



class SmartEmailProcessor:

    def __init__(self):

        self.classifier = EmailClassifier()

        self.gmail = GmailConnector()

        self.processed_emails = set()



    def process_new_emails(self):

        """Process unread emails and classify them"""

        emails = self.gmail.get_recent_emails(max_results=20)



        results = []

        for email_data in emails:

            if email_data['id'] not in self.processed_emails:

                classification = self.classifier.classify_email(

                    email_data['subject'], 

                    email_data['body']

                )



                result = {

                    'timestamp': datetime.now().isoformat(),

                    'email_id': email_data['id'],

                    'sender': email_data['sender'],

                    'subject': email_data['subject'],

                    'category': classification['category'],

                    'confidence': classification['confidence']

                }



                results.append(result)

                self.processed_emails.add(email_data['id'])



                # Log high-confidence urgent emails

                if ('urgent' in classification['category'].lower() and 

                    classification['confidence'] > 0.8):

                    print(f"🚨 URGENT EMAIL DETECTED: {email_data['subject']}")



        return results



    def run_continuous(self, check_interval=300):  # 5 minutes

        """Run as a background service"""

        print(f"Starting email classifier service...")

        print(f"Checking for new emails every {check_interval} seconds")



        while True:

            try:

                results = self.process_new_emails()

                if results:

                    print(f"Processed {len(results)} new emails")

                    # Here you could save to database, send notifications, etc.



                time.sleep(check_interval)



            except KeyboardInterrupt:

                print("Service stopped by user")

                break

            except Exception as e:

                print(f"Error processing emails: {e}")

                time.sleep(60)  # Wait a minute before retrying



# Run the processor

if __name__ == "__main__":

    processor = SmartEmailProcessor()



    # Test with current emails

    results = processor.process_new_emails()

    for result in results:

        print(f"📧 {result['subject'][:50]}... -> {result['category']} ({result['confidence']:.2f})")



    # Uncomment to run as service

    # processor.run_continuous()

[Unit]

Description=Smart Email Classifier

After=network.target



[Service]

Type=simple

User=your-username

WorkingDirectory=/path/to/your/script

ExecStart=/usr/bin/python3 /path/to/your/script/email_processor.py

Restart=always

RestartSec=10



[Install]

WantedBy=multi-user.target

sudo systemctl enable email-classifier

sudo systemctl start email-classifier