MediaCrawler/db.py

# 声明：本代码仅供学习和研究目的使用。使用者应遵守以下原则：  
# 1. 不得用于任何商业用途。  
# 2. 使用时应遵守目标平台的使用条款和robots.txt规则。  
# 3. 不得进行大规模爬取或对平台造成运营干扰。  
# 4. 应合理控制请求频率，避免给目标平台带来不必要的负担。   
# 5. 不得用于任何非法或不当的用途。
#   
# 详细许可条款请参阅项目根目录下的LICENSE文件。  
# 使用本代码即表示您同意遵守上述原则和LICENSE中的所有条款。  


# -*- coding: utf-8 -*-
# @Author  : relakkes@gmail.com
# @Time    : 2024/4/6 14:54
# @Desc    : mediacrawler db 管理
import asyncio
from typing import Dict
from urllib.parse import urlparse

import aiofiles
import aiomysql

import config
from async_db import AsyncMysqlDB
from async_sqlite_db import AsyncSqliteDB
from tools import utils
from var import db_conn_pool_var, media_crawler_db_var


async def init_mediacrawler_db():
    """
    初始化数据库链接池对象，并将该对象塞给media_crawler_db_var上下文变量
    Returns:

    """
    pool = await aiomysql.create_pool(
        host=config.MYSQL_DB_HOST,
        port=config.MYSQL_DB_PORT,
        user=config.MYSQL_DB_USER,
        password=config.MYSQL_DB_PWD,
        db=config.MYSQL_DB_NAME,
        autocommit=True,
    )
    async_db_obj = AsyncMysqlDB(pool)

    # 将连接池对象和封装的CRUD sql接口对象放到上下文变量中
    db_conn_pool_var.set(pool)
    media_crawler_db_var.set(async_db_obj)


async def init_sqlite_db():
    """
    初始化SQLite数据库对象，并将该对象塞给media_crawler_db_var上下文变量
    Returns:

    """
    async_db_obj = AsyncSqliteDB(config.SQLITE_DB_PATH)
    
    # 将SQLite数据库对象放到上下文变量中
    media_crawler_db_var.set(async_db_obj)


async def init_db():
    """
    初始化db连接池
    Returns:

    """
    utils.logger.info("[init_db] start init mediacrawler db connect object")
    if config.SAVE_DATA_OPTION == "sqlite":
        await init_sqlite_db()
        utils.logger.info("[init_db] end init sqlite db connect object")
    else:
        await init_mediacrawler_db()
        utils.logger.info("[init_db] end init mysql db connect object")


async def close():
    """
    关闭数据库连接
    Returns:

    """
    utils.logger.info("[close] close mediacrawler db connection")
    if config.SAVE_DATA_OPTION == "sqlite":
        # SQLite数据库连接会在AsyncSqliteDB对象销毁时自动关闭
        utils.logger.info("[close] sqlite db connection will be closed automatically")
    else:
        # MySQL连接池关闭
        db_pool: aiomysql.Pool = db_conn_pool_var.get()
        if db_pool is not None:
            db_pool.close()
            utils.logger.info("[close] mysql db pool closed")


async def init_table_schema(db_type: str = None):
    """
    用来初始化数据库表结构，请在第一次需要创建表结构的时候使用，多次执行该函数会将已有的表以及数据全部删除
    Args:
        db_type: 数据库类型，可选值为 'sqlite' 或 'mysql'，如果不指定则使用配置文件中的设置
    Returns:

    """
    # 如果没有指定数据库类型，则使用配置文件中的设置
    if db_type is None:
        db_type = config.SAVE_DATA_OPTION
    
    if db_type == "sqlite":
        utils.logger.info("[init_table_schema] begin init sqlite table schema ...")
        
        # 检查并删除可能存在的损坏数据库文件
        import os
        if os.path.exists(config.SQLITE_DB_PATH):
            try:
                # 尝试删除现有的数据库文件
                os.remove(config.SQLITE_DB_PATH)
                utils.logger.info(f"[init_table_schema] removed existing sqlite db file: {config.SQLITE_DB_PATH}")
            except Exception as e:
                utils.logger.warning(f"[init_table_schema] failed to remove existing sqlite db file: {e}")
                # 如果删除失败，尝试重命名文件
                try:
                    backup_path = f"{config.SQLITE_DB_PATH}.backup_{utils.get_current_timestamp()}"
                    os.rename(config.SQLITE_DB_PATH, backup_path)
                    utils.logger.info(f"[init_table_schema] renamed existing sqlite db file to: {backup_path}")
                except Exception as rename_e:
                    utils.logger.error(f"[init_table_schema] failed to rename existing sqlite db file: {rename_e}")
                    raise rename_e
        
        await init_sqlite_db()
        async_db_obj: AsyncSqliteDB = media_crawler_db_var.get()
        async with aiofiles.open("schema/sqlite_tables.sql", mode="r", encoding="utf-8") as f:
            schema_sql = await f.read()
            await async_db_obj.executescript(schema_sql)
            utils.logger.info("[init_table_schema] sqlite table schema init successful")
    elif db_type == "mysql":
        utils.logger.info("[init_table_schema] begin init mysql table schema ...")
        await init_mediacrawler_db()
        async_db_obj: AsyncMysqlDB = media_crawler_db_var.get()
        async with aiofiles.open("schema/tables.sql", mode="r", encoding="utf-8") as f:
            schema_sql = await f.read()
            await async_db_obj.execute(schema_sql)
            utils.logger.info("[init_table_schema] mysql table schema init successful")
            await close()
    else:
        utils.logger.error(f"[init_table_schema] 不支持的数据库类型: {db_type}")
        raise ValueError(f"不支持的数据库类型: {db_type}，支持的类型: sqlite, mysql")


def show_database_options():
    """
    显示支持的数据库选项
    """
    print("\n=== MediaCrawler 数据库初始化工具 ===")
    print("支持的数据库类型:")
    print("1. sqlite  - SQLite 数据库 (轻量级，无需额外配置)")
    print("2. mysql   - MySQL 数据库 (需要配置数据库连接信息)")
    print("3. config  - 使用配置文件中的设置")
    print("4. exit    - 退出程序")
    print("="*50)


def get_user_choice():
    """
    获取用户选择的数据库类型
    Returns:
        str: 用户选择的数据库类型
    """
    while True:
        choice = input("请输入数据库类型 (sqlite/mysql/config/exit): ").strip().lower()
        
        if choice in ['sqlite', 'mysql', 'config', 'exit']:
            return choice
        else:
            print("❌ 无效的选择，请输入: sqlite, mysql, config 或 exit")


async def main():
    """
    主函数，处理用户交互和数据库初始化
    """
    try:
        show_database_options()
        
        while True:
            choice = get_user_choice()
            
            if choice == 'exit':
                print("👋 程序已退出")
                break
            elif choice == 'config':
                print(f"📋 使用配置文件中的设置: {config.SAVE_DATA_OPTION}")
                await init_table_schema()
                print("✅ 数据库表结构初始化完成！")
                break
            else:
                print(f"🚀 开始初始化 {choice.upper()} 数据库...")
                await init_table_schema(choice)
                print("✅ 数据库表结构初始化完成！")
                break
                
    except KeyboardInterrupt:
        print("\n\n⚠️  用户中断操作")
    except Exception as e:
        print(f"\n❌ 初始化失败: {str(e)}")
        utils.logger.error(f"[main] 数据库初始化失败: {str(e)}")


if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())