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

refactored token_usage to general usage_tracker class that allows for implementation of audio and image usage tracking

Browse Source
This commit is contained in:
AlexHTW
2023-03-12 00:27:37 +01:00
parent 48d303b775
commit 896b09feb7
4 changed files with 179 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -2,4 +2,4 @@
/.idea
.env
.DS_Store
/token_usage
/usage_logs

31
telegram_bot.py
View File

@@ -8,7 +8,7 @@ from telegram.ext import ApplicationBuilder, ContextTypes, CommandHandler, Messa
from pydub import AudioSegment
from openai_helper import OpenAIHelper
import token_usage
from usage_tracker import UsageTracker
class ChatGPT3TelegramBot:
@@ -32,6 +32,7 @@ class ChatGPT3TelegramBot:
]
self.disallowed_message = "Sorry, you are not allowed to use this bot. You can check out the source code at " \
"https://github.com/n3d1117/chatgpt-telegram-bot"
self.usage = {}
async def help(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""
@@ -59,10 +60,13 @@ class ChatGPT3TelegramBot:
logging.info(f'User {update.message.from_user.name} requested their token usage statistics')
tokens_today, tokens_month = token_usage.get_token_usage(update.message.from_user.id)
cost_today = token_usage.cost_tokens(tokens_today, self.config['token_price'])
cost_month = token_usage.cost_tokens(tokens_month, self.config['token_price'])
user_id = update.message.from_user.id
if user_id not in self.usage:
self.usage[user_id] = UsageTracker(user_id)
tokens_today, tokens_month, cost_today, cost_month = self.usage[
user_id].get_token_count_and_cost(token_price=self.config['token_price'])
usage_text = f"Today you used {tokens_today} tokens (worth ${cost_today:.3f})."+\
f"\nThis month you used {tokens_month} tokens (worth ${cost_month:.3f})."
await update.message.reply_text(usage_text)
@@ -143,6 +147,16 @@ class ChatGPT3TelegramBot:
)
return
logging.info(f'New transcribe request received from user {update.message.from_user.name}')
user_id = update.message.from_user.id
if user_id not in self.usage:
self.usage[user_id] = UsageTracker(user_id)
chat_id = update.effective_chat.id
await context.bot.send_chat_action(chat_id=chat_id, action=constants.ChatAction.TYPING)
filename = update.message.voice.file_unique_id if update.message.voice else update.message.audio.file_unique_id
filename_ogg = f'{filename}.ogg'
filename_mp3 = f'{filename}.mp3'
try:
@@ -174,7 +188,7 @@ class ChatGPT3TelegramBot:
else:
# Send the response of the transcript
response, total_tokens = self.openai.get_chat_response(chat_id=chat_id, query=transcript)
token_usage.add_token_usage(update.message.from_user.id, total_tokens)
self.usage[user_id].add_chat_tokens(total_tokens)
await context.bot.send_message(
chat_id=chat_id,
reply_to_message_id=update.message.message_id,
@@ -209,7 +223,12 @@ class ChatGPT3TelegramBot:
await context.bot.send_chat_action(chat_id=chat_id, action=constants.ChatAction.TYPING)
response, total_tokens = self.openai.get_chat_response(chat_id=chat_id, query=update.message.text)
token_usage.add_token_usage(update.message.from_user.id, total_tokens)
user_id = update.message.from_user.id
if user_id not in self.usage:
self.usage[user_id] = UsageTracker(user_id)
self.usage[user_id].add_chat_tokens(total_tokens)
await context.bot.send_message(
chat_id=chat_id,
reply_to_message_id=update.message.message_id,

82
token_usage.py
View File

@@ -1,82 +0,0 @@
import os.path
import pathlib
import json
from datetime import date
from itertools import chain
def year_month(date):
# extract string of year-month from date, eg: '2023-03'
return str(date)[:7]
def add_token_usage(user_id, tokens):
"""
Adds used tokens from a request to a users usage file.
Enables tracking of daily/monthly token usage per user.
User files are stored as JSON in /token_usage directory.
JSON schema:
{'year-month':{'day': [tokens, tokens, ...], ...}, ...}
:param user_id: Telegram user ID
:param tokens: total tokens used in last request
"""
# path to usage file of given user
user_file = f"token_usage/{user_id}.json"
# current year-month as string
month = year_month(date.today())
# current day as string, no leading zero
day = str(date.today().day)
if os.path.isfile(user_file):
with open(user_file, "r") as infile:
usage = json.load(infile)
if month in usage:
if day in usage[month]:
# add token usage to current month and day
usage[month][day].append(tokens)
else:
# create new entry for current day
usage[month][day] = [tokens]
else:
# create new entry for current month and day
usage[month] = {day: [tokens]}
else:
# ensure directory exists
pathlib.Path("token_usage").mkdir(exist_ok=True)
# create new dictionary for this user and add used tokens
usage = {month: {day: [tokens]}}
# write updated token usage to user file
with open(user_file, "w") as outfile:
json.dump(usage, outfile)
def get_token_usage(user_id, date=date.today()):
"""
Sums tokens used per day and per month of given date.
Returns both values.
:param user_id: Telegram user ID
:param date: datetime.date object, default today
"""
# path to usage file of given user
user_file = f"token_usage/{user_id}.json"
# year-month as string
month = year_month(date)
# day as string, no leading zero
day = str(date.day)
if os.path.isfile(user_file):
with open(user_file, "r") as infile:
usage = json.load(infile)
usage_day = sum(usage[month][day])
usage_month = sum(chain.from_iterable(list(usage[month].values())))
return usage_day, usage_month
else:
return 0, 0
def cost_tokens(tokens, price_1k_tokens=0.002):
"""
cost of token amount in USD
current price gpt-3.5-turbo: $0.002/1000 tokens
:param tokens: number of tokens
:param price_1k_tokens: price of 1000 tokens (https://openai.com/pricing)
"""
return tokens*(price_1k_tokens/1000)

153
usage_tracker.py Normal file
View File

@@ -0,0 +1,153 @@
import os.path
import pathlib
import json
from datetime import date
def year_month(date):
# extract string of year-month from date, eg: '2023-03'
return str(date)[:7]
class UsageTracker:
"""
UsageTracker class
Enables tracking of daily/monthly token usage per user.
User files are stored as JSON in /token_usage directory.
JSON schema:
{"chat_tokens":{year-month:{day: total_tokens_used}, ...},
{"audio_seconds":{year-month:{day: total_seconds_transcribed}, ...},
{"image_count":{year-month:{day: [nr_images_small, nr_images_medium, nr_images_large]}, ...}}
"""
def __init__(self, user_id, logs_dir="usage_logs"):
"""
Initializes UsageTracker for a user with current date.
Loads usage data from usage log file.
:param user_id: Telegram ID of the user
:param logs_dir: path to directory of usage logs, default "usage_logs"
"""
self.user_id = user_id
self.logs_dir = logs_dir
# path to usage file of given user
self.user_file = f"{logs_dir}/{user_id}.json"
# current year-month as string
self.current_month = year_month(date.today())
# current day as string, no leading zero
self.current_day = str(date.today().day)
if os.path.isfile(self.user_file):
with open(self.user_file, "r") as file:
self.usage = json.load(file)
else:
# ensure directory exists
pathlib.Path(logs_dir).mkdir(exist_ok=True)
# create empty dictionary for this user
self.usage = {}
# token usage functions:
def add_chat_tokens(self, tokens):
"""
Adds used tokens from a request to a users usage .
:param tokens: total tokens used in last request
"""
if "chat_tokens" in self.usage:
if self.current_month in self.usage["chat_tokens"]:
if self.current_day in self.usage["chat_tokens"][self.current_month]:
# add token usage to existing month and day
self.usage["chat_tokens"][self.current_month][self.current_day] += tokens
else:
# create new entry for current day
self.usage["chat_tokens"][self.current_month][self.current_day] = tokens
else:
# create new entry for current month and day
self.usage["chat_tokens"][self.current_month] = {self.current_day: tokens}
else: # add chat_tokens key and token usage for current month and day
self.usage["chat_tokens"] = {self.current_month: {self.current_day: tokens}}
# write updated token usage to user file
with open(self.user_file, "w") as outfile:
json.dump(self.usage, outfile)
def get_token_usage(self, date=date.today()):
"""
Gets tokens used per day and sums tokens per month of given date.
Returns both values.
:param date: datetime.date object, default today
"""
# year-month as string
month = year_month(date)
# day as string, no leading zero
day = str(date.day)
usage_day = self.usage["chat_tokens"][month][day]
usage_month = sum(list(self.usage["chat_tokens"][month].values()))
return usage_day, usage_month
@staticmethod
def cost_tokens(tokens, token_price=0.002):
# cost of token amount in USD
# current price gpt-3.5-turbo: $0.002/1000 tokens
price_per_token = token_price*0.001
return tokens * price_per_token
def get_token_count_and_cost(self, date=date.today(), token_price=0.002):
"""
Gets total cost of tokens used per day and per month of given date.
:param date: datetime.date object, default today
:param token_price: price of 1000 tokens
:returns: 4 values (token count day, token count month, token cost day,
token cost month)
"""
tokens_day, tokens_month = self.get_token_usage(date)
cost_day = self.cost_tokens(tokens_day, token_price)
cost_month = self.cost_tokens(tokens_month, token_price)
return tokens_day, tokens_month, cost_day, cost_month
# transcription usage functions:
def add_audio_seconds(self, seconds):
# TODO: implement
pass
def get_transcription_usage(self, date=date.today()):
# TODO: implement
pass
@staticmethod
def cost_tokens(seconds, minute_price=0.006):
# cost of audio seconds transcribed, amount in USD
# current price Whisper: $0.002/1000 tokens
second_price = minute_price/60
return seconds * second_price
def get_audio_seconds_and_cost(self, date=date.today(), minute_price=0.006):
# TODO: implement
pass
# image usage functions:
def add_image_request(self, seconds):
# TODO: implement
pass
def get_image_count(self, date=date.today()):
# TODO: implement
pass
@staticmethod
def cost_images(image_counts, image_prices=[0.016, 0.018, 0.02]):
# TODO: implement
pass
def get_image_counts_and_costs(self, date=date.today(), image_prices=[0.016, 0.018, 0.02]):
# TODO: implement
pass
# general functions
def get_all_stats(self, date=date.today(), token_price=0.002, minute_price=0.006,
image_prices=[0.016, 0.018, 0.02]):
# TODO: implement
pass
def summarize_past_daily_usage(self):
# TODO: implement
pass