=== TASK 4 REVIEW PACKAGE ===
lines in extract: 240

def _kb_format_ms(ms: int) -> str:
    """将毫秒时长格式化为 mm:ss，供切片详情等展示使用。"""
    s = max(0, int(ms or 0)) // 1000
    return f"{s // 60:02d}:{s % 60:02d}"


def _kb_status_label(status: str) -> str:
    """映射知识库文档处理状态为本地化文案。"""
    mapping = {
        "ready": tr("KB Doc Status Ready"),
        "processing": tr("KB Doc Status Processing"),
        "uploading": tr("KB Doc Status Processing"),
        "failed": tr("KB Doc Status Failed"),
        "local_only": tr("KB Doc Status Local"),
    }
    return mapping.get(status or "", status or "?")


def _render_kb_documents_tab(api_base: str, files: list):
    """原始文档 Tab：上传、文件夹、按文件名过滤、列表与重试/删除/查看切片。"""
    import requests as _req

    name_filter = st.text_input(
        tr("KB Search Filename"),
        key="kb_doc_name_filter",
        placeholder=tr("KB Search Filename"),
    )

    with st.expander(tr("KB Upload Document"), expanded=False):
        with st.form("kb_upload_form"):
            uploaded = st.file_uploader(
                tr("Upload Videos"),
                type=["mp4", "mov", "avi", "flv", "mkv"],
                accept_multiple_files=True,
                key="kb_uploader",
            )
            submitted = st.form_submit_button(tr("Upload and Slice"))
        if submitted and uploaded:
            for f in uploaded:
                try:
                    r = _req.post(
                        f"{api_base}/videos",
                        files={"file": (f.name, f.getvalue(), "video/mp4")},
                        timeout=120,
                    )
                    if r.status_code == 200:
                        st.success(
                            f"{f.name} -> {r.json().get('data', {}).get('doc_id')}"
                        )
                    else:
                        st.error(f"{f.name}: {r.text}")
                except Exception as exc:
                    st.error(f"{f.name}: {exc}")
            st.rerun()

    with st.expander(tr("Upload Folder"), expanded=False):
        folder_path = st.text_input(
            tr("Local Folder Path"),
            key="kb_folder_path",
            placeholder="/path/to/videos",
            help=tr("Upload Folder Help"),
        )
        if st.button(tr("Upload Folder and Slice"), width="stretch"):
            if not folder_path.strip():
                st.error(tr("Please Enter Folder Path"))
            else:
                with st.spinner(tr("Uploading Folder")):
                    try:
                        r = _req.post(
                            f"{api_base}/videos/folder",
                            json={"path": folder_path.strip()},
                            timeout=600,
                        )
                        if r.status_code == 200:
                            data = r.json().get("data") or {}
                            st.success(
                                tr("Upload Folder Done").format(
                                    ok=data.get("ok", 0),
                                    total=data.get("total", 0),
                                    failed=data.get("failed", 0),
                                )
                            )
                            for item in data.get("results") or []:
                                if not item.get("ok"):
                                    st.warning(
                                        f"{item.get('filename')}: {item.get('error')}"
                                    )
                        else:
                            st.error(r.text)
                    except Exception as exc:
                        st.error(str(exc))
                st.rerun()

    filtered = files
    needle = (name_filter or "").strip().lower()
    if needle:
        filtered = [
            item
            for item in files
            if needle in str(item.get("filename") or "").lower()
        ]

    st.subheader(tr("Knowledge Base Videos"))
    if not filtered:
        st.info(tr("No Knowledge Base Videos"))
        return

    for item in filtered:
        doc_id = item.get("doc_id", "")
        slice_count = len(item.get("slices") or []) or item.get("point_num") or 0
        remote_tag = "remote" if item.get("remote") else "local"
        status_text = _kb_status_label(item.get("status", ""))
        label = (
            f"{item.get('filename', '?')} — {status_text} "
            f"({tr('KB Slice Count')}: {slice_count}, {remote_tag})"
        )
        with st.expander(label):
            st.caption(f"doc_id={doc_id}")
            if item.get("title"):
                st.caption(item["title"])
            if item.get("error"):
                st.error(item["error"])
            a1, a2, a3 = st.columns(3)
            if a1.button(tr("KB View Slices"), key=f"kb_view_{doc_id}"):
                st.session_state["kb_slice_filter_doc"] = doc_id
                st.session_state["kb_force_slices"] = True
                st.rerun()
            if a2.button(tr("Retry"), key=f"kb_retry_{doc_id}"):
                try:
                    _req.post(f"{api_base}/videos/{doc_id}/retry", timeout=30)
                except Exception as exc:
                    st.error(str(exc))
                st.rerun()
            if a3.button(tr("Delete"), key=f"kb_del_{doc_id}"):
                try:
                    _req.delete(f"{api_base}/videos/{doc_id}", timeout=30)
                except Exception as exc:
                    st.error(str(exc))
                st.rerun()


def _render_kb_slices_tab(files: list):
    """切片详情 Tab 占位；Task 5 将实现完整卡片/筛选/分页。"""
    _ = files
    st.caption(tr("KB Tab Slices"))


def _render_kb_search_tab(api_base: str, files: list):
    """知识检索 Tab 占位；Task 5 将对接 POST /search。"""
    _ = (api_base, files)
    st.caption(tr("KB Tab Search"))


