#!/usr/bin/env python3 """ Debug script: diagnose WHY ownership/history returns "Unknown". Tests WHOIS, DuckDuckGo search, about-page scraping, and full research pipelines. Usage: python3 debug_research.py # test default outlet python3 debug_research.py "Frontiers in Anesthesiology" "frontiersin.org" python3 debug_research.py --all # test first 10 from dataset """ import sys import json import time import logging logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(message)s") logger = logging.getLogger(__name__) def test_whois(domain: str) -> dict: """Test 1: Is WHOIS working?""" result = {"test": "whois", "domain": domain, "status": "unknown"} try: import whois w = whois.whois(domain) creation = w.creation_date if isinstance(creation, list): creation = creation[0] result["creation_date"] = str(creation) if creation else None result["registrar"] = getattr(w, "registrar", None) result["org"] = getattr(w, "org", None) result["status"] = "ok" if creation else "no_creation_date" except ImportError: result["status"] = "not_installed" result["fix"] = "pip install python-whois" except Exception as e: result["status"] = "error" result["error"] = f"{type(e).__name__}: {e}" return result def test_ddg_search(query: str) -> dict: """Test 2: Is DuckDuckGo search returning results?""" result = {"test": "ddg_search", "query": query, "status": "unknown"} try: from search_backends import DDGSearchBackend backend = DDGSearchBackend() results = backend.search(query, max_results=5) result["num_results"] = len(results) if results else 0 result["status"] = "ok" if results else "no_results" if results: result["top_results"] = [ {"title": r.get("title", "")[:80], "body": r.get("body", "")[:120]} for r in results[:3] ] except Exception as e: result["status"] = "error" result["error"] = f"{type(e).__name__}: {e}" return result def test_about_page_scrape(domain: str) -> dict: """Test 3: Can we scrape the about page?""" result = {"test": "about_page_scrape", "domain": domain, "status": "unknown"} try: from research import MediaResearcher researcher = MediaResearcher(model="gpt-5-mini-2025-08-07") about_text = researcher._scrape_about_page(domain) result["status"] = "ok" if about_text else "no_content" if about_text: result["content_length"] = len(about_text) result["preview"] = about_text[:300] except AttributeError as e: result["status"] = "no_method" result["note"] = f"MediaResearcher has no _scrape_about_page: {e}" except Exception as e: result["status"] = "error" result["error"] = f"{type(e).__name__}: {e}" return result def test_research_history(outlet_name: str, domain: str) -> dict: """Test 4: Does the full history research pipeline work?""" result = {"test": "research_history", "outlet": outlet_name, "status": "unknown"} try: from research import MediaResearcher researcher = MediaResearcher(model="gpt-5-mini-2025-08-07") t0 = time.time() history = researcher.research_history(outlet_name, domain) elapsed = time.time() - t0 result["elapsed_s"] = round(elapsed, 2) result["summary"] = history.summary[:300] if history.summary else None result["confidence"] = history.confidence result["founded"] = getattr(history, "founded", None) result["founder"] = getattr(history, "founder", None) result["status"] = "ok" if history.confidence and history.confidence > 0 else "low_confidence" except Exception as e: result["status"] = "error" result["error"] = f"{type(e).__name__}: {e}" return result def test_research_ownership(outlet_name: str, domain: str) -> dict: """Test 5: Does the full ownership research pipeline work?""" result = {"test": "research_ownership", "outlet": outlet_name, "status": "unknown"} try: from research import MediaResearcher researcher = MediaResearcher(model="gpt-5-mini-2025-08-07") t0 = time.time() ownership = researcher.research_ownership(outlet_name, domain) elapsed = time.time() - t0 result["elapsed_s"] = round(elapsed, 2) result["notes"] = ownership.notes[:300] if ownership.notes else None result["confidence"] = ownership.confidence result["owner"] = getattr(ownership, "owner", None) result["parent_company"] = getattr(ownership, "parent_company", None) result["status"] = "ok" if ownership.confidence and ownership.confidence > 0 else "low_confidence" except Exception as e: result["status"] = "error" result["error"] = f"{type(e).__name__}: {e}" return result def test_wikipedia(outlet_name: str, domain: str) -> dict: """Test 6: Does the Wikipedia API return useful content?""" import requests as _requests result = {"test": "wikipedia_api", "outlet": outlet_name, "status": "unknown"} API = "https://en.wikipedia.org/w/api.php" HEADERS = { "User-Agent": "MediaProfilingBot/1.0 (research tool; contact: media-profiling@example.com)", "Accept": "application/json", } # Build same search variants as _fetch_wikipedia does search_terms = [outlet_name] if domain: search_terms.append(domain) domain_base = domain.split(".")[0] if domain_base.lower() not in ("www", "the"): search_terms.append(domain_base) result["tried_terms"] = search_terms for term in search_terms: try: resp = _requests.get(API, headers=HEADERS, params={ "action": "query", "list": "search", "srsearch": term, "srlimit": 3, "format": "json", }, timeout=10) if resp.status_code != 200: result[f"search_{term}"] = f"HTTP {resp.status_code}" continue data = resp.json() search_results = data.get("query", {}).get("search", []) if not search_results: result[f"search_{term}"] = "0 results" continue page_title = search_results[0]["title"] result[f"search_{term}"] = f"found '{page_title}'" # Get extract resp2 = _requests.get(API, headers=HEADERS, params={ "action": "query", "titles": page_title, "prop": "extracts", "exintro": False, "explaintext": True, "exlimit": 1, "format": "json", }, timeout=10) if resp2.status_code != 200: continue pages = resp2.json().get("query", {}).get("pages", {}) for page in pages.values(): extract = page.get("extract", "") if extract and len(extract) > 100: result["status"] = "ok" result["matched_term"] = term result["article_title"] = page_title result["content_length"] = len(extract) result["preview"] = extract[:300] return result else: result[f"extract_{term}"] = f"too short ({len(extract)} chars)" except Exception as e: result[f"error_{term}"] = f"{type(e).__name__}: {e}" result["status"] = "no_content" return result def test_full_profiler(outlet_name: str, url: str) -> dict: """Test 6: Does the full MediaProfiler pipeline work end-to-end?""" result = {"test": "full_profiler", "outlet": outlet_name, "status": "unknown"} try: from research import MediaProfiler profiler = MediaProfiler(model="gpt-5-mini-2025-08-07") t0 = time.time() report = profiler.analyze(url, outlet_name=outlet_name) elapsed = time.time() - t0 result["elapsed_s"] = round(elapsed, 2) # Check key fields result["history_summary"] = getattr(report, "history_summary", None) result["ownership_owner"] = getattr(report, "ownership_owner", None) result["ownership_hq"] = getattr(report, "ownership_headquarters", None) # Try to dig into sub-results if hasattr(report, "researcher_results"): rr = report.researcher_results if hasattr(rr, "history"): h = rr.history result["history_founded"] = getattr(h, "founded", None) result["history_founder"] = getattr(h, "founder", None) result["history_confidence"] = getattr(h, "confidence", None) if hasattr(rr, "ownership"): o = rr.ownership result["ownership_owner"] = getattr(o, "owner", None) result["ownership_parent"] = getattr(o, "parent_company", None) result["ownership_confidence"] = getattr(o, "confidence", None) result["status"] = "ok" except Exception as e: result["status"] = "error" result["error"] = f"{type(e).__name__}: {e}" return result def run_diagnostics(outlet_name: str, domain: str, source_url: str = ""): if not source_url: source_url = f"https://{domain}" print(f"\n{'='*70}") print(f"DIAGNOSING: {outlet_name} ({domain})") print(f"{'='*70}\n") tests = [ ("WHOIS", lambda: test_whois(domain)), ("DDG: history query", lambda: test_ddg_search( f'"{outlet_name}" about us founded history media news organization' )), ("DDG: ownership query", lambda: test_ddg_search( f'"{outlet_name}" ownership owner parent company funded by headquarters' )), ("About page scrape", lambda: test_about_page_scrape(domain)), ("Wikipedia API", lambda: test_wikipedia(outlet_name, domain)), ("Full history pipeline", lambda: test_research_history(outlet_name, domain)), ("Full ownership pipeline", lambda: test_research_ownership(outlet_name, domain)), ] results = [] for name, test_fn in tests: print(f"\n--- {name} ---") t0 = time.time() r = test_fn() elapsed = time.time() - t0 r["wall_time_s"] = round(elapsed, 2) results.append(r) status_icon = { "ok": "PASS", "no_results": "FAIL", "no_content": "FAIL", "low_confidence": "WARN", "error": "ERROR", "not_installed": "SKIP", "no_method": "SKIP", }.get(r["status"], "????") print(f" [{status_icon}] {r['status']} ({r['wall_time_s']}s)") for k, v in r.items(): if k not in ("test", "status", "wall_time_s", "domain", "query", "outlet"): if isinstance(v, str) and len(v) > 200: print(f" {k}: {v[:200]}...") else: print(f" {k}: {v}") # Summary print(f"\n{'='*70}") print("SUMMARY") print(f"{'='*70}") for r in results: icon = {"ok": "OK", "error": "ERR", "no_results": "FAIL", "no_content": "FAIL", "low_confidence": "WARN", "not_installed": "SKIP", "no_method": "SKIP"}.get(r["status"], "???") print(f" [{icon:4s}] {r['test']}: {r['status']}") return results def main(): if len(sys.argv) >= 3: outlet_name = sys.argv[1] domain = sys.argv[2] source_url = sys.argv[3] if len(sys.argv) >= 4 else "" run_diagnostics(outlet_name, domain, source_url) elif len(sys.argv) == 2 and sys.argv[1] == "--all": with open("evaluation_dataset.json") as f: data = json.load(f) for item in data[:10]: name = item["name"] url = item["source_url"] from urllib.parse import urlparse domain = urlparse(url).netloc.replace("www.", "") run_diagnostics(name, domain, url) else: # Default: test the outlet from the user's result run_diagnostics( "Frontiers in Anesthesiology", "frontiersin.org", "https://www.frontiersin.org/journals/anesthesiology", ) if __name__ == "__main__": main()