diff --git a/fam-edge/src/fam_edge/person_service.py b/fam-edge/src/fam_edge/person_service.py index 77b0b08..7457ab3 100644 --- a/fam-edge/src/fam_edge/person_service.py +++ b/fam-edge/src/fam_edge/person_service.py @@ -25,6 +25,34 @@ from . import oracle_db logger = setup_logger('fam-edge.person_service') +# 年龄档分桶:未成年与成年是稳定特征,跨桶绝不可能是同一个人 +_AGE_CHILD = {'幼儿', '儿童', '婴儿', '小孩', '少年', '青少年'} +_AGE_ADULT = {'青年', '中年', '老年', '成年'} + + +def _age_bucket(v: str) -> str: + if v in _AGE_CHILD: + return 'child' + if v in _AGE_ADULT: + return 'adult' + return '' + + +def _features_conflict(fa: dict, fb: dict) -> str: + """硬冲突检测:性别不同 / 年龄档跨未成年-成年的两个人绝不合并。 + + LLM 合并只作为提议,落库前必须通过本校验(弱模型会把 爷爷/宝宝 全并进 + 人物A 这类错误)。返回冲突原因,无冲突返回空串。 + """ + ga, gb = str(fa.get('gender', '')), str(fb.get('gender', '')) + if ga and gb and 'unknown' not in (ga.lower(), gb.lower()) and ga != gb: + return f'性别冲突 {ga} vs {gb}' + ba = _age_bucket(str(fa.get('age_band', ''))) + bb = _age_bucket(str(fb.get('age_band', ''))) + if ba and bb and ba != bb: + return f'年龄档冲突 {fa.get("age_band")} vs {fb.get("age_band")}' + return '' + class PersonService: def __init__(self, db: oracle_db.OracleDB): @@ -82,19 +110,37 @@ class PersonService: # 4. 落库:canonical 若是另一个 label(target),解析为其已有 canonical,保证同一身份统一显示名 label_to_canonical = {r['label']: (r['canonical_name'] or r['label']) for r in rows} - updated = 0 + + def _parse_feats(js): + try: + f = json.loads(js or '{}') + return f if isinstance(f, dict) else {} + except (ValueError, TypeError): + return {} + + feat_map = {r['label']: _parse_feats(r['features_json']) for r in rows} + updated = blocked = 0 for label, canonical in mapping.items(): if label in manual: continue # 手动命名优先 if canonical and canonical != label: resolved = label_to_canonical.get(canonical, canonical) + conflict = _features_conflict( + feat_map.get(label, {}), feat_map.get(resolved, {})) + if conflict: + # 硬规则否决 LLM 提议:性别/年龄档冲突的绝不可能是同一人 + logger.warning( + f"PersonService: 否决 LLM 合并 {label} -> {resolved}({conflict})") + blocked += 1 + continue self.db.set_canonical(label, resolved, source='llm') updated += 1 self.db.record_activity( 'person', 'merge_done', f"校准 {len(label_videos)} 标签 + 聚合 {len(uid_features)} 特征," - f"LLM 合并更新 {updated} 条({', '.join(list(mapping)[:6])})") - logger.info(f"PersonService: LLM 合并完成,更新 {updated} 条") + f"LLM 合并更新 {updated} 条 / 硬规则否决 {blocked} 条" + f"({', '.join(list(mapping)[:6])})") + logger.info(f"PersonService: LLM 合并完成,更新 {updated} 条,硬规则否决 {blocked} 条") def _aggregate_features(self) -> Dict[str, Dict]: """从 events.person_appearances_json 聚合每个 uid 的合并特征。