[데이터 수집하고 분석하기] 유튜브 api를 통해 데이터 수집 및 분석 3차

1.데이터 합치기

이제 생성한 데이터를 합치고, 추출한 데이터를 바탕으로 생성형 ai를 통해 컨텐츠를 제작해보자.

# patches/06_merge_audio_features.py
import pandas as pd
from pathlib import Path

# 두 파일 읽기
audio = pd.read_csv("data/processed/audio_features.csv")
melody = pd.read_csv("data/processed/melody_key_features.csv")

# video_id 기준 병합
df = audio.merge(melody, on="video_id", how="left")

# 저장
Path("data/processed").mkdir(parents=True, exist_ok=True)
out_path = "data/processed/audio_full_features.csv"
df.to_csv(out_path, index=False, encoding="utf-8-sig")

print(f"[OK] saved merged features -> {out_path} (rows={len(df)})")

이 코드를 통해 audio_feature.csv파일과 melody_key_features.csv 두개를 하나의 csv파일로 합친다.

 

그다음

pip install numpy soundfile

로 csv를 해석하고 음악으로 바꿔줄 패키지를 받아두자.

2.데이터에서 가사,새 음율, 노래 생성하기

가사 생성

OpenAi와 Ollama중 로컬로 사용가능한 llm인 Ollama를 사용했다.

 

• https://ollama.com 설치 후, 예: ollama pull llama3.1
• 파이썬 패키지:pip install ollama

스크립트: 규칙 초안이 없으면 만들고 → LLM으로 창의 리라이팅 → SRT 생성

# -*- coding: utf-8 -*-
"""
데이터 기반 가사 생성 파이프라인
1) prompt_packs.jsonl 로부터 곡 조건 로드
2) (없으면) 규칙 기반 초안 lyrics.txt 생성
3) LLM(OPENAI or OLLAMA)으로 창의 리라이팅 -> lyrics_creative.txt
4) SRT 생성 -> lyrics.srt (줄당 기본 2.5초)

Usage:
  python patches/13_make_creative_lyrics.py --take 5 --provider openai --model gpt-4o-mini
  python patches/13_make_creative_lyrics.py --take 5 --provider ollama --model llama3.1

필요 파일:
- data/processed/prompt_packs.jsonl (이미 생성됨)
출력:
- outputs/<video_id>/lyrics.txt (규칙 초안)
- outputs/<video_id>/lyrics_creative.txt (LLM 리라이팅 결과)
- outputs/<video_id>/lyrics.srt (간이 싱크 자막)
"""
import os, re, json, time, random
from pathlib import Path
import argparse

PACKS = Path("data/processed/prompt_packs.jsonl")
OUTROOT = Path("outputs")
OUTROOT.mkdir(parents=True, exist_ok=True)

# ---------- 공통 유틸 ----------
def syllables(s: str) -> int:
    """아주 단순한 음절 길이 근사: 공백/구두점 제거한 길이."""
    return len(re.sub(r"[\s\.\,\!\?\-]", "", s))

def shorten_to_target(line: str, max_len=10):
    toks = line.split()
    while syllables(" ".join(toks)) > max_len and len(toks) > 1:
        toks.pop()
    return " ".join(toks)

def read_packs():
    packs = []
    with PACKS.open("r", encoding="utf-8") as f:
        for line in f:
            if line.strip():
                packs.append(json.loads(line))
    return packs

