#MongoDB 实战教程

from pymongo import MongoClient
from bson import ObjectId
from datetime import datetime
import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

# 同步连接
client = MongoClient(
    "mongodb://admin:password@localhost:27017/",
    serverSelectionTimeoutMS=5000  # 5秒超时
)

# 获取数据库
db = client.myapp_database
users_collection = db.users

# 异步连接
async def get_async_client():
    client = AsyncIOMotorClient("mongodb://admin:password@localhost:27017/")
    return client

# 插入文档
def insert_user(user_data):
    result = users_collection.insert_one(user_data)
    return result.inserted_id

# 批量插入
def insert_multiple_users(users_list):
    result = users_collection.insert_many(users_list)
    return result.inserted_ids

# 查询文档
def find_user_by_email(email):
    user = users_collection.find_one({"email": email})
    return user

def find_users_by_age(min_age, max_age):
    users = list(users_collection.find({
        "age": {"$gte": min_age, "$lte": max_age}
    }).sort("age", 1))
    return users

# 更新文档
def update_user_age(user_id, new_age):
    result = users_collection.update_one(
        {"_id": ObjectId(user_id)},
        {"$set": {"age": new_age, "updatedAt": datetime.now()}}
    )
    return result.modified_count

# 删除文档
def delete_user(user_id):
    result = users_collection.delete_one({"_id": ObjectId(user_id)})
    return result.deleted_count

# 聚合查询
def get_user_statistics():
    pipeline = [
        {"$group": {
            "_id": None,
            "total_users": {"$sum": 1},
            "avg_age": {"$avg": "$age"},
            "max_age": {"$max": "$age"},
            "min_age": {"$min": "$age"}
        }}
    ]
    result = list(users_collection.aggregate(pipeline))
    return result[0] if result else {}

# 使用连接池
from pymongo.errors import ConnectionFailure

def create_client_with_pool():
    client = MongoClient(
        "mongodb://admin:password@localhost:27017/",
        maxPoolSize=50,  # 最大连接池大小
        minPoolSize=10,  # 最小连接池大小
        maxIdleTimeMS=30000,  # 最大空闲时间
        serverSelectionTimeoutMS=5000,
        connectTimeoutMS=20000,
        socketTimeoutMS=20000
    )
    
    # 测试连接
    try:
        client.admin.command('ping')
        print("MongoDB connection successful")
    except ConnectionFailure:
        print("MongoDB server not available")
    
    return client

# 事务操作（MongoDB 4.0+）
def perform_transaction():
    client = create_client_with_pool()
    db = client.myapp_database
    
    with client.start_session() as session:
        with session.start_transaction():
            # 扣款操作
            result1 = db.accounts.update_one(
                {"account_id": "acc1", "balance": {"$gte": 100}},
                {"$inc": {"balance": -100}},
                session=session
            )
            
            if result1.modified_count == 0:
                raise Exception("Insufficient funds")
            
            # 入账操作
            result2 = db.accounts.update_one(
                {"account_id": "acc2"},
                {"$inc": {"balance": 100}},
                session=session
            )
            
            print("Transaction completed successfully")

# 异步操作
async def async_operations():
    client = AsyncIOMotorClient("mongodb://admin:password@localhost:27017/")
    db = client.myapp_database
    collection = db.users
    
    # 异步插入
    user_doc = {
        "name": "异步用户",
        "email": "async@example.com",
        "created_at": datetime.now()
    }
    result = await collection.insert_one(user_doc)
    
    # 异步查询
    user = await collection.find_one({"_id": result.inserted_id})
    
    # 异常处理
    try:
        await collection.find_one_and_update(
            {"email": "async@example.com"},
            {"$set": {"updated_at": datetime.now()}}
        )
    except Exception as e:
        print(f"Update error: {e}")
    
    return user

# 监听变更流
def watch_changes():
    collection = db.users
    
    # 监听集合变更
    with collection.watch() as stream:
        for change in stream:
            print(f"Change detected: {change}")
            # 处理变更事件
            if change['operationType'] == 'insert':
                print(f"New document inserted: {change['fullDocument']}")
            elif change['operationType'] == 'update':
                print(f"Document updated: {change['documentKey']}")
            elif change['operationType'] == 'delete':
                print(f"Document deleted: {change['documentKey']}")

