366 lines
14 KiB
JavaScript
366 lines
14 KiB
JavaScript
require('dotenv').config();
|
||
const fs = require('fs');
|
||
const path = require('path');
|
||
const cron = require('node-cron');
|
||
const { google } = require('googleapis');
|
||
|
||
const TOKENS_DIR = path.join(__dirname, 'tokens');
|
||
const CREDENTIALS_PATH = path.join(__dirname, 'credentials.json');
|
||
|
||
function getCredentials() {
|
||
if (!fs.existsSync(CREDENTIALS_PATH) || fs.statSync(CREDENTIALS_PATH).isDirectory()) {
|
||
throw new Error('Google OAuth credentials.json is missing. Please create the gmail-agent-auth Kubernetes Secret with your Google Cloud OAuth credentials.');
|
||
}
|
||
return JSON.parse(fs.readFileSync(CREDENTIALS_PATH));
|
||
}
|
||
|
||
const GEMINI_API_KEY = process.env.GEMINI_API_KEY || 'AIzaSyDKYcgVPN2oJirwzf_td3sBYMnXHWfVphU';
|
||
const WHATSAPP_GATEWAY_URL = process.env.WHATSAPP_GATEWAY_URL || 'http://whatsapp-gateway.ai-agents.svc.cluster.local:5001/api/send-message';
|
||
const WHATSAPP_TO = process.env.WHATSAPP_TO;
|
||
|
||
let baseOAuth2Client;
|
||
let accounts = [];
|
||
|
||
function loadAuth() {
|
||
try {
|
||
if (!fs.existsSync(CREDENTIALS_PATH) || fs.statSync(CREDENTIALS_PATH).isDirectory()) {
|
||
console.warn('⚠️ Missing credentials.json. Gmail Agent will not be able to process emails or authenticate new accounts until this file is provided.');
|
||
return;
|
||
}
|
||
if (!fs.existsSync(TOKENS_DIR)) {
|
||
fs.mkdirSync(TOKENS_DIR, { recursive: true });
|
||
console.log('Created empty tokens directory.');
|
||
}
|
||
|
||
const creds = getCredentials();
|
||
const {client_secret, client_id, redirect_uris} = creds.installed || creds.web;
|
||
baseOAuth2Client = { client_id, client_secret, redirect_uri: redirect_uris[0] || 'urn:ietf:wg:oauth:2.0:oob' };
|
||
|
||
const tokenFiles = fs.readdirSync(TOKENS_DIR).filter(file => file.endsWith('.json'));
|
||
if (tokenFiles.length === 0) {
|
||
console.log('ℹ️ No token files found in tokens/. Waiting for accounts to be added via UI.');
|
||
return;
|
||
}
|
||
|
||
for (const file of tokenFiles) {
|
||
const accountName = file.replace('.json', '');
|
||
const tokenContent = JSON.parse(fs.readFileSync(path.join(TOKENS_DIR, file)));
|
||
|
||
const client = new google.auth.OAuth2(baseOAuth2Client.client_id, baseOAuth2Client.client_secret, baseOAuth2Client.redirect_uri);
|
||
client.setCredentials(tokenContent);
|
||
|
||
accounts.push({ name: accountName, client });
|
||
console.log(`✅ Loaded credentials for account: ${accountName}`);
|
||
}
|
||
} catch (err) {
|
||
console.error('❌ Error loading auth files:', err.message);
|
||
}
|
||
}
|
||
|
||
async function ensureAgentReadLabel(account) {
|
||
const gmail = google.gmail({ version: 'v1', auth: account.client });
|
||
const res = await gmail.users.labels.list({ userId: 'me' });
|
||
const labels = res.data.labels || [];
|
||
let agentReadLabel = labels.find(l => l.name === 'AGENT_READ');
|
||
|
||
if (!agentReadLabel) {
|
||
console.log(`Creating AGENT_READ label for account ${account.name}...`);
|
||
const createRes = await gmail.users.labels.create({
|
||
userId: 'me',
|
||
requestBody: {
|
||
name: 'AGENT_READ',
|
||
labelListVisibility: 'labelShow',
|
||
messageListVisibility: 'show'
|
||
}
|
||
});
|
||
agentReadLabel = createRes.data;
|
||
}
|
||
return agentReadLabel.id;
|
||
}
|
||
|
||
async function getUnreadEmails(account) {
|
||
const gmail = google.gmail({ version: 'v1', auth: account.client });
|
||
const res = await gmail.users.messages.list({
|
||
userId: 'me',
|
||
q: '-label:AGENT_READ newer_than:2d',
|
||
maxResults: 10
|
||
});
|
||
|
||
const messages = res.data.messages || [];
|
||
const emails = [];
|
||
|
||
for (const msg of messages) {
|
||
const fullMsg = await gmail.users.messages.get({ userId: 'me', id: msg.id, format: 'full' });
|
||
const headers = fullMsg.data.payload.headers;
|
||
const subject = headers.find(h => h.name === 'Subject')?.value || 'No Subject';
|
||
const from = headers.find(h => h.name === 'From')?.value || 'Unknown';
|
||
|
||
let body = '';
|
||
if (fullMsg.data.payload.parts) {
|
||
const part = fullMsg.data.payload.parts.find(p => p.mimeType === 'text/plain');
|
||
if (part && part.body.data) {
|
||
body = Buffer.from(part.body.data, 'base64').toString('utf8');
|
||
}
|
||
} else if (fullMsg.data.payload.body && fullMsg.data.payload.body.data) {
|
||
body = Buffer.from(fullMsg.data.payload.body.data, 'base64').toString('utf8');
|
||
}
|
||
|
||
emails.push({ id: msg.id, subject, from, body: body.substring(0, 1000), accountName: account.name, client: account.client });
|
||
}
|
||
return emails;
|
||
}
|
||
|
||
async function markAsAgentRead(email, labelId) {
|
||
const gmail = google.gmail({ version: 'v1', auth: email.client });
|
||
await gmail.users.messages.modify({
|
||
userId: 'me',
|
||
id: email.id,
|
||
requestBody: { addLabelIds: [labelId] }
|
||
});
|
||
}
|
||
|
||
async function analyzeWithGemini(email) {
|
||
const prompt = `You are a strict AI assistant. Your ONLY job is to filter emails to send to WhatsApp.
|
||
|
||
CRITICAL RULES:
|
||
1. STRICTLY IGNORE ALL promotional emails, newsletters, daily updates, market recaps, deals, marketing, product updates, surveys, or general spam. If it looks like a mass email or newsletter, it is NOT important.
|
||
2. ONLY flag truly personal messages, urgent requests from real people, travel updates, direct security alerts, or specific receipts.
|
||
|
||
Email Subject: ${email.subject}
|
||
Email From: ${email.from}
|
||
Email Body Preview:
|
||
${email.body}
|
||
|
||
Determine if this is an important email. Return a strict JSON response. Do not include markdown formatting.
|
||
Format:
|
||
{
|
||
"isImportant": true or false,
|
||
"summary": "1-sentence summary of what the user needs to know if true, otherwise leave empty"
|
||
}`;
|
||
|
||
try {
|
||
const res = await fetch(`https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-flash:generateContent?key=${GEMINI_API_KEY}`, {
|
||
method: 'POST',
|
||
headers: { 'Content-Type': 'application/json' },
|
||
body: JSON.stringify({
|
||
contents: [{ parts: [{ text: prompt }] }],
|
||
generationConfig: {
|
||
temperature: 0.1,
|
||
responseMimeType: "application/json"
|
||
}
|
||
})
|
||
});
|
||
|
||
const data = await res.json();
|
||
if (data.candidates && data.candidates.length > 0) {
|
||
const text = data.candidates[0].content.parts[0].text;
|
||
const parsed = JSON.parse(text);
|
||
return { isImportant: parsed.isImportant, summary: parsed.summary || '' };
|
||
}
|
||
return { isImportant: false, summary: '' };
|
||
} catch (err) {
|
||
console.error('Gemini Error:', err.message);
|
||
return { isImportant: false, summary: '' };
|
||
}
|
||
}
|
||
|
||
async function sendWhatsAppNotification(email, summary) {
|
||
const message = `📧 *[${email.accountName}] Important Email*\n*From:* ${email.from}\n*Subject:* ${email.subject}\n\n🤖 *Summary:* ${summary}`;
|
||
|
||
try {
|
||
await fetch(WHATSAPP_GATEWAY_URL, {
|
||
method: 'POST',
|
||
headers: {
|
||
'Content-Type': 'application/json',
|
||
'x-app-name': 'Gmail Agent'
|
||
},
|
||
body: JSON.stringify({
|
||
to: 'default',
|
||
message: message
|
||
})
|
||
});
|
||
console.log(`Notification sent for: [${email.accountName}] ${email.subject}`);
|
||
} catch (err) {
|
||
console.error('WhatsApp Gateway Error:', err.message);
|
||
}
|
||
}
|
||
|
||
const ORCHESTRATOR_URL = process.env.ORCHESTRATOR_URL || 'http://agents-api.agents-runtime.svc.cluster.local:8000/webhook';
|
||
|
||
async function forwardToOrchestrator(email) {
|
||
console.log(`Forwarding email from owner to orchestrator: ${email.subject}`);
|
||
try {
|
||
await fetch(ORCHESTRATOR_URL, {
|
||
method: 'POST',
|
||
headers: { 'Content-Type': 'application/json' },
|
||
body: JSON.stringify({
|
||
text: `Subject: ${email.subject}\n\n${email.body}`,
|
||
sender: email.from,
|
||
targetJid: 'EMAIL'
|
||
})
|
||
});
|
||
} catch (err) {
|
||
console.error('Failed to forward to orchestrator:', err.message);
|
||
}
|
||
}
|
||
|
||
async function processEmails() {
|
||
console.log(`[${new Date().toISOString()}] Checking for new emails across ${accounts.length} accounts...`);
|
||
try {
|
||
for (const account of accounts) {
|
||
console.log(`Checking account: ${account.name}...`);
|
||
if (!account.agentReadLabelId) {
|
||
account.agentReadLabelId = await ensureAgentReadLabel(account);
|
||
}
|
||
|
||
const emails = await getUnreadEmails(account);
|
||
if (emails.length === 0) {
|
||
console.log(`No new emails for ${account.name}.`);
|
||
continue;
|
||
}
|
||
|
||
console.log(`Found ${emails.length} unread emails for ${account.name}.`);
|
||
|
||
for (const email of emails) {
|
||
if (email.from.toLowerCase().includes('deepkoluguri@gmail.com')) {
|
||
await forwardToOrchestrator(email);
|
||
} else {
|
||
const analysis = await analyzeWithGemini(email);
|
||
if (analysis.isImportant) {
|
||
await sendWhatsAppNotification(email, analysis.summary);
|
||
} else {
|
||
console.log(`Skipped non-important email: ${email.subject}`);
|
||
}
|
||
}
|
||
// Mark as processed using our custom label
|
||
await markAsAgentRead(email, account.agentReadLabelId);
|
||
}
|
||
}
|
||
} catch (err) {
|
||
console.error('Error processing emails:', err.message);
|
||
}
|
||
}
|
||
|
||
// ─── Express Server for UI Integration ─────────────────────────────────────────
|
||
const express = require('express');
|
||
const cors = require('cors');
|
||
const app = express();
|
||
app.use(cors());
|
||
app.use(express.json());
|
||
|
||
// Send an email (called by orchestrator)
|
||
app.post('/api/send-email', async (req, res) => {
|
||
const { to, subject, body } = req.body;
|
||
if (!to || !subject || !body) return res.status(400).json({ error: 'Missing to, subject, or body' });
|
||
|
||
// Use the first connected account to send the email
|
||
if (accounts.length === 0) return res.status(500).json({ error: 'No Gmail accounts connected' });
|
||
const account = accounts[0];
|
||
|
||
try {
|
||
const gmail = google.gmail({ version: 'v1', auth: account.client });
|
||
const rawMessage = [
|
||
`To: ${to}`,
|
||
`Subject: ${subject}`,
|
||
'Content-Type: text/plain; charset="UTF-8"',
|
||
'',
|
||
body
|
||
].join('\n');
|
||
|
||
const encodedMessage = Buffer.from(rawMessage).toString('base64').replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '');
|
||
|
||
await gmail.users.messages.send({
|
||
userId: 'me',
|
||
requestBody: {
|
||
raw: encodedMessage
|
||
}
|
||
});
|
||
console.log(`Sent email to ${to} with subject: ${subject}`);
|
||
res.json({ success: true });
|
||
} catch (err) {
|
||
console.error('Error sending email:', err.message);
|
||
res.status(500).json({ error: err.message });
|
||
}
|
||
});
|
||
|
||
// Get all connected accounts
|
||
app.get('/api/accounts', (req, res) => {
|
||
res.json({
|
||
success: true,
|
||
accounts: accounts.map(a => ({ name: a.name }))
|
||
});
|
||
});
|
||
|
||
// Generate Auth URL for a new account
|
||
app.post('/api/auth/url', (req, res) => {
|
||
const { accountName } = req.body;
|
||
if (!accountName) return res.status(400).json({ success: false, error: 'accountName is required' });
|
||
|
||
try {
|
||
const creds = getCredentials();
|
||
const {client_secret, client_id, redirect_uris} = creds.installed || creds.web;
|
||
const oAuth2Client = new google.auth.OAuth2(client_id, client_secret, redirect_uris[0] || 'urn:ietf:wg:oauth:2.0:oob');
|
||
|
||
const SCOPES = ['https://www.googleapis.com/auth/gmail.readonly', 'https://www.googleapis.com/auth/gmail.modify'];
|
||
const authUrl = oAuth2Client.generateAuthUrl({
|
||
access_type: 'offline',
|
||
prompt: 'consent',
|
||
scope: SCOPES,
|
||
state: accountName // pass the accountName in state so we don't lose it across redirects
|
||
});
|
||
|
||
res.json({ success: true, url: authUrl });
|
||
} catch (e) {
|
||
res.status(500).json({ success: false, error: e.message });
|
||
}
|
||
});
|
||
|
||
// Verify token and hot-reload account
|
||
app.post('/api/auth/token', (req, res) => {
|
||
const { accountName, code } = req.body;
|
||
if (!accountName || !code) return res.status(400).json({ success: false, error: 'accountName and code are required' });
|
||
|
||
try {
|
||
const creds = getCredentials();
|
||
const {client_secret, client_id, redirect_uris} = creds.installed || creds.web;
|
||
const oAuth2Client = new google.auth.OAuth2(client_id, client_secret, redirect_uris[0] || 'urn:ietf:wg:oauth:2.0:oob');
|
||
|
||
oAuth2Client.getToken(code, (err, token) => {
|
||
if (err) return res.status(400).json({ success: false, error: err.message });
|
||
|
||
const tokenPath = path.join(TOKENS_DIR, `${accountName}.json`);
|
||
fs.writeFileSync(tokenPath, JSON.stringify(token));
|
||
|
||
// Reload accounts list
|
||
accounts = [];
|
||
loadAuth();
|
||
|
||
res.json({ success: true, message: `Account ${accountName} added successfully!` });
|
||
});
|
||
} catch (e) {
|
||
res.status(500).json({ success: false, error: e.message });
|
||
}
|
||
});
|
||
|
||
const PORT = process.env.PORT || 5002;
|
||
app.listen(PORT, '0.0.0.0', () => {
|
||
console.log(`Express server listening on 0.0.0.0:${PORT}`);
|
||
});
|
||
|
||
// Start Application
|
||
try {
|
||
loadAuth();
|
||
console.log('Gmail Agent started successfully.');
|
||
console.log(`Cron job scheduled to check every 15 minutes.`);
|
||
|
||
// Check immediately on startup
|
||
processEmails();
|
||
|
||
// Schedule every 15 minutes
|
||
cron.schedule('*/15 * * * *', () => {
|
||
processEmails();
|
||
});
|
||
} catch (e) {
|
||
console.error('Failed to start:', e.message);
|
||
}
|