rag/wwwroot/knowledge_bases_list/media_cards.dspy

import json

ns = params_kw.copy()
kb_id = ns.get('kb_id', '')
kind = ns.get('kind', 'face')

db = DBPools()
dbname = get_module_dbname('rag')

rows = []
try:
    async with db.sqlorContext(dbname) as sor:
        if kind == 'voice':
            recs = await sor.sqlExe(
                "SELECT id, file_name, file_path, metadata, created_at FROM documents "
                "WHERE kb_id=${kb_id}$ AND (LOWER(file_name) LIKE '%%.mp3' OR LOWER(file_name) LIKE '%%.wav' "
                "OR LOWER(file_name) LIKE '%%.m4a' OR LOWER(file_name) LIKE '%%.aac' OR LOWER(file_name) LIKE '%%.ogg' "
                "OR LOWER(file_name) LIKE '%%.flac') ORDER BY created_at DESC",
                {"kb_id": kb_id})
        else:
            recs = await sor.sqlExe(
                "SELECT id, file_name, file_path, metadata, created_at FROM documents "
                "WHERE kb_id=${kb_id}$ AND (LOWER(file_name) LIKE '%%.jpg' OR LOWER(file_name) LIKE '%%.jpeg' "
                "OR LOWER(file_name) LIKE '%%.png' OR LOWER(file_name) LIKE '%%.gif' OR LOWER(file_name) LIKE '%%.webp' "
                "OR LOWER(file_name) LIKE '%%.bmp') ORDER BY created_at DESC",
                {"kb_id": kb_id})
        for r in recs:
            rows.append({
                "id": r.id,
                "file_name": r.file_name or '',
                "file_path": r.file_path or '',
                "metadata": r.metadata or '{}',
                "created_at": str(r.created_at)[:16] if r.created_at else ''
            })

        # 查出入介质标签（从标签表中取出来），按 doc_id 编组
        doc_ids = [r["id"] for r in rows]
        doc_tags = {}
        if doc_ids:
            mt_recs = await sor.sqlExe(
                "SELECT mt.media_id, t.name, t.color FROM media_tags mt "
                "JOIN tags t ON mt.tag_id=t.id "
                "WHERE mt.media_type='document' AND mt.media_id IN (${ids}$)",
                {"ids": doc_ids})
            for mt in mt_recs:
                doc_tags.setdefault(mt.media_id, []).append({"name": mt.name, "color": mt.color or '#3b82f6'})
        for r in rows:
            r["media_tags"] = doc_tags.get(r["id"], [])
except Exception:
    return {"widgettype": "Text", "options": {"text": "加载失败", "cfontsize": 14, "color": "#e74c3c", "halign": "center", "marginTop": "40px"}}

def safe_url(p):
    if not p:
        return ''
    if p.startswith('/idfile'):
        return p
    return '/idfile' + p

kind_cn = '人脸' if kind != 'voice' else '声纹'

cards = []
for f in rows:
    doc_id = f["id"]
    fname = f["file_name"]
    media_url = safe_url(f["file_path"])
    tags = f.get("media_tags", [])

    if kind == 'voice':
        media_widget = {"widgettype": "Html", "options": {"html": "<audio controls preload=\"metadata\" style=\"width:100%\" src=\"" + media_url + "\"></audio>", "padding": "8px 12px"}}
    else:
        media_widget = {"widgettype": "Image", "options": {"url": media_url, "width": "100%", "cheight": 12, "objectFit": "contain", "bgcolor": "#f5f5f5"}}

    if tags:
        tag_widgets = []
        for t in tags:
            tag_widgets.append({"widgettype": "Text", "options": {"text": "#" + str(t["name"]), "cfontsize": 10,
                               "color": "#1565c0", "bgcolor": "#e3f2fd",
                               "padding": "1px 8px", "borderRadius": "8px", "margin": "2px 4px 2px 0"}})
        tag_row = {"widgettype": "HBox", "options": {"wrap": True, "padding": "0 12px", "alignItems": "center"}, "subwidgets": tag_widgets}
    else:
        tag_row = {"widgettype": "Text", "options": {"text": "暂无标签", "cfontsize": 10, "color": "#bbb", "padding": "0 12px"}}

    tag_url = "./tag_form.dspy?doc_id=" + doc_id + "&kb_id=" + kb_id + "&kind=" + kind

    cards.append({
        "widgettype": "VBox",
        "options": {"css": "card", "cwidth": 20, "padding": "0", "spacing": "0", "overflow": "hidden"},
        "subwidgets": [
            media_widget,
            {"widgettype": "Text", "options": {"text": fname, "cfontsize": 12, "fontWeight": "bold", "color": "#333", "padding": "8px 12px 2px 12px", "wrap": True, "halign": "left"}},
            {"widgettype": "Text", "options": {"text": f["created_at"], "cfontsize": 10, "color": "#999", "padding": "0 12px", "halign": "left"}},
            tag_row,
            {"widgettype": "HBox", "options": {"padding": "8px 12px", "alignItems": "center"}, "subwidgets": [
                {"widgettype": "Text", "options": {"text": "", "css": "filler"}},
                {"widgettype": "Button", "options": {"label": "🏷 添加标签", "cfontsize": 11, "bgcolor": "#e3f2fd", "color": "#1565c0", "padding": "3px 10px", "borderRadius": "12px"},
                 "binds": [{"wid": "self", "event": "click", "actiontype": "urlwidget", "target": "PopupWindow",
                            "popup_options": {"title": "为「" + fname + "」添加标签", "width": "34%", "height": "auto",
                                              "movable": True, "resizable": True, "modal": True},
                            "options": {"url": tag_url}}]}
            ]}
        ]
    })

if not cards:
    cards.append({"widgettype": "Text", "options": {"text": "该知识库暂无" + kind_cn + "文件，请先在文件列表上传对应图片/音频", "cfontsize": 13, "color": "#aaa", "halign": "center", "marginTop": "60px"}})

result = {
    "widgettype": "VScrollPanel",
    "id": "media_card_panel",
    "options": {"css": "filler", "width": "100%", "height": "100%", "padding": "12px"},
    "subwidgets": [
        {
            "widgettype": "DynamicColumn",
            "options": {"width": "100%", "col_cwidth": 22, "col_cgap": 1},
            "subwidgets": cards
        }
    ]
}
return result