jira:NYJ-1540 desc: add api.py

@@ -193,6 +193,7 @@ files/
 cache_data/
 data/
 test/
+tests/
 # Media files
 *.jpg
 *.jpeg

+input_kafka:
+  servers: prod-kafka:9092
+  group_id: ai_match_service_prod
+  topic: football_replay_match_req
+  # username: xxx
+  # password: xxx
+
+output_kafka:
+  servers: prod-kafka:9092
+  topic: football_replay_match_resp
+
+cache_dir: /data/cache
+llm_base_url: http://prod-llm:11434
+llm_model: qwen3.6:35b-a3b-q8_0
+llm_temperature: 0.7

-app_name: "Football Replay Service"
-app_version: "1.0.0"
+input_kafka:
+  servers: 192.168.0.14:9092
+  group_id: ai_match_service
+  topic: football_replay_match_req
+  # username: xxx
+  # password: xxx
+
+output_kafka:
+  servers: 192.168.0.14:9092
+  topic: football_replay_match_resp
+
+
+llm:
+  base_url: http://192.168.1.59:11434
+  model_name: qwen3.6:35b-a3b-q8_0
+  temperature: 0.7
+
+common:
+  cache_dir: ./cache
\ No newline at end of file

-from aabd.base.cfg_loader import load_yaml_by_file_with_env
-from aabd.base.enhance_dict import EnhanceDict, read_prefixed_env_vars
+import os
 from pathlib import Path
 from logging import getLogger
+
+from aabd.base.cfg_loader import load_yaml_by_file_with_env
+from aabd.base.enhance_dict import EnhanceDict, read_prefixed_env_vars
+
 logger = getLogger(__name__)
 cfg_dir = Path(__file__).parent.absolute() / 'cfg'
-settings = EnhanceDict(load_yaml_by_file_with_env((cfg_dir / 'config.yaml').as_posix()))
+
+# 加载基础配置
+base_config = load_yaml_by_file_with_env((cfg_dir / 'config.yaml').as_posix())
+
+# 根据 ENV 环境变量加载环境特定配置并合并（如 config.dev.yaml / config.prod.yaml）
+env = os.getenv('ENV', 'dev')
+env_file = cfg_dir / f'config.{env}.yaml'
+if env_file.exists():
+    env_config = load_yaml_by_file_with_env(env_file.as_posix())
+    if isinstance(base_config, dict) and isinstance(env_config, dict):
+        for key, value in env_config.items():
+            if key in base_config and isinstance(base_config[key], dict) and isinstance(value, dict):
+                base_config[key].update(value)
+            else:
+                base_config[key] = value
+    logger.info(f"已加载环境配置: {env_file.name}")
+
+settings = EnhanceDict(base_config)
 settings.update(read_prefixed_env_vars('FRM_'))
 
 logger.info(f'Settings: {settings}')

