批量提问程序开发方案：基于Python的百度文小言接口实现

1. 项目概述

1.1 项目背景

在现代信息检索和自动化办公场景中，批量提问功能已成为提高工作效率的重要工具。本项目旨在开发一个基于Python的批量提问程序，专门针对百度文小言平台（假设为百度旗下的某个问答或AI对话平台），实现自动化、批量化的问题提交和答案收集功能。

1.2 需求分析

客户核心需求包括：

能够批量导入问题列表
自动向百度文小言平台提交问题
收集并整理返回的答案
提供友好的图形用户界面(GUI)
打包为可执行文件(exe和app)便于分发
支持多线程处理提高效率
具备错误处理和日志记录功能

1.3 技术选型

开发语言：Python 3.8+
GUI框架：PyQt5
网络请求：requests/httpx
多线程处理：concurrent.futures
打包工具：PyInstaller (exe), Py2App (app)
配置文件：JSON/YAML
日志系统：logging模块

2. 系统设计

2.1 架构设计

本系统采用三层架构设计：

表示层：GUI界面，负责用户交互
业务逻辑层：核心处理模块，包括问题处理、网络请求、答案解析等
数据访问层：负责问题列表的读取和答案的存储

2.2 模块划分

主控模块：程序入口和流程控制
界面模块：GUI设计和事件处理
网络模块：与百度文小言平台的API交互
数据处理模块：问题列表和答案的导入导出
配置模块：程序配置管理
日志模块：运行日志记录
打包模块：生成可执行文件

2.3 类设计

class BatchQuestionApp:"""主应用程序类"""class MainWindow(QMainWindow):"""主窗口类"""class QuestionProcessor:"""问题处理核心类"""class WenxiaoyanAPI:"""百度文小言API封装类"""class DataManager:"""数据管理类"""class ConfigManager:"""配置管理类"""

3. 详细实现

3.1 环境准备

首先创建Python虚拟环境并安装必要依赖：

python -m venv venv
source venv/bin/activate  # Linux/Mac
venv\Scripts\activate  # Windowspip install PyQt5 requests httpx pyinstaller py2app openpyxl pandas

3.2 主程序实现

