First commit

.env.example

@@ -0,0 +1,3 @@
+TELEGRAM_TOKEN=""
+SLEEP_TIME=900
+TELEGRAM_GROUP_ID=''

.gitignore

@@ -0,0 +1 @@
+.env

docker-compose.yml

@@ -0,0 +1,17 @@
+version: "3"
+
+services:
+  inventtesla:
+    build: inventtesla
+    image: inventtesla:multiuser
+    container_name: inventtesla
+    volumes:
+      - ./data:/app/data
+    restart: unless-stopped
+    environment:
+      - TZ="Europe/Madrid"
+      - TELEGRAM_TOKEN=${TELEGRAM_TOKEN}
+      - SLEEP_TIME=${SLEEP_TIME}
+      - TELEGRAM_GROUP_ID=${TELEGRAM_GROUP_ID}
+    dns:
+      - 8.8.8.8

inventtesla/Dockerfile

@@ -0,0 +1,9 @@
+FROM python:3.11
+
+RUN mkdir /app
+ADD . /app
+RUN pip install -r /app/requirements.txt
+
+WORKDIR /app
+
+CMD [ "python", "/app/inventtesla.py" ]

inventtesla/constants.py

@@ -0,0 +1,8 @@
+import os
+
+TELEGRAM_ESCAPE_CHARACTERS = ['_', '*', '[', ']', '(', ')', '~', '>', '+', '-', '=', '|', '{', '}', '.', '!']
+TELEGRAM_REMOVE_CHARACTERS = ['#']
+TELEGRAM_GROUP_ID = os.getenv("TELEGRAM_GROUP_ID")
+TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN")
+SLEEP_TIME = int(os.getenv("SLEEP_TIME"))
+MODELS = ["m3", "my"]

inventtesla/helpers.py

@@ -0,0 +1,94 @@
+import time
+import requests
+import logging
+import constants
+
+from telegram import InlineKeyboardButton, InlineKeyboardMarkup
+from telegram.ext import Application
+from telegram.constants import ParseMode
+
+# Enable logging
+logging.basicConfig(
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO
+)
+
+logger = logging.getLogger(__name__)
+
+def telegram_escape_characters(text):
+    for character in constants.TELEGRAM_ESCAPE_CHARACTERS:
+        try:
+            text = text.replace(character, f'\\{character}')
+        except:
+            pass
+    for character in constants.TELEGRAM_REMOVE_CHARACTERS:
+        try:
+            text = text.replace(character, '')
+        except:
+            pass
+    return text
+
+def get_inventory(model="m3"):
+    ret_json = False
+    headers = {'authority': 'www.tesla.com'}
+    url = f"https://www.tesla.com/inventory/api/v1/inventory-results?query=%7B%22query%22%3A%7B%22model%22%3A%22{model}%22%2C%22condition%22%3A%22new%22%2C%22options%22%3A%7B%7D%2C%22arrangeby%22%3A%22Price%22%2C%22order%22%3A%22asc%22%2C%22market%22%3A%22ES%22%2C%22language%22%3A%22es%22%2C%22super_region%22%3A%22north%20america%22%2C%22lng%22%3A2.1449%2C%22lat%22%3A41.4888%2C%22zip%22%3A%2208290%22%2C%22range%22%3A0%2C%22region%22%3A%22ES%22%7D%2C%22offset%22%3A0%2C%22count%22%3A50%2C%22outsideOffset%22%3A0%2C%22outsideSearch%22%3Afalse%7D"
+    for _ in range(10):
+        try:
+            response = requests.get(url, headers=headers, timeout=10)
+            if response.status_code == 200:
+                try:
+                    ret_json = response.json()['results']
+                    break
+                except Exception as e:
+                    logging.error(f"Something went wrong: {e} Response status code: {response.status_code} Response content: {response.content}")
+        except Exception as e:
+            logging.error(f"Error requesting inventory: {e}")
+        time.sleep(5)
+    
+    return ret_json
+
+def get_image(car_options, model):
+    logging.info(f"Car options: {car_options}")
+    photo_url = f"https://static-assets.tesla.com/configurator/compositor?&bkba_opt=1&view=STUD_3QTR&size=1000&model={model}&options={car_options}&crop=1100,650,400,230&"
+    return photo_url
+
+def check_inventory(new_inventory, old_inventory):
+    new_deals = []
+    for new_car in new_inventory:
+        found = False
+        for old_car in old_inventory:
+            if new_car['VIN'] == old_car['VIN']:
+                found = True
+                break
+        if not found:
+            new_deals.append(new_car)
+    return new_deals
+
+async def send_deal(car, model):
+    application = Application.builder().get_updates_http_version('1.1').http_version('1.1').token(constants.TELEGRAM_TOKEN).build()
+    title = telegram_escape_characters(car['TrimName'])
+    price = f"*💰 Precio:* {telegram_escape_characters(car['Price'])}€"
+    odometer = f"*Cuentakilómetros:* {car['Odometer']}{car['OdometerType']}"
+
+    options = "*Opciones:*"
+    for option in car['OptionCodeSpecs']['C_OPTS']['options']:
+        options = f"{options}\n• {telegram_escape_characters(option['name'])}"
+
+    specs = "*Especificaciones:*"
+    for spec in car['OptionCodeSpecs']['C_SPECS']['options']:
+        specs = f"{specs}\n• {telegram_escape_characters(spec['description'])}: {telegram_escape_characters(spec['name'])}"
+
+    callouts = "*Extras:*"
+    for callout in car['OptionCodeSpecs']['C_CALLOUTS']['options']:
+        callouts = f"{callouts}\n• {telegram_escape_characters(callout['name'])}"
+
+    message = f"{title}\n\n{odometer}\n\n{options}\n\n{specs}\n\n{callouts}\n\n{price}"
+
+    image = get_image(car['OptionCodeList'], model)
+   
+    keyboard = [[InlineKeyboardButton("Ir a la web", url=f"https://www.tesla.com/es_ES/{model}/order/{car['VIN']}")]]
+    markup = InlineKeyboardMarkup(keyboard)
+    await application.bot.send_photo(chat_id=constants.TELEGRAM_GROUP_ID, photo=image, caption=message, parse_mode=ParseMode.MARKDOWN_V2, reply_markup=markup)
+
+async def send_message(telegram_user_id, message):
+    application = Application.builder().get_updates_http_version('1.1').http_version('1.1').token(constants.TELEGRAM_TOKEN).build()
+    await application.bot.send_message(chat_id=telegram_user_id, text=message)

