pyr0ball
5f5cb99db6
Plant registry, grow event calendar, and gardening knowledge base populated via cf-harvest ingest endpoint. All 23 tests passing. Stack: - FastAPI + SQLite (cf-core migrations) on port 8522 - Vue 3 + Vite + Pinia SPA on port 8521 - 5 SQL migrations: locations → plants → grow_events, knowledge_sources → knowledge_facts - Earthy green theme (--color-primary: #4a7c59) matching CF theme system Knowledge ingest contract: - POST /api/v1/knowledge/ingest — idempotent on fact_hash (sha256 of type+payload+video_path) - 8 fact types: companion_planting, soil_amendment, propagation, pest_diagnosis, harvest_timing, instruction, medicinal, history_context - Partial batch failure: malformed facts collected in rejected[], valid facts inserted - Re-ingesting same video batch: facts_skipped_duplicate > 0, no duplicates Wired views: RegistryView (plant CRUD), KnowledgeView (facts + type filter chips) Stub views: CalendarView, SettingsView
100 lines
2.7 KiB
Python
100 lines
2.7 KiB
Python
"""cf-harvest knowledge ingest + query endpoints."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
|
|
from app.core.config import settings
|
|
from app.db.store import Store
|
|
from app.models.schemas.knowledge import (
|
|
IngestResponse,
|
|
KnowledgeFactOut,
|
|
KnowledgeIngestRequest,
|
|
)
|
|
from app.services.knowledge.ingest import ingest
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _get_store() -> Store:
|
|
s = Store(settings.DB_PATH)
|
|
try:
|
|
yield s
|
|
finally:
|
|
s.close()
|
|
|
|
|
|
@router.post("/ingest", response_model=IngestResponse)
|
|
async def ingest_knowledge(
|
|
body: KnowledgeIngestRequest, store: Store = Depends(_get_store)
|
|
):
|
|
return ingest(store.conn, body)
|
|
|
|
|
|
@router.get("", response_model=list[KnowledgeFactOut])
|
|
async def query_knowledge(
|
|
fact_type: Optional[str] = Query(None),
|
|
subject: Optional[str] = Query(None),
|
|
presenter: Optional[str] = Query(None),
|
|
min_confidence: Optional[float] = Query(None, ge=0.0, le=1.0),
|
|
limit: int = Query(50, ge=1, le=500),
|
|
offset: int = Query(0, ge=0),
|
|
store: Store = Depends(_get_store),
|
|
):
|
|
conditions = []
|
|
params: list = []
|
|
|
|
if fact_type:
|
|
conditions.append("kf.fact_type = ?")
|
|
params.append(fact_type)
|
|
if subject:
|
|
conditions.append("kf.subject LIKE ?")
|
|
params.append(f"%{subject}%")
|
|
if presenter:
|
|
conditions.append("kf.presenter LIKE ?")
|
|
params.append(f"%{presenter}%")
|
|
if min_confidence is not None:
|
|
conditions.append("kf.confidence >= ?")
|
|
params.append(min_confidence)
|
|
|
|
where = ("WHERE " + " AND ".join(conditions)) if conditions else ""
|
|
params += [limit, offset]
|
|
|
|
rows = store.conn.execute(
|
|
f"""
|
|
SELECT kf.*, ks.video_path, ks.location
|
|
FROM knowledge_facts kf
|
|
LEFT JOIN knowledge_sources ks ON ks.id = kf.source_id
|
|
{where}
|
|
ORDER BY kf.created_at DESC
|
|
LIMIT ? OFFSET ?
|
|
""",
|
|
params,
|
|
).fetchall()
|
|
|
|
result = []
|
|
for row in rows:
|
|
d = store._row_to_dict(row)
|
|
d["payload"] = json.loads(d["payload"])
|
|
result.append(d)
|
|
return result
|
|
|
|
|
|
@router.get("/{fact_id}", response_model=KnowledgeFactOut)
|
|
async def get_knowledge_fact(fact_id: int, store: Store = Depends(_get_store)):
|
|
row = store.conn.execute(
|
|
"""
|
|
SELECT kf.*, ks.video_path, ks.location
|
|
FROM knowledge_facts kf
|
|
LEFT JOIN knowledge_sources ks ON ks.id = kf.source_id
|
|
WHERE kf.id = ?
|
|
""",
|
|
(fact_id,),
|
|
).fetchone()
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Knowledge fact not found")
|
|
d = store._row_to_dict(row)
|
|
d["payload"] = json.loads(d["payload"])
|
|
return d
|