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

notifications: make db lookup common name agnostic

Browse Source
This commit is contained in:
frederik
2025-10-05 10:17:54 +02:00
parent 5029a417c2
commit 01fbef475e
1 changed files with 42 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/utils/notifications.py
+42 -41
View File
@@ -126,49 +126,50 @@ def sendAppriseNotifications(species, confidence, confidencepct, path,
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: numberDetections = get_todays_count_for(db_path, sciName)
con = sqlite3.connect(db_path) if 0 < numberDetections <= APPRISE_NOTIFICATION_NEW_SPECIES_DAILY_COUNT_LIMIT:
cur = con.cursor() print("send the notification")
today = datetime.now().strftime("%Y-%m-%d") notify_body = render_template(body, "first time today")
cur.execute(f"SELECT DISTINCT(Com_Name), COUNT(Com_Name) FROM detections WHERE Date = DATE('{today}') GROUP BY Com_Name") notify_title = render_template(title, "first time today")
known_species = cur.fetchall() notify(notify_body, notify_title, image_url)
detections = [d[1] for d in known_species if d[0] == comName] species_last_notified[comName] = int(timeim.time())
numberDetections = 0
if len(detections):
numberDetections = detections[0]
if numberDetections > 0 and numberDetections <= APPRISE_NOTIFICATION_NEW_SPECIES_DAILY_COUNT_LIMIT:
print("send the notification")
notify_body = render_template(body, "first time today")
notify_title = render_template(title, "first time today")
notify(notify_body, notify_title, image_url)
species_last_notified[comName] = int(timeim.time())
con.close()
except sqlite3.Error as e:
print(e)
print("Database busy")
timeim.sleep(2)
if settings_dict.get('APPRISE_NOTIFY_NEW_SPECIES') == "1": if settings_dict.get('APPRISE_NOTIFY_NEW_SPECIES') == "1":
try: numberDetections = get_this_weeks_count_for(db_path, sciName)
con = sqlite3.connect(db_path) if 0 < numberDetections <= 5:
cur = con.cursor() reason = f"only seen {numberDetections} times in last 7d"
today = datetime.now().strftime("%Y-%m-%d") notify_body = render_template(body, reason)
cur.execute(f"SELECT DISTINCT(Com_Name), COUNT(Com_Name) FROM detections WHERE Date >= DATE('{today}', '-7 day') GROUP BY Com_Name") notify_title = render_template(title, reason)
known_species = cur.fetchall() notify(notify_body, notify_title, image_url)
detections = [d[1] for d in known_species if d[0] == comName] species_last_notified[comName] = int(timeim.time())
numberDetections = 0
if len(detections):
numberDetections = detections[0] def get_todays_count_for(db_path, sci_name):
if numberDetections > 0 and numberDetections <= 5: today = datetime.now().strftime("%Y-%m-%d")
reason = f"only seen {numberDetections} times in last 7d" select_sql = f"SELECT COUNT(*) FROM detections WHERE Date = DATE('{today}') AND Sci_name = '{sci_name}'"
notify_body = render_template(body, reason) records = get_records(db_path, select_sql)
notify_title = render_template(title, reason) return records[0][0] if records else 0
notify(notify_body, notify_title, image_url)
species_last_notified[comName] = int(timeim.time())
con.close() def get_this_weeks_count_for(db_path, sci_name):
except sqlite3.Error: today = datetime.now().strftime("%Y-%m-%d")
print("Database busy") select_sql = f"SELECT COUNT(*) FROM detections WHERE Date >= DATE('{today}', '-7 day') AND Sci_name = '{sci_name}'"
timeim.sleep(2) records = get_records(db_path, select_sql)
return records[0][0] if records else 0
def get_records(db_path, select_sql):
try:
con = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
cur = con.cursor()
cur.execute(select_sql)
records = cur.fetchall()
con.close()
except sqlite3.Error:
print("Database busy")
timeim.sleep(2)
records = []
return records
if __name__ == "__main__": if __name__ == "__main__":