""" Hunter Companion - Code-First Architecture This module handles the Hunter Companion chat WITHOUT relying on LLM for intent detection. All routing is done via code (regex, keywords). LLM is only used as a last resort. """ from __future__ import annotations import re from dataclasses import dataclass from enum import Enum from typing import Any, Optional from sqlalchemy.ext.asyncio import AsyncSession from app.models.user import User from app.services.llm_tools import execute_tool class Intent(Enum): GREETING = "greeting" HELP = "help" ANALYZE_DOMAIN = "analyze_domain" WATCHLIST = "watchlist" PORTFOLIO = "portfolio" AUCTIONS = "auctions" DROPS = "drops" LISTINGS = "listings" INBOX = "inbox" SNIPER = "sniper" YIELD = "yield" UNCLEAR = "unclear" @dataclass class DetectedIntent: intent: Intent params: dict[str, Any] confidence: float # ============================================================================ # INTENT DETECTION (Pure Code - No LLM) # ============================================================================ def detect_intent(text: str) -> DetectedIntent: """Detect user intent using pattern matching. No LLM involved.""" t = (text or "").strip().lower() # 1. Greeting if _is_greeting(t): return DetectedIntent(Intent.GREETING, {}, 1.0) # 2. Help / Capabilities if _is_help(t): return DetectedIntent(Intent.HELP, {}, 1.0) # 3. Domain Analysis (detect domain in text) domain = _extract_domain(text) if domain: return DetectedIntent(Intent.ANALYZE_DOMAIN, {"domain": domain}, 0.95) # 4. Watchlist queries if _matches_keywords(t, ["watchlist", "tracking", "monitored", "watching"]): return DetectedIntent(Intent.WATCHLIST, {}, 0.9) # 5. Portfolio queries if _matches_keywords(t, ["portfolio", "my domains", "owned", "roi", "investment", "value"]): return DetectedIntent(Intent.PORTFOLIO, {}, 0.9) # 6. Auction queries if _matches_keywords(t, ["auction", "ending", "deals", "bid", "expir"]): hours = _extract_hours(t) return DetectedIntent(Intent.AUCTIONS, {"hours": hours}, 0.9) # 7. Drops if _matches_keywords(t, ["drop", "dropped", "zone file", "expired"]): return DetectedIntent(Intent.DROPS, {}, 0.9) # 8. Listings / For Sale if _matches_keywords(t, ["listing", "for sale", "selling", "leads"]): return DetectedIntent(Intent.LISTINGS, {}, 0.9) # 9. Inbox if _matches_keywords(t, ["inbox", "message", "unread", "conversation"]): return DetectedIntent(Intent.INBOX, {}, 0.9) # 10. Sniper alerts if _matches_keywords(t, ["sniper", "alert", "notification", "filter"]): return DetectedIntent(Intent.SNIPER, {}, 0.9) # 11. Yield if _matches_keywords(t, ["yield", "earning", "revenue", "monetiz"]): return DetectedIntent(Intent.YIELD, {}, 0.9) # Fallback: unclear intent return DetectedIntent(Intent.UNCLEAR, {"original": text}, 0.3) def _is_greeting(t: str) -> bool: greetings = {"hi", "hello", "hey", "yo", "sup", "hola", "hallo", "moin", "servus"} clean = re.sub(r'[!?.,]', '', t).strip() return clean in greetings or (len(clean) <= 10 and any(g in clean for g in greetings)) def _is_help(t: str) -> bool: patterns = [ "what can you do", "what do you do", "help", "how can you help", "capabilities", "features", "was kannst du", "hilfe", ] return any(p in t for p in patterns) def _matches_keywords(t: str, keywords: list[str]) -> bool: return any(kw in t for kw in keywords) def _extract_domain(text: str) -> Optional[str]: """Extract a domain from text using regex.""" # Match patterns like: example.com, sub.example.co.uk, etc. pattern = r'\b([a-zA-Z0-9][-a-zA-Z0-9]*\.)+[a-zA-Z]{2,}\b' matches = re.findall(pattern, text) if not matches: return None # Reconstruct full domain from the match full_match = re.search(pattern, text) if full_match: domain = full_match.group(0).lower() # Filter out common non-domains if domain not in {"example.com", "test.com", "domain.com", "e.g.", "i.e."}: return domain return None def _extract_hours(t: str) -> int: """Extract hours from text like '2 hours' or 'next 4h'.""" match = re.search(r'(\d+)\s*(?:hour|h\b|hr)', t) if match: return min(int(match.group(1)), 168) # Max 1 week return 24 # Default # ============================================================================ # RESPONSE HANDLERS (No LLM - Template Based) # ============================================================================ RESPONSES = { Intent.GREETING: """Hey! I'm your Hunter Companion. I can help you with: • Analyze any domain - just type it • Show auctions ending soon • Check your watchlist or portfolio • Find recently dropped domains What would you like to do?""", Intent.HELP: """Here's what I can do: • Domain Analysis - Type any domain (e.g., "startup.io") • Auctions - "show auctions" or "ending soon" • Watchlist - "show my watchlist" • Portfolio - "portfolio summary" or "my ROI" • Drops - "show drops" or "dropped domains" • Listings - "my listings" or "leads" Just ask naturally or type a domain to analyze!""", Intent.UNCLEAR: """I'm not sure what you're looking for. Try one of these: • Type a domain to analyze (e.g., "crypto.io") • "show auctions ending soon" • "my watchlist" • "portfolio summary" What would you like to do?""", } async def handle_intent( db: AsyncSession, user: User, detected: DetectedIntent, path: str, ) -> str: """Handle the detected intent and return a response.""" # Static responses if detected.intent in RESPONSES: return RESPONSES[detected.intent] # Dynamic responses (tool-based) try: if detected.intent == Intent.ANALYZE_DOMAIN: return await _handle_analyze(db, user, detected.params.get("domain", ""), path) if detected.intent == Intent.WATCHLIST: return await _handle_watchlist(db, user, path) if detected.intent == Intent.PORTFOLIO: return await _handle_portfolio(db, user, path) if detected.intent == Intent.AUCTIONS: return await _handle_auctions(db, user, detected.params.get("hours", 24), path) if detected.intent == Intent.DROPS: return await _handle_drops(db, user, path) if detected.intent == Intent.LISTINGS: return await _handle_listings(db, user, path) if detected.intent == Intent.INBOX: return await _handle_inbox(db, user, path) if detected.intent == Intent.SNIPER: return await _handle_sniper(db, user, path) if detected.intent == Intent.YIELD: return await _handle_yield(db, user, path) except Exception as e: return f"Sorry, something went wrong: {str(e)}" return RESPONSES[Intent.UNCLEAR] async def _handle_analyze(db: AsyncSession, user: User, domain: str, path: str) -> str: """Handle domain analysis.""" if not domain: return "Please provide a domain to analyze." result = await execute_tool(db, user, "analyze_domain", {"domain": domain}, path=path) if "error" in result: return f"Could not analyze {domain}: {result['error']}" # Format the analysis result score = result.get("pounce_score", 0) # Determine recommendation if score >= 70: rec = "BUY" rec_emoji = "✓" elif score >= 50: rec = "CONSIDER" rec_emoji = "~" else: rec = "SKIP" rec_emoji = "✗" lines = [ f"Analysis: {domain}", f"", f"Pounce Score: {score}/100", f"Recommendation: {rec_emoji} {rec}", ] # Availability avail = result.get("availability", {}) if avail: status = "Available" if avail.get("is_available") else "Taken" lines.append(f"Status: {status}") # Value estimate value = result.get("value", {}) if value and value.get("estimated_value"): lines.append(f"Est. Value: ${value['estimated_value']:,.0f}") # Risk risk = result.get("risk", {}) if risk and risk.get("risk_level"): lines.append(f"Risk: {risk['risk_level']}") # Radio test radio = result.get("radio_test", {}) if radio and radio.get("score") is not None: lines.append(f"Radio Test: {radio['score']}/100") return "\n".join(lines) async def _handle_watchlist(db: AsyncSession, user: User, path: str) -> str: """Handle watchlist query.""" result = await execute_tool(db, user, "list_watchlist", {"per_page": 10}, path=path) if "error" in result: return f"Could not load watchlist: {result['error']}" domains = result.get("domains", []) total = result.get("total", 0) if not domains: return "Your watchlist is empty.\n\nAdd domains from the Hunt page to start tracking." lines = [f"Your Watchlist ({total} domains)", ""] for d in domains[:10]: status = "✓" if d.get("is_available") else "•" lines.append(f"{status} {d.get('name', 'unknown')}") if total > 10: lines.append(f"\n... and {total - 10} more") return "\n".join(lines) async def _handle_portfolio(db: AsyncSession, user: User, path: str) -> str: """Handle portfolio query.""" result = await execute_tool(db, user, "portfolio_summary", {}, path=path) if "error" in result: return f"Could not load portfolio: {result['error']}" total = result.get("total_domains", 0) if total == 0: return "Your portfolio is empty.\n\nAdd domains you own to track ROI and renewals." lines = [ f"Portfolio Summary", "", f"Total Domains: {total}", f"Active: {result.get('active_domains', 0)}", f"Sold: {result.get('sold_domains', 0)}", "", f"Total Cost: ${result.get('total_cost', 0):,.0f}", f"Est. Value: ${result.get('total_estimated_value', 0):,.0f}", ] roi = result.get("overall_roi_percent") if roi is not None: sign = "+" if roi >= 0 else "" lines.append(f"ROI: {sign}{roi:.1f}%") return "\n".join(lines) async def _handle_auctions(db: AsyncSession, user: User, hours: int, path: str) -> str: """Handle auctions query.""" result = await execute_tool( db, user, "market_feed", {"ending_within_hours": hours, "limit": 10, "sort_by": "time"}, path=path ) if "error" in result: return f"Could not load auctions: {result['error']}" items = result.get("items", []) if not items: return f"No auctions ending in the next {hours} hours." lines = [f"Auctions Ending Soon ({len(items)} found)", ""] for a in items[:10]: domain = a.get("domain", "unknown") bid = a.get("current_bid", 0) platform = a.get("platform", "") lines.append(f"• {domain} - ${bid:.0f} ({platform})") return "\n".join(lines) async def _handle_drops(db: AsyncSession, user: User, path: str) -> str: """Handle drops query.""" result = await execute_tool( db, user, "get_drops", {"hours": 24, "limit": 15}, path=path ) if "error" in result: return f"Could not load drops: {result['error']}" domains = result.get("domains", []) total = result.get("total", 0) if not domains: return "No recently dropped domains found." lines = [f"Recently Dropped ({total} total)", ""] for d in domains[:15]: name = d if isinstance(d, str) else d.get("domain", "unknown") lines.append(f"• {name}") if total > 15: lines.append(f"\n... and {total - 15} more") return "\n".join(lines) async def _handle_listings(db: AsyncSession, user: User, path: str) -> str: """Handle listings query.""" result = await execute_tool(db, user, "list_my_listings", {"limit": 10}, path=path) if "error" in result: return f"Could not load listings: {result['error']}" items = result.get("items", []) if not items: return "You have no active listings.\n\nGo to For Sale to create your first listing." lines = [f"Your Listings ({len(items)})", ""] for l in items[:10]: domain = l.get("domain", "unknown") status = l.get("status", "") price = l.get("asking_price") price_str = f"${price:,.0f}" if price else "Make Offer" lines.append(f"• {domain} - {price_str} [{status}]") return "\n".join(lines) async def _handle_inbox(db: AsyncSession, user: User, path: str) -> str: """Handle inbox query.""" result = await execute_tool(db, user, "get_inbox_counts", {}, path=path) if "error" in result: return f"Could not load inbox: {result['error']}" buyer = result.get("buyer_unread", 0) seller = result.get("seller_unread", 0) total = buyer + seller if total == 0: return "No unread messages.\n\nYour inbox is all caught up!" lines = [ f"Inbox Summary", "", f"Unread: {total}", f"• As Buyer: {buyer}", f"• As Seller: {seller}", "", "Go to Inbox to view conversations.", ] return "\n".join(lines) async def _handle_sniper(db: AsyncSession, user: User, path: str) -> str: """Handle sniper alerts query.""" result = await execute_tool(db, user, "list_sniper_alerts", {"limit": 10}, path=path) if "error" in result: return f"Could not load alerts: {result['error']}" items = result.get("items", []) if not items: return "No sniper alerts configured.\n\nGo to Sniper to create custom alerts." active = sum(1 for a in items if a.get("is_active")) lines = [f"Sniper Alerts ({active} active)", ""] for a in items[:10]: name = a.get("name", "Unnamed") status = "✓" if a.get("is_active") else "○" lines.append(f"{status} {name}") return "\n".join(lines) async def _handle_yield(db: AsyncSession, user: User, path: str) -> str: """Handle yield query.""" result = await execute_tool(db, user, "yield_dashboard", {}, path=path) if "error" in result: return f"Could not load yield data: {result['error']}" stats = result.get("stats", {}) if stats.get("total_domains", 0) == 0: return "No yield domains configured.\n\nGo to Yield to activate monetization." lines = [ "Yield Summary", "", f"Active Domains: {stats.get('active_domains', 0)}", f"Monthly Revenue: ${stats.get('monthly_revenue', 0):.2f}", f"Monthly Clicks: {stats.get('monthly_clicks', 0)}", f"Conversions: {stats.get('monthly_conversions', 0)}", ] return "\n".join(lines) # ============================================================================ # MAIN ENTRY POINT # ============================================================================ async def process_message( db: AsyncSession, user: User, message: str, path: str, ) -> str: """ Process a user message and return a response. This is the main entry point for the Hunter Companion. """ # Step 1: Detect intent (pure code, no LLM) detected = detect_intent(message) # Step 2: Handle the intent and generate response response = await handle_intent(db, user, detected, path) return response