.gitignore

@@ -1,23 +0,0 @@
-__pycache__/
-*.pyc
-*.pyo
-*.pyd
-.Python
-env/
-venv/
-.env
-*.log
-*.log.*
-*.pid
-*.seed
-*.pid.lock
-coverage/
-.nyc_output/
-.pytest_cache/
-.mypy_cache/
-.idea/
-.vscode/
-*.swp
-*.swo
-*~
-.DS_Store

README.md

@@ -1,84 +1,2 @@
-# verify-delivery-service
+# verify_delivery
 
-KTV产线交付质检HTTP服务 — 4项QA检查
-
-## 服务说明
-
-将KTV产线的交付质检流程封装为HTTP服务，提供统一的质检入口。
-
-### 4项QA检查
-
-1. **QA1 字幕时间精准性** — 时间单调递增、无重叠、无gap>2s
-2. **QA2 字幕歌词正确性** — ASS中歌词与原始歌词逐行比对
-3. **QA3 MTV使用原音频** — 单轨、duration匹配原曲
-4. **QA4 KTV双轨音序** — Track1=伴奏(Accompaniment), Track2=原唱(Original)
-
-## 部署
-
-```bash
-git clone git@git.opencomputing.cn:yumoqing/verify_delivery.git
-cd verify_delivery
-chmod +x start.sh stop.sh
-./start.sh
-```
-
-服务运行在 `http://0.0.0.0:9085`
-
-## API端点
-
-### GET /api/status
-服务状态检查
-
-### POST /api/submit
-提交质检任务
-
-参数：
-- `mtv_path`: MTV视频路径（可选）
-- `ktv_path`: KTV视频路径（可选）
-- `ass_path`: ASS字幕路径（可选）
-- `lyrics_path`: 原始歌词路径（可选）
-- `calibrated_path`: calibrated.json路径（可选，用于QA1）
-- `original_duration`: 原曲时长（秒，可选）
-
-返回：
-```json
-{
-  "status": "submitted",
-  "task_id": "xxx",
-  "message": "质检任务已提交"
-}
-```
-
-### GET /api/task?task_id=xxx
-查询质检任务结果
-
-返回：
-```json
-{
-  "task_id": "xxx",
-  "status": "completed",
-  "result": {
-    "status": "PASSED",
-    "qa_results": {
-      "QA1_timeline": {"passed": true, "errors": []},
-      "QA2_lyrics": {"passed": true, "errors": []},
-      "QA3_mtv": {"passed": true, "errors": []},
-      "QA4_ktv": {"passed": true, "errors": []}
-    },
-    "total_errors": 0,
-    "errors": []
-  }
-}
-```
-
-## 技术栈
-
-- ahserver (ahserver.webapp)
-- longtasks (异步任务队列)
-- ffprobe (视频/音频元数据提取)
-- Python 3.10
-
-## Git仓库
-
-- 远端：`git@git.opencomputing.cn:yumoqing/verify_delivery.git`
-- 本地：`~/test/verify-delivery-service/`

ah.py

