282 lines
8.6 KiB
Python
Executable file
282 lines
8.6 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""
|
|
Parlobyg Email Client (info@parlobyg.dk)
|
|
|
|
Usage:
|
|
python3 email_client.py inbox --limit 10
|
|
python3 email_client.py unread
|
|
python3 email_client.py read --id 123
|
|
python3 email_client.py search --query "from:example@mail.dk"
|
|
python3 email_client.py folders
|
|
python3 email_client.py send --to "to@example.dk" --subject "Subject" --body "Body"
|
|
"""
|
|
|
|
import argparse
|
|
import imaplib
|
|
import smtplib
|
|
import ssl
|
|
import email
|
|
from email.mime.text import MIMEText
|
|
from email.mime.multipart import MIMEMultipart
|
|
from email.header import decode_header
|
|
from datetime import datetime
|
|
import json
|
|
import sys
|
|
|
|
# Configuration
|
|
IMAP_HOST = "imap.one.com"
|
|
IMAP_PORT = 993
|
|
SMTP_HOST = "send.one.com"
|
|
SMTP_PORT = 465
|
|
EMAIL_USER = "info@parlobyg.dk"
|
|
EMAIL_PASS = "Q223Ln57TyyC4"
|
|
|
|
|
|
def decode_mime_header(header):
|
|
"""Decode MIME encoded header."""
|
|
if header is None:
|
|
return ""
|
|
decoded_parts = decode_header(header)
|
|
result = []
|
|
for part, encoding in decoded_parts:
|
|
if isinstance(part, bytes):
|
|
result.append(part.decode(encoding or 'utf-8', errors='replace'))
|
|
else:
|
|
result.append(part)
|
|
return ' '.join(result)
|
|
|
|
|
|
def get_imap_connection():
|
|
"""Create IMAP connection."""
|
|
context = ssl.create_default_context()
|
|
mail = imaplib.IMAP4_SSL(IMAP_HOST, IMAP_PORT, ssl_context=context)
|
|
mail.login(EMAIL_USER, EMAIL_PASS)
|
|
return mail
|
|
|
|
|
|
def list_inbox(limit=10):
|
|
"""List recent inbox messages."""
|
|
mail = get_imap_connection()
|
|
mail.select("INBOX")
|
|
|
|
status, messages = mail.search(None, "ALL")
|
|
if status != "OK":
|
|
return {"error": "Failed to search inbox"}
|
|
|
|
msg_ids = messages[0].split()
|
|
msg_ids = msg_ids[-limit:] if len(msg_ids) > limit else msg_ids
|
|
msg_ids.reverse() # Most recent first
|
|
|
|
results = []
|
|
for msg_id in msg_ids:
|
|
status, msg_data = mail.fetch(msg_id, "(RFC822.HEADER)")
|
|
if status == "OK":
|
|
msg = email.message_from_bytes(msg_data[0][1])
|
|
results.append({
|
|
"id": msg_id.decode(),
|
|
"from": decode_mime_header(msg["From"]),
|
|
"subject": decode_mime_header(msg["Subject"]),
|
|
"date": msg["Date"]
|
|
})
|
|
|
|
mail.logout()
|
|
return {"count": len(results), "messages": results}
|
|
|
|
|
|
def list_unread():
|
|
"""List unread messages."""
|
|
mail = get_imap_connection()
|
|
mail.select("INBOX")
|
|
|
|
status, messages = mail.search(None, "UNSEEN")
|
|
if status != "OK":
|
|
return {"error": "Failed to search unread"}
|
|
|
|
msg_ids = messages[0].split() if messages[0] else []
|
|
|
|
results = []
|
|
for msg_id in msg_ids:
|
|
status, msg_data = mail.fetch(msg_id, "(RFC822.HEADER)")
|
|
if status == "OK":
|
|
msg = email.message_from_bytes(msg_data[0][1])
|
|
results.append({
|
|
"id": msg_id.decode(),
|
|
"from": decode_mime_header(msg["From"]),
|
|
"subject": decode_mime_header(msg["Subject"]),
|
|
"date": msg["Date"]
|
|
})
|
|
|
|
mail.logout()
|
|
return {"unread_count": len(results), "messages": results}
|
|
|
|
|
|
def read_email(msg_id):
|
|
"""Read a specific email by ID."""
|
|
mail = get_imap_connection()
|
|
mail.select("INBOX")
|
|
|
|
status, msg_data = mail.fetch(str(msg_id).encode(), "(RFC822)")
|
|
if status != "OK":
|
|
return {"error": f"Failed to fetch email {msg_id}"}
|
|
|
|
msg = email.message_from_bytes(msg_data[0][1])
|
|
|
|
# Extract body
|
|
body = ""
|
|
if msg.is_multipart():
|
|
for part in msg.walk():
|
|
if part.get_content_type() == "text/plain":
|
|
payload = part.get_payload(decode=True)
|
|
if payload:
|
|
body = payload.decode(part.get_content_charset() or 'utf-8', errors='replace')
|
|
break
|
|
else:
|
|
payload = msg.get_payload(decode=True)
|
|
if payload:
|
|
body = payload.decode(msg.get_content_charset() or 'utf-8', errors='replace')
|
|
|
|
mail.logout()
|
|
return {
|
|
"id": msg_id,
|
|
"from": decode_mime_header(msg["From"]),
|
|
"to": decode_mime_header(msg["To"]),
|
|
"subject": decode_mime_header(msg["Subject"]),
|
|
"date": msg["Date"],
|
|
"body": body[:5000] # Truncate long bodies
|
|
}
|
|
|
|
|
|
def search_emails(query, limit=20):
|
|
"""Search emails with IMAP search criteria."""
|
|
mail = get_imap_connection()
|
|
mail.select("INBOX")
|
|
|
|
# Parse simple query format
|
|
search_criteria = "ALL"
|
|
if query.startswith("from:"):
|
|
search_criteria = f'FROM "{query[5:]}"'
|
|
elif query.startswith("subject:"):
|
|
search_criteria = f'SUBJECT "{query[8:]}"'
|
|
elif query.startswith("to:"):
|
|
search_criteria = f'TO "{query[3:]}"'
|
|
else:
|
|
search_criteria = f'BODY "{query}"'
|
|
|
|
status, messages = mail.search(None, search_criteria)
|
|
if status != "OK":
|
|
return {"error": "Search failed"}
|
|
|
|
msg_ids = messages[0].split() if messages[0] else []
|
|
msg_ids = msg_ids[-limit:] if len(msg_ids) > limit else msg_ids
|
|
msg_ids.reverse()
|
|
|
|
results = []
|
|
for msg_id in msg_ids:
|
|
status, msg_data = mail.fetch(msg_id, "(RFC822.HEADER)")
|
|
if status == "OK":
|
|
msg = email.message_from_bytes(msg_data[0][1])
|
|
results.append({
|
|
"id": msg_id.decode(),
|
|
"from": decode_mime_header(msg["From"]),
|
|
"subject": decode_mime_header(msg["Subject"]),
|
|
"date": msg["Date"]
|
|
})
|
|
|
|
mail.logout()
|
|
return {"query": query, "count": len(results), "messages": results}
|
|
|
|
|
|
def list_folders():
|
|
"""List all folders."""
|
|
mail = get_imap_connection()
|
|
|
|
status, folders = mail.list()
|
|
if status != "OK":
|
|
return {"error": "Failed to list folders"}
|
|
|
|
folder_names = []
|
|
for folder in folders:
|
|
# Parse folder response
|
|
parts = folder.decode().split(' "/" ')
|
|
if len(parts) >= 2:
|
|
folder_names.append(parts[1].strip('"'))
|
|
|
|
mail.logout()
|
|
return {"count": len(folder_names), "folders": folder_names[:100]} # Limit output
|
|
|
|
|
|
def send_email(to, subject, body, cc=None):
|
|
"""Send an email."""
|
|
msg = MIMEMultipart()
|
|
msg["From"] = EMAIL_USER
|
|
msg["To"] = to
|
|
msg["Subject"] = subject
|
|
if cc:
|
|
msg["Cc"] = cc
|
|
|
|
msg.attach(MIMEText(body, "plain", "utf-8"))
|
|
|
|
try:
|
|
context = ssl.create_default_context()
|
|
with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, context=context) as server:
|
|
server.login(EMAIL_USER, EMAIL_PASS)
|
|
recipients = [to] + ([cc] if cc else [])
|
|
server.sendmail(EMAIL_USER, recipients, msg.as_string())
|
|
return {"status": "sent", "to": to, "subject": subject}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Parlobyg Email Client")
|
|
subparsers = parser.add_subparsers(dest="command", help="Available commands")
|
|
|
|
# Inbox
|
|
inbox_parser = subparsers.add_parser("inbox", help="List recent inbox messages")
|
|
inbox_parser.add_argument("--limit", type=int, default=10)
|
|
|
|
# Unread
|
|
subparsers.add_parser("unread", help="List unread messages")
|
|
|
|
# Read
|
|
read_parser = subparsers.add_parser("read", help="Read a specific email")
|
|
read_parser.add_argument("--id", required=True, help="Email ID")
|
|
|
|
# Search
|
|
search_parser = subparsers.add_parser("search", help="Search emails")
|
|
search_parser.add_argument("--query", required=True, help="Search query (from:, subject:, to:, or text)")
|
|
search_parser.add_argument("--limit", type=int, default=20)
|
|
|
|
# Folders
|
|
subparsers.add_parser("folders", help="List folders")
|
|
|
|
# Send
|
|
send_parser = subparsers.add_parser("send", help="Send an email")
|
|
send_parser.add_argument("--to", required=True, help="Recipient")
|
|
send_parser.add_argument("--subject", required=True, help="Subject")
|
|
send_parser.add_argument("--body", required=True, help="Body text")
|
|
send_parser.add_argument("--cc", help="CC recipient")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == "inbox":
|
|
result = list_inbox(args.limit)
|
|
elif args.command == "unread":
|
|
result = list_unread()
|
|
elif args.command == "read":
|
|
result = read_email(args.id)
|
|
elif args.command == "search":
|
|
result = search_emails(args.query, args.limit)
|
|
elif args.command == "folders":
|
|
result = list_folders()
|
|
elif args.command == "send":
|
|
result = send_email(args.to, args.subject, args.body, args.cc)
|
|
else:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|