competitive_intel_agent.py

"""
Competitive Intelligence Agent — researches competitors and market using SerpApi + OpenAI.
Optional HubSpot integration for internal CRM context. Produces briefings with citations.
"""
import os
import json
import logging
import asyncio
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from datetime import datetime

from openai import OpenAI
import serpapi
from dotenv import load_dotenv
from urllib3.util import Retry
from hubspot import HubSpot
from hubspot.crm.companies import (
    ApiException as CompaniesApiException,
    Filter,
    FilterGroup,
    PublicObjectSearchRequest,
)
from hubspot.crm.contacts import ApiException as ContactsApiException

load_dotenv()


@dataclass
class CompetitiveIntelConfig:
    """Configuration for the competitive intelligence agent."""
    openai_api_key: str
    serpapi_api_key: str
    hubspot_access_token: Optional[str] = None
    model: str = "gpt-4o"
    topn: int = 10
    report_dir: Optional[str] = None
    debug: bool = False

    @classmethod
    def from_env(cls) -> "CompetitiveIntelConfig":
        required = ["OPENAI_API_KEY", "SERPAPI_API_KEY"]
        missing = [v for v in required if not os.getenv(v)]
        if missing:
            raise ValueError(f"Missing required environment variables: {missing}")

        report_dir = os.getenv("REPORT_DIR")
        if report_dir and not os.path.isdir(report_dir):
            os.makedirs(report_dir, exist_ok=True)

        return cls(
            openai_api_key=os.getenv("OPENAI_API_KEY"),
            serpapi_api_key=os.getenv("SERPAPI_API_KEY"),
            hubspot_access_token=os.getenv("HUBSPOT_ACCESS_TOKEN") or None,
            model=os.getenv("OPENAI_MODEL", "gpt-4o"),
            topn=int(os.getenv("RESULT_LIMIT", "10")),
            report_dir=report_dir,
            debug=os.getenv("DEBUG", "false").lower() == "true",
        )


def search_web(api_key: str, query: str, topn: int = 10) -> str:
    """Search Google and return structured results."""
    try:
        params = {
            "engine": "google",
            "q": query,
            "num": topn,
            "api_key": api_key,
        }
        res = serpapi.search(params)
        items = (res.get("organic_results") or [])[:topn]
        data = [
            {
                "title": r.get("title"),
                "snippet": r.get("snippet"),
                "link": r.get("link"),
                "source": "google",
            }
            for r in items
        ]
        return json.dumps({"type": "web", "query": query, "results": data}, ensure_ascii=False)
    except Exception as e:
        return json.dumps({"type": "web", "query": query, "error": str(e)})


def search_news(api_key: str, query: str, date_range: str = "last_month", topn: int = 10) -> str:
    """Search Google News."""
    try:
        params = {
            "engine": "google_news",
            "q": query,
            "api_key": api_key,
        }
        if date_range == "last_week":
            params["tbs"] = "qdr:w"
        elif date_range == "last_month":
            params["tbs"] = "qdr:m"
        res = serpapi.search(params)
        items = (res.get("news_results") or [])[:topn]
        data = [
            {
                "title": r.get("title"),
                "snippet": r.get("snippet"),
                "link": r.get("link"),
                "date": r.get("date"),
                "source": r.get("source"),
            }
            for r in items
        ]
        return json.dumps({"type": "news", "query": query, "results": data}, ensure_ascii=False)
    except Exception as e:
        return json.dumps({"type": "news", "query": query, "error": str(e)})


