import json import os import shutil import subprocess import tempfile import threading import uuid from concurrent.futures import ThreadPoolExecutor from datetime import datetime from pathlib import Path from django.conf import settings from django.core.files import File from django.db import close_old_connections from learn.models import Mock_Test, Question_Bank from .question_video_service import ( explanation_video_storage, mock_video_folder, parse_question_number, ) _executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="question-video") _status_lock = threading.Lock() _JOB_ID_PATTERN = __import__("re").compile(r"^[a-f0-9]{32}$") def spool_root(): configured = getattr(settings, "QUESTION_VIDEO_SPOOL_ROOT", "") return Path(configured or os.path.join(tempfile.gettempdir(), "cet-question-video-jobs")) def ffmpeg_binary(): configured = getattr(settings, "FFMPEG_BINARY", "") if configured and os.path.isfile(configured) and os.access(configured, os.X_OK): return configured return shutil.which("ffmpeg") def _job_dir(job_id): if not _JOB_ID_PATTERN.fullmatch(str(job_id or "")): raise ValueError("Invalid upload job id.") return spool_root() / job_id def _status_path(job_id): return _job_dir(job_id) / "status.json" def _write_status(job_id, status): job_dir = _job_dir(job_id) job_dir.mkdir(parents=True, exist_ok=True) status["updated_at"] = datetime.utcnow().isoformat() + "Z" path = _status_path(job_id) temporary = path.with_suffix(".tmp") with _status_lock: temporary.write_text(json.dumps(status), encoding="utf-8") os.replace(str(temporary), str(path)) def read_job_status(job_id): path = _status_path(job_id) if not path.exists(): return None with _status_lock: return json.loads(path.read_text(encoding="utf-8")) def create_upload_job(mock_test, uploaded_files): questions = list( Question_Bank.objects.filter(test_id=mock_test.id) .order_by("section_id", "id") .values("id", "tag") ) if not questions: raise ValueError("This mock test has no questions.") max_files = int(getattr(settings, "QUESTION_VIDEO_MAX_FILES_PER_JOB", 500)) if not uploaded_files: raise ValueError("Select one or more video files.") if len(uploaded_files) > max_files: raise ValueError("A maximum of {0} videos can be uploaded at once.".format(max_files)) first_number_by_tag = {} for number, question in enumerate(questions, start=1): tag = str(question.get("tag") or "").strip() if tag and tag.lower() not in ("null", "none"): first_number_by_tag.setdefault(tag, number) parsed_by_group = {} seen_numbers = set() errors = [] for uploaded in uploaded_files: try: number, extension = parse_question_number(uploaded.name) if number < 1 or number > len(questions): raise ValueError( "Question number {0} is outside this mock's 1-{1} range.".format( number, len(questions) ) ) if number in seen_numbers: raise ValueError("More than one file was supplied for question {0}.".format(number)) seen_numbers.add(number) question = questions[number - 1] tag = str(question.get("tag") or "").strip() if tag.lower() in ("null", "none"): tag = "" canonical_number = first_number_by_tag.get(tag, number) if tag else number candidate = (uploaded, number, extension, question, canonical_number, tag) existing = parsed_by_group.get(canonical_number) # Prefer the first question's numbered file (26.mp4 for a 26-30 group). if existing is None or number == canonical_number: parsed_by_group[canonical_number] = candidate except ValueError as exc: errors.append("{0}: {1}".format(uploaded.name, exc)) if errors: raise ValueError("\n".join(errors)) job_id = uuid.uuid4().hex job_dir = _job_dir(job_id) input_dir = job_dir / "input" input_dir.mkdir(parents=True, exist_ok=False) manifest = [] try: for uploaded, source_number, extension, question, canonical_number, tag in sorted( parsed_by_group.values(), key=lambda row: row[4] ): input_path = input_dir / ("{0}{1}".format(canonical_number, extension)) with input_path.open("wb") as destination: for chunk in uploaded.chunks(): destination.write(chunk) manifest.append({ "number": canonical_number, "source_number": source_number, "question_id": question["id"], "tag": tag, "input_path": str(input_path), "original_name": uploaded.name, }) except Exception: shutil.rmtree(str(job_dir), ignore_errors=True) raise status = { "job_id": job_id, "mock_test_id": mock_test.id, "mock_title": mock_test.mock_title, "state": "queued", "total": len(manifest), "processed": 0, "uploaded": 0, "failed": 0, "linked_questions": 0, "ignored_tag_duplicates": len(uploaded_files) - len(manifest), "errors": [], "created_at": datetime.utcnow().isoformat() + "Z", } _write_status(job_id, status) _executor.submit(_process_upload_job, job_id, mock_test.id, manifest) return status def _compress_video(source, destination): binary = ffmpeg_binary() if not binary: raise RuntimeError("FFmpeg is not installed or FFMPEG_BINARY is not configured.") timeout = int(getattr(settings, "QUESTION_VIDEO_FFMPEG_TIMEOUT", 7200)) command = [ binary, "-y", "-i", source, "-map_metadata", "-1", "-c:v", "libx264", "-preset", "fast", "-crf", "28", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "96k", "-movflags", "+faststart", destination, ] result = subprocess.run( command, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, timeout=timeout, check=False, ) if result.returncode != 0: details = result.stderr.decode("utf-8", errors="replace")[-1500:] raise RuntimeError("FFmpeg compression failed: {0}".format(details)) def _process_upload_job(job_id, mock_test_id, manifest): close_old_connections() status = read_job_status(job_id) status["state"] = "processing" _write_status(job_id, status) job_dir = _job_dir(job_id) try: mock_test = Mock_Test.objects.get(id=mock_test_id) storage = explanation_video_storage() folder = mock_video_folder(mock_test) for item in sorted(manifest, key=lambda row: row["number"]): output_path = job_dir / ("compressed-{0}.mp4".format(item["number"])) status["current_question_number"] = item["number"] _write_status(job_id, status) try: _compress_video(item["input_path"], str(output_path)) storage_name = os.path.join(folder, "{0}.mp4".format(item["number"])).replace(os.sep, "/") if storage.exists(storage_name): storage.delete(storage_name) with output_path.open("rb") as compressed: saved_name = storage.save(storage_name, File(compressed)) if item.get("tag"): linked = Question_Bank.objects.filter( test_id=mock_test_id, tag=item["tag"], ).update(explanation_video=saved_name) else: linked = Question_Bank.objects.filter( id=item["question_id"], test_id=mock_test_id ).update(explanation_video=saved_name) status["linked_questions"] += linked status["uploaded"] += 1 except Exception as exc: status["failed"] += 1 status["errors"].append({ "file": item["original_name"], "question_number": item["number"], "error": str(exc), }) finally: status["processed"] += 1 for path in (Path(item["input_path"]), output_path): try: path.unlink() except OSError: pass _write_status(job_id, status) status["state"] = "completed" if status["failed"] == 0 else "completed_with_errors" status.pop("current_question_number", None) except Exception as exc: status["state"] = "failed" status["errors"].append({"error": str(exc)}) finally: status["finished_at"] = datetime.utcnow().isoformat() + "Z" _write_status(job_id, status) shutil.rmtree(str(job_dir / "input"), ignore_errors=True) close_old_connections()