import sys
import os
import json
import logging
from concurrent.futures import ThreadPoolExecutor
from PyQt5.QtWidgets import (QApplication, QMainWindow, QVBoxLayout, QHBoxLayout, QPushButton, QTextEdit, QLabel, QFileDialog, QProgressBar, QWidget, QMessageBox, QComboBox)
from PyQt5.QtCore import QThread, pyqtSignal
import requests
from configparser import ConfigParser
from datetime import datetime# 初始化日志系统
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',handlers=[logging.FileHandler('batch_question.log'),logging.StreamHandler()]
)
logger = logging.getLogger(__name__)class WenxiaoyanAPI:"""百度文小言API封装类"""def __init__(self, api_key=None, base_url="https://api.wenxiaoyan.baidu.com/v1"):self.api_key = api_keyself.base_url = base_urlself.session = requests.Session()self.session.headers.update({"Content-Type": "application/json","Authorization": f"Bearer {self.api_key}"})def ask_question(self, question, max_retries=3, timeout=30):"""向文小言提问"""url = f"{self.base_url}/question"payload = {"question": question,"timestamp": int(datetime.now().timestamp())}for attempt in range(max_retries):try:response = self.session.post(url, json=payload,timeout=timeout)response.raise_for_status()return response.json()except requests.exceptions.RequestException as e:logger.warning(f"提问失败(尝试 {attempt + 1}/{max_retries}): {str(e)}")if attempt == max_retries - 1:logger.error(f"提问'{question[:20]}...'最终失败: {str(e)}")return {"error": str(e)}def batch_ask(self, questions, max_workers=5):"""批量提问"""results = []with ThreadPoolExecutor(max_workers=max_workers) as executor:futures = {executor.submit(self.ask_question, q): q for q in questions}for future in concurrent.futures.as_completed(futures):question = futures[future]try:result = future.result()results.append((question, result))except Exception as e:logger.error(f"处理问题'{question[:20]}...'时出错: {str(e)}")results.append((question, {"error": str(e)}))return resultsclass QuestionProcessor(QThread):"""问题处理线程"""progress_updated = pyqtSignal(int, int)  # 当前进度, 总数question_done = pyqtSignal(str, dict)   # 问题, 结果finished = pyqtSignal(list)             # 所有结果def __init__(self, questions, api, parent=None):super().__init__(parent)self.questions = questionsself.api = apiself.results = []def run(self):total = len(self.questions)for i, question in enumerate(self.questions):try:result = self.api.ask_question(question)self.results.append((question, result))self.progress_updated.emit(i+1, total)self.question_done.emit(question, result)except Exception as e:logger.error(f"处理问题'{question[:20]}...'时出错: {str(e)}")self.results.append((question, {"error": str(e)}))self.finished.emit(self.results)class DataManager:"""数据管理类"""@staticmethoddef load_questions(file_path):"""从文件加载问题列表"""if not os.path.exists(file_path):raise FileNotFoundError(f"文件不存在: {file_path}")ext = os.path.splitext(file_path)[1].lower()try:if ext == '.txt':with open(file_path, 'r', encoding='utf-8') as f:return [line.strip() for line in f if line.strip()]elif ext == '.json':with open(file_path, 'r', encoding='utf-8') as f:data = json.load(f)if isinstance(data, list):return dataelif isinstance(data, dict) and 'questions' in data:return data['questions']else:raise ValueError("JSON格式不支持")elif ext in ('.xlsx', '.xls'):import pandas as pddf = pd.read_excel(file_path)if 'question' in df.columns:return df['question'].tolist()else:return df.iloc[:, 0].tolist()else:raise ValueError(f"不支持的文件格式: {ext}")except Exception as e:logger.error(f"加载问题文件失败: {str(e)}")raise@staticmethoddef save_results(results, file_path, format='json'):"""保存结果到文件"""try:if format == 'json':with open(file_path, 'w', encoding='utf-8') as f:json.dump(results, f, ensure_ascii=False, indent=2)elif format == 'txt':with open(file_path, 'w', encoding='utf-8') as f:for q, a in results:f.write(f"问题: {q}\n")if 'answer' in a:f.write(f"答案: {a['answer']}\n\n")elif 'error' in a:f.write(f"错误: {a['error']}\n\n")else:f.write(f"响应: {str(a)}\n\n")elif format in ('xlsx', 'xls'):import pandas as pddata = []for q, a in results:row = {'问题': q}if 'answer' in a:row.update({'答案': a['answer']})elif 'error' in a:row.update({'错误': a['error']})else:row.update({'响应': str(a)})data.append(row)df = pd.DataFrame(data)df.to_excel(file_path, index=False)else:raise ValueError(f"不支持的保存格式: {format}")except Exception as e:logger.error(f"保存结果失败: {str(e)}")raiseclass ConfigManager:"""配置管理类"""def __init__(self, config_file='config.ini'):self.config_file = config_fileself.config = ConfigParser()# 默认配置self.config['DEFAULT'] = {'api_key': '','api_url': 'https://api.wenxiaoyan.baidu.com/v1','max_workers': '5','timeout': '30','max_retries': '3','last_input_dir': '','last_output_dir': '','output_format': 'json'}if os.path.exists(self.config_file):self.config.read(self.config_file)else:self.save_config()def save_config(self):"""保存配置到文件"""with open(self.config_file, 'w') as f:self.config.write(f)def get(self, section, option):"""获取配置项"""return self.config.get(section, option)def set(self, section, option, value):"""设置配置项"""if not self.config.has_section(section):self.config.add_section(section)self.config.set(section, option, str(value))self.save_config()class MainWindow(QMainWindow):"""主窗口类"""def __init__(self):super().__init__()self.setWindowTitle("百度文小言批量提问工具")self.resize(800, 600)# 初始化配置self.config = ConfigManager()self.api = WenxiaoyanAPI(api_key=self.config.get('DEFAULT', 'api_key'),base_url=self.config.get('DEFAULT', 'api_url'))# 初始化UIself.init_ui()# 当前状态self.questions = []self.results = []def init_ui(self):"""初始化用户界面"""# 主窗口部件central_widget = QWidget()self.setCentralWidget(central_widget)# 主布局main_layout = QVBoxLayout()central_widget.setLayout(main_layout)# API设置区域api_layout = QHBoxLayout()main_layout.addLayout(api_layout)api_layout.addWidget(QLabel("API Key:"))self.api_key_edit = QTextEdit()self.api_key_edit.setMaximumHeight(30)self.api_key_edit.setPlainText(self.config.get('DEFAULT', 'api_key'))api_layout.addWidget(self.api_key_edit)# 输入区域input_layout = QHBoxLayout()main_layout.addLayout(input_layout)self.input_edit = QTextEdit()self.input_edit.setPlaceholderText("在此输入问题，每行一个...")input_layout.addWidget(self.input_edit)# 按钮区域button_layout = QHBoxLayout()main_layout.addLayout(button_layout)self.load_btn = QPushButton("加载问题文件")self.load_btn.clicked.connect(self.load_questions)button_layout.addWidget(self.load_btn)self.clear_btn = QPushButton("清空问题")self.clear_btn.clicked.connect(self.clear_questions)button_layout.addWidget(self.clear_btn)self.start_btn = QPushButton("开始提问")self.start_btn.clicked.connect(self.start_processing)button_layout.addWidget(self.start_btn)self.stop_btn = QPushButton("停止")self.stop_btn.setEnabled(False)self.stop_btn.clicked.connect(self.stop_processing)button_layout.addWidget(self.stop_btn)self.save_btn = QPushButton("保存结果")self.save_btn.setEnabled(False)self.save_btn.clicked.connect(self.save_results)button_layout.addWidget(self.save_btn)# 输出区域output_layout = QVBoxLayout()main_layout.addLayout(output_layout)output_layout.addWidget(QLabel("输出结果:"))self.output_edit = QTextEdit()self.output_edit.setReadOnly(True)output_layout.addWidget(self.output_edit)# 进度条self.progress_bar = QProgressBar()main_layout.addWidget(self.progress_bar)# 状态栏self.statusBar().showMessage("准备就绪")# 输出格式选择format_layout = QHBoxLayout()main_layout.addLayout(format_layout)format_layout.addWidget(QLabel("输出格式:"))self.format_combo = QComboBox()self.format_combo.addItems(["JSON", "TXT", "Excel"])self.format_combo.setCurrentText(self.config.get('DEFAULT', 'output_format').upper())format_layout.addWidget(self.format_combo)def load_questions(self):"""加载问题文件"""last_dir = self.config.get('DEFAULT', 'last_input_dir') or os.path.expanduser('~')file_path, _ = QFileDialog.getOpenFileName(self, "选择问题文件", last_dir, "文本文件 (*.txt);;JSON文件 (*.json);;Excel文件 (*.xlsx *.xls)")if file_path:try:self.config.set('DEFAULT', 'last_input_dir', os.path.dirname(file_path))self.questions = DataManager.load_questions(file_path)self.input_edit.setPlainText("\n".join(self.questions))self.statusBar().showMessage(f"已加载 {len(self.questions)} 个问题")except Exception as e:QMessageBox.critical(self, "错误", f"加载问题文件失败: {str(e)}")def clear_questions(self):"""清空问题列表"""self.questions = []self.input_edit.clear()self.statusBar().showMessage("问题列表已清空")def start_processing(self):"""开始处理问题"""# 更新API配置api_key = self.api_key_edit.toPlainText().strip()if not api_key:QMessageBox.warning(self, "警告", "请输入API Key")returnself.config.set('DEFAULT', 'api_key', api_key)self.api.api_key = api_key# 获取问题列表input_text = self.input_edit.toPlainText().strip()if input_text:self.questions = [q.strip() for q in input_text.split('\n') if q.strip()]if not self.questions:QMessageBox.warning(self, "警告", "没有可处理的问题")return# 清空之前的结果self.results = []self.output_edit.clear()# 更新UI状态self.start_btn.setEnabled(False)self.stop_btn.setEnabled(True)self.load_btn.setEnabled(False)self.clear_btn.setEnabled(False)self.progress_bar.setMaximum(len(self.questions))self.progress_bar.setValue(0)# 启动处理线程self.processor = QuestionProcessor(self.questions, self.api)self.processor.progress_updated.connect(self.update_progress)self.processor.question_done.connect(self.display_result)self.processor.finished.connect(self.on_processing_finished)self.processor.start()self.statusBar().showMessage("正在处理问题...")def stop_processing(self):"""停止处理问题"""if hasattr(self, 'processor') and self.processor.isRunning():self.processor.terminate()self.statusBar().showMessage("处理已停止")self.on_processing_finished([])def update_progress(self, current, total):"""更新进度"""self.progress_bar.setValue(current)self.statusBar().showMessage(f"处理中: {current}/{total}")def display_result(self, question, result):"""显示单个结果"""self.results.append((question, result))output = f"问题: {question}\n"if 'answer' in result:output += f"答案: {result['answer']}\n\n"elif 'error' in result:output += f"错误: {result['error']}\n\n"else:output += f"响应: {str(result)}\n\n"self.output_edit.append(output)def on_processing_finished(self, results):"""处理完成"""self.results = results or self.resultsself.start_btn.setEnabled(True)self.stop_btn.setEnabled(False)self.load_btn.setEnabled(True)self.clear_btn.setEnabled(True)self.save_btn.setEnabled(True)success = sum(1 for _, r in self.results if 'answer' in r)failed = len(self.results) - successself.statusBar().showMessage(f"处理完成! 成功: {success}, 失败: {failed}")def save_results(self):"""保存结果"""if not self.results:QMessageBox.warning(self, "警告", "没有可保存的结果")returnlast_dir = self.config.get('DEFAULT', 'last_output_dir') or os.path.expanduser('~')default_name = f"results_{datetime.now().strftime('%Y%m%d_%H%M%S')}"format = self.format_combo.currentText().lower()self.config.set('DEFAULT', 'output_format', format)if format == 'excel':ext = '.xlsx'file_filter = "Excel文件 (*.xlsx)"elif format == 'txt':ext = '.txt'file_filter = "文本文件 (*.txt)"else:  # jsonext = '.json'file_filter = "JSON文件 (*.json)"file_path, _ = QFileDialog.getSaveFileName(self,"保存结果",os.path.join(last_dir, default_name + ext),file_filter)if file_path:try:DataManager.save_results(self.results, file_path, format)self.config.set('DEFAULT', 'last_output_dir', os.path.dirname(file_path))QMessageBox.information(self, "成功", "结果保存成功!")except Exception as e:QMessageBox.critical(self, "错误", f"保存结果失败: {str(e)}")def main():"""主函数"""app = QApplication(sys.argv)# 设置应用程序样式app.setStyle('Fusion')# 创建并显示主窗口window = MainWindow()window.show()# 运行应用程序sys.exit(app.exec_())if __name__ == '__main__':main()