def search_jobs(api_key: str, query: str, topn: int = 10) -> str:
    """Search Google Jobs (e.g. competitor hiring signals)."""
    try:
        params = {
            "engine": "google_jobs",
            "q": query,
            "api_key": api_key,
        }
        res = serpapi.search(params)
        jobs = res.get("jobs_results") or []
        listings = jobs[:topn] if isinstance(jobs, list) else []
        data = []
        for j in listings:
            if isinstance(j, dict):
                data.append({
                    "title": j.get("title"),
                    "company": j.get("company_name"),
                    "location": j.get("location"),
                    "via": j.get("via"),
                    "description": (j.get("description") or "")[:500],
                })
            else:
                data.append({"raw": str(j)})
        return json.dumps({"type": "jobs", "query": query, "results": data}, ensure_ascii=False)
    except Exception as e:
        return json.dumps({"type": "jobs", "query": query, "error": str(e)})


TOOLS: List[Dict[str, Any]] = [
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "Search Google for company info, positioning, reviews, pricing pages, and general competitive research.",
            "parameters": {
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_news",
            "description": "Search for recent news about companies: funding, product launches, exec moves, partnerships.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "date_range": {
                        "type": "string",
                        "description": "e.g. 'last_week', 'last_month'",
                        "enum": ["last_week", "last_month"],
                    },
                },
                "required": ["query"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "search_jobs",
            "description": "Search job listings (e.g. competitor hiring, role growth, skills they invest in).",
            "parameters": {
                "type": "object",
                "properties": {"query": {"type": "string"}},
                "required": ["query"],
            },
        },
    },
]

