"""Learnings and insights storage.
This module provides internal functions for recording and retrieving learnings.
MCP resource and tool registration is handled by resources.py and tools.py.
"""
import json
from datetime import datetime
from typing import Any
from .database import get_db_connection
def add_learning(
category: str,
insight: str,
evidence: str | None = None,
confidence: str = "medium",
actionable_recommendations: list[str] | None = None,
related_decisions: list[int] | None = None,
) -> int:
"""Record a learning or insight.
Args:
category: Category of learning (audience, keywords, creative, bidding, timing, etc.)
insight: The insight or learning.
evidence: Data or observations supporting this insight.
confidence: Confidence level (low, medium, high).
actionable_recommendations: Suggested actions based on this learning.
related_decisions: IDs of related decisions.
Returns:
The ID of the recorded learning.
"""
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO learnings
(category, insight, evidence, confidence, actionable_recommendations, related_decisions)
VALUES (?, ?, ?, ?, ?, ?)
""",
(
category,
insight,
evidence,
confidence,
json.dumps(actionable_recommendations) if actionable_recommendations else None,
json.dumps(related_decisions) if related_decisions else None,
),
)
conn.commit()
return cursor.lastrowid
def get_learnings(
category: str | None = None,
confidence: str | None = None,
limit: int = 30,
since: datetime | None = None,
) -> list[dict[str, Any]]:
"""Retrieve recorded learnings.
Args:
category: Optional filter by category.
confidence: Optional filter by confidence level.
limit: Maximum number of learnings to return.
since: Optional filter for learnings after this timestamp.
Returns:
List of learning entries.
"""
with get_db_connection() as conn:
cursor = conn.cursor()
query = "SELECT * FROM learnings WHERE 1=1"
params = []
if category:
query += " AND category = ?"
params.append(category)
if confidence:
query += " AND confidence = ?"
params.append(confidence)
if since:
query += " AND timestamp > ?"
params.append(since.isoformat())
query += " ORDER BY timestamp DESC LIMIT ?"
params.append(limit)
cursor.execute(query, params)
rows = cursor.fetchall()
return [
{
"id": row["id"],
"timestamp": row["timestamp"],
"category": row["category"],
"insight": row["insight"],
"evidence": row["evidence"],
"confidence": row["confidence"],
"actionable_recommendations": (
json.loads(row["actionable_recommendations"])
if row["actionable_recommendations"]
else None
),
"related_decisions": (
json.loads(row["related_decisions"])
if row["related_decisions"]
else None
),
}
for row in rows
]
