"""Streamlit 主应用入口（``core/app.py``）— 需求 15 / 17.1 / 17.2。 底座主框架：**仅负责加载、导航与分发**，不承载任何科学计算（计算在 Skill 的 ``compute`` 阶段，纯 Python）。运行方式（与设计一致）:: streamlit run core/app.py --server.port=7860 --server.address=0.0.0.0 此时 ``core/`` 目录位于 ``sys.path[0]``，故 ``kernel`` / ``services`` 以**顶层包** 方式导入。``skills`` 为仓库根下的顶层包，由 ``SkillRegistry.discover("skills")`` 自动发现（需求 1.1）。 职责装配（对应 design.md「Architecture / app.py」与需求 15）： 1. **构建 Services 容器**（:func:`build_services`）：实例化全部底座共享服务； ``LLMService`` 从环境变量读取密钥；任一可选重依赖缺失时优雅降级为 ``None``， 不导致应用崩溃（需求 6.5）。 2. **发现 Skill 并合并 i18n**（:func:`discover_skills` / :func:`merge_skill_i18n`）： 扫描 ``skills/`` 注册 Skill，并把每个 Skill 的 ``i18n`` 资源合并进 ``I18nService`` （需求 17.6）。 3. **统一布局**（:func:`main` 及各 ``render_*``）：侧边栏参数区（导航 + 各 Skill 的 ``render_inputs``）、中央主画布区（结果 / 报告）、可折叠审计追踪区（需求 15.1）。 4. **角色可见性**（:func:`visible_skills_for_role` / :func:`can_view_audit`）：按登录 用户角色显示可见 Skill 与审计区（接 ``AuthService``，需求 15.5）。管理员初始密码 取自环境变量 ``ADMIN_INIT_PASSWORD``（需求 5.1，明文不入代码 / 日志）。 5. **顶部语言切换**（:func:`set_session_language` / :func:`render_language_switcher`）： 写入 ``Services.lang`` 并触发重渲染；界面文案一律经 ``I18nService.t`` 取用 （需求 17.1 / 17.2 / 17.4）。 6. **分发**（:func:`dispatch`）：把选定 Skill + 收集到的 ``RawInput`` 交给固定 ``Pipeline`` 执行（extract→compute→explain→report），渲染返回的 ``ReportSections``。 可测试性取向：所有**装配逻辑**（构建服务、发现 + 合并 i18n、角色过滤、语言状态、 认证、分发）都放在**纯函数 / 可导入的辅助函数**中，与 ``st.*`` 渲染调用分离， 便于在无 Streamlit 的环境下用 pytest 直接测试（见 ``tests/test_app.py``）。 安全说明：本应用自身**不暴露任何免认证的网络端点**——它是 Streamlit 渲染的 UI； 角色隔离在应用层完成：审计区仅 ``admin`` / ``auditor`` 可见，分析 Skill 的可见性 由 ``AuthService.visible_skills`` 决定。管理员密码来自环境变量、加盐存储， 日志中不出现任何明文密码或密钥。 """ from __future__ import annotations import logging import os import re import sys from typing import Any, Callable, Optional # --------------------------------------------------------------------------- # 导入路径自举（关键）： # 运行 ``streamlit run core/app.py`` 时，Streamlit 仅把 ``core/`` 目录放入 # ``sys.path[0]``，使 ``kernel`` / ``services`` 能以顶层包导入；但顶层包 # ``skills`` / ``utils`` / ``layers`` 位于**仓库根目录**，默认不可导入，导致 # Skill 发现为空（导航「暂无数据」）与注册 / 登录报 ``No module named 'utils'``。 # 因此在模块导入期把「``core/`` 目录」与「仓库根目录」都补进 ``sys.path``， # 使两类顶层包均可被发现（与 tests/conftest.py 的路径设置保持一致）。 _CORE_DIR = os.path.dirname(os.path.abspath(__file__)) _REPO_ROOT = os.path.dirname(_CORE_DIR) for _p in (_CORE_DIR, _REPO_ROOT): if _p not in sys.path: sys.path.insert(0, _p) # 底座以**顶层包** kernel / services 导入（core/ 位于 sys.path[0]）。 from kernel.pipeline import Pipeline from kernel.registry import SkillRegistry from kernel.services import Services from kernel.skill_base import PharmaSkill, RawInput, ReportSections # Streamlit 为前端运行期依赖；导入失败（如纯逻辑单测环境）时降级为 None， # 使本模块的装配逻辑仍可被导入与测试。 try: # pragma: no cover - 取决于运行环境是否安装 streamlit import streamlit as st except Exception: # noqa: BLE001 - 无 streamlit 时仅禁用渲染，不影响逻辑导入 st = None # type: ignore[assignment] logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # 常量 # --------------------------------------------------------------------------- DEFAULT_LANG = "zh" #: 默认会话角色（未登录）：研发人员——可使用分析 Skill，但不可查看审计区（需求 15.5）。 DEFAULT_ROLE = "researcher" #: Streamlit 会话状态键。 SESSION_CTX = "pk_app_context" # (svc, registry) 缓存，跨重渲染复用 SESSION_LANG = "pk_lang" # 当前会话语言 SESSION_USER = "pk_user" # 当前登录用户名 SESSION_ROLE = "pk_role" # 当前会话角色 SESSION_SECTIONS = "pk_sections" # 最近一次分析返回的 ReportSections SESSION_SKILL = "pk_skill_id" # 当前选中的 Skill id SESSION_USER_MODEL = "pk_user_model" # 用户自配模型 (UserModelConfig) SESSION_TASK_SHEET = "pk_task_sheet" # 最近一次理解层产出的任务单 (dict) SESSION_SHEET_STATE = "pk_sheet_state" # "none" | "pending_confirm" | "confirmed" # =========================================================================== # 一、服务构建（需求 7 / 6.5：缺库优雅降级） # =========================================================================== def build_services( env: Optional[dict] = None, *, skill_ids: Optional[list] = None, audit_db_path: str = ":memory:", ) -> Services: """构建并返回底座 :class:`Services` 容器。 实例化全部底座共享服务；``LLMService`` 从环境变量读取密钥（无密钥则自然无可用 提供商，``complete`` 返回结构化错误而非静默）。**任一服务构建失败都被捕获**并把 对应字段降级为 ``None``（需求 6.5：可选重依赖缺失不得崩溃），失败信息收集到 ``svc._build_warnings`` 供 UI 提示。 参数： - ``env``：环境映射，默认 ``os.environ``。 - ``skill_ids``：已注册 Skill 的 id 列表，用于 ``AuthService.visible_skills`` 的角色可见性判定；省略时回退到 AuthService 的默认集合。 - ``audit_db_path``：审计库路径，默认 ``":memory:"``（按会话保存）。 """ env = os.environ if env is None else env warnings: list[tuple[str, str]] = [] def _try(name: str, factory: Callable[[], Any]) -> Any: try: return factory() except Exception as exc: # noqa: BLE001 - 单个服务失败即降级，不阻断整体 logger.warning("构建服务 %s 失败，已降级为不可用：%s", name, exc) warnings.append((name, str(exc))) return None i18n = _try("i18n", _build_i18n) usage = _try("usage", _build_usage) llm = _try("llm", lambda: _build_llm(env, usage)) audit = _try("audit", lambda: _build_audit(audit_db_path)) report = _try("report", _build_report) chart = _try("chart", _build_chart) file = _try("file", _build_file) prompt_guard = _try("prompt_guard", _build_prompt_guard) auth = _try("auth", lambda: _build_auth(skill_ids)) svc = Services( llm=llm, report=report, chart=chart, file=file, auth=auth, audit=audit, usage=usage, prompt_guard=prompt_guard, i18n=i18n, lang=(getattr(i18n, "lang", DEFAULT_LANG) if i18n is not None else DEFAULT_LANG), ) # 把构建告警挂在实例上供 UI 友好提示（Services 为普通 dataclass，可附加属性）。 setattr(svc, "_build_warnings", warnings) return svc # 各服务的惰性构建器：在函数内部 import，使任一服务的重依赖问题被 build_services # 的 try/except 隔离，而不在模块导入期就连带失败。 def _build_i18n(): from services.i18n_service import I18nService return I18nService() def _build_usage(): from services.usage_service import UsageService return UsageService() def _build_llm(env: dict, usage: Any): from services.llm_service import LLMService, SlidingWindowRateLimiter # 全局滑动窗口限流：默认 60 秒内最多 30 次 LLM 调用（需求 12.1）。 limiter = SlidingWindowRateLimiter(max_calls=30, window=60.0) return LLMService.from_env(env=env, usage=usage, rate_limiter=limiter) def _build_audit(db_path: str): from services.audit_service import AuditService return AuditService(db_path=db_path) def _build_report(): from services.report_service import ReportService return ReportService() def _build_chart(): from services.chart_service import ChartService return ChartService() def _build_file(): from services.file_service import FileService return FileService() def _build_prompt_guard(): from services.prompt_guard import PromptGuard return PromptGuard() def _build_auth(skill_ids: Optional[list]): from services.auth_service import AuthService if skill_ids: return AuthService(all_skill_ids=skill_ids) return AuthService() # =========================================================================== # 二、技能发现与 i18n 合并（需求 1.1 / 17.6） # =========================================================================== def discover_skills(skills_source: str = "skills", svc: Optional[Services] = None) -> SkillRegistry: """扫描 ``skills_source`` 注册全部 Skill，返回 :class:`SkillRegistry`。 单个 Skill 导入失败由注册器内部容错跳过（需求 1.3）；``svc`` 注入后失败会落审计。 """ registry = SkillRegistry(svc) registry.discover(skills_source) return registry def merge_skill_i18n(svc: Services, registry: SkillRegistry) -> int: """把每个已注册 Skill 的 ``i18n`` 资源合并进底座 ``I18nService``（需求 17.6）。 兼容两种 Skill i18n 约定，避免命名空间前缀重复： - 若 Skill 的 CATALOG 键**已**全部以 ``"."`` 前缀（如 ``general_qa.title``）， 直接合并、不再加前缀。 - 否则（如稳定性 / 相容性的 ``name`` / ``section.results``）以 ``namespace=`` 合并，键自动加 ``"."`` 前缀。 返回成功合并资源的 Skill 数量。``i18n`` 不可用时安全返回 0。 """ i18n = getattr(svc, "i18n", None) merge = getattr(i18n, "merge", None) if not callable(merge): return 0 merged = 0 for skill in registry.all(): catalog = getattr(skill, "i18n", None) if not catalog: continue skill_id = getattr(skill.meta, "id", "") or "" prefix = f"{skill_id}." try: keys = list(catalog.keys()) already_prefixed = bool(keys) and all(str(k).startswith(prefix) for k in keys) if already_prefixed or not skill_id: merge(catalog) else: merge(catalog, namespace=skill_id) merged += 1 except Exception as exc: # noqa: BLE001 - 单个 Skill 文案合并失败不阻断其它 logger.warning("合并 Skill %s 的 i18n 资源失败：%s", skill_id, exc) return merged def bootstrap( env: Optional[dict] = None, skills_source: str = "skills", *, audit_db_path: str = ":memory:", ) -> tuple[Services, SkillRegistry]: """一次性装配：发现 Skill → 构建 Services（含角色可见性所需的 skill_ids）→ 合并 i18n。 返回 ``(svc, registry)``。本函数为纯装配、无 ``st.*`` 依赖，便于单测。 """ env = os.environ if env is None else env registry = discover_skills(skills_source) skill_ids = [s.meta.id for s in registry.all()] svc = build_services(env, skill_ids=skill_ids, audit_db_path=audit_db_path) registry.svc = svc # 后续失败可落审计 merge_skill_i18n(svc, registry) return svc, registry # =========================================================================== # 三、角色可见性与导航（需求 15.5） # =========================================================================== def visible_skills_for_role( registry: SkillRegistry, auth: Any, role: Optional[str] ) -> list: """返回该角色在导航中可见的 Skill 列表（按导航顺序）。 可见 id 集合由 ``AuthService.visible_skills(role)`` 决定（需求 15.5）： admin / researcher 可见全部分析 Skill；auditor 只读、不展示可执行 Skill。 ``auth`` 不可用时回退为展示全部已注册 Skill（避免空界面）。 """ all_skills = registry.all() visible = getattr(auth, "visible_skills", None) if not callable(visible): return all_skills try: visible_ids = set(visible(role)) except Exception as exc: # noqa: BLE001 - 判定异常时回退全部 logger.warning("计算角色可见 Skill 失败，回退展示全部：%s", exc) return all_skills return [s for s in all_skills if s.meta.id in visible_ids] def can_view_audit(auth: Any, role: Optional[str]) -> bool: """该角色是否可查看审计追踪区（admin / auditor 可见，需求 15.5）。""" checker = getattr(auth, "can_view_audit", None) if not callable(checker): return False try: return bool(checker(role)) except Exception: # noqa: BLE001 return False def skill_nav_label(skill: PharmaSkill, i18n: Any = None) -> str: """构造 Skill 的导航标签（图标 + 本地化名称），文案经 i18n 取用（需求 17.4）。 优先取底座导航键 ``nav.``，其次取 Skill 自身的 ``.name``，再退回 ``meta.display_name``。``i18n`` 不可用时直接用 display_name。 """ meta = skill.meta icon = getattr(meta, "icon", "") or "" display = getattr(meta, "display_name", meta.id) has = getattr(i18n, "has", None) t = getattr(i18n, "t", None) if callable(has) and callable(t): for key in (f"nav.{meta.id}", f"{meta.id}.name"): try: if has(key): display = t(key) break except Exception: # noqa: BLE001 - 文案取用失败回退 display_name pass return f"{icon} {display}".strip() def skill_description(skill: PharmaSkill, i18n: Any = None, lang: Optional[str] = None) -> str: """返回 Skill 的本地化说明文字（随语言切换，需求 17.3 / 1 修复）。 优先取底座 ``desc.``，其次取 Skill 自身的 ``.description``， 再退回 ``meta.description``（硬编码中文兜底）。 """ meta = skill.meta fallback = getattr(meta, "description", "") or "" has = getattr(i18n, "has", None) t = getattr(i18n, "t", None) if callable(t): for key in (f"desc.{meta.id}", f"{meta.id}.description"): try: if not callable(has) or has(key): return t(key, lang) if lang else t(key) except Exception: # noqa: BLE001 - 取用失败回退 pass return fallback def skill_action_label(skill: PharmaSkill, svc: Services) -> str: """返回 Skill 主操作按钮文案：问答类用「开始提问」，其余用「开始分析」（需求 5 修复）。""" action_key = "btn.ask" if getattr(skill.meta, "id", "") == "general_qa" else "btn.analyze" return _t(svc, action_key) # =========================================================================== # 四、语言状态（需求 17.1 / 17.2） # =========================================================================== def set_session_language(svc: Services, requested: Optional[str]) -> str: """设置当前会话语言：写入 ``Services.lang`` 与 ``I18nService.lang``，返回归一语言码。 未知 / 空输入回退默认语言（不报错，需求 17.5）。这是「写入 Services.lang 并触发 重渲染」的纯逻辑部分；触发重渲染由 :func:`render_language_switcher` 在 UI 侧完成。 """ i18n = getattr(svc, "i18n", None) set_lang = getattr(i18n, "set_language", None) if callable(set_lang): lang = set_lang(requested) else: lang = (requested or DEFAULT_LANG) svc.lang = lang return lang # =========================================================================== # 五、认证（需求 5.1 / 5.4 / 15.5） # =========================================================================== def authenticate( username: str, password: str, *, user_lookup: Callable[[str], Optional[dict]], auth: Any = None, ) -> Optional[dict]: """校验用户名 / 密码，成功返回 ``{"user", "role"}``，失败返回 ``None``。 - ``user_lookup(email) -> record``：返回含 ``password_hash``（或 ``password``）与 ``role`` 的记录；返回 ``None`` 表示用户不存在。注入此参数使认证逻辑与具体 存储后端解耦，便于无数据库的单测。 - ``auth``：:class:`AuthService`，用于加盐哈希校验（``verify_password``，需求 5.4） 与角色归一（``normalize_role``）。省略时按需构造。 本函数不记录任何明文密码（需求 5.2）。 """ if not username or not password: return None if auth is None: from services.auth_service import AuthService auth = AuthService() try: record = user_lookup(username) except Exception as exc: # noqa: BLE001 - 存储查询异常视为登录失败（不回显细节） logger.warning("用户查询失败，登录被拒绝：%s", exc) return None if not record: return None stored = record.get("password_hash") or record.get("password") or "" if not auth.verify_password(password, stored): return None role = auth.normalize_role(record.get("role")) return {"user": username, "role": role} def ensure_admin_from_env(env: Optional[dict] = None) -> bool: """尽力按环境变量 ``ADMIN_INIT_PASSWORD`` 初始化管理员账户（需求 5.1）。 复用 ``utils.database``：``get_db_connection`` 在建表时会按环境变量创建管理员 （加盐哈希、明文不入库 / 日志）。数据库不可用时安全返回 ``False``，不抛出。 """ try: from utils import database as db except Exception as exc: # noqa: BLE001 - 无数据库模块时跳过 logger.info("用户数据库不可用，跳过管理员初始化：%s", exc) return False try: conn = db.get_db_connection() conn.close() return True except Exception as exc: # noqa: BLE001 - 初始化失败不阻断应用 logger.warning("管理员初始化失败（已忽略）：%s", exc) return False def _db_user_lookup(username: str) -> Optional[dict]: """基于 ``utils.database`` 的用户查询（供 UI 登录使用）；不可用时返回 ``None``。""" try: from utils import database as db return db.get_user_by_email(username) except Exception as exc: # noqa: BLE001 logger.warning("数据库用户查询不可用：%s", exc) return None # =========================================================================== # 六、模型配置解析（需求 5：用户 > 管理员 > 环境变量） # =========================================================================== def get_admin_model_config() -> Optional[dict]: """读取管理员级模型配置（来自 ``utils.database``）；不可用时返回 ``None``。""" try: from utils import database as db return db.get_default_llm_config() except Exception as exc: # noqa: BLE001 - 数据库不可用时忽略管理员档 logger.info("读取管理员模型配置失败（已忽略）：%s", exc) return None def resolve_session_llm(svc: Services, user_model: Any = None) -> str: """按 用户 > 管理员 > 环境变量 解析并重建 ``svc.llm``，返回首选档来源。 - ``user_model``：本会话用户自配（:class:`UserModelConfig`）或 ``None``。 - 解析出有序 Fallback 链后用 :func:`build_llm_service` 重建 ``svc.llm``， 保留底座的用量计量与限流（沿用原 ``svc.llm`` 的 usage / rate_limiter）。 返回来源标识：``"user"`` / ``"admin"`` / ``"env"`` / ``"none"``。 """ try: from services.model_config import build_llm_service, resolve_chain admin_cfg = get_admin_model_config() chain = resolve_chain(user_cfg=user_model, admin_cfg=admin_cfg, env=os.environ) # 复用既有 usage / rate_limiter，保持计量与限流连续。 old = getattr(svc, "llm", None) usage = getattr(old, "_usage", None) or getattr(svc, "usage", None) rate_limiter = getattr(old, "_rate_limiter", None) svc.llm = build_llm_service(chain, usage=usage, rate_limiter=rate_limiter) return chain.source except Exception as exc: # noqa: BLE001 - 解析失败不应阻断应用 logger.warning("解析会话模型配置失败（保留原 LLM）：%s", exc) return "env" # =========================================================================== # 七、分发（需求 2.2：固定流水线） # =========================================================================== def dispatch( skill: PharmaSkill, raw: RawInput, svc: Services, *, pipeline: Optional[Pipeline] = None, user: str = "anonymous", ) -> ReportSections: """把选定 Skill + ``RawInput`` 交给固定 ``Pipeline`` 执行并返回 ``ReportSections``。 底座流水线顺序固定（extract→compute→explain→report），``compute`` 不接收 ``svc`` （需求 2.1 由契约保证）。本函数另对「分析执行」事件做尽力而为的审计记录 （需求 7.4）。 """ pipeline = pipeline or Pipeline() sections = pipeline.run(skill, raw, svc) _audit_analysis(svc, user, getattr(skill.meta, "id", "")) return sections def _audit_analysis(svc: Services, user: str, skill_id: str) -> None: audit = getattr(svc, "audit", None) recorder = getattr(audit, "record_analysis", None) if callable(recorder): try: recorder(user or "anonymous", skill_id, step="execute") except Exception: # noqa: BLE001 - 审计失败不得影响主流程 logger.debug("记录分析执行审计失败（已忽略）。", exc_info=True) # =========================================================================== # 七·b、理解层两阶段编排（intent-understanding-layer） # =========================================================================== def build_orchestrator(registry: SkillRegistry, svc: Services): """构造 ``AnalysisOrchestrator``（理解→确认→分析两阶段编排）。 放在独立纯函数中便于单测；导入失败（理解层未就绪）时返回 ``None``，调用方据此 回退到旧的直接分发路径，保证向后兼容。 """ try: from kernel.orchestrator import AnalysisOrchestrator return AnalysisOrchestrator(registry, svc) except Exception as exc: # noqa: BLE001 - 理解层不可用时回退 logger.info("理解层编排器不可用，回退直接分发：%s", exc) return None def stale_session_keys_on_switch(prev_skill_id: Optional[str], new_skill_id: str) -> list[str]: """切换技能时应清理的会话键（纯函数，供 UI 与测试）。 首次进入（``prev_skill_id`` 为空）或技能未变化时返回空列表；技能切换时返回需清理的 残留键（上一技能的报告结果与任务单状态），避免旧报告串显到新技能。 """ if not prev_skill_id or prev_skill_id == new_skill_id: return [] return [SESSION_SECTIONS, SESSION_TASK_SHEET, SESSION_SHEET_STATE] def sheet_is_confirmable(sheet_dict: Optional[dict]) -> bool: """任务单是否可被确认执行（存在、且非澄清态）。 纯函数，供 UI 与测试判定「确认并开始分析」按钮是否可用（需求 5.4 / 5.7）。 """ if not sheet_dict: return False clar = sheet_dict.get("clarification") if clar and clar.get("needed"): return False return True def items_to_rows(items: list[dict]) -> list[dict]: """把任务单提取项转为可编辑表格行（纯函数，供 UI 与测试，需求 5.3）。""" rows: list[dict] = [] for it in items or []: group = it.get("group") or {} rows.append({ "use": True, "field": it.get("field", ""), "value": it.get("value", ""), "strength": group.get("strength", ""), "batch": group.get("batch", ""), "attribute": group.get("attribute", ""), "source_ref": it.get("source_ref", ""), "origin": it.get("source", "llm"), }) return rows def rows_to_items(rows: list[dict], original_items: Optional[list[dict]] = None) -> list[dict]: """把编辑后的表格行转回任务单提取项（纯函数，需求 5.5）。 判定每行是否被用户改动：与对应的原始项逐字段比对；改动或新增的行标记 ``source="user_edited"``（回填校验据此豁免）。未勾选 ``use`` 的行被丢弃（删除）。 """ original_items = original_items or [] # 以 (field,value,attribute) 粗略对齐原始项，判断是否被编辑。 def _orig_signature(it: dict) -> tuple: g = it.get("group") or {} return (str(it.get("field", "")), str(it.get("value", "")), str(g.get("strength", "")), str(g.get("batch", "")), str(g.get("attribute", "")), str(it.get("source_ref", ""))) orig_sigs = {_orig_signature(it): it for it in original_items} items: list[dict] = [] for row in rows or []: if not row.get("use", True): continue # 删除：未勾选纳入 field = str(row.get("field", "") or "").strip() value = str(row.get("value", "") or "").strip() if not field and not value: continue # 空行跳过 sig = (field, value, str(row.get("strength", "") or ""), str(row.get("batch", "") or ""), str(row.get("attribute", "") or ""), str(row.get("source_ref", "") or "")) if sig in orig_sigs: # 未改动：保留原项（含其 source / located）。 items.append(orig_sigs[sig]) else: # 改动 / 新增：标记 user_edited，豁免回填校验。 items.append({ "field": field, "value": value, "source_ref": str(row.get("source_ref", "") or ""), "located": True, "group": { "strength": str(row.get("strength", "") or ""), "batch": str(row.get("batch", "") or ""), "attribute": str(row.get("attribute", "") or ""), }, "source": "user_edited", }) return items def apply_sheet_edits( sheet_dict: dict, edited_rows: list[dict], proposed_skill_id: Optional[str] = None, ) -> dict: """把用户编辑（行 + 功能切换）应用回任务单 dict（纯函数，需求 5.5）。""" new_sheet = dict(sheet_dict or {}) new_sheet["extracted_items"] = rows_to_items(edited_rows, sheet_dict.get("extracted_items")) if proposed_skill_id is not None: new_sheet["proposed_skill_id"] = proposed_skill_id return new_sheet # —— 意图槽位确认（意图保真，阶段四）—— #: 槽位名 → 本地化展示标签（中文兜底）。 _SLOT_LABELS = { "primary_cqa": "主要质量指标", "target_timepoints": "目标外推时间点", "spec_limit": "规格限度", "target_attributes": "目标质量属性", } #: 来源态 → 展示文案。 _SLOT_STATE_LABELS = { "stated": "用户明示", "inferred": "系统推断", "missing": "缺失（待确认）", } def slots_to_rows(sheet_dict: dict) -> list[dict]: """把任务单意图槽位转为可编辑确认表格行（纯函数，供 UI 与测试）。 每行暴露：槽位名 / 标签 / 当前值 / 来源态 / 依据 / 是否锁定。``missing`` 槽位 的值为空字符串供用户填写。仅返回影响结论的槽位之外的全部槽位（含规格）。 """ rows: list[dict] = [] for s in sheet_dict.get("intent_slots") or []: name = str(s.get("name", "")) val = s.get("value") if isinstance(val, list): val_txt = ", ".join(str(v) for v in val) else: val_txt = "" if val is None else str(val) rows.append({ "name": name, "label": _SLOT_LABELS.get(name, name), "value": val_txt, "state": str(s.get("state", "missing")), "state_label": _SLOT_STATE_LABELS.get(str(s.get("state", "missing")), ""), "evidence": str(s.get("evidence", "")), "locked": bool(s.get("locked", False)), "affects_result": bool(s.get("affects_result", True)), }) return rows def _parse_slot_value(name: str, raw_value: str): """把编辑框文本解析回槽位值类型（时间点 → int 列表；其余 → 去空白字符串）。""" txt = str(raw_value or "").strip() if not txt: return None if name == "target_timepoints": nums: list[int] = [] for tok in re.split(r"[，,、\s]+", txt): tok = tok.strip() if not tok: continue try: nums.append(int(float(tok))) except ValueError: continue return nums or None return txt def apply_slot_edits(sheet_dict: dict, edited_slot_rows: list[dict]) -> dict: """把用户对意图槽位的确认/编辑写回任务单（纯函数，采纳真实意图的硬保证，需求 5.5）。 规则：用户填入/修改了某槽位的值，或显式勾选锁定 → 该槽位变为 ``stated`` + ``locked`` （置信 1.0），下游计算据此覆盖派生值，且后续重推断不得推翻。未改动且未锁定的槽位 保持原状。这是「用户说了算」的工程落地。 """ new_sheet = dict(sheet_dict or {}) by_name = {str(r.get("name", "")): r for r in (edited_slot_rows or [])} new_slots: list[dict] = [] for s in sheet_dict.get("intent_slots") or []: slot = dict(s) name = str(slot.get("name", "")) row = by_name.get(name) if row is not None: new_val = _parse_slot_value(name, row.get("value")) user_locked = bool(row.get("locked", False)) old_val = slot.get("value") changed = new_val is not None and new_val != old_val if changed or user_locked: if new_val is not None: slot["value"] = new_val # 有有效值才视为已确认明示；仅勾锁却无值则保持缺失（避免锁定空值）。 if slot.get("value") not in (None, "", []): slot["state"] = "stated" slot["locked"] = True slot["confidence"] = 1.0 new_slots.append(slot) new_sheet["intent_slots"] = new_slots return new_sheet # =========================================================================== # 八、Streamlit 渲染层（统一布局，需求 15.1） # =========================================================================== # # 以下函数依赖 ``st``（Streamlit 模块）。装配逻辑均在上方纯函数中完成，这里只做 # 渲染编排：顶部语言切换 → 侧边栏（登录 + 导航 + 当前 Skill 输入面板）→ 中央主画布 # （结果 / 报告）→ 可折叠审计追踪区。无 streamlit 时这些函数不会被调用（``main`` # 在入口处守护），故纯逻辑单测不受影响。 def _t(svc: Services, key: str, **params) -> str: """经底座 I18nService 取用文案的便捷封装（缺失安全回退，需求 17.4 / 17.5）。""" i18n = getattr(svc, "i18n", None) t = getattr(i18n, "t", None) if callable(t): try: return t(key, **params) if params else t(key) except Exception: # noqa: BLE001 - 取用失败回退 key 本身 return key return key def get_or_create_context() -> tuple[Services, SkillRegistry]: """在 Streamlit 会话中构建并缓存 ``(svc, registry)``，跨重渲染复用。""" if SESSION_CTX not in st.session_state: ensure_admin_from_env() svc, registry = bootstrap( os.environ, "skills", audit_db_path=_default_audit_db_path() ) st.session_state[SESSION_CTX] = (svc, registry) svc, registry = st.session_state[SESSION_CTX] # 把会话语言同步进 Services（语言切换后重渲染时生效，需求 17.2）。 lang = st.session_state.get(SESSION_LANG, svc.lang or DEFAULT_LANG) set_session_language(svc, lang) return svc, registry def _default_audit_db_path() -> str: """审计库落盘路径（``/data/audit.db``）；目录不可用则回退内存库。""" try: from pathlib import Path data_dir = Path(__file__).resolve().parents[1] / "data" data_dir.mkdir(parents=True, exist_ok=True) return str(data_dir / "audit.db") except Exception: # noqa: BLE001 - 落盘不可用回退内存库 return ":memory:" def render_language_switcher(svc: Services) -> None: """顶部语言切换入口：写入会话语言与 ``Services.lang`` 并触发重渲染（需求 17.1/17.2）。""" i18n = getattr(svc, "i18n", None) langs = list(getattr(i18n, "available_languages", lambda: ("zh", "en"))()) current = st.session_state.get(SESSION_LANG, svc.lang or DEFAULT_LANG) cols = st.columns([6, 1]) with cols[1]: choice = st.selectbox( _t(svc, "label.language"), options=langs, index=langs.index(current) if current in langs else 0, format_func=lambda code: _t(svc, f"lang.{code}"), key="pk_lang_select", ) if choice != current: st.session_state[SESSION_LANG] = choice set_session_language(svc, choice) _rerun() def render_login(svc: Services) -> None: """侧边栏账户区：登录 / 注册 / 退出；明文密码不记录（需求 5.1 / 5.2）。""" user = st.session_state.get(SESSION_USER) if user: role = st.session_state.get(SESSION_ROLE, DEFAULT_ROLE) st.sidebar.caption(f"👤 {user} · {_t(svc, f'role.{role}')}") if st.sidebar.button(_t(svc, "btn.logout"), key="pk_logout", use_container_width=True): for k in (SESSION_USER, SESSION_ROLE, SESSION_USER_MODEL, "pk_user_model_loaded"): st.session_state.pop(k, None) _rerun() return with st.sidebar.expander(_t(svc, "auth.account"), expanded=False): login_tab, register_tab = st.tabs( [_t(svc, "auth.login_tab"), _t(svc, "auth.register_tab")] ) # —— 登录 —— with login_tab: username = st.text_input(_t(svc, "label.username"), key="pk_login_user") password = st.text_input( _t(svc, "label.password"), type="password", key="pk_login_pass" ) if st.button(_t(svc, "btn.login"), key="pk_login_submit", use_container_width=True): result = authenticate( username, password, user_lookup=_db_user_lookup, auth=getattr(svc, "auth", None), ) if result: st.session_state[SESSION_USER] = result["user"] st.session_state[SESSION_ROLE] = result["role"] _audit_login(svc, result["user"], True) _rerun() else: _audit_login(svc, username or "(unknown)", False) st.error(_t(svc, "auth.login_fail")) # —— 注册 —— with register_tab: reg_email = st.text_input(_t(svc, "auth.email"), key="pk_reg_email") reg_pass = st.text_input( _t(svc, "label.password"), type="password", key="pk_reg_pass" ) if st.button(_t(svc, "btn.register"), key="pk_reg_submit", use_container_width=True): ok, msg = _register_user(reg_email, reg_pass) if ok: st.success(_t(svc, "auth.register_ok")) else: st.error(f"{_t(svc, 'auth.register_fail')}: {msg}") def _register_user(email: str, password: str): """注册新用户（复用 ``utils.auth``）；返回 ``(ok, message)``，不可用时友好失败。""" try: from utils.auth import register_user return register_user(email, password) except Exception as exc: # noqa: BLE001 - 注册后端不可用时不崩溃 logger.warning("注册功能不可用：%s", exc) return False, str(exc) def _render_login_legacy_unused(svc: Services) -> None: """（已废弃，保留空壳避免历史引用报错）""" return None def _audit_login(svc: Services, user: str, success: bool) -> None: audit = getattr(svc, "audit", None) recorder = getattr(audit, "record_login", None) if callable(recorder): try: recorder(user, success=success) except Exception: # noqa: BLE001 logger.debug("记录登录审计失败（已忽略）。", exc_info=True) def _load_persisted_user_model(svc: Services) -> None: """登录用户首次进入会话时，把持久化的自配模型载入会话（若有且加密可用）。 仅在会话尚未载入过、且尚无会话级用户配置时执行，避免覆盖用户本次的临时修改。 """ user = st.session_state.get(SESSION_USER) if not user: return if st.session_state.get("pk_user_model_loaded") == user: return st.session_state["pk_user_model_loaded"] = user if st.session_state.get(SESSION_USER_MODEL) is not None: return try: from services.model_config import UserModelConfig from utils import database as db rec = db.get_user_llm_config(user) if rec and rec.get("enabled"): cfg = UserModelConfig( provider=rec.get("provider", ""), model=rec.get("model", ""), api_key=rec.get("api_key", ""), base_url=rec.get("base_url", ""), ) if cfg.is_usable(): st.session_state[SESSION_USER_MODEL] = cfg except Exception as exc: # noqa: BLE001 - 持久化不可用时静默退回仅会话 logger.info("载入持久化用户模型配置不可用：%s", exc) def _persist_user_model(cfg: Any) -> bool: """把登录用户的自配模型加密持久化；未登录 / 加密不可用时返回 False（仅会话）。""" user = st.session_state.get(SESSION_USER) if not user: return False try: from utils import database as db return db.set_user_llm_config( user, cfg.provider, cfg.api_key, model=cfg.model, base_url=cfg.base_url, enabled=True, ) except Exception as exc: # noqa: BLE001 logger.info("持久化用户模型配置不可用：%s", exc) return False def _clear_persisted_user_model() -> None: """删除登录用户持久化的自配模型（若已登录）。""" user = st.session_state.get(SESSION_USER) if not user: return try: from utils import database as db db.clear_user_llm_config(user) except Exception as exc: # noqa: BLE001 logger.info("清除持久化用户模型配置不可用：%s", exc) def render_model_settings(svc: Services) -> None: """侧边栏「模型设置」面板：用户自配模型（最高优先级），并显示当前生效来源。 用户可选择提供商 / 模型并填写自己的密钥；留空则使用管理员 / 环境配置（需求 5）。 登录用户在配置了 ``PHARMAK_SECRET_KEY`` 时，密钥会**加密后持久化**并随 DB 同步到 私有 Dataset，跨会话免重填；否则仅存会话状态（``SESSION_USER_MODEL``）。 """ try: from layers.llm_providers import AVAILABLE_MODELS, LLMProvider, get_available_providers from services.model_config import UserModelConfig except Exception as exc: # noqa: BLE001 - 依赖不可用时跳过该面板 logger.info("模型设置面板不可用：%s", exc) return # 登录用户：尝试载入持久化配置（仅首次）。 _load_persisted_user_model(svc) providers = get_available_providers() with st.sidebar.expander(_t(svc, "model.settings"), expanded=False): st.caption(_t(svc, "model.use_admin_hint")) # 提示持久化能力：登录 + 加密可用时密钥跨会话保留，否则仅本次会话。 try: from utils import crypto if st.session_state.get(SESSION_USER) and crypto.is_available(): st.caption(_t(svc, "model.persist_on")) else: st.caption(_t(svc, "model.persist_off")) except Exception: # noqa: BLE001 st.caption(_t(svc, "model.persist_off")) ids = [p["id"] for p in providers] names = {p["id"]: p["name"] for p in providers} # 默认选中：会话已存用户配置则定位之，否则第一个。 existing: Optional[UserModelConfig] = st.session_state.get(SESSION_USER_MODEL) default_idx = ids.index(existing.provider) if (existing and existing.provider in ids) else 0 pid = st.selectbox( _t(svc, "model.provider"), options=ids, index=default_idx, format_func=lambda i: names.get(i, i), key="pk_user_provider", ) # 模型：已知提供商给下拉 + "其他(手动)"；custom 直接手填。 model_value = "" try: prov_enum = LLMProvider(pid) except ValueError: prov_enum = None known_models = list(AVAILABLE_MODELS.get(prov_enum, [])) if prov_enum else [] if pid == "custom": model_value = st.text_input(_t(svc, "model.model_custom"), key="pk_user_model_custom") or "" base_url = st.text_input( _t(svc, "model.base_url"), help=_t(svc, "model.base_url_hint"), key="pk_user_base_url", ) or "" else: base_url = "" if known_models: choice = st.selectbox( _t(svc, "model.model"), options=known_models + ["—"], key="pk_user_model_pick", ) model_value = "" if choice == "—" else choice custom_m = st.text_input(_t(svc, "model.model_custom"), key="pk_user_model_free") or "" if custom_m.strip(): model_value = custom_m.strip() api_key = st.text_input(_t(svc, "model.api_key"), type="password", key="pk_user_api_key") or "" cols = st.columns(2) with cols[0]: if st.button(_t(svc, "model.save"), key="pk_user_model_save"): cfg = UserModelConfig( provider=pid, model=model_value, api_key=api_key, base_url=base_url ) if pid == "custom" and not _custom_base_url_ok(svc, base_url): pass # 错误已在 _custom_base_url_ok 内提示 elif cfg.is_usable(): st.session_state[SESSION_USER_MODEL] = cfg persisted = _persist_user_model(cfg) st.success(_t(svc, "model.saved_persisted" if persisted else "model.saved")) _rerun() else: st.warning(_t(svc, "model.test_fail")) with cols[1]: if st.button(_t(svc, "model.clear"), key="pk_user_model_clear"): st.session_state.pop(SESSION_USER_MODEL, None) _clear_persisted_user_model() st.info(_t(svc, "model.cleared")) _rerun() if st.button(_t(svc, "model.test"), key="pk_user_model_test"): cfg = UserModelConfig(provider=pid, model=model_value, api_key=api_key, base_url=base_url) if pid == "custom" and not _custom_base_url_ok(svc, base_url): pass # 错误已提示 elif _test_model_config(cfg): st.success(_t(svc, "model.test_ok")) else: st.error(_t(svc, "model.test_fail")) # 显示当前生效来源（用户/管理员/环境/无）。 source = resolve_session_llm(svc, st.session_state.get(SESSION_USER_MODEL)) st.caption(_t(svc, f"model.source.{source}")) def _custom_base_url_ok(svc: Any, base_url: str) -> bool: """校验用户填写的 custom base_url 是否安全（防 SSRF）；不安全时在 UI 提示并返回 False。""" try: from utils.url_guard import is_safe_public_url ok, reason = is_safe_public_url(base_url) if not ok: st.error(f"{_t(svc, 'model.base_url_unsafe')}：{reason}") return ok except Exception as exc: # noqa: BLE001 - 校验器不可用时保守拒绝自定义端点 logger.warning("base_url 安全校验不可用：%s", exc) st.error(_t(svc, "model.base_url_unsafe")) return False def _test_model_config(cfg: Any) -> bool: """对单个用户模型配置做一次轻量真实调用测试（失败返回 False，不抛出）。""" try: from services.llm_service import LLMService from services.model_config import UserModelConfig # noqa: F401 spec = cfg.to_spec() if spec is None: return False svc = LLMService([spec]) res = svc.complete("只回复一个词。", "OK", scope="model-test") return bool(res.ok) except Exception as exc: # noqa: BLE001 logger.info("模型连接测试失败：%s", exc) return False def render_admin_model_settings(svc: Services) -> None: """管理员「全局模型配置」面板：仅管理员可见，设置后供未自配的用户使用（需求 5）。""" role = st.session_state.get(SESSION_ROLE, DEFAULT_ROLE) auth = getattr(svc, "auth", None) is_admin = getattr(auth, "is_admin", None) if not (callable(is_admin) and is_admin(role)): return try: from layers.llm_providers import get_available_providers from utils import database as db except Exception as exc: # noqa: BLE001 logger.info("管理员模型面板不可用：%s", exc) return providers = get_available_providers() ids = [p["id"] for p in providers] names = {p["id"]: p["name"] for p in providers} current = db.get_default_llm_config() with st.sidebar.expander(_t(svc, "admin.model.title"), expanded=False): # 持久化状态提示 + Space Secrets 引导（HF 免费档）。 try: from utils import hf_storage if hf_storage.is_enabled(): st.caption(_t(svc, "admin.storage.synced")) else: st.caption(_t(svc, "admin.storage.ephemeral")) except Exception: # noqa: BLE001 st.caption(_t(svc, "admin.storage.ephemeral")) st.caption(_t(svc, "admin.model.secrets_hint")) enabled = st.checkbox( _t(svc, "admin.model.enable"), value=bool(current.get("enabled")), key="pk_admin_enable", ) idx = ids.index(current.get("provider")) if current.get("provider") in ids else 0 pid = st.selectbox( _t(svc, "model.provider"), options=ids, index=idx, format_func=lambda i: names.get(i, i), key="pk_admin_provider", ) model = st.text_input(_t(svc, "model.model_custom"), value=current.get("model", ""), key="pk_admin_model") or "" api_key = st.text_input(_t(svc, "model.api_key"), type="password", key="pk_admin_api_key") or "" base_url = "" if pid == "custom": base_url = st.text_input( _t(svc, "model.base_url"), value=current.get("base_url", ""), help=_t(svc, "model.base_url_hint"), key="pk_admin_base_url", ) or "" if st.button(_t(svc, "admin.model.save"), key="pk_admin_model_save"): # 密钥留空时保留原密钥（避免每次都要重输）。 key_to_save = api_key or current.get("api_key", "") db.set_default_llm_config( pid, key_to_save, model=model, base_url=base_url, enabled=enabled ) st.success(_t(svc, "admin.model.saved")) _rerun() def render_sidebar(svc: Services, registry: SkillRegistry) -> Optional[PharmaSkill]: """渲染侧边栏参数区：登录 → 导航（按角色过滤）→ 当前 Skill 输入面板（需求 15.1/15.5）。 返回当前选中的 Skill（无可见 Skill 时返回 ``None``）。 """ st.sidebar.title(_t(svc, "app.title")) render_login(svc) # 模型设置（用户自配，最高优先级）+ 管理员全局配置（仅管理员可见）。 render_model_settings(svc) render_admin_model_settings(svc) role = st.session_state.get(SESSION_ROLE, DEFAULT_ROLE) skills = visible_skills_for_role(registry, getattr(svc, "auth", None), role) if not skills: st.sidebar.info(_t(svc, "msg.no_data")) return None i18n = getattr(svc, "i18n", None) labels = [skill_nav_label(s, i18n) for s in skills] st.sidebar.markdown(f"**{_t(svc, 'nav.section')}**") idx = st.sidebar.radio( _t(svc, "nav.section"), options=list(range(len(skills))), format_func=lambda i: labels[i], key="pk_nav", label_visibility="collapsed", ) selected = skills[idx] # 切换功能时清理上一个技能的残留报告 / 任务单状态，避免旧报告串显到新技能（用户反馈）。 prev_id = st.session_state.get(SESSION_SKILL) for k in stale_session_keys_on_switch(prev_id, selected.meta.id): st.session_state.pop(k, None) st.session_state[SESSION_SKILL] = selected.meta.id # 当前 Skill 的输入面板（参数区）。 st.sidebar.markdown("---") st.sidebar.subheader(skill_nav_label(selected, i18n)) return selected def render_main(svc: Services, skill: PharmaSkill, registry: Optional[SkillRegistry] = None) -> None: """中央主画布区：收集输入 → 理解层产出任务单 → 用户确认 → 分析 → 渲染报告。 两阶段流程（intent-understanding-layer）： 1. 用户点击主操作按钮 → 理解层 ``build_task_sheet`` 产出《分析任务单》； 2. 展示任务单（含画像 / 意图 / 提取项 / 缺失），用户**每次都需确认**后才执行； 3. 澄清态任务单展示澄清提示与可选功能，不进入计算（需求 6）。 理解层 / 编排器不可用时回退到旧的直接分发路径（向后兼容）。 """ i18n = getattr(svc, "i18n", None) lang = getattr(svc, "lang", None) st.header(skill_nav_label(skill, i18n)) st.caption(skill_description(skill, i18n, lang)) raw = _safe_render_inputs(skill, st.sidebar, svc) orchestrator = build_orchestrator(registry, svc) if registry is not None else None # 输入签名确定的技能（通用问答 / 相容性）：尊重用户在侧边栏的显式选择， # 直接走固定流水线（extract→compute→explain），不经理解层意图澄清—— # 修复「问答被『请选择质量属性梳理/货架期外推』澄清拦截」与「相容性多余确认」。 # 理解层 / 编排器不可用时同样回退直接分发（向后兼容）。 if orchestrator is None or not getattr(skill.meta, "two_phase", True): _render_main_legacy(svc, skill, raw) return action_label = skill_action_label(skill, svc) if st.button(action_label, key=f"pk_run_{skill.meta.id}", type="primary", use_container_width=True): resolve_session_llm(svc, st.session_state.get(SESSION_USER_MODEL)) with st.spinner(_t(svc, "app.loading")): try: sheet = orchestrator.build_task_sheet(raw) st.session_state[SESSION_TASK_SHEET] = sheet.to_dict() st.session_state[SESSION_SHEET_STATE] = "pending_confirm" st.session_state.pop(SESSION_SECTIONS, None) except Exception as exc: # noqa: BLE001 logger.warning("理解层处理失败：%s", exc, exc_info=True) st.error(_t(svc, "error.generic")) state = st.session_state.get(SESSION_SHEET_STATE, "none") sheet_dict = st.session_state.get(SESSION_TASK_SHEET) if state == "pending_confirm" and sheet_dict: if sheet_is_confirmable(sheet_dict): # 展示文档画像（只读）+ 可编辑提取项 + 可切换功能。 _render_task_sheet(svc, sheet_dict) edited_rows, chosen_skill = _render_task_sheet_editor(svc, sheet_dict, registry) edited_slot_rows = _render_intent_slots_editor(svc, sheet_dict) st.caption(_t(svc, "task.confirm_required")) cols = st.columns(2) with cols[0]: if st.button(_t(svc, "task.btn.confirm"), key="pk_sheet_confirm", type="primary", use_container_width=True): final_sheet = apply_sheet_edits(sheet_dict, edited_rows, chosen_skill) final_sheet = apply_slot_edits(final_sheet, edited_slot_rows) _audit_sheet_edit(svc, sheet_dict, final_sheet) _run_confirmed_sheet(svc, orchestrator, final_sheet, raw) with cols[1]: if st.button(_t(svc, "task.btn.cancel"), key="pk_sheet_cancel", use_container_width=True): st.session_state[SESSION_SHEET_STATE] = "none" st.session_state.pop(SESSION_TASK_SHEET, None) _rerun() else: _render_task_sheet(svc, sheet_dict) chosen_intent = _render_clarification(svc, sheet_dict) if chosen_intent: # 用户在澄清态选定意图 → 解析为可确认任务单（清除澄清、派生槽位）， # 下一轮进入完整可编辑确认面板（提取项编辑 + 槽位确认）。 st.session_state[SESSION_TASK_SHEET] = _resolve_clarified_sheet( sheet_dict, chosen_intent, raw ) st.session_state[SESSION_SHEET_STATE] = "pending_confirm" _rerun() return sections = st.session_state.get(SESSION_SECTIONS) if sections is not None: render_report(svc, sections) else: st.info(_t(svc, "main.empty_hint")) def _run_confirmed_sheet(svc: Services, orchestrator, sheet_dict: dict, raw: RawInput) -> None: """把用户确认的任务单交编排器执行并渲染（需求 5.4 / 5.5）。""" from kernel.task_sheet import AnalysisTaskSheet with st.spinner(_t(svc, "app.loading")): try: sheet = AnalysisTaskSheet.from_dict(sheet_dict) sections = orchestrator.run_confirmed( sheet, raw, user=st.session_state.get(SESSION_USER, "anonymous") ) st.session_state[SESSION_SECTIONS] = sections st.session_state[SESSION_SHEET_STATE] = "confirmed" except Exception as exc: # noqa: BLE001 logger.warning("确认执行失败：%s", exc, exc_info=True) st.error(_t(svc, "error.generic")) _rerun() def _render_task_sheet(svc: Services, sheet_dict: dict) -> None: """展示《分析任务单》：意图 / 文档画像 / 提取项 / 缺失项（需求 5.2）。""" st.subheader(_t(svc, "task.sheet.title")) st.caption(_t(svc, "task.sheet.intro")) intent = sheet_dict.get("intent", "unknown") st.markdown(f"**{_t(svc, 'task.intent')}**: {_t(svc, f'task.intent.{intent}')}") profile = sheet_dict.get("document_profile") or {} tables = profile.get("tables") or [] if tables: st.markdown(f"**{_t(svc, 'task.profile')}**") st.dataframe( [ { _t(svc, "task.profile.table"): t.get("title", ""), _t(svc, "task.profile.type"): t.get("table_type", ""), _t(svc, "task.profile.source"): t.get("source_document", ""), _t(svc, "task.profile.is_stability"): "✓" if t.get("is_stability_time_series") else "—", } for t in tables ], use_container_width=True, ) if not profile.get("has_stability_time_series", False): st.caption(_t(svc, "task.profile.no_stability")) items = sheet_dict.get("extracted_items") or [] missing = sheet_dict.get("missing_items") or [] if missing: st.markdown(f"**{_t(svc, 'task.missing')}**: " + "、".join(str(m) for m in missing)) def _render_task_sheet_editor(svc: Services, sheet_dict: dict, registry: Optional[SkillRegistry]): """可编辑提取项 + 可切换分析功能（需求 5.3 / 5.5）。 返回 ``(edited_rows, chosen_skill_id)``。无 ``st.data_editor`` 的旧版 Streamlit 回退为只读展示并返回原始行 / 原功能（保证不崩溃）。 """ st.markdown(f"**{_t(svc, 'task.edit.items')}**") st.caption(_t(svc, "task.edit.hint")) rows = items_to_rows(sheet_dict.get("extracted_items") or []) column_config = None data_editor = getattr(st, "data_editor", None) or getattr(st, "experimental_data_editor", None) if callable(data_editor): edited = data_editor( rows, num_rows="dynamic", # 允许增 / 删行 use_container_width=True, key="pk_sheet_editor", column_config=column_config, ) edited_rows = list(edited) if isinstance(edited, list) else rows else: # 旧版回退：只读 st.dataframe(rows, use_container_width=True) edited_rows = rows # 分析功能切换。 chosen_skill = sheet_dict.get("proposed_skill_id", "") if registry is not None: ids = [s.meta.id for s in registry.all()] if ids: default_idx = ids.index(chosen_skill) if chosen_skill in ids else 0 i18n = getattr(svc, "i18n", None) def _skill_label(sid: str) -> str: # 优先 nav.，否则取 Skill 自身本地化名 / display_name（避免暴露原始键）。 has = getattr(i18n, "has", None) if callable(has) and has(f"nav.{sid}"): return _t(svc, f"nav.{sid}") sk = registry.get(sid) return skill_nav_label(sk, i18n) if sk is not None else sid chosen_skill = st.selectbox( _t(svc, "task.edit.skill"), options=ids, index=default_idx, format_func=_skill_label, key="pk_sheet_skill_pick", ) return edited_rows, chosen_skill def _render_intent_slots_editor(svc: Services, sheet_dict: dict) -> list[dict]: """渲染意图槽位确认子面板（意图保真，阶段四）；返回可回写的槽位行。 展示每个意图维度的当前值、来源态（明示/推断/缺失）与依据，允许用户**修正并锁定**。 用户填值/锁定后经 :func:`apply_slot_edits` 写回为 ``stated+locked``，下游计算据此 采纳用户真实意图、不被模型推断推翻。无槽位时不渲染（返回空列表）。 """ rows = slots_to_rows(sheet_dict) if not rows: return [] st.markdown(f"**{_t(svc, 'task.slots.title')}**") st.caption(_t(svc, "task.slots.hint")) data_editor = getattr(st, "data_editor", None) or getattr(st, "experimental_data_editor", None) # 仅暴露可编辑的「值 / 锁定」列；其余为只读上下文。 display = [ { "label": r["label"], "value": r["value"], "state_label": r["state_label"], "evidence": r["evidence"], "locked": r["locked"], } for r in rows ] if callable(data_editor): edited = data_editor( display, use_container_width=True, key="pk_slot_editor", disabled=["label", "state_label", "evidence"], ) edited_list = list(edited) if isinstance(edited, list) else display else: # 旧版回退：只读 st.dataframe(display, use_container_width=True) edited_list = display # 回填 name（展示层用 label，回写需 name）。 out: list[dict] = [] for r, e in zip(rows, edited_list): out.append({ "name": r["name"], "value": e.get("value", r["value"]), "locked": bool(e.get("locked", r["locked"])), }) return out def _audit_sheet_edit(svc: Services, original: dict, edited: dict) -> None: """记录用户对任务单的编辑（需求 8.2 可追溯）。""" orig_items = original.get("extracted_items") or [] new_items = edited.get("extracted_items") or [] user_edited = sum(1 for it in new_items if it.get("source") == "user_edited") audit = getattr(svc, "audit", None) recorder = getattr(audit, "record_event", None) or getattr(audit, "record_analysis", None) if callable(recorder): try: recorder("task_sheet_edited", edited.get("proposed_skill_id", ""), f"orig={len(orig_items)} final={len(new_items)} user_edited={user_edited}") except Exception: # noqa: BLE001 - 审计失败不影响主流程 logger.debug("记录任务单编辑审计失败（已忽略）。", exc_info=True) def _render_clarification(svc: Services, sheet_dict: dict): """澄清态：提示 + **交互式意图选择** + 确认按钮（需求 6）。 返回用户选定的意图值（如 ``"shelf_life_extrapolation"``）当且仅当用户点击确认按钮； 否则返回 ``None``。本函数本身不进入计算——选定后由调用方解析为可确认任务单。 """ clar = sheet_dict.get("clarification") or {} st.warning(_t(svc, "understanding.clarify.title")) msg_key = clar.get("message_key") or "" if msg_key: st.info(_t(svc, msg_key)) options = [o for o in (clar.get("options") or []) if o] if not options: return None st.caption(_t(svc, "understanding.clarify.choose")) labels = {o: _t(svc, f"task.intent.{o}") for o in options} radio = getattr(st, "radio", None) if callable(radio): choice = radio( _t(svc, "understanding.clarify.choose"), options=options, format_func=lambda o: labels.get(o, o), key="pk_clarify_choice", ) else: # 极简回退：选择框 choice = st.selectbox( _t(svc, "understanding.clarify.choose"), options=options, format_func=lambda o: labels.get(o, o), key="pk_clarify_choice", ) if st.button(_t(svc, "task.clarify.apply"), key="pk_clarify_apply", type="primary", use_container_width=True): return choice return None def _resolve_clarified_sheet(sheet_dict: dict, intent_value: str, raw: RawInput) -> dict: """把用户在澄清态选定的意图解析为**可确认任务单**（纯函数，供 UI 与测试）。 - 设定 ``intent`` 与 ``proposed_skill_id``（按意图映射），并**清除澄清态**； - 依选定意图与现有提取项派生意图槽位（``intent_slots``），使后续确认面板完整呈现 来源标记与待确认事项。意图解析 / 槽位派生失败时安全降级（不抛异常）。 """ new_sheet = dict(sheet_dict or {}) new_sheet["intent"] = intent_value new_sheet["proposed_skill_id"] = _INTENT_SKILL_MAP.get(intent_value, "") new_sheet["clarification"] = None try: from kernel.task_sheet import ExtractedItem, Intent from kernel.understanding import build_intent_slots intent = Intent(intent_value) items = [ExtractedItem.from_dict(it) for it in (sheet_dict.get("extracted_items") or [])] goal = str(getattr(raw, "goal", "") or "") new_sheet["intent_slots"] = [s.to_dict() for s in build_intent_slots(goal, intent, items)] except Exception: # noqa: BLE001 - 槽位派生失败不阻断澄清解析 new_sheet.setdefault("intent_slots", []) return new_sheet #: 意图值 → Skill id（与 kernel.router.INTENT_TO_SKILL 一致）。 _INTENT_SKILL_MAP = { "descriptive_summary": "descriptive_summary", "shelf_life_extrapolation": "stability", "compatibility": "compatibility", } def _render_main_legacy(svc: Services, skill: PharmaSkill, raw: RawInput) -> None: """旧的直接分发路径（理解层不可用时回退，保持向后兼容）。""" action_label = skill_action_label(skill, svc) if st.button(action_label, key=f"pk_run_legacy_{skill.meta.id}", type="primary", use_container_width=True): resolve_session_llm(svc, st.session_state.get(SESSION_USER_MODEL)) with st.spinner(_t(svc, "app.loading")): try: sections = dispatch( skill, raw, svc, user=st.session_state.get(SESSION_USER, "anonymous"), ) st.session_state[SESSION_SECTIONS] = sections except Exception as exc: # noqa: BLE001 logger.warning("分析执行失败：%s", exc, exc_info=True) st.error(_t(svc, "error.generic")) st.session_state.pop(SESSION_SECTIONS, None) sections = st.session_state.get(SESSION_SECTIONS) if sections is not None: render_report(svc, sections) else: st.info(_t(svc, "main.empty_hint")) def _safe_render_inputs(skill: PharmaSkill, st_ctx, svc: Optional[Services] = None) -> RawInput: """调用 Skill 的 ``render_inputs``，异常时降级空输入（不崩溃）。 渲染前把一个**已绑定底座 i18n 与命名空间**的翻译函数注入到 Skill 实例 （``skill._i18n_t``），使各 Skill 的输入面板文案经底座 ``I18nService`` 取用、 随语言切换（需求 3 / 17.4）。注入失败或无 ``svc`` 时，面板按其内置中文兜底， 不影响无 GUI 的纯逻辑单测。 """ if svc is not None: skill_id = getattr(skill.meta, "id", "") or "" def _translate(key: str) -> str: # 面板传入局部键（如 "input.goal"）；底座目录以 "." 前缀合并。 return _t(svc, f"{skill_id}.{key}") try: setattr(skill, "_i18n_t", _translate) except Exception: # noqa: BLE001 - 注入失败不阻断渲染 pass try: raw = skill.render_inputs(st_ctx) return raw if isinstance(raw, RawInput) else RawInput() except Exception as exc: # noqa: BLE001 logger.warning("渲染 Skill %s 输入面板失败：%s", skill.meta.id, exc) return RawInput() def render_report(svc: Services, sections: ReportSections) -> None: """渲染分析返回的 ``ReportSections``：优先完整 HTML，否则逐段展示，并提供下载。""" html = getattr(sections, "html", "") or "" if html: # 用 iframe 承载完整自包含报告（含样式）；跨 Streamlit 版本兼容。 _render_html_component(html, height=760) cols = st.columns(2) with cols[0]: st.download_button( _t(svc, "btn.download_report"), data=html, file_name="report.html", mime="text/html", key="pk_download_report", ) # 专业 Word(.docx) 下载：表格化、分级标题、去 AI/Markdown 噪声。 with cols[1]: _render_word_download(svc, html) return section_map = getattr(sections, "sections", {}) or {} if not section_map: st.info(_t(svc, "msg.no_data")) return for key, value in section_map.items(): st.subheader(_t(svc, f"report.section.{key}") if key else key) text = value if isinstance(value, str) else str(value) if text.strip().startswith("<"): st.markdown(text, unsafe_allow_html=True) else: st.write(text) def _render_word_download(svc: Services, html: str) -> None: """生成并提供专业 Word(.docx) 下载；python-docx 不可用时给出友好提示。""" try: report = getattr(svc, "report", None) to_docx = getattr(report, "to_docx", None) data = to_docx(html) if callable(to_docx) else None except Exception: # noqa: BLE001 - 导出失败不影响主流程 data = None if data: st.download_button( _t(svc, "btn.download_word"), data=data, file_name="report.docx", mime="application/vnd.openxmlformats-officedocument.wordprocessingml.document", key="pk_download_word", ) else: st.caption(_t(svc, "msg.word_unavailable")) def _render_html_component(html: str, *, height: int = 760) -> None: """跨版本兼容地嵌入 HTML：优先 ``st.components.v1.html``，回退 ``st.html``。""" try: comp = getattr(getattr(st, "components", None), "v1", None) if comp is not None and hasattr(comp, "html"): comp.html(html, height=height, scrolling=True) return except Exception: # noqa: BLE001 - 回退到 st.html pass html_fn = getattr(st, "html", None) if callable(html_fn): html_fn(html) else: st.markdown(html, unsafe_allow_html=True) def render_audit(svc: Services) -> None: """可折叠审计追踪区：仅 admin / auditor 可见（需求 15.1 / 15.5）。""" role = st.session_state.get(SESSION_ROLE, DEFAULT_ROLE) if not can_view_audit(getattr(svc, "auth", None), role): return audit = getattr(svc, "audit", None) recent = getattr(audit, "recent", None) with st.expander(_t(svc, "nav.audit"), expanded=False): if not callable(recent): st.info(_t(svc, "msg.no_data")) return try: events = recent(50) except Exception as exc: # noqa: BLE001 logger.warning("查询审计事件失败：%s", exc) events = [] if not events: st.info(_t(svc, "msg.no_data")) return rows = [ { "ts": getattr(e, "ts_iso", ""), "user": getattr(e, "user", ""), "event": getattr(e, "event_type", ""), "skill": getattr(e, "skill_id", "") or "", "step": getattr(e, "step", "") or "", } for e in events ] st.dataframe(rows, use_container_width=True) def render_build_warnings(svc: Services) -> None: """渲染服务构建期降级告警（如某可选重依赖缺失），帮助用户理解功能降级。""" warnings = getattr(svc, "_build_warnings", None) or [] if not warnings: return with st.expander("⚠️ " + _t(svc, "warn.pdf_fallback"), expanded=False): for name, msg in warnings: st.caption(f"{name}: {msg}") def _rerun() -> None: """跨 Streamlit 版本兼容的重渲染触发。""" rerun = getattr(st, "rerun", None) or getattr(st, "experimental_rerun", None) if callable(rerun): rerun() def load_css() -> None: """注入 Nordic 极简主题 CSS（``assets/style.css``）；缺失时静默跳过。""" try: from pathlib import Path css_path = Path(__file__).resolve().parents[1] / "assets" / "style.css" if css_path.is_file(): css = css_path.read_text(encoding="utf-8") st.markdown(f"", unsafe_allow_html=True) except Exception as exc: # noqa: BLE001 - 样式缺失不应影响功能 logger.info("加载样式表失败（已忽略）：%s", exc) def render_top_bar(svc: Services) -> None: """顶部品牌条 + 右侧语言切换：左品牌名/副标题，右语言选择（修复 #0 布局）。""" left, right = st.columns([5, 1]) with left: st.markdown( f"
" f"" f"💊 {_html_escape(_t(svc, 'app.title'))}" f"" f"{_html_escape(_t(svc, 'app.subtitle'))}
", unsafe_allow_html=True, ) with right: i18n = getattr(svc, "i18n", None) langs = list(getattr(i18n, "available_languages", lambda: ("zh", "en"))()) current = st.session_state.get(SESSION_LANG, svc.lang or DEFAULT_LANG) choice = st.selectbox( _t(svc, "label.language"), options=langs, index=langs.index(current) if current in langs else 0, format_func=lambda code: _t(svc, f"lang.{code}"), key="pk_lang_select", label_visibility="collapsed", ) if choice != current: st.session_state[SESSION_LANG] = choice set_session_language(svc, choice) _rerun() st.markdown( "
", unsafe_allow_html=True, ) def _html_escape(s: str) -> str: import html as _h return _h.escape(str(s or "")) def main() -> None: """Streamlit 应用主入口：装配上下文并按统一布局渲染。""" if st is None: # pragma: no cover - 无 streamlit 环境下不渲染 raise RuntimeError( "Streamlit 未安装，无法渲染界面。请通过 " "`streamlit run core/app.py` 启动。" ) # set_page_config 必须是第一个 Streamlit 调用。 st.set_page_config( page_title="Pharma K", page_icon="💊", layout="wide", initial_sidebar_state="expanded", ) load_css() svc, registry = get_or_create_context() # 顶部品牌条 + 语言切换（需求 17.1 / 17.2；修复 #0 布局）。 render_top_bar(svc) render_build_warnings(svc) # 侧边栏参数区（账户 + 导航 + 输入面板，需求 15.1 / 15.5）。 selected = render_sidebar(svc, registry) # 中央主画布区（结果 / 报告，需求 15.1）。 if selected is not None: render_main(svc, selected, registry) else: st.info(_t(svc, "msg.no_data")) # 可折叠审计追踪区（按角色，需求 15.1 / 15.5）。 render_audit(svc) if __name__ == "__main__": # pragma: no cover - 由 `streamlit run` 调用 main()