3.3 配置文件设计

创建config.ini配置文件：

[DEFAULT]
api_key = your_api_key_here
api_url = https://api.wenxiaoyan.baidu.com/v1
max_workers = 5
timeout = 30
max_retries = 3
last_input_dir = 
last_output_dir = 
output_format = json

3.4 打包为可执行文件

3.4.1 打包为Windows EXE

创建setup_pyinstaller.py:

# setup_pyinstaller.py
import PyInstaller.__main__PyInstaller.__main__.run(['--name=BatchQuestionTool','--onefile','--windowed','--icon=app.ico',  # 可选，指定图标'--add-data=config.ini;.','main.py'
])

运行命令：

python setup_pyinstaller.py

3.4.2 打包为MacOS APP

创建setup.py:

# setup.py
from setuptools import setupAPP = ['main.py']
DATA_FILES = ['config.ini']
OPTIONS = {'argv_emulation': True,'iconfile': 'app.icns',  # 可选，指定图标'plist': {'CFBundleName': 'BatchQuestionTool','CFBundleDisplayName': '批量提问工具','CFBundleVersion': '1.0.0','CFBundleIdentifier': 'com.yourcompany.batchquestiontool',}
}setup(app=APP,data_files=DATA_FILES,options={'py2app': OPTIONS},setup_requires=['py2app'],
)

