#!/usr/bin/env python3
"""SeedVault Inventory Watchdog — checks for low stock and data integrity issues."""
import sqlite3
import json
import os
import sys
import logging
import urllib.parse
from datetime import datetime
# Setup
BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
DB_PATH = os.path.join(BASE_DIR, "inventory.db")
STATE_PATH = os.path.join(os.path.dirname(__file__), "watchdog_state.json")
LOG_PATH = os.path.join(os.path.dirname(__file__), "watchdog.log")
ENV_PATH = os.path.join(BASE_DIR, ".env")
logging.basicConfig(
filename=LOG_PATH,
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
logging.info("Watchdog run started.")
# Load env vars
TELEGRAM_BOT_TOKEN=""
TELEGRAM_CHAT_ID = ""
if os.path.exists(ENV_PATH):
with open(ENV_PATH) as f:
for line in f:
line = line.strip()
if line.startswith("TELEGRAM_BOT_TOKEN="):
TELEGRAM_BOT_TOKEN = line.split("=", 1)[1]
elif line.startswith("TELEGRAM_CHAT_ID="):
TELEGRAM_CHAT_ID = line.split("=", 1)[1]
# Load previous state
prev_state = {}
if os.path.exists(STATE_PATH):
with open(STATE_PATH, "r") as f:
prev_state = json.load(f)
prev_low_ids = set(prev_state.get("last_low_stock_ids", []))
prev_out_ids = set(prev_state.get("last_out_stock_ids", []))
prev_integrity_errors = set(prev_state.get("last_integrity_errors", []))
# Query database
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("""
SELECT id, name, sku, product_type, bulk_weight_g, weight_per_packet_g,
current_stock_g, current_stock_qty, low_stock_threshold_g, is_unit_based
FROM product
""")
products = cursor.fetchall()
conn.close()
# Checks
low_stock_items = []
out_stock_items = []
integrity_errors = []
for p in products:
pid = p["id"]
name = p["name"] or ""
sku = p["sku"] or "N/A"
is_unit = bool(p["is_unit_based"])
# Current stock value
if is_unit:
stock_val = p["current_stock_qty"] or 0
threshold = 1.0
unit_label = "units"
else:
stock_val = p["current_stock_g"] or 0.0
threshold = p["low_stock_threshold_g"] or 10.0
unit_label = "g"
# Stock status
if stock_val <= 0:
out_stock_items.append({
"id": pid, "name": name, "sku": sku,
"stock": stock_val, "unit": unit_label
})
elif stock_val <= threshold:
low_stock_items.append({
"id": pid, "name": name, "sku": sku,
"stock": stock_val, "threshold": threshold, "unit": unit_label
})
# Integrity checks
if is_unit and (p["current_stock_qty"] or 0) < 0:
err_key = f"NEGATIVE_STOCK|{pid}|{sku}"
integrity_errors.append({
"key": err_key, "type": "NEGATIVE_STOCK",
"name": name, "sku": sku,
"detail": f"Stock is {p['current_stock_qty']} (must be >= 0)"
})
elif not is_unit and (p["current_stock_g"] or 0) < 0:
err_key = f"NEGATIVE_STOCK|{pid}|{sku}"
integrity_errors.append({
"key": err_key, "type": "NEGATIVE_STOCK",
"name": name, "sku": sku,
"detail": f"Stock is {p['current_stock_g']}g (must be >= 0)"
})
if not is_unit and (p["weight_per_packet_g"] or 0) <= 0:
err_key = f"INVALID_WEIGHT|{pid}|{sku}"
integrity_errors.append({
"key": err_key, "type": "INVALID_WEIGHT",
"name": name, "sku": sku,
"detail": f"Weight per packet is {p['weight_per_packet_g']}g (must be > 0)"
})
if not is_unit and (p["low_stock_threshold_g"] or 0) <= 0:
err_key = f"INVALID_THRESHOLD|{pid}|{sku}"
integrity_errors.append({
"key": err_key, "type": "INVALID_THRESHOLD",
"name": name, "sku": sku,
"detail": f"Threshold is {p['low_stock_threshold_g']} (must be > 0)"
})
if not name.strip():
err_key = f"MISSING_NAME|{pid}|{sku}"
integrity_errors.append({
"key": err_key, "type": "MISSING_NAME",
"name": f"Product #{pid}", "sku": sku,
"detail": "Product name is empty"
})
# Current state
curr_low_ids = set(item["id"] for item in low_stock_items)
curr_out_ids = set(item["id"] for item in out_stock_items)
curr_integrity_keys = set(e["key"] for e in integrity_errors)
# Detect changes
new_low = curr_low_ids - prev_low_ids
resolved_low = prev_low_ids - curr_low_ids
new_out = curr_out_ids - prev_out_ids
resolved_out = prev_out_ids - curr_out_ids
new_integrity = curr_integrity_keys - prev_integrity_errors
resolved_integrity = prev_integrity_errors - curr_integrity_keys
has_changes = bool(new_low or resolved_low or new_out or resolved_out or new_integrity or resolved_integrity)
# Save state
new_state = {
"last_low_stock_ids": sorted(curr_low_ids),
"last_out_stock_ids": sorted(curr_out_ids),
"last_integrity_errors": sorted(curr_integrity_keys),
"last_check": datetime.now().isoformat()
}
with open(STATE_PATH, "w") as f:
json.dump(new_state, f)
# Build report
now = datetime.now().strftime("%Y-%m-%d %H:%M")
report_parts = []
report_parts.append(f"📦 **SeedVault Inventory Report** — {now}\n")
if new_out:
report_parts.append("🔴 **NEW OUT OF STOCK:**")
for item in out_stock_items:
if item["id"] in new_out:
report_parts.append(f" • {item['name']} ({item['sku']})")
report_parts.append("")
if resolved_out:
report_parts.append("✅ **RESTOCKED (was OUT):**")
for pid in resolved_out:
name = next((p["name"] for p in products if p["id"] == pid), f"Product #{pid}")
sku = next((p["sku"] or "N/A" for p in products if p["id"] == pid), "N/A")
report_parts.append(f" • {name} ({sku})")
report_parts.append("")
if new_low:
report_parts.append("🟡 **NEW LOW STOCK:**")
for item in low_stock_items:
if item["id"] in new_low:
report_parts.append(f" • {item['name']} ({item['sku']}): {item['stock']}{item['unit']} (threshold: {item['threshold']}{item['unit']})")
report_parts.append("")
if resolved_low:
report_parts.append("✅ **RESTOCKED (was LOW):**")
for pid in resolved_low:
name = next((p["name"] for p in products if p["id"] == pid), f"Product #{pid}")
sku = next((p["sku"] or "N/A" for p in products if p["id"] == pid), "N/A")
report_parts.append(f" • {name} ({sku})")
report_parts.append("")
if new_integrity:
report_parts.append("⚠️ **NEW INTEGRITY ISSUES:**")
for err in integrity_errors:
if err["key"] in new_integrity:
report_parts.append(f" • {err['type']}: {err['name']} ({err['sku']}) — {err['detail']}")
report_parts.append("")
if resolved_integrity:
report_parts.append("✅ **RESOLVED ISSUES:**")
for key in resolved_integrity:
report_parts.append(f" • {key}")
report_parts.append("")
report_parts.append(f"---")
report_parts.append(f"Total products: {len(products)} | Out: {len(out_stock_items)} | Low: {len(low_stock_items)} | Errors: {len(integrity_errors)}")
report = "\n".join(report_parts)
# Send Telegram if changes or errors
def send_telegram(message):
"""Send message via Telegram Bot API."""
import urllib.request
import urllib.error
if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID:
logging.warning("Telegram credentials missing. Skipping notification.")
return False
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
data = {
"chat_id": TELEGRAM_CHAT_ID,
"text": message,
"parse_mode": "Markdown"
}
try:
req = urllib.request.Request(
url,
data=urllib.parse.urlencode(data).encode(),
headers={"Content-Type": "application/x-www-form-urlencoded"}
)
with urllib.request.urlopen(req, timeout=15) as resp:
result = json.loads(resp.read())
if result.get("ok"):
return True
else:
logging.error(f"Telegram API error: {result.get('description', 'unknown')}")
return False
except urllib.error.HTTPError as e:
logging.error(f"Failed to send Telegram message: {e.code} {e.reason} for url: {url}")
return False
except Exception as e:
logging.error(f"Failed to send Telegram message: {str(e)}")
return False
if has_changes or integrity_errors:
sent = send_telegram(report)
if sent:
logging.info("Consolidated report sent.")
else:
logging.error("Failed to send report.")
else:
logging.info("No changes detected.")
# Log details
for item in out_stock_items:
logging.info(f"OUT: {item['name']} ({item['sku']}) = {item['stock']}{item['unit']}")
for item in low_stock_items:
logging.info(f"LOW: {item['name']} ({item['sku']}) = {item['stock']}{item['unit']}/{item['threshold']}{item['unit']}")
for err in integrity_errors:
logging.error(f"Integrity Error: {err['type']} | Item: {err['name']} | SKU: {err['sku']} | Issue: {err['detail']}")
logging.info("Watchdog run completed.")
# Print report to stdout for the cron job output
print(report)
print(f"\n---META---")
print(f"has_changes: {has_changes}")
print(f"total_products: {len(products)}")
print(f"out_of_stock: {len(out_stock_items)}")
print(f"low_stock: {len(low_stock_items)}")
print(f"integrity_errors: {len(integrity_errors)}")