我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用slackclient.SlackClient()。
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))
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)
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 = {}
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
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, 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()