# ---------- 1) 규칙 기반 초안 ----------
def build_rule_based_lyrics(pack):
    mood   = pack["music"].get("mood", "Playful")
    bpm    = int(pack["music"].get("bpm", 120))
    themes = pack["lyrics"].get("themes", [])
    style  = pack["lyrics"].get("style", "Playful, 반복 후렴")

    # 한글 학습 토큰 보강
    if not any(ch in "".join(themes) for ch in list("가나다라마바사아자차카타파하")):
        themes = ["가", "나", "다", "라"] + themes

    # 코러스: 반복성 강조
    base = themes[:3] or ["가","나","다"]
    chorus = [
        f"{base[0]} {base[1]} {base[2]}! (짝짝)",
        f"같이 불러봐, {base[0]} {base[1]} {base[2]}!",
        f"{base[0]} {base[1]} {base[2]}! (빙글)"
    ]
    chorus = [shorten_to_target(x, 12) for x in chorus]

    # 동작 단어 세트
    actions_play = ["뛰어", "돌아", "박수", "웃어", "손들어", "점프", "폴짝"]
    actions_calm = ["살금", "조용히", "포근히", "토닥", "꿈꾸자"]
    acts = actions_play if "Calm" not in mood else actions_calm

    toks = (themes + acts)[:10]

    def verse():
        pool = toks[:]
        random.shuffle(pool)
        lines = []
        for i in range(4):
            a = pool[i % len(pool)]
            b = pool[(i+1) % len(pool)]
            lines.append(shorten_to_target(f"{a} {b}", 10))
        return lines

    v1 = verse()
    v2 = verse()

    lines = []
    lines.append("[Chorus]"); lines += chorus; lines.append("")
    lines.append("[Verse 1]"); lines += v1; lines.append("")
    lines.append("[Chorus]"); lines += chorus; lines.append("")
    lines.append("[Verse 2]"); lines += v2; lines.append("")
    lines.append("[Chorus]"); lines += chorus; lines.append("")
    return "\n".join(lines).strip()

# ---------- 2) LLM 리라이팅 ----------
OPENAI_CLIENT = None
def ensure_openai():
    global OPENAI_CLIENT
    if OPENAI_CLIENT is None:
        try:
            from openai import OpenAI
            OPENAI_CLIENT = OpenAI()
        except Exception as e:
            raise RuntimeError("openai 패키지가 필요합니다: pip install openai") from e
    return OPENAI_CLIENT

def call_openai(model, system_prompt, user_prompt, temperature=0.9):
    client = ensure_openai()
    resp = client.chat.completions.create(
        model=model,
        temperature=temperature,
        messages=[
            {"role":"system","content":system_prompt},
            {"role":"user","content":user_prompt}
        ]
    )
    return resp.choices[0].message.content.strip()

def call_ollama(model, system_prompt, user_prompt, temperature=0.9):
    try:
        import ollama
    except Exception as e:
        raise RuntimeError("ollama 패키지가 필요합니다: pip install ollama") from e
    prompt = f"<<SYS>>{system_prompt}<<SYS>>\n{user_prompt}"
    resp = ollama.chat(model=model, messages=[{"role":"user","content":prompt}],
                       options={"temperature": temperature})
    return resp["message"]["content"].strip()

SYSTEM_PROMPT = (
"당신은 0-7세 아동용 한국어 노래 작사가입니다. 짧고 쉬운 어휘, 반복적인 후렴, "
"밝고 따뜻한 이미지를 사용하세요. 연령 적합성을 지키고, 폭력/공포/비속어를 금지합니다."
)

def build_user_prompt(pack, draft):
    music = pack["music"]
    lyrics = pack["lyrics"]
    bpm = int(music.get("bpm", 120))
    mood = music.get("mood", "Playful")
    key  = music.get("key", "C major")
    pr   = int(round(float(music.get("pitch_range_semitones", 10))))
    themes = ", ".join(lyrics.get("themes", [])[:6])

    return f"""
아래 조건을 만족하도록, 초안 가사를 구조는 유지하되 더 창의적이고 노래하기 쉽게 리라이팅하세요.

[제약]
- 연령: 3-6세
- Mood: {mood}, Tempo: ~{bpm} BPM, Key: {key}, 음역 범위: ~{pr} 반음(너무 높지 않게)
- 후렴은 강하게 반복, 멜로디컬하고 리듬감 있게
- 문장은 3~9음절 정도로 짧게, 발음 쉬운 단어 사용
- 한글 학습 요소 포함(가-나-다-라 등, 혹은 숫자/색/동물/신체)
- 긍정적이고 따뜻한 이미지, 행동 유도(박수/점프/빙글 등)
- 형식 유지: [Chorus], [Verse 1], [Chorus], [Verse 2], [Chorus]

[주제/키워드]
{themes}

[초안]
{draft}

[출력]
- 위 형식 그대로 섹션 헤더 포함
- 한국어 가사만 출력
""".strip()

