※2026/2/1現在の情報になります
※スクリプトや構成のアップデートに合わせて、記事も随時アップデートしています
以前よりUbuntuでTV録画・視聴サーバー（以降、TVサーバーと表記）を設置し、CMを抜いたmp4への変換にはWindowsのAmatsukazeを利用し、バックアップ処理も含めて自動化していました。しかし、TVサーバーのために2台のPCを使うのはもったいないと考え、Ubuntu上で完結するシステムへの切り替えを検討していました。
ところが、録画後にAmatsukazeを使ってCM抜きmp4を生成する自動化処理は、Windows単体もしくはWindowsとUbuntuを2台利用したものがほとんどで、Ubuntuのみで完結したシステムの記事は見つかりませんでした。
このため、TVサーバーをUbuntuで実装する方法を一から検討し、「録画→不要ファイル削除→CM抜き→mp4/m2tsバックアップ」の手順を完全自動化することができました。
本記事はUbuntu上で自動化したTVサーバー仕様をメモしたものになります。
本記事のTVサーバーのシステム構成
構築したTVサーバーのシステム構成は以下のとおりです。
mp4ファイル再生用のメディアサーバーはTVサーバー上に構築することでPC一台でシステムを完結させることもできます。しかし、mp4ファイルは最終的にNASへバックアップコピーするので、録画中のDrop発生要因を減らすことも考慮し、バックアップ先のNAS上で整理・再生するようにしました。
NASへのコピーはSFTPを利用しました。Exportされたディレクトリをマウントする形式だと電源を入れる順番などでマウントに失敗することもあるため、コピー開始直前にNAS側の異常を検知できるSFTPを使います。
m2tsはファイルサイズが大きくLAN経由でのコピーは時間がかかる上に、保存後は頻繁にアクセスすることもないことから、NASへのバックアップコピーではなくPCに直接接続したHDDを保存先にしています。
TVサーバー構築用ソフトウェア
TVサーバーを構築するソフトは、録画用に「EPGStation」、リアルタイム視聴・m2ts再生用に「KonomiTV」を利用します。EPGStationは録画した番組をm2ts形式で保存し、KonomiTVでm2tsファイルを直接再生して視聴できます。
録画用ソフトにEPGStationを使う理由は、RaspberryPi上でも動作するためです。拙宅ではTVサーバーが何らかのトラブルで録画に失敗する場合に備えてRaspberryPiで録画バックアップサーバーを動かしているため、録画ルール(時刻・条件・録画ファイル格納ディレクトリ)のデータを同期しやすいEPGStationを選択しました。
このような縛りがなければ、KonomiTVが録画予約までサポートしているEDCBを使うほうが便利かと思います。
TVサーバー構築用OS
録画用ソフトにEDCBを使えばWindows PCのみでTVサーバーを構築することはできますが、セキュリティアップデートでアプリが起動しなくなる、録画途中に負荷がかかりDrop発生の可能性がある、再起動が発生する、突発的なブルースクリーンが発生するなどで録画が失敗するリスクがつきまといます。
Ubuntuは自動アップデートなどの自動処理を抑止できる上に、コンソールを使ってファイル操作やログのチェックもできることから、UbuntuでTVサーバーを構築しています。
m2tsファイルのmp4変換とバックアップ
録画した番組を保存する場合、m2tsファイルをほかの媒体にコピーするだけで終わります。しかし、m2tsファイルは下記の点で保存・視聴には不向きです。
- 100時間で約76GBとなり、1年でTB単位でトータル保存ファイルサイズが増加するため、ストレージコストが高くつく
- スマホやタブレットへの保存（オフライン視聴）時に保存できる番組数が少なくなる
- m2tsファイルを直接再生できるプレイヤー・メディアサーバーが少ない
- CMや番宣など不要な部分が含まれているため、流し見に向いていない
この問題は、Amatsukazeを利用しCMを抜いたmp4ファイルに変換し、m2tsの代わりにmp4を保存することで解決できます。mp4に変換するとファイルサイズはおおよそ10-30%程度まで縮小できます。
ただし、mp4に変換するとm2tsに記録されているオリジナルの字幕情報、ニコニコ動画のコメント、CMなどの付帯情報が消えてしまうため、これらを残したい場合はm2tsも残しておく必要があります。また、録画予約当初は削除を前提としていても、後になってずっと保存しておきたくなることもあるので、m2tsファイルもしばらくの間は保持していたほうが安心です。
メンテナンスフリーのための自動処理の要件
録画したファイルを保存する際には、通常下記の作業を行います。
- SSD/HDD/NASの残容量確認と不要ファイルの削除
- CM抜きmp4ファイルへ変換
- NAS,HDDへのコピー
本システムはメンテナンスフリーを目的としているので、これらを自動で行う手順を考える必要があります。変換、転送処理に条件はないため自動化は簡単ですが、ファイル・フォルダの整理と不要ファイルの削除については削除したいファイルとそうでないファイルがあり保存先のストレージ容量の制約もあるため、要求に沿って削除ルールを策定し条件ごとにパラメータを変えてコントロールします。
不要ファイルの削除
EPGStationで録画したm2tsファイルの削除は、EPGstationの削除機能にまかせます。EPGStation以外からファイルを削除してしまうとEGPStationのデータベースとの不整合が発生し、トラブルの原因になるためです。
保存が必要なm2tsファイルは、自動削除処理が実行される前に外付けHDDへコピーして保存します。コピー先のディレクトリサイズが指定容量を超過していた場合は、指定容量以内に収まるまで更新日時の古いファイルから順番に自動で削除して容量オーバーを回避します。これにより、即削除によるファイルの喪失を防ぎます。最大ディレクトリサイズはディレクトリごとに定義できるようにします。
各録画ファイルへ適用する削除ルールは、EPGStationの録画予約時の保存先ディレクトリ指定機能を利用します。ファイル削除ルールごとにディレクトリを以下の4つに分け、録画時にこれらのディレクトリを指定して録画ファイルを自動で分類します。
- no_conversion: CMカットも変換も不要な番組（ニュースなど）
- delete: 一定期間保持する番組（バラエティなど）
- delete_after_watch: 視聴するまで自動消去しない番組（連ドラなど）
- keep: 再エンコードに備えてm2tsを残したい番組（ドキュメンタリー・映画など）
NAS上のmp4ファイルについてもm2tsファイルと同様のルールで削除し、コピー開始前にルールに従って削除します。
ファイル削除によって空ディレクトリもできるため、HDDコピー時に空ディレクトリも削除する処理を実行するようにします。
mp4ファイルへの変換
CMカット機能があるAmatsukazeを使用します。AmatsukazeはUbuntuに対応しているため、Ubuntu上で動作させます。
変換したmp4ファイルを格納するディレクトリはm2tsファイルと同じ構造を維持するようにし、管理の手間を省きます。ここでは/mnt/converted_filesに保存するものとします。
NAS・HDDへコピー
Pythonスクリプトで以下のように処理をします。
- m2tsファイル：直接接続された外付けHDDへコピー
- mp4ファイル： sftp経由でNASへアップロード
一度変換・コピーしたファイルは繰り返し同じ処理が走らないようにデータベースに記録し管理します。
ファイルをコピーする際には録画ファイルのディレクトリ構造を維持してコピーし、ファイル整理の管理の手間を省きます。
TVサーバー構築
必要なPCスペック
録画・再生だけであればN100等の低スペックPCでも可能ですが、Amatsukazeでの変換を裏で同時に行う場合はCPU性能が必要になります。
Core i9のミニPCをTVサーバーにして運用した場合、変換中に録画ファイルのDropは発生しておらず、1時間番組の場合ですと録画終了からNASへのアップロード完了までは10分程度で終わります。
N100の場合は同時録画をしている最中に変換処理をしているとDropが発生することがありますが、変換処理を深夜などの無録画時間帯で実行することで回避できます。
メモリは8GBでも問題はないですが、選択するCodecやアップデートによって使用メモリが増えることもあるので、Swap発生を抑えるためにも16GBは搭載していた方が安心です。
Ubuntuのインストール
インストールするPCによりインストール方法が変わるため、ここでの説明は割愛します。以下の記事などを参照してください。
EPGStation、KonomiTVのセットアップ
EPGStationとKonomiTVをUbuntuにインストールする方法は以下の記事にまとめていますので、こちらを参考にしてください。
ECDBは下記記事を参考にするとインストールできます。
Ubuntu上で動作するAmatsukazeをセットアップする
Amatsukazeのインストールや設定は環境に大きく依存するので、マニュアルを参考にしてインストールしてください。
Amatsukazeのセットアップ後、最終的に以下のコマンドで変換キューに追加し、出力先にmp4ファイルができていればよいです。
$ /home/ユーザー名/Amatsukaze/exe_files/AmatsukazeAddTask -f <m2tsファイル> -o <出力先ディレクトリ> -ip localhost -s <プロファイル名>Amatsukazeはサーバー・クライアント型のアプリなので、PC起動時にAmatsukazeServer.shを自動実行する必要があります。これはcrontabに以下の記述を追加します。
@reboot cd /home/tv-recorder/Amatsukaze/Amatsukaze && ./AmatsukazeServer.sh &自動処理用Pythonスクリプト
下記が上記要件を取り入れたPythonスクリプトになります。Windowsでも動作するとは思いますが、動作は未確認です。
スクリプトを1ファイルに収めたので、処理内容の確認やカスタマイズはGeminiやCopilotなどを利用すると簡単です。
環境依存パラメータについては、ファイル先頭に定義されているConfigクラスにまとめていますので、環境に応じて変更してください。
【重要】スクリプトにはファイル一括削除処理が入っているため、パラメーターを間違えたりバグによって意図しないファイルを削除することがあります。スクリプト利用の際には、DryRunで十分に動作を理解・確認してください。デフォルトで「dry_run: bool = True」としています。動作確認後にFalse設定にしてください。
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
TV録画統合管理スクリプト (Integrated TV Recorder Manager)
【構成】
- Config: 設定管理
- StateManager: 状態管理 (INI)
- DiskOperations (Abstract): ディスク操作の抽象化
- LocalDiskOperations: ローカル操作
- RemoteDiskOperations: SFTP操作
- Cleaner: ディスクの掃除・容量管理ロジック
- Pipelines:
- M2tsConverterPipeline (Phase 1)
- Mp4UploadPipeline (Phase 2)
- M2tsBackupPipeline (Phase 3)
- Logger: 条件付きバッファリングロガー
"""
import configparser
import shutil
import subprocess
import logging
import logging.handlers
import sys
import unicodedata
import fcntl
import posixpath
import stat
import re
import time
from abc import ABC, abstractmethod
from pathlib import Path
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import List, Dict, Generator, Optional, Tuple, Any, Union
try:
import paramiko
except ImportError:
print("【エラー】'paramiko' ライブラリが必要です: pip install paramiko", file=sys.stderr)
sys.exit(1)
# =========================================================
# 0. データ構造 & ユーティリティ
# =========================================================
@dataclass
class FileEntry:
path: Union[Path, str]
name: str
size: int
mtime: datetime
def format_bytes(size: int) -> str:
for unit in ["Byte", "KB", "MB"]:
if size < 1024:
return f"{size:.2f}{unit}" if unit != "Byte" else f"{size}{unit}"
size /= 1024
return f"{size:.2f}GB"
def normalize_str(s: str) -> str:
return unicodedata.normalize('NFC', s) if isinstance(s, str) else s
# =========================================================
# 1. 設定管理 (Configuration)
# =========================================================
@dataclass
class Config:
# --- パス設定 ---
source_dir_m2ts: Path = Path("/mnt/tv-recorder/recorded_files")
dest_dir_hdd: Path = Path("/mnt/hdd/m2ts")
converted_dir: Path = Path("/mnt/converted_files")
ini_file: Path = Path("/home/tv-recorder/Scripts/status_TvRecorder.ini")
log_file: Path = Path("/home/tv-recorder/Scripts/process_TvRecorder.log")
lock_file: Path = Path("/tmp/TvRecorder.lock")
# --- 動作設定 ---
dry_run: bool = True
write_log: bool = False
scan_threshold_sec: int = 10
m2ts_delete_days: int = 30
# --- HDDコピー実行許可時間帯 ---
copy_window_start_hour: int = 2
# --- NAS接続設定 ---
nas_config: Dict = field(default_factory=lambda: {
'host': '192.168.1.3',
'port': 22,
'user': 'tv-recorder',
'password': '********',
'key_file': None,
'dest_dir': '/tv_program/converted_files'
})
# --- 除外設定 ---
skip_folders_m2ts: List[str] = field(default_factory=lambda: ['no_conversion'])
hdd_exclude_dirs: List[str] = field(default_factory=lambda: ['keep'])
converted_exclude_dirs: List[str] = field(default_factory=list)
nas_exclude_dirs: List[str] = field(default_factory=lambda: ['/keep/', '/delete_after_watch/'])
# --- 容量上限設定 (GB) ---
max_size_hdd_gb: float = 4500.0
max_size_converted_gb: float = 180.0
max_size_nas_gb: float = 4000.0
# --- 保持ポリシー ---
retention_policies_hdd: List[Dict] = field(default_factory=lambda: [
{"dir": "delete", "days": 182, "limit_gb": 1000},
{"dir": "delete_after_watch", "days": 365, "limit_gb": 1500}
])
retention_policies_nas: List[Dict] = field(default_factory=lambda: [
{"dir": "delete", "days": 365, "limit_gb": 1500},
])
# --- Amatsukaze設定 ---
amatsukaze_cmd: str = "/home/tv-recorder/Amatsukaze/Amatsukaze/exe_files/AmatsukazeAddTask"
amatsukaze_ip: str = "localhost"
amatsukaze_service: str = "QsvEnc"
@property
def is_hdd_copy_time_window(self) -> bool:
return datetime.now().hour == self.copy_window_start_hour
# =========================================================
# 2. ロガー & 状態管理 & 排他制御
# =========================================================
class ConditionalBufferHandler(logging.Handler):
def __init__(self, log_file: Path):
super().__init__()
self.buffer = []
self.passthrough = False
if not log_file.parent.exists():
log_file.parent.mkdir(parents=True, exist_ok=True)
self.stream_handler = logging.StreamHandler(sys.stdout)
self.file_handler = logging.FileHandler(log_file, encoding='utf-8')
def setFormatter(self, fmt):
super().setFormatter(fmt)
self.stream_handler.setFormatter(fmt)
self.file_handler.setFormatter(fmt)
def emit(self, record):
if getattr(self, 'passthrough', False):
self.stream_handler.handle(record)
self.file_handler.handle(record)
else:
if not hasattr(self, 'buffer'):
self.buffer = []
self.buffer.append(record)
def flush_all(self, should_write: bool):
if should_write:
self.passthrough = True
if hasattr(self, 'buffer'):
for record in self.buffer:
self.stream_handler.handle(record)
self.file_handler.handle(record)
self.stream_handler.flush()
self.file_handler.flush()
self.buffer = []
def close(self):
self.stream_handler.close()
self.file_handler.close()
super().close()
def setup_logger(log_path: Path) -> ConditionalBufferHandler:
root = logging.getLogger()
if root.hasHandlers(): root.handlers.clear()
root.setLevel(logging.INFO)
fmt = logging.Formatter('%(asctime)s [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
handler = ConditionalBufferHandler(log_path)
handler.setFormatter(fmt)
root.addHandler(handler)
return handler
def activate_realtime_log():
root = logging.getLogger()
for h in root.handlers:
if isinstance(h, ConditionalBufferHandler):
h.flush_all(True)
def acquire_lock(lock_path: Path):
try:
lock_fd = open(lock_path, 'w')
fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
return lock_fd
except IOError:
return None
class StateManager:
SECTION_COPY_HDD = 'CopyHDD'
SECTION_CONVERT = 'Convert'
SECTION_UPLOAD_NAS = 'UploadNAS'
def __init__(self, ini_path: Path):
self.ini_path = ini_path
self.config = configparser.ConfigParser()
self.config.optionxform = str
self._load()
def _load(self):
if self.ini_path.exists():
try:
self.config.read(self.ini_path, encoding='utf-8')
except Exception: pass
for sec in [self.SECTION_COPY_HDD, self.SECTION_CONVERT, self.SECTION_UPLOAD_NAS]:
if sec not in self.config: self.config[sec] = {}
def is_recorded(self, section: str, file_name: str, file_size: int) -> bool:
if section not in self.config: return False
norm_name = normalize_str(file_name)
sec_data = {normalize_str(k): v for k, v in self.config[section].items()}
return sec_data.get(norm_name) == str(file_size)
def is_key_exists(self, section: str, file_name: str) -> bool:
if section not in self.config: return False
norm_name = normalize_str(file_name)
return norm_name in {normalize_str(k) for k in self.config[section].keys()}
def update_entry(self, section: str, file_name: str, file_size: int, dry_run: bool = False):
if dry_run:
logging.info(f" [DryRun] INI更新 [{section}]: {file_name}")
return
try:
sec_data = self.config[section]
if file_name not in sec_data:
items = list(sec_data.items())
sec_data.clear()
sec_data[file_name] = str(file_size)
for k, v in items: sec_data[k] = v
else:
sec_data[file_name] = str(file_size)
with open(self.ini_path, 'w', encoding='utf-8') as f:
self.config.write(f)
except Exception as e:
logging.error(f"INI Update Failed: {e}")
# =========================================================
# 3. ディスク操作の抽象化 (Disk Operations & Cleaner)
# =========================================================
class DiskOperations(ABC):
@abstractmethod
def list_files_recursive(self, root_dir: Union[Path, str], exclude_dirs: List[str] = None) -> Generator[FileEntry, None, None]:
pass
@abstractmethod
def delete_file(self, path: Union[Path, str]) -> bool:
pass
@abstractmethod
def remove_empty_dir(self, path: Union[Path, str]) -> bool:
pass
@abstractmethod
def exists(self, path: Union[Path, str]) -> bool:
pass
class LocalDiskOperations(DiskOperations):
def __init__(self, dry_run: bool):
self.dry_run = dry_run
def list_files_recursive(self, root_dir: Path, exclude_dirs: List[str] = None) -> Generator[FileEntry, None, None]:
exclude_dirs = exclude_dirs or []
excludes = [root_dir / d for d in exclude_dirs]
for p in root_dir.rglob('*'):
if p.is_file():
if any(ex in p.parents for ex in excludes): continue
yield FileEntry(
path=p, name=p.name, size=p.stat().st_size,
mtime=datetime.fromtimestamp(p.stat().st_mtime)
)
def delete_file(self, path: Path) -> bool:
size_str = "Unknown"
try:
size_str = format_bytes(path.stat().st_size)
except Exception: pass
if self.dry_run:
logging.info(f" [DryRun] 削除: {path} ({size_str})")
return True
try:
path.unlink()
logging.info(f" 削除: {path} ({size_str})")
return True
except Exception: return False
def remove_empty_dir(self, path: Path) -> bool:
if self.dry_run: return True
try:
path.rmdir()
return True
except OSError: return False
def exists(self, path: Path) -> bool:
return path.exists()
class RemoteDiskOperations(DiskOperations):
def __init__(self, sftp, dry_run: bool):
self.sftp = sftp
self.dry_run = dry_run
def list_files_recursive(self, root_dir: str, exclude_dirs: List[str] = None) -> Generator[FileEntry, None, None]:
exclude_markers = exclude_dirs or []
try:
for path, attr, is_dir in self._sftp_walk(root_dir):
if not is_dir:
if any(m in path for m in exclude_markers): continue
yield FileEntry(
path=path, name=posixpath.basename(path), size=attr.st_size,
mtime=datetime.fromtimestamp(attr.st_mtime)
)
except Exception: pass
def delete_file(self, path: str) -> bool:
size_str = "Unknown"
try:
size_str = format_bytes(self.sftp.stat(path).st_size)
except Exception: pass
if self.dry_run:
logging.info(f" [DryRun] 削除(NAS): {posixpath.basename(path)} ({size_str})")
return True
try:
self.sftp.remove(path)
logging.info(f" 削除(NAS): {posixpath.basename(path)} ({size_str})")
if path.endswith('.mp4'):
ass = str(Path(path).with_suffix('.ass'))
if self.exists(ass):
self.sftp.remove(ass)
return True
except Exception: return False
def remove_empty_dir(self, path: str) -> bool:
if self.dry_run: return True
try:
self.sftp.rmdir(path)
return True
except Exception: return False
def exists(self, path: str) -> bool:
try:
self.sftp.stat(path)
return True
except FileNotFoundError: return False
def _sftp_walk(self, remote_path):
try:
for attr in self.sftp.listdir_attr(remote_path):
full_path = posixpath.join(remote_path, attr.filename)
if stat.S_ISDIR(attr.st_mode):
yield from self._sftp_walk(full_path)
yield (full_path, attr, True)
else:
yield (full_path, attr, False)
except Exception: pass
class Cleaner:
def __init__(self, config: Config, ops: DiskOperations, label: str):
self.cfg = config
self.ops = ops
self.label = label
def enforce_size_limit(self, target_dir: Union[Path, str], limit_gb: float, exclude_dirs: List[str]):
if not self.ops.exists(target_dir): return
all_files = list(self.ops.list_files_recursive(target_dir, exclude_dirs=[]))
total_size = sum(f.size for f in all_files)
logging.info(f" {self.label} 現在の全体容量 [{target_dir}]: {format_bytes(total_size)}")
deletable_files = []
for f in all_files:
is_excluded = False
if isinstance(target_dir, Path):
if exclude_dirs and any(ex in f.path.parts for ex in exclude_dirs):
is_excluded = True
else:
if exclude_dirs and any(ex in f.path for ex in exclude_dirs):
is_excluded = True
if not is_excluded:
deletable_files.append(f)
deletable_files.sort(key=lambda x: x.mtime)
limit_bytes = limit_gb * (1024**3)
delete_list = []
if total_size > limit_bytes:
for f in deletable_files:
delete_list.append(f)
total_size -= f.size
if total_size <= limit_bytes: break
if delete_list:
self.cfg.write_log = True
activate_realtime_log()
logging.info(f" {self.label} 全体容量制限チェック [{target_dir}]: 上限 {limit_gb}GB -> {len(delete_list)}ファイル削除")
for f in delete_list:
self.ops.delete_file(f.path)
def apply_retention_policy(self, parent_dir: Union[Path, str], dir_name: str, days: int, limit_gb: int):
if isinstance(parent_dir, Path):
target = parent_dir / dir_name
else:
target = posixpath.join(parent_dir, dir_name)
if not self.ops.exists(target): return
files = list(self.ops.list_files_recursive(target))
total_size = sum(f.size for f in files)
logging.info(f" {self.label} 現在の容量 [{target}]: {format_bytes(total_size)}")
cutoff = datetime.now() - timedelta(days=days)
delete_list = [f for f in files if f.mtime < cutoff]
keep_list = [f for f in files if f.mtime >= cutoff]
keep_list.sort(key=lambda x: x.mtime)
limit_bytes = limit_gb * (1024**3)
current_keep_size = sum(f.size for f in keep_list)
while current_keep_size > limit_bytes and keep_list:
target_f = keep_list.pop(0)
delete_list.append(target_f)
current_keep_size -= target_f.size
if delete_list:
self.cfg.write_log = True
activate_realtime_log()
logging.info(f" {self.label} ポリシー適用 [{dir_name}]: {len(delete_list)}ファイル削除")
for f in delete_list:
self.ops.delete_file(f.path)
def delete_old_files_by_pattern(self, target_dir: Union[Path, str], days: int, pattern: str):
if not self.ops.exists(target_dir): return
logging.info(f" {self.label} 古いファイルの削除チェック: {target_dir} ({days}日以上, {pattern})")
all_files = self.ops.list_files_recursive(target_dir)
cutoff = datetime.now() - timedelta(days=days)
count = 0
for f in all_files:
if isinstance(target_dir, Path):
if not f.path.match(pattern): continue
else:
if not f.name.endswith('.m2ts'): continue
if f.mtime < cutoff:
self.cfg.write_log = True
activate_realtime_log()
if self.ops.delete_file(f.path):
count += 1
if isinstance(f.path, Path) and f.name.endswith('.m2ts'):
avs_path = f.path.with_name(f.name + ".trim.avs")
if self.ops.exists(avs_path):
self.ops.delete_file(avs_path)
if count == 0:
logging.info(" -> 削除対象なし")
else:
logging.info(f" -> {count}ファイル削除")
# =========================================================
# 4. パイプライン基底クラス
# =========================================================
class BasePipeline(ABC):
def __init__(self, config: Config, state: StateManager):
self.cfg = config
self.state = state
self.ops = LocalDiskOperations(config.dry_run)
self.cleaner = Cleaner(config, self.ops, "[Local]")
@abstractmethod
def run(self):
pass
def _cleanup_empty_dirs_local(self, root: Path, excludes: List[str] = None):
if not root.exists(): return
excludes = excludes or []
deleted_count = 0
all_dirs = sorted([p for p in root.rglob('*') if p.is_dir()], key=lambda p: len(p.parts), reverse=True)
exclude_paths = [root / e for e in excludes]
for d in all_dirs:
if d in exclude_paths or d == root: continue
try:
if not any(d.iterdir()):
if self.ops.remove_empty_dir(d):
logging.info(f" [Local] 空ディレクトリ削除: {d}")
deleted_count += 1
except OSError: pass
if deleted_count > 0:
self.cfg.write_log = True
activate_realtime_log()
# =========================================================
# 5. 各フェーズの実装
# =========================================================
class M2tsConverterPipeline(BasePipeline):
"""Phase 1: M2TS -> MP4 変換"""
def run(self):
tasks = list(self._scan())
self._cleanup()
if tasks:
self.cfg.write_log = True
activate_realtime_log()
logging.info("=== Phase 1: M2TS変換 (Converter) ===")
logging.info(f"変換対象数: {len(tasks)}")
for task in tasks:
self._process(task)
def _scan(self):
if not self.cfg.source_dir_m2ts.exists(): return
threshold = datetime.now() - timedelta(seconds=self.cfg.scan_threshold_sec)
for p in self.cfg.source_dir_m2ts.rglob('*.m2ts'):
if not p.is_file(): continue
try:
rel = p.relative_to(self.cfg.source_dir_m2ts)
if len(rel.parts) > 1 and rel.parts[0] in self.cfg.skip_folders_m2ts: continue
except ValueError: pass
if datetime.fromtimestamp(p.stat().st_mtime) >= threshold: continue
if self.state.is_recorded(StateManager.SECTION_CONVERT, p.name, p.stat().st_size):
continue
yield {'src': p, 'rel': rel, 'size': p.stat().st_size}
def _cleanup(self):
self.cleaner.enforce_size_limit(self.cfg.converted_dir, self.cfg.max_size_converted_gb, self.cfg.converted_exclude_dirs)
self._cleanup_empty_dirs_local(self.cfg.converted_dir)
def _process(self, task):
src = task['src']
logging.info(f"[Converter] 登録: {src.name} ({format_bytes(task['size'])})")
out_dir = self.cfg.converted_dir / task['rel'].parent
if self._prepare_out_dir(out_dir, src.stem) and self._exec_amatsukaze(src, out_dir):
log_file = out_dir / f"{src.stem}-enc.log"
try:
if not self.cfg.dry_run:
log_file.touch()
logging.info(f" -> エンコードログファイル作成: {log_file.name}")
except Exception as e:
logging.warning(f" -> エンコードログファイル作成失敗: {e}")
self.state.update_entry(StateManager.SECTION_CONVERT, src.name, task['size'], self.cfg.dry_run)
def _prepare_out_dir(self, out_dir, stem):
if self.cfg.dry_run: return True
try:
out_dir.mkdir(parents=True, exist_ok=True)
norm = normalize_str(stem)
for p in out_dir.iterdir():
if p.is_file() and normalize_str(p.stem).startswith(norm): p.unlink()
return True
except Exception: return False
def _exec_amatsukaze(self, src, out_dir):
cmd = [self.cfg.amatsukaze_cmd, "-ip", self.cfg.amatsukaze_ip,
"-s", self.cfg.amatsukaze_service, "-o", str(out_dir), "-f", str(src)]
if self.cfg.dry_run:
logging.info(f" [DryRun] CMD: {cmd}")
return True
try:
subprocess.run(cmd, capture_output=True, text=True, check=True)
return True
except Exception as e:
logging.error(f" -> コマンド失敗: {e}")
return False
class Mp4UploadPipeline(BasePipeline):
"""Phase 2: MP4 -> NAS 転送"""
def __init__(self, config: Config, state: StateManager):
super().__init__(config, state)
self.sftp = None
self.transport = None
def run(self):
if not self.cfg.converted_dir.exists(): return
candidates = list(self._scan_candidates())
if not candidates: return
self.cfg.write_log = True
activate_realtime_log()
logging.info("=== Phase 2: MP4転送 (Upload) ===")
self.sftp, self.transport = self._connect_sftp()
if not self.sftp: return
self.ops = RemoteDiskOperations(self.sftp, self.cfg.dry_run)
self.cleaner = Cleaner(self.cfg, self.ops, "[NAS]")
try:
self._cleanup()
processed = 0
for task in candidates:
if self._check_remote_status(task):
self._process(task)
processed += 1
if processed > 0:
logging.info(f"MP4転送完了数: {processed}")
finally:
if self.transport: self.transport.close()
def _connect_sftp(self):
c = self.cfg.nas_config
try:
t = paramiko.Transport((c['host'], c['port']))
if c['key_file']:
k = paramiko.RSAKey.from_private_key_file(c['key_file'])
t.connect(username=c['user'], pkey=k)
else:
t.connect(username=c['user'], password=c['password'])
return paramiko.SFTPClient.from_transport(t), t
except Exception: return None, None
def _scan_candidates(self) -> Generator[Dict, None, None]:
dest_root = self.cfg.nas_config['dest_dir']
threshold = datetime.now() - timedelta(seconds=self.cfg.scan_threshold_sec)
exclude_ptn = re.compile(r'-\d+$')
for p in self.cfg.converted_dir.rglob('*.mp4'):
if not p.is_file(): continue
if exclude_ptn.search(p.stem): continue
if datetime.fromtimestamp(p.stat().st_mtime) >= threshold: continue
size = p.stat().st_size
if self.state.is_recorded(StateManager.SECTION_UPLOAD_NAS, p.name, size): continue
rel = p.relative_to(self.cfg.converted_dir).as_posix()
dest = posixpath.join(dest_root, rel)
yield {'src': p, 'dest': dest, 'name': p.name, 'size': size, 'type': 'mp4'}
ass = p.with_suffix('.ass')
if ass.exists():
ass_dest = posixpath.join(dest_root, ass.relative_to(self.cfg.converted_dir).as_posix())
yield {'src': ass, 'dest': ass_dest, 'size': ass.stat().st_size, 'type': 'ass'}
def _check_remote_status(self, task) -> bool:
try:
r_size = self.sftp.stat(task['dest']).st_size
task['reason'] = "サイズ変更(NAS既存)"
return True
except FileNotFoundError:
task['reason'] = "新規"
return True
def _cleanup(self):
root = self.cfg.nas_config['dest_dir']
nas_excludes = self.cfg.nas_exclude_dirs.copy()
for pol in self.cfg.retention_policies_nas:
self.cleaner.apply_retention_policy(root, pol['dir'], pol['days'], pol['limit_gb'])
nas_excludes.append(f"/{pol['dir']}/")
self.cleaner.enforce_size_limit(root, self.cfg.max_size_nas_gb, nas_excludes)
self._cleanup_empty_dirs_remote(root)
def _cleanup_empty_dirs_remote(self, root: str):
dirs = []
try:
for path, _, is_dir in self.ops._sftp_walk(root):
if is_dir: dirs.append(path)
except Exception: return
dirs.sort(key=lambda s: len(s), reverse=True)
norm_root = root.rstrip('/')
count = 0
for d in dirs:
if posixpath.dirname(d.rstrip('/')) == norm_root: continue
if self.ops.remove_empty_dir(d):
logging.info(f" [NAS] 空ディレクトリ削除: {d}")
count += 1
if count > 0:
self.cfg.write_log = True
activate_realtime_log()
def _process(self, task):
dest_dir = posixpath.dirname(task['dest'])
logging.info(f"[NAS] Upload ({task['reason']}): {task['src'].name} -> {dest_dir}")
if self._upload(task['src'], task['dest']):
if task['type'] == 'mp4':
self.state.update_entry(StateManager.SECTION_UPLOAD_NAS, task['name'], task['size'], self.cfg.dry_run)
def _upload(self, src, dest):
if self.cfg.dry_run:
logging.info(" [DryRun] Uploaded.")
return True
try:
self._mkdir_p(posixpath.dirname(dest))
size_str = format_bytes(src.stat().st_size)
logging.info(f" -> Upload開始 ({size_str})")
start = time.time()
self.sftp.put(str(src), dest)
dur = time.time() - start
logging.info(f" -> Upload完了 (所要時間: {dur:.1f}秒)")
return True
except Exception as e:
logging.error(f" -> Upload失敗: {e}")
return False
def _mkdir_p(self, remote_dir):
if remote_dir in ['/', '.']: return
try: self.sftp.stat(remote_dir)
except FileNotFoundError:
self._mkdir_p(posixpath.dirname(remote_dir))
try: self.sftp.mkdir(remote_dir)
except OSError: pass
class M2tsBackupPipeline(BasePipeline):
"""Phase 3: M2TS -> HDD バックアップ"""
def run(self):
if not self.cfg.is_hdd_copy_time_window and not self.cfg.dry_run: return
tasks = list(self._scan())
self._cleanup()
if tasks:
self.cfg.write_log = True
activate_realtime_log()
dest_path = self.cfg.dest_dir_hdd
pre_size_str = self._get_dir_size_str(dest_path)
logging.info("=== Phase 3: M2TSバックアップ (Backup) ===")
logging.info(f"コピー対象数: {len(tasks)} (現在のディレクトリサイズ [{dest_path}]: {pre_size_str})")
for task in tasks:
self._process(task)
self.cleaner.delete_old_files_by_pattern(self.cfg.source_dir_m2ts, self.cfg.m2ts_delete_days, "*.m2ts")
post_size_str = self._get_dir_size_str(dest_path)
logging.info(f"バックアップ完了後のディレクトリサイズ [{dest_path}]: {post_size_str}")
def _get_dir_size_str(self, target_path: Path) -> str:
if not target_path.exists():
return format_bytes(0)
all_files = self.ops.list_files_recursive(target_path, exclude_dirs=[])
return format_bytes(sum(f.size for f in all_files))
def _scan(self):
if not self.cfg.source_dir_m2ts.exists(): return
threshold = datetime.now() - timedelta(seconds=self.cfg.scan_threshold_sec)
for p in self.cfg.source_dir_m2ts.rglob('*.m2ts'):
if not p.is_file(): continue
try:
rel = p.relative_to(self.cfg.source_dir_m2ts)
if len(rel.parts) > 1 and rel.parts[0] in self.cfg.skip_folders_m2ts: continue
except ValueError: pass
if datetime.fromtimestamp(p.stat().st_mtime) >= threshold: continue
size = p.stat().st_size
dest = self.cfg.dest_dir_hdd / rel
if self.state.is_recorded(StateManager.SECTION_COPY_HDD, p.name, size):
continue
if dest.exists() and dest.stat().st_size == size: continue
yield {'src': p, 'dest': dest, 'size': size, 'rel': rel}
def _cleanup(self):
for pol in self.cfg.retention_policies_hdd:
self.cleaner.apply_retention_policy(self.cfg.dest_dir_hdd, pol["dir"], pol["days"], pol["limit_gb"])
self.cleaner.enforce_size_limit(self.cfg.dest_dir_hdd, self.cfg.max_size_hdd_gb, self.cfg.hdd_exclude_dirs)
self._cleanup_empty_dirs_local(self.cfg.source_dir_m2ts)
self._cleanup_empty_dirs_local(self.cfg.dest_dir_hdd)
def _process(self, task):
src, dest = task['src'], task['dest']
logging.info(f"[Backup] コピー: {src.name} ({format_bytes(task['size'])})")
if self._copy(src, dest):
self.state.update_entry(StateManager.SECTION_COPY_HDD, src.name, task['size'], self.cfg.dry_run)
self._create_trim_avs(task)
def _create_trim_avs(self, task):
src_stem = task['src'].stem
log_file = self.cfg.converted_dir / task['rel'].parent / f"{src_stem}-enc.log"
if not log_file.exists(): return
avs_file = Path(f"{task['dest']}.trim.avs")
if self.cfg.dry_run:
logging.info(f" [DryRun] AVS作成: {avs_file.name}")
return
try:
trim_line = None
with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
for line in f:
if "Trim(" in line:
trim_line = line.strip()
break
if trim_line:
with open(avs_file, 'w', encoding='utf-8') as f:
f.write(trim_line)
logging.info(f" -> AVS作成完了: {avs_file.name}")
except Exception as e:
logging.warning(f" -> AVS作成失敗: {e}")
def _copy(self, src, dest):
if self.cfg.dry_run:
logging.info(f" [DryRun] Copy to: {dest}")
return True
try:
dest.parent.mkdir(parents=True, exist_ok=True)
if dest.exists(): dest.unlink()
file_size = src.stat().st_size
free_space = shutil.disk_usage(dest.parent).free
if free_space < file_size:
logging.error(f" -> HDD容量不足 (空き: {format_bytes(free_space)}, 必要: {format_bytes(file_size)})")
return False
logging.info(f" -> Copy開始 ({format_bytes(file_size)})")
start = time.time()
shutil.copy2(src, dest)
dur = time.time() - start
logging.info(f" -> Copy完了 (所要時間: {dur:.1f}秒)")
return True
except Exception as e:
logging.error(f" -> Copy失敗: {e}")
return False
# =========================================================
# メイン処理
# =========================================================
def main():
cfg = Config()
lock_fd = acquire_lock(cfg.lock_file)
if lock_fd is None: return
logger = setup_logger(cfg.log_file)
try:
state = StateManager(cfg.ini_file)
mode = "DryRun" if cfg.dry_run else "Production"
logging.info(f"=== TV Recorder Manager Start [{mode}] ===")
M2tsConverterPipeline(cfg, state).run()
Mp4UploadPipeline(cfg, state).run()
M2tsBackupPipeline(cfg, state).run()
logging.info("=== All Finished ===")
except Exception as e:
cfg.write_log = True
logging.exception(f"Unexpected Error: {e}")
finally:
logger.flush_all(cfg.write_log)
logger.close()
if lock_fd: lock_fd.close()
if __name__ == "__main__":
main()このスクリプトを起動すると下記タスクを順次実行して終了します。
- Amatsukazeへ変換タスクを追加
- mp4をNASへコピー
- m2tsをHDDへコピー・空ディレクトリの削除
処理対象のファイルがない場合は何もせずに終了します。常駐するタイプのスクリプトではないので、実行開始にはトリガーが必要になります。
実行開始のトリガーはcronを利用します。通常のテレビ番組は0,15,30,45分台に終了することが多いため、録画終了からできるだけ早く変換・コピー処理が始まるように、毎時1分を起点として3分おきに実行するようにします。下記はcrontabの設定例になります。
1-59/3 * * * * /usr/bin/python3 /home/tv-recorder/Scripts/TvRecorder.pyN100などスペックの低いCPUの場合は、録画中に変換処理が走っているとDropが発生することがあるため、変換実行時間を録画番組のない時間帯に限定してスクリプトを起動するように設定すればよいです。
このPythonスクリプトは以下についても考慮しています。
- 録画中のファイルは処理対象から除外する
- ニュースなど、長期に保持する必要がないファイルでHDD/NASの容量を圧迫しないように、最長ファイル保持期間をディレクトリごとに設定できるようにする
- HDDアクセスによるDropを避けるため、HDDへコピーする時間帯を制限する
- 変換・コピー処理のログをリアルタイムでログに保存し、コンソール上からもリアルタイムで動作状況を確認できるようにする
- ログの肥大化を避けるため、処理が行われない場合はログ出力しないようにする
- 実際のファイル操作無しで動作確認できるようにDryRunで動作確認できるようにする
- コピー処理実行時にはHDD/NASにコピー可能かどうかを確認し、コピーできない場合はHDD/NASが復帰したときに処理を再実行する
- 後でCMカット位置を変更できるようにm2tsファイルコピー時にカット位置を示す*.avsファイルも合わせて生成してm2tsと同じディレクトリにコピーする
- 変換・コピー後にファイル移動などで再処理が走らないように、処理を実行したファイルをiniファイルに記録し、次回以降の処理でiniファイルを参照してスキップする
TVサーバー稼働後の運用とメンテナンス
KonomiTVはディレクトリ階層構造でファイルをではないため、ディレクトリ構造で整理された1000単位の数の録画ファイルを選択するには不向きです。そこで直近1週間程度の録画番組はKonomiTVで再生し、録画後のCM抜きmp4ファイルの再生や古いファイルはNAS上で動作するJellyfinなどのメディアサーバーを使って視聴します。
録画・変換したファイルは自動で削除されるので短い期間でNASやHDDの容量が足りなくなることはないですが、保存しておく録画ファイルが増えるとHDD/NASを圧迫してシステムが機能しなくなるため、保存するファイルのトータルサイズとHDD/NASの空き容量は、適宜チェックしておく必要があります。
録画予約時に保存するかどうか未定の番組については、いったん削除予定で保存しておき、保存対象になった時点で録画予約の修正と録画変更済みファイルをkeepに移動すればいいです。コピー処理が走らないので、作業はすぐに完了します。
CMカットのやり直し
本スクリプトはCMカット位置を定義したavsファイル(*.m2ts.trim.avs)をm2tsファイル同じ場所に保存しています。Amatsukazeでm2tsを変換するとき、m2tsファイルとavsを同じディレクトリに置いておくことで、強制的にavsファイルの定義に従ってCMがカットされます。
avsファイルはテキストファイルでTrim(<トリミング開始フレーム>,<トリミング終了フレーム>)を「++」で連結することでCM位置がカットできます。以下は定義のサンプルです。
Trim(108,26601) ++ Trim(29299,44403) ++ Trim(48000,53124) ++ Trim(53575,54053)CM位置を変更する場合はこれらの数値を変更し、トリム場所を増やす・減らす場合はTrim定義を追加・削除して、再変換すればよいことになります。
このフォーマットの編集にはTrimViewerが便利です。ただし、m2tsファイルによっては読み込みができない問題があります。この場合はAvsPmodを使用してカットします。
AvsPmodでのTrim定義の生成手順については下記サイトが詳しいのでこちらを参考にされるといいと思います。
https://enctools.com/amatsukaze-manual-cm-cut
参考になれば幸いです。
