|
|
"""
|
|
|
MongoDB Database Connection Manager
|
|
|
Author: AI Generated
|
|
|
Created: 2025-11-24
|
|
|
Purpose: Handle MongoDB connection and collection access
|
|
|
"""
|
|
|
|
|
|
from pymongo import MongoClient
|
|
|
from pymongo.database import Database
|
|
|
from pymongo.collection import Collection
|
|
|
from config import settings
|
|
|
|
|
|
|
|
|
class DatabaseManager:
|
|
|
"""Singleton MongoDB connection manager"""
|
|
|
|
|
|
_instance = None
|
|
|
_client: MongoClient = None
|
|
|
_db: Database = None
|
|
|
|
|
|
def __new__(cls):
|
|
|
if cls._instance is None:
|
|
|
cls._instance = super().__new__(cls)
|
|
|
return cls._instance
|
|
|
|
|
|
def __init__(self):
|
|
|
if self._client is None:
|
|
|
self.connect()
|
|
|
|
|
|
@property
|
|
|
def client(self):
|
|
|
"""Get MongoDB client"""
|
|
|
return self._client
|
|
|
|
|
|
@property
|
|
|
def db_name(self):
|
|
|
"""Get database name"""
|
|
|
return settings.DB_NAME
|
|
|
|
|
|
def connect(self):
|
|
|
"""Establish connection to MongoDB"""
|
|
|
self._client = MongoClient(settings.MONGODB_URI)
|
|
|
self._db = self._client[settings.DB_NAME]
|
|
|
print(f"✓ Connected to MongoDB: {settings.DB_NAME}")
|
|
|
|
|
|
def get_collection(self, collection_name: str) -> Collection:
|
|
|
"""Get a MongoDB collection"""
|
|
|
return self._db[collection_name]
|
|
|
|
|
|
def close(self):
|
|
|
"""Close MongoDB connection"""
|
|
|
if self._client:
|
|
|
self._client.close()
|
|
|
print("✓ MongoDB connection closed")
|
|
|
|
|
|
|
|
|
@property
|
|
|
def users(self) -> Collection:
|
|
|
"""User collection (contains embedded UserFollows array)"""
|
|
|
return self.get_collection(settings.COLLECTION_USERS)
|
|
|
|
|
|
@property
|
|
|
def payments(self) -> Collection:
|
|
|
"""Payment collection"""
|
|
|
return self.get_collection(settings.COLLECTION_PAYMENTS)
|
|
|
|
|
|
@property
|
|
|
def event_versions(self) -> Collection:
|
|
|
"""EventVersion collection"""
|
|
|
return self.get_collection(settings.COLLECTION_EVENT_VERSIONS)
|
|
|
|
|
|
@property
|
|
|
def post_social_media(self) -> Collection:
|
|
|
"""PostSocialMedia collection (contains nested Images.UserCommentPosts)"""
|
|
|
return self.get_collection(settings.COLLECTION_POST_SOCIAL_MEDIA)
|
|
|
|
|
|
|
|
|
@property
|
|
|
def audience_segments(self) -> Collection:
|
|
|
"""AudienceSegment collection (DEPRECATED - use event_audience_segments)"""
|
|
|
return self.get_collection(settings.COLLECTION_AUDIENCE_SEGMENTS)
|
|
|
|
|
|
@property
|
|
|
def user_segment_assignments(self) -> Collection:
|
|
|
"""UserSegmentAssignment collection"""
|
|
|
return self.get_collection(settings.COLLECTION_USER_SEGMENT_ASSIGNMENTS)
|
|
|
|
|
|
@property
|
|
|
def sentiment_results(self) -> Collection:
|
|
|
"""SentimentAnalysisResult collection"""
|
|
|
return self.get_collection(settings.COLLECTION_SENTIMENT_RESULTS)
|
|
|
|
|
|
@property
|
|
|
def event_insights(self) -> Collection:
|
|
|
"""EventInsightReport collection"""
|
|
|
return self.get_collection(settings.COLLECTION_EVENT_INSIGHTS)
|
|
|
|
|
|
|
|
|
@property
|
|
|
def event_audience_segments(self) -> Collection:
|
|
|
"""EventAudienceSegment collection"""
|
|
|
return self.get_collection("EventAudienceSegment")
|
|
|
|
|
|
@property
|
|
|
def event_sentiment_summary(self) -> Collection:
|
|
|
"""EventSentimentSummary collection"""
|
|
|
return self.get_collection("EventSentimentSummary")
|
|
|
|
|
|
|
|
|
|
|
|
db = DatabaseManager()
|
|
|
|
