claudecode/AvianVisitors
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

linted

Browse Source
This commit is contained in:
mcguirepr89
2022-06-24 07:46:37 -04:00
parent b662545845
commit 97e7b19aa2
1 changed files with 23 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/utils/notifications.py
+21 -14
View File
@@ -2,14 +2,14 @@ import apprise
import os import os
import socket import socket
import sqlite3 import sqlite3
from sqlite3 import Error
import time import time
from datetime import datetime, timedelta from datetime import datetime
userDir = os.path.expanduser('~') userDir = os.path.expanduser('~')
APPRISE_CONFIG = userDir + '/BirdNET-Pi/apprise.txt' APPRISE_CONFIG = userDir + '/BirdNET-Pi/apprise.txt'
DB_PATH = userDir + '/BirdNET-Pi/scripts/birds.db' DB_PATH = userDir + '/BirdNET-Pi/scripts/birds.db'
def notify(body, title): def notify(body, title):
apobj = apprise.Apprise() apobj = apprise.Apprise()
config = apprise.AppriseConfig() config = apprise.AppriseConfig()
@@ -20,6 +20,7 @@ def notify(body, title):
title=title, title=title,
) )
def sendAppriseNotifications(species, confidence, path, settings_dict, db_path=DB_PATH): def sendAppriseNotifications(species, confidence, path, settings_dict, db_path=DB_PATH):
# print(sendAppriseNotifications) # print(sendAppriseNotifications)
# print(settings_dict) # print(settings_dict)
@@ -33,17 +34,17 @@ def sendAppriseNotifications(species, confidence, path, settings_dict, db_path=D
websiteurl = settings_dict.get('BIRDNETPI_URL') websiteurl = settings_dict.get('BIRDNETPI_URL')
if len(websiteurl) == 0: if len(websiteurl) == 0:
raise ValueError('Blank URL') raise ValueError('Blank URL')
except Exception as e: except Exception:
websiteurl = "http://"+socket.gethostname()+".local" websiteurl = "http://"+socket.gethostname()+".local"
listenurl = websiteurl+"?filename="+path listenurl = websiteurl+"?filename="+path
if settings_dict.get('APPRISE_NOTIFY_EACH_DETECTION') == "1": if settings_dict.get('APPRISE_NOTIFY_EACH_DETECTION') == "1":
notify_body=body.replace("$sciname", sciName).replace("$comname", comName).replace("$confidence", confidence).replace("$listenurl", listenurl) notify_body = body.replace("$sciname", sciName).replace("$comname", comName).replace("$confidence", confidence).replace("$listenurl", listenurl)
notify_title=title.replace("$sciname", sciName).replace("$comname", comName).replace("$confidence", confidence).replace("$listenurl", listenurl) notify_title = title.replace("$sciname", sciName).replace("$comname", comName).replace("$confidence", confidence).replace("$listenurl", listenurl)
notify(notify_body, notify_title) notify(notify_body, notify_title)
APPRISE_NOTIFICATION_NEW_SPECIES_DAILY_COUNT_LIMIT = 1 # Notifies the first N per day. APPRISE_NOTIFICATION_NEW_SPECIES_DAILY_COUNT_LIMIT = 1 # Notifies the first N per day.
if settings_dict.get('APPRISE_NOTIFY_NEW_SPECIES_EACH_DAY') == "1": if settings_dict.get('APPRISE_NOTIFY_NEW_SPECIES_EACH_DAY') == "1":
try: try:
con = sqlite3.connect(db_path) con = sqlite3.connect(db_path)
@@ -51,14 +52,18 @@ def sendAppriseNotifications(species, confidence, path, settings_dict, db_path=D
today = datetime.now().strftime("%Y-%m-%d") today = datetime.now().strftime("%Y-%m-%d")
cur.execute(f"SELECT DISTINCT(Com_Name), count(Com_Name) FROM detections WHERE Date = date('{today}') GROUP BY Com_Name") cur.execute(f"SELECT DISTINCT(Com_Name), count(Com_Name) FROM detections WHERE Date = date('{today}') GROUP BY Com_Name")
known_species = cur.fetchall() known_species = cur.fetchall()
detections = [d[1] for d in known_species if d[0] == comName.replace("'","")] detections = [d[1] for d in known_species if d[0] == comName.replace("'", "")]
numberDetections = 0 numberDetections = 0
if len(detections): if len(detections):
numberDetections = detections[0] numberDetections = detections[0]
if numberDetections > 0 and numberDetections <= APPRISE_NOTIFICATION_NEW_SPECIES_DAILY_COUNT_LIMIT: if numberDetections > 0 and numberDetections <= APPRISE_NOTIFICATION_NEW_SPECIES_DAILY_COUNT_LIMIT:
print("send the notification") print("send the notification")
notify_body=body.replace("$sciname", sciName).replace("$comname", comName).replace("$confidence", confidence).replace("$listenurl", listenurl) + " (first time today)" notify_body = body.replace("$sciname", sciName).replace("$comname", comName)\
notify_title=title.replace("$sciname", sciName).replace("$comname", comName).replace("$confidence", confidence).replace("$listenurl", listenurl) + " (first time today)" .replace("$confidence", confidence).replace("$listenurl", listenurl)\
+ " (first time today)"
notify_title = title.replace("$sciname", sciName).replace("$comname", comName)\
.replace("$confidence", confidence).replace("$listenurl", listenurl)\
+ " (first time today)"
notify(notify_body, notify_title) notify(notify_body, notify_title)
con.close() con.close()
except sqlite3.Error as e: except sqlite3.Error as e:
@@ -73,13 +78,17 @@ def sendAppriseNotifications(species, confidence, path, settings_dict, db_path=D
today = datetime.now().strftime("%Y-%m-%d") today = datetime.now().strftime("%Y-%m-%d")
cur.execute(f"SELECT DISTINCT(Com_Name), count(Com_Name) FROM detections WHERE Date >= date('{today}', '-7 day') GROUP BY Com_Name") cur.execute(f"SELECT DISTINCT(Com_Name), count(Com_Name) FROM detections WHERE Date >= date('{today}', '-7 day') GROUP BY Com_Name")
known_species = cur.fetchall() known_species = cur.fetchall()
detections = [d[1] for d in known_species if d[0] == comName.replace("'","")] detections = [d[1] for d in known_species if d[0] == comName.replace("'", "")]
numberDetections = 0 numberDetections = 0
if len(detections): if len(detections):
numberDetections = detections[0] numberDetections = detections[0]
if numberDetections > 0 and numberDetections <= 5: if numberDetections > 0 and numberDetections <= 5:
notify_body=body.replace("$sciname", sciName).replace("$comname", comName).replace("$confidence", confidence).replace("$listenurl", listenurl) + " (only seen "+str(int(numberDetections))+" times in last 7d)" notify_body = body.replace("$sciname", sciName).replace("$comname", comName)\
notify_title=title.replace("$sciname", sciName).replace("$comname", comName).replace("$confidence", confidence).replace("$listenurl", listenurl) + " (only seen "+str(int(numberDetections))+" times in last 7d)" .replace("$confidence", confidence).replace("$listenurl", listenurl)\
+ " (only seen " + str(int(numberDetections)) + " times in last 7d)"
notify_title = title.replace("$sciname", sciName).replace("$comname", comName)\
.replace("$confidence", confidence).replace("$listenurl", listenurl)\
+ " (only seen " + str(int(numberDetections)) + " times in last 7d)"
notify(notify_body, notify_title) notify(notify_body, notify_title)
con.close() con.close()
except sqlite3.Error: except sqlite3.Error:
@@ -89,5 +98,3 @@ def sendAppriseNotifications(species, confidence, path, settings_dict, db_path=D
if __name__ == "__main__": if __name__ == "__main__":
print("notfications") print("notfications")