Jocaru/wallamanta
1
0
Fork 0
Code Issues 15 Packages Projects Releases Wiki Activity

Modified to be used in homelabs vip

Browse Source
This commit is contained in:
Joan
2024-03-22 10:48:14 +01:00
parent dfbe346cb2
commit b8c7aaf126
9 changed files with 303 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
wallamanta/wallamanta.py
View File

@@ -26,6 +26,10 @@ ACTION, ADD_PRODUCT_NAME, ADD_PRODUCT_MIN_PRICE, ADD_PRODUCT_MAX_PRICE, \
ADD_PRODUCT_CATEGORY, ADD_PRODUCT_TITLE_EXCLUDE, ADD_PRODUCT_DESCRIPTION_EXCLUDE, \
ADD_PRODUCT_COORDS, ADD_PRODUCT_DISTANCE, REMOVE_PRODUCT, LIST, FINISH, CONTINUE_OR_FINISH = range(13)
MAX_PRODUCTS_TESTING = 5
MAX_PRODUCTS_PREMIUM = 30
MAX_PRODUCTS_HOMELABS = 5
# Enable logging
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO
@@ -37,10 +41,10 @@ httpx_logger = logging.getLogger('httpx')
httpx_logger.setLevel(logging.WARNING)
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if walladb.is_user_valid(helpers.get_telegram_user_id(update)):
if walladb.is_user_active(helpers.get_telegram_user_id(update)):
message = "Utiliza el menú de la conversación para añadir un producto y sigue los pasos indicados. Si tienes cualquier duda contacta con @jocarduck para más información."
else:
message = "Activa tu periodo de prueba de 7 días con `/test` o contacta con @jocarduck para más información."
message = "No tienes permitido usar este bot."
await update.message.reply_markdown_v2(helpers.telegram_escape_characters(message))
async def main_menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
@@ -58,21 +62,27 @@ async def main_menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
async def menu_click_handler(update: Update, context: CallbackContext):
await update.callback_query.edit_message_reply_markup(None)
telegram_user_id = helpers.get_telegram_user_id(update)
if walladb.is_user_valid(telegram_user_id):
if walladb.is_user_active(telegram_user_id):
context.user_data['telegram_user_id'] = telegram_user_id
query = update.callback_query
number_of_products = walladb.count_user_products(telegram_user_id)
if query.data == 'add':
valid = False
number_of_products = walladb.count_user_products(telegram_user_id)
if walladb.is_user_testing(telegram_user_id):
valid = True
if number_of_products >= 5:
message = "Ya tienes 5 productos en seguimiento. Con premium puedes tener hasta 20."
if number_of_products >= MAX_PRODUCTS_TESTING:
message = f"Ya tienes {MAX_PRODUCTS_TESTING} productos en seguimiento. Borra algunos para añadir más."
valid = False
elif walladb.is_user_premium(telegram_user_id):
valid = True
if number_of_products >= 20:
message = "Ya tienes 20 productos en seguimiento. Borra algunos para añadir más."
if number_of_products >= MAX_PRODUCTS_PREMIUM:
message = f"Ya tienes {MAX_PRODUCTS_PREMIUM} productos en seguimiento. Borra algunos para añadir más."
valid = False
elif walladb.is_user_homelabs(telegram_user_id):
valid = True
if number_of_products >= MAX_PRODUCTS_HOMELABS:
message = f"Ya tienes {MAX_PRODUCTS_HOMELABS} productos en seguimiento. Borra algunos para añadir más."
valid = False
if valid:
await context.bot.send_message(chat_id=update.effective_chat.id,
@@ -82,14 +92,22 @@ async def menu_click_handler(update: Update, context: CallbackContext):
await context.bot.send_message(chat_id=update.effective_chat.id, text=f'{message}')
return ConversationHandler.END
if query.data == 'remove':
await send_list(update, context)
return REMOVE_PRODUCT
if number_of_products != 0:
await send_list(update, context)
return REMOVE_PRODUCT
message = "No tienes ninguna búsqueda activa"
await context.bot.send_message(chat_id=update.effective_chat.id, text=f'{message}')
return ConversationHandler.END
if query.data == 'list':
await send_list(update, context)
return LIST
if number_of_products != 0:
await send_list(update, context)
return LIST
message = "No tienes ninguna búsqueda activa"
await context.bot.send_message(chat_id=update.effective_chat.id, text=f'{message}')
return ConversationHandler.END
else:
await context.bot.send_message(chat_id=update.effective_chat.id,
text=helpers.telegram_escape_characters('Activa tu periodo de prueba de 7 días con `/test` o contacta con @jocarduck para más información.'), parse_mode=ParseMode.MARKDOWN_V2)
text=helpers.telegram_escape_characters('No tienes permiso para usar este bot.'), parse_mode=ParseMode.MARKDOWN_V2)
return ConversationHandler.END
async def add_product_name(update: Update, context: CallbackContext):
@@ -340,7 +358,7 @@ async def product_details(update: Update, context: ContextTypes.DEFAULT_TYPE) ->
async def send_list(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
telegram_user_id = helpers.get_telegram_user_id(update)
if walladb.is_user_valid(telegram_user_id):
if walladb.is_user_active(telegram_user_id):
if walladb.is_user_testing(telegram_user_id) or walladb.is_user_premium(telegram_user_id):
query = update.callback_query
if query.data == 'remove':
@@ -373,28 +391,21 @@ async def remove_user_command(update: Update, context: ContextTypes.DEFAULT_TYPE
async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
telegram_user_id = helpers.get_telegram_user_id(update)
if walladb.is_user_valid(telegram_user_id):
if walladb.is_user_active(telegram_user_id):
type = walladb.get_user_type(telegram_user_id)
until = walladb.get_user_until(telegram_user_id)
message = f"Tu cuenta es tipo: {type} y caduca el {helpers.get_spanish_date(until)}."
await update.message.reply_markdown_v2(helpers.telegram_escape_characters(message))
async def test_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
telegram_user_id = helpers.get_telegram_user_id(update)
telegram_user_name = helpers.get_telegram_user_name(update)
#if not walladb.is_user_valid(telegram_user_id):
if telegram_user_id < 0:
message = "Este bot no se puede usar en grupos."
elif walladb.get_user(telegram_user_id) == "NoName":
until = helpers.get_date_ahead(7)
walladb.add_test_user(telegram_user_id, telegram_user_name, until)
message = f"Periodo de prueba activado hasta el {helpers.get_spanish_date(until)}."
else:
message = "Ya has utilizado el periodo de prueba."
if walladb.is_user_testing(telegram_user_id):
message = "Ya estás en el periodo de prueba."
elif walladb.is_user_premium(telegram_user_id):
message = "Ya eres premium. No puedes volver al periodo de prueba."
walladb.enable_user(telegram_user_id, telegram_user_name)
message = f"Bienvenido a @wallamanta, usa el comando /help para más información."
await update.message.reply_markdown_v2(helpers.telegram_escape_characters(message))
async def add_to_db_and_send(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
@@ -428,6 +439,36 @@ async def list_threads(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No
else:
await update.message.reply_markdown_v2(helpers.telegram_escape_characters(f"{threads_string}"))
async def message_to_all(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
if helpers.is_user_admin(update.message.chat_id):
message = update.message.text.split('/message_to_all ')[1]
helpers.send_message_to_all(message)
await update.message.reply_markdown_v2(helpers.telegram_escape_characters(f"Messages sent to all users"))
async def code(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
code = update.message.text.split('/code ')[1]
telegram_user_id = update.message.chat_id
if len(code) != 12:
await update.message.reply_markdown_v2(helpers.telegram_escape_characters(f"El código no tiene el formato correcto"))
else:
code = walladb.get_code(code)
if code:
until = helpers.get_date_ahead(code['days'])
try:
if code['type'] == 'premium':
walladb.add_premium_user(telegram_user_id, until)
await update.message.reply_markdown_v2(helpers.telegram_escape_characters(f"Activado periodo premium hasta el {until}"))
elif code['type'] == 'testing':
walladb.add_test_user(telegram_user_id, until)
await update.message.reply_markdown_v2(helpers.telegram_escape_characters(f"Activado periodo testing hasta el {until}"))
walladb.use_code(code)
except Exception as e:
error_code = helpers.get_random_string(8)
logging.info(f"Error trying to checkout code {code}: {e}. {error_code}")
await update.message.reply_markdown_v2(helpers.telegram_escape_characters(f"Ha habido un error, contacta con @jocarduck y dale el siguiente código de error: `{error_code}`"))
else:
await update.message.reply_markdown_v2(helpers.telegram_escape_characters(f"El código no es válido o ya ha sido usado"))
def count_threads():
time.sleep(10)
while True:
@@ -447,17 +488,19 @@ def main()->None:
p = threading.Thread(target=account_checker.work, args=(3600, ))
p.start()
p = threading.Thread(target=count_threads)
p.start()
#p = threading.Thread(target=count_threads)
#p.start()
# on different commands - answer in Telegram
application.add_handler(CommandHandler("start", start_command))
application.add_handler(CommandHandler("help", help_command))
application.add_handler(CommandHandler("admin", admin_command))
application.add_handler(CommandHandler("add_premium_user", add_premium_user_command))
application.add_handler(CommandHandler("remove_user", remove_user_command))
application.add_handler(CommandHandler("status", status_command))
application.add_handler(CommandHandler("test", test_command))
application.add_handler(CommandHandler("list_threads", list_threads))
application.add_handler(CommandHandler("message_to_all", message_to_all))
application.add_handler(CommandHandler("code", code))
#application.add_handler(CallbackQueryHandler("list", send_list()))
#application.add_handler(CallbackQueryHandler(pattern="list", callback=send_list()))