-def replay_match_event(data):
-    pass
\ No newline at end of file
+import json
+import os
+from pathlib import Path
+
+try:
+    from .llm_image import Video2Frame, frames2content
+    from ..util.football_replay_match_live import FootballReplayMatchLive
+    from config import settings
+except ImportError:
+    import sys
+    sys.path.insert(0, str(Path(__file__).parent.parent))
+    from core.llm_image import Video2Frame, frames2content
+    from util.football_replay_match_live import FootballReplayMatchLive
+    from config import settings
+
+
+# 从配置中读取参数（配置项需在 config.yaml 或对应环境配置中定义）
+_cache_dir = settings.common.get('cache_dir', None)
+_llm_base_url = settings.llm.get('base_url', None)
+_llm_model = settings.llm.get('model_name', None)
+_llm_temperature = settings.llm.get('temperature', None)
+
+# 初始化视频处理与匹配器
+v2f = Video2Frame(_cache_dir)
+matcher = FootballReplayMatchLive(
+    base_url=_llm_base_url,
+    model=_llm_model,
+    temperature=_llm_temperature,
+)
+
+
+def replay_match_event(data: dict) -> dict:
+    """
+    足球回看与直播事件匹配接口。
+
+    请求参数:
+        id         : str   任务唯一键
+        match_id   : str   比赛id
+        replay     : dict  回看信息
+            |- id        : str  回看id
+            |- url       : str  回看url(m3u8)
+            |- start_utc : int  回看开始utc时间
+            |- end_utc   : int  回看结束utc时间
+        events     : list  赛事事件片段（直播候选）
+            |- id        : str  事件id
+            |- type      : str  事件类型（1：进球）
+            |- url       : str  事件片段url(m3u8)
+            |- event_utc : int  事件发生的utc时间点
+
+    响应参数:
+        id         : str   任务唯一键
+        match_id   : str   比赛id
+        replay_id  : str   回看id
+        event_id   : str|null  匹配到的比赛片段id，null表示无匹配
+    """
+    task_id = data.get("id")
+    match_id = data.get("match_id")
+    replay = data.get("replay", {})
+    events = data.get("events", [])
+
+    replay_id = replay.get("id")
+    replay_url = replay.get("url")
+    replay_start = replay.get("start_utc")
+    replay_end = replay.get("end_utc")
+
+    if not replay_url or replay_start is None or replay_end is None:
+        raise ValueError("replay 参数不完整，需要 url、start_utc、end_utc")
+
+    # 1. 为回看视频提取帧并构建 LLM content
+    try:
+        replay_frames = v2f.to_frames(
+            url=replay_url,
+            start_utc=replay_start,
+            end_utc=replay_end,
+            fps=2,
+        )
+        replay_contents = frames2content(replay_frames)
+        # 保持与 football_replay_match_live.py 中一致的提示结构
+        replay_contents.insert(0, {"type": "text", "text": "\n【回放片段信息】\n"})
+        replay_contents.append({"type": "text", "text": "\n回放解说内容：无\n"})
+    except Exception as e:
+        raise RuntimeError(f"回看视频处理失败: {e}")
+
+    # 2. 为每个 event（直播候选）提取帧并构建 LLM content
+    #    event 的 url 本身已是视频片段，直接处理完整片段
+    live_videos = []
+    for event in events:
+        event_id = event.get("id")
+        event_url = event.get("url")
+
+        if not event_url:
+            continue
+
+        try:
+            event_frames = v2f.to_frames(
+                url=event_url,
+                start_utc=0,
+                end_utc=999999,   # 传极大值，实际以视频真实长度为准
+                fps=2,
+            )
+            event_contents = frames2content(event_frames)
+            event_contents.insert(
+                0, {"type": "text", "text": f"### 候选片段 video_id: {event_id} ###"}
+            )
+            event_contents.append({"type": "text", "text": "\n该片段解说内容: 无\n"})
+        except Exception:
+            # 单个 event 处理失败不影响整体流程
+            continue
+
+        live_videos.append({
+            "video_id": event_id,
+            "video_path": "",          # 已提供 llm_contents，无需本地路径
+            "asr_text": "",
+            "llm_contents": event_contents,
+        })
+
+    # 3. 执行匹配
+    if len(live_videos) == 0:
+        matched_event_id = None
+    elif len(live_videos) == 1:
+        # 只有一个候选片段，直接视为匹配结果
+        matched_event_id = live_videos[0]["video_id"]
+    else:
+        try:
+            result = matcher._match_batch(replay_contents, live_videos, max_parallel=3)
+            matched_event_id = result.get("video_id") if result else None
+        except Exception:
+            matched_event_id = None
+
+    return {
+        "id": task_id,
+        "match_id": match_id,
+        "replay_id": replay_id,
+        "event_id": matched_event_id,
+    }

@@ -55,9 +55,11 @@ def kafka_message_iterator_thread(message_iterator, config, callback_func):
         for message in message_iterator:
             try:
                 json_message = json.loads(message)
-                replay_match_event(json_message)
-                logger.info(message)
-            except:
+                result = replay_match_event(json_message)
+                task_id = json_message.get("id")
+                callback_func(task_id, json.dumps(result, ensure_ascii=False))
+                logger.info(f"Processed task_id={task_id}, result={result}")
+            except Exception:
                 logger.exception(f"Error processing message: {message}")