83 lines
3.8 KiB
Python
83 lines
3.8 KiB
Python
import os
|
|
import requests
|
|
from apps.bitwarden.utils import bitwardenUtils
|
|
from .models import TelegramData
|
|
from datetime import datetime, timedelta
|
|
|
|
class telegramUtils():
|
|
|
|
token_uid = "a88dbe52-b862-422a-a116-b479009733ab"
|
|
chat_id_uid = "13e948ab-494a-4b09-bb3a-b47900974cef"
|
|
|
|
__bitwarden_access_token = os.environ.get('bitwarden_access_token')
|
|
__bitwarden_organization_id = os.environ.get('bitwarden_organization_id')
|
|
|
|
bw = bitwardenUtils(__bitwarden_access_token,__bitwarden_organization_id)
|
|
|
|
__TOKEN = bw.read_value_secret(token_uid)
|
|
__CHAT_ID = bw.read_value_secret(chat_id_uid)
|
|
|
|
|
|
|
|
def get_chat_id(self):
|
|
# Source - https://stackoverflow.com/a/75118239
|
|
# Posted by Yasantha Biyuranga, modified by community. See post 'Timeline' for change history
|
|
# Retrieved 2026-06-30, License - CC BY-SA 4.0
|
|
|
|
url = f"https://api.telegram.org/bot{self.__TOKEN}/getUpdates"
|
|
print(requests.get(url).json())
|
|
return 0
|
|
|
|
def send_message(self,message):
|
|
# Source - https://stackoverflow.com/a/75118239
|
|
# Posted by Yasantha Biyuranga, modified by community. See post 'Timeline' for change history
|
|
# Retrieved 2026-06-30, License - CC BY-SA 4.0
|
|
|
|
url = f"https://api.telegram.org/bot{self.__TOKEN}/sendMessage?chat_id={self.__CHAT_ID}&text={message}"
|
|
|
|
result = requests.get(url).json() # this sends the message
|
|
send_result = result['ok']
|
|
TelegramData.objects.create(message_id=result['result']['message_id'],chat_id=self.__CHAT_ID,description=message[0:99],creation_date=datetime.now(),expiration_date=datetime.now()+timedelta(days=1))
|
|
return send_result
|
|
|
|
def send_document(self,file,message=''):
|
|
url = 'https://api.telegram.org/bot{}/sendDocument'.format(self.__TOKEN)
|
|
data = {'chat_id': self.__CHAT_ID ,'caption': message}
|
|
document = open(file, 'rb')
|
|
files = {'document': document}
|
|
response = requests.post(url, data=data, files=files)
|
|
result = response.json()
|
|
|
|
#Create data , for the dercsription use a message of set , or use the filename , cut it to grab the end if the filename is more then 99 characters
|
|
TelegramData.objects.create(message_id=result['result']['message_id'],chat_id=self.__CHAT_ID,description=message[0:99] or str(file)[-99:],creation_date=datetime.now(),expiration_date=datetime.now()+timedelta(days=1))
|
|
return result['ok'],result['result']['message_id'],result
|
|
|
|
#example send_document('apps/Le_journal_10h00_du_30_06_2026.mp3')
|
|
|
|
def delete_message(self,message_id_list):
|
|
send_result = []
|
|
for message_id in message_id_list:
|
|
url = f"https://api.telegram.org/bot{self.__TOKEN}/deleteMessage?chat_id={self.__CHAT_ID}&message_id={message_id}"
|
|
|
|
result = requests.get(url).json()
|
|
send_result.append(["Message id: "+str(message_id),result['ok']])
|
|
if result['ok']:
|
|
message = TelegramData.objects.filter(message_id=message_id)
|
|
message.delete()
|
|
return send_result
|
|
|
|
def clear_expired_messages(self):
|
|
yesterday=datetime.now()+timedelta(days=-1)
|
|
#filter all the messages that expired yesterday or before
|
|
expired_messages_list = []
|
|
expired_messages = TelegramData.objects.filter(expiration_date__lte=yesterday)
|
|
|
|
#create a list with only the message_id
|
|
for expired_message in expired_messages:
|
|
expired_messages_list.append(expired_message.message_id)
|
|
|
|
#run the delete query with the list created before
|
|
result = self.delete_message(expired_messages_list)
|
|
return result[0]
|
|
|