vrsub/nodes/ass.py

"""SRT 转 ASS 节点。

单体版中作为进程内节点模块，由调度器直接调用。解析标准 SRT 后生成 ASS
文件，其中同一句字幕同时输出 LeftEye 与 RightEye 两个样式，分别落在屏幕
左右两半，形成 VR 双眼叠加效果。
"""

from __future__ import annotations

from pathlib import Path

from wov_sdk.models import InvokeRequest, InvokeResponse

# 统一 ASS 样式常量（单一事实来源）：新生成字幕（write_ass/invoke）与历史
# 字幕统一脚本（scripts/unify_ass_style.py）共用，调整样式只改这里，两条输出
# 路径不会各自漂移。
DEFAULT_MARGIN_TOP = 700

# 左右眼样式行字段（列顺序与 ASS Style Format 一一对应）：
# - PrimaryColour &HB3FFFFFF：约 70% 透明文字填充，降低对画面的遮挡；
# - OutlineColour &H80000000：半透明黑描边，兼顾可读性与不产生生硬黑框；
# - Alignment 8（\an8 顶部居中）：配合 MarginV 形成顶部安全区，避开画面中央
#   人脸区，并落在视线自然高度。
_ASS_FONT = "Arial"
_ASS_FONT_SIZE = 50
_ASS_PRIMARY = "&HB3FFFFFF"
_ASS_SECONDARY = "&H000000FF"
_ASS_OUTLINE = "&H80000000"
_ASS_BACK = "&H80000000"
_ASS_SCALE_X = 50
_ASS_SCALE_Y = 100
_ASS_BORDER_STYLE = 1
_ASS_OUTLINE_WIDTH = 4
_ASS_SHADOW = 0
_ASS_ALIGN = 8
_ASS_ENCODING = 1
_EYE_PAD = 50  # 左眼距屏幕左缘 / 右眼距右缘的水平内边距

# 样式/事件表头格式行（列顺序即上面注释的顺序，勿改动）。
ASS_STYLE_FORMAT = (
    "Format: Name,Fontname,Fontsize,PrimaryColour,SecondaryColour,OutlineColour,"
    "BackColour,Bold,Italic,Underline,StrikeOut,ScaleX,ScaleY,Spacing,Angle,"
    "BorderStyle,Outline,Shadow,Alignment,MarginL,MarginR,MarginV,Encoding"
)
ASS_EVENT_FORMAT = "Format: Layer,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text"


def style_row(eye: str, width: int, margin_top: int = DEFAULT_MARGIN_TOP) -> str:
    """生成单眼（LeftEye/RightEye）的完整 ASS 样式行。

    左眼占左半幅（左缘留 _EYE_PAD 内边距、向中线收 50%），右眼占右半幅；
    两眼水平相对位置一致 → 零视差，字幕落在屏幕平面，不引入额外景深冲突。
    margin_top 即该眼样式的 MarginV（距画面上缘的安全边距）。"""
    mid = width // 2
    margin_l, margin_r = (_EYE_PAD, mid) if eye == "LeftEye" else (mid, _EYE_PAD)
    return (
        f"Style: {eye},{_ASS_FONT},{_ASS_FONT_SIZE},{_ASS_PRIMARY},{_ASS_SECONDARY},"
        f"{_ASS_OUTLINE},{_ASS_BACK},0,0,0,0,{_ASS_SCALE_X},{_ASS_SCALE_Y},0,0,"
        f"{_ASS_BORDER_STYLE},{_ASS_OUTLINE_WIDTH},{_ASS_SHADOW},{_ASS_ALIGN},"
        f"{margin_l},{margin_r},{margin_top},{_ASS_ENCODING}"
    )


def ass_header(width: int, height: int, margin_top: int = DEFAULT_MARGIN_TOP) -> str:
    """生成标准 ASS 文件头：Script Info（含分辨率）+ 左右眼样式 + 事件格式行。

    新生成字幕与历史字幕统一脚本共用此函数，保证两者输出样式完全一致。"""
    return (
        "[Script Info]\n"
        "Title: VR Dual-Eye Subtitle\n"
        "ScriptType: v4.00+\n"
        "Collisions: Normal\n"
        f"PlayResX: {width}\n"
        f"PlayResY: {height}\n"
        "WrapStyle: 1\n"
        "ScaledBorderAndShadow: yes\n"
        "\n"
        "[V4+ Styles]\n"
        f"{ASS_STYLE_FORMAT}\n"
        f"{style_row('LeftEye', width, margin_top)}\n"
        f"{style_row('RightEye', width, margin_top)}\n"
        "\n"
        "[Events]\n"
        f"{ASS_EVENT_FORMAT}\n"
    )


