89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
import requests
|
||
import json
|
||
import os
|
||
from dotenv import load_dotenv
|
||
|
||
# Load information for owencloud and telegram
|
||
load_dotenv()
|
||
#Send To Telegram
|
||
def send_telegram(text: str):
|
||
token = os.getenv("TELEGRAM_TOKEN")
|
||
channel_id = os.getenv("TELEGRAM_CHANNEL_ID")
|
||
if not token or not channel_id:
|
||
raise ValueError("API Bot or ID-chat not true .env")
|
||
|
||
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
||
|
||
r = requests.post(url, data={
|
||
"chat_id": channel_id,
|
||
"text": text
|
||
})
|
||
print(r.text)
|
||
if r.status_code != 200:
|
||
raise Exception(f"Can't send message: {r.text}")
|
||
#Get Token for OwenCloud
|
||
def get_token():
|
||
login = os.getenv("LOGIN")
|
||
password = os.getenv("PASSWORD")
|
||
if not login or not password:
|
||
raise ValueError("Логин или пароль не заданы в .env")
|
||
|
||
data = {'login': login, 'password': password}
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json"
|
||
}
|
||
req = requests.post("https://api.owencloud.ru/v1/auth/open/", json=data, headers=headers)
|
||
if req.status_code == 200:
|
||
answer = req.json()
|
||
token = "Bearer " + answer["token"]
|
||
return token
|
||
else:
|
||
print(f"Ошибка авторизации: {req.status_code}, {req.text}")
|
||
return None
|
||
#Device parsing
|
||
def main():
|
||
token = get_token()
|
||
if not token:
|
||
print("Не удалось получить токен. Завершаем работу.")
|
||
return
|
||
|
||
device_id = "296613"
|
||
headers = {'Authorization': token}
|
||
data = json.dumps({'filter': ''})
|
||
|
||
req = requests.post(f"https://api.owencloud.ru/v1/device/{device_id}", data=data, headers=headers)
|
||
if req.status_code != 200:
|
||
print(f"Ошибка запроса устройства: {req.status_code}, {req.text}")
|
||
return
|
||
|
||
nswer = req.json()
|
||
|
||
if "parameters" not in nswer:
|
||
print("Не удалось найти параметры в ответе.")
|
||
return
|
||
|
||
parameters = nswer['parameters']
|
||
|
||
#Find temperature
|
||
try:
|
||
temperature_1 = next(param for param in parameters if param['name'] == "Температура_1 (НИЗ)")
|
||
temperature_2 = next(param for param in parameters if param['name'] == "Температура_2 (ВЕРХ)")
|
||
except StopIteration:
|
||
print("Не удалось найти необходимые параметры.")
|
||
return
|
||
|
||
message = (f"Шача:\n"
|
||
f"Температура_1 (НИЗ): {temperature_1['value']}\n"
|
||
f"Температура_2 (ВЕРХ): {temperature_2['value']}")
|
||
|
||
print(message)
|
||
|
||
# Send to Telegram
|
||
try:
|
||
send_telegram(message)
|
||
except Exception as e:
|
||
print(f"Ошибка отправки сообщения: {e}")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|