#!/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.")