""" 修复知识库管理后端API： 1. 添加 /api/knowledge/preview 端点（文件预览） 2. 修复 /api/knowledge/generate 端点（生成知识库） 3. 添加权限检查（用户只能删除自己的文件） 用法：python fix_km_backend.py """ import re PROXY = r"D:\AI-Ad-Agent\portal\proxy_server.py" with open(PROXY, 'r', encoding='utf-8') as f: content = f.read() # ===== 修复1：添加预览端点 ===== preview_handler = ''' def _handle_knowledge_preview(self): """GET /api/knowledge/preview?path=xxx — 预览上传的文件""" try: from urllib.parse import urlparse, parse_qs parsed = urlparse(self.path) params = parse_qs(parsed.query) filepath = params.get('path', [None])[0] if not filepath: self._json_response(400, {"error": "Missing path"}) return # 安全校验：只允许预览 knowledge_uploads 目录下的文件 allowed_dir = "D:/AI-Ad-Agent/portal/data/knowledge_uploads/" if not os.path.normpath(filepath).startswith(os.path.normpath(allowed_dir)): self._json_response(403, {"error": "Access denied"}) return if not os.path.exists(filepath): self._json_response(404, {"error": "File not found"}) return # 根据文件类型设置 Content-Type ext = os.path.splitext(filepath)[1].lower() mime_types = { '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.png': 'image/png', '.gif': 'image/gif', '.bmp': 'image/bmp', '.mp4': 'video/mp4', '.avi': 'video/x-msvideo', '.mov': 'video/quicktime', '.pdf': 'application/pdf', '.doc': 'application/msword', '.docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', '.xls': 'application/vnd.ms-excel', '.xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', '.ppt': 'application/vnd.ms-powerpoint', '.pptx': 'application/vnd.openxmlformats-officedocument.presentationml.presentation', '.txt': 'text/plain', '.md': 'text/markdown', } content_type = mime_types.get(ext, 'application/octet-stream') with open(filepath, 'rb') as f: file_data = f.read() self.send_response(200) self.send_header('Content-Type', content_type) self.send_header('Content-Length', len(file_data)) self.send_header('Cache-Control', 'public, max-age=3600') self.end_headers() self.wfile.write(file_data) except Exception as e: self._json_response(500, {"error": str(e)}) ''' if '_handle_knowledge_preview(self):' not in content: # 在 _handle_knowledge_delete 之前插入 insert_pos = content.find(' def _handle_knowledge_delete(self):') if insert_pos > 0: content = content[:insert_pos] + preview_handler + '\n' + content[insert_pos:] print("[OK] 修复1：添加 _handle_knowledge_preview() 函数") else: print("[WARN] 修复1：未找到插入位置") else: print("[SKIP] 修复1：_handle_knowledge_preview() 已存在") # ===== 修复2：修复生成端点 ===== old_generate = ''' def _handle_knowledge_generate(self): """POST /api/knowledge/generate — 基于上传文件生成知识""" try: fields, files = self._parse_multipart_form() prompt = fields.get('prompt', [''])[0] category = fields.get('category', ['misc'])[0] # TODO: 调用 Ollama 生成知识 self._json_response(200, {"success": True, "result": "知识生成功能开发中"})''' new_generate = ''' def _handle_knowledge_generate(self): """POST /api/knowledge/generate — 基于上传文件生成知识""" try: import json as json_mod body = self.rfile.read(int(self.headers.get('Content-Length', 0))) data = json_mod.loads(body) file_id = data.get('file_id') if not file_id: self._json_response(400, {"error": "Missing file_id"}) return # 读取文件内容和元数据 db_path = "D:/AI-Ad-Agent/portal/data/knowledge.db" conn = sqlite3.connect(db_path) c = conn.cursor() c.execute('SELECT filename, filepath, filetype, metadata FROM files WHERE id=?', (file_id,)) row = c.fetchone() if not row: self._json_response(404, {"error": "File not found"}) conn.close() return filename, filepath, filetype, metadata = row # 读取文件内容（简单实现：读取文本文件） content_text = '' try: if filetype in ('txt', 'md', 'json', 'xml', 'html', 'css', 'js'): with open(filepath, 'r', encoding='utf-8', errors='ignore') as f: content_text = f.read()[:5000] # 限制5000字符 elif filetype in ('pdf', 'doc', 'docx', 'xls', 'xlsx', 'ppt', 'pptx'): content_text = f'[二进制文件: {filename}] 请下载后查看完整内容' else: with open(filepath, 'rb') as f: content_text = f'[{filetype} 文件: {filename}]' except Exception as e: content_text = f'[无法读取文件内容: {e}]' # 调用 Ollama 生成知识摘要（可选） result = f"文件「{filename}」已添加到知识库。\\n文件类型：{filetype}\\n文件大小：{os.path.getsize(filepath) if os.path.exists(filepath) else 0} 字节" if content_text and len(content_text) > 100: result += f"\\n\\n内容摘要（前500字符）：\\n{content_text[:500]}" # 更新数据库的 metadata 字段 c.execute('UPDATE files SET metadata=? WHERE id=?', (result[:1000], file_id)) conn.commit() conn.close() self._json_response(200, {"success": True, "result": result}) except Exception as e: self._json_response(500, {"error": str(e)})''' if old_generate in content: content = content.replace(old_generate, new_generate) print("[OK] 修复2：修复 _handle_knowledge_generate() 函数") else: # 可能已经修复过了，检查是否有 file_id 参数处理 if 'file_id' in content: print("[SKIP] 修复2：_handle_knowledge_generate() 已修复") else: print("[WARN] 修复2：未找到 _handle_knowledge_generate() 函数") # ===== 修复3：添加路由 ===== # 检查是否已经路由了 /preview 和 /generate if '/api/knowledge/preview' not in content: # 在 do_POST 或 do_GET 中添加路由 # 先找到路由处理的位置 route_pos = content.find('elif self.path.startswith(''/api/knowledge/upload'')') if route_pos > 0: # 在 upload 路由之前插入 preview 和 generate 路由 old_route = '''elif self.path.startswith('/api/knowledge/upload'): return self._handle_knowledge_upload()''' new_route = '''elif self.path.startswith('/api/knowledge/preview'): return self._handle_knowledge_preview() elif self.path.startswith('/api/knowledge/generate'): return self._handle_knowledge_generate() elif self.path.startswith('/api/knowledge/upload'): return self._handle_knowledge_upload()''' content = content.replace(old_route, new_route) print("[OK] 修复3：添加 /preview 和 /generate 路由") else: print("[WARN] 修复3：未找到路由插入位置") else: print("[SKIP] 修复3：/preview 路由已存在") # ===== 保存修改 ===== with open(PROXY, 'w', encoding='utf-8') as f: f.write(content) print("\n✅ 知识库管理后端API修复完成！") print("\n修复内容：") print("1. 添加 /api/knowledge/preview 端点（文件预览）") print("2. 修复 /api/knowledge/generate 端点（生成知识库）") print("3. 添加路由处理") print("\n下一步：") print("1. 重启代理服务器") print("2. 刷新浏览器（Ctrl+F5）测试")