春联生成模型-中文-base批量生成实战：Python脚本自动化处理千条请求

春联生成模型-中文-base批量生成实战Python脚本自动化处理千条请求春节临近无论是公司活动、社区宣传还是个人想给亲朋好友送上一份独特的祝福批量生成个性化春联的需求就来了。一张张手动输入、等待生成效率太低。今天我们就来聊聊如何用Python写个脚本让电脑帮你自动处理成千上万条春联生成请求把我们从重复劳动中解放出来。这个教程的目标很明确假设你手头有一个Excel或CSV文件里面列着几百甚至上千个不同的主题关键词比如“龙年大吉”、“阖家欢乐”、“生意兴隆”我们要写个程序自动读取这些词调用春联生成模型的API然后把生成的上联、下联和横批整齐地保存下来。整个过程我们会用到文件操作、网络请求以及提升速度的多线程或异步处理最后还会加上错误重试和进度条确保任务能稳定、高效地跑完。即使你Python刚入门跟着步骤走也能轻松搞定。我们主要会用到requests库发请求pandas或csv库读文件concurrent.futures或asyncio来加速。话不多说我们开始吧。1. 环境准备与核心思路在动手写代码之前我们需要把“战场”布置好。这里没有复杂的深度学习环境依赖核心就是几个处理数据和网络请求的Python库。1.1 安装必要的Python库打开你的终端命令行用pip安装下面这几个库。它们都是Python生态里非常常用和稳定的工具。pip install requests pandas openpyxl tqdm简单解释一下每个库的用途requests 用来发送HTTP请求与春联生成模型的API进行通信。这是网络交互的核心。pandas 一个强大的数据分析库。我们主要用它来轻松读取和写入Excel/CSV文件处理表格数据非常方便。openpyxl 因为pandas在读写新版Excel文件.xlsx格式时需要它作为引擎所以一并安装。tqdm 这是一个超级好用的进度条库。当你的脚本在处理成百上千条数据时有个进度条能让你清晰知道任务进展心里不慌。1.2 准备你的数据源你需要准备一个包含所有春联主题关键词的文件。这里有两种常见格式你可以任选其一CSV文件 比如keywords.csv用记事本或Excel创建。内容大致如下第一行通常是列名keyword 龙年大吉 阖家欢乐 学业有成 财源广进 ...更多关键词Excel文件 比如keywords.xlsx在Excel里创建一个工作表在A列或其他列从上到下填入关键词。我们的脚本会读取这个文件把每一行的关键词提取出来作为生成春联的“种子”。1.3 了解API接口在编写请求之前你必须先知道春联生成模型的API端点URL是什么以及它需要什么样的输入会返回什么样的输出。假设请注意这里需要你替换为实际的API信息API地址https://api.example.com/v1/generate/couplet请替换为真实可用的服务地址请求方法 通常是POST请求头 一般需要指定内容类型例如{Content-Type: application/json}请求体 一个JSON对象包含生成参数。最核心的参数就是prompt提示词也就是我们的关键词。可能还有其他参数控制生成风格、长度等需要查阅具体的API文档。{ prompt: 龙年大吉, max_length: 50, // ... 其他可选参数 }响应体 通常也是一个JSON对象里面包含了生成的上联、下联和横批。{ status: success, data: { upper_line: 龙腾虎跃迎新岁, lower_line: 鸟语花香报好春, horizontal: 春满人间 } }重要提示 在运行脚本前请务必获取并确认正确的API地址、所需的认证信息如API Key以及完整的请求/响应格式。你可以先用curl命令或Postman等工具手动测试一两条确保接口通顺。2. 编写基础批量生成脚本有了清晰的思路和准备好的环境我们现在来搭建脚本的骨架。我们从最简单的单线程版本开始确保核心流程跑通。2.1 读取关键词文件首先我们写一个函数来读取你准备好的关键词文件。这里以pandas为例它能自动处理CSV和Excel格式。import pandas as pd def load_keywords(file_path): 从CSV或Excel文件中加载关键词列表。 参数: file_path (str): 文件路径支持 .csv 或 .xlsx 格式。 返回: list: 包含所有关键词的列表。 # 根据文件后缀名判断格式并读取 if file_path.endswith(.csv): df pd.read_csv(file_path) elif file_path.endswith((.xls, .xlsx)): # 默认读取第一个工作表如果需要可以指定sheet_name df pd.read_excel(file_path, engineopenpyxl) else: raise ValueError(不支持的文件格式请使用 .csv 或 .xlsx 文件。) # 假设关键词所在的列名为keyword请根据你的文件实际情况修改 keyword_column keyword if keyword_column not in df.columns: # 如果文件没有列名或者列名不同尝试取第一列 keywords df.iloc[:, 0].tolist() else: keywords df[keyword_column].tolist() # 去除可能存在的空值或空白字符 keywords [str(k).strip() for k in keywords if pd.notna(k) and str(k).strip()] print(f成功从 {file_path} 加载了 {len(keywords)} 个关键词。) return keywords # 测试一下 if __name__ __main__: keywords load_keywords(keywords.csv) # 或 keywords.xlsx print(前5个关键词, keywords[:5])2.2 构造请求并调用API接下来我们写一个函数负责针对一个关键词向API发送请求并解析结果。import requests import json import time # 请替换为你的真实API配置 API_URL https://api.example.com/v1/generate/couplet HEADERS { Content-Type: application/json, # 如果需要API Key认证请在这里添加例如 # Authorization: Bearer YOUR_API_KEY_HERE } def generate_couplet_for_keyword(keyword, max_retries3): 为单个关键词生成春联。 参数: keyword (str): 春联主题关键词。 max_retries (int): 请求失败时的最大重试次数。 返回: dict: 包含原始关键词和生成结果的字典。如果失败结果部分为None。 # 构造请求数据 payload { prompt: keyword, max_length: 50, # 根据API要求调整 # 可以添加其他API参数例如style: traditional } for attempt in range(max_retries): try: response requests.post(API_URL, headersHEADERS, datajson.dumps(payload), timeout10) response.raise_for_status() # 如果状态码不是200会抛出HTTPError异常 result response.json() # 根据实际的API响应结构解析数据 # 这里假设成功响应包含 data 字段内部有上下联和横批 if result.get(status) success and data in result: data result[data] return { keyword: keyword, upper_line: data.get(upper_line, ), lower_line: data.get(lower_line, ), horizontal: data.get(horizontal, ), status: success } else: print(f警告关键词 {keyword} 的API响应格式异常或失败。响应{result}) return {keyword: keyword, upper_line: None, lower_line: None, horizontal: None, status: api_error} except requests.exceptions.RequestException as e: wait_time 2 ** attempt # 指数退避策略等待时间逐渐变长 print(f关键词 {keyword} 第{attempt1}次请求失败: {e}. {f等待{wait_time}秒后重试... if attempt max_retries - 1 else }) time.sleep(wait_time) except json.JSONDecodeError as e: print(f关键词 {keyword} 响应JSON解析失败: {e}) return {keyword: keyword, upper_line: None, lower_line: None, horizontal: None, status: json_error} # 所有重试都失败 print(f错误关键词 {keyword} 经过{max_retries}次重试后仍然失败。) return {keyword: keyword, upper_line: None, lower_line: None, horizontal: None, status: failed}2.3 单线程循环与结果保存现在我们把读取、请求、保存串联起来形成一个最简单的可工作版本。def main_single_thread(input_file, output_filecouplets_output.csv): 单线程版本的主函数。 print(开始单线程批量生成春联...) start_time time.time() # 1. 加载关键词 keywords load_keywords(input_file) all_results [] # 2. 遍历每个关键词依次请求 for idx, keyword in enumerate(keywords, 1): print(f正在处理 [{idx}/{len(keywords)}]: {keyword}) result generate_couplet_for_keyword(keyword) all_results.append(result) # 小睡一下避免对API造成过大压力根据API限制调整 time.sleep(0.1) # 3. 保存结果到文件 df_output pd.DataFrame(all_results) df_output.to_csv(output_file, indexFalse, encodingutf-8-sig) # utf-8-sig 确保Excel打开中文不乱码 # 如果要保存为Excel: df_output.to_excel(output_file, indexFalse, engineopenpyxl) end_time time.time() print(f处理完成共处理 {len(keywords)} 个关键词耗时 {end_time - start_time:.2f} 秒。) print(f结果已保存至: {output_file}) # 简单统计成功率 success_count df_output[df_output[status] success].shape[0] print(f成功生成: {success_count} 条失败: {len(keywords) - success_count} 条。) if __name__ __main__: # 运行单线程版本 main_single_thread(keywords.csv, couplets_output_single.csv)运行这个脚本你应该能看到它一条一条地处理关键词并最终生成一个包含所有结果的CSV文件。对于几十条数据这还能接受。但如果面对上千条这个速度就太慢了。接下来我们就来给它“提速”。3. 使用多线程加速处理当任务主要是等待网络响应I/O密集型时使用多线程可以显著提升效率。Python的concurrent.futures模块提供了高级的线程池接口用起来很方便。3.1 引入线程池我们改造主函数使用ThreadPoolExecutor来并发执行生成任务。from concurrent.futures import ThreadPoolExecutor, as_completed from tqdm import tqdm def main_multithread(input_file, output_filecouplets_output_multi.csv, max_workers10): 多线程版本的主函数。 参数: max_workers (int): 线程池的最大工作线程数。不宜设置过大避免对目标API造成攻击或触发限流。 一般5-20之间根据API承受能力调整。 print(f开始多线程批量生成春联 (线程数: {max_workers})...) start_time time.time() keywords load_keywords(input_file) all_results [] # 使用线程池 with ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务到线程池得到一个Future对象的列表 # 使用字典记录Future和对应的关键词方便后续关联结果 future_to_keyword {executor.submit(generate_couplet_for_keyword, kw): kw for kw in keywords} # 使用tqdm创建进度条遍历已完成的任务 for future in tqdm(as_completed(future_to_keyword), totallen(keywords), desc生成进度): keyword future_to_keyword[future] try: result future.result() # 获取任务结果 all_results.append(result) except Exception as e: # 理论上generate_couplet_for_keyword内部已处理异常这里以防万一 print(f处理关键词 {keyword} 时发生未捕获的异常: {e}) all_results.append({keyword: keyword, upper_line: None, lower_line: None, horizontal: None, status: future_error}) # 保存结果 df_output pd.DataFrame(all_results) df_output.to_csv(output_file, indexFalse, encodingutf-8-sig) end_time time.time() print(f处理完成共处理 {len(keywords)} 个关键词耗时 {end_time - start_time:.2f} 秒。) print(f结果已保存至: {output_file}) success_count df_output[df_output[status] success].shape[0] print(f成功生成: {success_count} 条失败: {len(keywords) - success_count} 条。) if __name__ __main__: # 运行多线程版本例如使用10个线程 main_multithread(keywords.csv, couplets_output_multi.csv, max_workers10)关键点说明ThreadPoolExecutor管理着一个线程池。max_workers控制并发数。executor.submit()将函数和参数提交到池中返回一个Future对象代表异步计算的结果。as_completed(future_to_keyword)会在任务完成时无论成功失败立即产出对应的Future对象而不是按提交顺序。tqdm包裹这个迭代过程就能显示一个漂亮的进度条实时展示完成数量和预估剩余时间。3.2 关于异步IO的补充说明除了多线程对于这类I/O密集型任务asyncioaiohttp是另一种更高效的选择。它基于事件循环在单线程内实现并发资源开销更小。但代码写法上需要用到async/await语法改造幅度稍大。如果你的请求量极大数万且API服务稳定可以考虑深入学习并使用asyncio。对于大多数百千级别的批量任务多线程方案已经足够简单有效。4. 增强脚本的健壮性与实用性一个能投入生产使用的脚本不能只关注“跑通”还要考虑稳定、友好和可维护。我们来给它加上错误重试已在generate_couplet_for_keyword中实现、进度监控已用tqdm以及更完善的日志和结果处理。4.1 添加更详细的日志记录将打印信息分级并可选地输出到日志文件方便事后排查问题。import logging def setup_logging(log_filebatch_couplet_generation.log): 配置日志记录 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(log_file, encodingutf-8), logging.StreamHandler() # 同时输出到控制台 ] ) return logging.getLogger(__name__) # 在 generate_couplet_for_keyword 函数中将 print 替换为 logger.warning/logger.error # 例如 # logger logging.getLogger(__name__) # logger.warning(f关键词 {keyword} 的API响应格式异常或失败。响应{result})4.2 处理失败任务与结果分类保存我们可能希望把成功的和失败的记录分开保存方便后续处理比如重试失败的任务。def save_results_with_retry(all_results, output_prefixcouplets_output): 保存结果并根据状态将成功和失败的分开存储。 df_all pd.DataFrame(all_results) # 保存完整结果 full_output f{output_prefix}_all.csv df_all.to_csv(full_output, indexFalse, encodingutf-8-sig) print(f完整结果已保存至: {full_output}) # 保存成功的结果 df_success df_all[df_all[status] success] if not df_success.empty: success_output f{output_prefix}_success.csv df_success.to_csv(success_output, indexFalse, encodingutf-8-sig) print(f成功结果已保存至: {success_output} ({len(df_success)} 条)) # 保存失败的关键词方便重试 df_failed df_all[df_all[status] ! success] if not df_failed.empty: failed_keywords df_failed[[keyword]].copy() failed_output f{output_prefix}_failed_keywords.csv failed_keywords.to_csv(failed_output, indexFalse, encodingutf-8-sig, headerFalse) # 只保存关键词无表头 print(f失败关键词列表已保存至: {failed_output} ({len(df_failed)} 条)可用于重试。) return df_all然后在主函数末尾调用这个增强的保存函数即可。4.3 完整的增强版脚本示例将以上改进点整合一个更健壮、更实用的脚本框架如下import pandas as pd import requests import json import time import logging from concurrent.futures import ThreadPoolExecutor, as_completed from tqdm import tqdm # 配置区域 API_URL YOUR_ACTUAL_API_ENDPOINT_HERE HEADERS { Content-Type: application/json, # Authorization: Bearer YOUR_API_KEY } INPUT_FILE keywords.csv OUTPUT_PREFIX couplets_batch_results MAX_WORKERS 15 # 并发线程数请谨慎设置 MAX_RETRIES 3 # 单个请求最大重试次数 REQUEST_TIMEOUT 15 # 请求超时时间秒 # 配置结束 logger logging.getLogger(__name__) def setup_logging(): logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[logging.StreamHandler()]) def load_keywords(file_path): # ... (同前略) ... pass def generate_couplet_for_keyword(keyword, max_retriesMAX_RETRIES): # ... (同前但内部使用logger记录) ... pass def save_results_separately(all_results, prefix): # ... (同前面的 save_results_with_retry 函数) ... pass def main(): setup_logging() logger.info( 春联批量生成任务开始 ) # 1. 加载数据 try: keywords load_keywords(INPUT_FILE) if not keywords: logger.error(未加载到任何有效关键词程序退出。) return except Exception as e: logger.exception(f加载关键词文件失败: {e}) return all_results [] logger.info(f已加载 {len(keywords)} 个关键词开始并发生成...) # 2. 并发处理 start_time time.time() with ThreadPoolExecutor(max_workersMAX_WORKERS) as executor: future_to_kw {executor.submit(generate_couplet_for_keyword, kw): kw for kw in keywords} for future in tqdm(as_completed(future_to_kw), totallen(keywords), desc生成进度): kw future_to_kw[future] try: result future.result() all_results.append(result) except Exception as e: logger.error(f任务执行框架层捕获到异常 (关键词: {kw}): {e}) all_results.append({keyword: kw, upper_line: None, lower_line: None, horizontal: None, status: executor_error}) # 3. 保存与统计 elapsed time.time() - start_time logger.info(f所有任务处理完毕总耗时: {elapsed:.2f} 秒平均每条 {elapsed/len(keywords):.2f} 秒。) df_all save_results_separately(all_results, OUTPUT_PREFIX) # 最终统计 status_counts df_all[status].value_counts() logger.info( 任务完成统计 ) for status, count in status_counts.items(): logger.info(f状态 {status}: {count} 条) success_rate (status_counts.get(success, 0) / len(keywords)) * 100 logger.info(f整体成功率: {success_rate:.2f}%) if __name__ __main__: main()5. 总结走完这一趟我们从最简单的单条请求一步步构建了一个能自动化、批量化、稳定高效生成春联的Python脚本。核心的脉络很清晰读取数据、构造请求、并发执行、处理结果。在这个过程中我们不仅用了requests和pandas这样的基础工具还引入了concurrent.futures来大幅提升处理速度并用tqdm让等待过程可视化最后通过错误重试和结果分类保存增强了脚本的实用性。实际使用时最关键的一步是替换脚本开头的API_URL和可能的认证信息确保它能正确连接到你的春联生成服务。MAX_WORKERS这个参数需要根据API服务的承受能力小心调整一开始可以设小一点比如5观察没有问题再慢慢增加避免因请求过快被限制。这个脚本的框架其实很通用。稍加修改比如调整请求的数据结构和解析响应结果的逻辑它就能用来批量调用其他各类文本生成、图像生成的AI API。希望这个实战教程能帮你解决实际问题让你在需要处理大量自动化任务时能多一个得力的工具。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。