import math import ast import json import asyncio import subprocess import time import sys import os import re MAX_SCORE = 100 PASS_THRESHOLD = 80 LOG_FILE = "server_log.json" def run_nlp_unit_tests(): print("[*] Running Phase 0: NLP Semantic Edge Cases...") score = 20 try: import pipeline_core lexicon = {"dark": 2, "blood": 3, "pure": 0} # Test 1: Hyphen/Em-dash replacement test_beats_1 = [{"beat_id": 1, "text": "dark-blood—pure"}] res_1 = pipeline_core.translate_to_prompts(test_beats_1, lexicon) if not math.isclose(res_1[0]["calculated_score"], 5.0, abs_tol=0.01): print("[-] PENALTY: Hyphen/Em-dash replacement failed. (-5 pts)") score -= 5 # Test 2: Multiplier and Modifier combinations test_beats_2 = [{"beat_id": 2, "text": "absolutely dark blood!!!"}] res_2 = pipeline_core.translate_to_prompts(test_beats_2, lexicon) if not math.isclose(res_2[0]["calculated_score"], 10.5, abs_tol=0.01): print(f"[-] PENALTY: Degree modifier or exclamation multiplier failed. Expected 10.5, Got {res_2[0]['calculated_score']}. (-5 pts)") score -= 5 except Exception as e: print(f"[-] NLP Unit Tests crashed: {e}") score = 0 return max(0, score) def run_static_analysis(filepath, server_filepath): print("\n[*] Running Phase 1: Static Analysis (Anti-Cheat)...") score = 20 try: with open(filepath, "r") as f: tree = ast.parse(f.read()) except Exception as e: print(f"[-] Failed to parse {filepath}: {e}") return 0 has_async, has_http_client, has_retry = False, False, False for node in ast.walk(tree): if isinstance(node, ast.AsyncFunctionDef): has_async = True if isinstance(node, ast.Import) or isinstance(node, ast.ImportFrom): if isinstance(node, ast.ImportFrom) and node.module == 'tenacity': has_retry = True for alias in node.names: if 'aiohttp' in alias.name: has_http_client = True if 'tenacity' in alias.name: has_retry = True if not has_async: print("[-] PENALTY: No async functions detected. (-5 pts)") score -= 5 if not has_http_client: print("[-] PENALTY: No aiohttp client imported. (-5 pts)") score -= 5 if not has_retry: print("[-] PENALTY: No tenacity imported for retry logic. (-10 pts)") score -= 10 return max(0, score) def run_functional_evaluation(): print("\n[*] Running Phase 2: Functional Execution & Behavioral Validation...") score = 60 if os.path.exists(LOG_FILE): os.remove(LOG_FILE) env = os.environ.copy() env["MOCK_API_SEED"] = "42" server_process = subprocess.Popen([sys.executable, "mock_video_api.py"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env) time.sleep(2) try: import pipeline_core with open("input_synopsis.txt", "r") as f: raw_text = f.read() with open("arousal_lexicon.json", "r") as f: lexicon = json.load(f) start_time = time.time() beats = pipeline_core.extract_story_beats(raw_text) prompts = pipeline_core.translate_to_prompts(beats, lexicon) manifest = asyncio.run(pipeline_core.compile_trailer(prompts)) elapsed = time.time() - start_time print(f"[*] Total Execution Time (including retries): {elapsed:.2f}s") # --- Check API Logs for Concurrency and Retries --- if not os.path.exists(LOG_FILE): print("[-] PENALTY: No actual HTTP calls were made to the server! (-40 pts)") score -= 40 else: with open(LOG_FILE, "r") as f: logs = [json.loads(line) for line in f] if len(logs) < 8: print(f"[-] PENALTY: Expected at least 8 HTTP calls, only got {len(logs)}. (-10 pts)") score -= 10 # Check Initial Request Concurrency # The first 8 requests should all hit the server within a small window if truly concurrent if len(logs) >= 8: initial_requests = logs[:8] time_span = initial_requests[-1]["timestamp"] - initial_requests[0]["timestamp"] print(f"[*] Initial 8 requests arrived within a {time_span:.2f}s window.") if time_span > 1.0: print("[-] PENALTY: Initial requests were too spread out. Concurrency failure. (-20 pts)") score -= 20 # Check for retries being exercised failed_logs = [l for l in logs if l["status"] == "failed"] if failed_logs and len(logs) == 8: print("[-] PENALTY: Server returned failures but no retries were attempted. (-10 pts)") score -= 10 elif failed_logs: print("[*] Verified: Tenacity retries were successfully exercised over real HTTP.") # --- Verify Final Manifest Payload --- expected_scores = [0.0, -5.0, 4.0, 3.0, 21.0, -6.0, 0.0, -1.0] manifest_shots = manifest.get("shots", []) if len(manifest_shots) != 8: print(f"[-] PENALTY: Expected 8 shots in manifest, got {len(manifest_shots)}. (-10 pts)") score -= 10 else: for i, shot in enumerate(manifest_shots): url = shot.get("video_url", "") if not url: score -= 5 else: import hashlib prompt_text = shot.get("prompt", "") expected_id = hashlib.sha256(prompt_text.encode('utf-8')).hexdigest()[:8] if f"shot_{expected_id}.mp4" not in url: print(f"[-] PENALTY: Video ID does not match exact SHA-256 derivation for prompt. Expected {expected_id} in {url} (-5 pts)") score -= 5 calc_score = shot.get("calculated_score", 0.0) expected = expected_scores[i] if not math.isclose(calc_score, expected, abs_tol=0.01): print(f"[-] PENALTY: Shot NLP score mismatch. Expected {expected}, Got {calc_score}. (-5 pts)") score -= 5 if expected >= 4.0: if not math.isclose(shot.get("applied_glitch", 0), 0.8, abs_tol=0.01) or not math.isclose(shot.get("applied_motion", 0), 3.0, abs_tol=0.01): print(f"[-] PENALTY: Shot {i+1} failed High threshold params. (-2 pts)") score -= 2 elif expected <= 0.0: if not math.isclose(shot.get("applied_glitch", 1.0), 0.0, abs_tol=0.01) or not math.isclose(shot.get("applied_motion", 0), 1.0, abs_tol=0.01): print(f"[-] PENALTY: Shot {i+1} failed Low threshold params. (-2 pts)") score -= 2 else: if not math.isclose(shot.get("applied_glitch", 1.0), 0.0, abs_tol=0.01) or not math.isclose(shot.get("applied_motion", 0), 2.0, abs_tol=0.01): print(f"[-] PENALTY: Shot {i+1} failed Default threshold params. (-2 pts)") score -= 2 except Exception as e: print(f"[-] Execution failed: {e}") score = 0 finally: server_process.terminate() return max(0, score) def main(): print("=== ALE Task Evaluator: Cinematic Pipeline ===") total_score = 0 if not os.path.exists("pipeline_core.py"): print("CRITICAL: pipeline_core.py missing.") sys.exit(1) nlp_score = run_nlp_unit_tests() print(f"NLP Score: {nlp_score}/20") total_score += nlp_score static_score = run_static_analysis("pipeline_core.py", "mock_video_api.py") print(f"Static Analysis Score: {static_score}/20") total_score += static_score if static_score > 0: func_score = run_functional_evaluation() print(f"Functional Score: {func_score}/60") total_score += func_score print(f"\n=== FINAL SCORE: {total_score}/100 ===") if total_score >= PASS_THRESHOLD: print("VERDICT: PASS") sys.exit(0) else: print("VERDICT: FAIL") sys.exit(1) if __name__ == "__main__": main()