如何使用 python 向 ms 团队中的特定用户发送直接消息？

Question

我想通过 python 和 Odoo 向 ms 团队中的特定用户发送直接消息？

Answer 1

首先 ：

如果需要，在您的团队中创建一个新频道。新频道将阻止 Python 生成的通知接管其他对话频道。 img1

在所需频道上添加新连接器。 img 2

找到 Webhook 连接器并进行配置。 img 3

所需的配置只是 webhook 的名称和可选的图像。 img4

img 5 单击创建并复制生成的 webhook URL。 img6

将此代码添加到您的 Python 项目中，以便它可以向 Teams 写入消息。

Install pymsteams with pip.

pip install pymsteams

将此代码添加到您的 Python 项目以启用向 Teams 写入消息，将 URL 替换为您的 webhook：

import pymsteams
myTeamsMessage = pymsteams.connectorcard("INSERT WEBHOOK URL HERE")
Use this code to generate messages:

myTeamsMessage.text("This message was generated from Python!")
myTeamsMessage.send()

output

Answer 2

这是一个稍微复杂一点的任务，但绝对可行。使用 python 发送单个消息并不像使用 Webhook 在 Teams 中发送消息那么容易。您必须通过身份验证，获取您的令牌，然后发送聊天。以下是我使用 selenium 获取身份验证然后使用 requests.post 发送个人按摩的步骤。

顺便说一句，如果您认为通过 Microsoft 进行身份验证很容易，那么您是 100% 错误的，这里是从 microsofthttps://learn.microsoft.com/en-us/graph/auth-v2-user 获取访问令牌的链接

在 Azure 中注册您的应用程序后，您可以使用以下代码使用 selenium 获取身份验证令牌。

设置硒

import gc
gc.disable() # 03/23/2022 added this one to prevent trash collection and avoide crashing the notebooks
from selenium import webdriver
from selenium.webdriver.firefox.options import Options
from selenium.webdriver.firefox.service import Service
from selenium.common.exceptions import WebDriverException # to catch WebDriverException after changing the host from postman to local

Options.binary_location = "/usr/bin/firefox"
ser = Service("/tmp/geckodriver")
options = Options()
options.binary_location = "location for firefix bynary .. /firefox_binary.py"
options.headless = True
driver = webdriver.Firefox(options=options, service=ser)

使用以下代码进行身份验证。 （您需要一些 selenium 技能来编写手动过程的代码），我将分享起点作为示例，您可以编写自己的代码，因为它与我使用的链接不同。

import time
import re
import json
import requests
import pandas as pd
Options.binary_location = "/usr/bin/firefox"
ser = Service("/tmp/geckodriver")
options = Options()
options.headless = True
driver = webdriver.Firefox(options=options, service=ser)


tenant_id = 'your tenant id for Microsoft graph'
client_id = "your client id for Microsoft graph"
url = f"http://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/authorize?client_id={client_id}&response_type=code&response_mode=query&scope=user.read%20chat.read&state=12345"
driver.get(url)
print(driver.current_url)
#open_tab
time.sleep(3)

这是您将使用您的 selenium 技能的部分，我只是提出了 Microsoft 要求的第一步，根据您/您的公司设置，您可能需要更多步骤才能登录。

element_id = webdriver.common.by.By.ID
email_locator=(element_id,"i0116")
driver.find_element(*email_locator).send_keys("your email address")
next_button_locator = (element_id,"idSIButton9")
driver.find_element(*next_button_locator).click()
time.sleep(9)
print(driver.current_url)

我建议在每个步骤之后打印 driver.current_url，以便您可以手动检查链接，然后相应地对其进行编码。

登录后，现在您可以使用以下代码获取您的身份验证令牌。我使用 localhost:5000 作为返回 URL，但您可以在应用注册页面中设置任何您想要的 URL。

此身份验证令牌仅在一小时内有效，因此我使用以下 while 循环每 30 分钟刷新一次令牌

