feat: auto-mapped CIS report embedding, section embed, ADB RAW(16) fix
- Auto-embed report: maps each CSV to its ADB table (summaryreportcsvvector, identityandaccess, networking, etc.)
- Section embed: new endpoint POST /api/embeddings/report/{rid}/section with per-section button in UI
- Table validation: checks registered ADB tables before embedding, reports missing tables
- Auto-detect embedding dimension: reads table dim and selects correct model (3072→large, 1536→small)
- ADB RAW(16) ID fix: all vector inserts use HEXTORAW() for UUID (fixes ORA-01465)
- Float32 vectors: all inserts use array.array('f') for FLOAT32 compatibility
- Embedding status: real-time progress polling (Embedding X/Y — table: Z)
- Loading per section: spinner only on the section being embedded, not all
- Removed individual file embed buttons (only section + full report)
- Summary CSV chunking: groups by CIS section with tenancy + extract_date metadata
- Findings CSV chunking: each row becomes a document with structured content
- README: documented all 11 required ADB vector tables with descriptions
This commit is contained in:
nogueiraguh
409
backend/app.py
409
backend/app.py
@@ -41,6 +41,7 @@ for d in [DATA, OCI_DIR, REPORTS, MCP_DIR, WALLET_DIR]:
|
||||
|
||||
_running_reports: dict[str, asyncio.subprocess.Process] = {} # rid → subprocess
|
||||
_running_terraform: dict[str, asyncio.subprocess.Process] = {} # wid → subprocess
|
||||
_embedding_status: dict[str, dict] = {} # task_id → {status, message, table, tenancy, inserted, total}
|
||||
TERRAFORM_DIR = DATA / "terraform"
|
||||
TERRAFORM_DIR.mkdir(parents=True, exist_ok=True)
|
||||
_chat_executor = concurrent.futures.ThreadPoolExecutor(max_workers=10, thread_name_prefix="chat")
|
||||
@@ -3479,24 +3480,29 @@ def _auto_register_table(adb_config_id: str, table_name: str, description: str =
|
||||
|
||||
def _ingest_documents_task(cfg: dict, genai_cfg: dict, documents: list, user_id: str, username: str,
|
||||
table_name: str = None, tenancy: str = None, compartments: str = None,
|
||||
report_date: str = None):
|
||||
report_date: str = None, task_id: str = None):
|
||||
"""Background task: embed and insert documents into ADB via OCI GenAI.
|
||||
Tenancy and compartments are stored in METADATA as structured JSON for filtering."""
|
||||
import array
|
||||
emb_model = cfg.get("embedding_model_id", "cohere.embed-v4.0")
|
||||
table_name = table_name or cfg.get("table_name", "")
|
||||
total = len(documents)
|
||||
# Track status
|
||||
if task_id:
|
||||
_embedding_status[task_id] = {"status": "running", "table": table_name, "tenancy": tenancy or "",
|
||||
"inserted": 0, "total": total, "message": "Iniciando embedding..."}
|
||||
# Auto-register table so it appears in multi-table RAG search
|
||||
_auto_register_table(cfg["id"], table_name)
|
||||
conn = _get_adb_connection(cfg)
|
||||
try:
|
||||
cur = conn.cursor()
|
||||
inserted = 0
|
||||
for doc in documents:
|
||||
for i, doc in enumerate(documents):
|
||||
try:
|
||||
content = doc.get("content", "")
|
||||
if not content: continue
|
||||
embedding = _embed_text(content, genai_cfg, emb_model)
|
||||
vec = array.array('d', embedding)
|
||||
vec = array.array('f', [float(x) for x in embedding])
|
||||
# Build structured metadata with tenancy isolation
|
||||
doc_tenancy = tenancy or doc.get("tenancy", "")
|
||||
doc_compartments = compartments or doc.get("compartments", "")
|
||||
@@ -3509,22 +3515,27 @@ def _ingest_documents_task(cfg: dict, genai_cfg: dict, documents: list, user_id:
|
||||
)
|
||||
cur.execute(f"""
|
||||
INSERT INTO "{table_name}" (ID, TEXT, EMBEDDING, METADATA)
|
||||
VALUES (:1, :2, :3, :4)
|
||||
""", [str(uuid.uuid4()), content, vec, metadata])
|
||||
VALUES (HEXTORAW(:1), :2, :3, :4)
|
||||
""", [uuid.uuid4().hex.upper(), content, vec, metadata])
|
||||
inserted += 1
|
||||
if task_id:
|
||||
_embedding_status[task_id].update({"inserted": inserted, "message": f"Embedding {inserted}/{total}..."})
|
||||
except Exception as e:
|
||||
log.error(f"Failed to ingest document: {e}")
|
||||
conn.commit()
|
||||
cur.close()
|
||||
log.info(f"Ingested {inserted}/{len(documents)} documents into {table_name}" +
|
||||
msg = f"{inserted}/{total} documentos ingeridos em {table_name}" + (f" (tenancy: {tenancy})" if tenancy else "")
|
||||
log.info(f"Ingested {inserted}/{total} documents into {table_name}" +
|
||||
(f" (tenancy={tenancy})" if tenancy else ""))
|
||||
_audit(user_id, username, "ingest_documents", cfg["id"], f"{inserted} documents")
|
||||
_config_log("adb", cfg["id"], cfg.get("config_name"), "success", "ingest",
|
||||
f"{inserted}/{len(documents)} documentos ingeridos em {table_name}" +
|
||||
(f" (tenancy: {tenancy})" if tenancy else ""), user_id, username)
|
||||
_config_log("adb", cfg["id"], cfg.get("config_name"), "success", "ingest", msg, user_id, username)
|
||||
if task_id:
|
||||
_embedding_status[task_id].update({"status": "done", "inserted": inserted, "message": msg})
|
||||
except Exception as e:
|
||||
log.error(f"Ingestion task failed: {e}")
|
||||
_config_log("adb", cfg["id"], cfg.get("config_name"), "error", "ingest", str(e)[:500], user_id, username)
|
||||
if task_id:
|
||||
_embedding_status[task_id].update({"status": "error", "message": str(e)[:300]})
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
@@ -3627,40 +3638,357 @@ async def preview_report_chunks(rid: str, u=Depends(current_user)):
|
||||
"total_chunks": len(documents),
|
||||
"chunks": documents}
|
||||
|
||||
def _chunk_summary_csv(csv_path: str, tenancy: str, extract_date: str) -> list:
|
||||
"""Chunk the cis_summary_report.csv into documents with tenancy/date metadata.
|
||||
Each row becomes a document with structured content for vector search."""
|
||||
import csv as csvmod
|
||||
p = Path(csv_path)
|
||||
if not p.exists():
|
||||
return []
|
||||
documents = []
|
||||
with open(p, "r", encoding="utf-8") as f:
|
||||
rows = list(csvmod.DictReader(f))
|
||||
if not rows:
|
||||
return []
|
||||
# Group rows by section for richer context
|
||||
sections: dict = {}
|
||||
for row in rows:
|
||||
sec = row.get("Section", "Unknown")
|
||||
sections.setdefault(sec, []).append(row)
|
||||
for sec_name, sec_rows in sections.items():
|
||||
lines = []
|
||||
for r in sec_rows:
|
||||
rec = r.get("Recommendation #", "")
|
||||
title = r.get("Title", "")
|
||||
compliant = r.get("Compliant", "")
|
||||
pct = r.get("Compliance Percentage Per Recommendation", "")
|
||||
findings = r.get("Findings", "0")
|
||||
total = r.get("Total", "0")
|
||||
lines.append(f"Recommendation {rec}: {title} | Status: {compliant} | Compliance: {pct}% | Findings: {findings}/{total}")
|
||||
content = (
|
||||
f"Tenancy: {tenancy}\n"
|
||||
f"Extract Date: {extract_date}\n"
|
||||
f"Section: {sec_name}\n"
|
||||
f"Total Recommendations: {len(sec_rows)}\n\n"
|
||||
+ "\n".join(lines)
|
||||
)
|
||||
documents.append({
|
||||
"content": content,
|
||||
"section": sec_name,
|
||||
"tenancy": tenancy,
|
||||
"metadata": json.dumps({"tenancy": tenancy, "extract_date": extract_date, "section": sec_name})
|
||||
})
|
||||
return documents
|
||||
|
||||
|
||||
# ── CIS Report CSV → ADB Table Mapping ──
|
||||
_CIS_TABLE_MAP = {
|
||||
"Identity_and_Access_Management": "identityandaccess",
|
||||
"Networking": "networking",
|
||||
"Compute": "computeinstances",
|
||||
"Logging_and_Monitoring": "loggingandmonitoring",
|
||||
"Storage_Object_Storage": "objectstorage",
|
||||
"Storage_Block_Volumes": "storageblockvolume",
|
||||
"Storage_File_Storage_Service": "filestorageservice",
|
||||
"Asset_Management": "assetmanagement",
|
||||
}
|
||||
|
||||
def _resolve_table_for_csv(filename: str) -> str | None:
|
||||
"""Map a CIS report CSV filename to its ADB vector table."""
|
||||
if filename == "cis_summary_report.csv":
|
||||
return "summaryreportcsvvector"
|
||||
for pattern, table in _CIS_TABLE_MAP.items():
|
||||
if pattern in filename:
|
||||
return table
|
||||
return None
|
||||
|
||||
|
||||
def _chunk_findings_csv(csv_path: str, tenancy: str, extract_date: str) -> list:
|
||||
"""Chunk a CIS findings CSV into documents. Each row becomes a document with structured content."""
|
||||
import csv as csvmod
|
||||
p = Path(csv_path)
|
||||
if not p.exists():
|
||||
return []
|
||||
documents = []
|
||||
with open(p, "r", encoding="utf-8") as f:
|
||||
rows = list(csvmod.DictReader(f))
|
||||
if not rows:
|
||||
return []
|
||||
skip_cols = {"extract_date", "deep_link", "domain_deeplink", "defined_tags",
|
||||
"freeform_tags", "system_tags", "external_identifier"}
|
||||
for row in rows:
|
||||
parts = []
|
||||
parts.append(f"Tenancy: {tenancy}")
|
||||
parts.append(f"Extract Date: {extract_date}")
|
||||
for col, val in row.items():
|
||||
if col.lower() in skip_cols or not val or not val.strip():
|
||||
continue
|
||||
# Clean HYPERLINK formulas
|
||||
if val.startswith("=HYPERLINK"):
|
||||
import re
|
||||
m = re.search(r',\s*"([^"]+)"', val)
|
||||
val = m.group(1) if m else val
|
||||
parts.append(f"{col}: {val}")
|
||||
content = "\n".join(parts)
|
||||
if len(content) > 50:
|
||||
documents.append({
|
||||
"content": content,
|
||||
"tenancy": tenancy,
|
||||
"metadata": json.dumps({"tenancy": tenancy, "extract_date": extract_date, "source": p.name})
|
||||
})
|
||||
return documents
|
||||
|
||||
|
||||
@app.post("/api/embeddings/report/{rid}")
|
||||
async def embed_report(rid: str, req: dict, bg: BackgroundTasks, u=Depends(require("admin","user"))):
|
||||
"""Auto-embed all CIS report CSVs into their mapped ADB vector tables."""
|
||||
vid = req.get("adb_config_id")
|
||||
if not vid: raise HTTPException(400, "adb_config_id is required")
|
||||
with db() as c:
|
||||
r = c.execute("SELECT json_path,tenancy_name,config_id FROM reports WHERE id=? AND status='completed'", (rid,)).fetchone()
|
||||
r = c.execute("SELECT tenancy_name, config_id FROM reports WHERE id=? AND status='completed'", (rid,)).fetchone()
|
||||
if not r: raise HTTPException(404, "Report not found or not completed")
|
||||
json_path = r["json_path"]
|
||||
if not json_path or not Path(json_path).exists():
|
||||
raise HTTPException(400, "Report JSON file not found")
|
||||
try:
|
||||
report_data = json.loads(Path(json_path).read_text())
|
||||
except Exception:
|
||||
raise HTTPException(400, "Invalid report data")
|
||||
documents = _chunk_report_by_section(report_data)
|
||||
if not documents: raise HTTPException(400, "No sections found in report")
|
||||
# Optional section filter — embed only a specific section
|
||||
section_filter = req.get("section")
|
||||
if section_filter:
|
||||
documents = [d for d in documents if d.get("section") == section_filter]
|
||||
if not documents: raise HTTPException(400, f"Section '{section_filter}' not found in report")
|
||||
cfg, gc = _get_adb_and_genai(vid, oci_config_id=r.get("config_id"))
|
||||
target_table = req.get("table_name") or None
|
||||
# Extract tenancy and compartments for isolation
|
||||
tenancy = report_data.get("tenancy", r["tenancy_name"] or "unknown")
|
||||
report_date = report_data.get("generated_at", "")
|
||||
compartments_list = report_data.get("compartments", [])
|
||||
compartments_str = json.dumps(compartments_list) if compartments_list else ""
|
||||
bg.add_task(_ingest_documents_task, cfg, gc, documents, u["id"], u["username"],
|
||||
table_name=target_table, tenancy=tenancy, compartments=compartments_str,
|
||||
report_date=report_date)
|
||||
label = f"section={section_filter}" if section_filter else f"{len(documents)} sections"
|
||||
_audit(u["id"], u["username"], "embed_report", rid, f"{label}, tenancy={tenancy}")
|
||||
return {"ok": True, "message": f"Embedding de {len(documents)} seção(ões) iniciado (tenancy: {tenancy})", "sections": len(documents), "tenancy": tenancy}
|
||||
|
||||
rdir = REPORTS / rid
|
||||
tenancy = r["tenancy_name"] or "unknown"
|
||||
|
||||
# Read extract_date from summary CSV
|
||||
import csv as csvmod
|
||||
summary_csv = rdir / "cis_summary_report.csv"
|
||||
extract_date = ""
|
||||
if summary_csv.exists():
|
||||
with open(summary_csv, "r", encoding="utf-8") as f:
|
||||
first = next(csvmod.DictReader(f), {})
|
||||
extract_date = first.get("extract_date", "")
|
||||
|
||||
cfg, gc = _get_adb_and_genai(vid, oci_config_id=r["config_id"] if r["config_id"] else None)
|
||||
|
||||
# Load registered tables for validation
|
||||
with db() as c:
|
||||
registered = {t["table_name"].lower() for t in
|
||||
c.execute("SELECT table_name FROM adb_vector_tables WHERE adb_config_id=? AND is_active=1", (vid,)).fetchall()}
|
||||
|
||||
# Scan all CSVs and map to tables
|
||||
task_id = str(uuid.uuid4())
|
||||
table_docs: dict[str, list] = {}
|
||||
missing_tables: list[str] = []
|
||||
skipped_tables: list[str] = []
|
||||
|
||||
for csv_file in sorted(rdir.glob("cis_*.csv")):
|
||||
table = _resolve_table_for_csv(csv_file.name)
|
||||
if not table:
|
||||
continue
|
||||
if table.lower() not in registered:
|
||||
if table not in skipped_tables:
|
||||
skipped_tables.append(table)
|
||||
continue
|
||||
if csv_file.name == "cis_summary_report.csv":
|
||||
docs = _chunk_summary_csv(str(csv_file), tenancy, extract_date)
|
||||
else:
|
||||
docs = _chunk_findings_csv(str(csv_file), tenancy, extract_date)
|
||||
if docs:
|
||||
table_docs.setdefault(table, []).extend(docs)
|
||||
|
||||
if not table_docs:
|
||||
if skipped_tables:
|
||||
raise HTTPException(400, f"Tabela(s) não registrada(s) no ADB: {', '.join(skipped_tables)}. Registre em Configurações > ADB Vector.")
|
||||
raise HTTPException(400, "No CSV files found to embed")
|
||||
|
||||
# Calculate totals for status tracking
|
||||
total_docs = sum(len(d) for d in table_docs.values())
|
||||
tables_used = list(table_docs.keys())
|
||||
_embedding_status[task_id] = {
|
||||
"status": "running", "table": ", ".join(tables_used), "tenancy": tenancy,
|
||||
"inserted": 0, "total": total_docs,
|
||||
"message": f"Embedding {total_docs} documentos em {len(tables_used)} tabelas..."
|
||||
}
|
||||
|
||||
def _bg_embed_all():
|
||||
"""Background: embed documents into their respective tables sequentially."""
|
||||
import array
|
||||
default_model = cfg.get("embedding_model_id", "cohere.embed-v4.0")
|
||||
inserted_total = 0
|
||||
errors = []
|
||||
for tbl, docs in table_docs.items():
|
||||
_auto_register_table(cfg["id"], tbl)
|
||||
# Auto-detect embedding model based on table dimension
|
||||
emb_model = default_model
|
||||
try:
|
||||
actual_dim = _get_table_embedding_dim(cfg, tbl)
|
||||
if actual_dim and actual_dim in _DIM_TO_MODEL:
|
||||
emb_model = _DIM_TO_MODEL[actual_dim]
|
||||
log.info(f"Table {tbl}: dim={actual_dim}, using model {emb_model}")
|
||||
except Exception as e:
|
||||
log.warning(f"Could not detect dim for {tbl}: {e}")
|
||||
try:
|
||||
conn = _get_adb_connection(cfg)
|
||||
cur = conn.cursor()
|
||||
for doc in docs:
|
||||
try:
|
||||
content = doc.get("content", "")
|
||||
if not content: continue
|
||||
embedding = _embed_text(content, gc, emb_model)
|
||||
vec = array.array('f', [float(x) for x in embedding])
|
||||
metadata = _build_metadata_json(
|
||||
tenancy=doc.get("tenancy", tenancy),
|
||||
section=doc.get("section", ""),
|
||||
report_date=extract_date,
|
||||
)
|
||||
cur.execute(f'INSERT INTO "{tbl}" (ID, TEXT, EMBEDDING, METADATA) VALUES (HEXTORAW(:1), :2, :3, :4)',
|
||||
[uuid.uuid4().hex.upper(), content, vec, metadata])
|
||||
inserted_total += 1
|
||||
_embedding_status[task_id].update({
|
||||
"inserted": inserted_total,
|
||||
"message": f"Embedding {inserted_total}/{total_docs} — tabela: {tbl}"
|
||||
})
|
||||
except Exception as e:
|
||||
log.error(f"Failed to embed doc in {tbl}: {e}")
|
||||
conn.commit()
|
||||
cur.close()
|
||||
conn.close()
|
||||
log.info(f"Embedded {len(docs)} docs into {tbl} (tenancy={tenancy})")
|
||||
except Exception as e:
|
||||
log.error(f"Failed to connect/embed to {tbl}: {e}")
|
||||
errors.append(f"{tbl}: {str(e)[:100]}")
|
||||
|
||||
msg = f"{inserted_total}/{total_docs} documentos em {len(tables_used)} tabelas ({', '.join(tables_used)})"
|
||||
if tenancy: msg += f" — tenancy: {tenancy}"
|
||||
if errors: msg += f" | Erros: {'; '.join(errors)}"
|
||||
_embedding_status[task_id].update({
|
||||
"status": "done" if not errors else "done",
|
||||
"inserted": inserted_total, "message": msg
|
||||
})
|
||||
_audit(u["id"], u["username"], "embed_report_auto", rid, msg)
|
||||
_config_log("adb", cfg["id"], cfg.get("config_name"),
|
||||
"success" if not errors else "error", "ingest", msg, u["id"], u["username"])
|
||||
|
||||
_chat_executor.submit(_bg_embed_all)
|
||||
msg = f"Embedding iniciado — {total_docs} documentos em {len(tables_used)} tabelas ({', '.join(tables_used)})"
|
||||
if skipped_tables:
|
||||
msg += f". Ignoradas (não registradas): {', '.join(skipped_tables)}"
|
||||
return {
|
||||
"ok": True, "task_id": task_id, "message": msg,
|
||||
"tables": tables_used, "skipped": skipped_tables, "total_documents": total_docs, "tenancy": tenancy
|
||||
}
|
||||
|
||||
|
||||
@app.get("/api/embeddings/status/{task_id}")
|
||||
async def embedding_status(task_id: str, u=Depends(current_user)):
|
||||
"""Check embedding task progress."""
|
||||
st = _embedding_status.get(task_id)
|
||||
if not st:
|
||||
return {"status": "unknown", "message": "Task not found"}
|
||||
return st
|
||||
|
||||
|
||||
@app.post("/api/embeddings/report/{rid}/section")
|
||||
async def embed_report_section(rid: str, req: dict, bg: BackgroundTasks, u=Depends(require("admin","user"))):
|
||||
"""Embed all CSV files from a specific report section into the mapped ADB table."""
|
||||
import csv as csvmod
|
||||
vid = req.get("adb_config_id")
|
||||
file_names: list = req.get("file_names", [])
|
||||
if not vid:
|
||||
# Auto-detect first active ADB
|
||||
with db() as c:
|
||||
adb = c.execute("SELECT id FROM adb_vector_configs WHERE is_active=1 LIMIT 1").fetchone()
|
||||
if not adb: raise HTTPException(400, "No active ADB config found")
|
||||
vid = adb["id"]
|
||||
if not file_names: raise HTTPException(400, "file_names is required")
|
||||
|
||||
with db() as c:
|
||||
r = c.execute("SELECT tenancy_name, config_id FROM reports WHERE id=? AND status='completed'", (rid,)).fetchone()
|
||||
if not r: raise HTTPException(404)
|
||||
rdir = REPORTS / rid
|
||||
tenancy = r["tenancy_name"] or "unknown"
|
||||
|
||||
# Read extract_date
|
||||
summary = rdir / "cis_summary_report.csv"
|
||||
extract_date = ""
|
||||
if summary.exists():
|
||||
with open(summary, "r", encoding="utf-8") as f:
|
||||
first = next(csvmod.DictReader(f), {})
|
||||
extract_date = first.get("extract_date", "")
|
||||
|
||||
cfg, gc = _get_adb_and_genai(vid, oci_config_id=r["config_id"] if r["config_id"] else None)
|
||||
|
||||
# Load registered tables for validation
|
||||
with db() as c:
|
||||
registered = {t["table_name"].lower() for t in
|
||||
c.execute("SELECT table_name FROM adb_vector_tables WHERE adb_config_id=? AND is_active=1", (vid,)).fetchall()}
|
||||
|
||||
# Group files by target table and validate
|
||||
import array
|
||||
table_docs: dict[str, list] = {}
|
||||
missing_tables: list[str] = []
|
||||
for fname in file_names:
|
||||
csv_path = rdir / fname
|
||||
if not csv_path.exists(): continue
|
||||
table = _resolve_table_for_csv(fname)
|
||||
if not table: continue
|
||||
if table.lower() not in registered:
|
||||
if table not in missing_tables:
|
||||
missing_tables.append(table)
|
||||
continue
|
||||
if fname == "cis_summary_report.csv":
|
||||
docs = _chunk_summary_csv(str(csv_path), tenancy, extract_date)
|
||||
else:
|
||||
docs = _chunk_findings_csv(str(csv_path), tenancy, extract_date)
|
||||
if docs:
|
||||
table_docs.setdefault(table, []).extend(docs)
|
||||
|
||||
if missing_tables and not table_docs:
|
||||
raise HTTPException(400, f"Tabela(s) não registrada(s) no ADB: {', '.join(missing_tables)}. Registre em Configurações > ADB Vector.")
|
||||
if not table_docs:
|
||||
raise HTTPException(400, "No embeddable content found in the selected files")
|
||||
|
||||
total_docs = sum(len(d) for d in table_docs.values())
|
||||
tables_used = list(table_docs.keys())
|
||||
task_id = str(uuid.uuid4())
|
||||
_embedding_status[task_id] = {
|
||||
"status": "running", "table": ", ".join(tables_used), "tenancy": tenancy,
|
||||
"inserted": 0, "total": total_docs, "message": f"Embedding {total_docs} docs em {', '.join(tables_used)}..."
|
||||
}
|
||||
|
||||
def _bg():
|
||||
default_model = cfg.get("embedding_model_id", "cohere.embed-v4.0")
|
||||
inserted = 0
|
||||
for tbl, docs in table_docs.items():
|
||||
_auto_register_table(cfg["id"], tbl)
|
||||
# Auto-detect embedding model based on table dimension
|
||||
emb_model = default_model
|
||||
try:
|
||||
actual_dim = _get_table_embedding_dim(cfg, tbl)
|
||||
if actual_dim and actual_dim in _DIM_TO_MODEL:
|
||||
emb_model = _DIM_TO_MODEL[actual_dim]
|
||||
log.info(f"Table {tbl}: dim={actual_dim}, using model {emb_model}")
|
||||
except Exception as e:
|
||||
log.warning(f"Could not detect dim for {tbl}: {e}")
|
||||
try:
|
||||
conn = _get_adb_connection(cfg)
|
||||
cur = conn.cursor()
|
||||
for doc in docs:
|
||||
try:
|
||||
content = doc.get("content", "")
|
||||
if not content: continue
|
||||
embedding = _embed_text(content, gc, emb_model)
|
||||
vec = array.array('f', [float(x) for x in embedding])
|
||||
metadata = _build_metadata_json(tenancy=doc.get("tenancy", tenancy), section=doc.get("section", ""), report_date=extract_date)
|
||||
cur.execute(f'INSERT INTO "{tbl}" (ID, TEXT, EMBEDDING, METADATA) VALUES (HEXTORAW(:1), :2, :3, :4)',
|
||||
[uuid.uuid4().hex.upper(), content, vec, metadata])
|
||||
inserted += 1
|
||||
_embedding_status[task_id].update({"inserted": inserted, "message": f"Embedding {inserted}/{total_docs} — {tbl}"})
|
||||
except Exception as e:
|
||||
log.error(f"Section embed error in {tbl}: {e}")
|
||||
conn.commit(); cur.close(); conn.close()
|
||||
except Exception as e:
|
||||
log.error(f"Section embed connection error {tbl}: {e}")
|
||||
msg = f"{inserted}/{total_docs} docs em {', '.join(tables_used)} (tenancy: {tenancy})"
|
||||
_embedding_status[task_id].update({"status": "done", "inserted": inserted, "message": msg})
|
||||
_audit(u["id"], u["username"], "embed_section", rid, msg)
|
||||
|
||||
_chat_executor.submit(_bg)
|
||||
return {"ok": True, "task_id": task_id, "tables": tables_used, "total": total_docs,
|
||||
"message": f"Embedding de {total_docs} docs iniciado ({', '.join(tables_used)})"}
|
||||
|
||||
|
||||
@app.post("/api/embeddings/report/{rid}/file/{fid}")
|
||||
async def embed_report_file(rid: str, fid: str, req: dict, bg: BackgroundTasks, u=Depends(require("admin","user"))):
|
||||
@@ -3686,13 +4014,14 @@ async def embed_report_file(rid: str, fid: str, req: dict, bg: BackgroundTasks,
|
||||
if not content.strip(): raise HTTPException(400, "File is empty")
|
||||
documents = _chunk_text_file(content, f["file_name"])
|
||||
if not documents: raise HTTPException(400, "No content chunks found")
|
||||
cfg, gc = _get_adb_and_genai(vid, oci_config_id=r.get("config_id"))
|
||||
cfg, gc = _get_adb_and_genai(vid, oci_config_id=r["config_id"] if r["config_id"] else None)
|
||||
target_table = req.get("table_name") or None
|
||||
tenancy = r["tenancy_name"] or "unknown"
|
||||
task_id = str(uuid.uuid4())
|
||||
bg.add_task(_ingest_documents_task, cfg, gc, documents, u["id"], u["username"],
|
||||
table_name=target_table, tenancy=tenancy)
|
||||
table_name=target_table, tenancy=tenancy, task_id=task_id)
|
||||
_audit(u["id"], u["username"], "embed_report_file", f"{rid}/{fid}", f"{f['file_name']}, {len(documents)} chunks, tenancy={tenancy}")
|
||||
return {"ok": True, "message": f"Embedding de {f['file_name']} iniciado ({len(documents)} chunks)", "chunks": len(documents)}
|
||||
return {"ok": True, "task_id": task_id, "message": f"Embedding de {f['file_name']} iniciado ({len(documents)} chunks)", "chunks": len(documents)}
|
||||
|
||||
def _extract_pdf_text(file_bytes: bytes) -> str:
|
||||
"""Extract text from a PDF file using PyPDF2 or pdfplumber."""
|
||||
|
||||
Reference in New Issue
Block a user