As a seasoned expert in AI, Machine Learning, Generative AI, IoT and Robotics, I empower innovators and businesses to harness the potential of emerging technologies. With a passion for sharing knowledge, I curate insightful articles, tutorials and news on the latest advancements in AI, Robotics, Data Science, Cloud Computing and Open Source technologies. Hire Me Unlock cutting-edge solutions for your business. With expertise spanning AI, GenAI, IoT and Robotics, I deliver tailor services.
Friday
Chatbot and Local CoPilot with Local LLM, RAG, LangChain, and Guardrail
Saturday
Telegram Bot for Monitoring Summarizing and Sending Periodic Qverviews of Channel Posts
To develop a Telegram bot for monitoring, summarizing, and sending periodic overviews of channel posts, follow these steps:
Step 1: Set Up Your Environment
1. Install Python: Ensure you have Python installed on your system.
2. Install Required Libraries:
```python
pip install python-telegram-bot requests beautifulsoup4
```
Step 2: Create the Telegram Bot
1. Create a Bot on Telegram: Talk to [@BotFather](https://telegram.me/BotFather) to create a new bot. Note the API token provided.
Step 3: Develop the Bot
1. Monitor Telegram Channels:
```python
from telegram import Bot, Update
from telegram.ext import Updater, CommandHandler, MessageHandler, Filters, CallbackContext
import requests
from bs4 import BeautifulSoup
TOKEN = 'YOUR_TELEGRAM_BOT_TOKEN'
CHANNELS = ['@example_channel_1', '@example_channel_2']
SUMMARY_PERIOD = 60 * 60 # in seconds (1 hour)
bot = Bot(token=TOKEN)
def summarize_text(text):
# Use a simple summarization logic or integrate with an NLP model
return text[:100] + '...'
def monitor_channels(context: CallbackContext):
summaries = []
for channel in CHANNELS:
url = f'https://t.me/s/{channel.strip("@")}'
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')
posts = soup.find_all('div', class_='tgme_widget_message_text')
for post in posts:
summaries.append(summarize_text(post.get_text()))
summary = '\n\n'.join(summaries)
bot.send_message(chat_id=context.job.context, text=summary)
def start(update: Update, context: CallbackContext):
context.job_queue.run_repeating(monitor_channels, SUMMARY_PERIOD, context=update.message.chat_id)
update.message.reply_text('Bot started! You will receive periodic summaries.')
updater = Updater(token=TOKEN, use_context=True)
dp = updater.dispatcher
dp.add_handler(CommandHandler('start', start))
updater.start_polling()
updater.idle()
```
2. Customize Channels and Summary Period:
```python
def add_channel(update: Update, context: CallbackContext):
new_channel = context.args[0]
if new_channel not in CHANNELS:
CHANNELS.append(new_channel)
update.message.reply_text(f'Channel {new_channel} added.')
else:
update.message.reply_text(f'Channel {new_channel} already in the list.')
def remove_channel(update: Update, context: CallbackContext):
channel = context.args[0]
if channel in CHANNELS:
CHANNELS.remove(channel)
update.message.reply_text(f'Channel {channel} removed.')
else:
update.message.reply_text(f'Channel {channel} not found.')
def set_period(update: Update, context: CallbackContext):
global SUMMARY_PERIOD
try:
new_period = int(context.args[0]) * 60
SUMMARY_PERIOD = new_period
update.message.reply_text(f'Summary period set to {new_period // 60} minutes.')
except ValueError:
update.message.reply_text('Invalid period. Please provide a number.')
dp.add_handler(CommandHandler('add_channel', add_channel))
dp.add_handler(CommandHandler('remove_channel', remove_channel))
dp.add_handler(CommandHandler('set_period', set_period))
```
3. Documentation:
Provide clear instructions on how to use the bot, including commands to add/remove channels and set the summary period.
Step 4: Ensure Security and Compliance
- Secure Your Bot: Implement security measures to ensure the bot only responds to authorized users.
- Adhere to Telegram's API Usage Policies: Follow Telegram's guidelines and avoid actions that may lead to the bot being banned.
Step 5: Deployment and Support
- Deploy: Host your bot on a server to keep it running continuously.
- Ongoing Support: Be prepared to troubleshoot issues and update the bot as needed.
By following these steps, you can create a robust Telegram bot for monitoring, summarizing, and sending periodic overviews of channel posts.
Thursday
Bot State with Azure
We can use Azure Bot Application using FastAPI that integrates with Azure Cache for Redis for session management and uses Azure Cosmos DB for state management. Here are the steps to achieve this:
State Management with Azure Cosmos DB:
- Why do you need state?
- Maintaining state allows your bot to have more meaningful conversations by remembering certain things about a user or conversation.
- For example, if you’ve talked to a user previously, you can save previous information about them, so that you don’t have to ask for it again.
- State also keeps data for longer than the current turn, so your bot retains information over the course of a multi-turn conversation.
- Why do you need state?
Storage Layer:
- The backend storage layer is where the state information is actually stored.
- You can choose from different storage options:
- Memory Storage: For local testing only; volatile and temporary.
- Azure Blob Storage: Connects to an Azure Blob Storage object database.
- Azure Cosmos DB Partitioned Storage: Connects to a partitioned Cosmos DB NoSQL database.
- Note: The legacy Cosmos DB storage class has been deprecated.
State Management:
- State management automates reading and writing bot state to the underlying storage layer.
- State is stored as state properties (key-value pairs).
- The Bot Framework SDK abstracts the underlying implementation.
- You can use state property accessors to read and write state without worrying about storage specifics.
Setting Up Azure Cosmos DB for Bot State:
- Create an Azure Cosmos DB Account (globally distributed, multi-model database service).
- Within the Cosmos DB account, create a SQL Database to store the state of your bot effectively.
Implementing in Your Bot:
- In your bot code, use the appropriate storage provider (e.g., Cosmos DB) to manage state.
- Initialize state management and property accessors.
- Example (using FastAPI):
import azure.functions as func from WrapperFunction import app as fastapi_app from bot_state import BotState, CosmosDbPartitionedStorage # Initialize Cosmos DB storage cosmos_db_storage = CosmosDbPartitionedStorage( cosmos_db_endpoint="your_cosmos_db_endpoint", cosmos_db_key="your_cosmos_db_key", database_id="your_database_id", container_id="your_container_id" ) # Initialize bot state bot_state = BotState(cosmos_db_storage)
# Example: Writing a user-specific property async def save_user_preference(turn_context, preference_value): user_id = turn_context.activity.from_property.id await bot_state.user_state.set_property(turn_context, f"user_preference_{user_id}", preference_value) # Example: Reading a user-specific property async def get_user_preference(turn_context): user_id = turn_context.activity.from_property.id preference_value = await bot_state.user_state.get_property(turn_context, f"user_preference_{user_id}") return preference_value # Usage in your bot logic async def on_message_activity(turn_context): # Get user preference preference = await get_user_preference(turn_context) await turn_context.send_activity(f"Your preference: {preference}") # Set user preference await save_user_preference(turn_context, "New Preference Value") await turn_context.send_activity("Preference updated!") app = func.AsgiFunctionApp(app=fastapi_app, http_auth_level=func.AuthLevel.ANONYMOUS)mple: Writing a user-specific property async def save_user_preference(turn_context, preference_value): user_id = turn_context.activity.from_property.id await bot_state.user_state.set_property(turn_context, f"user_preference_{user_id}", preference_value) # Example: Reading a user-specific property async def get_user_preference(turn_context): user_id = turn_context.activity.from_property.id preference_value = await bot_state.user_state.get_property(turn_context, f"user_preference_{user_id}") return preference_value # Usage in your bot logic async def on_message_activity(turn_context): # Get user preference preference = await get_user_preference(turn_context) await turn_context.send_activity(f"Your preference: {preference}") # Set user preference await save_user_preference(turn_context, "New Preference Value") await turn_context.send_activity("Preference updated!")
Testing Locally and Deployment:
- Test your bot locally using VS Code or Azure CLI.
- Deploy your bot to Azure using the VS Code Azure Functions extension or Azure CLI.
-
The client want to have a shop where regular customers to be able to see products with their retail price, while Wholesale partners to see t...
-
URL based session management does not only have additional security risks compared to cookie based session management, but it can cause also...
-
Widgets and gadgets are small applications that run on your desktop or in your web browser which enable you to keep track of things like the...