diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..304eb3d --- /dev/null +++ b/.env.example @@ -0,0 +1,16 @@ +# MiAir Docker 部署配置 +# 复制此文件为 .env 并填入实际值 + +# 小米账号 (手机号或邮箱) +MI_USER= + +# 小米密码 +MI_PASS= + +# 设备 DID (可选，留空则自动搜索) +# 多个设备用逗号分隔 +MI_DID= + +# 局域网 IP (通常自动检测，如有需要可手动指定) +# 例如：192.168.1.1 +# HOST_IP=192.168.1.1 diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml new file mode 100644 index 0000000..007a602 --- /dev/null +++ b/.github/workflows/build.yml @@ -0,0 +1,65 @@ +name: Build and Push MiAir Docker Image + +on: + push: + branches: + - docker + - main + paths-ignore: + - '**.md' + - 'docs/**' + - '.gitignore' + - '.env.example' + - 'LICENSE' + - '*.png' + - '*.sh' + workflow_dispatch: + +env: + REGISTRY: ghcr.io + IMAGE_NAME: ${{ github.repository_owner }}/miair + +jobs: + build: + runs-on: ubuntu-latest + permissions: + contents: read + packages: write + + steps: + - uses: actions/checkout@v4 + + - name: Set up QEMU (multi-arch support) + uses: docker/setup-qemu-action@v3 + + - uses: docker/setup-buildx-action@v3 + + - uses: docker/login-action@v3 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - uses: docker/metadata-action@v5 + id: meta + with: + images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} + tags: | + type=ref,event=branch + type=sha,prefix=,format=short + type=raw,value=latest + type=semver,pattern={{version}},enable=${{ startsWith(github.ref, 'refs/tags/') }} + + - uses: docker/build-push-action@v6 + with: + context: . + file: ./Dockerfile + platforms: linux/amd64,linux/arm64 + push: true + tags: ${{ steps.meta.outputs.tags }} + labels: ${{ steps.meta.outputs.labels }} + cache-from: type=gha + cache-to: type=gha,mode=max + build-args: | + BUILD_DATE=${{ github.event.head_commit.timestamp }} + VCS_REF=${{ github.sha }} diff --git a/.gitignore b/.gitignore index 3f80168..71908f8 100644 --- a/.gitignore +++ b/.gitignore @@ -1,33 +1,6 @@ -# Python -__pycache__/ -*.py[cod] -*$py.class -*.egg-info/ -dist/ -build/ -*.egg - -# IDE -.vscode/ -.idea/ -*.swp -*.swo - -# 配置与敏感信息 -conf/config.json -conf/.mi.token -conf/*.log -conf/*.log.* - -# 运行时产物 -*.log -*.pcm -nul - -# 系统文件 -.DS_Store -Thumbs.dbf - -*.exe - - \ No newline at end of file +.env + +# Python byte-compiled / optimized / DLL files +__pycache__/ +*.py[codz] +*$py.class \ No newline at end of file diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..8193f4d --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,30 @@ +# Repository Instructions + +## Project Shape +- Python package `miair` plus top-level launcher `miair.py`; installed CLI entrypoint is `miair = miair.cli:main`. +- `miair.app.MiAir` wires the app: Web UI/API starts first, then DLNA and AirPlay start only when Xiaomi account/cookie and `mi_did` are configured. +- Major protocol code lives under `miair/dlna/` and `miair/airplay/`; Web API routes and settings masking live in `miair/web/api.py`. + +## Commands +- Runtime requires Python `>=3.10` per `pyproject.toml`; README says Python 3.12+ for Windows usage. +- Install editable package/deps with `python3 -m pip install -e .` before importing modules or running tests; `miair.py` auto-installs runtime deps only when launched directly. +- Start locally with `python3 miair.py` or `python3 miair.py --conf-path conf --web-port 8300 --dlna-port 8200`. +- Run the existing test script with `python3 tests/test_audio_seek.py`; this is not a pytest-configured repo and `pytest` is not declared as a dependency. +- Focused pytest-style execution may still work if pytest is installed: `python3 -m pytest tests/test_audio_seek.py -k detect_audio_format`. + +## Runtime And Config +- Config is loaded from `/config.json`; relative `--conf-path` is normalized to an absolute path in `Config.load`. +- Env fallback names are `MI_USER`, `MI_PASS`, and `MI_DID`; `MIAIR_HOSTNAME` explicitly overrides saved `hostname`, while `HOST_IP` in `.env.example` is only a commented hint and is not read by `Config`. +- Default ports are DLNA HTTP `8200` and Web UI/API `8300`; Docker exposes both and runs with host networking. +- Secrets/cookies must stay masked in API responses; preserve `_mask_cookie`, `_unmask_cookie`, and `_mask_devices` behavior when touching settings endpoints. + +## Docker And Deploy +- CI only builds/pushes Docker images on `main`, `docker`, or manual dispatch; Markdown, docs, shell scripts, `.env.example`, `.gitignore`, license, and PNG changes are ignored by the workflow. +- The checked-in `Dockerfile` copies `config-example.json` and `.env.example` into the image, seeds `/app/conf/config.json` and `/app/conf/.env` if absent, then runs `python miair.py --conf-path /app/conf`. +- `deploy.sh` is interactive and root-only; it rewrites `Dockerfile` via heredoc, builds `miair:latest`, mounts a user-selected config directory to `/app/conf`, and starts container `miair` with `--network=host` and `--restart unless-stopped`. +- `manage.sh update` downloads `main.tar.gz`, rebuilds the image, removes the existing container, then re-runs `deploy.sh`; avoid assuming it is a non-interactive update path. + +## Testing Quirks +- `tests/test_audio_seek.py` imports production modules and needs runtime dependencies such as `miservice-fork`, `aiohttp`, `zeroconf`, `pycryptodome`, and `av` installed. +- ffmpeg-dependent portions of the test script are best-effort: they skip or warn when ffmpeg is unavailable; pure-Python seek tests should still run once Python deps are installed. +- The current macOS system Python in this workspace is 3.9.6, below the project requirement, so use a Python 3.10+ interpreter for meaningful verification. diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..a7ca814 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,30 @@ +FROM python:3.12-slim + +LABEL maintainer="MiAir" +LABEL description="DLNA/AirPlay receiver for Xiaomi AI Speaker" + +RUN apt-get update && apt-get install -y --no-install-recommends \ + ffmpeg \ + libportaudio2 \ + dnsutils \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app + +COPY pyproject.toml . +RUN pip install --no-cache-dir . --root-user-action=ignore + +# 这里明确把示例配置文件也拷贝进镜像的备用区 +COPY config-example.json .env.example ./ +COPY miair.py ./ +COPY miair/ ./miair/ + +RUN mkdir -p /app/conf + +EXPOSE 8200 8300 + +# 【核心修改】智能启动命令： +# 1. 检查 /app/conf/config.json 是否存在，不存在就复制一份示例文件过去 +# 2. 检查 /app/conf/.env 是否存在，不存在也复制过去 +# 3. 最后使用 exec 移交进程控制权并启动真正的 Python 服务 +ENTRYPOINT ["/bin/sh", "-c", "if [ ! -f /app/conf/config.json ]; then cp /app/config-example.json /app/conf/config.json; fi && if [ ! -f /app/conf/.env ]; then cp /app/.env.example /app/conf/.env; fi && exec python miair.py --conf-path /app/conf"] diff --git a/README.md b/README.md index 249d114..9e6c2c5 100644 --- a/README.md +++ b/README.md @@ -5,22 +5,61 @@ **[XiaoMusic](https://github.com/hanxi/xiaomusic "XiaoMusic")**   **[AirPlay2 Receiver](https://github.com/openairplay/airplay2-receiver "AirPlay2 Receiver")**   **[MaCast](https://github.com/xfangfang/Macast "MaCast")** ## 快速开始 - +## Windows *确保设备已安装 Python 3.12+* +前往Release页面下载最新版本\ 进入项目目录，使用终端执行 ```python python miair.py ``` -程序将自动安装相关依赖库，请确保网络畅通 +程序将自动安装相关依赖库，请确保网络畅通\ +安装完成后访问 `http://主机IP:8300` 即可打开 Web 管理界面。 + +## Docker (Thanks @SyunSS) + +支持平台：Linux / OpenWrt / macOS + +### 使用脚本部署 +```bash +# 安装 Git +opkg update +opkg install git +opkg install git-http + +# 克隆项目 +rm -rf MiAir # 如果是更新，需要清理旧的部署目录 +git clone https://github.com/KiriChen-Wind/MiAir.git +cd MiAir + +# 赋予权限并运行安装脚本 +chmod +x deploy.sh manage.sh +./deploy.sh +``` + +安装完成后访问 `http://容器宿主机IP:8300` 即可打开 Web 管理界面。 +请确保容器网络为Host。\ +请确保 Docker 容器重启策略设定为 Unless Stopped。\ +部分情况下，修改配置后容器可能无法自动重启，请手动重启容器。 + + +### Docker 相关命令 +```bash +docker logs -f miair # 查看日志 +docker stop miair # 停止 +docker start miair # 启动 +docker restart miair # 重启 +``` +## 我们 +**[需要帮助&交流&测试版本发布](https://qun.qq.com/universal-share/share?ac=1&authKey=1zXhx2zxgw9GG2mkecypT9clD7q0B3W3l4K0D4fQirmpDWakz0Oy2BI3ocDrgzbh&busi_data=eyJncm91cENvZGUiOiI3NDEyNjcyOTgiLCJ0b2tlbiI6InYwbitXQTF5cE9MaUJCR0hMUk03OWV0WkFoMThxbjJRaWI4dHVlbUpGdW5OdEZBVEpXMXF0T1dQUnRmRXRzYVgiLCJ1aW4iOiIxODQxOTM4MDQwIn0%3D&data=_OrA-eASJMwYwx-Uj-BReC1Xh3zGAdkn8CQskbEsQ5S66bhqvvO6dJ-QrSlRl-Ks00l5XDw1FANE8Um0w5yB8Q&svctype=4&tempid=h5_group_info "需要帮助&交流&测试版本发布")** -### 后续 可能 添加的功能 +## 后续 可能 添加的功能 -- 支持 Docker 部署 +- ~~支持 Docker 部署~~ ✅ 已支持 - 支持 OpenWrt 部署 - ~~支持 MacOS 部署~~ - ...... -[![preview](https://raw.githubusercontent.com/KiriChen-Wind/MiAir/main/preview.png "preview")](https://raw.githubusercontent.com/KiriChen-Wind/MiAir/main/preview.png "preview") \ No newline at end of file +[![preview](https://raw.githubusercontent.com/KiriChen-Wind/MiAir/main/preview.png "preview")](https://raw.githubusercontent.com/KiriChen-Wind/MiAir/main/preview.png "preview") diff --git a/deploy.sh b/deploy.sh new file mode 100644 index 0000000..e86e974 --- /dev/null +++ b/deploy.sh @@ -0,0 +1,366 @@ +#!/bin/bash +# MiAir Docker 部署脚本 +# 支持 Linux / OpenWrt / iStoreOS / macOS 等平台 + +set -e + +echo "==========================================" +echo " MiAir Docker 部署脚本" +echo "==========================================" +echo "" + +# 颜色定义 +RED='\033[0;31m' +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +NC='\033[0m' # No Color + +# 配置变量 +APP_DIR="$(cd "$(dirname "$0")" && pwd)" +PARENT_DIR="$(dirname "$APP_DIR")" +CONTAINER_NAME="miair" +IMAGE_NAME="miair:latest" + +# 检测设备架构 +ARCH=$(uname -m) +echo -e "${GREEN}检测到设备架构: $ARCH${NC}" + +# 根据架构设置基础镜像 +case "$ARCH" in + arm*|aarch64) + BASE_IMAGE="python:3.12-slim" + ;; + x86_64|amd64) + BASE_IMAGE="python:3.12-slim" + ;; + *) + echo -e "${YELLOW}警告: 未知架构 $ARCH，使用默认基础镜像${NC}" + BASE_IMAGE="python:3.12-slim" + ;; +esac +echo -e "${GREEN}使用基础镜像: $BASE_IMAGE${NC}" + +# 询问用户配置目录位置 +echo -e "${YELLOW}配置文件目录设置${NC}" +echo "建议使用持久化存储路径，如: /root/miair_conf 或 /opt/miair_conf" +read -p "请输入配置目录路径（直接回车使用默认: /tmp/miair_conf）: " CONFIG_DIR +if [ -z "$CONFIG_DIR" ]; then + CONFIG_DIR="/tmp/miair_conf" +fi + +# 确保配置目录存在 +mkdir -p "$CONFIG_DIR" +echo -e "${GREEN}✓ 配置目录: $CONFIG_DIR${NC}" + +# 进入脚本所在目录 +cd "$APP_DIR" + +# 检测是否以 root 运行 +if [ "$EUID" -ne 0 ]; then + echo -e "${RED}错误: 请使用 root 权限运行此脚本${NC}" + echo "运行: sudo ./deploy.sh" + exit 1 +fi + +# ============================================ +# 步骤 1: 检查 Docker 是否安装 +# ============================================ +echo -e "${GREEN}[1/8] 检查 Docker 环境...${NC}" + +if command -v docker &> /dev/null; then + echo -e "${GREEN}✓ Docker 已安装: $(docker --version)${NC}" +else + echo -e "${YELLOW}Docker 未安装，正在安装...${NC}" + opkg update + opkg install dockerd docker-compose + /etc/init.d/dockerd start + /etc/init.d/dockerd enable + echo -e "${GREEN}✓ Docker 安装完成${NC}" +fi + +# 等待 Docker 服务就绪 +echo "等待 Docker 服务启动..." +for i in {1..30}; do + if docker info &> /dev/null; then + echo -e "${GREEN}✓ Docker 服务运行正常${NC}" + break + fi + if [ $i -eq 30 ]; then + echo -e "${RED}✗ Docker 服务启动超时${NC}" + exit 1 + fi + sleep 1 +done + +# ============================================ +# 步骤 2: 获取宿主机 IP +# ============================================ +echo -e "${GREEN}[2/8] 获取局域网 IP...${NC}" + +# 获取 LAN 口 IP 的函数：优先枚举局域网私有地址，默认出网地址只作为最后兜底 +get_lan_ip() { + local ip + + # 1. 优先从非虚拟接口中选择私有 LAN 地址，避免多网卡时误选 WAN/default route + ip=$(ip -o -4 addr show scope global 2>/dev/null | awk ' + $2 !~ /^(lo|docker[0-9]*|tailscale[0-9]*|veth.*|br-[a-f0-9]+|wg[0-9]*|tun[0-9]*|tap[0-9]*|ppp[0-9]*)$/ { + split($4, a, "/"); candidate=a[1] + if (candidate ~ /^192\.168\./) { best=candidate; found=1; exit } + if (candidate ~ /^10\./ && best == "") best=candidate + if (candidate ~ /^172\.(1[6-9]|2[0-9]|3[0-1])\./ && best == "") best=candidate + if (fallback == "") fallback=candidate + } + END { if (best != "") print best; else if (fallback != "") print fallback } + ') + [ -n "$ip" ] && echo "$ip" && return + + # 2. 如果 ip addr 不可用，再尝试 hostname -I 中的私有地址 + if command -v hostname &> /dev/null; then + ip=$(hostname -I 2>/dev/null | awk '{ + for (i=1; i<=NF; i++) { + if ($i ~ /^192\.168\./) { print $i; exit } + if ($i ~ /^10\./ && best == "") best=$i + if ($i ~ /^172\.(1[6-9]|2[0-9]|3[0-1])\./ && best == "") best=$i + } + if (best != "") print best + }') + [ -n "$ip" ] && echo "$ip" && return + fi + + # 3. 最后才使用默认出网 IP，它可能是 WAN 地址，后续仍会要求用户确认 + ip=$(ip -4 route get 8.8.8.8 2>/dev/null | awk '/src/ {for (i=1; i<=NF; i++) if ($i == "src") {print $(i+1); exit}}') + [ -n "$ip" ] && echo "$ip" && return + + # 兜底 + echo "127.0.0.1" +} + +HOST_IP=$(get_lan_ip) + +if [ -z "$HOST_IP" ] || [ "$HOST_IP" = "127.0.0.1" ]; then + echo -e "${YELLOW}警告: 无法自动获取到有效的局域网 IP。${NC}" + read -p "请输入宿主机的局域网 IP 地址: " HOST_IP + while [ -z "$HOST_IP" ] || [ "$HOST_IP" = "127.0.0.1" ]; do + echo -e "${RED}IP 地址无效，请重新输入${NC}" + read -p "请输入宿主机的局域网 IP 地址: " HOST_IP + done +else + echo -e "${GREEN}获取到的局域网 IP 地址: ${YELLOW}$HOST_IP${NC}" + echo "这个局域网 IP 地址是正确的吗？" + read -p "如果此局域网 IP 地址正确，请按下回车以确认使用。否则请输入正确的局域网 IP 地址并按下回车： " USER_IP + if [ -n "$USER_IP" ]; then + HOST_IP="$USER_IP" + fi +fi +echo -e "${GREEN}✓ 最终确认宿主机 IP: $HOST_IP${NC}" + +# ============================================ +# 步骤 3: 检查并下载 MiAir 代码 +# ============================================ +echo -e "${GREEN}[3/8] 检查 MiAir 代码...${NC}" + +# 检查必要文件是否存在 +if [ -f "miair.py" ] && [ -f "pyproject.toml" ] && [ -d "miair" ]; then + echo -e "${GREEN}✓ MiAir 代码已存在${NC}" +else + echo -e "${YELLOW}未找到 MiAir 代码，开始从 GitHub 下载...${NC}" + + # 下载代码 + echo "下载 MiAir 源代码..." + if command -v wget &> /dev/null; then + wget -O miair.tar.gz https://github.com/KiriChen-Wind/MiAir/archive/refs/heads/main.tar.gz + elif command -v curl &> /dev/null; then + curl -L https://github.com/KiriChen-Wind/MiAir/archive/refs/heads/main.tar.gz -o miair.tar.gz + else + opkg install wget + wget -O miair.tar.gz https://github.com/KiriChen-Wind/MiAir/archive/refs/heads/main.tar.gz + fi + + # 解压并整理 + echo "解压文件..." + tar -xzf miair.tar.gz + cp -r MiAir-main/* . 2>/dev/null || true + cp MiAir-main/.* . 2>/dev/null || true + rm -rf MiAir-main miair.tar.gz + + # 验证下载 + if [ -f "miair.py" ] && [ -f "pyproject.toml" ] && [ -d "miair" ]; then + echo -e "${GREEN}✓ MiAir 代码下载并准备完成${NC}" + else + echo -e "${RED}✗ MiAir 代码下载失败${NC}" + echo "请手动下载: https://github.com/KiriChen-Wind/MiAir" + exit 1 + fi +fi + +# ============================================ +# 步骤 4: 配置参数 +# ============================================ + +# 加载 .env 文件（如果存在） +if [ -f ".env" ]; then + echo "加载 .env 配置文件..." + set -a + source .env + set +a +fi + +# ============================================ +# 步骤 5: 创建 Dockerfile +# ============================================ +echo -e "${GREEN}[5/8] 创建 Dockerfile...${NC}" + +# 确保在正确目录 +cd "$APP_DIR" + +# 检查 pyproject.toml 是否存在 +if [ ! -f "pyproject.toml" ]; then + echo -e "${RED}错误: pyproject.toml 不存在${NC}" + echo "请确保 MiAir 代码已正确放置" + exit 1 +fi + +# 创建 Dockerfile +cat > Dockerfile << DOCKERFILE_EOF +FROM $BASE_IMAGE + +LABEL maintainer="MiAir" +LABEL description="DLNA/AirPlay receiver for Xiaomi AI Speaker" + +# 安装系统依赖 +RUN apt-get update && apt-get install -y --no-install-recommends \ + ffmpeg \ + libportaudio2 \ + dnsutils \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app + +# 安装 Python 依赖 +COPY pyproject.toml . +RUN pip install --no-cache-dir . + +# 复制应用代码 +COPY miair.py ./ +COPY miair/ ./miair/ + +# 创建配置目录 +RUN mkdir -p /app/conf + +# 暴露端口 +EXPOSE 8200 8300 + +ENTRYPOINT ["python", "miair.py", "--conf-path", "/app/conf"] +DOCKERFILE_EOF + +echo -e "${GREEN}✓ Dockerfile 创建完成${NC}" + +# ============================================ +# 步骤 6: 构建 Docker 镜像 +# ============================================ +echo -e "${GREEN}[6/8] 构建 Docker 镜像 (可能需要几分钟)...${NC}" + +docker build -t miair:latest . + +if [ $? -eq 0 ]; then + echo -e "${GREEN}✓ 镜像构建成功${NC}" +else + echo -e "${RED}✗ 镜像构建失败${NC}" + exit 1 +fi + +# ============================================ +# 步骤 7: 停止并删除旧容器（如存在） +# ============================================ +echo -e "${GREEN}[7/8] 清理旧容器...${NC}" + +# 询问用户是否保留之前的配置 +echo -e "${YELLOW}是否保留之前的配置文件？${NC}" +echo "1) 保留配置" +echo "2) 重置配置" +read -p "请选择 (1/2): " KEEP_CONFIG + +# 根据用户选择处理配置 +if [ "$KEEP_CONFIG" = "1" ]; then + # 备份配置 + echo -e "${GREEN}保留配置，备份现有配置...${NC}" + TEMP_CONF="/tmp/miair_conf_backup_$$" + if [ -d "$CONFIG_DIR" ] && [ -n "$(ls -A "$CONFIG_DIR" 2>/dev/null)" ]; then + cp -r "$CONFIG_DIR" "$TEMP_CONF" + echo -e "${GREEN}✓ 配置已备份${NC}" + else + TEMP_CONF="" + echo -e "${YELLOW}未找到现有配置，将使用默认配置${NC}" + fi +else + echo -e "${YELLOW}重置配置文件...${NC}" + rm -rf "$CONFIG_DIR" 2>/dev/null || true + TEMP_CONF="" +fi + +# 清理旧容器 +docker rm -f "$CONTAINER_NAME" 2>/dev/null || true + +# 确保配置目录存在 +mkdir -p "$CONFIG_DIR" +echo -e "${GREEN}✓ 清理完成${NC}" + +# ============================================ +# 步骤 8: 启动容器 +# ============================================ +echo -e "${GREEN}[8/8] 启动 MiAir 容器...${NC}" + +# 构建环境变量参数 +ENV_VARS="-e TZ=Asia/Shanghai -e MIAIR_HOSTNAME=$HOST_IP" +[ -n "$MI_USER" ] && ENV_VARS="$ENV_VARS -e MI_USER=$MI_USER" +[ -n "$MI_PASS" ] && ENV_VARS="$ENV_VARS -e MI_PASS=$MI_PASS" +[ -n "$MI_DID" ] && ENV_VARS="$ENV_VARS -e MI_DID=$MI_DID" + +# 启动命令 +docker run -d \ + --name "$CONTAINER_NAME" \ + --network=host \ + $ENV_VARS \ + -v "$CONFIG_DIR:/app/conf" \ + --restart unless-stopped \ + --cap-add=NET_ADMIN \ + --cap-add=NET_BIND_SERVICE \ + --cap-add=NET_BROADCAST \ + "$IMAGE_NAME" + +if [ $? -eq 0 ]; then + echo "" + echo "==========================================" + echo -e "${GREEN} 🎉 MiAir 部署成功！${NC}" + echo "==========================================" + echo "" + + # 恢复备份的配置 + if [ -n "$TEMP_CONF" ] && [ -d "$TEMP_CONF" ]; then + echo -e "${GREEN}恢复配置...${NC}" + cp -r "$TEMP_CONF"/* "$CONFIG_DIR/" 2>/dev/null || true + rm -rf "$TEMP_CONF" + echo -e "${GREEN}✓ 配置已恢复${NC}" + fi + + echo -e "配置文件目录: ${GREEN}$CONFIG_DIR${NC}" + echo -e "Web 管理界面: ${GREEN}http://$HOST_IP:8300${NC}" + echo -e "DLNA 服务端口: ${GREEN}8200${NC}" + echo "" + echo "查看日志命令:" + echo " docker logs -f miair" + echo "" + echo "停止服务:" + echo " docker stop miair" + echo "" + + # 显示日志前 20 行 + echo "最近日志:" + echo "----------------------------------------" + docker logs miair 2>&1 | tail -20 + echo "----------------------------------------" +else + echo -e "${RED}✗ 容器启动失败，请查看日志${NC}" + echo "docker logs miair" + exit 1 +fi diff --git a/manage.sh b/manage.sh new file mode 100644 index 0000000..8264fb9 --- /dev/null +++ b/manage.sh @@ -0,0 +1,71 @@ +#!/bin/bash +# MiAir 管理脚本 +# 用法: ./manage.sh [start|stop|restart|logs|status|update|uninstall] + +APP_DIR="$(cd "$(dirname "$0")" && pwd)" +CONTAINER_NAME="miair" + +case "$1" in + start) + echo "启动 MiAir..." + docker start "$CONTAINER_NAME" + ;; + stop) + echo "停止 MiAir..." + docker stop "$CONTAINER_NAME" + ;; + restart) + echo "重启 MiAir..." + docker restart "$CONTAINER_NAME" + ;; + logs) + if [ "$2" = "-f" ]; then + docker logs -f "$CONTAINER_NAME" + else + docker logs "$CONTAINER_NAME" + fi + ;; + status) + docker ps -a | grep "$CONTAINER_NAME" + ;; + update) + echo "更新 MiAir..." + cd "$APP_DIR" + # 备份配置 + cp -r conf conf.bak + + # 下载最新代码 + wget -O miair.tar.gz https://github.com/KiriChen-Wind/MiAir/archive/refs/heads/main.tar.gz + tar -xzf miair.tar.gz + cp -r MiAir-main/* . + rm -rf MiAir-main miair.tar.gz + + # 重新构建 + docker build -t miair:latest . + docker rm -f "$CONTAINER_NAME" + ./deploy.sh + ;; + uninstall) + echo "卸载 MiAir..." + read -p "确定要删除容器和配置吗? (y/N): " confirm + if [ "$confirm" = "y" ]; then + docker rm -f "$CONTAINER_NAME" + docker rmi miair:latest + rm -rf "$APP_DIR" + echo "卸载完成" + fi + ;; + *) + echo "用法: $0 {start|stop|restart|logs|status|update|uninstall}" + echo "" + echo " start - 启动服务" + echo " stop - 停止服务" + echo " restart - 重启服务" + echo " logs - 查看日志" + echo " logs -f - 实时查看日志" + echo " status - 查看状态" + echo " update - 更新到最新版本" + echo " uninstall - 卸载" + exit 1 + ;; +esac diff --git a/miair/airplay/audio_stream.py b/miair/airplay/audio_stream.py index 1628013..1de70d7 100644 --- a/miair/airplay/audio_stream.py +++ b/miair/airplay/audio_stream.py @@ -20,8 +20,8 @@ # --- 队列参数 --- # 每个 ALAC 包约 8ms (352 samples @ 44100Hz) -# 20 个包 ≈ 160ms 的缓冲上限 -_QUEUE_MAXSIZE = 20 +# 100 个包 ≈ 800ms 的缓冲上限，兼顾低延迟与抗 WiFi 抖动 +_QUEUE_MAXSIZE = 100 class AudioStreamServer: @@ -112,15 +112,19 @@ def stop_streaming(self): log.info("音频流: 停止接收 PCM 数据") def write_pcm(self, data: bytes): - """写入 PCM 音频数据 — 非阻塞，队列满时丢弃旧数据""" + """写入 PCM 音频数据 — 非阻塞，队列满时批量丢弃旧数据腾出空间""" if not self._active: return try: self._audio_queue.put_nowait(data) except queue.Full: - # 丢弃最旧数据，保证新数据优先 (降低延迟) + # 批量丢弃旧数据，一次性腾出足够空间，避免反复 put/get 开销 + dropped = 0 + target = _QUEUE_MAXSIZE // 4 # 丢弃 25% 腾出充裕空间 try: - self._audio_queue.get_nowait() + for _ in range(target): + self._audio_queue.get_nowait() + dropped += 1 except queue.Empty: pass try: @@ -132,6 +136,15 @@ def write_pcm(self, data: bytes): # WAV 模式 — 直接输出 PCM，零编码延迟 # ============================================================ + def _reject_if_inactive(self) -> web.Response | None: + """流已停止（TEARDOWN 后）返回 404 拒绝拉取，避免音箱把空响应当成 EOF + 反复重连该 URL 造成死循环（氛围灯闪烁、固件崩溃）。未停止返回 None。 + """ + if not self._active: + log.info("AirPlay: 流已停止，拒绝拉取 (404)") + return web.Response(status=404, headers={"Connection": "close"}) + return None + def _build_wav_header(self, data_size: int = 0x7FFFFF00) -> bytes: byte_rate = self._sample_rate * self._channels * self._sample_width block_align = self._channels * self._sample_width @@ -150,6 +163,10 @@ async def _handle_stream_wav(self, request: web.Request) -> web.StreamResponse: 使用专用写入线程从队列批量读取数据，通过 asyncio 事件写回 HTTP 响应， 避免每个包都经过 asyncio.to_thread 的调度开销。 """ + rejected = self._reject_if_inactive() + if rejected is not None: + return rejected + response = web.StreamResponse( status=200, headers={ @@ -193,18 +210,18 @@ def _reader_thread(): chunk = self._audio_queue.get(timeout=0.02) if chunk is None: break - with data_lock: - pending_data.append(chunk) - # 批量读取更多数据 (减少唤醒次数) - for _ in range(11): + # 本地批量收集，避免每个 chunk 都加锁 + local_batch = [chunk] + for _ in range(31): try: extra = self._audio_queue.get_nowait() if extra is None: break - with data_lock: - pending_data.append(extra) + local_batch.append(extra) except queue.Empty: break + with data_lock: + pending_data.extend(local_batch) loop.call_soon_threadsafe(data_ready.set) empty_streak = 0 except queue.Empty: @@ -217,7 +234,7 @@ def _reader_thread(): loop.call_soon_threadsafe(data_ready.set) continue except Exception as e: - pass + log.error(f"WAV reader 异常: {e}") finally: writer_done = True loop.call_soon_threadsafe(data_ready.set) @@ -233,8 +250,8 @@ def _reader_thread(): await data_ready.wait() data_ready.clear() with data_lock: - chunks = pending_data[:] - pending_data.clear() + chunks = pending_data + pending_data = [] if chunks: # 批量写入：合并所有 chunk 一次性写出 await response.write(b"".join(chunks)) @@ -259,6 +276,10 @@ def _reader_thread(): # ============================================================ async def _handle_stream_mp3(self, request: web.Request) -> web.StreamResponse: + rejected = self._reject_if_inactive() + if rejected is not None: + return rejected + response = web.StreamResponse( status=200, headers={ @@ -348,16 +369,18 @@ def feed_ffmpeg(): chunk = self._audio_queue.get(timeout=0.05) if chunk is None: break - proc.stdin.write(chunk) - # 批量喂数据 - for _ in range(4): + # 批量收集数据后一次性写入，减少 write 系统调用次数 + chunks = [chunk] + for _ in range(15): try: extra = self._audio_queue.get_nowait() if extra is None: - break - proc.stdin.write(extra) + proc.stdin.close() + return + chunks.append(extra) except queue.Empty: break + proc.stdin.write(b"".join(chunks)) empty_streak = 0 except queue.Empty: empty_streak += 1 @@ -383,7 +406,7 @@ def feed_ffmpeg(): # 从 ffmpeg stdout 读取并写给客户端 try: while self._active and not self._abort: - audio_data = await asyncio.to_thread(proc.stdout.read, 1024) + audio_data = await asyncio.to_thread(proc.stdout.read, 8192) if not audio_data or self._abort: break await response.write(audio_data) diff --git a/miair/airplay/mdns.py b/miair/airplay/mdns.py index 0d1e27b..85f4c25 100644 --- a/miair/airplay/mdns.py +++ b/miair/airplay/mdns.py @@ -12,6 +12,8 @@ from zeroconf import ServiceInfo, Zeroconf, IPVersion from zeroconf._exceptions import ServiceNameAlreadyRegistered, NonUniqueNameException +from miair.config import Config + log = logging.getLogger("miair") @@ -39,9 +41,10 @@ def start(self): def _run_mdns(self): """在独立线程中运行 mDNS""" try: - # 获取本机 IP 地址 + # mDNS 发布地址必须和 DLNA/AirPlay 流 URL 使用同一个配置 IP。 ip = self._get_ip() ip_bytes = socket.inet_aton(ip) + server_name = self._get_server_name() log.info(f"AirPlay mDNS 启动中，IP: {ip}:{self.rtsp_port}") @@ -91,7 +94,7 @@ def _run_mdns(self): addresses=[ip_bytes], port=self.rtsp_port, properties=airplay_properties, - server=f"{self.hostname}.local.", + server=server_name, ) # ===== RAOP 服务 (_raop._tcp) ===== @@ -121,7 +124,7 @@ def _run_mdns(self): addresses=[ip_bytes], port=self.rtsp_port, properties=raop_properties, - server=f"{self.hostname}.local.", + server=server_name, ) # 只注册 RAOP 服务，强制 iOS 使用 AirPlay 1 (RAOP) 协议 @@ -183,16 +186,33 @@ def stop(self): self._thread.join(timeout=2) def _get_ip(self) -> str: - """获取本机 IP 地址""" + """获取 mDNS 发布 IP，优先使用传入的统一 hostname。""" try: - s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - s.connect(("8.8.8.8", 80)) - ip = s.getsockname()[0] - s.close() - return ip - except Exception: - return "127.0.0.1" + socket.inet_aton(self.hostname) + if self.hostname not in ("0.0.0.0", "127.0.0.1"): + return self.hostname + except OSError: + pass + return Config._detect_local_ip() + + def _get_server_name(self) -> str: + """mDNS server 字段使用主机名，不能把 IP 拼成 192.168.x.x.local。""" + safe_device = re_safe_hostname(self.device_name) + device_id = self.device_id.replace(":", "").lower() + return f"miair-{safe_device[:40]}-{device_id}.local." def update_port(self, port: int): """更新 RTSP 端口（动态分配后调用）""" self.rtsp_port = port + + +def re_safe_hostname(value: str) -> str: + """生成可用于 mDNS server 字段的 ASCII 主机名片段。""" + chars = [] + for ch in value.lower(): + if "a" <= ch <= "z" or "0" <= ch <= "9": + chars.append(ch) + elif chars and chars[-1] != "-": + chars.append("-") + hostname = "".join(chars).strip("-") + return hostname or "speaker" diff --git a/miair/airplay/server.py b/miair/airplay/server.py index 1b4fcd2..5debd20 100644 --- a/miair/airplay/server.py +++ b/miair/airplay/server.py @@ -6,8 +6,10 @@ import asyncio import base64 +from collections import deque +from dataclasses import dataclass import logging -import os +import queue import socket import struct import subprocess @@ -23,9 +25,224 @@ from miair.airplay.audio_stream import AudioStreamServer from miair.airplay.mdns import AirPlayMDNS from miair.airplay.playfair import PlayFair +from miair.config import Config log = logging.getLogger("miair") + +# ============================================================ +# 核心数据结构 +# ============================================================ + +@dataclass +class PacketData: + """解码队列中的音频包""" + seq: int # 序列号；-1 = 静音帧 + timestamp: int # RTP 时间戳 (32-bit)；0 = 无时间戳 + payload: bytes # 加密音频数据或静音帧 + + +class JitterBuffer: + """RTP 包环形缓冲区 + + 512 条目 (~4.1 秒 @ 44100Hz/352spf)，自动淘汰最旧条目。 + 替代 dict-based jitter_buffer，提供更大的抖动容忍窗口。 + """ + + BUFFER_SIZE = 512 + + def __init__(self, max_size: int = BUFFER_SIZE): + self._max_size = max_size + self._packets: dict[int, tuple[int, bytes]] = {} # seq -> (timestamp, payload) + self._order: deque[int] = deque() # 插入顺序，用于淘汰 + + def insert(self, seq: int, rtp_timestamp: int, payload: bytes) -> None: + """插入包。满时淘汰最旧条目。重复 seq 覆盖。""" + if seq in self._packets: + self._packets[seq] = (rtp_timestamp, payload) + return + while len(self._packets) >= self._max_size: + old_seq = self._order.popleft() + self._packets.pop(old_seq, None) + self._packets[seq] = (rtp_timestamp, payload) + self._order.append(seq) + + def has(self, seq: int) -> bool: + return seq in self._packets + + def pop(self, seq: int) -> tuple[int, bytes] | None: + """取出并删除指定 seq 的包。""" + pkt = self._packets.pop(seq, None) + return pkt + + def drain(self, start_seq: int) -> list[tuple[int, int, bytes]]: + """从 start_seq 开始按序取出连续的包，遇到缺口停止。 + 返回 [(seq, timestamp, payload), ...]""" + result = [] + seq = start_seq + while seq in self._packets: + ts, payload = self._packets.pop(seq) + result.append((seq, ts, payload)) + seq = (seq + 1) & 0xFFFF + return result + + def gap_missing(self, next_seq: int) -> list[int]: + """返回 next_seq 到下一个可用包之间的缺失 seq 列表。最多扫描 32 个位置。""" + missing = [] + seq = next_seq + for _ in range(32): + if seq in self._packets: + break + missing.append(seq) + seq = (seq + 1) & 0xFFFF + else: + # 扫描了 32 个位置都没找到，只返回前几个 + missing = missing[:8] + return missing + + def next_available_after(self, seq: int) -> int | None: + """找 seq 之后最近的可用 seq。""" + s = seq + for _ in range(self._max_size): + if s in self._packets: + return s + s = (s + 1) & 0xFFFF + return None + + def clear(self) -> None: + self._packets.clear() + self._order.clear() + + def __len__(self) -> int: + return len(self._packets) + + def __contains__(self, seq: int) -> bool: + return seq in self._packets + + +class PlaybackPacer: + """帧释放调度器 — 将 RTP 时间戳映射到本地时钟，控制解码线程的帧释放时机 + + 时间锚点由 D4 sync 包 (RTCP TIME_ANNOUNCE) 提供： + playAtRtpTimestamp 对应 NTP 时间 → 映射到本地 perf_counter。 + 无锚点时退化为启动缓冲模式（累积 32 帧后立即释放）。 + """ + + def __init__(self, sample_rate: int = 44100): + self._sample_rate = sample_rate + # 锚点: RTP 时间戳 → 本地 perf_counter 时间 + self._anchor_rtp_ts: int | None = None + self._anchor_perf: float = 0.0 + self._lock = threading.Lock() + # 漂移校正: 实际/期望时间比的 EMA + self._drift_rate: float = 1.0 + # 目标延迟: 帧释放提前量，补偿 HTTP 缓冲 + 网络 + 音箱缓冲 + self._target_latency_sec: float = 0.200 + # 启动缓冲 + self._startup_count: int = 0 + self._startup_target: int = 32 # ~256ms + self._started: bool = False + + @property + def has_anchor(self) -> bool: + return self._anchor_rtp_ts is not None + + def update_anchor(self, sender_rtp_ts: int, ntp_time: float, + play_at_rtp_ts: int) -> None: + """从 D4 sync 包更新时间锚点。 + + D4 包含义: sender 在 NTP 时间 ntp_time 时，RTP 时钟为 sender_rtp_ts， + 且 play_at_rtp_ts 对应的音频应该被播放。 + 我们用 play_at_rtp_ts 作为锚点，因为它直接告诉我们「这个 RTP 时间戳 + 应该在什么时刻播放」。 + """ + now_perf = time.perf_counter() + with self._lock: + if self._anchor_rtp_ts is None: + # 首次同步: 建立锚点 + self._anchor_rtp_ts = play_at_rtp_ts + self._anchor_perf = now_perf + self._target_latency_sec + self._started = False + self._startup_count = 0 + else: + # 后续同步: 计算漂移率 + audio_elapsed = (play_at_rtp_ts - self._anchor_rtp_ts) / self._sample_rate + real_elapsed = now_perf - self._anchor_perf + if audio_elapsed > 0.5: + measured_rate = real_elapsed / audio_elapsed + # EMA 更新 (alpha=0.05 温和收敛) + self._drift_rate += 0.05 * (measured_rate - self._drift_rate) + # 定期重锚点防止累积误差 + self._anchor_rtp_ts = play_at_rtp_ts + self._anchor_perf = now_perf + self._target_latency_sec + + def wait_for_frame(self, rtp_timestamp: int) -> bool: + """解码线程调用。等到帧应该释放的时刻。 + 返回 True = 播放，False = 太晚了跳过。 + """ + if rtp_timestamp == 0: + return True # 静音帧直接播放 + + with self._lock: + if self._anchor_rtp_ts is None: + # 无锚点: 启动缓冲模式 + self._startup_count += 1 + if self._startup_count >= self._startup_target: + self._started = True + return True + + # 计算此帧应该释放的时刻 + audio_offset = (rtp_timestamp - self._anchor_rtp_ts) / self._sample_rate + target_perf = self._anchor_perf + audio_offset * self._drift_rate + + now = time.perf_counter() + wait_time = target_perf - now + + if wait_time > 0.005: # 超过 5ms 才 sleep + time.sleep(wait_time) + return True + elif wait_time < -0.100: # 超过 100ms 过期 + return False # 跳过 + else: + return True # 稍微迟到但可接受 + + def reset(self) -> None: + """FLUSH 时重置。""" + with self._lock: + self._anchor_rtp_ts = None + self._anchor_perf = 0.0 + self._drift_rate = 1.0 + self._startup_count = 0 + self._started = False + + +class NTPClockSync: + """NTP 时钟同步 — 跟踪本机与 iPhone 的网络延迟""" + + NTP_EPOCH_OFFSET = 2208988800.0 # 1900-01-01 到 1970-01-01 的秒数 + + def __init__(self): + self._latency_ms: float = 50.0 # 估计的单向延迟 (ms) + self._rtt_samples: deque[float] = deque(maxlen=20) + self._lock = threading.Lock() + + @property + def latency_ms(self) -> float: + return self._latency_ms + + def update_latency(self, request_sent_mono: float, response_recv_mono: float) -> None: + """从 timing exchange 的往返时间更新延迟估计。""" + rtt = response_recv_mono - request_sent_mono + if rtt <= 0 or rtt > 1.0: # 丢弃异常值 + return + with self._lock: + self._rtt_samples.append(rtt) + if len(self._rtt_samples) >= 3: + sorted_rtts = sorted(self._rtt_samples) + median_rtt = sorted_rtts[len(sorted_rtts) // 2] + self._latency_ms = (median_rtt / 2.0) * 1000.0 + + # AirPort 私钥 (用于 AirPlay 1 RSA 认证) AIRPORT_PRIVATE_KEY = ( "-----BEGIN RSA PRIVATE KEY-----\n" @@ -138,10 +355,12 @@ def __init__(self, hostname: str, device_name: str = "MiAir", shared_zeroconf=No self._resampler = None self._session_key: bytes | None = None self._session_iv: bytes | None = None + self._session_iv16: bytes | None = None # 预切片的 16 字节 IV，避免每包切片 self._audio_format = 0 self._sample_rate = 44100 self._channels = 2 self._fmtp_params: list[str] = [] # SDP fmtp 参数 + self._silence_frame: bytes = b'\x00' * 1408 # 默认 1 帧静音 (44100Hz/2ch/16bit/352spf) # 回调 self.on_play_start: Callable | None = None @@ -157,6 +376,23 @@ def __init__(self, hostname: str, device_name: str = "MiAir", shared_zeroconf=No self._is_playing: bool = False # 是否正在播放 self._loop: asyncio.AbstractEventLoop | None = None # 事件循环引用（用于跨线程回调） + # RTP 状态跟踪（用于 FLUSH/RECORD 响应的 RTP-Info 头） + self._last_rtp_seq: int = 0 + self._last_rtp_timestamp: int = 0 + # RTCP 重传请求 + self._rtcp_control_socket: socket.socket | None = None + self._rtcp_control_addr: tuple | None = None # iPhone 的 control 地址 + self._rtp_data_socket: socket.socket | None = None # RTP 数据 socket（用于注入重传包） + self._flush_flag = threading.Event() # FLUSH 请求标志（跨线程通知） + + # 时间同步（新增） + self._timing_pacer: PlaybackPacer | None = None + self._clock_sync: NTPClockSync | None = None + self._timing_client_addr: tuple | None = None # iPhone timing 端口地址 + self._timing_socket: socket.socket | None = None + self._timing_request_seq: int = 0 + self._rtsp_client_addr: tuple | None = None # RTSP 客户端 IP + def _generate_device_id(self) -> str: """生成设备 MAC 地址格式的 ID @@ -173,15 +409,14 @@ def device_id_bin(self) -> bytes: return int(self.device_id.replace(":", ""), base=16).to_bytes(6, "big") def _get_ipv4(self) -> str: - """获取本机 IPv4 地址""" + """获取 AirPlay 认证使用的 IPv4，保持与服务发布地址一致。""" try: - s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - s.connect(("8.8.8.8", 80)) - ip = s.getsockname()[0] - s.close() - return ip - except Exception: - return "127.0.0.1" + socket.inet_pton(socket.AF_INET, self.hostname) + if self.hostname not in ("0.0.0.0", "127.0.0.1"): + return self.hostname + except OSError: + pass + return Config._detect_local_ip() @property def ipv4_bin(self) -> bytes: @@ -271,6 +506,7 @@ def _safe_call_on_play_stop(self): def _handle_rtsp_client(self, sock: socket.socket, addr: tuple): """处理 RTSP 客户端连接""" log.info(f"AirPlay 客户端连接: {addr}") + self._rtsp_client_addr = addr # 存储客户端地址供 SETUP 解析端口使用 session_active = False rtp_socket = None rtp_thread = None @@ -280,6 +516,8 @@ def _handle_rtsp_client(self, sock: socket.socket, addr: tuple): # 设置客户端 socket 超时，防止无限阻塞导致线程卡死 sock.settimeout(30.0) + # 关闭 Nagle 算法，RTSP 响应立即发送 + sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) try: while self._running: @@ -336,7 +574,6 @@ def _handle_rtsp_client(self, sock: socket.socket, addr: tuple): body = body[:content_length] cseq = headers.get("CSeq", "0") - log.info(f"RTSP {method} {path} CSeq={cseq} body={len(body)} bytes") if method == "OPTIONS": response_headers = { @@ -346,16 +583,12 @@ def _handle_rtsp_client(self, sock: socket.socket, addr: tuple): # AirPlay 1 认证: 响应 Apple-Challenge apple_challenge = headers.get("Apple-Challenge") if apple_challenge: - log.info(f"Apple-Challenge: {apple_challenge}") apple_response = AP1Security.compute_apple_response( apple_challenge, self.ipv4_bin, self.device_id_bin, ) - log.info(f"Apple-Response: {apple_response[:50]}...") response_headers["Apple-Response"] = apple_response - else: - log.info("OPTIONS: 无 Apple-Challenge") self._send_rtsp_response(sock, 200, cseq, response_headers) elif method == "ANNOUNCE": @@ -390,10 +623,38 @@ def _handle_rtsp_client(self, sock: socket.socket, addr: tuple): break elif method == "FLUSH": - log.info("RTSP FLUSH: 清空音频缓冲区") - # 仅清空队列，不停止流服务器，避免断开客户端 - self._stream_server.start_streaming() - self._send_rtsp_response(sock, 200, cseq) + # FLUSH 中的 RTP-Info 头告知接收端：从此 seq/rtptime 开始新的播放 + rtp_info = headers.get("RTP-Info", "") + flush_seq = 0 + flush_rtptime = 0 + if rtp_info: + for part in rtp_info.split(";"): + part = part.strip() + if part.startswith("seq="): + try: flush_seq = int(part[4:]) + except ValueError: pass + elif part.startswith("rtptime="): + try: flush_rtptime = int(part[8:]) + except ValueError: pass + log.info(f"RTSP FLUSH: seq={flush_seq} rtptime={flush_rtptime}") + # 清空音频缓冲区但不停止流服务器 + self._stream_server.start_streaming() + # 通知 RTP 接收线程重置 jitter buffer + self._flush_flag.set() + # 响应中返回当前接收端的 RTP 状态 + self._send_rtsp_response(sock, 200, cseq, { + "RTP-Info": f"seq={self._last_rtp_seq};rtptime={self._last_rtp_timestamp}", + }) + + elif method == "FLUSHBUFFERED": + # FLUSHBUFFERED 用于 buffered 模式，包含 from_seq/until_seq 范围 + rtp_info = headers.get("RTP-Info", "") + log.info(f"RTSP FLUSHBUFFERED: RTP-Info={rtp_info}") + self._stream_server.start_streaming() + self._flush_flag.set() + self._send_rtsp_response(sock, 200, cseq, { + "RTP-Info": f"seq={self._last_rtp_seq};rtptime={self._last_rtp_timestamp}", + }) elif method == "GET_PARAMETER": vol_body = f"volume: {self._last_volume_db:.2f}\r\n".encode() @@ -403,9 +664,11 @@ def _handle_rtsp_client(self, sock: socket.socket, addr: tuple): }) sock.sendall(vol_body) + elif method == "SET_VOLUME_NOTIFICATION": + self._send_rtsp_response(sock, 200, cseq) + elif method == "SET_PARAMETER": content_type = headers.get("Content-Type", "") - log.info(f"SET_PARAMETER: Content-Type={content_type}, body size={len(body)}") if not content_type.startswith("image/"): body_str = body.decode("utf-8", errors="replace") @@ -505,8 +768,6 @@ def _send_rtsp_binary_response(self, sock: socket.socket, status: int, def _handle_announce(self, sock: socket.socket, headers: dict, body: bytes, cseq: str): """处理 ANNOUNCE 请求 - 解析 SDP""" sdp = body.decode("utf-8", errors="replace") - log.info(f"ANNOUNCE SDP:\n{sdp}") - log.info(f"ANNOUNCE headers: {headers}") # 解析 SDP 提取音频参数 self._sample_rate = 44100 @@ -529,24 +790,17 @@ def _handle_announce(self, sock: socket.socket, headers: dict, body: bytes, cseq log.info(f"从 SDP 中识别到客户端名称: {self._client_name}") if line.startswith("a=rtpmap:"): - # 例如: a=rtpmap:96 AppleLossless parts = line.split() - log.info(f"Found rtpmap: {parts}") if len(parts) >= 2: fmt = parts[1] if "AppleLossless" in fmt: - self._audio_format = 0x2 # ALAC - log.info(f"识别到 ALAC 格式") + self._audio_format = 0x2 elif "mpeg4-generic" in fmt: - self._audio_format = 0x4 # AAC - log.info(f"识别到 AAC 格式") + self._audio_format = 0x4 elif "L16" in fmt or "PCM" in fmt: - self._audio_format = 0x1 # PCM - log.info(f"识别到 PCM 格式") + self._audio_format = 0x1 elif line.startswith("a=fmtp:"): - # ALAC fmtp: a=fmtp:96 352 0 16 40 10 14 2 255 0 0 44100 parts = line.split() - log.info(f"Found fmtp: {parts}") self._fmtp_params = parts[1:] # 保存完整 fmtp 参数（去掉 payload type） if len(parts) >= 12: try: @@ -570,8 +824,6 @@ def _handle_announce(self, sock: socket.socket, headers: dict, body: bytes, cseq iv_data += "=" * (4 - len(iv_data) % 4) if len(iv_data) % 4 else "" aes_iv = base64.b64decode(iv_data) - log.info(f"解析结果: audio_format={self._audio_format}, sr={self._sample_rate}, ch={self._channels}") - if aes_key and aes_iv: if aes_key_type == "rsa": # 解密 RSA AES 密钥 @@ -590,9 +842,11 @@ def _handle_announce(self, sock: socket.socket, headers: dict, body: bytes, cseq self._session_key = None self._session_iv = aes_iv + self._session_iv16 = aes_iv[:16] if aes_iv else None else: self._session_key = None self._session_iv = None + self._session_iv16 = None log.info(f"音频未加密") # 初始化音频解码器 @@ -692,6 +946,8 @@ def _init_decoder(self): ) self._stream_server.set_audio_params(self._sample_rate, self._channels, 2) + # 预计算单帧静音数据 (352 samples * ch * 2 bytes)，避免循环中反复分配 + self._silence_frame = b'\x00' * (self._sample_rate * self._channels * 2 * 352 // self._sample_rate) log.info(f"音频解码器初始化: fmt={self._audio_format}, sr={self._sample_rate}, ch={self._channels}, bits={bitdepth}") except Exception as e: log.error(f"解码器初始化失败: {e}") @@ -708,13 +964,24 @@ def _handle_setup(self, sock: socket.socket, headers: dict, cseq: str) -> tuple: - timing_port: 接收/发送 NTP timing 包 """ transport = headers.get("Transport", "") - log.info(f"SETUP Transport: {transport}") + + # 解析 iPhone 的 control_port 和 timing_port + client_timing_port = 0 + client_control_port = 0 + for part in transport.split(";"): + part = part.strip() + if "timing_port" in part: + try: client_timing_port = int(part.split("=")[1]) + except (ValueError, IndexError): pass + elif "control_port" in part: + try: client_control_port = int(part.split("=")[1]) + except (ValueError, IndexError): pass # 创建 RTP 接收 socket (server_port - 音频数据) rtp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) rtp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - # 增大内核 UDP 接收缓冲区，防止高频小包场景下内核丢包 - rtp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 524288) + # 增大内核 UDP 接收缓冲区到 1MB，防止高频小包场景下内核丢包 + rtp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1048576) rtp_socket.settimeout(1.0) rtp_socket.bind(("0.0.0.0", 0)) server_port = rtp_socket.getsockname()[1] @@ -726,13 +993,26 @@ def _handle_setup(self, sock: socket.socket, headers: dict, cseq: str) -> tuple: control_socket.bind(("0.0.0.0", 0)) control_port = control_socket.getsockname()[1] + # 存储 socket 引用，供 RTCP 重传使用 + self._rtcp_control_socket = control_socket + self._rtp_data_socket = rtp_socket + # 创建 timing socket (NTP 时间同步) timing_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) timing_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - timing_socket.settimeout(2.0) + timing_socket.settimeout(1.0) # 短超时，timing_loop 内部管理发送间隔 timing_socket.bind(("0.0.0.0", 0)) timing_port = timing_socket.getsockname()[1] + # 初始化时间同步对象 + self._timing_socket = timing_socket + self._clock_sync = NTPClockSync() + self._timing_pacer = PlaybackPacer(self._sample_rate) + if client_timing_port > 0 and self._rtsp_client_addr: + self._timing_client_addr = (self._rtsp_client_addr[0], client_timing_port) + if client_control_port > 0 and self._rtsp_client_addr: + self._rtcp_control_addr = (self._rtsp_client_addr[0], client_control_port) + timing_thread = threading.Thread( target=self._timing_loop, args=(timing_socket,), @@ -767,23 +1047,70 @@ def _handle_setup(self, sock: socket.socket, headers: dict, cseq: str) -> tuple: return True, rtp_socket, control_socket, timing_socket + def _request_retransmit(self, start_seq: int, count: int): + """向 iPhone 发送 RTCP REXMIT_REQUEST 请求重传丢失的 RTP 包 + + RTCP type 0xd5 (213) 格式: + byte 0: 0x80 (version=2) + byte 1: 0xd5 (type) + byte 2-3: length in 32-bit words + byte 4-5: start sequence number + byte 6-7: amount of following missing packets + """ + sock = self._rtcp_control_socket + addr = self._rtcp_control_addr + if not sock or not addr: + return + try: + req = bytearray(8) + req[0] = 0x80 + req[1] = 0xd5 # REXMIT_REQUEST + req[2:4] = (2).to_bytes(2, 'big') # length = 8 bytes / 4 = 2 words + req[4:6] = start_seq.to_bytes(2, 'big') + req[6:8] = count.to_bytes(2, 'big') + sock.sendto(bytes(req), addr) + except Exception as e: + log.debug(f"RTCP 重传请求失败: {e}") + def _rtcp_loop(self, rtcp_socket: socket.socket): - """RTCP 控制包接收循环""" + """RTCP 控制包接收循环 + + 处理: + - TIME_ANNOUNCE (0xd4/212): NTP 时间同步 + - REXMIT_RESPONSE (0xd6/214): iPhone 返回的重传包，注入 RTP 数据流 + """ log.info("RTCP 线程启动") try: while self._running: try: - data, addr = rtcp_socket.recvfrom(256) + data, addr = rtcp_socket.recvfrom(1500) if not data or len(data) < 4: continue - # RTCP 包处理 - 主要用于时间同步 - # AirPlay 1 使用 RTCP 类型 212 (0xd4) 发送时间信息 - if len(data) >= 8: - ptype = data[1] - if ptype == 212: # TIME_ANNOUNCE_NTP - # 提取 sender RTP timestamp 和 playAt timestamp - sender_ts = int.from_bytes(data[4:8], 'big') - play_at_ts = int.from_bytes(data[16:20], 'big') if len(data) >= 20 else 0 + + # 记录 iPhone 控制端口地址（用于发送重传请求） + if not self._rtcp_control_addr: + self._rtcp_control_addr = addr + + ptype = data[1] + if ptype == 212: # TIME_ANNOUNCE_NTP (D4 sync 包) + if len(data) >= 20: + sender_rtp_ts = int.from_bytes(data[4:8], 'big') + ntp_sec = int.from_bytes(data[8:12], 'big') + ntp_frac = int.from_bytes(data[12:16], 'big') + ntp_time = ntp_sec + (ntp_frac * 2**-32) + play_at_rtp_ts = int.from_bytes(data[16:20], 'big') + if self._timing_pacer: + self._timing_pacer.update_anchor( + sender_rtp_ts, ntp_time, play_at_rtp_ts) + elif ptype == 214: # REXMIT_RESPONSE — iPhone 重传的 RTP 包 + # data[4:] 是完整的 RTP 包，注入 RTP 数据 socket + rtp_data = data[4:] + if len(rtp_data) >= 12 and self._rtp_data_socket: + try: + rtp_port = self._rtp_data_socket.getsockname()[1] + self._rtp_data_socket.sendto(rtp_data, ('127.0.0.1', rtp_port)) + except Exception as e: + log.debug(f"RTCP 重传包注入失败: {e}") except socket.timeout: continue except OSError: @@ -806,65 +1133,109 @@ def _handle_record(self, sock: socket.socket, cseq: str): self._send_rtsp_response(sock, 200, cseq, { "Audio-Latency": "0", + "RTP-Info": f"seq={self._last_rtp_seq};rtptime={self._last_rtp_timestamp}", }) def _timing_loop(self, timing_socket: socket.socket): - """RAOP NTP 时间同步响应循环 + """双向 NTP 时间同步循环 - AirPlay 1 timing 使用 RTP 格式: - 请求: 0x80 0x52 (type=0x52=82 即 TIME_REQUEST) + seq(2) + zero(8) + ref_time(8) + recv_time(8) - 响应: 0x80 0x53 (type=0x53=83 即 TIME_RESPONSE) + seq(2) + ref_time(8) + recv_time(8) + send_time(8) - 共 32 字节 + - 响应 iPhone 的 0x52 timing 请求（原有逻辑） + - 主动向 iPhone 发送 0x52 timing 请求（每 3 秒一次，前 3 次 300ms 间隔） + - 从 0x53 响应计算 RTT，更新 NTPClockSync 延迟估计 """ - log.info("Timing 线程启动") + NTP_EPOCH = NTPClockSync.NTP_EPOCH_OFFSET + # 发送间隔: 前 3 次 300ms（快速收敛），之后 3 秒 + _fast_pings_left = 3 + _fast_interval = 0.3 + _normal_interval = 3.0 + _last_send_time = 0.0 + _pending_seq: int | None = None + _send_mono: float = 0.0 + try: while self._running: + # --- 主动发送 timing request --- + now = time.time() + client_addr = self._timing_client_addr + interval = _fast_interval if _fast_pings_left > 0 else _normal_interval + if client_addr and (now - _last_send_time) >= interval: + self._timing_request_seq = (self._timing_request_seq + 1) & 0xFFFF + _pending_seq = self._timing_request_seq + _send_mono = time.perf_counter() + + req = bytearray(32) + req[0] = 0x80 + req[1] = 0x52 # TIME_REQUEST + req[2:4] = self._timing_request_seq.to_bytes(2, 'big') + # bytes 24-31: our send time as NTP timestamp + ntp_now = now + NTP_EPOCH + ntp_sec = int(ntp_now) + ntp_frac = int((ntp_now - ntp_sec) * (2**32)) + req[24:28] = ntp_sec.to_bytes(4, 'big') + req[28:32] = ntp_frac.to_bytes(4, 'big') + + try: + timing_socket.sendto(bytes(req), client_addr) + if _fast_pings_left > 0: + _fast_pings_left -= 1 + except Exception: + pass + _last_send_time = now + + # --- 接收并处理包 --- try: data, addr = timing_socket.recvfrom(256) - if not data or len(data) < 32: - continue + except socket.timeout: + continue + if not data or len(data) < 32: + continue - # 检查是否为 timing request (type byte = 0x52 or 0xd2) - ptype = data[1] & 0x7f # 去掉 marker bit - if ptype != 0x52: - continue + ptype = data[1] & 0x7f - now = time.time() - # NTP 时间戳 (从 1900-01-01 开始的秒数) - ntp_now = now + 2208988800.0 - ntp_sec = int(ntp_now) - ntp_frac = int((ntp_now - ntp_sec) * (2**32)) + if ptype == 0x52: + # iPhone 发来 timing request → 回复 response + recv_now = time.time() + ntp_recv = recv_now + NTP_EPOCH + ntp_sec = int(ntp_recv) + ntp_frac = int((ntp_recv - ntp_sec) * (2**32)) response = bytearray(32) - response[0] = 0x80 # RTP version 2 - response[1] = 0xd3 # timing response type (0x53 | 0x80 marker) + response[0] = 0x80 + response[1] = 0xd3 # timing response (0x53 | marker) response[2:4] = data[2:4] # 复制 sequence number - - # bytes 4-11: 复制请求中的 reference send time (来自请求的 bytes 24-31) - if len(data) >= 32: - response[4:12] = data[24:32] - # bytes 12-19: receive timestamp (我们收到请求的时间) + response[4:12] = data[24:32] # 复制 reference send time response[12:16] = ntp_sec.to_bytes(4, 'big') response[16:20] = ntp_frac.to_bytes(4, 'big') - # bytes 20-27: send timestamp (我们发送响应的时间) - send_now = time.time() + 2208988800.0 + send_now = time.time() + NTP_EPOCH send_sec = int(send_now) send_frac = int((send_now - send_sec) * (2**32)) response[20:24] = send_sec.to_bytes(4, 'big') response[24:28] = send_frac.to_bytes(4, 'big') timing_socket.sendto(bytes(response), addr) - except socket.timeout: - continue - except OSError: - break - except Exception as e: + + elif ptype == 0x53: + # iPhone 回复 timing response → 计算 RTT + resp_seq = (data[2] << 8) | data[3] + if _pending_seq is not None and resp_seq == _pending_seq: + recv_mono = time.perf_counter() + if self._clock_sync: + self._clock_sync.update_latency(_send_mono, recv_mono) + _pending_seq = None + + except Exception: pass finally: timing_socket.close() def _rtp_receive_loop(self, rtp_socket: socket.socket): - """RTP 音频数据接收循环""" + """RTP 音频数据接收循环 — 两阶段管道 + 时间调度 + + Stage 1 (receiver): recvfrom → 环形缓冲区 → 按序发送到解码队列 + Stage 2 (decoder): 解码队列 → PlaybackPacer 调度 → 解密 → ALAC 解码 → write_pcm + + 接收线程只做 UDP 读取和轻量操作，永远不被解码阻塞。 + """ log.info("RTP 接收线程启动") # 等待流媒体激活 @@ -879,96 +1250,185 @@ def _rtp_receive_loop(self, rtp_socket: socket.socket): return log.info("RTP: 开始接收音频数据") - packet_count = 0 - error_count = 0 - last_seq = 0 + # 预计算常用值 + _session_key = self._session_key + _session_iv16 = self._session_iv16 + _write_pcm = self._stream_server.write_pcm + _decode_audio = self._decode_audio + _silence_frame = self._silence_frame + _recv_buf = bytearray(2048) + + # 解码队列 + decode_queue: queue.Queue[PacketData | None] = queue.Queue(maxsize=200) + running = True + + # ---- Stage 2: 解码线程（带 pacer 调度） ---- + def _decoder_worker(): + _pacer = self._timing_pacer + try: + while running: + try: + item = decode_queue.get(timeout=0.5) + except queue.Empty: + continue + if item is None: + break + + # 时间调度: 等到正确时刻再释放帧 + if _pacer and item.timestamp > 0: + if not _pacer.wait_for_frame(item.timestamp): + continue # 太晚了，跳过 + + # 静音帧直接写入 + if item.seq == -1: + _write_pcm(item.payload) + continue + + # 解密 + payload = item.payload + if _session_key and _session_iv16: + try: + plen = len(payload) + decrypt_len = plen & ~0xF + if decrypt_len > 0: + cipher = AES.new(_session_key, AES.MODE_CBC, _session_iv16) + decrypted = cipher.decrypt(payload[:decrypt_len]) + if decrypt_len < plen: + decrypted = decrypted + bytes(memoryview(payload)[decrypt_len:]) + payload = decrypted + except Exception: + _write_pcm(_silence_frame) + continue + + # 解码并输出 + decoded = _decode_audio(payload) + if decoded: + _write_pcm(decoded) + except Exception as e: + log.error(f"RTP 解码线程异常: {e}") + + decoder_thread = threading.Thread(target=_decoder_worker, daemon=True) + decoder_thread.start() + + # ---- Stage 1: 接收 + 环形缓冲 + 指数退避重传 ---- try: - jitter_buffer = {} # seq -> pcm_data + jb = JitterBuffer(max_size=512) next_seq = -1 - buffer_threshold = 2 # 最小缓冲阈值以降低延迟 (约 16ms) - # 缓冲区最大上限，防止极端情况下内存无限增长 - max_jitter_size = 100 + _pacer = self._timing_pacer + + # 启动缓冲 + STARTUP_BUFFER_TARGET = 32 # ~256ms @ 44100Hz/352spf + startup_buffered = False + + # 重传状态: seq -> (首次请求 perf_counter 时间, 重试次数) + _retransmit_state: dict[int, tuple[float, int]] = {} + _RETRANSMIT_BASE_INTERVAL = 0.040 # 40ms + _RETRANSMIT_MAX_INTERVAL = 1.000 + _RETRANSMIT_GIVE_UP_TIME = 2.0 # 2秒后放弃 while self._running: try: - data, addr = rtp_socket.recvfrom(2048) - if not data or len(data) < 12: + nbytes, addr = rtp_socket.recvfrom_into(_recv_buf) + if nbytes < 12: continue - # RTP 头解析 - seq = int.from_bytes(data[2:4], 'big') - payload_type = data[1] & 0x7f - payload = data[12:] + seq = (_recv_buf[2] << 8) | _recv_buf[3] + rtp_timestamp = (_recv_buf[4] << 24) | (_recv_buf[5] << 16) | \ + (_recv_buf[6] << 8) | _recv_buf[7] + payload = bytes(_recv_buf[12:nbytes]) + + # 跟踪最新 RTP 状态（供 FLUSH/RECORD 响应使用） + self._last_rtp_seq = seq + self._last_rtp_timestamp = rtp_timestamp - # 初始对齐 next_seq if next_seq == -1: next_seq = seq - log.info(f"RTP: 初始序列号 {next_seq}") - - # 将原始 payload 放入抖动缓冲区 - jitter_buffer[seq] = payload - packet_count += 1 - - # 缓冲区过大时强制清理最老的包，防止内存泄漏 - if len(jitter_buffer) > max_jitter_size: - # 丢弃最老的包，跳转到最新的包 - oldest = min(jitter_buffer.keys(), key=lambda x: (x - next_seq) & 0xFFFF) - while len(jitter_buffer) > buffer_threshold and oldest != next_seq: - jitter_buffer.pop(oldest, None) - oldest = min(jitter_buffer.keys(), key=lambda x: (x - next_seq) & 0xFFFF) if jitter_buffer else next_seq - next_seq = min(jitter_buffer.keys(), key=lambda x: (x - next_seq) & 0xFFFF) if jitter_buffer else next_seq - - # 当缓冲区达到一定大小或已收到下一个期望的包时，开始输出 - while True: - if next_seq in jitter_buffer: - ordered_payload = jitter_buffer.pop(next_seq) - - # 解密 — IV 每包相同，但 CBC 要求每次新建 cipher - if self._session_key and self._session_iv: - try: - cipher = AES.new(self._session_key, AES.MODE_CBC, self._session_iv[:16]) - plen = len(ordered_payload) - decrypt_len = plen & ~0xF - if decrypt_len > 0: - decrypted = cipher.decrypt(ordered_payload[:decrypt_len]) - if decrypt_len < plen: - # 用 memoryview 避免尾部切片拷贝 - decrypted = decrypted + bytes(memoryview(ordered_payload)[decrypt_len:]) - ordered_payload = decrypted - # else: 不足 16 字节无需解密 - except Exception as e: - next_seq = (next_seq + 1) & 0xFFFF + + # 插入环形缓冲区 + jb.insert(seq, rtp_timestamp, payload) + + # 启动预缓冲: 等待足够帧数 + if not startup_buffered: + if len(jb) < STARTUP_BUFFER_TARGET: + continue + # 等待 pacer 锚点（如果还没收到 D4 sync） + if _pacer and not _pacer.has_anchor: + if len(jb) < STARTUP_BUFFER_TARGET * 2: + continue + startup_buffered = True + + # 检查 FLUSH 请求 + if self._flush_flag.is_set(): + self._flush_flag.clear() + jb.clear() + next_seq = seq + _retransmit_state.clear() + if _pacer: + _pacer.reset() + continue + + # 按序 drain 到解码队列 + drained = jb.drain(next_seq) + for d_seq, d_ts, d_payload in drained: + try: + decode_queue.put_nowait(PacketData(d_seq, d_ts, d_payload)) + except queue.Full: + pass + if drained: + next_seq = (drained[-1][0] + 1) & 0xFFFF + for d_seq, _, _ in drained: + _retransmit_state.pop(d_seq, None) + + # 丢包检测 + 指数退避重传 + if len(jb) > 8: + missing_seqs = jb.gap_missing(next_seq) + now_mono = time.perf_counter() + + for missing_seq in missing_seqs: + if missing_seq in _retransmit_state: + first_time, retry_count = _retransmit_state[missing_seq] + # 放弃条件 + if now_mono - first_time > _RETRANSMIT_GIVE_UP_TIME: + _retransmit_state.pop(missing_seq, None) + next_avail = jb.next_available_after(missing_seq) + if next_avail is not None: + gap = (next_avail - missing_seq) & 0xFFFF + gap = min(gap, 64) + for _ in range(gap): + try: + decode_queue.put_nowait( + PacketData(-1, 0, _silence_frame)) + except queue.Full: + pass + next_seq = next_avail + # 清理跳过范围的重传状态 + for s in range(missing_seq, next_avail): + _retransmit_state.pop(s & 0xFFFF, None) continue - # 解码音频 - pcm_data = self._decode_audio(ordered_payload) - if pcm_data: - self._stream_server.write_pcm(pcm_data) + # 指数退避重试 + backoff = min( + _RETRANSMIT_BASE_INTERVAL * (2 ** retry_count), + _RETRANSMIT_MAX_INTERVAL + ) + if now_mono - first_time >= backoff * (retry_count + 1): + self._request_retransmit(missing_seq, 1) + _retransmit_state[missing_seq] = (first_time, retry_count + 1) + try: + decode_queue.put_nowait( + PacketData(-1, 0, _silence_frame)) + except queue.Full: + pass else: - error_count += 1 - if error_count > 100: - log.warning(f"RTP: 连续解码失败 {error_count} 次") - error_count = 0 - - last_seq = next_seq - next_seq = (next_seq + 1) & 0xFFFF - - elif len(jitter_buffer) > buffer_threshold: - # 缓冲区过大，说明中间丢包了，跳过丢失的包 - missing_seq = next_seq - next_seq = min(jitter_buffer.keys(), key=lambda x: (x - missing_seq) & 0xFFFF) - if self._codec_context: + # 首次检测: 请求重传 + self._request_retransmit(missing_seq, 1) + _retransmit_state[missing_seq] = (now_mono, 1) try: - self._codec_context.flush_buffers() - except Exception as e: + decode_queue.put_nowait( + PacketData(-1, 0, _silence_frame)) + except queue.Full: pass - continue - else: - break - - if packet_count % 500 == 0: - log.info(f"RTP: 已接收 {packet_count} 个音频包") except socket.timeout: continue @@ -977,25 +1437,30 @@ def _rtp_receive_loop(self, rtp_socket: socket.socket): except Exception as e: log.error(f"RTP 接收错误: {e}") - import traceback - log.error(traceback.format_exc()) finally: + running = False + if self._timing_pacer: + self._timing_pacer.reset() + try: + decode_queue.put_nowait(None) + except queue.Full: + pass rtp_socket.close() - log.info(f"RTP 接收线程结束，共接收 {packet_count} 个包，最后 seq={last_seq}") + log.info("RTP 接收线程结束") def _decode_audio(self, data: bytes) -> bytes | None: """解码音频数据为 PCM""" if not self._codec_context: - # PCM 模式直接返回 (假设是 s16le) return data + silence = self._silence_frame + try: packet = av.packet.Packet(data) frames = self._codec_context.decode(packet) if not frames: - return None + return silence - # 用 memoryview 零拷贝截取有效音频数据 ch2 = self._channels * 2 parts = [] for frame in frames: @@ -1007,12 +1472,9 @@ def _decode_audio(self, data: bytes) -> bytes | None: else: mv = memoryview(resampled.planes[0]) parts.append(bytes(mv[:resampled.samples * ch2])) - return b"".join(parts) if parts else None - except Exception as e: - # 解码失败时返回静音数据，避免音频流中断 - # 返回 10ms 静音数据 - silence_len = self._sample_rate * self._channels * 2 // 100 - return b'\x00' * silence_len + return b"".join(parts) if parts else silence + except Exception: + return silence def _send_rtsp_response(self, sock: socket.socket, code: int, cseq: str, headers: dict | None = None): """发送 RTSP 响应""" diff --git a/miair/airplay/speaker_airplay.py b/miair/airplay/speaker_airplay.py index 16ca0d6..df286aa 100644 --- a/miair/airplay/speaker_airplay.py +++ b/miair/airplay/speaker_airplay.py @@ -93,14 +93,46 @@ async def _play_on_speaker(self, stream_url: str): self._play_grace_until = time.time() + 10.0 # 10秒宽限期 success = await self.controller.play_url(stream_url) if success: - log.info(f"AirPlay 音频已在 {self.device_name} 开始播放") - # 启动状态轮询（打断续播） + log.info(f"AirPlay 音频已在 {self.device_name} 开始播放: {stream_url}") self._start_poll() + if self.config: + default_vol = getattr(self.config, 'default_volume', 0) + follow_dev_vol = getattr(self.config, 'follow_device_volume', False) + if follow_dev_vol: + try: + current_vol = await self.controller.get_volume() + if self.airplay_server: + self.airplay_server._last_volume_db = self._vol_pct_to_db(current_vol) + log.info(f"AirPlay 已跟随设备当前音量到 {self.device_name}: {current_vol}%") + except Exception as e: + log.error(f"AirPlay 获取当前音量失败: {e}") + elif default_vol > 0: + await asyncio.sleep(0.5) + await self.controller.set_volume(default_vol) + if self.airplay_server: + self.airplay_server._last_volume_db = self._vol_pct_to_db(default_vol) + log.info(f"AirPlay 已应用默认音量到 {self.device_name}: {default_vol}%") else: log.warning(f"AirPlay 音频在 {self.device_name} 播放失败") except Exception as e: log.error(f"AirPlay 播放到 {self.device_name} 失败: {e}") + @staticmethod + def _vol_pct_to_db(volume: int) -> float: + """音箱百分比 → AirPlay dB 值（线性映射逆运算） + + iOS 步骤 1-16 线性映射: -28.125 dB ~ 0 dB → 音箱 6% ~ 100% + 逆向: dB = (volume - 6) / 94 * 28.125 - 28.125 + """ + import math + if volume <= 0: + return -144.0 + if volume >= 100: + return 0.0 + if volume <= 6: + return -28.125 + return (volume - 6) / 94.0 * 28.125 - 28.125 + def _on_play_stop(self): """AirPlay 停止播放 @@ -129,7 +161,10 @@ async def _stop_speaker(self): pass def _start_poll(self): - """启动 AirPlay 状态轮询任务""" + """启动 AirPlay 状态轮询任务（仅在 auto_resume_on_interrupt 开启时）""" + # 未开启自动续播则不启动轮询，避免无意义的 API 调用 + if self.config and not getattr(self.config, 'auto_resume_on_interrupt', False): + return if self._poll_task and not self._poll_task.done(): return # 已在运行 self._poll_task = asyncio.create_task(self._poll_speaker_state()) @@ -201,22 +236,23 @@ async def _poll_speaker_state(self): def _on_volume_change(self, vol_db: float): """处理音量改变 - + 注意: 这个回调从 RTSP 线程调用，不在 asyncio 事件循环中。 + + iOS 步骤 1-16 线性映射 dB → 音箱百分比: + 步骤 0 (静音) → 0% + 步骤 1 (-28.125 dB) → 6% + 步骤 8 (-15.0 dB) → 50% + 步骤 16 (0 dB) → 100% """ - # AirPlay 音量范围: -144 (静音) 到 0 (最大) if vol_db <= -144: volume = 0 elif vol_db >= 0: volume = 100 else: - # 使用声压级对数映射 (10^(dB/20))，这更符合人耳听觉和 iOS 的滑动曲线 - # 0dB -> 1.0 (100%) - # -20dB -> 0.1 (10%) - # -40dB -> 0.01 (1%) - volume = int(pow(10, vol_db / 20) * 100) - - # 确保即使在低分贝下也有基本的映射，避免由于 int() 导致的过早归零 + # 线性映射: -28.125 dB ~ 0 dB → 6% ~ 100% + volume = int(6 + (vol_db + 28.125) / 28.125 * 94) + volume = max(0, min(100, volume)) if volume == 0 and vol_db > -144: volume = 1 diff --git a/miair/airplay/utils.py b/miair/airplay/utils.py index 74c7be8..25ca58f 100644 --- a/miair/airplay/utils.py +++ b/miair/airplay/utils.py @@ -192,7 +192,7 @@ def get_volume(): vol = interpolate(pct, 0, 100, -30, 0) elif subsys == "Linux": line_pct = subprocess.check_output(["amixer", "get", "Master"]).splitlines()[-1] - m = re.search(b"\[([0-9]+)%\]", line_pct) + m = re.search(rb"\[([0-9]+)%\]", line_pct) if m: pct = int(m.group(1)) if pct < 45: diff --git a/miair/app.py b/miair/app.py index 9445425..4663067 100644 --- a/miair/app.py +++ b/miair/app.py @@ -53,6 +53,37 @@ async def get_all_devices(self) -> list[dict]: log.warning(f"获取设备列表失败: {e}") return [] + async def _periodic_device_check(self): + """每分钟自主检查设备列表，如果为空且启动超过5分钟，则触发重启""" + import time + start_time = time.time() + while True: + await asyncio.sleep(60) + + # 如果没有开启自动重启，不执行此逻辑 + if not self.config.auto_restart: + continue + + # 如果没有配置账号密码，不检查 + if not self.config.account and not self.config.cookie: + continue + + uptime = time.time() - start_time + if uptime < 300: + continue + + try: + devices = await self.get_all_devices() + if not devices: + log.error("定期检查发现设备列表突然为空，判定为故障，触发自动重启以恢复服务...") + from miair.web.api import _restart_process + try: + asyncio.get_running_loop().call_soon(_restart_process) + except RuntimeError: + _restart_process() + except Exception as e: + log.warning(f"定期检查设备列表异常: {e}") + async def start(self): """启动所有服务""" self._setup_logging() @@ -80,16 +111,48 @@ async def start(self): log.info("未选择音箱设备，请打开 Web 管理界面选择设备") log.info(f"请访问 http://{self.config.hostname}:{self.config.web_port} 进行配置") + self._device_check_task = asyncio.create_task(self._periodic_device_check()) + async def _start_dlna_services(self): """启动 DLNA 相关服务 (登录、初始化音箱、SSDP、HTTP)""" try: # 登录小米 await self.auth.login() + # 检查登录状态 + if not self.auth.is_logged_in(): + log.warning("登录失败，无法启动 DLNA 服务") + # 清空渲染器和控制器，避免显示旧设备 + self.renderers.clear() + self._did_to_udn.clear() + if hasattr(self, 'speaker_manager'): + self.speaker_manager.controllers.clear() + return + + # 获取设备列表，确保能正常获取新账号的设备 + device_list = await self.auth.get_device_list() + if not device_list: + log.warning("未获取到设备列表，无法启动 DLNA 服务") + # 清空渲染器和控制器，避免显示旧设备 + self.renderers.clear() + self._did_to_udn.clear() + if hasattr(self, 'speaker_manager'): + self.speaker_manager.controllers.clear() + + # 如果开启了自动重启，则尝试重启 + if self.config.auto_restart: + log.warning("未获取到设备列表，正在尝试自动重启程序...") + from miair.web.api import _restart_process + asyncio.get_running_loop().call_later(5, _restart_process) + return + # 初始化音箱 await self.speaker_manager.init_speakers() if not self.speaker_manager.controllers: log.warning("没有可用的音箱，请检查配置或重新选择设备") + # 清空渲染器和控制器，避免显示旧设备 + self.renderers.clear() + self._did_to_udn.clear() return # 为每个音箱创建 DLNA 渲染器 @@ -101,7 +164,7 @@ async def _start_dlna_services(self): udn = speaker.udn friendly_name = speaker.get_dlna_name() - renderer = DLNARenderer(udn, friendly_name, controller) + renderer = DLNARenderer(udn, friendly_name, controller, self.config.default_volume, config=self.config) self.renderers[udn] = renderer self._did_to_udn[did] = udn @@ -126,6 +189,13 @@ async def _start_dlna_services(self): except Exception as e: log.error(f"启动 DLNA 服务失败: {e}") + # 确保 dlna_running 为 False + self.dlna_running = False + # 清空渲染器和控制器，避免显示旧设备 + self.renderers.clear() + self._did_to_udn.clear() + if hasattr(self, 'speaker_manager'): + self.speaker_manager.controllers.clear() async def _start_airplay_for_speakers(self): """为每个音箱启动独立的 AirPlay 接收服务""" @@ -143,6 +213,9 @@ async def restart_dlna_services(self): """重启 DLNA 服务 (用户通过 Web 修改配置后调用)""" # 先停止现有服务 await self._stop_dlna_services() + # 关闭并重新初始化 auth，确保账号切换生效 + await self.auth.close() + self.auth = AuthManager(self.config) # 重建 speaker manager self.speaker_manager = SpeakerManager(self.config, self.auth) # 启动 @@ -167,6 +240,9 @@ async def stop(self): """停止所有服务""" log.info("MiAir 正在关闭...") + if hasattr(self, '_device_check_task') and self._device_check_task: + self._device_check_task.cancel() + await self._stop_dlna_services() if self.airplay_manager: await self.airplay_manager.stop() diff --git a/miair/auth.py b/miair/auth.py index 16890d9..b8b90d6 100644 --- a/miair/auth.py +++ b/miair/auth.py @@ -80,6 +80,9 @@ async def login(self): self.config.password, token_store=token_store, ) + # 确保 token 不为 None，避免后续操作出错 + if not hasattr(self.account, 'token') or self.account.token is None: + self.account.token = {"deviceId": "miair_device"} # 显式调用 login # 如果使用 cookie 登录，跳过 login 调用，直接标记为已登录 @@ -93,6 +96,9 @@ async def login(self): log.info("小米账号登录成功") except Exception as e: self._logged_in = False + # 确保 token 不为 None，避免后续操作出错 + if not hasattr(self.account, 'token') or self.account.token is None: + self.account.token = {"deviceId": "miair_device"} err_msg = str(e) err_code = self._extract_error_code(err_msg) if err_code == "87001" or "captcha" in err_msg.lower(): @@ -116,6 +122,18 @@ async def login(self): ) else: log.error(f"登录失败: {e}") + + # 如果开启了自动重启，则在严重错误时尝试重启程序 + if self.config.auto_restart: + log.warning("检测到登录失败，正在尝试自动重启程序以恢复服务...") + from miair.web.api import _restart_process + import asyncio + try: + loop = asyncio.get_running_loop() + loop.call_later(5, _restart_process) + except RuntimeError: + # 如果没有正在运行的 loop，则直接重启 + _restart_process() # 无论是否登录成功，都设置 service (方便后续重试) self.mina_service = MiNAService(self.account) @@ -123,7 +141,7 @@ async def login(self): async def ensure_login(self): """确保已登录，未登录则尝试登录""" - if self.mina_service is None: + if self.mina_service is None or not self._logged_in: await self.login() @staticmethod diff --git a/miair/config.py b/miair/config.py index 60a8792..137a896 100644 --- a/miair/config.py +++ b/miair/config.py @@ -1,12 +1,20 @@ from __future__ import annotations +import ipaddress import json +import logging import os +import re +import socket +import subprocess import threading import uuid from dataclasses import asdict, dataclass, field +log = logging.getLogger("miair") + + @dataclass class Speaker: """单个小爱音箱的配置""" @@ -18,11 +26,22 @@ class Speaker: dlna_name: str = "" udn: str = "" use_music_api: bool = False + compatibility_mode: bool | None = None enabled: bool = True # 不支持无损格式的音箱型号列表 _NON_LOSSLESS_HARDWARE = {"L05B", "L05C", "LX06", "L16A"} + def is_compatibility_mode(self) -> bool: + if self.compatibility_mode is not None: + return self.compatibility_mode + # 默认：如果 hardware 在 NEED_USE_PLAY_MUSIC_API 中，则为 False，否则为 True + from miair.const import NEED_USE_PLAY_MUSIC_API + for model in NEED_USE_PLAY_MUSIC_API: + if model in self.hardware: + return False + return True + def get_dlna_name(self) -> str: return self.dlna_name or self.name or f"XiaoAI-{self.did}" @@ -66,8 +85,14 @@ class Config: # 实验性功能：打断后续播 auto_resume_on_interrupt: bool = False resume_delay_seconds: int = 5 + # 默认音量 (1-100) + default_volume: int = 38 + # 实验性功能：跟随设备当前音量 + follow_device_volume: bool = True # 语音控制 enable_voice_control: bool = False + # 自动重启（当登录失败或服务异常时） + auto_restart: bool = False voice_poll_interval: int = 1 speakers: dict = field(default_factory=dict) @@ -80,30 +105,190 @@ def log_file(self) -> str: return os.path.join(self.conf_path, "miair.log") def __post_init__(self): + self.resume_delay_seconds = max(1, min(15, self.resume_delay_seconds)) if not self.account: self.account = os.getenv("MI_USER", "") if not self.password: self.password = os.getenv("MI_PASS", "") if not self.mi_did: self.mi_did = os.getenv("MI_DID", "") - if not self.hostname: - self.hostname = os.getenv("MIAIR_HOSTNAME", "") + env_hostname = os.getenv("MIAIR_HOSTNAME", "").strip() + if env_hostname: + self.hostname = env_hostname if not self.hostname: self.hostname = self._detect_local_ip() @staticmethod def _detect_local_ip() -> str: - """自动检测本机局域网 IP""" - import socket + """自动检测本机局域网 IP，避免多网卡时误选默认 WAN 出口。""" + candidates = Config._collect_local_ipv4_candidates() + usable = [] + seen = set() + for ip, source in candidates: + if ip in seen or not Config._is_usable_local_ipv4(ip): + continue + seen.add(ip) + usable.append((ip, source, Config._score_local_ipv4(ip, source))) + + if usable: + usable.sort(key=lambda item: item[2], reverse=True) + selected = usable[0][0] + summary = ", ".join(f"{ip}({source})" for ip, source, _ in usable[:6]) + log.info(f"自动检测局域网 IP: {selected}; 候选: {summary}") + return selected + + ip = Config._detect_default_route_ip() + if Config._is_usable_local_ipv4(ip): + log.info(f"自动检测局域网 IP fallback: {ip}") + return ip + return "127.0.0.1" + + @staticmethod + def _collect_local_ipv4_candidates() -> list[tuple[str, str]]: + """枚举本机 IPv4 候选地址，返回 (ip, 来源/接口名)。""" + candidates: list[tuple[str, str]] = [] + + # Linux/OpenWrt: 最可靠，可拿到接口名用于降低 Docker/VPN 等虚拟网卡优先级。 + candidates.extend(Config._collect_from_ip_addr()) + + # Windows/macOS fallback: 解析系统网络配置输出。 + candidates.extend(Config._collect_from_ipconfig()) + candidates.extend(Config._collect_from_ifconfig()) + + # 标准库 fallback: 不依赖外部命令，但通常拿不到接口名。 + for host in {socket.gethostname(), socket.getfqdn()}: + if not host: + continue + try: + infos = socket.getaddrinfo(host, None, socket.AF_INET, socket.SOCK_DGRAM) + except OSError: + continue + for info in infos: + candidates.append((info[4][0], f"hostname:{host}")) + + # 最后保留旧逻辑作为 fallback 候选，不能作为唯一的首选依据。 + candidates.append((Config._detect_default_route_ip(), "default-route")) + return candidates + + @staticmethod + def _collect_from_ip_addr() -> list[tuple[str, str]]: + try: + proc = subprocess.run( + ["ip", "-o", "-4", "addr", "show", "scope", "global"], + capture_output=True, + text=True, + timeout=2, + check=False, + ) + except (OSError, subprocess.SubprocessError): + return [] + result = [] + for line in proc.stdout.splitlines(): + parts = line.split() + if len(parts) >= 4: + result.append((parts[3].split("/", 1)[0], parts[1])) + return result + @staticmethod + def _collect_from_ipconfig() -> list[tuple[str, str]]: + try: + proc = subprocess.run( + ["ipconfig"], + capture_output=True, + text=True, + timeout=2, + check=False, + ) + except (OSError, subprocess.SubprocessError): + return [] + result = [] + adapter = "ipconfig" + for raw_line in proc.stdout.splitlines(): + line = raw_line.strip() + if line.endswith(":") and not re.search(r"\d+\.\d+\.\d+\.\d+", line): + adapter = line[:-1] + if "ipv4" not in line.lower(): + continue + match = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", line) + if match: + result.append((match.group(1), adapter)) + return result + + @staticmethod + def _collect_from_ifconfig() -> list[tuple[str, str]]: + try: + proc = subprocess.run( + ["ifconfig"], + capture_output=True, + text=True, + timeout=2, + check=False, + ) + except (OSError, subprocess.SubprocessError): + return [] + result = [] + iface = "ifconfig" + for raw_line in proc.stdout.splitlines(): + if raw_line and not raw_line[0].isspace(): + iface = raw_line.split(":", 1)[0].strip() + match = re.search(r"\binet\s+(\d{1,3}(?:\.\d{1,3}){3})", raw_line) + if match: + result.append((match.group(1), iface)) + return result + + @staticmethod + def _detect_default_route_ip() -> str: + s = None try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) - ip = s.getsockname()[0] - s.close() - return ip - except Exception: + return s.getsockname()[0] + except OSError: return "127.0.0.1" + finally: + if s: + s.close() + + @staticmethod + def _is_usable_local_ipv4(ip: str) -> bool: + try: + addr = ipaddress.ip_address(ip) + except ValueError: + return False + return not ( + addr.is_loopback + or addr.is_link_local + or addr.is_multicast + or addr.is_unspecified + or addr.is_reserved + ) + + @staticmethod + def _score_local_ipv4(ip: str, source: str = "") -> int: + addr = ipaddress.ip_address(ip) + source_lower = source.lower() + score = 0 + if addr.is_private: + score += 100 + if ip.startswith("192.168."): + score += 50 + elif ip.startswith("10."): + score += 40 + elif ipaddress.ip_address("172.16.0.0") <= addr <= ipaddress.ip_address("172.31.255.255"): + score += 30 + + virtual_markers = ( + "docker", "veth", "br-", "vmware", "virtualbox", "vbox", "hyper-v", + "wsl", "tailscale", "zerotier", "vpn", "tun", "tap", "wg", "ppp", + "utun", "awdl", "llw", "anpi", + ) + if any(marker in source_lower for marker in virtual_markers): + score -= 120 + if ip.startswith("172.17.") or ip.startswith("172.18."): + score -= 30 + if source_lower == "default-route": + score -= 20 + return score @property def mi_token_home(self) -> str: diff --git a/miair/const.py b/miair/const.py index fafe202..ed18477 100644 --- a/miair/const.py +++ b/miair/const.py @@ -1,5 +1,8 @@ """MiAir 常量定义""" +VERSION = "0.8.5-alpha" + + # SSDP 相关 SSDP_ADDR = "239.255.255.250" SSDP_PORT = 1900 @@ -40,13 +43,16 @@ "X8F", "X4B", "LX05", + "LX05A", "OH2", "OH2P", "X6A", + "L15A", + "L07A", ] # 默认 audio_id (用于 play_by_music_url) -DEFAULT_AUDIO_ID = "1582971365183456177" +DEFAULT_AUDIO_ID = "448161862632079419" # 支持的协议信息 (ConnectionManager GetProtocolInfo) - 仅音频 SUPPORTED_PROTOCOLS = ( diff --git a/miair/dlna/device_server.py b/miair/dlna/device_server.py index 91593d0..e9fc5f9 100644 --- a/miair/dlna/device_server.py +++ b/miair/dlna/device_server.py @@ -67,6 +67,8 @@ def __init__(self, hostname: str, dlna_port: int, config: "Config | None" = None self._resume_tasks: dict[str, asyncio.Task] = {} # udn -> resume task # 追踪活跃的代理任务，用于强制中止 self._active_proxy_tasks: dict[str, set[asyncio.Task]] = {} # udn -> set of tasks + # 暂停中屏蔽集合：暂停时加入，播放/停止时移除，阻止音箱重连拉取音频 + self._paused_proxy_udns: set[str] = set() # 周期性缓冲清理任务 self._buffer_cleanup_task: asyncio.Task | None = None # 内存上限: 缓冲总大小不超过 200MB @@ -131,6 +133,7 @@ def register_renderer(self, renderer: DLNARenderer): renderer.seek_url_func = self.create_seek_url renderer.pre_buffer_func = self.start_buffering renderer.abort_proxy_func = self.abort_proxy_for_renderer + renderer.resume_proxy_func = self.resume_proxy_for_renderer # ---- 音频缓冲/代理系统 ---- @@ -677,7 +680,8 @@ async def _periodic_buffer_cleanup(self): pass def abort_proxy_for_renderer(self, udn: str): - """立即中止指定渲染器的所有活跃代理连接""" + """立即中止指定渲染器的所有活跃代理连接，并屏蔽后续重连请求""" + self._paused_proxy_udns.add(udn) tasks = self._active_proxy_tasks.get(udn) if tasks: log.info(f"[{udn}] 正在中止 {len(tasks)} 个活跃的媒体代理连接...") @@ -686,6 +690,10 @@ def abort_proxy_for_renderer(self, udn: str): task.cancel() tasks.clear() + def resume_proxy_for_renderer(self, udn: str): + """恢复渲染器的代理访问（播放/停止时调用）""" + self._paused_proxy_udns.discard(udn) + async def _handle_media_proxy(self, request: web.Request) -> web.StreamResponse: """媒体代理处理器 - 从内存缓冲提供音频，支持 Range/Seek""" token = request.match_info.get("token", "") @@ -695,6 +703,12 @@ async def _handle_media_proxy(self, request: web.Request) -> web.StreamResponse: return web.Response(status=404, text="Not Found") buffer_id, base_offset, udn = entry + + # 暂停中的渲染器，拒绝代理请求（防止音箱重连拉取音频） + if udn and udn in self._paused_proxy_udns: + log.info(f"[{udn}] 暂停中，拒绝代理请求") + return web.Response(status=403, text="Paused") + buf = self._media_buffers.get(buffer_id) if not buf: log.warning(f"代理请求缓冲不存在: {buffer_id}") @@ -1069,6 +1083,9 @@ async def _handle_control(self, request: web.Request) -> web.Response: if not renderer: return web.Response(status=404, text="Device not found") + # 记录最后活动时间 + renderer._last_control_time = time.time() + # 解析 SOAPAction soap_action = request.headers.get("SOAPAction", "") if not soap_action: @@ -1099,6 +1116,11 @@ async def _handle_subscribe(self, request: web.Request) -> web.Response: if not event_manager: return web.Response(status=404) + # 记录活动时间（续订或新订阅都说明控制端仍在连接） + renderer = self.renderers.get(udn) + if renderer: + renderer._last_control_time = time.time() + # 检查是否是续订 (有 SID header) sid = request.headers.get("SID", "") if sid: @@ -1127,6 +1149,9 @@ async def _handle_subscribe(self, request: web.Request) -> web.Response: # 发送初始事件 (参照 MaCast: 在后台发送完整状态，不阻塞 SUBSCRIBE 响应) renderer = self.renderers.get(udn) if renderer: + # 客户端建立新订阅，视为新会话或重连，重置音量初始化标志 + # 以便在接下来的播放中重新应用默认音量 + renderer._volume_initialized = False task = asyncio.get_running_loop().create_task( self._send_initial_event(event_manager, sid, renderer, service) ) @@ -1186,7 +1211,44 @@ async def _poll_speaker_states(self): while True: await asyncio.sleep(5) for udn, renderer in self.renderers.items(): - if not renderer.speaker or not renderer.current_uri: + if not renderer.speaker: + continue + + # 检测控制端断开：渲染器处于终端状态（PAUSED/STOPPED）且有媒体URI + # 1. PAUSED 振荡：音箱 STOP 但被强制保持 PAUSED 超过 30 秒 + # 语音打断场景（_user_stopped=False）：直接重置 + # 用户手动暂停（_user_stopped=True）：需确认无订阅者后重置 + # 2. STOPPED 持久：有 URI 但无控制活动超过 60 秒且无订阅者 + idle = False + if ( + renderer.current_uri + and renderer.transport_state == TRANSPORT_STATE_PAUSED + and renderer._stuck_paused_since > 0 + and (time.time() - renderer._stuck_paused_since) > 30 + ): + if not renderer._user_stopped: + idle = True + elif ( + renderer.event_manager + and not renderer.event_manager.has_subscribers() + ): + idle = True + elif ( + renderer.current_uri + and renderer.transport_state == TRANSPORT_STATE_STOPPED + and renderer._last_control_time > 0 + and (time.time() - renderer._last_control_time) > 60 + and renderer.event_manager + and not renderer.event_manager.has_subscribers() + ): + idle = True + + if idle: + await renderer.reset_to_idle() + renderer._stuck_paused_since = 0.0 + continue + + if not renderer.current_uri: continue try: status = await asyncio.wait_for( @@ -1264,7 +1326,15 @@ def _handle_state_transition( renderer._play_start_time = 0.0 renderer.transport_state = TRANSPORT_STATE_PAUSED - if self.config and self.config.auto_resume_on_interrupt: + if renderer._stuck_paused_since == 0.0: + renderer._stuck_paused_since = time.time() + + if ( + self.config + and self.config.auto_resume_on_interrupt + and not renderer._user_stopped + and (time.time() - renderer._stuck_paused_since) < 15 + ): if udn in self._resume_tasks: self._resume_tasks[udn].cancel() delay = self.config.resume_delay_seconds diff --git a/miair/dlna/eventing.py b/miair/dlna/eventing.py index 4cf38ec..a1dc04c 100644 --- a/miair/dlna/eventing.py +++ b/miair/dlna/eventing.py @@ -78,6 +78,10 @@ def unsubscribe(self, sid: str) -> bool: return True return False + def has_subscribers(self) -> bool: + """检查是否有未过期的订阅者""" + return any(not sub.expired for sub in self._subscriptions.values()) + async def notify_all(self, event_xml: str): """向所有活跃订阅者发送事件通知 (fire-and-forget，不等待慢订阅者)""" expired_sids = [] diff --git a/miair/dlna/renderer.py b/miair/dlna/renderer.py index 774d504..d55abe6 100644 --- a/miair/dlna/renderer.py +++ b/miair/dlna/renderer.py @@ -25,10 +25,11 @@ class DLNARenderer: """每个音箱对应一个 DLNA 渲染器实例，管理传输状态""" - def __init__(self, udn: str, friendly_name: str, speaker: SpeakerController): + def __init__(self, udn: str, friendly_name: str, speaker: SpeakerController, default_volume: int = 50, config=None): self.udn = udn self.friendly_name = friendly_name self.speaker = speaker + self.config = config # 保存did以便快速访问 self.did = speaker.did self._lock = asyncio.Lock() @@ -41,9 +42,9 @@ def __init__(self, udn: str, friendly_name: str, speaker: SpeakerController): self.play_speed = "1" # 音量/静音 - self.volume = 50 + self.volume = default_volume self.mute = False - self._pre_mute_volume = 50 + self._pre_mute_volume = default_volume # 事件管理器 (由 DeviceServer 注入) self.event_manager = None @@ -55,6 +56,8 @@ def __init__(self, udn: str, friendly_name: str, speaker: SpeakerController): self.pre_buffer_func = None # 代理中止回调 (由 DeviceServer 注入，替代双向引用) self.abort_proxy_func: Callable[[str], None] | None = None + # 代理恢复回调 (由 DeviceServer 注入，暂停后恢复代理访问) + self.resume_proxy_func: Callable[[str], None] | None = None # 位置追踪 (基于定时器的近似值) self._play_start_time: float = 0.0 @@ -69,6 +72,15 @@ def __init__(self, udn: str, friendly_name: str, speaker: SpeakerController): self._play_check_task: asyncio.Task | None = None # play() 后的宽限期，在此时间之前轮询不覆盖 PLAYING 状态 self._play_grace_until: float = 0.0 + # 用户主动暂停/停止标志：True 表示用户手动暂停/停止，不应自动续播 + self._user_stopped: bool = False + # 最后一次接收 DLNA 控制命令的时间（用于检测控制端断开） + self._last_control_time: float = 0.0 + # 检测 "音箱已停而渲染器被强制保持PAUSED" 的起始时间 + # 当此模式持续超过阈值时，判定控制端已断开并重置为空闲 + self._stuck_paused_since: float = 0.0 + # 标记是否已应用过默认音量（避免切歌时重复重置） + self._volume_initialized: bool = False # 视频格式扩展名列表 VIDEO_EXTENSIONS = {'.mp4', '.mov', '.avi', '.mkv', '.flv', '.wmv', '.m4v', '.3gp', '.ts', '.mts', '.m2ts'} @@ -144,6 +156,10 @@ async def _check_play_status(self): async def play(self) -> bool: """开始播放 (DLNA Play)""" + # 恢复代理访问（暂停时会屏蔽，此处解除以便音箱重新拉取音频） + if self.resume_proxy_func: + self.resume_proxy_func(self.udn) + needs_transcode = self._needs_transcode() play_url = None @@ -161,6 +177,8 @@ async def play(self) -> bool: self._play_check_task = None self.transport_state = TRANSPORT_STATE_TRANSITIONING + self._user_stopped = False + self._stuck_paused_since = 0.0 log.info(f"[{self.friendly_name}] Play: {self.current_uri}") # 计算当前播放位置（用于从暂停位置继续播放） @@ -207,6 +225,7 @@ async def play(self) -> bool: self._play_grace_until = time.time() + 8.0 log.info(f"[{self.friendly_name}] 播放成功") self._play_check_task = asyncio.create_task(self._check_play_status()) + asyncio.create_task(self._apply_default_volume()) else: self.transport_state = TRANSPORT_STATE_STOPPED self._play_grace_until = 0.0 @@ -214,8 +233,37 @@ async def play(self) -> bool: await self.notify_state_change() return success + async def _apply_default_volume(self): + """播放开始后应用默认音量（仅首次连接时应用，切歌不重置）""" + try: + if getattr(self.config, 'follow_device_volume', False): + return + + # 仅首次播放时应用默认音量，避免切歌时覆盖用户调整的音量 + if self._volume_initialized: + return + + default_vol = getattr(self.config, 'default_volume', 50) + if default_vol <= 0: + return + + await asyncio.sleep(0.5) + if self.speaker: + await self.speaker.set_volume(default_vol) + self.volume = default_vol + self._volume_initialized = True + log.info(f"[{self.friendly_name}] 已应用默认音量: {default_vol}%") + await self.notify_state_change() + except Exception as e: + log.error(f"[{self.friendly_name}] 应用默认音量失败: {e}") + async def pause(self) -> bool: """暂停播放 (DLNA Pause)""" + # 中止媒体代理连接，让音箱失去音频源后停止播放 + # 解决部分型号（如 L05C）通过 MiNA API 发送 stop/pause 无效的问题 + if self.abort_proxy_func: + self.abort_proxy_func(self.udn) + async with self._lock: if not self.speaker: self.transport_state = TRANSPORT_STATE_PAUSED @@ -227,6 +275,9 @@ async def pause(self) -> bool: self._accumulated_time += time.time() - self._play_start_time self._play_start_time = 0.0 self.transport_state = TRANSPORT_STATE_PAUSED + self._user_stopped = True + # 启动 idle 超时计时器，以便控制端断开后可以正常重置 + self._stuck_paused_since = time.time() # 取消播放状态检查任务 if self._play_check_task: self._play_check_task.cancel() @@ -240,7 +291,11 @@ async def stop(self) -> bool: # 立即中止所有活跃的媒体代理连接，防止音箱在断开后播放缓存残余 if self.abort_proxy_func: self.abort_proxy_func(self.udn) - + + # 停止后解除暂停屏蔽（暂停时会加入屏蔽集合） + if self.resume_proxy_func: + self.resume_proxy_func(self.udn) + async with self._lock: if not self.speaker: self.transport_state = TRANSPORT_STATE_STOPPED @@ -250,6 +305,7 @@ async def stop(self) -> bool: self.transport_state = TRANSPORT_STATE_STOPPED self._accumulated_time = 0.0 self._play_start_time = 0.0 + self._user_stopped = True # 取消播放状态检查任务 if self._play_check_task: self._play_check_task.cancel() @@ -258,6 +314,29 @@ async def stop(self) -> bool: await self.notify_state_change() return success + async def reset_to_idle(self): + """重置为空闲状态（控制端断开连接后调用）""" + async with self._lock: + if self.transport_state == TRANSPORT_STATE_NO_MEDIA: + return + old_state = self.transport_state + self.transport_state = TRANSPORT_STATE_NO_MEDIA + self.current_uri = "" + self.current_uri_metadata = "" + self.next_uri = "" + self.next_uri_metadata = "" + self._accumulated_time = 0.0 + self._play_start_time = 0.0 + self._track_duration = 0.0 + self._user_stopped = False + self._stuck_paused_since = 0.0 + self._volume_initialized = False + if self._play_check_task: + self._play_check_task.cancel() + self._play_check_task = None + log.info(f"[{self.friendly_name}] 控制端断开，重置为空闲 ({old_state})") + await self.notify_state_change() + async def seek(self, unit: str, target: str) -> bool: """Seek - 生成格式正确的 seeked 音频并重新播放 diff --git a/miair/dlna/ssdp.py b/miair/dlna/ssdp.py index a1a0a91..9b70cbc 100644 --- a/miair/dlna/ssdp.py +++ b/miair/dlna/ssdp.py @@ -1,5 +1,7 @@ """SSDP 组播发现服务""" +from __future__ import annotations + import asyncio import logging import random @@ -113,10 +115,17 @@ async def start(self): self._sock.bind(("", SSDP_PORT)) # 加入组播组 + multicast_interface = self._get_multicast_interface() + if multicast_interface: + self._sock.setsockopt( + socket.IPPROTO_IP, + socket.IP_MULTICAST_IF, + socket.inet_aton(multicast_interface), + ) mreq = struct.pack( "4s4s", socket.inet_aton(SSDP_ADDR), - socket.inet_aton("0.0.0.0"), # 使用INADDR_ANY + socket.inet_aton(multicast_interface or "0.0.0.0"), ) self._sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) self._sock.setblocking(False) @@ -132,7 +141,20 @@ async def start(self): # 启动定期 alive 任务 self._alive_task = asyncio.create_task(self._periodic_alive()) - log.info(f"SSDP 服务已启动 (监听 {SSDP_ADDR}:{SSDP_PORT})") + if multicast_interface: + log.info(f"SSDP 服务已启动 (监听 {SSDP_ADDR}:{SSDP_PORT}, 接口 {multicast_interface})") + else: + log.info(f"SSDP 服务已启动 (监听 {SSDP_ADDR}:{SSDP_PORT}, 接口自动选择)") + + def _get_multicast_interface(self) -> str | None: + """返回 SSDP 组播使用的本地接口地址。""" + try: + socket.inet_aton(self.hostname) + except OSError: + return None + if self.hostname in ("0.0.0.0", "127.0.0.1"): + return None + return self.hostname async def stop(self): """停止 SSDP 服务""" diff --git a/miair/dlna/templates.py b/miair/dlna/templates.py index 9b8bf8b..bbaf80b 100644 --- a/miair/dlna/templates.py +++ b/miair/dlna/templates.py @@ -24,7 +24,7 @@ def device_description_xml(udn: str, friendly_name: str, base_url: str) -> str: {DEVICE_TYPE} {escape(friendly_name)} MiAir - https://github.com/miair + https://github.com/KiriChen-Wind/MiAir MiAir - Xiaomi Speaker DLNA Audio Renderer MiAir Speaker 1.0 @@ -32,6 +32,7 @@ def device_description_xml(udn: str, friendly_name: str, base_url: str) -> str: uuid:{udn} DMR-1.50 audio-only + QPlay:2 {AVTRANSPORT_URN} diff --git a/miair/speaker.py b/miair/speaker.py index 19758ac..12c4b22 100644 --- a/miair/speaker.py +++ b/miair/speaker.py @@ -1,5 +1,6 @@ """小爱音箱控制模块""" +import asyncio import json import logging @@ -13,11 +14,28 @@ class SpeakerController: """单个小爱音箱的控制接口""" + # 连续登录失败计数（所有实例共享，因为登录状态是全局的） + _consecutive_login_failures: int = 0 + _LOGIN_FAILURE_RESTART_THRESHOLD = 6 # 连续失败 6 次后触发重启 + def __init__(self, speaker: Speaker, auth: AuthManager): self.speaker = speaker self.auth = auth self._last_volume: int = 50 # 用于 unmute 恢复 + @classmethod + def _check_and_trigger_restart(cls): + """检查连续登录失败次数，达到阈值时触发进程重启""" + if cls._consecutive_login_failures >= cls._LOGIN_FAILURE_RESTART_THRESHOLD: + log.error( + f"连续 {cls._consecutive_login_failures} 次登录失败，正在重启程序以恢复服务..." + ) + from miair.web.api import _restart_process + try: + asyncio.get_running_loop().call_soon(_restart_process) + except RuntimeError: + _restart_process() + @property def device_id(self) -> str: return self.speaker.device_id @@ -27,10 +45,9 @@ def did(self) -> str: return self.speaker.did def _should_use_music_api(self) -> bool: - return ( - self.speaker.use_music_api - or self.speaker.hardware in NEED_USE_PLAY_MUSIC_API - ) + if self.speaker.is_compatibility_mode(): + return False + return True async def play_url(self, url: str) -> bool: """让音箱播放指定 URL""" @@ -47,17 +64,59 @@ async def play_url(self, url: str) -> bool: return ret is not None except Exception as e: log.error(f"play_url 失败: {e}") + # 检查是否是登录失败的错误 + if "Login failed" in str(e) or "登录验证失败" in str(e): + log.info("检测到登录失败，尝试重新登录...") + # 重置登录状态并重新登录 + self.auth._logged_in = False + await self.auth.login() + # 重新尝试播放 + try: + await self.auth.ensure_login() + if self._should_use_music_api(): + ret = await self.auth.mina_service.play_by_music_url( + self.device_id, url, audio_id=DEFAULT_AUDIO_ID + ) + else: + ret = await self.auth.mina_service.play_by_url(self.device_id, url) + return ret is not None + except Exception as e2: + log.error(f"重新登录后 play_url 仍然失败: {e2}") + return False return False async def pause(self) -> bool: """暂停播放""" try: await self.auth.ensure_login() - ret = await self.auth.mina_service.player_pause(self.device_id) - log.info(f"player_pause device_id={self.device_id} ret={ret}") + if self._should_use_music_api(): + # 某些使用 play_by_music_url 的设备，调用 pause 后 API 状态不会 + # 正确更新为 paused (status=2)，需改用 stop 来实现暂停语义 + ret = await self.auth.mina_service.player_stop(self.device_id) + log.info(f"player_stop(as pause) device_id={self.device_id} ret={ret}") + else: + ret = await self.auth.mina_service.player_pause(self.device_id) + log.info(f"player_pause device_id={self.device_id} ret={ret}") return True except Exception as e: log.error(f"pause 失败: {e}") + # 检查是否是登录失败的错误 + if "Login failed" in str(e) or "登录验证失败" in str(e): + log.info("检测到登录失败，尝试重新登录...") + # 重置登录状态并重新登录 + self.auth._logged_in = False + await self.auth.login() + # 重新尝试暂停 + try: + await self.auth.ensure_login() + if self._should_use_music_api(): + await self.auth.mina_service.player_stop(self.device_id) + else: + await self.auth.mina_service.player_pause(self.device_id) + return True + except Exception as e2: + log.error(f"重新登录后 pause 仍然失败: {e2}") + return False return False async def stop(self) -> bool: @@ -67,11 +126,28 @@ async def stop(self) -> bool: # 某些型号的小爱音箱在 stop 后仍会残留缓存， # 连续调用 stop + pause 可以更彻底地清空播放状态。 ret = await self.auth.mina_service.player_stop(self.device_id) - await self.pause() + await self.pause() log.info(f"player_stop device_id={self.device_id} ret={ret}") return True except Exception as e: log.error(f"stop 失败: {e}") + # 检查是否是登录失败的错误 + if "Login failed" in str(e) or "登录验证失败" in str(e): + log.info("检测到登录失败，尝试重新登录...") + # 重置登录状态并重新登录 + self.auth._logged_in = False + await self.auth.login() + # 重新尝试停止 + try: + await self.auth.ensure_login() + await self.auth.mina_service.player_stop(self.device_id) + await self.pause() + return True + except Exception as e2: + log.error(f"重新登录后 stop 仍然失败: {e2}") + SpeakerController._consecutive_login_failures += 1 + SpeakerController._check_and_trigger_restart() + return False return False async def set_volume(self, volume: int) -> bool: @@ -86,6 +162,22 @@ async def set_volume(self, volume: int) -> bool: return True except Exception as e: log.error(f"set_volume 失败: {e}") + # 检查是否是登录失败的错误 + if "Login failed" in str(e) or "登录验证失败" in str(e): + log.info("检测到登录失败，尝试重新登录...") + # 重置登录状态并重新登录 + self.auth._logged_in = False + await self.auth.login() + # 重新尝试设置音量 + try: + await self.auth.ensure_login() + await self.auth.mina_service.player_set_volume(self.device_id, volume) + if volume > 0: + self._last_volume = volume + return True + except Exception as e2: + log.error(f"重新登录后 set_volume 仍然失败: {e2}") + return False return False async def get_volume(self) -> int: @@ -100,6 +192,24 @@ async def get_volume(self) -> int: return volume except Exception as e: log.error(f"get_volume 失败: {e}") + # 检查是否是登录失败的错误 + if "Login failed" in str(e) or "登录验证失败" in str(e): + log.info("检测到登录失败，尝试重新登录...") + # 重置登录状态并重新登录 + self.auth._logged_in = False + await self.auth.login() + # 重新尝试获取音量 + try: + await self.auth.ensure_login() + status = await self.auth.mina_service.player_get_status(self.device_id) + info = json.loads(status.get("data", {}).get("info", "{}")) + volume = int(info.get("volume", 0)) + if volume > 0: + self._last_volume = volume + return volume + except Exception as e2: + log.error(f"重新登录后 get_volume 仍然失败: {e2}") + return self._last_volume return self._last_volume async def get_status(self) -> dict: @@ -127,11 +237,42 @@ async def get_status(self) -> dict: raise Exception(f"Mina API response missing 'info': {playing_info}") info = json.loads(info_str) + # 获取成功，重置连续登录失败计数 + SpeakerController._consecutive_login_failures = 0 return { "status": info.get("status", 0), "volume": int(info.get("volume", 0)), } except Exception as e: + # 检查是否是登录失败的错误 + if "Login failed" in str(e) or "登录验证失败" in str(e): + log.info("检测到登录失败，尝试重新登录...") + # 重置登录状态并重新登录 + self.auth._logged_in = False + await self.auth.login() + # 重新尝试获取状态 + try: + await self.auth.ensure_login() + playing_info = await self.auth.mina_service.player_get_status( + self.device_id + ) + if playing_info.get("code") != 0: + raise Exception(f"Mina API Error: {playing_info}") + data = playing_info.get("data", {}) + info_str = data.get("info") + if not info_str: + raise Exception(f"Mina API response missing 'info': {playing_info}") + info = json.loads(info_str) + # 重试成功，重置计数 + SpeakerController._consecutive_login_failures = 0 + return { + "status": info.get("status", 0), + "volume": int(info.get("volume", 0)), + } + except Exception as e2: + log.error(f"重新登录后 get_status 仍然失败: {e2}") + SpeakerController._consecutive_login_failures += 1 + SpeakerController._check_and_trigger_restart() # 向上抛出异常，让调用者（如 DeviceServer 的轮询任务）捕获并忽略本次轮询 raise Exception(f"get_status 失败: {e}") diff --git a/miair/web/api.py b/miair/web/api.py index 58beb80..5229de6 100644 --- a/miair/web/api.py +++ b/miair/web/api.py @@ -4,25 +4,192 @@ import logging import os import sys +import tarfile +import tempfile +import urllib.request +import urllib.error from aiohttp import web import asyncio from miair.config import Config +from miair.const import VERSION + log = logging.getLogger("miair") +# passToken 在返回给前端时使用的完整脱敏占位符（不是真实凭据） +MASKED_TOKEN = "********" + + +def _mask_value(key: str, value: str) -> str: + """按字段生成脱敏后的展示值。 + + - passToken: 完整敏感凭据，全部替换为占位符； + - userId: 仅为账号标识，保留最后 3 位明文，其余用 * 覆盖（如 *****238）， + 便于用户确认当前账号又不暴露完整 ID。 + 其它字段保持原样。 + """ + if not value: + return value + if key == "passToken": + return MASKED_TOKEN + if key == "userId": + if len(value) <= 3: + return MASKED_TOKEN + return MASKED_TOKEN + value[-3:] + return value + + +def _mask_cookie(cookie: str) -> str: + """对通过 /api/setting 返回给前端的 cookie 进行脱敏，隐藏 passToken 与 userId 的敏感部分。""" + if not cookie: + return cookie + parts = [] + for item in cookie.split(";"): + stripped = item.strip() + if not stripped: + continue + if "=" in stripped: + key, value = stripped.split("=", 1) + key = key.strip() + parts.append(f"{key}={_mask_value(key, value.strip())}") + continue + parts.append(stripped) + return "; ".join(parts) + + +def _unmask_cookie(new_cookie: str, current_cookie: str) -> str: + """将前端回写的 cookie 还原为真实值。 + + 脱敏值中一定含有 `*`（passToken/userId 的真实值不含 `*`）。若某字段回写值仍带 + `*`（用户未修改），则用当前已存储的真实值替换，避免脱敏值被写坏凭据；用户填入 + 的新值不含 `*`，按原样保存。 + """ + if not new_cookie or MASKED_TOKEN not in new_cookie: + return new_cookie + + # 解析当前存储的真实值 + current = {} + for item in (current_cookie or "").split(";"): + item = item.strip() + if "=" in item: + k, v = item.split("=", 1) + current[k.strip()] = v.strip() + + parts = [] + for item in new_cookie.split(";"): + stripped = item.strip() + if not stripped: + continue + if "=" in stripped: + key, value = stripped.split("=", 1) + key = key.strip() + value = value.strip() + if MASKED_TOKEN in value and current.get(key): + parts.append(f"{key}={current[key]}") + continue + parts.append(f"{key}={value}") + continue + parts.append(stripped) + return "; ".join(parts) + + +def _mask_devices(device_list, required_fields=['miotDID','hardware','name']): + """按白名单裁剪设备信息，仅保留 required_fields 指定的字段。 + + Args: + device_list: 单个设备 dict，或设备 dict 组成的列表。 + required_fields: 需要保留的字段名列表，支持用点号表示嵌套路径 + （如 "capabilities.multiroom_music"）。 + + Returns: + 仅含指定字段、并保持原嵌套结构的设备。设备中不存在的字段会被跳过。 + 输入为列表时返回列表，输入为单个 dict 时返回单个 dict。 + """ + single = not isinstance(device_list, list) + devices = [device_list] if single else device_list + + _MISSING = object() + result = [] + for device in devices: + masked = {} + for field in required_fields: + keys = field.split(".") + + # 沿路径逐级取值，任一级缺失或非 dict 则跳过该字段 + value = device + for k in keys: + if isinstance(value, dict) and k in value: + value = value[k] + else: + value = _MISSING + break + if value is _MISSING: + continue + + # 沿路径逐级写入，重建嵌套结构 + target = masked + for k in keys[:-1]: + nested = target.get(k) + if not isinstance(nested, dict): + nested = {} + target[k] = nested + target = nested + target[keys[-1]] = value + + result.append(masked) + + return result[0] if single else result + + +def _is_docker(): + """检测是否在 Docker 容器中运行""" + # 1. 环境变量显式指定（最可靠） + if os.environ.get("MIAIR_DOCKER"): + return True + # 2. Docker 会在容器根目录创建 .dockerenv 文件 + if os.path.exists("/.dockerenv"): + return True + # 3. 检查 cgroup（兼容 cgroup v1 和 v2） + try: + with open('/proc/1/cgroup', 'r') as f: + content = f.read() + return any(k in content for k in ('docker', 'containerd', 'kubepods')) + except Exception: + pass + # 4. 检查 /proc/self/mountinfo 中的 overlay/docker 挂载 + try: + with open('/proc/self/mountinfo', 'r') as f: + content = f.read() + return 'docker' in content or '/docker/' in content + except Exception: + pass + return False + + +def _get_app_dir(): + """获取应用根目录（miair 包的上级目录）""" + return os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + + def _restart_process(): """重启当前 Python 进程""" log.info(f"重启进程: {sys.executable} {sys.argv}") - if sys.platform == "win32": + + # 检测是否在 Docker 容器中 + if _is_docker(): + # Docker 环境下，直接退出进程 + # Docker 容器已设置 restart=unless-stopped，会自动重启 + log.info("在 Docker 环境中，退出进程，Docker 会自动重启容器") + # 使用 exit code 0，unless-stopped 策略下任何退出都会重启 + os._exit(0) + elif sys.platform == "win32": # Windows 上 os.execv 行为不同，使用 subprocess 重启 import subprocess subprocess.Popen([sys.executable] + sys.argv) - # 退出当前进程 - import os os._exit(0) else: os.execv(sys.executable, [sys.executable] + sys.argv) @@ -46,6 +213,7 @@ async def handle_get_setting(request): need_device_list = request.query.get("need_device_list", "false") == "true" data = { + "version": VERSION, "hostname": config.hostname, "dlna_port": config.dlna_port, "web_port": config.web_port, @@ -53,12 +221,15 @@ async def handle_get_setting(request): "auto_play_on_set_uri": config.auto_play_on_set_uri, "mi_did": config.mi_did, "has_account": bool(config.account or config.cookie), - "cookie": config.cookie, + "cookie": _mask_cookie(config.cookie), "dlna_running": app_instance.dlna_running, "renderers_count": len(app_instance.renderers), # 实验性功能 "auto_resume_on_interrupt": config.auto_resume_on_interrupt, "resume_delay_seconds": config.resume_delay_seconds, + "default_volume": config.default_volume, + "follow_device_volume": config.follow_device_volume, + "auto_restart": config.auto_restart, } # 返回已配置的 speakers 信息 @@ -71,12 +242,16 @@ async def handle_get_setting(request): "dlna_name": speaker.get_dlna_name(), "hardware": speaker.hardware, "enabled": speaker.enabled, + "compatibility_mode": speaker.is_compatibility_mode(), } data["speakers"] = speakers_info + from miair.const import NEED_USE_PLAY_MUSIC_API + data["need_use_play_music_api"] = NEED_USE_PLAY_MUSIC_API + if need_device_list: device_list = await app_instance.get_all_devices() - data["device_list"] = device_list + data["device_list"] = _mask_devices(device_list) return web.json_response(data) @@ -90,7 +265,8 @@ async def handle_save_setting(request): if "password" in data: config.password = data["password"] if "cookie" in data: - config.cookie = data["cookie"] + # 若前端回写的是脱敏占位符（未修改 passToken等），还原为已存储的真实值 + config.cookie = _unmask_cookie(data["cookie"], config.cookie) # 更新设备选择 if "mi_did" in data: @@ -100,18 +276,32 @@ async def handle_save_setting(request): if "auto_play_on_set_uri" in data: config.auto_play_on_set_uri = data["auto_play_on_set_uri"] + # 更新端口配置 + if "dlna_port" in data: + config.dlna_port = data["dlna_port"] + if "web_port" in data: + config.web_port = data["web_port"] + # 更新实验性功能配置 if "auto_resume_on_interrupt" in data: config.auto_resume_on_interrupt = data["auto_resume_on_interrupt"] if "resume_delay_seconds" in data: - config.resume_delay_seconds = data["resume_delay_seconds"] - - # 更新 speaker 名称 + config.resume_delay_seconds = max(1, min(15, int(data["resume_delay_seconds"]))) + if "default_volume" in data: + config.default_volume = max(1, min(100, int(data["default_volume"]))) + if "follow_device_volume" in data: + config.follow_device_volume = data["follow_device_volume"] + if "auto_restart" in data: + config.auto_restart = data["auto_restart"] + + # 更新 speaker 名称和兼容模式 if "speakers" in data: for did, speaker_data in data["speakers"].items(): speaker = config.get_speaker(did) if "dlna_name" in speaker_data: speaker.dlna_name = speaker_data["dlna_name"] + if "compatibility_mode" in speaker_data: + speaker.compatibility_mode = speaker_data["compatibility_mode"] config.save() @@ -139,7 +329,7 @@ async def handle_get_devices(request): "devices": [], "error": "登录失败，请检查账号密码或尝试使用 Cookie 登录" }) - return web.json_response({"devices": devices}) + return web.json_response({"devices": _mask_devices(devices)}) except Exception as e: return web.json_response( {"error": f"获取设备列表失败: {e}"}, status=500 @@ -203,7 +393,7 @@ async def handle_rename_speaker(request): async def handle_status(request): """系统状态""" return web.json_response({ - "version": "0.1.0", + "version": VERSION, "dlna_running": app_instance.dlna_running, "renderers_count": len(app_instance.renderers), "hostname": config.hostname, @@ -211,6 +401,85 @@ async def handle_status(request): "web_port": config.web_port, }) + async def handle_execute_update(request): + """执行一键更新：从 GitHub 下载最新代码覆盖后重启""" + app_dir = _get_app_dir() + in_docker = _is_docker() + url = "https://github.com/KiriChen-Wind/MiAir/archive/refs/heads/main.tar.gz" + + log.info(f"开始一键更新 (目录: {app_dir}, Docker: {in_docker})") + + tmp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".tar.gz") + try: + # 下载最新代码压缩包 + log.info(f"正在下载更新: {url}") + urllib.request.urlretrieve(url, tmp_file.name) + tmp_file.close() + + # 解压并覆盖当前代码 + log.info("正在解压更新...") + import shutil + with tarfile.open(tmp_file.name, "r:gz") as tar: + members = tar.getmembers() + prefix = members[0].name.split("/")[0] + "/" + + for member in members: + if not member.name.startswith(prefix): + continue + member.name = member.name[len(prefix):] + if not member.name: + continue + + target = os.path.join(app_dir, member.name) + if member.isdir(): + os.makedirs(target, exist_ok=True) + elif member.isfile(): + parent = os.path.dirname(target) + if parent: + os.makedirs(parent, exist_ok=True) + src = tar.extractfile(member) + if src: + with open(target, "wb") as dst: + shutil.copyfileobj(src, dst) + + # Docker 环境下重新安装依赖（可能有新增依赖） + if in_docker: + import subprocess + try: + pyproject = os.path.join(app_dir, "pyproject.toml") + if os.path.exists(pyproject): + subprocess.run( + [sys.executable, "-m", "pip", "install", "--no-cache-dir", "."], + cwd=app_dir, + capture_output=True, + timeout=120, + ) + except Exception as e: + log.warning(f"更新后重新安装依赖失败: {e}") + + log.info("更新下载并解压完成") + + except Exception as e: + log.error(f"更新失败: {e}") + return web.json_response( + {"ok": False, "error": f"更新失败: {e}"}, + status=500, + ) + finally: + try: + os.unlink(tmp_file.name) + except OSError: + pass + + # 更新完成后重启 + resp = web.json_response({"ok": True, "message": "更新完成，正在重启..."}) + await resp.prepare(request) + await resp.write_eof() + + log.info("一键更新完成，正在重启进程...") + asyncio.get_running_loop().call_soon(_restart_process) + return resp + # 注册路由 web_app.router.add_get("/", handle_index) web_app.router.add_get("/api/setting", handle_get_setting) @@ -219,6 +488,7 @@ async def handle_status(request): web_app.router.add_get("/api/speakers", handle_get_speakers) web_app.router.add_post("/api/speakers/{did}/rename", handle_rename_speaker) web_app.router.add_get("/api/status", handle_status) + web_app.router.add_post("/api/update", handle_execute_update) # 静态文件 import os diff --git a/miair/web/static/index.html b/miair/web/static/index.html index f87987b..d98993d 100644 --- a/miair/web/static/index.html +++ b/miair/web/static/index.html @@ -1,721 +1,2010 @@ + MiAir + + + + + + -
-
- -
为小爱音箱添加 DLNA / AirPlay 投送支持
+
+
+
+
+
+ - -
- - 加载中... -
+
+ + 正在同步状态... +
-
- -
-
+
+ +
- 小米账号 + + 设备列表 +
-
- - + +
+
+ 正在加载设备列表... +
-
- - + +
+
-