运行命令：

python setup.py py2app

4. 功能扩展与优化

4.1 高级功能实现

4.1.1 代理支持

在WenxiaoyanAPI类中添加代理支持：

class WenxiaoyanAPI:def __init__(self, api_key=None, base_url="https://api.wenxiaoyan.baidu.com/v1", proxy=None):self.proxy = proxy# ...其他初始化代码...def ask_question(self, question, max_retries=3, timeout=30):proxies = Noneif self.proxy:proxies = {'http': self.proxy,'https': self.proxy}try:response = self.session.post(url, json=payload,timeout=timeout,proxies=proxies)# ...其余代码...

4.1.2 结果后处理

添加结果后处理功能，如关键词提取、情感分析等：

class ResultProcessor:"""结果后处理类"""@staticmethoddef extract_keywords(text, top_n=5):"""提取关键词"""import jieba.analysereturn jieba.analyse.extract_tags(text, topK=top_n)@staticmethoddef sentiment_analysis(text):"""情感分析"""from snownlp import SnowNLPreturn SnowNLP(text).sentiments@staticmethoddef summarize(text, ratio=0.2):"""文本摘要"""from gensim.summarization import summarizereturn summarize(text, ratio=ratio)

4.1.3 问题模板支持

实现问题模板功能，支持变量替换：

