#!/usr/bin/env python3 """ Django management command for AI chat with readline support Usage: python manage.py chat """ import os import readline import atexit from django.core.management.base import BaseCommand from products.aichat import AIChatService class ReadlineConfig: """Readline配置类""" def __init__(self, history_file: str = None): self.history_file = history_file or os.path.expanduser("~/.aichat_history") self.commands = ['help', 'history', 'clear', 'readline-help', 'stats', 'quit', 'exit', 'q'] self.setup_readline() def setup_readline(self): """设置readline功能""" # 设置历史文件 try: readline.read_history_file(self.history_file) except FileNotFoundError: pass # 设置历史文件大小 readline.set_history_length(1000) # 注册退出时保存历史 atexit.register(self.save_history) # 设置tab补全 readline.parse_and_bind('tab: complete') # 设置其他有用的绑定 readline.parse_and_bind('set editing-mode emacs') # 使用emacs编辑模式 readline.parse_and_bind('set completion-ignore-case on') # 忽略大小写补全 readline.parse_and_bind('set show-all-if-ambiguous on') # 显示所有可能的补全 # 设置自定义补全器 readline.set_completer(self.completer) # 设置补全分隔符 readline.set_completer_delims(' \t\n`!@#$%^&*()=+[{]}\\|;:\'",<>?') def completer(self, text, state): """自定义补全器""" options = [i for i in self.commands if i.startswith(text)] if state < len(options): return options[state] return None def save_history(self): """保存历史到文件""" try: readline.write_history_file(self.history_file) except Exception as e: print(f"⚠️ 保存历史记录失败: {e}") def add_to_history(self, line: str): """添加行到历史""" if line.strip(): readline.add_history(line.strip()) def get_history_stats(self): """获取历史统计信息""" try: history_length = readline.get_current_history_length() return { 'current_length': history_length, 'max_length': readline.get_history_length(), 'history_file': self.history_file } except Exception: return {'error': '无法获取历史统计信息'} def clear_history(self): """清空readline历史""" try: readline.clear_history() print("🗑️ Readline历史已清空") except Exception as e: print(f"⚠️ 清空历史失败: {e}") class Command(BaseCommand): """AI聊天管理命令""" help = '启动AI聊天系统，支持readline功能' def add_arguments(self, parser): """添加命令行参数""" parser.add_argument( '--history-file', type=str, help='指定历史文件路径 (默认: ~/.aichat_history)' ) parser.add_argument( '--chat-type', type=str, choices=['platform', 'product'], default='platform', help='聊天类型: platform(平台客服) 或 product(产品客服) (默认: platform)' ) parser.add_argument( '--product-id', type=int, help='产品ID，当chat-type为product时必须提供' ) def handle(self, *args, **options): """执行命令""" # 参数验证 chat_type = options.get('chat_type') product_id = options.get('product_id') if chat_type == 'product' and not product_id: self.stderr.write("❌ 错误: 当chat-type为product时，必须提供--product-id参数") self.stderr.write("示例: python manage.py chat --chat-type product --product-id 12345") return # 显示系统信息 self.stdout.write("🚀 徵象防伪验证平台 AI客服系统") self.stdout.write("=" * 50) self.stdout.write(f"聊天类型: {chat_type.upper()}") if product_id: self.stdout.write(f"产品ID: {product_id}") self.stdout.write("输入 'help' 查看帮助信息") self.stdout.write("输入 'history' 查看对话历史") self.stdout.write("输入 'clear' 清空对话历史") self.stdout.write("输入 'quit' 或 'exit' 退出系统") self.stdout.write("=" * 50) self.stdout.write("💡 提示: 使用 ↑↓ 键浏览历史，Tab 键补全，Ctrl+A/E 跳转行首/尾") self.stdout.write("=" * 50) try: # 初始化readline readline_config = ReadlineConfig(options.get('history_file')) # 初始化AI服务，传递聊天类型和产品ID ai_service = AIChatService( chat_type=chat_type, product_id=product_id ) self.stdout.write("✅ AI服务初始化成功！") self.stdout.write("") # 启动聊天循环 self._chat_loop(ai_service, readline_config) except Exception as e: self.stderr.write(f"❌ 系统初始化失败: {str(e)}") self.stderr.write("请检查API密钥配置") def _chat_loop(self, ai_service, readline_config): """聊天主循环""" while True: try: # 获取用户输入 user_input = input("👤 您: ").strip() if not user_input: continue # 添加到readline历史 readline_config.add_to_history(user_input) # 处理特殊命令 if user_input.lower() in ['quit', 'exit', 'q']: self.stdout.write("👋 感谢使用，再见！") break elif user_input.lower() == 'help': self._print_help() continue elif user_input.lower() == 'history': ai_service.print_history() continue elif user_input.lower() == 'clear': ai_service.clear_history() self.stdout.write("🗑️ 对话历史已清空") continue elif user_input.lower() == 'readline-help': self._print_readline_help() continue elif user_input.lower() == 'stats': self._print_readline_stats(readline_config) continue # 发送消息给AI self.stdout.write("🤖 AI助手正在思考...") response = ai_service.chat(user_input) self.stdout.write(f"🤖 AI助手: {response}") self.stdout.write("") except KeyboardInterrupt: self.stdout.write("\n\n👋 感谢使用，再见！") break except EOFError: self.stdout.write("\n👋 再见！") break except Exception as e: self.stderr.write(f"❌ 发生错误: {str(e)}") self.stderr.write("请重试或输入 'quit' 退出") def _print_help(self): """打印帮助信息""" help_text = """ 📖 帮助信息 ========== 基本命令: - 直接输入文字与AI助手对话 - help: 显示此帮助信息 - history: 查看对话历史 - clear: 清空对话历史 - readline-help: 显示readline快捷键帮助 - stats: 显示readline统计信息 - quit/exit/q: 退出系统 功能特性: - 智能客服问答 - 二维码扫描验证（输入 '扫描' 或 '验证' 触发） - 防伪平台信息咨询 - 多轮对话支持 - 持久化输入历史记录 使用提示: 1. 可以询问平台功能、防伪验证等问题 2. 需要验证产品时，AI会主动提供扫描功能 3. 支持中文对话，表达自然即可 4. 使用 ↑↓ 键快速访问历史输入 """ self.stdout.write(help_text) def _print_readline_help(self): """打印readline快捷键帮助""" readline_help = """ ⌨️ Readline 快捷键帮助 ======================= 导航快捷键: ↑/↓ - 浏览历史记录 Ctrl+A - 跳转到行首 Ctrl+E - 跳转到行尾 Ctrl+B - 向后移动一个字符 Ctrl+F - 向前移动一个字符 Ctrl+P - 上一个历史记录 (同 ↑) Ctrl+N - 下一个历史记录 (同 ↓) 编辑快捷键: Ctrl+U - 删除从光标到行首的内容 Ctrl+K - 删除从光标到行尾的内容 Ctrl+W - 删除光标前的一个单词 Ctrl+D - 删除光标后的一个字符 Ctrl+H - 删除光标前的一个字符 (同 Backspace) 历史搜索: Ctrl+R - 反向搜索历史记录 Ctrl+S - 正向搜索历史记录 补全功能: Tab - 自动补全 双击Tab - 显示所有可能的补全 其他: Ctrl+L - 清屏 Ctrl+C - 中断当前操作 Ctrl+D - 退出 (EOF) 💡 提示: 这些快捷键基于 Emacs 编辑模式，与大多数终端应用保持一致 """ self.stdout.write(readline_help) def _print_readline_stats(self, readline_config): """打印readline统计信息""" stats = readline_config.get_history_stats() if 'error' in stats: self.stderr.write(f"❌ {stats['error']}") return self.stdout.write("\n📊 Readline 统计信息") self.stdout.write("=" * 30) self.stdout.write(f"当前历史记录数: {stats['current_length']}") self.stdout.write(f"最大历史记录数: {stats['max_length']}") self.stdout.write(f"历史文件路径: {stats['history_file']}") # 检查历史文件大小 try: if os.path.exists(stats['history_file']): file_size = os.path.getsize(stats['history_file']) self.stdout.write(f"历史文件大小: {file_size} 字节") else: self.stdout.write("历史文件: 尚未创建") except Exception: self.stdout.write("历史文件: 无法获取信息") self.stdout.write("=" * 30)