import base64 import hashlib import hmac import json import math import mimetypes import os import secrets import sqlite3 import time import uuid from contextlib import contextmanager from datetime import datetime, timedelta from typing import List, Optional import markdown as md from fastapi import FastAPI, File, Form, HTTPException, Query, Request, UploadFile from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from starlette.middleware.base import BaseHTTPMiddleware DB_PATH = "/data/garden.db" PASSWORD = os.environ.get("GARDEN_PASSWORD", "garden") API_TOKEN = os.environ.get("GARDEN_API_TOKEN", "changeme") SECRET_KEY = os.environ.get("GARDEN_SECRET_KEY", "dev-secret-change-me") GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "") GEMINI_TRANSCRIBE_MODEL = os.environ.get("GEMINI_TRANSCRIBE_MODEL", "gemini-2.5-flash") GEMINI_CHAT_MODEL = os.environ.get("GEMINI_CHAT_MODEL", "gemini-2.5-flash") # Hard cap on how many round-trips the function-calling loop may take per # user message. Each round-trip is a Gemini call; keep this generous enough # for "find empty space → check 3 areas → fetch notes" style chains, but low # enough to bound latency + cost when the model goes off the rails. CHAT_MAX_TOOL_ITERATIONS = 6 COOKIE_NAME = "garden_session" SESSION_DAYS_DEFAULT = 14 SESSION_DAYS_REMEMBER = 365 PUBLIC_PATHS = {"/login", "/logout", "/health"} # Plant physical state. The model: # idea — a planning placeholder; the plant may not physically exist yet. # unplanted — physically exists (purchased / seeded indoors) but not in ground. # Can be placed in an area for layout planning while staying unplanted. # planted — in the ground; must have planting_id set. # removed — terminal (died, pulled, given away). Excluded from active views. # Old verbose states (active, growing, flowering, fruiting, harvesting, dormant, # dead, planned) were normalized into this set on the 2026-05-14 migration. PLANT_STATUSES = ["idea", "unplanted", "planted", "removed"] SUPPLY_CATEGORIES = [ "", "fertilizer", "soil", "amendment", "mulch", "pesticide", "seed", "tool", "other", ] PLANTING_SOURCES = [ "", "seed", "sapling", "transplant", "runner", "division", "cutting", "purchase", "volunteer", "other", ] # Planting status was previously a workflow state (idea/planned/planted/skipped) # used as a workaround for the NOT NULL planting_id constraint on plants. Under # the unplanted-plants model (2026-05-14), plantings only exist when something # is actually planted, so "planted" is the only meaningful value. Kept as a # single-element list for backward compatibility with form handlers. PLANTING_STATUSES = ["planted"] # Species classification. Each species is either a plant or an animal, with a # primary function constrained by the type. Varieties inherit from their species. SPECIES_TYPES = ["plant", "animal"] SPECIES_FUNCTION_OPTIONS = { "plant": ["edible", "ornamental", "support", "weed"], "animal": ["beneficial", "pest"], } def species_function_valid(type_val: str, fn_val: str) -> bool: return fn_val in SPECIES_FUNCTION_OPTIONS.get(type_val, []) def normalize_species_kind(type_val: str, fn_val: str) -> tuple: """Coerce an arbitrary (type, primary_function) pair to a valid combo. Falls back to (plant, edible) if the type is unknown; otherwise picks the first valid function for the type.""" type_val = (type_val or "").strip().lower() if type_val not in SPECIES_FUNCTION_OPTIONS: type_val = "plant" fn_val = (fn_val or "").strip().lower() if not species_function_valid(type_val, fn_val): fn_val = SPECIES_FUNCTION_OPTIONS[type_val][0] return type_val, fn_val # Inline click-to-edit configuration. For each entity type, the keys are # field names that can be edited via POST /api/edit/{entity}/{id}; the values # describe how to validate / render. Type meanings: # "text" plain string # "textarea" multi-line string (overview-style fields use snapshot=True so # every change records an overview_versions row) # "int" integer # "float" floating point # "select" one of an enumerated set (`options` is the allowed list) # "fk" foreign key into another table (`fk_table` + `fk_id_col`); the # client posts the id, server validates the row exists. # A "snapshot" key triggers `snapshot_overview` before the update so the # previous overview/notes value is captured in `overview_versions`. EDITABLE_FIELDS = { "area": { "name": {"type": "text", "required": True}, "short_name": {"type": "text"}, "length_ft": {"type": "float", "min": 0}, "width_ft": {"type": "float", "min": 0}, "garden_area_sqft": {"type": "float", "min": 0}, "structure_features": {"type": "textarea"}, "sunlight": {"type": "textarea"}, "soil_environment": {"type": "textarea"}, "notes": {"type": "textarea", "snapshot": True}, "sketch_rotation": {"type": "select", "options": [0, 45, 90, 135, 180, 225, 270, 315]}, }, "planting": { "name": {"type": "text", "required": True}, "source": {"type": "text"}, # Status retired in the unplanted-plants model — kept as a single-value # select to preserve the click-to-edit endpoint shape; no UI exposes it. "status": {"type": "select", "options": ["planted"]}, "notes": {"type": "textarea", "snapshot": True}, "year_id": {"type": "fk", "fk_table": "years", "nullable": True}, }, "year": { "year": {"type": "int", "required": True, "min": 1900, "max": 2100}, "notes": {"type": "textarea", "snapshot": True}, }, "species": { "name": {"type": "text", "required": True}, "common_name": {"type": "text"}, "description": {"type": "textarea", "snapshot": True}, "type": {"type": "select", "options": ["plant", "animal"]}, "primary_function": {"type": "select", "options": [ "edible", "ornamental", "support", "weed", "beneficial", "pest"]}, }, "variety": { "name": {"type": "text", "required": True}, "description": {"type": "textarea", "snapshot": True}, "species_id": {"type": "fk", "fk_table": "species", "nullable": True}, }, "station": { "name": {"type": "text", "required": True}, "notes": {"type": "textarea"}, "current_schedule": {"type": "textarea"}, }, } # Tables for entity_type -> SQL identifier (used by /api/edit handler). EDITABLE_TABLES = { "area": "areas", "planting": "plantings", "year": "years", "species": "species", "variety": "varieties", "station": "watering_stations", } COMMENT_KINDS = ["", "observation", "question", "done", "issue", "idea", "note"] ARTIFACT_KINDS = ["planting-plan", "fertilization-schedule", "shopping-list", "watering-plan", "other"] COMMENT_TARGET_TYPES = {"area", "plant", "species", "variety", "planting", "year", "station"} ARTIFACTS_DIR = "/data/artifacts" NOTES_DIR = "/data/notes" NOTE_STATUSES = ["new", "processing", "processed", "archived"] app = FastAPI() ASSET_VERSION = datetime.utcnow().strftime("%Y%m%d%H%M%S") templates = Jinja2Templates(directory="templates") templates.env.globals["ASSET_VERSION"] = ASSET_VERSION templates.env.globals["PLANT_STATUSES"] = PLANT_STATUSES templates.env.globals["SUPPLY_CATEGORIES"] = SUPPLY_CATEGORIES templates.env.globals["PLANTING_SOURCES"] = PLANTING_SOURCES templates.env.globals["PLANTING_STATUSES"] = PLANTING_STATUSES templates.env.globals["COMMENT_KINDS"] = COMMENT_KINDS templates.env.globals["ARTIFACT_KINDS"] = ARTIFACT_KINDS templates.env.globals["SPECIES_TYPES"] = SPECIES_TYPES templates.env.globals["SPECIES_FUNCTION_OPTIONS"] = SPECIES_FUNCTION_OPTIONS templates.env.globals["SKETCH_ROTATION_STEPS"] = (0, 45, 90, 135, 180, 225, 270, 315) # Species icon registry. For each known species name, either a unicode emoji # (rendered as plain text) or an SVG symbol id (rendered via ). Anything # not in this map falls back to the leaf symbol. # Per-species spacing defaults (plants_per_unit, space_per_unit_sqft). # `space_per_unit_sqft` is the size of one icon's footprint on the area sketch # in square feet (so a 2 sqft tomato icon takes up a 2x2 ft visual block); # `plants_per_unit` is how many actual plants that one icon represents. # The area-sketch renderer uses these to size icons proportionally and to # compute icon counts for grouped crops (e.g. 30 carrots → 10 icons of 3). SPECIES_SPACING_DEFAULTS = [ # warm-season anchors ("Tomato", 1, 2.0), ("Bell Peppers", 1, 1.5), ("Hot Pepper", 1, 1.5), ("Eggplant", 1, 2.0), # brassicas ("Cabbage", 1, 2.0), ("Cauliflower", 1, 2.0), ("Broccoli", 1, 2.0), ("Brussels Sprouts", 1, 2.0), ("Kale", 1, 1.5), # greens / lettuces ("Lettuce", 1, 0.5), ("Spinach", 2, 0.5), # roots ("Carrots", 3, 0.5), ("Radish", 9, 0.5), ("Beet", 4, 0.5), ("Onion", 4, 0.5), ("Green Onions", 4, 0.5), ("Garlic", 4, 0.5), ("Potato", 1, 1.5), ("Potatoes", 1, 1.5), ("Yams", 1, 2.0), # legumes ("String Beans", 4, 0.5), ("Snap Peas", 4, 0.5), # cucurbits ("Cucumbers", 1, 1.0), ("Squash", 1, 4.0), ("Zucchini", 1, 4.0), ("Cantaloupe", 1, 4.0), # corn (block) ("Corn", 1, 1.0), # berries / shrubs ("Strawberry", 1, 1.0), ("Blueberry", 1, 16.0), ("Raspberry", 1, 3.0), ("Currant", 1, 16.0), ("Gooseberry", 1, 16.0), ("Jostaberry", 1, 16.0), ("Haskap", 1, 16.0), ("Lingonberry", 1, 4.0), ("Blackberry", 1, 16.0), ("Marionberry", 1, 16.0), ("Tayberry", 1, 16.0), ("Loganberry", 1, 16.0), ("Branchberry", 1, 16.0), ("Pawpaw", 1, 100.0), # tree fruits (mature spread) ("Apple", 1, 100.0), ("Pear", 1, 100.0), ("Peach", 1, 100.0), ("Plum", 1, 100.0), ("Apricot", 1, 100.0), ("Cherry", 1, 25.0), # vines ("Grape", 1, 25.0), ("Kiwi", 1, 36.0), # herbs ("Basil", 1, 1.0), ("Parsley", 1, 0.5), ("Cilantro", 4, 0.5), ("Chives", 1, 1.0), # flowers / companions ("Marigold", 1, 0.5), ("Calendula", 1, 1.0), ("Borage", 1, 3.0), ("Nasturtium", 1, 1.0), ("Alyssum", 1, 0.5), ("Creeper", 1, 0.5), # cover crops / soil fixers ("Buckwheat", 4, 0.5), ("White Clover", 16, 0.5), ("Comfrey", 1, 9.0), # wildlife ("Mason Bee", 1, 1.0), ] SPECIES_ICONS = { # Emoji (preferred when one exists) "Strawberry": {"type": "emoji", "char": "\U0001F353"}, "Blueberry": {"type": "emoji", "char": "\U0001FAD0"}, "Apple": {"type": "emoji", "char": "\U0001F34E"}, "Pear": {"type": "emoji", "char": "\U0001F350"}, "Peach": {"type": "emoji", "char": "\U0001F351"}, "Cherry": {"type": "emoji", "char": "\U0001F352"}, "Grape": {"type": "emoji", "char": "\U0001F347"}, "Kiwi": {"type": "emoji", "char": "\U0001F95D"}, "Tomato": {"type": "emoji", "char": "\U0001F345"}, "Brocolli": {"type": "emoji", "char": "\U0001F966"}, "Corn": {"type": "emoji", "char": "\U0001F33D"}, "Cucumbers": {"type": "emoji", "char": "\U0001F952"}, "Cabbage": {"type": "emoji", "char": "\U0001F96C"}, "Eggplant": {"type": "emoji", "char": "\U0001F346"}, "Bell Peppers": {"type": "emoji", "char": "\U0001FAD1"}, "Lettuce": {"type": "emoji", "char": "\U0001F96C"}, "Potatoes": {"type": "emoji", "char": "\U0001F954"}, "Yams": {"type": "emoji", "char": "\U0001F360"}, "Green Onions": {"type": "emoji", "char": "\U0001F9C5"}, "Basil": {"type": "emoji", "char": "\U0001F33F"}, "Parsley": {"type": "emoji", "char": "\U0001F33F"}, "Cilantro": {"type": "emoji", "char": "\U0001F33F"}, "String Beans": {"type": "emoji", "char": "\U0001FAD8"}, "Snap Peas": {"type": "emoji", "char": "\U0001FADB"}, "Cantaloupe": {"type": "emoji", "char": "\U0001F348"}, "Zucchini": {"type": "emoji", "char": "\U0001F952"}, "Onion": {"type": "emoji", "char": "\U0001F9C5"}, "Hot Pepper": {"type": "emoji", "char": "\U0001F336"}, "Carrots": {"type": "emoji", "char": "\U0001F955"}, "Spinach": {"type": "emoji", "char": "\U0001F96C"}, "Garlic": {"type": "emoji", "char": "\U0001F9C4"}, "Marigold": {"type": "emoji", "char": "\U0001F33C"}, "Buckwheat": {"type": "emoji", "char": "\U0001F33E"}, "White Clover": {"type": "emoji", "char": "\U0001F340"}, # New custom SVG symbols (added 2026-05-05) — give species without good # emoji their own distinct artwork. "Beet": {"type": "svg", "id": "icon-beet"}, "Radish": {"type": "svg", "id": "icon-radish"}, "Chives": {"type": "svg", "id": "icon-chives"}, "Comfrey": {"type": "svg", "id": "icon-comfrey"}, "Borage": {"type": "svg", "id": "icon-borage"}, "Nasturtium": {"type": "svg", "id": "icon-nasturtium"}, "Calendula": {"type": "svg", "id": "icon-calendula"}, "Mason Bee": {"type": "svg", "id": "icon-mason-bee"}, "Broccoli": {"type": "emoji", "char": "\U0001F966"}, # canonical spelling now "Brussels Sprouts": {"type": "emoji", "char": "\U0001F96C"}, "Kale": {"type": "emoji", "char": "\U0001F33F"}, "Potato": {"type": "emoji", "char": "\U0001F954"}, # singular vs existing "Potatoes" # SVG symbols defined in _species_icons.html "Raspberry": {"type": "svg", "id": "icon-raspberry"}, "Currant": {"type": "svg", "id": "icon-currant"}, "Haskap": {"type": "svg", "id": "icon-haskap"}, "Jostaberry": {"type": "svg", "id": "icon-jostaberry"}, "Gooseberry": {"type": "svg", "id": "icon-gooseberry"}, "Lingonberry": {"type": "svg", "id": "icon-lingonberry"}, "Blackberry": {"type": "svg", "id": "icon-blackberry"}, "Marionberry": {"type": "svg", "id": "icon-marionberry"}, "Tayberry": {"type": "svg", "id": "icon-tayberry"}, "Loganberry": {"type": "svg", "id": "icon-loganberry"}, "Branchberry": {"type": "svg", "id": "icon-blackberry"}, "Pawpaw": {"type": "svg", "id": "icon-pawpaw"}, "Plum": {"type": "svg", "id": "icon-plum"}, "Apricot": {"type": "svg", "id": "icon-apricot"}, "Alyssum": {"type": "svg", "id": "icon-alyssum"}, "Creeper": {"type": "svg", "id": "icon-creeper"}, "Squash": {"type": "svg", "id": "icon-squash"}, "Cauliflower": {"type": "svg", "id": "icon-cauliflower"}, } DEFAULT_SPECIES_ICON = {"type": "svg", "id": "icon-leaf"} def species_icon(name: Optional[str]) -> dict: """Template helper — returns {'type': 'emoji'|'svg', 'char'|'id': ...}.""" if not name: return DEFAULT_SPECIES_ICON return SPECIES_ICONS.get(name, DEFAULT_SPECIES_ICON) templates.env.globals["species_icon"] = species_icon app.mount("/static", StaticFiles(directory="static"), name="static") def init_db(): os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) with sqlite3.connect(DB_PATH) as con: # FK must be OFF during schema changes tables = { r[0] for r in con.execute( "SELECT name FROM sqlite_master WHERE type='table'" ) } if "zones" in tables and "areas" not in tables: con.execute("ALTER TABLE zones RENAME TO areas") con.executescript(""" CREATE TABLE IF NOT EXISTS areas ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, notes TEXT DEFAULT '', parent_id INTEGER, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS watering_stations ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, notes TEXT DEFAULT '', current_schedule TEXT DEFAULT '', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS area_stations ( area_id INTEGER NOT NULL REFERENCES areas(id) ON DELETE CASCADE, station_id INTEGER NOT NULL REFERENCES watering_stations(id) ON DELETE CASCADE, PRIMARY KEY (area_id, station_id) ); CREATE TABLE IF NOT EXISTS supplies ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, category TEXT DEFAULT '', amount TEXT DEFAULT '', notes TEXT DEFAULT '', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS species ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, common_name TEXT DEFAULT '', description TEXT DEFAULT '', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS varieties ( id INTEGER PRIMARY KEY AUTOINCREMENT, species_id INTEGER REFERENCES species(id) ON DELETE SET NULL, name TEXT NOT NULL, description TEXT DEFAULT '', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS years ( id INTEGER PRIMARY KEY AUTOINCREMENT, year INTEGER NOT NULL UNIQUE, notes TEXT DEFAULT '', created_at TEXT NOT NULL ); """) # Drop legacy tables if they still exist (were empty in prod). con.execute("DROP TABLE IF EXISTS event_plants") con.execute("DROP TABLE IF EXISTS events") con.execute("DROP TABLE IF EXISTS schedule") area_cols = {r[1] for r in con.execute("PRAGMA table_info(areas)")} if "sort_order" not in area_cols: con.execute("ALTER TABLE areas ADD COLUMN sort_order INTEGER") con.execute( "UPDATE areas SET sort_order = id WHERE sort_order IS NULL" ) if "short_name" not in area_cols: con.execute("ALTER TABLE areas ADD COLUMN short_name TEXT DEFAULT ''") if "length_ft" not in area_cols: con.execute("ALTER TABLE areas ADD COLUMN length_ft REAL DEFAULT 0") if "width_ft" not in area_cols: con.execute("ALTER TABLE areas ADD COLUMN width_ft REAL DEFAULT 0") if "garden_area_sqft" not in area_cols: con.execute("ALTER TABLE areas ADD COLUMN garden_area_sqft REAL DEFAULT 0") if "structure_features" not in area_cols: con.execute("ALTER TABLE areas ADD COLUMN structure_features TEXT DEFAULT ''") if "sunlight" not in area_cols: con.execute("ALTER TABLE areas ADD COLUMN sunlight TEXT DEFAULT ''") if "soil_environment" not in area_cols: con.execute("ALTER TABLE areas ADD COLUMN soil_environment TEXT DEFAULT ''") if "featured_image_path" not in area_cols: con.execute("ALTER TABLE areas ADD COLUMN featured_image_path TEXT DEFAULT ''") if "sketch_rotation" not in area_cols: # Degrees clockwise applied to the natural rectangle (length-horizontal). # 0 = length runs east-west; 90 = length runs north-south. The detail # page draws compass markers (E/W/N/S) so the rotation is meaningful. con.execute("ALTER TABLE areas ADD COLUMN sketch_rotation INTEGER DEFAULT 0") # Position of the child's center within the parent's natural-orientation # rectangle (feet from parent's top-left). NULL = not yet sited; renderer # falls back to a default grid layout. if "parent_pos_x" not in area_cols: con.execute("ALTER TABLE areas ADD COLUMN parent_pos_x REAL") if "parent_pos_y" not in area_cols: con.execute("ALTER TABLE areas ADD COLUMN parent_pos_y REAL") # Migrate supplies table supply_cols = {r[1] for r in con.execute("PRAGMA table_info(supplies)")} if "quantity_on_hand" not in supply_cols: con.execute("ALTER TABLE supplies ADD COLUMN quantity_on_hand TEXT DEFAULT ''") con.execute("ALTER TABLE supplies ADD COLUMN quantity_needed TEXT DEFAULT ''") con.execute("ALTER TABLE supplies ADD COLUMN needed INTEGER DEFAULT 0") con.execute("UPDATE supplies SET quantity_on_hand = amount WHERE amount != ''") # Featured image support across additional entities. Areas already have # the column (added earlier); these mirror it on every other entity that # benefits from a photo (planting projects, individual plant groups, # watering stations, species, varieties, supplies, years). for tbl in ("plantings", "plants", "watering_stations", "species", "varieties", "supplies", "years"): cols = {r[1] for r in con.execute(f"PRAGMA table_info({tbl})")} if "featured_image_path" not in cols: con.execute( f"ALTER TABLE {tbl} ADD COLUMN featured_image_path TEXT DEFAULT ''" ) # Plant positions within an area's sketch — JSON list of {x, y} pairs # in rect-local coordinates (feet from the natural-orientation top-left # of the rectangle, so x in [0, length_ft], y in [0, width_ft]). # Length should equal the plant_group's quantity; renderer falls back # to a default grid for missing entries. plant_cols_now = {r[1] for r in con.execute("PRAGMA table_info(plants)")} if "positions" not in plant_cols_now: con.execute("ALTER TABLE plants ADD COLUMN positions TEXT DEFAULT ''") # Per-species spacing: how much area one icon represents and how many # plants fit in that footprint. Used by the area-sketch icon sizing # so a tomato visually occupies its 2-sqft block while a carrot icon # (which represents 3 carrots) occupies just 0.5 sqft. species_cols_for_spacing = {r[1] for r in con.execute("PRAGMA table_info(species)")} if "plants_per_unit" not in species_cols_for_spacing: con.execute("ALTER TABLE species ADD COLUMN plants_per_unit INTEGER DEFAULT 1") if "space_per_unit_sqft" not in species_cols_for_spacing: con.execute("ALTER TABLE species ADD COLUMN space_per_unit_sqft REAL DEFAULT 1.0") # Varieties can override species spacing — both nullable so NULL means # "inherit the species value." load_plants_by_area uses COALESCE to # compute the effective spacing per plant_group. variety_cols_for_spacing = {r[1] for r in con.execute("PRAGMA table_info(varieties)")} if "plants_per_unit" not in variety_cols_for_spacing: con.execute("ALTER TABLE varieties ADD COLUMN plants_per_unit INTEGER") if "space_per_unit_sqft" not in variety_cols_for_spacing: con.execute("ALTER TABLE varieties ADD COLUMN space_per_unit_sqft REAL") # Seed reasonable defaults for known species. Idempotent: only writes # rows still at the (1, 1.0) initial-default — never overrides a value # the user (or an earlier seed pass) has already changed. for name, ppu, sqft in SPECIES_SPACING_DEFAULTS: con.execute( """UPDATE species SET plants_per_unit = ?, space_per_unit_sqft = ? WHERE LOWER(name) = LOWER(?) AND COALESCE(plants_per_unit, 1) = 1 AND COALESCE(space_per_unit_sqft, 1.0) = 1.0""", (ppu, sqft, name), ) # Rebuild plantings table if old schema (has variety_id column) planting_cols = {r[1] for r in con.execute("PRAGMA table_info(plantings)")} if "variety_id" in planting_cols: # Old schema — rebuild con.executescript(""" DROP TABLE IF EXISTS plantings; CREATE TABLE plantings ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, year_id INTEGER REFERENCES years(id) ON DELETE SET NULL, source TEXT DEFAULT '', notes TEXT DEFAULT '', created_at TEXT NOT NULL ); """) elif "plantings" not in tables: con.executescript(""" CREATE TABLE plantings ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, year_id INTEGER REFERENCES years(id) ON DELETE SET NULL, source TEXT DEFAULT '', notes TEXT DEFAULT '', created_at TEXT NOT NULL ); """) # Add Planting.status (idea/planned/planted/skipped) for planning-phase workflow planting_cols_now = {r[1] for r in con.execute("PRAGMA table_info(plantings)")} if "status" not in planting_cols_now: con.execute("ALTER TABLE plantings ADD COLUMN status TEXT DEFAULT 'planted'") con.execute("UPDATE plantings SET status = 'planted' WHERE status IS NULL OR status = ''") # Overview version history — snapshot taken before any overview save. # entity_type covers anything with a "notes/overview" field: area, planting, # plant, species, variety, year. author is 'user' or 'ai'. con.executescript(""" CREATE TABLE IF NOT EXISTS overview_versions ( id INTEGER PRIMARY KEY AUTOINCREMENT, entity_type TEXT NOT NULL, entity_id INTEGER NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL, author TEXT NOT NULL DEFAULT 'user', change_note TEXT DEFAULT '' ); CREATE INDEX IF NOT EXISTS idx_overview_versions_entity ON overview_versions (entity_type, entity_id, created_at DESC); """) # Artifacts (AI-generated markdown outputs: plans, schedules, lists) con.executescript(""" CREATE TABLE IF NOT EXISTS artifacts ( id INTEGER PRIMARY KEY AUTOINCREMENT, kind TEXT NOT NULL, title TEXT NOT NULL, filename TEXT NOT NULL UNIQUE, generated_at TEXT NOT NULL, notes TEXT DEFAULT '' ); CREATE INDEX IF NOT EXISTS idx_artifacts_kind ON artifacts (kind); """) os.makedirs(ARTIFACTS_DIR, exist_ok=True) # Field notes — unified table holding both raw audio/photo/text captures # AND finalized "comment-style" notes (the post-processing form). Originally # two separate tables (`field_notes` for captures, `comments` for finalized # notes); merged 2026-05-04 so one timeline holds everything. A row can be: # - a capture awaiting processing (status='new', text/transcript/audio_path # populated, body empty) # - a processed capture (status='processed', body filled in by the user # during processing, original text/transcript preserved) # - a directly-authored note (status='processed', body filled in directly, # captured_at/audio_path/photo_paths empty) # Multi-target via field_note_targets (replaces comment_targets). con.executescript(""" CREATE TABLE IF NOT EXISTS field_notes ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL, captured_at TEXT, audio_path TEXT, photo_paths TEXT DEFAULT '', text TEXT DEFAULT '', transcript TEXT DEFAULT '', status TEXT NOT NULL DEFAULT 'new', processed_at TEXT, processing_notes TEXT DEFAULT '' ); CREATE INDEX IF NOT EXISTS idx_field_notes_status ON field_notes (status, created_at DESC); """) os.makedirs(NOTES_DIR, exist_ok=True) # Add comment-style columns to field_notes (additive; idempotent). fn_cols = {r[1] for r in con.execute("PRAGMA table_info(field_notes)")} if "body" not in fn_cols: con.execute("ALTER TABLE field_notes ADD COLUMN body TEXT DEFAULT ''") if "kind" not in fn_cols: con.execute("ALTER TABLE field_notes ADD COLUMN kind TEXT DEFAULT ''") if "comment_date" not in fn_cols: con.execute("ALTER TABLE field_notes ADD COLUMN comment_date TEXT") if "parent_id" not in fn_cols: con.execute("ALTER TABLE field_notes ADD COLUMN parent_id INTEGER REFERENCES field_notes(id) ON DELETE SET NULL") # Index on comment_date supports the per-entity timeline queries that # used to be ORDER BY comment_date DESC against the comments table. con.execute("CREATE INDEX IF NOT EXISTS idx_field_notes_comment_date " "ON field_notes (comment_date DESC, created_at DESC)") # field_note_targets replaces comment_targets. Same shape, comment_id # column renamed to field_note_id. con.executescript(""" CREATE TABLE IF NOT EXISTS field_note_targets ( id INTEGER PRIMARY KEY AUTOINCREMENT, field_note_id INTEGER NOT NULL REFERENCES field_notes(id) ON DELETE CASCADE, target_type TEXT NOT NULL, target_id INTEGER NOT NULL, is_primary INTEGER DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_field_note_targets_target ON field_note_targets (target_type, target_id); CREATE INDEX IF NOT EXISTS idx_field_note_targets_fnid ON field_note_targets (field_note_id); CREATE UNIQUE INDEX IF NOT EXISTS idx_field_note_targets_unique ON field_note_targets (field_note_id, target_type, target_id); """) # Rebuild plants table — new schema requires planting_id plant_cols = {r[1] for r in con.execute("PRAGMA table_info(plants)")} needs_rebuild = ("name" in plant_cols) or ("planting_id" not in plant_cols) if needs_rebuild: con.executescript(""" DROP TABLE IF EXISTS plants; CREATE TABLE plants ( id INTEGER PRIMARY KEY AUTOINCREMENT, planting_id INTEGER NOT NULL REFERENCES plantings(id) ON DELETE CASCADE, species_id INTEGER REFERENCES species(id) ON DELETE SET NULL, variety_id INTEGER REFERENCES varieties(id) ON DELETE SET NULL, area_id INTEGER REFERENCES areas(id) ON DELETE SET NULL, quantity TEXT DEFAULT '', source TEXT DEFAULT '', status TEXT DEFAULT '', notes TEXT DEFAULT '', created_at TEXT NOT NULL ); """) elif "plants" not in tables: con.executescript(""" CREATE TABLE plants ( id INTEGER PRIMARY KEY AUTOINCREMENT, planting_id INTEGER REFERENCES plantings(id) ON DELETE SET NULL, species_id INTEGER REFERENCES species(id) ON DELETE SET NULL, variety_id INTEGER REFERENCES varieties(id) ON DELETE SET NULL, area_id INTEGER REFERENCES areas(id) ON DELETE SET NULL, quantity TEXT DEFAULT '', source TEXT DEFAULT '', status TEXT DEFAULT 'planted', notes TEXT DEFAULT '', created_at TEXT NOT NULL ); """) # 2026-05-14: Unplanted-plants model. Two structural changes to plants: # 1. planting_id NOT NULL → nullable (plants can exist without a # planting). ON DELETE CASCADE → SET NULL (deleting a planting # detaches its plants instead of nuking them). # 2. plants.status enum is now {idea, unplanted, planted, removed}; # previous values get normalized. # Detect by inspecting the notnull flag on planting_id and rebuild the # table only if needed. Idempotent. plant_pragma = list(con.execute("PRAGMA table_info(plants)")) planting_id_col = next((c for c in plant_pragma if c[1] == "planting_id"), None) if planting_id_col and planting_id_col[3] == 1: # SQLite can't change NOT NULL in place — rebuild via rename+swap. # Copy ALL existing columns so nothing is lost. con.executescript(""" ALTER TABLE plants RENAME TO _plants_old_v6; CREATE TABLE plants ( id INTEGER PRIMARY KEY AUTOINCREMENT, planting_id INTEGER REFERENCES plantings(id) ON DELETE SET NULL, species_id INTEGER REFERENCES species(id) ON DELETE SET NULL, variety_id INTEGER REFERENCES varieties(id) ON DELETE SET NULL, area_id INTEGER REFERENCES areas(id) ON DELETE SET NULL, quantity TEXT DEFAULT '', source TEXT DEFAULT '', status TEXT DEFAULT 'planted', notes TEXT DEFAULT '', created_at TEXT NOT NULL, featured_image_path TEXT DEFAULT '', positions TEXT DEFAULT '', is_archived INTEGER DEFAULT 0 ); INSERT INTO plants (id, planting_id, species_id, variety_id, area_id, quantity, source, status, notes, created_at, featured_image_path, positions, is_archived) SELECT id, planting_id, species_id, variety_id, area_id, quantity, source, status, notes, created_at, featured_image_path, positions, is_archived FROM _plants_old_v6; DROP TABLE _plants_old_v6; CREATE INDEX IF NOT EXISTS idx_plants_archived ON plants (is_archived); """) # One-shot normalization of plants.status + cleanup of retired planting # statuses (idea/planned/skipped). Guards on existence of any retired # status so reruns are no-ops. old_status_plantings = con.execute( "SELECT id, status FROM plantings WHERE status IN ('idea', 'planned', 'skipped')" ).fetchall() if old_status_plantings: # Detach: each plant inherits a status that mirrors the planting's # old intent. Rows are tuples here (no Row factory inside init_db). map_status = {"idea": "idea", "planned": "unplanted", "skipped": "unplanted"} for pid_, pstatus_ in old_status_plantings: ns = map_status.get(pstatus_, "unplanted") con.execute( "UPDATE plants SET status = ?, planting_id = NULL WHERE planting_id = ?", (ns, pid_), ) con.execute("DELETE FROM plantings WHERE id = ?", (pid_,)) # Normalize legacy plant statuses (idempotent — only affects rows still # holding old values). Plants in planted plantings with a vague status # collapse to 'planted'; 'dead' becomes terminal 'removed'. con.execute( """UPDATE plants SET status = 'planted' WHERE status IN ('', 'active', 'growing', 'flowering', 'fruiting', 'harvesting', 'dormant', 'planned')""" ) con.execute("UPDATE plants SET status = 'removed' WHERE status = 'dead'") # 2026-05-14b: Merge duplicate UNATTACHED unplanted/idea plant_groups. # When the user detaches plants from areas, they should accumulate into # one row per (species, variety, status, planting). Older detaches that # happened before the auto-merge was wired up leave duplicates behind; # this one-shot reconciliation collapses them. # # Lossless: sums quantities, keeps the lowest id as the survivor, deletes # the rest. Only operates on rows with area_id IS NULL, no positions, # status IN (unplanted, idea), not archived. dup_groups = con.execute( """SELECT COALESCE(species_id, -1) AS s, COALESCE(variety_id, -1) AS v, status, COALESCE(planting_id, -1) AS pl, GROUP_CONCAT(id) AS ids, COUNT(*) AS n FROM plants WHERE area_id IS NULL AND COALESCE(positions, '') IN ('', '[]') AND status IN ('unplanted', 'idea') AND COALESCE(is_archived, 0) = 0 GROUP BY s, v, status, pl HAVING n > 1""" ).fetchall() for s, v, st, pl, ids_csv, n in dup_groups: ids = sorted(int(x) for x in ids_csv.split(",")) keeper = ids[0] losers = ids[1:] # Sum quantities (treat blank/non-numeric as 0). total = 0 for i in [keeper] + losers: qstr = con.execute("SELECT quantity FROM plants WHERE id = ?", (i,)).fetchone() try: total += int((qstr[0] or "0").strip()) except (TypeError, ValueError): pass con.execute( "UPDATE plants SET quantity = ? WHERE id = ?", (str(total), keeper), ) for li in losers: con.execute("DELETE FROM plants WHERE id = ?", (li,)) # Species classification: type (plant|animal) + primary_function. # Default new rows to plant/edible. Function values are constrained by # type at the application layer (see SPECIES_FUNCTION_OPTIONS): # plant → edible, ornamental, support, weed # animal → beneficial, pest species_cols_for_kind = {r[1] for r in con.execute("PRAGMA table_info(species)")} if "type" not in species_cols_for_kind: con.execute( "ALTER TABLE species ADD COLUMN type TEXT NOT NULL DEFAULT 'plant'" ) if "primary_function" not in species_cols_for_kind: con.execute( "ALTER TABLE species ADD COLUMN primary_function TEXT NOT NULL DEFAULT 'edible'" ) # One-time backfill: Mason Bee → animal/beneficial, plus the # support/weed species we know about. Idempotent: only updates rows # still at the (plant, edible) defaults so a user-set value is never # clobbered. _SPECIES_KIND_BACKFILL = { "Mason Bee": ("animal", "beneficial"), "Alyssum": ("plant", "support"), "Borage": ("plant", "support"), "Calendula": ("plant", "support"), "Marigold": ("plant", "support"), "Nasturtium": ("plant", "support"), "White Clover": ("plant", "support"), "Buckwheat": ("plant", "support"), "Comfrey": ("plant", "support"), "Creeper": ("plant", "weed"), } for name, (kind, fn) in _SPECIES_KIND_BACKFILL.items(): con.execute( """UPDATE species SET type = ?, primary_function = ? WHERE LOWER(name) = LOWER(?) AND type = 'plant' AND primary_function = 'edible'""", (kind, fn, name), ) # AI chat — sessions + messages + a meta table for tool gaps the model # flags via the `suggest_new_tool` meta-tool. Sessions are linked back # to activity_log via `activity_log_id` so the chat row in the log can # be updated (title/summary) when the session ends. con.executescript(""" CREATE TABLE IF NOT EXISTS chat_sessions ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL, ended_at TEXT, title TEXT DEFAULT '', summary TEXT DEFAULT '', activity_log_id INTEGER ); CREATE INDEX IF NOT EXISTS idx_chat_sessions_created ON chat_sessions (created_at DESC); CREATE TABLE IF NOT EXISTS chat_messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id INTEGER NOT NULL REFERENCES chat_sessions(id) ON DELETE CASCADE, created_at TEXT NOT NULL, role TEXT NOT NULL, text TEXT DEFAULT '', photo_paths TEXT DEFAULT '', tool_calls TEXT DEFAULT '', tokens_used INTEGER DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_chat_messages_session ON chat_messages (session_id, id); CREATE TABLE IF NOT EXISTS tool_suggestions ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL, session_id INTEGER REFERENCES chat_sessions(id) ON DELETE SET NULL, name TEXT NOT NULL, signature TEXT DEFAULT '', why_useful TEXT DEFAULT '', example_question TEXT DEFAULT '', status TEXT NOT NULL DEFAULT 'pending' ); CREATE INDEX IF NOT EXISTS idx_tool_suggestions_status ON tool_suggestions (status, created_at DESC); """) # Activity log — append-only record of every user-driven mutation. # Excludes purely-mechanical flows (sketch positioning, transcription). # `category` is one of the broad buckets shown in the UI: # add | edit | delete | archive | unarchive # `field`/`old_value`/`new_value` are populated for `edit` rows; the # rest leave them blank. `summary` is a short human-readable line for # the detailed-log timeline view. con.executescript(""" CREATE TABLE IF NOT EXISTS activity_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL, category TEXT NOT NULL, entity_type TEXT NOT NULL, entity_id INTEGER, summary TEXT NOT NULL, field TEXT, old_value TEXT, new_value TEXT ); CREATE INDEX IF NOT EXISTS idx_activity_log_created ON activity_log (created_at DESC); CREATE INDEX IF NOT EXISTS idx_activity_log_entity ON activity_log (entity_type, entity_id, created_at DESC); CREATE INDEX IF NOT EXISTS idx_activity_log_category ON activity_log (category, created_at DESC); """) # is_archived flag for soft-delete on every primary entity. Default 0. # Default queries hide archived rows; views can opt in to show them. for tbl in ("areas", "plantings", "plants", "species", "varieties", "watering_stations", "supplies", "years", "field_notes"): cols = {r[1] for r in con.execute(f"PRAGMA table_info({tbl})")} if "is_archived" not in cols: con.execute( f"ALTER TABLE {tbl} ADD COLUMN is_archived INTEGER DEFAULT 0" ) for tbl in ("areas", "plantings", "plants", "species", "varieties", "watering_stations", "supplies", "years", "field_notes"): con.execute( f"CREATE INDEX IF NOT EXISTS idx_{tbl}_archived " f"ON {tbl} (is_archived)" ) # One-time cleanup: plants with quantity=0 should be marked removed and # detached from their area so they don't count toward area totals. con.execute( """UPDATE plants SET area_id = NULL, status = 'removed' WHERE TRIM(quantity) = '0' AND status != 'removed'""" ) con.commit() init_db() @contextmanager def db(): con = sqlite3.connect(DB_PATH) con.row_factory = sqlite3.Row con.execute("PRAGMA foreign_keys = ON") try: yield con con.commit() finally: con.close() def now(): return datetime.utcnow().isoformat(timespec="seconds") # --- Activity log ------------------------------------------------------- # # Append-only record of user-driven mutations (add / edit / delete / archive # / unarchive). The intent is the "Detailed log" view on /timeline plus a # searchable record we can mine later for patterns. Sketch repositioning # (plant.positions, areas.parent_pos_*) is intentionally NOT logged — those # are visual layout tweaks, not data changes worth auditing. # # Helpers below wrap the raw INSERT so callers stay tiny at each mutation # site. They never raise: a failing log insert should never break the user- # facing operation. def _short(value, limit: int = 80) -> str: """Stringify and truncate a value for compact log display.""" if value is None: return "" s = str(value) if len(s) > limit: s = s[: limit - 1] + "…" return s def log_activity( con, category: str, entity_type: str, entity_id: Optional[int], summary: str, field: Optional[str] = None, old_value=None, new_value=None, ) -> None: try: con.execute( """INSERT INTO activity_log (created_at, category, entity_type, entity_id, summary, field, old_value, new_value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", ( now(), category, entity_type, entity_id, summary, field, None if old_value is None else _short(old_value, 500), None if new_value is None else _short(new_value, 500), ), ) except sqlite3.Error: # Logging must never break the underlying mutation. pass def log_add(con, entity_type: str, entity_id: int, label: str) -> None: log_activity( con, "add", entity_type, entity_id, f"Added {entity_type}: {_short(label, 80)}", ) def log_delete(con, entity_type: str, entity_id: Optional[int], label: str) -> None: log_activity( con, "delete", entity_type, entity_id, f"Deleted {entity_type}: {_short(label, 80)}", ) def log_edit( con, entity_type: str, entity_id: int, field: str, old_value, new_value, label: str = "", ) -> None: head = label or f"{entity_type} #{entity_id}" summary = f"{head}: {field} → {_short(new_value, 60)}" log_activity( con, "edit", entity_type, entity_id, summary, field=field, old_value=old_value, new_value=new_value, ) def log_archive(con, entity_type: str, entity_id: int, label: str) -> None: log_activity( con, "archive", entity_type, entity_id, f"Archived {entity_type}: {_short(label, 80)}", ) def log_unarchive(con, entity_type: str, entity_id: int, label: str) -> None: log_activity( con, "unarchive", entity_type, entity_id, f"Unarchived {entity_type}: {_short(label, 80)}", ) # Best-effort short label lookup for any logged entity. Defined here (next to # the log helpers) rather than near `target_label` because callers in # archive_entity / unarchive_entity / _maybe_archive_planting reference it # before target_label is in scope. _LOG_LABEL_LOOKUP = { "area": ("areas", "name"), "planting": ("plantings", "name"), "species": ("species", "name"), "variety": ("varieties", "name"), "station": ("watering_stations", "name"), "year": ("years", "year"), "supply": ("supplies", "name"), } def _entity_label_safe(con, entity_type: str, entity_id: Optional[int]) -> str: """Fetch a short, human-readable label for an entity. Falls back to ` #` when the row can't be found or the type is unknown.""" if entity_id is None: return entity_type fallback = f"{entity_type} #{entity_id}" try: if entity_type == "plant": r = con.execute( """SELECT COALESCE(v.name, s.name) AS nm, p.quantity FROM plants p LEFT JOIN species s ON s.id = p.species_id LEFT JOIN varieties v ON v.id = p.variety_id WHERE p.id = ?""", (entity_id,), ).fetchone() if r and r[0]: qty = (r[1] or "").strip() return f"{qty}× {r[0]}" if qty else r[0] return fallback if entity_type == "comment": r = con.execute( "SELECT body FROM field_notes WHERE id = ?", (entity_id,) ).fetchone() if r and r[0]: return _short(r[0], 60) return fallback spec = _LOG_LABEL_LOOKUP.get(entity_type) if not spec: return fallback table, col = spec r = con.execute( f"SELECT {col} FROM {table} WHERE id = ?", (entity_id,) ).fetchone() if r and r[0] is not None: return str(r[0]) except sqlite3.Error: pass return fallback # --- Auth --- def sign_session(data: dict) -> str: raw = json.dumps(data, separators=(",", ":")).encode() payload = base64.urlsafe_b64encode(raw).rstrip(b"=").decode() sig = hmac.new(SECRET_KEY.encode(), payload.encode(), hashlib.sha256).hexdigest() return f"{payload}.{sig}" def verify_session(cookie: Optional[str]) -> Optional[dict]: if not cookie or "." not in cookie: return None try: payload, sig = cookie.rsplit(".", 1) expected = hmac.new( SECRET_KEY.encode(), payload.encode(), hashlib.sha256 ).hexdigest() if not hmac.compare_digest(sig, expected): return None padding = "=" * ((4 - len(payload) % 4) % 4) data = json.loads(base64.urlsafe_b64decode(payload + padding)) if data.get("exp", 0) < time.time(): return None return data except Exception: return None def require_session(request: Request): """Enforce session cookie on /api/* endpoints (the AuthMiddleware skips /api/*).""" if not verify_session(request.cookies.get(COOKIE_NAME)): raise HTTPException(401, "auth required") def safe_next(value: Optional[str]) -> str: if not value or not value.startswith("/") or value.startswith("//"): return "/" return value class AuthMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): path = request.url.path if ( path.startswith("/static") or path.startswith("/api/") or path in PUBLIC_PATHS ): return await call_next(request) if not verify_session(request.cookies.get(COOKIE_NAME)): return RedirectResponse(f"/login?next={path}", status_code=303) return await call_next(request) app.add_middleware(AuthMiddleware) @app.get("/login", response_class=HTMLResponse) def login_form(request: Request, next: str = "/", error: int = 0): if verify_session(request.cookies.get(COOKIE_NAME)): return RedirectResponse(safe_next(next), status_code=303) return templates.TemplateResponse( "login.html", {"request": request, "next": safe_next(next), "error": bool(error)}, ) @app.post("/login") def login_submit( password: str = Form(...), remember: str = Form(""), next: str = Form("/"), ): target = safe_next(next) if not secrets.compare_digest(password, PASSWORD): return RedirectResponse(f"/login?next={target}&error=1", status_code=303) days = SESSION_DAYS_REMEMBER if remember else SESSION_DAYS_DEFAULT exp = int(time.time()) + days * 86400 token = sign_session({"exp": exp}) resp = RedirectResponse(target, status_code=303) resp.set_cookie( COOKIE_NAME, token, max_age=days * 86400, httponly=True, secure=True, samesite="lax", path="/", ) return resp @app.post("/logout") def logout(): resp = RedirectResponse("/login", status_code=303) resp.delete_cookie(COOKIE_NAME, path="/") return resp @app.get("/health") def health(): return {"ok": True} # --- Helpers --- def _parse_optional_int(value: str): value = (value or "").strip() return int(value) if value else None def load_areas(con, include_archived: bool = False) -> list: where = "" if include_archived else "WHERE is_archived = 0" return [ dict(r) for r in con.execute( f"SELECT * FROM areas {where} ORDER BY COALESCE(sort_order, 999999), name" ).fetchall() ] def build_tree(area_rows: list): by_id = {a["id"]: dict(a, children=[]) for a in area_rows} roots = [] for a in by_id.values(): pid = a.get("parent_id") if pid and pid in by_id: by_id[pid]["children"].append(a) else: roots.append(a) def sort_key(x): order = x.get("sort_order") return (order if order is not None else 999999, x["name"].lower()) def sort_children(node): node["children"].sort(key=sort_key) for c in node["children"]: sort_children(c) roots.sort(key=sort_key) for r in roots: sort_children(r) return roots, by_id def get_ancestors(area_id: int, by_id: dict) -> list: path = [] cur = by_id.get(area_id) while cur: path.append(cur) pid = cur.get("parent_id") cur = by_id.get(pid) if pid else None return list(reversed(path)) def get_descendant_ids(area_id: int, by_id: dict) -> set: result = set() def walk(aid): node = by_id.get(aid) if not node: return for child in node["children"]: result.add(child["id"]) walk(child["id"]) walk(area_id) return result def _qty(plant: dict) -> int: try: return int(plant["quantity"]) except (ValueError, TypeError): return 1 def get_plant_count_map(areas_tree: list, plants_by_area: dict) -> dict: """Sum of quantities for each area including all descendants.""" result = {} def walk(node): total = sum(_qty(p) for p in plants_by_area.get(node["id"], [])) for c in node["children"]: total += walk(c) result[node["id"]] = total return total for root in areas_tree: walk(root) return result def get_species_summary_map(areas_tree: list, plants_by_area: dict) -> dict: """For each area (id) return a list of {name, count} species summaries over the full subtree, sorted by count desc then name asc. `count` is the sum of plant quantities for that species within the subtree.""" result = {} def walk(node): own = {} for p in plants_by_area.get(node["id"], []): sp = p.get("species_name") if not sp: continue own[sp] = own.get(sp, 0) + _qty(p) merged = dict(own) for c in node["children"]: walk(c) for name, count in result[c["id"]]["_counts"].items(): merged[name] = merged.get(name, 0) + count ordered = sorted( ((n, c) for n, c in merged.items() if c > 0), key=lambda kv: (-kv[1], kv[0]), ) result[node["id"]] = { "_counts": merged, "summary": [{"name": n, "count": c} for n, c in ordered], } for root in areas_tree: walk(root) return {aid: info["summary"] for aid, info in result.items()} def get_area_plant_summary( con, area_id: int, by_id: dict, plant_statuses: Optional[set] = None, ) -> dict: """Build summary stats and consolidated plant list for an area and all descendants. `plant_statuses`: if given, only include plants whose `plants.status` is in this set. Pass None to include everything. Matches `load_plants_by_area`. """ desc_ids = [area_id] + list(get_descendant_ids(area_id, by_id)) placeholders = ",".join("?" * len(desc_ids)) rows = con.execute( f"""SELECT p.quantity, p.status AS plant_status, s.name AS species_name, v.name AS variety_name, s.id AS species_id, v.id AS variety_id, pl.id AS planting_id, pl.name AS planting_name FROM plants p LEFT JOIN species s ON s.id = p.species_id LEFT JOIN varieties v ON v.id = p.variety_id LEFT JOIN plantings pl ON pl.id = p.planting_id WHERE p.area_id IN ({placeholders})""", desc_ids, ).fetchall() total_qty = 0 species_set = set() variety_set = set() planting_set = set() consolidated = {} # (species_name, variety_name) -> qty for r in rows: if ( plant_statuses is not None and (r["plant_status"] or "planted") not in plant_statuses ): continue qty = 1 try: qty = int(r["quantity"]) except (ValueError, TypeError): pass total_qty += qty sp = r["species_name"] or "Unknown" vr = r["variety_name"] or "Unknown" if r["species_id"]: species_set.add(r["species_id"]) if r["variety_id"]: variety_set.add(r["variety_id"]) if r["planting_id"]: planting_set.add(r["planting_id"]) consolidated[(sp, vr)] = consolidated.get((sp, vr), 0) + qty plant_list = sorted( [{"species": sp, "variety": vr, "qty": q} for (sp, vr), q in consolidated.items()], key=lambda x: (-x["qty"], x["species"], x["variety"]), ) return { "total_qty": total_qty, "species_count": len(species_set), "variety_count": len(variety_set), "planting_count": len(planting_set), "plant_list": plant_list, } def area_path_name(area_id: int, by_id: dict) -> str: return " > ".join(a["name"] for a in get_ancestors(area_id, by_id)) def area_compact_path(area_id: int, by_id: dict) -> dict: """Return {parent: 'abbreviated parent path', name: 'area name'} for compact display.""" ancestors = get_ancestors(area_id, by_id) if not ancestors: return {"parent": "", "name": "?", "full": "?"} area = ancestors[-1] name = area["name"] # Build parent path, skipping top-level "Home" unless it IS the area parents = ancestors[:-1] if parents and parents[0]["name"].lower() == "home": parents = parents[1:] parts = [] for p in parents: sn = (p.get("short_name") or "").strip() parts.append(sn if sn else p["name"]) parent = " > ".join(parts) full = (parent + " > " + name) if parent else name return {"parent": parent, "name": name, "full": full} def area_tree_payload(areas_tree: list, by_id: dict, exclude_ids: Optional[set] = None) -> dict: """Compact tree data for the drill-down area picker. Returns `{"roots": [ids], "areas": {id: {...}}}`. `short_name` is always present (empty string when unset). `path_short` joins the short-names of each ancestor for rendering compact breadcrumbs client-side. """ exclude_ids = exclude_ids or set() areas = {} for aid, a in by_id.items(): if aid in exclude_ids: continue children = [c["id"] for c in a["children"] if c["id"] not in exclude_ids] ancestors = get_ancestors(aid, by_id)[:-1] path_short = " › ".join( (p.get("short_name") or p["name"]) for p in ancestors if p["id"] not in exclude_ids ) areas[aid] = { "id": aid, "name": a["name"], "short_name": (a.get("short_name") or "").strip(), "parent_id": a.get("parent_id"), "children": children, "path_short": path_short, } roots = [a["id"] for a in areas_tree if a["id"] not in exclude_ids] return {"roots": roots, "areas": areas} def flattened_area_options(by_id: dict, exclude_ids: Optional[set] = None): exclude_ids = exclude_ids or set() options = [] for aid in by_id: if aid in exclude_ids: continue cp = area_compact_path(aid, by_id) options.append({ "id": aid, "path": area_path_name(aid, by_id), "parent": cp["parent"], "name": cp["name"], "compact": cp["full"], }) options.sort(key=lambda x: x["path"].lower()) return options def load_plants_by_area( con, plant_statuses: Optional[set] = None, include_archived: bool = False, ) -> dict: """Map area_id → list of plants. `plant_statuses`: if given, only include plants whose `plants.status` is in this set. Pass None to include everything. Used by the overview pages to optionally hide unplanted / idea / removed plants. `include_archived`: when False (default), archived plants AND plants whose planting (if any) is archived are skipped. """ result = {} arch_clause = "" if include_archived else ( " WHERE p.is_archived = 0 AND COALESCE(pl.is_archived, 0) = 0" ) rows = con.execute( f"""SELECT p.*, s.name AS species_name, v.name AS variety_name, COALESCE(v.plants_per_unit, s.plants_per_unit, 1) AS species_plants_per_unit, COALESCE(v.space_per_unit_sqft, s.space_per_unit_sqft, 1.0) AS species_space_per_unit_sqft, pl.name AS planting_name FROM plants p LEFT JOIN species s ON s.id = p.species_id LEFT JOIN varieties v ON v.id = p.variety_id LEFT JOIN plantings pl ON pl.id = p.planting_id {arch_clause} ORDER BY COALESCE(v.name, s.name, '')""" ).fetchall() for p in rows: ps = (p["status"] or "planted") if plant_statuses is not None and ps not in plant_statuses: continue result.setdefault(p["area_id"], []).append(dict(p)) return result def delete_area_recursive(con, area_id: int): children = con.execute( "SELECT id FROM areas WHERE parent_id = ?", (area_id,) ).fetchall() for child in children: delete_area_recursive(con, child["id"]) # Unlink plants from this area (don't delete — they belong to plantings) con.execute("UPDATE plants SET area_id = NULL WHERE area_id = ?", (area_id,)) delete_target_refs(con, "area", area_id) con.execute("DELETE FROM areas WHERE id = ?", (area_id,)) # Mapping of entity_type → (table_name, id_column). Single source of truth for # archive routes and helpers. ARCHIVABLE_ENTITIES = { "area": "areas", "planting": "plantings", "plant": "plants", "species": "species", "variety": "varieties", "station": "watering_stations", "supply": "supplies", "year": "years", "comment": "field_notes", } def archive_entity(con, entity_type: str, entity_id: int) -> None: """Set is_archived=1 on an entity and apply down-cascades. Cascades (down only — structural children, not plants-as-tenants): - planting → all its plants - area → all descendant areas (recursive) - species → all its varieties After archiving a plant, also runs the planting auto-archive check (if every plant in the planting is now archived, archive the planting too). """ table = ARCHIVABLE_ENTITIES.get(entity_type) if not table: return label = _entity_label_safe(con, entity_type, entity_id) con.execute(f"UPDATE {table} SET is_archived = 1 WHERE id = ?", (entity_id,)) log_archive(con, entity_type, entity_id, label) if entity_type == "planting": # Log each plant individually so the cascade is visible in the log. plant_rows = con.execute( "SELECT id FROM plants WHERE planting_id = ? AND is_archived = 0", (entity_id,), ).fetchall() con.execute( "UPDATE plants SET is_archived = 1 WHERE planting_id = ?", (entity_id,), ) for r in plant_rows: plabel = _entity_label_safe(con, "plant", r[0]) log_archive(con, "plant", r[0], plabel + " (cascade from planting)") elif entity_type == "area": # Walk descendants; areas table has no recursive CTE setup, easier to BFS. frontier = [entity_id] while frontier: children = [ r[0] for r in con.execute( f"SELECT id FROM areas WHERE parent_id IN ({','.join('?' * len(frontier))})", tuple(frontier), ).fetchall() ] if not children: break child_labels = { cid: _entity_label_safe(con, "area", cid) for cid in children } con.execute( f"UPDATE areas SET is_archived = 1 " f"WHERE id IN ({','.join('?' * len(children))})", tuple(children), ) for cid in children: log_archive(con, "area", cid, child_labels[cid] + " (cascade from area)") frontier = children elif entity_type == "species": var_rows = con.execute( "SELECT id FROM varieties WHERE species_id = ? AND is_archived = 0", (entity_id,), ).fetchall() con.execute( "UPDATE varieties SET is_archived = 1 WHERE species_id = ?", (entity_id,), ) for r in var_rows: vlabel = _entity_label_safe(con, "variety", r[0]) log_archive(con, "variety", r[0], vlabel + " (cascade from species)") elif entity_type == "plant": row = con.execute( "SELECT planting_id FROM plants WHERE id = ?", (entity_id,) ).fetchone() if row and row[0]: _maybe_archive_planting(con, row[0]) def unarchive_entity(con, entity_type: str, entity_id: int) -> None: """Set is_archived=0 on a single entity. No cascades on unarchive — the user explicitly unarchives each thing they want active. (Auto-cascading unarchive would surprise users by reviving children they intended to keep archived.)""" table = ARCHIVABLE_ENTITIES.get(entity_type) if not table: return label = _entity_label_safe(con, entity_type, entity_id) con.execute(f"UPDATE {table} SET is_archived = 0 WHERE id = ?", (entity_id,)) log_unarchive(con, entity_type, entity_id, label) def _maybe_archive_planting(con, planting_id: int) -> None: """If every plant in this planting is archived (and there is at least one), archive the planting too. Called after any plant change that could leave the planting empty: archive, delete, status→removed.""" row = con.execute( """SELECT COUNT(*) AS total, SUM(CASE WHEN is_archived = 1 THEN 1 ELSE 0 END) AS archived FROM plants WHERE planting_id = ?""", (planting_id,), ).fetchone() if not row or not row[0]: return if row[0] == row[1]: # Only archive if not already archived (avoid redundant write + log). cur = con.execute( "SELECT is_archived FROM plantings WHERE id = ?", (planting_id,) ).fetchone() if cur and not cur[0]: label = _entity_label_safe(con, "planting", planting_id) con.execute( "UPDATE plantings SET is_archived = 1 WHERE id = ?", (planting_id,) ) log_archive(con, "planting", planting_id, label + " (auto: all plants archived)") def target_label(con, target_type: str, target_id: int, areas_by_id: Optional[dict] = None) -> str: """Short human label for any commentable entity. Used in 'also on: X' hints.""" if target_type == "area": if areas_by_id and target_id in areas_by_id: return area_path_name(target_id, areas_by_id) row = con.execute("SELECT name FROM areas WHERE id = ?", (target_id,)).fetchone() return row["name"] if row else f"area #{target_id}" if target_type == "species": row = con.execute("SELECT name FROM species WHERE id = ?", (target_id,)).fetchone() return row["name"] if row else f"species #{target_id}" if target_type == "variety": row = con.execute( """SELECT v.name, s.name AS species_name FROM varieties v LEFT JOIN species s ON s.id = v.species_id WHERE v.id = ?""", (target_id,), ).fetchone() if row: return f"{row['name']} ({row['species_name']})" if row["species_name"] else row["name"] return f"variety #{target_id}" if target_type == "planting": row = con.execute( """SELECT p.name, y.year AS year_value FROM plantings p LEFT JOIN years y ON y.id = p.year_id WHERE p.id = ?""", (target_id,), ).fetchone() if row: return f"{row['year_value']} — {row['name']}" if row["year_value"] else row["name"] return f"planting #{target_id}" if target_type == "year": row = con.execute("SELECT year FROM years WHERE id = ?", (target_id,)).fetchone() return str(row["year"]) if row else f"year #{target_id}" if target_type == "plant": row = con.execute( """SELECT p.quantity, s.name AS species_name, v.name AS variety_name FROM plants p LEFT JOIN species s ON s.id = p.species_id LEFT JOIN varieties v ON v.id = p.variety_id WHERE p.id = ?""", (target_id,), ).fetchone() if row: name = row["variety_name"] or row["species_name"] or f"plant #{target_id}" return f"{row['quantity']}× {name}" if row["quantity"] else name return f"plant #{target_id}" if target_type == "station": row = con.execute("SELECT name FROM watering_stations WHERE id = ?", (target_id,)).fetchone() return row["name"] if row else f"station #{target_id}" return f"{target_type} #{target_id}" def target_short_label(con, target_type: str, target_id: int, areas_by_id: Optional[dict] = None) -> str: """Short label for chips — for areas, leaf name (or short_name) only, not the full path.""" if target_type == "area": if areas_by_id and target_id in areas_by_id: a = areas_by_id[target_id] return a.get("short_name") or a["name"] row = con.execute("SELECT name, short_name FROM areas WHERE id = ?", (target_id,)).fetchone() if row: return row["short_name"] or row["name"] return f"area #{target_id}" return target_label(con, target_type, target_id, areas_by_id) def target_url(target_type: str, target_id: int) -> str: return { "area": f"/areas/{target_id}", "species": f"/species/{target_id}", "variety": f"/varieties/{target_id}", "planting": f"/plantings/{target_id}", "year": f"/years/{target_id}", "plant": f"/plants/{target_id}", "station": f"/stations/{target_id}", }.get(target_type, "/") def _attach_media_fields(c: dict) -> dict: """Split photo_paths into a list and ensure audio_path/photos exist on the dict so templates can render media uniformly across captures and comments.""" photos = [p for p in (c.get("photo_paths") or "").split("|") if p] c["photos"] = photos c.setdefault("audio_path", None) return c def load_comments( con, target_type: str, target_id: int, areas_by_id: Optional[dict] = None, include_archived: bool = False, ) -> list: """Load comments for a target via field_note_targets join. Each comment includes `other_targets` — list of {type, id, label, url} for targets other than this one.""" arch_clause = "" if include_archived else " AND c.is_archived = 0" rows = con.execute( f"""SELECT c.*, ct.is_primary FROM field_notes c JOIN field_note_targets ct ON ct.field_note_id = c.id WHERE ct.target_type = ? AND ct.target_id = ?{arch_clause} ORDER BY c.comment_date DESC, c.created_at DESC""", (target_type, target_id), ).fetchall() result = [] for r in rows: c = _attach_media_fields(dict(r)) others = con.execute( """SELECT target_type, target_id FROM field_note_targets WHERE field_note_id = ? AND NOT (target_type = ? AND target_id = ?)""", (c["id"], target_type, target_id), ).fetchall() c["other_targets"] = [ { "type": t["target_type"], "id": t["target_id"], "label": target_label(con, t["target_type"], t["target_id"], areas_by_id), "short_label": target_short_label(con, t["target_type"], t["target_id"], areas_by_id), "url": target_url(t["target_type"], t["target_id"]), } for t in others ] # Resolve primary target for display header prim = con.execute( """SELECT target_type, target_id FROM field_note_targets WHERE field_note_id = ? AND is_primary = 1""", (c["id"],), ).fetchone() if prim: c["primary_type"] = prim["target_type"] c["primary_id"] = prim["target_id"] c["primary_label"] = target_label(con, prim["target_type"], prim["target_id"], areas_by_id) c["primary_short_label"] = target_short_label(con, prim["target_type"], prim["target_id"], areas_by_id) c["primary_url"] = target_url(prim["target_type"], prim["target_id"]) else: # No primary in field_note_targets — orphan note (e.g. capture without # any target yet). Templates handle missing primary_label gracefully. c["primary_type"] = None c["primary_id"] = None c["primary_label"] = "" c["primary_short_label"] = "" c["primary_url"] = "" result.append(c) return result def collect_expansion_targets( con, target_type: str, target_id: int, include_children: bool = False, include_related: bool = False, by_id: Optional[dict] = None, ) -> list: """Return (target_type, target_id) tuples for expanded comment-thread lookup. Always includes the direct target. Children and related are gated by flags. Per entity: area children = descendant areas related = plantings/species/varieties/stations/plants in the area tree planting children = plants related = areas/species/varieties of this planting's plants species children = varieties related = plantings/areas/plants that use this species variety children = (none) related = parent species + plantings/areas/plants using this variety station children = (none) related = areas this station covers year children = (none) related = plantings in this year + their plants/areas/species/varieties plant children = (none) related = planting/area/species/variety attached to the plant """ direct: list = [(target_type, target_id)] extras: list = [] def _add(t: str, i): if i is not None: extras.append((t, int(i))) if target_type == "area": if include_children and by_id: for aid in get_descendant_ids(target_id, by_id): extras.append(("area", aid)) if include_related: area_ids = {target_id} if by_id: area_ids |= get_descendant_ids(target_id, by_id) if area_ids: ph = ",".join("?" * len(area_ids)) params = tuple(area_ids) for r in con.execute(f"SELECT DISTINCT planting_id FROM plants WHERE area_id IN ({ph}) AND planting_id IS NOT NULL", params).fetchall(): _add("planting", r[0]) for r in con.execute(f"SELECT DISTINCT species_id FROM plants WHERE area_id IN ({ph}) AND species_id IS NOT NULL", params).fetchall(): _add("species", r[0]) for r in con.execute(f"SELECT DISTINCT variety_id FROM plants WHERE area_id IN ({ph}) AND variety_id IS NOT NULL", params).fetchall(): _add("variety", r[0]) for r in con.execute(f"SELECT id FROM plants WHERE area_id IN ({ph})", params).fetchall(): _add("plant", r[0]) for r in con.execute(f"SELECT DISTINCT station_id FROM area_stations WHERE area_id IN ({ph})", params).fetchall(): _add("station", r[0]) elif target_type == "planting": if include_children: for r in con.execute("SELECT id FROM plants WHERE planting_id = ?", (target_id,)).fetchall(): _add("plant", r[0]) if include_related: for r in con.execute("SELECT DISTINCT area_id FROM plants WHERE planting_id = ? AND area_id IS NOT NULL", (target_id,)).fetchall(): _add("area", r[0]) for r in con.execute("SELECT DISTINCT species_id FROM plants WHERE planting_id = ? AND species_id IS NOT NULL", (target_id,)).fetchall(): _add("species", r[0]) for r in con.execute("SELECT DISTINCT variety_id FROM plants WHERE planting_id = ? AND variety_id IS NOT NULL", (target_id,)).fetchall(): _add("variety", r[0]) elif target_type == "species": if include_children: for r in con.execute("SELECT id FROM varieties WHERE species_id = ?", (target_id,)).fetchall(): _add("variety", r[0]) if include_related: for r in con.execute("SELECT DISTINCT planting_id FROM plants WHERE species_id = ? AND planting_id IS NOT NULL", (target_id,)).fetchall(): _add("planting", r[0]) for r in con.execute("SELECT DISTINCT area_id FROM plants WHERE species_id = ? AND area_id IS NOT NULL", (target_id,)).fetchall(): _add("area", r[0]) for r in con.execute("SELECT id FROM plants WHERE species_id = ?", (target_id,)).fetchall(): _add("plant", r[0]) elif target_type == "variety": if include_related: sp = con.execute("SELECT species_id FROM varieties WHERE id = ?", (target_id,)).fetchone() if sp and sp[0]: _add("species", sp[0]) for r in con.execute("SELECT DISTINCT planting_id FROM plants WHERE variety_id = ? AND planting_id IS NOT NULL", (target_id,)).fetchall(): _add("planting", r[0]) for r in con.execute("SELECT DISTINCT area_id FROM plants WHERE variety_id = ? AND area_id IS NOT NULL", (target_id,)).fetchall(): _add("area", r[0]) for r in con.execute("SELECT id FROM plants WHERE variety_id = ?", (target_id,)).fetchall(): _add("plant", r[0]) elif target_type == "station": if include_related: for r in con.execute("SELECT area_id FROM area_stations WHERE station_id = ?", (target_id,)).fetchall(): _add("area", r[0]) elif target_type == "year": if include_related: pids = [r[0] for r in con.execute("SELECT id FROM plantings WHERE year_id = ?", (target_id,)).fetchall()] for pid in pids: _add("planting", pid) if pids: ph = ",".join("?" * len(pids)) for r in con.execute(f"SELECT id, species_id, variety_id, area_id FROM plants WHERE planting_id IN ({ph})", tuple(pids)).fetchall(): _add("plant", r[0]) _add("species", r[1]) _add("variety", r[2]) _add("area", r[3]) elif target_type == "plant": if include_related: row = con.execute("SELECT planting_id, area_id, species_id, variety_id FROM plants WHERE id = ?", (target_id,)).fetchone() if row: _add("planting", row[0]) _add("area", row[1]) _add("species", row[2]) _add("variety", row[3]) seen = set() out = [] for t in direct + extras: if t not in seen: seen.add(t) out.append(t) return out def load_comments_expanded( con, target_type: str, target_id: int, include_children: bool = False, include_related: bool = False, by_id: Optional[dict] = None, primary_only_for_direct: bool = False, include_archived: bool = False, ) -> list: """Same shape as load_comments but expanded across children/related entities. `primary_only_for_direct` restricts matches against the direct target to is_primary=1 (used for /years/* so the year page defaults to 'comments posted directly about the year'). Children/related matches are never restricted to primary — they're already indirect. """ targets = collect_expansion_targets(con, target_type, target_id, include_children, include_related, by_id) if not targets: return [] # Find matching comment IDs. Handle the primary-only restriction for the direct target separately. comment_ids: set = set() direct_target = (target_type, target_id) indirect_targets = [t for t in targets if t != direct_target] if primary_only_for_direct: for r in con.execute( "SELECT field_note_id FROM field_note_targets WHERE target_type = ? AND target_id = ? AND is_primary = 1", (target_type, target_id), ).fetchall(): comment_ids.add(r[0]) else: for r in con.execute( "SELECT field_note_id FROM field_note_targets WHERE target_type = ? AND target_id = ?", (target_type, target_id), ).fetchall(): comment_ids.add(r[0]) if indirect_targets: # Chunk to avoid hitting SQLite's 999-param limit. CHUNK = 100 for i in range(0, len(indirect_targets), CHUNK): chunk = indirect_targets[i : i + CHUNK] where = " OR ".join(["(target_type = ? AND target_id = ?)"] * len(chunk)) params: list = [] for t, tid in chunk: params += [t, tid] for r in con.execute( f"SELECT DISTINCT field_note_id FROM field_note_targets WHERE {where}", tuple(params), ).fetchall(): comment_ids.add(r[0]) if not comment_ids: return [] id_list = list(comment_ids) # Chunked fetch rows = [] CHUNK = 400 arch_clause = "" if include_archived else " AND is_archived = 0" for i in range(0, len(id_list), CHUNK): chunk = id_list[i : i + CHUNK] ph = ",".join("?" * len(chunk)) rows.extend( con.execute( f"SELECT * FROM field_notes WHERE id IN ({ph}){arch_clause} " f"ORDER BY comment_date DESC, created_at DESC", tuple(chunk), ).fetchall() ) rows.sort(key=lambda r: (r["comment_date"], r["created_at"]), reverse=True) result = [] for r in rows: c = _attach_media_fields(dict(r)) others = con.execute( """SELECT target_type, target_id, is_primary FROM field_note_targets WHERE field_note_id = ? ORDER BY is_primary DESC, id""", (c["id"],), ).fetchall() prim = next((o for o in others if o["is_primary"]), None) sec = [o for o in others if not o["is_primary"]] if prim: c["primary_type"] = prim["target_type"] c["primary_id"] = prim["target_id"] c["primary_label"] = target_label(con, prim["target_type"], prim["target_id"], by_id) c["primary_short_label"] = target_short_label(con, prim["target_type"], prim["target_id"], by_id) c["primary_url"] = target_url(prim["target_type"], prim["target_id"]) else: c["primary_type"] = None c["primary_id"] = None c["primary_label"] = "" c["primary_short_label"] = "" c["primary_url"] = "" c["other_targets"] = [ { "type": t["target_type"], "id": t["target_id"], "label": target_label(con, t["target_type"], t["target_id"], by_id), "short_label": target_short_label(con, t["target_type"], t["target_id"], by_id), "url": target_url(t["target_type"], t["target_id"]), } for t in sec if not (t["target_type"] == c["primary_type"] and t["target_id"] == c["primary_id"]) ] c["is_primary"] = (c["primary_type"] == target_type and c["primary_id"] == target_id) result.append(c) return result def load_comment_targets(con, comment_id: int, areas_by_id: Optional[dict] = None) -> list: """All targets for a comment, with labels and URLs. Used on edit page.""" rows = con.execute( """SELECT target_type, target_id, is_primary FROM field_note_targets WHERE field_note_id = ? ORDER BY is_primary DESC, id""", (comment_id,), ).fetchall() return [ { "type": r["target_type"], "id": r["target_id"], "is_primary": bool(r["is_primary"]), "label": target_label(con, r["target_type"], r["target_id"], areas_by_id), "url": target_url(r["target_type"], r["target_id"]), } for r in rows ] # Entity types that have an overview (notes column). Maps to (table_name, column_name). OVERVIEW_ENTITIES = { "area": ("areas", "notes"), "planting": ("plantings", "notes"), "plant": ("plants", "notes"), "species": ("species", "description"), "variety": ("varieties", "description"), "year": ("years", "notes"), } def get_current_overview(con, entity_type: str, entity_id: int) -> Optional[str]: """Return the current overview text for an entity, or None if entity not found.""" if entity_type not in OVERVIEW_ENTITIES: return None table, col = OVERVIEW_ENTITIES[entity_type] row = con.execute( f"SELECT {col} FROM {table} WHERE id = ?", (entity_id,) ).fetchone() if not row: return None return row[col] or "" def snapshot_overview( con, entity_type: str, entity_id: int, new_content: str, author: str = "user", change_note: str = "", ): """If the current overview differs from new_content, snapshot the current value into overview_versions before the caller writes the new content. Idempotent: returns silently if entity is unknown or content is unchanged.""" if entity_type not in OVERVIEW_ENTITIES: return current = get_current_overview(con, entity_type, entity_id) if current is None: return # entity not found if (current or "") == (new_content or ""): return # no change con.execute( """INSERT INTO overview_versions (entity_type, entity_id, content, created_at, author, change_note) VALUES (?, ?, ?, ?, ?, ?)""", (entity_type, entity_id, current or "", now(), author, change_note), ) def load_overview_versions(con, entity_type: str, entity_id: int) -> list: return [ dict(r) for r in con.execute( """SELECT * FROM overview_versions WHERE entity_type = ? AND entity_id = ? ORDER BY created_at DESC, id DESC""", (entity_type, entity_id), ).fetchall() ] def _ancestor_ids(area_id: int, by_id: dict) -> list: """Return [root, ..., parent, self] area ids, or empty list if unknown.""" chain = get_ancestors(area_id, by_id) return [a["id"] for a in chain] def lca_area_ids(area_ids: list, by_id: dict) -> list: """Given a set of area ids, return the ancestor chain of their lowest common ancestor (inclusive). If one area, returns that area's full chain. Used to cache 'this comment belongs to the LCA and everything above it'.""" area_ids = [a for a in area_ids if a in by_id] if not area_ids: return [] chains = [_ancestor_ids(a, by_id) for a in area_ids] common = [] for items in zip(*chains): if len(set(items)) == 1: common.append(items[0]) else: break return common def stations_for_area_set(con, area_ids: list) -> set: """Return all station ids attached to any area in this set or any descendant. Used for area → station inference (single station case).""" if not area_ids: return set() placeholders = ",".join("?" * len(area_ids)) rows = con.execute( f"SELECT DISTINCT station_id FROM area_stations WHERE area_id IN ({placeholders})", area_ids, ).fetchall() return {r["station_id"] for r in rows} def inferred_targets_for( con, target_type: str, target_id: int, comment_date: Optional[str] = None ) -> list: """Return inherited targets for a comment on a primary entity. Cached at comment-creation time; represents state at time of posting. Rules: - year: derived from comment_date (not from the entity). Year row created if missing. - plant: species, variety, area + area ancestors, planting. - planting: year; if all plants same variety → variety (→ species); else if all plants same species → species; area LCA + ancestors; if a single station covers LCA → station. - area: area ancestors; if all plants here+descendants same variety/species → that; if single station covers this area → station. - species: all plantings using this species (if only one → that planting); common area (LCA + ancestors) across plants; stations feeding those areas. - variety: species; plus species-style inference scoped to this variety. - station: areas it covers (LCA + ancestors); all species planted in covered areas. """ inferred: list = [] area_rows = load_areas(con) _, by_id = build_tree(area_rows) # Year: always derived from comment_date when provided if comment_date: try: year_val = int(comment_date[:4]) year_id = get_or_create_year(con, year_val) inferred.append(("year", year_id)) except (ValueError, TypeError): pass # Inference uses only active rows: archived plants/varieties/species/areas/ # stations should not become inferred targets. The post-filter at the bottom # drops any archived (type, id) pairs that slip through; the per-branch # queries below also filter `plants.is_archived = 0` so that subtree/aggregate # queries don't surface ghost data from an archived plant group. if target_type == "plant": # Rules (owner spec, 2026-04-20): # - Variety if present, else species (one or the other, never both). # - Direct area only, no ancestors. # - All stations feeding the area (coverage cascades through # ancestors inside load_stations_for_area). # - Year from comment_date (appended above). # - Planting not auto-linked. row = con.execute( "SELECT species_id, variety_id, area_id FROM plants WHERE id = ?", (target_id,), ).fetchone() if not row: return [] if row["variety_id"]: inferred.append(("variety", row["variety_id"])) elif row["species_id"]: inferred.append(("species", row["species_id"])) if row["area_id"]: inferred.append(("area", row["area_id"])) for s in load_stations_for_area(con, row["area_id"], by_id): inferred.append(("station", s["id"])) elif target_type == "planting": # Rules (per owner, 2026-04-20 v2): # - Every plant group in the planting is a target. # - Variety/species selection: # * multi-species planting → per species: if that species has # exactly one variety across its plants, link the variety; # otherwise link the species. # * otherwise, if exactly one distinct variety across the # whole planting → link that variety. # * otherwise (many varieties but one species) → link that species. # - Every direct area a plant lives in is a target (no ancestors). # - Every watering station feeding any of those areas is a target # (station coverage already cascades from ancestor areas inside # load_stations_for_area). # - Year derived from comment_date (already appended above). plants = con.execute( """SELECT id, species_id, variety_id, area_id FROM plants WHERE planting_id = ? AND is_archived = 0""", (target_id,), ).fetchall() # All plant groups in the planting for p in plants: inferred.append(("plant", p["id"])) varieties = {p["variety_id"] for p in plants if p["variety_id"]} species_ids = {p["species_id"] for p in plants if p["species_id"]} if len(species_ids) > 1: for sp_id in species_ids: sp_vars = { p["variety_id"] for p in plants if p["species_id"] == sp_id and p["variety_id"] } if len(sp_vars) == 1: inferred.append(("variety", next(iter(sp_vars)))) else: inferred.append(("species", sp_id)) elif len(varieties) == 1: inferred.append(("variety", next(iter(varieties)))) elif len(species_ids) == 1: inferred.append(("species", next(iter(species_ids)))) direct_area_ids = {p["area_id"] for p in plants if p["area_id"]} for aid in direct_area_ids: inferred.append(("area", aid)) for aid in direct_area_ids: for s in load_stations_for_area(con, aid, by_id): inferred.append(("station", s["id"])) elif target_type == "area": # Rules (owner spec, 2026-04-20 v3): # Computed over the target area's full subtree (self + descendants): # P = plant-group count, S = species count, V = variety count. # # - Plant groups: if 0 < P <= PLANT_CAP, link every plant group in # the subtree; else none. Plant groups are preferred over plantings # because plantings can split; plant IDs travel with the plant. # - Variety / species (pick at most one approach): # V == 1 → link that variety. # S == 1 → link that species. # S <= SPECIES_CAP → per species: if that species has # exactly one variety in the subtree, # link that variety; else link the # species. # otherwise → nothing (area too broad). # - Station: exactly one station serves the area (directly or # inherited from an ancestor) → link it; else none. # - Year: from comment_date (appended above). # - No area ancestors and no planting links. PLANT_CAP = 10 SPECIES_CAP = 8 full_subtree = get_descendant_ids(target_id, by_id) | {target_id} placeholders = ",".join("?" * len(full_subtree)) subtree_plants = con.execute( f"SELECT id, species_id, variety_id FROM plants " f"WHERE area_id IN ({placeholders}) AND is_archived = 0", list(full_subtree), ).fetchall() if 0 < len(subtree_plants) <= PLANT_CAP: for p in subtree_plants: inferred.append(("plant", p["id"])) sub_varieties = {p["variety_id"] for p in subtree_plants if p["variety_id"]} sub_species = {p["species_id"] for p in subtree_plants if p["species_id"]} if len(sub_varieties) == 1: inferred.append(("variety", next(iter(sub_varieties)))) elif len(sub_species) == 1: inferred.append(("species", next(iter(sub_species)))) elif 0 < len(sub_species) <= SPECIES_CAP: for sp_id in sub_species: sp_vars = { p["variety_id"] for p in subtree_plants if p["species_id"] == sp_id and p["variety_id"] } if len(sp_vars) == 1: inferred.append(("variety", next(iter(sp_vars)))) else: inferred.append(("species", sp_id)) stations = load_stations_for_area(con, target_id, by_id) if len(stations) == 1: inferred.append(("station", stations[0]["id"])) elif target_type == "species": plantings = { r["planting_id"] for r in con.execute( "SELECT DISTINCT planting_id FROM plants " "WHERE species_id = ? AND is_archived = 0", (target_id,), ).fetchall() } if len(plantings) == 1: inferred.append(("planting", next(iter(plantings)))) plant_areas = [ r["area_id"] for r in con.execute( "SELECT area_id FROM plants " "WHERE species_id = ? AND area_id IS NOT NULL AND is_archived = 0", (target_id,), ).fetchall() ] if plant_areas: for aid in lca_area_ids(plant_areas, by_id): inferred.append(("area", aid)) stations = stations_for_area_set(con, list(set(plant_areas))) for sid in stations: inferred.append(("station", sid)) elif target_type == "variety": vrow = con.execute( "SELECT species_id FROM varieties WHERE id = ?", (target_id,) ).fetchone() if vrow and vrow["species_id"]: inferred.append(("species", vrow["species_id"])) plantings = { r["planting_id"] for r in con.execute( "SELECT DISTINCT planting_id FROM plants " "WHERE variety_id = ? AND is_archived = 0", (target_id,), ).fetchall() } if len(plantings) == 1: inferred.append(("planting", next(iter(plantings)))) plant_areas = [ r["area_id"] for r in con.execute( "SELECT area_id FROM plants " "WHERE variety_id = ? AND area_id IS NOT NULL AND is_archived = 0", (target_id,), ).fetchall() ] if plant_areas: for aid in lca_area_ids(plant_areas, by_id): inferred.append(("area", aid)) stations = stations_for_area_set(con, list(set(plant_areas))) for sid in stations: inferred.append(("station", sid)) elif target_type == "station": # Rules (owner spec, 2026-04-20): # - Each area this station is directly attached to. # - Every plant group in the served subtree (attached areas + all # descendants, since station coverage cascades down). # - Year from comment_date (appended above). attached = [ r["area_id"] for r in con.execute( "SELECT area_id FROM area_stations WHERE station_id = ?", (target_id,) ).fetchall() ] for aid in attached: inferred.append(("area", aid)) if attached: served = set(attached) for a in attached: served |= get_descendant_ids(a, by_id) placeholders = ",".join("?" * len(served)) for p in con.execute( f"SELECT id FROM plants " f"WHERE area_id IN ({placeholders}) AND is_archived = 0", list(served), ).fetchall(): inferred.append(("plant", p["id"])) # de-dupe + drop self seen = set() out = [] for t, i in inferred: if (t, i) == (target_type, target_id) or (t, i) in seen: continue seen.add((t, i)) out.append((t, i)) # Drop any archived entities — inference should never auto-link a soft-deleted # row. (The primary target itself is set elsewhere and is allowed to be # archived: posting a comment ON an archived entity is a valid action.) if not out: return out table_for = { "area": "areas", "planting": "plantings", "plant": "plants", "species": "species", "variety": "varieties", "station": "watering_stations", "year": "years", } by_table: dict = {} for t, i in out: tbl = table_for.get(t) if tbl: by_table.setdefault(tbl, set()).add(i) archived: set = set() for tbl, ids in by_table.items(): ph = ",".join("?" * len(ids)) for r in con.execute( f"SELECT id FROM {tbl} WHERE id IN ({ph}) AND is_archived = 1", tuple(ids), ).fetchall(): # Map back to (type, id) — invert table_for lookup. for t2, tbl2 in table_for.items(): if tbl2 == tbl: archived.add((t2, r[0])) break return [pair for pair in out if pair not in archived] def add_inferred_targets( con, comment_id: int, target_type: str, target_id: int, comment_date: Optional[str] = None ): """Insert inherited secondary targets for a newly created comment. Idempotent.""" for t_type, t_id in inferred_targets_for(con, target_type, target_id, comment_date): con.execute( """INSERT OR IGNORE INTO field_note_targets (field_note_id, target_type, target_id, is_primary) VALUES (?, ?, ?, 0)""", (comment_id, t_type, t_id), ) def post_comment( con, target_type: str, target_id: int, comment_date: str, body: str, kind: str = "", created_at: Optional[str] = None, audio_path: Optional[str] = None, photo_paths: str = "", transcript: str = "", captured_at: Optional[str] = None, explicit_secondaries: Optional[list] = None, ) -> int: """Standard entry point for creating a comment — inserts the field_note row, attaches the primary target, and applies the inferred-target rules. Optional media (audio, photos, transcript, captured_at) lets the comment form attach the same kinds of media as the /capture page — comments and captures share the unified field_notes table. `explicit_secondaries`: when provided (a list of (type, id) tuples), use that list as-is for secondaries and skip auto-inference. When None (default), the algorithmic inference runs. The floating capture modal sets this when the user has edited the inferred-targets list pre-submit. Use this from HTTP handlers AND from any script/backfill that creates comments, so inference always runs. Returns the new comment id. """ if kind and kind not in COMMENT_KINDS: kind = "" ts = created_at or now() cur = con.execute( """INSERT INTO field_notes (comment_date, body, kind, created_at, status, processed_at, audio_path, photo_paths, transcript, captured_at) VALUES (?, ?, ?, ?, 'processed', ?, ?, ?, ?, ?)""", (comment_date, body, kind, ts, ts, audio_path, photo_paths or '', transcript or '', captured_at), ) comment_id = cur.lastrowid con.execute( """INSERT INTO field_note_targets (field_note_id, target_type, target_id, is_primary) VALUES (?, ?, ?, 1)""", (comment_id, target_type, target_id), ) if explicit_secondaries is not None: for t_type, t_id in explicit_secondaries: if (t_type, t_id) == (target_type, target_id): continue # don't double-insert the primary con.execute( """INSERT OR IGNORE INTO field_note_targets (field_note_id, target_type, target_id, is_primary) VALUES (?, ?, ?, 0)""", (comment_id, t_type, t_id), ) else: add_inferred_targets(con, comment_id, target_type, target_id, comment_date) # Activity log: one row per comment posted. Summary mentions the primary # target so the detailed log says "Posted note on area: FG — short body". primary_label = _entity_label_safe(con, target_type, target_id) body_short = _short(body or transcript or "(media-only)", 60) log_activity( con, "add", "comment", comment_id, f"Posted note on {target_type}: {primary_label} — {body_short}", ) return comment_id def delete_target_refs(con, target_type: str, target_id: int): """Remove all field_note_targets pointing at this entity. Delete orphaned comments (no remaining targets). Re-point primary if lost.""" affected = [ r[0] for r in con.execute( "SELECT field_note_id FROM field_note_targets WHERE target_type = ? AND target_id = ?", (target_type, target_id), ).fetchall() ] con.execute( "DELETE FROM field_note_targets WHERE target_type = ? AND target_id = ?", (target_type, target_id), ) for cid in affected: remaining = con.execute( """SELECT target_type, target_id, is_primary FROM field_note_targets WHERE field_note_id = ? ORDER BY is_primary DESC, id""", (cid,), ).fetchall() if not remaining: con.execute("DELETE FROM field_notes WHERE id = ?", (cid,)) continue if not any(r["is_primary"] for r in remaining): np = remaining[0] con.execute( """UPDATE field_note_targets SET is_primary = 1 WHERE field_note_id = ? AND target_type = ? AND target_id = ?""", (cid, np["target_type"], np["target_id"]), ) # (No legacy target_type/target_id columns on field_notes; primary is # carried entirely by field_note_targets.is_primary now.) def load_species(con, include_archived: bool = False) -> list: where = "" if include_archived else "WHERE is_archived = 0" return [ dict(r) for r in con.execute( f"SELECT * FROM species {where} ORDER BY name" ).fetchall() ] def load_varieties(con, include_archived: bool = False) -> list: where = "" if include_archived else "WHERE v.is_archived = 0" return [ dict(r) for r in con.execute( f"""SELECT v.*, s.name AS species_name, s.type AS species_type, s.primary_function AS species_primary_function FROM varieties v LEFT JOIN species s ON s.id = v.species_id {where} ORDER BY v.name""" ).fetchall() ] def load_years(con, include_archived: bool = False) -> list: where = "" if include_archived else "WHERE is_archived = 0" return [ dict(r) for r in con.execute( f"SELECT * FROM years {where} ORDER BY year DESC" ).fetchall() ] def load_plantings(con, include_archived: bool = False) -> list: where = "" if include_archived else "WHERE p.is_archived = 0" rows = con.execute( f"""SELECT p.*, y.year AS year_value FROM plantings p LEFT JOIN years y ON y.id = p.year_id {where} ORDER BY y.year DESC, p.name""" ).fetchall() out = [] for r in rows: d = dict(r) if not d.get("status"): d["status"] = "planted" out.append(d) return out def load_plants_for_planting( con, planting_id: int, include_archived: bool = False ) -> list: arch_clause = "" if include_archived else " AND p.is_archived = 0" return [ dict(r) for r in con.execute( f"""SELECT p.*, s.name AS species_name, v.name AS variety_name, a.name AS area_name FROM plants p LEFT JOIN species s ON s.id = p.species_id LEFT JOIN varieties v ON v.id = p.variety_id LEFT JOIN areas a ON a.id = p.area_id WHERE p.planting_id = ?{arch_clause} ORDER BY p.id""", (planting_id,), ).fetchall() ] def plant_display_name(plant: dict) -> str: parts = [] if plant.get("variety_name"): parts.append(plant["variety_name"]) if plant.get("species_name"): parts.append(plant["species_name"]) return " ".join(parts) if parts else f"Plant #{plant['id']}" def planting_total_qty( con, planting_id: int, include_archived: bool = False ) -> tuple: """Return (group_count, total_quantity) for a planting.""" arch = "" if include_archived else " AND is_archived = 0" rows = con.execute( f"SELECT quantity FROM plants WHERE planting_id = ?{arch}", (planting_id,), ).fetchall() groups = len(rows) total = 0 for r in rows: try: total += int(r[0]) except (ValueError, TypeError): total += 1 # count as 1 if quantity is not a number return groups, total def species_options(species: list) -> list: return [(s["id"], s["name"]) for s in species] def variety_options(varieties: list) -> list: return [ (v["id"], (v["species_name"] + " — " if v["species_name"] else "") + v["name"]) for v in varieties ] def load_stations(con, include_archived: bool = False) -> list: where = "" if include_archived else "WHERE is_archived = 0" return [ dict(r) for r in con.execute( f"SELECT * FROM watering_stations {where} ORDER BY name" ).fetchall() ] def load_stations_for_area( con, area_id: int, by_id: dict, include_archived: bool = False ) -> list: """Get stations explicitly assigned to this area. Assignment is fully explicit: when the user checks a parent area in the station form, the picker UI auto-checks every descendant, and each is saved individually. So a station only appears here if it's directly assigned to this exact area_id. """ arch_clause = "" if include_archived else " AND s.is_archived = 0" return [ dict(r) for r in con.execute( f"""SELECT s.* FROM watering_stations s JOIN area_stations asx ON asx.station_id = s.id WHERE asx.area_id = ?{arch_clause} ORDER BY s.name""", (area_id,), ).fetchall() ] def load_areas_for_station( con, station_id: int, include_archived: bool = False ) -> list: arch_clause = "" if include_archived else " AND a.is_archived = 0" return [ dict(r) for r in con.execute( f"""SELECT a.* FROM areas a JOIN area_stations asx ON asx.area_id = a.id WHERE asx.station_id = ?{arch_clause}""", (station_id,), ).fetchall() ] def load_supplies(con, include_archived: bool = False) -> list: where = "" if include_archived else "WHERE is_archived = 0" return [ dict(r) for r in con.execute( f"SELECT * FROM supplies {where} ORDER BY category, name" ).fetchall() ] def get_or_create_year(con, year_value: int) -> int: """Get or create a year record, return its id.""" row = con.execute("SELECT id FROM years WHERE year = ?", (year_value,)).fetchone() if row: return row[0] cur = con.execute( "INSERT INTO years (year, notes, created_at) VALUES (?, '', ?)", (year_value, now()), ) return cur.lastrowid # --- AI chat (Gemini function-calling) ----------------------------------- # # Architecture: every user turn carries a small "starter context" describing # the current garden state, plus a set of tools the model can call to fetch # more detail when it needs it. The Gemini wrapper loops: # model_call -> if function_call: dispatch tool -> model_call ... # until plain text is returned. Tool calls are recorded on the assistant # chat_messages row so the replay view can show what the model looked at. def build_starter_context(con) -> str: """Compact garden snapshot prepended to every chat turn. Targeted at ~3-5K tokens — current date + season, area names + dims, species + type/ function, what's actually planted right now (by species + by planting), and the last 14 days of one-line note summaries. Anything deeper happens through tools.""" today = datetime.utcnow() cutoff = (today - timedelta(days=14)).date().isoformat() today_iso = today.date().isoformat() # A short hemisphere/season hint based on month — northern-hemisphere # default. Helps the model anchor seasonal questions ("now" = which month). month = today.month season = ( "winter (dormant)" if month in (12, 1, 2) else "early spring (cool-season planting)" if month == 3 else "spring (last frost / transplant season)" if month in (4, 5) else "early summer (warm-season established)" if month == 6 else "summer (peak growth, harvest, fertilize heavy feeders)" if month in (7, 8) else "early fall (cool-crop second planting / harvest preserving)" if month == 9 else "fall (winding down, plant garlic/cover crops)" if month in (10, 11) else "unknown" ) parts: list = [] parts.append(f"Today: {today_iso} — {season} (northern hemisphere assumption)") # Areas: id, full path, dims. Keep the path compact. areas = load_areas(con) _, by_id = build_tree(areas) parts.append("\n## Areas") for a in sorted(areas, key=lambda x: area_path_name(x["id"], by_id).lower()): path = area_path_name(a["id"], by_id) L, W = a.get("length_ft") or 0, a.get("width_ft") or 0 dims = f" {L:g}×{W:g}ft" if (L and W) else "" sn = f" [{a['short_name']}]" if a.get("short_name") else "" parts.append(f"- #{a['id']} {path}{sn}{dims}") # Species CATALOGUE: id, name, type/function. NOT the same as what's # currently planted — that's the next section. Animals + non-edibles are # the interesting cases for the model so they need to be visible. parts.append("\n## Species catalogue (types defined; not what's currently planted)") sp_rows = con.execute( """SELECT id, name, common_name, type, primary_function FROM species WHERE is_archived = 0 ORDER BY name""" ).fetchall() for s in sp_rows: cn = f" ({s['common_name']})" if s["common_name"] else "" parts.append( f"- #{s['id']} {s['name']}{cn} — {s['type']}/{s['primary_function']}" ) # WHAT'S ACTUALLY PLANTED. Aggregated by species so the model knows what # fruit/vegetable/etc. is in the ground right now without having to call # tools. Sums plant.quantity over all plant_groups with status='planted'. parts.append("\n## Currently planted (status=planted, by species)") planted_rows = con.execute( """SELECT s.name AS species_name, s.type AS species_type, COUNT(DISTINCT pl.id) AS group_count, GROUP_CONCAT(DISTINCT v.name) AS varieties, SUM(CAST(NULLIF(pl.quantity, '') AS INTEGER)) AS total_qty FROM plants pl LEFT JOIN species s ON s.id = pl.species_id LEFT JOIN varieties v ON v.id = pl.variety_id WHERE pl.is_archived = 0 AND pl.status = 'planted' GROUP BY pl.species_id ORDER BY total_qty DESC, s.name""" ).fetchall() if planted_rows: for r in planted_rows: sp = r["species_name"] or "(unknown species)" qty = r["total_qty"] or "?" grps = r["group_count"] varieties = r["varieties"] or "" varieties_short = "" if varieties: vs = [v for v in varieties.split(",") if v][:4] more = max(0, len([v for v in varieties.split(",") if v]) - len(vs)) varieties_short = f" — varieties: {', '.join(vs)}" + (f" +{more} more" if more else "") parts.append(f"- {sp}: {qty} plants in {grps} group{'s' if grps != 1 else ''}{varieties_short}") else: parts.append("- (nothing currently planted; check Unplanted/Ideas pools below)") # UNPLANTED + IDEA pools. Same aggregation but for staging plants. pool_rows = con.execute( """SELECT pl.status AS pst, s.name AS species_name, COUNT(DISTINCT pl.id) AS group_count, SUM(CAST(NULLIF(pl.quantity, '') AS INTEGER)) AS total_qty FROM plants pl LEFT JOIN species s ON s.id = pl.species_id WHERE pl.is_archived = 0 AND pl.status IN ('unplanted', 'idea') GROUP BY pl.status, pl.species_id ORDER BY pl.status, total_qty DESC""" ).fetchall() if pool_rows: parts.append("\n## Staged (not yet planted)") for r in pool_rows: sp = r["species_name"] or "(unknown)" parts.append(f"- {r['pst']}: {sp} — {r['total_qty'] or '?'} plants ({r['group_count']} groups)") # Plantings: id, year, name, status, plant count + species breakdown. parts.append("\n## Plantings (each represents one ground-planting event)") pl_rows = con.execute( """SELECT p.id, p.name, p.status, y.year AS year_value, COUNT(pt.id) AS plant_count, GROUP_CONCAT(DISTINCT s.name) AS species_list FROM plantings p LEFT JOIN years y ON y.id = p.year_id LEFT JOIN plants pt ON pt.planting_id = p.id AND pt.is_archived = 0 LEFT JOIN species s ON s.id = pt.species_id WHERE p.is_archived = 0 GROUP BY p.id ORDER BY y.year DESC, p.name""" ).fetchall() for p in pl_rows: yr = f"{p['year_value']} — " if p["year_value"] else "" species_str = "" if p["species_list"]: sp_list = sorted({s for s in p["species_list"].split(",") if s}) shown = sp_list[:5] extra = len(sp_list) - len(shown) species_str = f" [{', '.join(shown)}" + (f" +{extra} more]" if extra else "]") parts.append( f"- #{p['id']} {yr}{p['name']} ({p['status'] or 'planted'}, " f"{p['plant_count']} plants){species_str}" ) # Recent CARE ACTIONS — done-tagged notes from the last 60 days, with # longer bodies so the model sees what was actually done. This is the # critical context for "should I fertilize / water / treat?" questions: # without it the model will recommend things the user already did. care_cutoff = (today - timedelta(days=60)).date().isoformat() parts.append( "\n## Recent care actions (done-tagged notes, last 60 days)\n" "Use this to know what the user has ALREADY done. Don't recommend " "actions that have already been taken recently." ) care_rows = con.execute( """SELECT c.id, c.comment_date, c.body, ct.target_type, ct.target_id FROM field_notes c LEFT JOIN field_note_targets ct ON ct.field_note_id = c.id AND ct.is_primary = 1 WHERE c.is_archived = 0 AND c.kind = 'done' AND c.body IS NOT NULL AND c.body != '' AND c.comment_date >= ? ORDER BY c.comment_date DESC, c.id DESC LIMIT 80""", (care_cutoff,), ).fetchall() if care_rows: for n in care_rows: try: tgt = ( target_short_label(con, n["target_type"], n["target_id"], by_id) if n["target_type"] else "?" ) except Exception: tgt = "?" # Longer bodies (300 chars) so dosage/coverage detail survives. body = _short(n["body"], 300) parts.append( f"- {n['comment_date']} on {n['target_type'] or '?'}:{tgt}: {body}" ) else: parts.append("- (no care actions logged in the last 60 days)") # Recent ISSUES — same window, problems the user flagged. Helps the # model give grounded advice ("you mentioned aphids on the cherries…"). issue_rows = con.execute( """SELECT c.id, c.comment_date, c.body, ct.target_type, ct.target_id FROM field_notes c LEFT JOIN field_note_targets ct ON ct.field_note_id = c.id AND ct.is_primary = 1 WHERE c.is_archived = 0 AND c.kind = 'issue' AND c.body IS NOT NULL AND c.body != '' AND c.comment_date >= ? ORDER BY c.comment_date DESC, c.id DESC LIMIT 30""", (care_cutoff,), ).fetchall() if issue_rows: parts.append("\n## Open / recent issues (issue-tagged notes, last 60 days)") for n in issue_rows: try: tgt = ( target_short_label(con, n["target_type"], n["target_id"], by_id) if n["target_type"] else "?" ) except Exception: tgt = "?" body = _short(n["body"], 300) parts.append( f"- {n['comment_date']} on {n['target_type'] or '?'}:{tgt}: {body}" ) # Other recent notes: observations, questions, ideas, generic notes. # Wider net than the care/issue sections so the model sees ambient # context. 30-day window — older than that, use search_notes. parts.append("\n## Other recent notes (observations/questions/ideas, last 30 days)") note_cutoff_30 = (today - timedelta(days=30)).date().isoformat() note_rows = con.execute( """SELECT c.id, c.comment_date, c.body, c.kind, ct.target_type, ct.target_id FROM field_notes c LEFT JOIN field_note_targets ct ON ct.field_note_id = c.id AND ct.is_primary = 1 WHERE c.is_archived = 0 AND c.body IS NOT NULL AND c.body != '' AND c.comment_date >= ? AND COALESCE(c.kind, '') NOT IN ('done', 'issue') ORDER BY c.comment_date DESC, c.id DESC LIMIT 40""", (note_cutoff_30,), ).fetchall() if note_rows: for n in note_rows: try: tgt = ( target_short_label(con, n["target_type"], n["target_id"], by_id) if n["target_type"] else "?" ) except Exception: tgt = "?" kind = f" [{n['kind']}]" if n["kind"] else "" body = _short(n["body"], 180) parts.append( f"- {n['comment_date']} on {n['target_type'] or '?'}:{tgt}{kind}: {body}" ) else: parts.append("- (no other notes in the last 30 days)") return "\n".join(parts) # --- Chat tools ----------------------------------------------------------- # # Each tool is a Python callable that takes a sqlite connection plus # kwargs from the model, returns a JSON-serializable dict. The schema # entries are passed verbatim to Gemini's `tools` parameter so the model # knows what's available. Keep tool output compact — tokens still cost # money even if cheap, and large blobs make the model lose focus. def _chat_truncate_text(s: Optional[str], limit: int = 600) -> str: if not s: return "" s = str(s) return s if len(s) <= limit else s[:limit - 1] + "…" def _tool_get_planting_detail(con, *, planting_id: int) -> dict: row = con.execute( """SELECT p.*, y.year AS year_value FROM plantings p LEFT JOIN years y ON y.id = p.year_id WHERE p.id = ?""", (planting_id,), ).fetchone() if not row: return {"error": f"planting #{planting_id} not found"} plants = con.execute( """SELECT pl.id, pl.quantity, pl.status, pl.area_id, pl.notes, s.name AS species, v.name AS variety FROM plants pl LEFT JOIN species s ON s.id = pl.species_id LEFT JOIN varieties v ON v.id = pl.variety_id WHERE pl.planting_id = ? AND pl.is_archived = 0 ORDER BY pl.id""", (planting_id,), ).fetchall() areas = load_areas(con) _, by_id = build_tree(areas) return { "id": row["id"], "name": row["name"], "year": row["year_value"], "status": row["status"], "source": row["source"] or "", "notes": _chat_truncate_text(row["notes"], 1500), "plants": [ { "id": p["id"], "qty": p["quantity"], "status": p["status"] or "", "species": p["species"], "variety": p["variety"], "area_id": p["area_id"], "area_path": ( area_path_name(p["area_id"], by_id) if p["area_id"] else None ), "notes": _chat_truncate_text(p["notes"], 300), } for p in plants ], } def _tool_get_area_detail(con, *, area_id: int) -> dict: areas = load_areas(con) _, by_id = build_tree(areas) if area_id not in by_id: return {"error": f"area #{area_id} not found"} a = by_id[area_id] desc_ids = get_descendant_ids(area_id, by_id) | {area_id} ph = ",".join("?" * len(desc_ids)) plants = con.execute( f"""SELECT pl.id, pl.quantity, pl.area_id, pl.status, s.name AS species, v.name AS variety, pt.name AS planting, pt.id AS planting_id FROM plants pl LEFT JOIN species s ON s.id = pl.species_id LEFT JOIN varieties v ON v.id = pl.variety_id LEFT JOIN plantings pt ON pt.id = pl.planting_id WHERE pl.area_id IN ({ph}) AND pl.is_archived = 0 ORDER BY pl.id""", list(desc_ids), ).fetchall() stations = load_stations_for_area(con, area_id, by_id) return { "id": a["id"], "name": a["name"], "short_name": a.get("short_name") or "", "path": area_path_name(area_id, by_id), "length_ft": a.get("length_ft") or 0, "width_ft": a.get("width_ft") or 0, "garden_area_sqft": a.get("garden_area_sqft") or 0, "structure_features": _chat_truncate_text(a.get("structure_features"), 500), "sunlight": _chat_truncate_text(a.get("sunlight"), 500), "soil_environment": _chat_truncate_text(a.get("soil_environment"), 500), "notes": _chat_truncate_text(a.get("notes"), 1500), "subareas": [ {"id": c["id"], "name": c["name"]} for c in a["children"] ], "plants": [ { "id": p["id"], "qty": p["quantity"], "species": p["species"], "variety": p["variety"], "planting": p["planting"], "planting_id": p["planting_id"], "area_id": p["area_id"], } for p in plants ], "stations": [{"id": s["id"], "name": s["name"]} for s in stations], } def _tool_get_species_detail(con, *, species_id: int) -> dict: s = con.execute( "SELECT * FROM species WHERE id = ?", (species_id,) ).fetchone() if not s: return {"error": f"species #{species_id} not found"} varieties = con.execute( """SELECT id, name, description FROM varieties WHERE species_id = ? AND is_archived = 0 ORDER BY name""", (species_id,), ).fetchall() plantings = con.execute( """SELECT DISTINCT p.id, p.name, y.year FROM plants pl JOIN plantings p ON p.id = pl.planting_id LEFT JOIN years y ON y.id = p.year_id WHERE pl.species_id = ? AND pl.is_archived = 0 AND p.is_archived = 0 ORDER BY y.year DESC, p.name""", (species_id,), ).fetchall() return { "id": s["id"], "name": s["name"], "common_name": s["common_name"] or "", "type": s["type"] or "plant", "primary_function": s["primary_function"] or "edible", "plants_per_unit": s["plants_per_unit"] or 1, "space_per_unit_sqft": s["space_per_unit_sqft"] or 1.0, "description": _chat_truncate_text(s["description"], 1500), "varieties": [ {"id": v["id"], "name": v["name"], "description": _chat_truncate_text(v["description"], 200)} for v in varieties ], "current_plantings": [ {"id": p["id"], "name": p["name"], "year": p["year"]} for p in plantings ], } def _tool_find_plants(con, *, species: str = "", variety: str = "", status: str = "") -> dict: """Find plant groups. With no filters, returns ALL non-archived plants (capped at 50) so the model can browse what's in the garden when the snapshot's species summary isn't enough. Filters narrow by species name (substring), variety name (substring), and/or status (planted | unplanted | idea | removed).""" species = (species or "").strip() variety = (variety or "").strip() status = (status or "").strip() clauses = ["pl.is_archived = 0"] params: list = [] if species: clauses.append("LOWER(s.name) LIKE ?") params.append(f"%{species.lower()}%") if variety: clauses.append("LOWER(v.name) LIKE ?") params.append(f"%{variety.lower()}%") if status: clauses.append("pl.status = ?") params.append(status) rows = con.execute( f"""SELECT pl.id, pl.quantity, pl.area_id, s.name AS species, v.name AS variety, pt.id AS planting_id, pt.name AS planting, y.year AS year FROM plants pl LEFT JOIN species s ON s.id = pl.species_id LEFT JOIN varieties v ON v.id = pl.variety_id LEFT JOIN plantings pt ON pt.id = pl.planting_id LEFT JOIN years y ON y.id = pt.year_id WHERE {' AND '.join(clauses)} ORDER BY y.year DESC, pl.id LIMIT 50""", params, ).fetchall() areas = load_areas(con) _, by_id = build_tree(areas) return { "matches": [ { "plant_id": r["id"], "qty": r["quantity"], "species": r["species"], "variety": r["variety"], "planting_id": r["planting_id"], "planting": r["planting"], "year": r["year"], "area_id": r["area_id"], "area_path": ( area_path_name(r["area_id"], by_id) if r["area_id"] else None ), } for r in rows ], "count": len(rows), } def _tool_find_empty_space( con, *, area_id: Optional[int] = None, min_sqft: float = 1.0, ) -> dict: """Heuristic: per area, sum the species_space_per_unit_sqft of plants occupying it; subtract from area sqft (length×width) or the explicit `garden_area_sqft` to get rough free space. Return areas whose free space ≥ `min_sqft`.""" areas = load_areas(con) _, by_id = build_tree(areas) if area_id is not None and area_id not in by_id: return {"error": f"area #{area_id} not found"} if area_id is not None: candidate_ids = list(get_descendant_ids(area_id, by_id) | {area_id}) else: candidate_ids = [a["id"] for a in areas] if not candidate_ids: return {"matches": [], "count": 0} ph = ",".join("?" * len(candidate_ids)) occ_rows = con.execute( f"""SELECT pl.area_id, COALESCE(v.space_per_unit_sqft, s.space_per_unit_sqft, 1.0) AS sqft, COALESCE(v.plants_per_unit, s.plants_per_unit, 1) AS ppu, pl.quantity FROM plants pl LEFT JOIN species s ON s.id = pl.species_id LEFT JOIN varieties v ON v.id = pl.variety_id WHERE pl.area_id IN ({ph}) AND pl.is_archived = 0""", candidate_ids, ).fetchall() occ_by_area: dict = {} for r in occ_rows: try: qty = int((r["quantity"] or "1").strip()) or 1 except (TypeError, ValueError): qty = 1 ppu = max(int(r["ppu"] or 1), 1) sqft = float(r["sqft"] or 1.0) * (qty / ppu) occ_by_area[r["area_id"]] = occ_by_area.get(r["area_id"], 0) + sqft matches: list = [] for aid in candidate_ids: a = by_id[aid] # Prefer explicit garden_area_sqft; else compute from L×W; else skip. cap = float(a.get("garden_area_sqft") or 0) if not cap: L = float(a.get("length_ft") or 0) W = float(a.get("width_ft") or 0) cap = L * W if not cap: continue # Skip non-leaf areas with sub-areas — their plants live in children. if a["children"]: continue used = occ_by_area.get(aid, 0) free = cap - used if free >= min_sqft: matches.append({ "area_id": aid, "path": area_path_name(aid, by_id), "capacity_sqft": round(cap, 1), "used_sqft": round(used, 1), "free_sqft": round(free, 1), "sunlight": _chat_truncate_text(a.get("sunlight"), 200), "soil_environment": _chat_truncate_text(a.get("soil_environment"), 200), }) matches.sort(key=lambda m: m["free_sqft"], reverse=True) return {"matches": matches[:25], "count": len(matches)} def _tool_get_recent_notes( con, *, target_type: str, target_id: int, limit: int = 20, ) -> dict: if target_type not in COMMENT_TARGET_TYPES: return {"error": f"unknown target_type {target_type!r}"} limit = max(1, min(int(limit), 100)) rows = con.execute( """SELECT c.id, c.comment_date, c.body, c.kind, ct.is_primary FROM field_notes c JOIN field_note_targets ct ON ct.field_note_id = c.id WHERE ct.target_type = ? AND ct.target_id = ? AND c.is_archived = 0 AND c.body IS NOT NULL AND c.body != '' ORDER BY c.comment_date DESC, c.id DESC LIMIT ?""", (target_type, target_id, limit), ).fetchall() return { "notes": [ { "id": r["id"], "date": r["comment_date"], "kind": r["kind"] or "", "body": _chat_truncate_text(r["body"], 800), "is_primary": bool(r["is_primary"]), } for r in rows ], "count": len(rows), } def _tool_search_notes(con, *, query: str, limit: int = 25) -> dict: q = (query or "").strip() if not q: return {"error": "supply a non-empty query"} limit = max(1, min(int(limit), 100)) rows = con.execute( """SELECT c.id, c.comment_date, c.body, c.kind, ct.target_type, ct.target_id FROM field_notes c LEFT JOIN field_note_targets ct ON ct.field_note_id = c.id AND ct.is_primary = 1 WHERE c.is_archived = 0 AND c.body LIKE ? ORDER BY c.comment_date DESC, c.id DESC LIMIT ?""", (f"%{q}%", limit), ).fetchall() return { "matches": [ { "id": r["id"], "date": r["comment_date"], "primary_type": r["target_type"], "primary_id": r["target_id"], "body": _chat_truncate_text(r["body"], 400), } for r in rows ], "count": len(rows), } def _tool_list_artifacts(con) -> dict: rows = con.execute( """SELECT id, kind, title, filename, generated_at, notes FROM artifacts ORDER BY generated_at DESC""" ).fetchall() return { "artifacts": [ { "id": r["id"], "kind": r["kind"], "title": r["title"], "filename": r["filename"], "generated_at": r["generated_at"], "notes": _chat_truncate_text(r["notes"], 200), } for r in rows ], "count": len(rows), } def _tool_read_artifact(con, *, artifact_id: int) -> dict: row = con.execute( "SELECT * FROM artifacts WHERE id = ?", (artifact_id,) ).fetchone() if not row: return {"error": f"artifact #{artifact_id} not found"} path = os.path.join(ARTIFACTS_DIR, row["filename"]) try: with open(path, "r", encoding="utf-8") as f: content = f.read() except OSError as e: return {"error": f"could not read file: {e}"} # Cap length so a giant plan doesn't blow the context. The model can # ask for specific sections in a follow-up if needed. return { "id": row["id"], "kind": row["kind"], "title": row["title"], "generated_at": row["generated_at"], "content": _chat_truncate_text(content, 8000), "truncated": len(content) > 8000, "full_length": len(content), } def _tool_suggest_new_tool( con, *, name: str, signature: str, why_useful: str, example_question: str, _session_id: Optional[int] = None, ) -> dict: """Meta-tool. The model calls this when it wishes a tool existed that would help answer the current question. Logged for later review at /chat/tool-suggestions; never affects the actual response.""" name = (name or "").strip() if not name: return {"error": "name required"} con.execute( """INSERT INTO tool_suggestions (created_at, session_id, name, signature, why_useful, example_question) VALUES (?, ?, ?, ?, ?, ?)""", (now(), _session_id, name, signature, why_useful, example_question), ) return { "ok": True, "message": ( f"Tool suggestion '{name}' recorded. " "Continue answering the user's question with the tools you have." ), } # Registry: name -> (callable, JSON-schema dict for Gemini's `tools` array). # The schema follows Gemini's `function_declarations` shape. CHAT_TOOLS = { "get_planting_detail": ( _tool_get_planting_detail, { "name": "get_planting_detail", "description": "Full detail for one planting: status, source, " "overview, and every plant group in it (with " "species/variety/area/qty).", "parameters": { "type": "object", "properties": { "planting_id": {"type": "integer", "description": "id from the planting list"}, }, "required": ["planting_id"], }, }, ), "get_area_detail": ( _tool_get_area_detail, { "name": "get_area_detail", "description": "Full detail for one area: dimensions, sunlight, " "soil, structure, sub-areas, every plant in the " "subtree, and stations serving it.", "parameters": { "type": "object", "properties": { "area_id": {"type": "integer"}, }, "required": ["area_id"], }, }, ), "get_species_detail": ( _tool_get_species_detail, { "name": "get_species_detail", "description": "Species info: type/function, spacing, varieties, " "and which plantings currently grow it.", "parameters": { "type": "object", "properties": { "species_id": {"type": "integer"}, }, "required": ["species_id"], }, }, ), "find_plants": ( _tool_find_plants, { "name": "find_plants", "description": "Browse plant groups. All filters optional — call " "with no args to get up to 50 non-archived plants " "across the whole garden. Filter species/variety " "by substring (case-insensitive). Filter status by " "exact value: 'planted' (in the ground), 'unplanted' " "(staged), 'idea' (planning), 'removed' (terminal).", "parameters": { "type": "object", "properties": { "species": {"type": "string"}, "variety": {"type": "string"}, "status": {"type": "string", "description": "planted | unplanted | idea | removed"}, }, }, }, ), "find_empty_space": ( _tool_find_empty_space, { "name": "find_empty_space", "description": "List leaf areas with free space (capacity minus " "occupied sqft). Pass min_sqft to filter; pass " "area_id to restrict to one subtree.", "parameters": { "type": "object", "properties": { "area_id": {"type": "integer"}, "min_sqft": {"type": "number", "description": "default 1.0"}, }, }, }, ), "get_recent_notes": ( _tool_get_recent_notes, { "name": "get_recent_notes", "description": "Up to 100 recent field notes attached to a " "specific entity. Pass target_type ('area', " "'planting', 'plant', 'species', 'variety', " "'station', 'year') and target_id.", "parameters": { "type": "object", "properties": { "target_type": {"type": "string"}, "target_id": {"type": "integer"}, "limit": {"type": "integer", "description": "default 20"}, }, "required": ["target_type", "target_id"], }, }, ), "search_notes": ( _tool_search_notes, { "name": "search_notes", "description": "Substring search across all field-note bodies " "(case-insensitive). Returns recent matches first.", "parameters": { "type": "object", "properties": { "query": {"type": "string"}, "limit": {"type": "integer", "description": "default 25"}, }, "required": ["query"], }, }, ), "list_artifacts": ( _tool_list_artifacts, { "name": "list_artifacts", "description": "List AI-generated artifacts (planting plans, " "fertilization schedules, watering plans, etc.).", "parameters": {"type": "object", "properties": {}}, }, ), "read_artifact": ( _tool_read_artifact, { "name": "read_artifact", "description": "Read the full markdown content of one artifact " "by id (truncated to ~8K chars).", "parameters": { "type": "object", "properties": { "artifact_id": {"type": "integer"}, }, "required": ["artifact_id"], }, }, ), "suggest_new_tool": ( _tool_suggest_new_tool, { "name": "suggest_new_tool", "description": "Meta-tool: call this when answering would be " "easier with a tool that doesn't exist yet. The " "suggestion is logged for the developer; this " "does NOT affect your current answer.", "parameters": { "type": "object", "properties": { "name": {"type": "string", "description": "snake_case identifier"}, "signature": {"type": "string", "description": "args + return shape"}, "why_useful": {"type": "string"}, "example_question": { "type": "string", "description": "user question that would benefit", }, }, "required": ["name", "why_useful"], }, }, ), } def chat_dispatch_tool( con, name: str, args: dict, session_id: Optional[int] = None, ) -> dict: """Execute a tool call by name. Returns whatever the tool returns, wrapped with `{"error": ...}` if the tool isn't registered or raises.""" spec = CHAT_TOOLS.get(name) if not spec: return {"error": f"unknown tool {name!r}"} fn, _schema = spec try: # The meta-tool needs the session id; pass it via a private kwarg so # the model can't accidentally trigger that branch with a manual arg. if name == "suggest_new_tool": return fn(con, _session_id=session_id, **(args or {})) return fn(con, **(args or {})) except TypeError as e: return {"error": f"bad arguments for {name}: {e}"} except sqlite3.Error as e: return {"error": f"database error in {name}: {e}"} CHAT_TOOL_DECLARATIONS = [spec[1] for spec in CHAT_TOOLS.values()] def gemini_chat_complete( con, history: list, user_parts: list, system_prompt: str, session_id: Optional[int] = None, max_iterations: int = CHAT_MAX_TOOL_ITERATIONS, ) -> dict: """Run one user turn through Gemini with function calling. `history` — prior conversation as Gemini `contents` entries (each is `{role: 'user'|'model'|'function', parts: [...]}`). `user_parts` — parts for the new user message (text + inline_data for images). Appended to history before the call. `system_prompt` — instruction text passed via `systemInstruction`. Loops: send → if model returns a functionCall, execute it via chat_dispatch_tool, append result, repeat. Stops at max_iterations or when the model returns plain text. Returns: {"text": , "tool_calls": [{name, args, result}, ...], "raw_history_added": []} """ import urllib.request import urllib.error if not GEMINI_API_KEY: raise HTTPException(500, "GEMINI_API_KEY is not configured") contents = list(history) contents.append({"role": "user", "parts": user_parts}) added: list = [{"role": "user", "parts": user_parts}] tool_calls_record: list = [] url = ( f"https://generativelanguage.googleapis.com/v1beta/models/" f"{GEMINI_CHAT_MODEL}:generateContent?key={GEMINI_API_KEY}" ) for _ in range(max_iterations): payload = { "contents": contents, "systemInstruction": {"parts": [{"text": system_prompt}]}, "tools": [{"function_declarations": CHAT_TOOL_DECLARATIONS}], "generationConfig": {"temperature": 0.4}, } req = urllib.request.Request( url, data=json.dumps(payload).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=120) as resp: body = resp.read().decode("utf-8") except urllib.error.HTTPError as e: detail = e.read().decode("utf-8", errors="replace")[:500] raise HTTPException(502, f"Gemini API error {e.code}: {detail}") except urllib.error.URLError as e: raise HTTPException(502, f"Gemini API unreachable: {e.reason}") data = json.loads(body) candidates = data.get("candidates") or [] if not candidates: return { "text": "(no response from model)", "tool_calls": tool_calls_record, "raw_history_added": added, } first = candidates[0] content = first.get("content") or {} parts = content.get("parts") or [] # If any part is a functionCall, execute every functionCall in this # response and feed the results back. Gemini may emit text + a # function call together; we treat the call as authoritative for the # loop and discard accompanying text (the model will produce final # text after seeing the tool result). function_calls = [p["functionCall"] for p in parts if "functionCall" in p] if function_calls: # Append the model's tool-call message to the running history. contents.append({"role": "model", "parts": parts}) added.append({"role": "model", "parts": parts}) tool_response_parts: list = [] for fc in function_calls: name = fc.get("name", "") args = fc.get("args", {}) or {} result = chat_dispatch_tool(con, name, args, session_id=session_id) tool_calls_record.append({ "name": name, "args": args, "result": result, }) tool_response_parts.append({ "functionResponse": {"name": name, "response": result}, }) contents.append({"role": "function", "parts": tool_response_parts}) added.append({"role": "function", "parts": tool_response_parts}) continue # Plain text response — extract and return. texts = [p["text"] for p in parts if "text" in p] final_text = "\n".join(t for t in texts if t).strip() contents.append({"role": "model", "parts": parts}) added.append({"role": "model", "parts": parts}) return { "text": final_text or "(empty reply)", "tool_calls": tool_calls_record, "raw_history_added": added, } # Hit the iteration cap. Return whatever we have so the user isn't # left hanging; the assistant message will explain the cap was hit. return { "text": ( "(stopped after exceeding tool-call iteration limit; the model " "may need narrower questions or additional tools)" ), "tool_calls": tool_calls_record, "raw_history_added": added, } CHAT_SYSTEM_PROMPT = ( "You are a garden assistant for the user, who is the gardener.\n\n" "THE SNAPSHOT BELOW IS YOUR GROUND TRUTH. Top to bottom it has: " "today's date and season; the user's areas; the species CATALOGUE " "(types defined, NOT what's planted); WHAT IS CURRENTLY PLANTED " "(by species, with qty + varieties); WHAT IS STAGED (unplanted/idea " "pools); plantings (each with the species in it); RECENT CARE " "ACTIONS (done-tagged notes, 60 days); OPEN/RECENT ISSUES " "(issue-tagged notes, 60 days); and other recent notes (30 days). " "Treat all of this as authoritative — do not claim you don't know " "what's in the garden when this is right here.\n\n" "FOR ONGOING-CARE QUESTIONS (fertilize? water? treat? prune? " "transplant?) follow this loop EVERY TIME:\n" " 1. Read 'Recent care actions' for matching keywords first. If " " the user already did the action recently (within a typical " " cadence — fertilize ~4-6 weeks, water 2-7 days, etc.), say " " so plainly and recommend what's next, not a repeat. Cite the " " specific date and what was done.\n" " 2. Read 'Open / recent issues' for problems that should " " influence your advice (aphids, deficiency, transplant shock, " " etc.). Mention them by date.\n" " 3. Cross-check against 'Currently planted' to be specific to " " species the user actually has — NOT hypothetical species.\n" " 4. If care relevant to the question is older than 60 days OR " " scattered across many notes, CALL search_notes with a " " keyword like 'fertiliz', 'water', 'mulch', 'prune', " " 'aphid' (substring match) before answering.\n\n" "USE TOOLS for: drill-down on a specific planting/area/species " "(get_planting_detail, get_area_detail, get_species_detail); the " "full notes on one entity (get_recent_notes); finding plants of a " "given species across the whole garden (find_plants); locating " "free space (find_empty_space); reading an artifact " "(read_artifact); searching all notes by keyword (search_notes).\n\n" "GENERAL GARDENING KNOWLEDGE is yours to use freely — but always " "ground it in the user's actual data above, not a generic answer.\n\n" "IMAGES: when the user uploads a photo, INSPECT IT and describe " "what you see — plants, pests, leaves, soil, structures, etc. " "Identify species/varieties as best you can and offer practical " "observations (disease signs, ripeness, growth stage, weeds, " "etc.). If ambiguous, say what you see and what would clarify. " "Never refuse to look at an image.\n\n" "ALWAYS produce a useful reply. suggest_new_tool is a developer " "side-channel — never use it as an excuse to skip answering.\n\n" "Tone: concise, practical, friendly. Cite entity ids (e.g. " "'planting #5') and dates ('you fed them on 2026-04-18') when " "you reference garden data. Don't invent data — if a tool or " "snapshot section is empty, say so plainly.\n\n" "GARDEN SNAPSHOT:\n" ) # --- Area routes --- @app.get("/", response_class=HTMLResponse) def index(request: Request, status: str = Query("planted")): payload = _overview_payload(None, status=status) with db() as con: recent_plantings = [ dict(r) for r in con.execute( """SELECT p.*, y.year AS year_value FROM plantings p LEFT JOIN years y ON y.id = p.year_id ORDER BY p.created_at DESC LIMIT 5""" ).fetchall() ] for rp in recent_plantings: groups, total = planting_total_qty(con, rp["id"]) rp["plant_count"] = total rp["group_count"] = groups payload["request"] = request payload["recent_plantings"] = recent_plantings payload["is_home"] = True return templates.TemplateResponse("overview_tiles.html", payload) def _fmt_dims(area: dict) -> str: parts = [] l, w = area.get("length_ft") or 0, area.get("width_ft") or 0 if l and w: def _n(x): return str(int(x)) if float(x).is_integer() else f"{x:.1f}" parts.append(f"{_n(l)}×{_n(w)} ft") ga = area.get("garden_area_sqft") or 0 if ga: parts.append(f"{int(ga)} sqft") return " · ".join(parts) # Status filter for area / overview / planting pages. Values map to a set of # `plants.status` values; None means "no filter" (show everything including # removed). `idea` and `unplanted` are only shown when the user explicitly # asks for them. OVERVIEW_STATUS_CHOICES = { "planted": {"planted"}, "unplanted": {"unplanted"}, "idea": {"idea"}, "active": {"planted", "unplanted"}, # everything that physically or virtually exists "all": None, } def _overview_payload(area_id: Optional[int], status: str = "planted"): statuses = OVERVIEW_STATUS_CHOICES.get(status, OVERVIEW_STATUS_CHOICES["planted"]) with db() as con: area_rows = load_areas(con) plants_by_area = load_plants_by_area(con, plant_statuses=statuses) areas_tree, by_id = build_tree(area_rows) plant_counts = get_plant_count_map(areas_tree, plants_by_area) species_summary = get_species_summary_map(areas_tree, plants_by_area) if area_id is None: roots = areas_tree if len(roots) == 1: # Skip the singleton root — land directly on its children. current = roots[0] children = current["children"] else: current = None children = roots else: current = by_id.get(area_id) if not current: raise HTTPException(404) children = current["children"] breadcrumbs = [] if current: breadcrumbs = get_ancestors(current["id"], by_id) enriched_children = [] for c in children: enriched_children.append({ "id": c["id"], "name": c["name"], "dims": _fmt_dims(c), "child_count": len(c["children"]), "plant_count": plant_counts.get(c["id"], 0), "summary": species_summary.get(c["id"], []), }) return { "current": current, "current_dims": _fmt_dims(current) if current else "", "current_total": plant_counts.get(current["id"], 0) if current else sum(plant_counts.get(r["id"], 0) for r in areas_tree), "current_summary": species_summary.get(current["id"], []) if current else [], "breadcrumbs": breadcrumbs, "children": enriched_children, "status_filter": status if status in OVERVIEW_STATUS_CHOICES else "planted", } @app.get("/overview/tiles") @app.get("/overview/tiles/{area_id}") def overview_tiles_redirect( area_id: Optional[int] = None, status: str = Query("planted"), ): """Legacy URL — the tile navigation is now part of the area detail page.""" qs = f"?status={status}" if status and status != "planted" else "" target = f"/areas/{area_id}{qs}" if area_id else ("/" + qs) return RedirectResponse(target, status_code=301) @app.get("/areas/arrange", response_class=HTMLResponse) def areas_arrange(request: Request): with db() as con: area_rows = load_areas(con) plants_by_area = load_plants_by_area(con) areas_tree, _ = build_tree(area_rows) plant_counts = get_plant_count_map(areas_tree, plants_by_area) return templates.TemplateResponse( "areas_arrange.html", { "request": request, "areas_tree": areas_tree, "plant_counts": plant_counts, }, ) @app.get("/areas/new", response_class=HTMLResponse) def area_new(request: Request, parent_id: Optional[int] = None): with db() as con: area_rows = load_areas(con) areas_tree, by_id = build_tree(area_rows) parent_options = flattened_area_options(by_id) area_tree = area_tree_payload(areas_tree, by_id) return templates.TemplateResponse( "area_form.html", { "request": request, "area": None, "parent_options": parent_options, "area_tree": area_tree, "selected_parent_id": parent_id, }, ) @app.post("/areas/reorder") async def areas_reorder(request: Request): data = await request.json() area_id = int(data["area_id"]) parent_id = data.get("parent_id") if parent_id is not None and parent_id != "": parent_id = int(parent_id) else: parent_id = None new_index = int(data["new_index"]) with db() as con: if parent_id is not None: area_rows = load_areas(con) _, by_id = build_tree(area_rows) exclude = {area_id} | get_descendant_ids(area_id, by_id) if parent_id in exclude: raise HTTPException(400, "Invalid parent (cycle)") con.execute( "UPDATE areas SET parent_id = ? WHERE id = ?", (parent_id, area_id), ) if parent_id is None: siblings = con.execute( """SELECT id FROM areas WHERE parent_id IS NULL AND id != ? ORDER BY COALESCE(sort_order, 999999), name""", (area_id,), ).fetchall() else: siblings = con.execute( """SELECT id FROM areas WHERE parent_id = ? AND id != ? ORDER BY COALESCE(sort_order, 999999), name""", (parent_id, area_id), ).fetchall() sibling_ids = [s["id"] for s in siblings] new_index = max(0, min(new_index, len(sibling_ids))) sibling_ids.insert(new_index, area_id) for i, sid in enumerate(sibling_ids): con.execute( "UPDATE areas SET sort_order = ? WHERE id = ?", (i * 10, sid), ) return {"ok": True} SKETCH_ROTATION_STEPS = (0, 45, 90, 135, 180, 225, 270, 315) def _normalize_rotation(v: str) -> int: try: n = int(v) except (TypeError, ValueError): return 0 n = n % 360 return n if n in SKETCH_ROTATION_STEPS else 0 def resolve_child_area_positions(parent: dict, children: list) -> list: """For each child area with dimensions, return a dict augmented with `resolved_x` and `resolved_y` (in parent natural-orientation coords, feet) representing the child's CENTER position. Uses the saved parent_pos_x/y when present; otherwise default-grid layout fits children evenly into the parent's rectangle. """ parent_L = float(parent.get("length_ft") or 0) parent_W = float(parent.get("width_ft") or 0) if parent_L <= 0 or parent_W <= 0 or not children: return [] sized = [c for c in children if (c.get("length_ft") or 0) > 0 and (c.get("width_ft") or 0) > 0] n = len(sized) if n == 0: return [] cols = max(1, round(math.sqrt(n * parent_L / parent_W))) rows = math.ceil(n / cols) cell_w = parent_L / cols cell_h = parent_W / rows out = [] for i, c in enumerate(sized): d = dict(c) if d.get("parent_pos_x") is not None and d.get("parent_pos_y") is not None: d["resolved_x"] = float(d["parent_pos_x"]) d["resolved_y"] = float(d["parent_pos_y"]) else: col = i % cols row = i // cols d["resolved_x"] = (col + 0.5) * cell_w d["resolved_y"] = (row + 0.5) * cell_h out.append(d) return out def resolve_plant_positions(plant: dict, area_length_ft: float, area_width_ft: float) -> list: """Return positions (x, y, plants_in_icon, sqft_per_icon) for each visible icon. Icon count = ceil(quantity / plants_per_unit) so multi-plant species (carrots, radish, white clover) collapse into fewer, smaller icons. Saved positions are honored when their length matches the icon count. Otherwise we generate a default grid sized to the icon footprint. """ try: qty = max(1, int((plant.get("quantity") or "1").strip())) except (TypeError, ValueError): qty = 1 ppu = max(1, int(plant.get("species_plants_per_unit") or 1)) sqft = float(plant.get("species_space_per_unit_sqft") or 1.0) if sqft <= 0: sqft = 1.0 icon_count = max(1, math.ceil(qty / ppu)) side = math.sqrt(sqft) # icon footprint side length in ft # Each icon represents `ppu` plants except possibly the last (partial). icon_plants = [ppu] * icon_count leftover = qty - ppu * (icon_count - 1) if leftover > 0: icon_plants[-1] = leftover raw = plant.get("positions") or "" saved = [] if raw: try: data = json.loads(raw) for p in data: if isinstance(p, dict) and "x" in p and "y" in p: saved.append((float(p["x"]), float(p["y"]))) except (ValueError, TypeError): saved = [] out = [] if len(saved) >= icon_count: for i in range(icon_count): out.append((saved[i][0], saved[i][1], icon_plants[i], sqft)) return out L = max(side, float(area_length_ft or 1)) W = max(side, float(area_width_ft or 1)) # Grid cells sized to the icon's footprint, then scaled to fit the area. cols = max(1, min(icon_count, math.floor(L / side) or 1)) rows = math.ceil(icon_count / cols) col_w = L / cols row_h = W / max(rows, 1) for i in range(icon_count): if i < len(saved): x, y = saved[i] else: col = i % cols row = i // cols x = (col + 0.5) * col_w y = (row + 0.5) * row_h out.append((x, y, icon_plants[i], sqft)) return out def sketch_geometry(length_ft: float, width_ft: float, rotation_deg, pad_ratio: float = 0.1) -> dict: """Layout values for an SVG sketch of a rectangle rotated by rotation_deg. Returns a viewBox sized to fit the rotated bounding box plus padding, plus a centered rect (in natural orientation) and the rotation origin (viewBox center). The caller draws the rect with these coords and applies a single wrapper. `pad_ratio` controls breathing room around the rotated bounding box. Default 0.1 (10%) for the regular sketch view; layout mode passes a smaller value (~0.02) to use more of the available canvas. """ L = float(length_ft or 0) W = float(width_ft or 0) try: rot = int(rotation_deg or 0) except (TypeError, ValueError): rot = 0 theta = math.radians(rot) c = abs(math.cos(theta)) s = abs(math.sin(theta)) bbox_w = L * c + W * s bbox_h = L * s + W * c pad = max(bbox_w, bbox_h) * pad_ratio vb_w = bbox_w + 2 * pad vb_h = bbox_h + 2 * pad cx = vb_w / 2 cy = vb_h / 2 return { "vb_w": vb_w, "vb_h": vb_h, "cx": cx, "cy": cy, "rect_x": cx - L / 2, "rect_y": cy - W / 2, "rect_w": L, "rect_h": W, "pad": pad, "rotation": rot, } templates.env.globals["sketch_geometry"] = sketch_geometry def species_icon_color(name) -> str: """Stable HSL color from a species name. Used to tint plant icons in the area sketch so different species are visually distinguishable.""" s = (name or "").strip() if not s: return "hsl(0, 0%, 70%)" digest = hashlib.md5(s.lower().encode("utf-8")).hexdigest() hue = int(digest[:4], 16) % 360 return f"hsl({hue}, 55%, 50%)" templates.env.globals["species_icon_color"] = species_icon_color def _parse_float(v: str) -> float: v = (v or "").strip() if not v: return 0.0 try: return float(v) except ValueError: return 0.0 @app.post("/areas") def area_create( name: str = Form(...), notes: str = Form(""), parent_id: str = Form(""), short_name: str = Form(""), length_ft: str = Form(""), width_ft: str = Form(""), garden_area_sqft: str = Form(""), structure_features: str = Form(""), sunlight: str = Form(""), soil_environment: str = Form(""), sketch_rotation: str = Form("0"), featured_image: Optional[UploadFile] = File(None), ): pid = int(parent_id) if parent_id.strip() else None rotation = _normalize_rotation(sketch_rotation) image_rel = "" if featured_image is not None and (featured_image.filename or "").strip(): image_rel = _save_upload(featured_image, "image") with db() as con: max_order = con.execute( "SELECT COALESCE(MAX(sort_order), 0) FROM areas WHERE " + ("parent_id IS NULL" if pid is None else "parent_id = ?"), () if pid is None else (pid,), ).fetchone()[0] cur = con.execute( """INSERT INTO areas (name, notes, parent_id, sort_order, short_name, length_ft, width_ft, garden_area_sqft, structure_features, sunlight, soil_environment, featured_image_path, sketch_rotation, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", (name.strip(), notes.strip(), pid, max_order + 10, short_name.strip(), _parse_float(length_ft), _parse_float(width_ft), _parse_float(garden_area_sqft), structure_features.strip(), sunlight.strip(), soil_environment.strip(), image_rel, rotation, now()), ) new_id = cur.lastrowid log_add(con, "area", new_id, name.strip()) return RedirectResponse(f"/areas/{new_id}", status_code=303) @app.get("/areas/{area_id}", response_class=HTMLResponse) def area_detail( area_id: int, request: Request, children: int = Query(1), related: int = Query(0), status: str = Query("planted"), archived: int = Query(0), ): include_children = bool(children) include_related = bool(related) show_archived = bool(archived) statuses = OVERVIEW_STATUS_CHOICES.get(status, OVERVIEW_STATUS_CHOICES["planted"]) with db() as con: # Visiting an archived area's URL directly should still work — pin the # archived flag in the URL so children/comments load consistently. target_row = con.execute( "SELECT is_archived FROM areas WHERE id = ?", (area_id,) ).fetchone() if not target_row: raise HTTPException(404) if target_row["is_archived"] and not show_archived: return RedirectResponse(f"/areas/{area_id}?archived=1", status_code=303) area_rows = load_areas(con, include_archived=show_archived) areas_tree, by_id = build_tree(area_rows) area = by_id.get(area_id) if not area: raise HTTPException(404) plants_by_area = load_plants_by_area( con, plant_statuses=statuses, include_archived=show_archived ) stations = load_stations_for_area( con, area_id, by_id, include_archived=show_archived ) area_comments = load_comments_expanded( con, "area", area_id, include_children=include_children, include_related=include_related, by_id=by_id, include_archived=show_archived, ) area_summary = get_area_plant_summary(con, area_id, by_id, plant_statuses=statuses) # Get planting info for plants directly in this area plants = plants_by_area.get(area_id, []) # Sketch shows ALL plant_groups in this area regardless of the # ?status= filter (which is for the inventory table). Plants in # `idea`/`planned` plantings should still be visible on the sketch. sketch_plants_by_area = load_plants_by_area(con, plant_statuses=None) sketch_plants = sketch_plants_by_area.get(area_id, []) # Resolve sketch positions (used by the area-detail dialog drag UI). L = float(area.get("length_ft") or 0) W = float(area.get("width_ft") or 0) if L > 0 and W > 0: for p in sketch_plants: p["positions_resolved"] = resolve_plant_positions(p, L, W) # Direct children with their own dimensions, positioned inside the # parent for the layout sketch. sketch_children = resolve_child_area_positions(area, area["children"]) # Uniform label font size in SVG user units (feet). Scaling by max # parent dimension keeps the rendered pixel size roughly constant # across views (Home's labels and Backyard's labels both look ~13px). parent_dim = max(L or 0, W or 0) child_label_size_ft = round(parent_dim * 0.014, 3) if parent_dim > 0 else 1.5 planting_map = {} if plants: planting_ids = {p["planting_id"] for p in plants if p["planting_id"] is not None} for pid in planting_ids: row = con.execute( """SELECT p.*, y.year AS year_value FROM plantings p LEFT JOIN years y ON y.id = p.year_id WHERE p.id = ?""", (pid,), ).fetchone() if row: planting_map[pid] = dict(row) # Sort the plant list so rows from the same planting cluster # together; the template inserts a divider row each time the # planting changes. NULL planting_id (future "unplanted/idea" # plants) sort to the bottom in a single group. def _planting_sort_key(p): pid = p["planting_id"] if pid is None: return (1, "", "", "", p["id"]) pl = planting_map.get(pid, {}) yr = pl.get("year_value") or 0 # Newer year first, then planting name, then species/variety. return (0, -int(yr or 0), (pl.get("name") or "").lower(), (p.get("species_name") or "").lower(), (p.get("variety_name") or "").lower()) plants = sorted(plants, key=_planting_sort_key) breadcrumbs = get_ancestors(area_id, by_id) plant_counts = get_plant_count_map(areas_tree, plants_by_area) species_summary = get_species_summary_map(areas_tree, plants_by_area) enriched_children = [ { "id": c["id"], "name": c["name"], "dims": _fmt_dims(c), "child_count": len(c["children"]), "plant_count": plant_counts.get(c["id"], 0), "summary": species_summary.get(c["id"], []), "featured_image_path": c.get("featured_image_path") or "", } for c in area["children"] ] area_options = flattened_area_options(by_id) area_tree = area_tree_payload(areas_tree, by_id) with db() as con: species_list = load_species(con, include_archived=show_archived) varieties_list = load_varieties(con, include_archived=show_archived) return templates.TemplateResponse( "area_detail.html", { "request": request, "area": area, "breadcrumbs": breadcrumbs, "children": area["children"], "enriched_children": enriched_children, "current_dims": _fmt_dims(area), "current_total": plant_counts.get(area_id, 0), "current_summary": species_summary.get(area_id, []), "status_filter": status if status in OVERVIEW_STATUS_CHOICES else "planted", "plants": plants, "sketch_plants": sketch_plants, "sketch_children": sketch_children, "child_label_size_ft": child_label_size_ft, "plant_counts": plant_counts, "stations": stations, "area_comments": area_comments, "planting_map": planting_map, "by_id": by_id, "summary": area_summary, "area_options": area_options, "area_tree": area_tree, "species_list": species_list, "varieties_list": varieties_list, "include_children": include_children, "include_related": include_related, "supports_children": True, "show_archived": show_archived, }, ) # --------------------------------------------------------------------------- # Layout mode — mobile-optimized full-viewport editor for an area's plant # positions. Tap-pick-tap-place, inventory tray, auto-rotate, auto-arrange. # --------------------------------------------------------------------------- def _plant_positions_strict(plant: dict) -> list: """Return ONLY the placed positions (positions JSON as-is), no auto-fill. Layout mode treats the positions array as the source of truth; anything beyond `quantity - len(positions)` lives in the inventory tray.""" raw = plant.get("positions") or "" if not raw: return [] try: data = json.loads(raw) except (ValueError, TypeError): return [] out = [] for p in data: if isinstance(p, dict) and "x" in p and "y" in p: try: out.append({"x": float(p["x"]), "y": float(p["y"])}) except (TypeError, ValueError): continue return out @app.get("/areas/{area_id}/layout", response_class=HTMLResponse) def area_layout(area_id: int, request: Request): """Full-viewport mobile-first plant layout editor. Differences from /areas/{id} sketch dialog: - Its own page (not a ), so the URL is shareable and back works. - Positions are strict (no auto-fill); plants without positions sit in the inventory tray. - Tap-pick-tap-place model (no drag thresholds for touch confusion). - Auto-rotate to fit viewport (computed client-side). - Companion-aware auto-arrange via Gemini. """ with db() as con: area_rows = load_areas(con, include_archived=True) _, by_id = build_tree(area_rows) area = by_id.get(area_id) if not area: raise HTTPException(404) L = float(area.get("length_ft") or 0) W = float(area.get("width_ft") or 0) if L <= 0 or W <= 0: raise HTTPException(400, "Area must have dimensions to use layout mode.") sketch_plants_by_area = load_plants_by_area(con, plant_statuses=None) sketch_plants = sketch_plants_by_area.get(area_id, []) # Global pools — unplanted / idea plants NOT currently attached to # THIS area. (Plants of those statuses that ARE attached to this area # already show up in sketch_plants → Inventory tab.) When the user # places one of these, the JS calls /api/plants/{id}/attach-to-area # which sets area_id + first position. pool_rows = con.execute( """SELECT p.*, s.name AS species_name, v.name AS variety_name, COALESCE(v.plants_per_unit, s.plants_per_unit, 1) AS species_plants_per_unit, COALESCE(v.space_per_unit_sqft, s.space_per_unit_sqft, 1.0) AS species_space_per_unit_sqft, a2.name AS source_area_name FROM plants p LEFT JOIN species s ON s.id = p.species_id LEFT JOIN varieties v ON v.id = p.variety_id LEFT JOIN areas a2 ON a2.id = p.area_id WHERE p.is_archived = 0 AND p.status IN ('unplanted', 'idea') AND (p.area_id IS NULL OR p.area_id != ?) ORDER BY p.status, COALESCE(v.name, s.name, ''), p.id""", (area_id,), ).fetchall() unplanted_pool = [dict(r) for r in pool_rows if r["status"] == "unplanted"] idea_pool = [dict(r) for r in pool_rows if r["status"] == "idea"] # Build placed list + inventory. # Each plant_group is split into "icons" of up to ppu plants each. For # qty=15, ppu=4: 4 icons sized [4, 4, 4, 3] (partial last). The inventory # tile displays *plants* not icons; placing/removing icons add or subtract # the icon's plant_count from the display total. placed_icons = [] # one entry per canvas icon inventory_tiles = [] # one entry per plant_group (always emitted, even if empty) for p in sketch_plants: try: qty = max(1, int((p.get("quantity") or "1").strip())) except (TypeError, ValueError): qty = 1 ppu = max(1, int(p.get("species_plants_per_unit") or 1)) sqft = float(p.get("species_space_per_unit_sqft") or 1.0) side = math.sqrt(max(sqft, 0.05)) positions = _plant_positions_strict(p) icon_count_total = max(1, math.ceil(qty / ppu)) icon_plants_each = [ppu] * icon_count_total leftover = qty - ppu * (icon_count_total - 1) if leftover > 0: icon_plants_each[-1] = leftover # Plant physical state — controls visual style on the canvas. # 'planted' = solid (default), 'unplanted' = dashed, 'idea' = dotted. pstatus = (p.get("status") or "planted") placed_n = min(len(positions), icon_count_total) for i, pos in enumerate(positions[:icon_count_total]): placed_icons.append({ "plant_id": p["id"], "icon_index": i, "x": pos["x"], "y": pos["y"], "side": side, "sqft": sqft, "plant_count": icon_plants_each[i], "species_name": p.get("species_name") or "", "variety_name": p.get("variety_name") or "", "species_id": p.get("species_id"), "variety_id": p.get("variety_id"), "status": pstatus, "icon": species_icon(p.get("species_name")), "color": species_icon_color(p.get("species_name") or ""), }) # Inventory queue: the unplaced icons (FIFO). Sending a placed icon # back appends its plant_count; placing pops from the front. inv_plants_each = icon_plants_each[placed_n:] inv_count_plants = sum(inv_plants_each) # Always emit a tile so user can send placed icons back even when # inventory is empty. inventory_tiles.append({ "plant_id": p["id"], "species_name": p.get("species_name") or "?", "variety_name": p.get("variety_name") or "", "species_id": p.get("species_id"), "variety_id": p.get("variety_id"), "status": pstatus, "qty_total": qty, "ppu": ppu, "sqft": sqft, "side": side, "count": inv_count_plants, # in PLANTS (for display) "inv_plants_each": inv_plants_each, # queue of plant_counts "icon": species_icon(p.get("species_name")), "color": species_icon_color(p.get("species_name") or ""), }) # Build pool tiles for the global Unplanted + Idea trays. Same shape as # inventory_tiles, but the queue is the FULL icon schedule (none placed # in this area yet) and the click-handler in JS calls attach-to-area # instead of just updating positions. def _build_pool_tile(p, pool_status): try: q = max(1, int((p.get("quantity") or "1").strip())) except (TypeError, ValueError): q = 1 ppu = max(1, int(p.get("species_plants_per_unit") or 1)) sqft = float(p.get("species_space_per_unit_sqft") or 1.0) side = math.sqrt(max(sqft, 0.05)) icon_count_total = max(1, math.ceil(q / ppu)) icon_plants_each = [ppu] * icon_count_total leftover = q - ppu * (icon_count_total - 1) if leftover > 0: icon_plants_each[-1] = leftover return { "plant_id": p["id"], "species_name": p.get("species_name") or "?", "variety_name": p.get("variety_name") or "", "species_id": p.get("species_id"), "variety_id": p.get("variety_id"), "status": pool_status, "source_area_name": p.get("source_area_name") or "", "qty_total": q, "ppu": ppu, "sqft": sqft, "side": side, "count": sum(icon_plants_each), "inv_plants_each": icon_plants_each, "icon": species_icon(p.get("species_name")), "color": species_icon_color(p.get("species_name") or ""), } unplanted_pool_tiles = [_build_pool_tile(p, "unplanted") for p in unplanted_pool] idea_pool_tiles = [_build_pool_tile(p, "idea") for p in idea_pool] # Layout mode renders the rect axis-aligned (NOT compass-rotated). The # compass overlay tells the user where N actually is. This way the SVG # viewBox is tight to the rect (no wasted space from rotation bounding # box) and the screen-fit rotation only needs to consider 0° vs 90°. compass_rot = ((area.get("sketch_rotation") or 0) % 360) g = sketch_geometry(L, W, 0, pad_ratio=0.02) return templates.TemplateResponse( "layout_mode.html", { "request": request, "area": area, "L": L, "W": W, "compass_rot": compass_rot, "g": g, "placed_icons": placed_icons, "inventory_tiles": inventory_tiles, "unplanted_pool_tiles": unplanted_pool_tiles, "idea_pool_tiles": idea_pool_tiles, }, ) @app.post("/api/areas/{area_id}/auto-arrange") async def api_area_auto_arrange(area_id: int, request: Request): """Ask Gemini to suggest companion-aware placements for the inventory plants in this area. Returns proposed placements; client applies them via the existing /api/plants/{id}/positions endpoint.""" if not verify_session(request.cookies.get(COOKIE_NAME)): raise HTTPException(401, "auth required") if not GEMINI_API_KEY: raise HTTPException(500, "GEMINI_API_KEY not configured") with db() as con: area = con.execute("SELECT * FROM areas WHERE id = ?", (area_id,)).fetchone() if not area: raise HTTPException(404) L = float(area["length_ft"] or 0) W = float(area["width_ft"] or 0) if L <= 0 or W <= 0: raise HTTPException(400, "area lacks dimensions") plants = load_plants_by_area(con, plant_statuses=None).get(area_id, []) # Build inventory + placed lists for the prompt. inv = [] placed = [] for p in plants: try: qty = max(1, int((p.get("quantity") or "1").strip())) except (TypeError, ValueError): qty = 1 ppu = max(1, int(p.get("species_plants_per_unit") or 1)) sqft = float(p.get("species_space_per_unit_sqft") or 1.0) positions = _plant_positions_strict(p) icon_count = max(1, math.ceil(qty / ppu)) inv_count = max(0, icon_count - len(positions)) for pos in positions: placed.append({ "plant_id": p["id"], "species": p.get("species_name"), "variety": p.get("variety_name"), "x": round(pos["x"], 2), "y": round(pos["y"], 2), "sqft": sqft, }) if inv_count > 0: inv.append({ "plant_id": p["id"], "species": p.get("species_name"), "variety": p.get("variety_name"), "count": inv_count, "sqft_per_icon": sqft, "plants_per_icon": ppu, }) if not inv: return {"ok": True, "message": "Inventory empty — nothing to arrange.", "assignments": []} prompt = { "task": "Suggest companion-aware positions for the inventory plants.", "area": { "name": area["name"], "length_ft": L, "width_ft": W, "structure_features": area["structure_features"] or "", "sunlight": area["sunlight"] or "", }, "placed": placed, "inventory": inv, "instructions": ( "For each inventory item, propose `count` (x, y) coordinates in " "rect-local feet (0..length_ft × 0..width_ft). Group same-species " "plants together. Place companions adjacent where reasonable " "(e.g., basil near tomato, marigolds at perimeter). Avoid " "overlap with placed items by at least sqrt(sqft) feet. Respect " "sunlight/structure where it matters. Return ONLY valid JSON " "with shape: {assignments: [{plant_id, positions: [{x, y}, ...]}]}" ), } import urllib.request, urllib.error payload = { "contents": [{"parts": [{"text": json.dumps(prompt)}]}], "generationConfig": { "temperature": 0.2, "responseMimeType": "application/json", }, } url = (f"https://generativelanguage.googleapis.com/v1beta/models/" f"{GEMINI_TRANSCRIBE_MODEL}:generateContent?key={GEMINI_API_KEY}") req = urllib.request.Request( url, data=json.dumps(payload).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=60) as resp: body = resp.read().decode("utf-8") except urllib.error.HTTPError as e: detail = e.read().decode("utf-8", errors="replace")[:500] raise HTTPException(502, f"Gemini error {e.code}: {detail}") except urllib.error.URLError as e: raise HTTPException(502, f"Gemini unreachable: {e.reason}") try: data = json.loads(body) text = data["candidates"][0]["content"]["parts"][0]["text"] parsed = json.loads(text) except (KeyError, IndexError, TypeError, ValueError): raise HTTPException(502, f"Unexpected Gemini response: {body[:300]}") assignments = parsed.get("assignments", []) # Clamp positions to area bounds. for a in assignments: for p in a.get("positions", []): if "x" in p: p["x"] = max(0.0, min(L, float(p["x"]))) if "y" in p: p["y"] = max(0.0, min(W, float(p["y"]))) return {"ok": True, "assignments": assignments} @app.post("/areas/{area_id}") def area_update( area_id: int, name: str = Form(...), notes: str = Form(""), parent_id: str = Form(""), short_name: str = Form(""), length_ft: str = Form(""), width_ft: str = Form(""), garden_area_sqft: str = Form(""), structure_features: str = Form(""), sunlight: str = Form(""), soil_environment: str = Form(""), sketch_rotation: str = Form("0"), featured_image: Optional[UploadFile] = File(None), remove_featured_image: str = Form(""), ): rotation = _normalize_rotation(sketch_rotation) pid = int(parent_id) if parent_id.strip() else None with db() as con: if pid is not None: area_rows = load_areas(con) _, by_id = build_tree(area_rows) exclude = {area_id} | get_descendant_ids(area_id, by_id) if pid in exclude: raise HTTPException(400, "Invalid parent (would create a cycle)") existing = con.execute( "SELECT featured_image_path FROM areas WHERE id = ?", (area_id,) ).fetchone() current_image = (existing["featured_image_path"] if existing else "") or "" new_upload = featured_image is not None and (featured_image.filename or "").strip() if new_upload: image_rel = _save_upload(featured_image, "image") _delete_note_file(current_image) elif remove_featured_image: image_rel = "" _delete_note_file(current_image) else: image_rel = current_image snapshot_overview(con, "area", area_id, notes.strip()) con.execute( """UPDATE areas SET name = ?, notes = ?, parent_id = ?, short_name = ?, length_ft = ?, width_ft = ?, garden_area_sqft = ?, structure_features = ?, sunlight = ?, soil_environment = ?, featured_image_path = ?, sketch_rotation = ? WHERE id = ?""", (name.strip(), notes.strip(), pid, short_name.strip(), _parse_float(length_ft), _parse_float(width_ft), _parse_float(garden_area_sqft), structure_features.strip(), sunlight.strip(), soil_environment.strip(), image_rel, rotation, area_id), ) return RedirectResponse(f"/areas/{area_id}", status_code=303) # Map (entity_type → DB table, URL plural). Keep in sync with the # `featured_image_block` macro in templates/_featured_image.html. FEATURED_IMAGE_ENTITIES = { "area": ("areas", "areas"), "planting": ("plantings", "plantings"), "plant": ("plants", "plants"), "station": ("watering_stations", "stations"), "species": ("species", "species"), "variety": ("varieties", "varieties"), "supply": ("supplies", "supplies"), "year": ("years", "years"), } def _set_featured_image(table: str, entity_id: int, upload: UploadFile, return_url: str): """Save an uploaded image, point the entity row at it, and delete the previous file from disk. `table` is a code-controlled identifier so the f-string is safe (not user input).""" if not (upload.filename or "").strip(): raise HTTPException(400, "no file provided") image_rel = _save_upload(upload, "image") # Map plural table → singular logged entity_type. _table_to_entity = { "areas": "area", "plantings": "planting", "plants": "plant", "species": "species", "varieties": "variety", "watering_stations": "station", "supplies": "supply", "years": "year", } with db() as con: row = con.execute( f"SELECT featured_image_path FROM {table} WHERE id = ?", (entity_id,) ).fetchone() if not row: raise HTTPException(404) old = (row["featured_image_path"] or "").strip() con.execute( f"UPDATE {table} SET featured_image_path = ? WHERE id = ?", (image_rel, entity_id), ) ent = _table_to_entity.get(table) if ent: log_edit( con, ent, entity_id, "featured_image_path", old, image_rel, label=_entity_label_safe(con, ent, entity_id), ) if old: _delete_note_file(old) return RedirectResponse(return_url, status_code=303) @app.post("/areas/{area_id}/featured-image") def area_set_featured_image(area_id: int, featured_image: UploadFile = File(...)): return _set_featured_image("areas", area_id, featured_image, f"/areas/{area_id}") @app.post("/plantings/{planting_id}/featured-image") def planting_set_featured_image(planting_id: int, featured_image: UploadFile = File(...)): return _set_featured_image("plantings", planting_id, featured_image, f"/plantings/{planting_id}") @app.post("/plants/{plant_id}/featured-image") def plant_set_featured_image(plant_id: int, featured_image: UploadFile = File(...)): return _set_featured_image("plants", plant_id, featured_image, f"/plants/{plant_id}") @app.post("/stations/{station_id}/featured-image") def station_set_featured_image(station_id: int, featured_image: UploadFile = File(...)): return _set_featured_image("watering_stations", station_id, featured_image, f"/stations/{station_id}") @app.post("/species/{species_id}/featured-image") def species_set_featured_image(species_id: int, featured_image: UploadFile = File(...)): return _set_featured_image("species", species_id, featured_image, f"/species/{species_id}") @app.post("/varieties/{variety_id}/featured-image") def variety_set_featured_image(variety_id: int, featured_image: UploadFile = File(...)): return _set_featured_image("varieties", variety_id, featured_image, f"/varieties/{variety_id}") @app.post("/supplies/{supply_id}/featured-image") def supply_set_featured_image(supply_id: int, featured_image: UploadFile = File(...)): # Supplies have no detail page; redirect to the edit form (where the # featured-image block is rendered). return _set_featured_image("supplies", supply_id, featured_image, f"/supplies/{supply_id}/edit") @app.post("/years/{year_id}/featured-image") def year_set_featured_image(year_id: int, featured_image: UploadFile = File(...)): return _set_featured_image("years", year_id, featured_image, f"/years/{year_id}") @app.post("/areas/{area_id}/delete") def area_delete(area_id: int): parent = None with db() as con: row = con.execute( "SELECT parent_id FROM areas WHERE id = ?", (area_id,) ).fetchone() if row: parent = row["parent_id"] label = _entity_label_safe(con, "area", area_id) delete_area_recursive(con, area_id) log_delete(con, "area", area_id, label) target = f"/areas/{parent}" if parent else "/" return RedirectResponse(target, status_code=303) # --- Year routes --- @app.get("/years", response_class=HTMLResponse) def year_list(request: Request, archived: int = Query(0)): show_archived = bool(archived) arch_clause = "" if show_archived else " AND pl.is_archived = 0" with db() as con: years = load_years(con, include_archived=show_archived) for y in years: y["planting_count"] = con.execute( f"SELECT COUNT(*) FROM plantings pl " f"WHERE pl.year_id = ?{arch_clause}", (y["id"],), ).fetchone()[0] qty_row = con.execute( f"""SELECT COALESCE(SUM(CASE WHEN p.quantity != '' THEN CAST(p.quantity AS INTEGER) ELSE 1 END), 0) FROM plants p JOIN plantings pl ON pl.id = p.planting_id WHERE pl.year_id = ?{arch_clause}""", (y["id"],), ).fetchone() y["plant_count"] = qty_row[0] return templates.TemplateResponse( "year_list.html", {"request": request, "years": years, "show_archived": show_archived}, ) @app.get("/years/new", response_class=HTMLResponse) def year_new(request: Request): return templates.TemplateResponse( "year_form.html", {"request": request, "year_obj": None} ) @app.post("/years") def year_create(year: int = Form(...), notes: str = Form("")): with db() as con: cur = con.execute( "INSERT INTO years (year, notes, created_at) VALUES (?, ?, ?)", (year, notes.strip(), now()), ) new_id = cur.lastrowid log_add(con, "year", new_id, str(year)) return RedirectResponse(f"/years/{new_id}", status_code=303) @app.get("/years/{year_id}", response_class=HTMLResponse) def year_detail( year_id: int, request: Request, related: int = Query(0), archived: int = Query(0), ): include_related = bool(related) show_archived = bool(archived) with db() as con: row = con.execute("SELECT * FROM years WHERE id = ?", (year_id,)).fetchone() if not row: raise HTTPException(404) if row["is_archived"] and not show_archived: return RedirectResponse(f"/years/{year_id}?archived=1", status_code=303) year_obj = dict(row) pl_arch = "" if show_archived else " AND is_archived = 0" plantings = [ dict(r) for r in con.execute( f"SELECT * FROM plantings WHERE year_id = ?{pl_arch} ORDER BY name", (year_id,), ).fetchall() ] for p in plantings: groups, total = planting_total_qty( con, p["id"], include_archived=show_archived ) p["plant_count"] = total p["group_count"] = groups area_rows_y = load_areas(con, include_archived=show_archived) _, by_id_y = build_tree(area_rows_y) comments = load_comments_expanded( con, "year", year_id, include_children=False, include_related=include_related, by_id=by_id_y, primary_only_for_direct=not include_related, include_archived=show_archived, ) return templates.TemplateResponse( "year_detail.html", { "request": request, "year_obj": year_obj, "plantings": plantings, "year_comments": comments, "include_related": include_related, "include_children": False, "supports_children": False, "show_archived": show_archived, }, ) @app.post("/years/{year_id}") def year_update(year_id: int, year: int = Form(...), notes: str = Form("")): with db() as con: snapshot_overview(con, "year", year_id, notes.strip()) con.execute( "UPDATE years SET year = ?, notes = ? WHERE id = ?", (year, notes.strip(), year_id), ) return RedirectResponse(f"/years/{year_id}", status_code=303) @app.post("/years/{year_id}/delete") def year_delete(year_id: int): with db() as con: label = _entity_label_safe(con, "year", year_id) delete_target_refs(con, "year", year_id) con.execute("DELETE FROM years WHERE id = ?", (year_id,)) log_delete(con, "year", year_id, label) return RedirectResponse("/years", status_code=303) # --- Planting routes --- @app.get("/plantings", response_class=HTMLResponse) def planting_list(request: Request, archived: int = Query(0)): show_archived = bool(archived) with db() as con: plantings = load_plantings(con, include_archived=show_archived) for p in plantings: groups, total = planting_total_qty( con, p["id"], include_archived=show_archived ) p["plant_count"] = total p["group_count"] = groups years = load_years(con, include_archived=show_archived) # Group plantings by year, sort within year by status order status_order = {"idea": 0, "planned": 1, "planted": 2, "skipped": 3} by_year = {} no_year = [] for p in plantings: p.setdefault("status", "planted") if p["year_value"]: by_year.setdefault(p["year_value"], []).append(p) else: no_year.append(p) for group in list(by_year.values()) + [no_year]: group.sort(key=lambda x: (status_order.get(x.get("status") or "planted", 2), x["name"].lower())) year_groups = sorted(by_year.items(), key=lambda x: x[0], reverse=True) return templates.TemplateResponse( "planting_list.html", { "request": request, "year_groups": year_groups, "no_year": no_year, "total_plantings": len(plantings), "show_archived": show_archived, }, ) @app.get("/unplanted", response_class=HTMLResponse) def unplanted_list(request: Request, tab: str = Query("unplanted")): """List + quick-add for plants outside the planted state. The unplanted-plants model uses plant.status to distinguish: unplanted = physically exists (bought / seeded indoors) but not in ground idea = planning placeholder; may not exist physically yet Both share this page via a tab toggle. """ tab = tab if tab in ("unplanted", "idea") else "unplanted" with db() as con: rows = con.execute( """SELECT p.*, s.name AS species_name, v.name AS variety_name, a.name AS area_name, a.id AS attached_area_id, COALESCE(v.plants_per_unit, s.plants_per_unit, 1) AS plants_per_unit FROM plants p LEFT JOIN species s ON s.id = p.species_id LEFT JOIN varieties v ON v.id = p.variety_id LEFT JOIN areas a ON a.id = p.area_id WHERE p.is_archived = 0 AND p.status IN ('unplanted', 'idea') ORDER BY p.status, COALESCE(v.name, s.name, ''), p.id""" ).fetchall() unplanted = [dict(r) for r in rows if r["status"] == "unplanted"] ideas = [dict(r) for r in rows if r["status"] == "idea"] species_list = load_species(con) varieties_list = load_varieties(con) # All non-archived plantings — used by the "mark as planted" picker. plantings = con.execute( """SELECT pl.id, pl.name, y.year AS year_value FROM plantings pl LEFT JOIN years y ON y.id = pl.year_id WHERE COALESCE(pl.is_archived, 0) = 0 ORDER BY COALESCE(y.year, 0) DESC, pl.name""" ).fetchall() return templates.TemplateResponse( "unplanted_list.html", { "request": request, "tab": tab, "unplanted": unplanted, "ideas": ideas, "species_list": species_list, "varieties_list": varieties_list, "plantings": [dict(p) for p in plantings], }, ) @app.post("/unplanted") async def unplanted_create(request: Request): """Create a new unplanted or idea plant. No planting_id — that's the point.""" form = await request.form() status = (form.get("status") or "unplanted").strip() if status not in ("unplanted", "idea"): raise HTTPException(400, "status must be 'unplanted' or 'idea'") species_id = _parse_optional_int(form.get("species_id", "")) variety_id = _parse_optional_int(form.get("variety_id", "")) qty = (form.get("quantity") or "").strip() or "1" notes = (form.get("notes") or "").strip() source = (form.get("source") or "").strip() if not species_id and not variety_id: raise HTTPException(400, "species or variety required") with db() as con: if variety_id and species_id: v = con.execute("SELECT species_id FROM varieties WHERE id = ?", (variety_id,)).fetchone() if v and v["species_id"] and v["species_id"] != species_id: variety_id = None cur = con.execute( """INSERT INTO plants (planting_id, species_id, variety_id, area_id, quantity, source, status, notes, created_at) VALUES (NULL, ?, ?, NULL, ?, ?, ?, ?, ?)""", (species_id, variety_id, qty, source, status, notes, now()), ) new_id = cur.lastrowid log_add(con, "plant", new_id, _entity_label_safe(con, "plant", new_id)) return RedirectResponse(f"/unplanted?tab={status}", status_code=303) @app.post("/plants/{plant_id}/set-pool-status") async def plant_set_pool_status(plant_id: int, request: Request): """Flip an unplanted ↔ idea plant. Used by the small toggle buttons on /unplanted (e.g., "Promote to unplanted"). Refuses to operate on planted or removed plants — those have their own lifecycle endpoints.""" form = await request.form() new_status = (form.get("status") or "").strip() next_url = (form.get("next") or "").strip() if new_status not in ("idea", "unplanted"): raise HTTPException(400, "status must be 'idea' or 'unplanted'") with db() as con: prev = con.execute("SELECT status FROM plants WHERE id = ?", (plant_id,)).fetchone() if not prev: raise HTTPException(404) old = prev[0] or "" if old not in ("idea", "unplanted"): raise HTTPException(400, "this endpoint only flips idea<->unplanted; this plant is " + (old or "(empty)")) if old != new_status: con.execute("UPDATE plants SET status = ? WHERE id = ?", (new_status, plant_id)) label = _entity_label_safe(con, "plant", plant_id) log_edit(con, "plant", plant_id, "status", old, new_status, label=label) return RedirectResponse(safe_next(next_url) if next_url else f"/plants/{plant_id}", status_code=303) @app.post("/plants/{plant_id}/mark-planted") async def plant_mark_planted(plant_id: int, request: Request): """Convert an unplanted/idea plant to status='planted'. Body: planting_id (existing) OR planting_name + year_id (new planting). Sets status='planted', planting_id=. Keeps area_id+positions so the placement transfers from a "plan" to a real plant. """ form = await request.form() next_url = (form.get("next") or "").strip() existing_pid = _parse_optional_int(form.get("planting_id", "")) new_name = (form.get("planting_name") or "").strip() new_year_id = _parse_optional_int(form.get("year_id", "")) with db() as con: prev = con.execute("SELECT * FROM plants WHERE id = ?", (plant_id,)).fetchone() if not prev: raise HTTPException(404) if existing_pid: target_pid = existing_pid exists = con.execute("SELECT 1 FROM plantings WHERE id = ?", (target_pid,)).fetchone() if not exists: raise HTTPException(400, "planting not found") elif new_name: cur = con.execute( """INSERT INTO plantings (name, year_id, source, status, notes, created_at) VALUES (?, ?, '', 'planted', '', ?)""", (new_name, new_year_id, now()), ) target_pid = cur.lastrowid log_add(con, "planting", target_pid, new_name) else: raise HTTPException(400, "planting_id or planting_name required") con.execute( "UPDATE plants SET planting_id = ?, status = 'planted' WHERE id = ?", (target_pid, plant_id), ) label = _entity_label_safe(con, "plant", plant_id) log_edit(con, "plant", plant_id, "status", prev["status"], "planted", label=label) log_edit(con, "plant", plant_id, "planting_id", prev["planting_id"], target_pid, label=label) return RedirectResponse(safe_next(next_url) if next_url else f"/plants/{plant_id}", status_code=303) @app.get("/plantings/new", response_class=HTMLResponse) def planting_new(request: Request, area: Optional[int] = None): current_year = datetime.utcnow().year with db() as con: default_year_id = get_or_create_year(con, current_year) all_species = load_species(con) all_varieties = load_varieties(con) all_years = load_years(con) area_rows = load_areas(con) _, by_id = build_tree(area_rows) area_options = flattened_area_options(by_id) initial_plants = [] if area is not None and area in by_id: initial_plants = [{ "area_id": area, "area_path": area_path_name(area, by_id), "species_id": None, "variety_id": None, "quantity": 1, "notes": "", }] return templates.TemplateResponse( "planting_form.html", { "request": request, "planting": None, "plants": initial_plants, "species_list": all_species, "variety_list": all_varieties, "year_list": all_years, "area_options": area_options, "default_year_id": default_year_id, }, ) @app.post("/plantings") async def planting_create(request: Request): form = await request.form() name = form.get("name", "").strip() if not name: raise HTTPException(400, "Name required") source = form.get("source", "").strip() notes = form.get("notes", "").strip() status = form.get("status", "planned").strip() if status not in PLANTING_STATUSES: status = "planned" # Handle year: can be existing id or new year value year_id = None year_select = form.get("year_id", "").strip() new_year_value = form.get("new_year", "").strip() with db() as con: if year_select == "new" and new_year_value: year_id = get_or_create_year(con, int(new_year_value)) elif year_select: year_id = int(year_select) cur = con.execute( """INSERT INTO plantings (name, year_id, source, status, notes, created_at) VALUES (?, ?, ?, ?, ?, ?)""", (name, year_id, source, status, notes, now()), ) planting_id = cur.lastrowid log_add(con, "planting", planting_id, name) # Parse plant rows plant_count = int(form.get("plant_count", "0")) for i in range(plant_count): qty = form.get(f"plant_qty_{i}", "").strip() species_id = _parse_optional_int(form.get(f"plant_species_{i}", "")) variety_id = _parse_optional_int(form.get(f"plant_variety_{i}", "")) area_id = _parse_optional_int(form.get(f"plant_area_{i}", "")) plant_source = form.get(f"plant_source_{i}", "").strip() plant_notes = form.get(f"plant_notes_{i}", "").strip() if not qty and not species_id and not variety_id and not area_id and not plant_notes: continue # skip empty rows pcur = con.execute( """INSERT INTO plants (planting_id, species_id, variety_id, area_id, quantity, source, status, notes, created_at) VALUES (?, ?, ?, ?, ?, ?, '', ?, ?)""", (planting_id, species_id, variety_id, area_id, qty, plant_source, plant_notes, now()), ) new_pid = pcur.lastrowid plabel = _entity_label_safe(con, "plant", new_pid) log_add(con, "plant", new_pid, plabel) return RedirectResponse(f"/plantings/{planting_id}", status_code=303) @app.get("/plantings/{planting_id}", response_class=HTMLResponse) def planting_detail( planting_id: int, request: Request, children: int = Query(1), related: int = Query(0), archived: int = Query(0), ): include_children = bool(children) include_related = bool(related) show_archived = bool(archived) with db() as con: row = con.execute( """SELECT p.*, y.year AS year_value FROM plantings p LEFT JOIN years y ON y.id = p.year_id WHERE p.id = ?""", (planting_id,), ).fetchone() if not row: raise HTTPException(404) if row["is_archived"] and not show_archived: return RedirectResponse(f"/plantings/{planting_id}?archived=1", status_code=303) plants = load_plants_for_planting( con, planting_id, include_archived=show_archived ) area_rows = load_areas(con, include_archived=show_archived) _, by_id = build_tree(area_rows) comments = load_comments_expanded( con, "planting", planting_id, include_children=include_children, include_related=include_related, by_id=by_id, include_archived=show_archived, ) all_species = load_species(con, include_archived=show_archived) all_varieties = load_varieties(con, include_archived=show_archived) all_years = load_years(con, include_archived=show_archived) area_options = flattened_area_options(by_id) # Enrich plants with area paths total_qty = 0 for p in plants: if p["area_id"] and p["area_id"] in by_id: cp = area_compact_path(p["area_id"], by_id) p["area_parent"] = cp["parent"] p["area_name"] = cp["name"] p["area_compact"] = cp["full"] p["area_path"] = area_path_name(p["area_id"], by_id) else: p["area_parent"] = "" p["area_name"] = "" p["area_compact"] = "" p["area_path"] = "" p["display_name"] = plant_display_name(p) try: total_qty += int(p["quantity"]) except (ValueError, TypeError): total_qty += 1 return templates.TemplateResponse( "planting_detail.html", { "request": request, "planting": dict(row), "year_list": all_years, "plants": plants, "planting_comments": comments, "species_list": all_species, "variety_list": all_varieties, "area_options": area_options, "total_qty": total_qty, "group_count": len(plants), "include_children": include_children, "include_related": include_related, "supports_children": True, "show_archived": show_archived, }, ) @app.post("/plantings/{planting_id}") async def planting_update(planting_id: int, request: Request): form = await request.form() name = form.get("name", "").strip() if not name: raise HTTPException(400, "Name required") source = form.get("source", "").strip() notes = form.get("notes", "").strip() status = form.get("status", "planned").strip() if status not in PLANTING_STATUSES: status = "planned" year_id = None year_select = form.get("year_id", "").strip() new_year_value = form.get("new_year", "").strip() with db() as con: if year_select == "new" and new_year_value: year_id = get_or_create_year(con, int(new_year_value)) elif year_select: year_id = int(year_select) snapshot_overview(con, "planting", planting_id, notes) con.execute( """UPDATE plantings SET name = ?, year_id = ?, source = ?, status = ?, notes = ? WHERE id = ?""", (name, year_id, source, status, notes, planting_id), ) # Parse plant rows — track which existing plant IDs are kept plant_count = int(form.get("plant_count", "0")) seen_plant_ids = set() for i in range(plant_count): plant_id = _parse_optional_int(form.get(f"plant_id_{i}", "")) qty = form.get(f"plant_qty_{i}", "").strip() species_id = _parse_optional_int(form.get(f"plant_species_{i}", "")) variety_id = _parse_optional_int(form.get(f"plant_variety_{i}", "")) area_id = _parse_optional_int(form.get(f"plant_area_{i}", "")) plant_source = form.get(f"plant_source_{i}", "").strip() plant_notes = form.get(f"plant_notes_{i}", "").strip() if not qty and not species_id and not variety_id and not area_id and not plant_notes: continue if variety_id: v_row = con.execute( "SELECT species_id FROM varieties WHERE id = ?", (variety_id,) ).fetchone() if v_row and species_id and v_row["species_id"] != species_id: variety_id = None if plant_id: # Update existing seen_plant_ids.add(plant_id) snapshot_overview(con, "plant", plant_id, plant_notes) con.execute( """UPDATE plants SET species_id=?, variety_id=?, area_id=?, quantity=?, source=?, notes=? WHERE id=? AND planting_id=?""", (species_id, variety_id, area_id, qty, plant_source, plant_notes, plant_id, planting_id), ) else: # Insert new cur = con.execute( """INSERT INTO plants (planting_id, species_id, variety_id, area_id, quantity, source, status, notes, created_at) VALUES (?, ?, ?, ?, ?, ?, '', ?, ?)""", (planting_id, species_id, variety_id, area_id, qty, plant_source, plant_notes, now()), ) seen_plant_ids.add(cur.lastrowid) # Delete plant rows that were removed from the form existing = con.execute( "SELECT id FROM plants WHERE planting_id = ?", (planting_id,) ).fetchall() for row in existing: if row["id"] not in seen_plant_ids: delete_target_refs(con, "plant", row["id"]) con.execute("DELETE FROM plants WHERE id = ?", (row["id"],)) return RedirectResponse(f"/plantings/{planting_id}", status_code=303) @app.post("/plantings/{planting_id}/delete") def planting_delete(planting_id: int): with db() as con: label = _entity_label_safe(con, "planting", planting_id) plant_rows = con.execute( "SELECT id FROM plants WHERE planting_id = ?", (planting_id,) ).fetchall() for r in plant_rows: plabel = _entity_label_safe(con, "plant", r[0]) delete_target_refs(con, "plant", r[0]) log_delete(con, "plant", r[0], plabel + " (cascade from planting delete)") delete_target_refs(con, "planting", planting_id) con.execute("DELETE FROM plantings WHERE id = ?", (planting_id,)) log_delete(con, "planting", planting_id, label) return RedirectResponse("/plantings", status_code=303) # --- Planting status update (inline from planting detail/list) --- @app.post("/plantings/{planting_id}/status") def planting_status_update(planting_id: int, status: str = Form(...), next: str = Form("")): status = status.strip() if status not in PLANTING_STATUSES: raise HTTPException(400, "invalid status") with db() as con: prev = con.execute( "SELECT status FROM plantings WHERE id = ?", (planting_id,) ).fetchone() prev_status = prev[0] if prev else None con.execute("UPDATE plantings SET status = ? WHERE id = ?", (status, planting_id)) if prev_status != status: label = _entity_label_safe(con, "planting", planting_id) log_edit(con, "planting", planting_id, "status", prev_status, status, label=label) return RedirectResponse(safe_next(next) if next else f"/plantings/{planting_id}", status_code=303) # --- Plant inline CRUD (from planting detail) --- @app.post("/plantings/{planting_id}/plants") async def plant_create(planting_id: int, request: Request): form = await request.form() qty = form.get("quantity", "").strip() species_id = _parse_optional_int(form.get("species_id", "")) variety_id = _parse_optional_int(form.get("variety_id", "")) area_id = _parse_optional_int(form.get("area_id", "")) source = form.get("source", "").strip() notes = form.get("notes", "").strip() next_url = form.get("next", "") with db() as con: exists = con.execute( "SELECT 1 FROM plantings WHERE id = ?", (planting_id,) ).fetchone() if not exists: raise HTTPException(404) if variety_id and species_id: v_row = con.execute( "SELECT species_id FROM varieties WHERE id = ?", (variety_id,) ).fetchone() if v_row and v_row["species_id"] and v_row["species_id"] != species_id: variety_id = None cur = con.execute( """INSERT INTO plants (planting_id, species_id, variety_id, area_id, quantity, source, status, notes, created_at) VALUES (?, ?, ?, ?, ?, ?, '', ?, ?)""", (planting_id, species_id, variety_id, area_id, qty, source, notes, now()), ) new_pid = cur.lastrowid log_add(con, "plant", new_pid, _entity_label_safe(con, "plant", new_pid)) return RedirectResponse( safe_next(next_url) if next_url else f"/plantings/{planting_id}", status_code=303, ) @app.post("/plants/{plant_id}") async def plant_update(plant_id: int, request: Request): form = await request.form() qty = form.get("quantity", "").strip() species_id = _parse_optional_int(form.get("species_id", "")) variety_id = _parse_optional_int(form.get("variety_id", "")) area_id = _parse_optional_int(form.get("area_id", "")) source = form.get("source", "").strip() notes = form.get("notes", "").strip() next_url = form.get("next", "") with db() as con: prev = con.execute( "SELECT * FROM plants WHERE id = ?", (plant_id,) ).fetchone() if not prev: raise HTTPException(404) if variety_id and species_id: v_row = con.execute( "SELECT species_id FROM varieties WHERE id = ?", (variety_id,) ).fetchone() if v_row and v_row["species_id"] and v_row["species_id"] != species_id: variety_id = None snapshot_overview(con, "plant", plant_id, notes) # When a plant group's quantity drops to zero, mark it removed and detach # it from its area so it no longer counts toward area totals or species pills. new_status = prev["status"] try: qty_int = int(qty) if qty else None except ValueError: qty_int = None if qty_int is not None and qty_int <= 0: new_status = "removed" area_id = None con.execute( """UPDATE plants SET quantity=?, species_id=?, variety_id=?, area_id=?, source=?, notes=?, status=? WHERE id=?""", (qty, species_id, variety_id, area_id, source, notes, new_status, plant_id), ) # Per-field log entries for whichever fields actually changed. new_vals = {"quantity": qty, "species_id": species_id, "variety_id": variety_id, "area_id": area_id, "source": source, "notes": notes, "status": new_status} label = _entity_label_safe(con, "plant", plant_id) for fld, new_val in new_vals.items(): old_val = prev[fld] if fld in prev.keys() else None if (old_val or "") != (new_val or ""): log_edit(con, "plant", plant_id, fld, old_val, new_val, label=label) pid = prev["planting_id"] return RedirectResponse( safe_next(next_url) if next_url else f"/plantings/{pid}", status_code=303, ) # --- Plant detail page (individual plant group) --- @app.get("/plants/{plant_id}", response_class=HTMLResponse) def plant_detail( plant_id: int, request: Request, related: int = Query(0), archived: int = Query(0), ): include_related = bool(related) show_archived = bool(archived) with db() as con: row = con.execute( """SELECT p.*, s.name AS species_name, v.name AS variety_name, pl.name AS planting_name, pl.id AS planting_id_, y.year AS year_value, y.id AS year_id_ FROM plants p LEFT JOIN species s ON s.id = p.species_id LEFT JOIN varieties v ON v.id = p.variety_id LEFT JOIN plantings pl ON pl.id = p.planting_id LEFT JOIN years y ON y.id = pl.year_id WHERE p.id = ?""", (plant_id,), ).fetchone() if not row: raise HTTPException(404) if row["is_archived"] and not show_archived: return RedirectResponse(f"/plants/{plant_id}?archived=1", status_code=303) area_rows = load_areas(con, include_archived=show_archived) _, by_id = build_tree(area_rows) comments = load_comments_expanded( con, "plant", plant_id, include_children=False, include_related=include_related, by_id=by_id, include_archived=show_archived, ) area_info = None if row["area_id"] and row["area_id"] in by_id: area_info = { "id": row["area_id"], "path": area_path_name(row["area_id"], by_id), "short": area_compact_path(row["area_id"], by_id)["full"], } plant = dict(row) return templates.TemplateResponse( "plant_detail.html", { "request": request, "plant": plant, "area_info": area_info, "plant_comments": comments, "include_children": False, "include_related": include_related, "supports_children": False, "show_archived": show_archived, }, ) # --- Plant status update (inline from planting detail) --- @app.post("/plants/{plant_id}/status") def plant_status_update(plant_id: int, status: str = Form(""), next: str = Form("")): new_status = status.strip() with db() as con: prev = con.execute( "SELECT status FROM plants WHERE id = ?", (plant_id,) ).fetchone() prev_status = prev[0] if prev else None con.execute("UPDATE plants SET status = ? WHERE id = ?", (new_status, plant_id)) if (prev_status or "") != new_status: label = _entity_label_safe(con, "plant", plant_id) log_edit(con, "plant", plant_id, "status", prev_status, new_status, label=label) return RedirectResponse(safe_next(next) if next else "/plantings", status_code=303) @app.post("/api/areas/{area_id}/parent-position") async def api_area_set_parent_position(area_id: int, request: Request): """Save a child area's center position within its parent's natural- orientation rectangle. Body: JSON {"x": float, "y": float} in feet, rect-local. Coordinates clamp to the parent's dimensions.""" if not verify_session(request.cookies.get(COOKIE_NAME)): raise HTTPException(401, "auth required") try: payload = await request.json() x = float(payload.get("x")) y = float(payload.get("y")) except (ValueError, TypeError): raise HTTPException(400, "expected JSON {x: float, y: float}") with db() as con: row = con.execute( """SELECT a.parent_id, p.length_ft, p.width_ft FROM areas a LEFT JOIN areas p ON p.id = a.parent_id WHERE a.id = ?""", (area_id,), ).fetchone() if not row: raise HTTPException(404, "area not found") if not row["parent_id"]: raise HTTPException(400, "area has no parent — nothing to position within") parent_L = float(row["length_ft"] or 0) or 1.0 parent_W = float(row["width_ft"] or 0) or 1.0 x = max(0.0, min(parent_L, x)) y = max(0.0, min(parent_W, y)) con.execute( "UPDATE areas SET parent_pos_x = ?, parent_pos_y = ? WHERE id = ?", (round(x, 3), round(y, 3), area_id), ) return {"ok": True, "id": area_id, "x": round(x, 3), "y": round(y, 3)} @app.post("/api/plants/{plant_id}/attach-to-area") async def api_plant_attach_to_area(plant_id: int, request: Request): """Attach part of an unplanted/idea plant_group to an area. Two modes depending on whether the source plant is already in the target area: A) source.area_id == new_area_id: append a position (no split). Used when re-positioning within the current area's pool tile (rare — the Inventory tab usually handles this). B) source.area_id != new_area_id (the common pool→canvas flow): SPLIT. Peel off `icon_plants` plants from the source group into a new (or merged) plant_group in the target area, with one new position. The source group's qty decrements by `icon_plants`; if it reaches 0 the source row is deleted. If a plant_group already exists in the target area with matching species/variety/planting/status, we merge into it (bump qty + append position) rather than creating a tiny new row each time. Body JSON: {area_id: int, position: {x: float, y: float}, icon_plants: int}. """ if not verify_session(request.cookies.get(COOKIE_NAME)): raise HTTPException(401, "auth required") try: payload = await request.json() except Exception: raise HTTPException(400, "invalid JSON body") new_area_id = payload.get("area_id") if not isinstance(new_area_id, int): raise HTTPException(400, "area_id required (int)") pos = payload.get("position") or {} try: icon_plants = max(1, int(payload.get("icon_plants") or 1)) except (TypeError, ValueError): icon_plants = 1 with db() as con: src = con.execute("SELECT * FROM plants WHERE id = ?", (plant_id,)).fetchone() if not src: raise HTTPException(404, "plant not found") area_row = con.execute( "SELECT length_ft, width_ft FROM areas WHERE id = ?", (new_area_id,) ).fetchone() if not area_row: raise HTTPException(400, "area not found") L = float(area_row["length_ft"] or 0) or 1.0 W = float(area_row["width_ft"] or 0) or 1.0 # Validate + clamp the new position. try: x = max(0.0, min(L, float(pos.get("x")))) y = max(0.0, min(W, float(pos.get("y")))) except (TypeError, ValueError): raise HTTPException(400, "position required (x, y)") new_pos = {"x": round(x, 3), "y": round(y, 3)} try: src_qty = int((src["quantity"] or "0").strip()) except (TypeError, ValueError): src_qty = 0 # Mode A: same area — just append the position. if src["area_id"] == new_area_id: try: positions = json.loads(src["positions"] or "[]") if not isinstance(positions, list): positions = [] except (ValueError, TypeError): positions = [] positions.append(new_pos) con.execute( "UPDATE plants SET positions = ? WHERE id = ?", (json.dumps(positions), plant_id), ) return {"ok": True, "id": plant_id, "split": False, "saved": len(positions)} # Mode B: cross-area split. Cap how many we peel off. peel = min(icon_plants, max(0, src_qty)) if peel < 1: raise HTTPException(400, "source group has no plants left to place") # Look for an existing target group to merge into (same species, # variety, planting, status — all coalesced so NULL matches NULL). target = con.execute( """SELECT * FROM plants WHERE area_id = ? AND status = ? AND COALESCE(species_id, -1) = COALESCE(?, -1) AND COALESCE(variety_id, -1) = COALESCE(?, -1) AND COALESCE(planting_id, -1) = COALESCE(?, -1) AND COALESCE(is_archived, 0) = 0 LIMIT 1""", (new_area_id, src["status"], src["species_id"], src["variety_id"], src["planting_id"]), ).fetchone() if target: try: tpositions = json.loads(target["positions"] or "[]") if not isinstance(tpositions, list): tpositions = [] except (ValueError, TypeError): tpositions = [] tpositions.append(new_pos) try: t_qty = int((target["quantity"] or "0").strip()) except (TypeError, ValueError): t_qty = 0 con.execute( "UPDATE plants SET quantity = ?, positions = ? WHERE id = ?", (str(t_qty + peel), json.dumps(tpositions), target["id"]), ) target_id = target["id"] else: cur = con.execute( """INSERT INTO plants (planting_id, species_id, variety_id, area_id, quantity, source, status, notes, created_at, featured_image_path, positions, is_archived) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, '', ?, 0)""", (src["planting_id"], src["species_id"], src["variety_id"], new_area_id, str(peel), src["source"] or "", src["status"], src["notes"] or "", now(), json.dumps([new_pos])), ) target_id = cur.lastrowid # Decrement source. If exhausted, delete the row (no plants left). remaining = src_qty - peel if remaining <= 0: con.execute("DELETE FROM plants WHERE id = ?", (plant_id,)) else: con.execute( "UPDATE plants SET quantity = ? WHERE id = ?", (str(remaining), plant_id), ) return {"ok": True, "split": True, "source_id": plant_id, "target_id": target_id, "peeled": peel, "remaining": remaining} @app.post("/api/plants/{plant_id}/detach-from-area") async def api_plant_detach_from_area(plant_id: int, request: Request): """Clear a plant's area_id and positions. Used in layout mode to push an unplanted/idea plant back to the global Unplanted/Ideas pile. For unplanted/idea plants we also auto-merge with any existing unattached plant_group of the same species/variety/status/planting — so e.g. a "4× Marigold" detach combined with an already-floating "1× Marigold" becomes a single "5× Marigold" row, instead of accumulating siblings. Planted plants are NOT auto-merged (their planting identity matters). """ if not verify_session(request.cookies.get(COOKIE_NAME)): raise HTTPException(401, "auth required") with db() as con: src = con.execute("SELECT * FROM plants WHERE id = ?", (plant_id,)).fetchone() if not src: raise HTTPException(404, "plant not found") con.execute( "UPDATE plants SET area_id = NULL, positions = '' WHERE id = ?", (plant_id,), ) merged_into = None if src["status"] in ("unplanted", "idea"): target = con.execute( """SELECT * FROM plants WHERE id != ? AND area_id IS NULL AND COALESCE(is_archived, 0) = 0 AND status = ? AND COALESCE(species_id, -1) = COALESCE(?, -1) AND COALESCE(variety_id, -1) = COALESCE(?, -1) AND COALESCE(planting_id, -1) = COALESCE(?, -1) ORDER BY id LIMIT 1""", (plant_id, src["status"], src["species_id"], src["variety_id"], src["planting_id"]), ).fetchone() if target: try: src_qty = int((src["quantity"] or "0").strip()) except (TypeError, ValueError): src_qty = 0 try: t_qty = int((target["quantity"] or "0").strip()) except (TypeError, ValueError): t_qty = 0 con.execute( "UPDATE plants SET quantity = ? WHERE id = ?", (str(t_qty + src_qty), target["id"]), ) con.execute("DELETE FROM plants WHERE id = ?", (plant_id,)) merged_into = target["id"] return {"ok": True, "id": plant_id, "merged_into": merged_into} @app.post("/api/plants/{plant_id}/positions") async def api_plant_set_positions(plant_id: int, request: Request): """Save the per-plant icon positions from the area-sketch drag-drop UI. Body: JSON {"positions": [{"x": float, "y": float}, ...]}. Coordinates are in feet, rect-local (0..length_ft × 0..width_ft). Length should match the plant_group's quantity; we accept up to that many. """ if not verify_session(request.cookies.get(COOKIE_NAME)): raise HTTPException(401, "auth required") try: payload = await request.json() except Exception: raise HTTPException(400, "invalid JSON body") raw_positions = payload.get("positions") if not isinstance(raw_positions, list): raise HTTPException(400, "expected 'positions' to be a list") cleaned = [] for p in raw_positions: if not isinstance(p, dict): continue try: x = float(p.get("x")) y = float(p.get("y")) except (TypeError, ValueError): continue cleaned.append({"x": round(x, 3), "y": round(y, 3)}) with db() as con: row = con.execute( """SELECT p.id, p.quantity, a.length_ft, a.width_ft FROM plants p LEFT JOIN areas a ON a.id = p.area_id WHERE p.id = ?""", (plant_id,), ).fetchone() if not row: raise HTTPException(404, "plant not found") # Trim/clamp to the area's natural bounds + the group's quantity. try: qty = max(1, int((row["quantity"] or "1").strip())) except (TypeError, ValueError): qty = 1 L = float(row["length_ft"] or 0) or 1.0 W = float(row["width_ft"] or 0) or 1.0 for p in cleaned: p["x"] = max(0.0, min(L, p["x"])) p["y"] = max(0.0, min(W, p["y"])) cleaned = cleaned[:qty] con.execute( "UPDATE plants SET positions = ? WHERE id = ?", (json.dumps(cleaned), plant_id), ) return {"ok": True, "id": plant_id, "saved": len(cleaned)} @app.post("/plants/{plant_id}/delete") def plant_delete(plant_id: int, next: str = Form("")): with db() as con: label = _entity_label_safe(con, "plant", plant_id) delete_target_refs(con, "plant", plant_id) con.execute("DELETE FROM plants WHERE id = ?", (plant_id,)) log_delete(con, "plant", plant_id, label) return RedirectResponse(safe_next(next) if next else "/plantings", status_code=303) @app.post("/plants/{plant_id}/move") def plant_move( plant_id: int, target_area_id: int = Form(...), move_qty: int = Form(...), next: str = Form(""), ): """Move some or all of a plant group to a different area. If `move_qty` equals the plant group's current quantity, the row is updated in place. Otherwise the row is split: the original row is reduced by `move_qty`, and a new row is inserted in the same planting with the moved quantity at the target area. """ if move_qty <= 0: raise HTTPException(400, "move_qty must be positive") with db() as con: row = con.execute( "SELECT * FROM plants WHERE id = ?", (plant_id,) ).fetchone() if not row: raise HTTPException(404) area_row = con.execute( "SELECT id FROM areas WHERE id = ?", (target_area_id,) ).fetchone() if not area_row: raise HTTPException(400, "Target area not found") try: current_qty = int((row["quantity"] or "").strip()) except ValueError: current_qty = 0 if current_qty <= 0: raise HTTPException(400, "Can't move plants with a non-numeric quantity") if move_qty > current_qty: raise HTTPException(400, "move_qty exceeds current quantity") label = _entity_label_safe(con, "plant", plant_id) prev_area = row["area_id"] if move_qty == current_qty: con.execute( "UPDATE plants SET area_id = ? WHERE id = ?", (target_area_id, plant_id), ) log_edit( con, "plant", plant_id, "area_id", prev_area, target_area_id, label=f"{label} (moved all {move_qty})", ) else: con.execute( "UPDATE plants SET quantity = ? WHERE id = ?", (str(current_qty - move_qty), plant_id), ) cur = con.execute( """INSERT INTO plants (planting_id, species_id, variety_id, area_id, quantity, source, status, notes, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, '', ?)""", ( row["planting_id"], row["species_id"], row["variety_id"], target_area_id, str(move_qty), row["source"] or "", row["status"] or "", now(), ), ) split_pid = cur.lastrowid log_activity( con, "edit", "plant", plant_id, f"{label}: split {move_qty} of {current_qty} to area #{target_area_id}", field="quantity", old_value=current_qty, new_value=current_qty - move_qty, ) log_add( con, "plant", split_pid, _entity_label_safe(con, "plant", split_pid) + f" (split from #{plant_id})", ) return RedirectResponse(safe_next(next) if next else f"/areas/{target_area_id}", status_code=303) # --- Inventory route --- @app.get("/inventory", response_class=HTMLResponse) def inventory(request: Request, archived: int = Query(0)): show_archived = bool(archived) with db() as con: supplies = load_supplies(con, include_archived=show_archived) supplies_by_category: dict = {} for s in supplies: cat = s["category"] or "other" supplies_by_category.setdefault(cat, []).append(s) shopping = [s for s in supplies if s.get("needed")] return templates.TemplateResponse( "inventory.html", { "request": request, "supplies": supplies, "supplies_by_category": supplies_by_category, "shopping": shopping, "show_archived": show_archived, }, ) # --- Supply routes --- @app.get("/supplies/new", response_class=HTMLResponse) def supply_new(request: Request): return templates.TemplateResponse( "supply_form.html", {"request": request, "supply": None} ) @app.post("/supplies") def supply_create( name: str = Form(...), category: str = Form(""), quantity_on_hand: str = Form(""), quantity_needed: str = Form(""), needed: str = Form(""), notes: str = Form(""), ): with db() as con: cur = con.execute( """INSERT INTO supplies (name, category, amount, quantity_on_hand, quantity_needed, needed, notes, created_at) VALUES (?, ?, '', ?, ?, ?, ?, ?)""", ( name.strip(), category.strip(), quantity_on_hand.strip(), quantity_needed.strip(), 1 if needed else 0, notes.strip(), now(), ), ) log_add(con, "supply", cur.lastrowid, name.strip()) return RedirectResponse("/inventory", status_code=303) @app.get("/supplies/{supply_id}/edit", response_class=HTMLResponse) def supply_edit(supply_id: int, request: Request): with db() as con: row = con.execute( "SELECT * FROM supplies WHERE id = ?", (supply_id,) ).fetchone() if not row: raise HTTPException(404) return templates.TemplateResponse( "supply_form.html", {"request": request, "supply": dict(row)} ) @app.post("/supplies/{supply_id}") def supply_update( supply_id: int, name: str = Form(...), category: str = Form(""), quantity_on_hand: str = Form(""), quantity_needed: str = Form(""), needed: str = Form(""), notes: str = Form(""), ): with db() as con: con.execute( """UPDATE supplies SET name=?, category=?, quantity_on_hand=?, quantity_needed=?, needed=?, notes=? WHERE id=?""", ( name.strip(), category.strip(), quantity_on_hand.strip(), quantity_needed.strip(), 1 if needed else 0, notes.strip(), supply_id, ), ) return RedirectResponse("/inventory", status_code=303) @app.post("/supplies/{supply_id}/delete") def supply_delete(supply_id: int): with db() as con: label = _entity_label_safe(con, "supply", supply_id) con.execute("DELETE FROM supplies WHERE id = ?", (supply_id,)) log_delete(con, "supply", supply_id, label) return RedirectResponse("/inventory", status_code=303) # --- Station routes --- @app.get("/stations", response_class=HTMLResponse) def station_list(request: Request, archived: int = Query(0)): show_archived = bool(archived) with db() as con: stations = load_stations(con, include_archived=show_archived) counts = {} for s in stations: counts[s["id"]] = con.execute( "SELECT COUNT(*) FROM area_stations WHERE station_id = ?", (s["id"],), ).fetchone()[0] return templates.TemplateResponse( "station_list.html", { "request": request, "stations": stations, "counts": counts, "show_archived": show_archived, }, ) @app.get("/stations/new", response_class=HTMLResponse) def station_new(request: Request): with db() as con: area_rows = load_areas(con) areas_tree, _ = build_tree(area_rows) return templates.TemplateResponse( "station_form.html", { "request": request, "station": None, "areas_tree": areas_tree, "selected_area_ids": set(), }, ) @app.post("/stations") async def station_create(request: Request): form = await request.form() name = form.get("name", "").strip() notes = form.get("notes", "").strip() current_schedule = form.get("current_schedule", "").strip() area_ids = [int(v) for v in form.getlist("area_ids")] if not name: raise HTTPException(400, "Name required") with db() as con: cur = con.execute( """INSERT INTO watering_stations (name, notes, current_schedule, created_at) VALUES (?, ?, ?, ?)""", (name, notes, current_schedule, now()), ) sid = cur.lastrowid for aid in area_ids: con.execute( "INSERT OR IGNORE INTO area_stations (area_id, station_id) VALUES (?, ?)", (aid, sid), ) log_add(con, "station", sid, name) return RedirectResponse(f"/stations/{sid}", status_code=303) @app.get("/stations/{station_id}", response_class=HTMLResponse) def station_detail( station_id: int, request: Request, related: int = Query(0), archived: int = Query(0), ): include_related = bool(related) show_archived = bool(archived) with db() as con: station_row = con.execute( "SELECT * FROM watering_stations WHERE id = ?", (station_id,) ).fetchone() if not station_row: raise HTTPException(404) if station_row["is_archived"] and not show_archived: return RedirectResponse(f"/stations/{station_id}?archived=1", status_code=303) area_rows = load_areas(con, include_archived=show_archived) station_areas = load_areas_for_station( con, station_id, include_archived=show_archived ) areas_tree, by_id = build_tree(area_rows) comments = load_comments_expanded( con, "station", station_id, include_children=False, include_related=include_related, by_id=by_id, include_archived=show_archived, ) station_area_paths = [ (a["id"], area_path_name(a["id"], by_id)) for a in station_areas ] station_area_paths.sort(key=lambda x: x[1].lower()) selected_area_ids = {a["id"] for a in station_areas} return templates.TemplateResponse( "station_detail.html", { "request": request, "station": dict(station_row), "station_areas": station_area_paths, "station_comments": comments, "include_children": False, "include_related": include_related, "supports_children": False, "show_archived": show_archived, "areas_tree": areas_tree, "selected_area_ids": selected_area_ids, }, ) @app.post("/stations/{station_id}") async def station_update(station_id: int, request: Request): form = await request.form() name = form.get("name", "").strip() notes = form.get("notes", "").strip() current_schedule = form.get("current_schedule", "").strip() area_ids = [int(v) for v in form.getlist("area_ids")] if not name: raise HTTPException(400, "Name required") with db() as con: con.execute( """UPDATE watering_stations SET name = ?, notes = ?, current_schedule = ? WHERE id = ?""", (name, notes, current_schedule, station_id), ) con.execute( "DELETE FROM area_stations WHERE station_id = ?", (station_id,) ) for aid in area_ids: con.execute( "INSERT OR IGNORE INTO area_stations (area_id, station_id) VALUES (?, ?)", (aid, station_id), ) return RedirectResponse(f"/stations/{station_id}", status_code=303) @app.post("/stations/{station_id}/delete") def station_delete(station_id: int): with db() as con: label = _entity_label_safe(con, "station", station_id) con.execute( "DELETE FROM watering_stations WHERE id = ?", (station_id,) ) log_delete(con, "station", station_id, label) return RedirectResponse("/stations", status_code=303) # --- Catalog routes (species / varieties) --- @app.get("/catalog", response_class=HTMLResponse) def catalog_index(request: Request): with db() as con: s_count = con.execute("SELECT COUNT(*) FROM species").fetchone()[0] v_count = con.execute("SELECT COUNT(*) FROM varieties").fetchone()[0] return templates.TemplateResponse( "catalog_index.html", { "request": request, "species_count": s_count, "variety_count": v_count, }, ) # Species @app.get("/species", response_class=HTMLResponse) def species_list( request: Request, archived: int = Query(0), type: str = Query(""), ): show_archived = bool(archived) arch_clause = "" if show_archived else " AND is_archived = 0" type_filter = type.strip().lower() if type.strip().lower() in SPECIES_TYPES else "" with db() as con: species = load_species(con, include_archived=show_archived) if type_filter: species = [s for s in species if (s.get("type") or "plant") == type_filter] counts = {} for s in species: counts[s["id"]] = con.execute( f"SELECT COUNT(*) FROM varieties WHERE species_id = ?{arch_clause}", (s["id"],), ).fetchone()[0] return templates.TemplateResponse( "species_list.html", { "request": request, "species": species, "counts": counts, "show_archived": show_archived, "type_filter": type_filter, }, ) @app.get("/species/new", response_class=HTMLResponse) def species_new(request: Request): return templates.TemplateResponse( "species_form.html", {"request": request, "species": None} ) def _parse_spacing(plants_per_unit_str: str, space_str: str, default_plants: int = 1, default_space: float = 1.0): """Parse spacing form fields. Returns (plants_per_unit, space_per_unit_sqft). Empty strings mean 'use default'. Invalid input falls back to defaults.""" try: ppu = int((plants_per_unit_str or "").strip()) if (plants_per_unit_str or "").strip() else default_plants except ValueError: ppu = default_plants try: sqft = float((space_str or "").strip()) if (space_str or "").strip() else default_space except ValueError: sqft = default_space return max(1, ppu), max(0.05, sqft) @app.post("/species") def species_create( name: str = Form(...), common_name: str = Form(""), description: str = Form(""), plants_per_unit: str = Form("1"), space_per_unit_sqft: str = Form("1.0"), type: str = Form("plant"), primary_function: str = Form("edible"), ): ppu, sqft = _parse_spacing(plants_per_unit, space_per_unit_sqft) type_v, fn_v = normalize_species_kind(type, primary_function) with db() as con: cur = con.execute( """INSERT INTO species (name, common_name, description, created_at, plants_per_unit, space_per_unit_sqft, type, primary_function) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", (name.strip(), common_name.strip(), description.strip(), now(), ppu, sqft, type_v, fn_v), ) new_id = cur.lastrowid log_add(con, "species", new_id, name.strip()) return RedirectResponse(f"/species/{new_id}", status_code=303) @app.get("/species/{species_id}", response_class=HTMLResponse) def species_detail( species_id: int, request: Request, children: int = Query(1), related: int = Query(0), archived: int = Query(0), ): include_children = bool(children) include_related = bool(related) show_archived = bool(archived) with db() as con: row = con.execute( "SELECT * FROM species WHERE id = ?", (species_id,) ).fetchone() if not row: raise HTTPException(404) if row["is_archived"] and not show_archived: return RedirectResponse(f"/species/{species_id}?archived=1", status_code=303) var_arch = "" if show_archived else " AND is_archived = 0" varieties = [ dict(v) for v in con.execute( f"SELECT * FROM varieties " f"WHERE species_id = ?{var_arch} ORDER BY name", (species_id,), ).fetchall() ] plant_arch = "" if show_archived else " AND is_archived = 0" plant_count = con.execute( f"SELECT COUNT(*) FROM plants WHERE species_id = ?{plant_arch}", (species_id,), ).fetchone()[0] area_rows_s = load_areas(con, include_archived=show_archived) _, by_id_s = build_tree(area_rows_s) comments = load_comments_expanded( con, "species", species_id, include_children=include_children, include_related=include_related, by_id=by_id_s, include_archived=show_archived, ) return templates.TemplateResponse( "species_detail.html", { "request": request, "species": dict(row), "varieties": varieties, "plant_count": plant_count, "species_comments": comments, "include_children": include_children, "include_related": include_related, "supports_children": True, "show_archived": show_archived, }, ) @app.post("/species/{species_id}") def species_update( species_id: int, name: str = Form(...), common_name: str = Form(""), description: str = Form(""), plants_per_unit: str = Form("1"), space_per_unit_sqft: str = Form("1.0"), type: str = Form("plant"), primary_function: str = Form("edible"), ): ppu, sqft = _parse_spacing(plants_per_unit, space_per_unit_sqft) type_v, fn_v = normalize_species_kind(type, primary_function) with db() as con: snapshot_overview(con, "species", species_id, description.strip()) con.execute( """UPDATE species SET name = ?, common_name = ?, description = ?, plants_per_unit = ?, space_per_unit_sqft = ?, type = ?, primary_function = ? WHERE id = ?""", (name.strip(), common_name.strip(), description.strip(), ppu, sqft, type_v, fn_v, species_id), ) return RedirectResponse(f"/species/{species_id}", status_code=303) @app.post("/species/{species_id}/spacing") def species_set_spacing( species_id: int, plants_per_unit: str = Form("1"), space_per_unit_sqft: str = Form("1.0"), next: str = Form(""), ): """Inline-edit endpoint — updates ONLY spacing fields (no name/overview).""" ppu, sqft = _parse_spacing(plants_per_unit, space_per_unit_sqft) with db() as con: cur = con.execute( "UPDATE species SET plants_per_unit = ?, space_per_unit_sqft = ? WHERE id = ?", (ppu, sqft, species_id), ) if cur.rowcount == 0: raise HTTPException(404) return RedirectResponse(safe_next(next) if next else f"/species/{species_id}", status_code=303) @app.post("/species/{species_id}/delete") def species_delete(species_id: int): with db() as con: label = _entity_label_safe(con, "species", species_id) delete_target_refs(con, "species", species_id) con.execute("DELETE FROM species WHERE id = ?", (species_id,)) log_delete(con, "species", species_id, label) return RedirectResponse("/species", status_code=303) # Varieties @app.get("/varieties", response_class=HTMLResponse) def variety_list(request: Request, archived: int = Query(0)): show_archived = bool(archived) with db() as con: varieties = load_varieties(con, include_archived=show_archived) return templates.TemplateResponse( "variety_list.html", {"request": request, "varieties": varieties, "show_archived": show_archived}, ) @app.get("/varieties/new", response_class=HTMLResponse) def variety_new(request: Request, species_id: Optional[int] = None): with db() as con: all_species = load_species(con) species_default = None if species_id: sp = con.execute( "SELECT plants_per_unit, space_per_unit_sqft FROM species WHERE id = ?", (species_id,), ).fetchone() if sp: species_default = dict(sp) return templates.TemplateResponse( "variety_form.html", { "request": request, "variety": None, "species_choices": species_options(all_species), "selected_species_id": species_id, "species_default": species_default, }, ) def _parse_optional_spacing(plants_str: str, space_str: str): """For variety overrides — empty input means NULL (inherit species).""" ppu = None sqft = None s_p = (plants_str or "").strip() s_s = (space_str or "").strip() if s_p: try: ppu = max(1, int(s_p)) except ValueError: ppu = None if s_s: try: sqft = max(0.05, float(s_s)) except ValueError: sqft = None return ppu, sqft @app.post("/varieties") def variety_create( name: str = Form(...), species_id: str = Form(""), description: str = Form(""), plants_per_unit: str = Form(""), space_per_unit_sqft: str = Form(""), ): sid = _parse_optional_int(species_id) ppu, sqft = _parse_optional_spacing(plants_per_unit, space_per_unit_sqft) with db() as con: cur = con.execute( """INSERT INTO varieties (name, species_id, description, created_at, plants_per_unit, space_per_unit_sqft) VALUES (?, ?, ?, ?, ?, ?)""", (name.strip(), sid, description.strip(), now(), ppu, sqft), ) new_id = cur.lastrowid log_add(con, "variety", new_id, name.strip()) return RedirectResponse(f"/varieties/{new_id}", status_code=303) @app.get("/varieties/{variety_id}", response_class=HTMLResponse) def variety_detail( variety_id: int, request: Request, related: int = Query(0), archived: int = Query(0), ): include_related = bool(related) show_archived = bool(archived) with db() as con: row = con.execute( """SELECT v.*, s.name AS species_name, s.plants_per_unit AS species_plants_per_unit, s.space_per_unit_sqft AS species_space_per_unit_sqft, s.type AS species_type, s.primary_function AS species_primary_function FROM varieties v LEFT JOIN species s ON s.id = v.species_id WHERE v.id = ?""", (variety_id,), ).fetchone() if not row: raise HTTPException(404) if row["is_archived"] and not show_archived: return RedirectResponse(f"/varieties/{variety_id}?archived=1", status_code=303) plant_arch = "" if show_archived else " AND is_archived = 0" plant_count = con.execute( f"SELECT COUNT(*) FROM plants WHERE variety_id = ?{plant_arch}", (variety_id,), ).fetchone()[0] area_rows_v = load_areas(con, include_archived=show_archived) _, by_id_v = build_tree(area_rows_v) comments = load_comments_expanded( con, "variety", variety_id, include_children=False, include_related=include_related, by_id=by_id_v, include_archived=show_archived, ) all_species = load_species(con, include_archived=show_archived) species_default = None if row["species_plants_per_unit"] is not None or row["species_space_per_unit_sqft"] is not None: species_default = { "plants_per_unit": row["species_plants_per_unit"] or 1, "space_per_unit_sqft": row["species_space_per_unit_sqft"] or 1.0, } return templates.TemplateResponse( "variety_detail.html", { "request": request, "variety": dict(row), "species_default": species_default, "species_list": all_species, "plant_count": plant_count, "variety_comments": comments, "include_children": False, "include_related": include_related, "supports_children": False, "show_archived": show_archived, }, ) @app.post("/varieties/{variety_id}") def variety_update( variety_id: int, name: str = Form(...), species_id: str = Form(""), description: str = Form(""), plants_per_unit: str = Form(""), space_per_unit_sqft: str = Form(""), ): sid = _parse_optional_int(species_id) ppu, sqft = _parse_optional_spacing(plants_per_unit, space_per_unit_sqft) with db() as con: snapshot_overview(con, "variety", variety_id, description.strip()) con.execute( """UPDATE varieties SET name = ?, species_id = ?, description = ?, plants_per_unit = ?, space_per_unit_sqft = ? WHERE id = ?""", (name.strip(), sid, description.strip(), ppu, sqft, variety_id), ) return RedirectResponse(f"/varieties/{variety_id}", status_code=303) @app.post("/varieties/{variety_id}/spacing") def variety_set_spacing( variety_id: int, plants_per_unit: str = Form(""), space_per_unit_sqft: str = Form(""), next: str = Form(""), ): """Inline-edit endpoint — updates ONLY the variety's spacing override. Empty inputs mean 'inherit from species'.""" ppu, sqft = _parse_optional_spacing(plants_per_unit, space_per_unit_sqft) with db() as con: cur = con.execute( "UPDATE varieties SET plants_per_unit = ?, space_per_unit_sqft = ? WHERE id = ?", (ppu, sqft, variety_id), ) if cur.rowcount == 0: raise HTTPException(404) return RedirectResponse(safe_next(next) if next else f"/varieties/{variety_id}", status_code=303) @app.post("/varieties/{variety_id}/delete") def variety_delete(variety_id: int): with db() as con: label = _entity_label_safe(con, "variety", variety_id) delete_target_refs(con, "variety", variety_id) con.execute("DELETE FROM varieties WHERE id = ?", (variety_id,)) log_delete(con, "variety", variety_id, label) return RedirectResponse("/varieties", status_code=303) # --- Comment routes --- @app.post("/comments") def comment_create( target_type: str = Form(...), target_id: int = Form(...), comment_date: str = Form(...), body: str = Form(""), kind: str = Form(""), transcript: str = Form(""), captured_at: str = Form(""), next: str = Form("/"), audio: Optional[UploadFile] = File(None), photos: List[UploadFile] = File(default=[]), secondary_target: List[str] = Form(default=[]), override_inference: str = Form(""), ): if target_type not in COMMENT_TARGET_TYPES: raise HTTPException(400, "invalid target type") body = (body or "").strip() transcript = (transcript or "").strip() # Save any uploaded media first; media-only notes (e.g. just a photo) are valid. audio_rel = "" if audio is not None and (audio.filename or "").strip(): audio_rel = _save_upload(audio, "audio") photo_rels = [] for ph in photos or []: if ph and (ph.filename or "").strip(): photo_rels.append(_save_upload(ph, "photo")) photo_paths_str = "|".join(photo_rels) has_media = bool(audio_rel or photo_paths_str) if not body and not has_media: return RedirectResponse(safe_next(next), status_code=303) # Parse explicit secondaries from the floating capture modal. When # `override_inference` is set, the supplied list is the final word and # auto-inference is skipped — even if the list is empty (the user # explicitly removed everything). explicit_secondaries = None if override_inference == "1": explicit_secondaries = [] for s in secondary_target: if ":" not in s: continue t_str, i_str = s.split(":", 1) if t_str not in COMMENT_TARGET_TYPES: continue try: explicit_secondaries.append((t_str, int(i_str))) except ValueError: continue with db() as con: post_comment( con, target_type, target_id, comment_date, body, kind=kind, audio_path=audio_rel or None, photo_paths=photo_paths_str, transcript=transcript, captured_at=captured_at or None, explicit_secondaries=explicit_secondaries, ) return RedirectResponse(safe_next(next), status_code=303) @app.get("/comments/{comment_id}/edit", response_class=HTMLResponse) def comment_edit(comment_id: int, request: Request, next: str = "/"): with db() as con: row = con.execute( "SELECT * FROM field_notes WHERE id = ?", (comment_id,) ).fetchone() if not row: raise HTTPException(404) area_rows = load_areas(con) _, by_id = build_tree(area_rows) targets = load_comment_targets(con, comment_id, by_id) all_species = load_species(con) all_varieties = load_varieties(con) all_plantings = load_plantings(con) all_years = load_years(con) area_options = flattened_area_options(by_id) return templates.TemplateResponse( "comment_form.html", { "request": request, "comment": dict(row), "targets": targets, "back_url": safe_next(next), "species_list": all_species, "variety_list": all_varieties, "planting_list": all_plantings, "year_list": all_years, "area_options": area_options, }, ) @app.post("/comments/{comment_id}") async def comment_update( comment_id: int, request: Request, comment_date: str = Form(...), body: str = Form(""), kind: str = Form(""), next: str = Form("/"), audio: Optional[UploadFile] = File(None), photos: List[UploadFile] = File(default=[]), remove_audio: str = Form(""), ): if kind and kind not in COMMENT_KINDS: kind = "" # FastAPI's List[Form] only captures the first value when fields share a # name; pull `remove_photos` straight from the multipart form to get all. form = await request.form() remove_photos_set = set(form.getlist("remove_photos")) with db() as con: row = con.execute( "SELECT audio_path, photo_paths FROM field_notes WHERE id = ?", (comment_id,), ).fetchone() if not row: raise HTTPException(404, "note not found") cur_audio = row["audio_path"] cur_photos = [p for p in (row["photo_paths"] or "").split("|") if p] # Audio: remove (if checked or being replaced) → optionally add new. new_audio_uploaded = audio is not None and (audio.filename or "").strip() if (remove_audio or new_audio_uploaded) and cur_audio: _delete_note_file(cur_audio) cur_audio = None if new_audio_uploaded: cur_audio = _save_upload(audio, "audio") # Photos: drop any in remove_photos (and delete from disk), then append uploads. kept_photos = [] for p in cur_photos: if p in remove_photos_set: _delete_note_file(p) else: kept_photos.append(p) for ph in photos or []: if ph and (ph.filename or "").strip(): kept_photos.append(_save_upload(ph, "photo")) new_photo_str = "|".join(kept_photos) con.execute( """UPDATE field_notes SET comment_date = ?, body = ?, kind = ?, audio_path = ?, photo_paths = ? WHERE id = ?""", (comment_date, body.strip(), kind, cur_audio, new_photo_str, comment_id), ) body_short = _short(body or "(media-only)", 60) log_activity( con, "edit", "comment", comment_id, f"Edited note: {body_short}", ) return RedirectResponse(safe_next(next), status_code=303) @app.post("/comments/{comment_id}/delete") def comment_delete(comment_id: int, next: str = Form("/")): with db() as con: label = _entity_label_safe(con, "comment", comment_id) con.execute("DELETE FROM field_notes WHERE id = ?", (comment_id,)) log_delete(con, "comment", comment_id, label) return RedirectResponse(safe_next(next), status_code=303) @app.post("/comments/{comment_id}/targets") def comment_target_add( comment_id: int, target_type: str = Form(...), target_id: int = Form(...), next: str = Form("/"), ): if target_type not in COMMENT_TARGET_TYPES: raise HTTPException(400, "invalid target type") with db() as con: exists = con.execute( "SELECT 1 FROM field_notes WHERE id = ?", (comment_id,) ).fetchone() if not exists: raise HTTPException(404) con.execute( """INSERT OR IGNORE INTO field_note_targets (field_note_id, target_type, target_id, is_primary) VALUES (?, ?, ?, 0)""", (comment_id, target_type, target_id), ) return RedirectResponse(safe_next(next), status_code=303) @app.post("/comments/{comment_id}/targets/{target_type}/{target_id}/delete") def comment_target_delete( comment_id: int, target_type: str, target_id: int, next: str = Form("/"), ): with db() as con: count = con.execute( "SELECT COUNT(*) FROM field_note_targets WHERE field_note_id = ?", (comment_id,) ).fetchone()[0] if count <= 1: raise HTTPException(400, "cannot remove last target; delete comment instead") was_primary = con.execute( """SELECT is_primary FROM field_note_targets WHERE field_note_id = ? AND target_type = ? AND target_id = ?""", (comment_id, target_type, target_id), ).fetchone() if not was_primary: return RedirectResponse(safe_next(next), status_code=303) con.execute( """DELETE FROM field_note_targets WHERE field_note_id = ? AND target_type = ? AND target_id = ?""", (comment_id, target_type, target_id), ) if was_primary["is_primary"]: remaining = con.execute( """SELECT target_type, target_id FROM field_note_targets WHERE field_note_id = ? ORDER BY id LIMIT 1""", (comment_id,), ).fetchone() if remaining: con.execute( """UPDATE field_note_targets SET is_primary = 1 WHERE field_note_id = ? AND target_type = ? AND target_id = ?""", (comment_id, remaining["target_type"], remaining["target_id"]), ) # (No legacy target_type/target_id columns on field_notes; primary # is carried entirely by field_note_targets.is_primary now.) return RedirectResponse(safe_next(next), status_code=303) # --- Timeline view (chronological comments — global or per-entity) --- TIMELINE_FILTER_TYPES = {"area", "planting", "plant", "species", "variety", "station", "year"} TIMELINE_TYPES_WITH_CHILDREN = {"area", "planting", "species"} def _build_filter_options(con, by_id: dict) -> dict: """Options for the timeline filter dropdown, grouped by entity type.""" opts: dict = {"areas": [], "plantings": [], "plants": [], "species": [], "varieties": [], "stations": [], "years": []} for a in sorted(load_areas(con), key=lambda r: area_path_name(r["id"], by_id).lower()): opts["areas"].append((a["id"], area_path_name(a["id"], by_id))) for p in con.execute( "SELECT p.id, p.name, y.year FROM plantings p LEFT JOIN years y ON y.id = p.year_id ORDER BY y.year DESC, p.name" ).fetchall(): label = p["name"] + (f" ({p['year']})" if p["year"] else "") opts["plantings"].append((p["id"], label)) for r in con.execute( """SELECT pl.id, s.name AS sp, v.name AS var, a.name AS area, pt.name AS planting FROM plants pl LEFT JOIN species s ON s.id = pl.species_id LEFT JOIN varieties v ON v.id = pl.variety_id LEFT JOIN areas a ON a.id = pl.area_id LEFT JOIN plantings pt ON pt.id = pl.planting_id ORDER BY pt.name, s.name, v.name, pl.id""" ).fetchall(): bits = [] if r["sp"]: bits.append(r["sp"]) if r["var"]: bits.append(r["var"]) label = " — ".join(bits) if bits else f"plant #{r['id']}" if r["planting"]: label += f" ({r['planting']})" opts["plants"].append((r["id"], label)) for r in con.execute("SELECT id, name FROM species ORDER BY name").fetchall(): opts["species"].append((r["id"], r["name"])) for r in con.execute( """SELECT v.id, v.name, s.name AS sp FROM varieties v LEFT JOIN species s ON s.id = v.species_id ORDER BY s.name, v.name""" ).fetchall(): label = f"{r['name']} ({r['sp']})" if r["sp"] else r["name"] opts["varieties"].append((r["id"], label)) for r in con.execute("SELECT id, name FROM watering_stations ORDER BY name").fetchall(): opts["stations"].append((r["id"], r["name"])) for r in con.execute("SELECT id, year FROM years ORDER BY year DESC").fetchall(): opts["years"].append((r["id"], str(r["year"]))) return opts @app.get("/timeline", response_class=HTMLResponse) def global_timeline( request: Request, scope_type: Optional[str] = None, scope_id: Optional[int] = None, children: Optional[int] = None, related: int = 0, archived: int = Query(0), view: str = Query("notes"), ): # Normalise filter if scope_type and scope_type not in TIMELINE_FILTER_TYPES: scope_type = None scope_id = None supports_children = scope_type in TIMELINE_TYPES_WITH_CHILDREN if children is None: include_children = supports_children else: include_children = bool(children) and supports_children include_related = bool(related) show_archived = bool(archived) view = view if view in ("notes", "log") else "notes" with db() as con: area_rows = load_areas(con, include_archived=show_archived) _, by_id = build_tree(area_rows) filter_options = _build_filter_options(con, by_id) scope_label = None log_entries: list = [] if view == "log": # Activity-log view. Filter by scope when set so e.g. # /timeline?scope_type=area&scope_id=6&view=log shows only that # area's mutations. params: list = [] where = "" if scope_type and scope_id: where = "WHERE entity_type = ? AND entity_id = ?" params = [scope_type, scope_id] log_entries = [ dict(r) for r in con.execute( f"""SELECT id, created_at, category, entity_type, entity_id, summary, field, old_value, new_value FROM activity_log {where} ORDER BY id DESC LIMIT 500""", params, ).fetchall() ] # Attach entity_url for clickability where the entity still exists. for e in log_entries: if e["entity_id"]: e["entity_url"] = target_url(e["entity_type"], e["entity_id"]) else: e["entity_url"] = "" if scope_type and scope_id: scope_label = target_label(con, scope_type, scope_id, by_id) comments: list = [] if view == "log": pass elif scope_type and scope_id: primary_only_for_direct = (scope_type == "year" and not include_related) comments = load_comments_expanded( con, scope_type, scope_id, include_children=include_children, include_related=include_related, by_id=by_id, primary_only_for_direct=primary_only_for_direct, include_archived=show_archived, ) scope_label = target_label(con, scope_type, scope_id, by_id) else: # Unfiltered global timeline — load every comment-style row (those # with a body). Capture rows (raw audio/photo/text awaiting # processing) live in /capture/queue, not the comment timeline. arch_w = "" if show_archived else " AND c.is_archived = 0" rows = con.execute( f"""SELECT c.* FROM field_notes c WHERE c.body IS NOT NULL AND c.body != ''{arch_w} ORDER BY c.comment_date DESC, c.created_at DESC, c.id DESC""" ).fetchall() comments = [] for r in rows: c = _attach_media_fields(dict(r)) prim = con.execute( """SELECT target_type, target_id FROM field_note_targets WHERE field_note_id = ? AND is_primary = 1""", (c["id"],), ).fetchone() if prim: c["primary_type"] = prim["target_type"] c["primary_id"] = prim["target_id"] c["primary_label"] = target_label(con, prim["target_type"], prim["target_id"], by_id) c["primary_short_label"] = target_short_label(con, prim["target_type"], prim["target_id"], by_id) c["primary_url"] = target_url(prim["target_type"], prim["target_id"]) else: c["primary_type"] = None c["primary_id"] = None c["primary_label"] = "" c["primary_short_label"] = "" c["primary_url"] = "" others = con.execute( """SELECT target_type, target_id FROM field_note_targets WHERE field_note_id = ? AND is_primary = 0 ORDER BY id""", (c["id"],), ).fetchall() c["other_targets"] = [ { "type": t["target_type"], "id": t["target_id"], "label": target_label(con, t["target_type"], t["target_id"], by_id), "short_label": target_short_label(con, t["target_type"], t["target_id"], by_id), "url": target_url(t["target_type"], t["target_id"]), } for t in others ] c["is_primary"] = True comments.append(c) return templates.TemplateResponse( "timeline.html", { "request": request, "entity_type": scope_type, "entity_id": scope_id, "entity_label": scope_label or "All garden commentary", "comments": comments, "log_entries": log_entries, "view_mode": view, "back_url": "/", "is_global": scope_type is None, "filter_options": filter_options, "scope_type": scope_type, "scope_id": scope_id, "include_children": include_children, "include_related": include_related, "supports_children": supports_children, "show_archived": show_archived, }, ) @app.get("/timeline/{entity_type}/{entity_id}", response_class=HTMLResponse) def entity_timeline( entity_type: str, entity_id: int, request: Request, archived: int = Query(0), view: str = Query("notes"), ): if entity_type not in OVERVIEW_ENTITIES: raise HTTPException(404, "unknown entity type") show_archived = bool(archived) view = view if view in ("notes", "log") else "notes" with db() as con: area_rows = load_areas(con, include_archived=show_archived) _, by_id = build_tree(area_rows) if view == "log": comments = [] log_entries = [ dict(r) for r in con.execute( """SELECT id, created_at, category, entity_type, entity_id, summary, field, old_value, new_value FROM activity_log WHERE entity_type = ? AND entity_id = ? ORDER BY id DESC LIMIT 500""", (entity_type, entity_id), ).fetchall() ] for e in log_entries: e["entity_url"] = target_url(e["entity_type"], e["entity_id"]) else: log_entries = [] comments = load_comments( con, entity_type, entity_id, by_id, include_archived=show_archived ) label = target_label(con, entity_type, entity_id, by_id) if label.startswith(entity_type + " #"): raise HTTPException(404) back_url = target_url(entity_type, entity_id) return templates.TemplateResponse( "timeline.html", { "request": request, "entity_type": entity_type, "entity_id": entity_id, "entity_label": label, "comments": comments, "log_entries": log_entries, "view_mode": view, "back_url": back_url, "is_global": False, "show_archived": show_archived, }, ) # --- Overview history (version log per entity) --- @app.get("/overview-history/{entity_type}/{entity_id}", response_class=HTMLResponse) def overview_history(entity_type: str, entity_id: int, request: Request): if entity_type not in OVERVIEW_ENTITIES: raise HTTPException(404, "unknown entity type") with db() as con: current = get_current_overview(con, entity_type, entity_id) if current is None: raise HTTPException(404) versions = load_overview_versions(con, entity_type, entity_id) label = target_label(con, entity_type, entity_id) back_url = target_url(entity_type, entity_id) return templates.TemplateResponse( "overview_history.html", { "request": request, "entity_type": entity_type, "entity_id": entity_id, "entity_label": label, "current_content": current, "versions": versions, "back_url": back_url, }, ) # --- Artifacts (AI-generated markdown outputs) --- def _slugify(text: str) -> str: out = [] for c in text.lower().strip(): if c.isalnum(): out.append(c) elif c in " -_": out.append("-") slug = "".join(out).strip("-") while "--" in slug: slug = slug.replace("--", "-") return slug or "artifact" def _artifact_read(filename: str) -> str: path = os.path.join(ARTIFACTS_DIR, filename) if not os.path.isfile(path): return "" with open(path, "r", encoding="utf-8") as f: return f.read() def _artifact_write(filename: str, content: str): os.makedirs(ARTIFACTS_DIR, exist_ok=True) path = os.path.join(ARTIFACTS_DIR, filename) with open(path, "w", encoding="utf-8") as f: f.write(content) @app.get("/artifacts", response_class=HTMLResponse) def artifact_list(request: Request): with db() as con: rows = [ dict(r) for r in con.execute( "SELECT * FROM artifacts ORDER BY generated_at DESC, id DESC" ).fetchall() ] by_kind: dict = {} for r in rows: by_kind.setdefault(r["kind"], []).append(r) kind_order = [k for k in ARTIFACT_KINDS if k in by_kind] return templates.TemplateResponse( "artifact_list.html", { "request": request, "artifacts": rows, "by_kind": by_kind, "kind_order": kind_order, }, ) @app.get("/artifacts/{artifact_id}", response_class=HTMLResponse) def artifact_detail(artifact_id: int, request: Request): with db() as con: row = con.execute( "SELECT * FROM artifacts WHERE id = ?", (artifact_id,) ).fetchone() if not row: raise HTTPException(404) content_md = _artifact_read(row["filename"]) content_html = md.markdown( content_md, extensions=["tables", "fenced_code", "nl2br", "sane_lists"], ) if content_md else "" return templates.TemplateResponse( "artifact_detail.html", { "request": request, "artifact": dict(row), "content_md": content_md, "content_html": content_html, }, ) @app.post("/artifacts/{artifact_id}/delete") def artifact_delete(artifact_id: int): with db() as con: row = con.execute( "SELECT filename FROM artifacts WHERE id = ?", (artifact_id,) ).fetchone() if row: path = os.path.join(ARTIFACTS_DIR, row["filename"]) if os.path.isfile(path): try: os.remove(path) except OSError: pass con.execute("DELETE FROM artifacts WHERE id = ?", (artifact_id,)) return RedirectResponse("/artifacts", status_code=303) # --- Archive / Unarchive --- _ARCHIVE_FALLBACK_URLS = { "comment": "/timeline", "supply": "/inventory", } def _archive_redirect_target(entity_type: str, entity_id: int, next_url: str) -> str: """Pick a sensible redirect for an archive/unarchive POST. Prefer the `next` form value (the originating page); fall back to the entity's detail page or a list page if it has no detail URL.""" safe = safe_next(next_url) if safe and safe != "/": return safe fallback = target_url(entity_type, entity_id) if fallback != "/": return fallback return _ARCHIVE_FALLBACK_URLS.get(entity_type, "/") @app.post("/archive/{entity_type}/{entity_id}") def entity_archive( entity_type: str, entity_id: int, next: str = Form(""), ): if entity_type not in ARCHIVABLE_ENTITIES: raise HTTPException(404) with db() as con: archive_entity(con, entity_type, entity_id) return RedirectResponse( _archive_redirect_target(entity_type, entity_id, next), status_code=303 ) @app.post("/unarchive/{entity_type}/{entity_id}") def entity_unarchive( entity_type: str, entity_id: int, next: str = Form(""), ): if entity_type not in ARCHIVABLE_ENTITIES: raise HTTPException(404) with db() as con: unarchive_entity(con, entity_type, entity_id) return RedirectResponse( _archive_redirect_target(entity_type, entity_id, next), status_code=303 ) # Legacy /schedule kept as redirect to /artifacts (old table unused) # --- API --- def check_token(token: str): if not secrets.compare_digest(token or "", API_TOKEN): raise HTTPException(401, "invalid token") @app.get("/api/export") def api_export(token: str = ""): check_token(token) with db() as con: areas = [ dict(r) for r in con.execute( "SELECT * FROM areas ORDER BY COALESCE(sort_order, 999999), name" ).fetchall() ] station_areas = {} for r in con.execute( "SELECT station_id, area_id FROM area_stations" ).fetchall(): station_areas.setdefault(r["station_id"], []).append(r["area_id"]) stations = [] for s in con.execute( "SELECT * FROM watering_stations ORDER BY name" ).fetchall(): sd = dict(s) sd["area_ids"] = station_areas.get(sd["id"], []) stations.append(sd) supplies = [dict(r) for r in con.execute( "SELECT * FROM supplies ORDER BY category, name" ).fetchall()] species = load_species(con) varieties = load_varieties(con) years = load_years(con) plantings = load_plantings(con) plants = [dict(r) for r in con.execute("SELECT * FROM plants").fetchall()] all_comments = [ dict(r) for r in con.execute( "SELECT * FROM field_notes ORDER BY comment_date DESC, created_at DESC" ).fetchall() ] field_note_targets = [ dict(r) for r in con.execute( "SELECT * FROM field_note_targets ORDER BY field_note_id, id" ).fetchall() ] artifacts = [ dict(r) for r in con.execute( "SELECT * FROM artifacts ORDER BY generated_at DESC" ).fetchall() ] return { "areas": areas, "stations": stations, "supplies": supplies, "species": species, "varieties": varieties, "years": years, "plantings": plantings, "plants": plants, "comments": all_comments, "field_note_targets": field_note_targets, "artifacts": artifacts, "exported_at": now(), } @app.get("/api/areas") def api_areas(token: str = ""): """JSON list of areas for autocomplete.""" check_token(token) with db() as con: area_rows = load_areas(con) _, by_id = build_tree(area_rows) return flattened_area_options(by_id) # Endpoints powering the floating capture modal. Session-authed via the # AuthMiddleware (any authenticated request reaches here). @app.get("/api/current-year") def api_current_year(): """Resolve (creating if absent) the year row for today, used by the floating capture modal as the fallback primary target.""" today = datetime.utcnow().year with db() as con: year_id = get_or_create_year(con, today) return {"id": year_id, "year": today, "type": "year", "label": str(today)} # Map URL prefix → primary-target type. Used by api_page_context to derive # the current page's primary target from the path. Comment/edit URLs aren't # included; clicking the bubble there falls back to current year. _PAGE_TARGET_PREFIXES = { "areas": "area", "plantings": "planting", "plants": "plant", "species": "species", "varieties": "variety", "stations": "station", "years": "year", } @app.get("/api/page-context") def api_page_context(path: str = ""): """Given a frontend pathname (e.g. '/areas/7'), return the primary target that the floating capture modal should default to: a {type, id, label} dict, or {} when the page has no associated entity.""" parts = (path or "").strip("/").split("/") if len(parts) < 2: return {} type_singular = _PAGE_TARGET_PREFIXES.get(parts[0]) if not type_singular: return {} try: eid = int(parts[1]) except ValueError: return {} with db() as con: # `target_short_label` collapses the area path to just the leaf name # (or short_name) — perfect for a compact pill display. try: label = target_short_label(con, type_singular, eid) except Exception: return {} if label.startswith(f"{type_singular} #"): # target_short_label returns this format when the row doesn't exist. return {} return {"type": type_singular, "id": eid, "label": label} @app.post("/api/client-log") async def api_client_log(request: Request): """Diagnostic sink for tiny JSON beacons from client JS. Used to trace where photo-upload submits hang when the main multipart POST never arrives at the server. Payload: {"event": "save_clicked"|..., "data": {...}}. Intentionally session-gated (require_session) so it's not a public log fire-hose, but otherwise dead simple — we just print to stderr so the breadcrumbs show up in `docker logs garden-app`. Body is capped at 4KB via FastAPI's default body size; we don't need more than that. """ require_session(request) try: payload = await request.json() except Exception: payload = {"raw": "(invalid JSON body)"} event = (payload.get("event") or "?") data = payload.get("data") or {} import sys print(f"[client-log] {event}: {json.dumps(data)[:600]}", file=sys.stderr, flush=True) return {"ok": True} @app.get("/api/picker-targets") def api_picker_targets(): """All primary-target options grouped by type. Powers the swap-target picker in the floating capture modal.""" with db() as con: area_rows = load_areas(con) _, by_id = build_tree(area_rows) opts = _build_filter_options(con, by_id) return opts @app.get("/api/inferred-targets") def api_inferred_targets(type: str, id: int, date: str = ""): """Preview the secondary targets that would be auto-attached to a comment on this primary. Powers the 'will also link to:' row in the floating capture modal — the same `inferred_targets_for` rules that run at post time, but exposed read-only so the UI can show them before submit.""" if type not in COMMENT_TARGET_TYPES: raise HTTPException(404, "unknown target type") cd = (date or "").strip() or datetime.utcnow().date().isoformat() with db() as con: try: primary_label = target_short_label(con, type, id) except Exception: return {"primary_label": "", "secondaries": []} if primary_label.startswith(f"{type} #"): return {"primary_label": "", "secondaries": []} out = [] for ttype, tid in inferred_targets_for(con, type, id, cd): try: label = target_short_label(con, ttype, tid) except Exception: continue out.append({"type": ttype, "id": tid, "label": label}) return {"primary_label": primary_label, "secondaries": out} @app.post("/api/species") async def api_species_create(request: Request): """Inline species creation from planting form (session-authed, not token).""" data = await request.json() name = (data.get("name") or "").strip() if not name: raise HTTPException(400, "name required") with db() as con: existing = con.execute( "SELECT id FROM species WHERE name = ?", (name,) ).fetchone() if existing: return {"id": existing[0], "name": name} cur = con.execute( "INSERT INTO species (name, common_name, description, created_at) VALUES (?, '', '', ?)", (name, now()), ) log_add(con, "species", cur.lastrowid, name) return {"id": cur.lastrowid, "name": name} @app.post("/api/varieties") async def api_variety_create(request: Request): """Inline variety creation from planting form (session-authed, not token).""" data = await request.json() name = (data.get("name") or "").strip() species_id = data.get("species_id") if not name: raise HTTPException(400, "name required") if species_id: species_id = int(species_id) else: species_id = None with db() as con: cur = con.execute( "INSERT INTO varieties (name, species_id, description, created_at) VALUES (?, ?, '', ?)", (name, species_id, now()), ) log_add(con, "variety", cur.lastrowid, name) return {"id": cur.lastrowid, "name": name, "species_id": species_id} @app.post("/api/edit/{entity_type}/{entity_id}") async def api_edit_field(entity_type: str, entity_id: int, request: Request): """Generic per-field click-to-edit endpoint. Body: {"field": "name", "value": "..."}. Returns {"ok": true, "value": } on success. Validates that (entity_type, field) is whitelisted in EDITABLE_FIELDS, coerces the incoming value to the field's declared type, validates constraints (required, min/max, options, fk row existence), snapshots the previous overview/notes if the field has snapshot=True, and applies a single UPDATE. For species, the (type, primary_function) combo is constrained: changing `type` clamps `primary_function` to the first valid option for that type if the existing function is incompatible (mirrors `normalize_species_kind`). """ if entity_type not in EDITABLE_FIELDS: raise HTTPException(404, "unknown entity type") fields = EDITABLE_FIELDS[entity_type] table = EDITABLE_TABLES[entity_type] data = await request.json() field = (data.get("field") or "").strip() if field not in fields: raise HTTPException(400, f"field {field!r} is not editable") spec = fields[field] raw = data.get("value") # Type coercion + validation. Each branch returns the value to STORE. def coerce(spec, raw): ftype = spec["type"] if ftype in ("text", "textarea"): v = "" if raw is None else str(raw).strip() if spec.get("required") and not v: raise HTTPException(400, f"{field} is required") return v if ftype == "int": if raw in (None, ""): if spec.get("required"): raise HTTPException(400, f"{field} is required") return None try: v = int(raw) except (TypeError, ValueError): raise HTTPException(400, f"{field} must be an integer") if "min" in spec and v < spec["min"]: raise HTTPException(400, f"{field} must be ≥ {spec['min']}") if "max" in spec and v > spec["max"]: raise HTTPException(400, f"{field} must be ≤ {spec['max']}") return v if ftype == "float": if raw in (None, ""): return 0.0 # historical default for numeric area columns try: v = float(raw) except (TypeError, ValueError): raise HTTPException(400, f"{field} must be a number") if "min" in spec and v < spec["min"]: raise HTTPException(400, f"{field} must be ≥ {spec['min']}") return v if ftype == "select": allowed = spec["options"] # Compare loosely: ints compare to int strings (rotation is int). sval = str(raw).strip() if raw is not None else "" for opt in allowed: if str(opt) == sval: return opt raise HTTPException(400, f"{field} must be one of {allowed}") if ftype == "fk": if raw in (None, "", 0, "0"): if not spec.get("nullable", False): raise HTTPException(400, f"{field} is required") return None try: v = int(raw) except (TypeError, ValueError): raise HTTPException(400, f"{field} must be an integer id") return v raise HTTPException(500, f"unknown field type {ftype}") new_value = coerce(spec, raw) side_effects: dict = {} with db() as con: # Verify the row exists. existing = con.execute( f"SELECT * FROM {table} WHERE id = ?", (entity_id,) ).fetchone() if not existing: raise HTTPException(404, f"{entity_type} #{entity_id} not found") prev_value = existing[field] if field in existing.keys() else None ent_label = _entity_label_safe(con, entity_type, entity_id) # FK existence check. if spec["type"] == "fk" and new_value is not None: fk_table = spec["fk_table"] row = con.execute( f"SELECT id FROM {fk_table} WHERE id = ?", (new_value,) ).fetchone() if not row: raise HTTPException(400, f"{field} references missing {fk_table} id") # Snapshot the previous overview/notes value before the change. if spec.get("snapshot"): cur_text = existing[field] if field in existing.keys() else "" snapshot_overview(con, entity_type, entity_id, cur_text or "") con.execute( f"UPDATE {table} SET {field} = ? WHERE id = ?", (new_value, entity_id), ) # Species type/function consistency: if the user just changed `type`, # clamp `primary_function` to a valid option if it doesn't match. # Return the clamped value as a side_effect so the client can update # the function display + its dropdown options without a page reload. if entity_type == "species" and field == "type": cur_fn = (existing["primary_function"] or "edible").strip().lower() valid = SPECIES_FUNCTION_OPTIONS.get(new_value, []) if cur_fn not in valid and valid: con.execute( "UPDATE species SET primary_function = ? WHERE id = ?", (valid[0], entity_id), ) side_effects["primary_function"] = valid[0] if entity_type == "species" and field == "primary_function": # Inverse: if function is set to one that doesn't match the # current type, reject. The client filters options at activate # time so this is a defense-in-depth check, not the primary one. cur_type = (existing["type"] or "plant").strip().lower() if not species_function_valid(cur_type, str(new_value)): raise HTTPException( 400, f"{new_value} is not a valid function for type {cur_type}" ) # Activity log: one row per primary edit. Side-effects (e.g. clamped # primary_function on type change) get their own log row so cascades # are visible in the detailed view. log_edit(con, entity_type, entity_id, field, prev_value, new_value, label=ent_label) for se_field, se_value in side_effects.items(): se_prev = existing[se_field] if se_field in existing.keys() else None log_edit( con, entity_type, entity_id, se_field, se_prev, se_value, label=ent_label + " (auto)", ) return {"ok": True, "value": new_value, "side_effects": side_effects} @app.post("/api/station/{station_id}/areas") async def api_station_set_areas(station_id: int, request: Request): """Replace a station's area-served set in one shot. Body: {"area_ids": [1, 2, 3]}. Used by the inline 'Edit areas served' toggle on the station detail page.""" data = await request.json() raw = data.get("area_ids") or [] if not isinstance(raw, list): raise HTTPException(400, "area_ids must be a list of integers") try: area_ids = [int(x) for x in raw] except (TypeError, ValueError): raise HTTPException(400, "area_ids must be integers") with db() as con: if not con.execute( "SELECT 1 FROM watering_stations WHERE id = ?", (station_id,) ).fetchone(): raise HTTPException(404) # Validate every supplied id exists in `areas`. Skip the rest if any # fail — partial updates would surprise the user. if area_ids: ph = ",".join("?" * len(area_ids)) found = {r[0] for r in con.execute( f"SELECT id FROM areas WHERE id IN ({ph})", area_ids ).fetchall()} missing = [i for i in area_ids if i not in found] if missing: raise HTTPException(400, f"unknown area ids: {missing}") prev_ids = sorted( r[0] for r in con.execute( "SELECT area_id FROM area_stations WHERE station_id = ?", (station_id,), ).fetchall() ) con.execute( "DELETE FROM area_stations WHERE station_id = ?", (station_id,) ) for aid in area_ids: con.execute( "INSERT OR IGNORE INTO area_stations (area_id, station_id) " "VALUES (?, ?)", (aid, station_id), ) new_sorted = sorted(area_ids) if prev_ids != new_sorted: label = _entity_label_safe(con, "station", station_id) log_edit( con, "station", station_id, "areas_served", prev_ids, new_sorted, label=label, ) return {"ok": True, "area_ids": area_ids} # --- Chat (AI assistant) endpoints --------------------------------------- # # A chat session is a conversation with Gemini grounded in the garden # starter context. Tool calls within a session are recorded on each # assistant message so the replay view can show what the model looked at. # One activity_log row per session — created at /sessions, updated with # the title and ended_at when /end is called. def _chat_history_for_session(con, session_id: int) -> list: """Rebuild Gemini-format `contents` from chat_messages. Tool calls are persisted as a JSON array on the assistant row; replay them as role:'model' (the function-call) followed by role:'function' (the response) so the model sees the same context it produced last turn.""" rows = con.execute( """SELECT role, text, photo_paths, tool_calls FROM chat_messages WHERE session_id = ? ORDER BY id""", (session_id,), ).fetchall() contents: list = [] for r in rows: if r["role"] == "user": parts: list = [] if r["text"]: parts.append({"text": r["text"]}) for p in (r["photo_paths"] or "").split("|"): if not p: continue abs_p = _safe_note_path(p) if not abs_p: continue try: with open(abs_p, "rb") as f: b64 = base64.b64encode(f.read()).decode("ascii") mime = mimetypes.guess_type(abs_p)[0] or "image/jpeg" parts.append({"inline_data": {"mime_type": mime, "data": b64}}) except OSError: pass if parts: contents.append({"role": "user", "parts": parts}) else: # Assistant turn: re-emit each tool call as a model functionCall + # function functionResponse pair, then the final text. try: tcs = json.loads(r["tool_calls"]) if r["tool_calls"] else [] except json.JSONDecodeError: tcs = [] for tc in tcs: contents.append({ "role": "model", "parts": [{"functionCall": { "name": tc.get("name", ""), "args": tc.get("args", {}) or {}, }}], }) contents.append({ "role": "function", "parts": [{"functionResponse": { "name": tc.get("name", ""), "response": tc.get("result", {}) or {}, }}], }) if r["text"]: contents.append({"role": "model", "parts": [{"text": r["text"]}]}) return contents @app.post("/api/chat/sessions") def api_chat_session_create(): """Create a new chat session and return its id. Logs a single 'add' row in activity_log (category extended to 'chat' to make it visually distinct in the detailed log; falls back to 'add' if the UI doesn't know about 'chat').""" with db() as con: ts = now() cur = con.execute( """INSERT INTO chat_sessions (created_at) VALUES (?)""", (ts,), ) session_id = cur.lastrowid # Use category='chat' — the detailed-log view styles it via the same # cat-* class system; missing CSS just falls back to default chip. log_cur = con.execute( """INSERT INTO activity_log (created_at, category, entity_type, entity_id, summary) VALUES (?, 'chat', 'chat_session', ?, ?)""", (ts, session_id, f"Chat session #{session_id} started"), ) log_id = log_cur.lastrowid con.execute( "UPDATE chat_sessions SET activity_log_id = ? WHERE id = ?", (log_id, session_id), ) return {"id": session_id, "activity_log_id": log_id} @app.post("/api/chat/sessions/{session_id}/messages") async def api_chat_send_message(session_id: int, request: Request): """Append a user message + photos, run Gemini, persist the assistant reply (with its tool-call record), and return the reply.""" form = await request.form() text = (form.get("text") or "").strip() photos = form.getlist("photos") if hasattr(form, "getlist") else [] photo_rels: list = [] inline_image_parts: list = [] for ph in photos: if hasattr(ph, "filename") and (ph.filename or "").strip(): rel = _save_upload(ph, "photo") photo_rels.append(rel) abs_p = _safe_note_path(rel) if abs_p: try: with open(abs_p, "rb") as f: b64 = base64.b64encode(f.read()).decode("ascii") mime = mimetypes.guess_type(abs_p)[0] or "image/jpeg" inline_image_parts.append( {"inline_data": {"mime_type": mime, "data": b64}} ) except OSError: pass if not text and not inline_image_parts: raise HTTPException(400, "message must include text or at least one photo") with db() as con: sess = con.execute( "SELECT id, ended_at FROM chat_sessions WHERE id = ?", (session_id,) ).fetchone() if not sess: raise HTTPException(404, "chat session not found") if sess["ended_at"]: raise HTTPException(400, "chat session is already ended") history = _chat_history_for_session(con, session_id) starter = build_starter_context(con) system_prompt = CHAT_SYSTEM_PROMPT + starter user_parts: list = [] if text: user_parts.append({"text": text}) user_parts.extend(inline_image_parts) # Persist the user message immediately — even if Gemini errors out, # the user's input shouldn't vanish. ts_user = now() con.execute( """INSERT INTO chat_messages (session_id, created_at, role, text, photo_paths) VALUES (?, ?, 'user', ?, ?)""", (session_id, ts_user, text, "|".join(photo_rels)), ) result = gemini_chat_complete( con, history, user_parts, system_prompt, session_id=session_id, ) ts_assist = now() con.execute( """INSERT INTO chat_messages (session_id, created_at, role, text, tool_calls) VALUES (?, ?, 'assistant', ?, ?)""", ( session_id, ts_assist, result["text"], json.dumps(result["tool_calls"]), ), ) return { "ok": True, "session_id": session_id, "user_message": {"text": text, "photo_paths": photo_rels, "created_at": ts_user}, "assistant_message": { "text": result["text"], "tool_calls": result["tool_calls"], "created_at": ts_assist, }, } @app.post("/api/chat/sessions/{session_id}/end") def api_chat_session_end(session_id: int): """Finalize the session: ask Gemini for a one-line title from the transcript, store it, and update the activity_log row's summary so the detailed log gets a meaningful title.""" with db() as con: sess = con.execute( "SELECT id, ended_at, activity_log_id FROM chat_sessions WHERE id = ?", (session_id,), ).fetchone() if not sess: raise HTTPException(404, "chat session not found") if sess["ended_at"]: return {"ok": True, "already_ended": True} msgs = con.execute( """SELECT role, text, tool_calls FROM chat_messages WHERE session_id = ? ORDER BY id""", (session_id,), ).fetchall() if not msgs: # Empty session — drop it instead of leaving litter. con.execute("DELETE FROM chat_sessions WHERE id = ?", (session_id,)) if sess["activity_log_id"]: con.execute( "DELETE FROM activity_log WHERE id = ?", (sess["activity_log_id"],), ) return {"ok": True, "discarded": True} # Build a compact transcript for the title-generation call. transcript_lines: list = [] tool_count = 0 for m in msgs: role = "User" if m["role"] == "user" else "Assistant" transcript_lines.append(f"{role}: {_short(m['text'], 200)}") if m["role"] == "assistant" and m["tool_calls"]: try: tc = json.loads(m["tool_calls"]) tool_count += len(tc) except json.JSONDecodeError: pass transcript = "\n".join(transcript_lines[:40]) # safety cap title = _chat_summarize_title(transcript) if tool_count: log_summary = ( f"Chat: {title} ({len(msgs)} messages, {tool_count} tools)" ) else: log_summary = f"Chat: {title} ({len(msgs)} messages)" ts = now() con.execute( """UPDATE chat_sessions SET ended_at = ?, title = ?, summary = ? WHERE id = ?""", (ts, title, log_summary, session_id), ) if sess["activity_log_id"]: con.execute( "UPDATE activity_log SET summary = ? WHERE id = ?", (log_summary, sess["activity_log_id"]), ) return {"ok": True, "title": title, "summary": log_summary} def _chat_summarize_title(transcript: str) -> str: """Single-shot Gemini call for a one-line session title. Falls back to a generic label on any failure — never block session-end on this.""" import urllib.request import urllib.error if not GEMINI_API_KEY or not transcript.strip(): return "Untitled chat" payload = { "contents": [{"parts": [{"text": "Summarize this garden-assistant chat as a single short title " "(max 8 words, no quotation marks, no period). Reply with the " "title only.\n\n" + transcript }]}], "generationConfig": {"temperature": 0.0, "responseMimeType": "text/plain"}, } url = ( f"https://generativelanguage.googleapis.com/v1beta/models/" f"{GEMINI_CHAT_MODEL}:generateContent?key={GEMINI_API_KEY}" ) req = urllib.request.Request( url, data=json.dumps(payload).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read().decode("utf-8")) title = data["candidates"][0]["content"]["parts"][0]["text"] return _short(title.strip().replace("\n", " "), 80) or "Untitled chat" except Exception: return "Untitled chat" @app.get("/api/chat/sessions/{session_id}") def api_chat_session_get(session_id: int): """Replay payload — session metadata + every message.""" with db() as con: sess = con.execute( "SELECT * FROM chat_sessions WHERE id = ?", (session_id,) ).fetchone() if not sess: raise HTTPException(404) rows = con.execute( """SELECT id, created_at, role, text, photo_paths, tool_calls FROM chat_messages WHERE session_id = ? ORDER BY id""", (session_id,), ).fetchall() msgs: list = [] for r in rows: try: tcs = json.loads(r["tool_calls"]) if r["tool_calls"] else [] except json.JSONDecodeError: tcs = [] msgs.append({ "id": r["id"], "created_at": r["created_at"], "role": r["role"], "text": r["text"], "photos": [p for p in (r["photo_paths"] or "").split("|") if p], "tool_calls": tcs, }) return { "id": sess["id"], "created_at": sess["created_at"], "ended_at": sess["ended_at"], "title": sess["title"] or "", "summary": sess["summary"] or "", "messages": msgs, } def api_artifact_list(token: str = ""): check_token(token) with db() as con: rows = [ dict(r) for r in con.execute( "SELECT * FROM artifacts ORDER BY generated_at DESC, id DESC" ).fetchall() ] return {"artifacts": rows} @app.post("/api/artifacts") async def api_artifact_create(request: Request, token: str = ""): check_token(token) data = await request.json() kind = (data.get("kind") or "other").strip() title = (data.get("title") or "").strip() content = data.get("content_md") or "" notes = (data.get("notes") or "").strip() if kind not in ARTIFACT_KINDS: raise HTTPException(400, f"invalid kind; must be one of {ARTIFACT_KINDS}") if not title: raise HTTPException(400, "title required") if not content.strip(): raise HTTPException(400, "content_md required") generated_at = now() date_part = generated_at[:10] base = f"{date_part}-{_slugify(title)}" # Ensure unique filename filename = f"{base}.md" i = 2 while os.path.exists(os.path.join(ARTIFACTS_DIR, filename)): filename = f"{base}-{i}.md" i += 1 _artifact_write(filename, content) with db() as con: cur = con.execute( """INSERT INTO artifacts (kind, title, filename, generated_at, notes) VALUES (?, ?, ?, ?, ?)""", (kind, title, filename, generated_at, notes), ) artifact_id = cur.lastrowid return {"ok": True, "id": artifact_id, "filename": filename} # --- Field notes (capture page) --- ALLOWED_AUDIO_EXT = {".webm", ".ogg", ".oga", ".m4a", ".mp4", ".mp3", ".wav"} ALLOWED_IMAGE_EXT = {".jpg", ".jpeg", ".png", ".heic", ".webp", ".gif"} MAX_AUDIO_BYTES = 50 * 1024 * 1024 # 50 MB MAX_IMAGE_BYTES = 25 * 1024 * 1024 # 25 MB per image def _ext_for(filename: str, content_type: Optional[str], allowed: set, default: str) -> str: base = (filename or "").lower() for ext in allowed: if base.endswith(ext): return ext if content_type: guessed = mimetypes.guess_extension(content_type.split(";")[0].strip()) or "" if guessed.lower() in allowed: return guessed.lower() return default def _save_upload(upload: UploadFile, kind: str) -> str: """Save upload under data/notes/YYYY-MM/. Returns relative path.""" if kind == "audio": allowed = ALLOWED_AUDIO_EXT default = ".webm" max_bytes = MAX_AUDIO_BYTES else: allowed = ALLOWED_IMAGE_EXT default = ".jpg" max_bytes = MAX_IMAGE_BYTES ext = _ext_for(upload.filename or "", upload.content_type, allowed, default) month = datetime.utcnow().strftime("%Y-%m") rel_dir = month abs_dir = os.path.join(NOTES_DIR, rel_dir) os.makedirs(abs_dir, exist_ok=True) name = f"{uuid.uuid4().hex}{ext}" rel_path = f"{rel_dir}/{name}" abs_path = os.path.join(NOTES_DIR, rel_path) size = 0 with open(abs_path, "wb") as f: while True: chunk = upload.file.read(1024 * 1024) if not chunk: break size += len(chunk) if size > max_bytes: f.close() os.remove(abs_path) raise HTTPException(413, f"{kind} upload exceeds {max_bytes // (1024*1024)} MB") f.write(chunk) return rel_path def _delete_note_file(rel_path: str) -> None: """Best-effort delete of a file under NOTES_DIR. Silent on failure.""" if not rel_path: return abs_path = _safe_note_path(rel_path) if abs_path: try: os.remove(abs_path) except OSError: pass def _safe_note_path(rel_path: str) -> Optional[str]: """Resolve a note file path safely under NOTES_DIR, returning absolute path or None.""" if not rel_path or rel_path.startswith("/") or ".." in rel_path.split("/"): return None abs_path = os.path.normpath(os.path.join(NOTES_DIR, rel_path)) notes_root = os.path.normpath(NOTES_DIR) if not abs_path.startswith(notes_root + os.sep) and abs_path != notes_root: return None if not os.path.isfile(abs_path): return None return abs_path def _row_to_note(row: sqlite3.Row) -> dict: d = dict(row) photos = [p for p in (d.get("photo_paths") or "").split("|") if p] d["photos"] = photos return d def _load_recent_notes(con, limit: int = 10, status: Optional[str] = None) -> list: where = "" args: list = [] if status: where = "WHERE status = ?" args.append(status) rows = con.execute( f"SELECT * FROM field_notes {where} ORDER BY created_at DESC, id DESC LIMIT ?", (*args, limit), ).fetchall() return [_row_to_note(r) for r in rows] @app.get("/capture", response_class=HTMLResponse) def capture_page(request: Request): with db() as con: recent = _load_recent_notes(con, limit=10) new_count = con.execute( "SELECT COUNT(*) FROM field_notes WHERE status = 'new'" ).fetchone()[0] return templates.TemplateResponse( "capture.html", {"request": request, "recent": recent, "new_count": new_count}, ) @app.get("/capture/queue", response_class=HTMLResponse) def capture_queue(request: Request, status: str = "new"): if status not in NOTE_STATUSES and status != "all": status = "new" with db() as con: if status == "all": rows = con.execute( "SELECT * FROM field_notes ORDER BY created_at DESC, id DESC LIMIT 200" ).fetchall() else: rows = con.execute( "SELECT * FROM field_notes WHERE status = ? ORDER BY created_at DESC, id DESC LIMIT 200", (status,), ).fetchall() notes = [_row_to_note(r) for r in rows] counts = {s: 0 for s in NOTE_STATUSES} for r in con.execute( "SELECT status, COUNT(*) AS n FROM field_notes GROUP BY status" ).fetchall(): counts[r["status"]] = r["n"] return templates.TemplateResponse( "capture_queue.html", {"request": request, "notes": notes, "status": status, "counts": counts}, ) @app.get("/notes/{rel_path:path}") def serve_note(request: Request, rel_path: str): if not verify_session(request.cookies.get(COOKIE_NAME)): raise HTTPException(401, "auth required") abs_path = _safe_note_path(rel_path) if not abs_path: raise HTTPException(404, "not found") return FileResponse(abs_path) def _transcribe_audio_via_gemini(abs_audio_path: str) -> str: """Transcribe an audio file via Gemini's generateContent (multimodal). Uses inline base64 audio (fine up to ~20 MB; field-note audio is bounded at 50 MB by _save_upload, so for larger files this would need switching to the Files API). Stdlib only — no httpx/requests dependency. """ import urllib.request import urllib.error if not GEMINI_API_KEY: raise HTTPException(500, "GEMINI_API_KEY is not configured") ext = os.path.splitext(abs_audio_path)[1].lower() mime_map = { ".webm": "audio/webm", ".ogg": "audio/ogg", ".oga": "audio/ogg", ".m4a": "audio/mp4", ".mp4": "audio/mp4", ".mp3": "audio/mpeg", ".wav": "audio/wav", } mime = mime_map.get(ext, "audio/webm") with open(abs_audio_path, "rb") as f: audio_b64 = base64.b64encode(f.read()).decode("ascii") payload = { "contents": [{ "parts": [ {"text": ( "Transcribe this audio recording verbatim. It is a casual " "voice memo about gardening (varieties, areas, observations). " "Return ONLY the transcript text. No preamble, no explanation, " "no quotation marks. Preserve the speaker's exact words including " "any 'um's or 'uh's removed. Resolve obvious mis-hearings of " "garden terms when context makes them clear." )}, {"inline_data": {"mime_type": mime, "data": audio_b64}}, ] }], "generationConfig": { "temperature": 0.0, "responseMimeType": "text/plain", }, } url = ( f"https://generativelanguage.googleapis.com/v1beta/models/" f"{GEMINI_TRANSCRIBE_MODEL}:generateContent?key={GEMINI_API_KEY}" ) req = urllib.request.Request( url, data=json.dumps(payload).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=120) as resp: body = resp.read().decode("utf-8") except urllib.error.HTTPError as e: detail = e.read().decode("utf-8", errors="replace")[:500] raise HTTPException(502, f"Gemini API error {e.code}: {detail}") except urllib.error.URLError as e: raise HTTPException(502, f"Gemini API unreachable: {e.reason}") data = json.loads(body) try: text = data["candidates"][0]["content"]["parts"][0]["text"] except (KeyError, IndexError, TypeError): raise HTTPException(502, f"Unexpected Gemini response shape: {body[:300]}") return text.strip() @app.post("/api/notes/{note_id}/transcribe") def api_note_transcribe(note_id: int, request: Request): if not verify_session(request.cookies.get(COOKIE_NAME)): raise HTTPException(401, "auth required") with db() as con: row = con.execute( "SELECT id, audio_path, transcript FROM field_notes WHERE id = ?", (note_id,), ).fetchone() if not row: raise HTTPException(404, "note not found") if not row["audio_path"]: raise HTTPException(400, "note has no audio attached") abs_path = _safe_note_path(row["audio_path"]) if not abs_path: raise HTTPException(404, "audio file missing on disk") transcript = _transcribe_audio_via_gemini(abs_path) con.execute( "UPDATE field_notes SET transcript = ? WHERE id = ?", (transcript, note_id), ) return {"ok": True, "id": note_id, "transcript": transcript} @app.post("/api/notes") async def api_note_create( request: Request, audio: Optional[UploadFile] = File(None), photos: List[UploadFile] = File(default=[]), text: str = Form(""), transcript: str = Form(""), captured_at: str = Form(""), ): require_session(request) text = (text or "").strip() transcript = (transcript or "").strip() audio_rel: Optional[str] = None photo_rels: list = [] saved: list = [] try: if audio is not None and audio.filename: audio_rel = _save_upload(audio, "audio") saved.append(audio_rel) for ph in photos or []: if ph and ph.filename: rel = _save_upload(ph, "photo") saved.append(rel) photo_rels.append(rel) except HTTPException: for rel in saved: ap = _safe_note_path(rel) if ap: try: os.remove(ap) except OSError: pass raise if not (audio_rel or photo_rels or text): raise HTTPException(400, "note must include audio, a photo, or text") with db() as con: cur = con.execute( """INSERT INTO field_notes (created_at, captured_at, audio_path, photo_paths, text, transcript, status) VALUES (?, ?, ?, ?, ?, ?, 'new')""", ( now(), captured_at or None, audio_rel, "|".join(photo_rels), text, transcript, ), ) note_id = cur.lastrowid return {"ok": True, "id": note_id} @app.get("/api/notes") def api_notes_list(request: Request, status: str = "new", limit: int = 50): require_session(request) if status not in NOTE_STATUSES and status != "all": status = "new" limit = max(1, min(200, limit)) with db() as con: if status == "all": rows = con.execute( "SELECT * FROM field_notes ORDER BY created_at DESC, id DESC LIMIT ?", (limit,), ).fetchall() else: rows = con.execute( "SELECT * FROM field_notes WHERE status = ? ORDER BY created_at DESC, id DESC LIMIT ?", (status, limit), ).fetchall() notes = [_row_to_note(r) for r in rows] return {"notes": notes} @app.patch("/api/notes/{note_id}") async def api_note_update(note_id: int, request: Request): require_session(request) data = await request.json() fields = [] args: list = [] if "status" in data: s = (data.get("status") or "").strip() if s not in NOTE_STATUSES: raise HTTPException(400, f"invalid status; must be one of {NOTE_STATUSES}") fields.append("status = ?") args.append(s) if s == "processed": fields.append("processed_at = ?") args.append(now()) if "transcript" in data: fields.append("transcript = ?") args.append(data.get("transcript") or "") if "text" in data: fields.append("text = ?") args.append(data.get("text") or "") if "processing_notes" in data: fields.append("processing_notes = ?") args.append(data.get("processing_notes") or "") if not fields: raise HTTPException(400, "no updatable fields provided") args.append(note_id) with db() as con: cur = con.execute( f"UPDATE field_notes SET {', '.join(fields)} WHERE id = ?", args ) if cur.rowcount == 0: raise HTTPException(404, "note not found") return {"ok": True} @app.delete("/api/notes/{note_id}") def api_note_delete(note_id: int, request: Request): require_session(request) with db() as con: row = con.execute( "SELECT audio_path, photo_paths FROM field_notes WHERE id = ?", (note_id,) ).fetchone() if not row: raise HTTPException(404, "note not found") paths = [] if row["audio_path"]: paths.append(row["audio_path"]) for p in (row["photo_paths"] or "").split("|"): if p: paths.append(p) con.execute("DELETE FROM field_notes WHERE id = ?", (note_id,)) for rel in paths: ap = _safe_note_path(rel) if ap: try: os.remove(ap) except OSError: pass return {"ok": True}