拉取历史频道消息 python
Pulling historical channel messages python
我正在尝试通过从我所属的松弛通道中拉取 messages/responses 来创建一个小型数据集。我想使用 python 从通道中提取数据,但是我在计算 api 密钥时遇到了问题。我在 slack 上创建了一个应用程序,但我不确定如何找到我的 api 密钥。我看到了我的客户端密钥、签名密钥和验证令牌,但找不到我的 api 密钥
这是我认为我正在努力完成的基本示例:
import slack
sc = slack.SlackClient("api key")
sc.api_call(
"channels.history",
channel="C0XXXXXX"
)
如果可能的话,我也愿意手动下载数据。非常感谢任何帮助。
这是使用 slack webapi。您需要安装请求包。这应该获取频道中的所有消息。您需要一个可以从应用程序管理页面获取的令牌。您可以使用 getChannels() 函数。获取所有消息后,您需要查看谁写了什么消息,您需要进行 ID 匹配(将 ID 映射到用户名),您可以使用 getUsers() 函数。如果您不想使用应用程序中的令牌,请按照此 https://api.slack.com/custom-integrations/legacy-tokens 生成 legacy-token。
def getMessages(token, channelId):
print("Getting Messages")
# this function get all the messages from the slack team-search channel
# it will only get all the messages from the team-search channel
slack_url = "https://slack.com/api/conversations.history?token=" + token + "&channel=" + channelId
messages = requests.get(slack_url).json()
return messages
def getChannels(token):
'''
function returns an object containing a object containing all the
channels in a given workspace
'''
channelsURL = "https://slack.com/api/conversations.list?token=%s" % token
channelList = requests.get(channelsURL).json()["channels"] # an array of channels
channels = {}
# putting the channels and their ids into a dictonary
for channel in channelList:
channels[channel["name"]] = channel["id"]
return {"channels": channels}
def getUsers(token):
# this function get a list of users in workplace including bots
users = []
channelsURL = "https://slack.com/api/users.list?token=%s&pretty=1" % token
members = requests.get(channelsURL).json()["members"]
return members
消息
请参阅下面的示例代码,了解如何从 Python 中的频道提取消息。
- 它使用官方 Python Slack 库并调用
conversations_history
有分页。因此它将与
任何类型的渠道,如果
需要。
- 结果将作为 JSON 数组写入文件。
- 您可以指定要检索的频道和最大消息数
线程
请注意,conversations.history
endpoint will not return thread messages. Those have to be retrieved additionaly with one call to conversations.replies
您要为其检索消息的每个线程。
通过检查消息中的 threads_ts
属性,可以在每个频道的消息中识别线程。如果它存在,则有一个线程附加到它。有关线程如何工作的更多详细信息,请参阅此 page。
ID
虽然此脚本不会用名称替换 ID。如果您需要,这里有一些如何实现它的指导:
- 您需要替换用户、频道、机器人、用户组的 ID(如果是付费计划)
- 您可以分别使用
users_list
、conversations_list
和 usergroups_list
从 API 中获取用户、频道和用户组的列表,机器人需要通过一个 bots_info(如果需要)
- ID 出现在邮件中的许多地方:
- 用户顶级属性
- bot_id 顶级 属性
- as link 在任何允许文本的 属性 中,例如
<@U12345678>
表示用户,<#C1234567>
表示频道。这些可以出现在顶层 text
属性,也可以出现在附件和块中。
示例代码
import os
import slack
import json
from time import sleep
CHANNEL = "C12345678"
MESSAGES_PER_PAGE = 200
MAX_MESSAGES = 1000
# init web client
client = slack.WebClient(token=os.environ['SLACK_TOKEN'])
# get first page
page = 1
print("Retrieving page {}".format(page))
response = client.conversations_history(
channel=CHANNEL,
limit=MESSAGES_PER_PAGE,
)
assert response["ok"]
messages_all = response['messages']
# get additional pages if below max message and if they are any
while len(messages_all) + MESSAGES_PER_PAGE <= MAX_MESSAGES and response['has_more']:
page += 1
print("Retrieving page {}".format(page))
sleep(1) # need to wait 1 sec before next call due to rate limits
response = client.conversations_history(
channel=CHANNEL,
limit=MESSAGES_PER_PAGE,
cursor=response['response_metadata']['next_cursor']
)
assert response["ok"]
messages = response['messages']
messages_all = messages_all + messages
print(
"Fetched a total of {} messages from channel {}".format(
len(messages_all),
CHANNEL
))
# write the result to a file
with open('messages.json', 'w', encoding='utf-8') as f:
json.dump(
messages_all,
f,
sort_keys=True,
indent=4,
ensure_ascii=False
)
我正在尝试通过从我所属的松弛通道中拉取 messages/responses 来创建一个小型数据集。我想使用 python 从通道中提取数据,但是我在计算 api 密钥时遇到了问题。我在 slack 上创建了一个应用程序,但我不确定如何找到我的 api 密钥。我看到了我的客户端密钥、签名密钥和验证令牌,但找不到我的 api 密钥
这是我认为我正在努力完成的基本示例:
import slack
sc = slack.SlackClient("api key")
sc.api_call(
"channels.history",
channel="C0XXXXXX"
)
如果可能的话,我也愿意手动下载数据。非常感谢任何帮助。
这是使用 slack webapi。您需要安装请求包。这应该获取频道中的所有消息。您需要一个可以从应用程序管理页面获取的令牌。您可以使用 getChannels() 函数。获取所有消息后,您需要查看谁写了什么消息,您需要进行 ID 匹配(将 ID 映射到用户名),您可以使用 getUsers() 函数。如果您不想使用应用程序中的令牌,请按照此 https://api.slack.com/custom-integrations/legacy-tokens 生成 legacy-token。
def getMessages(token, channelId):
print("Getting Messages")
# this function get all the messages from the slack team-search channel
# it will only get all the messages from the team-search channel
slack_url = "https://slack.com/api/conversations.history?token=" + token + "&channel=" + channelId
messages = requests.get(slack_url).json()
return messages
def getChannels(token):
'''
function returns an object containing a object containing all the
channels in a given workspace
'''
channelsURL = "https://slack.com/api/conversations.list?token=%s" % token
channelList = requests.get(channelsURL).json()["channels"] # an array of channels
channels = {}
# putting the channels and their ids into a dictonary
for channel in channelList:
channels[channel["name"]] = channel["id"]
return {"channels": channels}
def getUsers(token):
# this function get a list of users in workplace including bots
users = []
channelsURL = "https://slack.com/api/users.list?token=%s&pretty=1" % token
members = requests.get(channelsURL).json()["members"]
return members
消息
请参阅下面的示例代码,了解如何从 Python 中的频道提取消息。
- 它使用官方 Python Slack 库并调用
conversations_history
有分页。因此它将与 任何类型的渠道,如果 需要。 - 结果将作为 JSON 数组写入文件。
- 您可以指定要检索的频道和最大消息数
线程
请注意,conversations.history
endpoint will not return thread messages. Those have to be retrieved additionaly with one call to conversations.replies
您要为其检索消息的每个线程。
通过检查消息中的 threads_ts
属性,可以在每个频道的消息中识别线程。如果它存在,则有一个线程附加到它。有关线程如何工作的更多详细信息,请参阅此 page。
ID
虽然此脚本不会用名称替换 ID。如果您需要,这里有一些如何实现它的指导:
- 您需要替换用户、频道、机器人、用户组的 ID(如果是付费计划)
- 您可以分别使用
users_list
、conversations_list
和usergroups_list
从 API 中获取用户、频道和用户组的列表,机器人需要通过一个 bots_info(如果需要) - ID 出现在邮件中的许多地方:
- 用户顶级属性
- bot_id 顶级 属性
- as link 在任何允许文本的 属性 中,例如
<@U12345678>
表示用户,<#C1234567>
表示频道。这些可以出现在顶层text
属性,也可以出现在附件和块中。
示例代码
import os
import slack
import json
from time import sleep
CHANNEL = "C12345678"
MESSAGES_PER_PAGE = 200
MAX_MESSAGES = 1000
# init web client
client = slack.WebClient(token=os.environ['SLACK_TOKEN'])
# get first page
page = 1
print("Retrieving page {}".format(page))
response = client.conversations_history(
channel=CHANNEL,
limit=MESSAGES_PER_PAGE,
)
assert response["ok"]
messages_all = response['messages']
# get additional pages if below max message and if they are any
while len(messages_all) + MESSAGES_PER_PAGE <= MAX_MESSAGES and response['has_more']:
page += 1
print("Retrieving page {}".format(page))
sleep(1) # need to wait 1 sec before next call due to rate limits
response = client.conversations_history(
channel=CHANNEL,
limit=MESSAGES_PER_PAGE,
cursor=response['response_metadata']['next_cursor']
)
assert response["ok"]
messages = response['messages']
messages_all = messages_all + messages
print(
"Fetched a total of {} messages from channel {}".format(
len(messages_all),
CHANNEL
))
# write the result to a file
with open('messages.json', 'w', encoding='utf-8') as f:
json.dump(
messages_all,
f,
sort_keys=True,
indent=4,
ensure_ascii=False
)