monitor.py

import os
import time
import asyncio
import json
import re
from dotenv import load_dotenv
from datetime import datetime, timezone

load_dotenv()

from process_utils import transcribe_worker



# 监控配置
ROOT_DIR = os.getenv("AUDIO_DIR", os.path.join(os.getcwd(), "record"))
POLL_INTERVAL = int(os.getenv("MONITOR_POLL_INTERVAL", "5"))  # 秒
IDLE_THRESHOLD = int(os.getenv("MONITOR_IDLE_SECONDS", str(5 * 60)))  # 5 分钟
MAX_WAIT = int(os.getenv("MONITOR_MAX_SECONDS", str(10 * 60)))  # 10 分钟
AUDIO_EXTS = {".wav"}

print(f"Monitor config: ROOT_DIR={ROOT_DIR}, POLL_INTERVAL={POLL_INTERVAL}, IDLE_THRESHOLD={IDLE_THRESHOLD}, MAX_WAIT={MAX_WAIT}")

# 可选的文件名正则（针对不含扩展名的 basename）
AUDIO_MATCH_PATTERN = os.getenv("AUDIO_MATCH", None)
AUDIO_MATCH_RE = None
if AUDIO_MATCH_PATTERN:
    try:
        AUDIO_MATCH_RE = re.compile(AUDIO_MATCH_PATTERN)
    except Exception as e:
        print(f"Invalid AUDIO_MATCH regex '{AUDIO_MATCH_PATTERN}': {e}. Ignoring name filter.")
        AUDIO_MATCH_RE = None

def is_audio_file(fname):
    _, ext = os.path.splitext(fname.lower())
    if ext not in AUDIO_EXTS:
        return False
    # 如果设置了 AUDIO_MATCH，检查 basename（不含扩展名）是否匹配（使用 fullmatch 以保证格式严格）
    if AUDIO_MATCH_RE:
        base = os.path.splitext(os.path.basename(fname))[0]
        try:
            return bool(AUDIO_MATCH_RE.fullmatch(base))
        except Exception:
            # 兜底：如果匹配出现异常，视为不匹配
            return False
    return True


def list_unprocessed_audio_files(d):
    files = []
    try:
        for f in os.listdir(d):
            full = os.path.join(d, f)
            if not os.path.isfile(full):
                continue
            # 跳过 marker 文件本身
            if f.endswith(".processed"):
                continue
                # 如果是音频文件但存在同名的 .processed 标记文件，则跳过
            if is_audio_file(f):
                marker = full + ".processed"
                if os.path.exists(marker):
                    continue
                files.append(full)
            if is_audio_file(f):
                files.append(full)
    except FileNotFoundError:
        return []
    files.sort(key=lambda p: os.path.getmtime(p))
    return files

