PyScript for Automation CBMS
This commit is contained in:
@@ -0,0 +1,2 @@
|
|||||||
|
TOKEN = "194|TBreiVY324x1zNTm3cVCfqbGzUbd8Igu9xIuecou40d20a16"
|
||||||
|
BASE_URL = "http://192.168.26.11:83"
|
||||||
+62
@@ -0,0 +1,62 @@
|
|||||||
|
import os
|
||||||
|
import requests
|
||||||
|
import threading
|
||||||
|
import logging
|
||||||
|
import schedule
|
||||||
|
import time
|
||||||
|
from dotenv import load_dotenv
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||||||
|
|
||||||
|
def load_env_vars() -> tuple[Optional[str], Optional[str]]:
|
||||||
|
with load_dotenv():
|
||||||
|
base_url = os.getenv("BASE_URL")
|
||||||
|
token = os.getenv("TOKEN")
|
||||||
|
|
||||||
|
if not base_url:
|
||||||
|
logging.error("BASE_URL not set in environment variables.")
|
||||||
|
if not token:
|
||||||
|
logging.error("TOKEN not set in environment variables.")
|
||||||
|
|
||||||
|
return base_url, token
|
||||||
|
|
||||||
|
def make_requests() -> None:
|
||||||
|
base_url, token = load_env_vars()
|
||||||
|
|
||||||
|
if not base_url or not token:
|
||||||
|
return
|
||||||
|
|
||||||
|
headers = {"Authorization": f"Bearer {token}"}
|
||||||
|
full_url = f"{base_url}/api/fetch"
|
||||||
|
# add /data every saturday ignore /fetch for saturday
|
||||||
|
|
||||||
|
with requests.Session() as session:
|
||||||
|
try:
|
||||||
|
response = session.get(full_url, headers=headers, timeout=10)
|
||||||
|
response.raise_for_status()
|
||||||
|
|
||||||
|
if response.status_code == 200:
|
||||||
|
logging.info("Data Updated!")
|
||||||
|
logging.info(f"Success: {full_url} - Response: {response.json()}")
|
||||||
|
except requests.exceptions.Timeout:
|
||||||
|
logging.error(f"Request to {full_url} timed out")
|
||||||
|
except requests.exceptions.HTTPError as e:
|
||||||
|
logging.error(f"HTTP error occurred: {e}")
|
||||||
|
except requests.exceptions.RequestException as e:
|
||||||
|
logging.error(f"Error while requesting {full_url}: {e}")
|
||||||
|
|
||||||
|
def run_threaded(job_func):
|
||||||
|
job_thread = threading.Thread(target=job_func)
|
||||||
|
job_thread.start()
|
||||||
|
|
||||||
|
def schedule_daily_request():
|
||||||
|
schedule.every().day.at("00:00:00").do(run_threaded, make_requests)
|
||||||
|
|
||||||
|
while True:
|
||||||
|
schedule.run_pending()
|
||||||
|
time.sleep(1)
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
logging.info("Starting daily request scheduler...")
|
||||||
|
schedule_daily_request()
|
||||||
@@ -0,0 +1,3 @@
|
|||||||
|
requests
|
||||||
|
python-dotenv
|
||||||
|
schedule
|
||||||
Reference in New Issue
Block a user