@@ -1,46 +0,0 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""KTV产线交付质检服务 — verify_delivery as HTTP service"""
-import sys, os
-sys.path.insert(0, os.path.dirname(__file__))
-
-from ahserver.webapp import webapp
-from ahserver.serverenv import ServerEnv
-from ahserver.configuredServer import add_startup
-from longtasks.longtasks import LongTasks, schedule_once
-from appPublic.log import debug
-from workers.verify import VerifyWorker
-
-
-class VerifyTasks(LongTasks):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-
-    async def process_task(self, payload, workid=None):
-        import json
-        if isinstance(payload, str):
-            payload = json.loads(payload)
-        worker = VerifyWorker(self)
-        result = await worker.run_task(payload)
-        return result
-
-
-async def on_app_built(app):
-    env = ServerEnv()
-    lt = env.longtasks
-    if lt:
-        schedule_once(0.1, lt.run)
-        debug('Verify delivery longtasks worker started')
-
-
-def init():
-    env = ServerEnv()
-    env.longtasks = VerifyTasks(
-        'redis://127.0.0.1:6379', 'verify_delivery',
-        worker_cnt=1, stuck_seconds=600, max_age_hours=24
-    )
-    add_startup(on_app_built)
-
-
-if __name__ == '__main__':
-    webapp(init)

app/api/status/index.dspy

@@ -1,9 +0,0 @@
-# -*- coding: utf-8 -*-
-import json
-result = {
-    "service": "verify-delivery",
-    "status": "running",
-    "checks": ["QA1_timeline", "QA2_lyrics", "QA3_mtv", "QA4_ktv"],
-    "description": "KTV产线交付质检服务 — 4项QA检查"
-}
-return json.dumps(result, ensure_ascii=False, indent=2)

app/api/submit/index.dspy

@@ -1,60 +0,0 @@
-# -*- coding:utf-8 -*-
-# POST /api/submit - 提交KTV交付质检任务
-
-import json
-import uuid
-from ahserver.serverenv import ServerEnv
-
-method = request.method
-
-if method == "POST":
-    mtv_path = params_kw.get("mtv_path", "")
-    ktv_path = params_kw.get("ktv_path", "")
-    ass_path = params_kw.get("ass_path", "")
-    lyrics_path = params_kw.get("lyrics_path", "")
-    calibrated_path = params_kw.get("calibrated_path", "")
-    original_duration = params_kw.get("original_duration", 0)
-
-    if not any([mtv_path, ktv_path]):
-        return json.dumps({"error": "至少需要提供mtv_path或ktv_path"}, ensure_ascii=False)
-
-    task_id = params_kw.get("task_id", str(uuid.uuid4()).replace("-", "")[:12])
-
-    payload = {
-        "task_type": "verify",
-        "task_id": task_id,
-        "mtv_path": mtv_path,
-        "ktv_path": ktv_path,
-        "ass_path": ass_path,
-        "lyrics_path": lyrics_path,
-        "calibrated_path": calibrated_path,
-        "original_duration": float(original_duration) if original_duration else 0
-    }
-
-    env = ServerEnv()
-    longtasks = env.longtasks
-    if longtasks is None:
-        return json.dumps({"error": "service not ready"}, ensure_ascii=False)
-
-    result = await longtasks.submit_task(payload)
-    real_task_id = result.get("task_id", str(result)) if isinstance(result, dict) else str(result)
-
-    return json.dumps({
-        "task_id": real_task_id,
-        "status": "queued",
-        "message": "质检任务已提交",
-        "check_url": f"/api/task?task_id={real_task_id}"
-    }, ensure_ascii=False)
-else:
-    return json.dumps({
-        "usage": "POST with JSON body",
-        "params": {
-            "mtv_path": "string (optional, MTV视频路径)",
-            "ktv_path": "string (optional, KTV视频路径)",
-            "ass_path": "string (optional, ASS字幕路径)",
-            "lyrics_path": "string (optional, 原始歌词路径)",
-            "calibrated_path": "string (optional, calibrated.json路径)",
-            "original_duration": "float (optional, 原曲时长秒)",
-            "task_id": "string (optional, auto-generated)"
-        }
-    }, ensure_ascii=False)

app/api/task/index.dspy

@@ -1,17 +0,0 @@
-# -*- coding:utf-8 -*-
-# GET /api/task?task_id=xxx - 查询任务状态
-
-import json
-from ahserver.serverenv import ServerEnv
-
-task_id = params_kw.get("task_id", "")
-if not task_id:
-    return json.dumps({"error": "task_id is required"}, ensure_ascii=False)
-
-env = ServerEnv()
-longtasks = env.longtasks
-if longtasks is None:
-    return json.dumps({"error": "service not ready"}, ensure_ascii=False)
-
-status = await longtasks.get_status(task_id)
-return json.dumps(status)

build.sh

@@ -1,150 +0,0 @@
-#!/bin/bash
-# 一键部署脚本模板
-# 用法: ./build.sh [deploy|update|stop|status]
-
-set -e
-
-SERVICE_NAME="verify_delivery"
-GIT_REPO="git@git.opencomputing.cn:yumoqing/verify_delivery.git"
-SERVICE_PORT=9085
-DEPLOY_DIR="/data/ymq/$SERVICE_NAME"
-VENV_PATH="/data/ymq/wan22-service/py3"
-GPU_ID="CPU"
-
-# 颜色输出
-RED='\033[0;31m'
-GREEN='\033[0;32m'
-YELLOW='\033[1;33m'
-NC='\033[0m'
-
-log_info() { echo -e "${GREEN}[INFO]${NC} $1"; }
-log_warn() { echo -e "${YELLOW}[WARN]${NC} $1"; }
-log_error() { echo -e "${RED}[ERROR]${NC} $1"; }
-
-check_deps() {
-    command -v git >/dev/null || { log_error "git not found"; exit 1; }
-    [ -f "$VENV_PATH/bin/python" ] || { log_error "Python venv not found: $VENV_PATH"; exit 1; }
-}
-
-deploy() {
-    log_info "Deploying $SERVICE_NAME..."
-    
-    # 检查依赖
-    check_deps
-    
-    # 克隆或更新代码
-    if [ -d "$DEPLOY_DIR/.git" ]; then
-        log_info "Updating existing deployment..."
-        cd "$DEPLOY_DIR"
-        git fetch origin
-        git reset --hard origin/master
-    else
-        log_info "Cloning repository..."
-        cd /data/ymq
-        git clone "$GIT_REPO" "$SERVICE_NAME"
-        cd "$DEPLOY_DIR"
-    fi
-    
-    # 创建必要目录
-    mkdir -p "$DEPLOY_DIR/app/api/status"
-    mkdir -p "$DEPLOY_DIR/app/api/submit"
-    mkdir -p "$DEPLOY_DIR/app/api/task"
-    
-    # 设置权限
-    chmod +x start.sh stop.sh 2>/dev/null || true
-    
-    # 启动服务
-    start_service
-}
-
-start_service() {
-    log_info "Starting $SERVICE_NAME on port $SERVICE_PORT..."
-    
-    # 停止旧进程
-    if [ -f stop.sh ]; then
-        bash stop.sh 2>/dev/null || true
-        sleep 2
-    fi
-    
-    # 启动新进程
-    bash start.sh
-    
-    # 等待启动
-    sleep 3
-    
-    # 验证
-    if ss -tlnp | grep -q ":$SERVICE_PORT "; then
-        log_info "✓ Service started successfully"
-        verify_api
-    else
-        log_error "✗ Service failed to start"
-        log_error "Check logs: $DEPLOY_DIR/nohup.out"
-        exit 1
-    fi
-}
-
-verify_api() {
-    log_info "Verifying API endpoints..."
-    
-    # 检查 status endpoint
-    if curl -s "http://127.0.0.1:$SERVICE_PORT/api/status" | grep -q "service"; then
-        log_info "✓ /api/status OK"
-    else
-        log_warn "✗ /api/status failed"
-    fi
-}
-
-stop_service() {
-    log_info "Stopping $SERVICE_NAME..."
-    if [ -f "$DEPLOY_DIR/stop.sh" ]; then
-        cd "$DEPLOY_DIR"
-        bash stop.sh
-        log_info "✓ Service stopped"
-    else
-        log_warn "stop.sh not found"
-    fi
-}
-
-show_status() {
-    echo "=== $SERVICE_NAME Status ==="
-    echo "Port: $SERVICE_PORT"
-    echo "Deploy Dir: $DEPLOY_DIR"
-    echo ""
-    
-    # 检查进程
-    if ss -tlnp | grep -q ":$SERVICE_PORT "; then
-        echo -e "Status: ${GREEN}RUNNING${NC}"
-        PID=$(ss -tlnp | grep ":$SERVICE_PORT " | grep -oP 'pid=\K[0-9]+')
-        echo "PID: $PID"
-    else
-        echo -e "Status: ${RED}STOPPED${NC}"
-    fi
-    echo ""
-    
-    # 检查 API
-    echo "API Endpoints:"
-    curl -s "http://127.0.0.1:$SERVICE_PORT/api/status" 2>/dev/null | python3 -m json.tool 2>/dev/null || echo "  (not responding)"
-}
-
-# 主入口
-case "${1:-deploy}" in
-    deploy|install)
-        deploy
-        ;;
-    update|upgrade)
-        deploy
-        ;;
-    stop)
-        stop_service
-        ;;
-    start)
-        start_service
-        ;;
-    status)
-        show_status
-        ;;
-    *)
-        echo "Usage: $0 {deploy|update|stop|start|status}"
-        exit 1
-        ;;
-esac

conf/config.json

@@ -1,16 +0,0 @@
-{
-    "password_key": "VerifyDeliveryService2026Key",
-    "databases": {},
-    "session_redis": {"host": "127.0.0.1", "port": 6379, "db": 1},
-    "website": {
-        "paths": [["$[workdir]$/app", ""]],
-        "host": "0.0.0.0",
-        "port": 9085,
-        "coding": "utf-8",
-        "indexes": ["index.html", "index.dspy"],
-        "processors": [[".dspy", "dspy"]],
-        "startswiths": [{"leading": "/idfile", "registerfunction": "idfile"}]
-    },
-    "hot_reload": false,
-    "filesroot": "/tmp/verify-delivery-outputs"
-}

start.sh

@@ -1,5 +0,0 @@
-#!/bin/bash
-cd "$(dirname "$0")"
-nohup /data/ymq/wan22-service/py3/bin/python ah.py > service.log 2>&1 &
-echo $! > service.pid
-echo "verify-delivery-service started, PID: $!"

stop.sh

@@ -1,14 +0,0 @@
-#!/bin/bash
-cd "$(dirname "$0")"
-if [ -f service.pid ]; then
-    pid=$(cat service.pid)
-    if kill -0 $pid 2>/dev/null; then
-        kill $pid
-        echo "verify-delivery-service stopped, PID: $pid"
-    else
-        echo "Process $pid not running"
-    fi
-    rm -f service.pid
-else
-    echo "service.pid not found"
-fi

workers/verify.py

@@ -1,194 +0,0 @@
-# -*- coding: utf-8 -*-
-"""KTV产线交付质检Worker — 4项QA检查"""
-import json, os, subprocess, re
-
-def run_ffprobe(filepath, args):
-    cmd = ["ffprobe", "-v", "error"] + args + [filepath]
-    r = subprocess.run(cmd, capture_output=True, text=True)
-    return r.stdout.strip()
-
-class VerifyWorker:
-    def __init__(self, longtasks):
-        self.lt = longtasks
-
-    async def run_task(self, payload):
-        mtv_path = payload.get("mtv_path", "")
-        ktv_path = payload.get("ktv_path", "")
-        ass_path = payload.get("ass_path", "")
-        lyrics_path = payload.get("lyrics_path", "")
-        original_duration = float(payload.get("original_duration", 0))
-
-        # 检查文件是否存在
-        missing = []
-        for label, p in [("MTV", mtv_path), ("KTV", ktv_path), ("ASS", ass_path), ("歌词", lyrics_path)]:
-            if p and not os.path.exists(p):
-                missing.append(f"{label}: {p}")
-        if missing:
-            return {"status": "FAILED", "error": "文件不存在", "missing": missing}
-
-        all_errors = []
-        qa_results = {}
-
-        # QA-1: 字幕时间精准性（从calibrated数据检查）
-        # 如果有calibrated_path，检查时间轴
-        calibrated_path = payload.get("calibrated_path", "")
-        if calibrated_path and os.path.exists(calibrated_path):
-            qa1_errors = self._check_timeline(calibrated_path)
-            qa_results["QA1_timeline"] = {"passed": len(qa1_errors) == 0, "errors": qa1_errors}
-            all_errors.extend(qa1_errors)
-        else:
-            qa_results["QA1_timeline"] = {"passed": True, "errors": [], "skipped": "no calibrated_path"}
-
-        # QA-2: 字幕歌词正确性
-        qa2_errors = self._check_subtitles(ass_path, lyrics_path)
-        qa_results["QA2_lyrics"] = {"passed": len(qa2_errors) == 0, "errors": qa2_errors}
-        all_errors.extend(qa2_errors)
-
-        # QA-3: MTV使用原音频
-        if mtv_path and os.path.exists(mtv_path):
-            qa3_errors = self._check_mtv(mtv_path, original_duration)
-            qa_results["QA3_mtv"] = {"passed": len(qa3_errors) == 0, "errors": qa3_errors}
-            all_errors.extend(qa3_errors)
-        else:
-            qa_results["QA3_mtv"] = {"passed": True, "errors": [], "skipped": "no MTV file"}
-
-        # QA-4: KTV双轨音序
-        if ktv_path and os.path.exists(ktv_path):
-            qa4_errors = self._check_ktv(ktv_path, original_duration)
-            qa_results["QA4_ktv"] = {"passed": len(qa4_errors) == 0, "errors": qa4_errors}
-            all_errors.extend(qa4_errors)
-        else:
-            qa_results["QA4_ktv"] = {"passed": True, "errors": [], "skipped": "no KTV file"}
-
-        # 汇总
-        passed = len(all_errors) == 0
-        return {
-            "status": "PASSED" if passed else "FAILED",
-            "qa_results": qa_results,
-            "total_errors": len(all_errors),
-            "errors": all_errors
-        }
-
-    def _check_timeline(self, calibrated_path):
-        """QA-1: 字幕时间精准性 — 时间单调递增、无重叠"""
-        errors = []
-        try:
-            with open(calibrated_path, 'r') as f:
-                data = json.load(f)
-            lyrics = data.get("lyrics", data.get("segments", data))
-            if not isinstance(lyrics, list):
-                return ["calibrated数据格式错误，应为list"]
-
-            for i in range(len(lyrics)):
-                seg = lyrics[i]
-                start = seg.get("start", 0)
-                end = seg.get("end", 0)
-                if end <= start:
-                    errors.append(f"行{i+1}: end({end:.2f})<=start({start:.2f})")
-                if i > 0:
-                    prev_end = lyrics[i-1].get("end", 0)
-                    if start < prev_end - 0.1:
-                        errors.append(f"行{i+1}: start({start:.2f})与前一行end({prev_end:.2f})重叠")
-        except Exception as e:
-            errors.append(f"读取calibrated文件失败: {e}")
-        return errors
-
-    def _check_subtitles(self, ass_path, lyrics_path):
-        """QA-2: 字幕歌词正确性 — ASS中歌词与原始歌词逐行比对"""
-        errors = []
-        if not ass_path or not lyrics_path:
-            return errors
-        if not os.path.exists(ass_path):
-            return [f"ASS文件不存在: {ass_path}"]
-        if not os.path.exists(lyrics_path):
-            return [f"原始歌词文件不存在: {lyrics_path}"]
-
-        with open(lyrics_path, 'r') as f:
-            original_lines = [l.strip() for l in f.readlines() if l.strip()]
-
-        with open(ass_path, 'r') as f:
-            ass_content = f.read()
-
-        karaoke_lines = []
-        for line in ass_content.split('\n'):
-            if 'Karaoke' in line and 'Dialogue' in line:
-                parts = line.split(',', 9)
-                if len(parts) >= 10:
-                    text = parts[9]
-                    clean = re.sub(r'\{\\kf\d+\}', '', text).strip()
-                    if clean:
-                        karaoke_lines.append(clean)
-
-        if len(karaoke_lines) != len(original_lines):
-            errors.append(f"歌词行数不匹配: ASS={len(karaoke_lines)}行, 原始={len(original_lines)}行")
-
-        mismatch_count = 0
-        for i in range(min(len(karaoke_lines), len(original_lines))):
-            kl = karaoke_lines[i].replace(' ', '')
-            ol = original_lines[i].replace(' ', '')
-            if kl != ol:
-                mismatch_count += 1
-                if mismatch_count <= 5:
-                    errors.append(f"行{i+1}: ASS='{kl}' vs 原始='{ol}'")
-        if mismatch_count > 5:
-            errors.append(f"... 还有{mismatch_count-5}处不匹配")
-
-        return errors
-
-    def _check_mtv(self, path, original_duration):
-        """QA-3: MTV使用原音频（单轨，duration匹配）"""
-        errors = []
-        try:
-            streams = run_ffprobe(path, [
-                "-show_entries", "stream=index,codec_type,duration",
-                "-show_entries", "stream_tags=handler_name",
-                "-of", "json"
-            ])
-            data = json.loads(streams)
-            audio_streams = [s for s in data.get("streams", []) if s.get("codec_type") == "audio"]
-            if len(audio_streams) != 1:
-                errors.append(f"MTV应有1条音轨，实际{len(audio_streams)}条")
-            if audio_streams and original_duration > 0:
-                dur = float(audio_streams[0].get("duration", 0))
-                if abs(dur - original_duration) > 2:
-                    errors.append(f"音频时长{dur:.1f}s与原曲{original_duration:.1f}s差距>2s")
-
-            fmt = run_ffprobe(path, ["-show_entries", "format=duration", "-of", "csv=p=0"])
-            if fmt and original_duration > 0:
-                video_dur = float(fmt)
-                if abs(video_dur - original_duration) > 3:
-                    errors.append(f"视频时长{video_dur:.1f}s与原曲{original_duration:.1f}s差距>3s")
-        except Exception as e:
-            errors.append(f"ffprobe失败: {e}")
-        return errors
-
-    def _check_ktv(self, path, original_duration):
-        """QA-4: KTV双轨音序 — Track1=伴奏, Track2=原唱"""
-        errors = []
-        try:
-            streams = run_ffprobe(path, [
-                "-show_entries", "stream=index,codec_type,duration",
-                "-show_entries", "stream_tags=handler_name",
-                "-of", "json"
-            ])
-            data = json.loads(streams)
-            audio_streams = [s for s in data.get("streams", []) if s.get("codec_type") == "audio"]
-            if len(audio_streams) != 2:
-                errors.append(f"KTV应有2条音轨，实际{len(audio_streams)}条")
-                return errors
-
-            t1 = audio_streams[0].get("tags", {}).get("handler_name", "")
-            if "伴奏" not in t1 and "Accompaniment" not in t1:
-                errors.append(f"Track1应为伴奏，实际标签: '{t1}'")
-            t2 = audio_streams[1].get("tags", {}).get("handler_name", "")
-            if "原唱" not in t2 and "Original" not in t2:
-                errors.append(f"Track2应为原唱，实际标签: '{t2}'")
-
-            if original_duration > 0:
-                for i, a in enumerate(audio_streams):
-                    dur = float(a.get("duration", 0))
-                    if abs(dur - original_duration) > 2:
-                        errors.append(f"音轨{i+1}时长{dur:.1f}s与原曲{original_duration:.1f}s差距>2s")
-        except Exception as e:
-            errors.append(f"ffprobe失败: {e}")
-        return errors