"""Campaign data access — reads from local JSON file (dev). File location: settings.campaigns_file (default: data/campaigns.json). """ from __future__ import annotations import json import logging from pathlib import Path from typing import Any, cast from marketing_intelligence.core.config import settings from marketing_intelligence.core.models import Campaign logger = logging.getLogger(__name__) def _fetch_local() -> list[dict[str, Any]]: path = Path(settings.campaigns_file) if not path.exists(): logger.warning("campaigns file not found: %s", path) return [] return cast(list[dict[str, Any]], json.loads(path.read_text(encoding="utf-8"))) def _to_campaigns(rows: list[dict[str, Any]]) -> list[Campaign]: return [Campaign.model_validate(r) for r in rows if r.get("status") == "active"] def list_active_campaigns() -> list[Campaign]: """Return all active campaigns from the local JSON file.""" return _to_campaigns(_fetch_local()) def get_campaign(campaign_id: str) -> Campaign | None: """Return one campaign by campaign_id or campaign_key.""" for c in list_active_campaigns(): if c.campaign_id == campaign_id or c.campaign_key == campaign_id: return c return None