class QuestionTemplate:"""问题模板处理器"""def __init__(self, template_file=None):self.templates = {}if template_file:self.load_templates(template_file)def load_templates(self, file_path):"""加载模板文件"""with open(file_path, 'r', encoding='utf-8') as f:self.templates = json.load(f)def apply_template(self, template_name, variables):"""应用模板"""if template_name not in self.templates:raise ValueError(f"模板不存在: {template_name}")template = self.templates[template_name]return template.format(**variables)

4.2 性能优化

4.2.1 异步请求处理

使用aiohttp实现异步请求，提高大批量问题处理效率：

import aiohttp
import asyncioclass AsyncWenxiaoyanAPI:"""异步API客户端"""def __init__(self, api_key, base_url="https://api.wenxiaoyan.baidu.com/v1", max_concurrent=10):self.api_key = api_keyself.base_url = base_urlself.max_concurrent = max_concurrentself.semaphore = asyncio.Semaphore(max_concurrent)async def ask_question(self, session, question):url = f"{self.base_url}/question"payload = {"question": question,"timestamp": int(datetime.now().timestamp())}headers = {"Content-Type": "application/json","Authorization": f"Bearer {self.api_key}"}async with self.semaphore:try:async with session.post(url, json=payload, headers=headers) as response:response.raise_for_status()return await response.json()except Exception as e:logger.error(f"提问'{question[:20]}...'失败: {str(e)}")return {"error": str(e)}async def batch_ask(self, questions):connector = aiohttp.TCPConnector(limit=self.max_concurrent)async with aiohttp.ClientSession(connector=connector) as session:tasks = [self.ask_question(session, q) for q in questions]return await asyncio.gather(*tasks)

4.2.2 缓存机制

实现简单的缓存机制，避免重复提问相同问题：

import hashlib
from functools import lru_cacheclass CachedWenxiaoyanAPI(WenxiaoyanAPI):"""带缓存的API客户端"""def __init__(self, *args, cache_file='question_cache.json', **kwargs):super().__init__(*args, **kwargs)self.cache_file = cache_fileself.cache = self._load_cache()def _load_cache(self):if os.path.exists(self.cache_file):with open(self.cache_file, 'r', encoding='utf-8') as f:return json.load(f)return {}def _save_cache(self):with open(self.cache_file, 'w', encoding='utf-8') as f:json.dump(self.cache, f, ensure_ascii=False, indent=2)def _get_cache_key(self, question):return hashlib.md5(question.encode('utf-8')).hexdigest()def ask_question(self, question, max_retries=3, timeout=30):cache_key = self._get_cache_key(question)if cache_key in self.cache:return self.cache[cache_key]result = super().ask_question(question, max_retries, timeout)self.cache[cache_key] = resultself._save_cache()return result

4.3 安全性增强

4.3.1 API密钥加密存储

from cryptography.fernet import Fernetclass SecureConfigManager(ConfigManager):"""安全的配置管理器，加密敏感信息"""def __init__(self, config_file='config.ini', key_file='config.key'):self.key_file = key_fileself.cipher = self._init_cipher()super().__init__(config_file)def _init_cipher(self):if os.path.exists(self.key_file):with open(self.key_file, 'rb') as f:key = f.read()else:key = Fernet.generate_key()with open(self.key_file, 'wb') as f:f.write(key)return Fernet(key)def _encrypt(self, text):return self.cipher.encrypt(text.encode('utf-8')).decode('utf-8')def _decrypt(self, text):return self.cipher.decrypt(text.encode('utf-8')).decode('utf-8')def get(self, section, option):value = super().get(section, option)if option in ('api_key', 'password'):try:return self._decrypt(value)except:return valuereturn valuedef set(self, section, option, value):if option in ('api_key', 'password'):value = self._encrypt(value)super().set(section, option, value)

4.3.2 请求签名验证

