Python结合SQLite构建一个完整数据驱动应用的终极指南

"""
高级博客系统 - 展示sqlite在python中的完整应用
"""
import sqlite3
import json
import hashlib
import uuid
from datetime import datetime, timedelta
from contextlib import contextmanager
from dataclasses import dataclass
from typing import list, optional, dict, any
import logging

# 配置日志系统
logging.basicconfig(
    level=logging.info,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getlogger(__name__)

@dataclass
class user:
    """用户数据类"""
    id: optional[int] = none
    username: str = ""
    email: str = ""
    password_hash: str = ""
    bio: optional[str] = none
    created_at: optional[str] = none
    last_login: optional[str] = none

@dataclass
class blogpost:
    """博客文章数据类"""
    id: optional[int] = none
    title: str = ""
    content: str = ""
    author_id: int = 0
    status: str = "draft"  # draft, published, archived
    tags: optional[list[str]] = none
    metadata: optional[dict[str, any]] = none
    created_at: optional[str] = none
    updated_at: optional[str] = none
    published_at: optional[str] = none

class databaseconnection:
    """高级数据库连接管理器"""
    
    def __init__(self, db_path: str = ":memory:", enable_wal: bool = true):
        """
        初始化数据库连接
        
        args:
            db_path: 数据库路径，默认为内存数据库
            enable_wal: 是否启用write-ahead logging（提高并发性能）
        """
        self.db_path = db_path
        self.enable_wal = enable_wal
        
        # 注册自定义类型适配器
        self._register_adapters()
    
    def _register_adapters(self):
        """注册自定义类型适配器"""
        # 列表适配器（存储为json）
        def adapt_list(lst):
            return json.dumps(lst)
        
        def convert_list(text):
            return json.loads(text) if text else []
        
        # 字典适配器
        def adapt_dict(dct):
            return json.dumps(dct)
        
        def convert_dict(text):
            return json.loads(text) if text else {}
        
        sqlite3.register_adapter(list, adapt_list)
        sqlite3.register_adapter(dict, adapt_dict)
        sqlite3.register_converter("list", convert_list)
        sqlite3.register_converter("dict", convert_dict)
    
    @contextmanager
    def get_connection(self):
        """
        获取数据库连接的上下文管理器
        
        使用示例：
            with db.get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute("select * from users")
        """
        conn = none
        try:
            # 创建连接，启用类型检测和自定义转换器
            conn = sqlite3.connect(
                self.db_path,
                detect_types=sqlite3.parse_decltypes | sqlite3.parse_colnames,
                isolation_level=none  # 使用自动提交模式
            )
            
            # 设置行工厂为字典类型
            conn.row_factory = sqlite3.row
            
            # 启用外键约束
            conn.execute("pragma foreign_keys = on")
            
            # 启用wal模式（提高并发性能）
            if self.enable_wal:
                conn.execute("pragma journal_mode = wal")
                conn.execute("pragma synchronous = normal")
            
            # 设置繁忙超时
            conn.execute("pragma busy_timeout = 5000")
            
            logger.info(f"数据库连接已建立: {self.db_path}")
            
            yield conn
            
        except sqlite3.error as e:
            logger.error(f"数据库错误: {e}")
            if conn:
                conn.rollback()
            raise
        
        finally:
            if conn:
                # 如果使用wal模式，执行检查点
                if self.enable_wal:
                    conn.execute("pragma wal_checkpoint(passive)")
                conn.close()
                logger.info("数据库连接已关闭")

class blogdatabase:
    """博客系统数据库管理器"""
    
    def __init__(self, db_path: str = "blog_system.db"):
        self.db_conn = databaseconnection(db_path)
        self._init_database()
    
    def _init_database(self):
        """初始化数据库表结构"""
        with self.db_conn.get_connection() as conn:
            cursor = conn.cursor()
            
            # 启用外键和性能优化
            cursor.execute("pragma foreign_keys = on")
            cursor.execute("pragma journal_mode = wal")
            
            # 创建用户表
            cursor.execute("""
                create table if not exists users (
                    id integer primary key autoincrement,
                    username text unique not null,
                    email text unique not null,
                    password_hash text not null,
                    bio text,
                    avatar_url text,
                    is_admin boolean default 0,
                    is_active boolean default 1,
                    created_at timestamp default current_timestamp,
                    last_login timestamp,
                    metadata dict default '{}',
                    
                    -- 添加索引以提高查询性能
                    constraint chk_username_length check (length(username) >= 3),
                    constraint chk_email_format check (email like '%@%.%')
                )
            """)
            
            # 创建用户表的索引
            cursor.execute("create index if not exists idx_users_username on users(username)")
            cursor.execute("create index if not exists idx_users_email on users(email)")
            cursor.execute("create index if not exists idx_users_created on users(created_at)")
            
            # 创建博客文章表
            cursor.execute("""
                create table if not exists blog_posts (
                    id integer primary key autoincrement,
                    title text not null,
                    slug text unique not null,
                    content text not null,
                    excerpt text,
                    author_id integer not null,
                    status text default 'draft' check (status in ('draft', 'published', 'archived')),
                    tags list default '[]',
                    category text,
                    view_count integer default 0,
                    like_count integer default 0,
                    metadata dict default '{}',
                    created_at timestamp default current_timestamp,
                    updated_at timestamp default current_timestamp,
                    published_at timestamp,
                    
                    -- 全文搜索列（虚拟列）
                    content_search text generated always as (lower(title || ' ' || content)),
                    
                    -- 外键约束
                    foreign key (author_id) references users(id) on delete cascade,
                    
                    -- 约束条件
                    constraint chk_title_length check (length(title) >= 5),
                    constraint chk_slug_format check (slug glob '[a-z0-9-]*')
                )
            """)
            
            # 创建博客文章表的索引
            cursor.execute("create index if not exists idx_posts_author on blog_posts(author_id)")
            cursor.execute("create index if not exists idx_posts_status on blog_posts(status)")
            cursor.execute("create index if not exists idx_posts_created on blog_posts(created_at)")
            cursor.execute("create index if not exists idx_posts_published on blog_posts(published_at)")
            cursor.execute("create index if not exists idx_posts_category on blog_posts(category)")
            
            # 创建全文搜索虚拟表的触发器
            cursor.execute("""
                create virtual table if not exists posts_fts using fts5(
                    title, content, content_search,
                    tokenize="porter unicode61"
                )
            """)
            
            # 创建触发器以同步fts表
            cursor.execute("""
                create trigger if not exists posts_ai after insert on blog_posts
                begin
                    insert into posts_fts(rowid, title, content, content_search)
                    values (new.id, new.title, new.content, new.content_search);
                end
            """)
            
            cursor.execute("""
                create trigger if not exists posts_ad after delete on blog_posts
                begin
                    delete from posts_fts where rowid = old.id;
                end
            """)
            
            cursor.execute("""
                create trigger if not exists posts_au after update on blog_posts
                begin
                    delete from posts_fts where rowid = old.id;
                    insert into posts_fts(rowid, title, content, content_search)
                    values (new.id, new.title, new.content, new.content_search);
                end
            """)
            
            # 创建评论表
            cursor.execute("""
                create table if not exists comments (
                    id integer primary key autoincrement,
                    post_id integer not null,
                    user_id integer not null,
                    parent_id integer,  -- 支持嵌套评论
                    content text not null,
                    is_approved boolean default 0,
                    created_at timestamp default current_timestamp,
                    updated_at timestamp default current_timestamp,
                    
                    foreign key (post_id) references blog_posts(id) on delete cascade,
                    foreign key (user_id) references users(id) on delete cascade,
                    foreign key (parent_id) references comments(id) on delete cascade
                )
            """)
            
            # 创建评论表的索引
            cursor.execute("create index if not exists idx_comments_post on comments(post_id)")
            cursor.execute("create index if not exists idx_comments_user on comments(user_id)")
            cursor.execute("create index if not exists idx_comments_parent on comments(parent_id)")
            
            conn.commit()
            logger.info("数据库表结构初始化完成")
    
    @staticmethod
    def _hash_password(password: str) -> str:
        """生成密码哈希值"""
        salt = uuid.uuid4().hex
        return hashlib.sha256(salt.encode() + password.encode()).hexdigest() + ':' + salt
    
    @staticmethod
    def _check_password(hashed_password: str, user_password: str) -> bool:
        """验证密码"""
        password_hash, salt = hashed_password.split(':')
        return password_hash == hashlib.sha256(salt.encode() + user_password.encode()).hexdigest()
    
    def create_user(self, username: str, email: str, password: str, bio: str = none) -> optional[int]:
        """
        创建新用户
        
        returns:
            新用户的id，如果失败则返回none
        """
        try:
            with self.db_conn.get_connection() as conn:
                cursor = conn.cursor()
                
                # 检查用户名和邮箱是否已存在
                cursor.execute(
                    "select count(*) from users where username = ? or email = ?",
                    (username, email)
                )
                if cursor.fetchone()[0] > 0:
                    logger.warning(f"用户名或邮箱已存在: {username}, {email}")
                    return none
                
                # 创建用户
                password_hash = self._hash_password(password)
                cursor.execute("""
                    insert into users (username, email, password_hash, bio, last_login)
                    values (?, ?, ?, ?, current_timestamp)
                """, (username, email, password_hash, bio))
                
                user_id = cursor.lastrowid
                conn.commit()
                
                logger.info(f"用户创建成功: {username} (id: {user_id})")
                return user_id
                
        except sqlite3.error as e:
            logger.error(f"创建用户失败: {e}")
            return none
    
    def authenticate_user(self, username: str, password: str) -> optional[user]:
        """用户认证"""
        try:
            with self.db_conn.get_connection() as conn:
                cursor = conn.cursor()
                
                cursor.execute("""
                    select id, username, email, password_hash, bio, created_at, last_login
                    from users 
                    where username = ? and is_active = 1
                """, (username,))
                
                row = cursor.fetchone()
                if not row:
                    return none
                
                # 验证密码
                if self._check_password(row['password_hash'], password):
                    # 更新最后登录时间
                    cursor.execute(
                        "update users set last_login = current_timestamp where id = ?",
                        (row['id'],)
                    )
                    conn.commit()
                    
                    return user(
                        id=row['id'],
                        username=row['username'],
                        email=row['email'],
                        password_hash=row['password_hash'],
                        bio=row['bio'],
                        created_at=row['created_at'],
                        last_login=row['last_login']
                    )
                return none
                
        except sqlite3.error as e:
            logger.error(f"用户认证失败: {e}")
            return none
    
    def create_blog_post(self, title: str, content: str, author_id: int, 
                        tags: list[str] = none, category: str = none) -> optional[int]:
        """
        创建博客文章
        
        returns:
            新文章的id，如果失败则返回none
        """
        try:
            with self.db_conn.get_connection() as conn:
                cursor = conn.cursor()
                
                # 生成slug
                import re
                slug = re.sub(r'[^a-z0-9]+', '-', title.lower()).strip('-')
                
                # 确保slug唯一
                counter = 1
                original_slug = slug
                while true:
                    cursor.execute("select count(*) from blog_posts where slug = ?", (slug,))
                    if cursor.fetchone()[0] == 0:
                        break
                    slug = f"{original_slug}-{counter}"
                    counter += 1
                
                # 创建文章摘要
                excerpt = content[:150] + "..." if len(content) > 150 else content
                
                cursor.execute("""
                    insert into blog_posts 
                    (title, slug, content, excerpt, author_id, tags, category, created_at)
                    values (?, ?, ?, ?, ?, ?, ?, current_timestamp)
                """, (title, slug, content, excerpt, author_id, 
                      tags or [], category))
                
                post_id = cursor.lastrowid
                conn.commit()
                
                logger.info(f"博客文章创建成功: {title} (id: {post_id})")
                return post_id
                
        except sqlite3.error as e:
            logger.error(f"创建博客文章失败: {e}")
            return none
    
    def publish_blog_post(self, post_id: int) -> bool:
        """发布博客文章"""
        try:
            with self.db_conn.get_connection() as conn:
                cursor = conn.cursor()
                
                cursor.execute("""
                    update blog_posts 
                    set status = 'published', 
                        published_at = current_timestamp,
                        updated_at = current_timestamp
                    where id = ? and status != 'published'
                """, (post_id,))
                
                success = cursor.rowcount > 0
                conn.commit()
                
                if success:
                    logger.info(f"博客文章已发布: {post_id}")
                else:
                    logger.warning(f"博客文章发布失败或已发布: {post_id}")
                
                return success
                
        except sqlite3.error as e:
            logger.error(f"发布博客文章失败: {e}")
            return false
    
    def search_posts(self, query: str, limit: int = 10, offset: int = 0) -> list[dict]:
        """
        全文搜索博客文章
        """
        try:
            with self.db_conn.get_connection() as conn:
                cursor = conn.cursor()
                
                # 使用fts5进行全文搜索
                cursor.execute("""
                    select 
                        p.id,
                        p.title,
                        p.slug,
                        p.excerpt,
                        p.author_id,
                        u.username as author_name,
                        p.category,
                        p.tags,
                        p.view_count,
                        p.like_count,
                        p.created_at,
                        p.published_at,
                        snippet(posts_fts, 0, '<b>', '</b>', '...', 64) as snippet,
                        rank
                    from posts_fts f
                    join blog_posts p on f.rowid = p.id
                    join users u on p.author_id = u.id
                    where posts_fts match ?
                        and p.status = 'published'
                    order by rank
                    limit ? offset ?
                """, (f"{query}*", limit, offset))
                
                results = []
                for row in cursor.fetchall():
                    results.append(dict(row))
                
                return results
                
        except sqlite3.error as e:
            logger.error(f"搜索文章失败: {e}")
            return []
    
    def get_popular_posts(self, days: int = 30, limit: int = 10) -> list[dict]:
        """
        获取热门文章（基于浏览量和点赞数）
        """
        try:
            with self.db_conn.get_connection() as conn:
                cursor = conn.cursor()
                
                cursor.execute("""
                    select 
                        p.id,
                        p.title,
                        p.slug,
                        p.excerpt,
                        p.author_id,
                        u.username as author_name,
                        p.category,
                        p.tags,
                        p.view_count,
                        p.like_count,
                        p.created_at,
                        p.published_at,
                        -- 热度分数 = 浏览量 + 点赞数*5
                        (p.view_count + p.like_count * 5) as hot_score
                    from blog_posts p
                    join users u on p.author_id = u.id
                    where p.status = 'published'
                        and p.published_at >= datetime('now', ?)
                    order by hot_score desc
                    limit ?
                """, (f"-{days} days", limit))
                
                results = []
                for row in cursor.fetchall():
                    results.append(dict(row))
                
                return results
                
        except sqlite3.error as e:
            logger.error(f"获取热门文章失败: {e}")
            return []
    
    def increment_view_count(self, post_id: int) -> bool:
        """增加文章浏览量"""
        try:
            with self.db_conn.get_connection() as conn:
                cursor = conn.cursor()
                cursor.execute(
                    "update blog_posts set view_count = view_count + 1 where id = ?",
                    (post_id,)
                )
                conn.commit()
                return cursor.rowcount > 0
                
        except sqlite3.error as e:
            logger.error(f"增加浏览量失败: {e}")
            return false
    
    def add_comment(self, post_id: int, user_id: int, content: str, 
                   parent_id: int = none) -> optional[int]:
        """
        添加评论
        
        returns:
            新评论的id，如果失败则返回none
        """
        try:
            with self.db_conn.get_connection() as conn:
                cursor = conn.cursor()
                
                cursor.execute("""
                    insert into comments (post_id, user_id, parent_id, content)
                    values (?, ?, ?, ?)
                """, (post_id, user_id, parent_id, content))
                
                comment_id = cursor.lastrowid
                conn.commit()
                
                logger.info(f"评论添加成功: 文章 {post_id}, 用户 {user_id}")
                return comment_id
                
        except sqlite3.error as e:
            logger.error(f"添加评论失败: {e}")
            return none
    
    def get_post_with_comments(self, post_id: int) -> optional[dict]:
        """获取文章及其评论（嵌套结构）"""
        try:
            with self.db_conn.get_connection() as conn:
                cursor = conn.cursor()
                
                # 获取文章信息
                cursor.execute("""
                    select 
                        p.*,
                        u.username as author_name,
                        u.avatar_url as author_avatar
                    from blog_posts p
                    join users u on p.author_id = u.id
                    where p.id = ?
                """, (post_id,))
                
                post_row = cursor.fetchone()
                if not post_row:
                    return none
                
                # 获取评论（使用递归cte获取嵌套结构）
                cursor.execute("""
                    with recursive comment_tree as (
                        -- 顶级评论
                        select 
                            c.id,
                            c.post_id,
                            c.user_id,
                            c.parent_id,
                            c.content,
                            c.is_approved,
                            c.created_at,
                            u.username,
                            u.avatar_url,
                            1 as level,
                            printf('%010d', c.id) as path
                        from comments c
                        join users u on c.user_id = u.id
                        where c.post_id = ? and c.parent_id is null
                        
                        union all
                        
                        -- 递归获取子评论
                        select 
                            c.id,
                            c.post_id,
                            c.user_id,
                            c.parent_id,
                            c.content,
                            c.is_approved,
                            c.created_at,
                            u.username,
                            u.avatar_url,
                            ct.level + 1,
                            ct.path || printf('-%010d', c.id)
                        from comments c
                        join users u on c.user_id = u.id
                        join comment_tree ct on c.parent_id = ct.id
                        where c.post_id = ?
                    )
                    select * from comment_tree
                    order by path
                """, (post_id, post_id))
                
                comments = []
                for row in cursor.fetchall():
                    comments.append(dict(row))
                
                # 构建结果
                result = dict(post_row)
                result['comments'] = comments
                
                return result
                
        except sqlite3.error as e:
            logger.error(f"获取文章详情失败: {e}")
            return none
    
    def backup_database(self, backup_path: str) -> bool:
        """备份数据库"""
        try:
            with self.db_conn.get_connection() as conn:
                backup_conn = sqlite3.connect(backup_path)
                with backup_conn:
                    conn.backup(backup_conn)
                backup_conn.close()
                
                logger.info(f"数据库备份成功: {backup_path}")
                return true
                
        except sqlite3.error as e:
            logger.error(f"数据库备份失败: {e}")
            return false
    
    def get_statistics(self) -> dict[str, any]:
        """获取系统统计信息"""
        try:
            with self.db_conn.get_connection() as conn:
                cursor = conn.cursor()
                
                stats = {}
                
                # 用户统计
                cursor.execute("""
                    select 
                        count(*) as total_users,
                        count(case when is_admin then 1 end) as admin_users,
                        count(case when last_login >= datetime('now', '-7 days') then 1 end) as active_users_7d,
                        count(case when created_at >= datetime('now', '-30 days') then 1 end) as new_users_30d
                    from users
                    where is_active = 1
                """)
                stats.update(dict(cursor.fetchone()))
                
                # 文章统计
                cursor.execute("""
                    select 
                        count(*) as total_posts,
                        count(case when status = 'published' then 1 end) as published_posts,
                        count(case when status = 'draft' then 1 end) as draft_posts,
                        sum(view_count) as total_views,
                        sum(like_count) as total_likes,
                        count(case when published_at >= datetime('now', '-7 days') then 1 end) as new_posts_7d
                    from blog_posts
                """)
                stats.update(dict(cursor.fetchone()))
                
                # 评论统计
                cursor.execute("""
                    select 
                        count(*) as total_comments,
                        count(case when is_approved then 1 end) as approved_comments,
                        count(case when created_at >= datetime('now', '-7 days') then 1 end) as new_comments_7d
                    from comments
                """)
                stats.update(dict(cursor.fetchone()))
                
                # 热门分类
                cursor.execute("""
                    select 
                        category,
                        count(*) as post_count,
                        sum(view_count) as total_views
                    from blog_posts
                    where category is not null and status = 'published'
                    group by category
                    order by total_views desc
                    limit 5
                """)
                stats['top_categories'] = [dict(row) for row in cursor.fetchall()]
                
                return stats
                
        except sqlite3.error as e:
            logger.error(f"获取统计信息失败: {e}")
            return {}

def main():
    """主函数 - 演示完整的博客系统"""
    print("=" * 60)
    print("sqlite博客系统演示")
    print("=" * 60)
    
    # 创建数据库实例
    blog_db = blogdatabase("blog_demo.db")
    
    # 创建测试用户
    print("\n1. 创建测试用户...")
    user_id = blog_db.create_user(
        username="techwriter",
        email="writer@example.com",
        password="securepassword123",
        bio="技术作家，热爱分享知识"
    )
    
    if not user_id:
        print("用户创建失败或用户已存在，尝试认证...")
        user = blog_db.authenticate_user("techwriter", "securepassword123")
        if user:
            user_id = user.id
            print(f"用户认证成功: {user.username}")
        else:
            print("无法创建或认证用户，退出演示")
            return
    
    # 创建博客文章
    print("\n2. 创建博客文章...")
    post_content = """
    # sqlite在现代web开发中的革命性作用
    
    在当今快速迭代的开发环境中，sqlite已经从简单的嵌入式数据库演变为一个能够支撑中等规模生产应用的强大工具。
    
    ## 为什么sqlite被低估了？
    
    大多数开发者认为sqlite只适合小型项目或原型开发，但这种看法已经过时了。sqlite 3.0引入的wal模式、全文搜索和json支持，使其具备了处理复杂应用的能力。
    
    ## 关键技术特性
    
    1. **无服务器架构**：不需要独立的数据库服务器
    2. **零配置**：开箱即用
    3. **acid合规**：完整的事务支持
    4. **全文搜索**：内置fts5扩展
    5. **json支持**：直接处理json数据
    6. **并发控制**：wal模式支持多读单写
    
    ## 生产环境最佳实践
    
    - 使用wal模式提高并发性能
    - 合理设计索引优化查询
    - 定期备份和优化数据库
    - 使用连接池管理数据库连接
    
    sqlite正在改变我们对轻量级数据库的认知，它不再是"玩具数据库"，而是现代应用架构中的重要组件。
    """
    
    post_id = blog_db.create_blog_post(
        title="sqlite在现代web开发中的革命性作用",
        content=post_content,
        author_id=user_id,
        tags=["sqlite", "database", "web-development", "python"],
        category="technology"
    )
    
    if post_id:
        print(f"文章创建成功，id: {post_id}")
        
        # 发布文章
        if blog_db.publish_blog_post(post_id):
            print("文章已成功发布")
        
        # 增加浏览量
        for _ in range(5):
            blog_db.increment_view_count(post_id)
        print("模拟了5次文章浏览")
    
    # 创建第二篇文章
    print("\n3. 创建第二篇文章...")
    post2_id = blog_db.create_blog_post(
        title="python数据持久化策略比较",
        content="本文比较了python中各种数据持久化方案...",
        author_id=user_id,
        tags=["python", "database", "persistence"],
        category="programming"
    )
    
    if post2_id:
        blog_db.publish_blog_post(post2_id)
        print("第二篇文章创建并发布成功")
    
    # 添加评论
    print("\n4. 添加评论...")
    blog_db.add_comment(
        post_id=post_id,
        user_id=user_id,
        content="非常好的文章！sqlite确实被很多人低估了。"
    )
    
    blog_db.add_comment(
        post_id=post_id,
        user_id=user_id,
        content="请问wal模式和普通模式的具体性能差异有多大？",
        parent_id=1  # 回复第一条评论
    )
    
    # 搜索文章
    print("\n5. 搜索文章...")
    search_results = blog_db.search_posts("sqlite 开发")
    print(f"搜索到 {len(search_results)} 篇文章:")
    for result in search_results:
        print(f"  - {result['title']} (相关性分数: {result['rank']:.2f})")
        print(f"    摘要: {result['snippet'][:100]}...")
    
    # 获取热门文章
    print("\n6. 热门文章排名...")
    popular_posts = blog_db.get_popular_posts(days=30, limit=3)
    print("本月热门文章:")
    for i, post in enumerate(popular_posts, 1):
        print(f"  {i}. {post['title']}")
        print(f"     浏览量: {post['view_count']}, 点赞: {post['like_count']}, 热度: {post['hot_score']}")
    
    # 获取文章详情
    print("\n7. 获取文章详情...")
    post_details = blog_db.get_post_with_comments(post_id)
    if post_details:
        print(f"文章标题: {post_details['title']}")
        print(f"作者: {post_details['author_name']}")
        print(f"标签: {', '.join(post_details['tags'])}")
        print(f"浏览量: {post_details['view_count']}")
        print(f"评论数: {len(post_details['comments'])}")
        
        # 显示评论
        for comment in post_details['comments']:
            indent = "  " * (comment['level'] - 1)
            print(f"{indent}├─ {comment['username']}: {comment['content'][:50]}...")
    
    # 获取系统统计
    print("\n8. 系统统计信息...")
    stats = blog_db.get_statistics()
    print(f"总用户数: {stats.get('total_users', 0)}")
    print(f"总文章数: {stats.get('total_posts', 0)}")
    print(f"已发布文章: {stats.get('published_posts', 0)}")
    print(f"总浏览量: {stats.get('total_views', 0)}")
    print(f"总评论数: {stats.get('total_comments', 0)}")
    
    # 备份数据库
    print("\n9. 备份数据库...")
    if blog_db.backup_database("blog_demo_backup.db"):
        print("数据库备份成功")
    
    print("\n" + "=" * 60)
    print("演示完成！")
    print("数据库文件: blog_demo.db")
    print("备份文件: blog_demo_backup.db")
    print("=" * 60)

if __name__ == "__main__":
    main()

# 启用wal模式
conn.execute("pragma journal_mode = wal")

# 创建全文搜索虚拟表
create virtual table posts_fts using fts5(...)

# 自定义类型适配器
sqlite3.register_adapter(dict, adapt_dict)
sqlite3.register_converter("dict", convert_dict)

# 使用递归cte获取嵌套评论
with recursive comment_tree as (...)