File size: 7,481 Bytes
ac649ea |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 |
"""
test_social_tools.py
Test for social media tools (session required)
Will identify which sessions need to be created/refreshed
"""
import sys
import os
import io
import json
# Force UTF-8 output
if sys.platform == 'win32':
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
print("="*60)
print("[TEST] SOCIAL MEDIA TOOLS (Session Required)")
print("="*60)
# Check for existing sessions
print("\n[SESSIONS] Checking for existing session files...")
session_dirs = [
"src/utils/.sessions",
".sessions",
]
session_files = {}
for dir in session_dirs:
if os.path.exists(dir):
for f in os.listdir(dir):
if f.endswith('.json'):
session_files[f] = os.path.join(dir, f)
if session_files:
print(f"[OK] Found {len(session_files)} session files:")
for name, path in session_files.items():
print(f" - {name}: {path}")
else:
print("[WARN] No session files found!")
print(" Please create sessions using the session manager")
# Try to create ToolSet with profile scrapers
print("\n[TEST] Creating ToolSet with profile scrapers...")
try:
from src.utils.tool_factory import create_tool_set
tools = create_tool_set(include_profile_scrapers=True)
print("[OK] ToolSet created")
except Exception as e:
print(f"[FAIL] Could not create ToolSet: {e}")
sys.exit(1)
# Test Twitter
print("\n[1] Testing scrape_twitter (keyword search)...")
try:
twitter_tool = tools.get("scrape_twitter")
if twitter_tool:
result = twitter_tool.invoke({"query": "sri lanka", "max_items": 2})
parsed = json.loads(result) if isinstance(result, str) else result
if isinstance(parsed, dict) and "error" in parsed:
if "session" in str(parsed['error']).lower():
print(f"[SESSION] Twitter: Session not found or expired")
print(f" Error: {parsed['error'][:100]}")
else:
print(f"[WARN] Twitter error: {parsed['error'][:100]}")
elif isinstance(parsed, dict):
count = len(parsed.get('results', []))
print(f"[OK] Twitter: {count} tweets fetched")
elif isinstance(parsed, list):
print(f"[OK] Twitter: {len(parsed)} tweets fetched")
else:
print(f"[OK] Twitter returned: {type(parsed)}")
else:
print("[WARN] scrape_twitter not found in ToolSet")
except Exception as e:
print(f"[FAIL] Twitter: {e}")
# Test Facebook
print("\n[2] Testing scrape_facebook (keyword search)...")
try:
fb_tool = tools.get("scrape_facebook")
if fb_tool:
result = fb_tool.invoke({"keywords": ["sri lanka"], "max_items": 2})
parsed = json.loads(result) if isinstance(result, str) else result
if isinstance(parsed, dict) and "error" in parsed:
if "session" in str(parsed['error']).lower():
print(f"[SESSION] Facebook: Session not found or expired")
print(f" Error: {parsed['error'][:100]}")
else:
print(f"[WARN] Facebook error: {parsed['error'][:100]}")
elif isinstance(parsed, dict):
count = len(parsed.get('results', []))
print(f"[OK] Facebook: {count} posts fetched")
elif isinstance(parsed, list):
print(f"[OK] Facebook: {len(parsed)} posts fetched")
else:
print(f"[OK] Facebook returned: {type(parsed)}")
else:
print("[WARN] scrape_facebook not found in ToolSet")
except Exception as e:
print(f"[FAIL] Facebook: {e}")
# Test LinkedIn
print("\n[3] Testing scrape_linkedin (keyword search)...")
try:
linkedin_tool = tools.get("scrape_linkedin")
if linkedin_tool:
result = linkedin_tool.invoke({"keywords": ["sri lanka"], "max_items": 2})
parsed = json.loads(result) if isinstance(result, str) else result
if isinstance(parsed, dict) and "error" in parsed:
if "session" in str(parsed['error']).lower():
print(f"[SESSION] LinkedIn: Session not found or expired")
print(f" Error: {parsed['error'][:100]}")
else:
print(f"[WARN] LinkedIn error: {parsed['error'][:100]}")
elif isinstance(parsed, dict):
count = len(parsed.get('results', []))
print(f"[OK] LinkedIn: {count} posts fetched")
elif isinstance(parsed, list):
print(f"[OK] LinkedIn: {len(parsed)} posts fetched")
else:
print(f"[OK] LinkedIn returned: {type(parsed)}")
else:
print("[WARN] scrape_linkedin not found in ToolSet")
except Exception as e:
print(f"[FAIL] LinkedIn: {e}")
# Test Instagram
print("\n[4] Testing scrape_instagram (hashtag search)...")
try:
instagram_tool = tools.get("scrape_instagram")
if instagram_tool:
result = instagram_tool.invoke({"keywords": ["srilanka"], "max_items": 2})
parsed = json.loads(result) if isinstance(result, str) else result
if isinstance(parsed, dict) and "error" in parsed:
if "session" in str(parsed['error']).lower():
print(f"[SESSION] Instagram: Session not found or expired")
print(f" Error: {parsed['error'][:100]}")
else:
print(f"[WARN] Instagram error: {parsed['error'][:100]}")
elif isinstance(parsed, dict):
count = len(parsed.get('results', []))
print(f"[OK] Instagram: {count} posts fetched")
elif isinstance(parsed, list):
print(f"[OK] Instagram: {len(parsed)} posts fetched")
else:
print(f"[OK] Instagram returned: {type(parsed)}")
else:
print("[WARN] scrape_instagram not found in ToolSet")
except Exception as e:
print(f"[FAIL] Instagram: {e}")
# Test Profile Scrapers
print("\n[5] Testing scrape_twitter_profile (specific account)...")
try:
from src.utils.profile_scrapers import scrape_twitter_profile
result = scrape_twitter_profile.invoke({"username": "SLTMobitel", "max_items": 2})
parsed = json.loads(result) if isinstance(result, str) else result
if isinstance(parsed, dict) and "error" in parsed:
if "session" in str(parsed['error']).lower():
print(f"[SESSION] Twitter Profile: Session not found or expired")
print(f" Error: {parsed['error'][:100]}")
elif "timeout" in str(parsed['error']).lower():
print(f"[TIMEOUT] Twitter Profile: Navigation timed out (X blocks automation)")
print(f" Error: {parsed['error'][:100]}")
else:
print(f"[WARN] Twitter Profile error: {parsed['error'][:100]}")
elif isinstance(parsed, dict):
count = len(parsed.get('results', []))
print(f"[OK] Twitter Profile: {count} tweets fetched from @SLTMobitel")
else:
print(f"[OK] Twitter Profile returned: {type(parsed)}")
except Exception as e:
print(f"[FAIL] Twitter Profile: {e}")
print("\n" + "="*60)
print("[SUMMARY]")
print("="*60)
print("If you see [SESSION] errors, please create new sessions using:")
print(" - Twitter: Run session manager with Twitter login")
print(" - Facebook: Run session manager with Facebook login")
print(" - LinkedIn: Run session manager with LinkedIn login")
print(" - Instagram: Run session manager with Instagram login")
print("\nSession manager: python src/utils/session_manager.py")
print("="*60)
|