from pymongo import MongoClient
from bson import ObjectId
from datetime import datetime
from typing import Optional, List, Dict
import hashlib

class UserManager:
    def __init__(self, connection_string: str, database_name: str):
        self.client = MongoClient(connection_string)
        self.db = self.client[database_name]
        self.collection = self.db.users
        
        # 创建索引
        self.collection.create_index("email", unique=True)
        self.collection.create_index("username", unique=True)
        self.collection.create_index("created_at")
    
    def hash_password(self, password: str) -> str:
        """哈希密码"""
        return hashlib.sha256(password.encode()).hexdigest()
    
    def create_user(self, user_data: Dict) -> str:
        """创建用户"""
        user_data['password'] = self.hash_password(user_data['password'])
        user_data['created_at'] = datetime.now()
        user_data['updated_at'] = datetime.now()
        user_data['is_active'] = True
        
        result = self.collection.insert_one(user_data)
        return str(result.inserted_id)
    
    def get_user_by_email(self, email: str) -> Optional[Dict]:
        """通过邮箱获取用户"""
        user = self.collection.find_one({"email": email})
        return user
    
    def authenticate_user(self, email: str, password: str) -> Optional[Dict]:
        """用户认证"""
        hashed_password = self.hash_password(password)
        user = self.collection.find_one({
            "email": email,
            "password": hashed_password,
            "is_active": True
        })
        return user
    
    def update_user_profile(self, user_id: str, profile_data: Dict) -> int:
        """更新用户资料"""
        result = self.collection.update_one(
            {"_id": ObjectId(user_id)},
            {
                "$set": {
                    **profile_data,
                    "updated_at": datetime.now()
                }
            }
        )
        return result.modified_count
    
    def get_users_paginated(self, page: int = 1, limit: int = 10) -> List[Dict]:
        """分页获取用户"""
        skip = (page - 1) * limit
        users = list(
            self.collection.find({})
            .skip(skip)
            .limit(limit)
            .sort("created_at", -1)
        )
        return users
    
    def get_user_statistics(self) -> Dict:
        """获取用户统计信息"""
        pipeline = [
            {
                "$group": {
                    "_id": None,
                    "total_users": {"$sum": 1},
                    "active_users": {
                        "$sum": {"$cond": [{"$eq": ["$is_active", True]}, 1, 0]}
                    },
                    "avg_age": {"$avg": "$age"},
                    "users_by_city": {
                        "$push": "$address.city"
                    }
                }
            }
        ]
        result = list(self.collection.aggregate(pipeline))
        return result[0] if result else {}

# 使用示例
user_manager = UserManager(
    "mongodb://admin:password@localhost:27017/", 
    "myapp_database"
)

# 创建用户
user_id = user_manager.create_user({
    "username": "john_doe",
    "email": "john@example.com",
    "password": "secure_password",
    "age": 25,
    "address": {
        "city": "北京",
        "district": "朝阳区"
    }
})

# 认证用户
user = user_manager.authenticate_user("john@example.com", "secure_password")
if user:
    print(f"用户认证成功: {user['username']}")

