"""Perplexity Sonar-Pro client. Two call shapes used by the pipeline: - city_intelligence(city, country, show_date) → CityIntelligence - local_news(city, country, since_date, show_date) → PerplexityNewsResult (also extracts competitor_warnings from the returned list) The competitor-warnings extraction happens here because Perplexity is the only provider that can scan current news sources; the Gemini fact-check stage only verifies what comes out of this stage. """ from __future__ import annotations import asyncio import json import logging from datetime import date from typing import Any import httpx from app.config import settings from app.models import ( CityIntelligence, NewsItem, PerplexityNewsResult, ) from app.providers._common import ( JsonExtractionError, _build_semaphore, extract_json, http_post_json, ) logger = logging.getLogger(__name__) _PERPLEXITY_URL = "https://api.perplexity.ai/chat/completions" class PerplexityClient: """Sonar-Pro HTTP client (OpenAI-compatible chat.completions endpoint).""" def __init__(self) -> None: self._client = httpx.AsyncClient(timeout=settings.PERPLEXITY_TIMEOUT) self._sem = _build_semaphore(settings.PERPLEXITY_MAX_CONCURRENCY) async def close(self) -> None: await self._client.aclose() def _headers(self) -> dict[str, str]: return { "Authorization": f"Bearer {settings.PERPLEXITY_API_KEY}", "Content-Type": "application/json", "Accept": "application/json", } async def _chat(self, *, system: str, user: str) -> dict[str, Any]: payload = { "model": settings.PERPLEXITY_MODEL, "messages": [ {"role": "system", "content": system}, {"role": "user", "content": user}, ], "temperature": 0.2, } async with self._sem: data = await http_post_json( client=self._client, url=_PERPLEXITY_URL, headers=self._headers(), payload=payload, timeout=float(settings.PERPLEXITY_TIMEOUT), provider_name="perplexity", ) try: text = data["choices"][0]["message"]["content"] except (KeyError, IndexError) as exc: raise JsonExtractionError("perplexity: missing choices[0].message.content") from exc citations = data.get("citations", []) return {"text": text, "citations": citations} # ── Stage 5: city_intelligence ───────────────────────────────────── async def city_intelligence(self, *, city: str, country: str, show_date: date) -> CityIntelligence: system = ( "You are a city-intelligence analyst for a stand-up comedy tour. " "Only return claims that you can support with current online sources. " "Do not speculate. If you cannot find a fact, omit it. " "Return STRICT JSON only — no commentary, no markdown fences." ) user = ( f"I am preparing a comedy show in {city}, {country} on {show_date.isoformat()}. " "Produce a JSON object with these keys: " 'points_of_pride (up to 5 strings the city is known for locally), ' 'sensitivities_to_avoid (up to 5 topics a visiting comedian should not poke at), ' "local_conversation_today (one sentence describing the biggest local " "conversation in this city right now), " "audience_competitors (list of up to 5 other comedians or shows " "currently competing for this city's live-comedy audience)." ) data = await self._chat(system=system, user=user) try: parsed = extract_json(data["text"]) except JsonExtractionError: logger.warning("perplexity city_intel returned non-JSON; storing raw text") parsed = { "local_conversation_today": data["text"][:1000], "points_of_pride": [], "sensitivities_to_avoid": [], "audience_competitors": [], } if isinstance(parsed.get("points_of_pride"), str): parsed["points_of_pride"] = [parsed["points_of_pride"]] rows: list[str] = [] competitors: list[str] = [] if isinstance(parsed.get("audience_competitors"), list): competitors = [str(x) for x in parsed["audience_competitors"]] return CityIntelligence( city=city, points_of_pride=[str(x) for x in parsed.get("points_of_pride", [])][:5], sensitivities_to_avoid=[str(x) for x in parsed.get("sensitivities_to_avoid", [])][:5], local_conversation_today=str(parsed.get("local_conversation_today") or None), audience_competitors=competitors[:5], raw_provider_payload={"text": data["text"][:4000], "citations": data["citations"]}, ) # ── Stage 6: local_news (+ competitor_warnings extraction) ────────── async def local_news(self, *, city: str, country: str, since_date: date, show_date: date) -> PerplexityNewsResult: system = ( "You are a local-news summarizer for a visiting comedian. Cite the " "original URLs when possible. Return STRICT JSON only — no commentary, " "no markdown fences." ) user = ( f"Return local news for {city}, {country} published between " f"{since_date.isoformat()} and {show_date.isoformat()}. " "Produce a JSON object with two keys:\n" ' "recent_news": array of objects {title, summary, date, url} ' "of up to 8 items most relevant to a visiting comedian.\n" ' "competitor_warnings": array of objects {title, summary, date, url} ' "of up to 5 items describing OTHER comedy shows or comedians " f"performing in {city} within ±3 days of {show_date.isoformat()}. " "If no competition is found, return an empty array." ) data = await self._chat(system=system, user=user) try: parsed = extract_json(data["text"]) except JsonExtractionError: logger.warning("perplexity local_news returned non-JSON; storing raw text") parsed = {"recent_news": [], "competitor_warnings": []} if not isinstance(parsed, dict): parsed = {"recent_news": [], "competitor_warnings": []} news_rows = [] for item in parsed.get("recent_news", [])[:8]: news_rows.append(NewsItem( title=str(item.get("title", "") or "")[:300], summary=str(item.get("summary", "") or "")[:1000], date=str(item.get("date") or None), url=str(item.get("url") or None), is_competitor=False, )) competitor_rows = [] for item in parsed.get("competitor_warnings", [])[:5]: competitor_rows.append(NewsItem( title=str(item.get("title", "") or "")[:300], summary=str(item.get("summary", "") or "")[:1000], date=str(item.get("date") or None), url=str(item.get("url") or None), is_competitor=True, )) return PerplexityNewsResult(recent_news=news_rows, competitor_warnings=competitor_rows)