207 lines
9.6 KiB
Plaintext
207 lines
9.6 KiB
Plaintext
ns = params_kw.copy()
|
|
kb_id = ns.get('kb_id', '41c9f1cd45e0')
|
|
env = request._run_ns
|
|
|
|
# Get unprocessed documents
|
|
rows = []
|
|
async with get_sor_context(env, 'rag') as sor:
|
|
recs = await sor.sqlExe(
|
|
"SELECT id, file_name, file_size, file_path FROM documents WHERE kb_id=${kb_id}$ AND (chunk_count IS NULL OR chunk_count=0 OR status='pending') LIMIT 20",
|
|
{"kb_id": kb_id})
|
|
rows = [dict(r) for r in recs]
|
|
|
|
if not rows:
|
|
return json.dumps({"widgettype": "Text", "options": {"text": "No unprocessed documents found"}}, ensure_ascii=False)
|
|
|
|
import aiohttp, io, os
|
|
|
|
FS = '/d/rag/ragserver/files'
|
|
LOCAL = '/d/rag/ragserver/pkgs/rag/rag/files'
|
|
results = []
|
|
|
|
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=60)) as session:
|
|
for r in rows:
|
|
fid = r['id']
|
|
fn = r['file_name']
|
|
fp = r['file_path'] or ''
|
|
|
|
# Resolve file
|
|
fpath = None
|
|
if fp.startswith('/idfile/'):
|
|
fname = fp.replace('/idfile/', '')
|
|
if os.path.exists(os.path.join(LOCAL, fname)):
|
|
fpath = os.path.join(LOCAL, fname)
|
|
else:
|
|
full = os.path.join(FS, fp.lstrip('/'))
|
|
if os.path.exists(full):
|
|
fpath = full
|
|
|
|
if not fpath:
|
|
results.append(f"✗ {fn}: not found")
|
|
continue
|
|
|
|
ext = os.path.splitext(fpath)[1].lower()
|
|
text_exts = {'.txt','.md','.csv','.json','.xml','.html','.htm','.py','.js','.css','.yaml','.yml','.log','.rst'}
|
|
with open(fpath, 'rb') as f:
|
|
data = f.read()
|
|
|
|
text = ''
|
|
if ext in text_exts:
|
|
text = data.decode('utf-8', errors='replace')
|
|
elif ext == '.pdf':
|
|
try:
|
|
from PyPDF2 import PdfReader; reader = PdfReader(io.BytesIO(data))
|
|
text = '\n'.join(p.extract_text() or '' for p in reader.pages)
|
|
except: text = ''
|
|
elif ext == '.docx':
|
|
try:
|
|
from docx import Document; doc = Document(io.BytesIO(data))
|
|
text = '\n'.join(p.text for p in doc.paragraphs if p.text.strip())
|
|
except: text = ''
|
|
elif ext == '.pptx':
|
|
try:
|
|
from pptx import Presentation; prs = Presentation(io.BytesIO(data))
|
|
parts = [sh.text for s in prs.slides for sh in s.shapes if hasattr(sh,'text') and sh.text.strip()]
|
|
text = '\n'.join(parts)
|
|
except: text = ''
|
|
elif ext in {'.xlsx','.xls'}:
|
|
try:
|
|
from openpyxl import load_workbook
|
|
wb = load_workbook(io.BytesIO(data), read_only=True, data_only=True)
|
|
rows2 = [' | '.join(str(c) if c else '' for c in row) for ws in wb.worksheets for row in ws.iter_rows(values_only=True) if any(c for c in row)]
|
|
text = '\n'.join(rows2)
|
|
except: text = ''
|
|
|
|
if not text or len(text.strip()) < 10:
|
|
ext_l = os.path.splitext(fpath)[1].lower()
|
|
image_exts = {'.jpg','.jpeg','.png','.bmp','.gif','.webp'}
|
|
audio_exts = {'.mp3','.wav','.flac','.ogg','.m4a','.aac'}
|
|
video_exts = {'.mp4','.avi','.mov','.mkv','.webm'}
|
|
face_count = 0
|
|
speakers = 0
|
|
import base64, subprocess
|
|
|
|
# IMAGE: face detection
|
|
if ext_l in image_exts:
|
|
try:
|
|
img_b64 = base64.b64encode(data).decode()
|
|
r_img = await session.post('https://media.opencomputing.net:10443/face/api/detect',
|
|
json={"images": [img_b64]})
|
|
if r_img.status == 200:
|
|
fd = await r_img.json()
|
|
res = fd.get("results", [])
|
|
if res and isinstance(res[0], dict):
|
|
face_count = len(res[0].get("faces", res[0].get("detections", [])))
|
|
except: pass
|
|
|
|
# VIDEO: frame → face + audio → voiceprint
|
|
if ext_l in video_exts:
|
|
try:
|
|
tmp_img = '/tmp/' + fid + '_frame.jpg'
|
|
subprocess.run(['ffmpeg','-y','-i',fpath,'-vframes','1','-q:v','2',tmp_img],
|
|
capture_output=True, timeout=30)
|
|
if os.path.exists(tmp_img):
|
|
with open(tmp_img, 'rb') as fi:
|
|
fd2 = fi.read()
|
|
img_b64 = base64.b64encode(fd2).decode()
|
|
r_vid = await session.post('https://media.opencomputing.net:10443/face/api/detect',
|
|
json={"images": [img_b64]})
|
|
if r_vid.status == 200:
|
|
fdd = await r_vid.json()
|
|
res = fdd.get("results", [])
|
|
if res and isinstance(res[0], dict):
|
|
face_count = len(res[0].get("faces", res[0].get("detections", [])))
|
|
os.remove(tmp_img)
|
|
except: pass
|
|
try:
|
|
tmp_audio = '/tmp/' + fid + '_audio.wav'
|
|
subprocess.run(['ffmpeg','-y','-i',fpath,'-vn','-acodec','pcm_s16le','-ar','16000','-ac','1',tmp_audio],
|
|
capture_output=True, timeout=60)
|
|
if os.path.exists(tmp_audio) and os.path.getsize(tmp_audio) > 44:
|
|
with open(tmp_audio, 'rb') as fa:
|
|
aud = fa.read()
|
|
form = aiohttp.FormData()
|
|
form.add_field('file', aud, filename='audio.wav')
|
|
r_au = await session.post('https://media.opencomputing.net:10443/voiceprint/extract/submit',
|
|
data=form)
|
|
if r_au.status == 200:
|
|
vd = await r_au.json()
|
|
if vd.get('status') == 'SUCCEEDED' or vd.get('embedding'):
|
|
speakers = vd.get('speakers', 1)
|
|
os.remove(tmp_audio)
|
|
except: pass
|
|
|
|
# AUDIO: voiceprint
|
|
if ext_l in audio_exts:
|
|
try:
|
|
form = aiohttp.FormData()
|
|
form.add_field('file', data, filename=fn)
|
|
r_au = await session.post('https://media.opencomputing.net:10443/voiceprint/extract/submit',
|
|
data=form)
|
|
if r_au.status == 200:
|
|
vd = await r_au.json()
|
|
if vd.get('status') == 'SUCCEEDED' or vd.get('embedding'):
|
|
speakers = vd.get('speakers', 1)
|
|
except: pass
|
|
|
|
import json as j2
|
|
meta = j2.dumps({"face": face_count, "speakers": speakers}, ensure_ascii=False)
|
|
async with get_sor_context(env, 'rag') as sor:
|
|
await sor.sqlExe(
|
|
"UPDATE documents SET status='done',chunk_count=0,metadata=${meta}$ WHERE id=${id}$",
|
|
{"meta": meta, "id": fid})
|
|
results.append(f"🎬 {fn}: faces={face_count} speakers={speakers}")
|
|
continue
|
|
|
|
# Chunk
|
|
paragraphs = text.split('\n')
|
|
chunks, cur = [], ''
|
|
for p in paragraphs:
|
|
p = p.strip()
|
|
if not p:
|
|
if cur: chunks.append(cur); cur = ''
|
|
continue
|
|
if len(cur) + len(p) < 500: cur = (cur+'\n'+p).strip()
|
|
else:
|
|
if cur: chunks.append(cur)
|
|
cur = p
|
|
if cur: chunks.append(cur)
|
|
|
|
if not chunks:
|
|
results.append(f"~ {fn}: empty chunks")
|
|
continue
|
|
|
|
# Embed
|
|
try:
|
|
r2 = await session.post('https://embedding.opencomputing.net:10443/api/embed',
|
|
json={"texts": chunks, "model": "CLIP-ViT-H-14"})
|
|
emb = await r2.json() if r2.status == 200 else {}
|
|
embeddings = emb.get("text_embeddings", emb.get("embeddings", []))
|
|
except:
|
|
embeddings = []
|
|
|
|
# VDB
|
|
if embeddings:
|
|
try:
|
|
await session.post('https://vectordb.opencomputing.net:10443/v1/createcollection',
|
|
json={"colname": kb_id, "fields": [{"name": "id", "type": "str", "is_primary": True, "max_length": 64}, {"name": "vector", "type": "fvector", "dim": 1024}, {"name": "text", "type": "str", "max_length": 65535}], "description": "RAG kb", "metric": "COSINE"})
|
|
await session.post('https://vectordb.opencomputing.net:10443/v1/upsert',
|
|
json={"colname": kb_id, "data": [{"id": fid + "_c" + str(i), "vector": e, "text": chunks[i]} for i, e in enumerate(embeddings)]})
|
|
except: pass
|
|
|
|
# DB
|
|
async with get_sor_context(env, 'rag') as sor:
|
|
for i, ct in enumerate(chunks):
|
|
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())",
|
|
{"id": f"{fid}_c{i}", "doc_id": fid, "kb_id": kb_id, "idx": i, "content": ct[:2000], "vid": ""})
|
|
await sor.sqlExe("UPDATE documents SET status='done',chunk_count=${n}$ WHERE id=${id}$", {"n": len(chunks), "id": fid})
|
|
await sor.sqlExe("UPDATE knowledge_bases SET chunk_count=chunk_count+${n}$ WHERE id=${kb_id}$", {"n": len(chunks), "kb_id": kb_id})
|
|
|
|
results.append(f"✓ {fn}: {len(text)} chars → {len(chunks)} chunks")
|
|
|
|
return json.dumps({
|
|
"widgettype": "VBox", "options": {"padding": "20px"},
|
|
"subwidgets": [{"widgettype": "Text", "options": {"text": r, "cfontsize": 13, "color": "#333"}} for r in results]
|
|
}, ensure_ascii=False)
|