# ---------- 3) SRT 생성 ----------
def text_to_srt(text: str, sec_per_line=2.5):
    def fmt(t):
        ms=int((t-int(t))*1000); s=int(t)%60; m=(int(t)//60)%60; h=(int(t)//3600)
        return f"{h:02}:{m:02}:{s:02},{ms:03}"
    lines = [ln for ln in text.splitlines() if ln.strip() and not ln.strip().startswith('[')]
    out=[]; t=0.0; n=1
    for ln in lines:
        start=t; end=t+sec_per_line
        out.append(str(n)); n+=1
        out.append(f"{fmt(start)} --> {fmt(end)}")
        out.append(ln.strip()); out.append("")
        t=end
    return "\n".join(out)

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--take", type=int, default=5)
    ap.add_argument("--provider", choices=["openai","ollama"], default="openai")
    ap.add_argument("--model", default="gpt-4o-mini")  # openai 기본
    ap.add_argument("--temp", type=float, default=0.9)
    ap.add_argument("--sec", type=float, default=2.5, help="SRT 한 줄당 초")
    args = ap.parse_args()

    packs = read_packs()
    take = min(args.take, len(packs))

    for i, pack in enumerate(packs[:take], 1):
        vid = str(pack["video_id"])
        outdir = OUTROOT / vid
        outdir.mkdir(parents=True, exist_ok=True)

        draft_path = outdir/"lyrics.txt"
        if not draft_path.exists():
            draft = build_rule_based_lyrics(pack)
            draft_path.write_text(draft, encoding="utf-8")
        else:
            draft = draft_path.read_text(encoding="utf-8")

        user_prompt = build_user_prompt(pack, draft)

        # LLM 호출
        try:
            if args.provider == "openai":
                result = call_openai(args.model, SYSTEM_PROMPT, user_prompt, temperature=args.temp)
            else:
                result = call_ollama(args.model, SYSTEM_PROMPT, user_prompt, temperature=args.temp)
        except Exception as e:
            # 실패 시 draft 그대로 사용
            result = draft

        # 저장
        creative_path = outdir/"lyrics_creative.txt"
        creative_path.write_text(result, encoding="utf-8")

        # SRT 생성
        srt_text = text_to_srt(result, sec_per_line=args.sec)
        (outdir/"lyrics.srt").write_text(srt_text, encoding="utf-8")

        print(f"[{i}/{take}] {vid} -> lyrics_creative.txt, lyrics.srt")

if __name__ == "__main__":
    main()

 

다음 규칙으로 노래를 만들었다.

lyrics_v1.srt

0.00MB

lyrics_v1.txt

0.00MB

 

이런 느낌이다. 현재 "가나다" 같은 가사가 주로 반복되다 보니 매우 유치하고 재미없는 노래가 나옴을 확인했다.

 

프롬프트를 다음과 같이 수정한다.

# -*- coding: utf-8 -*-
"""
데이터 기반 가사 생성 파이프라인 (버전 관리 포함)

기능:
1) prompt_packs.jsonl 로부터 곡 조건 로드
2) 규칙 기반 초안 lyrics_vN.txt 자동 생성 (테마 중심 코러스, '가-나-다'는 보조 1줄만)
3) LLM(OpenAI/Ollama)으로 창의 리라이팅 -> lyrics_creative_vN.txt
4) SRT 생성 -> lyrics_vN.srt (간단 시간 분배)
5) 실행 때마다 v1, v2, ... 로 누적 저장 (이전 버전 보존)

사용법:
  # OpenAI 사용 (환경변수 OPENAI_API_KEY 필요)
  python patches/13_make_creative_lyrics.py --take 5 --provider openai --model gpt-4o-mini --sec 2.5

  # Ollama 사용 (로컬 LLM)
  python patches/13_make_creative_lyrics.py --take 5 --provider ollama --model llama3.1 --sec 2.5
"""
import os
import re
import json
import random
from pathlib import Path
import argparse

PACKS = Path("data/processed/prompt_packs.jsonl")
OUTROOT = Path("outputs")
OUTROOT.mkdir(parents=True, exist_ok=True)

# ------------------------ 유틸 ------------------------
def next_versioned_path(outdir: Path, stem: str, ext: str) -> Path:
    """
    outdir 안에 stem_vN.ext 형식의 다음 버전 파일 경로 반환
    """
    existing = list(outdir.glob(f"{stem}_v*{ext}"))
    if not existing:
        return outdir / f"{stem}_v1{ext}"
    nums = []
    for f in existing:
        m = re.search(r"_v(\d+)", f.stem)
        if m:
            nums.append(int(m.group(1)))
    next_num = (max(nums) + 1) if nums else 1
    return outdir / f"{stem}_v{next_num}{ext}"

def read_packs():
    packs = []
    with PACKS.open("r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                packs.append(json.loads(line))
    return packs

def text_to_srt(text: str, sec_per_line=2.5) -> str:
    """
    섹션 헤더([Chorus] 등)는 제외하고 줄당 고정 길이로 SRT 생성
    """
    def fmt(t):
        ms = int((t - int(t)) * 1000)
        s = int(t) % 60
        m = (int(t) // 60) % 60
        h = (int(t) // 3600)
        return f"{h:02}:{m:02}:{s:02},{ms:03}"

    lines = [ln for ln in text.splitlines() if ln.strip() and not ln.strip().startswith('[')]
    out = []
    t = 0.0
    n = 1
    for ln in lines:
        start = t
        end = t + sec_per_line
        out.append(str(n)); n += 1
        out.append(f"{fmt(start)} --> {fmt(end)}")
        out.append(ln.strip())
        out.append("")
        t = end
    return "\n".join(out)

# --------------------- 규칙 기반 초안 ---------------------
def build_rule_based_lyrics(pack: dict) -> str:
    """
    테마 중심 코러스 + 짧은 절(4행)로 구성.
    '가-나-다-라'는 학습 보조로 1줄만 살짝 삽입.
    """
    raw_themes = pack.get("lyrics", {}).get("themes", [])
    mood   = pack.get("music", {}).get("mood", "Playful")
    bpm    = int(pack.get("music", {}).get("bpm", 120))

    # 영어→한국어 간단 매핑(필요시 추가)
    KO_MAP = {
        "head":"머리","shoulder":"어깨","knee":"무릎","toe":"발가락","foot":"발","hand":"손","face":"얼굴","eye":"눈","mouth":"입",
        "dog":"강아지","cat":"고양이","bunny":"토끼","rabbit":"토끼","duck":"오리","bird":"새","bear":"곰","lion":"사자","tiger":"호랑이",
        "jump":"점프","spin":"빙글","dance":"춤","clap":"박수","run":"달려","sleep":"잠자","dream":"꿈","hug":"포옹","smile":"웃어",
        "red":"빨강","blue":"파랑","yellow":"노랑","green":"초록","pink":"분홍","purple":"보라","orange":"주황","black":"검정","white":"하양",
        "one":"하나","two":"둘","three":"셋","four":"넷","five":"다섯","star":"별","moon":"달","sun":"해","night":"밤","day":"낮","happy":"행복",
        "baby":"아기","friend":"친구"
    }

    # 토큰 정리 + 한글화
    themes_ko = []
    for t in raw_themes:
        tt = str(t).strip().lower()
        if len(tt) <= 2:
            continue
        themes_ko.append(KO_MAP.get(tt, t))  # 이미 한글이면 그대로

    # 카테고리 보강(테마 부족 시 다양성 확보)
    BODY   = ["머리","어깨","무릎","발","눈","코","입","손","발가락"]
    ANIM   = ["토끼","강아지","고양이","오리","곰","사자","호랑이"]
    ACTS_P = ["박수","점프","빙글","춤춰","달려","손들어","웃어","폴짝"]
    ACTS_C = ["살금","조용히","포근히","토닥","꿈꾸자","쿨쿨"]
    ACTS   = ACTS_P if "Calm" not in mood else ACTS_C

    def ensure_rich(themes):
        s = list(dict.fromkeys(themes))  # 순서 유지 중복 제거
        if len(s) < 4:
            s += random.sample(BODY, k=min(3, len(BODY)))
        if len(s) < 6:
            s += random.sample(ANIM, k=min(2, len(ANIM)))
        if len(s) < 8:
            s += random.sample(ACTS, k=min(2, len(ACTS)))
        return list(dict.fromkeys(s))

    themes_ko = ensure_rich(themes_ko)

    # 코러스: 테마 기반 훅(lead 2~3개)
    lead = []
    for cand in themes_ko:
        if any(x in cand for x in ["머리","어깨","무릎","발","토끼","강아지","고양이","별","달","춤","박수","점프","빙글"]):
            lead.append(cand)
        if len(lead) >= 3:
            break
    if not lead:
        lead = themes_ko[:3] or ["리듬","노래","친구"]

    hook = f"{lead[0]} {lead[1]} {lead[2]}" if len(lead) >= 3 else " ".join(lead)

    def shorten_to_target(line: str, max_len=12) -> str:
        def syl(s): return len(re.sub(r"[\s\.\,\!\?\-]", "", s))
        toks = line.split()
        while syl(" ".join(toks)) > max_len and len(toks) > 1:
            toks.pop()
        return " ".join(toks)

    chorus = [
        shorten_to_target(f"{hook}! (짝짝)", 12),
        shorten_to_target(f"{hook}! (빙글)", 12),
        "가-나-다-라! (한 번 더!)"
    ]

    pool = list(dict.fromkeys(themes_ko + ACTS))
    random.shuffle(pool)

    def verse_block():
        lines=[]
        for i in range(4):
            a = pool[(i*2) % len(pool)]
            b = pool[(i*2+1) % len(pool)]
            line = random.choice([
                f"{a} {b}",
                f"{a} 하고 {b}",
                f"{a} {b} 같이",
            ])
            lines.append(shorten_to_target(line, 10))
        return lines

    v1 = verse_block()
    random.shuffle(pool)
    v2 = verse_block()

    parts = []
    parts.append("[Chorus]"); parts += chorus; parts.append("")
    parts.append("[Verse 1]"); parts += v1; parts.append("")
    parts.append("[Chorus]"); parts += chorus; parts.append("")
    parts.append("[Verse 2]"); parts += v2; parts.append("")
    parts.append("[Chorus]"); parts += chorus; parts.append("")
    return "\n".join(parts).strip()

# --------------------- LLM 리라이팅 ---------------------
OPENAI_CLIENT = None

SYSTEM_PROMPT = (
    "당신은 0-7세 아동용 한국어 노래 작사가입니다. "
    "짧고 쉬운 어휘, 반복적인 후렴, 밝고 따뜻한 이미지를 사용하세요. "
    "연령 적합성을 지키고, 폭력/공포/비속어를 금지합니다."
)

def ensure_openai():
    global OPENAI_CLIENT
    if OPENAI_CLIENT is None:
        try:
            from openai import OpenAI
            OPENAI_CLIENT = OpenAI()
        except Exception as e:
            raise RuntimeError("openai 패키지가 필요합니다: pip install openai") from e
    return OPENAI_CLIENT

def call_openai(model: str, system_prompt: str, user_prompt: str, temperature=0.9) -> str:
    client = ensure_openai()
    resp = client.chat.completions.create(
        model=model,
        temperature=temperature,
        messages=[
            {"role":"system","content":system_prompt},
            {"role":"user","content":user_prompt}
        ]
    )
    return resp.choices[0].message.content.strip()

def call_ollama(model: str, system_prompt: str, user_prompt: str, temperature=0.9) -> str:
    try:
        import ollama
    except Exception as e:
        raise RuntimeError("ollama 패키지가 필요합니다: pip install ollama") from e
    prompt = f"<<SYS>>{system_prompt}<<SYS>>\n{user_prompt}"
    resp = ollama.chat(model=model, messages=[{"role":"user","content":prompt}],
                       options={"temperature": float(temperature)})
    return resp["message"]["content"].strip()

def build_user_prompt(pack: dict, draft: str) -> str:
    music  = pack.get("music", {})
    lyrics = pack.get("lyrics", {})
    bpm = int(music.get("bpm", 120))
    mood = music.get("mood", "Playful")
    key  = music.get("key", "C major")
    pr   = int(round(float(music.get("pitch_range_semitones", 10))))
    themes = ", ".join(lyrics.get("themes", [])[:8])

    return f"""
아래 조건을 만족하도록, 초안 가사를 구조는 유지하되 더 창의적이고 노래하기 쉽게 리라이팅하세요.

[제약]
- 연령: 3-6세
- Mood: {mood}, Tempo: ~{bpm} BPM, Key: {key}, 음역 범위: ~{pr} 반음(너무 높지 않게)
- [중요] 코러스는 '가-나-다-라' 대신 테마 어휘로 만든 '후크'를 반복하세요.
- '가-나-다-라'는 보조 학습 요소로 1회 정도 자연스럽게 삽입만 허용합니다.
- 문장은 3~9음절 정도로 짧게, 발음 쉬운 단어 사용
- 긍정적이고 따뜻한 이미지, 행동 유도(박수/점프/빙글 등)
- 형식 유지: [Chorus], [Verse 1], [Chorus], [Verse 2], [Chorus]

[주제/키워드]
{themes}

[초안]
{draft}

[출력]
- 위 형식 그대로 섹션 헤더 포함
- 한국어 가사만 출력
""".strip()

# ------------------------ 메인 ------------------------
def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--take", type=int, default=5, help="처리할 팩 개수")
    ap.add_argument("--provider", choices=["openai","ollama"], default="openai")
    ap.add_argument("--model", default="gpt-4o-mini")
    ap.add_argument("--temp", type=float, default=0.9)
    ap.add_argument("--sec", type=float, default=2.5, help="SRT 한 줄 시간(초)")
    ap.add_argument("--start", type=int, default=0, help="시작 인덱스 (샘플 오프셋)")
    args = ap.parse_args()

    packs = read_packs()
    subset = packs[args.start: args.start + args.take]

    for idx, pack in enumerate(subset, 1):
        vid = str(pack.get("video_id"))
        outdir = OUTROOT / vid
        outdir.mkdir(parents=True, exist_ok=True)

        # 1) 규칙 기반 초안 -> versioned
        draft_text = build_rule_based_lyrics(pack)
        draft_path = next_versioned_path(outdir, "lyrics", ".txt")
        draft_path.write_text(draft_text, encoding="utf-8")

        # 2) LLM 리라이팅 -> versioned
        user_prompt = build_user_prompt(pack, draft_text)
        try:
            if args.provider == "openai":
                creative_text = call_openai(args.model, SYSTEM_PROMPT, user_prompt, temperature=args.temp)
            else:
                creative_text = call_ollama(args.model, SYSTEM_PROMPT, user_prompt, temperature=args.temp)
        except Exception as e:
            # 실패하면 초안 사용
            creative_text = draft_text

        creative_path = next_versioned_path(outdir, "lyrics_creative", ".txt")
        creative_path.write_text(creative_text, encoding="utf-8")

        # 3) SRT -> versioned (LLM 결과 기준으로 만듦)
        srt_text = text_to_srt(creative_text, sec_per_line=args.sec)
        srt_path = next_versioned_path(outdir, "lyrics", ".srt")
        srt_path.write_text(srt_text, encoding="utf-8")

        print(f"[{idx}/{len(subset)}] {vid} -> {draft_path.name}, {creative_path.name}, {srt_path.name}")

if __name__ == "__main__":
    main()

lyrics_v1.txt
lyrics_creative_v1.txt
lyrics_v1.srt
lyrics_v2.txt
lyrics_creative_v2.txt
lyrics_v2.srt
...

이렇게 새 가사=새 파일 로 누적되도록 변경시켰다.

---

음악화

아이디어 요약 (어떻게 다르게 만들까?)

• Key: C/G/etc. major → 해당 조의 음계(도레미) 사용
• BPM: 곡 속도 결정 (가사마다 다르게)
• pitch_range: 한 옥타브 이내로 제한 (아이들 따라 부르기 쉽게)
• repeat_score: 반복성이 높을수록 모티프(짧은 패턴) 재사용 ↑, 낮으면 변형 ↑
• mood: 리듬 밀도/쉼표 비율에 반영 (Playful/Excited=노트 많게, Calm=적게)

스크립트: 가사 → 멜로디(MIDI+WAV) 자동 생성