HUBSPOT_TOOLS: List[Dict[str, Any]] = [
    {
        "type": "function",
        "function": {
            "name": "hubspot_search_company_by_domain",
            "description": "[HUBSPOT] Check if we have this company in HubSpot (competitor or account we track). Returns company id, name, domain.",
            "parameters": {
                "type": "object",
                "properties": {"domain": {"type": "string", "description": "Company domain to search for"}},
                "required": ["domain"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "hubspot_get_contact_by_email",
            "description": "[HUBSPOT] Get contact details by email for internal context (who we know at this company).",
            "parameters": {
                "type": "object",
                "properties": {"email": {"type": "string", "description": "Contact email address"}},
                "required": ["email"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "hubspot_get_contact_activity_history",
            "description": "[HUBSPOT] Get our notes, calls, and emails with this contact for internal competitive context.",
            "parameters": {
                "type": "object",
                "properties": {"email": {"type": "string", "description": "Contact email address"}},
                "required": ["email"],
            },
        },
    },
]


class HubSpotService:
    """HubSpot CRM service for internal competitor/account context."""

    def __init__(self, config: CompetitiveIntelConfig) -> None:
        self.config = config
        retry = Retry(
            total=5,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
        )
        self.client = HubSpot(access_token=config.hubspot_access_token, retry=retry)
        self._log = logging.getLogger("HubSpotService")
        if config.debug:
            self._log.setLevel(logging.DEBUG)

    def search_company_by_domain(self, domain: str) -> Optional[Dict[str, Any]]:
        try:
            domain_filter = Filter(property_name="domain", operator="EQ", value=domain)
            search_request = PublicObjectSearchRequest(
                filter_groups=[FilterGroup(filters=[domain_filter])],
                properties=["name", "domain", "hs_object_id"],
                limit=1,
            )
            response = self.client.crm.companies.search_api.do_search(search_request)
            if response.results:
                company = response.results[0]
                return {
                    "id": company.id,
                    "name": company.properties.get("name"),
                    "domain": company.properties.get("domain"),
                }
            return None
        except CompaniesApiException as e:
            self._log.error("Company search failed: %s", e)
            return None

    def get_contact_by_email(self, email: str) -> Optional[Dict[str, Any]]:
        try:
            contact = self.client.crm.contacts.basic_api.get_by_id(
                contact_id=email,
                id_property="email",
                properties=["firstname", "lastname", "email", "phone", "lifecyclestage"],
            )
            return contact.to_dict()
        except ContactsApiException as e:
            if e.status == 404:
                return None
            self._log.error("Contact retrieval error: %s", e)
            raise

    def get_contact_activity_history(self, email: str) -> List[Dict[str, Any]]:
        contact = self.get_contact_by_email(email)
        if not contact:
            return []
        contact_id = contact["id"]
        history: List[Dict[str, Any]] = []
        activity_types = {"note": "note", "email": "email", "call": "call"}
        for object_type in activity_types:
            try:
                assoc = self.client.crm.associations.v4.basic_api.get_page(
                    object_type="contact",
                    object_id=contact_id,
                    to_object_type=object_type,
                )
                ids = [r.to_object_id for r in assoc.results]
                if not ids:
                    continue
                props = ["hs_timestamp", "hubspot_owner_id"]
                if object_type == "note":
                    props.append("hs_note_body")
                elif object_type == "email":
                    props.append("hs_email_subject")
                elif object_type == "call":
                    props.append("hs_call_title")
                batch = getattr(self.client.crm.objects, f"{object_type}s").batch_api.read(
                    batch_read_input_simple_public_object_id={
                        "properties": props,
                        "inputs": [{"id": oid} for oid in ids],
                    }
                )
                for eng in batch.results:
                    history.append({
                        "type": object_type,
                        "timestamp": eng.properties.get("hs_timestamp"),
                        "content": self._extract_content(eng.properties, object_type),
                    })
            except Exception as e:
                self._log.warning("Failed to get %s history: %s", object_type, e)
                continue
        return sorted(history, key=lambda x: x.get("timestamp") or "", reverse=True)

    @staticmethod
    def _extract_content(properties: Dict[str, Any], object_type: str) -> str:
        if object_type == "note":
            return properties.get("hs_note_body") or ""
        if object_type == "email":
            return properties.get("hs_email_subject") or ""
        if object_type == "call":
            return properties.get("hs_call_title") or ""
        return ""

    async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> str:
        try:
            if tool_name == "hubspot_search_company_by_domain":
                result = self.search_company_by_domain(arguments.get("domain", ""))
                return json.dumps(result or {"error": "Company not found"})
            if tool_name == "hubspot_get_contact_by_email":
                result = self.get_contact_by_email(arguments.get("email", ""))
                return json.dumps(result or {"error": "Contact not found"})
            if tool_name == "hubspot_get_contact_activity_history":
                result = self.get_contact_activity_history(arguments.get("email", ""))
                return json.dumps({"activities": result})
            return json.dumps({"error": f"Unknown HubSpot tool: {tool_name}"})
        except Exception as e:
            self._log.error("HubSpot tool %s failed: %s", tool_name, e)
            return json.dumps({"error": str(e)})


class CompetitiveIntelligenceAgent:
    """Agent that researches competitors and produces briefings using SerpApi + OpenAI; optional HubSpot for internal context."""

    def __init__(self, config: Optional[CompetitiveIntelConfig] = None):
        self.config = config or CompetitiveIntelConfig.from_env()
        self.client = OpenAI(api_key=self.config.openai_api_key)
        self.tools = list(TOOLS)
        self.hubspot_service: Optional[HubSpotService] = None
        if self.config.hubspot_access_token:
            self.hubspot_service = HubSpotService(self.config)
            self.tools = self.tools + list(HUBSPOT_TOOLS)
        self.messages: List[Dict[str, Any]] = []
        level = logging.DEBUG if self.config.debug else logging.INFO
        logging.basicConfig(level=level, format="[%(levelname)s] %(message)s")
        self.log = logging.getLogger("CompetitiveIntelAgent")
        self._setup_system_prompt()

    def _setup_system_prompt(self) -> None:
        today_str = datetime.now().strftime("%Y-%m-%d")
        hubspot_note = ""
        if self.hubspot_service:
            hubspot_note = (
                " When HubSpot is available, you can look up companies by domain, contacts by email, "
                "and activity history (notes, calls, emails) to combine external intel with what we already know in CRM."
            )
        sys_prompt = (
            f"Today's date is {today_str}. "
            "You are an AI competitive intelligence analyst. "
            "You have access to web search, news search, and job search."
            f"{hubspot_note} "
            "Your job is to:\n"
            "1. Research companies and competitors using search_web (positioning, pricing, reviews).\n"
            "2. Find recent news with search_news (funding, launches, exec changes).\n"
            "3. Use search_jobs to infer hiring focus and growth (e.g. 'Company X jobs').\n"
            "4. When HubSpot tools are available, use them to add internal context (companies we track, contacts, our past interactions).\n"
            "5. Synthesize a clear, actionable briefing with key findings and citations [1], [2], ...\n\n"
            "Always cite sources by number. Be concise; highlight differentiators, pricing/positioning, and recent moves."
        )
        self.messages = [{"role": "system", "content": sys_prompt}]

    async def _call_tool(self, name: str, arguments: Dict[str, Any]) -> str:
        if name == "search_web":
            return search_web(
                self.config.serpapi_api_key,
                arguments.get("query", ""),
                self.config.topn,
            )
        if name == "search_news":
            return search_news(
                self.config.serpapi_api_key,
                arguments.get("query", ""),
                arguments.get("date_range", "last_month"),
                self.config.topn,
            )
        if name == "search_jobs":
            return search_jobs(
                self.config.serpapi_api_key,
                arguments.get("query", ""),
                self.config.topn,
            )
        if name.startswith("hubspot_") and self.hubspot_service:
            return await self.hubspot_service.call_tool(name, arguments)
        return json.dumps({"error": f"Unknown tool: {name}"})

    async def run_once(self, user_input: str) -> str:
        """Process one user message and return the final briefing."""
        self.messages.append({"role": "user", "content": user_input})

        while True:
            response = self.client.chat.completions.create(
                model=self.config.model,
                messages=self.messages,
                tools=self.tools,
            )
            msg = response.choices[0].message
            tool_calls = msg.tool_calls

            if not tool_calls:
                answer = msg.content or ""
                self.messages.append({"role": "assistant", "content": answer})
                return answer

            self.messages.append(msg)
            for tc in tool_calls:
                name = tc.function.name
                args = json.loads(tc.function.arguments)
                content = await self._call_tool(name, args)
                self.messages.append({
                    "tool_call_id": tc.id,
                    "role": "tool",
                    "name": name,
                    "content": content,
                })

    def save_conversation(self, path: str) -> None:
        """Save conversation history to a JSON file."""
        with open(path, "w") as f:
            json.dump(self.messages, f, indent=2)
        self.log.info("Conversation saved to %s", path)


async def main() -> None:
    import argparse

    parser = argparse.ArgumentParser(description="Competitive Intelligence Agent")
    parser.add_argument("-q", "--query", help="Single query mode")
    parser.add_argument("-m", "--model", default="gpt-4o", choices=["gpt-4o", "gpt-4o-mini"])
    parser.add_argument("-n", "--topn", type=int, default=10, help="Max results per search")
    parser.add_argument("-d", "--debug", action="store_true", help="Enable debug logging")
    parser.add_argument("-o", "--outfile", help="Save conversation to JSON file")
    args = parser.parse_args()

    config = CompetitiveIntelConfig.from_env()
    config.model = args.model
    config.topn = args.topn
    config.debug = args.debug

    agent = CompetitiveIntelligenceAgent(config)

    try:
        if args.query:
            result = await agent.run_once(args.query)
            print(result)
        else:
            print("Competitive Intelligence Agent started. Type 'exit' to quit.")
            while True:
                try:
                    user_input = input("\n> ")
                    if user_input.strip().lower() in {"exit", "quit"}:
                        print("Goodbye!")
                        break
                    result = await agent.run_once(user_input)
                    print(result)
                except (KeyboardInterrupt, EOFError):
                    print("\nExiting...")
                    break

        if args.outfile:
            agent.save_conversation(args.outfile)
            print(f"Conversation saved to {args.outfile}")
    except Exception as e:
        logging.error("Application failed: %s", e)
        raise


if __name__ == "__main__":
    asyncio.run(main())