"""Report history storage and retrieval.
This module provides internal functions for storing and retrieving report history.
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 store_report(
report_type: str,
date_range: str,
data: dict[str, Any],
filters: dict[str, Any] | None = None,
summary: str | None = None,
) -> int:
"""Store a report snapshot for historical comparison.
Args:
report_type: Type of report (campaign, keyword, search_terms, performance).
date_range: The date range the report covers.
data: The report data.
filters: Optional filters applied to the report.
summary: Optional human-readable summary.
Returns:
The ID of the stored report.
"""
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO report_history (report_type, date_range, filters, data, summary)
VALUES (?, ?, ?, ?, ?)
""",
(
report_type,
date_range,
json.dumps(filters) if filters else None,
json.dumps(data),
summary,
),
)
conn.commit()
return cursor.lastrowid
def get_report_history(
report_type: str | None = None,
limit: int = 20,
since: datetime | None = None,
) -> list[dict[str, Any]]:
"""Retrieve historical reports.
Args:
report_type: Optional filter by report type.
limit: Maximum number of reports to return.
since: Optional filter for reports after this timestamp.
Returns:
List of historical report entries.
"""
with get_db_connection() as conn:
cursor = conn.cursor()
query = "SELECT * FROM report_history WHERE 1=1"
params = []
if report_type:
query += " AND report_type = ?"
params.append(report_type)
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"],
"report_type": row["report_type"],
"date_range": row["date_range"],
"filters": json.loads(row["filters"]) if row["filters"] else None,
"data": json.loads(row["data"]),
"summary": row["summary"],
}
for row in rows
]
