Teams集成-订阅事件处理

在Teams会议侧边栏应用开发-会议转写-CSDN博客的基础上，使用/delta接口尝试获取实时转写，发现只能更新了一次，然后就不再更新了，想尝试使用订阅事件去获取转写，发现也不是实时的，当会议结束时，订阅事件才会产生，且有一定的延迟。本文主要针对订阅事件如何创建及事件接收后如何处理进行描述。

转写免费配额（很容易使用完，可以重新注册一个新的应用再次获取免费配额）：

转写计费成本：

订阅事件的创建：

// 定义 /subscribe 端点
server.post('/subscribe', async (req, res) => {try {const onlineMeetingId = req.query.meetingId;console.log('/subscribe meetingId:',onlineMeetingId);const response = await fetch("https://graph.microsoft.com/beta/subscriptions", {method: "GET",headers: {"Content-Type": "application/json",Authorization: `Bearer ${app_token}`,}});if (!response.ok) {const errorData = await response.json();throw new Error(errorData.error.message);}const data = await response.json();let noSub = true;data.value.forEach(sub=>{if (sub.resource === `communications/onlineMeetings/${onlineMeetingId}/transcripts`){noSub = false;}})if (noSub){const subscription = {changeType: "created",notificationUrl: "https://shortly-adapted-akita.ngrok-free.app/webhook",resource: `communications/onlineMeetings/${onlineMeetingId}/transcripts`,expirationDateTime: new Date(Date.now() +  60 * 60 * 1000).toISOString(), //1小时clientState: "secretClientValue",};const response2 = await fetch("https://graph.microsoft.com/beta/subscriptions", {method: "POST",headers: {"Content-Type": "application/json",Authorization: `Bearer ${app_token}`,},body: JSON.stringify(subscription),});if (!response2.ok) {const errorData = await response2.json();throw new Error(errorData.error.message);}const data2 = await response2.json();console.log("Subscription created:", data2.id);res.send(200, data2);}} catch (error) {console.error("Error creating subscription:", error);res.send(500, { error: error.message });}
});

订阅事件的接收（订阅事件没有使用加密方式，需要调用2次API获取转写）：

// 定义 /webhook 端点
server.post('/webhook', async (req, res) => {const validationToken = req.query.validationToken;console.log('/webhook validationToken:', validationToken);// 如果是验证请求，返回 validationTokenif (validationToken) {res.setHeader('Content-Type', 'text/plain');res.send(200, validationToken);return;}const notification = req.body.value[0];const odataId = notification.resourceData['@odata.id'];const url = `https://graph.microsoft.com/beta/${odataId}`.replace('communications','me');//console.log('url:', url);const response = await fetch('https://shortly-adapted-akita.ngrok-free.app/get_user_token');const token_data = await response.json();// 获取具体的转写数据try {const response = await fetch(url, {headers: {Authorization: `Bearer ${token_data.user_token}`,},});if (!response.ok) {const errorData = await response.json();//console.log(errorData);throw new Error(errorData.error.message);}const transcriptData = await response.json();//console.log("Transcript Data:", transcriptData);// 获取转写内容const transcriptContentUrl = transcriptData.transcriptContentUrl;const contentResponse = await fetch(`${transcriptContentUrl}?$format=text/vtt`, {headers: {Authorization: `Bearer ${token_data.user_token}`,},});if (!contentResponse.ok) {const errorData = await contentResponse.text();//console.log(errorData);throw new Error(errorData);}const transcriptContent = await contentResponse.text();console.log("Transcript Content:", transcriptContent);// 将转写数据发送到所有连接的 WebSocket 客户端wss.clients.forEach((client) => {if (client.readyState === WebSocket.OPEN) {client.send(JSON.stringify({ type: 'newTranscript', data: transcriptContent }));}});} catch (error) {console.error("Error fetching transcript data:", error);}res.send(200, "OK");
});

完整的服务端代码如下：

import restify from "restify";
import send from "send";
import fs from "fs";
import fetch from "node-fetch";
import path from 'path';
import { fileURLToPath } from 'url';
import { storeToken, getToken } from './redisClient.js';
import { WebSocketServer, WebSocket } from 'ws';const __filename = fileURLToPath(import.meta.url);
console.log('__filename: ', __filename);const __dirname = path.dirname(__filename);
console.log('__dirname: ', __dirname);// Create HTTP server.
const server = restify.createServer({key: process.env.SSL_KEY_FILE ? fs.readFileSync(process.env.SSL_KEY_FILE) : undefined,certificate: process.env.SSL_CRT_FILE ? fs.readFileSync(process.env.SSL_CRT_FILE) : undefined,formatters: {"text/html": function (req, res, body) {return body;},},
});server.use(restify.plugins.bodyParser());
server.use(restify.plugins.queryParser());server.get("/static/*",restify.plugins.serveStatic({directory: __dirname,})
);server.listen(process.env.port || process.env.PORT || 3000, function () {console.log(`\n${server.name} listening to ${server.url}`);
});// Adding tabs to our app. This will setup routes to various views
// Setup home page
server.get("/config", (req, res, next) => {send(req, __dirname + "/config/config.html").pipe(res);
});// Setup the static tab
server.get("/meetingTab", (req, res, next) => {send(req, __dirname + "/panel/panel.html").pipe(res);
});//获得用户token
server.get('/auth', (req, res, next) => {res.status(200);res.send(`
<!DOCTYPE html>
<html>
<head><script>// Function to handle the token storageasync function handleToken() {const hash = window.location.hash.substring(1);const hashParams = new URLSearchParams(hash);const access_token = hashParams.get('access_token');console.log('Received hash parameters:', hashParams);if (access_token) {console.log('Access token found:', access_token);localStorage.setItem("access_token", access_token);console.log('Access token stored in localStorage');try {const response = await fetch('https://shortly-adapted-akita.ngrok-free.app/store_user_token', {method: 'POST',headers: {'Content-Type': 'application/json'},body: JSON.stringify({ "user_token" : access_token })});if (response.ok) {console.log('Token stored successfully');} else {console.error('Failed to store token:', response.statusText);}} catch (error) {console.error('Error storing token:', error);}} else {console.log('No access token found');}window.close();}// Call the function to handle the tokenhandleToken();</script>
</head>
<body></body>
</html>`);next();
});// 存储 user_token
server.post('/store_user_token', async (req, res) => {const user_token = req.body.user_token;if (!user_token) {res.status(400);res.send('user_token are required');}try {// Store user tokenawait storeToken('user_token', user_token);console.log('user_token stored in Redis');} catch (err) {console.error('user_token store Error:', err);}res.status(200);   res.send('Token stored successfully');
});// 获取 user_token
server.get('/get_user_token', async (req, res) => {try {// Store user tokenconst user_token = await getToken('user_token');console.log('user_token get in Redis');res.send({"user_token": user_token});} catch (err) {console.error('user_token get Error:', err);}
});//应用token
let app_token = '';
let app_token_expires_at = 0;
const app_token_refresh_interval = 3000 * 1000; // 3000秒const getAppToken = async () => {try {// 构建请求体const requestBody = new URLSearchParams({"grant_type": "client_credentials","client_id": "Azure注册应用ID","client_secret": "Azure注册应用密码","scope": "https://graph.microsoft.com/.default",}).toString();// 获取app令牌const tokenUrl = `https://login.microsoftonline.com/注册应用租户ID/oauth2/v2.0/token`;const tokenResponse = await fetch(tokenUrl, {method: 'POST',headers: {'Content-Type': 'application/x-www-form-urlencoded',},body: requestBody,});if (!tokenResponse.ok) {const errorData = await tokenResponse.json();throw new Error(errorData.error_description);}const tokenData = await tokenResponse.json();app_token = tokenData.access_token;app_token_expires_at = Date.now() + app_token_refresh_interval;console.log("app_token received!");} catch (error) {console.error('Error getting app token:', error);}
};// 定期刷新 app_token
setInterval(getAppToken, app_token_refresh_interval);// 定义 /getTranscripts 端点
server.get('/getTranscripts', async (req, res) => {try {const url = req.query.url;if (!url) {res.send(400, { error: 'URL is required' });return;}// 调用 Microsoft Graph APIconst graphResponse = await fetch(url, {headers: {Authorization: `Bearer ${app_token}`,},});if (!graphResponse.ok) {const errorData = await graphResponse.json();res.send(500, { error: errorData.error.message });return;}const data = await graphResponse.json();const currentTime = new Date().toISOString(); // 获取当前时间console.log(currentTime, ', getTranscripts length:',data.value.length);// 返回转录文本res.send(200, data);} catch (error) {// 返回错误res.send(500, { error: error.message });}
});// 定义 /getTranscriptContent 端点
let callCount = 0; // 调用计数器server.get('/getTranscriptContent', async (req, res) => {try {const response = await fetch('https://shortly-adapted-akita.ngrok-free.app/get_user_token');const token_data = await response.json();const transcriptContentUrl = req.query.transcriptContentUrl;if (!transcriptContentUrl) {res.send(400, { error: 'transcriptContentUrl is required' });return;}const content_url = `${transcriptContentUrl}?$format=text/vtt`;// 调用 Microsoft Graph APIconst graphResponse = await fetch(content_url, {headers: {Authorization: `Bearer ${token_data.user_token}`,},});if (!graphResponse.ok) {const errorData = await graphResponse.text();res.send(500, { error: errorData });return;}const data = await graphResponse.text();callCount++; // 增加正确调用计数const currentTime = new Date().toISOString(); // 获取当前时间console.log(`getTranscriptContent called at ${currentTime}, call count: ${callCount}`); // 输出日志console.log('content:', data)// 返回转录文本res.send(200, data);} catch (error) {// 返回错误res.send(500, { error: error.message });}
});// 确保在服务器启动时获取 app_token
getAppToken();// 初始化 WebSocket 服务器
const wss = new WebSocketServer({ server: server.server });wss.on('connection', (ws) => {console.log('A client connected');ws.on('message', (message) => {console.log(`Received message: ${message}`);});ws.on('close', () => {console.log('A client disconnected');});
});// 定义 /subscribe 端点
server.post('/subscribe', async (req, res) => {try {const onlineMeetingId = req.query.meetingId;console.log('/subscribe meetingId:',onlineMeetingId);const response = await fetch("https://graph.microsoft.com/beta/subscriptions", {method: "GET",headers: {"Content-Type": "application/json",Authorization: `Bearer ${app_token}`,}});if (!response.ok) {const errorData = await response.json();throw new Error(errorData.error.message);}const data = await response.json();let noSub = true;data.value.forEach(sub=>{if (sub.resource === `communications/onlineMeetings/${onlineMeetingId}/transcripts`){noSub = false;}})if (noSub){const subscription = {changeType: "created",notificationUrl: "https://shortly-adapted-akita.ngrok-free.app/webhook",resource: `communications/onlineMeetings/${onlineMeetingId}/transcripts`,expirationDateTime: new Date(Date.now() +  60 * 60 * 1000).toISOString(), //1小时clientState: "secretClientValue",};const response2 = await fetch("https://graph.microsoft.com/beta/subscriptions", {method: "POST",headers: {"Content-Type": "application/json",Authorization: `Bearer ${app_token}`,},body: JSON.stringify(subscription),});if (!response2.ok) {const errorData = await response2.json();throw new Error(errorData.error.message);}const data2 = await response2.json();console.log("Subscription created:", data2.id);res.send(200, data2);}} catch (error) {console.error("Error creating subscription:", error);res.send(500, { error: error.message });}
});// 定义 /webhook 端点
server.post('/webhook', async (req, res) => {const validationToken = req.query.validationToken;console.log('/webhook validationToken:', validationToken);// 如果是验证请求，返回 validationTokenif (validationToken) {res.setHeader('Content-Type', 'text/plain');res.send(200, validationToken);return;}const notification = req.body.value[0];const odataId = notification.resourceData['@odata.id'];const url = `https://graph.microsoft.com/beta/${odataId}`.replace('communications','me');//console.log('url:', url);const response = await fetch('https://shortly-adapted-akita.ngrok-free.app/get_user_token');const token_data = await response.json();// 获取具体的转写数据try {const response = await fetch(url, {headers: {Authorization: `Bearer ${token_data.user_token}`,},});if (!response.ok) {const errorData = await response.json();//console.log(errorData);throw new Error(errorData.error.message);}const transcriptData = await response.json();//console.log("Transcript Data:", transcriptData);// 获取转写内容const transcriptContentUrl = transcriptData.transcriptContentUrl;const contentResponse = await fetch(`${transcriptContentUrl}?$format=text/vtt`, {headers: {Authorization: `Bearer ${token_data.user_token}`,},});if (!contentResponse.ok) {const errorData = await contentResponse.text();//console.log(errorData);throw new Error(errorData);}const transcriptContent = await contentResponse.text();console.log("Transcript Content:", transcriptContent);// 将转写数据发送到所有连接的 WebSocket 客户端wss.clients.forEach((client) => {if (client.readyState === WebSocket.OPEN) {client.send(JSON.stringify({ type: 'newTranscript', data: transcriptContent }));}});} catch (error) {console.error("Error fetching transcript data:", error);}res.send(200, "OK");
});

完整的页面代码如下：

<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>Meeting Transcripts</title><script src="https://res.cdn.office.net/teams-js/2.0.0/js/MicrosoftTeams.min.js"></script><script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script><style>.subtitle {display: flex;align-items: center;margin-bottom: 10px;}.speaker-photo {width: 20px;height: 20px;border-radius: 50%;margin-right: 10px;}</style>
</head>
<body><h2>Meeting Transcripts</h2><div id="transcripts"></div><script>const clientId = 'Azure注册应用ID';const tenantId = 'Azure注册应用租户ID';const authUrl = `https://login.microsoftonline.com/${tenantId}/oauth2/v2.0/authorize`;const redirectUri = '你的https域名/auth'; // 确保与服务器端一致const scope = 'user.read';let user_token = null;let meetingOrganizerUserId = null;let participants = {}; // 用于存储参会者的信息let nextLink = null;let deltaLink = null;let userPhotoCache = {}; // 用于缓存用户头像let tokenFetched = false; // 标志变量，用于跟踪是否已经获取了 user_tokenconst getUserInfo = async (userId, accessToken) => {const graphUrl = `https://graph.microsoft.com/v1.0/users/${userId}`;const response = await fetch(graphUrl, {headers: {'Authorization': `Bearer ${accessToken}`}});if (response.status === 401) {// 如果 token 超期，重新触发 initAuthenticationinitAuthentication();return null;}const userInfo = await response.json();return userInfo;};const getUserPhoto = async (userId, accessToken) => {if (userPhotoCache[userId]) {return userPhotoCache[userId];}const graphUrl = `https://graph.microsoft.com/v1.0/users/${userId}/photo/$value`;const response = await fetch(graphUrl, {headers: {'Authorization': `Bearer ${accessToken}`}});if (!response.ok) {const errorData = await response.json();console.error('Error fetching user photo:', errorData);return null;}const photoBlob = await response.blob();const photoUrl = URL.createObjectURL(photoBlob);userPhotoCache[userId] = photoUrl; // 缓存头像 URLreturn photoUrl;};const getMeetingDetails = async (user_token, joinMeetingId) => {const apiUrl = `https://graph.microsoft.com/v1.0/me/onlineMeetings?$filter=joinMeetingIdSettings/joinMeetingId eq '${joinMeetingId}'`;const response = await fetch(apiUrl, {method: 'GET',headers: {'Authorization': `Bearer ${user_token}`,'Content-Type': 'application/json'}});if (!response.ok) {const errorData = await response.json();throw new Error(`HTTP error! status: ${response.status}, message: ${errorData.error}`);}const data = await response.json();return data.value[0];};const getTranscriptContent = async (content) => {const lines = content.trim().split('\n');const subtitles = [];let currentSpeaker = null;for (let i = 0; i < lines.length; i++) {const line = lines[i].trim();if (line.includes('-->')) {const [startTime, endTime] = line.split(' --> ');const text = lines[i + 1].trim();const speakerMatch = text.match(/<v\s*([^>]+)>/);const speaker = speakerMatch ? speakerMatch[1] : null;const content = text.replace(/<v\s*[^>]*>/, '').replace(/<\/v>/, '');if (speaker && speaker !== currentSpeaker) {currentSpeaker = speaker;}subtitles.push({ startTime, endTime, speaker: currentSpeaker, content });i++; // Skip the next line as it's the text content}}return subtitles;};const displaySubtitle = async (subtitle, transcriptElement, accessToken) => {const subtitleElement = document.createElement('div');subtitleElement.classList.add('subtitle');// 获取说话者的头像const speakerUserId = participants[subtitle.speaker];const speakerPhotoUrl = speakerUserId ? await getUserPhoto(speakerUserId, accessToken) : 'default-avatar.png';// 创建头像元素const speakerPhotoElement = document.createElement('img');speakerPhotoElement.src = speakerPhotoUrl;speakerPhotoElement.alt = subtitle.speaker;speakerPhotoElement.classList.add('speaker-photo');// 创建输出字符串const output = `${subtitle.startTime}\n${subtitle.content}`;subtitleElement.appendChild(speakerPhotoElement);subtitleElement.appendChild(document.createTextNode(output));transcriptElement.appendChild(subtitleElement);};const subscribeToWebhook = async (meetingId) => {try {const subscribeResponse = await fetch(`https://shortly-adapted-akita.ngrok-free.app/subscribe?meetingId=${meetingId}`, {method: 'POST'});if (!subscribeResponse.ok) {const errorData = await subscribeResponse.json();console.error('Webhook subscription failed:', errorData);} else {console.log('Webhook subscription successful');}} catch (error) {console.error('Error subscribing to webhook:', error);}};const init = async () => {try {if (!tokenFetched) {const response = await fetch('https://shortly-adapted-akita.ngrok-free.app/get_user_token');const data = await response.json();if (response.ok) {user_token = data.user_token;console.log('user token retrieved:', user_token);tokenFetched = true;} else {console.error('Failed to get token:', response.statusText);return;}}const joinMeetingId = '45756456529'; // 替换为你要查询的 joinMeetingIdtry {const meetingDetails = await getMeetingDetails(user_token, joinMeetingId);console.log('Meeting Details:', meetingDetails);meetingOrganizerUserId = meetingDetails.participants.organizer.identity.user.id;const meetingId = meetingDetails.id; // 获取会议 IDconsole.log('Organizer User ID:', meetingOrganizerUserId);console.log('Meeting ID:', meetingId);// 获取主持人信息const organizerInfo = await getUserInfo(meetingOrganizerUserId, user_token);const organizerDisplayName = organizerInfo.displayName;participants[organizerDisplayName] = meetingOrganizerUserId;// 获取参会者信息const attendeesPromises = meetingDetails.participants.attendees.map(async attendee => {const userId = attendee.identity.user.id;const userInfo = await getUserInfo(userId, user_token);const displayName = userInfo.displayName;participants[displayName] = userId;});await Promise.all(attendeesPromises);// 初始化历史转写// await fetchTranscripts();// 订阅 Webhookawait subscribeToWebhook(meetingId);// 设置每小时调用一次订阅 WebhooksetInterval(() => subscribeToWebhook(meetingId), 3600000); // 每3600秒（1小时）调用一次} catch (error) {console.error('Error fetching meeting details:', error);}console.log('User Token:', user_token);} catch (error) {console.error('Error getting token:', error);}};const initAuthentication = () => {microsoftTeams.app.initialize();microsoftTeams.authentication.authenticate({url: `${authUrl}?client_id=${clientId}&response_type=token&redirect_uri=${encodeURIComponent(redirectUri)}&scope=${encodeURIComponent(scope)}`,width: 600,height: 535,successCallback: async (result) => {console.log('Authentication success:', result);},failureCallback: (error) => {console.error('Authentication failed:', error);}});};// 设置较长的轮询时间来防止 user_token 的超期setInterval(initAuthentication, 3000000); // 每3000秒（50分钟）轮询一次initAuthentication();init();// 初始化 WebSocket 客户端const socket = new WebSocket('wss://shortly-adapted-akita.ngrok-free.app');socket.onopen = () => {console.log('WebSocket connection established');};socket.onmessage = (event) => {const message = JSON.parse(event.data);console.log(message);if (message.type === 'newTranscript') {const transcriptData = message.data;console.log('New Transcript Data:', transcriptData);const transcriptsContainer = document.getElementById('transcripts');const transcriptElement = document.createDocumentFragment(); // 使用 DocumentFragment 优化 DOM 操作getTranscriptContent(transcriptData).then(subtitles => {subtitles.forEach(subtitle => {displaySubtitle(subtitle, transcriptElement, user_token);});}).catch(error => {const errorElement = document.createElement('div');errorElement.innerHTML = `<strong>${error}</strong>`;transcriptElement.appendChild(errorElement);}).finally(() => {transcriptsContainer.appendChild(transcriptElement); // 一次性插入 DOM});}};socket.onclose = () => {console.log('WebSocket connection closed');};socket.onerror = (error) => {console.error('WebSocket error:', error);};</script>
</body>
</html>

输出效果：