- 登录 account.xiaomi.com → F12 开发者工具 → Application → Cookies → 复制 userId 和 passToken -

-
- + +
+
+ + +
+
+ + 渲染器媒体状态 + +
+
+
+ 暂无媒体 +
-

-
+ +
- 实验性功能 + + 配置选项 +
-
-
-
打断后续播
-
被语音唤醒打断后自动恢复播放
+ +
+ 实验性配置 +
+
+
打断后续播 (Auto-Resume)
+
被语音唤醒等操作打断后自动恢复播放
+
+ +
+ +
+
+
自动故障应对
+
当检测到服务或网络发生故障时，将尝试自动重新启动
+
+ +
+ +
+ 续播延迟 (秒) + +
+ +
+
+ 默认音量级别 + 38% +
+ +
+ 1 + 38 + 100 +
+
+ +
+
+
跟随设备当前音量级别
+
+
-
-
- 恢复延迟 - - 秒 + +
+ 网络设置 +
+ + +
+
+ + +
-
- + +
+
-

+ +
- -
-
-
- 选择音箱 -
-

勾选要作为媒体接收器的小爱音箱，点击名称可自定义显示名称

-
-
请先登录小米账号
+
+ MiAir + V0.8.5-alpha by + KiriChen +
+
+ + + + + +