import hmac
import base64class SignedWenxiaoyanAPI(WenxiaoyanAPI):"""带请求签名的API客户端"""def __init__(self, api_key, api_secret, *args, **kwargs):self.api_secret = api_secretsuper().__init__(api_key, *args, **kwargs)def _generate_signature(self, payload):timestamp = str(payload['timestamp'])message = timestamp + self.api_keysignature = hmac.new(self.api_secret.encode('utf-8'),message.encode('utf-8'),hashlib.sha256).digest()return base64.b64encode(signature).decode('utf-8')def ask_question(self, question, max_retries=3, timeout=30):payload = {"question": question,"timestamp": int(datetime.now().timestamp())}payload['signature'] = self._generate_signature(payload)# 其余部分与父类相同...

5. 测试方案

5.1 单元测试

创建test_wenxiaoyan.py:

import unittest
from unittest.mock import patch, MagicMock
from main import WenxiaoyanAPI, DataManagerclass TestWenxiaoyanAPI(unittest.TestCase):def setUp(self):self.api = WenxiaoyanAPI(api_key="test_key")@patch('requests.Session.post')def test_ask_question_success(self, mock_post):mock_response = MagicMock()mock_response.status_code = 200mock_response.json.return_value = {"answer": "测试答案"}mock_post.return_value = mock_responseresult = self.api.ask_question("测试问题")self.assertEqual(result, {"answer": "测试答案"})@patch('requests.Session.post')def test_ask_question_failure(self, mock_post):mock_post.side_effect = requests.exceptions.RequestException("请求失败")result = self.api.ask_question("测试问题")self.assertIn("error", result)class TestDataManager(unittest.TestCase):def test_load_questions_txt(self):with open('test_questions.txt', 'w', encoding='utf-8') as f:f.write("问题1\n问题2\n问题3\n")questions = DataManager.load_questions('test_questions.txt')self.assertEqual(questions, ["问题1", "问题2", "问题3"])def test_load_questions_json(self):with open('test_questions.json', 'w', encoding='utf-8') as f:json.dump({"questions": ["问题1", "问题2"]}, f)questions = DataManager.load_questions('test_questions.json')self.assertEqual(questions, ["问题1", "问题2"])if __name__ == '__main__':unittest.main()

5.2 集成测试

创建test_integration.py:

import unittest
from main import MainWindow, QApplication
from PyQt5.QtTest import QTest
from PyQt5.QtCore import Qtclass TestMainWindow(unittest.TestCase):@classmethoddef setUpClass(cls):cls.app = QApplication([])def setUp(self):self.window = MainWindow()self.window.show()def test_load_questions(self):# 模拟点击加载按钮QTest.mouseClick(self.window.load_btn, Qt.LeftButton)# TODO: 添加文件选择模拟和验证def tearDown(self):self.window.close()if __name__ == '__main__':unittest.main()

5.3 性能测试

创建benchmark.py:

import time
from main import WenxiaoyanAPIdef benchmark():api = WenxiaoyanAPI(api_key="test_key")questions = [f"测试问题{i}" for i in range(100)]# 测试单线程性能start = time.time()for q in questions:api.ask_question(q)single_thread_time = time.time() - start# 测试多线程性能start = time.time()api.batch_ask(questions)multi_thread_time = time.time() - startprint(f"单线程耗时: {single_thread_time:.2f}秒")print(f"多线程耗时: {multi_thread_time:.2f}秒")print(f"性能提升: {single_thread_time/multi_thread_time:.1f}倍")if __name__ == '__main__':benchmark()

6. 部署与维护

6.1 安装程序制作

6.1.1 Windows安装程序

使用Inno Setup创建安装程序：

创建setup.iss脚本：

[Setup]
AppName=百度文小言批量提问工具
AppVersion=1.0
DefaultDirName={pf}\BatchQuestionTool
DefaultGroupName=批量提问工具
OutputDir=output
OutputBaseFilename=BatchQuestionTool_Setup
Compression=lzma
SolidCompression=yes[Files]
Source: "dist\BatchQuestionTool.exe"; DestDir: "{app}"; Flags: ignoreversion
Source: "config.ini"; DestDir: "{app}"; Flags: onlyifdoesntexist
Source: "README.md"; DestDir: "{app}"; Flags: isreadme[Icons]
Name: "{group}\批量提问工具"; Filename: "{app}\BatchQuestionTool.exe"
Name: "{commondesktop}\批量提问工具"; Filename: "{app}\BatchQuestionTool.exe"

使用Inno Setup Compiler编译安装程序

6.1.2 MacOS安装程序

使用create-dmg创建DMG安装包：

安装create-dmg:

