| |
| """Quick test for trending topics: AI occasions + niche trends. Run from repo root: python scripts/test_trends.py""" |
|
|
| import asyncio |
| import sys |
| import os |
|
|
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) |
|
|
|
|
| async def main(): |
| from services.current_occasions import get_current_occasions |
|
|
| print("=== 1. Current occasions (AI) ===\n") |
| occasions = await get_current_occasions() |
| if not occasions: |
| print("No occasions returned (AI may have failed or returned empty).") |
| else: |
| for i, o in enumerate(occasions, 1): |
| print(f" {i}. {o['title']}") |
| print(f" {o['summary']}") |
| print() |
| print(f"Total: {len(occasions)} occasions\n") |
|
|
| try: |
| from services.trend_monitor import trend_monitor |
| print("=== 2. Full trends for niche 'home_insurance' (occasions + news) ===\n") |
| data = await trend_monitor.get_relevant_trends_for_niche("home_insurance") |
| trends = data.get("relevant_trends") or [] |
| for i, t in enumerate(trends[:8], 1): |
| print(f" {i}. [{t.get('category', '?')}] {t.get('title', '')}") |
| print(f" {(t.get('summary') or '')[:120]}...") |
| print() |
| print(f"Total trends: {len(trends)}") |
| except ImportError as e: |
| print("=== 2. Skipping full trends (missing dependency):", e) |
| print(" Install with: pip install gnews") |
|
|
|
|
| if __name__ == "__main__": |
| asyncio.run(main()) |
|
|