333 lines
12 KiB
Python
333 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
import datetime
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
import requests
|
|
from requests_toolbelt.multipart.encoder import MultipartEncoder, MultipartEncoderMonitor
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(message)s",
|
|
handlers=[logging.StreamHandler(sys.stdout)]
|
|
)
|
|
logger = logging.getLogger("peertube-uploader")
|
|
|
|
# ----------------- Configurations -----------------
|
|
WATCH_DIR = Path(os.getenv("WATCH_DIR", "/media"))
|
|
UPLOAD_HISTORY_FILE = WATCH_DIR / ".uploaded_history.txt"
|
|
|
|
PEERTUBE_URL = os.getenv("PEERTUBE_URL", "http://peertube:9000").rstrip("/")
|
|
PEERTUBE_USER = os.getenv("PEERTUBE_USER", "").strip()
|
|
PEERTUBE_PASSWORD = os.getenv("PEERTUBE_PASSWORD", "").strip()
|
|
PEERTUBE_CHANNEL = os.getenv("PEERTUBE_CHANNEL", "").strip()
|
|
PEERTUBE_HOST_HEADER = os.getenv("PEERTUBE_HOST_HEADER", "").strip()
|
|
|
|
# Video Metadata Template
|
|
TITLE_PREFIX = os.getenv("TITLE_PREFIX", "Practice")
|
|
VIDEO_CATEGORY = int(os.getenv("VIDEO_CATEGORY", "5")) # 5 = Sports
|
|
VIDEO_PRIVACY = int(os.getenv("VIDEO_PRIVACY", "4")) # 1=Public, 2=Unlisted, 3=Private, 4=Internal
|
|
VIDEO_LANGUAGE = os.getenv("VIDEO_LANGUAGE", "en")
|
|
VIDEO_TAGS = [t.strip() for t in os.getenv("VIDEO_TAGS", "volleyball,practice").split(",") if t.strip()]
|
|
VIDEO_DESCRIPTION_TEMPLATE = os.getenv(
|
|
"VIDEO_DESCRIPTION",
|
|
"Recorded on {date}.\nAutomated ingest and upload from GoPro HERO13."
|
|
)
|
|
|
|
POLL_INTERVAL = int(os.getenv("POLL_INTERVAL", "15"))
|
|
MOVE_TO_UPLOADED = os.getenv("MOVE_TO_UPLOADED", "false").lower() == "true"
|
|
DELETE_AFTER_UPLOAD = os.getenv("DELETE_AFTER_UPLOAD", "false").lower() == "true"
|
|
UPLOADED_DIR = WATCH_DIR / "uploaded"
|
|
|
|
# ----------------- Helpers -----------------
|
|
def format_bytes(size_bytes: int | float) -> str:
|
|
if size_bytes <= 0:
|
|
return "0 B"
|
|
units = ["B", "KB", "MB", "GB", "TB"]
|
|
unit_idx = 0
|
|
val = float(size_bytes)
|
|
while val >= 1024.0 and unit_idx < len(units) - 1:
|
|
val /= 1024.0
|
|
unit_idx += 1
|
|
return f"{val:.2f} {units[unit_idx]}"
|
|
|
|
def load_upload_history() -> set[str]:
|
|
if not UPLOAD_HISTORY_FILE.exists():
|
|
return set()
|
|
try:
|
|
with open(UPLOAD_HISTORY_FILE, "r") as f:
|
|
return {line.strip() for line in f if line.strip()}
|
|
except Exception as e:
|
|
logger.warning(f"Failed to load upload history: {e}")
|
|
return set()
|
|
|
|
def record_upload_history(filename: str):
|
|
try:
|
|
with open(UPLOAD_HISTORY_FILE, "a") as f:
|
|
f.write(f"{filename}\n")
|
|
except Exception as e:
|
|
logger.error(f"Failed to record upload history for {filename}: {e}")
|
|
|
|
def get_take_number(filepath: Path) -> int:
|
|
"""Extract numeric sequence ID (GX_0061 -> 61) for deterministic sorting."""
|
|
m = re.search(r'(\d{4})', filepath.stem)
|
|
if m:
|
|
return int(m.group(1))
|
|
return 0
|
|
|
|
def get_video_recording_datetime(filepath: Path) -> datetime.datetime:
|
|
"""Extract true recording datetime from video tags, stream tags, or timecode."""
|
|
cmd = [
|
|
"ffprobe", "-v", "quiet",
|
|
"-print_format", "json",
|
|
"-show_entries", "format_tags=creation_time:stream_tags=creation_time:stream_tags=timecode",
|
|
str(filepath)
|
|
]
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
data = json.loads(result.stdout)
|
|
|
|
# 1. Format creation_time
|
|
creation_str = data.get("format", {}).get("tags", {}).get("creation_time")
|
|
|
|
# 2. Stream creation_time
|
|
if not creation_str:
|
|
for stream in data.get("streams", []):
|
|
creation_str = stream.get("tags", {}).get("creation_time")
|
|
if creation_str:
|
|
break
|
|
|
|
if creation_str:
|
|
creation_str = creation_str.replace("Z", "+00:00")
|
|
dt = datetime.datetime.fromisoformat(creation_str)
|
|
return dt.astimezone()
|
|
|
|
except Exception as err:
|
|
logger.debug(f"ffprobe metadata extraction failed for {filepath.name}: {err}")
|
|
|
|
# Fallback to filesystem mtime (synced with original capture timestamp)
|
|
mtime = filepath.stat().st_mtime
|
|
return datetime.datetime.fromtimestamp(mtime).astimezone()
|
|
|
|
# ----------------- PeerTube API Client -----------------
|
|
class PeerTubeClient:
|
|
def __init__(self, base_url, username, password, host_header=""):
|
|
self.base_url = base_url
|
|
self.username = username
|
|
self.password = password
|
|
self.host_header = host_header
|
|
self.session = requests.Session()
|
|
self.channel_id = None
|
|
|
|
if self.host_header:
|
|
self.session.headers.update({
|
|
"Host": self.host_header,
|
|
"X-Forwarded-Proto": "https"
|
|
})
|
|
|
|
def authenticate(self):
|
|
logger.info(f"Authenticating with PeerTube at {self.base_url} as '{self.username}'...")
|
|
oauth_resp = self.session.get(f"{self.base_url}/api/v1/oauth-clients/local", timeout=10)
|
|
oauth_resp.raise_for_status()
|
|
client_data = oauth_resp.json()
|
|
|
|
token_payload = {
|
|
"client_id": client_data["client_id"],
|
|
"client_secret": client_data["client_secret"],
|
|
"grant_type": "password",
|
|
"response_type": "code",
|
|
"username": self.username,
|
|
"password": self.password
|
|
}
|
|
|
|
token_resp = self.session.post(
|
|
f"{self.base_url}/api/v1/users/token",
|
|
data=token_payload,
|
|
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
|
timeout=10
|
|
)
|
|
if token_resp.status_code != 200:
|
|
raise RuntimeError(f"Authentication failed ({token_resp.status_code}): {token_resp.text}")
|
|
|
|
access_token = token_resp.json()["access_token"]
|
|
self.session.headers.update({"Authorization": f"Bearer {access_token}"})
|
|
logger.info(f"Authenticated successfully as '{self.username}'")
|
|
|
|
def resolve_channel_id(self, channel_name: str) -> int:
|
|
if self.channel_id:
|
|
return self.channel_id
|
|
|
|
if not channel_name:
|
|
me_resp = self.session.get(f"{self.base_url}/api/v1/users/me", timeout=10)
|
|
me_resp.raise_for_status()
|
|
channels = me_resp.json().get("videoChannels", [])
|
|
if not channels:
|
|
raise RuntimeError("No channel found on this PeerTube account.")
|
|
self.channel_id = channels[0]["id"]
|
|
logger.info(f"Using default channel: {channels[0]['name']} (ID: {self.channel_id})")
|
|
return self.channel_id
|
|
|
|
ch_resp = self.session.get(f"{self.base_url}/api/v1/video-channels/{channel_name}", timeout=10)
|
|
if ch_resp.status_code != 200:
|
|
raise RuntimeError(f"Channel '{channel_name}' not found: {ch_resp.text}")
|
|
self.channel_id = ch_resp.json()["id"]
|
|
logger.info(f"Resolved channel '{channel_name}' -> ID {self.channel_id}")
|
|
return self.channel_id
|
|
|
|
def upload_video(self, file_path: Path, title: str, description: str, category: int, privacy: int, tags: list[str], language: str):
|
|
total_size = file_path.stat().st_size
|
|
logger.info(f"Uploading '{title}' ({format_bytes(total_size)})...")
|
|
|
|
fields = [
|
|
("channelId", str(self.channel_id)),
|
|
("name", title[:120]),
|
|
("category", str(category)),
|
|
("privacy", str(privacy)),
|
|
("language", language),
|
|
("description", description),
|
|
("videofile", (file_path.name, open(file_path, "rb"), "video/mp4"))
|
|
]
|
|
for tag in tags[:5]:
|
|
fields.append(("tags[]", tag))
|
|
|
|
encoder = MultipartEncoder(fields=fields)
|
|
|
|
start_time = time.time()
|
|
last_log_time = start_time
|
|
last_bytes = 0
|
|
|
|
def progress_callback(monitor):
|
|
nonlocal last_log_time, last_bytes
|
|
now = time.time()
|
|
if now - last_log_time >= 5:
|
|
delta_t = now - last_log_time
|
|
delta_b = monitor.bytes_read - last_bytes
|
|
speed = (delta_b / delta_t) if delta_t > 0 else 0
|
|
pct = (monitor.bytes_read / monitor.len) * 100
|
|
logger.info(f"Upload [{title}]: {pct:.1f}% ({format_bytes(monitor.bytes_read)} / {format_bytes(monitor.len)}) @ {format_bytes(speed)}/s")
|
|
last_log_time = now
|
|
last_bytes = monitor.bytes_read
|
|
|
|
monitor = MultipartEncoderMonitor(encoder, progress_callback)
|
|
headers = {"Content-Type": monitor.content_type}
|
|
if self.host_header:
|
|
headers["Host"] = self.host_header
|
|
headers["X-Forwarded-Proto"] = "https"
|
|
|
|
resp = self.session.post(
|
|
f"{self.base_url}/api/v1/videos/upload",
|
|
data=monitor,
|
|
headers=headers,
|
|
timeout=None
|
|
)
|
|
if resp.status_code not in (200, 201, 202):
|
|
raise RuntimeError(f"Upload failed ({resp.status_code}): {resp.text}")
|
|
|
|
elapsed = max(time.time() - start_time, 0.1)
|
|
video_uuid = resp.json().get("video", {}).get("uuid", "UNKNOWN")
|
|
logger.info(f"Published '{title}' (UUID: {video_uuid}) in {elapsed:.1f}s ({format_bytes(total_size / elapsed)}/s)")
|
|
return video_uuid
|
|
|
|
# ----------------- Processing Routine -----------------
|
|
def scan_and_upload():
|
|
history = load_upload_history()
|
|
|
|
pending_files = []
|
|
for f in WATCH_DIR.glob("*.mp4"):
|
|
if f.name.startswith(".") or f.is_dir():
|
|
continue
|
|
if f.name not in history:
|
|
size1 = f.stat().st_size
|
|
time.sleep(0.5)
|
|
size2 = f.stat().st_size
|
|
if size1 == size2 and size1 > 0:
|
|
pending_files.append(f)
|
|
|
|
if not pending_files:
|
|
return
|
|
|
|
# Group files by true recording date
|
|
date_groups = defaultdict(list)
|
|
for f in pending_files:
|
|
dt = get_video_recording_datetime(f)
|
|
take_num = get_take_number(f)
|
|
date_str = dt.strftime("%Y-%m-%d")
|
|
date_groups[date_str].append((take_num, dt, f))
|
|
|
|
client = PeerTubeClient(
|
|
base_url=PEERTUBE_URL,
|
|
username=PEERTUBE_USER,
|
|
password=PEERTUBE_PASSWORD,
|
|
host_header=PEERTUBE_HOST_HEADER
|
|
)
|
|
try:
|
|
client.authenticate()
|
|
client.resolve_channel_id(PEERTUBE_CHANNEL)
|
|
except Exception as e:
|
|
logger.error(f"PeerTube connection failed: {e}")
|
|
return
|
|
|
|
for date_str, items in date_groups.items():
|
|
# Sort strictly by GoPro take number (0061 before 0062)
|
|
items.sort(key=lambda x: (x[0], x[1]))
|
|
total_parts = len(items)
|
|
|
|
for idx, (take_num, dt, video_path) in enumerate(items, 1):
|
|
if total_parts > 1:
|
|
title = f"{TITLE_PREFIX} - {date_str} - Part {idx}"
|
|
else:
|
|
title = f"{TITLE_PREFIX} - {date_str}"
|
|
|
|
description = VIDEO_DESCRIPTION_TEMPLATE.format(
|
|
date=date_str,
|
|
time=dt.strftime("%H:%M:%S"),
|
|
filename=video_path.name,
|
|
part=idx,
|
|
total_parts=total_parts
|
|
)
|
|
|
|
try:
|
|
client.upload_video(
|
|
file_path=video_path,
|
|
title=title,
|
|
description=description,
|
|
category=VIDEO_CATEGORY,
|
|
privacy=VIDEO_PRIVACY,
|
|
tags=VIDEO_TAGS,
|
|
language=VIDEO_LANGUAGE
|
|
)
|
|
|
|
record_upload_history(video_path.name)
|
|
|
|
if DELETE_AFTER_UPLOAD:
|
|
logger.info(f"Removing local file: {video_path.name}")
|
|
video_path.unlink(missing_ok=True)
|
|
elif MOVE_TO_UPLOADED:
|
|
UPLOADED_DIR.mkdir(parents=True, exist_ok=True)
|
|
dest = UPLOADED_DIR / video_path.name
|
|
logger.info(f"Moving to archive: {dest}")
|
|
video_path.rename(dest)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to upload {video_path.name}: {e}")
|
|
|
|
def main():
|
|
logger.info(f"PeerTube Auto-Uploader started. Monitoring '{WATCH_DIR}' -> '{PEERTUBE_URL}'")
|
|
while True:
|
|
try:
|
|
scan_and_upload()
|
|
except Exception as err:
|
|
logger.error(f"Scanner error: {err}")
|
|
time.sleep(POLL_INTERVAL)
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except KeyboardInterrupt:
|
|
logger.info("Uploader service stopped.") |