微软OAuth2 IMAP收件

八岁鱼

游客，如果您要查看本帖隐藏内容请回复

xxbml666

谢谢分享

u88营销软件下载

感谢楼主的分享,学习一下

萧楚楠

import email as email_reader
import requests
import imaplib
from email.header import decode_header
from bs4 import BeautifulSoup  # 添加HTML解析库

def get_access_token_from_refresh_token(refresh_token, client_id):
headers = {
'Content-Type': 'application/x-www-form-urlencoded;charset=UTF-8',
}
data = {
"client_id": client_id,
"refresh_token": refresh_token,
"grant_type": "refresh_token"
}

try:
    rr = requests.post(
        "https://login.microsoftonline.com/common/oauth2/v2.0/token",
        headers=headers,
        data=data,
        timeout=10
    )
    rr.raise_for_status()

    response_data = rr.json()
    if "error" not in response_data:
        return {
            "code": 0,
            "access_token": response_data["access_token"],
            "refresh_token": response_data.get("refresh_token", refresh_token)
        }
    elif "error_description" in response_data and "User account is found to be in service abuse mode" in \
            response_data["error_description"]:
        return {"code": 1, "message": "账户被封锁或凭证错误"}
    else:
        return {"code": 1, "message": f"获取访问令牌失败: {response_data.get('error_description', '未知错误')}"}

except requests.exceptions.RequestException as e:
    return {"code": 1, "message": f"网络请求错误: {str(e)}"}

def imap_authenticate_with_oauth2(username, access_token):
try:
auth_string = f"user={username}\1auth=Bearer {access_token}\1\1"
mail = imaplib.IMAP4_SSL("outlook.office365.com")
mail.authenticate("XOAUTH2", lambda x: auth_string)
return mail
except Exception as e:
raise Exception(f"IMAP认证失败: {str(e)}")

def decode_str(s):
value, charset = decode_header(s)[0]
if charset:
value = value.decode(charset)
return value

def html_to_text(html_content):
"""将HTML内容转换为纯文本"""
if not html_content:
return ""

soup = BeautifulSoup(html_content, 'html.parser')
# 处理pre标签，保留换行
for pre in soup.find_all('pre'):
    pre.replace_with(f"\n{pre.get_text()}\n")

# 处理br标签，转换为换行
for br in soup.find_all('br'):
    br.replace_with("\n")

# 获取纯文本并处理多余空白
text = soup.get_text()
# 合并多个空白为一个
text = ' '.join(text.split())
# 恢复换行
text = text.replace(' \n ', '\n')
text = text.replace('\n ', '\n')
text = text.replace(' \n', '\n')

return text

def get_latest_email(email, access_token):
mail = imap_authenticate_with_oauth2(email, access_token)

try:
    mail.select("inbox")
    status, messages = mail.search(None, 'ALL')
    if not messages[0]:
        return {"code": 1, "message": "收件箱为空"}

    latest_mail_id = messages[0].split()[-1]
    status, msg_data = mail.fetch(latest_mail_id, '(RFC822)')

    raw_email = msg_data[0][1]
    msg = email_reader.message_from_bytes(raw_email)

    subject = decode_str(msg["Subject"])

    content = ""
    content_type = ""

    if msg.is_multipart():
        # 优先获取text/plain部分
        for part in msg.walk():
            part_content_type = part.get_content_type()
            content_disposition = str(part.get("Content-Disposition"))

            try:
                part_body = part.get_payload(decode=True)
                if not part_body:
                    continue

                # 尝试解码内容
                charset = part.get_content_charset() or 'utf-8'
                try:
                    decoded_body = part_body.decode(charset, errors='replace')
                except LookupError:
                    decoded_body = part_body.decode('utf-8', errors='replace')

                if part_content_type == "text/plain" and "attachment" not in content_disposition:
                    content = decoded_body
                    content_type = "text/plain"
                    break
                elif part_content_type == "text/html" and "attachment" not in content_disposition and not content:
                    content = decoded_body
                    content_type = "text/html"
            except Exception as e:
                continue
    else:
        body = msg.get_payload(decode=True)
        if body:
            charset = msg.get_content_charset() or 'utf-8'
            try:
                content = body.decode(charset, errors='replace')
            except LookupError:
                content = body.decode('utf-8', errors='replace')
            content_type = msg.get_content_type()

    # 如果是HTML内容，转换为纯文本
    if content_type == "text/html":
        content = html_to_text(content)

    return {
        "code": 0,
        "subject": subject,
        "content": content
    }
except Exception as e:
    return {"code": 1, "message": f"读取邮件失败: {str(e)}"}
finally:
    mail.close()
    mail.logout()

def example():
email = ''
client_id = 'your_client_id'
refresh_token = "your_refresh_token"

token_result = get_access_token_from_refresh_token(refresh_token, client_id)
if token_result["code"] != 0:
    print(f"获取令牌失败: {token_result['message']}")
    return

email_result = get_latest_email(email, token_result["access_token"])
if email_result["code"] == 0:
    print(f"主题: {email_result['subject']}")
    print(f"内容: {email_result['content'][:2000]}...")
else:
    print(f"获取邮件失败: {email_result['message']}")

if name == 'main':
example()

萧楚楠

感谢分享，很给力！~

757674712

牛逼牛逼牛逼牛逼牛逼牛逼

通大小不点

        感谢分享，很给力！~

miwang8812

太强了吧必须需要

一个人的故事

谢谢分享！！！

aikang

感谢分享，很给力！~

HuangQAQ

1111111111111