""" GEO Search Service - Unified DataForSEO Integration """ from typing import List, Dict, Any, Optional def execute_search( query: str, engine: str, # google, bing, etc. save_id: int, user_id: int, num_results: int = 10, location_code: Optional[int] = None ) -> Dict[str, Any]: """ Execute search query via DataForSEO API. Returns: { "query_id": int, # Database ID "results": [...], # Formatted SERP data "position": int | None, # User's rank "raw_response": {...} # Full API response } """ # Single DataForSEO implementation # Rate limiting, error handling, retry logic # Position calculation # Database storage pass

# geo/api/search.py (BEFORE) url = "https://api.dataforseo.com/v3/serp/google/organic/live/advanced" response = requests.post(url, headers=headers, json=post_data) # ... 50 lines of processing # geo/api/search.py (AFTER) from geo.services.search_service import execute_search result = execute_search(query, "google", save_id, user_id)

# geo/pipeline/mention_scraper.py (BEFORE) DATAFORSEO_SEARCH_URL = "https://api.dataforseo.com/..." resp = requests.post(DATAFORSEO_SEARCH_URL, ...) # ... duplicate DataForSEO code # geo/pipeline/mention_scraper.py (AFTER) from geo.services.search_service import execute_search result = execute_search(query_text, "google", scan_id, user_id)

-- Migration: 016_unify_geo_saves.sql BEGIN; -- 1. Copy data from public.geo_saves to geo.geo_saves INSERT INTO geo.geo_saves SELECT * FROM public.geo_saves ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name, url = EXCLUDED.url, -- ... all fields updated_at = NOW(); -- 2. Update foreign key references ALTER TABLE public.geo_queries DROP CONSTRAINT IF EXISTS geo_queries_save_id_fkey; ALTER TABLE public.geo_queries ADD CONSTRAINT geo_queries_save_id_fkey FOREIGN KEY (save_id) REFERENCES geo.geo_saves(id) ON DELETE CASCADE; -- 3. Drop old table DROP TABLE public.geo_saves; -- 4. Update sequence SELECT setval('geo.geo_saves_id_seq', (SELECT MAX(id) FROM geo.geo_saves)); COMMIT;

# Before: Mixed references db.query("SELECT * FROM geo_saves WHERE ...") # public schema db.query("SELECT * FROM geo.geo_saves WHERE ...") # geo schema # After: Consistent db.query("SELECT * FROM geo.geo_saves WHERE ...") # Always geo schema

-- Migration: 017_centralize_progress.sql BEGIN; -- 1. Remove redundant status columns ALTER TABLE geo.mention_scans DROP COLUMN IF EXISTS status, DROP COLUMN IF EXISTS current_stage, DROP COLUMN IF EXISTS progress_percentage; ALTER TABLE geo.reports DROP COLUMN IF EXISTS status; -- 2. Keep only completion timestamp -- (status is in async_operations, completion is immutable metadata) ALTER TABLE geo.mention_scans ADD COLUMN completed_at TIMESTAMPTZ; ALTER TABLE geo.reports ADD COLUMN completed_at TIMESTAMPTZ; COMMIT;

# Before: Dual writes db.update("geo.mention_scans", scan_id, {"status": "scanning", "progress": 25}) db.update("shared.async_operations", op_id, {"status": "in_progress", "progress": 25}) # After: Single source of truth db.update("shared.async_operations", op_id, { "status": "in_progress", "progress": 25, "current_step": "Scanning web sources..." }) # Frontend queries async_operations only async_op = db.query_one( "SELECT * FROM shared.async_operations WHERE operation_id = %s", (scan_id,) ) return {"status": async_op["status"], "progress": async_op["progress"]}

""" GEO Analysis Service - Sentiment, Relevance, Authority """ from typing import List, Dict, Any def analyze_sentiment( text: str, context: Dict[str, Any] ) -> float: """ Analyze sentiment of text. Returns: -1.0 (very negative) to 1.0 (very positive) """ # GPT-4o analysis pass def score_relevance( result: Dict[str, Any], profile: Dict[str, Any] ) -> int: """ Score how relevant a search result is to user's business. Returns: 0-100 relevance score """ pass def calculate_authority( domain: str, result_metadata: Dict[str, Any] ) -> int: """ Calculate authority score of source domain. Returns: 0-100 authority score """ pass def analyze_search_results( results: List[Dict[str, Any]], profile: Dict[str, Any], include_sentiment: bool = True, include_relevance: bool = True, include_authority: bool = True ) -> List[Dict[str, Any]]: """ Batch analyze search results with optional features. Returns: Results enriched with sentiment, relevance, authority scores """ pass

# Mentions pipeline (always uses classification) from geo.services.analysis_service import analyze_search_results enriched_results = analyze_search_results( results=raw_search_results, profile=mention_profile, include_sentiment=True, include_relevance=True, include_authority=True ) # Reports (optional classification) if user_settings.get("enable_classification"): enriched_results = analyze_search_results( results=report_search_results, profile=report_profile, include_sentiment=False, # Maybe not useful for ranking reports include_relevance=True, # But relevance scoring could be valuable include_authority=True ) else: enriched_results = report_search_results # Use raw