"""Garden Tracker: plants, seeds, schedules, hydro tanks, planting guide.""" import html import io import json import logging import os import re import threading import time import urllib.error import urllib.parse import urllib.request from datetime import date, datetime, timedelta from decimal import Decimal import pymysql from flask import Blueprint, Response, jsonify, request, send_from_directory from werkzeug.utils import secure_filename from blueprints.alerts import ensure_alert_state_table, send_sms_alert_detailed from config import ( GARDEN_CLIENT_DIST, GARDEN_LAT, GARDEN_LON, GARDEN_LOCATION_LABEL, GARDEN_REMINDER_CHECK_MINUTES, GARDEN_ZIP, MYSQL_DATABASE, MYSQL_HOST, MYSQL_PASSWORD, MYSQL_PORT, MYSQL_USER, PARTSINV_IMAGE_SEARCH_API_KEY, PARTSINV_IMAGE_SEARCH_CX, WEATHER_LAT, WEATHER_LON, ) logger = logging.getLogger(__name__) bp = Blueprint("gardentracker", __name__) DB_CONFIG = { "host": MYSQL_HOST, "port": MYSQL_PORT, "user": MYSQL_USER, "password": MYSQL_PASSWORD, "database": MYSQL_DATABASE, "cursorclass": pymysql.cursors.DictCursor, } TASK_TYPES = {"fertilize", "test_tank", "water", "prune", "harvest_check", "custom"} PLANT_STATUSES = {"active", "harvested", "died", "archived"} PLANT_SOURCE_TYPES = {"seed", "store_bought"} EVENT_TYPES = { "seed_start", "germinated", "true_leaves", "seeded", "transplanted", "fertilized", "harvested", "pruned", "watered", "died", "note", "custom", } _reminder_thread_started = False _reminder_lock = threading.Lock() _seed_type_fetch_inflight = set() _seed_type_fetch_lock = threading.Lock() _seed_type_image_retry_done = set() _seed_type_preserved_images = {} MAX_SEED_TYPE_IMAGES = 12 WIKI_IMAGE_LIMIT = 1 SEED_IMAGE_UPLOAD_DIR = "gardentracker_seed_uploads" def get_db(): return pymysql.connect(**DB_CONFIG) def _serialize(value): if isinstance(value, datetime): return value.isoformat() if isinstance(value, date): return value.isoformat() if isinstance(value, Decimal): return float(value) if isinstance(value, bytes): return value.decode("utf-8", errors="replace") return value def _serialize_row(row): if not row: return row return {k: _serialize(v) for k, v in row.items()} def _serialize_rows(rows): return [_serialize_row(r) for r in rows] def _parse_date(value): if value is None or value == "": return None if isinstance(value, date): return value try: return datetime.fromisoformat(str(value).replace("Z", "+00:00")).date() except ValueError: return None def _parse_datetime(value): if value is None or value == "": return None if isinstance(value, datetime): return value try: return datetime.fromisoformat(str(value).replace("Z", "+00:00")) except ValueError: return None def ensure_garden_tables(conn): with conn.cursor() as cur: cur.execute( """ CREATE TABLE IF NOT EXISTS GardenSettings ( id INT PRIMARY KEY DEFAULT 1, zip_code VARCHAR(10) NOT NULL DEFAULT '39429', lat DECIMAL(10, 6) DEFAULT NULL, lng DECIMAL(10, 6) DEFAULT NULL, notifications_enabled TINYINT(1) NOT NULL DEFAULT 1, low_seed_threshold INT NOT NULL DEFAULT 5, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS GardenPlant ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(120) NOT NULL, variety VARCHAR(120) DEFAULT NULL, location VARCHAR(120) DEFAULT NULL, status VARCHAR(20) NOT NULL DEFAULT 'active', planted_date DATE DEFAULT NULL, moisture_device_id VARCHAR(16) DEFAULT NULL, seed_id INT DEFAULT NULL, source_type VARCHAR(20) NOT NULL DEFAULT 'store_bought', label_id VARCHAR(16) DEFAULT NULL, notes TEXT DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) """ ) for col_sql in ( "ALTER TABLE GardenPlant ADD COLUMN seed_id INT DEFAULT NULL", "ALTER TABLE GardenPlant ADD COLUMN source_type VARCHAR(20) NOT NULL DEFAULT 'store_bought'", "ALTER TABLE GardenPlant ADD COLUMN label_id VARCHAR(16) DEFAULT NULL", ): try: cur.execute(col_sql) except pymysql.OperationalError: pass try: cur.execute("CREATE UNIQUE INDEX idx_garden_plant_label_id ON GardenPlant (label_id)") except pymysql.OperationalError: pass cur.execute( """ CREATE TABLE IF NOT EXISTS GardenPlantEvent ( id INT AUTO_INCREMENT PRIMARY KEY, plant_id INT NOT NULL, event_type VARCHAR(32) NOT NULL, event_date DATE NOT NULL, notes TEXT DEFAULT NULL, qty DECIMAL(10, 2) DEFAULT NULL, unit VARCHAR(32) DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_plant (plant_id), FOREIGN KEY (plant_id) REFERENCES GardenPlant(id) ON DELETE CASCADE ) """ ) for col_sql in ( "ALTER TABLE GardenPlantEvent ADD COLUMN days_since_start INT DEFAULT NULL", "ALTER TABLE GardenPlantEvent ADD COLUMN days_since_anchor VARCHAR(16) DEFAULT NULL", "ALTER TABLE GardenPlantEvent ADD COLUMN location VARCHAR(120) DEFAULT NULL", "ALTER TABLE GardenPlantEvent ADD COLUMN height DECIMAL(10, 2) DEFAULT NULL", ): try: cur.execute(col_sql) except pymysql.OperationalError: pass cur.execute( """ CREATE TABLE IF NOT EXISTS GardenSchedule ( id INT AUTO_INCREMENT PRIMARY KEY, plant_id INT DEFAULT NULL, tank_id INT DEFAULT NULL, task_type VARCHAR(32) NOT NULL, title VARCHAR(120) NOT NULL, interval_days INT NOT NULL DEFAULT 14, next_due DATETIME NOT NULL, enabled TINYINT(1) NOT NULL DEFAULT 1, notes TEXT DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_next_due (next_due), INDEX idx_plant (plant_id), FOREIGN KEY (plant_id) REFERENCES GardenPlant(id) ON DELETE CASCADE ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS GardenTaskCompletion ( id INT AUTO_INCREMENT PRIMARY KEY, schedule_id INT NOT NULL, completed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, notes TEXT DEFAULT NULL, INDEX idx_schedule (schedule_id), FOREIGN KEY (schedule_id) REFERENCES GardenSchedule(id) ON DELETE CASCADE ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS GardenSeedInventory ( id INT AUTO_INCREMENT PRIMARY KEY, crop_name VARCHAR(120) NOT NULL, variety VARCHAR(120) DEFAULT NULL, qty DECIMAL(10, 2) NOT NULL DEFAULT 0, unit VARCHAR(32) NOT NULL DEFAULT 'seeds', source VARCHAR(120) DEFAULT NULL, purchase_date DATE DEFAULT NULL, expiry_date DATE DEFAULT NULL, location VARCHAR(120) DEFAULT NULL, low_stock_threshold INT DEFAULT NULL, notes TEXT DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS GardenTank ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(120) NOT NULL, volume_gal DECIMAL(10, 2) DEFAULT NULL, solution_type VARCHAR(120) DEFAULT NULL, target_ph_min DECIMAL(4, 2) DEFAULT NULL, target_ph_max DECIMAL(4, 2) DEFAULT NULL, target_ec_min DECIMAL(6, 2) DEFAULT NULL, target_ec_max DECIMAL(6, 2) DEFAULT NULL, test_interval_days INT NOT NULL DEFAULT 7, notes TEXT DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS GardenTankReading ( id INT AUTO_INCREMENT PRIMARY KEY, tank_id INT NOT NULL, ph DECIMAL(4, 2) DEFAULT NULL, ec DECIMAL(6, 2) DEFAULT NULL, temp_f DECIMAL(5, 2) DEFAULT NULL, volume_gal DECIMAL(10, 2) DEFAULT NULL, notes TEXT DEFAULT NULL, recorded_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, INDEX idx_tank (tank_id), FOREIGN KEY (tank_id) REFERENCES GardenTank(id) ON DELETE CASCADE ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS GardenExternalCache ( cache_key VARCHAR(128) PRIMARY KEY, payload JSON NOT NULL, fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, expires_at TIMESTAMP NOT NULL ) """ ) cur.execute( """ CREATE TABLE IF NOT EXISTS GardenSeedTypeInfo ( cache_key VARCHAR(256) PRIMARY KEY, crop_name VARCHAR(120) NOT NULL, variety VARCHAR(120) DEFAULT NULL, summary TEXT DEFAULT NULL, wikipedia_title VARCHAR(256) DEFAULT NULL, wikipedia_url VARCHAR(512) DEFAULT NULL, info_source VARCHAR(32) DEFAULT NULL, source_url VARCHAR(512) DEFAULT NULL, source_name VARCHAR(120) DEFAULT NULL, images JSON DEFAULT NULL, planting JSON DEFAULT NULL, fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """ ) for col_sql in ( "ALTER TABLE GardenSeedTypeInfo ADD COLUMN info_source VARCHAR(32) DEFAULT NULL", "ALTER TABLE GardenSeedTypeInfo ADD COLUMN source_url VARCHAR(512) DEFAULT NULL", "ALTER TABLE GardenSeedTypeInfo ADD COLUMN source_name VARCHAR(120) DEFAULT NULL", "ALTER TABLE GardenSeedTypeInfo ADD COLUMN images_user_managed TINYINT(1) NOT NULL DEFAULT 0", ): try: cur.execute(col_sql) except pymysql.OperationalError: pass cur.execute( """ CREATE TABLE IF NOT EXISTS GardenSeedCompany ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(120) NOT NULL, base_url VARCHAR(512) NOT NULL, search_url_template VARCHAR(512) DEFAULT NULL, enabled TINYINT(1) NOT NULL DEFAULT 1, sort_order INT NOT NULL DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) """ ) cur.execute("SELECT COUNT(*) AS cnt FROM GardenSeedCompany") if int((cur.fetchone() or {}).get("cnt") or 0) == 0: for idx, (name, base_url, template) in enumerate(DEFAULT_SEED_COMPANIES): cur.execute( """ INSERT INTO GardenSeedCompany (name, base_url, search_url_template, enabled, sort_order) VALUES (%s, %s, %s, 1, %s) """, (name, base_url, template, idx), ) cur.execute("SELECT id FROM GardenSettings WHERE id = 1") if not cur.fetchone(): cur.execute( """ INSERT INTO GardenSettings (id, zip_code, lat, lng) VALUES (1, %s, %s, %s) """, (GARDEN_ZIP, GARDEN_LAT, GARDEN_LON), ) conn.commit() ensure_alert_state_table(conn) _backfill_plant_label_ids(conn) def _allocate_plant_label_ids(cur, count=1): cur.execute( """ SELECT MAX(CAST(SUBSTRING(label_id, 3) AS UNSIGNED)) AS n FROM GardenPlant WHERE label_id LIKE 'P-%' """ ) row = cur.fetchone() start = int(row["n"] or 0) + 1 return [f"P-{i:05d}" for i in range(start, start + count)] def _backfill_plant_label_ids(conn): with conn.cursor() as cur: cur.execute( "SELECT id FROM GardenPlant WHERE label_id IS NULL OR label_id = '' ORDER BY id" ) rows = cur.fetchall() if not rows: return labels = _allocate_plant_label_ids(cur, len(rows)) for row, label in zip(rows, labels): cur.execute("UPDATE GardenPlant SET label_id = %s WHERE id = %s", (label, row["id"])) conn.commit() _PLANT_SELECT = """ SELECT p.*, s.crop_name AS seed_crop_name, s.variety AS seed_variety, s.location AS seed_bin, s.source AS seed_source, EXISTS ( SELECT 1 FROM GardenPlantEvent dead_event WHERE dead_event.plant_id = p.id AND dead_event.event_type = 'died' ) AS has_died_event FROM GardenPlant p LEFT JOIN GardenSeedInventory s ON s.id = p.seed_id """ def _plant_type_crop_variety(row): if not row: return None, None crop = (row.get("seed_crop_name") or row.get("name") or "").strip() variety = (row.get("seed_variety") or row.get("variety") or "").strip() or None return crop or None, variety def _get_primary_type_image(conn, crop_name, variety): type_info = _get_seed_type_info(conn, crop_name, variety) images = (type_info or {}).get("images") or [] return images[0] if images else None def _serialize_plant(row, conn=None, include_primary_image=False): if not row: return row out = _serialize_row(row) if out.get("source_type") == "seed" and out.get("seed_id"): out["seed_label"] = _seed_display_label(row) if include_primary_image and conn is not None: crop_name, variety = _plant_type_crop_variety(row) if crop_name: out["primary_image"] = _get_primary_type_image(conn, crop_name, variety) return out def _seed_display_label(row): parts = [] if row.get("seed_bin"): parts.append(f"bin #{row['seed_bin']}") crop = row.get("seed_crop_name") or row.get("crop_name") if crop: parts.append(crop) variety = row.get("seed_variety") or row.get("variety") if variety: parts.append(variety) return " · ".join(parts) if parts else None def _serialize_plants(rows, conn=None, include_primary_image=False): return [_serialize_plant(r, conn=conn, include_primary_image=include_primary_image) for r in rows] def _as_date(value): if value is None: return None if isinstance(value, datetime): return value.date() if isinstance(value, date): return value return _parse_date(value) def _compute_days_since_start(cur, plant_id, plant, event_date): """Days from seed start or planting to event_date.""" event_date = _as_date(event_date) if not event_date: return None, None anchors = [] planted = _as_date(plant.get("planted_date") if plant else None) if planted and planted <= event_date: anchors.append(planted) cur.execute( """ SELECT event_date FROM GardenPlantEvent WHERE plant_id = %s AND event_type IN ('seed_start', 'seeded', 'transplanted') AND event_date <= %s ORDER BY event_date ASC, id ASC """, (plant_id, event_date), ) for row in cur.fetchall(): anchor = _as_date(row.get("event_date")) if anchor: anchors.append(anchor) if anchors: anchor = min(anchors) days = (event_date - anchor).days anchor_type = "seed_start" if (plant or {}).get("source_type") == "seed" else "planting" return max(0, days), anchor_type return None, None def _enrich_event(cur, plant_id, plant, event): if not event: return event out = _serialize_row(event) if out.get("days_since_start") is not None: if not out.get("days_since_anchor"): _, anchor = _compute_days_since_start(cur, plant_id, plant, event.get("event_date")) if anchor: out["days_since_anchor"] = anchor return out days, anchor = _compute_days_since_start(cur, plant_id, plant, event.get("event_date")) if days is not None: out["days_since_start"] = days out["days_since_anchor"] = anchor return out def _enrich_events(cur, plant_id, plant, events): return [_enrich_event(cur, plant_id, plant, ev) for ev in events] def _normalize_location(value): if value is None: return None loc = str(value).strip() return loc or None def _optional_float(value): if value is None or value == "": return None try: return float(value) except (TypeError, ValueError): return None def _sync_transplant_location(cur, plant_id, event_type, location): if event_type != "transplanted": return cur.execute( "UPDATE GardenPlant SET location = %s WHERE id = %s", (_normalize_location(location), plant_id), ) def _get_settings(conn): with conn.cursor() as cur: cur.execute("SELECT * FROM GardenSettings WHERE id = 1") return cur.fetchone() def _sql_dump_garden_tables(conn): out = io.StringIO() out.write("-- Garden Tracker backup\n") out.write(f"-- Generated at {datetime.now().isoformat()}\n\n") out.write("SET FOREIGN_KEY_CHECKS = 0;\n\n") with conn.cursor() as cur: cur.execute( """ SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_TYPE = 'BASE TABLE' AND TABLE_NAME LIKE 'Garden%' ORDER BY TABLE_NAME """ ) table_names = [row["TABLE_NAME"] for row in cur.fetchall()] for table_name in table_names: cur.execute(f"SHOW CREATE TABLE `{table_name}`") create_row = cur.fetchone() create_sql = create_row.get("Create Table") if create_row else None if not create_sql: continue out.write(f"-- Table: {table_name}\n") out.write(f"DROP TABLE IF EXISTS `{table_name}`;\n") out.write(f"{create_sql};\n\n") cur.execute(f"SELECT * FROM `{table_name}`") rows = cur.fetchall() if not rows: continue columns = list(rows[0].keys()) col_sql = ", ".join([f"`{c}`" for c in columns]) for row in rows: vals = ", ".join([conn.literal(row.get(c)) for c in columns]) out.write(f"INSERT INTO `{table_name}` ({col_sql}) VALUES ({vals});\n") out.write("\n") out.write("SET FOREIGN_KEY_CHECKS = 1;\n") return out.getvalue() def _fetch_json_url(url, timeout=15): req = urllib.request.Request(url, headers={"User-Agent": "GardenTracker/1.0"}) with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode()) def _cache_get(conn, key): with conn.cursor() as cur: cur.execute( "SELECT payload, expires_at FROM GardenExternalCache WHERE cache_key = %s", (key,), ) row = cur.fetchone() if not row: return None if row["expires_at"] and row["expires_at"] > datetime.utcnow(): payload = row["payload"] if isinstance(payload, str): return json.loads(payload) return payload return None def _cache_set(conn, key, payload, ttl_hours): expires = datetime.utcnow() + timedelta(hours=ttl_hours) with conn.cursor() as cur: cur.execute( """ INSERT INTO GardenExternalCache (cache_key, payload, expires_at) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE payload = VALUES(payload), fetched_at = CURRENT_TIMESTAMP, expires_at = VALUES(expires_at) """, (key, json.dumps(payload), expires), ) conn.commit() WIKIPEDIA_API = "https://en.wikipedia.org" WIKIMEDIA_IMAGE_HOSTS = frozenset({"upload.wikimedia.org", "commons.wikimedia.org"}) SEED_COMPANY_CDN_HOSTS = frozenset({"cdn.commercev3.net"}) SEED_COMPANY_IMAGE_SKIP_TOKENS = ( "logo", "favicon", "mobilemenu", "mobilecart", "empty_cart", "green-lush", "header", "footer", "newsletter", "search-icon", "group.png", "ngb-footer", "fontawesome", "brands_sec", "solid_sec", "regular_sec", ) DEFAULT_SEED_COMPANIES = ( ("Johnny's Selected Seeds", "https://www.johnnyseeds.com/", "{base_url}/search/?q={query}"), ("Totally Tomatoes", "https://www.totallytomato.com/", "{base_url}/search?q={query}"), ) GENERIC_VARIETY_TERMS = frozenset({ "hybrid", "heirloom", "organic", "open pollinated", "open-pollinated", "op", "f1", "f2", "gmo", "non-gmo", "non gmo", "treated", "untreated", "pelleted", "raw", "standard", "common", "mixed", "mix", "blend", "variety", "unknown", "generic", "regular", "traditional", "improved", }) CROP_WIKI_ALIASES = { "arugula": ("eruca", "rocket", "roquette"), "corn": ("maize", "zea mays", "zea"), "eggplant": ("aubergine", "solanum melongena"), "cilantro": ("coriander", "coriandrum"), "scallion": ("green onion", "allium fistulosum"), "zucchini": ("courgette", "cucurbita"), } def _normalize_image_url(url): u = (url or "").strip() if not u: return "" if u.startswith("//"): return "https:" + u if u.startswith("http://"): return "https://" + u[7:] return u def _wikimedia_image_identity(url): """Canonical identity for deduping the same file at different thumbnail sizes.""" url = _normalize_image_url(url) if not url: return "" path = urllib.parse.urlparse(url).path if "/thumb/" in path: rest = path.split("/thumb/", 1)[1] parts = rest.split("/") if len(parts) >= 3: return "/".join(parts[:-1]).lower() for marker in ("/wikipedia/commons/", "/wikipedia/en/", "/commons/"): if marker in path: return path.split(marker, 1)[1].lower() return path.lower() def _append_unique_image(images, seen_identities, img, limit=2): if not img or len(images) >= limit: return url = _normalize_image_url(img.get("url")) if not url: return ident = _wikimedia_image_identity(url) or url if ident in seen_identities: return seen_identities.add(ident) images.append({ "url": url, "title": img.get("title"), "source": img.get("source") or "wikipedia", }) def _seed_upload_root(): base = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) return os.path.join(base, SEED_IMAGE_UPLOAD_DIR) def _seed_upload_subdir(cache_key): safe = re.sub(r"[^\w\-|]+", "_", cache_key or "unknown")[:80] return safe or "unknown" def _seed_upload_display_url(file_path): if not file_path: return None return f"/gardentracker/api/seed-type-upload/{urllib.parse.quote(file_path, safe='/')}" def _seed_image_display_url(url, file_path=None): if file_path: return _seed_upload_display_url(file_path) normalized = _normalize_image_url(url) if not normalized: return None return f"/gardentracker/api/seed-image?url={urllib.parse.quote(normalized, safe='')}" def _parse_images_raw(images): if images is None: return [] if isinstance(images, (bytes, bytearray)): images = images.decode("utf-8", errors="replace") if isinstance(images, str): try: images = json.loads(images) except json.JSONDecodeError: return [] if not isinstance(images, list): return [] out = [] for img in images: if isinstance(img, str): img = {"url": img, "source": "wikipedia"} if isinstance(img, dict) and (img.get("url") or img.get("file_path")): out.append(img) return out def _image_list_identity(img): if not isinstance(img, dict): return None if img.get("file_path"): return f"upload:{img['file_path']}" url = _normalize_image_url(img.get("url")) if not url: return None return _wikimedia_image_identity(url) or url def _requested_image_url(): data = request.get_json(force=True, silent=True) or {} return (request.args.get("url") or data.get("url") or "").strip() def _requested_image_index(): data = request.get_json(force=True, silent=True) or {} raw = request.args.get("index", data.get("index")) if raw is None or raw == "": return None try: return int(raw) except (TypeError, ValueError): return None def _image_matches_request(stored_img, requested_url): req = _normalize_image_url(requested_url) if not req or not isinstance(stored_img, dict): return False stored_norm = _normalize_image_url(stored_img.get("url")) if stored_norm and stored_norm == req: return True file_path = (stored_img.get("file_path") or "").strip() if file_path and req in (f"seed-upload://{file_path}", stored_norm): return True return _image_list_identity(stored_img) == _image_list_identity({"url": req}) def _type_info_images_user_managed(conn, cache_key): if not _has_images_user_managed_column(conn): return False with conn.cursor() as cur: cur.execute( "SELECT images_user_managed FROM GardenSeedTypeInfo WHERE cache_key = %s", (cache_key,), ) row = cur.fetchone() return bool(row and row.get("images_user_managed")) def _normalize_seed_images(images): if images is None: return [] if isinstance(images, (bytes, bytearray)): images = images.decode("utf-8", errors="replace") if isinstance(images, str): try: images = json.loads(images) except json.JSONDecodeError: return [] if not isinstance(images, list): return [] out = [] seen = set() for img in images: if isinstance(img, str): img = {"url": img, "source": "wikipedia"} if not isinstance(img, dict): continue file_path = (img.get("file_path") or "").strip() url = _normalize_image_url(img.get("url")) if not url and not file_path: continue ident = _image_list_identity(img) if not ident or ident in seen: continue seen.add(ident) entry = { "url": url or f"seed-upload://{file_path}", "title": img.get("title"), "source": img.get("source") or "wikipedia", "display_url": _seed_image_display_url(url, file_path=file_path or None), } if file_path: entry["file_path"] = file_path out.append(entry) if len(out) >= MAX_SEED_TYPE_IMAGES: break return out def _seed_type_cache_key(crop_name, variety): crop = (crop_name or "").strip().lower() var = (variety or "").strip().lower() return f"{crop}|{var}" def _map_cropgraph_guide(crop_data): return { "slug": crop_data.get("slug"), "common_name": crop_data.get("commonName"), "scientific_name": crop_data.get("scientificName"), "category": crop_data.get("category"), "season": crop_data.get("season"), "days_to_harvest": crop_data.get("daysToHarvest"), "windows": crop_data.get("windows") or [], "notes": crop_data.get("notes"), "zone_range": crop_data.get("zoneRange"), } def _fetch_cropgraph_planting(conn, crop_name): crop = (crop_name or "").strip() if not crop: return None slug = crop.lower().replace(" ", "-") cache_key = f"crop_detail:{slug}" cached = _cache_get(conn, cache_key) if cached: return cached try: data = _fetch_json_url(f"https://api.cropgraph.com/api/crop/{slug}") except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError): try: search = _fetch_json_url( f"https://api.cropgraph.com/api/search?q={urllib.parse.quote(crop)}" ) results = search.get("results") or [] if not results: return None slug = results[0].get("slug") or slug data = _fetch_json_url(f"https://api.cropgraph.com/api/crop/{slug}") except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError): return None crop_data = data.get("crop") or data guide = _map_cropgraph_guide(crop_data) _cache_set(conn, cache_key, guide, 24) return guide def _wiki_page_summary(title): encoded = urllib.parse.quote((title or "").replace(" ", "_"), safe="") if not encoded: return None try: return _fetch_json_url(f"{WIKIPEDIA_API}/api/rest_v1/page/summary/{encoded}") except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError): return None def _wiki_search_titles(query, limit=5): q = (query or "").strip() if not q: return [] try: data = _fetch_json_url( f"{WIKIPEDIA_API}/w/rest.php/v1/search/title?q={urllib.parse.quote(q)}&limit={limit}" ) return [p.get("title") for p in (data.get("pages") or []) if p.get("title")] except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError): return [] def _image_from_wiki_summary(summary): if not summary: return None thumb = summary.get("thumbnail") or {} url = thumb.get("source") or (summary.get("originalimage") or {}).get("source") url = _normalize_image_url(url) if not url: return None return { "url": url, "title": summary.get("title"), "source": "wikipedia", } def _wiki_page_images(title, limit=2): title = (title or "").strip() if not title: return [] params = urllib.parse.urlencode( { "action": "query", "titles": title, "prop": "pageimages", "format": "json", "pithumbsize": 500, "piprop": "thumbnail|name", } ) try: data = _fetch_json_url(f"{WIKIPEDIA_API}/w/api.php?{params}") except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError): return [] images = [] for page in (data.get("query") or {}).get("pages", {}).values(): if page.get("missing"): continue thumb = _normalize_image_url((page.get("thumbnail") or {}).get("source")) if thumb: images.append({ "url": thumb, "title": page.get("title"), "source": "wikipedia", }) return images[:limit] def _is_generic_variety(variety): v = (variety or "").strip().lower() if not v or len(v) <= 2: return True return v in GENERIC_VARIETY_TERMS def _crop_match_terms(crop_name): crop = (crop_name or "").strip().lower() terms = set() if crop: terms.add(crop) for word in crop.split(): if len(word) >= 3: terms.add(word) for alias in CROP_WIKI_ALIASES.get(crop, ()): terms.add(alias.lower()) return terms def _wiki_title_matches_crop(title, crop_name, variety=None): title_l = (title or "").lower() if not title_l: return False for term in _crop_match_terms(crop_name): if term in title_l: return True variety = (variety or "").strip().lower() if variety and not _is_generic_variety(variety) and variety in title_l: return True return False def _wiki_crop_title_candidates(crop_name, limit=8): crop_name = (crop_name or "").strip() if not crop_name: return [] seen = set() out = [] for title in _wiki_search_titles(crop_name, limit=limit): if title and title.lower() not in seen: seen.add(title.lower()) out.append(title) if len(out) < 3: for title in _wiki_fulltext_titles(crop_name, limit=limit): if title and title.lower() not in seen: seen.add(title.lower()) out.append(title) return out def _filter_titles_for_crop(titles, crop_name, variety=None): filtered = [t for t in (titles or []) if _wiki_title_matches_crop(t, crop_name, variety)] if filtered: return filtered allowlist = {t.lower() for t in _wiki_crop_title_candidates(crop_name)} return [t for t in (titles or []) if (t or "").lower() in allowlist] def _variety_for_wiki_search(variety): variety = (variety or "").strip() if _is_generic_variety(variety): return "" return variety def _wiki_search_queries(crop_name, variety): crop_name = (crop_name or "").strip() variety = _variety_for_wiki_search(variety) queries = [] if variety: queries.append(f"{variety} {crop_name}") queries.append(f"{crop_name} {variety}") if crop_name: queries.append(crop_name) seen = set() out = [] for q in queries: key = q.lower() if key and key not in seen: seen.add(key) out.append(q) return out def _wiki_resolve_titles(crop_name, variety, limit=5): variety_for_match = variety for query in _wiki_search_queries(crop_name, variety): titles = _wiki_search_titles(query, limit=limit) relevant = _filter_titles_for_crop(titles, crop_name, variety_for_match) if relevant: return relevant, query crop_titles = _wiki_crop_title_candidates(crop_name, limit=limit) if crop_titles: return crop_titles, crop_name return [], None def _wiki_fulltext_titles(query, limit=5): q = (query or "").strip() if not q: return [] params = urllib.parse.urlencode( { "action": "query", "list": "search", "srsearch": q, "srlimit": limit, "format": "json", } ) try: data = _fetch_json_url(f"{WIKIPEDIA_API}/w/api.php?{params}") except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError): return [] return [hit.get("title") for hit in (data.get("query") or {}).get("search", []) if hit.get("title")] def _wiki_file_is_photo(file_title): name = (file_title or "").lower() if not name.startswith("file:"): return False if not any(ext in name for ext in (".jpg", ".jpeg", ".png", ".webp")): return False skip_tokens = ( "commons-logo", "edit-clear", "symbol", "icon-", "flag", "map", "logo", "pictogram", "ambox", "wikimedia", "question_book", "crystal", ) return not any(token in name for token in skip_tokens) def _wiki_extra_page_images(title, skip_identities=None, limit=2): title = (title or "").strip() if not title or limit <= 0: return [] skip_identities = set(skip_identities or []) params = urllib.parse.urlencode( { "action": "query", "titles": title, "prop": "images", "imlimit": 30, "format": "json", } ) try: data = _fetch_json_url(f"{WIKIPEDIA_API}/w/api.php?{params}") except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError): return [] file_titles = [] for page in (data.get("query") or {}).get("pages", {}).values(): for item in page.get("images") or []: file_title = item.get("title") if _wiki_file_is_photo(file_title): file_titles.append(file_title) if not file_titles: return [] images = [] batch_size = 8 for start in range(0, min(len(file_titles), 24), batch_size): batch = file_titles[start:start + batch_size] info_params = urllib.parse.urlencode( { "action": "query", "titles": "|".join(batch), "prop": "imageinfo", "iiprop": "url", "iiurlwidth": 500, "format": "json", } ) try: info_data = _fetch_json_url(f"{WIKIPEDIA_API}/w/api.php?{info_params}") except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError): continue for page in (info_data.get("query") or {}).get("pages", {}).values(): info = (page.get("imageinfo") or [{}])[0] url = _normalize_image_url(info.get("thumburl") or info.get("url")) if not url: continue ident = _wikimedia_image_identity(url) or url if ident in skip_identities: continue skip_identities.add(ident) images.append({ "url": url, "title": page.get("title"), "source": "wikipedia", }) if len(images) >= limit: return images return images def _collect_wiki_images(titles, crop_name, variety=None, limit=WIKI_IMAGE_LIMIT): images = [] seen_identities = set() for title in _filter_titles_for_crop(titles, crop_name, variety): if len(images) >= limit: break page_imgs = _wiki_page_images(title, limit=1) if page_imgs: _append_unique_image(images, seen_identities, page_imgs[0], limit) else: primary = _wiki_page_summary(title) _append_unique_image(images, seen_identities, _image_from_wiki_summary(primary), limit) if len(images) >= limit: break if limit > 1: for extra in _wiki_extra_page_images( title, skip_identities=seen_identities, limit=limit - len(images), ): _append_unique_image(images, seen_identities, extra, limit) if len(images) >= limit: break return images[:limit] def _get_seed_companies(conn, enabled_only=False): with conn.cursor() as cur: sql = "SELECT * FROM GardenSeedCompany" if enabled_only: sql += " WHERE enabled = 1" sql += " ORDER BY sort_order, name" cur.execute(sql) return cur.fetchall() def _company_host(base_url): return (urllib.parse.urlparse(base_url).hostname or "").lower() PRODUCT_URL_MIN_MATCH_SCORE = 50 def _is_allowed_product_url(conn, url): parsed = urllib.parse.urlparse(url) host = (parsed.hostname or "").lower() if not host or parsed.scheme not in ("http", "https"): return False for company in _get_seed_companies(conn, enabled_only=True): company_host = _company_host(company.get("base_url")) if company_host and (host == company_host or host.endswith("." + company_host)): return True return False def _source_name_for_url(conn, url): host = _company_host(url) for company in _get_seed_companies(conn, enabled_only=True): if _company_host(company.get("base_url")) == host: return company.get("name") return None def _allowed_image_hosts(conn): hosts = set(WIKIMEDIA_IMAGE_HOSTS) | set(SEED_COMPANY_CDN_HOSTS) for company in _get_seed_companies(conn, enabled_only=True): host = _company_host(company.get("base_url")) if host: hosts.add(host) return hosts def _image_host_allowed(host, allowed_hosts): host = (host or "").lower() if not host: return False if host in allowed_hosts: return True for allowed in allowed_hosts: if host == allowed or host.endswith("." + allowed): return True return False def _build_seed_search_query(crop_name, variety): crop_name = (crop_name or "").strip() variety = _variety_for_wiki_search(variety) if variety: return f"{variety} {crop_name}".strip() return crop_name def _crop_plural_slug(crop_name): slug = re.sub(r"[^a-z0-9]+", "-", (crop_name or "").strip().lower()).strip("-") if not slug: return slug if slug.endswith("s"): return slug if slug.endswith("y") and len(slug) > 1 and slug[-2] not in "aeiou": return slug[:-1] + "ies" if slug.endswith(("ch", "sh", "x", "z")): return slug + "es" return slug + "s" def _seed_company_search_queries(crop_name, variety): crop_name = (crop_name or "").strip() queries = [] eff = _effective_variety(variety) if eff: apostrophe_free = eff.replace("'", "").replace("’", "").strip() if apostrophe_free and apostrophe_free.lower() != eff.lower(): queries.append(f"{apostrophe_free} {crop_name}".strip()) queries.append(apostrophe_free) if re.search(r"\bog\b", eff, re.I): organic_var = re.sub(r"\bOG\b", "Organic", eff) queries.append(f"{organic_var} {crop_name}".strip()) organic_free = organic_var.replace("'", "").replace("’", "").strip() if organic_free and organic_free.lower() != organic_var.lower(): queries.append(f"{organic_free} {crop_name}".strip()) stripped = re.sub(r"\s+OG\s*$", "", eff, flags=re.I).strip() if stripped and stripped.lower() != eff.lower(): queries.append(f"{stripped} {crop_name}".strip()) queries.append(stripped) stripped_free = stripped.replace("'", "").replace("’", "").strip() if stripped_free and stripped_free.lower() != stripped.lower(): queries.append(f"{stripped_free} {crop_name}".strip()) queries.append(stripped_free) queries.append(f"{eff} {crop_name}".strip()) queries.append(eff) if crop_name: queries.append(crop_name) seen = set() out = [] for query in queries: key = query.lower() if query and key not in seen: seen.add(key) out.append(query) return out def _is_company_product_path(path): path = (path or "").lower() if any( skip in path for skip in ( "/gift-certificate", "'+produrl+'", "produrl", "/viewcart", "/create_review", ) ): return False if "/product/" in path: return bool(re.search(r"/product/[^/]+(?:/\d+)?/?$", path)) if ".html" in path or "/herbs/" in path or "/vegetables/" in path: return True return False def _is_seed_company_image_url(url): u = (url or "").lower() if not u: return False if any(skip in u for skip in SEED_COMPANY_IMAGE_SKIP_TOKENS): return False if "/images/popup/" in u or "/images/products/" in u: return True if "cdn.commercev3.net" in u and re.search( r"/images/[^/]*\d{3,}[^/]*\.(?:jpg|jpeg|png|webp)", u, ): return True return False def _company_post_search_links(company, search_queries): host = _company_host(company.get("base_url")) if host != "www.totallytomato.com": return [] base_url = (company.get("base_url") or "").rstrip("/") links = [] for query in search_queries: try: payload = urllib.parse.urlencode({ "action": "Search", "search_type": "prodcat", "keyword": query, }).encode() req = urllib.request.Request( f"{base_url}/index.php", data=payload, headers={ "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36" ), "Content-Type": "application/x-www-form-urlencoded", "Accept": "text/html,application/xhtml+xml;q=0.9,*/*;q=0.8", "Referer": f"{base_url}/", }, method="POST", ) with urllib.request.urlopen(req, timeout=20) as resp: search_html = resp.read().decode("utf-8", errors="replace") links.extend(_extract_company_product_links(search_html, company)) except (urllib.error.URLError, urllib.error.HTTPError, OSError): continue return list(dict.fromkeys(links)) def _collect_company_product_links(company, crop_name, variety, search_queries): links = [] seen = set() def add(new_links): for link in new_links or []: if link and link not in seen: seen.add(link) links.append(link) for search_query in search_queries: try: search_html = _fetch_html_url(_company_search_url(company, search_query)) except (urllib.error.URLError, urllib.error.HTTPError, OSError): search_html = "" add(_extract_company_product_links(search_html, company)) add(_company_browse_links(company, crop_name)) add(_company_post_search_links(company, search_queries)) domain = _company_host(company.get("base_url")) if domain: for search_query in search_queries: add(_google_site_search_links(domain, search_query)) return links def _company_browse_links(company, crop_name): host = _company_host(company.get("base_url")) base_url = (company.get("base_url") or "").rstrip("/") crop_name = (crop_name or "").strip() if not host or not base_url or not crop_name: return [] paths = [] if host == "www.johnnyseeds.com": plural = _crop_plural_slug(crop_name) if plural: paths.append(f"/vegetables/{plural}/") elif host == "www.totallytomato.com": plural = _crop_plural_slug(crop_name) if plural: paths.append(f"/vegetables/vegetable-seeds/{plural}-seeds") links = [] for path in paths: try: browse_html = _fetch_html_url(base_url + path) except (urllib.error.URLError, urllib.error.HTTPError, OSError): continue links.extend(_extract_company_product_links(browse_html, company)) return list(dict.fromkeys(links)) def _variety_match_forms(variety): norm_var = _normalize_match_text(variety) forms = [norm_var] if re.search(r"\bog\b", norm_var): forms.append(re.sub(r"\bog\b", "organic", norm_var)) compact = norm_var.replace(" ", "") if compact: forms.append(compact) code_match = re.match(r"^([a-z])\s+(\d+)\b", norm_var) if code_match: forms.append(f"{code_match.group(1)}{code_match.group(2)}") return list(dict.fromkeys(forms)) def _effective_variety(variety): variety = (variety or "").strip() if not variety or _is_generic_variety(variety): return "" return variety def _normalize_match_text(text): text = (text or "").lower() text = text.replace("-", " ").replace("_", " ") text = re.sub(r"[^a-z0-9]+", " ", text) return " ".join(text.split()) def _variety_match_score(variety, *texts): variety = _effective_variety(variety) if not variety: return 100 combined = _normalize_match_text(" ".join(texts)) if not combined: return 0 compact_text = combined.replace(" ", "") for form in _variety_match_forms(variety): if form in combined: return 100 compact_form = form.replace(" ", "") if len(compact_form) >= 4 and compact_form in compact_text: return 98 norm_var = _normalize_match_text(variety) var_tokens = [] for token in norm_var.split(): if len(token) >= 3: var_tokens.append(token) elif token in ("og", "op", "f1", "f2"): var_tokens.append(token) code_match = re.match(r"^([a-z])\s+(\d+)\b", norm_var) if code_match: var_tokens.append(f"{code_match.group(1)}{code_match.group(2)}") if not var_tokens: return 100 if norm_var.replace(" ", "") in compact_text else 0 expanded = combined if "og" in var_tokens: expanded += " organic" matched = sum(1 for token in var_tokens if token in expanded) ratio = matched / len(var_tokens) if ratio >= 1.0: return 95 if ratio >= 0.5: return 65 if matched: return 35 return 0 def _crop_match_score(crop_name, *texts): crop_name = (crop_name or "").strip() if not crop_name: return 100 norm_crop = _normalize_match_text(crop_name) combined = _normalize_match_text(" ".join(texts)) if not combined: return 0 if norm_crop in combined: return 100 crop_tokens = [t for t in norm_crop.split() if len(t) >= 3] if not crop_tokens: return 100 if norm_crop in combined.replace(" ", "") else 0 matched = sum(1 for token in crop_tokens if token in combined) return int(100 * matched / len(crop_tokens)) def _score_seed_candidate(url, crop_name, variety, title=None, summary=None): texts = [url or "", title or "", summary or ""] crop_score = _crop_match_score(crop_name, *texts) if crop_score < 40: return 0 variety_score = _variety_match_score(variety, *texts) if _effective_variety(variety) and variety_score < 50: return 0 if _effective_variety(variety): score = variety_score * 0.75 + crop_score * 0.25 else: score = float(crop_score) if "seed" in _normalize_match_text(url): score += 5 return score def _company_search_url(company, query): base_url = (company.get("base_url") or "").rstrip("/") template = ( company.get("search_url_template") or "{base_url}/search/?q={query}" ) encoded = urllib.parse.quote(query) return template.format(base_url=base_url, query=encoded, encoded_query=encoded) def _fetch_html_url(url, timeout=15): req = urllib.request.Request( url, headers={ "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36" ), "Accept": "text/html,application/xhtml+xml;q=0.9,*/*;q=0.8", "Accept-Language": "en-US,en;q=0.9", }, ) with urllib.request.urlopen(req, timeout=timeout) as resp: return resp.read().decode("utf-8", errors="replace") def _html_meta_content(page_html, name=None, prop=None): if prop: attr = f'property="{prop}"' else: attr = f'name="{name}"' match = re.search( rf"{re.escape(attr)}\s+content=(['\"])(.*?)\1", page_html, re.I | re.S, ) if not match: return None return html.unescape(match.group(2).strip()) def _score_seed_link(url, crop_name, variety): return _score_seed_candidate(url, crop_name, variety) def _extract_company_product_links(html, company): base_url = (company.get("base_url") or "").rstrip("/") base_host = _company_host(base_url) links = [] for href in re.findall(r'href="([^"]+)"', html or ""): href = href.replace("&", "&") if href.startswith("/"): href = base_url + href if not href.startswith("http"): continue parsed = urllib.parse.urlparse(href) if base_host and parsed.hostname and base_host not in parsed.hostname: continue path = (parsed.path or "").lower() if any( skip in path for skip in ( "/search", "/customer-", "/growers-library", "/shop-by-color", "/cart", "/account", "/login", "/shipping", "/privacy", ) ): continue if _is_company_product_path(path): links.append(parsed.scheme + "://" + parsed.netloc + parsed.path.rstrip("/")) return list(dict.fromkeys(links)) def _google_site_search_links(domain, query, limit=5): if not PARTSINV_IMAGE_SEARCH_API_KEY or not PARTSINV_IMAGE_SEARCH_CX: return [] params = urllib.parse.urlencode( { "key": PARTSINV_IMAGE_SEARCH_API_KEY, "cx": PARTSINV_IMAGE_SEARCH_CX, "q": f"site:{domain} {query}", "num": min(limit, 10), } ) try: data = _fetch_json_url( f"https://www.googleapis.com/customsearch/v1?{params}", timeout=12, ) except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError): return [] links = [] for item in data.get("items") or []: link = item.get("link") if link: links.append(link.split("?")[0]) return list(dict.fromkeys(links))[:limit] def _extract_product_page_info(page_url): try: page_html = _fetch_html_url(page_url) except (urllib.error.URLError, urllib.error.HTTPError, OSError): return None def meta_property(prop): return _html_meta_content(page_html, prop=prop) def meta_name(name): return _html_meta_content(page_html, name=name) title = meta_property("og:title") or meta_name("title") if not title: title_match = re.search(r"([^<]+)", page_html, re.I) if title_match: title = html.unescape(title_match.group(1).strip()) title = re.sub(r"\s*\|.*$", "", title).strip() if not title: h1_match = re.search(r"]*>(.*?)", page_html, re.I | re.S) if h1_match: title = html.unescape(re.sub(r"<[^>]+>", "", h1_match.group(1))) title = " ".join(title.split()) summary = meta_property("og:description") or meta_name("description") if not summary: lead_match = re.search( r"]*>\s*((?:A [^<]{40,600}|(?:A blend of|This)[^<]{20,600}))

", page_html, re.I | re.S, ) if lead_match: summary = html.unescape(re.sub(r"<[^>]+>", "", lead_match.group(1))) summary = " ".join(summary.split()) images = [] seen = set() def add_image(url): normalized = _normalize_image_url(url) if not normalized or not _is_seed_company_image_url(normalized): return ident = _wikimedia_image_identity(normalized) or normalized if ident in seen: return seen.add(ident) images.append({ "url": normalized, "title": title, "source": "seed_company", }) for url in re.findall( r'https://[^"\']+/images/products/[^"\']+\.(?:jpg|jpeg|png|webp)', page_html, re.I, ): add_image(url) for url in re.findall( r'https://cdn\.commercev3\.net/[^"\']+\.(?:jpg|jpeg|png|webp)', page_html, re.I, ): add_image(url) og_image = _normalize_image_url(meta_property("og:image")) if og_image: add_image(og_image) for match in re.finditer( r'', page_html, re.S, ): try: payload = json.loads(match.group(1)) except json.JSONDecodeError: continue items = payload if isinstance(payload, list) else [payload] for item in items: if not isinstance(item, dict) or item.get("@type") != "Product": continue if not summary: summary = item.get("description") if not title: title = item.get("name") image = item.get("image") if isinstance(image, list): image = image[0] if image else None if isinstance(image, dict): image = image.get("url") image = _normalize_image_url(image) if image: add_image(image) if not summary and title: summary = title if not summary and not images: return None if title: title = html.unescape(title) title = " ".join(title.split()) if summary: summary = html.unescape(summary) summary = " ".join(summary.split()) return { "title": title, "summary": summary, "images": images[:2], "source_url": page_url, } def _merge_seed_images(existing, new_images, limit=MAX_SEED_TYPE_IMAGES): merged = list(existing or []) seen = {_image_list_identity(img) for img in merged} seen.discard(None) for img in new_images or []: ident = _image_list_identity(img) if not ident or ident in seen: continue seen.add(ident) merged.append(img) if len(merged) >= limit: break return merged[:limit] def _ensure_seed_type_info_row(conn, cache_key, crop_name, variety): with conn.cursor() as cur: cur.execute("SELECT cache_key FROM GardenSeedTypeInfo WHERE cache_key = %s", (cache_key,)) if cur.fetchone(): return cur.execute( """ INSERT INTO GardenSeedTypeInfo (cache_key, crop_name, variety, images) VALUES (%s, %s, %s, %s) """, (cache_key, crop_name, variety or None, json.dumps([])), ) conn.commit() def _has_images_user_managed_column(conn): with conn.cursor() as cur: cur.execute( """ SELECT COUNT(*) AS cnt FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'GardenSeedTypeInfo' AND COLUMN_NAME = 'images_user_managed' """ ) return int((cur.fetchone() or {}).get("cnt") or 0) > 0 def _load_type_info_images(conn, cache_key): with conn.cursor() as cur: cur.execute("SELECT images FROM GardenSeedTypeInfo WHERE cache_key = %s", (cache_key,)) row = cur.fetchone() return _parse_images_raw(row.get("images") if row else None) def _save_type_info_images(conn, cache_key, crop_name, variety, images, user_managed=False): _ensure_seed_type_info_row(conn, cache_key, crop_name, variety) with conn.cursor() as cur: payload = json.dumps(images[:MAX_SEED_TYPE_IMAGES]) if user_managed and _has_images_user_managed_column(conn): cur.execute( """ UPDATE GardenSeedTypeInfo SET images = %s, images_user_managed = 1 WHERE cache_key = %s """, (payload, cache_key), ) else: cur.execute( "UPDATE GardenSeedTypeInfo SET images = %s WHERE cache_key = %s", (payload, cache_key), ) conn.commit() def _delete_seed_upload_file(file_path): if not file_path or ".." in file_path or file_path.startswith("/"): return upload_root = _seed_upload_root() full = os.path.join(upload_root, file_path) if os.path.isfile(full): try: os.remove(full) except OSError: logger.warning("Could not delete seed upload %s", file_path) def _fetch_seed_company_info(conn, crop_name, variety): companies = _get_seed_companies(conn, enabled_only=True) if not companies: return { "summary": None, "images": [], "info_source": None, "source_url": None, "source_name": None, } query = _build_seed_search_query(crop_name, variety) if not query: return { "summary": None, "images": [], "info_source": None, "source_url": None, "source_name": None, } best = None best_score = -1 search_queries = _seed_company_search_queries(crop_name, variety) for company in companies: links = _collect_company_product_links(company, crop_name, variety, search_queries) candidates = [] for link in dict.fromkeys(links): url_score = _score_seed_candidate(link, crop_name, variety) if url_score <= 0: continue candidates.append((url_score, link)) candidates.sort(reverse=True) for url_score, link in candidates[:6]: page = _extract_product_page_info(link) if not page: continue full_score = _score_seed_candidate( link, crop_name, variety, title=page.get("title"), summary=page.get("summary"), ) if full_score <= 0: continue total_score = full_score + (4 if page.get("summary") else 0) + len(page.get("images") or []) if total_score > best_score: best_score = total_score best = { "summary": page.get("summary"), "images": page.get("images") or [], "info_source": "seed_company", "source_url": page.get("source_url"), "source_name": company.get("name"), } if best and best.get("summary") and len(best.get("images") or []) >= 2: break if best and best.get("summary") and len(best.get("images") or []) >= 2: break if best: extra_images = [] seen = { _wikimedia_image_identity(img.get("url")) or img.get("url") for img in best.get("images") or [] } for company in companies: if len(best.get("images") or []) >= 2: break try: search_html = _fetch_html_url(_company_search_url(company, query)) except (urllib.error.URLError, urllib.error.HTTPError, OSError): continue links = _extract_company_product_links(search_html, company) for link in dict.fromkeys(links): if link == best.get("source_url"): continue url_score = _score_seed_candidate(link, crop_name, variety) if url_score <= 0: continue page = _extract_product_page_info(link) if not page: continue full_score = _score_seed_candidate( link, crop_name, variety, title=page.get("title"), summary=page.get("summary"), ) if full_score <= 0: continue for img in page.get("images") or []: ident = _wikimedia_image_identity(img.get("url")) or img.get("url") if ident and ident not in seen: seen.add(ident) extra_images.append(img) if len(best.get("images") or []) + len(extra_images) >= 2: break if len(best.get("images") or []) + len(extra_images) >= 2: break best["images"] = _merge_seed_images(best.get("images"), extra_images, limit=2) return best return { "summary": None, "images": [], "info_source": None, "source_url": None, "source_name": None, } def _fetch_wikipedia_seed_info(crop_name, variety): crop_name = (crop_name or "").strip() variety = (variety or "").strip() if not crop_name and not variety: return { "summary": None, "wikipedia_title": None, "wikipedia_url": None, "images": [], } titles, matched_query = _wiki_resolve_titles(crop_name, variety) if not titles and crop_name: titles = _wiki_fulltext_titles(crop_name) matched_query = crop_name images = [] summary_text = None wiki_title = None wiki_url = None if titles: primary = _wiki_page_summary(titles[0]) if primary: summary_text = primary.get("extract") wiki_title = primary.get("title") wiki_url = (primary.get("content_urls") or {}).get("desktop", {}).get("page") images = _collect_wiki_images(titles, crop_name, variety, limit=WIKI_IMAGE_LIMIT) return { "summary": summary_text, "wikipedia_title": wiki_title, "wikipedia_url": wiki_url, "images": images[:WIKI_IMAGE_LIMIT], } def _serialize_seed_type_info(row): if not row: return None images = _normalize_seed_images(row.get("images")) planting = row.get("planting") if isinstance(planting, (bytes, bytearray)): planting = planting.decode("utf-8", errors="replace") if isinstance(planting, str): planting = json.loads(planting) return { "summary": row.get("summary"), "wikipedia_title": row.get("wikipedia_title"), "wikipedia_url": row.get("wikipedia_url"), "info_source": row.get("info_source"), "source_url": row.get("source_url"), "source_name": row.get("source_name"), "images": images, "planting": planting, "fetched_at": _serialize(row.get("fetched_at")), } def _get_seed_type_info(conn, crop_name, variety): cache_key = _seed_type_cache_key(crop_name, variety) with conn.cursor() as cur: cur.execute("SELECT * FROM GardenSeedTypeInfo WHERE cache_key = %s", (cache_key,)) row = cur.fetchone() return _serialize_seed_type_info(row) def _seed_type_info_has_content(type_info): if not type_info: return False if (type_info.get("summary") or "").strip(): return True if type_info.get("images"): return True if type_info.get("planting"): return True if type_info.get("wikipedia_url"): return True return False def _apply_product_page_info(conn, crop_name, variety, page_info): cache_key = _seed_type_cache_key(crop_name, variety) _ensure_seed_type_info_row(conn, cache_key, crop_name, variety) with conn.cursor() as cur: cur.execute("SELECT * FROM GardenSeedTypeInfo WHERE cache_key = %s", (cache_key,)) existing = cur.fetchone() or {} existing_images = _normalize_seed_images(existing.get("images")) if _type_info_images_user_managed(conn, cache_key): images = existing_images else: images = _merge_seed_images(existing_images, page_info.get("images") or []) source_url = page_info.get("source_url") summary = page_info.get("summary") or page_info.get("title") source_name = _source_name_for_url(conn, source_url) with conn.cursor() as cur: cur.execute( """ INSERT INTO GardenSeedTypeInfo (cache_key, crop_name, variety, summary, wikipedia_title, wikipedia_url, info_source, source_url, source_name, images, planting) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE crop_name = VALUES(crop_name), variety = VALUES(variety), summary = VALUES(summary), info_source = VALUES(info_source), source_url = VALUES(source_url), source_name = VALUES(source_name), images = VALUES(images), fetched_at = CURRENT_TIMESTAMP """, ( cache_key, crop_name, variety or None, summary, existing.get("wikipedia_title"), existing.get("wikipedia_url"), "seed_company", source_url, source_name, json.dumps(images), existing.get("planting"), ), ) conn.commit() return _get_seed_type_info(conn, crop_name, variety) def _fetch_and_store_seed_type_info(conn, crop_name, variety, preserved_images=None): cache_key = _seed_type_cache_key(crop_name, variety) if preserved_images is not None: existing_images = list(preserved_images) else: existing_images = _load_type_info_images(conn, cache_key) planting = _fetch_cropgraph_planting(conn, crop_name) company = _fetch_seed_company_info(conn, crop_name, variety) summary = company.get("summary") images = list(company.get("images") or []) info_source = company.get("info_source") source_url = company.get("source_url") source_name = company.get("source_name") wiki_title = None wiki_url = None need_summary = not summary need_images = len(images) < 2 if need_summary or need_images: wiki = _fetch_wikipedia_seed_info(crop_name, variety) if need_summary and wiki.get("summary"): summary = wiki.get("summary") wiki_title = wiki.get("wikipedia_title") wiki_url = wiki.get("wikipedia_url") if not info_source: info_source = "wikipedia" if need_images: images = _merge_seed_images( images, (wiki.get("images") or [])[:WIKI_IMAGE_LIMIT], limit=MAX_SEED_TYPE_IMAGES, ) if not info_source and (wiki.get("summary") or wiki.get("images")): info_source = "wikipedia" wiki_title = wiki.get("wikipedia_title") wiki_url = wiki.get("wikipedia_url") if _type_info_images_user_managed(conn, cache_key): images = _load_type_info_images(conn, cache_key) else: images = _merge_seed_images(existing_images, images, limit=MAX_SEED_TYPE_IMAGES) with conn.cursor() as cur: cur.execute( """ INSERT INTO GardenSeedTypeInfo (cache_key, crop_name, variety, summary, wikipedia_title, wikipedia_url, info_source, source_url, source_name, images, planting) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE crop_name = VALUES(crop_name), variety = VALUES(variety), summary = VALUES(summary), wikipedia_title = VALUES(wikipedia_title), wikipedia_url = VALUES(wikipedia_url), info_source = VALUES(info_source), source_url = VALUES(source_url), source_name = VALUES(source_name), images = VALUES(images), planting = VALUES(planting), fetched_at = CURRENT_TIMESTAMP """, ( cache_key, crop_name, variety or None, summary, wiki_title, wiki_url, info_source, source_url, source_name, json.dumps(images), json.dumps(planting) if planting else None, ), ) conn.commit() def _is_seed_type_fetch_inflight(cache_key): with _seed_type_fetch_lock: return cache_key in _seed_type_fetch_inflight def _queue_seed_type_fetch(crop_name, variety, preserved_images=None): cache_key = _seed_type_cache_key(crop_name, variety) with _seed_type_fetch_lock: if preserved_images is not None: _seed_type_preserved_images[cache_key] = preserved_images if cache_key in _seed_type_fetch_inflight: return _seed_type_fetch_inflight.add(cache_key) def worker(): preserved = None try: with _seed_type_fetch_lock: preserved = _seed_type_preserved_images.pop(cache_key, None) conn = get_db() try: ensure_garden_tables(conn) _fetch_and_store_seed_type_info( conn, crop_name, variety, preserved_images=preserved, ) finally: conn.close() except Exception: logger.exception("Seed type enrichment failed for %s", cache_key) finally: with _seed_type_fetch_lock: _seed_type_fetch_inflight.discard(cache_key) threading.Thread( target=worker, daemon=True, name=f"seed-type-{cache_key[:32]}", ).start() def _serialize_seed_with_stock(conn, row): settings = _get_settings(conn) or {} threshold = int(settings.get("low_seed_threshold") or 5) item = _serialize_row(row) limit = row.get("low_stock_threshold") or threshold item["low_stock"] = float(row["qty"]) <= float(limit) return item def _get_coords(conn): settings = _get_settings(conn) or {} lat = settings.get("lat") or GARDEN_LAT lng = settings.get("lng") or GARDEN_LON zip_code = settings.get("zip_code") or GARDEN_ZIP return float(lat), float(lng), zip_code def _format_mmdd(value): """Format MM-DD or date-like strings as MM/DD.""" if not value: return value s = str(value).strip() parts = s.split("-") if len(parts) == 2 and all(p.isdigit() for p in parts): return f"{int(parts[0]):02d}/{int(parts[1]):02d}" if len(parts) == 3 and all(p.isdigit() for p in parts): return f"{int(parts[1]):02d}/{int(parts[2]):02d}" return s def _get_zone_info(conn): cached = _cache_get(conn, "zone_info:v2") if cached: return cached settings = _get_settings(conn) or {} zip_code = settings.get("zip_code") or GARDEN_ZIP lat, lng, _ = _get_coords(conn) try: phzm = _fetch_json_url(f"https://phzmapi.org/{zip_code}.json") except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError): phzm = {} try: cropgraph = _fetch_json_url( f"https://api.cropgraph.com/api/zone?lat={lat}&lng={lng}" ) except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError): cropgraph = {} zone = cropgraph.get("zone") or {} frost = cropgraph.get("frostDates") or {} result = { "zip": zip_code, "zone": zone.get("zone") or phzm.get("zone"), "temperature_range": phzm.get("temperature_range"), "last_spring_frost": _format_mmdd(frost.get("lastSpring")), "first_fall_frost": _format_mmdd(frost.get("firstFall")), "season_days": cropgraph.get("seasonDays"), "climate_type": cropgraph.get("climateType"), "coordinates": {"lat": lat, "lng": lng}, } _cache_set(conn, "zone_info:v2", result, 24) return result def _get_weather(): lat = WEATHER_LAT lng = WEATHER_LON url = ( "https://api.open-meteo.com/v1/forecast?" f"latitude={lat}&longitude={lng}¤t=temperature_2m,relative_humidity_2m" "&daily=temperature_2m_max,temperature_2m_min&temperature_unit=fahrenheit" "&timezone=auto&forecast_days=2" ) try: data = _fetch_json_url(url) current = data.get("current") or {} daily = data.get("daily") or {} return { "temp_f": current.get("temperature_2m"), "humidity": current.get("relative_humidity_2m"), "today_max": (daily.get("temperature_2m_max") or [None])[0], "today_min": (daily.get("temperature_2m_min") or [None])[0], } except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError): return None def _advance_schedule_next_due(schedule): interval = int(schedule.get("interval_days") or 14) return datetime.utcnow() + timedelta(days=interval) def _schedule_label(schedule, plant_name=None, tank_name=None): title = schedule.get("title") or schedule.get("task_type", "Task") if plant_name: return f"{title} ({plant_name})" if tank_name: return f"{title} ({tank_name})" return title def _check_reminders(): conn = get_db() try: ensure_garden_tables(conn) settings = _get_settings(conn) or {} if not settings.get("notifications_enabled", 1): return low_threshold = int(settings.get("low_seed_threshold") or 5) now = datetime.utcnow() today_start = datetime.combine(date.today(), datetime.min.time()) with conn.cursor() as cur: cur.execute( """ SELECT s.*, p.name AS plant_name, t.name AS tank_name FROM GardenSchedule s LEFT JOIN GardenPlant p ON p.id = s.plant_id LEFT JOIN GardenTank t ON t.id = s.tank_id WHERE s.enabled = 1 AND s.next_due <= %s """, (now,), ) due_schedules = cur.fetchall() for sched in due_schedules: sid = str(sched["id"]) cur.execute( """ SELECT last_notified_at FROM alert_state WHERE source_type = 'garden_schedule' AND source_id = %s AND metric = 'reminder' """, (sid,), ) alert_row = cur.fetchone() last_notified = alert_row["last_notified_at"] if alert_row else None if last_notified and last_notified >= today_start: continue cur.execute( """ SELECT id FROM GardenTaskCompletion WHERE schedule_id = %s AND completed_at >= %s LIMIT 1 """, (sched["id"], today_start), ) if cur.fetchone(): continue label = _schedule_label( sched, sched.get("plant_name"), sched.get("tank_name") ) msg = f"Garden Tracker: {label} is due" ok, err = send_sms_alert_detailed(msg) if ok: cur.execute( """ INSERT INTO alert_state (source_type, source_id, metric, last_state, last_notified_at) VALUES ('garden_schedule', %s, 'reminder', 'due', %s) ON DUPLICATE KEY UPDATE last_state = 'due', last_notified_at = VALUES(last_notified_at) """, (sid, now), ) cur.execute( """ SELECT s.* FROM GardenSeedInventory s WHERE s.qty <= COALESCE(s.low_stock_threshold, %s) """, (low_threshold,), ) for seed in cur.fetchall(): sid = f"seed_{seed['id']}" cur.execute( """ SELECT last_notified_at FROM alert_state WHERE source_type = 'garden_seed' AND source_id = %s AND metric = 'low_stock' """, (sid,), ) alert_row = cur.fetchone() last_notified = alert_row["last_notified_at"] if alert_row else None if last_notified and last_notified >= today_start: continue msg = ( f"Garden Tracker: Low seed stock — {seed['crop_name']}" f"{(' (' + seed['variety'] + ')') if seed.get('variety') else ''}: " f"{seed['qty']} {seed['unit']} remaining" ) ok, _ = send_sms_alert_detailed(msg) if ok: cur.execute( """ INSERT INTO alert_state (source_type, source_id, metric, last_state, last_notified_at) VALUES ('garden_seed', %s, 'low_stock', 'low', %s) ON DUPLICATE KEY UPDATE last_state = 'low', last_notified_at = VALUES(last_notified_at) """, (sid, now), ) cur.execute("SELECT * FROM GardenTank") tanks = cur.fetchall() for tank in tanks: cur.execute( """ SELECT recorded_at FROM GardenTankReading WHERE tank_id = %s ORDER BY recorded_at DESC LIMIT 1 """, (tank["id"],), ) last_reading = cur.fetchone() interval = int(tank.get("test_interval_days") or 7) overdue = False if not last_reading: overdue = True else: due_by = last_reading["recorded_at"] + timedelta(days=interval) overdue = due_by <= now if not overdue: continue tid = str(tank["id"]) cur.execute( """ SELECT last_notified_at FROM alert_state WHERE source_type = 'garden_tank' AND source_id = %s AND metric = 'test_overdue' """, (tid,), ) alert_row = cur.fetchone() last_notified = alert_row["last_notified_at"] if alert_row else None if last_notified and last_notified >= today_start: continue msg = f"Garden Tracker: Tank '{tank['name']}' needs testing (pH/EC)" ok, _ = send_sms_alert_detailed(msg) if ok: cur.execute( """ INSERT INTO alert_state (source_type, source_id, metric, last_state, last_notified_at) VALUES ('garden_tank', %s, 'test_overdue', 'due', %s) ON DUPLICATE KEY UPDATE last_state = 'due', last_notified_at = VALUES(last_notified_at) """, (tid, now), ) conn.commit() except Exception: logger.exception("Garden reminder check failed") finally: conn.close() def _reminder_loop(): interval_secs = max(60, GARDEN_REMINDER_CHECK_MINUTES * 60) time.sleep(10) while True: _check_reminders() time.sleep(interval_secs) def start_reminder_thread(): global _reminder_thread_started with _reminder_lock: if _reminder_thread_started: return t = threading.Thread(target=_reminder_loop, daemon=True, name="garden-reminders") t.start() _reminder_thread_started = True @bp.before_app_request def _ensure_reminder_thread(): start_reminder_thread() # --- Health & Settings --- @bp.route("/api/health") def health(): try: conn = get_db() except pymysql.Error as exc: return jsonify({"ok": False, "error": str(exc)}), 500 try: ensure_garden_tables(conn) return jsonify({"ok": True, "service": "gardentracker"}) except pymysql.Error as exc: return jsonify({"ok": False, "error": str(exc)}), 500 finally: conn.close() @bp.route("/api/settings", methods=["GET"]) def get_settings(): conn = get_db() try: ensure_garden_tables(conn) return jsonify({"ok": True, "settings": _serialize_row(_get_settings(conn))}) finally: conn.close() @bp.route("/api/settings", methods=["PUT"]) def update_settings(): data = request.get_json(force=True, silent=True) or {} conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute( """ UPDATE GardenSettings SET zip_code = COALESCE(%s, zip_code), lat = COALESCE(%s, lat), lng = COALESCE(%s, lng), notifications_enabled = COALESCE(%s, notifications_enabled), low_seed_threshold = COALESCE(%s, low_seed_threshold) WHERE id = 1 """, ( data.get("zip_code"), data.get("lat"), data.get("lng"), data.get("notifications_enabled"), data.get("low_seed_threshold"), ), ) conn.commit() with conn.cursor() as cur: cur.execute( "DELETE FROM GardenExternalCache WHERE cache_key IN ('zone_info', 'zone_info:v2')" ) conn.commit() return jsonify({"ok": True, "settings": _serialize_row(_get_settings(conn))}) finally: conn.close() @bp.route("/api/settings/backup/full", methods=["GET"]) def settings_backup_full(): conn = get_db() try: ensure_garden_tables(conn) dump_sql = _sql_dump_garden_tables(conn) filename = f"garden_tracker_backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.sql" return Response( dump_sql, mimetype="application/sql", headers={"Content-Disposition": f'attachment; filename="{filename}"'}, ) except Exception as exc: return jsonify({"ok": False, "error": str(exc)}), 500 finally: conn.close() @bp.route("/api/seed-companies", methods=["GET"]) def list_seed_companies(): conn = get_db() try: ensure_garden_tables(conn) companies = _serialize_rows(_get_seed_companies(conn)) return jsonify({"ok": True, "companies": companies}) finally: conn.close() @bp.route("/api/seed-companies", methods=["POST"]) def create_seed_company(): data = request.get_json(force=True, silent=True) or {} name = (data.get("name") or "").strip() base_url = (data.get("base_url") or "").strip().rstrip("/") if not name or not base_url: return jsonify({"ok": False, "error": "name and base_url required"}), 400 if not base_url.startswith("http"): base_url = "https://" + base_url conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute( """ INSERT INTO GardenSeedCompany (name, base_url, search_url_template, enabled, sort_order) VALUES (%s, %s, %s, %s, %s) """, ( name, base_url, data.get("search_url_template"), 1 if data.get("enabled", True) else 0, int(data.get("sort_order") or 0), ), ) company_id = cur.lastrowid cur.execute("SELECT * FROM GardenSeedCompany WHERE id = %s", (company_id,)) company = cur.fetchone() conn.commit() return jsonify({"ok": True, "company": _serialize_row(company)}), 201 finally: conn.close() @bp.route("/api/seed-companies/", methods=["PUT"]) def update_seed_company(company_id): data = request.get_json(force=True, silent=True) or {} conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT id FROM GardenSeedCompany WHERE id = %s", (company_id,)) if not cur.fetchone(): return jsonify({"ok": False, "error": "Not found"}), 404 base_url = data.get("base_url") if isinstance(base_url, str): base_url = base_url.strip().rstrip("/") if base_url and not base_url.startswith("http"): base_url = "https://" + base_url cur.execute( """ UPDATE GardenSeedCompany SET name = COALESCE(%s, name), base_url = COALESCE(%s, base_url), search_url_template = COALESCE(%s, search_url_template), enabled = COALESCE(%s, enabled), sort_order = COALESCE(%s, sort_order) WHERE id = %s """, ( data.get("name"), base_url, data.get("search_url_template"), data.get("enabled"), data.get("sort_order"), company_id, ), ) cur.execute("SELECT * FROM GardenSeedCompany WHERE id = %s", (company_id,)) company = cur.fetchone() conn.commit() return jsonify({"ok": True, "company": _serialize_row(company)}) finally: conn.close() @bp.route("/api/seed-companies/", methods=["DELETE"]) def delete_seed_company(company_id): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("DELETE FROM GardenSeedCompany WHERE id = %s", (company_id,)) if cur.rowcount == 0: return jsonify({"ok": False, "error": "Not found"}), 404 conn.commit() return jsonify({"ok": True}) finally: conn.close() # --- Plants --- @bp.route("/api/plants", methods=["GET"]) def list_plants(): status = request.args.get("status") conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: if status: cur.execute( _PLANT_SELECT + " WHERE p.status = %s ORDER BY p.name", (status,), ) else: cur.execute(_PLANT_SELECT + " ORDER BY p.name") rows = cur.fetchall() return jsonify({"ok": True, "plants": _serialize_plants(rows)}) finally: conn.close() @bp.route("/api/plants/by-label/", methods=["GET"]) def get_plant_by_label(label_id): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute(_PLANT_SELECT + " WHERE p.label_id = %s", (label_id.strip().upper(),)) plant = cur.fetchone() if not plant: return jsonify({"ok": False, "error": "Not found"}), 404 plant_id = plant["id"] cur.execute( "SELECT * FROM GardenPlantEvent WHERE plant_id = %s ORDER BY event_date DESC, id DESC", (plant_id,), ) events = _enrich_events(cur, plant_id, plant, cur.fetchall()) return jsonify({"ok": True, "plant": _serialize_plant(plant, conn=conn, include_primary_image=True), "events": events}) finally: conn.close() @bp.route("/api/plants/", methods=["GET"]) def get_plant(plant_id): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute(_PLANT_SELECT + " WHERE p.id = %s", (plant_id,)) plant = cur.fetchone() if not plant: return jsonify({"ok": False, "error": "Not found"}), 404 cur.execute( "SELECT * FROM GardenPlantEvent WHERE plant_id = %s ORDER BY event_date DESC, id DESC", (plant_id,), ) events = cur.fetchall() cur.execute( "SELECT * FROM GardenSchedule WHERE plant_id = %s ORDER BY next_due", (plant_id,), ) schedules = cur.fetchall() events = _enrich_events(cur, plant_id, plant, events) return jsonify( { "ok": True, "plant": _serialize_plant(plant, conn=conn, include_primary_image=True), "events": events, "schedules": _serialize_rows(schedules), } ) finally: conn.close() @bp.route("/api/plants", methods=["POST"]) def create_plant(): data = request.get_json(force=True, silent=True) or {} source_type = (data.get("source_type") or "store_bought").strip() if source_type not in PLANT_SOURCE_TYPES: return jsonify({"ok": False, "error": "invalid source_type"}), 400 status = data.get("status") or "active" if status not in PLANT_STATUSES: return jsonify({"ok": False, "error": "invalid status"}), 400 try: quantity = int(data.get("quantity") or 1) except (TypeError, ValueError): return jsonify({"ok": False, "error": "invalid quantity"}), 400 if quantity < 1 or quantity > 500: return jsonify({"ok": False, "error": "quantity must be between 1 and 500"}), 400 seed_id = data.get("seed_id") name = (data.get("name") or "").strip() variety = (data.get("variety") or "").strip() or None planted_date = _parse_date(data.get("planted_date")) or date.today() conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: seed = None if source_type == "seed": if not seed_id: return jsonify({"ok": False, "error": "seed_id required for seed-sourced plants"}), 400 cur.execute("SELECT * FROM GardenSeedInventory WHERE id = %s", (seed_id,)) seed = cur.fetchone() if not seed: return jsonify({"ok": False, "error": "Seed not found"}), 404 if not name: name = seed["crop_name"] if variety is None and seed.get("variety"): variety = seed["variety"] elif not name: return jsonify({"ok": False, "error": "name required for store-bought plants"}), 400 event_type = "seed_start" if source_type == "seed" else "transplanted" event_notes = None if source_type == "seed": event_notes = _seed_display_label(seed) elif data.get("notes"): event_notes = data.get("notes") created_ids = [] label_ids = _allocate_plant_label_ids(cur, quantity) for label_id in label_ids: cur.execute( """ INSERT INTO GardenPlant (label_id, name, variety, location, status, planted_date, moisture_device_id, notes, seed_id, source_type) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( label_id, name, variety, data.get("location"), status, planted_date, data.get("moisture_device_id"), data.get("notes"), seed_id if source_type == "seed" else None, source_type, ), ) plant_id = cur.lastrowid created_ids.append(plant_id) cur.execute( """ INSERT INTO GardenPlantEvent ( plant_id, event_type, event_date, notes, days_since_start, days_since_anchor ) VALUES (%s, %s, %s, %s, %s, %s) """, (plant_id, event_type, planted_date, event_notes, 0, "planting"), ) if source_type == "seed" and data.get("decrement_seed", True): cur.execute( "UPDATE GardenSeedInventory SET qty = GREATEST(0, qty - %s) WHERE id = %s", (quantity, seed_id), ) plants = [] for plant_id in created_ids: cur.execute(_PLANT_SELECT + " WHERE p.id = %s", (plant_id,)) row = cur.fetchone() if row: plants.append(_serialize_plant(row)) conn.commit() return jsonify( { "ok": True, "count": len(plants), "plant": plants[0] if plants else None, "plants": plants, } ), 201 finally: conn.close() @bp.route("/api/plants/", methods=["PUT"]) def update_plant(plant_id): data = request.get_json(force=True, silent=True) or {} conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT id FROM GardenPlant WHERE id = %s", (plant_id,)) if not cur.fetchone(): return jsonify({"ok": False, "error": "Not found"}), 404 status = data.get("status") if status is not None and status not in PLANT_STATUSES: return jsonify({"ok": False, "error": "invalid status"}), 400 cur.execute( """ UPDATE GardenPlant SET name = COALESCE(%s, name), variety = COALESCE(%s, variety), location = COALESCE(%s, location), status = COALESCE(%s, status), planted_date = COALESCE(%s, planted_date), moisture_device_id = COALESCE(%s, moisture_device_id), notes = COALESCE(%s, notes), seed_id = COALESCE(%s, seed_id), source_type = COALESCE(%s, source_type) WHERE id = %s """, ( data.get("name"), data.get("variety"), data.get("location"), status, _parse_date(data.get("planted_date")) if "planted_date" in data else None, data.get("moisture_device_id"), data.get("notes"), data.get("seed_id"), data.get("source_type"), plant_id, ), ) cur.execute(_PLANT_SELECT + " WHERE p.id = %s", (plant_id,)) plant = cur.fetchone() conn.commit() return jsonify({"ok": True, "plant": _serialize_plant(plant)}) finally: conn.close() @bp.route("/api/plants//duplicate", methods=["POST"]) def duplicate_plant(plant_id): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute(_PLANT_SELECT + " WHERE p.id = %s", (plant_id,)) plant = cur.fetchone() if not plant: return jsonify({"ok": False, "error": "Not found"}), 404 label_id = _allocate_plant_label_ids(cur, 1)[0] source_type = plant.get("source_type") or "store_bought" event_type = "seed_start" if source_type == "seed" else "transplanted" event_notes = None if source_type == "seed" and plant.get("seed_id"): cur.execute("SELECT * FROM GardenSeedInventory WHERE id = %s", (plant["seed_id"],)) seed = cur.fetchone() if seed: event_notes = _seed_display_label(seed) planted_date = plant.get("planted_date") or date.today() if isinstance(planted_date, datetime): planted_date = planted_date.date() cur.execute( """ INSERT INTO GardenPlant (label_id, name, variety, location, status, planted_date, moisture_device_id, notes, seed_id, source_type) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( label_id, plant["name"], plant.get("variety"), plant.get("location"), plant.get("status") or "active", planted_date, plant.get("moisture_device_id"), plant.get("notes"), plant.get("seed_id") if source_type == "seed" else None, source_type, ), ) new_id = cur.lastrowid cur.execute( """ INSERT INTO GardenPlantEvent ( plant_id, event_type, event_date, notes, days_since_start, days_since_anchor ) VALUES (%s, %s, %s, %s, %s, %s) """, (new_id, event_type, planted_date, event_notes, 0, "planting"), ) cur.execute(_PLANT_SELECT + " WHERE p.id = %s", (new_id,)) new_plant = cur.fetchone() conn.commit() return jsonify({"ok": True, "plant": _serialize_plant(new_plant)}), 201 finally: conn.close() @bp.route("/api/plants/", methods=["DELETE"]) def delete_plant(plant_id): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("DELETE FROM GardenPlant WHERE id = %s", (plant_id,)) if cur.rowcount == 0: return jsonify({"ok": False, "error": "Not found"}), 404 conn.commit() return jsonify({"ok": True}) finally: conn.close() @bp.route("/api/plants//events", methods=["GET"]) def list_plant_events(plant_id): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute(_PLANT_SELECT + " WHERE p.id = %s", (plant_id,)) plant = cur.fetchone() if not plant: return jsonify({"ok": False, "error": "Plant not found"}), 404 cur.execute( "SELECT * FROM GardenPlantEvent WHERE plant_id = %s ORDER BY event_date DESC, id DESC", (plant_id,), ) rows = cur.fetchall() rows = _enrich_events(cur, plant_id, plant, rows) return jsonify({"ok": True, "events": rows}) finally: conn.close() @bp.route("/api/plants//events", methods=["POST"]) def create_plant_event(plant_id): data = request.get_json(force=True, silent=True) or {} event_type = data.get("event_type") or "note" if event_type not in EVENT_TYPES: return jsonify({"ok": False, "error": "invalid event_type"}), 400 event_date = _parse_date(data.get("event_date")) or date.today() conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute(_PLANT_SELECT + " WHERE p.id = %s", (plant_id,)) plant = cur.fetchone() if not plant: return jsonify({"ok": False, "error": "Plant not found"}), 404 location = _normalize_location(data.get("location")) height = _optional_float(data.get("height")) days_since, anchor = _compute_days_since_start(cur, plant_id, plant, event_date) cur.execute( """ INSERT INTO GardenPlantEvent ( plant_id, event_type, event_date, notes, qty, unit, location, height, days_since_start, days_since_anchor ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( plant_id, event_type, event_date, data.get("notes"), data.get("qty"), data.get("unit"), location, height, days_since, anchor, ), ) _sync_transplant_location(cur, plant_id, event_type, location) event_id = cur.lastrowid cur.execute("SELECT * FROM GardenPlantEvent WHERE id = %s", (event_id,)) event = cur.fetchone() event = _enrich_event(cur, plant_id, plant, event) conn.commit() return jsonify({"ok": True, "event": event}), 201 finally: conn.close() @bp.route("/api/plants//events/", methods=["PUT"]) def update_plant_event(plant_id, event_id): data = request.get_json(force=True, silent=True) or {} conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute(_PLANT_SELECT + " WHERE p.id = %s", (plant_id,)) plant = cur.fetchone() if not plant: return jsonify({"ok": False, "error": "Plant not found"}), 404 cur.execute( "SELECT * FROM GardenPlantEvent WHERE id = %s AND plant_id = %s", (event_id, plant_id), ) event = cur.fetchone() if not event: return jsonify({"ok": False, "error": "Not found"}), 404 event_type = data.get("event_type", event["event_type"]) if event_type not in EVENT_TYPES: return jsonify({"ok": False, "error": "invalid event_type"}), 400 if "event_date" in data: event_date = _parse_date(data.get("event_date")) if not event_date: return jsonify({"ok": False, "error": "invalid event_date"}), 400 else: event_date = _as_date(event.get("event_date")) notes = data["notes"] if "notes" in data else event.get("notes") qty = data["qty"] if "qty" in data else event.get("qty") unit = data["unit"] if "unit" in data else event.get("unit") if "days_since_start" in data and data.get("days_since_start") is not None: try: days_since = max(0, int(data["days_since_start"])) except (TypeError, ValueError): return jsonify({"ok": False, "error": "invalid days_since_start"}), 400 anchor = "seed_start" if (plant or {}).get("source_type") == "seed" else "planting" else: days_since, anchor = _compute_days_since_start(cur, plant_id, plant, event_date) location = _normalize_location( data["location"] if "location" in data else event.get("location") ) if "height" in data: height = _optional_float(data.get("height")) else: height = event.get("height") cur.execute( """ UPDATE GardenPlantEvent SET event_type = %s, event_date = %s, notes = %s, qty = %s, unit = %s, location = %s, height = %s, days_since_start = %s, days_since_anchor = %s WHERE id = %s AND plant_id = %s """, ( event_type, event_date, notes, qty, unit, location, height, days_since, anchor, event_id, plant_id, ), ) _sync_transplant_location(cur, plant_id, event_type, location) cur.execute("SELECT * FROM GardenPlantEvent WHERE id = %s", (event_id,)) event = cur.fetchone() event = _enrich_event(cur, plant_id, plant, event) conn.commit() return jsonify({"ok": True, "event": event}) finally: conn.close() @bp.route("/api/plants//events/", methods=["DELETE"]) def delete_plant_event(plant_id, event_id): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute( "DELETE FROM GardenPlantEvent WHERE id = %s AND plant_id = %s", (event_id, plant_id), ) if cur.rowcount == 0: return jsonify({"ok": False, "error": "Not found"}), 404 conn.commit() return jsonify({"ok": True}) finally: conn.close() # --- Seeds --- @bp.route("/api/seed-type-upload/") def serve_seed_type_upload(rel): if not rel or ".." in rel.split("/"): return jsonify({"ok": False, "error": "invalid path"}), 400 upload_root = _seed_upload_root() full = os.path.join(upload_root, rel) if not os.path.isfile(full): return jsonify({"ok": False, "error": "not found"}), 404 return send_from_directory(upload_root, rel) @bp.route("/api/seed-image") def seed_image_proxy(): raw_url = (request.args.get("url") or "").strip() if not raw_url: return jsonify({"ok": False, "error": "url required"}), 400 url = _normalize_image_url(raw_url) parsed = urllib.parse.urlparse(url) host = (parsed.hostname or "").lower() conn = get_db() try: ensure_garden_tables(conn) allowed_hosts = _allowed_image_hosts(conn) finally: conn.close() if not _image_host_allowed(host, allowed_hosts): return jsonify({"ok": False, "error": "host not allowed"}), 400 if parsed.scheme != "https": return jsonify({"ok": False, "error": "invalid url"}), 400 referer = "https://en.wikipedia.org/" if host not in WIKIMEDIA_IMAGE_HOSTS: referer = f"https://{host}/" try: req = urllib.request.Request( url, headers={ "User-Agent": "Mozilla/5.0 GardenTracker/1.0", "Referer": referer, }, ) with urllib.request.urlopen(req, timeout=15) as resp: data = resp.read() content_type = resp.headers.get("Content-Type", "image/jpeg") return Response( data, mimetype=content_type, headers={"Cache-Control": "public, max-age=86400"}, ) except (urllib.error.URLError, urllib.error.HTTPError, OSError) as exc: return jsonify({"ok": False, "error": str(exc)}), 502 @bp.route("/api/seeds", methods=["GET"]) def list_seeds(): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT * FROM GardenSeedInventory ORDER BY crop_name, variety") rows = cur.fetchall() settings = _get_settings(conn) or {} threshold = int(settings.get("low_seed_threshold") or 5) out = [] for row in rows: item = _serialize_row(row) limit = row.get("low_stock_threshold") or threshold item["low_stock"] = float(row["qty"]) <= float(limit) out.append(item) return jsonify({"ok": True, "seeds": out}) finally: conn.close() @bp.route("/api/seeds", methods=["POST"]) def create_seed(): data = request.get_json(force=True, silent=True) or {} crop_name = (data.get("crop_name") or "").strip() if not crop_name: return jsonify({"ok": False, "error": "crop_name required"}), 400 conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute( """ INSERT INTO GardenSeedInventory (crop_name, variety, qty, unit, source, purchase_date, expiry_date, location, low_stock_threshold, notes) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( crop_name, data.get("variety"), data.get("qty", 0), data.get("unit") or "seeds", data.get("source"), _parse_date(data.get("purchase_date")), _parse_date(data.get("expiry_date")), data.get("location"), data.get("low_stock_threshold"), data.get("notes"), ), ) seed_id = cur.lastrowid cur.execute("SELECT * FROM GardenSeedInventory WHERE id = %s", (seed_id,)) seed = cur.fetchone() conn.commit() _queue_seed_type_fetch(crop_name, data.get("variety")) return jsonify({"ok": True, "seed": _serialize_row(seed)}), 201 finally: conn.close() @bp.route("/api/seeds/", methods=["GET"]) def get_seed(seed_id): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT * FROM GardenSeedInventory WHERE id = %s", (seed_id,)) row = cur.fetchone() if not row: return jsonify({"ok": False, "error": "Not found"}), 404 seed = _serialize_seed_with_stock(conn, row) type_info = _get_seed_type_info(conn, row["crop_name"], row.get("variety")) cache_key = _seed_type_cache_key(row["crop_name"], row.get("variety")) type_info_loading = False if not type_info: type_info_loading = _is_seed_type_fetch_inflight(cache_key) if not type_info_loading: _queue_seed_type_fetch(row["crop_name"], row.get("variety")) type_info_loading = True elif not (type_info.get("images") or []): if not _type_info_images_user_managed(conn, cache_key): if cache_key not in _seed_type_image_retry_done: _seed_type_image_retry_done.add(cache_key) if not _is_seed_type_fetch_inflight(cache_key): _queue_seed_type_fetch(row["crop_name"], row.get("variety")) type_info_loading = True else: type_info_loading = _is_seed_type_fetch_inflight(cache_key) return jsonify( { "ok": True, "seed": seed, "type_info": type_info, "type_info_loading": type_info_loading, } ) finally: conn.close() @bp.route("/api/seeds//product-url", methods=["POST"]) def set_seed_product_url(seed_id): data = request.get_json(silent=True) or {} url = (data.get("url") or "").strip() if not url: return jsonify({"ok": False, "error": "Product URL is required"}), 400 if not url.startswith("http://") and not url.startswith("https://"): url = "https://" + url clean_url = url.split("?")[0].rstrip("/") conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT * FROM GardenSeedInventory WHERE id = %s", (seed_id,)) row = cur.fetchone() if not row: return jsonify({"ok": False, "error": "Not found"}), 404 if not _is_allowed_product_url(conn, clean_url): return jsonify({ "ok": False, "error": "URL must be a product page on a configured seed company site", }), 400 page_info = _extract_product_page_info(clean_url) if not page_info: return jsonify({ "ok": False, "error": "Could not load product info from that page", }) crop_name = row["crop_name"] variety = row.get("variety") match_score = _score_seed_candidate( clean_url, crop_name, variety, title=page_info.get("title"), summary=page_info.get("summary"), ) matches = match_score >= PRODUCT_URL_MIN_MATCH_SCORE preview = { "title": page_info.get("title"), "summary": page_info.get("summary"), "source_url": clean_url, "source_name": _source_name_for_url(conn, clean_url), "image_count": len(page_info.get("images") or []), } existing = _get_seed_type_info(conn, crop_name, variety) has_existing = _seed_type_info_has_content(existing) apply = bool(data.get("apply")) force = bool(data.get("force")) if not apply: if not matches: return jsonify({ "ok": False, "error": ( "This page does not appear to match " f"{crop_name}" + (f" — {variety}" if variety else "") ), "match_score": match_score, "preview": preview, "has_existing": has_existing, "can_apply": True, }) return jsonify({ "ok": True, "matches": True, "match_score": match_score, "preview": preview, "has_existing": has_existing, }) if not matches and not force: return jsonify({ "ok": False, "error": "Page match not confirmed", }), 400 type_info = _apply_product_page_info(conn, crop_name, variety, page_info) return jsonify({"ok": True, "type_info": type_info}) finally: conn.close() @bp.route("/api/seeds//refresh-type-info", methods=["POST"]) def refresh_seed_type_info(seed_id): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT * FROM GardenSeedInventory WHERE id = %s", (seed_id,)) row = cur.fetchone() if not row: return jsonify({"ok": False, "error": "Not found"}), 404 cache_key = _seed_type_cache_key(row["crop_name"], row.get("variety")) preserved_images = _load_type_info_images(conn, cache_key) with conn.cursor() as cur: cur.execute("DELETE FROM GardenSeedTypeInfo WHERE cache_key = %s", (cache_key,)) conn.commit() _seed_type_image_retry_done.discard(cache_key) _queue_seed_type_fetch( row["crop_name"], row.get("variety"), preserved_images=preserved_images, ) return jsonify({"ok": True, "type_info_loading": True}) finally: conn.close() @bp.route("/api/seeds//type-images", methods=["POST"]) def upload_seed_type_image(seed_id): image = request.files.get("image") if not image: return jsonify({"ok": False, "error": "No file uploaded"}), 400 conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT * FROM GardenSeedInventory WHERE id = %s", (seed_id,)) row = cur.fetchone() if not row: return jsonify({"ok": False, "error": "Not found"}), 404 cache_key = _seed_type_cache_key(row["crop_name"], row.get("variety")) images = _load_type_info_images(conn, cache_key) if len(images) >= MAX_SEED_TYPE_IMAGES: return jsonify({"ok": False, "error": f"Maximum {MAX_SEED_TYPE_IMAGES} images"}), 400 safe_name = secure_filename(image.filename or "image.jpg") or "image.jpg" ext = os.path.splitext(safe_name)[1].lower() if ext not in (".jpg", ".jpeg", ".png", ".webp", ".gif"): ext = ".jpg" filename = f"{os.urandom(16).hex()}{ext}" rel = f"{_seed_upload_subdir(cache_key)}/{filename}" full = os.path.join(_seed_upload_root(), rel) os.makedirs(os.path.dirname(full), exist_ok=True) image.save(full) images.append({ "url": f"seed-upload://{rel}", "file_path": rel, "title": safe_name, "source": "uploaded", }) _save_type_info_images( conn, cache_key, row["crop_name"], row.get("variety"), images, user_managed=True, ) type_info = _get_seed_type_info(conn, row["crop_name"], row.get("variety")) return jsonify({"ok": True, "type_info": type_info}), 201 finally: conn.close() @bp.route("/api/seeds//type-images", methods=["DELETE"]) def delete_seed_type_image(seed_id): url = _requested_image_url() index = _requested_image_index() if url == "" and index is None: return jsonify({"ok": False, "error": "url or index required"}), 400 conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT * FROM GardenSeedInventory WHERE id = %s", (seed_id,)) row = cur.fetchone() if not row: return jsonify({"ok": False, "error": "Not found"}), 404 cache_key = _seed_type_cache_key(row["crop_name"], row.get("variety")) images = _load_type_info_images(conn, cache_key) removed = None kept = list(images) if index is not None: if index < 0 or index >= len(images): return jsonify({"ok": False, "error": "Image not found"}), 404 removed = kept.pop(index) else: kept = [] for img in images: if _image_matches_request(img, url) and removed is None: removed = img continue kept.append(img) if not removed: return jsonify({"ok": False, "error": "Image not found"}), 404 if removed.get("file_path"): _delete_seed_upload_file(removed["file_path"]) _save_type_info_images( conn, cache_key, row["crop_name"], row.get("variety"), kept, user_managed=True, ) type_info = _get_seed_type_info(conn, row["crop_name"], row.get("variety")) return jsonify({"ok": True, "type_info": type_info}) finally: conn.close() @bp.route("/api/seeds//type-images/primary", methods=["PATCH", "POST"]) def make_seed_type_image_primary(seed_id): url = _requested_image_url() index = _requested_image_index() if url == "" and index is None: return jsonify({"ok": False, "error": "url or index required"}), 400 conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT * FROM GardenSeedInventory WHERE id = %s", (seed_id,)) row = cur.fetchone() if not row: return jsonify({"ok": False, "error": "Not found"}), 404 cache_key = _seed_type_cache_key(row["crop_name"], row.get("variety")) images = _load_type_info_images(conn, cache_key) if index is not None: if index < 0 or index >= len(images): return jsonify({"ok": False, "error": "Image not found"}), 404 idx = index else: idx = next( (i for i, img in enumerate(images) if _image_matches_request(img, url)), None, ) if idx is None: return jsonify({"ok": False, "error": "Image not found"}), 404 if idx == 0: _save_type_info_images( conn, cache_key, row["crop_name"], row.get("variety"), images, user_managed=True, ) type_info = _get_seed_type_info(conn, row["crop_name"], row.get("variety")) return jsonify({"ok": True, "type_info": type_info}) chosen = images.pop(idx) images.insert(0, chosen) _save_type_info_images( conn, cache_key, row["crop_name"], row.get("variety"), images, user_managed=True, ) type_info = _get_seed_type_info(conn, row["crop_name"], row.get("variety")) return jsonify({"ok": True, "type_info": type_info}) finally: conn.close() @bp.route("/api/seeds/", methods=["PUT"]) def update_seed(seed_id): data = request.get_json(force=True, silent=True) or {} conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT id FROM GardenSeedInventory WHERE id = %s", (seed_id,)) if not cur.fetchone(): return jsonify({"ok": False, "error": "Not found"}), 404 cur.execute( """ UPDATE GardenSeedInventory SET crop_name = COALESCE(%s, crop_name), variety = COALESCE(%s, variety), qty = COALESCE(%s, qty), unit = COALESCE(%s, unit), source = COALESCE(%s, source), purchase_date = COALESCE(%s, purchase_date), expiry_date = COALESCE(%s, expiry_date), location = COALESCE(%s, location), low_stock_threshold = COALESCE(%s, low_stock_threshold), notes = COALESCE(%s, notes) WHERE id = %s """, ( data.get("crop_name"), data.get("variety"), data.get("qty"), data.get("unit"), data.get("source"), _parse_date(data.get("purchase_date")) if "purchase_date" in data else None, _parse_date(data.get("expiry_date")) if "expiry_date" in data else None, data.get("location"), data.get("low_stock_threshold"), data.get("notes"), seed_id, ), ) cur.execute("SELECT * FROM GardenSeedInventory WHERE id = %s", (seed_id,)) seed = cur.fetchone() conn.commit() return jsonify({"ok": True, "seed": _serialize_row(seed)}) finally: conn.close() @bp.route("/api/seeds/", methods=["DELETE"]) def delete_seed(seed_id): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("DELETE FROM GardenSeedInventory WHERE id = %s", (seed_id,)) if cur.rowcount == 0: return jsonify({"ok": False, "error": "Not found"}), 404 conn.commit() return jsonify({"ok": True}) finally: conn.close() @bp.route("/api/seeds//adjust-qty", methods=["PATCH"]) def adjust_seed_qty(seed_id): data = request.get_json(force=True, silent=True) or {} delta = data.get("delta") if delta is None: return jsonify({"ok": False, "error": "delta required"}), 400 conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT * FROM GardenSeedInventory WHERE id = %s", (seed_id,)) seed = cur.fetchone() if not seed: return jsonify({"ok": False, "error": "Not found"}), 404 new_qty = max(0, float(seed["qty"]) + float(delta)) cur.execute( "UPDATE GardenSeedInventory SET qty = %s WHERE id = %s", (new_qty, seed_id), ) cur.execute("SELECT * FROM GardenSeedInventory WHERE id = %s", (seed_id,)) seed = cur.fetchone() conn.commit() return jsonify({"ok": True, "seed": _serialize_row(seed)}) finally: conn.close() # --- Tanks --- @bp.route("/api/tanks", methods=["GET"]) def list_tanks(): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT * FROM GardenTank ORDER BY name") tanks = cur.fetchall() out = [] now = datetime.utcnow() for tank in tanks: item = _serialize_row(tank) cur.execute( """ SELECT * FROM GardenTankReading WHERE tank_id = %s ORDER BY recorded_at DESC LIMIT 1 """, (tank["id"],), ) latest = cur.fetchone() item["latest_reading"] = _serialize_row(latest) if latest else None interval = int(tank.get("test_interval_days") or 7) if latest: due_by = latest["recorded_at"] + timedelta(days=interval) item["test_overdue"] = due_by <= now item["next_test_due"] = _serialize(due_by) else: item["test_overdue"] = True item["next_test_due"] = None out.append(item) return jsonify({"ok": True, "tanks": out}) finally: conn.close() @bp.route("/api/tanks", methods=["POST"]) def create_tank(): data = request.get_json(force=True, silent=True) or {} name = (data.get("name") or "").strip() if not name: return jsonify({"ok": False, "error": "name required"}), 400 conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute( """ INSERT INTO GardenTank (name, volume_gal, solution_type, target_ph_min, target_ph_max, target_ec_min, target_ec_max, test_interval_days, notes) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( name, data.get("volume_gal"), data.get("solution_type"), data.get("target_ph_min"), data.get("target_ph_max"), data.get("target_ec_min"), data.get("target_ec_max"), data.get("test_interval_days", 7), data.get("notes"), ), ) tank_id = cur.lastrowid cur.execute("SELECT * FROM GardenTank WHERE id = %s", (tank_id,)) tank = cur.fetchone() conn.commit() return jsonify({"ok": True, "tank": _serialize_row(tank)}), 201 finally: conn.close() @bp.route("/api/tanks/", methods=["PUT"]) def update_tank(tank_id): data = request.get_json(force=True, silent=True) or {} conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT id FROM GardenTank WHERE id = %s", (tank_id,)) if not cur.fetchone(): return jsonify({"ok": False, "error": "Not found"}), 404 cur.execute( """ UPDATE GardenTank SET name = COALESCE(%s, name), volume_gal = COALESCE(%s, volume_gal), solution_type = COALESCE(%s, solution_type), target_ph_min = COALESCE(%s, target_ph_min), target_ph_max = COALESCE(%s, target_ph_max), target_ec_min = COALESCE(%s, target_ec_min), target_ec_max = COALESCE(%s, target_ec_max), test_interval_days = COALESCE(%s, test_interval_days), notes = COALESCE(%s, notes) WHERE id = %s """, ( data.get("name"), data.get("volume_gal"), data.get("solution_type"), data.get("target_ph_min"), data.get("target_ph_max"), data.get("target_ec_min"), data.get("target_ec_max"), data.get("test_interval_days"), data.get("notes"), tank_id, ), ) cur.execute("SELECT * FROM GardenTank WHERE id = %s", (tank_id,)) tank = cur.fetchone() conn.commit() return jsonify({"ok": True, "tank": _serialize_row(tank)}) finally: conn.close() @bp.route("/api/tanks/", methods=["DELETE"]) def delete_tank(tank_id): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("DELETE FROM GardenTank WHERE id = %s", (tank_id,)) if cur.rowcount == 0: return jsonify({"ok": False, "error": "Not found"}), 404 conn.commit() return jsonify({"ok": True}) finally: conn.close() @bp.route("/api/tanks//readings", methods=["GET"]) def list_tank_readings(tank_id): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute( """ SELECT * FROM GardenTankReading WHERE tank_id = %s ORDER BY recorded_at DESC LIMIT 100 """, (tank_id,), ) rows = cur.fetchall() return jsonify({"ok": True, "readings": _serialize_rows(rows)}) finally: conn.close() @bp.route("/api/tanks//readings", methods=["POST"]) def create_tank_reading(tank_id): data = request.get_json(force=True, silent=True) or {} conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT id FROM GardenTank WHERE id = %s", (tank_id,)) if not cur.fetchone(): return jsonify({"ok": False, "error": "Tank not found"}), 404 recorded_at = _parse_datetime(data.get("recorded_at")) or datetime.utcnow() cur.execute( """ INSERT INTO GardenTankReading (tank_id, ph, ec, temp_f, volume_gal, notes, recorded_at) VALUES (%s, %s, %s, %s, %s, %s, %s) """, ( tank_id, data.get("ph"), data.get("ec"), data.get("temp_f"), data.get("volume_gal"), data.get("notes"), recorded_at, ), ) reading_id = cur.lastrowid cur.execute("SELECT * FROM GardenTankReading WHERE id = %s", (reading_id,)) reading = cur.fetchone() conn.commit() return jsonify({"ok": True, "reading": _serialize_row(reading)}), 201 finally: conn.close() # --- Schedules --- @bp.route("/api/schedules", methods=["GET"]) def list_schedules(): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute( """ SELECT s.*, p.name AS plant_name, t.name AS tank_name FROM GardenSchedule s LEFT JOIN GardenPlant p ON p.id = s.plant_id LEFT JOIN GardenTank t ON t.id = s.tank_id ORDER BY s.next_due """ ) rows = cur.fetchall() return jsonify({"ok": True, "schedules": _serialize_rows(rows)}) finally: conn.close() @bp.route("/api/schedules", methods=["POST"]) def create_schedule(): data = request.get_json(force=True, silent=True) or {} task_type = data.get("task_type") or "custom" if task_type not in TASK_TYPES: return jsonify({"ok": False, "error": "invalid task_type"}), 400 title = (data.get("title") or task_type.replace("_", " ").title()).strip() interval_days = int(data.get("interval_days") or 14) next_due = _parse_datetime(data.get("next_due")) or datetime.utcnow() conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute( """ INSERT INTO GardenSchedule (plant_id, tank_id, task_type, title, interval_days, next_due, enabled, notes) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) """, ( data.get("plant_id"), data.get("tank_id"), task_type, title, interval_days, next_due, 1 if data.get("enabled", True) else 0, data.get("notes"), ), ) schedule_id = cur.lastrowid cur.execute("SELECT * FROM GardenSchedule WHERE id = %s", (schedule_id,)) schedule = cur.fetchone() conn.commit() return jsonify({"ok": True, "schedule": _serialize_row(schedule)}), 201 finally: conn.close() @bp.route("/api/schedules/", methods=["PUT"]) def update_schedule(schedule_id): data = request.get_json(force=True, silent=True) or {} conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT * FROM GardenSchedule WHERE id = %s", (schedule_id,)) if not cur.fetchone(): return jsonify({"ok": False, "error": "Not found"}), 404 task_type = data.get("task_type") if task_type is not None and task_type not in TASK_TYPES: return jsonify({"ok": False, "error": "invalid task_type"}), 400 enabled = data.get("enabled") cur.execute( """ UPDATE GardenSchedule SET plant_id = COALESCE(%s, plant_id), tank_id = COALESCE(%s, tank_id), task_type = COALESCE(%s, task_type), title = COALESCE(%s, title), interval_days = COALESCE(%s, interval_days), next_due = COALESCE(%s, next_due), enabled = COALESCE(%s, enabled), notes = COALESCE(%s, notes) WHERE id = %s """, ( data.get("plant_id"), data.get("tank_id"), task_type, data.get("title"), data.get("interval_days"), _parse_datetime(data.get("next_due")) if "next_due" in data else None, 1 if enabled is True else (0 if enabled is False else None), data.get("notes"), schedule_id, ), ) cur.execute("SELECT * FROM GardenSchedule WHERE id = %s", (schedule_id,)) schedule = cur.fetchone() conn.commit() return jsonify({"ok": True, "schedule": _serialize_row(schedule)}) finally: conn.close() @bp.route("/api/schedules/", methods=["DELETE"]) def delete_schedule(schedule_id): conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("DELETE FROM GardenSchedule WHERE id = %s", (schedule_id,)) if cur.rowcount == 0: return jsonify({"ok": False, "error": "Not found"}), 404 conn.commit() return jsonify({"ok": True}) finally: conn.close() @bp.route("/api/schedules//complete", methods=["POST"]) def complete_schedule(schedule_id): data = request.get_json(force=True, silent=True) or {} conn = get_db() try: ensure_garden_tables(conn) with conn.cursor() as cur: cur.execute("SELECT * FROM GardenSchedule WHERE id = %s", (schedule_id,)) schedule = cur.fetchone() if not schedule: return jsonify({"ok": False, "error": "Not found"}), 404 now = datetime.utcnow() cur.execute( """ INSERT INTO GardenTaskCompletion (schedule_id, completed_at, notes) VALUES (%s, %s, %s) """, (schedule_id, now, data.get("notes")), ) next_due = _advance_schedule_next_due(schedule) cur.execute( "UPDATE GardenSchedule SET next_due = %s WHERE id = %s", (next_due, schedule_id), ) if schedule.get("plant_id") and schedule.get("task_type") == "fertilize": cur.execute( """ INSERT INTO GardenPlantEvent (plant_id, event_type, event_date, notes) VALUES (%s, 'fertilized', %s, %s) """, ( schedule["plant_id"], date.today(), data.get("notes") or schedule.get("title"), ), ) cur.execute("SELECT * FROM GardenSchedule WHERE id = %s", (schedule_id,)) schedule = cur.fetchone() conn.commit() return jsonify({"ok": True, "schedule": _serialize_row(schedule)}) finally: conn.close() # --- Due tasks --- @bp.route("/api/tasks/due") def tasks_due(): days = int(request.args.get("days") or 7) conn = get_db() try: ensure_garden_tables(conn) until = datetime.utcnow() + timedelta(days=days) with conn.cursor() as cur: cur.execute( """ SELECT s.*, p.name AS plant_name, t.name AS tank_name FROM GardenSchedule s LEFT JOIN GardenPlant p ON p.id = s.plant_id LEFT JOIN GardenTank t ON t.id = s.tank_id WHERE s.enabled = 1 AND s.next_due <= %s ORDER BY s.next_due """, (until,), ) schedules = cur.fetchall() cur.execute("SELECT * FROM GardenTank ORDER BY name") tanks = cur.fetchall() now = datetime.utcnow() tank_tasks = [] for tank in tanks: cur.execute( """ SELECT recorded_at FROM GardenTankReading WHERE tank_id = %s ORDER BY recorded_at DESC LIMIT 1 """, (tank["id"],), ) latest = cur.fetchone() interval = int(tank.get("test_interval_days") or 7) overdue = not latest or ( latest["recorded_at"] + timedelta(days=interval) <= now ) if overdue: tank_tasks.append( { "type": "tank_test", "tank_id": tank["id"], "tank_name": tank["name"], "title": f"Test tank: {tank['name']}", "next_due": _serialize( latest["recorded_at"] + timedelta(days=interval) if latest else now ), "overdue": True, } ) zone = _get_zone_info(conn) weather = _get_weather() return jsonify( { "ok": True, "tasks": _serialize_rows(schedules), "tank_tasks": tank_tasks, "zone": zone, "weather": weather, } ) finally: conn.close() # --- Planting guide --- @bp.route("/api/zone-info") def zone_info(): conn = get_db() try: ensure_garden_tables(conn) return jsonify({"ok": True, "zone": _get_zone_info(conn), "weather": _get_weather()}) finally: conn.close() @bp.route("/api/planting-now") def planting_now(): conn = get_db() try: ensure_garden_tables(conn) settings = _get_settings(conn) or {} zip_code = str(settings.get("zip_code") or GARDEN_ZIP).strip() if zip_code != GARDEN_ZIP: return jsonify( { "ok": True, "show": False, "location": None, "data": None, "message": f"Plant now is only available for {GARDEN_LOCATION_LABEL} ({GARDEN_ZIP}).", } ) cache_key = f"planting_now:{GARDEN_ZIP}" cached = _cache_get(conn, cache_key) if cached: return jsonify( { "ok": True, "show": True, "location": {"city": GARDEN_LOCATION_LABEL, "zip": GARDEN_ZIP}, "data": cached, } ) try: data = _fetch_json_url( f"https://api.cropgraph.com/api/planting?lat={GARDEN_LAT}&lng={GARDEN_LON}" ) except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError) as exc: return jsonify({"ok": False, "error": str(exc)}), 502 _cache_set(conn, cache_key, data, 6) return jsonify( { "ok": True, "show": True, "location": {"city": GARDEN_LOCATION_LABEL, "zip": GARDEN_ZIP}, "data": data, } ) finally: conn.close() @bp.route("/api/crops/search") def search_crops(): q = (request.args.get("q") or "").strip() if not q: return jsonify({"ok": True, "results": []}) conn = get_db() try: ensure_garden_tables(conn) cache_key = f"crop_search:{q.lower()}" cached = _cache_get(conn, cache_key) if cached: return jsonify({"ok": True, "results": cached}) try: data = _fetch_json_url( f"https://api.cropgraph.com/api/search?q={urllib.parse.quote(q)}" ) except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError, OSError) as exc: return jsonify({"ok": False, "error": str(exc)}), 502 results = data.get("results") or data if isinstance(data, list) else [] _cache_set(conn, cache_key, results, 24) return jsonify({"ok": True, "results": results}) finally: conn.close() @bp.route("/api/planting-guide") def planting_guide(): crop = (request.args.get("crop") or "").strip() if not crop: return jsonify({"ok": False, "error": "crop query param required"}), 400 conn = get_db() try: ensure_garden_tables(conn) guide = _fetch_cropgraph_planting(conn, crop) if not guide: return jsonify({"ok": False, "error": "Crop not found"}), 404 return jsonify({"ok": True, "guide": guide, "zone": _get_zone_info(conn)}) finally: conn.close() # --- SPA --- @bp.route("/", defaults={"path": ""}) @bp.route("/") def spa(path): dist_dir = GARDEN_CLIENT_DIST if path.startswith("api/"): return jsonify({"error": "Not found"}), 404 if path and os.path.exists(os.path.join(dist_dir, path)): return send_from_directory(dist_dir, path) index_path = os.path.join(dist_dir, "index.html") if os.path.exists(index_path): return send_from_directory(dist_dir, "index.html") return jsonify( { "error": "Garden Tracker frontend not built", "message": "Build the React client and place output in GARDEN_CLIENT_DIST.", "dist": dist_dir, } ), 503