""" Footage Screener 鈥?Web UI (Database-backed) """ import json import os import sys import threading import time from pathlib import Path from werkzeug.utils import secure_filename from flask import Flask, render_template, jsonify, send_file, request, Response from db import ( init_db, insert_footage, list_footage, get_footage_by_id, get_stats, upsert_screener, delete_footage as db_delete_footage, soft_delete_footage, restore_footage, permanent_delete_footage, set_clip_status, create_task, update_task, complete_task, list_tasks, get_running_task, migrate_from_json, get_db, list_all_categories, create_category as db_create_category, move_footage_to_category, ) from analyzer import analyze_video as video_analyze sys.stdout.reconfigure(encoding="utf-8") app = Flask(__name__, template_folder=os.path.join(os.path.dirname(__file__), "templates"), static_folder=os.path.join(os.path.dirname(__file__), "static")) app.config["SEND_FILE_MAX_AGE_DEFAULT"] = 3600 # Cache static files 1 hour app.config["MAX_CONTENT_LENGTH"] = 4 * 1024 * 1024 * 1024 # 4GB max upload # Unified video storage directory VIDEO_STORAGE_DIR = os.environ.get("VIDEO_STORAGE_DIR") or os.path.join(os.path.dirname(__file__), "videos") UPLOAD_DIR = os.path.join(VIDEO_STORAGE_DIR, "uploads") THUMB_DIR = os.path.join(os.path.dirname(__file__), "static", "thumbs") ALLOWED_EXTENSIONS = {".mp4", ".mov", ".avi", ".mkv", ".mts", ".mxf", ".wmv", ".flv", ".mpg", ".mpeg", ".3gp", ".webm", ".ts", ".m4v"} # Global state _analysis_running = False _analysis_progress = {"current": 0, "total": 0, "file": "", "status": "idle"} # === Routes === @app.route("/") def index(): return render_template("index.html") @app.route("/api/footage") def api_footage(): category = request.args.get("category", "") search = request.args.get("search", "") ai_search = request.args.get("ai_search", "") min_dur = request.args.get("min_duration", type=float, default=0) max_dur = request.args.get("max_duration", type=float, default=99999) screener_filter = request.args.get("screener", "") page = request.args.get("page", type=int, default=1) per_page = request.args.get("per_page", type=int, default=60) items, total = list_footage( category=category, search=search, ai_search=ai_search, min_dur=min_dur, max_dur=max_dur, screener_filter=screener_filter, page=page, per_page=per_page, status=request.args.get("status", "active"), ) return jsonify({ "items": items, "total": total, "page": page, "per_page": per_page, "pages": max(1, (total + per_page - 1) // per_page), }) @app.route("/api/stats") def api_stats(): return jsonify(get_stats()) @app.route("/video/") def serve_video(clip_id): """Stream video file with range request support.""" clip = get_footage_by_id(clip_id) if not clip: return "Clip not found", 404 full_path = clip.get("filepath", "") if not os.path.exists(full_path): return "File not found", 404 ext = clip["filename"].lower().rsplit(".", 1)[-1] mime_map = {"mp4": "video/mp4", "mov": "video/quicktime", "avi": "video/x-msvideo", "mkv": "video/x-matroska", "mts": "video/mp2t"} mime = mime_map.get(ext, "video/mp4") file_size = os.path.getsize(full_path) range_header = request.headers.get("Range") if range_header: byte_start = int(range_header.replace("bytes=", "").split("-")[0]) byte_end = min(byte_start + 5 * 1024 * 1024, file_size - 1) length = byte_end - byte_start + 1 def generate_range(): with open(full_path, "rb") as f: f.seek(byte_start) remaining = length while remaining > 0: chunk = f.read(min(65536, remaining)) if not chunk: break remaining -= len(chunk) yield chunk return Response( generate_range(), status=206, mimetype=mime, headers={ "Content-Range": f"bytes {byte_start}-{byte_end}/{file_size}", "Accept-Ranges": "bytes", "Content-Length": str(length), }, ) return send_file(full_path, mimetype=mime) # === Analysis API === @app.route("/api/analyze", methods=["POST"]) def api_analyze(): global _analysis_running if _analysis_running: return jsonify({"error": "Analysis already running"}), 409 data = request.json or {} mode = data.get("mode", "all") selected_ids = set(data.get("selected_ids", [])) cv_only = data.get("cv_only", False) # Get targets from DB if mode == "selected" and selected_ids: targets = [get_footage_by_id(i) for i in selected_ids] targets = [t for t in targets if t] elif mode == "unanalyzed": items, _ = list_footage(screener_filter="未分析", per_page=9999) targets = items else: items, _ = list_footage(per_page=9999) targets = items if not targets: return jsonify({"error": "No videos to analyze"}), 400 _analysis_running = True task_id = create_task(mode, len(targets)) thread = threading.Thread(target=_run_analysis, args=(targets, cv_only, task_id), daemon=True) thread.start() return jsonify({"message": f"Analyzing {len(targets)} videos...", "total": len(targets), "task_id": task_id}) def _analyze_single(clip: dict, cv_only: bool, task_id: int, progress: dict) -> bool: """Analyze a single video using MiMo 2.5.""" filepath = clip.get("filepath", "") footage_id = clip.get("id", 0) if not filepath or not os.path.exists(filepath) or not footage_id: return False try: # Use new AI-only analyzer result = video_analyze(filepath) if result.error: print(f"Analysis error for {clip.get('filename')}: {result.error}") # Calculate decision score final_score = result.final_score # Determine decision based on quality if result.quality_score >= 7: decision = "KEEP" elif result.quality_score >= 4: decision = "REVIEW" else: decision = "DELETE" decision_reason = result.decision_reason or result.quality_reason upsert_screener(footage_id, { "cv_score": 0, "ai_score": round(final_score, 1), "final_score": round(final_score, 1), "decision": decision, "decision_reason": decision_reason, "blur_score": 0, "shake_score": 0, "black_frame_ratio": 0, "overexposed_ratio": 0, "underexposed_ratio": 0, "is_silent": 0, "ai_valid": 1 if decision == "KEEP" else 0, "ai_description": result.description, "ai_quality": result.quality_score, "ai_reason": result.quality_reason, "ai_mode": result.analysis_mode, "issues": [], "transcription": result.transcription, "scenes": result.shots, }) return True except Exception as e: print(f"Error analyzing {clip.get('filename')}: {e}") return False def _safe_scene_analysis(filepath: str) -> dict: """Scene analysis with error protection.""" try: from scene_analyzer import analyze_scenes return analyze_scenes(filepath) except Exception as e: print(f"Scene analysis error: {e}") return {"transcription": "", "scenes": [], "summary": str(e)} def _run_analysis(targets: list[dict], cv_only: bool = False, task_id: int = 0): global _analysis_running, _analysis_progress _analysis_progress = {"current": 0, "total": len(targets), "file": "", "status": "running"} error_msg = "" import concurrent.futures # Process videos in batches of 3 for parallel execution batch_size = 3 for batch_start in range(0, len(targets), batch_size): batch = targets[batch_start:batch_start + batch_size] with concurrent.futures.ThreadPoolExecutor(max_workers=batch_size) as executor: futures = {} for clip in batch: future = executor.submit(_analyze_single, clip, cv_only, task_id, _analysis_progress) futures[future] = clip for future in concurrent.futures.as_completed(futures): clip = futures[future] try: success = future.result(timeout=600) _analysis_progress["current"] += 1 _analysis_progress["file"] = clip.get("filename", "") if task_id: update_task(task_id, completed_count=_analysis_progress["current"], current_file=clip.get("filename", "")) except Exception as e: error_msg = str(e) _analysis_progress["current"] += 1 # count errors too print(f"Error analyzing {clip.get('filename')}: {e}") if task_id: update_task(task_id, completed_count=_analysis_progress["current"], current_file=clip.get("filename", "")) if task_id: complete_task(task_id, error_msg) _analysis_progress["status"] = "done" _analysis_running = False @app.route("/api/clip", methods=["POST"]) def api_clip(): """Clip a segment from a video.""" data = request.json or {} footage_id = data.get("id") start = data.get("start", 0) end = data.get("end", 0) filename = data.get("filename", "clip") if not footage_id or end <= start: return jsonify({"error": "Invalid parameters"}), 400 clip = get_footage_by_id(footage_id) if not clip: return jsonify({"error": "Clip not found"}), 404 filepath = clip.get("filepath", "") if not os.path.exists(filepath): return jsonify({"error": "File not found"}), 404 # Create output filename clip_dir = os.path.join(os.path.dirname(__file__), "clips") os.makedirs(clip_dir, exist_ok=True) stem = Path(filename).stem ext = Path(filename).suffix output_name = f"{stem}_clip_{int(start)}-{int(end)}{ext}" output_path = os.path.join(clip_dir, output_name) # FFmpeg clip cmd = [ "ffmpeg", "-y", "-v", "quiet", "-ss", str(start), "-i", filepath, "-t", str(end - start), "-c", "copy", output_path, ] try: proc = subprocess.run(cmd, capture_output=True, timeout=60, encoding="utf-8", errors="replace") if proc.returncode == 0 and os.path.exists(output_path): return jsonify({"ok": True, "output": output_name, "path": output_path}) else: return jsonify({"error": "FFmpeg failed"}), 500 except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/api/clip-status", methods=["POST"]) def api_clip_status(): """Mark footage as clipped or unclipped.""" data = request.json or {} footage_id = data.get("id") status = data.get("status", "clipped") # "clipped" or "" if not footage_id: return jsonify({"error": "No ID"}), 400 set_clip_status(footage_id, status) return jsonify({"ok": True}) @app.route("/api/analyze-scenes", methods=["POST"]) def api_analyze_scenes(): """Analyze all scenes/shots in a video.""" data = request.json or {} footage_id = data.get("id") if not footage_id: return jsonify({"error": "No footage ID"}), 400 clip = get_footage_by_id(footage_id) if not clip: return jsonify({"error": "Clip not found"}), 404 filepath = clip.get("filepath", "") if not os.path.exists(filepath): return jsonify({"error": "File not found"}), 404 from scene_analyzer import analyze_scenes result = analyze_scenes(filepath) return jsonify(result) @app.route("/api/tasks") def api_tasks(): """List analysis tasks.""" tasks = list_tasks(limit=20) running = get_running_task() return jsonify({"tasks": tasks, "running": running}) @app.route("/api/analyze/status") def api_analyze_status(): return jsonify(_analysis_progress) @app.route("/api/scan", methods=["POST"]) def api_scan(): data = request.json or {} base_dir = data.get("directory", "") if not base_dir or not os.path.exists(base_dir): return jsonify({"error": "Invalid directory"}), 400 from scanner import scan_directory footage = scan_directory(base_dir) return jsonify({"message": f"Scanned {len(footage)} clips", "count": len(footage)}) @app.route("/api/batch/delete", methods=["POST"]) def api_batch_delete(): data = request.json or {} confirm = data.get("confirm", False) ids = data.get("ids", []) if ids: # Soft delete specific items by ID (move to recycle bin) if not confirm: items = [get_footage_by_id(i) for i in ids] items = [i for i in items if i] return jsonify({"count": len(items), "files": [f["filename"] for f in items]}) moved = [] for fid in ids: clip = get_footage_by_id(fid) if not clip: continue soft_delete_footage(fid) moved.append(clip["filename"]) return jsonify({"deleted": len(moved), "files": moved}) else: # Soft delete all DELETE-category items items, _ = list_footage(screener_filter="DELETE", per_page=9999) if not confirm: return jsonify({"count": len(items), "files": [f["filename"] for f in items]}) moved = [] for f in items: soft_delete_footage(f["id"]) moved.append(f["filename"]) return jsonify({"deleted": len(moved), "files": moved}) @app.route("/api/restore", methods=["POST"]) def api_restore(): """Restore items from recycle bin.""" data = request.json or {} ids = data.get("ids", []) if not ids: return jsonify({"error": "No IDs provided"}), 400 restored = [] for fid in ids: clip = get_footage_by_id(fid) if clip: restore_footage(fid) restored.append(clip["filename"]) return jsonify({"restored": len(restored), "files": restored}) @app.route("/api/permanent-delete", methods=["POST"]) def api_permanent_delete(): """Permanently delete items from recycle bin.""" data = request.json or {} ids = data.get("ids", []) if not ids: return jsonify({"error": "No IDs provided"}), 400 deleted = [] for fid in ids: filepath = permanent_delete_footage(fid) if filepath and os.path.exists(filepath): try: os.remove(filepath) except Exception: pass clip = get_footage_by_id(fid) if clip: deleted.append(clip["filename"]) return jsonify({"deleted": len(deleted), "files": deleted}) @app.route("/api/export") def api_export(): selected_ids = request.args.get("ids", "") if selected_ids: ids = [int(x) for x in selected_ids.split(",") if x.strip()] items = [get_footage_by_id(i) for i in ids] items = [i for i in items if i] else: items, _ = list_footage(per_page=9999) report = { "total": len(items), "keep": len([f for f in items if f.get("screener", {}).get("decision") == "KEEP"]), "review": len([f for f in items if f.get("screener", {}).get("decision") == "REVIEW"]), "delete": len([f for f in items if f.get("screener", {}).get("decision") == "DELETE"]), "clips": items, } return jsonify(report) # === Upload API === @app.route("/api/upload", methods=["POST"]) def api_upload(): """Upload video files. Saves to uploads/ dir, extracts metadata, writes to DB.""" if "files" not in request.files: return jsonify({"error": "No files provided"}), 400 files = request.files.getlist("files") category = request.form.get("category", "涓婁紶绱犳潗") os.makedirs(UPLOAD_DIR, exist_ok=True) from scanner import generate_thumbnail uploaded = [] errors = [] for f in files: if not f.filename: continue ext = os.path.splitext(f.filename)[1].lower() if ext not in ALLOWED_EXTENSIONS: errors.append({"filename": f.filename, "error": f"Unsupported format: {ext}"}) continue # Save file (handle duplicates) filename = secure_filename(f.filename) if not filename: filename = f"upload_{int(time.time())}{ext}" dest_path = os.path.join(UPLOAD_DIR, filename) if os.path.exists(dest_path): name, ext2 = os.path.splitext(filename) filename = f"{name}_{int(time.time())}{ext2}" dest_path = os.path.join(UPLOAD_DIR, filename) try: f.save(dest_path) except Exception as e: errors.append({"filename": f.filename, "error": str(e)}) continue # Extract metadata from cv_analyzer import _get_video_info info = _get_video_info(dest_path) # Generate thumbnail thumb_name = f"{category}_{filename}.jpg" thumb_path = os.path.join(THUMB_DIR, thumb_name) thumb_rel = f"thumbs/{thumb_name}" generate_thumbnail(dest_path, thumb_path) # Insert to DB clip = { "filename": filename, "filepath": dest_path, "category": category, "thumbnail": thumb_rel, "duration": round(info["duration"], 2), "width": info["width"], "height": info["height"], "fps": round(info["fps"], 2), "total_frames": info["total_frames"], "size_mb": round(os.path.getsize(dest_path) / 1024 / 1024, 1), } clip_id = insert_footage(clip) clip["id"] = clip_id uploaded.append(clip) return jsonify({ "uploaded": len(uploaded), "errors": len(errors), "files": uploaded, "error_details": errors, }) @app.route("/api/upload/status") def api_upload_status(): """Check upload directory.""" if not os.path.exists(UPLOAD_DIR): return jsonify({"count": 0, "total_size_mb": 0}) files = [f for f in os.listdir(UPLOAD_DIR) if os.path.isfile(os.path.join(UPLOAD_DIR, f))] total_size = sum(os.path.getsize(os.path.join(UPLOAD_DIR, f)) for f in files) return jsonify({"count": len(files), "total_size_mb": round(total_size / 1024 / 1024, 1)}) @app.route("/api/scene-notes/batch", methods=["POST"]) def update_scene_notes_batch(): """Update notes for multiple scenes at once.""" data = request.json or {} footage_id = data.get("footage_id") updates = data.get("updates", []) if not footage_id or not updates: return jsonify({"error": "footage_id and updates required"}), 400 try: with get_db() as conn: # Get current scenes row = conn.execute("SELECT scenes FROM screener WHERE footage_id = ?", (footage_id,)).fetchone() if not row: return jsonify({"error": "No analysis found for this footage"}), 404 scenes = json.loads(row["scenes"]) if row["scenes"] else [] # Apply updates for update in updates: scene_index = update.get("scene_index") notes = update.get("notes", "") for scene in scenes: if scene.get("scene_index") == scene_index or scene.get("shot_index") == scene_index: scene["notes"] = notes break # Save back to database conn.execute("UPDATE screener SET scenes = ? WHERE footage_id = ?", (json.dumps(scenes, ensure_ascii=False), footage_id)) conn.commit() return jsonify({"success": True, "updated": len(updates)}) except Exception as e: return jsonify({"error": str(e)}), 500 # === Scene Notes API === @app.route("/api/scene-notes", methods=["POST"]) def update_scene_notes(): data = request.json or {} footage_id = data.get("footage_id") scene_index = data.get("scene_index") notes = data.get("notes", "") if not footage_id or scene_index is None: return jsonify({"error": "footage_id and scene_index required"}), 400 try: with get_db() as conn: # Get current scenes row = conn.execute("SELECT scenes FROM screener WHERE footage_id = ?", (footage_id,)).fetchone() if not row: return jsonify({"error": "No analysis found for this footage"}), 404 scenes = json.loads(row["scenes"]) if row["scenes"] else [] # Find and update the scene updated = False for scene in scenes: if scene.get("scene_index") == scene_index or scene.get("shot_index") == scene_index: scene["notes"] = notes updated = True break if not updated: return jsonify({"error": "Scene not found"}), 404 # Save back to database conn.execute("UPDATE screener SET scenes = ? WHERE footage_id = ?", (json.dumps(scenes, ensure_ascii=False), footage_id)) conn.commit() return jsonify({"success": True, "notes": notes}) except Exception as e: return jsonify({"error": str(e)}), 500 # === Category Management API === @app.route("/api/categories", methods=["GET"]) def list_categories(): """获取所有分类列表""" categories = list_all_categories() return jsonify({"categories": categories}) @app.route("/api/category/create", methods=["POST"]) def create_category(): """创建新分类（文件夹）""" data = request.json or {} name = data.get("name", "").strip() if not name: return jsonify({"error": "分类名称不能为空"}), 400 if len(name) > 50: return jsonify({"error": "分类名称不能超过50个字符"}), 400 if db_create_category(name): return jsonify({"success": True, "category": name}) else: return jsonify({"error": "分类已存在"}), 400 @app.route("/api/category/move", methods=["POST"]) def move_to_category(): """将选中的视频移动到指定分类""" data = request.json or {} ids = data.get("ids", []) category = data.get("category", "").strip() if not ids: return jsonify({"error": "请选择要移动的视频"}), 400 if not category: return jsonify({"error": "请选择目标分类"}), 400 moved = move_footage_to_category(ids, category) return jsonify({"success": True, "moved": moved}) if __name__ == "__main__": # Auto-migrate from JSON if DB is empty init_db() json_cache = os.path.join(os.path.dirname(__file__), "data", "footage_cache.json") stats = get_stats() if stats["categories"]["全部"]["count"] == 0 and os.path.exists(json_cache): print("Migrating from JSON cache...") count = migrate_from_json(json_cache) print(f"Migrated {count} items.") # Fix any stuck running tasks from previous session from db import get_db as _get_db with _get_db() as _conn: stuck = _conn.execute("SELECT COUNT(*) FROM tasks WHERE status='running'").fetchone()[0] if stuck > 0: _conn.execute("UPDATE tasks SET status='completed', completed_at=CURRENT_TIMESTAMP WHERE status='running'") print(f"Fixed {stuck} stuck task(s)") port = int(os.environ.get("PORT", 5000)) host = os.environ.get("HOST", "0.0.0.0") # 0.0.0.0 = 灞€鍩熺綉鍙闂? print(f"\n 馃幀 Footage Screener Web UI") print(f" 鏈満璁块棶: http://localhost:{port}") print(f" 灞€鍩熺綉璁块棶: http://<浣犵殑IP>:{port}") print(f" 鎸?Ctrl+C 鍋滄鏈嶅姟\n") app.run(host=host, port=port, debug=False)