import os
import sqlite3
import json
import requests
import logging
from datetime import datetime
from dotenv import load_dotenv
# Load environment variables from .env file if it exists
load_dotenv(os.path.join(os.path.dirname(__file__), '../.env'))
# Configuration
# The script lives in seedvault-inventory/scripts/
# DB lives in seedvault-inventory/inventory.db
BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))
DB_PATH = os.path.join(BASE_DIR, 'seedvault-inventory', 'inventory.db')
STATE_FILE = os.path.join(os.path.dirname(__file__), 'watchdog_state.json')
LOG_FILE = os.path.join(os.path.dirname(__file__), 'watchdog.log')
# Load Env Variables
# Users should set these in their environment or a .env file
TOKEN = os.getenv('TELEGRAM_BOT_TOKEN')
CHAT_ID = os.getenv('TELEGRAM_CHAT_ID')
# Logging setup
logging.basicConfig(
filename=LOG_FILE,
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
def get_inventory_data():
"""Returns all products and checks for data integrity issues."""
if not os.path.exists(DB_PATH):
logging.error(f"Database not found at {DB_PATH}")
raise FileNotFoundError(f"Database not found at {DB_PATH}")
try:
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# Fetch all products to perform integrity checks
query = "SELECT id, name, sku, current_stock_g, low_stock_threshold_g, weight_per_packet_g FROM Product"
cursor.execute(query)
rows = [dict(row) for row in cursor.fetchall()]
conn.close()
return rows
except Exception as e:
logging.error(f"Database error: {e}")
raise e
def check_integrity(products):
"""Analyzes products for data corruption or logical errors."""
integrity_errors = []
for p in products:
name = p.get('name') or "Unknown Product"
sku = p.get('sku') or "N/A"
if not p.get('name'):
integrity_errors.append({
'type': 'MISSING_NAME',
'sku': sku,
'name': name,
'details': 'Product name is empty'
})
if p.get('weight_per_packet_g', 0) <= 0:
integrity_errors.append({
'type': 'INVALID_WEIGHT',
'sku': sku,
'name': name,
'details': f"Weight per packet is {p.get('weight_per_packet_g')}g (must be > 0)"
})
if p.get('current_stock_g', 0) < 0:
integrity_errors.append({
'type': 'NEGATIVE_STOCK',
'sku': sku,
'name': name,
'details': f"Stock is negative: {p.get('current_stock_g')}g"
})
if p.get('low_stock_threshold_g', 0) < 0:
integrity_errors.append({
'type': 'INVALID_THRESHOLD',
'sku': sku,
'name': name,
'details': f"Low stock threshold is negative: {p.get('low_stock_threshold_g')}g"
})
return integrity_errors
def send_telegram(message):
if not TOKEN or not CHAT_ID:
logging.warning("Telegram credentials missing. Skipping notification.")
return False
url = f"https://api.telegram.org/bot{TOKEN}/sendMessage"
payload = {
"chat_id": CHAT_ID,
"text": message,
"parse_mode": "Markdown"
}
try:
response = requests.post(url, json=payload)
response.raise_for_status()
return True
except Exception as e:
logging.error(f"Failed to send Telegram message: {e}")
return False
def load_state():
if os.path.exists(STATE_FILE):
try:
with open(STATE_FILE, 'r') as f:
return json.load(f)
except Exception as e:
logging.error(f"Failed to load state: {e}")
return {"last_low_stock_ids": []}
def save_state(state):
try:
with open(STATE_FILE, 'w') as f:
json.dump(state, f)
except Exception as e:
logging.error(f"Failed to save state: {e}")
def main():
logging.info("Watchdog run started.")
print(f"DEBUG: LOG_FILE is {LOG_FILE}")
print(f"DEBUG: DB_PATH is {DB_PATH}")
print("DEBUG: Watchdog run started.")
try:
products = get_inventory_data()
print(f"DEBUG: Found {len(products)} products")
for p in products:
print(f"DEBUG: Product: {p}")
except Exception as e:
# Critical failure: Database unreachable
error_msg = f"🚨 *SeedVault System Error* 🚨\n\nDatabase connection failed: `{str(e)}`"
send_telegram(error_msg)
logging.error(f"Watchdog aborted due to database error: {e}")
return
# 1. Check Data Integrity
integrity_errors = check_integrity(products)
print(f"DEBUG: Found {len(integrity_errors)} integrity errors")
# 2. Check Stock Status
low_stock_products = [p for p in products if p['current_stock_g'] <= p['low_stock_threshold_g']]
current_ids = [p['id'] for p in low_stock_products]
state = load_state()
last_ids = state.get("last_low_stock_ids", [])
new_low_stock_ids = set(current_ids) - set(last_ids)
print(f"DEBUG: Found {len(new_low_stock_ids)} new low stock items")
recovered_ids = list(set(last_ids) - set(current_ids))
print(f"DEBUG: Found {len(recovered_ids)} recovered items")
# 3. Build Consolidated Message
message_parts = []
if integrity_errors:
message_parts.append("⚠️ *DATA INTEGRITY ERRORS* ⚠️")
for err in integrity_errors:
message_parts.append(f"❌ *{err['type']}*\n• Item: {err['name']}\n• SKU: `{err['sku']}`\n• Issue: {err['details']}")
if new_low_stock_ids:
message_parts.append("\n🚨 *NEW LOW STOCK ITEMS* 🚨")
for p in low_stock_products:
if p['id'] in new_low_stock_ids:
status = "🔴 OUT" if p['current_stock_g'] <= 0 else "🟡 LOW"
message_parts.append(f"• *{p['name']}* ({p['sku'] or 'No SKU'}): {p['current_stock_g']:.1f}g [{status}]")
if recovered_ids:
message_parts.append("\n✅ *STOCK RECOVERED* ✅")
message_parts.append("• Some items have been replenished.")
if message_parts:
full_msg = "📦 *SeedVault Inventory Report*\n\n" + "\n\n".join(message_parts)
if send_telegram(full_msg):
logging.info("Consolidated report sent.")
save_state({"last_low_stock_ids": current_ids})
else:
logging.error("Failed to send report.")
else:
logging.info("No changes detected.")
logging.info("Watchdog run completed.")
print("DEBUG: Reached end of main")
if __name__ == "__main__":
main()