import base64, os, subprocess
ns = params_kw.copy()
kb_id = ns.get('kb_id', '')
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()
file_size = len(file_data)

# ---- ORG STORAGE QUOTA CHECK (per-org limit, not global) ----
def fmt_bytes(n):
    if n < 1024: return str(n) + 'B'
    if n < 1048576: return str(round(n/1024, 1)) + 'KB'
    return str(round(n/1048576, 1)) + 'MB'

quota_limit = 104857600
used = 0
async with get_sor_context(env, 'rag') as sor:
    rec = await sor.sqlExe("SELECT COALESCE(SUM(file_size),0) AS used FROM rag_documents WHERE org_id=${org_id}$", {"org_id": userorgid})
    if rec: used = int(rec[0].used)
    lim = await sor.sqlExe("SELECT limit_bytes FROM rag_org_storage_limits WHERE org_id=${org_id}$", {"org_id": userorgid})
    if lim: quota_limit = int(lim[0].limit_bytes)
if used + file_size > quota_limit:
    return json.dumps({"status": "error", "error": "storage_quota_exceeded",
        "message": "存储配额超限：机构已用 " + fmt_bytes(used) + "，限额 " + fmt_bytes(quota_limit) + "，本文件 " + fmt_bytes(file_size)}, ensure_ascii=False)

# Save file via FileStorage (returns web path e.g. /idfile/191/193/197/97/xxx.txt)
web_path = await env.save_file(file_data, file_name)
real_path = env.realpath(web_path)

doc_id = uuid()
ext = '.' + file_name.rsplit('.', 1)[1] if '.' in file_name else '.bin'
ext_l = ext.lower()

# ── 后台入库：共享模块 rag.ingest（UI 上传与对外 API 上传同一实现）──
# background_reco 派发 asyncio 任务；ingest_one 只接原始类型参数、自建连接。
from rag.ingest import ingest_one
# SYNC PART — record document as 'pending', update KB counts,
# fire background ingestion, return immediately.
# ============================================================
async with get_sor_context(env, 'rag') as sor:
    await sor.sqlExe(
        "INSERT INTO rag_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', 'pending', 0, '{}', ${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": web_path, "org_id": userorgid})
    await sor.sqlExe(
        "UPDATE rag_knowledge_bases SET doc_count=doc_count+1, total_size=total_size+${size}$ WHERE id=${kb_id}$",
        {"size": file_size, "kb_id": kb_id})

# Fire background ingestion — pass primitives only (no env/request/proxy objects)
background_reco(ingest_one, doc_id, kb_id, file_name, ext_l, real_path)

result = {"status": "SUCCEEDED", "doc_id": doc_id, "file_name": file_name,
    "file_size": file_size, "folder_id": folder_id, "ingest": "pending"}
return json.dumps(result, ensure_ascii=False, default=str)