inventtesla/inventtesla.py

@@ -0,0 +1,28 @@
+import threading
+import logging
+import constants
+import inventtesla_checker
+
+from telegram.ext import (
+    Application
+)
+
+# Enable logging
+logging.basicConfig(
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO
+)
+
+logger = logging.getLogger(__name__)
+
+def main()->None:
+    p = threading.Thread(target=inventtesla_checker.inventtesla_checker, args=(constants.SLEEP_TIME, ))
+    p.start()
+
+    """Start the bot."""
+    # Create the Application and pass it your bot's token.
+    application = Application.builder().get_updates_http_version('1.1').http_version('1.1').token(constants.TELEGRAM_TOKEN).build()
+
+    application.run_polling()
+
+if __name__ == "__main__":
+    main()

inventtesla/inventtesla_checker.py

@@ -0,0 +1,39 @@
+import asyncio
+import logging
+import helpers
+import constants
+
+# Enable logging
+logging.basicConfig(
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO
+)
+
+async def work(sleep_time):
+    inventories = {}
+    for model in constants.MODELS:
+        inventories[model] = helpers.get_inventory(model)
+        if type(inventories[model]) is list:
+            inventories[model].pop(0)
+        else:
+            inventories[model] = list()
+    while True:
+        for model in constants.MODELS:
+            logging.info(f"Checking for new {model} Teslas")
+            new_inventory = helpers.get_inventory(model)
+            if type(new_inventory) is list:
+                new_deals = helpers.check_inventory(new_inventory, inventories[model])
+                inventories[model] = new_inventory
+
+                for deal in new_deals:
+                    await helpers.send_deal(deal, model)
+
+        await asyncio.sleep(sleep_time)
+
+def inventtesla_checker(sleep_time):
+    logging.info(f"Tesla checker starting... Checking every {sleep_time} seconds")
+    while True:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        loop.run_until_complete(work(sleep_time))
+        loop.close()
+        

inventtesla/requirements.txt

@@ -0,0 +1,4 @@
+python-telegram-bot==20.1
+python-telegram-bot[job-queue]==20.1
+requests==2.28.1
+Pillow==9.4.0