如何写一个脚本自动更新数据库

本文目录导读：

1. Python脚本方案（最通用）
2. Shell脚本方案（Linux）
3. Windows批处理方案
4. Crontab定时任务配置（Linux）
5. 高级功能示例
6. 使用说明

我来帮你编写一个自动更新数据库的脚本,以下提供几种常见方案：

Python脚本方案（最通用）

#!/usr/bin/env python3
"""
数据库自动更新脚本
支持MySQL/PostgreSQL/SQLite
"""
import os
import time
import logging
import schedule
from datetime import datetime
import pymysql  # MySQL
# import psycopg2  # PostgreSQL
# import sqlite3  # SQLite
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('db_update.log'),
        logging.StreamHandler()
    ]
)
class DatabaseUpdater:
    def __init__(self, db_config):
        self.config = db_config
        self.connection = None
    def connect(self):
        """连接数据库"""
        try:
            # MySQL连接示例
            self.connection = pymysql.connect(
                host=self.config['host'],
                user=self.config['user'],
                password=self.config['password'],
                database=self.config['database'],
                charset='utf8mb4'
            )
            logging.info("数据库连接成功")
        except Exception as e:
            logging.error(f"数据库连接失败: {e}")
            raise
    def disconnect(self):
        """断开数据库连接"""
        if self.connection:
            self.connection.close()
            logging.info("数据库连接已关闭")
    def check_and_update(self):
        """检查并更新数据库"""
        try:
            cursor = self.connection.cursor()
            # 1. 检查数据是否需要更新
            # 示例：检查最新记录的时间
            cursor.execute("SELECT MAX(update_time) FROM your_table")
            last_update = cursor.fetchone()[0]
            # 2. 如果数据过期，执行更新
            if self._needs_update(last_update):
                self._perform_update(cursor)
                self.connection.commit()
                logging.info("数据库更新成功")
            else:
                logging.info("数据已是最新，无需更新")
        except Exception as e:
            self.connection.rollback()
            logging.error(f"更新失败: {e}")
        finally:
            if cursor:
                cursor.close()
    def _needs_update(self, last_update):
        """判断是否需要更新"""
        if last_update is None:
            return True
        # 示例：如果最后更新超过1小时，需要更新
        time_diff = datetime.now() - last_update
        return time_diff.total_seconds() > 3600
    def _perform_update(self, cursor):
        """执行实际的更新操作"""
        # 示例1：插入新数据
        insert_query = """
        INSERT INTO your_table (column1, column2, update_time)
        VALUES (%s, %s, NOW())
        """
        cursor.execute(insert_query, ('value1', 'value2'))
        # 示例2：更新已有数据
        update_query = """
        UPDATE your_table 
        SET column1 = %s, update_time = NOW()
        WHERE condition = %s
        """
        cursor.execute(update_query, ('new_value', 'condition'))
        # 示例3：删除过期数据
        delete_query = """
        DELETE FROM your_table 
        WHERE update_time < DATE_SUB(NOW(), INTERVAL 30 DAY)
        """
        cursor.execute(delete_query)
def main():
    # 数据库配置
    db_config = {
        'host': 'localhost',
        'user': 'your_user',
        'password': 'your_password',
        'database': 'your_database'
    }
    updater = DatabaseUpdater(db_config)
    try:
        updater.connect()
        # 立即执行一次
        updater.check_and_update()
        # 定时执行（每小时执行一次）
        schedule.every(1).hours.do(updater.check_and_update)
        # 也可以设置特定时间执行
        # schedule.every().day.at("02:00").do(updater.check_and_update)
        logging.info("定时更新任务已启动")
        # 保持脚本运行
        while True:
            schedule.run_pending()
            time.sleep(60)
    except KeyboardInterrupt:
        logging.info("脚本被用户中断")
    finally:
        updater.disconnect()
if __name__ == "__main__":
    main()

Shell脚本方案（Linux）