while True:
  try:
    driver.get(url)
  except WebDriverException:
        time.sleep(3)
   
  url_pattern = 'http://localhost:5000/?code=(?P<code>[^&]*)&state=12345.*'
  re_match = re.match(url_pattern,driver.current_url)
  code = re_match.group('code')
  
  headers = { 'Content-Type': 'application/x-www-form-urlencoded'}
  body = {'grant_type': 'authorization_code',
         'code': code,
         'redirect_url':'your redirect url',
         'client_id': 'your client id',
         'client_secret':'your client secret'}
  
  response = requests.post('https://login.microsoftonline.com/yor tenant id/oauth2/v2.0/token',headers=headers,data = body)

  access_token = json.loads(response.text)['access_token']
 
  pd_response = pd.DataFrame([json.loads(response.text)])
  # I am saving the new token in spark dataframe where I can read it with other code, but you can save it different way
  sf_response = spark.createDataFrame(pd_response)
  sf_response.write.mode("overwrite").saveAsTable('graphTest')
  time.sleep(30*60)

现在，如果您克服了微软为验证您自己而设置的所有障碍，您可以使用以下行发布消息

#used this function to get updated token, but you can pass your token anyway that you want.

def getGraphAccessToken():
  """
  Gets access token that is saved in graphTest table
  """
  dataFrame = spark.sql("select * from graphTest")
  return dataFrame.collect()[0]['access_token']

最后，只要您有他们的聊天 ID，您就可以使用以下代码将消息发送给任何人或一群人。

def sendIndividualChat(text,chatIdList,contentType="html"):
  """
  sends individual chat through Microsoft Teams.
  Parameters:
  ----------
  text : str, message content
  chatIdList : list, list of chat id(id is in string format) for individual chat messages
  """
 
  headers = {'Content-type':'application/json',
            "Authorization": f"Bearer {getGraphAccessToken()}"
        }

  body = {
       "body": {
         "contentType": contentType,
         "content": text
  }}

  for chatId in chatIdList:
    requests.post(f"https://graph.microsoft.com/v1.0/chats/{chatId}/messages",headers=headers,data =json.dumps(body) )

如果您不知道如何获取个人或群聊的聊天 ID，有几种方法，您可以使用 Graph Explorer https://developer.microsoft.com/en-us/graph/graph-explorer 或以下代码获取您最近聊天的列表及其相应的聊天 ID。确保向要获取聊天 ID 的人/组发送消息，以便它显示在最近的聊天消息中。

def getCurrentChats():
  """
  Gets list of current chats for current user.
  Returns: 
  --------
  Pandas DataFrame with information about the current chats
  """
  headers = {'Content-type':'application/json',
            "Authorization": f"Bearer {getGraphAccessToken()}"}
  connection = http.client.HTTPSConnection("graph.microsoft.com")
  connection.request("GET","/v1.0/me/chats",headers=headers)
  resp = connection.getresponse()
  text = resp.read()
  return pd.DataFrame(json.loads(text)['value'])

def getCurrentChatMembers(currentChatsDataFrame,idColumn = 'id',debug=False):
  """
  Get dictionary of member emails for each chat id
  Parameters:
  ----------
  currentChatsDataFrame : Pandas DataFrame returned from getCurrentChats()
  idColumn : str, name of column with chat id (default= id)
  Returns:
  --------
  Pandas DataFrame with ['memberName','memberEmail','chatId'] columns
  """         
  memberList = []
  chatIdList = []
  emailList = []
  headers = {'Content-type':'application/json',
            "Authorization": f"Bearer {getGraphAccessToken()}"}
  connection = http.client.HTTPSConnection("graph.microsoft.com")
  for chat_id in currentChatsDataFrame[idColumn]:
    connection.request("GET",f"/v1.0/me/chats/{chat_id}/members",headers=headers)
    resp = connection.getresponse()
    text = resp.read()
    #chatIdList.append(chat_id)
    respJson = json.loads(text)['value']
    dprint(respJson,debug=debug)
    if respJson[1]['email'] =='your email address':# This returns the information about other chat member not the current user.
      chatIdList.append(chat_id)
      memberList.append(respJson[0]['displayName'])
      emailList.append(respJson[0]['email'])
    else:
      chatIdList.append(chat_id)
      memberList.append(respJson[1]['displayName'])
      emailList.append(respJson[1]['email'])
    dprint(text,debug=debug)
    dprint('---------------------------')
    #member_list.append(json.loads(text)['value'])
  dprint(f"chatIdList is {chatIdList}",debug=debug)
  dprint(f"memberlist is {memberList}",debug=debug)
  dprint(f"emaillist is {emailList}",debug=debug)
  return pd.DataFrame({'memberName':memberList,'memberEmail':emailList,'chatId': chatIdList})