"""Change log tracking for all modifications.
This module provides internal functions for logging and retrieving changes.
MCP resource registration is handled by resources.py.
"""
import json
from datetime import datetime
from typing import Any
from .database import get_db_connection
def log_change(
entity_type: str,
action: str,
entity_id: str | None = None,
details: dict[str, Any] | None = None,
user_context: str | None = None,
) -> int:
"""Record a change to the Google Ads account.
Args:
entity_type: Type of entity changed (campaign, ad_group, ad, keyword, etc.)
action: Action taken (create, update, delete, pause, enable, etc.)
entity_id: Optional ID of the affected entity.
details: Optional dictionary with change details.
user_context: Optional context about why the change was made.
Returns:
The ID of the created change log entry.
"""
with get_db_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"""
INSERT INTO change_log (entity_type, entity_id, action, details, user_context)
VALUES (?, ?, ?, ?, ?)
""",
(
entity_type,
entity_id,
action,
json.dumps(details) if details else None,
user_context,
),
)
conn.commit()
return cursor.lastrowid
def get_recent_changes(
limit: int = 50,
entity_type: str | None = None,
since: datetime | None = None,
) -> list[dict[str, Any]]:
"""Retrieve recent changes from the log.
Args:
limit: Maximum number of changes to return.
entity_type: Optional filter by entity type.
since: Optional filter for changes after this timestamp.
Returns:
List of change log entries.
"""
with get_db_connection() as conn:
cursor = conn.cursor()
query = "SELECT * FROM change_log WHERE 1=1"
params = []
if entity_type:
query += " AND entity_type = ?"
params.append(entity_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"],
"entity_type": row["entity_type"],
"entity_id": row["entity_id"],
"action": row["action"],
"details": json.loads(row["details"]) if row["details"] else None,
"user_context": row["user_context"],
}
for row in rows
]
