""" FAM-UI - 家庭多模态智能监控系统前端 v2（新架构 v2，2026-08-21）管理后台：仅从甲骨文同步镜像 (sync_videos / sync_events / sync_people) 读取展示，不处理任何视频。页面: - 🕒 事件时间轴: 视频会话列表 + 事件时间线（纯文本摘要，无帧图） - 💬 AI 对话: 基于同步事件上下文问答（走核心 /api/chat/ask -> 甲骨文编排） - 📝 对话历史 - 👤 人物管理: 命名 / 合并（回推甲骨文，不再有帧照片） - 📈 统计图表: 模型来源 / 关注事件 / 同步状态 """ import os import sys import json import html as _html import requests import streamlit as st import pymysql import pymysql.cursors from datetime import datetime, date sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) from config_loader import load_config _cfg = load_config() _core_url = _cfg.get('core_url', 'http://127.0.0.1:8000') _oracle_url = (_cfg.get('oracle_url') or '').rstrip('/') _oracle_token = _cfg.get('oracle_token') or '' _db_cfg = _cfg.get('database', {}) esc = _html.escape def get_db_conn(): """获取数据库连接""" return pymysql.connect( host=_db_cfg.get('host', '127.0.0.1'), port=_db_cfg.get('port', 3306), user=_db_cfg.get('user', 'root'), password=_db_cfg.get('password', ''), database=_db_cfg.get('database', 'sentinel_home_ai'), charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) def serialize_datetime(obj): if hasattr(obj, 'isoformat'): return obj.isoformat() return str(obj) def parse_ts(ts): """'2026-08-14 22:31:15' / datetime -> datetime（失败返回 None）""" if isinstance(ts, datetime): return ts if not ts: return None try: return datetime.strptime(str(ts)[:19], '%Y-%m-%d %H:%M:%S') except ValueError: # 兼容仅相对时间 '00:01:23' try: return datetime.strptime(str(ts)[:8], '%H:%M:%S') except ValueError: return None def parse_persons(person_list_json): """解析 person_list_json（字符串或数组）为名字集合""" if not person_list_json: return set() if isinstance(person_list_json, (list, tuple)): items = person_list_json else: try: items = json.loads(person_list_json) except (ValueError, TypeError): items = [person_list_json] out = set() for it in items: s = str(it).strip() if s and s != '无人': out.add(s) return out # ============================================================ # 页面配置 & 全局样式 # ============================================================ st.set_page_config( page_title="家庭智能监控", page_icon="🏠", layout="wide", initial_sidebar_state="expanded" ) GLOBAL_CSS = """ """ def page_header(icon: str, title: str, sub: str = ''): st.markdown( f'

' f'

{icon}

' f'

{esc(title)}

' f'{f"

{esc(sub)}

" if sub else ""}' f'

', unsafe_allow_html=True) def render_event_list(events: list): """渲染事件时间线：左相对时间 + 右事件卡（事件画面截图 + 人物/关注徽章 + 描述）""" items = [] for e in events: ts = parse_ts(e.get('ts')) time_label = ts.strftime('%H:%M:%S') if ts else (str(e.get('ts') or '')[:8] or '--:--') camera = e.get('camera_name') or '' persons = parse_persons(e.get('person_list_json')) attention = bool(e.get('is_attention_event')) desc = e.get('description') or '（无描述）' eid = e.get('id') # 事件对应时间点的画面截图（Oracle 带 token 接口；加载失败自动隐藏） img_html = '' if _oracle_url and eid: img_html = ( f'

') badges = ''.join( f'{esc(p)}' for p in sorted(persons)) if attention: badges = '⚠ 需关注' + badges items.append( f'

' f'

{esc(time_label)}' f'{f"{esc(camera)}" if camera else ""}

' f'

{img_html}{badges}' f'

{esc(desc)}

' f'

' ) st.markdown(f'

{"".join(items)}

', unsafe_allow_html=True) # ============================================================ # 侧边栏 # ============================================================ st.markdown(GLOBAL_CSS, unsafe_allow_html=True) st.sidebar.markdown( '

' '

🏠 家庭智能监控

' '

SENTINEL HOME AI · 管理后台

' '

', unsafe_allow_html=True) # 顶部横向导航 page = st.segmented_control( "功能页面", ["🕒 事件时间轴", "💬 AI 对话", "📝 对话历史", "👤 人物管理", "📈 统计图表", "🤖 模型统计"], default="🕒 事件时间轴", label_visibility="collapsed" ) or "🕒 事件时间轴" # 侧边栏底部：同步状态 try: resp = requests.get(f"{_core_url}/api/status", timeout=10) if resp.status_code == 200: sdata = resp.json().get('sync', {}) last = sdata.get('last_sync_at') err = sdata.get('last_error') cursor = sdata.get('cursor') cnt = sdata.get('last_count') cnt_str = f"视频+{cnt[0]} / 事件+{cnt[1]} / 人物+{cnt[2]}" if cnt else "—" status_cls = 'ok' if sdata.get('running') else 'err' status_txt = '同步中' if sdata.get('running') else '未运行' err_line = f'

⚠ {esc(err)}

' if err else '' st.sidebar.markdown('---') st.sidebar.markdown( f'

' f'同步状态
' f'状态 {esc(status_txt)}
' f'最近 {esc(str(last)[:19]) if last else "—"}
' f'本次增量 {esc(cnt_str)}
' f'游标 {esc(str(cursor)[:19]) if cursor else "（全量）"}' f'{err_line}' f'

', unsafe_allow_html=True) except Exception: pass # ============================================================ # 事件时间轴页 # ============================================================ if page == "🕒 事件时间轴": page_header('🕒', '事件时间轴', '视频会话 · 时间点 · 信息摘要') with st.container(): col_date, col_sp = st.columns([1, 3]) with col_date: date_filter = st.date_input("日期", value=None) date_str = date_filter.isoformat() if date_filter else None # 统计卡 try: conn = get_db_conn() try: cursor = conn.cursor() if date_str: cursor.execute( "SELECT COUNT(*) v FROM sync_videos WHERE status='done' AND processed_at LIKE %s", (f'{date_str}%',)) v = cursor.fetchone().get('v', 0) cursor.execute( "SELECT COUNT(*) e FROM sync_events se JOIN sync_videos sv ON se.video_id=sv.id WHERE sv.processed_at LIKE %s", (f'{date_str}%',)) e = cursor.fetchone().get('e', 0) cursor.execute( """SELECT COALESCE(SUM(se.is_attention_event),0) AS att FROM sync_events se JOIN sync_videos sv ON se.video_id=sv.id WHERE sv.processed_at LIKE %s""", (f'{date_str}%',)) att = cursor.fetchone().get('att', 0) else: cursor.execute("SELECT COUNT(*) v FROM sync_videos WHERE status='done'") v = cursor.fetchone().get('v', 0) cursor.execute("SELECT COUNT(*) e FROM sync_events") e = cursor.fetchone().get('e', 0) cursor.execute("SELECT COALESCE(SUM(is_attention_event),0) att FROM sync_events") att = cursor.fetchone().get('att', 0) # 出现人物数：distinct label 命中 sync_events cursor.execute("SELECT label, canonical_name FROM sync_people") people = cursor.fetchall() ps = 0 for p in people: name = p['canonical_name'] or p['label'] cursor.execute("SELECT COUNT(*) c FROM sync_events WHERE person_list_json LIKE %s", (f'%{name}%',)) if cursor.fetchone()['c'] > 0: ps += 1 finally: conn.close() st.markdown( f'

' f'

{v}

视频会话

' f'

{e}

事件数

' f'

{ps}

出现人物

' f'

{att}

需关注

' f'

', unsafe_allow_html=True) except Exception: pass # 视频会话列表 page_size = 15 if 'event_page' not in st.session_state: st.session_state.event_page = 0 offset = st.session_state.event_page * page_size videos = [] try: conn = get_db_conn() try: cursor = conn.cursor() if date_str: cursor.execute( """SELECT id, filename, camera_name, event_start_time, summary_json, compute_provider, processed_at, (SELECT COUNT(*) FROM sync_events se WHERE se.video_id=sync_videos.id) AS event_count FROM sync_videos WHERE status='done' AND processed_at LIKE %s ORDER BY COALESCE(processed_at, updated_at, created_at) DESC LIMIT %s OFFSET %s""", (f'{date_str}%', page_size, offset)) else: cursor.execute( """SELECT id, filename, camera_name, event_start_time, summary_json, compute_provider, processed_at, (SELECT COUNT(*) FROM sync_events se WHERE se.video_id=sync_videos.id) AS event_count FROM sync_videos WHERE status='done' ORDER BY COALESCE(processed_at, updated_at, created_at) DESC LIMIT %s OFFSET %s""", (page_size, offset)) videos = cursor.fetchall() finally: conn.close() except Exception as e: st.error(f"读取同步数据失败: {e}") if not videos: st.markdown( '

🗓' '该日期暂无监控会话（甲骨文尚未同步数据？）

', unsafe_allow_html=True) else: if 'selected_video_id' not in st.session_state: st.session_state.selected_video_id = videos[0]['id'] valid_ids = {v['id'] for v in videos} if st.session_state.selected_video_id not in valid_ids: st.session_state.selected_video_id = videos[0]['id'] selected_id = st.session_state.selected_video_id col_list, col_detail = st.columns([1, 2.35], gap='large') with col_list: st.markdown( f'

视频会话 · {len(videos)} 条

', unsafe_allow_html=True) for vid in videos: proc = vid.get('processed_at') ptime = parse_ts(proc) time_label = ptime.strftime('%m-%d %H:%M') if ptime else '--:--' date_label = ptime.strftime('%Y-%m-%d') if ptime else '' summary = (vid.get('summary_json') or '').strip() summary_short = summary if len(summary) <= 28 else summary[:28] + '…' if not summary_short: summary_short = '暂无摘要' is_selected = vid['id'] == selected_id if st.button( f"{'▶ ' if is_selected else ''}{time_label} · {vid.get('camera_name') or '未知'} · {vid['event_count']}事件", key=f"vidbtn_{vid['id']}", type="primary" if is_selected else "secondary", use_container_width=True ): st.session_state.selected_video_id = vid['id'] st.rerun() st.markdown( f'

{esc(date_label)} · {esc(summary_short)}

', unsafe_allow_html=True) nav1, nav2, nav3 = st.columns(3) with nav1: if st.button("← 上一页", use_container_width=True, disabled=st.session_state.event_page == 0): st.session_state.event_page -= 1 st.rerun() with nav3: if st.button("下一页 →", use_container_width=True, disabled=len(videos) < page_size): st.session_state.event_page += 1 st.rerun() with col_detail: vid = next(v for v in videos if v['id'] == selected_id) proc = parse_ts(vid.get('processed_at')) start = parse_ts(vid.get('event_start_time')) range_str = '' if proc: range_str = proc.strftime('%Y-%m-%d %H:%M:%S') if start: range_str += f"（起始 {start.strftime('%H:%M:%S')}）" provider = vid.get('compute_provider') or '' model_badges = '' if provider: model_badges = ''.join( f'{esc(p)}' for p in str(provider).split(',')) summary = vid.get('summary_json') or '暂无全局摘要' # 视频缩略图（Oracle 带 token 接口，无图时优雅降级） thumb_html = '' if _oracle_url: thumb_html = ( f'

') st.markdown( f'

' f'{thumb_html}' f'

' f'{esc(vid.get("camera_name") or "未知摄像头")}' f'会话 #{vid["id"]}' f'{model_badges}

' f'

⏱ {esc(range_str)} · 文件 {esc(vid.get("filename") or "")}

' f'

{esc(summary)}

' f'

', unsafe_allow_html=True) events = [] try: conn = get_db_conn() try: cursor = conn.cursor() cursor.execute( """SELECT e.id, e.ts, e.description, e.person_list_json, e.is_attention_event, v.camera_name FROM sync_events e JOIN sync_videos v ON e.video_id = v.id WHERE e.video_id=%s ORDER BY e.ts ASC""", (selected_id,)) events = cursor.fetchall() finally: conn.close() except Exception as e: st.error(f"读取事件失败: {e}") if events: render_event_list(events) else: st.markdown( '

🎞' '该会话暂无时间点事件

', unsafe_allow_html=True) # ============================================================ # AI 对话页 # ============================================================ elif page == "💬 AI 对话": page_header('💬', 'AI 对话', '基于同步事件上下文的智能问答') named = [] try: conn = get_db_conn() try: cursor = conn.cursor() cursor.execute( "SELECT DISTINCT canonical_name FROM sync_people " "WHERE canonical_name IS NOT NULL AND canonical_name != ''") named = [r['canonical_name'] for r in cursor.fetchall()] finally: conn.close() except Exception: pass col1, col2 = st.columns(2) with col1: queried_person = st.text_input("查询人物", value=named[0] if named else "") with col2: queried_date = st.date_input("查询日期", value=date.today()) if named: quick_person = st.selectbox("快捷选择成员", [""] + named) if quick_person: queried_person = quick_person quick_questions = [ f"{queried_person}今天干嘛了？", f"{queried_person}有没有发生什么需要注意的事情？", f"今天{queried_person}的活动时间线是什么？", ] selected_quick = st.selectbox("快捷提问", ["自定义"] + quick_questions) user_question = st.text_area("你的问题", value=selected_quick if selected_quick != "自定义" else "") if st.button("提问", type="primary"): if not user_question.strip(): st.warning("请输入问题") elif not queried_person.strip(): st.warning("请输入查询人物") else: with st.spinner("AI 正在思考..."): try: resp = requests.post( f"{_core_url}/api/chat/ask", json={ "question": user_question, "queried_person": queried_person, "queried_date": queried_date.isoformat() }, timeout=120 ) if resp.status_code == 200: data = resp.json() st.markdown( f'

❓ 提问

' f'{esc(user_question)}

', unsafe_allow_html=True) st.markdown( f'

🤖 回答

', unsafe_allow_html=True) st.markdown(data['answer']) if data.get('context_summary'): st.caption(f"上下文: {data['context_summary']}") else: st.error(f"请求失败: {resp.status_code} {resp.text}") except requests.ConnectionError: st.error(f"无法连接 FAM-Core ({_core_url})") except Exception as e: st.error(f"异常: {e}") # ============================================================ # 对话历史页 # ============================================================ elif page == "📝 对话历史": page_header('📝', '对话历史', '历史问答记录') if 'chat_page' not in st.session_state: st.session_state.chat_page = 0 page_size = 20 offset = st.session_state.chat_page * page_size history = [] try: conn = get_db_conn() try: cursor = conn.cursor() cursor.execute( """SELECT chat_id, user_question, ai_answer, context_summary, queried_date, queried_person, created_at FROM chat_history ORDER BY created_at DESC LIMIT %s OFFSET %s""", (page_size, offset)) history = cursor.fetchall() finally: conn.close() except Exception as e: st.error(f"读取对话历史失败: {e}") if not history: st.markdown( '

💬' '暂无对话记录

', unsafe_allow_html=True) else: for h in history: created = serialize_datetime(h['created_at']) st.markdown( f'

' f'👤 {esc(h.get("queried_person") or "未知")} · {esc(created)}

' f'{esc(h["user_question"])}

', unsafe_allow_html=True) st.markdown( f'

🤖 回答

', unsafe_allow_html=True) st.markdown(h['ai_answer'] or '') st.markdown('

', unsafe_allow_html=True) nav1, nav2, nav3 = st.columns([1, 1, 1]) with nav1: if st.button("← 上一页", disabled=st.session_state.chat_page == 0): st.session_state.chat_page -= 1 st.rerun() with nav2: st.markdown( f'

第 {st.session_state.chat_page + 1} 页

', unsafe_allow_html=True) with nav3: if st.button("下一页 →", disabled=len(history) < page_size): st.session_state.chat_page += 1 st.rerun() # ============================================================ # 人物管理页 # ============================================================ elif page == "👤 人物管理": page_header('👤', '人物管理', '所有出现的人物 · 命名与合并（回推甲骨文）') people = [] try: conn = get_db_conn() try: cursor = conn.cursor() cursor.execute( "SELECT id, label, canonical_name, first_seen, appearances, source " "FROM sync_people ORDER BY id ASC") people = cursor.fetchall() finally: conn.close() except Exception as e: st.error(f"读取人物失败: {e}") people = [] # 按规范名聚合（未命名按 label） groups = {} for p in people: key = p['canonical_name'] or p['label'] groups.setdefault(key, {'display': key, 'is_named': bool(p['canonical_name']), 'labels': [], 'appearances': 0, 'first_seen': None}) g = groups[key] g['labels'].append(p['label']) g['appearances'] += (p.get('appearances') or 0) fs = p.get('first_seen') if fs and (g['first_seen'] is None or str(fs) < str(g['first_seen'])): g['first_seen'] = fs if not groups: st.markdown( '

👤' '暂未发现任何人物（甲骨文尚未同步）

', unsafe_allow_html=True) else: st.markdown( f'

' f'共 {len(groups)} 个身份 · ' f'已命名 {sum(1 for g in groups.values() if g["is_named"])} · ' f'未命名 {sum(1 for g in groups.values() if not g["is_named"])}' f'

', unsafe_allow_html=True) # 命名/合并操作 def do_name(label, new_name): try: resp = requests.post( f"{_core_url}/api/member/name", json={"label": label, "canonical_name": new_name, "named_by": "UI管理员"}, timeout=30) if resp.status_code == 200: st.success(f"已保存：{label} → {new_name}（已回推甲骨文并即时同步）") st.rerun() else: st.error(f"失败: {resp.status_code} {resp.text[:150]}") except Exception as e: st.error(f"异常: {e}") def do_merge(src_label, tgt_key): try: resp = requests.post( f"{_core_url}/api/member/merge", json={"source": src_label, "target": tgt_key}, timeout=30) if resp.status_code == 200: st.success(f"已合并：{src_label} → {tgt_key}") st.rerun() else: st.error(f"合并失败: {resp.status_code} {resp.text[:150]}") except Exception as e: st.error(f"合并异常: {e}") label_options = [p['label'] for p in people] for key in sorted(groups, key=lambda k: -groups[k]['appearances']): g = groups[key] is_named = g['is_named'] fs = parse_ts(g['first_seen']) first_str = fs.strftime('%m-%d %H:%M') if fs else '--' tag = (f'已命名') if is_named else \ (f'未命名') label_str = ' · '.join(g['labels']) st.markdown( f'

' f'{esc(key)}{tag}

' f'

' f'标识: {esc(label_str)}

' f'

' f'出现 {g["appearances"]} 次 · 首次 {esc(first_str)}

', unsafe_allow_html=True) if not is_named: # 未命名身份：每个 label 都给一个命名框 for lb in g['labels']: col_in, col_btn, col_mg = st.columns([2, 1, 1.4]) with col_in: new_name = st.text_input( "名字", key=f"nm_{lb}", placeholder=f'为「{lb}」起名', label_visibility="collapsed") with col_btn: if st.button("命名", key=f"btn_{lb}", type="primary", use_container_width=True): if new_name.strip(): do_name(lb, new_name.strip()) else: st.warning("请输入名字") with col_mg: merge_to = st.selectbox( "合并到", options=[""] + [o for o in label_options if o != lb], key=f"mg_{lb}", label_visibility="collapsed", placeholder="合并到其他…") if merge_to: if st.button("合并", key=f"mbtn_{lb}", use_container_width=True): do_merge(lb, merge_to) else: # 已命名：仅提供合并到其他身份 merge_to = st.selectbox( "合并到", options=[""] + [o for o in label_options if o != key], key=f"mg2_{key}", label_visibility="collapsed", placeholder="合并到其他身份…") if merge_to: if st.button("合并", key=f"mbtn2_{key}", use_container_width=True): # 用该身份下任一 label 作为 source do_merge(g['labels'][0], merge_to) st.markdown('

', unsafe_allow_html=True) # ============================================================ # 统计图表页 # ============================================================ elif page == "📈 统计图表": page_header('📈', '统计图表', '模型来源 / 关注事件 / 同步状态') conn = None try: conn = get_db_conn() cursor = conn.cursor() st.markdown('

模型来源分布

', unsafe_allow_html=True) cursor.execute( "SELECT compute_provider, COUNT(*) AS count FROM sync_videos " "WHERE status='done' GROUP BY compute_provider") stats = cursor.fetchall() if stats: chart = {} for r in stats: prov = (r['compute_provider'] or 'unknown') # 可能是逗号分隔多个 for p in str(prov).split(','): p = p.strip() if p: chart[p] = chart.get(p, 0) + r['count'] st.bar_chart(chart) else: st.markdown('

暂无统计数据

', unsafe_allow_html=True) st.markdown('

关注事件统计

', unsafe_allow_html=True) cursor.execute( """SELECT v.processed_at, e.person_list_json FROM sync_events e JOIN sync_videos v ON e.video_id=v.id WHERE e.is_attention_event = 1 ORDER BY v.processed_at DESC""") attention = cursor.fetchall() if attention: import pandas as pd rows = [] for a in attention: persons = parse_persons(a['person_list_json']) ptime = parse_ts(a['processed_at']) rows.append({ "日期": ptime.strftime('%Y-%m-%d') if ptime else '?', "人物": ','.join(sorted(persons)) or '无人', }) df_att = pd.DataFrame(rows) st.dataframe(df_att, use_container_width=True, hide_index=True) else: st.markdown('

暂无关注事件

', unsafe_allow_html=True) st.markdown('

同步状态

', unsafe_allow_html=True) try: resp = requests.get(f"{_core_url}/api/status", timeout=10) if resp.status_code == 200: sdata = resp.json().get('sync', {}) cnt = sdata.get('last_count') cnt_str = f"视频+{cnt[0]} / 事件+{cnt[1]} / 人物+{cnt[2]}" if cnt else "—" err = sdata.get('last_error') cls = 'ok' if sdata.get('running') else 'err' err_line = f'

⚠ {esc(err)}

' if err else '' st.markdown( f'

' f'运行状态 {"同步中" if sdata.get("running") else "未运行"}
' f'最近同步 {esc(str(sdata.get("last_sync_at") or "—")[:19])}
' f'本次增量 {esc(cnt_str)}
' f'游标 {esc(str(sdata.get("cursor") or "（全量）")[:19])}
' f'周期 {esc(str(sdata.get("interval_sec")))}s' f'{err_line}

', unsafe_allow_html=True) except Exception: st.markdown('

无法获取同步状态

', unsafe_allow_html=True) except Exception as e: st.error(f"统计查询失败: {e}") finally: if conn: conn.close() # ============================================================ # 模型统计页（云端模型识别成功/失败统计） # ============================================================ elif page == "🤖 模型统计": page_header('🤖', '云端模型统计', '请求时间 · 耗时 · 成功/失败 · 失败原因') conn = None try: conn = get_db_conn() cursor = conn.cursor() # ---- 按模型聚合：成功/失败/成功率/平均耗时 ---- st.markdown('

按模型聚合

', unsafe_allow_html=True) cursor.execute( """SELECT provider, model, SUM(CASE WHEN success=1 THEN 1 ELSE 0 END) AS ok_cnt, SUM(CASE WHEN success=0 THEN 1 ELSE 0 END) AS fail_cnt, ROUND(AVG(duration_sec),1) AS avg_dur, COUNT(*) AS total, MAX(created_at) AS last_call FROM sync_model_calls GROUP BY provider, model ORDER BY total DESC""") agg = cursor.fetchall() if agg: rows = [] for r in agg: total = r['total'] or 0 ok = r['ok_cnt'] or 0 rate = (ok / total * 100) if total else 0 rows.append({ "模型": f"{r['provider']} / {r['model']}", "成功": ok, "失败": r['fail_cnt'] or 0, "成功率%": round(rate, 1), "平均耗时s": r['avg_dur'], "最后调用": str(r['last_call'] or '—')[:19], }) st.dataframe(rows, use_container_width=True, hide_index=True) else: st.markdown('

暂无模型调用记录（视频处理中或尚未同步）

', unsafe_allow_html=True) # ---- 最近调用明细 ---- st.markdown('