Initial Commit

.gitea/workflows/docker-publish.yml

@@ -0,0 +1,44 @@
+name: Publish Docker image
+
+on:
+  release:
+    types: [published]
+  push:
+    tags:
+      - "v*"
+
+jobs:
+  push_to_registry:
+    name: Build and Push Docker image
+    runs-on: ubuntu-latest
+    permissions:
+      packages: write
+      contents: read
+
+    steps:
+      - name: Check out the repo
+        uses: actions/checkout@v4
+
+      - name: Log in to the Container registry
+        uses: docker/login-action@v3
+        with:
+          registry: git.stws.cc
+          username: ${{ github.actor }}
+          password: ${{ secrets.REGISTRY_TOKEN }}
+
+      - name: Extract metadata (tags, labels)
+        id: meta
+        uses: docker/metadata-action@v5
+        with:
+          images: git.stws.cc/${{ github.repository }}
+          tags: |
+            type=ref,event=tag
+            type=raw,value=latest
+
+      - name: Build and push Docker images
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}

Dockerfile

@@ -0,0 +1,28 @@
+FROM python:3.12-alpine
+
+ENV PYTHONUNBUFFERED=1
+
+# Install system dependencies
+RUN apk add --no-cache \
+    usbutils \
+    iproute2 \
+    dhcpcd \
+    curl \
+    ffmpeg \
+    gcc \
+    musl-dev \
+    linux-headers \
+    libffi-dev
+
+WORKDIR /app
+
+# Install Python dependencies and strip compiler tools
+RUN pip install --no-cache-dir \
+    "open-gopro>=0.22.0" \
+    aiohttp \
+    && apk del gcc musl-dev linux-headers libffi-dev
+
+COPY ingest.py /app/ingest.py
+RUN chmod +x /app/ingest.py
+
+CMD ["python3", "/app/ingest.py"]

ingest.py

