aljaz/chatgpt-telegram-bot
1
1
Fork 0
mirror of https://github.com/aljazceru/chatgpt-telegram-bot.git synced 2026-01-10 00:16:09 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into main

Browse Source
This commit is contained in:
ned
2023-04-16 09:41:41 +02:00
committed by GitHub
parent a353073122 1ac0b248c2
commit 61ccd5df88
7 changed files with 477 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.github/workflows/publish.yaml vendored
View File

@@ -16,6 +16,12 @@ jobs:
steps:
- name: Check out the repo
uses: actions/checkout@v3
- name: Set up QEMU
uses: docker/setup-qemu-action@v2
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v2
- name: Log in to Docker Hub
uses: docker/login-action@f054a8b539a109f9f41c372932f1ae047eff08c9

32
README.md
View File

@@ -4,7 +4,7 @@
[![license](https://img.shields.io/badge/License-GPL%202.0-brightgreen.svg)](LICENSE)
[![Publish Docker image](https://github.com/n3d1117/chatgpt-telegram-bot/actions/workflows/publish.yaml/badge.svg)](https://github.com/n3d1117/chatgpt-telegram-bot/actions/workflows/publish.yaml)
A [Telegram bot](https://core.telegram.org/bots/api) that integrates with OpenAI's _official_ [ChatGPT](https://openai.com/blog/chatgpt/) APIs to provide answers. Ready to use with minimal configuration required.
A [Telegram bot](https://core.telegram.org/bots/api) that integrates with OpenAI's _official_ [ChatGPT](https://openai.com/blog/chatgpt/), [DALL·E](https://openai.com/product/dall-e-2) and [Whisper](https://openai.com/research/whisper) APIs to provide answers. Ready to use with minimal configuration required.
## Screenshots
![demo](https://user-images.githubusercontent.com/11541888/225114786-0d639854-b3e1-4214-b49a-e51ce8c40387.png)
@@ -17,9 +17,7 @@ A [Telegram bot](https://core.telegram.org/bots/api) that integrates with OpenAI
- [x] Docker and Proxy support
- [x] (NEW!) Image generation using DALL·E via the `/image` command
- [x] (NEW!) Transcribe audio and video messages using Whisper (may require [ffmpeg](https://ffmpeg.org))
- [x] (NEW!) Automatic conversation summary to avoid excessive token usage (fixes [#34](https://github.com/n3d1117/chatgpt-telegram-bot/issues/34))
- [x] (NEW!) Group chat support with inline queries
- To use this feature, enable inline queries for your bot in BotFather via the `/setinline` [command](https://core.telegram.org/bots/inline)
- [x] (NEW!) Automatic conversation summary to avoid excessive token usage
- [x] (NEW!) Track token usage per user - by [@AlexHTW](https://github.com/AlexHTW)
- [x] (NEW!) Get personal token usage statistics and cost per day/month via the `/stats` command - by [@AlexHTW](https://github.com/AlexHTW)
- [x] (NEW!) User budgets and guest budgets - by [@AlexHTW](https://github.com/AlexHTW)
@@ -27,7 +25,9 @@ A [Telegram bot](https://core.telegram.org/bots/api) that integrates with OpenAI
- [x] (NEW!) GPT-4 support
- If you have access to the GPT-4 API, simply change the `OPENAI_MODEL` parameter to `gpt-4`
- [x] (NEW!) Localized bot language
- Available languages :gb: :de: :ru: :tr: :it:
- Available languages :gb: :de: :ru: :tr: :it: :es:
- [x] (NEW!) Improved inline queries support for group and private chats - by [@bugfloyd](https://github.com/bugfloyd)
- To use this feature, enable inline queries for your bot in BotFather via the `/setinline` [command](https://core.telegram.org/bots/inline)
## Additional features - help needed!
If you'd like to help, check out the [issues](https://github.com/n3d1117/chatgpt-telegram-bot/issues) section and contribute!
@@ -49,21 +49,21 @@ Customize the configuration by copying `.env.example` and renaming it to `.env`,
|-----------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `OPENAI_API_KEY` | Your OpenAI API key, you can get it from [here](https://platform.openai.com/account/api-keys) |
| `TELEGRAM_BOT_TOKEN` | Your Telegram bot's token, obtained using [BotFather](http://t.me/botfather) (see [tutorial](https://core.telegram.org/bots/tutorial#obtain-your-bot-token)) |
| `ADMIN_USER_IDS` | Telegram user IDs of admins. These users have access to special admin commands, information and no budget restrictions. Admin IDs don't have to be added to `ALLOWED_TELEGRAM_USER_IDS`. **Note**: by default, no admin ('-') |
| `ADMIN_USER_IDS` | Telegram user IDs of admins. These users have access to special admin commands, information and no budget restrictions. Admin IDs don't have to be added to `ALLOWED_TELEGRAM_USER_IDS`. **Note**: by default, no admin (`-`) |
| `ALLOWED_TELEGRAM_USER_IDS` | A comma-separated list of Telegram user IDs that are allowed to interact with the bot (use [getidsbot](https://t.me/getidsbot) to find your user ID). **Note**: by default, *everyone* is allowed (`*`) |
### Optional configuration
The following parameters are optional and can be set in the `.env` file:
#### Budgets
| Parameter | Description | Default value |
|------------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------------|
| `BUDGET_PERIOD` | Determines the time frame all budgets are applied to. Available periods: `daily` *(resets budget every day)*, `monthly` *(resets budgets on the first of each month)*, `all-time` *(never resets budget)*. See the [Budget Manual](https://github.com/n3d1117/chatgpt-telegram-bot/discussions/184) for more information | `monthly` |
| `USER_BUDGETS` | A comma-separated list of $-amounts per user from list `ALLOWED_TELEGRAM_USER_IDS` to set custom usage limit of OpenAI API costs for each. For `*`- user lists the first `USER_BUDGETS` value is given to every user. **Note**: by default, *no limits* for any user (`*`). See the [Budget Manual](https://github.com/n3d1117/chatgpt-telegram-bot/discussions/184) for more information | `*` |
| `GUEST_BUDGET` | $-amount as usage limit for all guest users. Guest users are users in group chats that are not in the `ALLOWED_TELEGRAM_USER_IDS` list. Value is ignored if no usage limits are set in user budgets (`USER_BUDGETS`="*"). See the [Budget Manual](https://github.com/n3d1117/chatgpt-telegram-bot/discussions/184) for more information | `100.0` |
| `TOKEN_PRICE` | $-price per 1000 tokens used to compute cost information in usage statistics (https://openai.com/pricing) | `0.002` |
| `IMAGE_PRICES` | A comma-separated list with 3 elements of prices for the different image sizes: `256x256`, `512x512` and `1024x1024` | `0.016,0.018,0.02` |
| `TRANSCRIPTION_PRICE` | USD-price for one minute of audio transcription | `0.006` |
| Parameter | Description | Default value |
|-----------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------------------|
| `BUDGET_PERIOD` | Determines the time frame all budgets are applied to. Available periods: `daily` *(resets budget every day)*, `monthly` *(resets budgets on the first of each month)*, `all-time` *(never resets budget)*. See the [Budget Manual](https://github.com/n3d1117/chatgpt-telegram-bot/discussions/184) for more information | `monthly` |
| `USER_BUDGETS` | A comma-separated list of $-amounts per user from list `ALLOWED_TELEGRAM_USER_IDS` to set custom usage limit of OpenAI API costs for each. For `*`- user lists the first `USER_BUDGETS` value is given to every user. **Note**: by default, *no limits* for any user (`*`). See the [Budget Manual](https://github.com/n3d1117/chatgpt-telegram-bot/discussions/184) for more information | `*` |
| `GUEST_BUDGET` | $-amount as usage limit for all guest users. Guest users are users in group chats that are not in the `ALLOWED_TELEGRAM_USER_IDS` list. Value is ignored if no usage limits are set in user budgets (`USER_BUDGETS`=`*`). See the [Budget Manual](https://github.com/n3d1117/chatgpt-telegram-bot/discussions/184) for more information | `100.0` |
| `TOKEN_PRICE` | $-price per 1000 tokens used to compute cost information in usage statistics. Source: https://openai.com/pricing | `0.002` |
| `IMAGE_PRICES` | A comma-separated list with 3 elements of prices for the different image sizes: `256x256`, `512x512` and `1024x1024`. Source: https://openai.com/pricing | `0.016,0.018,0.02` |
| `TRANSCRIPTION_PRICE` | USD-price for one minute of audio transcription. Source: https://openai.com/pricing | `0.006` |
Check out the [Budget Manual](https://github.com/n3d1117/chatgpt-telegram-bot/discussions/184) for possible budget configurations.
@@ -87,9 +87,9 @@ Check out the [Budget Manual](https://github.com/n3d1117/chatgpt-telegram-bot/di
| `PRESENCE_PENALTY` | Number between -2.0 and 2.0. Positive values penalize new tokens based on whether they appear in the text so far | `0.0` |
| `FREQUENCY_PENALTY` | Number between -2.0 and 2.0. Positive values penalize new tokens based on their existing frequency in the text so far | `0.0` |
| `IMAGE_SIZE` | The DALL·E generated image size. Allowed values: `256x256`, `512x512` or `1024x1024` | `512x512` |
| `GROUP_TRIGGER_KEYWORD` | If set, the bot in group chats will only respond to messages that start with this keyword | "" |
| `GROUP_TRIGGER_KEYWORD` | If set, the bot in group chats will only respond to messages that start with this keyword | - |
| `IGNORE_GROUP_TRANSCRIPTIONS` | If set to true, the bot will not process transcriptions in group chats | `true` |
| `BOT_LANGUAGE` | Language of general bot messages. Currently available: `en`, `de`, `ru`, `tr`, `it`. **Note** [Contribute additional translations](https://github.com/n3d1117/chatgpt-telegram-bot/discussions/219) | `en` |
| `BOT_LANGUAGE` | Language of general bot messages. Currently available: `en`, `de`, `ru`, `tr`, `it`, `es`. **Note** [Contribute additional translations](https://github.com/n3d1117/chatgpt-telegram-bot/discussions/219) | `en` |
Check out the [official API reference](https://platform.openai.com/docs/api-reference/chat) for more details.

6
bot/main.py
View File

@@ -50,10 +50,10 @@ def main():
# remove support for old budget names at some point in the future
if os.environ.get('MONTHLY_USER_BUDGETS') is not None:
logging.warning('The environment variable MONTHLY_USER_BUDGETS is deprecated. '
'Please use USER_BUDGETS with BUDGET_PERIOD instead.')
'Please use USER_BUDGETS with BUDGET_PERIOD instead.')
if os.environ.get('MONTHLY_GUEST_BUDGET') is not None:
logging.warning('The environment variable MONTHLY_GUEST_BUDGET is deprecated. '
'Please use GUEST_BUDGET with BUDGET_PERIOD instead.')
'Please use GUEST_BUDGET with BUDGET_PERIOD instead.')
telegram_config = {
'token': os.environ['TELEGRAM_BOT_TOKEN'],
@@ -71,7 +71,7 @@ def main():
'ignore_group_transcriptions': os.environ.get('IGNORE_GROUP_TRANSCRIPTIONS', 'true').lower() == 'true',
'group_trigger_keyword': os.environ.get('GROUP_TRIGGER_KEYWORD', ''),
'token_price': float(os.environ.get('TOKEN_PRICE', 0.002)),
'image_prices': [float(i) for i in os.environ.get('IMAGE_PRICES',"0.016,0.018,0.02").split(",")],
'image_prices': [float(i) for i in os.environ.get('IMAGE_PRICES', "0.016,0.018,0.02").split(",")],
'transcription_price': float(os.environ.get('TOKEN_PRICE', 0.006)),
'bot_language': os.environ.get('BOT_LANGUAGE', 'en'),
}

22
bot/openai_helper.py
View File

@@ -26,9 +26,11 @@ def default_max_tokens(model: str) -> int:
"""
return 1200 if model in GPT_3_MODELS else 2400
with open('translations.json', 'r', encoding='utf-8') as f:
translations = json.load(f)
def localized_text(key, bot_language):
"""
Return translated text for a key in specified bot_language.
@@ -46,6 +48,7 @@ def localized_text(key, bot_language):
# return key as text
return key
class OpenAIHelper:
"""
ChatGPT helper class.
@@ -172,7 +175,7 @@ class OpenAIHelper:
frequency_penalty=self.config['frequency_penalty'],
stream=stream
)
except openai.error.RateLimitError as e:
raise Exception(f"⚠️ _{localized_text('openai_rate_limit', bot_language)}._ ⚠️\n{str(e)}") from e
@@ -198,7 +201,10 @@ class OpenAIHelper:
if 'data' not in response or len(response['data']) == 0:
logging.error(f'No response from GPT: {str(response)}')
raise Exception(f"⚠️ _{localized_text('error', bot_language)}._ ⚠️\n{localized_text('try_again', bot_language)}.")
raise Exception(
f"⚠️ _{localized_text('error', bot_language)}._ "
f"⚠️\n{localized_text('try_again', bot_language)}."
)
return response['data'][0]['url'], self.config['image_size']
except Exception as e:
@@ -253,8 +259,8 @@ class OpenAIHelper:
:return: The summary
"""
messages = [
{ "role": "assistant", "content": "Summarize this conversation in 700 characters or less" },
{ "role": "user", "content": str(conversation) }
{"role": "assistant", "content": "Summarize this conversation in 700 characters or less"},
{"role": "user", "content": str(conversation)}
]
response = await openai.ChatCompletion.acreate(
model=self.config['model'],
@@ -281,8 +287,8 @@ class OpenAIHelper:
:param messages: the messages to send
:return: the number of tokens required
"""
model = self.config['model']
try:
model = self.config['model']
encoding = tiktoken.encoding_for_model(model)
except KeyError:
encoding = tiktoken.get_encoding("gpt-3.5-turbo")
@@ -304,7 +310,7 @@ class OpenAIHelper:
num_tokens += tokens_per_name
num_tokens += 3 # every reply is primed with <|start|>assistant<|message|>
return num_tokens
def get_billing_current_month(self):
"""Gets billed usage for current month from OpenAI API.
@@ -324,5 +330,5 @@ class OpenAIHelper:
}
response = requests.get("https://api.openai.com/dashboard/billing/usage", headers=headers, params=params)
billing_data = json.loads(response.text)
usage_month = billing_data["total_usage"] / 100 # convert cent amount to dollars
return usage_month
usage_month = billing_data["total_usage"] / 100 # convert cent amount to dollars
return usage_month

475
bot/telegram_bot.py
View File

@@ -3,32 +3,35 @@ import logging
import os
import itertools
import asyncio
import json
import telegram
from uuid import uuid4
from telegram import constants, BotCommandScopeAllGroupChats
from telegram import Message, MessageEntity, Update, InlineQueryResultArticle, InputTextMessageContent, BotCommand, ChatMember
from telegram import InlineKeyboardMarkup, InlineKeyboardButton, InlineQueryResultArticle
from telegram import Message, MessageEntity, Update, InputTextMessageContent, BotCommand, ChatMember
from telegram.error import RetryAfter, TimedOut
from telegram.ext import ApplicationBuilder, ContextTypes, CommandHandler, MessageHandler, \
filters, InlineQueryHandler, Application, CallbackContext
filters, InlineQueryHandler, CallbackQueryHandler, Application, CallbackContext
from pydub import AudioSegment
from openai_helper import OpenAIHelper, localized_text
from usage_tracker import UsageTracker
def message_text(message: Message) -> str:
"""
Returns the text of a message, excluding any bot commands.
"""
message_text = message.text
if message_text is None:
message_txt = message.text
if message_txt is None:
return ''
for _, text in sorted(message.parse_entities([MessageEntity.BOT_COMMAND]).items(), key=(lambda item: item[0].offset)):
message_text = message_text.replace(text, '').strip()
for _, text in sorted(message.parse_entities([MessageEntity.BOT_COMMAND]).items(),
key=(lambda item: item[0].offset)):
message_txt = message_txt.replace(text, '').strip()
return message_txt if len(message_txt) > 0 else ''
return message_text if len(message_text) > 0 else ''
class ChatGPTTelegramBot:
"""
@@ -36,10 +39,10 @@ class ChatGPTTelegramBot:
"""
# Mapping of budget period to cost period
budget_cost_map = {
"monthly":"cost_month",
"daily":"cost_today",
"all-time":"cost_all_time"
}
"monthly": "cost_month",
"daily": "cost_today",
"all-time": "cost_all_time"
}
def __init__(self, config: dict, openai: OpenAIHelper):
"""
@@ -58,14 +61,16 @@ class ChatGPTTelegramBot:
BotCommand(command='resend', description=localized_text('resend_description', bot_language))
]
self.group_commands = [
BotCommand(command='chat', description=localized_text('chat_description', bot_language))
] + self.commands
BotCommand(command='chat',
description=localized_text('chat_description', bot_language))
] + self.commands
self.disallowed_message = localized_text('disallowed', bot_language)
self.budget_limit_message = localized_text('budget_limit', bot_language)
self.usage = {}
self.last_message = {}
self.inline_queries_cache = {}
async def help(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
async def help(self, update: Update, _: ContextTypes.DEFAULT_TYPE) -> None:
"""
Shows the help menu.
"""
@@ -73,40 +78,39 @@ class ChatGPTTelegramBot:
commands_description = [f'/{command.command} - {command.description}' for command in commands]
bot_language = self.config['bot_language']
help_text = (
localized_text('help_text', bot_language)[0] +
'\n\n' +
'\n'.join(commands_description) +
'\n\n' +
localized_text('help_text', bot_language)[1] +
'\n\n' +
localized_text('help_text', bot_language)[2]
localized_text('help_text', bot_language)[0] +
'\n\n' +
'\n'.join(commands_description) +
'\n\n' +
localized_text('help_text', bot_language)[1] +
'\n\n' +
localized_text('help_text', bot_language)[2]
)
await update.message.reply_text(help_text, disable_web_page_preview=True)
async def stats(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""
Returns token usage statistics for current day and month.
"""
if not await self.is_allowed(update, context):
logging.warning(f'User {update.message.from_user.name} (id: {update.message.from_user.id}) '
f'is not allowed to request their usage statistics')
f'is not allowed to request their usage statistics')
await self.send_disallowed_message(update, context)
return
logging.info(f'User {update.message.from_user.name} (id: {update.message.from_user.id}) '
f'requested their usage statistics')
f'requested their usage statistics')
user_id = update.message.from_user.id
if user_id not in self.usage:
self.usage[user_id] = UsageTracker(user_id, update.message.from_user.name)
tokens_today, tokens_month = self.usage[user_id].get_current_token_usage()
images_today, images_month = self.usage[user_id].get_current_image_count()
(transcribe_minutes_today, transcribe_seconds_today, transcribe_minutes_month,
transcribe_seconds_month) = self.usage[user_id].get_current_transcription_duration()
(transcribe_minutes_today, transcribe_seconds_today, transcribe_minutes_month,
transcribe_seconds_month) = self.usage[user_id].get_current_transcription_duration()
current_cost = self.usage[user_id].get_current_cost()
chat_id = update.effective_chat.id
chat_messages, chat_token_length = self.openai.get_conversation_stats(chat_id)
remaining_budget = self.get_remaining_budget(update)
@@ -136,13 +140,20 @@ class ChatGPTTelegramBot:
)
# text_budget filled with conditional content
text_budget = "\n\n"
budget_period =self.config['budget_period']
budget_period = self.config['budget_period']
if remaining_budget < float('inf'):
text_budget += f"{localized_text('stats_budget', bot_language)}{localized_text(budget_period, bot_language)}: ${remaining_budget:.2f}.\n"
text_budget += (
f"{localized_text('stats_budget', bot_language)}"
f"{localized_text(budget_period, bot_language)}: "
f"${remaining_budget:.2f}.\n"
)
# add OpenAI account information for admin request
if self.is_admin(update):
text_budget += f"{localized_text('stats_openai', bot_language)}{self.openai.get_billing_current_month():.2f}"
if self.is_admin(user_id):
text_budget += (
f"{localized_text('stats_openai', bot_language)}"
f"{self.openai.get_billing_current_month():.2f}"
)
usage_text = text_current_conversation + text_today + text_month + text_budget
await update.message.reply_text(usage_text, parse_mode=constants.ParseMode.MARKDOWN)
@@ -160,7 +171,8 @@ class ChatGPTTelegramBot:
if chat_id not in self.last_message:
logging.warning(f'User {update.message.from_user.name} (id: {update.message.from_user.id})'
f' does not have anything to resend')
await context.bot.send_message(chat_id=chat_id, text=localized_text('resend_failed', self.config['bot_language']))
await context.bot.send_message(chat_id=chat_id,
text=localized_text('resend_failed', self.config['bot_language']))
return
# Update message text, clear self.last_message and send the request to prompt
@@ -177,12 +189,12 @@ class ChatGPTTelegramBot:
"""
if not await self.is_allowed(update, context):
logging.warning(f'User {update.message.from_user.name} (id: {update.message.from_user.id}) '
f'is not allowed to reset the conversation')
f'is not allowed to reset the conversation')
await self.send_disallowed_message(update, context)
return
logging.info(f'Resetting the conversation for user {update.message.from_user.name} '
f'(id: {update.message.from_user.id})...')
f'(id: {update.message.from_user.id})...')
chat_id = update.effective_chat.id
reset_content = message_text(update.message)
@@ -193,17 +205,19 @@ class ChatGPTTelegramBot:
"""
Generates an image for the given prompt using DALL·E APIs
"""
if not self.config['enable_image_generation'] or not await self.check_allowed_and_within_budget(update, context):
if not self.config['enable_image_generation'] or not await self.check_allowed_and_within_budget(update,
context):
return
chat_id = update.effective_chat.id
image_query = message_text(update.message)
if image_query == '':
await context.bot.send_message(chat_id=chat_id, text=localized_text('image_no_prompt', self.config['bot_language']))
await context.bot.send_message(chat_id=chat_id,
text=localized_text('image_no_prompt', self.config['bot_language']))
return
logging.info(f'New image generation request received from user {update.message.from_user.name} '
f'(id: {update.message.from_user.id})')
f'(id: {update.message.from_user.id})')
async def _generate():
try:
@@ -229,7 +243,7 @@ class ChatGPTTelegramBot:
parse_mode=constants.ParseMode.MARKDOWN
)
await self.wrap_with_indicator(update, context, constants.ChatAction.UPLOAD_PHOTO, _generate)
await self.wrap_with_indicator(update, context, _generate, constants.ChatAction.UPLOAD_PHOTO)
async def transcribe(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""
@@ -256,7 +270,10 @@ class ChatGPTTelegramBot:
await context.bot.send_message(
chat_id=chat_id,
reply_to_message_id=self.get_reply_to_message_id(update),
text=f"{localized_text('media_download_fail', bot_language)[0]}: {str(e)}. {localized_text('media_download_fail', bot_language)[1]}",
text=(
f"{localized_text('media_download_fail', bot_language)[0]}: "
f"{str(e)}. {localized_text('media_download_fail', bot_language)[1]}"
),
parse_mode=constants.ParseMode.MARKDOWN
)
return
@@ -266,7 +283,7 @@ class ChatGPTTelegramBot:
audio_track = AudioSegment.from_file(filename)
audio_track.export(filename_mp3, format="mp3")
logging.info(f'New transcribe request received from user {update.message.from_user.name} '
f'(id: {update.message.from_user.id})')
f'(id: {update.message.from_user.id})')
except Exception as e:
logging.exception(e)
@@ -322,7 +339,10 @@ class ChatGPTTelegramBot:
self.usage["guests"].add_chat_tokens(total_tokens, self.config['token_price'])
# Split into chunks of 4096 characters (Telegram's message limit)
transcript_output = f"_{localized_text('transcript', bot_language)}:_\n\"{transcript}\"\n\n_{localized_text('answer', bot_language)}:_\n{response}"
transcript_output = (
f"_{localized_text('transcript', bot_language)}:_\n\"{transcript}\"\n\n"
f"_{localized_text('answer', bot_language)}:_\n{response}"
)
chunks = self.split_into_chunks(transcript_output)
for index, transcript_chunk in enumerate(chunks):
@@ -348,16 +368,20 @@ class ChatGPTTelegramBot:
if os.path.exists(filename):
os.remove(filename)
await self.wrap_with_indicator(update, context, constants.ChatAction.TYPING, _execute)
await self.wrap_with_indicator(update, context, _execute, constants.ChatAction.TYPING)
async def prompt(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""
React to incoming messages and respond accordingly.
"""
if update.edited_message or update.message.via_bot:
return
if not await self.check_allowed_and_within_budget(update, context):
return
logging.info(f'New message received from user {update.message.from_user.name} (id: {update.message.from_user.id})')
logging.info(
f'New message received from user {update.message.from_user.name} (id: {update.message.from_user.id})')
chat_id = update.effective_chat.id
user_id = update.message.from_user.id
prompt = message_text(update.message)
@@ -375,28 +399,30 @@ class ChatGPTTelegramBot:
return
try:
total_tokens = 0
if self.config['stream']:
await context.bot.send_chat_action(chat_id=chat_id, action=constants.ChatAction.TYPING)
is_group_chat = self.is_group_chat(update)
stream_response = self.openai.get_chat_response_stream(chat_id=chat_id, query=prompt)
i = 0
prev = ''
sent_message = None
backoff = 0
chunk = 0
stream_chunk = 0
async for content, tokens in stream_response:
if len(content.strip()) == 0:
continue
chunks = self.split_into_chunks(content)
if len(chunks) > 1:
content = chunks[-1]
if chunk != len(chunks) - 1:
chunk += 1
stream_chunks = self.split_into_chunks(content)
if len(stream_chunks) > 1:
content = stream_chunks[-1]
if stream_chunk != len(stream_chunks) - 1:
stream_chunk += 1
try:
await self.edit_message_with_retry(context, chat_id, sent_message.message_id, chunks[-2])
await self.edit_message_with_retry(context, chat_id, str(sent_message.message_id),
stream_chunks[-2])
except:
pass
try:
@@ -408,12 +434,7 @@ class ChatGPTTelegramBot:
pass
continue
if is_group_chat:
# group chats have stricter flood limits
cutoff = 180 if len(content) > 1000 else 120 if len(content) > 200 else 90 if len(content) > 50 else 50
else:
cutoff = 90 if len(content) > 1000 else 45 if len(content) > 200 else 25 if len(content) > 50 else 15
cutoff = self.get_stream_cutoff_values(update, content)
cutoff += backoff
if i == 0:
@@ -434,7 +455,7 @@ class ChatGPTTelegramBot:
try:
use_markdown = tokens != 'not_finished'
await self.edit_message_with_retry(context, chat_id, sent_message.message_id,
await self.edit_message_with_retry(context, chat_id, str(sent_message.message_id),
text=content, markdown=use_markdown)
except RetryAfter as e:
@@ -458,7 +479,6 @@ class ChatGPTTelegramBot:
total_tokens = int(tokens)
else:
total_tokens = 0
async def _reply():
nonlocal total_tokens
response, total_tokens = await self.openai.get_chat_response(chat_id=chat_id, query=prompt)
@@ -481,21 +501,12 @@ class ChatGPTTelegramBot:
reply_to_message_id=self.get_reply_to_message_id(update) if index == 0 else None,
text=chunk
)
except Exception as e:
raise e
except Exception as exception:
raise exception
await self.wrap_with_indicator(update, context, constants.ChatAction.TYPING, _reply)
await self.wrap_with_indicator(update, context, _reply, constants.ChatAction.TYPING)
try:
# add chat request to users usage tracker
self.usage[user_id].add_chat_tokens(total_tokens, self.config['token_price'])
# add guest chat request to guest usage tracker
allowed_user_ids = self.config['allowed_user_ids'].split(',')
if str(user_id) not in allowed_user_ids and 'guests' in self.usage:
self.usage["guests"].add_chat_tokens(total_tokens, self.config['token_price'])
except Exception as e:
logging.warning(f'Failed to add tokens to usage_logs: {str(e)}')
pass
self.add_chat_request_to_usage_tracker(user_id, total_tokens)
except Exception as e:
logging.exception(e)
@@ -511,24 +522,161 @@ class ChatGPTTelegramBot:
Handle the inline query. This is run when you type: @botusername <query>
"""
query = update.inline_query.query
if query == '':
if len(query) < 3:
return
if not await self.check_allowed_and_within_budget(update, context, is_inline=True):
return
results = [
InlineQueryResultArticle(
id=str(uuid4()),
title='Ask ChatGPT',
input_message_content=InputTextMessageContent(query),
description=query,
thumb_url='https://user-images.githubusercontent.com/11541888/223106202-7576ff11-2c8e-408d-94ea-b02a7a32149a.png'
callback_data_suffix = "gpt:"
result_id = str(uuid4())
self.inline_queries_cache[result_id] = query
callback_data = f'{callback_data_suffix}{result_id}'
await self.send_inline_query_result(update, result_id, message_content=query, callback_data=callback_data)
async def send_inline_query_result(self, update: Update, result_id, message_content, callback_data=""):
try:
reply_markup = None
bot_language = self.config['bot_language']
if callback_data:
reply_markup = InlineKeyboardMarkup([[
InlineKeyboardButton(text=f'🤖 {localized_text("answer_with_chatgpt", bot_language)}',
callback_data=callback_data)
]])
inline_query_result = InlineQueryResultArticle(
id=result_id,
title=localized_text("ask_chatgpt", bot_language),
input_message_content=InputTextMessageContent(message_content),
description=message_content,
thumb_url='https://user-images.githubusercontent.com/11541888/223106202-7576ff11-2c8e-408d-94ea'
'-b02a7a32149a.png',
reply_markup=reply_markup
)
]
await update.inline_query.answer(results)
await update.inline_query.answer([inline_query_result], cache_time=0)
except Exception as e:
logging.error(f'An error occurred while generating the result card for inline query {e}')
async def edit_message_with_retry(self, context: ContextTypes.DEFAULT_TYPE, chat_id: int,
message_id: int, text: str, markdown: bool = True):
async def handle_callback_inline_query(self, update: Update, context: CallbackContext):
callback_data = update.callback_query.data
user_id = update.callback_query.from_user.id
inline_message_id = update.callback_query.inline_message_id
name = update.callback_query.from_user.name
callback_data_suffix = "gpt:"
query = ""
bot_language = self.config['bot_language']
answer_tr = localized_text("answer", bot_language)
loading_tr = localized_text("loading", bot_language)
try:
if callback_data.startswith(callback_data_suffix):
unique_id = callback_data.split(':')[1]
total_tokens = 0
# Retrieve the prompt from the cache
query = self.inline_queries_cache.get(unique_id)
if query:
self.inline_queries_cache.pop(unique_id)
else:
error_message = (
f'{localized_text("error", bot_language)}. '
f'{localized_text("try_again", bot_language)}'
)
await self.edit_message_with_retry(context, chat_id=None, message_id=inline_message_id,
text=f'{query}\n\n_{answer_tr}:_\n{error_message}',
is_inline=True)
return
if self.config['stream']:
stream_response = self.openai.get_chat_response_stream(chat_id=user_id, query=query)
i = 0
prev = ''
sent_message = None
backoff = 0
async for content, tokens in stream_response:
if len(content.strip()) == 0:
continue
cutoff = self.get_stream_cutoff_values(update, content)
cutoff += backoff
if i == 0:
try:
if sent_message is not None:
await self.edit_message_with_retry(context, chat_id=None,
message_id=inline_message_id,
text=f'{query}\n\n{answer_tr}:\n{content}',
is_inline=True)
except:
continue
elif abs(len(content) - len(prev)) > cutoff or tokens != 'not_finished':
prev = content
try:
use_markdown = tokens != 'not_finished'
divider = '_' if use_markdown else ''
text = f'{query}\n\n{divider}{answer_tr}:{divider}\n{content}'
# We only want to send the first 4096 characters. No chunking allowed in inline mode.
text = text[:4096]
await self.edit_message_with_retry(context, chat_id=None, message_id=inline_message_id,
text=text, markdown=use_markdown, is_inline=True)
except RetryAfter as e:
backoff += 5
await asyncio.sleep(e.retry_after)
continue
except TimedOut:
backoff += 5
await asyncio.sleep(0.5)
continue
except Exception:
backoff += 5
continue
await asyncio.sleep(0.01)
i += 1
if tokens != 'not_finished':
total_tokens = int(tokens)
else:
async def _send_inline_query_response():
nonlocal total_tokens
# Edit the current message to indicate that the answer is being processed
await context.bot.edit_message_text(inline_message_id=inline_message_id,
text=f'{query}\n\n_{answer_tr}:_\n{loading_tr}',
parse_mode=constants.ParseMode.MARKDOWN)
logging.info(f'Generating response for inline query by {name}')
response, total_tokens = await self.openai.get_chat_response(chat_id=user_id, query=query)
text_content = f'{query}\n\n_{answer_tr}:_\n{response}'
# We only want to send the first 4096 characters. No chunking allowed in inline mode.
text_content = text_content[:4096]
# Edit the original message with the generated content
await self.edit_message_with_retry(context, chat_id=None, message_id=inline_message_id,
text=text_content, is_inline=True)
await self.wrap_with_indicator(update, context, _send_inline_query_response,
constants.ChatAction.TYPING, is_inline=True)
self.add_chat_request_to_usage_tracker(user_id, total_tokens)
except Exception as e:
logging.error(f'Failed to respond to an inline query via button callback: {e}')
logging.exception(e)
localized_answer = localized_text('chat_fail', self.config['bot_language'])
await self.edit_message_with_retry(context, chat_id=None, message_id=inline_message_id,
text=f"{query}\n\n_{answer_tr}:_\n{localized_answer} {str(e)}",
is_inline=True)
async def edit_message_with_retry(self, context: ContextTypes.DEFAULT_TYPE, chat_id: int | None,
message_id: str, text: str, markdown: bool = True, is_inline: bool = False):
"""
Edit a message with retry logic in case of failure (e.g. broken markdown)
:param context: The context to use
@@ -536,12 +684,14 @@ class ChatGPTTelegramBot:
:param message_id: The message id to edit
:param text: The text to edit the message with
:param markdown: Whether to use markdown parse mode
:param is_inline: Whether the message to edit is an inline message
:return: None
"""
try:
await context.bot.edit_message_text(
chat_id=chat_id,
message_id=message_id,
message_id=int(message_id) if not is_inline else None,
inline_message_id=message_id if is_inline else None,
text=text,
parse_mode=constants.ParseMode.MARKDOWN if markdown else None
)
@@ -562,36 +712,46 @@ class ChatGPTTelegramBot:
logging.warning(str(e))
raise e
async def wrap_with_indicator(self, update: Update, context: CallbackContext, chat_action: constants.ChatAction, coroutine):
async def wrap_with_indicator(self, update: Update, context: CallbackContext, coroutine,
chat_action: constants.ChatAction = "", is_inline=False):
"""
Wraps a coroutine while repeatedly sending a chat action to the user.
"""
task = context.application.create_task(coroutine(), update=update)
while not task.done():
context.application.create_task(update.effective_chat.send_action(chat_action))
if not is_inline:
context.application.create_task(update.effective_chat.send_action(chat_action))
try:
await asyncio.wait_for(asyncio.shield(task), 4.5)
except asyncio.TimeoutError:
pass
async def send_disallowed_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
async def send_disallowed_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE, is_inline=False):
"""
Sends the disallowed message to the user.
"""
await context.bot.send_message(
chat_id=update.effective_chat.id,
text=self.disallowed_message,
disable_web_page_preview=True
)
if not is_inline:
await context.bot.send_message(
chat_id=update.effective_chat.id,
text=self.disallowed_message,
disable_web_page_preview=True
)
else:
result_id = str(uuid4())
await self.send_inline_query_result(update, result_id, message_content=self.disallowed_message)
async def send_budget_reached_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
async def send_budget_reached_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE, is_inline=False):
"""
Sends the budget reached message to the user.
"""
await context.bot.send_message(
chat_id=update.effective_chat.id,
text=self.budget_limit_message
)
if not is_inline:
await context.bot.send_message(
chat_id=update.effective_chat.id,
text=self.budget_limit_message
)
else:
result_id = str(uuid4())
await self.send_inline_query_result(update, result_id, message_content=self.budget_limit_message)
async def error_handler(self, update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
"""
@@ -599,10 +759,24 @@ class ChatGPTTelegramBot:
"""
logging.error(f'Exception while handling an update: {context.error}')
def get_stream_cutoff_values(self, update: Update, content: str) -> int:
"""
Gets the stream cutoff values for the message length
"""
if self.is_group_chat(update):
# group chats have stricter flood limits
return 180 if len(content) > 1000 else 120 if len(content) > 200 else 90 if len(
content) > 50 else 50
else:
return 90 if len(content) > 1000 else 45 if len(content) > 200 else 25 if len(
content) > 50 else 15
def is_group_chat(self, update: Update) -> bool:
"""
Checks if the message was sent from a group chat
"""
if not update.effective_chat:
return False
return update.effective_chat.type in [
constants.ChatType.GROUP,
constants.ChatType.SUPERGROUP
@@ -623,23 +797,23 @@ class ChatGPTTelegramBot:
except Exception as e:
raise e
async def is_allowed(self, update: Update, context: CallbackContext) -> bool:
async def is_allowed(self, update: Update, context: CallbackContext, is_inline=False) -> bool:
"""
Checks if the user is allowed to use the bot.
"""
if self.config['allowed_user_ids'] == '*':
return True
if self.is_admin(update):
user_id = update.inline_query.from_user.id if is_inline else update.message.from_user.id
if self.is_admin(user_id):
return True
name = update.inline_query.from_user.name if is_inline else update.message.from_user.name
allowed_user_ids = self.config['allowed_user_ids'].split(',')
# Check if user is allowed
if str(update.message.from_user.id) in allowed_user_ids:
if str(user_id) in allowed_user_ids:
return True
# Check if it's a group a chat with at least one authorized member
if self.is_group_chat(update):
if not is_inline and self.is_group_chat(update):
admin_user_ids = self.config['admin_user_ids'].split(',')
for user in itertools.chain(allowed_user_ids, admin_user_ids):
if not user.strip():
@@ -647,12 +821,11 @@ class ChatGPTTelegramBot:
if await self.is_user_in_group(update, context, user):
logging.info(f'{user} is a member. Allowing group chat message...')
return True
logging.info(f'Group chat messages from user {update.message.from_user.name} '
f'(id: {update.message.from_user.id}) are not allowed')
logging.info(f'Group chat messages from user {name} '
f'(id: {user_id}) are not allowed')
return False
def is_admin(self, update: Update, log_no_admin=False) -> bool:
def is_admin(self, user_id: int, log_no_admin=False) -> bool:
"""
Checks if the user is the admin of the bot.
The first user in the user list is the admin.
@@ -665,22 +838,22 @@ class ChatGPTTelegramBot:
admin_user_ids = self.config['admin_user_ids'].split(',')
# Check if user is in the admin user list
if str(update.message.from_user.id) in admin_user_ids:
if str(user_id) in admin_user_ids:
return True
return False
def get_user_budget(self, update: Update) -> float | None:
def get_user_budget(self, user_id) -> float | None:
"""
Get the user's budget based on their user ID and the bot configuration.
:param update: Telegram update object
:param user_id: User id
:return: The user's budget as a float, or None if the user is not found in the allowed user list
"""
# no budget restrictions for admins and '*'-budget lists
if self.is_admin(update) or self.config['user_budgets'] == '*':
if self.is_admin(user_id) or self.config['user_budgets'] == '*':
return float('inf')
user_budgets = self.config['user_budgets'].split(',')
if self.config['allowed_user_ids'] == '*':
# same budget for all users, use value in first position of budget list
@@ -689,7 +862,6 @@ class ChatGPTTelegramBot:
'only the first value is used as budget for everyone.')
return float(user_budgets[0])
user_id = update.message.from_user.id
allowed_user_ids = self.config['allowed_user_ids'].split(',')
if str(user_id) in allowed_user_ids:
user_index = allowed_user_ids.index(str(user_id))
@@ -699,18 +871,20 @@ class ChatGPTTelegramBot:
return float(user_budgets[user_index])
return None
def get_remaining_budget(self, update: Update) -> float:
def get_remaining_budget(self, update: Update, is_inline=False) -> float:
"""
Calculate the remaining budget for a user based on their current usage.
:param update: Telegram update object
:param is_inline: Boolean flag for inline queries
:return: The remaining budget for the user as a float
"""
user_id = update.message.from_user.id
user_id = update.inline_query.from_user.id if is_inline else update.message.from_user.id
name = update.inline_query.from_user.name if is_inline else update.message.from_user.name
if user_id not in self.usage:
self.usage[user_id] = UsageTracker(user_id, update.message.from_user.name)
self.usage[user_id] = UsageTracker(user_id, name)
# Get budget for users
user_budget = self.get_user_budget(update)
user_budget = self.get_user_budget(user_id)
budget_period = self.config['budget_period']
if user_budget is not None:
cost = self.usage[user_id].get_current_cost()[self.budget_cost_map[budget_period]]
@@ -722,42 +896,60 @@ class ChatGPTTelegramBot:
cost = self.usage['guests'].get_current_cost()[self.budget_cost_map[budget_period]]
return self.config['guest_budget'] - cost
def is_within_budget(self, update: Update) -> bool:
def is_within_budget(self, update: Update, is_inline=False) -> bool:
"""
Checks if the user reached their usage limit.
Initializes UsageTracker for user and guest when needed.
:param update: Telegram update object
:param is_inline: Boolean flag for inline queries
:return: Boolean indicating if the user has a positive budget
"""
user_id = update.message.from_user.id
user_id = update.inline_query.from_user.id if is_inline else update.message.from_user.id
name = update.inline_query.from_user.name if is_inline else update.message.from_user.name
if user_id not in self.usage:
self.usage[user_id] = UsageTracker(user_id, update.message.from_user.name)
self.usage[user_id] = UsageTracker(user_id, name)
remaining_budget = self.get_remaining_budget(update)
remaining_budget = self.get_remaining_budget(update, is_inline=is_inline)
return remaining_budget > 0
async def check_allowed_and_within_budget(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool:
async def check_allowed_and_within_budget(self, update: Update, context: ContextTypes.DEFAULT_TYPE,
is_inline=False) -> bool:
"""
Checks if the user is allowed to use the bot and if they are within their budget
:param update: Telegram update object
:param context: Telegram context object
:param is_inline: Boolean flag for inline queries
:return: Boolean indicating if the user is allowed to use the bot
"""
if not await self.is_allowed(update, context):
logging.warning(f'User {update.message.from_user.name} (id: {update.message.from_user.id}) '
f'is not allowed to use the bot')
await self.send_disallowed_message(update, context)
return False
name = update.inline_query.from_user.name if is_inline else update.message.from_user.name
user_id = update.inline_query.from_user.id if is_inline else update.message.from_user.id
if not self.is_within_budget(update):
logging.warning(f'User {update.message.from_user.name} (id: {update.message.from_user.id}) '
f'reached their usage limit')
await self.send_budget_reached_message(update, context)
if not await self.is_allowed(update, context, is_inline=is_inline):
logging.warning(f'User {name} (id: {user_id}) '
f'is not allowed to use the bot')
await self.send_disallowed_message(update, context, is_inline)
return False
if not self.is_within_budget(update, is_inline=is_inline):
logging.warning(f'User {name} (id: {user_id}) '
f'reached their usage limit')
await self.send_budget_reached_message(update, context, is_inline)
return False
return True
def add_chat_request_to_usage_tracker(self, user_id, used_tokens):
try:
# add chat request to users usage tracker
self.usage[user_id].add_chat_tokens(used_tokens, self.config['token_price'])
# add guest chat request to guest usage tracker
allowed_user_ids = self.config['allowed_user_ids'].split(',')
if str(user_id) not in allowed_user_ids and 'guests' in self.usage:
self.usage["guests"].add_chat_tokens(used_tokens, self.config['token_price'])
except Exception as e:
logging.warning(f'Failed to add tokens to usage_logs: {str(e)}')
pass
def get_reply_to_message_id(self, update: Update):
"""
Returns the message id of the message to reply to
@@ -808,8 +1000,9 @@ class ChatGPTTelegramBot:
self.transcribe))
application.add_handler(MessageHandler(filters.TEXT & (~filters.COMMAND), self.prompt))
application.add_handler(InlineQueryHandler(self.inline_query, chat_types=[
constants.ChatType.GROUP, constants.ChatType.SUPERGROUP
constants.ChatType.GROUP, constants.ChatType.SUPERGROUP, constants.ChatType.PRIVATE
]))
application.add_handler(CallbackQueryHandler(self.handle_callback_inline_query))
application.add_error_handler(self.error_handler)

102
bot/usage_tracker.py
View File

@@ -3,9 +3,11 @@ import pathlib
import json
from datetime import date
def year_month(date):
def year_month(date_str):
# extract string of year-month from date, eg: '2023-03'
return str(date)[:7]
return str(date_str)[:7]
class UsageTracker:
"""
@@ -67,26 +69,14 @@ class UsageTracker:
# token usage functions:
def add_chat_tokens(self, tokens, tokens_price=0.002):
"""Adds used tokens from a request to a users usage history and updates current cost-
"""Adds used tokens from a request to a users usage history and updates current cost
:param tokens: total tokens used in last request
:param tokens_price: price per 1000 tokens, defaults to 0.002
"""
today = date.today()
last_update = date.fromisoformat(self.usage["current_cost"]["last_update"])
token_cost = round(tokens * tokens_price / 1000, 6)
# add to all_time cost, initialize with calculation of total_cost if key doesn't exist
self.usage["current_cost"]["all_time"] = self.usage["current_cost"].get("all_time", self.initialize_all_time_cost()) + token_cost
# add current cost, update new day
if today == last_update:
self.usage["current_cost"]["day"] += token_cost
self.usage["current_cost"]["month"] += token_cost
else:
if today.month == last_update.month:
self.usage["current_cost"]["month"] += token_cost
else:
self.usage["current_cost"]["month"] = token_cost
self.usage["current_cost"]["day"] = token_cost
self.usage["current_cost"]["last_update"] = str(today)
self.add_current_costs(token_cost)
# update usage_history
if str(today) in self.usage["usage_history"]["chat_tokens"]:
# add token usage to existing date
@@ -94,7 +84,7 @@ class UsageTracker:
else:
# create new entry for current date
self.usage["usage_history"]["chat_tokens"][str(today)] = tokens
# write updated token usage to user file
with open(self.user_file, "w") as outfile:
json.dump(self.usage, outfile)
@@ -109,7 +99,7 @@ class UsageTracker:
usage_day = self.usage["usage_history"]["chat_tokens"][str(today)]
else:
usage_day = 0
month = str(today)[:7] # year-month as string
month = str(today)[:7] # year-month as string
usage_month = 0
for today, tokens in self.usage["usage_history"]["chat_tokens"].items():
if today.startswith(month):
@@ -128,22 +118,8 @@ class UsageTracker:
sizes = ["256x256", "512x512", "1024x1024"]
requested_size = sizes.index(image_size)
image_cost = image_prices[requested_size]
today = date.today()
last_update = date.fromisoformat(self.usage["current_cost"]["last_update"])
# add to all_time cost, initialize with calculation of total_cost if key doesn't exist
self.usage["current_cost"]["all_time"] = self.usage["current_cost"].get("all_time", self.initialize_all_time_cost()) + image_cost
# add current cost, update new day
if today == last_update:
self.usage["current_cost"]["day"] += image_cost
self.usage["current_cost"]["month"] += image_cost
else:
if today.month == last_update.month:
self.usage["current_cost"]["month"] += image_cost
else:
self.usage["current_cost"]["month"] = image_cost
self.usage["current_cost"]["day"] = image_cost
self.usage["current_cost"]["last_update"] = str(today)
self.add_current_costs(image_cost)
# update usage_history
if str(today) in self.usage["usage_history"]["number_images"]:
@@ -153,7 +129,7 @@ class UsageTracker:
# create new entry for current date
self.usage["usage_history"]["number_images"][str(today)] = [0, 0, 0]
self.usage["usage_history"]["number_images"][str(today)][requested_size] += 1
# write updated image number to user file
with open(self.user_file, "w") as outfile:
json.dump(self.usage, outfile)
@@ -163,12 +139,12 @@ class UsageTracker:
:return: total number of images requested per day and per month
"""
today=date.today()
today = date.today()
if str(today) in self.usage["usage_history"]["number_images"]:
usage_day = sum(self.usage["usage_history"]["number_images"][str(today)])
else:
usage_day = 0
month = str(today)[:7] # year-month as string
month = str(today)[:7] # year-month as string
usage_month = 0
for today, images in self.usage["usage_history"]["number_images"].items():
if today.startswith(month):
@@ -179,25 +155,12 @@ class UsageTracker:
def add_transcription_seconds(self, seconds, minute_price=0.006):
"""Adds requested transcription seconds to a users usage history and updates current cost.
:param tokens: total tokens used in last request
:param tokens_price: price per minute transcription, defaults to 0.006
:param seconds: total seconds used in last request
:param minute_price: price per minute transcription, defaults to 0.006
"""
today = date.today()
last_update = date.fromisoformat(self.usage["current_cost"]["last_update"])
transcription_price = round(seconds * minute_price / 60, 2)
# add to all_time cost, initialize with calculation of total_cost if key doesn't exist
self.usage["current_cost"]["all_time"] = self.usage["current_cost"].get("all_time", self.initialize_all_time_cost()) + transcription_price
# add current cost, update new day
if today == last_update:
self.usage["current_cost"]["day"] += transcription_price
self.usage["current_cost"]["month"] += transcription_price
else:
if today.month == last_update.month:
self.usage["current_cost"]["month"] += transcription_price
else:
self.usage["current_cost"]["month"] = transcription_price
self.usage["current_cost"]["day"] = transcription_price
self.usage["current_cost"]["last_update"] = str(today)
self.add_current_costs(transcription_price)
# update usage_history
if str(today) in self.usage["usage_history"]["transcription_seconds"]:
@@ -206,11 +169,30 @@ class UsageTracker:
else:
# create new entry for current date
self.usage["usage_history"]["transcription_seconds"][str(today)] = seconds
# write updated token usage to user file
with open(self.user_file, "w") as outfile:
json.dump(self.usage, outfile)
def add_current_costs(self, request_cost):
today = date.today()
last_update = date.fromisoformat(self.usage["current_cost"]["last_update"])
# add to all_time cost, initialize with calculation of total_cost if key doesn't exist
self.usage["current_cost"]["all_time"] = \
self.usage["current_cost"].get("all_time", self.initialize_all_time_cost()) + request_cost
# add current cost, update new day
if today == last_update:
self.usage["current_cost"]["day"] += request_cost
self.usage["current_cost"]["month"] += request_cost
else:
if today.month == last_update.month:
self.usage["current_cost"]["month"] += request_cost
else:
self.usage["current_cost"]["month"] = request_cost
self.usage["current_cost"]["day"] = request_cost
self.usage["current_cost"]["last_update"] = str(today)
def get_current_transcription_duration(self):
"""Get minutes and seconds of audio transcribed for today and this month.
@@ -221,7 +203,7 @@ class UsageTracker:
seconds_day = self.usage["usage_history"]["transcription_seconds"][str(today)]
else:
seconds_day = 0
month = str(today)[:7] # year-month as string
month = str(today)[:7] # year-month as string
seconds_month = 0
for today, seconds in self.usage["usage_history"]["transcription_seconds"].items():
if today.startswith(month):
@@ -229,7 +211,7 @@ class UsageTracker:
minutes_day, seconds_day = divmod(seconds_day, 60)
minutes_month, seconds_month = divmod(seconds_month, 60)
return int(minutes_day), round(seconds_day, 2), int(minutes_month), round(seconds_month, 2)
# general functions
def get_current_cost(self):
"""Get total USD amount of all requests of the current day and month
@@ -257,18 +239,18 @@ class UsageTracker:
:param tokens_price: price per 1000 tokens, defaults to 0.002
:param image_prices: prices for images of sizes ["256x256", "512x512", "1024x1024"],
defaults to [0.016, 0.018, 0.02]
:param tokens_price: price per minute transcription, defaults to 0.006
:param minute_price: price per minute transcription, defaults to 0.006
:return: total cost of all requests
"""
total_tokens = sum(self.usage['usage_history']['chat_tokens'].values())
token_cost = round(total_tokens * tokens_price / 1000, 6)
total_images = [sum(values) for values in zip(*self.usage['usage_history']['number_images'].values())]
image_prices_list = [float(x) for x in image_prices.split(',')]
image_cost = sum([count * price for count, price in zip(total_images, image_prices_list)])
total_transcription_seconds = sum(self.usage['usage_history']['transcription_seconds'].values())
transcription_cost = round(total_transcription_seconds * minute_price / 60, 2)
all_time_cost = token_cost + transcription_cost + image_cost
return all_time_cost
return all_time_cost

67
translations.json
View File

@@ -36,7 +36,52 @@
"openai_rate_limit":"OpenAI Rate Limit exceeded",
"openai_invalid":"OpenAI Invalid request",
"error":"An error has occurred",
"try_again":"Please try again in a while"
"try_again":"Please try again in a while",
"answer_with_chatgpt":"Answer with ChatGPT",
"ask_chatgpt":"Ask ChatGPT",
"loading":"Loading..."
},
"es": {
"help_description":"Muestra el mensaje de ayuda",
"reset_description":"Reinicia la conversación. Opcionalmente, pasa instrucciones de alto nivel (por ejemplo, /reset Eres un asistente útil)",
"image_description":"Genera una imagen a partir de una sugerencia (por ejemplo, /image gato)",
"stats_description":"Obtén tus estadísticas de uso actuales",
"resend_description":"Reenvía el último mensaje",
"chat_description":"¡Chatea con el bot!",
"disallowed":"Lo siento, no tienes permiso para usar este bot. Puedes revisar el código fuente en https://github.com/n3d1117/chatgpt-telegram-bot",
"budget_limit":"Lo siento, has alcanzado tu límite de uso.",
"help_text":["Soy un bot de ChatGPT, ¡háblame!", "Envíame un mensaje de voz o un archivo y lo transcribiré para ti", "Código abierto en https://github.com/n3d1117/chatgpt-telegram-bot"],
"stats_conversation":["Conversación actual", "mensajes de chat en el historial", "tokens de chat en el historial"],
"usage_today":"Uso hoy",
"usage_month":"Uso este mes",
"stats_tokens":"tokens de chat usados",
"stats_images":"imágenes generadas",
"stats_transcribe":["minutos y", "segundos transcritos"],
"stats_total":"💰 Por un monto total de $",
"stats_budget":"Tu presupuesto restante",
"monthly":" para este mes",
"daily":" para hoy",
"all-time":"",
"stats_openai":"Este mes se facturó $ a tu cuenta de OpenAI",
"resend_failed":"No tienes nada que reenviar",
"reset_done":"¡Listo!",
"image_no_prompt":"¡Por favor proporciona una sugerencia! (por ejemplo, /image gato)",
"image_fail":"No se pudo generar la imagen",
"media_download_fail":["No se pudo descargar el archivo de audio", "Asegúrate de que el archivo no sea demasiado grande. (máx. 20MB)"],
"media_type_fail":"Tipo de archivo no compatible",
"transcript":"Transcripción",
"answer":"Respuesta",
"transcribe_fail":"No se pudo transcribir el texto",
"chat_fail":"No se pudo obtener la respuesta",
"prompt":"sugerencia",
"completion":"completado",
"openai_rate_limit":"Límite de tasa de OpenAI excedido",
"openai_invalid":"Solicitud inválida de OpenAI",
"error":"Ha ocurrido un error",
"try_again":"Por favor, inténtalo de nuevo más tarde",
"answer_with_chatgpt":"Responder con ChatGPT",
"ask_chatgpt":"Preguntar a ChatGPT",
"loading":"Cargando..."
},
"de": {
"help_description":"Zeige die Hilfenachricht",
@@ -75,7 +120,10 @@
"openai_rate_limit":"OpenAI Nutzungslimit überschritten",
"openai_invalid":"OpenAI ungültige Anfrage",
"error":"Ein Fehler ist aufgetreten",
"try_again":"Bitte versuche es später erneut"
"try_again":"Bitte versuche es später erneut",
"answer_with_chatgpt":"Antworte mit ChatGPT",
"ask_chatgpt":"Frage ChatGPT",
"loading":"Lade..."
},
"ru": {
"help_description":"Показать справочное сообщение",
@@ -114,7 +162,10 @@
"openai_rate_limit":"Превышен предел использования OpenAI",
"openai_invalid":"ошибочный запрос OpenAI",
"error":"Произошла ошибка",
"try_again":"Пожалуйста, повторите попытку позже"
"try_again":"Пожалуйста, повторите попытку позже",
"answer_with_chatgpt":"Ответить с помощью ChatGPT",
"ask_chatgpt":"Спросить ChatGPT",
"loading":"Загрузка..."
},
"tr": {
"help_description":"Yardım mesajını göster",
@@ -153,7 +204,10 @@
"openai_rate_limit":"OpenAI maksimum istek limiti aşıldı",
"openai_invalid":"OpenAI Geçersiz istek",
"error":"Bir hata oluştu",
"try_again":"Lütfen birazdan tekrar deneyiniz"
"try_again":"Lütfen birazdan tekrar deneyiniz",
"answer_with_chatgpt":"ChatGPT ile cevapla",
"ask_chatgpt":"ChatGPT'ye sor",
"loading":"Yükleniyor..."
},
"it": {
"help_description":"Mostra il messaggio di aiuto",
@@ -192,7 +246,10 @@
"openai_rate_limit":"Limite massimo di richieste OpenAI raggiunto",
"openai_invalid":"Richiesta OpenAI non valida",
"error":"Si è verificato un errore",
"try_again":"Riprova più tardi"
"try_again":"Riprova più tardi",
"answer_with_chatgpt":"Rispondi con ChatGPT",
"ask_chatgpt":"Chiedi a ChatGPT",
"loading":"Carico..."
},
"id": {
"help_description": "Menampilkan pesan bantuan",