# --- Care Count (Volunteers โ Visits โ Items) ---
# Modern UI/UX with industry-standard design patterns
# Enhanced version with improved user experience
from __future__ import annotations
import os, io, time, base64, re, uuid, json
from datetime import datetime, timedelta
from typing import Optional
import logging
import pytz
import requests
import pandas as pd
import streamlit as st
from PIL import Image, ImageOps, ImageEnhance
from supabase import create_client, Client
# Import our modern UI components
from ui_improvements import ModernUIComponents, apply_modern_ui, create_modern_layout
# ------------------------ Enhanced Logging ------------------------
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('care_count.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# ------------------------ App config ------------------------
os.environ.setdefault("HOME", "/tmp")
os.environ["STREAMLIT_BROWSER_GATHER_USAGE_STATS"] = "false"
TZ = os.getenv("APP_TZ", "America/Toronto")
CUTOFF_HOUR = int(os.getenv("CUTOFF_HOUR", "-1")) # -1 disables daily cutoff
INACTIVITY_MIN = int(os.getenv("INACTIVITY_MIN", "30"))
def local_now() -> datetime:
return datetime.now(pytz.timezone(TZ))
# Enhanced page config with modern settings
st.set_page_config(
page_title="Care Count - Volunteer Management",
page_icon="๐",
layout="wide",
initial_sidebar_state="expanded",
menu_items={
'Get Help': 'https://github.com/your-repo/care-count',
'Report a bug': "https://github.com/your-repo/care-count/issues",
'About': "# Care Count\nVolunteer management system for community impact tracking"
}
)
# Apply modern UI styling
apply_modern_ui()
# ------------------------ Enhanced Secrets Management ------------------------
try:
from dotenv import load_dotenv
load_dotenv()
except Exception:
pass
try:
_SECRETS = getattr(st, "secrets", {})
except Exception:
_SECRETS = {}
def _is_useful(val: Optional[str]) -> bool:
return val is not None and str(val).strip() != ""
def get_secret(name: str, default: Optional[str] = None) -> Optional[str]:
"""Enhanced secret management with logging"""
v = os.getenv(name)
if _is_useful(v):
logger.info(f"Secret {name} loaded from environment")
return v
try:
if hasattr(_SECRETS, "get"):
v = _SECRETS.get(name, default)
if _is_useful(v):
logger.info(f"Secret {name} loaded from Streamlit secrets")
return v
except Exception as e:
logger.warning(f"Failed to load secret {name} from Streamlit secrets: {e}")
return default
def require_secret(name: str) -> str:
"""Enhanced secret requirement with better error handling"""
v = get_secret(name, None)
if not _is_useful(v):
st.error(
f"๐ **Configuration Error**\n\n"
f"Missing required configuration: `{name}`\n\n"
f"Please ensure this is set in your environment variables or `.streamlit/secrets.toml` file.\n\n"
f"For help, check the [documentation](https://github.com/your-repo/care-count#configuration)."
)
st.stop()
return str(v)
# Required Supabase credentials
SUPABASE_URL = require_secret("SUPABASE_URL")
SUPABASE_KEY = require_secret("SUPABASE_KEY")
# Initialize Supabase client with error handling
try:
sb: Client = create_client(SUPABASE_URL, SUPABASE_KEY)
logger.info("Supabase client initialized successfully")
except Exception as e:
st.error(f"Failed to connect to Supabase: {e}")
logger.error(f"Supabase connection failed: {e}")
st.stop()
# Optional provider/config
PROVIDER = (get_secret("PROVIDER", "nebius") or "nebius").strip().lower()
GEMMA_MODEL = (get_secret("GEMMA_MODEL", "google/gemma-3-27b-it") or "google/gemma-3-27b-it").strip()
NEBIUS_API_KEY = get_secret("NEBIUS_API_KEY")
NEBIUS_BASE_URL = get_secret("NEBIUS_BASE_URL", "https://api.nebius.ai/v1")
FEATH_API_KEY = get_secret("FEATHERLESS_API_KEY")
FEATH_BASE_URL = get_secret("FEATHERLESS_BASE_URL", "https://api.featherless.ai/v1")
# ------------------------ Enhanced Event Logging ------------------------
def log_event(action: str, actor: Optional[str], details: dict, level: str = "info"):
"""Enhanced event logging with multiple levels"""
log_data = {
"happened_at": datetime.utcnow().isoformat(),
"action": action,
"actor_email": actor,
"details": json.dumps(details) if isinstance(details, dict) else str(details)
}
# Log to file
if level == "error":
logger.error(f"Event: {action} by {actor} - {details}")
elif level == "warning":
logger.warning(f"Event: {action} by {actor} - {details}")
else:
logger.info(f"Event: {action} by {actor} - {details}")
# Log to database (without level field to avoid schema issues)
try:
sb.table("events").insert(log_data).execute()
except Exception as e:
logger.warning(f"Failed to log event to database (continuing): {e}")
# ------------------------ Modern Authentication UI ------------------------
def auth_block() -> tuple[bool, Optional[str]]:
"""Enhanced authentication with modern UI and improved loading states"""
if "auth_email" not in st.session_state:
st.session_state["auth_email"] = None
if "user_email" in st.session_state:
return True, st.session_state["user_email"]
# Modern hero section for login
st.markdown(ModernUIComponents.create_hero_section(
"Care Count",
"Snap items, track visits, and make an impact."
), unsafe_allow_html=True)
# Modern login form with improved state management
with st.container():
st.markdown('', unsafe_allow_html=True)
st.subheader("Sign In")
st.markdown("Enter your email to receive a secure login code.")
# Use a single form for better state management
with st.form("auth_form", clear_on_submit=False):
email = st.text_input(
"Email Address",
value=st.session_state.get("auth_email") or "",
placeholder="your.email@example.com",
help="We'll send you a secure 6-digit code",
disabled=st.session_state.get("auth_email") is not None
)
# Show different buttons based on state
col1, col2, col3 = st.columns([1, 2, 1])
with col2:
if not st.session_state.get("auth_email"):
send = st.form_submit_button("๐ง Send Login Code", use_container_width=True, type="primary")
else:
send = st.form_submit_button("๐ง Resend Code", use_container_width=True, type="secondary")
if send:
if not email or "@" not in email:
st.error("Please enter a valid email address.")
else:
# Use placeholder for better UX during loading
status_placeholder = st.empty()
try:
with status_placeholder.container():
st.markdown(ModernUIComponents.create_status_message("Sending login code...", "loading"), unsafe_allow_html=True)
# Perform the actual API call
sb.auth.sign_in_with_otp({"email": email, "shouldCreateUser": True})
st.session_state["auth_email"] = email
# Clear status and show success
status_placeholder.empty()
st.markdown(ModernUIComponents.create_status_message("Login code sent! Check your email.", "success"), unsafe_allow_html=True)
log_event("otp_requested", email, {"method": "email"})
# Small delay to prevent immediate rerun
time.sleep(0.5)
st.rerun()
except Exception as e:
error_msg = str(e)
status_placeholder.empty()
if "429" in error_msg or "Too Many Requests" in error_msg:
st.warning("โณ Please wait a moment before requesting another code. For security, there's a brief delay between requests.")
elif "Invalid email" in error_msg:
st.error("โ Please enter a valid email address.")
else:
st.error(f"โ Could not send code: {error_msg}")
log_event("otp_failed", email, {"error": error_msg}, "error")
st.markdown('
', unsafe_allow_html=True)
# OTP verification form - only show if email is set
if st.session_state.get("auth_email"):
with st.container():
st.markdown('', unsafe_allow_html=True)
st.subheader("๐ข Verify Code")
st.markdown(f"Enter the 6-digit code sent to **{st.session_state['auth_email']}**")
with st.form("otp_verify", clear_on_submit=True):
code = st.text_input(
"Verification Code",
max_chars=6,
placeholder="123456",
help="Enter the 6-digit code from your email"
)
col1, col2, col3 = st.columns([1, 2, 1])
with col2:
verify_btn = st.form_submit_button("โ
Verify & Start Shift", use_container_width=True, type="primary")
if verify_btn:
if len(code) != 6 or not code.isdigit():
st.error("Please enter a valid 6-digit code.")
else:
# Use placeholder for verification status
verify_placeholder = st.empty()
try:
with verify_placeholder.container():
st.markdown(ModernUIComponents.create_status_message("Verifying code...", "loading"), unsafe_allow_html=True)
# Perform verification
res = sb.auth.verify_otp({"email": st.session_state["auth_email"], "token": code, "type": "email"})
if res and res.user:
# Clear verification status
verify_placeholder.empty()
# Ensure subsequent PostgREST requests carry the user's JWT for RLS
try:
token = getattr(getattr(res, "session", None), "access_token", None)
if token:
sb.postgrest.auth(token)
except Exception:
pass
email = st.session_state["auth_email"]
# Enhanced volunteer upsert
volunteer_data = {
"email": email,
"last_login_at": datetime.utcnow().isoformat(),
"shift_started_at": datetime.utcnow().isoformat(),
"shift_ended_at": None
}
sb.table("volunteers").upsert(volunteer_data, on_conflict="email").execute()
st.session_state["user_email"] = email
st.session_state["shift_started"] = True
st.session_state["last_activity_at"] = local_now()
log_event("login_success", email, {"method": "otp"})
st.success("๐ Welcome back! Your shift has started.")
# Custom Laurier Cares bags celebration
bags_html = """
"""
st.markdown(bags_html, unsafe_allow_html=True)
# Small delay before rerun for better UX
time.sleep(1)
st.rerun()
except Exception as e:
error_msg = str(e)
verify_placeholder.empty()
if "expired" in error_msg.lower() or "invalid" in error_msg.lower():
st.error("โ This code has expired or is invalid. Please request a new code.")
# Clear the auth_email to allow new code request
if "auth_email" in st.session_state:
del st.session_state["auth_email"]
# Small delay before rerun
time.sleep(0.5)
st.rerun()
else:
st.error(f"โ Verification failed: {error_msg}")
log_event("otp_verification_failed", st.session_state["auth_email"], {"error": error_msg}, "error")
# Add option to start over
col1, col2, col3 = st.columns([1, 1, 1])
with col2:
if st.button("๐ Use Different Email", use_container_width=True, type="secondary"):
if "auth_email" in st.session_state:
del st.session_state["auth_email"]
st.rerun()
st.markdown('
๐ก Quick Tips
When you're done, please End Visit and Sign out.
We'll auto-sign you out after inactivity or at 8pm local time. ๐
""", unsafe_allow_html=True)
# ------------------------ Helper Functions (Enhanced) ------------------------
def fetch_volunteer_row(email: str) -> dict | None:
"""Enhanced volunteer data fetching with error handling"""
try:
return sb.table("volunteers").select("*").eq("email", email).single().execute().data
except Exception as e:
logger.warning(f"Failed to fetch volunteer data for {email}: {e}")
return None
def items_today(email: str) -> int:
"""Enhanced item counting with better error handling"""
try:
day = local_now().strftime("%Y-%m-%d")
# Try partitioned table first
data = sb.table("visit_items_p").select("id,timestamp,volunteer") \
.gte("timestamp", f"{day} 00:00:00").lte("timestamp", f"{day} 23:59:59") \
.eq("volunteer", email).execute().data
return len(data or [])
except Exception:
try:
data = sb.table("visit_items").select("id,timestamp,volunteer") \
.gte("timestamp", f"{day} 00:00:00").lte("timestamp", f"{day} 23:59:59") \
.eq("volunteer", email).execute().data
return len(data or [])
except Exception as e:
logger.warning(f"Failed to count items for {email}: {e}")
return 0
def fallback_visit_code() -> str:
"""Enhanced visit code generation with better uniqueness"""
try:
day = local_now().strftime("%Y-%m-%d")
todays = sb.table("visits").select("id,started_at").gte("started_at", f"{day} 00:00:00") \
.lte("started_at", f"{day} 23:59:59").execute().data or []
seq = len(todays) + 1
except Exception:
seq = int(time.time()) % 1000
return f"V-{seq}-{local_now().strftime('%Y%m%d')}-{str(uuid.uuid4())[:6]}"
def _to_png_bytes(img: Image.Image) -> bytes:
"""Convert image to PNG bytes"""
b = io.BytesIO()
img.save(b, format="PNG")
return b.getvalue()
def preprocess_for_label(img: Image.Image) -> Image.Image:
"""Enhanced image preprocessing for better AI recognition"""
img = img.convert("RGB")
w, h = img.size
scale = 1024 / max(w, h) if max(w, h) > 1024 else 1.0
if scale < 1.0:
img = img.resize((int(w * scale), int(h * scale)))
img = ImageOps.autocontrast(img, cutoff=2)
img = ImageEnhance.Brightness(img).enhance(1.06)
img = ImageEnhance.Contrast(img).enhance(1.05)
img = ImageEnhance.Sharpness(img).enhance(1.15)
return img
def gemma_item_name(img_bytes: bytes) -> str:
"""Enhanced AI item identification with better error handling and fallback"""
try:
primary_err = None
if PROVIDER == "nebius":
try:
if not NEBIUS_API_KEY:
raise RuntimeError("NEBIUS_API_KEY missing")
return _openai_style_chat(NEBIUS_BASE_URL, NEBIUS_API_KEY, GEMMA_MODEL, img_bytes)
except Exception as e:
primary_err = e
if PROVIDER == "featherless":
try:
if not FEATH_API_KEY:
raise RuntimeError("FEATHERLESS_API_KEY missing")
return _openai_style_chat(FEATH_BASE_URL, FEATH_API_KEY, GEMMA_MODEL, img_bytes)
except Exception as e:
primary_err = e
# Fallback chain if primary failed or unknown provider
if FEATH_API_KEY:
return _openai_style_chat(FEATH_BASE_URL, FEATH_API_KEY, GEMMA_MODEL, img_bytes)
raise primary_err or RuntimeError(f"Unknown PROVIDER: {PROVIDER}")
except Exception as e:
logger.error(f"AI identification failed: {e}")
raise
def _openai_style_chat(base_url: str, api_key: str, model_id: str, img_bytes: bytes) -> str:
"""Enhanced API communication with better error handling"""
b64 = base64.b64encode(img_bytes).decode("utf-8")
url = f"{base_url.rstrip('/')}/chat/completions"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
payload = {
"model": model_id,
"temperature": 0,
"messages": [
{"role": "system", "content": "You label item being held in the image for a food bank. Return ONLY the item name."},
{"role": "user", "content": [
{"type": "text", "text": "What is the name of the item in the picture? Return only the item name."},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}}
]}
]
}
try:
r = requests.post(url, json=payload, headers=headers, timeout=90)
if r.status_code != 200:
raise RuntimeError(f"LLM HTTP {r.status_code}: {r.text[:200]}")
data = r.json()
return (data["choices"][0]["message"]["content"] or "").strip()
except requests.exceptions.Timeout:
raise RuntimeError("AI service timeout - please try again")
except requests.exceptions.RequestException as e:
raise RuntimeError(f"AI service error: {e}")
def normalize_item_name(s: str) -> str:
"""Enhanced item name normalization"""
s = (s or "").strip()
if not s:
return ""
# Enhanced brand and type recognition
BRANDS = {
"whiskas","tetley","kellogg's","kelloggs","campbell's","campbells","heinz",
"nestle","kraft","general mills","cheerios","oreo","oreos","pringles","lays","doritos",
"ice river","green bottle","great value","wheat thins","vegetable thins","raid"
}
GENERIC_TYPES = {
"water","toothpaste","deodorant","antiperspirant","soap","shampoo","conditioner",
"lotion","tea","coffee","cereal","pasta","rice","beans","sauce","salsa","cleaner",
"peanut butter","jam","jelly","tuna","chicken","beef","flour","sugar","salt","oil",
"crackers","cookies","soup","insect killer","spray"
}
low = re.sub(r"[ยฎโข]", "", s.lower())
for b in BRANDS:
low = low.replace(b, "")
chosen = None
for t in GENERIC_TYPES:
if t in low:
chosen = t
break
cleaned = " ".join(low.split())
return (chosen or cleaned.title())[:120]
def clean_text(v: Optional[str], maxlen: int = 120) -> Optional[str]:
"""Enhanced text cleaning"""
if not v:
return None
v = re.sub(r"\s+", " ", v).strip()
return v[:maxlen] if v else None
def deterministic_ingest_id(v_id: int, email: str, name: str, qty: int, ts_iso: str) -> str:
"""Enhanced ID generation for data integrity"""
key = f"visit_items::{v_id}::{email}::{name}::{qty}::{ts_iso}"
return str(uuid.uuid5(uuid.NAMESPACE_URL, key))
def try_rpc_ingest(email: str, v_id: int, name: str, qty: int,
category: Optional[str], unit: Optional[str],
barcode: Optional[str], ts_iso: str, ingest_id: str) -> tuple[bool,str]:
"""Enhanced RPC ingestion with better error handling"""
try:
res = sb.rpc("safe_ingest_visit_item", {
"p_email": email,
"p_visit_id": v_id,
"p_item_name": name,
"p_qty": qty,
"p_category": category,
"p_unit": unit,
"p_barcode": barcode,
"p_ts": ts_iso,
"p_ingest_id": ingest_id
}).execute()
rows = res.data if isinstance(res.data, list) else []
if rows:
r0 = rows[0]
ok = bool(r0.get("ok", False))
msg = str(r0.get("msg", ""))
return ok, msg or ("ok" if ok else "failed")
return True, "ok"
except Exception as e:
logger.warning(f"RPC ingest failed, using fallback: {e}")
raise e
def fallback_direct_insert(email: str, v_id: int, name: str, qty: int,
category: Optional[str], unit: Optional[str],
barcode: Optional[str], ts_iso: str, ingest_id: str,
weather_type: Optional[str] = None, temp_c: Optional[float] = None) -> None:
"""Enhanced fallback insertion with better error handling"""
payload = {
"visit_id": v_id,
"timestamp": ts_iso,
"volunteer": email,
"item_name": name,
"category": category,
"unit": unit,
"qty": qty,
"barcode": barcode,
"weather_type": weather_type,
"temp_c": temp_c,
"ingest_id": ingest_id
}
try:
sb.table("visit_items_p").insert(payload).execute()
except Exception:
# Legacy table fallback
payload.pop("ingest_id", None)
sb.table("visit_items").insert(payload).execute()
def load_items_for_visit(visit_id: int) -> list[dict]:
"""Enhanced item loading with better error handling"""
try:
return sb.table("visit_items_p").select("*").eq("visit_id", visit_id) \
.order("timestamp", desc=True).limit(500).execute().data or []
except Exception:
try:
return sb.table("visit_items").select("*").eq("visit_id", visit_id) \
.order("timestamp", desc=True).limit(500).execute().data or []
except Exception as e:
logger.warning(f"Failed to load items for visit {visit_id}: {e}")
return []
def delete_item(table: str, item_id: int):
"""Enhanced item deletion with better error handling"""
try:
sb.table(table).delete().eq("id", item_id).execute()
except Exception as e:
logger.error(f"Failed to delete item {item_id} from {table}: {e}")
raise e
# ------------------------ App Configuration Display ------------------------
st.sidebar.markdown("### โ๏ธ Configuration")
st.sidebar.info(f"""
**Provider:** `{PROVIDER}`
**Model:** `{GEMMA_MODEL}`
**Timezone:** `{TZ}`
**Cutoff:** {CUTOFF_HOUR}:00 PM
**Inactivity:** {INACTIVITY_MIN} min
""")
# ------------------------ Main App Execution ------------------------
if __name__ == "__main__":
try:
main()
except Exception as e:
logger.error(f"Application error: {e}")
st.error(f"โ Application error: {e}")
st.info("Please refresh the page or contact support if the issue persists.")