【切り抜き自動化/Python】YouTubeアーカイブのチャット/コメント取得、切り抜き箇所特定、動画編集(yt-dlp/moviepy)

#!/bin/sh
# 作業用ルートディレクトリで実行 source src/chat.sh arg1 arg2 [arg3 arg4 arg5 arg6]
# 引数(必須2個 + 任意4個): 作成するディレクトリのパス, YouTubeのChannel ID, 検索文字列(厳しめ), 検索文字列(緩め), 取得開始日(YYYYMMDD), 取得終了日(YYYYMMDD)

current_dir=`pwd`
source venv/yt-dlp_moviepy/bin/activate
if [ ! -d "$1"/live_chat ]
then
  mkdir -p "$1"/live_chat
  cd "$1"/live_chat
  if [ $# -ge 6 ]
  then
    yt-dlp --skip-download --write-subs --write-comments --dateafter $5 --datebefore $6 -o "%(upload_date)s[%(id)s]" "https://www.youtube.com/channel/$2"
  elif [ $# == 5 ]
  then
    yt-dlp --skip-download --write-subs --write-comments --dateafter $5 -o "%(upload_date)s[%(id)s]" "https://www.youtube.com/channel/$2"
  else
    yt-dlp --skip-download --write-subs --write-comments -o "%(upload_date)s[%(id)s]" "https://www.youtube.com/channel/$2"
  fi
  cd "${current_dir}"
fi
if [ $# -ge 3 ] && [ ! -e "$1"/extract/results.txt ]
then
  mkdir -p "$1"/extract
  cd "$1"
  if [ $# -ge 4 ]
  then
    python "${current_dir}"/src/clustering_chat.py "$3" "$4"
  else
    python "${current_dir}"/src/clustering_chat.py "$3"
  fi
  cd "${current_dir}"
fi
deactivate

import sys # argv
import os # listdir
import re # search
import unicodedata # normalize

def subStrBegin(str, str_begin, str_end): # 該当範囲の文字列を切り出し(開始文字列から検索)
  begin = str.find(str_begin) + len(str_begin)
  if begin < len(str_begin):
    return ""
  end = str[begin:].find(str_end) + begin
  return str[begin:end]

def subStrEnd(str, str_begin, str_end): # 該当範囲の文字列を切り出し(終了文字列から検索)
  end = str.find(str_end)
  if end < 0:
    return ""
  begin = str[:end].rfind(str_begin) + len(str_begin)
  return str[begin:end]

def getId(filename): # ファイル名からVideoIDを抽出
  return subStrEnd(filename, "[", "]")

def getDate(filename): # ファイル名から投稿日を抽出
  return subStrEnd(filename, "/", "[")

def getTitle(line): # 生データからタイトルを抽出
  return subStrBegin(line, '"title": "', '"')

def getReleaseDate(line): # 生データから公開日を抽出
  return subStrBegin(line, '"release_date": "', '"')

def getSecond(line): # 生データから時刻を抽出
  str = subStrBegin(line, '"videoOffsetTimeMsec": "', '"')[:-3] # mili sec で記載されているため下3文字削る
  if str == "":
    return 0
  return int(str)

def getText(line): # 生データからチャットを抽出
  text = ""
  str_begin = '"text": "'
  str_end = '"'
  while True: # チャットがいくつかに分割される場合があるためすべて抽出して結合
    begin = line.find(str_begin) + len(str_begin)
    if begin < len(str_begin):
      break
    end = line[begin:].find(str_end) + begin
    text += line[begin:end]
    line = line[end + len(str_end):]
  return text

def exchangeToYen(amount): # 円に両替した場合の金額
  YEN_PER = {"¥": 1, "$": 150, "€": 160, "₩": 0.1, "£": 190, "₱": 2.5, "₹": 1.8, "₫": 0.006, "₪": 41, "MYR": 32, "CA$": 110, "NT$": 4.7, "THB": 4, "HK$": 19, "A$": 97, "ARS": 0.17, "PLN": 37, "MX$": 8.7, "CLP": 0.15, "IDR": 0.01, "SGD": 110, "R$": 30, "ZAR": 7.9, "RON": 32, "NOK": 14, "NZ$": 90, "RUB": 1.6, "DKK": 22, "SEK": 14, "CHF": 170, "PEN": 40, "CRC": 0.3, "RSD": 1.4, "UYU": 3.8, "DOP": 2.5, "ISK": 1.1, "SAR": 40, "HUF": 40, "CZK": 6.4, "BGN": 82, "BYN": 45, "GTQ": 19, "BOB": 21, "PYG": 0.02, "TRY": 4.6, "COP": 0.038, "HRK": 20, "AED": 40, "KES": 1, "NIO": 4, "ден": 2.6}
  # 為替レートは2024/03/10時点、有効数字2桁程度。未記載の通貨は0円で換算
  amount_normalized = unicodedata.normalize("NFKC", amount).replace(" ", "").replace(",", "")
  if amount_normalized == "":
    return 0
  for symbol in YEN_PER:
    if amount_normalized[:len(symbol)] == symbol:
      return float(amount_normalized[len(symbol):]) * YEN_PER[symbol]
  return 0

def getYenSuperchat(line): # 生データからスーパーチャット金額を抽出
  return exchangeToYen(subStrBegin(line, '"purchaseAmountText": {"simpleText": "', '"'))

def isInfo(filename): # コメントならTrue, チャットならFalseを返す
  return subStrBegin(filename, "].", ".") == "info"

def isValidChat(line): # データが重複していないか確認
  if line.find("addChatItemAction") < 0:
    return False
  return True

def containStr(text, query): # 文字列が含まれるか判定
  return re.search(query, text) != None

def getCommentList(line, query): # 生データから該当コメントを抽出してリスト化
  list_comment = []
  str_begin = '"text": "'
  str_end = '"'
  while True: # 全コメントを抽出
    begin = line.find(str_begin) + len(str_begin)
    if begin < len(str_begin):
      break
    end = line[begin:].find(str_end) + begin
    comment = line[begin:end]
    if containStr(comment, query):
      list_comment.append(comment)
    line = line[end + len(str_end):]
  return list_comment

def secondToTime(second_src): # 秒数(int)から時間表示(str)に変換
  SECOND_PER_MINUTE = 60
  MINUTE_PER_HOUR = 60
  second = second_src % SECOND_PER_MINUTE
  second_src = int((second_src - second) / SECOND_PER_MINUTE)
  minute = second_src % MINUTE_PER_HOUR
  second_src = int((second_src - minute) / MINUTE_PER_HOUR)
  hour = second_src
  return str(hour) + ":" + str(minute).zfill(2) + ":" + str(second).zfill(2)

def timeToDisplayTime(str): # 時間を表示用に変換(yyyyMMdd -> yyyy/MM/dd)
  return str[0:4] + "/" + str[4:6] + "/" + str[6:8]

def clusteringChat(dir, list_query): # チャットを1事象ごとにクラスタリング
  SEC_CLUSTERING = 90 # チャット間隔が90秒未満の場合、同じ事象に対するコメントだと判定(初期値)
  SEC_PRE = 15 # lv0チャット1個目の何秒前からチャット数を数えるか
  MAX_EXAMPLE_CHAT = 24 # 参考例として出力する該当チャット数の最大値
  MAX_EXAMPLE_COMMENT = 24 # 参考例として出力する該当コメント数の最大値
  results = [] # [n][0]: VideoID, [n][1]: 投稿日, [n][2]: 開始秒数, [n][3]: 終了秒数, [n][4]: チャット/コメント数, [n][5][m]: チャット/コメント例 (m <= 12), [n][6]: 投げ銭金額, [n][7]: 公開日
  list_date_title = [] # [n][0]: VideoID, [n][1]: 投稿日, [n][2]: タイトル

  query_lv = [""] * 3 # 検索条件厳しい順
  for i in range(len(list_query)):
    query_lv[i] = list_query[i]
  for i in range(len(list_query), len(query_lv) - 1):
    query_lv[i] = query_lv[len(list_query) - 1]
  
  list_file = os.listdir(dir)
  list_file.sort() # 日付順に並び替え 日付が同じ場合はID順
  BUF_SIZE = 1000 # 下記リングバッファのサイズ
  for filename in list_file:
    buf_lv1_count = [(0, 0)] * BUF_SIZE # query_lv[1] に該当するチャットの秒数と金額を保持するリングバッファ
    buf_lv1_itr = 0 # 上記リングバッファのイテレータ
    sec_end_same = 0 # 同じ事象だと判定される期限
    release_date = "00000000"
    with open(dir + filename) as f:
    # 入力データは、1チャットにつき1行。コメントは全データが1行
      id = getId(filename)
      date = getDate(filename)
      if date.isdecimal() and int(release_date) < int(date): # 投稿日 < 公開日 が成り立つように。ただし、infoファイルから読み込まれるはずなので、このif文の有無で結果は変わらないはず
        release_date = date
      if isInfo(filename): # コメントの場合、開始終了秒数を0として追加
        for line in f:
          list_date_title.append((id, date, getTitle(line)))
          if getReleaseDate(line) != "":
            release_date = getReleaseDate(line)
          if query_lv[0] != "":
            list_comment = getCommentList(line, query_lv[0])
            if len(list_comment) > 0: # コメントに検索文字列が含まれていない場合は除外
              results.append([id, date, 0, 0, len(list_comment), list_comment[0:MAX_EXAMPLE_COMMENT], 0, release_date])
      else: # チャットの場合
        for line in f:
          if not isValidChat(line): # チャットが既存データと重複する場合は除外
            continue
          chat_text = getText(line)
          yen = getYenSuperchat(line)
          contain_lv1 = query_lv[0] != "" and containStr(chat_text, query_lv[1])
          if not contain_lv1 and yen <= 0: # チャットに検索文字列が含まれていない場合は除外
            continue
          chat_second = getSecond(line)
          is_same = len(results) > 0 and results[-1][0] == id and results[-1][3] > 0 and chat_second <= sec_end_same
          contain_lv0 = query_lv[0] != "" and containStr(chat_text, query_lv[0])
          if contain_lv0 and not is_same:
          # チャット間隔が SEC_CLUSTERING 秒未満の場合、同じ事象に対するコメントだと判定
            count_pre = 0 # 最初のlv0チャット以前のlv1チャット数 (SEC_PRE 秒前まで)
            yen_pre = 0
            for itr in range(buf_lv1_itr - 1, buf_lv1_itr - 1 - BUF_SIZE, -1):
              if buf_lv1_count[itr][0] <= max(0, chat_second - SEC_PRE):
                break
              count_pre += 1
              yen_pre += buf_lv1_count[itr][1]
            results.append([id, date, chat_second, chat_second, count_pre, [], yen_pre, release_date])
            sec_end_same = 0
            is_same = True
          if is_same:
            results[-1][3] = max(chat_second, results[-1][3])
            results[-1][4] += 1
            results[-1][6] += yen
            if len(results[-1][5]) < MAX_EXAMPLE_CHAT and contain_lv0: # チャット例を MAX_EXAMPLE_CHAT 個まで保持
              results[-1][5].append(chat_text)
            sec_end_same += (chat_second + SEC_CLUSTERING - sec_end_same) * SEC_CLUSTERING / (SEC_CLUSTERING + (chat_second - results[-1][2]) * 4) # 期限の加算時間を徐々に小さくする
          else:
            buf_lv1_count[buf_lv1_itr] = (chat_second, yen)
            buf_lv1_itr = (buf_lv1_itr + 1) % BUF_SIZE

  return results, list_date_title

def writeResults(results, path): # 結果を出力: URL チャット/コメント数 金額 開始秒数 終了秒数 投稿日 チャット例
  DELIMITER = " " # 出力時の区切り文字
  NEWLINE = "\n" # 改行文字
  SEC_BUFFER = 30 # チャット1個目の何秒前から動画を確認するか
  with open(path, "w") as f:
    for (id, _, sec_begin, sec_end, count, list_text, yen, release_date) in results:
      second = max(sec_begin - SEC_BUFFER, 0)
      f.write("https://youtu.be/" + id + "?t=" + str(second) + "s" + DELIMITER)
      f.write(str(count) + DELIMITER)
      f.write(str(int(yen)) + DELIMITER)
      f.write(secondToTime(sec_begin) + DELIMITER)
      f.write(secondToTime(sec_end) + DELIMITER)
      f.write(timeToDisplayTime(release_date) + DELIMITER)
      for text in list_text:
        f.write(text + DELIMITER)
      f.write(NEWLINE)

def writeListDateTitle(list_date_title, path): # ID,日付,タイトルを出力
  DELIMITER = " " # 出力時の区切り文字
  NEWLINE = "\n" # 改行文字
  with open(path, "w") as f:
    for (id, date, title) in list_date_title:
      f.write(id + DELIMITER)
      f.write(date + DELIMITER)
      f.write(title + NEWLINE)

def execute(dir_src, path_results, path_list_date_title, list_query):
  results, list_date_title = clusteringChat(dir_src, list_query)
  writeResults(results, path_results)
  writeListDateTitle(list_date_title, path_list_date_title)

def main():
  execute("live_chat/", "extract/results.txt", "extract/list_date_title.txt", sys.argv[1:])

if __name__ == "__main__":
  main()

#!/bin/sh
# 作業用ルートディレクトリで実行 source src/video.sh arg1 [option]
# 引数(必須1個 + オプション): 作成したディレクトリのパス, オプション (-a: 動画全編をダウンロードする場合, -d: ダウンロードした元動画を削除する場合)

current_dir=`pwd`
source venv/yt-dlp_moviepy/bin/activate
cd "$1"
mkdir download clip
python "${current_dir}"/src/download_clip.py ${@:2}
mkdir dst
cd dst
python "${current_dir}"/src/edit_video.py
cd "${current_dir}"
deactivate

import sys # argv
import os # remove, path.isfile
import operator # itemgetter
import csv # reader
import glob # glob, escape
import yt_dlp # YoutubeDL
from moviepy.editor import *

def subStrBegin(str, str_begin, str_end): # 該当範囲の文字列を切り出し(開始文字列から検索)
  begin = str.find(str_begin) + len(str_begin)
  if begin < len(str_begin):
    return ""
  end = str[begin:].find(str_end) + begin
  return str[begin:end]

def timeToSecond(str): # 時間表示(str)から秒数(float)に変換
  SECOND_PER_MINUTE = 60
  MINUTE_PER_HOUR = 60
  DELIMITER = ":" # 時間表示の区切り文字
  index_delimiter = str.find(DELIMITER)
  hour = int(str[:index_delimiter])
  str = str[index_delimiter + len(DELIMITER):]
  index_delimiter = str.find(DELIMITER)
  minute = int(str[:index_delimiter])
  str = str[index_delimiter + len(DELIMITER):]
  second = float(str)
  return (hour * MINUTE_PER_HOUR + minute) * SECOND_PER_MINUTE + second

def getVideoPath(filename): # 拡張子も含めた動画のパスを取得
  list_extension = ("mp4", "webm", "mkv")
  for extension in list_extension:
    path = filename + "." + extension
    if os.path.isfile(path):
      return path
  print("file not found: " + filename)
  return ""

def getDictDateTitle(path): # ファイルから日付とタイトルの辞書を取得
  if not os.path.isfile(path):
    return {}
  dict_date_title = {}
  with open(path) as f:
    for line in f:
      dict_date_title[line[:11]] = (line[12:20], line[21:-1])
  return dict_date_title

def getResults(path, dict_date_title): # ファイルから結果を取得
  DELIMITER = " " # 区切り文字
  results = [] # [n][0]: VideoID, [n][1]: 投稿日, [n][2]: 開始秒数, [n][3]: 終了秒数
  with open(path) as f:
    reader = csv.reader(f, delimiter=DELIMITER)
    for row in reader:
      if len(row) <= 0:
        continue
      if row[0][:4] != "http": # 行頭に変更がなければ除外
        id = subStrBegin(row[0], "youtu.be/", "?")
        date = row[5]
        if id in dict_date_title:
          date = dict_date_title[id][0]
        results.append((id, date, timeToSecond(row[3]), timeToSecond(row[4])))
  return results

def downloadAllAndClip(results, dir_download, dir_clip, remove_original): # 各動画のダウンロードと切り抜き
  option = {
    "outtmpl": dir_download + "%(upload_date)s[%(id)s].%(ext)s", # 出力形式 投稿日[動画ID].mp4
    #"format": "bv*[ext=mp4]+ba[ext=m4a]/b[ext=mp4] / bv*+ba/b", # Download the best mp4 video available, or the best video if no mp4 available
    "ignoreerrors": True, # エラーを無視して続行
  }
  MAX_RETRY_DOWNLOAD = 6 # ダウンロードに失敗した際の最大再試行回数
  results.sort(key=operator.itemgetter(1, 0, 2, 3)) # 日付順で、動画ごとにすべての切り抜きを作成
  len_results = len(results)
  with yt_dlp.YoutubeDL(option) as ydl:
    for i in range(len_results):
      (id, date, sec_begin, sec_end) = results[i]
      url = "https://www.youtube.com/watch?v=" + id
      filename = date + "[" + id + "]"
      filename_download = dir_download + filename
      path_download = ""
      str_sec_begin = str(int(sec_begin * 1000)).zfill(8)
      str_sec_end = str(int(sec_end * 1000)).zfill(8)
      path_clip = dir_clip + filename + "_" + str_sec_begin + "-" + str_sec_end + ".mp4"
      if not os.path.isfile(path_clip): # 出力先ファイルが既に存在する場合は動画生成しない
        retry_download = MAX_RETRY_DOWNLOAD
        while retry_download > 0:
          path_download = getVideoPath(filename_download)
          if path_download != "": # 存在しない場合のみダウンロード
            break
          retry_download -= 1
          ydl.download([url])
        if path_download != "": # ダウンロードに失敗した場合はスキップ
          if sec_begin < sec_end: # 時刻指定に誤りがある場合はスキップ
            videoclip = VideoFileClip(path_download).subclip(sec_begin, sec_end)
            videoclip.write_videofile(path_clip, codec="mpeg4", bitrate="1000000000")
      if remove_original and (i == len_results - 1 or id != results[i + 1][0]): # 用済みになった動画は、オプションがTrueなら削除
        if os.path.isfile(path_download):
          os.remove(path_download)

def downloadOnlyClip(results, dir_download, dir_clip, remove_original): # 切り抜き部分のみをダウンロード
  MAX_RETRY_DOWNLOAD = 6 # ダウンロードに失敗した際の最大再試行回数
  SEC_INTERVAL_BEGIN_DOWNLOAD = 12 # 切り抜き前に指定秒数を追加した部分をダウンロード
  SEC_INTERVAL_END_DOWNLOAD = 4 # 切り抜き後に指定秒数を追加した部分をダウンロード
  SEC_INTERVAL_BEGIN_CLIP = 8 # 前に指定秒数以上が存在するダウンロード動画から切り抜きを作成
  SEC_INTERVAL_END_CLIP = 0.1 # 後に指定秒数以上が存在するダウンロード動画から切り抜きを作成
  results.sort(key=operator.itemgetter(1, 0, 2, 3)) # 日付順にダウンロード
  len_results = len(results)
  for i in range(len_results):
    (id, date, sec_begin, sec_end) = results[i]
    if sec_begin >= sec_end: # 時刻指定に誤りがある場合はスキップ
      continue
    url = "https://www.youtube.com/watch?v=" + id
    filename = date + "[" + id + "]"
    filename_download_pre = dir_download + filename
    path_download = ""
    str_sec_begin = str(round(sec_begin * 1000)).zfill(8)
    str_sec_end = str(round(sec_end * 1000)).zfill(8)
    path_clip = dir_clip + filename + "_" + str_sec_begin + "-" + str_sec_end + ".mp4"
    if not os.path.isfile(path_clip): # 出力先ファイルが既に存在する場合は動画生成しない
      retry_download = MAX_RETRY_DOWNLOAD
      sec_begin_download = 0
      sec_end_download = 0
      while retry_download > 0:
        for downloaded in glob.glob(glob.escape(filename_download_pre) + "*"):
          index_extension = downloaded.rfind(".")
          index_id_end = downloaded[:index_extension].rfind("]")
          if index_id_end + 1 == index_extension: # 動画が開始から終了まですべてダウンロードされている場合
            sec_begin_download = 0
            sec_end_download = 0
            path_download = downloaded
            break
          else: # 切り抜き付近のみがダウンロードされている場合
            index_sec_end = downloaded[:index_extension].rfind("-")
            index_sec_begin = downloaded[:index_sec_end].rfind("_")
            if downloaded[index_sec_end + 1:index_extension].isdecimal(): # partファイルの場合は無視
              sec_begin_download = int(downloaded[index_sec_begin + 1:index_sec_end]) / 1000
              sec_end_download = int(downloaded[index_sec_end + 1:index_extension]) / 1000
              if (sec_begin_download <= sec_begin - SEC_INTERVAL_BEGIN_CLIP or sec_begin_download == 0) and sec_end_download >= sec_end + SEC_INTERVAL_END_CLIP: # 適するダウンロード動画が存在する場合
                path_download = downloaded
                break
        if path_download != "":
          break
        sec_begin_download = round(sec_begin - SEC_INTERVAL_BEGIN_DOWNLOAD)
        if sec_begin_download < 0:
          sec_begin_download = 0
        sec_end_download = round(sec_end + SEC_INTERVAL_END_DOWNLOAD)
        str_sec_begin_download = str(round(sec_begin_download * 1000)).zfill(8)
        str_sec_end_download = str(round(sec_end_download * 1000)).zfill(8)
        filename_download = dir_download + filename + "_" + str_sec_begin_download + "-" + str_sec_end_download
        retry_download -= 1
        option = {
          "outtmpl": filename_download + ".%(ext)s", # 出力形式 投稿日[動画ID]_開始ミリ秒-終了ミリ秒.mp4
          #"format": "bv*[ext=mp4]+ba[ext=m4a]/b[ext=mp4] / bv*+ba/b", # Download the best mp4 video available, or the best video if no mp4 available
          "ignoreerrors": True, # エラーを無視して続行
          "download_ranges": lambda info_dict, ydl: [{"start_time": sec_begin_download, "end_time": sec_end_download}]
        }
        with yt_dlp.YoutubeDL(option) as ydl:
          ydl.download([url])
      if path_download != "":
        videoclip = VideoFileClip(path_download) # ダウンロードの開始終了秒数をyt-dlpが厳守しない(少し長めになる)ので、切り抜き位置を調整する。現在は、終了秒数が一致している想定で調整している
        sec_begin_clip = videoclip.duration - sec_end_download + sec_begin # = videoclip.duration - (sec_end_download - sec_begin_download) + (sec_begin - sec_begin_download)
        sec_end_clip = videoclip.duration - sec_end_download + sec_end
        subclip = videoclip.subclip(sec_begin_clip, sec_end_clip)
        subclip.write_videofile(path_clip, codec="mpeg4", bitrate="1000000000")
    if remove_original and (i == len_results - 1 or id != results[i + 1][0]): # 用済みになった動画は、オプションがTrueなら削除
      for downloaded in glob.glob(glob.escape(filename_download_pre) + "*"):
        os.remove(downloaded)

def execute(path_results, path_list_date_title, dir_download, dir_clip, download_all, remove_original):
  dict_date_title = getDictDateTitle(path_list_date_title)
  results = getResults(path_results, dict_date_title)
  if download_all:
    downloadAllAndClip(results, dir_download, dir_clip, remove_original)
  else:
    downloadOnlyClip(results, dir_download, dir_clip, remove_original)

def main():
  download_all = False # 全編をダウンロードするか
  remove_original = False # ダウンロードした元動画を削除するか
  for arg in sys.argv[1:]:
    download_all = download_all or (arg == "-a" or arg == "-A" or arg == "--all")
    remove_original = remove_original or (arg == "-d" or arg == "-D" or arg == "--delete")
  execute("extract/results.txt", "extract/list_date_title.txt", "download/", "clip/", download_all, remove_original)

if __name__ == "__main__":
  main()

import os # path.isfile, rename
import csv # reader
import numpy # array, argmin, abs
from moviepy.editor import *
from moviepy.video.fx.resize import resize

def subStrBegin(str, str_begin, str_end): # 該当範囲の文字列を切り出し(開始文字列から検索)
  begin = str.find(str_begin) + len(str_begin)
  if begin < len(str_begin):
    return ""
  end = str[begin:].find(str_end) + begin
  return str[begin:end]

def timeToSecond(str): # 時間表示(str)から秒数(float)に変換
  SECOND_PER_MINUTE = 60
  MINUTE_PER_HOUR = 60
  DELIMITER = ":" # 時間表示の区切り文字
  index_delimiter = str.find(DELIMITER)
  hour = int(str[:index_delimiter])
  str = str[index_delimiter + len(DELIMITER):]
  index_delimiter = str.find(DELIMITER)
  minute = int(str[:index_delimiter])
  str = str[index_delimiter + len(DELIMITER):]
  second = float(str)
  return (hour * MINUTE_PER_HOUR + minute) * SECOND_PER_MINUTE + second

def secondToTime(second_src): # 秒数(float)から時間表示(str)に変換
  SECOND_PER_MINUTE = 60
  MINUTE_PER_HOUR = 60
  second_src = round(second_src) # 小数点以下は四捨五入
  second = second_src % SECOND_PER_MINUTE
  second_src = int((second_src - second) / SECOND_PER_MINUTE)
  minute = second_src % MINUTE_PER_HOUR
  second_src = int((second_src - minute) / MINUTE_PER_HOUR)
  hour = second_src
  return str(hour) + ":" + str(minute).zfill(2) + ":" + str(second).zfill(2)

def getVideoPath(filename): # 拡張子も含めた動画のパスを取得
  list_extension = ("mp4", "webm", "mkv")
  for extension in list_extension:
    path = filename + "." + extension
    if os.path.isfile(path):
      return path
  print("file not found: " + filename)
  return ""

def getDictDateTitle(path): # ファイルから日付とタイトルの辞書を取得
  if not os.path.isfile(path):
    return {}
  dict_date_title = {}
  with open(path) as f:
    for line in f:
      dict_date_title[line[:11]] = (line[12:20], line[21:-1])
  return dict_date_title

def getResults(path, dict_date_title): # ファイルから結果を取得
  DELIMITER = " " # 区切り文字
  results = [] # [n][0]: VideoID, [n][1]: 投稿日, [n][2]: 開始秒数, [n][3]: 終了秒数, [n][4]: チャット数, [n][5]: コメント数, [n][6]: 金額, [n][7]: タイトル, [n][8]: 公開日(表示用文字列)
  dict_count_comment = {}
  with open(path) as f:
    reader = csv.reader(f, delimiter=DELIMITER)
    for row in reader:
      if len(row) <= 0:
        continue
      is_comment = row[4] == "0:00:00" # 終了時刻が0ならコメントと判定
      id = subStrBegin(row[0], "youtu.be/", "?")
      if is_comment:
        dict_count_comment[id] = int(row[1])
      if row[0][:4] != "http": # 行頭に変更がなければ除外
        count_chat = 0
        if not is_comment:
          count_chat = int(row[1])
        date = row[5]
        title = ""
        if id in dict_date_title:
          date = dict_date_title[id][0]
          title = dict_date_title[id][1]
        results.append([id, date, timeToSecond(row[3]), timeToSecond(row[4]), count_chat, 0, int(row[2]), title, row[5]])
  for result in results: # コメント数をすべてに反映
    if result[0] in dict_count_comment:
      result[5] = dict_count_comment[result[0]]
  return results

def displayText(release_date, count_chat, count_comment, yen): # 切り抜き動画中に表示する文字
  DISPLAY_DATE = True # 公開日の表示オプション
  DISPLAY_COUNT = True # 該当チャット数の表示オプション
  DISPLAY_YEN = True # スーパーチャット金額(円)の表示オプション
  NEWLINE = "\n" # 改行文字
  display_text = ""
  if DISPLAY_DATE:
    display_text += release_date + NEWLINE
  if DISPLAY_COUNT and count_chat > 0:
    display_text += "関連チャット数: " + str(count_chat) + NEWLINE
  if DISPLAY_COUNT and count_comment > 0:
    display_text += "関連コメント数: " + str(count_comment) + NEWLINE
  if DISPLAY_YEN and yen > 0:
    display_text += "スパチャ総額: ¥" + str(yen) + NEWLINE
  if len(display_text) > 0:
    display_text = display_text[:- len(NEWLINE)]
  return display_text

def getResolution(results, dir): # 全切り抜きのうち、該当数が最も多い解像度を取得
  LIST_HEIGHT_16_9 = [144, 360, 480, 720, 1080, 1440, 2160, 4320] # 一般的な16:9の解像度(の高さ)一覧
  list_count = [0] * len(LIST_HEIGHT_16_9)
  list_resolution = []
  for (id, date, sec_begin, sec_end, _, _, _, _, _) in results:
    str_sec_begin = str(int(sec_begin * 1000)).zfill(8)
    str_sec_end = str(int(sec_end * 1000)).zfill(8)
    path = getVideoPath(dir + date + "[" + id + "]_" + str_sec_begin + "-" + str_sec_end)
    video = VideoFileClip(path)
    index_nearest = numpy.argmin(numpy.abs(numpy.array(LIST_HEIGHT_16_9) - video.h))
    list_count[index_nearest] += 1
    list_resolution.append((video.h, video.w))
  height = LIST_HEIGHT_16_9[list_count.index(max(list_count))]
  width = int(height * 16 / 9)
  list_target_resolution = []
  for (height_original, width_original) in list_resolution:
    height_target = height_original
    width_target = width_original
    if height_target > height or width_target > width or (height_target != height and width_target != width): # 共通解像度に一致するようサイズ変更
      height_target = min(height, int(width * height_original / width_original))
      width_target = min(width, int(height * width_original / height_original))
    list_target_resolution.append((height_target, width_target))
  return (width, height), list_target_resolution # sizeは(x, y), target_resolutionは(y, x)。不満はmoviepy開発陣へ

def subClip(resolution, target_resolution, title, text, path, path_font=""): # 各切り抜きのサイズを合わせ、文字とフェード効果を付与
  SEC_FADEIN = 1 # フェードイン秒数
  SEC_FADEOUT = 1 # フェードアウト秒数
  SEC_AUDIO_FADEIN = 0.1 # 音のフェードイン秒数
  SEC_AUDIO_FADEOUT = 0.1 # 音のフェードアウト秒数
  COLOR_FONT = "#ffffff" # フォント色
  COLOR_BACKGROUND = "#000000c0" # テキストの背景色
  (width, height) = resolution
  (height_target, width_target) = target_resolution
  videoclip = VideoFileClip(path, target_resolution=target_resolution).set_position((int((width - width_target) / 2), int((height - height_target) / 2))) # 指定の解像度で動画読み込み
  fontsize = int(height / 20)
  titlesize = int(fontsize / 2)
  position_title = (16, 8)
  position_text = (16, 16 + titlesize + 1)
  duration = videoclip.duration
  if path_font == "":
    titleclip = TextClip(txt=title, fontsize=titlesize, color=COLOR_FONT, bg_color=COLOR_BACKGROUND).set_position(position_title).set_duration(duration)
    textclip = TextClip(txt=text, fontsize=fontsize, color=COLOR_FONT, bg_color=COLOR_BACKGROUND).set_position(position_text).set_duration(duration)
  else:
    titleclip = TextClip(txt=title, font=path_font, fontsize=titlesize, color=COLOR_FONT, bg_color=COLOR_BACKGROUND).set_position(position_title).set_duration(duration)
    textclip = TextClip(txt=text, font=path_font, fontsize=fontsize, color=COLOR_FONT, bg_color=COLOR_BACKGROUND).set_position(position_text).set_duration(duration)
  return CompositeVideoClip(clips=[videoclip.audio_fadein(SEC_AUDIO_FADEIN).audio_fadeout(SEC_AUDIO_FADEOUT), titleclip, textclip], size=resolution).fadein(SEC_FADEIN).fadeout(SEC_FADEOUT)

def mergeClip(results, dir, path_dst_video, path_font): # 全切り抜きを結合
  SEC_PART = 60 * 12 # まとめて処理できる最大秒数。この秒数ごとの動画を生成し、最後に結合する
  list_video = []
  list_duration = []
  resolution, list_target_resolution = getResolution(results, dir)
  len_results = len(results)
  path_dst_video_pre = path_dst_video[:path_dst_video.rfind(".")] # 拡張子より前
  path_dst_video_extension = path_dst_video[path_dst_video.rfind("."):]
  part_current = 0
  duration_current = 0
  for i in range(len_results):
    (id, date, sec_begin, sec_end, count_chat, count_comment, yen, title, release_date) = results[i]
    str_sec_begin = str(int(sec_begin * 1000)).zfill(8)
    str_sec_end = str(int(sec_end * 1000)).zfill(8)
    path = getVideoPath(dir + date + "[" + id + "]_" + str_sec_begin + "-" + str_sec_end)
    text = displayText(release_date, count_chat, count_comment, yen)
    list_video.append(subClip(resolution, list_target_resolution[i], title, text, path, path_font))
    list_duration.append(list_video[-1].duration)
    duration_current += list_video[-1].duration
    if duration_current >= SEC_PART or i == len_results - 1: # まとめて処理できる許容量を超えた場合、現在のリスト内の動画をすべて結合して出力
      path_current = path_dst_video_pre + "_part" + str(part_current) + path_dst_video_extension
      concatenate_videoclips(list_video).write_videofile(path_current, codec="mpeg4", bitrate="1000000000")
      list_video.clear()
      part_current += 1
      duration_current = 0
  if part_current == 1: # 許容量に到達しなかった場合は結合処理が不要
    os.rename(path_dst_video_pre + "_part0" + path_dst_video_extension, path_dst_video)
  else:
    list_part = []
    for i in range(part_current):
      path_current = path_dst_video_pre + "_part" + str(i) + path_dst_video_extension
      list_part.append(VideoFileClip(path_current))
    concatenate_videoclips(list_part).write_videofile(path_dst_video, codec="mpeg4", bitrate="1000000000")
  return list_duration

def generateTimestamp(results, list_duration): # タイムスタンプ用の文字列を生成 各切り抜きの開始時刻、投稿日、URL
  NEWLINE = "\n" # 改行文字
  SEC_CLUSTERING = 90 # 間隔が90秒未満の場合、同じ事象に対する切り抜きだと判定(初期値)。その場合、それらのタイムスタンプのカウントを同じにする (例: 1. 2-1. 2-2. 3. 4. ...)
  timestamp = ""
  len_results = len(results)
  sec_sum = 0
  count_num = 0
  count_sequence = 0
  for i in range(len_results):
    (id, date, sec_begin, _, _, _, _, _, release_date) = results[i]
    url = "https://youtu.be/" + id + "?t=" + str(int(sec_begin)) + "s"
    if i > 0 and id == results[i - 1][0] and sec_begin < results[i - 1][3] + SEC_CLUSTERING:
      count_sequence += 1
      timestamp += secondToTime(sec_sum) + " " + str(count_num) + "-" + str(count_sequence) + ". " + release_date + " "
    else:
      count_num += 1
      count_sequence = 1
      timestamp += secondToTime(sec_sum) + " " + str(count_num) + ". " + release_date + " "
    timestamp += url + NEWLINE
    sec_sum += list_duration[i] # 実際の動画の duration と sec_end - sec_begin では誤差(前者が最大+0.05s程度)が発生し、累積すると数秒単位の誤差になってしまう。これを防ぐため、実際の動画の duration を使う
  return timestamp

def execute(path_results, path_list_date_title, dir_video, path_dst_video, path_dst_timestamp, path_font=""):
  dict_date_title = getDictDateTitle(path_list_date_title)
  results = getResults(path_results, dict_date_title)
  list_duration = mergeClip(results, dir_video, path_dst_video, path_font)
  timestamp = generateTimestamp(results, list_duration)
  with open(path_dst_timestamp, "w") as f:
    f.write(timestamp)

def main():
  execute("../extract/results.txt", "../extract/list_date_title.txt", "../clip/", "clip.mp4", "timestamp.txt", "../../../src/font/MPLUSRounded1c-Regular.ttf")

if __name__ == "__main__":
  main()

#!/bin/sh
# 作業用ルートディレクトリで実行 source setup.sh [arg1]
# 引数(任意1個): Pythonのバージョン(3.12等。デフォルト値は3)

python_version=${1:-3}

sudo apt update
sudo apt install -y "python${python_version}-venv" imagemagick ffmpeg

mkdir -p {src,data,venv}
rm -fr venv/yt-dlp_moviepy

"python${python_version}" -m venv venv/yt-dlp_moviepy
source venv/yt-dlp_moviepy/bin/activate
pip install -U pip
pip install -U yt-dlp
pip install -U moviepy
deactivate