def setup(self):
try:
config = self.container.config[constants.CONFIG_KEY]
except KeyError:
raise ConfigurationError(
'`{}` config key not found'.format(constants.CONFIG_KEY))
token = config.get('TOKEN')
clients = config.get('BOTS')
if token:
self.clients[constants.DEFAULT_BOT_NAME] = SlackClient(token)
if clients:
for bot_name, token in clients.items():
self.clients[bot_name] = SlackClient(token)
if not self.clients:
raise ConfigurationError(
'At least one token must be provided in `{}` config'
.format(constants.CONFIG_KEY))
python类SlackClient()的实例源码
def setup(self):
try:
config = self.container.config[constants.CONFIG_KEY]
except KeyError:
raise ConfigurationError(
'`{}` config key not found'.format(constants.CONFIG_KEY))
if self.bot_name:
try:
token = config['BOTS'][self.bot_name]
except KeyError:
raise ConfigurationError(
'No token for `{}` bot in `{}` config'
.format(self.bot_name, constants.CONFIG_KEY))
else:
token = (
config.get('BOTS', {}).get(constants.DEFAULT_BOT_NAME) or
config.get('TOKEN'))
if not token:
raise ConfigurationError(
'No token provided by `{}` config'
.format(constants.CONFIG_KEY))
self.client = SlackClient(token)
def sendSlack(channel, token, message, debug):
"""simple Slack sender for status reports"""
#
# simple input validation
#
if debug:
app.logger.debug("Not sending out SLACK message: " + str(message))
return 0
if channel is None or token is None or message is None:
return
sc = SlackClient(token)
sc.api_call(
"chat.postMessage",
channel=channel,
text=message
)
def slack_fileUpload(filename, file):
global slack_api_token
try:
slack_api_token
except NameError:
slack_api_token = os.environ["SLACK_API_TOKEN"]
sc = SlackClient(slack_api_token)
response = sc.api_call(
'files.upload',
channels=slack_channel,
filename=filename,
file=open(file, 'rb'),
title="Full scan results")
if not response['ok']:
print("Slack Error: {}".format(response['error']))
else:
print("+ Full report uploaded to Slack")
def _post_result():
slack = SlackClient(_config['slack_token'])
attachments = []
for key, value in _results.iteritems():
attachments.append({
'title': key,
'text': value,
'color': ('danger' if value != 'passed' else 'good')
})
formatted_result = _config.copy()
formatted_result.update({
'title': 'Execution Results',
'attachments': attachments,
'as_user': "false"
})
slack.api_call("chat.postMessage",
text="*Test results*",
**formatted_result)
def list_channels():
"""
helper method for listing all slack channels
:return: None
"""
bot_token = ENV_DICT.get('SLACKBOT_TOKEN')
if not bot_token:
return []
sc = SlackClient(bot_token)
channels = sc.api_call('channels.list')
# this is to make this function backwards-compatible with older version of SlackClient which returns a string
if isinstance(channels, basestring):
channels = json.loads(channels)
return channels
# construct a map from slack channel name to slack channel id (the format that the API expects)
def __init__(self, api_key):
"""
SlackWrapper constructor.
Connect to the real-time messaging API and
load the bot's login data.
"""
self.api_key = api_key
self.client = SlackClient(self.api_key)
self.connected = self.client.rtm_connect()
self.server = None
self.username = None
self.user_id = None
if self.connected:
self.server = self.client.server
self.username = self.server.username
self.user_id = self.server.login_data.get('self').get('id')
def slackMsg(item,color,link, size):
# line 101
if slack_token == "":
return
sc = SlackClient(slack_token)
text = item+"\n"
text += color+'\n'
text += size.title()+'\n'
text += link+'\n'
text += "Restock!"+'\n'
text += str(datetime.utcnow().strftime('%H:%M:%S.%f')[:-3])
sc.api_call(
"chat.postMessage",
channel="#test",
text=text
)
def __init__(self):
load_dotenv(".env")
#
BOT_CHANNEL_GENERAL = os.environ.get("BOT_CHANNEL_GENERAL")
SLACK_BOT_TOKEN = os.environ.get("SLACK_BOT_TOKEN")
#
self.BOT_ID = os.environ.get("BOT_ID")
self.AT_BOT = "<@" + str(self.BOT_ID) + ">"
self.WEATHER_API_TOKEN = os.environ.get("WEATHER_API_TOKEN")
self.slack_client = SlackClient(SLACK_BOT_TOKEN)
self.public_channels_list = [BOT_CHANNEL_GENERAL]
self.input = Input(self)
self.command = Command(self)
self.weather = Weather(self)
#
if self.slack_client.rtm_connect():
print("Bot connected and running!")
self.input.read()
else:
print("Connection failed. Invalid Slack token or bot ID?")
def initialize(update_everyone=False):
"""Used to initalize resources - both the 'shelf' store and slack client - and return them.
Args:
update_everyone (Optional[bool]): If True, updates all users in the EMAIL_DOMAIN
from slack. Defaults to False.
Returns:
store: A shelve instance
sc: A SlackClient instance
"""
store = open_store()
sc = get_slack_client()
if update_everyone:
update_everyone_from_slack(store, sc)
return store, sc
def update_everyone_from_slack(store, sc):
"""Updates our store's list of `everyone`.
This list is comprised of all slack users with
the specified EMAIL_DOMAIN in config.py.
Args:
store (instance): A persistent, dictionary-like object used to keep information about past/future meetings
sc (SlackClient): An instance of SlackClient
"""
if not sc:
sc = get_slack_client()
users = sc.api_call("users.list")
store['everyone'] = [m['name'] for m in users['members']
if not m['deleted'] and
m['profile'].get('email') and
m['profile']['email'].endswith('@' + EMAIL_DOMAIN)]
def send_to_slack(message, config, channel='#general'):
"""
Sends a message to the ToR slack.
:param message: String; the message that is to be encoded
:param channel: String; channel option, defaults to general
:param config: the global config dict.
:return: None.
"""
if config.slack_api_key:
slackc = SlackClient(config.slack_api_key)
slackc.api_call(
'chat.postMessage',
channel=channel,
text=message
)
return
def __init__(self, name=None, slack_client=None, plugin_config=None):
'''
A plugin in initialized with:
- name (str)
- slack_client - a connected instance of SlackClient - can be used to make API
calls within the plugins
- plugin config (dict) - (from the yaml config)
Values in config:
- DEBUG (bool) - this will be overridden if debug is set in config for this plugin
'''
if name is None:
self.name = type(self).__name__
else:
self.name = name
if plugin_config is None:
self.plugin_config = {}
else:
self.plugin_config = plugin_config
self.slack_client = slack_client
self.jobs = []
self.debug = self.plugin_config.get('DEBUG', False)
self.outputs = []
def __init__(self, slack_token, reporting_channel, state):
"""Connect to Slack and setup the object."""
self.channels = {}
self.client = SlackClient(slack_token)
if self.client.rtm_connect():
print("HueBot connected to Slack!")
else:
print("HueBot did not connect to Slack")
raise IOError
self.reporting_channel = reporting_channel
self.state = state
self.state.on_new_failure = self.message_failure
self.state.on_new_warning = self.message_warning
# Retrieve parse functions
self.parse_functions = []
for attr_name in dir(self):
attr = getattr(self, attr_name)
try:
if attr._parse_func:
self.parse_functions.append(attr)
except AttributeError:
pass
def slack_notify_message(message, channel_id=None):
"""
sends a slack message (for logging, and error notification)
:param message:
:return: None
"""
bot_token = SECRETS_DICT['CITIGROUP_SLACKBOT_TOKEN']
# bot_token = SECRETS_DICT['COMPUTERLAB_SLACKBOT_TOKEN']
sc = SlackClient(bot_token)
if not channel_id:
channel_id = 'C193UMR0V'
print message
sc.api_call('chat.postMessage', channel=channel_id,
text='{message}'.format(message=message), link_names=1,
as_user=True)
def __init__( self, actor, oid, apiToken, botToken ):
self.actor = actor
self.oid = oid
self.apiToken = apiToken
self.botToken = botToken
self.slack = Slacker( self.apiToken )
self.bot = SlackClient( self.botToken )
self.botId = None
self.invId = str( uuid.uuid4() )
self.taskId = 0
self.slackLinkRE = re.compile( r'<.+://.+\|(.+)>' )
self.history = { 'last_cmd' : [] }
self.makeChannel( '#detects' )
if not self.bot.rtm_connect():
raise Exception( 'failed to connect bot to Slack rtm API' )
self.actor.newThread( self.botThread )
def dispatch(self, request, *args, **kwargs):
try:
code = request.GET.get('code', '')
sc = SlackClient("")
result = sc.api_call("oauth.access", client_id=settings.SLACK_CLIENT_ID,
client_secret=settings.SLACK_CLIENT_SECRET, code=code,
redirect_uri=request.build_absolute_uri(reverse('oauth')))
if SlackAuth.objects.filter(team_id=result["team_id"]).exists():
SlackAuth.objects.get(team_id=result["team_id"]).delete()
slack_auth = SlackAuth.objects.create(access_token=result["access_token"], team_id=result["team_id"],
team_name=result["team_name"], bot_id=result["bot"]["bot_user_id"],
bot_access_token=result["bot"]["bot_access_token"])
retrieve_channel_users.delay(slack_auth.pk)
return HttpResponseRedirect(reverse("success"))
except Exception:
logger.error(traceback.format_exc())
return HttpResponseRedirect(reverse("failure"))
def retrieve_channel_users(slack_auth_id):
slack_auth = SlackAuth.objects.get(pk=slack_auth_id)
sc = SlackClient(slack_auth.bot_access_token)
channels = sc.api_call("channels.list")
for channel in channels["channels"]:
if channel["is_general"]:
slack_auth.general_channel_id = channel['id']
slack_auth.save()
break
users = sc.api_call("users.list")
for user in users["members"]:
if user["is_bot"] or user["id"] == "USLACKBOT":
continue
if not TalkUser.objects.filter(slack_id=user["id"], slack_auth=slack_auth).exists():
talk_user = TalkUser.objects.create(email=user["profile"]["email"],
name=user["profile"]["real_name"],
slack_id=user["id"],
slack_auth=slack_auth)
# sm = SlackMessage(slack_auth.bot_access_token, talk_user.slack_id)
# sm.send_text("Hi, my name is TickerBot. I am a conversational game on slack. You compete with your team"
# " for the best portfolio! To learn how to play, just ask :)",
# actions=[{"text": "how to play", "value": "how to play"}])
# sc = SlackClient(slack_auth.access_token)
# result = sc.api_call("channels.join", name=slack_auth.general_channel_id)
def pageParse(t=0):
custom = SlackClient("")
nowpage = 1
imgs = []
while True:
rep = custom.api_call(
"files.list",
channel="C045B1Y37",
type="images",
page=nowpage,
ts_from=t)
nowpage += 1
print(rep)
rep['files'].reverse()
imgs.extend(rep['files'])
if rep['paging']['pages'] < nowpage:
break
imgs.reverse()
return imgs
def __init__(self):
self.colorPrint = ColorPrint.setPrint("Root")
self.colorPrint("Test Unit", wantname)
modules = InitModule.modulesGet()
privacy = password_crypt.logIn(InitModule.requiresGet(modules))
# self.colorPrint("Secret",privacy,color="WARNING")
modules = [i for i in modules if i['name'] in ["", *wantname]]
self.modules, base = InitModule.initSet(privacy, modules)
self.slack = SlackClient(base.get('TOKEN'))
self.ignoretype = [
'user_typing',
'reconnect_url',
'pref_change',
'presence_change',
'emoji_changed',
'desktop_notification']
def _api_call(self, method, **kwargs):
# type: (str, **Any) -> Dict[str, Any]
'''
Performs a _validated_ Slack API call. After performing a normal API
call using SlackClient, validate that the call returned 'ok'. If not,
log and error.
Args:
method (str): The API endpoint to call.
**kwargs: Any arguments to pass on to the request.
Returns:
(dict): Parsed JSON from the response.
'''
response = self._slack.api_call(method, **kwargs)
if not ('ok' in response and response['ok']):
if kwargs:
logging.error('Bad Slack API request on {} with {}'.format(method, kwargs))
else:
logging.error('Bad Slack API request on {}'.format(method))
return response
def __init__(self):
super(Bot, self).__init__()
self.name = "pythonboardingbot"
self.emoji = ":robot_face:"
# When we instantiate a new bot object, we can access the app
# credentials we set earlier in our local development environment.
self.oauth = {"client_id": os.environ.get("CLIENT_ID"),
"client_secret": os.environ.get("CLIENT_SECRET"),
# Scopes provide and limit permissions to what our app
# can access. It's important to use the most restricted
# scope that your app will need.
"scope": "bot"}
self.verification = os.environ.get("VERIFICATION_TOKEN")
# NOTE: Python-slack requires a client connection to generate
# an oauth token. We can connect to the client without authenticating
# by passing an empty string as a token and then reinstantiating the
# client with a valid OAuth token once we have one.
self.client = SlackClient("")
# We'll use this dictionary to store the state of each message object.
# In a production envrionment you'll likely want to store this more
# persistantly in a database.
self.messages = {}
def run(self):
# create one at https://api.slack.com/web#authentication, and set the environment variable
sc = SlackClient(os.environ["SLACK_CLIENT_TOKEN"])
channel_id = 0
# Get Channel Information
for channel in sc.api_call("channels.list")["channels"]:
if (channel["name"] == self.channel_name):
channel_id = channel["id"]
# Get Channel History
if (self.last_message_id):
channel_history_chunk = sc.api_call("channels.history", channel=channel_id, count=1000, latest=self.last_message_id)
else:
channel_history_chunk = sc.api_call("channels.history", channel=channel_id, count=1000)
if (not channel_history_chunk["ok"]):
raise Exception('Channel not found, or permissions error', 'channel_name=' + self.channel_name)
channel_history_chunk_last_message = channel_history_chunk["messages"]
outputdata = channel_history_chunk
with self.output().open('w') as outfile:
json.dump(outputdata, outfile, sort_keys=True, indent=4, separators=(',', ': '))
def edubot_channel_post(sc, team_bot, message, employee):
from slackclient import SlackClient
channels_list = sc.api_call('channels.list', exclude_archived=1)
channel_names = [channel['name'] for channel in channels_list['channels']]
if 'question' in channel_names:
channel_id = '#question'
post_on_question_channel(sc, channel_id, message)
else:
owner_sc = SlackClient(team_bot.owner.slack_access_token)
msg_res = owner_sc.api_call('channels.create', name='question')
channel_id = msg_res['channel']['id']
mentioned_employees = extract_mentions("<!everyone>", team_bot.slack_team_id, team_bot.owner)
if mentioned_employees:
for employee in mentioned_employees:
owner_sc.api_call('channels.invite', channel=channel_id, user=employee.user.username)
post_on_question_channel(sc, channel_id, message)
def eval_teambot_events(token, team_bot_id, events):
from datetime import datetime
print("%s == Evaluating event" % (datetime.now().time()))
team_bot = TeamBot.objects.filter(id=int(team_bot_id)).first()
if len(events) == 0:
return None
sc = SlackClient(token)
for event in events:
dm_status, subtype = is_dm(sc, event)
if is_message(event) and ('subtype' not in event):
if ('user' in event) and (event['user'] != 'USLACKBOT'):
team = team_bot.team
save_ic_message(team, event, dm_status, team_bot)
if is_message(event):
#ignore your own and other bots' messages
if ((not subtype) or (subtype not in ('bot_message', 'message_changed'))):
if dm_status:
collaborate_with_user(sc, event['user'], event['team'], event['channel'], event['text'], team_bot)
else:
print("ignoring as echo or other bot talking")
else:
classify_and_act(sc, team_bot, event)
def do_scrape():
"""
Runs the craigslist scraper, and posts data to slack.
"""
# Create a slack client.
sc = SlackClient(settings.SLACK_TOKEN)
# Get all the results from craigslist.
all_results = []
for area in settings.AREAS:
all_results += scrape_area(area)
print("{}: Got {} results".format(time.ctime(), len(all_results)))
# Post each result to slack.
for result in all_results:
post_listing_to_slack(sc, result)
def get_redirect_url(self, *args, **kwargs):
if not SLACK_ID:
return reverse("project:edit", kwargs={'project': self.kwargs['project']})
slack = SlackClient("")
code = self.request.GET['code']
resp = slack.api_call(
"oauth.access",
code=code,
client_id=SLACK_ID,
client_secret=SLACK_SECRET,
redirect_uri="https://" + HOST + reverse("integration:slack:auth",
kwargs={'project': self.kwargs['project']}),
)
if resp['ok']:
si = SlackIntegration()
si.api_token = resp['access_token']
si.project = Project.objects.get(name_short=self.kwargs['project'])
si.save()
return reverse("integration:slack:update", kwargs={'project': self.kwargs['project'], 'pk': si.pk})
return reverse("project:edit", kwargs={'project': self.kwargs['project']})
def connect(self):
"""Convenience method that creates Server instance"""
self.slack_client = SlackClient(self.token)
self.slack_client.rtm_connect()
def __init__(self):
super(Bot, self).__init__()
self.name = "poorwebhook"
self.emoji = ":robot_face:"
# When we instantiate a new bot object, we can access the app
# credentials we set earlier in our local development environment.
self.oauth = {"client_id": os.environ.get("CLIENT_ID"),
"client_secret": os.environ.get("CLIENT_SECRET"),
# Scopes provide and limit permissions to what our app
# can access. It's important to use the most restricted
# scope that your app will need.
"scope": "bot"}
self.verification = os.environ.get("VERIFICATION_TOKEN")
# NOTE: Python-slack requires a client connection to generate
# an oauth token. We can connect to the client without authenticating
# by passing an empty string as a token and then reinstantiating the
# client with a valid OAuth token once we have one.
self.client = SlackClient('')
# We'll use this dictionary to store the state of each message object.
# In a production environment you'll likely want to store this more
# persistently in a database.
self.messages = {}
self.dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
self.table = self.dynamodb.Table('poor-webhook-tokens')
def message_channel(self, channel, message):
creds = self.auth_info()
self.client = SlackClient(creds['bot_token'])
response = message
res = self.client.api_call(
"chat.postMessage",
channel=channel,
text=response,
attachments=None
)
print(res)
