Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| Test Firebase connection and collection access with better error handling. | |
| """ | |
| import os | |
| import sys | |
| # Add the parent directory to Python path | |
| current_dir = os.path.dirname(os.path.abspath(__file__)) | |
| sys.path.insert(0, current_dir) | |
| def test_firebase_connection(): | |
| """Test Firebase connection and collection access.""" | |
| print("π Testing Firebase Connection") | |
| print("=" * 50) | |
| try: | |
| from cve_factchecker.firebase_loader import FirebaseNewsLoader, FirebaseConfig | |
| # Initialize loader | |
| loader = FirebaseNewsLoader() | |
| print(f"π‘ Project ID: {loader.project_id}") | |
| print(f"π API Key: {loader.api_key[:20]}...") | |
| # Test 1: Try to fetch a small number of articles from main collection | |
| print(f"\nπ Test 1: Fetching from 'articles' collection...") | |
| try: | |
| articles = loader.fetch_articles(limit=3, language=None) # No language filter first | |
| print(f" β Success: {len(articles)} articles fetched") | |
| if articles: | |
| sample = articles[0] | |
| print(f" π Sample title: {sample.title[:80]}...") | |
| print(f" π Sample content length: {len(sample.content)} chars") | |
| if hasattr(sample, 'language') and sample.language: | |
| print(f" π Sample language: {sample.language}") | |
| else: | |
| print(f" β οΈ No language field found") | |
| except Exception as e: | |
| print(f" β Error: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| # Test 2: Try language filtering | |
| print(f"\nπ Test 2: Testing language filtering...") | |
| try: | |
| english_articles = loader.fetch_articles(limit=3, language="English") | |
| print(f" β English filter: {len(english_articles)} articles") | |
| # Try different language values | |
| for lang in ["english", "en", "EN"]: | |
| try: | |
| test_articles = loader.fetch_articles(limit=1, language=lang) | |
| print(f" π Language '{lang}': {len(test_articles)} articles") | |
| except Exception as e: | |
| print(f" β Language '{lang}' failed: {e}") | |
| except Exception as e: | |
| print(f" β Language filtering error: {e}") | |
| # Test 3: Try dedicated English collection | |
| print(f"\nπ― Test 3: Testing dedicated English collection...") | |
| try: | |
| english_collection_articles = loader.fetch_english_articles(limit=5) | |
| print(f" β English collection: {len(english_collection_articles)} articles") | |
| if english_collection_articles: | |
| sample = english_collection_articles[0] | |
| print(f" π English sample: {sample.title[:80]}...") | |
| except Exception as e: | |
| print(f" β English collection error: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| # Test 4: Network connectivity check | |
| print(f"\nπ Test 4: Basic network connectivity...") | |
| try: | |
| import requests | |
| response = requests.get("https://httpbin.org/ip", timeout=10) | |
| if response.status_code == 200: | |
| print(f" β Internet connection OK") | |
| # Test Firebase endpoint | |
| firebase_test_url = f"https://firestore.googleapis.com/v1/projects/{loader.project_id}" | |
| firebase_response = requests.get(firebase_test_url, timeout=10) | |
| print(f" π‘ Firebase endpoint status: {firebase_response.status_code}") | |
| else: | |
| print(f" β οΈ Network test failed: {response.status_code}") | |
| except Exception as e: | |
| print(f" β Network test error: {e}") | |
| return True | |
| except ImportError as e: | |
| print(f"β Import error: {e}") | |
| return False | |
| except Exception as e: | |
| print(f"β Unexpected error: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| return False | |
| def test_alternative_collection_names(): | |
| """Test if collections exist with different naming patterns.""" | |
| print(f"\nπ Testing Alternative Collection Names") | |
| print("=" * 50) | |
| try: | |
| from cve_factchecker.firebase_loader import FirebaseNewsLoader | |
| loader = FirebaseNewsLoader() | |
| # Test different collection patterns by modifying the config | |
| alternative_names = [ | |
| "Articles", # Capital A | |
| "English_Articles", # Capital E and underscore | |
| "englishArticles", # camelCase | |
| "EnglishArticles", # PascalCase | |
| "news", # Simple name | |
| "documents" # Generic name | |
| ] | |
| for collection_name in alternative_names: | |
| print(f"π Testing collection: '{collection_name}'") | |
| try: | |
| # Temporarily modify the collection name | |
| original_collection = loader.config.ENGLISH_ARTICLES_COLLECTION | |
| loader.config.ENGLISH_ARTICLES_COLLECTION = collection_name | |
| # Try to fetch | |
| articles = loader.fetch_english_articles(limit=1) | |
| print(f" β Found {len(articles)} articles in '{collection_name}'") | |
| # Restore original | |
| loader.config.ENGLISH_ARTICLES_COLLECTION = original_collection | |
| if articles: | |
| return collection_name # Found working collection | |
| except Exception as e: | |
| # Restore original | |
| loader.config.ENGLISH_ARTICLES_COLLECTION = original_collection | |
| print(f" β Collection '{collection_name}' failed: {str(e)[:50]}...") | |
| return None | |
| except Exception as e: | |
| print(f"β Alternative collection test failed: {e}") | |
| return None | |
| def main(): | |
| """Main test function.""" | |
| print("π§ͺ CVE Fact Checker - Firebase Connection Test") | |
| print("=" * 80) | |
| # Basic connection test | |
| success = test_firebase_connection() | |
| if success: | |
| # Test alternative collection names | |
| working_collection = test_alternative_collection_names() | |
| if working_collection: | |
| print(f"\nπ Found working English collection: '{working_collection}'") | |
| else: | |
| print(f"\nπ‘ Recommendation: Use main 'articles' collection with language filtering") | |
| print(f"\nπ Test Summary:") | |
| print(f" Firebase connection: {'β OK' if success else 'β Failed'}") | |
| return success | |
| if __name__ == "__main__": | |
| success = main() | |
| sys.exit(0 if success else 1) |