基于Microsoft Graph实现Linux Teams PWA未读消息提醒的技术求助

问题背景 #

我在Linux系统上运行Microsoft Teams渐进式Web应用（PWA），该应用无系统托盘图标，消息通知仅显示2秒，离开屏幕时无法获知新消息。尝试通过Python结合Microsoft Graph API实现未读消息检测，但现有脚本存在两个问题：

• 无法获取消息的lastReadDateTime字段值；
• 每次启动脚本都需要重新验证登录。

我拥有O365管理员权限，可注册应用，也接受其他编程语言的解决方案。

问题1：获取lastReadDateTime字段的解决方法 #

原脚本无法获取lastReadDateTime是因为Graph API请求me/chats时默认不返回viewpoint字段，需要显式通过$select参数指定返回该字段，同时确保权限配置正确。

具体调整： #

1. 修改API请求参数：在获取聊天列表的请求中添加$select=id,chatType,viewpoint，强制返回包含lastMessageReadDateTime的viewpoint对象；
2. 确认权限配置：确保应用注册时已添加Chat.Read权限并完成管理员同意。

问题2：避免每次启动重新登录的解决方法 #

原脚本使用交互式登录且未持久化token，可通过MSAL的token缓存功能，将登录状态保存到本地文件，下次启动自动读取缓存并刷新token。

具体调整： #

1. 使用序列化token缓存：创建SerializableTokenCache并将缓存写入本地文件（如token_cache.bin）；
2. 优先从缓存获取token：启动时先尝试从缓存读取有效token，仅在缓存失效时触发交互式登录。

修改后的完整Python代码 #

import msal
import requests
import time
from datetime import datetime, timezone
import pickle

# ---------------- 配置项 ----------------
CLIENT_ID = "你的客户端ID"
TENANT_ID = "你的租户ID"
AUTHORITY = f"https://login.microsoftonline.com/{TENANT_ID}"
SCOPES = ["Chat.Read"]  # 读取聊天所需权限
POLL_INTERVAL = 10      # 检测间隔（秒）
TOKEN_CACHE_FILE = "token_cache.bin"  # token缓存文件路径
# ----------------------------------------

def load_token_cache():
    """加载本地token缓存"""
    try:
        with open(TOKEN_CACHE_FILE, "rb") as f:
            cache = msal.SerializableTokenCache()
            cache.deserialize(pickle.load(f))
            return cache
    except FileNotFoundError:
        return msal.SerializableTokenCache()

def save_token_cache(cache):
    """保存token缓存到本地文件"""
    with open(TOKEN_CACHE_FILE, "wb") as f:
        pickle.dump(cache.serialize(), f)

def get_access_token():
    """获取access token，优先使用缓存，失效则触发交互式登录"""
    cache = load_token_cache()
    app = msal.PublicClientApplication(
        CLIENT_ID,
        authority=AUTHORITY,
        token_cache=cache
    )

    # 尝试从缓存获取有效token
    accounts = app.get_accounts()
    if accounts:
        result = app.acquire_token_silent(SCOPES, account=accounts[0])
        if result.get("access_token"):
            save_token_cache(cache)
            return result["access_token"]
    
    # 缓存失效，触发交互式登录
    result = app.acquire_token_interactive(scopes=SCOPES)
    if "access_token" not in result:
        raise Exception(f"登录失败：{result.get('error_description')}")
    save_token_cache(cache)
    return result["access_token"]

def list_chats(token):
    """获取用户聊天列表，包含viewpoint字段"""
    headers = {"Authorization": f"Bearer {token}"}
    # 添加$select参数指定返回viewpoint字段
    url = "https://graph.microsoft.com/v1.0/me/chats?$select=id,chatType,viewpoint"
    resp = requests.get(url, headers=headers)
    resp.raise_for_status()
    return resp.json().get("value", [])

def get_latest_messages(token, chat_id):
    """获取指定聊天的最新消息"""
    headers = {"Authorization": f"Bearer {token}"}
    # 添加$top=1只获取最新一条消息，提升效率
    url = f"https://graph.microsoft.com/v1.0/chats/{chat_id}/messages?$top=1"
    resp = requests.get(url, headers=headers)
    resp.raise_for_status()
    return resp.json().get("value", [])

def check_unread_messages(token, chat_id):
    """检查指定聊天是否有未读消息"""
    # 获取聊天详情（包含viewpoint）
    headers = {"Authorization": f"Bearer {token}"}
    url = f"https://graph.microsoft.com/v1.0/me/chats/{chat_id}?$select=viewpoint,lastUpdatedDateTime"
    resp = requests.get(url, headers=headers)
    resp.raise_for_status()
    chat = resp.json()

    # 获取最新消息
    messages = get_latest_messages(token, chat_id)
    if not messages:
        return None, None
    
    latest_msg = messages[0]
    msg_time = datetime.fromisoformat(latest_msg["createdDateTime"].replace("Z", "+00:00"))
    last_read_time = chat["viewpoint"].get("lastMessageReadDateTime")
    
    if last_read_time:
        last_read_time = datetime.fromisoformat(last_read_time.replace("Z", "+00:00"))
        # 判断消息是否未读
        if msg_time > last_read_time:
            return latest_msg, True
    else:
        # 从未读过该聊天，所有消息都是未读
        return latest_msg, True
    
    return latest_msg, False

if __name__ == "__main__":
    try:
        print("初始化登录...")
        token = get_access_token()

        print("\n获取聊天列表...")
        chats = list_chats(token)
        if not chats:
            print("未找到任何聊天。")
            exit()

        # 显示聊天列表及最后读取时间
        for i, chat in enumerate(chats):
            chat_type = chat["chatType"]
            chat_id = chat["id"]
            last_read = chat["viewpoint"].get("lastMessageReadDateTime", "从未读取")
            print(f"{i+1}. {chat_type} - ID: {chat_id}")
            print(f"  最后读取时间: {last_read}\n")

        choix = int(input("请输入要监控的聊天编号：")) - 1
        chat_id = chats[choix]["id"]

        print(f"\n开始监控聊天 {chat_id}...")

        while True:
            latest_msg, is_unread = check_unread_messages(token, chat_id)
            if latest_msg and is_unread:
                sender = latest_msg["from"]["user"]["displayName"]
                content = latest_msg["body"]["content"]
                print(f"\n【未读消息】来自 {sender}：")
                print(content)
            else:
                print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - 无新未读消息")
            
            time.sleep(POLL_INTERVAL)

    except Exception as e:
        print("错误：", e)

内容的提问来源于stack exchange，提问作者user34224