main.py

import pandas as pd, soundfile as sf, json
from pathlib import Path
from pathlib import Path
from pydub import AudioSegment, silence
from sklearn.model_selection import train_test_split

root = Path("avdp-synth-corpus/")
df = pd.read_csv("avdp-synth-corpus/data/he/manifest.csv")
sp = df[df["project"] == "she_proves"].copy()

positive = {"SV", "IT"}
negative = {"NEU", "NEG"}

sp["binary_label"] = sp["violence_typology"].map(
    lambda x: 1 if x in positive else 0 if x in negative else None
)

sp = sp.dropna(subset=["binary_label"])
sp["binary_label"] = sp["binary_label"].astype(int)
# Tag backend per row (Google clips have "Chirp" in voice_families)
sp = sp.assign(backend=sp["voice_families"].str.contains("Chirp").map({True: "google", False: "azure"}))
print(sp.groupby("backend")["clip_id"].count())
# azure     10
# google     2

# Load audio for each row
audio = {row.clip_id: sf.read(root / row.wav_path) for row in sp.itertuples()}
print('.')

# =========================================================
# TXT PARSER
# =========================================================

import re

def read_segments_txt(txt_path):
    rows = []

    timing_pattern = re.compile(
        r"\[SPEAKER:.*?\|\s*ROLE:\s*(.*?)\s*\|\s*ONSET:\s*([\d.]+)\s*\|\s*OFFSET:\s*([\d.]+)\s*\]"
    )

    pending_segment = None

    with open(txt_path, "r", encoding="utf-8") as f:
        lines = [line.strip() for line in f if line.strip()]

    for line_idx, line in enumerate(lines):

        match = timing_pattern.match(line)

        if match:
            role = match.group(1)
            onset = float(match.group(2))
            offset = float(match.group(3))

            pending_segment = {
                "line_idx": line_idx,
                "role": role,
                "onset": onset,
                "offset": offset,
                "text": "",
                "action": "",
                "intensity": None,
            }

            continue

        if pending_segment is not None and not line.startswith("["):
            pending_segment["text"] = line
            continue

        if pending_segment is not None and line.startswith("[ACTION:"):
            action_match = re.match(
                r"\[ACTION:\s*(.*?)\s*\|\s*INTENSITY:\s*(\d+)\s*\]",
                line
            )

            if action_match:
                pending_segment["action"] = action_match.group(1)
                pending_segment["intensity"] = int(action_match.group(2))

            rows.append(pending_segment)
            pending_segment = None

    return rows

# =========================================================
# AUDIO SEGMENTATION
# =========================================================
out_dir = root / "data/he/she_proves_segments"
out_dir.mkdir(parents=True, exist_ok=True)

segment_rows = []

for row in sp.itertuples():

    wav_path = root / row.wav_path

    txt_path = wav_path.with_suffix(".txt")

    print(f"\nProcessing: {wav_path.name}")

    if not wav_path.exists():
        print("Missing wav:", wav_path)
        continue

    if not txt_path.exists():
        print("Missing txt:", txt_path)
        continue

    # Load audio
    audio = AudioSegment.from_file(wav_path)

    # Read txt annotations
    segments_info = read_segments_txt(txt_path)

    print("Found segments:", len(segments_info))

    for seg in segments_info:

        # =================================================
        # Convert seconds -> milliseconds
        # =================================================

        onset_ms = int(seg["onset"] * 1000)
        offset_ms = int(seg["offset"] * 1000)

        # Skip invalid segments
        if offset_ms <= onset_ms:
            continue

        # Cut audio
        chunk = audio[onset_ms:offset_ms]

        # =================================================
        # Save segment
        # =================================================

        segment_id = f"{row.clip_id}_seg_{seg['line_idx']:03d}"

        segment_filename = f"{segment_id}.wav"

        segment_path = out_dir / segment_filename

        chunk.export(segment_path, format="wav")

        # =================================================
        # Save metadata
        # =================================================

        segment_rows.append({
            "parent_clip_id": row.clip_id,
            "segment_id": segment_id,
            "wav_path": str(segment_path.relative_to(root)),
            "txt_path": str(txt_path.relative_to(root)),
            "onset": seg["onset"],
            "offset": seg["offset"],
            "duration_sec": seg["offset"] - seg["onset"],
            "sentence": seg["text"],
            "speaker_role": seg["role"],
            "action": seg["action"],
            "intensity": seg["intensity"],
            "label": row.violence_typology,
            "binary_label": row.binary_label,
        })

# =========================================================
# BUILD SEGMENTS MANIFEST
# =========================================================

segments_df = pd.DataFrame(segment_rows)

segments_df.to_csv('output_manifest.csv', index=False)

# =========================================================
# SUMMARY
# =========================================================

print("\n====================================")
print("DONE")
print("====================================")

print("\nSaved manifest:")
print('output_manifest.csv')

print("\nTotal segments:")
print(len(segments_df))

print("\nBinary label distribution:")
print(segments_df["binary_label"].value_counts())

print("\nOriginal label distribution:")
print(segments_df["label"].value_counts())

print("\nDuration stats:")
print(segments_df["duration_sec"].describe())

print("\nSample rows:")
print(segments_df.head())