phares3757/frigate-smtp
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added more filters for zones, cameras, and objects.

Browse Source
This commit is contained in:
Undercover Computing
2025-08-10 17:59:48 +12:00
committed by GitHub
parent 41719c4557
commit 68749e219b
8 changed files with 315 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
log.py
View File

@ -1,4 +1,3 @@
# Import necessary dependencies.
import paho.mqtt.client as mqtt
import smtplib
from email.mime.multipart import MIMEMultipart
@ -11,7 +10,6 @@ from io import BytesIO
import time
import threading
# Set up logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(levelname)s - %(message)s',
@ -21,7 +19,6 @@ logging.basicConfig(
]
)
# Load settings from config.json
with open('config.json', 'r') as f:
config = json.load(f)
@ -32,17 +29,64 @@ SMTP_PASSWORD = config["smtp"]["password"]
EMAIL_FROM = config["smtp"]["from"]
EMAIL_TO = config["smtp"]["to"]
HOMEASSISTANT_URL = config["homeassistant_url"]
HOMEASSISTANT_IP = config.get("homeassistant_ip", HOMEASSISTANT_URL)
# MQTT configuration
MQTT_BROKER_IP = config["mqtt"]["broker_ip"]
MQTT_PORT = config["mqtt"]["port"]
MQTT_USERNAME = config["mqtt"]["username"]
MQTT_PASSWORD = config["mqtt"]["password"]
# Dictionary to track event IDs and email state
try:
with open("alert_rules.json", "r") as f:
alert_rules_raw = json.load(f)
logging.info(f"Loaded alert_rules.json: {alert_rules_raw}")
except Exception as e:
logging.error(f"Failed to load alert_rules.json, no events will be processed: {e}")
alert_rules_raw = {}
alert_rules = {}
for cam, rules in alert_rules_raw.items():
alert_rules[cam.lower()] = {
"labels": [lbl.lower() for lbl in rules.get("labels", [])],
"ignore": [lbl.lower() for lbl in rules.get("ignore", [])],
"zones": [zone.lower() for zone in rules.get("zones", [])]
}
event_cache = {}
# Function to send email with attachment and clip link
def rule_allows_event(camera, label, zones):
cam_key = camera.lower()
lbl = label.lower()
zones_check = [z.lower() for z in zones] if zones else []
if cam_key not in alert_rules:
logging.debug(f"Camera '{camera}' not in alert_rules.json — event blocked")
return False
rule = alert_rules[cam_key]
if rule["labels"]:
if lbl not in rule["labels"]:
logging.debug(f"Label '{label}' not allowed for camera '{camera}' — event blocked")
return False
if rule["ignore"]:
if lbl in rule["ignore"]:
logging.debug(f"Label '{label}' is ignored for camera '{camera}' — event blocked")
return False
if rule["zones"]:
if not zones_check:
logging.debug(f"No zone info in event but zones filter present — event blocked")
return False
allowed_zones = [z.lower() for z in rule["zones"]]
if not any(zone in allowed_zones for zone in zones_check):
logging.debug(f"Zones {zones} not allowed for camera '{camera}' — event blocked")
return False
logging.debug(f"Event allowed for camera '{camera}', label '{label}', zones '{zones}'")
return True
def send_email(message, snapshot_urls, event_label, clip_url):
try:
subject = f"(Test) {event_label} detected!"
@ -51,19 +95,20 @@ def send_email(message, snapshot_urls, event_label, clip_url):
msg['From'] = EMAIL_FROM
msg['To'] = ", ".join(EMAIL_TO)
# Add the email body
body = f"{message}\n\nClip: {clip_url}"
msg.attach(MIMEText(body))
# Attach snapshots to the email
for snapshot_url in snapshot_urls:
response = requests.get(snapshot_url)
response.raise_for_status()
image_data = BytesIO(response.content)
msg.attach(MIMEImage(image_data.read(), name="snapshot.jpg"))
try:
logging.debug(f"Fetching snapshot: {snapshot_url}")
response = requests.get(snapshot_url, timeout=5)
response.raise_for_status()
image_data = BytesIO(response.content)
msg.attach(MIMEImage(image_data.read(), name="snapshot.jpg"))
except Exception as e:
logging.warning(f"Failed to fetch or attach snapshot {snapshot_url}: {e}")
# Send the email
with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
with smtplib.SMTP(SMTP_SERVER, SMTP_PORT, timeout=10) as server:
server.starttls()
server.login(SMTP_USERNAME, SMTP_PASSWORD)
server.sendmail(EMAIL_FROM, EMAIL_TO, msg.as_string())
@ -72,42 +117,71 @@ def send_email(message, snapshot_urls, event_label, clip_url):
except Exception as e:
logging.error(f"Failed to send email: {e}")
# Function to handle the event timeout and send email after waiting for new snapshots
def handle_event(event_id, event_label, snapshot_urls):
time.sleep(7.5) # Wait for more snapshots before sending email
def handle_event(event_id):
logging.debug(f"Event handler started for event ID: {event_id}")
time.sleep(7.5) # Delay to allow snapshots to accumulate
clip_url = f"{HOMEASSISTANT_URL}/api/frigate/notifications/{event_id}/gate/clip.mp4"
email_message = f"A {event_label} was detected.\nEvent ID: {event_id}"
send_email(email_message, snapshot_urls, event_label, clip_url)
event_info = event_cache.get(event_id)
if not event_info:
logging.warning(f"No cached info found for event ID: {event_id} on handle_event")
return
clip_url = f"{HOMEASSISTANT_URL}/api/frigate/notifications/{event_id}/{event_info['camera']}/clip.mp4"
email_message = f"A {event_info['event_label']} was detected on camera {event_info['camera']}.\nEvent ID: {event_id}"
send_email(email_message, event_info['snapshot_urls'], event_info['event_label'], clip_url)
# Remove event from cache after processing
event_cache.pop(event_id, None)
logging.debug(f"Event ID {event_id} processed and removed from cache")
# MQTT message callback
def on_message(client, userdata, message):
try:
event_data = json.loads(message.payload.decode("utf-8"))
logging.debug(f"Frigate event data: {event_data}")
logging.debug(f"Received MQTT message: {event_data}")
event_label = event_data["after"]["label"]
event_id = event_data["after"]["id"]
snapshot_url = f"{HOMEASSISTANT_URL}/api/frigate/notifications/{event_id}/snapshot.jpg"
if event_data.get("type") != "new":
logging.debug("Event type is not 'new', ignoring.")
return
after = event_data.get("after")
if not after:
logging.debug("No 'after' data in event, ignoring.")
return
event_label = after.get("label")
event_id = after.get("id")
camera = after.get("camera")
# Use correct zone info as list
zones = after.get("current_zones") or after.get("entered_zones") or []
if not event_label or not event_id or not camera:
logging.debug("Missing label, id or camera in event, ignoring.")
return
if not rule_allows_event(camera, event_label, zones):
logging.info(f"Event from camera '{camera}' with label '{event_label}' and zones '{zones}' blocked by alert rules.")
return
snapshot_url = f"{HOMEASSISTANT_IP}/api/frigate/notifications/{event_id}/snapshot.jpg"
if event_id in event_cache:
event_cache[event_id]['snapshot_urls'].append(snapshot_url)
logging.debug(f"Added snapshot to existing event cache: {event_id}")
else:
event_cache[event_id] = {
'event_label': event_label,
'camera': camera,
'snapshot_urls': [snapshot_url],
'timer': threading.Thread(target=handle_event, args=(event_id, event_label, [snapshot_url]))
'timer': threading.Thread(target=handle_event, args=(event_id,))
}
event_cache[event_id]['timer'].start()
logging.debug(f"Started new event handler thread for event ID: {event_id}")
logging.info(f"Event processed: {event_label} - Event ID: {event_id} from camera: {camera} Zones: {zones}")
logging.info(f"Event processed: {event_label} - Event ID: {event_id}")
except Exception as e:
logging.error(f"Error processing MQTT message: {e}")
# MQTT connection callback
def on_connect(client, userdata, flags, rc):
if rc == 0:
logging.info("Connected to MQTT broker successfully")
@ -115,7 +189,6 @@ def on_connect(client, userdata, flags, rc):
else:
logging.error(f"Failed to connect to MQTT broker. Return code: {rc}")
# MQTT setup
def connect_mqtt():
client = mqtt.Client()
client.username_pw_set(MQTT_USERNAME, MQTT_PASSWORD)
@ -130,12 +203,11 @@ def connect_mqtt():
logging.error(f"Error during MQTT connection: {e}")
if __name__ == "__main__":
# Print or log the warning message when the script is run
print("WARNING: USE THIS FOR TESTING AND DEBUGGING ONLY!")
logging.warning("WARNING: USE THIS FOR TESTING AND DEBUGGING ONLY!")
time.sleep(2)
print("WARNING: USE THIS FOR TESTING AND DEBUGGING ONLY!")
logging.warning("WARNING: USE THIS FOR TESTING AND DEBUGGING ONLY!")
logging.info("Starting Frigate Event Notifier...")
connect_mqtt()
connect_mqtt()