fix: use FileStorage for upload/delete — hashed paths /idfile/191/193/197/97/xxx.txt
This commit is contained in:
parent
c14588f33b
commit
3f3743842e
@ -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:
|
||||
|
||||
@ -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,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user