用微软365邮箱收发邮件【azure-应用注册】

前置条件：

- 有一个365邮箱，配置好许可证

- 在azure portal里有Microsoft Entra ID ，注册相关应用时graph API赋权

- 应用的应用程序(客户端) ID，目录(租户) ID，客户端的密码，邮箱的id，名称

1.收件：

import os
import json
import requests
from .auth2 import get_access_token

def load_token_from_file():
    TOKEN_FILE = 'token_cache.json'
    if os.path.exists(TOKEN_FILE):
        with open(TOKEN_FILE, 'r') as f:
            data = json.load(f)
            return data.get('access_token', None)
    return None

def fetch_unread_emails():
    try:
        access_token = get_access_token()
        if not access_token:
            print('无法加载访问令牌。')
            return

        headers = {
            'Authorization': f'Bearer {access_token}',
            'Content-Type': 'application/json'
        }

        user_id = '-------'
        url = f'https://graph.microsoft.com/v1.0/users/{user_id}/mailfolders/inbox/messages?$filter=isRead eq false'

        response = requests.get(url, headers=headers)
        print(f"Response Status Code: {response.status_code}")
        print(f"Response Content: {response.content}")

        if response.status_code != 200:
            print(f"Error fetching emails: {response.content}")
            return
###代码仅为示例，如想了解更多收发邮件功能，实现赋能gpt的自动收发，处理，function calling，add wx： MTMwMTE4MjY1OTI= （base64）
        emails = response.json().get('value', [])
        for email in emails:
            print(f"Email ID: {email['id']}")
            print(f"From: {email['from']['emailAddress']['address']}")
            print(f"Subject: {email['subject']}")
            print(f"Content: {email['body']['content']}")
            print(f"Received Time: {email['receivedDateTime']}")
            print("-" * 40)

    except Exception as ex:
        print("Error fetching unread emails: ", ex)

# Test fetching unread emails
if __name__ == "__main__":
    fetch_unread_emails()

2.发件：

from flask import Flask, request, jsonify
import requests
import base64
import msal
import os

app = Flask(__name__)

# Configuration variables
CLIENT_ID = 'your_client_id'
CLIENT_SECRET = 'your_client_secret'
TENANT_ID = 'your_tenant_id'
SCOPES = ['https://graph.microsoft.com/.default']
CACHE_FILE = 'token_cache.bin'
URL = 'https://graph.microsoft.com/v1.0/me/sendMail'

def load_cache():
    cache = msal.SerializableTokenCache()
    if os.path.exists(CACHE_FILE):
        cache.deserialize(open(CACHE_FILE, "r").read())
    return cache

def save_cache(cache):
    if cache.has_state_changed:
        with open(CACHE_FILE, "w") as f:
            f.write(cache.serialize())

def get_access_token():
    cache = load_cache()
    app = msal.ConfidentialClientApplication(
        CLIENT_ID,
        authority=f"https://login.microsoftonline.com/{TENANT_ID}",
        client_credential=CLIENT_SECRET,
        token_cache=cache
    )

    result = app.acquire_token_silent(SCOPES, account=None)
    if not result:
        result = app.acquire_token_for_client(scopes=SCOPES)

    if "access_token" in result:
        save_cache(cache)
        return result['access_token']
    else:
        raise ValueError(f"Failed to obtain access token: {result.get('error_description')}")

def create_attachment(file_path, content_id):
    with open(file_path, "rb") as f:
        content_bytes = base64.b64encode(f.read()).decode('utf-8')
    return {
        "@odata.type": "#microsoft.graph.fileAttachment",
        "name": os.path.basename(file_path),
        "contentBytes": content_bytes,
        "contentId": content_id
    }

@app.route('/send-email', methods=['POST'])
def send_email_api():
    data = request.json
    try:
        token = get_access_token()
        sender_email = data['sender']
        recipients_emails = data['recipients']
        subject = data['subject']
        body = data['body']
        attachments = [create_attachment(file_path, content_id) for file_path, content_id in data.get('attachments', [])]

        email_data = {
            "message": {
                "subject": subject,
                "body": {
                    "contentType": "HTML",
                    "content": body
                },
                "toRecipients": [{"emailAddress": {"address": recipient}} for recipient in recipients_emails],
                "from": {"emailAddress": {"address": sender_email}},
                "attachments": attachments
            }
        }

        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

        response = requests.post(URL, headers=headers, json=email_data)

        if response.status_code == 202:
            return jsonify({"message": "Email sent successfully"}), 200
        else:
            return jsonify({"error": f"Failed to send email: {response.status_code}"}), 500
    except Exception as e:
        return jsonify({"error": str(e)}), 500

if __name__ == "__main__":
    app.run(debug=True, port=5000)