gitlab-instance-0a899031_d2yexeapp/web2_camera.py

"""Physical camera capture config and FFmpeg device enumeration."""

from __future__ import annotations

import configparser
import re
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Any


def camera_ini_path(config_dir: Path) -> Path:
    return config_dir / "camera_live.ini"


def default_camera_config() -> dict[str, str]:
    return {
        "video_device": "",
        "audio_device": "",
        "width": "1280",
        "height": "720",
        "fps": "30",
        "bitrate": "2500",
        "use_gpu": "是",
        "youtube_watch_url": "",
    }


def load_camera_config(config_dir: Path) -> dict[str, str]:
    p = camera_ini_path(config_dir)
    out = default_camera_config()
    if not p.exists():
        return out
    cp = configparser.ConfigParser()
    try:
        cp.read(p, encoding="utf-8")
    except Exception:
        return out
    if not cp.has_section("camera"):
        return out
    for key in out:
        if cp.has_option("camera", key):
            out[key] = str(cp.get("camera", key) or "").strip()
    return out


def save_camera_config(config_dir: Path, patch: dict[str, Any]) -> tuple[bool, str]:
    config_dir = Path(config_dir)
    config_dir.mkdir(parents=True, exist_ok=True)
    cur = load_camera_config(config_dir)
    for k, v in patch.items():
        if k in cur:
            cur[k] = str(v or "").strip()
    cp = configparser.ConfigParser()
    cp["camera"] = cur
    p = camera_ini_path(config_dir)
    try:
        with p.open("w", encoding="utf-8") as f:
            cp.write(f)
        return True, "摄像头配置已保存"
    except Exception as e:
        return False, f"保存失败: {e}"


def camera_config_ready(config_dir: Path) -> str | None:
    cfg = load_camera_config(config_dir)
    if not str(cfg.get("video_device") or "").strip():
        return "请先选择摄像头设备"
    yp = Path(config_dir) / "youtube.ini"
    if not yp.exists():
        return "请先在 YouTube 配置中填写串流密钥"
    cp = configparser.ConfigParser()
    try:
        cp.read(yp, encoding="utf-8")
    except Exception:
        return "无法读取 youtube.ini"
    if not cp.has_section("youtube"):
        return "请先在 YouTube 配置中填写串流密钥"
    for name, val in cp.items("youtube"):
        v = (val or "").strip()
        if name in ("key", "stream_key") and v and not v.startswith("#"):
            return None
    return "请先在 YouTube 配置中填写串流密钥"


def _ffmpeg_exe() -> str | None:
    return shutil.which("ffmpeg")


def _run_ffmpeg_list(args: list[str], timeout: float = 12.0) -> str:
    exe = _ffmpeg_exe()
    if not exe:
        return ""
    try:
        proc = subprocess.run(
            [exe, *args],
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=timeout,
        )
        return (proc.stderr or "") + (proc.stdout or "")
    except Exception:
        return ""


def _parse_dshow_devices(stderr: str) -> list[dict[str, str]]:
    devices: list[dict[str, str]] = []
    in_video = False
    in_audio = False
    for line in stderr.splitlines():
        if re.search(r"DirectShow video devices", line, re.I):
            in_video, in_audio = True, False
            continue
        if re.search(r"DirectShow audio devices", line, re.I):
            in_audio, in_video = True, False
            continue
        if line.strip().startswith("[") and not re.search(r'\] "(.*)" \((video|audio)\)', line):
            in_video = in_audio = False
        m = re.search(r'\] "(.*)" \((video|audio)\)', line)
        if not m:
            continue
        name, kind = m.group(1), m.group(2)
        if kind == "video" and in_video:
            devices.append({"id": name, "name": name, "kind": "video"})
        elif kind == "audio" and in_audio:
            devices.append({"id": name, "name": name, "kind": "audio"})
    return devices


def _parse_v4l2_devices(stderr: str) -> list[dict[str, str]]:
    devices: list[dict[str, str]] = []
    for line in stderr.splitlines():
        m = re.search(r"\] (/dev/video\d+)", line)
        if m:
            dev = m.group(1)
            devices.append({"id": dev, "name": dev, "kind": "video"})
    return devices


def list_capture_devices() -> dict[str, Any]:
    exe = _ffmpeg_exe()
    if not exe:
        return {"ok": False, "error": "未找到 ffmpeg，请确认安装目录完整", "devices": []}
    if sys.platform == "win32":
        stderr = _run_ffmpeg_list(["-hide_banner", "-list_devices", "true", "-f", "dshow", "-i", "dummy"])
        devices = _parse_dshow_devices(stderr)
    elif sys.platform == "darwin":
        stderr = _run_ffmpeg_list(["-hide_banner", "-f", "avfoundation", "-list_devices", "true", "-i", ""])
        devices = []
        for line in stderr.splitlines():
            m = re.search(r"\[(\d+)\] (.+)", line)
            if m and re.search(r"AVFoundation|video devices", stderr, re.I):
                devices.append({"id": m.group(1), "name": m.group(2).strip(), "kind": "video"})
    else:
        stderr = _run_ffmpeg_list(["-hide_banner", "-list_devices", "true", "-f", "v4l2", "-i", "dummy"])
        devices = _parse_v4l2_devices(stderr)
    return {"ok": True, "devices": devices, "platform": sys.platform}