Python实现MySQL分批次数据导入 支持指定导入范围适配API日调用限制

核心逻辑 #

• 采用固定位置断点存储机制，不依赖本地临时文件、不依赖数据库现有数据量统计，直接以本地JSON数据源的数组下标为进度标记，从根源上避免重复处理、漏处理问题
• 单次运行严格截取1000条数据范围，完全匹配外部API每日1000次的调用限额，不会超额触发限流
• 整批数据处理完成并全部入库后，统一更新进度标记，保证程序中断后重跑不会出现数据缺失

前置准备：创建进度存储结构 #

首先在你的业务数据库中执行以下SQL，创建专门存储导入进度的表，避免本地进度文件丢失、损坏的问题：

CREATE TABLE IF NOT EXISTS import_progress (
    id INT PRIMARY KEY DEFAULT 1,
    last_processed_index INT NOT NULL DEFAULT 0,
    total_records INT NOT NULL DEFAULT 0,
    update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    CONSTRAINT check_single_row CHECK (id = 1)
);
-- 初始化单行进度记录
INSERT INTO import_progress (id, last_processed_index, total_records) 
VALUES (1, 0, 0) 
ON DUPLICATE KEY UPDATE id = 1;
-- 给业务表的电影名称字段加唯一索引，避免重复导入
ALTER TABLE serije_filmovi ADD UNIQUE INDEX idx_naslov (naslov);

调整后的可运行代码 #

替换原有逻辑为以下代码，自动实现范围控制与断点续跑：

import json
import mysql.connector
# 保留你原有导入的GetMovie依赖
from moviedb import GetMovie

# 配置参数：匹配API每日调用限额
DAILY_API_LIMIT = 1000

mysql = mysql.connector.connect(
    host=host_input,
    user=user_input,
    passwd=passwd_input,
    database=database_input
)

# 1. 读取上次运行的断点位置
cursor = mysql.cursor()
cursor.execute("SELECT last_processed_index, total_records FROM import_progress WHERE id = 1")
progress = cursor.fetchone()
start_index = progress[0]
old_total = progress[1]

# 2. 加载JSON数据源，计算本次处理范围
with open('sadrzaj.json') as json_file:
    data = json.load(json_file)
    film_list = data['filmovi']
    total_films = len(film_list)

# 同步更新总记录数（如果JSON文件有新增内容）
if old_total != total_films:
    cursor.execute(
        "UPDATE import_progress SET total_records = %s WHERE id = 1",
        (total_films,)
    )
    mysql.commit()

# 全部处理完成则直接退出
if start_index >= total_films:
    print("所有数据已全部导入完成，无需运行")
    exit()

# 计算本次处理的结束位置：不超过API限额、不超过数组总长度
end_index = min(start_index + DAILY_API_LIMIT, total_films)
print(f"本次将处理从第{start_index+1}条到第{end_index}条数据，共{end_index - start_index}条")

# 3. 遍历处理本次范围内的所有数据
insert_params = []  # 批量收集插入参数，减少数据库交互次数
for real_index in range(start_index, end_index):
    p = film_list[real_index]
    naslov = p['naslov']
    iframe = p['iframe']
    opis = p['opis']

    # API调用加异常兜底，单条失败不中断整体流程
    try:
        movie = GetMovie(title=naslov, api_key='5f8abea5', plot='full')
        info = movie.get_data(
            'Title', 'imdbRating', 'Genre', 'Year', 'Runtime', 
            'Country', 'Plot', 'Poster', 'Type', 'Language'
        )
        imdbRating = info["imdbRating"]
        genre = info["Genre"]
        year = info["Year"]
        runtime = info["Runtime"]
        country = info["Country"]
        poster = info["Poster"]
        typee = info['Type']
        language = info["Language"]

        insert_params.append(
            (naslov, iframe, opis, imdbRating, genre, year, 
             runtime, country, poster, typee, language)
        )
        print(f"已获取{naslov}的元数据")
    except Exception as e:
        print(f"获取{naslov}元数据失败，跳过，错误信息：{str(e)}")
        continue

# 4. 批量入库 + 更新断点
if insert_params:
    sql = """INSERT IGNORE INTO serije_filmovi(
        naslov, iframe, opis, imdbRating, genre, years, 
        runtime, country, poster, typee, language
    ) VALUES (%s,%s, %s, %s, %s, %s,%s, %s, %s, %s, %s)"""
    cursor.executemany(sql, insert_params)
    mysql.commit()
    print(f"本次共成功插入{cursor.rowcount}条数据")

# 所有数据入库完成后，再更新断点位置
cursor.execute(
    "UPDATE import_progress SET last_processed_index = %s WHERE id = 1",
    (end_index,)
)
mysql.commit()
print(f"本次运行结束，下次运行将从第{end_index + 1}条数据开始处理")

cursor.close()
mysql.close()

关键设计说明 #

• 导入时使用INSERT IGNORE语法，配合naslov字段的唯一索引，即使重跑断点位置，也不会因为唯一键冲突报错，重复数据会自动跳过
• 断点更新操作全部在整批数据处理完成后执行，避免处理到一半程序崩溃导致的漏数问题，最多仅可能出现最后一小批的重复处理，不会有数据缺失
• 用executemany批量插入替代原有的单条循环插入、单条commit逻辑，大幅降低数据库交互次数，提升执行效率
• 严格卡准每日1000次的API调用限额，不会超额触发限流，每次运行处理完就自动退出，后续运行直接接续上一次的位置即可

内容的提问来源于stack exchange，提问作者aleksa312