Created
January 4, 2023 19:00
-
-
Save vittodevit/8c3c3a4c855a799de53ca3625a5ed49e to your computer and use it in GitHub Desktop.
scraper per il timelapse
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
HOST_FTP = "----------------" # host del server ftp | |
PORT_FTP = 21 # porta del server ftp | |
USER_FTP = "-----------------------" # utente server ftp | |
PASS_FTP = "---------" # password del server ftp | |
CARTELLA_FTP = "-" # cartella che contiene il file da scaricare | |
FILE_FTP = "--------------" # nome del file da scaricare | |
ESTENSIONE_LOC = ".---" # estensione con cui salvare il file remoto in locale | |
CARTELLA_LOC = "----------------------------" # cartella locale in cui salvare i file | |
DELAY_CHECK = 60 # ogni quanti secondi controllare per le modifiche | |
TG_CHATID = "---------" # id della chat telegram per gli alert | |
# inserisci qui il token di telegram! | |
TG_TOKEN = "-----------------------------------------" | |
# ------- DIPENDENZE ------- # | |
import datetime | |
import time | |
import requests | |
import re | |
import sys | |
from ftplib import FTP | |
# ------- PROGRAMMA ------- # | |
# contiene l'ultima modifica del file | |
ultima_modifica = 0 | |
# stampa un errore | |
def log(testo: str, errore:bool): | |
if errore: | |
print("[{}][EE] {}".format(datetime.datetime.now(), testo)) | |
else: | |
print("[{}][==] {}".format(datetime.datetime.now(), testo)) | |
def ts(): | |
return str(int(time.time())) + "_" + str(datetime.datetime.now()).replace(" ", "T") | |
# info | |
log("Connettendo a " + HOST_FTP + ":" + str(PORT_FTP), False) | |
# tenta la connessione al server ftp | |
try: | |
ftp = FTP(HOST_FTP) | |
#ftp = FTP(HOST_FTP + ":" + str(PORT_FTP)) | |
ftp.login(user = USER_FTP, passwd = PASS_FTP) | |
ftp.cwd(CARTELLA_FTP) | |
except Exception as e: | |
log(e, True) | |
sys.exit(-1) # errore!! | |
# inizio del ciclo | |
while True: | |
try: | |
# controllo l'ultima modifica | |
timestamp = ftp.voidcmd("MDTM {}{}".format(CARTELLA_FTP, FILE_FTP))[4:].strip() | |
log("TS REMOTO: {} - TS LOCALE: {}".format(timestamp, ultima_modifica), False) | |
if timestamp != ultima_modifica: | |
ultima_modifica = timestamp | |
# scarica il file | |
log("Scaricando {}".format(CARTELLA_LOC + ts() + ESTENSIONE_LOC), False) | |
localfile = open(CARTELLA_LOC + ts() + ESTENSIONE_LOC, 'wb') | |
ftp.retrbinary('RETR ' + FILE_FTP, localfile.write, 1024) | |
log("File scaricato!", False) | |
localfile.close() | |
except Exception as e: | |
log(e, True) | |
# invia allerta su telegram | |
ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])') | |
message = ansi_escape.sub('', "ALLERTA SCRAPER: " + str(e)) | |
url = f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage?chat_id={TG_CHATID}&text={message}" | |
print(requests.get(url).json()) | |
time.sleep(DELAY_CHECK) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment