""" Oracle 本地库（SQLite） - 视频摘要 / 事件 / 人物 存储 表结构: videos : 每个被处理的视频一个记录（含全局摘要 + 事件列表 + 人物列表，JSON 冗余存储便于查询） events : 视频拆出的事件（时间点 + 描述 + 涉及人物） people : 规范人物表（canonical_name + 别名），由 person_service 维护 sync_cursor: 同步游标（NAS 拉取用，记录最后成功同步时间） 对外提供: - upsert_video / get_pending_videos / mark_video_processed - upsert_event - upsert_person / get_known_members_context - get_sync_delta(since_iso) -> 增量数据（供 NAS 拉取） - set_cursor / get_cursor """ import os import json import re import sqlite3 import threading from datetime import datetime, timezone, timedelta from typing import Dict, List, Optional logger = None # 延迟注入，避免循环 import def _now_iso() -> str: return datetime.now(timezone(timedelta(hours=8))).strftime('%Y-%m-%d %H:%M:%S') class OracleDB: def __init__(self, db_path: str): os.makedirs(os.path.dirname(db_path), exist_ok=True) self.db_path = db_path self._conn = sqlite3.connect(db_path, check_same_thread=False) self._conn.row_factory = sqlite3.Row self._conn.execute("PRAGMA journal_mode=WAL") self._conn.execute("PRAGMA busy_timeout=10000") self._write_lock = threading.Lock() # 复合写（如 DELETE+INSERT+commit）串行化 self._init_schema() # ------------------------------------------------------------------ def _init_schema(self): c = self._conn c.executescript(""" CREATE TABLE IF NOT EXISTS videos ( id INTEGER PRIMARY KEY AUTOINCREMENT, drive_file_id TEXT, filename TEXT UNIQUE, local_path TEXT, camera_name TEXT, duration_sec REAL, event_start_time TEXT, status TEXT DEFAULT 'pending', retry_count INTEGER DEFAULT 0, file_valid INTEGER DEFAULT 1, file_error TEXT, media_meta_json TEXT, last_fail_at TEXT, summary_json TEXT, events_json TEXT, people_json TEXT, compute_provider TEXT, created_at TEXT, updated_at TEXT, processed_at TEXT ); CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, video_id INTEGER, ts TEXT, description TEXT, person_list_json TEXT, person_appearances_json TEXT, is_attention_event INTEGER DEFAULT 0, FOREIGN KEY(video_id) REFERENCES videos(id) ); CREATE TABLE IF NOT EXISTS people ( id INTEGER PRIMARY KEY AUTOINCREMENT, label TEXT UNIQUE, canonical_name TEXT, first_seen TEXT, appearances INTEGER DEFAULT 0, source TEXT DEFAULT 'llm', features_json TEXT, display_uid TEXT, updated_at TEXT ); CREATE TABLE IF NOT EXISTS sync_cursor ( key TEXT PRIMARY KEY, value TEXT ); CREATE TABLE IF NOT EXISTS model_calls ( id INTEGER PRIMARY KEY AUTOINCREMENT, provider TEXT, model TEXT, video_id INTEGER, filename TEXT, started_at TEXT, duration_sec REAL, success INTEGER DEFAULT 0, error TEXT, created_at TEXT ); CREATE TABLE IF NOT EXISTS service_activity ( id INTEGER PRIMARY KEY AUTOINCREMENT, service TEXT, action TEXT, detail TEXT, ts TEXT ); CREATE INDEX IF NOT EXISTS idx_videos_updated ON videos(updated_at); CREATE INDEX IF NOT EXISTS idx_events_video ON events(video_id); CREATE INDEX IF NOT EXISTS idx_model_calls_created ON model_calls(created_at); CREATE INDEX IF NOT EXISTS idx_activity_ts ON service_activity(ts); """) # 兼容旧库：补 retry_count / file_valid / media 等列（生产-消费队列用） cols = [r[1] for r in c.execute("PRAGMA table_info(videos)").fetchall()] for col, ddl in [ ('retry_count', "ALTER TABLE videos ADD COLUMN retry_count INTEGER DEFAULT 0"), ('file_valid', "ALTER TABLE videos ADD COLUMN file_valid INTEGER DEFAULT 1"), ('file_error', "ALTER TABLE videos ADD COLUMN file_error TEXT"), ('media_meta_json', "ALTER TABLE videos ADD COLUMN media_meta_json TEXT"), ('last_fail_at', "ALTER TABLE videos ADD COLUMN last_fail_at TEXT"), ]: if col not in cols: c.execute(ddl) # 兼容旧库：events 表补 person_appearances_json（新架构 v3 加） ev_cols = [r[1] for r in c.execute("PRAGMA table_info(events)").fetchall()] if 'person_appearances_json' not in ev_cols: c.execute("ALTER TABLE events ADD COLUMN person_appearances_json TEXT") # 兼容旧库：people 表补 features_json / display_uid（新架构 v3 加） pe_cols = [r[1] for r in c.execute("PRAGMA table_info(people)").fetchall()] for col, ddl in [ ('features_json', "ALTER TABLE people ADD COLUMN features_json TEXT"), ('display_uid', "ALTER TABLE people ADD COLUMN display_uid TEXT"), ]: if col not in pe_cols: c.execute(ddl) self._conn.commit() # ------------------------------------------------------------------ # videos # ------------------------------------------------------------------ def get_video_by_filename(self, filename: str) -> Optional[sqlite3.Row]: cur = self._conn.execute("SELECT * FROM videos WHERE filename=?", (filename,)) return cur.fetchone() def get_video_by_id(self, video_id: int) -> Optional[sqlite3.Row]: cur = self._conn.execute("SELECT * FROM videos WHERE id=?", (video_id,)) return cur.fetchone() def record_model_call(self, provider: str, model: str, video_id, filename, started_at: str, duration_sec: float, success: bool, error: str = ''): """记录一次云端模型调用（前端统计成功/失败/耗时/失败原因）""" now = _now_iso() self._conn.execute( "INSERT INTO model_calls (provider, model, video_id, filename, " "started_at, duration_sec, success, error, created_at) " "VALUES (?,?,?,?,?,?,?,?,?)", (provider, model, video_id, filename, started_at, duration_sec, 1 if success else 0, error or '', now)) self._conn.commit() # ------------------------------------------------------------------ # 服务活动日志（实时状态界面用；只保留最近 7 天） # ------------------------------------------------------------------ def record_activity(self, service: str, action: str, detail: str = ''): """记录一条服务活动（queue/rclone/person/model...）。 写入时顺带清理 7 天前的旧记录（用户要求只保留最近七天）。 """ now = _now_iso() with self._write_lock: self._conn.execute( "INSERT INTO service_activity (service, action, detail, ts) " "VALUES (?,?,?,?)", (service, action, str(detail or '')[:500], now)) # 只保留最近 7 天 self._conn.execute( "DELETE FROM service_activity WHERE ts < ?", ((datetime.now(timezone(timedelta(hours=8))) - timedelta(days=7)) .strftime('%Y-%m-%d %H:%M:%S'),)) self._conn.commit() def get_recent_activities(self, limit: int = 50) -> List[Dict]: """最近活动（时间倒序）。""" rows = self._conn.execute( "SELECT id, service, action, detail, ts FROM service_activity " "ORDER BY id DESC LIMIT ?", (int(limit),)).fetchall() return [dict(r) for r in rows] def get_queue_status(self) -> Dict: """实时队列/处理状态（前端服务状态卡用）。""" total = self._conn.execute("SELECT COUNT(*) c FROM videos").fetchone()['c'] by_status = {} for r in self._conn.execute( "SELECT status, COUNT(*) c FROM videos GROUP BY status").fetchall(): by_status[r['status']] = r['c'] # 最近处理的视频（done/failed，按 updated_at 倒序） recent = self._conn.execute( "SELECT id, filename, status, compute_provider, updated_at, " "processed_at, event_start_time FROM videos " "ORDER BY COALESCE(updated_at, created_at) DESC LIMIT 5" ).fetchall() return { "total": total, "by_status": by_status, "recent": [dict(r) for r in recent], } def set_video_file_status(self, video_id: int, valid: bool, error: str = '', media_meta: dict = None): """登记/更新文件校验结果：valid / file_error / media_meta_json""" now = _now_iso() self._conn.execute( "UPDATE videos SET file_valid=?, file_error=?, media_meta_json=?, " "updated_at=? WHERE id=?", (1 if valid else 0, error or '', json.dumps(media_meta, ensure_ascii=False) if media_meta else None, now, video_id)) self._conn.commit() def ensure_video(self, filename: str, local_path: str, camera_name: str = '', event_start_time: str = '', duration_sec: float = 0.0, drive_file_id: str = '') -> int: """视频进入监听目录时登记；已存在则更新路径。返回 video_id。""" now = _now_iso() row = self.get_video_by_filename(filename) if row: self._conn.execute( "UPDATE videos SET local_path=?, camera_name=?, event_start_time=?, " "duration_sec=?, updated_at=? WHERE id=?", (local_path, camera_name, event_start_time, duration_sec, now, row['id'])) self._conn.commit() return row['id'] cur = self._conn.execute( "INSERT INTO videos (drive_file_id, filename, local_path, camera_name, " "duration_sec, event_start_time, status, created_at, updated_at) " "VALUES (?,?,?,?,?,?, 'pending', ?, ?)", (drive_file_id, filename, local_path, camera_name, duration_sec, event_start_time, now, now)) self._conn.commit() return cur.lastrowid def get_pending_videos(self, limit: int = 1) -> List[sqlite3.Row]: cur = self._conn.execute( "SELECT * FROM videos WHERE status IN ('pending','failed') " "ORDER BY id ASC LIMIT ?", (limit,)) return cur.fetchall() def mark_video_processed(self, video_id: int, summary: str, events: List[dict], people: List[str], compute_provider: str) -> List[int]: """落库视频结果；返回新插入事件的 id 列表（与 events 参数一一对应）。 events 内每条可含 person_appearances（[{uid, features, action}]）， 原样存到 events.person_appearances_json，供 person_service 聚合特征。 """ with self._write_lock: now = _now_iso() self._conn.execute( "UPDATE videos SET status='done', summary_json=?, events_json=?, " "people_json=?, compute_provider=?, updated_at=?, processed_at=? WHERE id=?", (summary, json.dumps(events, ensure_ascii=False), json.dumps(people, ensure_ascii=False), compute_provider, now, now, video_id)) # 事件落独立表，便于 NAS 拉取 self._conn.execute("DELETE FROM events WHERE video_id=?", (video_id,)) event_ids: List[int] = [] for ev in events: pa = ev.get('person_appearances') cur = self._conn.execute( "INSERT INTO events (video_id, ts, description, person_list_json, " "person_appearances_json, is_attention_event) VALUES (?,?,?,?,?,?)", (video_id, ev.get('timestamp', ''), ev.get('description', ''), json.dumps(ev.get('people', []), ensure_ascii=False), json.dumps(pa, ensure_ascii=False) if pa else None, 1 if ev.get('is_attention_event') else 0)) event_ids.append(cur.lastrowid) self._conn.commit() return event_ids def mark_video_failed(self, video_id: int, error: str = ''): now = _now_iso() self._conn.execute( "UPDATE videos SET status='failed', retry_count=retry_count+1, " "summary_json=?, updated_at=?, last_fail_at=? WHERE id=?", (error, now, now, video_id)) self._conn.commit() def set_event_start_time(self, video_id: int, event_start_time: str): """回填从文件名解析出的视频开始时间（补录/纠偏用）。""" self._conn.execute( "UPDATE videos SET event_start_time=?, updated_at=? WHERE id=?", (event_start_time, _now_iso(), video_id)) self._conn.commit() def mark_video_invalid(self, video_id: int, error: str = ''): """文件校验不通过（损坏/非视频等），标记 invalid，producer 不再重试。""" now = _now_iso() self._conn.execute( "UPDATE videos SET status='invalid', file_valid=0, file_error=?, " "updated_at=? WHERE id=?", (error or '', now, video_id)) self._conn.commit() def reset_video_to_pending(self, video_id: int): """文件被重新同步覆盖（mtime 变化）时，清掉旧分析结果重新排队处理。""" now = _now_iso() self._conn.execute( "UPDATE videos SET status='pending', retry_count=0, summary_json=NULL, " "events_json=NULL, people_json=NULL, compute_provider=NULL, " "processed_at=NULL, file_valid=1, updated_at=? WHERE id=?", (now, video_id)) self._conn.commit() def get_first_event_for_label(self, label: str): """找到某人物（canonical_name 或 UID label）最早一次出现的事件。 返回 dict{video_id, ts, features_text, event_start_time} 或 None。 features_text 从该事件 person_appearances_json 中对应 uid 的特征拼出， 供 frame_service 用大模型在画面中定位该人物。 """ # canonical_name -> 其下所有 label；否则按 label 本身匹配 rows = self._conn.execute( "SELECT label FROM people WHERE canonical_name=?", (label,)).fetchall() labels = {r['label'] for r in rows} if rows else {label} best = None for lb in labels: pattern = f'%{lb}%' row = self._conn.execute( """SELECT e.video_id, e.ts, e.person_appearances_json, v.event_start_time FROM events e JOIN videos v ON v.id=e.video_id WHERE v.status='done' AND (e.person_list_json LIKE ? OR e.person_appearances_json LIKE ?) ORDER BY e.ts ASC LIMIT 1""", (pattern, pattern)).fetchone() if row and row['video_id'] and \ (best is None or (row['ts'] or '') < (best['ts'] or '')): best = row if not best: return None features_text = '' try: pa = json.loads(best['person_appearances_json'] or '[]') except (ValueError, TypeError): pa = [] if isinstance(pa, list): for p in pa: uid = str((p.get('uid') or '')).strip() if uid and uid in labels and isinstance(p.get('features'), dict): bits = [str(v) for v in p['features'].values() if v and str(v).strip().lower() != 'unknown'] if bits: features_text = '，'.join(bits) break return {'video_id': best['video_id'], 'ts': best['ts'], 'features_text': features_text, 'event_start_time': best['event_start_time']} def get_events_for_label(self, label: str, limit: int = 6): """该人物（canonical_name 或 UID label）出现的候选事件，按时间倒序（最近优先）。 返回 [{video_id, ts, features_text, bbox}]（dict 列表）：features_text 是该 事件中该人物的结构化特征文本；bbox 是视频分析时随该人物一并产出的包围框 （[ymin,xmin,ymax,xmax]，0-1000 归一化，取不到为 None）——frame_service 直接 用它做头像裁剪，不再额外调用模型定位。取最近的事件而不是最早的：bbox 是新 加的字段，老事件普遍没有，最近优先能更快用上新数据，也更能反映人物当前样貌。 """ rows = self._conn.execute( "SELECT label FROM people WHERE canonical_name=?", (label,)).fetchall() labels = {r['label'] for r in rows} | {label} seen = set() out = [] for lb in labels: pattern = f'%{lb}%' rs = self._conn.execute( """SELECT e.video_id, e.ts, e.person_appearances_json, v.event_start_time FROM events e JOIN videos v ON v.id=e.video_id WHERE v.status='done' AND (e.person_list_json LIKE ? OR e.person_appearances_json LIKE ?) -- 排除历史遗留的畸形 ts（如缺日期的 "26:21"）：这类值既不能 -- 正确排序（字符串比较会排到最前面），extract_frame 也没法从 -- 中算出正确偏移，只会抽到视频开头的错误画面 AND e.ts GLOB '[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9] [0-9][0-9]:[0-9][0-9]:[0-9][0-9]' ORDER BY e.ts DESC LIMIT ?""", (pattern, pattern, limit)).fetchall() for r in rs: key = (r['video_id'], r['ts']) if key in seen: continue seen.add(key) d = dict(r) d['features_text'] = self._features_text_for( d.get('person_appearances_json'), labels) d['bbox'] = self._bbox_for_uids(d.get('person_appearances_json'), labels) out.append(d) out.sort(key=lambda r: (r['ts'] or ''), reverse=True) return out[:limit] @staticmethod def _bbox_for_uids(pa_json, uids): """从 person_appearances_json 里取属于 uids 身份组那个人物的 bbox （[ymin,xmin,ymax,xmax]，0-1000 归一化）。bbox 随视频分析一次性产出， 取不到/非法一律返回 None（调用方退回整帧兜底，不再额外调用模型定位）。""" try: pa = json.loads(pa_json or '[]') except (ValueError, TypeError): return None if not isinstance(pa, list): return None for p in pa: if not isinstance(p, dict): continue p_uid = re.sub(r'[（(][^()（）]*[)）]', '', str(p.get('uid', ''))).strip() if p_uid not in uids: continue bbox = p.get('bbox') if isinstance(bbox, list) and len(bbox) == 4: try: return [float(v) for v in bbox] except (TypeError, ValueError): return None return None @staticmethod def _features_text_for(pa_json, uids) -> str: """从 person_appearances_json 提取属于 uids 身份组的人物特征文本。 uid 先剥离括号再匹配（历史事件里存在 '人物A（别名：人物B）' 这类原始输出）， 只取该组人物的特征，避免把同帧其他人的特征混进头像定位 prompt。 """ try: pa = json.loads(pa_json or '[]') except (ValueError, TypeError): return '' if not isinstance(pa, list): return '' bits = [] for p in pa: if not isinstance(p, dict): continue uid = re.sub(r'[（(][^()（）]*[)）]', '', str(p.get('uid', ''))).strip() if uid not in uids or not isinstance(p.get('features'), dict): continue for v in p['features'].values(): if v and str(v).strip().lower() != 'unknown': bits.append(str(v)) return '，'.join(bits) def get_all_videos(self) -> List[sqlite3.Row]: return self._conn.execute( "SELECT * FROM videos WHERE status='done' ORDER BY id ASC").fetchall() # ------------------------------------------------------------------ # people # ------------------------------------------------------------------ def upsert_person(self, label: str, canonical_name: str = '', source: str = 'llm', first_seen: str = '', features: dict = None, display_uid: str = ''): """登记/更新人物。 features: 该人物的结构化特征 dict（gender/age_band/build/hair/clothing/face/ distinguishing）。与已有 features_json 合并（已有非 unknown 字段不被 覆盖，新非 unknown 字段补齐）。None 时不更新特征列。 display_uid: 大模型给的人物 UID（如 "人物A"）。label 本身就是 UID 时可省略。 """ # 剥离括号后缀（如 '人物A（别名/标识：人物B）' -> '人物A'），防止大模型 # 带备注的原始输出分裂出垃圾人物行 label = re.sub(r'[（(][^()（）]*[)）]', '', str(label)).strip() or str(label) now = _now_iso() row = self._conn.execute("SELECT * FROM people WHERE label=?", (label,)).fetchone() # 特征合并（在已有 features_json 基础上） merged_features = self._merge_features( row['features_json'] if row else None, features) if row else ( self._merge_features(None, features)) if row: # manual 覆盖 llm；llm 不覆盖 manual if source == 'manual' or row['source'] != 'manual': self._conn.execute( "UPDATE people SET canonical_name=?, source=?, appearances=appearances+1, " "features_json=?, display_uid=?, updated_at=? WHERE label=?", (canonical_name or row['canonical_name'], source, merged_features, display_uid or row['display_uid'] or label, now, label)) else: self._conn.execute( "UPDATE people SET appearances=appearances+1, features_json=?, " "display_uid=?, updated_at=? WHERE label=?", (merged_features, display_uid or row['display_uid'] or label, now, label)) else: self._conn.execute( "INSERT INTO people (label, canonical_name, first_seen, appearances, " "source, features_json, display_uid, updated_at) " "VALUES (?,?,?,1,?,?,?,?)", (label, canonical_name, first_seen or now, source, merged_features, display_uid or label, now)) self._conn.commit() @staticmethod def _merge_features(old_json: Optional[str], new_features: Optional[dict]) -> Optional[str]: """合并人物特征：已有非 unknown 字段不被覆盖；新字段在 old 为空/unknown 时补齐。 - old_json 为 None / 空 -> 直接用 new_features - new_features 为 None / 空 -> 不变 - 字段级：new 值非 'unknown' 且非空时覆盖 old（old 为 unknown/空）； new 值为 'unknown' 时保留 old（哪怕 old 也是 unknown） """ if not new_features: return old_json try: old = json.loads(old_json) if old_json else {} except (ValueError, TypeError): old = {} if not isinstance(old, dict): old = {} merged = dict(old) for k, v in new_features.items(): v_str = str(v).strip() if v is not None else '' if v_str and v_str.lower() != 'unknown': # 新值是有效特征，覆盖（无论 old 是什么） merged[k] = v_str elif k not in merged: # 新值 unknown 且 old 没该字段，至少把字段占位（写 unknown） merged[k] = v_str or 'unknown' return json.dumps(merged, ensure_ascii=False) def set_canonical(self, label: str, canonical_name: str, source: str = 'manual'): """设置规范名（label 可视为别名）。source 透传：llm 的可被后续纠正，manual 优先。""" self.upsert_person(label, canonical_name, source=source) def set_person_appearances(self, label: str, count: int, source: str = 'llm'): """覆盖设置出现次数（reconcile 时用 distinct 视频数校准，避免累加膨胀）。""" label = re.sub(r'[（(][^()（）]*[)）]', '', str(label)).strip() or str(label) now = _now_iso() row = self._conn.execute("SELECT * FROM people WHERE label=?", (label,)).fetchone() if row: if source == 'manual' or row['source'] != 'manual': self._conn.execute( "UPDATE people SET appearances=?, source=?, updated_at=? WHERE label=?", (int(count), source, now, label)) else: self._conn.execute( "UPDATE people SET appearances=?, updated_at=? WHERE label=?", (int(count), now, label)) else: self._conn.execute( "INSERT INTO people (label, canonical_name, first_seen, appearances, " "source, updated_at) VALUES (?,?,?,?,?,?)", (label, '', now, int(count), source, now)) self._conn.commit() def get_people(self) -> List[sqlite3.Row]: return self._conn.execute("SELECT * FROM people ORDER BY id ASC").fetchall() def get_known_members_context(self) -> str: """生成 known_members_context 文本，注入视频提示让模型用真名。""" rows = self.get_people() lines = [] for r in rows: name = r['canonical_name'] or r['label'] if name and name != r['label']: lines.append(f"- {name}（别名/标识：{r['label']}）") else: lines.append(f"- {name}") return '\n'.join(lines) if lines else '' # ------------------------------------------------------------------ # 同步导出（供 NAS 拉取） # ------------------------------------------------------------------ def get_sync_delta(self, since_iso: str) -> Dict: """返回 since 之后变更的 videos / events / people。""" videos = self._conn.execute( "SELECT * FROM videos WHERE updated_at > ? ORDER BY id ASC", (since_iso,) ).fetchall() # events 表本身没有 updated_at 列，变更判断借用所属 video 的 updated_at events = self._conn.execute( "SELECT e.* FROM events e JOIN videos v ON e.video_id=v.id " "WHERE v.updated_at > ? ORDER BY e.id ASC", (since_iso,)).fetchall() people = self._conn.execute( "SELECT * FROM people WHERE updated_at > ? ORDER BY id ASC", (since_iso,) ).fetchall() # 模型调用统计（created_at >= since 配合 NAS 端幂等 upsert 防漏同秒记录） model_calls = self._conn.execute( "SELECT * FROM model_calls WHERE created_at >= ? ORDER BY id ASC", (since_iso,)).fetchall() def _ser(row): d = dict(row) return d return { "videos": [_ser(v) for v in videos], "events": [_ser(e) for e in events], "people": [_ser(p) for p in people], "model_calls": [_ser(m) for m in model_calls], "server_time": _now_iso(), } # ------------------------------------------------------------------ # 同步游标 # ------------------------------------------------------------------ def get_cursor(self, key: str) -> str: row = self._conn.execute("SELECT value FROM sync_cursor WHERE key=?", (key,)).fetchone() return row['value'] if row else '' def set_cursor(self, key: str, value: str): self._conn.execute( "INSERT INTO sync_cursor (key, value) VALUES (?, ?) " "ON CONFLICT(key) DO UPDATE SET value=excluded.value", (key, value)) self._conn.commit() def close(self): self._conn.close()