#!/bin/bash
# 数据库自动更新脚本
# 配置
DB_HOST="localhost"
DB_USER="username"
DB_PASS="password"
DB_NAME="database_name"
LOG_FILE="/var/log/db_update.log"
# 日志函数
log() {
    echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a "$LOG_FILE"
}
# 备份数据库（更新前）
backup_database() {
    log "开始备份数据库..."
    BACKUP_FILE="/tmp/backup_${DB_NAME}_$(date +%Y%m%d_%H%M%S).sql"
    mysqldump -h "$DB_HOST" -u "$DB_USER" -p"$DB_PASS" "$DB_NAME" > "$BACKUP_FILE"
    if [ $? -eq 0 ]; then
        log "备份成功: $BACKUP_FILE"
        # 压缩备份文件
        gzip "$BACKUP_FILE"
    else
        log "备份失败"
        return 1
    fi
}
# 执行更新
perform_update() {
    log "开始更新数据库..."
    # 方法1：执行SQL文件
    mysql -h "$DB_HOST" -u "$DB_USER" -p"$DB_PASS" "$DB_NAME" < "/path/to/update.sql"
    # 方法2：直接执行SQL语句
    # mysql -h "$DB_HOST" -u "$DB_USER" -p"$DB_PASS" "$DB_NAME" -e "
    #     UPDATE your_table SET column1 = 'new_value' WHERE condition;
    #     INSERT INTO your_table (column1, column2) VALUES ('value1', 'value2');
    #     DELETE FROM your_table WHERE update_time < DATE_SUB(NOW(), INTERVAL 30 DAY);
    # "
    if [ $? -eq 0 ]; then
        log "数据库更新成功"
        return 0
    else
        log "数据库更新失败"
        return 1
    fi
}
# 检查是否需要更新
check_update_needed() {
    # 获取最近更新时间
    LAST_UPDATE=$(mysql -h "$DB_HOST" -u "$DB_USER" -p"$DB_PASS" "$DB_NAME" -N -e "
        SELECT MAX(update_time) FROM your_table
    ")
    if [ -z "$LAST_UPDATE" ]; then
        # 没有数据，需要更新
        return 0
    fi
    # 计算时间差（秒）
    CURRENT_TIME=$(date +%s)
    LAST_UPDATE_EPOCH=$(date -d "$LAST_UPDATE" +%s 2>/dev/null)
    TIME_DIFF=$((CURRENT_TIME - LAST_UPDATE_EPOCH))
    # 如果超过1小时未更新
    if [ $TIME_DIFF -gt 3600 ]; then
        return 0
    fi
    return 1
}
# 主函数
main() {
    log "=== 数据库更新脚本启动 ==="
    if check_update_needed; then
        backup_database
        if [ $? -eq 0 ]; then
            perform_update
        fi
    else
        log "数据已是最新，无需更新"
    fi
    log "=== 数据库更新脚本完成 ==="
}
# 执行主函数
main

Windows批处理方案

@echo off
REM 数据库自动更新脚本 (Windows)
set DB_HOST=localhost
set DB_USER=username
set DB_PASS=password
set DB_NAME=database_name
set LOG_FILE=C:\logs\db_update.log
echo [%date% %time%] 开始更新数据库 >> %LOG_FILE%
REM 备份数据库
echo 备份数据库...
mysqldump -h %DB_HOST% -u %DB_USER% -p%DB_PASS% %DB_NAME% > C:\backup\%DB_NAME%_%date:~0,4%%date:~5,2%%date:~8,2%.sql
if %ERRORLEVEL% equ 0 (
    echo 备份成功 >> %LOG_FILE%
) else (
    echo 备份失败 >> %LOG_FILE%
    exit /b 1
)
REM 执行更新
echo 执行更新...
mysql -h %DB_HOST% -u %DB_USER% -p%DB_PASS% %DB_NAME% < C:\scripts\update.sql
if %ERRORLEVEL% equ 0 (
    echo [%date% %time%] 更新成功 >> %LOG_FILE%
) else (
    echo [%date% %time%] 更新失败 >> %LOG_FILE%
)
echo 完成 >> %LOG_FILE%

Crontab定时任务配置（Linux）

# 编辑crontab
crontab -e
# 每天凌晨2点执行
0 2 * * * /path/to/update_script.sh
# 每小时的15分执行
15 * * * * /path/to/update_script.sh
# 每周一凌晨3点执行
0 3 * * 1 /path/to/update_script.sh
# 每月1号凌晨4点执行
0 4 1 * * /path/to/update_script.sh

高级功能示例

# 增量更新示例
class IncrementalUpdater(DatabaseUpdater):
    def _perform_update(self, cursor):
        """支持增量更新的逻辑"""
        # 1. 获取上次更新的标记
        cursor.execute("SELECT MAX(batch_id) FROM update_log")
        last_batch = cursor.fetchone()[0] or 0
        # 2. 获取需要更新的数据（从外部API）
        new_data = self.fetch_external_data(last_batch)
        # 3. 批量插入
        if new_data:
            insert_query = """
            INSERT INTO your_table (field1, field2, batch_id, update_time)
            VALUES (%s, %s, %s, NOW())
            """
            for item in new_data:
                cursor.execute(insert_query, 
                    (item['field1'], item['field2'], last_batch + 1))
            # 4. 记录更新批次
            cursor.execute("""
                INSERT INTO update_log (batch_id, update_time, records_updated)
                VALUES (%s, NOW(), %s)
            """, (last_batch + 1, len(new_data)))
    def fetch_external_data(self, last_batch):
        """从外部源获取数据"""
        # 实现你的数据获取逻辑
        return []
# 进度通知
def notify_update_status(status):
    """发送更新状态通知"""
    # 邮件通知
    # 钉钉/企业微信通知
    # Slack通知等
    pass

使用说明

1. 安装依赖（Python方案）：

   pip install pymysql schedule

2. 设置权限（Shell方案）：

   chmod +x update_script.sh

3. 测试运行：

   python db_updater.py  # Python
   ./update_script.sh     # Shell

4. 配置定时任务：

   • Linux: 使用crontab
   • Windows: 使用任务计划程序

根据你的具体需求,选择合适的方案并修改相应的数据库配置和更新逻辑。