def dialogue_line(style: str, start: str, end: str, text: str) -> str:
    """生成一行标准 Dialogue 事件。

    文本前缀固定 \\an8 顶部居中对齐（与样式 Alignment 一致），使每句字幕
    都落到样式定义的顶部安全区位置。"""
    return f"Dialogue: 0,{start},{end},{style},,0,0,0,,{{\\an8}}{text}"


def _ass_header(resolution: str, margin_top: int = DEFAULT_MARGIN_TOP) -> str:
    """兼容旧接口的 ASS 头生成：resolution 形如 "3840x1920"。

    内部委托给 ass_header()（统一样式出口），仅负责把字符串分辨率解析为
    整数宽高。"""
    width, height = (int(part) for part in resolution.lower().split("x", 1))
    return ass_header(width, height, margin_top=margin_top)


def parse_srt(text: str) -> list[tuple[str, str, str]]:
    """把 SRT 文本解析为 (开始时间, 结束时间, 文本) 条目列表。"""
    entries: list[tuple[str, str, str]] = []
    lines = text.splitlines()
    index = 0
    while index < len(lines):
        # 跳过序号前的空行，兼容文件开头有换行的情况。
        if not lines[index].strip():
            index += 1
            continue
        # 跳过序号行，直接读取下一行时间轴。
        index += 1
        if index >= len(lines):
            break
        time_line = lines[index].strip()
        index += 1
        # 时间轴必须包含分隔符，否则按畸形输入跳过。
        if " --> " not in time_line:
            continue
        # SRT 使用逗号毫秒，ASS 使用点号，需要转换。
        start, end = [part.replace(",", ".") for part in time_line.split(" --> ")]
        # 连续读取非空行作为字幕文本，多行用 ASS 换行符 \N 连接。
        text_lines: list[str] = []
        while index < len(lines) and lines[index].strip():
            text_lines.append(lines[index])
            index += 1
        entries.append((start, end, r"\N".join(text_lines)))
        index += 1
    return entries


def write_ass(
    entries: list[tuple[str, str, str]],
    output_path: Path,
    resolution: str,
    margin_top: int = DEFAULT_MARGIN_TOP,
) -> None:
    """把解析后的条目写入 ASS 文件，每个条目输出左右眼两行 Dialogue。

    margin_top 为字幕距画面顶部的安全边距，配合顶部对齐（\an8）让字幕落在
    顶部安全区内。左右眼使用相同文本与水平相对位置（零视差）。"""
    width, height = (int(part) for part in resolution.lower().split("x", 1))
    header = ass_header(width, height, margin_top=margin_top)
    # an8 对齐到屏幕顶部，配合 MarginV 形成顶部安全区，避开中央人脸区域。
    lines = [header.rstrip("\n")]
    for start, end, text in entries:
        lines.append(dialogue_line("LeftEye", start, end, text))
        lines.append(dialogue_line("RightEye", start, end, text))
    output_path.write_text("\n".join(lines) + "\n", encoding="utf-8")

def invoke(request: InvokeRequest) -> InvokeResponse:
    """把 cn_srt_uri 指向的 SRT 转为 dual_eye.ass 产物。"""
    srt_uri = request.inputs.get("cn_srt_uri")
    if not srt_uri:
        return InvokeResponse(status="failed", error="cn_srt_uri is required")

    srt_path = Path(srt_uri)
    if not srt_path.is_file():
        return InvokeResponse(status="failed", error="srt file not found")

    entries = parse_srt(srt_path.read_text(encoding="utf-8"))
    output_dir = Path(request.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    output_path = output_dir / "dual_eye.ass"
    # 分辨率默认 3840x1920，覆盖常见 VR 视频尺寸。
    resolution = str(request.params.get("resolution", "3840x1920"))
    # margin_top 可选：不同分辨率/内容可用工作流参数微调顶部安全边距。
    margin_top = int(request.params.get("margin_top", 700))
    write_ass(entries, output_path, resolution, margin_top=margin_top)
    return InvokeResponse(status="completed", outputs={"ass_uri": str(output_path)})