def __init__(self):
token = os.environ.get('SLACK_TOKEN')
if not token:
print('Need to define ENV variable "SLACK_TOKEN"')
sys.exit(-1)
self.sc = SlackClient(token)
self.contacts = []
self.channels = []
self.active = []
self.ims = {}
python类SlackClient()的实例源码
def __init__(self, host, port, botname, **kwargs):
self.sc = SlackClient(SLACKBOT_API_TOKEN)
self.sc.rtm_connect()
self.botname = botname
self.host = host
self.port = str(port)
self.lang = 'en'
self.icon_url = 'https://avatars.slack-edge.com/2016-05-30/46725216032_4983112db797f420c0b5_48.jpg'
self.session_manager = SessionManager()
self.weights = kwargs.get('weights')
def __init__(self, slack_token):
self.sc = SlackClient(slack_token)
pass
def __init__(self):
self.slack = SlackClient(_config['slack_token'])
pass
slack_messenger.py 文件源码
项目:yahoo-fantasy-football-metrics
作者: uberfastman
项目源码
文件源码
阅读 21
收藏 0
点赞 0
评论 0
def __init__(self):
with open("./authentication/private.txt", "r") as auth_file:
auth_data = auth_file.read().split("\n")
slack_api_token = auth_data[2]
token = slack_api_token # found at https://api.slack.com/web#authentication
self.sc = SlackClient(token)
def __init__(self, bot_id, debug=False):
self.sc = SlackClient(bot_id)
self.user = ""
self.connect = self.sc.rtm_connect()
self.debug = debug
self.threads = []
t = threading.Thread(target=self.watch_message)
self.threads.append(t)
t.start()
self.handlers = []
def __init__(self, bot_id, debug=False):
self.sc = SlackClient(bot_id)
self.user = ""
self.connect = self.sc.rtm_connect()
self.debug = debug
self.threads = []
t = threading.Thread(target=self.watch_message)
self.threads.append(t)
t.start()
self.handlers = []
def __init__(self, bot_id, debug=False):
self.sc = SlackClient(bot_id)
self.user = ""
self.connect = self.sc.rtm_connect()
self.debug = debug
self.threads = []
t = threading.Thread(target=self.watch_message)
self.threads.append(t)
t.start()
self.handlers = []
def __init__(self, bot_id, debug=False):
self.sc = SlackClient(bot_id)
self.user = ""
self.connect = self.sc.rtm_connect()
self.debug = debug
self.threads = []
t = threading.Thread(target=self.watch_message)
self.threads.append(t)
t.start()
self.handlers = []
def __init__(self, key='', type='slack'):
self.key = key
self.type = type
self.parameters = {
'icon' : ':space_invader:',
'username': 'Notifier Bot',
'channel' : 'general'
}
if self.key:
if self.type == 'slack':
from slackclient import SlackClient
self.slackClient = SlackClient(self.key);
def test_get_user(self):
user = SlackUser.get_user(SlackClient('xxx'), test_user_name)
assert isinstance(user, SlackUser)
def test_get_user_none():
user = SlackUser.get_user(SlackClient('xxx'), 'doesnotexist')
assert user is None
def test_get_channel(self):
channel = self.room_class.get_channel(SlackClient('xxx'), self.test_room_name)
assert isinstance(channel, self.room_class)
def test_get_channel_none():
channel = SlackChannel.get_channel(SlackClient('xxx'), 'doesnotexist')
assert channel is None
def post_message_to_slack(self, channel, message, username="algobot", debug=False):
results = self.build_def_hash("Display Error", "Not Run", {
})
try:
import slackclient
if debug:
self.lg("Posting to Slack(" + str(channel) + ") Message(" + str(message)[0:10] + ")", 6)
else:
slack = slackclient.SlackClient(self.m_slack_node["Token"])
slack.api_call("chat.postMessage", channel=channel, text=message, username=username, as_user=True)
if debug:
self.lg("Done Posting to Slack(" + str(channel) + ")", 6)
except Exception, e:
err_msg = "Failed to post message to slack with Ex(" + str(e) + ")"
self.lg("ERROR: " + str(err_msg), 0)
results = self.build_def_hash("Display Error", str(err_msg), {
})
results["Status"] = "FAILED"
# end of try/ex
return results
# end of post_message_to_slack
def post_message_to_slack(self, channel, message, debug=False):
slack_response = {}
try:
# SlackHQ Repository for integrating with the Slack API:
# https://github.com/slackhq/python-slackclient
import slackclient
slack = slackclient.SlackClient(self.m_slack_token)
slack_response = slack.api_call("chat.postMessage", channel=channel, text=message, username=self.m_bot_name, as_user=True)
if "error" in slack_response:
print "\nERROR: Slack API call had an error(" + str(slack_response["error"]) + ")\n"
except Exception, e:
err_msg = "Failed to post message to slack with Ex(" + str(e) + ") SlackResponse(" + str(slack_response) + ")"
print "ERROR: " + str(err_msg)
# end of try/ex
# end of post_message_to_slack
# end of SlkMsg
def __init__(self, slack_token, channel_name, bot_name="Notifier Bot"):
super().__init__()
self.bot_name = bot_name
self.channel_name = channel_name
self.slack_token = slack_token
self.slack_client = SlackClient(self.slack_token)
def __init__(self):
self._client = SlackClient(self._token())
self._channel = CHANNEL
self._icon = ICON
self._username = USERNAME
self._client.rtm_connect()
def __init__(self, token):
self.token = token
# Slacker is a Slack Web API Client
self.web = Slacker(token)
# SlackClient is a Slack Websocket RTM API Client
self.rtm = SlackClient(token)
def __init__(self, token, ping_interval=3, client=None, **kwargs):
"""
:param token: The slack API token.
:param ping_interval: The interval in seconds we should ping Slack
server to keep the connection alive.
:param client: By default, BigSlacker uses Slack official Python
client, but you can pass any other client here, provided it has
the same interface.
:param process_plugin_return: Defaults to True.
This process the output (return) of a plugin function we call as a
message to send. Therefore making it easy to "reply" to events.
If the plugin method returns None we do nothing.
Returning [(channel, message)] will make us send it. You can return one
or more messages.
:param sleeping_time: Every time we read data from the WebSocket we
sleep 1 second by default. If you have a different need, you can
override this setting.
"""
self.client = client or SlackClient(token)
self.ping_interval = ping_interval
self.process_plugin_return = kwargs.get('process_plugin_return', True)
self._injected_plugins = []
self._last_call = None
self._sleeping_time = kwargs.get('sleeping_time', 1)
self._custom_callbacks = {}
self._load_plugins()