diff --git a/fam-core/src/fam_core/oracle_sync.py b/fam-core/src/fam_core/oracle_sync.py index 90e6af1..4e921df 100644 --- a/fam-core/src/fam_core/oracle_sync.py +++ b/fam-core/src/fam_core/oracle_sync.py @@ -49,10 +49,16 @@ class OracleSync: self._last_sync_at = None self._last_error = None self._last_count = None + # 拉取互斥锁:trigger_now(命名后即时拉回)与后台 _run 并发时只允许一个执行 + self._pull_lock = threading.Lock() # ------------------------------------------------------------------ def _pull_once(self) -> bool: - """执行一次增量拉取。返回是否成功。""" + """执行一次增量拉取(加锁防并发双拉)。返回是否成功。""" + with self._pull_lock: + return self._pull_once_locked() + + def _pull_once_locked(self) -> bool: since = db_layer.get_sync_cursor() or '' params = {'since': since, 'token': self.token} try: diff --git a/fam-edge/config/config.yaml b/fam-edge/config/config.yaml index 89b4fa2..fce0faf 100644 --- a/fam-edge/config/config.yaml +++ b/fam-edge/config/config.yaml @@ -26,9 +26,9 @@ gdrive_sync: oracle_db: path: "/opt/fam-edge/data/oracle.db" -# NAS 拉取同步接口鉴权 token(与 NAS oracle_sync.token 一致) +# NAS 拉取同步接口鉴权 token(与 NAS oracle_sync.token 一致;明文直配,不再依赖 .env) sync_api: - token: "${ORACLE_SYNC_TOKEN}" + token: "MLH92wv5jSDdQtHfcWJgKt-YaStn3IttjlrYxW0DwXA" # 人物识别服务 person_service: @@ -56,7 +56,7 @@ models: model_name: "gemini-flash-latest" # 主模型(每日免费配额 20 请求,按模型独立) fallback_models: - "gemini-flash-lite-latest" - api_key: "${GEMINI_API_KEY}" + api_key: "AQ.Ab8RN6I0l8hC7hLnNHRY6qOXdch5CTWczDNlS4c1XrneGHipUQ" timeout: 600 # 模型级独立超时(最终值,不参与编排层 ×2 放大) # gemini-flash-lite 实测 ~22-34s,按用户要求放宽至 8 分钟(480s),避免大视频/排队时过早切断 @@ -79,7 +79,7 @@ models: - "nvidia/nemotron-nano-12b-v2-vl" - "meta/llama-3.2-11b-vision-instruct" base_url: "https://integrate.api.nvidia.com/v1" - api_key: "${NVIDIA_API_KEY}" + api_key: "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" timeout: 600 switch_interval_sec: 5 # 模型切换间隔:一个失败后等待再试下一个 model_timeouts: # 模型级独立超时(最终值,不参与 ×2) diff --git a/fam-edge/src/fam_edge/oracle_db.py b/fam-edge/src/fam_edge/oracle_db.py index c169b8e..e99c75e 100644 --- a/fam-edge/src/fam_edge/oracle_db.py +++ b/fam-edge/src/fam_edge/oracle_db.py @@ -243,6 +243,26 @@ class OracleDB: """手动命名:设置规范名(label 可视为别名)。""" self.upsert_person(label, canonical_name, source='manual') + def set_person_appearances(self, label: str, count: int, source: str = 'llm'): + """覆盖设置出现次数(reconcile 时用 distinct 视频数校准,避免累加膨胀)。""" + 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() diff --git a/fam-edge/src/fam_edge/person_service.py b/fam-edge/src/fam_edge/person_service.py index b0797d6..3f9e8e3 100644 --- a/fam-edge/src/fam_edge/person_service.py +++ b/fam-edge/src/fam_edge/person_service.py @@ -43,7 +43,8 @@ class PersonService: # ------------------------------------------------------------------ def reconcile(self): """汇总 + LLM 合并一次。可由定时或手动触发。""" - # 1. 先把所有视频的 people_mentioned 同步进 people 表(标签级) + # 1. 统计每个人物标签出现过的视频数(去重),校准 appearances(防每次 reconcile 累加膨胀) + label_videos: Dict[str, set] = {} for v in self.db.get_all_videos(): try: people = json.loads(v['people_json'] or '[]') @@ -51,7 +52,9 @@ class PersonService: people = [] for p in people: if p and p not in ('无人', '无'): - self.db.upsert_person(p, source='llm') + label_videos.setdefault(p, set()).add(v['id']) + for label, vids in label_videos.items(): + self.db.set_person_appearances(label, len(vids), source='llm') # 2. 收集未命名(无 canonical 或 canonical==label)的标签 + 描述样本 rows = self.db.get_people() @@ -66,12 +69,17 @@ class PersonService: if not mapping: return + # 3. 落库:canonical 若是另一个 label(target),解析为其已有 canonical,保证同一身份统一显示名 + label_to_canonical = {r['label']: (r['canonical_name'] or r['label']) for r in rows} + updated = 0 for label, canonical in mapping.items(): if label in manual: continue # 手动命名优先 if canonical and canonical != label: - self.db.set_canonical(label, canonical, source='llm') - logger.info(f"PersonService: LLM 合并完成,更新 {len(mapping)} 条") + resolved = label_to_canonical.get(canonical, canonical) + self.db.set_canonical(label, resolved, source='llm') + updated += 1 + logger.info(f"PersonService: LLM 合并完成,更新 {updated} 条") def _collect_descriptions(self, labels: List[str]) -> Dict[str, List[str]]: """从 events 表收集每个标签出现时的描述样本。"""