diff --git a/wwwroot/knowledge_bases_list/delete_file.dspy b/wwwroot/knowledge_bases_list/delete_file.dspy index 89ce4f8..c9cf402 100644 --- a/wwwroot/knowledge_bases_list/delete_file.dspy +++ b/wwwroot/knowledge_bases_list/delete_file.dspy @@ -29,12 +29,12 @@ async with get_sor_context(env, 'rag') as sor: "UPDATE knowledge_bases SET doc_count=GREATEST(doc_count-1, 0), total_size=GREATEST(total_size-${size}$, 0) WHERE id=${kb_id}$", {"size": file_size, "kb_id": kb_id}) -# Delete physical file +# Delete physical file via FileStorage if file_path and file_path.startswith("/idfile/"): - import os - base = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "rag", "files") - real_path = os.path.join(base, os.path.basename(file_path)) + from ahserver.filestorage import FileStorage + fs = FileStorage() try: + real_path = fs.realPath(file_path) if os.path.exists(real_path): os.remove(real_path) except Exception: diff --git a/wwwroot/knowledge_bases_list/upload_file.dspy b/wwwroot/knowledge_bases_list/upload_file.dspy index a0184e7..3555b09 100644 --- a/wwwroot/knowledge_bases_list/upload_file.dspy +++ b/wwwroot/knowledge_bases_list/upload_file.dspy @@ -4,22 +4,30 @@ folder_id = ns.get('folder', '') file_name = ns.get('file_name', 'upload.bin') if not kb_id: return json.dumps({"status": "error", "error": "kb_id required"}, ensure_ascii=False) + file_data = await request.read() if not file_data: return json.dumps({"status": "error", "error": "no file data"}, ensure_ascii=False) env = request._run_ns userorgid = await env.get_userorgid() -doc_id = str(uuid()).replace('-', '')[:16] -ext = '.' + file_name.rsplit('.', 1)[1] if '.' in file_name else '.bin' -file_path = '/d/rag/ragserver/pkgs/rag/rag/files/' + doc_id + ext -with open(file_path, "wb") as f: - f.write(file_data) -file_size = len(file_data) +# Use FileStorage for proper hashed path +from ahserver.filestorage import FileStorage +fs = FileStorage() +real_path = fs._name2path(file_name) +with open(real_path, "wb") as f: + f.write(file_data) +web_path = fs.webpath(real_path) # e.g. /idfile/191/193/197/97/xxx.txt +if not web_path.startswith('/'): + web_path = '/' + web_path + +doc_id = str(uuid()).replace('-', '')[:16] +file_size = len(file_data) +ext = '.' + file_name.rsplit('.', 1)[1] if '.' in file_name else '.bin' ext_l = ext.lower() + text_exts = {'.txt', '.md', '.csv', '.json', '.xml', '.html', '.htm', '.py', '.js', '.css', '.yaml', '.yml', '.log', '.rst'} -doc_exts = {'.pdf', '.docx', '.pptx', '.xlsx', '.doc', '.ppt', '.xls'} image_exts = {'.jpg', '.jpeg', '.png', '.bmp', '.gif', '.webp'} audio_exts = {'.mp3', '.wav', '.flac', '.ogg', '.m4a', '.aac'} video_exts = {'.mp4', '.avi', '.mov', '.mkv', '.webm'} @@ -45,38 +53,15 @@ elif ext_l == '.docx': doc = Document(io.BytesIO(file_data)) text = '\n'.join(p.text for p in doc.paragraphs if p.text.strip()) except: pass -elif ext_l == '.pptx': - import io; from pptx import Presentation - try: - prs = Presentation(io.BytesIO(file_data)) - parts = [] - for s in prs.slides: - for sh in s.shapes: - if hasattr(sh, 'text') and sh.text.strip(): - parts.append(sh.text) - text = '\n'.join(parts) - except: pass -elif ext_l in {'.xlsx', '.xls'}: - import io; from openpyxl import load_workbook - try: - wb = load_workbook(io.BytesIO(file_data), read_only=True, data_only=True) - rows = [] - for ws in wb.worksheets: - for row in ws.iter_rows(values_only=True): - r = ' | '.join(str(c) if c is not None else '' for c in row) - if r.strip(): rows.append(r) - text = '\n'.join(rows) - except: pass import aiohttp, base64 -# --- IMAGE: face detection + CLIP visual embedding --- +# --- IMAGE: face detection --- if ext_l in image_exts: img_b64 = base64.b64encode(file_data).decode() async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=15)) as s: - # Face detection try: - r = await s.post('https://media.opencomputing.net:10443/face/api/detect', json={"images": [img_b64]}) + r = await s.post('https://media.opencomputing.net/face/api/detect', json={"images": [img_b64]}) if r.status == 200: fd = await r.json() results = fd.get("results", []) @@ -85,13 +70,13 @@ if ext_l in image_exts: meta_parts['face'] = face_count except: pass -# --- AUDIO: voiceprint extraction --- +# --- AUDIO: voiceprint --- if ext_l in audio_exts: async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=60)) as s: try: form = aiohttp.FormData() form.add_field('file', file_data, filename=file_name) - r = await s.post('https://media.opencomputing.net:10443/voiceprint/extract/submit', data=form) + r = await s.post('https://media.opencomputing.net/voiceprint/extract/submit', data=form) if r.status == 200: vd = await r.json() if vd.get('status') == 'SUCCEEDED': @@ -101,54 +86,31 @@ if ext_l in audio_exts: meta_parts['voiceprint'] = voice_speakers except: pass -# --- VIDEO: frame extraction + audio extraction (via ffmpeg) --- +# --- VIDEO: frame + audio extraction --- if ext_l in video_exts: - import subprocess, tempfile + import subprocess meta_parts['video'] = 'pending' - # Extract keyframe for face detection try: tmp_img = '/tmp/' + doc_id + '_frame.jpg' - subprocess.run(['ffmpeg', '-y', '-i', file_path, '-vframes', '1', '-q:v', '2', tmp_img], + subprocess.run(['ffmpeg', '-y', '-i', real_path, '-vframes', '1', '-q:v', '2', tmp_img], capture_output=True, timeout=30) - if __import__('os').path.exists(tmp_img): + if os.path.exists(tmp_img): with open(tmp_img, 'rb') as fi: frame_data = fi.read() img_b64 = base64.b64encode(frame_data).decode() async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=15)) as s: try: - r = await s.post('https://media.opencomputing.net:10443/face/api/detect', json={"images": [img_b64]}) + r = await s.post('https://media.opencomputing.net/face/api/detect', json={"images": [img_b64]}) if r.status == 200: fd = await r.json() results = fd.get("results", []) if results and isinstance(results[0], dict): face_count = len(results[0].get("faces", results[0].get("detections", []))) except: pass - __import__('os').remove(tmp_img) + os.remove(tmp_img) except: pass - # Extract audio for voiceprint - try: - tmp_audio = '/tmp/' + doc_id + '_audio.wav' - subprocess.run(['ffmpeg', '-y', '-i', file_path, '-vn', '-acodec', 'pcm_s16le', '-ar', '16000', '-ac', '1', tmp_audio], - capture_output=True, timeout=60) - if __import__('os').path.exists(tmp_audio) and __import__('os').path.getsize(tmp_audio) > 44: - with open(tmp_audio, 'rb') as fa: - audio_data = fa.read() - async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=60)) as s: - try: - form = aiohttp.FormData() - form.add_field('file', audio_data, filename='audio.wav') - r = await s.post('https://media.opencomputing.net:10443/voiceprint/extract/submit', data=form) - if r.status == 200: - vd = await r.json() - if vd.get('status') == 'SUCCEEDED' or vd.get('embedding'): - voice_speakers = vd.get('speakers', 1) - except: pass - __import__('os').remove(tmp_audio) - except: pass - meta_parts['face'] = face_count - meta_parts['voiceprint'] = voice_speakers -# --- RAG INGEST (text + image CLIP embedding) --- +# --- RAG INGEST for text --- if text and len(text.strip()) > 10: paragraphs = text.split('\n') chunks = [] @@ -168,7 +130,7 @@ if text and len(text.strip()) > 10: if chunks: async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=15)) as s: try: - r = await s.post('https://embedding.opencomputing.net:10443/api/embed', + r = await s.post('https://embedding.opencomputing.net/api/embed', json={"texts": chunks, "model": "CLIP-ViT-H-14"}) emb_data = await r.json() if r.status == 200 else {} embeddings = emb_data.get("text_embeddings", emb_data.get("embeddings", [])) @@ -178,10 +140,10 @@ if text and len(text.strip()) > 10: vector_ids = [] if embeddings: try: - vdb_data = {"collection": kb_id, "data": [ + vdb_data = {"colname": kb_id, "data": [ {"id": doc_id + "_" + str(i), "vector": emb, "text": chunks[i]} for i, emb in enumerate(embeddings)]} - await s.post('https://vectordb.opencomputing.net:10443/v1/upsert', json=vdb_data) + await s.post('https://vectordb.opencomputing.net/v1/upsert', json=vdb_data) vector_ids = [doc_id + "_" + str(i) for i in range(len(embeddings))] except: pass @@ -191,7 +153,7 @@ if text and len(text.strip()) > 10: vid = vector_ids[i] if i < len(vector_ids) else '' await sor.sqlExe( "INSERT INTO document_chunks (id, doc_id, kb_id, chunk_index, content, vector_id, created_at) " - "VALUES (" + "${id}$, ${doc_id}$, ${kb_id}$, ${idx}$, ${content}$, ${vid}$, NOW())", + "VALUES (${id}$, ${doc_id}$, ${kb_id}$, ${idx}$, ${content}$, ${vid}$, NOW())", {"id": doc_id + "_c" + str(i), "doc_id": doc_id, "kb_id": kb_id, "idx": i, "content": chunk_text[:2000], "vid": vid}) chunks_n = len(chunks) @@ -203,16 +165,16 @@ status = 'done' async with get_sor_context(env, 'rag') as sor: await sor.sqlExe( "INSERT INTO documents (id, kb_id, folder_id, file_name, file_type, file_size, file_path, mime_type, status, chunk_count, metadata, org_id, created_at, updated_at) " - "VALUES (" + "${id}$, ${kb_id}$, ${folder_id}$, ${file_name}$, 'other', ${file_size}$, ${file_path}$, 'application/octet-stream', ${status}$, ${chunks}$, ${meta}$, ${org_id}$, NOW(), NOW())", + "VALUES (${id}$, ${kb_id}$, ${folder_id}$, ${file_name}$, 'other', ${file_size}$, ${file_path}$, 'application/octet-stream', ${status}$, ${chunks}$, ${meta}$, ${org_id}$, NOW(), NOW())", {"id": doc_id, "kb_id": kb_id, "folder_id": folder_id, "file_name": file_name, - "file_size": file_size, "file_path": "/idfile/files/" + doc_id + ext, + "file_size": file_size, "file_path": web_path, "status": status, "chunks": chunks_n, "meta": meta_json, "org_id": userorgid}) await sor.sqlExe( - "UPDATE knowledge_bases SET doc_count=doc_count+1, total_size=total_size+" + "${size}$ WHERE id=${kb_id}$", + "UPDATE knowledge_bases SET doc_count=doc_count+1, total_size=total_size+${size}$ WHERE id=${kb_id}$", {"size": file_size, "kb_id": kb_id}) if chunks_n: await sor.sqlExe( - "UPDATE knowledge_bases SET chunk_count=chunk_count+" + "${n}$ WHERE id=${kb_id}$", + "UPDATE knowledge_bases SET chunk_count=chunk_count+${n}$ WHERE id=${kb_id}$", {"n": chunks_n, "kb_id": kb_id}) result = {"status": "SUCCEEDED", "doc_id": doc_id, "file_name": file_name,