# python
async def run_transcription_for_files(files, out_dir):
    """
    调用 process_utils.transcribe_worker 处理文件列表（文件路径列表），
    保存 transcript 到 out_dir，并将返回的 speaker_map 与 out_dir 下的 speaker_map.json 合并更新后保存。
    返回 True/False 表示是否成功。
    该版本对 final_result / speaker_map / existing_map_list 增加了类型检查与容错，
    避免因返回非预期类型（例如字符串）而导致的 "string indices must be integers" 错误。
    """
    try:
        # transcribe_worker 期望收到相对于 AUDIO_DIR 的路径（而不是绝对路径），
        # 所以在此将绝对路径转换为相对路径；如果无法相对化则使用 basename 回退。
        rel_files = []
        for f in files:
            if os.path.isabs(f):
                try:
                    rp = os.path.relpath(f, ROOT_DIR)
                    # 如果结果包含上级路径，说明不在 ROOT_DIR 下，回退为 basename
                    if rp.startswith(os.pardir + os.sep) or rp == os.pardir:
                        rp = os.path.basename(f)
                except Exception:
                    rp = os.path.basename(f)
            else:
                rp = f
            rel_files.append(rp)

        # transcribe_worker 是 async generator，逐条读取状态
        final_result = None
        async for status in transcribe_worker(rel_files):
            # 增强型诊断：若 status 不是 dict，打印类型（但继续尝试处理）
            if not isinstance(status, dict):
                try:
                    print(f"[transcribe] received non-dict status: type={type(status)} value={repr(status)[:200]}")
                except Exception:
                    print(f"[transcribe] received non-dict status of type {type(status)}")
                # 如果是字符串且可能是 JSON，尝试解析为 dict
                if isinstance(status, str):
                    try:
                        parsed = json.loads(status)
                        if isinstance(parsed, dict):
                            status = parsed
                        else:
                            # 保留原始 status（后续判断会忽略）
                            pass
                    except Exception:
                        pass

            msg = None
            prog = None
            try:
                msg = status.get("message") or status.get("status")
                prog = status.get("progress")
            except Exception:
                # status 不是 dict，跳过
                pass

            if msg:
                print(f"[transcribe] {msg} ({prog})")
            try:
                if isinstance(status, dict) and status.get("status") == "done":
                    final_result = status
            except Exception:
                # 忽略非 dict 的 status
                pass

        if not final_result:
            print("Transcription ended without final result.")
            return False

        # 确保 final_result 为 dict（若为字符串尝试解析）
        if isinstance(final_result, str):
            try:
                final_result = json.loads(final_result)
            except Exception as e:
                print(f"Unexpected final_result string that is not JSON: {e}; value={repr(final_result)[:200]}")
                return False
        if not isinstance(final_result, dict):
            print(f"Unexpected final_result type: {type(final_result)}. Aborting.")
            return False

        transcript = final_result.get("transcript", "") or ""
        speaker_map = final_result.get("speaker_map", []) or []

        # 兼容多种 speaker_map 表现形式
        if isinstance(speaker_map, dict):
            speaker_map = [speaker_map]
        elif isinstance(speaker_map, str):
            try:
                parsed_sm = json.loads(speaker_map)
                if isinstance(parsed_sm, dict):
                    speaker_map = [parsed_sm]
                elif isinstance(parsed_sm, list):
                    speaker_map = parsed_sm
                else:
                    speaker_map = []
            except Exception:
                print("speaker_map is a string but not valid JSON; ignoring speaker_map.")
                speaker_map = []
        elif not isinstance(speaker_map, list):
            print(f"speaker_map has unexpected type {type(speaker_map)}; ignoring.")
            speaker_map = []


        # 计算时间范围用于 transcript 文件名：优先基于文件名中匹配 AUDIO_MATCH 的时间信息，解析失败则回退到文件的 mtime
        ts_list = []
        for f in files:
            ts = None
            try:
                base = os.path.splitext(os.path.basename(f))[0]
                if AUDIO_MATCH_RE:
                    m = AUDIO_MATCH_RE.search(base)
                    if m:
                        # 优先使用第一个捕获组（如果有），否则使用整个匹配
                        try:
                            ts_str = m.group(1) if m.lastindex and m.lastindex >= 1 else m.group(0)
                        except Exception:
                            ts_str = m.group(0)
                        try:
                            # 将文件名中安全的时间格式转换为 ISO 格式：
                            # 例如 2023-05-01T12_34_56.789+08_00 -> 2023-05-01T12:34:56.789+08:00
                            ts_iso = re.sub(r'(?P<date>\d{4}-\d{2}-\d{2}T)(?P<h>\d{2})_(?P<m>\d{2})_(?P<s>\d{2})(?P<rest>.*)',
                                            r'\g<date>\g<h>:\g<m>:\g<s>\g<rest>',
                                            ts_str)
                            ts_iso = re.sub(r'([+-]\d{2})_(\d{2})$', r'\1:\2', ts_iso)
                            dt = datetime.fromisoformat(ts_iso)
                            ts = dt.timestamp()
                        except Exception:
                            ts = None
            except Exception:
                ts = None

            if ts is None:
                try:
                    ts = os.path.getmtime(f)
                except Exception:
                    ts = None

            if ts is not None:
                ts_list.append(ts)

        start_ts = min(ts_list) if ts_list else time.time()
        end_ts = max(ts_list) if ts_list else time.time()
        start_str = datetime.fromtimestamp(start_ts, timezone.utc).strftime("%Y%m%d_%H%M%S")
        end_str = datetime.fromtimestamp(end_ts, timezone.utc).strftime("%Y%m%d_%H%M%S")

        # 保存 transcript（按时间段单独文件）
        os.makedirs(out_dir, exist_ok=True)
        transcript_fname = os.path.join(out_dir, f"transcript_{start_str}__{end_str}.txt")
        with open(transcript_fname, "w", encoding="utf-8") as outf:
            outf.write(transcript or "")
        print(f"Saved transcript -> {transcript_fname}")

        # 合并并保存全局 speaker_map 到 out_dir/speaker_map.json（每个子目录仅一个 speaker_map）
        speaker_map_path = os.path.join(out_dir, "speaker_map.json")
        existing_map_list = []
        if os.path.exists(speaker_map_path):
            try:
                with open(speaker_map_path, "r", encoding="utf-8") as inf:
                    existing_map_list = json.load(inf) or []
                if isinstance(existing_map_list, dict):
                    existing_map_list = [existing_map_list]
                elif not isinstance(existing_map_list, list):
                    print(f"Existing speaker_map.json has unexpected type {type(existing_map_list)}; will overwrite.")
                    existing_map_list = []
            except Exception as e:
                print(f"Failed to load existing speaker_map.json: {e}. Will overwrite with new map.")
                existing_map_list = []

        # 构建以 speaker 为 key 的字典并合并（优先使用新值非空字段）
        merged = {}
        def merge_sound_chars(old_sc, new_sc):
            if not isinstance(old_sc, dict):
                old_sc = {}
            if not isinstance(new_sc, dict):
                new_sc = {}
            merged_sc = dict(old_sc)  # start with old
            # override with new values if they are not empty
            for k, v in new_sc.items():
                if v is not None and v != "":
                    merged_sc[k] = v
            return merged_sc

        for entry in existing_map_list:
            if not isinstance(entry, dict):
                print(f"Skipping non-dict entry in existing_map_list: {repr(entry)[:100]}")
                continue
            sp = entry.get("speaker")
            if not sp:
                continue
            merged[sp] = dict(entry)

        for entry in speaker_map:
            if not isinstance(entry, dict):
                print(f"Skipping non-dict entry in speaker_map: {repr(entry)[:100]}")
                continue
            sp = entry.get("speaker")
            if not sp:
                continue
            if sp in merged:
                # 合并非空字段，特别处理 sound_characteristics
                old = merged[sp]
                new = entry
                old_sc = old.get("sound_characteristics", {}) or {}
                new_sc = new.get("sound_characteristics", {}) or {}
                merged_sc = merge_sound_chars(old_sc, new_sc)
                # 合并其它顶层字段（以 new 的非空为准）
                merged_entry = dict(old)
                for k, v in new.items():
                    if k == "sound_characteristics":
                        continue
                    if v is not None and v != "":
                        merged_entry[k] = v
                merged_entry["sound_characteristics"] = merged_sc
                merged[sp] = merged_entry
            else:
                merged[sp] = entry

        final_speaker_list = list(merged.values())

        # 原子写入 speaker_map.json
        try:
            tmp_path = speaker_map_path + ".tmp"
            with open(tmp_path, "w", encoding="utf-8") as outf:
                json.dump(final_speaker_list, outf, ensure_ascii=False, indent=2)
            os.replace(tmp_path, speaker_map_path)
            print(f"Updated speaker_map -> {speaker_map_path}")
        except Exception as e:
            print(f"Failed to save merged speaker_map: {e}")

        return True
    except Exception as e:
        print(f"Error during transcription task: {e}")
        return False