npm install --global create-dmg

创建DMG:

create-dmg \--volname "批量提问工具" \--volicon "app.icns" \--background "background.png" \--window-pos 200 120 \--window-size 800 400 \--icon-size 100 \--icon "BatchQuestionTool.app" 200 190 \--hide-extension "BatchQuestionTool.app" \--app-drop-link 600 190 \"BatchQuestionTool.dmg" \"dist/BatchQuestionTool.app"

6.2 自动更新机制

实现简单的自动更新检查功能：

class UpdateChecker:"""更新检查器"""def __init__(self, current_version, update_url):self.current_version = current_versionself.update_url = update_urldef check_update(self):try:response = requests.get(self.update_url, timeout=10)response.raise_for_status()latest_info = response.json()if latest_info['version'] > self.current_version:return {'available': True,'version': latest_info['version'],'release_notes': latest_info['release_notes'],'download_url': latest_info['download_url']}return {'available': False}except Exception as e:logger.error(f"检查更新失败: {str(e)}")return {'available': False, 'error': str(e)}def download_update(self, progress_callback=None):# 实现下载逻辑pass

在MainWindow中添加更新检查：

class MainWindow(QMainWindow):def __init__(self):# ...原有代码...self.check_for_updates()def check_for_updates(self):self.update_checker = UpdateChecker(current_version="1.0.0",update_url="https://api.yourdomain.com/update/check")def update_check_done(result):if result.get('available'):self.show_update_dialog(result)self.update_thread = QThread()self.update_worker = UpdateWorker(self.update_checker)self.update_worker.moveToThread(self.update_thread)self.update_worker.finished.connect(update_check_done)self.update_thread.started.connect(self.update_worker.run)self.update_thread.start()def show_update_dialog(self, update_info):# 显示更新对话框passclass UpdateWorker(QObject):finished = pyqtSignal(dict)def __init__(self, update_checker):super().__init__()self.update_checker = update_checkerdef run(self):result = self.update_checker.check_update()self.finished.emit(result)

6.3 错误报告系统

实现错误报告功能：

class ErrorReporter:"""错误报告系统"""def __init__(self, report_url, app_version):self.report_url = report_urlself.app_version = app_versiondef report_error(self, error, context=None):"""报告错误到服务器"""data = {'timestamp': datetime.now().isoformat(),'version': self.app_version,'error': str(error),'context': context or {},'system': {'platform': sys.platform,'python_version': sys.version}}try:response = requests.post(self.report_url,json=data,timeout=10)return response.status_code == 200except Exception:return False

在MainWindow中集成错误报告：

class MainWindow(QMainWindow):def __init__(self):self.error_reporter = ErrorReporter(report_url="https://api.yourdomain.com/error/report",app_version="1.0.0")sys.excepthook = self.handle_uncaught_exceptiondef handle_uncaught_exception(self, exc_type, exc_value, exc_traceback):"""处理未捕获的异常"""logger.critical("未捕获的异常", exc_info=(exc_type, exc_value, exc_traceback))# 报告错误self.error_reporter.report_error(exc_value, {'type': str(exc_type),'traceback': traceback.format_tb(exc_traceback)})# 显示错误对话框QMessageBox.critical(None,"未预期的错误",f"发生未预期的错误: {str(exc_value)}\n\n错误已自动报告给开发团队。")

7. 用户文档

7.1 使用说明

首次运行配置
- 启动程序后，在API Key输入框中输入您的百度文小言API密钥
- 点击"保存配置"按钮保存设置
加载问题
- 点击"加载问题文件"按钮选择包含问题的文本文件(.txt)、JSON文件(.json)或Excel文件(.xlsx)
- 或者直接在输入框中输入问题，每行一个问题
开始提问
- 点击"开始提问"按钮开始处理问题列表
- 进度条将显示处理进度
- 结果将实时显示在输出区域
保存结果
- 处理完成后，点击"保存结果"按钮
- 选择保存格式(JSON/TXT/Excel)和保存位置
高级设置
- 在菜单栏的"设置"中可以配置:
  - 并发线程数
  - 请求超时时间
  - 最大重试次数
  - 代理设置

7.2 常见问题解答

Q: 如何获取百度文小言的API Key?
A: 请登录百度云平台，在AI服务中找到文小言API服务，申请开通后即可获取API Key。

