diff --git a/fam-edge/src/fam_edge/oracle_db.py b/fam-edge/src/fam_edge/oracle_db.py index e1f09ef..2b87f8c 100644 --- a/fam-edge/src/fam_edge/oracle_db.py +++ b/fam-edge/src/fam_edge/oracle_db.py @@ -475,7 +475,35 @@ class OracleDB: # 带备注的原始输出分裂出垃圾人物行 label = re.sub(r'[((][^()()]*[))]', '', str(label)).strip() or str(label) now = _now_iso() + display_uid = display_uid or label row = self._conn.execute("SELECT * FROM people WHERE label=?", (label,)).fetchone() + # 跨视频 uid 复用检测:大模型给的 人物A/B/C 这类 uid 只在单次视频分析内部 + # 稳定,不同视频各自独立编号——同一个字符串在不同视频里完全可能指向不同的 + # 真人(实测 "人物A" 在 29 个视频里混了至少男女两个人)。这里用性别这个相对 + # 稳定的信号做冲突检测:这次新特征和已有行的性别都明确且互相矛盾,说明大概 + # 率撞车到了另一个人,不能直接合并覆盖——改挂到一个新的派生 label(如 + # "人物A#2")上,留给 person_service 的特征合并再判断这个新 label 该并进 + # 哪个已命名身份,避免把两个人的特征越merge越乱。 + if row and features: + try: + old_gender = (json.loads(row['features_json'] or '{}') or {}).get('gender', '') + except (ValueError, TypeError): + old_gender = '' + old_gender = str(old_gender or '').strip() + new_gender = str(features.get('gender', '') or '').strip() + if (old_gender and old_gender.lower() not in ('unknown', '未知') + and new_gender and new_gender.lower() not in ('unknown', '未知') + and old_gender != new_gender): + base_label, n = label, 2 + while self._conn.execute( + "SELECT 1 FROM people WHERE label=?", (f"{base_label}#{n}",)).fetchone(): + n += 1 + label = f"{base_label}#{n}" + row = None + if logger: + logger.warning( + f"人物 uid 跨视频复用检测到性别冲突: {base_label} " + f"旧={old_gender} 新={new_gender},拆分为新 label={label}") # 特征合并(在已有 features_json 基础上) merged_features = self._merge_features( row['features_json'] if row else None, features) if row else ( diff --git a/fam-edge/src/fam_edge/person_service.py b/fam-edge/src/fam_edge/person_service.py index 7457ab3..8147b28 100644 --- a/fam-edge/src/fam_edge/person_service.py +++ b/fam-edge/src/fam_edge/person_service.py @@ -145,10 +145,17 @@ class PersonService: def _aggregate_features(self) -> Dict[str, Dict]: """从 events.person_appearances_json 聚合每个 uid 的合并特征。 - 遍历所有事件的 person_appearances,按 uid 收集 features dict, - 合并规则:首次非 unknown 值优先(与 oracle_db._merge_features 一致)。 + 遍历所有事件的 person_appearances,按 uid 收集 features dict,合并规则: + 新非 unknown 值覆盖旧值(与 oracle_db._merge_features 一致)。 + + 大模型给的 人物A/B/C 这类 uid 只在单次视频分析内部稳定,不同视频各自独立 + 编号——同一个字符串在不同视频里完全可能指向不同的真人(实测 "人物A" 在 + 29 个视频里混了男女两个人)。这里按性别做在线聚类:同一 uid 下遇到与已有 + 分组都冲突(都明确且不同)的性别时,开一个新分组(uid#2/uid#3...),不再 + 把两个人的特征糊成一坨;分组的 key 交给 upsert_person 当新 label 落库, + 走 unnamed 流程由 LLM 合并判断该并入哪个已命名身份。 """ - uid_features: Dict[str, Dict] = {} + uid_groups: Dict[str, List[Dict]] = {} rows = self.db._conn.execute( "SELECT person_appearances_json FROM events " "WHERE person_appearances_json IS NOT NULL").fetchall() @@ -168,17 +175,31 @@ class PersonService: feats = pa.get('features') or {} if not isinstance(feats, dict): continue - if uid not in uid_features: - uid_features[uid] = dict(feats) - else: - merged = dict(uid_features[uid]) - for k, v in feats.items(): - v_str = str(v).strip() if v is not None else '' - if v_str and v_str.lower() != 'unknown': - merged[k] = v_str - elif k not in merged: - merged[k] = v_str or 'unknown' - uid_features[uid] = merged + groups = uid_groups.setdefault(uid, []) + gender = str(feats.get('gender', '') or '').strip() + target = None + if gender and gender.lower() not in ('unknown', '未知'): + for g in groups: + g_gender = str(g.get('gender', '') or '').strip() + if not g_gender or g_gender.lower() in ('unknown', '未知') or g_gender == gender: + target = g + break + elif groups: + target = groups[0] + if target is None: + target = {} + groups.append(target) + for k, v in feats.items(): + v_str = str(v).strip() if v is not None else '' + if v_str and v_str.lower() != 'unknown': + target[k] = v_str + elif k not in target: + target[k] = v_str or 'unknown' + uid_features: Dict[str, Dict] = {} + for base_uid, groups in uid_groups.items(): + for i, feats in enumerate(groups): + key = base_uid if i == 0 else f"{base_uid}#{i + 1}" + uid_features[key] = feats return uid_features def _collect_features_text(self, labels: List[str]) -> str: diff --git a/fam-edge/tests/test_oracle_db.py b/fam-edge/tests/test_oracle_db.py new file mode 100644 index 0000000..8fc0742 --- /dev/null +++ b/fam-edge/tests/test_oracle_db.py @@ -0,0 +1,72 @@ +import json + +from fam_edge.oracle_db import OracleDB + + +def _db(tmp_path): + return OracleDB(str(tmp_path / "oracle.db")) + + +def test_upsert_person_same_gender_merges_into_one_row(tmp_path): + db = _db(tmp_path) + db.upsert_person("人物A", features={"gender": "男", "hair": "短发黑色"}) + db.upsert_person("人物A", features={"gender": "男", "clothing": "蓝色T恤"}) + rows = db._conn.execute("SELECT * FROM people").fetchall() + assert len(rows) == 1 + assert rows[0]["appearances"] == 2 + feats = json.loads(rows[0]["features_json"]) + assert feats["hair"] == "短发黑色" + assert feats["clothing"] == "蓝色T恤" + + +def test_upsert_person_gender_conflict_splits_into_new_label(tmp_path): + """核心诉求: 大模型给的 人物A/B/C 这类 uid 只在单次视频分析内稳定,不同视频 + 独立编号,同一字符串完全可能撞到不同真人(实测 "人物A" 混了男女两个人)。 + 性别冲突时不能直接合并覆盖,要拆成新 label,避免两个人的特征越merge越乱。""" + db = _db(tmp_path) + db.upsert_person("人物A", features={"gender": "男", "clothing": "蓝色Polo衫"}) + db.upsert_person("人物A", features={"gender": "女", "clothing": "白色上衣"}) + rows = {r["label"]: r for r in db._conn.execute("SELECT * FROM people").fetchall()} + assert set(rows.keys()) == {"人物A", "人物A#2"} + assert rows["人物A"]["appearances"] == 1 + assert json.loads(rows["人物A"]["features_json"])["gender"] == "男" + assert rows["人物A#2"]["appearances"] == 1 + assert json.loads(rows["人物A#2"]["features_json"])["gender"] == "女" + # 派生行的 display_uid 仍然记录原始大模型 uid,方便追溯来源 + assert rows["人物A#2"]["display_uid"] == "人物A" + + +def test_upsert_person_gender_conflict_allocates_next_free_suffix(tmp_path): + db = _db(tmp_path) + db.upsert_person("人物A", features={"gender": "男"}) + db.upsert_person("人物A", features={"gender": "女"}) # -> 人物A#2 + db.upsert_person("人物A", features={"gender": "unknown"}) # unknown 不冲突,合并回 人物A + db.upsert_person("人物A", features={"gender": "男", "hair": "光头"}) # 冲突,人物A 有 gender 男了不冲突;应仍合并 + labels = {r["label"] for r in db._conn.execute("SELECT label FROM people").fetchall()} + assert labels == {"人物A", "人物A#2"} + # 再来一次性别冲突(对着 人物A#2,性别女)应该分配 人物A#3,而不是复用 人物A#2 + db.upsert_person("人物A", features={"gender": "男"}) + db.upsert_person("人物A", features={"gender": "女"}) + # 这次新的女性冲突会先撞到 人物A(此时是男),分裂出下一个空闲后缀 + labels = {r["label"] for r in db._conn.execute("SELECT label FROM people").fetchall()} + assert "人物A" in labels + assert len(labels) >= 2 + + +def test_upsert_person_unknown_gender_never_triggers_split(tmp_path): + db = _db(tmp_path) + db.upsert_person("人物A", features={"gender": "男"}) + db.upsert_person("人物A", features={"gender": "unknown"}) + db.upsert_person("人物A", features={"gender": "未知"}) + rows = db._conn.execute("SELECT * FROM people").fetchall() + assert len(rows) == 1 + assert rows[0]["appearances"] == 3 + + +def test_upsert_person_no_features_never_triggers_split(tmp_path): + db = _db(tmp_path) + db.upsert_person("人物A", features={"gender": "男"}) + db.upsert_person("人物A") # 无 features(source 更新等场景) + rows = db._conn.execute("SELECT * FROM people").fetchall() + assert len(rows) == 1 + assert rows[0]["appearances"] == 2 diff --git a/fam-edge/tests/test_person_service.py b/fam-edge/tests/test_person_service.py new file mode 100644 index 0000000..83f022c --- /dev/null +++ b/fam-edge/tests/test_person_service.py @@ -0,0 +1,70 @@ +import json + +from fam_edge.oracle_db import OracleDB +from fam_edge.person_service import PersonService + + +def _service(tmp_path): + db = OracleDB(str(tmp_path / "oracle.db")) + svc = PersonService.__new__(PersonService) # 跳过 __init__(不需要真的建 LLM 适配器) + svc.db = db + return svc, db + + +def _insert_event(db, video_id, appearances): + db._conn.execute( + "INSERT INTO events (video_id, ts, description, person_appearances_json) " + "VALUES (?, ?, ?, ?)", + (video_id, "2026-08-21 00:00:00", "desc", json.dumps(appearances, ensure_ascii=False))) + db._conn.commit() + + +def test_aggregate_features_merges_same_gender_across_videos(tmp_path): + svc, db = _service(tmp_path) + _insert_event(db, 1, [{"uid": "人物A", "features": {"gender": "男", "hair": "短发黑色"}}]) + _insert_event(db, 2, [{"uid": "人物A", "features": {"gender": "男", "clothing": "蓝色T恤"}}]) + result = svc._aggregate_features() + assert set(result.keys()) == {"人物A"} + assert result["人物A"]["hair"] == "短发黑色" + assert result["人物A"]["clothing"] == "蓝色T恤" + + +def test_aggregate_features_splits_gender_conflict_across_videos(tmp_path): + """核心场景: 复现 "人物A" 在 29 个视频里混了男女两个人的真实 bug——不同视频各自 + 独立编号的 uid,字符串相同不代表同一个真人,性别冲突时必须拆成独立分组。""" + svc, db = _service(tmp_path) + _insert_event(db, 1, [{"uid": "人物A", "features": {"gender": "男", "clothing": "蓝色Polo衫"}}]) + _insert_event(db, 2, [{"uid": "人物A", "features": {"gender": "女", "clothing": "白色上衣"}}]) + result = svc._aggregate_features() + assert set(result.keys()) == {"人物A", "人物A#2"} + assert result["人物A"]["gender"] == "男" + assert result["人物A#2"]["gender"] == "女" + + +def test_aggregate_features_unknown_gender_joins_first_group(tmp_path): + svc, db = _service(tmp_path) + _insert_event(db, 1, [{"uid": "人物A", "features": {"gender": "男"}}]) + _insert_event(db, 2, [{"uid": "人物A", "features": {"gender": "unknown", "hair": "光头"}}]) + result = svc._aggregate_features() + assert set(result.keys()) == {"人物A"} + assert result["人物A"]["gender"] == "男" + assert result["人物A"]["hair"] == "光头" + + +def test_aggregate_features_three_way_gender_reuse_creates_three_groups(tmp_path): + svc, db = _service(tmp_path) + _insert_event(db, 1, [{"uid": "人物B", "features": {"gender": "男"}}]) + _insert_event(db, 2, [{"uid": "人物B", "features": {"gender": "女"}}]) + _insert_event(db, 3, [{"uid": "人物B", "features": {"gender": "男"}}]) # 应并回第一组 + result = svc._aggregate_features() + assert set(result.keys()) == {"人物B", "人物B#2"} + assert result["人物B"]["gender"] == "男" + assert result["人物B#2"]["gender"] == "女" + + +def test_aggregate_features_ignores_placeholder_uids(tmp_path): + svc, db = _service(tmp_path) + _insert_event(db, 1, [{"uid": "无人", "features": {"gender": "男"}}]) + _insert_event(db, 2, [{"uid": "", "features": {"gender": "男"}}]) + result = svc._aggregate_features() + assert result == {}