rag/wwwroot/knowledge_bases_list/batch_ingest.dspy

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 rag_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]
    krecs = await sor.sqlExe("SELECT embedding_engine FROM rag_knowledge_bases WHERE id=${kb_id}$", {"kb_id": kb_id})
    emb_engine = (getattr(krecs[0], 'embedding_engine', '') or 'clip-vith14').strip() if krecs else 'clip-vith14'
if emb_engine == 'bge-m3':
    emb_url = 'https://embedding.opencomputing.net:10443/txte/api/embed'
    emb_model = 'bge-m3'
else:
    emb_url = 'https://embedding.opencomputing.net:10443/mme/api/embed'
    emb_model = 'CLIP-ViT-H-14'

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 rag_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(emb_url,
                json={"texts": chunks, "model": emb_model})
            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 rag_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 rag_documents SET status='done',chunk_count=${n}$ WHERE id=${id}$", {"n": len(chunks), "id": fid})
            await sor.sqlExe("UPDATE rag_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)