Q: 支持哪些格式的问题文件?
A: 程序支持纯文本(.txt)、JSON(.json)和Excel(.xlsx, .xls)格式。文本文件每行一个问题；JSON文件可以是问题数组或包含questions字段的对象；Excel文件可以是第一列或名为"question"的列。

Q: 处理大量问题时程序无响应怎么办?
A: 这是正常现象，程序正在后台处理问题。如需停止处理，可以点击"停止"按钮。建议处理大量问题时减少并发线程数。

Q: 如何查看详细的错误信息?
A: 程序目录下的batch_question.log文件记录了详细运行日志，包括错误信息。

8. 开发路线图

8.1 短期计划 (1-2个月)

功能增强
- 实现问题分类和标签功能
- 添加答案质量评分机制
- 支持自定义请求头参数
性能优化
- 实现请求批量化处理
- 优化内存使用效率
- 改进多线程任务调度
用户体验
- 添加黑暗模式支持
- 实现界面语言切换
- 优化结果展示格式

8.2 中期计划 (3-6个月)

高级分析功能
- 集成答案相似度分析
- 实现答案自动摘要
- 添加情感分析功能
云服务集成
- 支持百度云对象存储(BOS)直接存取
- 实现与百度大脑其他AI服务集成
- 添加团队协作功能
扩展性增强
- 开发插件系统支持第三方扩展
- 提供RESTful API供其他程序调用
- 支持自定义脚本处理流程

8.3 长期计划 (6-12个月)

AI增强
- 集成问题自动生成功能
- 实现智能问题推荐
- 添加自动追问和深度问答功能
企业版功能
- 开发用户权限管理系统
- 实现审计日志和操作追溯
- 支持LDAP/AD集成
生态建设
- 开发移动端应用
- 创建模板市场和插件市场
- 建立开发者社区

9. 项目总结

本项目实现了一个功能完善的百度文小言批量提问工具，具有以下特点和优势：

功能全面
- 支持多种问题输入格式
- 提供灵活的答案输出选项
- 具备完善的配置和错误处理机制
性能优异
- 多线程并发处理提高效率
- 智能缓存减少重复请求
- 资源占用优化
易于使用
- 直观的图形用户界面
- 详细的文档和帮助信息
- 自动更新和错误报告
可扩展性强
- 模块化设计便于功能扩展
- 清晰的API接口文档
- 完善的测试体系保障质量
跨平台支持
- 支持Windows和MacOS平台
- 提供标准安装程序
- 适配不同屏幕尺寸

通过本项目的实施，客户可以显著提高使用百度文小言平台的工作效率，实现问题的批量处理和答案的自动化收集，为数据分析和知识管理提供有力支持。

10. 附录

10.1 完整代码结构

BatchQuestionTool/
│
├── main.py                 # 主程序入口
├── config.ini              # 配置文件
├── requirements.txt        # 依赖列表
│
├── src/                    # 源代码目录
│   ├── api/                # API相关代码
│   │   ├── __init__.py
│   │   ├── wenxiaoyan.py   # 文小言API封装
│   │   └── auth.py         # 认证相关
│   │
│   ├── core/               # 核心逻辑
│   │   ├── processor.py    # 问题处理器
│   │   └── datamanager.py  # 数据管理
│   │
│   ├── ui/                 # 用户界面
│   │   ├── mainwindow.py   # 主窗口
│   │   └── dialogs.py      # 各种对话框
│   │
│   ├── utils/              # 工具类
│   │   ├── logger.py       # 日志工具
│   │   └── config.py       # 配置工具
│   │
│   └── resources/          # 资源文件
│       ├── icons/          # 图标资源
│       └── styles/         # 样式表
│
├── tests/                  # 测试代码
│   ├── unit/               # 单元测试
│   └── integration/        # 集成测试
│
├── docs/                   # 文档
│   ├── user_manual.md      # 用户手册
│   └── api_reference.md    # API参考
│
├── scripts/                # 实用脚本
│   ├── build_exe.py        # 构建Windows EXE
│   └── build_app.py        # 构建MacOS APP
│
└── dist/                   # 构建输出目录

10.2 第三方库清单

PyQt5==5.15.7
requests==2.26.0
httpx==0.19.0
openpyxl==3.0.9
pandas==1.3.4
PyInstaller==4.6
py2app==0.28
python-dotenv==0.19.1
cryptography==36.0.0
jieba==0.42.1
snownlp==0.12.3
gensim==4.1.2
aiohttp==3.8.1