async def handle_subdir_processing(subdir, files, state):
    """
    在独立任务中处理某个子目录的文件列表：调用转录，保存结果，标记已处理文件，重置状态。
    """
    print(f"Starting processing for {subdir}, {len(files)} files.")
    success = await run_transcription_for_files(files, subdir)
    # 标记已处理文件，避免重复处理
    for f in files:
        try:
            marker = f + ".processed"
            with open(marker, "w", encoding="utf-8") as m:
                m.write(f"processed: {time.time()}\n")
        except Exception as e:
            print(f"Failed to mark {f} as processed: {e}")
    # 重置状态
    state["files_set"].clear()
    state["first_seen"] = None
    state["last_added"] = None
    state["processing"] = False
    print(f"Finished processing for {subdir}. success={success}")


async def monitor_loop(root_dir):
    """
    主轮询循环：遍历 root_dir 下的子目录并监控新增音频文件。
    """
    print(f"Monitoring root: {root_dir}")
    # 每个子目录的状态记录
    states = {}

    while True:
        try:
            if not os.path.isdir(root_dir):
                os.makedirs(root_dir, exist_ok=True)

            subdirs = [os.path.join(root_dir, d) for d in os.listdir(root_dir)
                       if os.path.isdir(os.path.join(root_dir, d))]
            now = time.time()
            for sd in subdirs:
                st = states.setdefault(sd, {
                    "files_set": set(),
                    "first_seen": None,
                    "last_added": None,
                    "processing": False
                })
                # 扫描当前目录的未处理音频文件
                current_files = list_unprocessed_audio_files(sd)
                current_set = set(current_files)

                # 新加入的文件集合 = current_set - previously tracked
                new_files = current_set - st["files_set"]
                if new_files:
                    # 更新 tracked set 与时间点
                    for nf in new_files:
                        st["files_set"].add(nf)
                    if st["first_seen"] is None:
                        st["first_seen"] = now
                    st["last_added"] = now
                    print(f"[{sd}] Detected {len(new_files)} new files. total tracked: {len(st['files_set'])}")

                # 如果有被外部删除的文件，从 tracked 中移除
                removed = {f for f in st["files_set"] if not os.path.exists(f)}
                if removed:
                    for r in removed:
                        st["files_set"].discard(r)

                if st["processing"]:
                    continue  # 正在处理，跳过检查

                # 如果没有任何待处理文件，跳过
                if not st["files_set"]:
                    st["first_seen"] = None
                    st["last_added"] = None
                    continue

                # 判断是否满足触发条件：空闲超过 IDLE_THRESHOLD 或 从 first_seen 起超过 MAX_WAIT
                since_last = now - (st["last_added"] or now)
                since_first = now - (st["first_seen"] or now)
                if since_last >= IDLE_THRESHOLD or since_first >= MAX_WAIT:
                    # 触发异步处理任务
                    files_to_process = sorted(list(st["files_set"]), key=lambda p: os.path.getmtime(p))
                    st["processing"] = True
                    asyncio.create_task(handle_subdir_processing(sd, files_to_process, st))
        except Exception as e:
            print(f"Error in monitor loop: {e}")

        await asyncio.sleep(POLL_INTERVAL)


def main():
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(monitor_loop(ROOT_DIR))
    except KeyboardInterrupt:
        print("Monitor stopped by user.")
    finally:
        try:
            loop.stop()
        except Exception:
            pass


if __name__ == "__main__":
    main()