def _render_volcano_kb_view():
    """火山方舟知识库控制台：Header + Tabs（原始文档 / 切片 / 检索 / 问答）。"""
    import requests as _req
    import time as _time

    if not volcano_kb_service.is_enabled():
        st.warning(tr("Volcano KB Not Configured"))
        return

    api_base = f"http://127.0.0.1:{config.listen_port}/api/v1/volcano_kb"
    collection = config.app.get("volcengine_kb_collection", "") or ""
    project = config.app.get("volcengine_kb_project", "") or ""

    h1, h2 = st.columns([3, 1])
    with h1:
        st.subheader(collection or tr("Volcano Knowledge Base"))
        st.caption(
            f"{tr('KB Official Version')} · {tr('KB Build Success')} · {tr('KB Data Type')}"
        )
        st.caption(f"project={project} · collection={collection}")
        if st.session_state.get("kb_last_sync_at"):
            st.caption(
                tr("KB Last Sync").format(time=st.session_state["kb_last_sync_at"])
            )
    with h2:
        if st.button(tr("Sync Knowledge Base"), type="primary", width="stretch"):
            with st.spinner(tr("Syncing Knowledge Base")):
                try:
                    r = _req.post(f"{api_base}/videos/sync", timeout=120)
                    if r.status_code == 200:
                        data = r.json().get("data") or {}
                        st.success(
                            tr("Sync Knowledge Base Done").format(
                                docs=data.get("doc_count", 0),
                                slices=data.get("slice_count", 0),
                            )
                        )
                        st.session_state["kb_last_sync_at"] = _time.strftime(
                            "%Y-%m-%d %H:%M:%S"
                        )
                    else:
                        st.error(r.text)
                except Exception as exc:
                    st.error(str(exc))
            st.rerun()
        if st.button(tr("Refresh List"), width="stretch"):
            st.rerun()

    # 使用稳定 tab id，避免语言切换后 radio 状态与文案错位。
    tab_ids = ["documents", "slices", "search", "qa"]
    tab_labels = {
        "documents": tr("KB Tab Documents"),
        "slices": tr("KB Tab Slices"),
        "search": tr("KB Tab Search"),
        "qa": tr("KB Tab QA"),
    }
    if st.session_state.get("kb_console_tab") not in tab_ids:
        st.session_state["kb_console_tab"] = "documents"
    if st.session_state.pop("kb_force_slices", False):
        st.session_state["kb_console_tab"] = "slices"

    choice = st.radio(
        "kb_console_tabs",
        tab_ids,
        horizontal=True,
        label_visibility="collapsed",
        key="kb_console_tab",
        format_func=lambda tab_id: tab_labels.get(tab_id, tab_id),
    )

    try:
        r = _req.get(f"{api_base}/videos", timeout=30)
        files = r.json().get("data", {}).get("files", []) if r.status_code == 200 else []
    except Exception as exc:
        st.error(str(exc))
        files = []

    if choice == "documents":
        _render_kb_documents_tab(api_base, files)
    elif choice == "slices":
        _render_kb_slices_tab(files)
    elif choice == "search":
        _render_kb_search_tab(api_base, files)
    else:
        st.info(tr("KB QA Coming Soon"))




=== test/webui/test_volcano_kb_view.py ===
import json
import sys
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent.parent))


class TestVolcanoKBWebUI(unittest.TestCase):
    def _main_py_text(self):
        return (Path(__file__).parent.parent.parent / "webui" / "Main.py").read_text(
            encoding="utf-8"
        )

    def test_video_sources_includes_volcano_kb(self):
        text = self._main_py_text()
        self.assertIn('"volcano_kb"', text)
        self.assertIn('tr("Volcano Knowledge Base")', text)

    def test_validation_accepts_volcano_kb(self):
        text = self._main_py_text()
        self.assertIn(
            '"pexels", "pixabay", "coverr", "local", "volcano_kb"', text
        )

    def test_kb_view_function_defined(self):
        text = self._main_py_text()
        self.assertIn("def _render_volcano_kb_view():", text)
        self.assertIn("def _render_main_nav()", text)
        self.assertIn("volcano_kb_service.is_enabled()", text)
        self.assertIn('key="main_nav"', text)
        self.assertIn('initial_sidebar_state="collapsed"', text)
        self.assertNotIn("st.sidebar.container(key=\"main_nav\")", text)
        self.assertNotIn("st.sidebar.radio(", text)

    def test_zh_json_has_keys(self):
        data = json.loads(
            (Path(__file__).parent.parent.parent / "webui" / "i18n" / "zh.json").read_text(
                encoding="utf-8"
            )
        )
        table = data.get("Translation", data)
        for key in [
            "Volcano Knowledge Base",
            "Upload Videos",
            "Upload and Slice",
            "Uploaded Videos",
            "Volcano KB Not Configured",
            "View",
            "Generate",
            "Retry",
            "Delete",
        ]:
            self.assertIn(key, table)

    def test_en_json_has_keys(self):
        data = json.loads(
            (Path(__file__).parent.parent.parent / "webui" / "i18n" / "en.json").read_text(
                encoding="utf-8"
            )
        )
        table = data.get("Translation", data)
        for key in [
            "Volcano Knowledge Base",
            "Volcano KB Not Configured",
            "Upload and Slice",
        ]:
            self.assertIn(key, table)


if __name__ == "__main__":
    unittest.main()