class ContentManager:
    def __init__(self, connection_string: str, database_name: str):
        self.client = MongoClient(connection_string)
        self.db = self.client[database_name]
        self.posts_collection = self.db.posts
        self.comments_collection = self.db.comments
        
        # 创建索引
        self.posts_collection.create_index("slug", unique=True)
        self.posts_collection.create_index("created_at", expireAfterSeconds=3600*24*30)  # 30天过期
        self.comments_collection.create_index("post_id")
        self.posts_collection.create_index([("title", "text"), ("content", "text")])
    
    def create_post(self, post_data: Dict) -> str:
        """创建文章"""
        post_data['created_at'] = datetime.now()
        post_data['updated_at'] = datetime.now()
        post_data['slug'] = self.generate_slug(post_data['title'])
        post_data['view_count'] = 0
        post_data['like_count'] = 0
        post_data['comment_count'] = 0
        
        result = self.posts_collection.insert_one(post_data)
        return str(result.inserted_id)
    
    def generate_slug(self, title: str) -> str:
        """生成URL友好slug"""
        import re
        slug = re.sub(r'[^\w\s-]', '', title.lower())
        slug = re.sub(r'[-\s]+', '-', slug)
        return slug
    
    def get_post_with_comments(self, post_id: str) -> Optional[Dict]:
        """获取文章及评论"""
        post = self.posts_collection.find_one({"_id": ObjectId(post_id)})
        if not post:
            return None
        
        comments = list(self.comments_collection.find({
            "post_id": ObjectId(post_id)
        }).sort("created_at", -1))
        
        post['comments'] = comments
        return post
    
    def add_comment(self, post_id: str, comment_data: Dict) -> str:
        """添加评论"""
        comment_data['post_id'] = ObjectId(post_id)
        comment_data['created_at'] = datetime.now()
        
        result = self.comments_collection.insert_one(comment_data)
        
        # 更新文章评论计数
        self.posts_collection.update_one(
            {"_id": ObjectId(post_id)},
            {"$inc": {"comment_count": 1}}
        )
        
        return str(result.inserted_id)
    
    def search_posts(self, query: str, page: int = 1, limit: int = 10) -> List[Dict]:
        """搜索文章"""
        skip = (page - 1) * limit
        
        posts = list(
            self.posts_collection.find({
                "$text": {"$search": query}
            })
            .skip(skip)
            .limit(limit)
            .sort([("score", {"$meta": "textScore"}), ("created_at", -1)])
        )
        
        return posts
    
    def get_popular_posts(self, days: int = 7, limit: int = 10) -> List[Dict]:
        """获取热门文章"""
        from datetime import timedelta
        
        cutoff_date = datetime.now() - timedelta(days=days)
        
        pipeline = [
            {
                "$match": {
                    "created_at": {"$gte": cutoff_date}
                }
            },
            {
                "$addFields": {
                    "popularity_score": {
                        "$add": [
                            {"$multiply": ["$view_count", 1]},
                            {"$multiply": ["$like_count", 2]},
                            {"$multiply": ["$comment_count", 3]}
                        ]
                    }
                }
            },
            {
                "$sort": {"popularity_score": -1}
            },
            {
                "$limit": limit
            }
        ]
        
        posts = list(self.posts_collection.aggregate(pipeline))
        return posts

# 使用示例
content_manager = ContentManager(
    "mongodb://admin:password@localhost:27017/",
    "blog_database"
)

# 创建文章
post_id = content_manager.create_post({
    "title": "MongoDB实战指南",
    "content": "MongoDB是一个强大的NoSQL数据库...",
    "author": "张三",
    "tags": ["mongodb", "nosql", "database"],
    "category": "技术"
})

# 添加评论
content_manager.add_comment(post_id, {
    "author": "李四",
    "email": "lisi@example.com",
    "content": "写得很详细，感谢分享！",
    "parent_id": None  # 支持嵌套评论
})

# 搜索文章
search_results = content_manager.search_posts("mongodb")
print(f"找到 {len(search_results)} 篇相关文章")

# 连接副本集
client = MongoClient(
    "mongodb://mongo1:27017,mongo2:27017,mongo3:27017",
    replicaSet="rs0",
    readPreference="secondaryPreferred"
)

# 连接分片集群
client = MongoClient(
    "mongodb://mongos1:27017,mongos2:27017",
    readPreference="nearest"
)

# 读取偏好设置
from pymongo.read_preferences import ReadPreference

# 主节点读取（默认）
db.collection.find().read_preference(ReadPreference.PRIMARY)

# 从节点读取（降低主节点负载）
db.collection.find().read_preference(ReadPreference.SECONDARY)

# 最近节点读取
db.collection.find().read_preference(ReadPreference.NEAREST)

"""
MongoDB设计最佳实践：
1. 合理的文档结构：嵌套与引用的权衡
2. 索引策略：避免过度索引，关注查询模式
3. 分片键选择：均匀分布数据
4. 读写关注：根据业务需求调整一致性级别
5. 连接池管理：合理配置连接数
"""

# 嵌套vs引用示例
# 嵌套 - 适合数据变化少、查询频繁的场景
user_with_address = {
    "name": "张三",
    "address": {
        "street": "北京路123号",
        "city": "北京",
        "country": "中国"
    }
}

# 引用 - 适合数据变化频繁、共享数据的场景
user_with_address_ref = {
    "name": "张三",
    "address_id": ObjectId("address_object_id")
}