import time import datetime import logging from slack_bolt import App from slack_bolt.adapter.socket_mode import SocketModeHandler from slack_sdk.errors import SlackApiError from sqlalchemy.orm import Session from sqlalchemy import func from markdown_to_mrkdwn import SlackMarkdownConverter from community_qa_bot.pingcap_qa_service import CommunityQAService, CommunityNewQaService from community_qa_bot.settings import ( SLACK_APP_TOKEN, SLACK_DM_HELP_TEXT, SLACK_REACH_LIMIT_TEXT, SLACK_AI_FEATURE_NOT_ENABLED_TEXT, SLACK_NEW_HELP_MESSAGE_ALERT_CHANNEL_IDS, SLACK_NEW_HELP_MESSAGE_FEISHU_WEBHOOK, SLACK_QA_DAILY_LIMIT, TIDB_AI_API_KEY, TIDB_AI_SLACK_CHAT_ENGINE, ) from community_qa_bot.models import get_or_create, User, QaHistory, QaFeedback, SlackAIChannel from community_qa_bot.db import Session as DBSession from community_qa_bot.types import FeedbackType from community_qa_bot.notifier import FeishuNotifier, sanitize_and_shorten from community_qa_bot.text import split_text from community_qa_bot.slack.bot import installation_store, Bot, WebClient logger = logging.getLogger(__name__) qa_service = CommunityQAService( api_key=TIDB_AI_API_KEY, chat_engine=TIDB_AI_SLACK_CHAT_ENGINE, ) new_qa_service = CommunityNewQaService() app = App(installation_store=installation_store) def _get_user_identity_from_slack(client: WebClient, user_id: str): username = user_id display_name = None real_name = None name = None email = None is_email_confirmed = False try: user_info = client.users_info(user=user_id)['user'] if 'profile' in user_info: if user_info['profile'].get('display_name'): display_name = user_info['profile']['display_name'] if user_info['profile'].get('email'): email = user_info['profile']['email'] real_name = user_info.get('real_name') name = user_info.get('name') is_email_confirmed = user_info.get('is_email_confirmed') if display_name: username = f'{display_name}-{user_id}' elif real_name: username = f'{real_name}-{user_id}' elif name: username = f'{name}-{user_id}' except SlackApiError as e: raise Exception(f"Failed to get user info: {e.response['error']}") return { 'username': username, 'display_name': display_name, 'real_name': real_name, 'name': name, 'email': email, 'is_email_confirmed': is_email_confirmed } def get_user(session: Session, bot: Bot, client: WebClient, user_id: str) -> User: app_user_id = f'{user_id}-{bot.team_id}' user = session.query(User).filter(User.id == app_user_id).first() if not user: identity = _get_user_identity_from_slack(client, user_id) identity['id'] = app_user_id user = User(**identity) session.add(user) session.commit() return user def get_thread_history_from_slack(client: WebClient, channel_id: str, thread_ts: str) -> list: try: thread_history = client.conversations_replies(channel=channel_id, ts=thread_ts) return thread_history['messages'] except SlackApiError as e: logger.error(f"Got an error: {e.response['error']}") except Exception as e: logger.error(e) return [] def pure_mention_text(text: str, bot_user_id: str): return text.replace(f'<@{bot_user_id}>', '').strip() def build_qa_context(bot_user_id: str, message_history: list): qa_context = [] for m in message_history: if m.get('user') == bot_user_id: qa_context.append({ 'role': 'assistant', 'content': m['text'] }) elif m.get('type') == 'message': # only accept mention message if not m.get('text').startswith(f'<@{bot_user_id}>'): continue content = pure_mention_text(m['text'], bot_user_id) # skip empty message if not content: continue qa_context.append({ 'role': 'user', 'content': content }) else: continue qa_context = qa_context[-5:] should_skip = False if not qa_context or qa_context[-1]['role'] == 'assistant': should_skip = True return qa_context, should_skip def reach_daily_limit(session: Session, user: User): today = datetime.datetime.now().strftime('%Y-%m-%d') count = ( session.query(QaHistory) .filter( QaHistory.user_id == user.id, func.date(QaHistory.created_at) == today ).count() ) return count >= SLACK_QA_DAILY_LIMIT def is_channel_ai_enabled(session: Session, team_id: str, channel_id: str) -> bool: return session.query(SlackAIChannel).filter( SlackAIChannel.team_id == team_id, SlackAIChannel.channel_id == channel_id, ).first() is not None @app.middleware def db_session_middlerware(context, next): with DBSession() as session: context.db_session = session next() @app.event("app_mention") def handle_app_mention(event, say, context): logger.info(event) team_id = event.get('team') channel_id = event.get('channel') client_msg_id = event['client_msg_id'] thread_ts = event.get('thread_ts', event['ts']) bot, client = installation_store.find_bot_with_client(team_id) if not bot: logger.warning(f'bot not found for team_id: {team_id}') return if not is_channel_ai_enabled(context.db_session, team_id, channel_id): say(SLACK_AI_FEATURE_NOT_ENABLED_TEXT, thread_ts=thread_ts) client.conversations_leave(channel=channel_id) return user = get_user(context.db_session, bot, client, event['user']) if reach_daily_limit(context.db_session, user): say(SLACK_REACH_LIMIT_TEXT, thread_ts=thread_ts) return message_history = get_thread_history_from_slack(client, event['channel'], thread_ts) qa_context, should_skip = build_qa_context(bot.bot_user_id, message_history) if should_skip: return qa_history, _ = get_or_create( context.db_session, QaHistory, slack_message_id=client_msg_id, user_id=user.id, thread_ts=thread_ts, request_body=qa_context, ) # send thinking message to mark the message is being processed thinking_message_text = 'Thinking... :thinking_face:' thinking_slack_resp = say( blocks=[ { "type": "section", "text": { "type": "mrkdwn", "text": thinking_message_text }, }, ], text=thinking_message_text, thread_ts=thread_ts, ) thinking_message_ts = thinking_slack_resp['ts'] start_time = time.time() if team_id == 'T04AQPYPM': # PingCAP Workspace answer, ai_chat_id, ai_chat_message_id = qa_service.ask(origin="slack", qa_context=qa_context) else: answer, ai_chat_id, ai_chat_message_id = qa_service.ask(origin="slack", qa_context=qa_context) spend_time = time.time() - start_time qa_history.spend_time = spend_time qa_history.response_text = answer qa_history.ai_chat_id = ai_chat_id qa_history.ai_chat_message_id = ai_chat_message_id if answer: md_converter = SlackMarkdownConverter() answer = md_converter.convert(answer) # slack message max length is 3001 texts = split_text(answer, 3000) for i, text in enumerate(texts): blocks = [ { "type": "section", "text": {"type": "mrkdwn", "text": text}, }, ] # add like and dislike button at the last message if i == len(texts) - 1: blocks.append({ "type": "actions", "elements": [ { "type": "button", "text": {"type": "plain_text", "text": ":thumbsup:"}, "action_id": "click_like" }, { "type": "button", "text": {"type": "plain_text", "text": ":thumbsdown:"}, "action_id": "click_dislike" } ] }) slack_resp = say( blocks=blocks, text=text, thread_ts=event['ts'] ) time.sleep(0.2) qa_history.response_message_ts = slack_resp['ts'] client.chat_delete( channel=event['channel'], ts=thinking_message_ts, ) context.db_session.commit() @app.action("click_like") def handle_click_like(ack, body, context): team_id = body['team']['id'] bot, client = installation_store.find_bot_with_client(team_id) if not bot: logger.warning(f'bot not found for team_id: {team_id}') return channel_id = body['channel']['id'] user_id = body['user']['id'] message_ts = body['message']['ts'] thread_ts = body['message']['thread_ts'] user = get_user(context.db_session, bot, client, user_id) qa_history = ( context.db_session.query(QaHistory) .filter(QaHistory.response_message_ts == message_ts) .first() ) if qa_history: context.db_session.add(QaFeedback( qa_history_id=qa_history.id, user_id=user.id, feedback_type=FeedbackType.LIKE.value, )) context.db_session.commit() qa_service.like( origin="slack", chat_message_id=qa_history.ai_chat_message_id, username=user.username ) client.chat_postEphemeral( channel=channel_id, thread_ts=thread_ts, user=user_id, text='Thanks for your feedback!' ) ack() @app.action("click_dislike") def handle_click_dislike(ack, body, context): team_id = body['team']['id'] bot, client = installation_store.find_bot_with_client(team_id) if not bot: logger.warning(f'bot not found for team_id: {team_id}') return user_id = body['user']['id'] channel_id = body['channel']['id'] message_ts = body['message']['ts'] thread_ts = body['message']['thread_ts'] user = get_user(context.db_session, bot, client, user_id) qa_history = ( context.db_session.query(QaHistory) .filter(QaHistory.response_message_ts == message_ts) .first() ) if qa_history: context.db_session.add(QaFeedback( qa_history_id=qa_history.id, user_id=user.id, feedback_type=FeedbackType.DISLIKE.value, )) context.db_session.commit() qa_service.dislike( origin="slack", chat_message_id=qa_history.ai_chat_message_id, username=user.username, comment=f"From slack, user {user.username} click dislike button" ) client.chat_postEphemeral( channel=channel_id, thread_ts=thread_ts, user=user_id, text=f'Thanks for your feedback, could you please give me some advice, {qa_history.get_signature_url(user.id)}' ) ack() @app.event("message") def handle_message_events(event, say, logger): logger.info(event) team_id = event.get('team') or event.get('message', {}).get('team') bot, client = installation_store.find_bot_with_client(team_id) if not bot: logger.warning(f'bot not found for team_id: {team_id}') return channle = event['channel'] message_ts = event['ts'] channel_type = event.get('channel_type') if channel_type == 'im': say(SLACK_DM_HELP_TEXT) elif channel_type == 'channel': channle = event['channel'] message_ts = event['ts'] message_content = event.get('text', '') thread_ts = event.get('thread_ts') subtype = event.get('subtype') # skip thread message and message which is not sent by user if thread_ts or subtype: return if channle in SLACK_NEW_HELP_MESSAGE_ALERT_CHANNEL_IDS: link = client.chat_getPermalink( channel=channle, message_ts=message_ts )['permalink'] shorten_content = sanitize_and_shorten(message_content) FeishuNotifier( SLACK_NEW_HELP_MESSAGE_FEISHU_WEBHOOK ).send_markdown( title='New help message', content=f'Platform: Slack\nContent: {shorten_content}\nLink: {link}' ) def run_app(): SocketModeHandler(app, SLACK_APP_TOKEN).start()