@@ -0,0 +1,515 @@
+#!/usr/bin/env python3
+import asyncio
+import logging
+import os
+import re
+import subprocess
+import sys
+import tempfile
+import time
+import urllib.request
+import warnings
+import shutil
+from collections import defaultdict
+from pathlib import Path
+import aiohttp
+from open_gopro import WiredGoPro
+
+warnings.filterwarnings("ignore", category=UserWarning)
+warnings.filterwarnings("ignore", message=".*unregister_all_services.*")
+
+DEBUG_MODE = os.getenv("DEBUG", "false").lower() == "true"
+LOG_LEVEL = logging.DEBUG if DEBUG_MODE else logging.INFO
+
+logging.basicConfig(
+    level=LOG_LEVEL,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    handlers=[logging.StreamHandler(sys.stdout)]
+)
+logger = logging.getLogger("gopro-ingest")
+
+if not DEBUG_MODE:
+    for lib in ["open_gopro", "zeroconf", "asyncio", "aiohttp", "urllib3"]:
+        logging.getLogger(lib).setLevel(logging.ERROR)
+
+DEST_DIR = Path(os.getenv("INGEST_DIR", "/media"))
+TEMP_DIR = DEST_DIR / "temp"
+HISTORY_FILE = DEST_DIR / ".ingested_history.txt"
+
+POLL_INTERVAL = int(os.getenv("POLL_INTERVAL", "5"))
+LOG_PROGRESS_INTERVAL = int(os.getenv("LOG_PROGRESS_INTERVAL", "5"))
+DELETE_AFTER_INGEST = os.getenv("DELETE_AFTER_INGEST", "false").lower() == "true"
+KEEP_RAW_CHAPTERS = os.getenv("KEEP_RAW_CHAPTERS", "false").lower() == "true"
+
+GOPRO_USB_VENDOR_ID = "2672"
+last_synced_fingerprint = ""
+
+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_history() -> set[str]:
+    if not HISTORY_FILE.exists():
+        return set()
+    try:
+        with open(HISTORY_FILE, "r") as f:
+            return {line.strip() for line in f if line.strip()}
+    except Exception as e:
+        logger.warning(f"Could not read history file: {e}")
+        return set()
+
+def record_history(filenames: list[str] | set[str]):
+    try:
+        with open(HISTORY_FILE, "a") as f:
+            for fn in filenames:
+                f.write(f"{fn}\n")
+    except Exception as e:
+        logger.error(f"Failed to update history file: {e}")
+
+def get_source_creation_time(filepath: Path) -> str | None:
+    """Extract creation_time string from the original raw GoPro file."""
+    cmd = [
+        "ffprobe", "-v", "quiet",
+        "-print_format", "json",
+        "-show_entries", "format_tags=creation_time:stream_tags=creation_time",
+        str(filepath)
+    ]
+    try:
+        res = subprocess.run(cmd, capture_output=True, text=True, check=True)
+        import json
+        data = json.loads(res.stdout)
+        c_time = data.get("format", {}).get("tags", {}).get("creation_time")
+        if not c_time:
+            for s in data.get("streams", []):
+                c_time = s.get("tags", {}).get("creation_time")
+                if c_time:
+                    break
+        return c_time
+    except Exception:
+        return None
+
+def is_carrier_ready(iface: str) -> bool:
+    carrier_file = Path(f"/sys/class/net/{iface}/carrier")
+    try:
+        return carrier_file.read_text().strip() == "1"
+    except (FileNotFoundError, OSError):
+        return False
+
+def auto_configure_gopro_interfaces():
+    net_path = Path("/sys/class/net")
+    if not net_path.exists():
+        return
+
+    for iface_dir in net_path.iterdir():
+        iface = iface_dir.name
+        if iface.startswith(("lo", "docker", "br", "veth", "tun", "wg")):
+            continue
+
+        dev_path = iface_dir / "device"
+        if not dev_path.exists():
+            continue
+
+        curr = dev_path.resolve()
+        is_gopro = False
+        for _ in range(4):
+            vendor_file = curr / "idVendor"
+            if vendor_file.exists():
+                try:
+                    if vendor_file.read_text().strip().lower() == GOPRO_USB_VENDOR_ID:
+                        is_gopro = True
+                        break
+                except Exception:
+                    pass
+            curr = curr.parent
+
+        if is_gopro:
+            try:
+                addr_output = subprocess.check_output(["ip", "-4", "addr", "show", "dev", iface], text=True)
+                if "inet " in addr_output:
+                    continue  # Already has an IP address
+
+                # Bring link up administratively
+                subprocess.run(["ip", "link", "set", iface, "up"], check=False)
+
+                # Wait up to 5 seconds for the hardware carrier to settle
+                carrier_detected = False
+                for _ in range(10):
+                    time.sleep(0.5)
+                    if is_carrier_ready(iface):
+                        carrier_detected = True
+                        break
+
+                if not carrier_detected:
+                    logger.debug(f"GoPro interface '{iface}' present, but has NO-CARRIER. Is camera turned on and set to GoPro Connect?")
+                    continue
+
+                logger.info(f"Carrier active on '{iface}'. Requesting DHCP lease...")
+                # udhcpc in Alpine exits cleanly and does not lock sockets
+                if shutil.which("udhcpc"):
+                    subprocess.run(["udhcpc", "-i", iface, "-n", "-q", "-t", "4"], check=False)
+                elif shutil.which("dhcpcd"):
+                    subprocess.run(["dhcpcd", "-1", "-4", "-q", "-t", "10", iface], check=False)
+
+            except Exception as e:
+                logger.debug(f"Interface configuration error on {iface}: {e}")
+
+def detect_gopro_ip() -> str | None:
+    candidates = set()
+    try:
+        out = subprocess.check_output(["ip", "-4", "addr", "show"], text=True)
+        subnets = re.findall(r'inet (172\.(?:2[0-9]|3[0-1])\.\d+)\.\d+/\d+', out)
+        for subnet in subnets:
+            candidates.add(f"{subnet}.51")
+    except Exception:
+        pass
+
+    try:
+        out = subprocess.check_output(["ip", "-4", "route"], text=True)
+        subnets = re.findall(r'(172\.(?:2[0-9]|3[0-1])\.\d+)\.0/\d+', out)
+        for subnet in subnets:
+            candidates.add(f"{subnet}.51")
+    except Exception:
+        pass
+
+    for ip in candidates:
+        try:
+            req = urllib.request.Request(f"http://{ip}:8080/gopro/camera/state", method="GET")
+            with urllib.request.urlopen(req, timeout=1.0) as resp:
+                if resp.status == 200:
+                    return ip
+        except Exception:
+            continue
+    return None
+
+def extract_media_files(media_data) -> list[str]:
+    files = []
+    if isinstance(media_data, dict):
+        for group in media_data.get("media", []):
+            dir_name = group.get("directory", group.get("d", ""))
+            entries = group.get("file_system") or group.get("fs") or group.get("files") or []
+            for entry in entries:
+                if isinstance(entry, dict):
+                    fn = entry.get("filename") or entry.get("n")
+                    if fn:
+                        files.append(f"{dir_name}/{fn}" if (dir_name and "/" not in fn) else fn)
+                elif isinstance(entry, str):
+                    files.append(f"{dir_name}/{entry}" if dir_name else entry)
+    elif hasattr(media_data, "files"):
+        for f in media_data.files:
+            files.append(f.filename if hasattr(f, "filename") else str(f))
+    return files
+
+def parse_gopro_filename(filename: str) -> tuple[str, int] | None:
+    m = re.match(r'^(GX|GH)(\d{2})(\d{4})\.MP4$', filename, re.IGNORECASE)
+    if m:
+        prefix, chapter, take_num = m.group(1).upper(), int(m.group(2)), m.group(3)
+        return (f"{prefix}_{take_num}", chapter)
+
+    m = re.match(r'^GOPR(\d{4})\.MP4$', filename, re.IGNORECASE)
+    if m:
+        return (f"GOPR_{m.group(1)}", 1)
+
+    m = re.match(r'^GP(\d{2})(\d{4})\.MP4$', filename, re.IGNORECASE)
+    if m:
+        return (f"GOPR_{m.group(2)}", int(m.group(1)) + 1)
+
+    return None
+
+def is_file_already_processed(cam_path: str, history: set[str]) -> bool:
+    file_name = Path(cam_path).name
+    if file_name in history or cam_path in history:
+        return True
+
+    parsed = parse_gopro_filename(file_name)
+    if parsed:
+        take_id, _ = parsed
+        if (DEST_DIR / f"{take_id}.mp4").exists():
+            return True
+
+    if (TEMP_DIR / file_name).exists() or (DEST_DIR / file_name).exists():
+        return True
+
+    return False
+
+async def download_file_async(gopro_ip: str, cam_path: str, temp_file: Path, max_retries: int = 3) -> bool:
+    file_name = Path(cam_path).name
+    url = f"http://{gopro_ip}:8080/videos/DCIM/{cam_path}"
+    # Increase socket read timeout to prevent disconnects during SD card lag
+    timeout = aiohttp.ClientTimeout(total=None, sock_connect=15, sock_read=120)
+
+    for attempt in range(1, max_retries + 1):
+        try:
+            async with aiohttp.ClientSession(timeout=timeout) as session:
+                async with session.get(url) as resp:
+                    if resp.status != 200:
+                        logger.error(f"Failed to fetch {cam_path}: HTTP {resp.status} (attempt {attempt}/{max_retries})")
+                        if attempt < max_retries:
+                            await asyncio.sleep(2 * attempt)
+                            continue
+                        return False
+
+                    total_bytes = int(resp.headers.get("Content-Length", 0))
+                    size_label = f" ({format_bytes(total_bytes)})" if total_bytes > 0 else ""
+                    logger.info(f"Downloading {file_name}{size_label} [Attempt {attempt}/{max_retries}]...")
+
+                    start_time = time.time()
+                    last_log_time = start_time
+                    last_log_bytes = 0
+                    downloaded = 0
+
+                    with open(temp_file, "wb") as f:
+                        async for chunk in resp.content.iter_chunked(1024 * 1024):
+                            f.write(chunk)
+                            downloaded += len(chunk)
+
+                            now = time.time()
+                            if now - last_log_time >= LOG_PROGRESS_INTERVAL:
+                                delta_t = now - last_log_time
+                                delta_b = downloaded - last_log_bytes
+                                speed = (delta_b / delta_t) if delta_t > 0 else 0
+                                speed_str = f"{format_bytes(speed)}/s"
+
+                                if total_bytes > 0:
+                                    pct = (downloaded / total_bytes) * 100
+                                    logger.info(
+                                        f"Progress [{file_name}]: {pct:.1f}% "
+                                        f"({format_bytes(downloaded)} / {format_bytes(total_bytes)}) @ {speed_str}"
+                                    )
+                                else:
+                                    logger.info(f"Progress [{file_name}]: {format_bytes(downloaded)} @ {speed_str}")
+
+                                last_log_time = now
+                                last_log_bytes = downloaded
+
+                    # Verify received bytes match Content-Length header if provided
+                    if total_bytes > 0 and downloaded < total_bytes:
+                        raise aiohttp.ClientPayloadError(f"Incomplete download: got {downloaded}/{total_bytes} bytes")
+
+                    elapsed = max(time.time() - start_time, 0.1)
+                    avg_speed = downloaded / elapsed
+                    logger.info(f"Completed {file_name}: {format_bytes(downloaded)} in {elapsed:.1f}s ({format_bytes(avg_speed)}/s)")
+                    return True
+
+        except Exception as err:
+            logger.warning(f"Error transferring {cam_path} (attempt {attempt}/{max_retries}): {err}")
+            if temp_file.exists():
+                temp_file.unlink(missing_ok=True)
+            if attempt < max_retries:
+                await asyncio.sleep(3 * attempt)
+            else:
+                logger.error(f"Exhausted all retries for {cam_path}")
+                return False
+
+    return False
+
+async def process_and_finalize_videos(all_camera_files: list[str]):
+    TEMP_DIR.mkdir(parents=True, exist_ok=True)
+    
+    # 1. Map what chapters exist in the camera inventory
+    camera_takes = defaultdict(set)
+    for p in all_camera_files:
+        fn = Path(p).name
+        parsed = parse_gopro_filename(fn)
+        if parsed:
+            take_id, chapter = parsed
+            camera_takes[take_id].add(chapter)
+
+    # 2. Map what staged chapters we have on disk in TEMP_DIR
+    staged_takes = defaultdict(dict)
+    for p in TEMP_DIR.glob("*.[mM][pP]4"):
+        parsed = parse_gopro_filename(p.name)
+        if parsed:
+            take_id, chapter = parsed
+            staged_takes[take_id][chapter] = p
+
+    for take_id, chapters in staged_takes.items():
+        expected_chapters = camera_takes.get(take_id, set())
+        staged_chapters = set(chapters.keys())
+
+        # If any chapter for this take is still missing from TEMP_DIR, wait for future run
+        if expected_chapters and not expected_chapters.issubset(staged_chapters):
+            missing = sorted(list(expected_chapters - staged_chapters))
+            logger.info(f"Skipping finalize for '{take_id}': waiting for missing chapter(s) {missing}")
+            continue
+
+        sorted_chapters = [chapters[k] for k in sorted(chapters.keys())]
+        output_file = DEST_DIR / f"{take_id}.mp4"
+        temp_output_file = DEST_DIR / f".{take_id}.tmp.mp4"
+
+        if output_file.exists():
+            continue
+
+        is_hevc = take_id.startswith("GX")
+        tag_args = ["-tag:v", "hvc1"] if is_hevc else []
+        primary_source_file = sorted_chapters[0]
+
+        creation_time = get_source_creation_time(primary_source_file)
+        meta_args = ["-metadata", f"creation_time={creation_time}"] if creation_time else []
+
+        if len(sorted_chapters) == 1:
+            logger.info(f"Finalizing single-take '{take_id}' -> {output_file.name}...")
+            cmd = [
+                "ffmpeg", "-y",
+                "-i", str(primary_source_file),
+                "-map", "0:v:0",
+                "-map", "0:a?",
+                *meta_args,
+                "-c", "copy",
+                "-sn", "-dn",
+                *tag_args,
+                "-movflags", "+faststart",
+                str(temp_output_file)
+            ]
+        else:
+            logger.info(f"Stitching {len(sorted_chapters)} chapters for '{take_id}' -> {output_file.name} (Timestamp: {creation_time})...")
+            with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as concat_list:
+                for ch in sorted_chapters:
+                    concat_list.write(f"file '{ch.resolve()}'\n")
+                concat_path = concat_list.name
+
+            cmd = [
+                "ffmpeg", "-y",
+                "-f", "concat",
+                "-safe", "0",
+                "-i", concat_path,
+                "-map", "0:v:0",
+                "-map", "0:a?",
+                *meta_args,
+                "-c", "copy",
+                "-sn", "-dn",
+                *tag_args,
+                "-movflags", "+faststart",
+                str(temp_output_file)
+            ]
+
+        try:
+            proc = await asyncio.create_subprocess_exec(
+                *cmd,
+                stdout=asyncio.subprocess.DEVNULL,
+                stderr=asyncio.subprocess.PIPE
+            )
+            _, stderr = await proc.communicate()
+
+            if proc.returncode == 0 and temp_output_file.exists():
+                temp_output_file.rename(output_file)
+                src_stat = primary_source_file.stat()
+                os.utime(output_file, (src_stat.st_atime, src_stat.st_mtime))
+
+                logger.info(f"Ready for upload: {output_file.name} ({format_bytes(output_file.stat().st_size)})")
+
+                raw_names = [ch.name for ch in sorted_chapters]
+                record_history(raw_names)
+
+                if not KEEP_RAW_CHAPTERS:
+                    logger.info(f"Cleaning raw staging files for '{take_id}' from temp...")
+                    for ch in sorted_chapters:
+                        ch.unlink(missing_ok=True)
+            else:
+                logger.error(f"FFmpeg processing failed for {take_id}: {stderr.decode()}")
+                if temp_output_file.exists():
+                    temp_output_file.unlink(missing_ok=True)
+        finally:
+            if 'concat_path' in locals() and Path(concat_path).exists():
+                Path(concat_path).unlink(missing_ok=True)
+
+async def sync_gopro(gopro_ip: str):
+    global last_synced_fingerprint
+    DEST_DIR.mkdir(parents=True, exist_ok=True)
+    TEMP_DIR.mkdir(parents=True, exist_ok=True)
+
+    try:
+        async with WiredGoPro(ip=gopro_ip) as gopro:
+            resp = await gopro.http_command.get_media_list()
+            if not resp.ok:
+                return
+
+            all_files = extract_media_files(resp.data)
+            if not all_files:
+                return
+
+            history = load_history()
+            pending_files = [f for f in all_files if not is_file_already_processed(f, history)]
+            current_fingerprint = f"{gopro_ip}:{len(all_files)}:{len(pending_files)}"
+
+            if not pending_files:
+                if last_synced_fingerprint != current_fingerprint:
+                    logger.info(f"GoPro connected ({len(all_files)} files found). All media up to date. Idle...")
+                    last_synced_fingerprint = current_fingerprint
+                return
+
+            logger.info(f"GoPro connected: {len(pending_files)} new file(s) to ingest (Total on SD: {len(all_files)}).")
+
+            ingested_files_this_run = []
+            for cam_path in pending_files:
+                file_name = Path(cam_path).name
+                
+                if file_name.upper().endswith(".MP4"):
+                    target_file = TEMP_DIR / file_name
+                    temp_file = TEMP_DIR / f".{file_name}.tmp"
+                else:
+                    target_file = DEST_DIR / file_name
+                    temp_file = DEST_DIR / f".{file_name}.tmp"
+
+                try:
+                    success = await download_file_async(
+                        gopro_ip=gopro_ip,
+                        cam_path=cam_path,
+                        temp_file=temp_file
+                    )
+
+                    if success and temp_file.exists():
+                        temp_file.rename(target_file)
+                        logger.info(f"Ingested to staging: {file_name}")
+                        ingested_files_this_run.append(file_name)
+
+                        if not file_name.upper().endswith(".MP4"):
+                            record_history([file_name])
+
+                        if DELETE_AFTER_INGEST:
+                            logger.info(f"Deleting {cam_path} from camera SD card...")
+                            await gopro.http_command.delete_file(path=cam_path)
+                    else:
+                        logger.error(f"Failed to ingest: {cam_path}")
+                        if temp_file.exists():
+                            temp_file.unlink()
+
+                except Exception as err:
+                    logger.error(f"Error transferring {cam_path}: {err}")
+                    if temp_file.exists():
+                        temp_file.unlink()
+
+            logger.info("Checking staged media for completed takes...")
+            await process_and_finalize_videos(all_files)
+
+            logger.info("All media finalized in root directory. Camera can be safely unplugged.")
+            last_synced_fingerprint = f"{gopro_ip}:{len(all_files)}:0"
+
+    except (asyncio.TimeoutError, ConnectionRefusedError, OSError):
+        last_synced_fingerprint = ""
+    except Exception as e:
+        logger.debug(f"Sync event: {e}")
+
+async def main():
+    global last_synced_fingerprint
+    logger.info(f"GoPro Ingest Daemon active. Target folder: '{DEST_DIR}'")
+    while True:
+        auto_configure_gopro_interfaces()
+        gopro_ip = detect_gopro_ip()
+        if gopro_ip:
+            await sync_gopro(gopro_ip)
+        else:
+            last_synced_fingerprint = ""
+        await asyncio.sleep(POLL_INTERVAL)
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        logger.info("Ingest daemon stopped.")