Oliver/ai_qc
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
ai_qc/backend/api_server.py
nickviljoen 71bb9a6295 fix(hp_copy_review): correct llm casing + route HP reports to /hp/ folder 
Two bugs surfaced by the first dev smoke test:

1. Profile JSON declared "llm": "gemini" (lowercase). llm_config's
   dispatcher compares model_name == "Gemini" case-sensitively
   (matches the rest of the codebase), so the check fell through to
   "Invalid model selected" and never reached the API. Every other
   profile uses "Gemini" with capital G. Spec mistake — fixed.

2. get_client_from_profile() resolves the per-report output folder
   from the profile_id via hardcoded prefix matches. No 'hp_' branch
   existed, so hp_copy_review reports landed under output-dev/general/
   instead of output-dev/hp/ — the UI then couldn't find them. Added
   'hp_' → 'hp' alongside the existing mappings.

The check itself works correctly otherwise: profile_source was
user_selected, brand resolved to 'hp', and the reference asset was
successfully attached. Bug 1 just prevented Gemini from being called.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-17 22:07:25 +02:00

6828 lines
No EOL
307 KiB
Python
Executable file
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""
API server for Visual AI QC application.
Provides API endpoints for visual quality control checks without web UI.
"""
import os
import sys
import json
import base64
import collections
import html
import importlib
import traceback
import re
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta
from pathlib import Path
from flask import Flask, request, jsonify, Response, make_response, g, redirect
from dotenv import load_dotenv
# Determine environment and load appropriate config
def load_environment_config():
"""Load environment-specific configuration"""
# Check for environment variable first
environment = os.environ.get('ENVIRONMENT', 'development')
# Determine config file path based on environment
base_dir = os.path.dirname(os.path.abspath(__file__))
# Try new config structure first
config_path = os.path.join(base_dir, 'config', f'{environment}.env')
# Fall back to old config.env if new structure doesn't exist
if not os.path.exists(config_path):
old_config_path = os.path.join(base_dir, 'config.env')
if os.path.exists(old_config_path):
config_path = old_config_path
environment = 'production' # Assume production for backward compatibility
print(f"Using legacy config file: {config_path}")
else:
print(f"No configuration file found. Checked: {config_path} and {old_config_path}")
return environment
# Load the configuration
load_dotenv(config_path)
print(f"Environment: {environment}")
print(f"Loaded configuration from: {config_path}")
print(f"OPENAI_API_KEY set: {'OPENAI_API_KEY' in os.environ}")
print(f"GOOGLE_API_KEY set: {'GOOGLE_API_KEY' in os.environ}")
print(f"Port: {os.environ.get('PORT', 'not set')}")
return environment
# Load environment configuration
current_environment = load_environment_config()
# Add the parent directory to the Python path to ensure imports work correctly
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
# Import QC utilities and model configuration
from visual_qc_apps.utils import get_image_from_asset
from llm_config import run_visual_qc, get_model_info
from profile_config import QC_CHECKS, PROFILES, get_profile, get_check_llm_map
from brand_guidelines_db import BrandGuidelinesDB
from auth_middleware import AuthMiddleware
from technical_check import inspect as technical_inspect, format_for_llm_prompt as technical_to_prompt
import box_jwt_client
from PIL import Image
import io
# Create Flask app
app = Flask(__name__)
# Configure app based on environment
upload_folder = os.environ.get('UPLOAD_FOLDER', 'uploads')
output_folder = os.environ.get('OUTPUT_FOLDER', 'output')
debug_mode = os.environ.get('DEBUG_MODE', 'false').lower() == 'true'
app.config['UPLOAD_FOLDER'] = upload_folder
app.config['OUTPUT_FOLDER'] = output_folder
app.config['MAX_CONTENT_LENGTH'] = 100 * 1024 * 1024 # 100MB max file size
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'default-secret-change-this')
app.debug = debug_mode
print(f"Upload folder: {upload_folder}")
print(f"Output folder: {output_folder}")
print(f"Debug mode: {debug_mode}")
# Ensure directories exist
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
os.makedirs(app.config['OUTPUT_FOLDER'], exist_ok=True)
# Initialize authentication middleware
auth = AuthMiddleware(app)
# Initialize brand guidelines database
brand_db = BrandGuidelinesDB()
# Initialize media plans storage
MEDIA_PLANS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'media_plans')
os.makedirs(MEDIA_PLANS_DIR, exist_ok=True)
MEDIA_PLANS_DB_FILE = os.path.join(MEDIA_PLANS_DIR, 'plans_db.json')
def _load_media_plans_db():
if os.path.exists(MEDIA_PLANS_DB_FILE):
try:
with open(MEDIA_PLANS_DB_FILE, 'r') as f:
return json.load(f)
except (json.JSONDecodeError, FileNotFoundError):
pass
return {}
def _save_media_plans_db(db):
with open(MEDIA_PLANS_DB_FILE, 'w') as f:
json.dump(db, f, indent=2)
def _get_active_media_plan(client_id):
"""Get the parsed media plan data for a client, or None."""
db = _load_media_plans_db()
plan_info = db.get(client_id)
if not plan_info:
return None
json_path = plan_info.get('json_path')
if json_path and os.path.exists(json_path):
try:
with open(json_path, 'r') as f:
return json.load(f)
except (json.JSONDecodeError, FileNotFoundError):
pass
return None
# Global progress tracking
progress_tracker = {}
# Dictionary to store QC app instances and prompts
qc_apps = {}
# Define QC checks that require reference assets
REFERENCE_ASSET_REQUIRED_CHECKS = {'brand_assets_visibility', 'visual_hierarchy', 'logo_visibility'}
def extract_json_from_response(response_text):
"""Extract JSON objects from the LLM's response"""
# First, try to find JSON blocks (```json ... ```)
json_pattern = r'```json\s*(.*?)\s*```'
json_matches = re.finditer(json_pattern, response_text, re.DOTALL)
# Get all JSON blocks as a list
json_objects = []
for match in json_matches:
try:
json_data = json.loads(match.group(1).strip())
json_objects.append(json_data)
except Exception as e:
print(f"Could not parse JSON block: {e}")
# If we found multiple JSON blocks, merge them (later blocks override earlier blocks)
if json_objects:
merged_json = {}
for json_obj in json_objects:
if json_obj: # If not empty
merged_json.update(json_obj)
if merged_json:
return merged_json
# If we couldn't extract JSON blocks or they were empty, look for JSON directly
try:
# Try to find pure JSON in response (without code blocks)
# Remove markdown code formatting first
clean_response = re.sub(r'```.*?```', '', response_text, flags=re.DOTALL)
# Look for text that looks like JSON (between { and })
json_pattern = r'\{.*\}'
json_match = re.search(json_pattern, clean_response, re.DOTALL)
if json_match:
try:
json_data = json.loads(json_match.group(0))
return json_data
except json.JSONDecodeError:
pass
except Exception as e:
print(f"Failed to extract direct JSON: {e}")
# If we couldn't find valid JSON, return an empty dict
return {}
def detect_and_crop_main_element(image_path, file_type_hint=None):
"""Detect main element in POS files and return cropped region coordinates"""
try:
if not file_type_hint or 'pos' not in file_type_hint.lower():
return None
# Create a prompt to detect the main marketing element
crop_detection_prompt = """
This appears to be a POS (Point of Sale) material. Please identify the main marketing element that should be used for QC analysis.
Look for the primary branded content area (usually the center panel or main marketing message) and ignore peripheral elements like:
- Side panels with fine print
- Edge decorations
- Background patterns
- Border elements
Respond with JSON format:
{
"main_element_detected": true/false,
"crop_coordinates": {
"x": left_position_percentage,
"y": top_position_percentage,
"width": width_percentage,
"height": height_percentage
},
"description": "description of the main element found"
}
Coordinates should be percentages (0-100) of the total image dimensions.
"""
# Run AI analysis to detect main element
result = run_visual_qc(
prompt=crop_detection_prompt,
asset_path=image_path,
model_name="Gemini"
)
# Extract crop coordinates
crop_data = extract_json_from_response(result['response'])
if crop_data.get('main_element_detected') and 'crop_coordinates' in crop_data:
return crop_data['crop_coordinates']
return None
except Exception as e:
print(f"Error detecting main element: {e}")
return None
def apply_crop_to_analysis(image_path, crop_coordinates):
"""Apply cropping to image for analysis (if coordinates provided)"""
if not crop_coordinates:
return image_path
try:
from PIL import Image
import os
# Load original image
with Image.open(image_path) as img:
width, height = img.size
# Convert percentage coordinates to pixels
x = int((crop_coordinates['x'] / 100) * width)
y = int((crop_coordinates['y'] / 100) * height)
crop_width = int((crop_coordinates['width'] / 100) * width)
crop_height = int((crop_coordinates['height'] / 100) * height)
# Ensure coordinates are within image bounds
x = max(0, min(x, width))
y = max(0, min(y, height))
crop_width = min(crop_width, width - x)
crop_height = min(crop_height, height - y)
# Crop the image
cropped_img = img.crop((x, y, x + crop_width, y + crop_height))
# Save cropped version
filename, ext = os.path.splitext(image_path)
cropped_path = f"{filename}_cropped{ext}"
cropped_img.save(cropped_path)
return cropped_path
except Exception as e:
print(f"Error applying crop: {e}")
return image_path
return image_path
def extract_score_from_result(result, profile_config=None, check_name=None):
"""Extract score from LLM result response with Unilever-specific logic"""
score = None
try:
# Use our extraction function to get score from JSON blocks
json_data = extract_json_from_response(result['response'])
# Unilever Key Visual profile specific logic
if (profile_config and
((hasattr(profile_config, 'name') and profile_config.name == 'Unilever Key Visual') or
(hasattr(profile_config, 'get') and profile_config.get('name') == 'Unilever Key Visual')) and
check_name in ['face_visibility', 'new_visibility', 'face_gaze_direction']):
# Check for zero score conditions based on missing elements
if check_name == 'face_visibility' and json_data.get('face_present') == False:
print(f"Unilever profile: No face detected for {check_name}, setting score to 0")
return 0
elif check_name == 'new_visibility' and json_data.get('new_present') == False:
print(f"Unilever profile: No 'new' element detected for {check_name}, setting score to 0")
return 0
elif check_name == 'face_gaze_direction' and json_data.get('face_present') == False:
print(f"Unilever profile: No face detected for {check_name}, setting score to 0")
return 0
# Standard scoring logic
if 'score' in json_data:
score = json_data.get('score')
print(f"Extracted score from JSON block: {score}")
# If we still don't have a score, look for any score in text
if score is None:
# Try to find a score pattern in the text
score_pattern = r'["\']score["\']\s*:\s*(\d+)'
score_match = re.search(score_pattern, result['response'])
if score_match:
score = int(score_match.group(1))
print(f"Extracted score from regex: {score}")
else:
# Look for descriptive scores in text
descriptive_score_pattern = r'score(?:\s+is|\s*:\s*|\s+of\s+)(?:\s*)(\d+)(?:\s*out\s*of\s*10)?'
descriptive_match = re.search(descriptive_score_pattern, result['response'].lower())
if descriptive_match:
score = int(descriptive_match.group(1))
print(f"Extracted score from descriptive text: {score}")
else:
# Try to determine score from pass/fail status (legacy mode)
result_text = result.get('response', '').upper()
if "PASS" in result_text:
score = 10 # Pass = 10/10
print("Detected PASS keyword, setting score to 10")
elif "FAIL" in result_text:
score = 3 # Fail = 3/10
print("Detected FAIL keyword, setting score to 3")
else:
score = 5 # Default middle score
print(f"Could not extract score, using default of 5")
except Exception as parse_error:
print(f"Error parsing score from response: {parse_error}")
score = 5 # Default to middle score
return score if score is not None else 5
def determine_grade(overall_score):
"""Determine Pass/Fail based on overall score"""
# Convert overall score to individual check average (1-10 scale)
avg_individual_score = overall_score / 10
if avg_individual_score >= 6:
return 'Pass'
else:
return 'Fail'
def _run_dj_file_naming_check(check_name, file_path, profile_weights):
"""
Deterministic file-naming check for the Dow Jones / OLIVER convention.
Bypasses the LLM dispatch entirely; returns a result dict shaped like an LLM check
so it slots into the existing scoring + report pipeline.
"""
from file_naming_validator import validate_filename
filename = os.path.basename(file_path) if file_path else ''
validation = validate_filename(filename)
weight = profile_weights.get(check_name, 0.1)
score = validation['score']
weighted_score = score * weight
return {
'check_name': check_name,
'status': 'success',
'response': validation['explanation'],
'json_data': validation,
'score': score,
'weight': weight,
'weighted_score': weighted_score,
'model_used': {'model': 'deterministic', 'provider': 'internal'},
'token_usage': {'input_tokens': 0, 'output_tokens': 0},
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'display_name': 'DJ File Naming',
'brand_guidelines_status': None,
'requires_brand_guidelines': False,
}
def process_single_check(check_name, qc_apps, profile_config, profile_weights, file_path,
analysis_reference_asset, brand_db, progress_tracker, session_id,
check_index, total_checks, model_version=None, media_plan_context=None,
ocr_context=None):
"""Process a single QC check - designed to run in parallel"""
try:
# Deterministic file-naming check — short-circuit before any LLM dispatch.
if check_name == 'dj_file_naming':
return _run_dj_file_naming_check(check_name, file_path, profile_weights)
# Check if this check requires a reference asset but none is provided
if check_name in REFERENCE_ASSET_REQUIRED_CHECKS and not analysis_reference_asset:
# Return automatic fail with score 0
fail_response = f"Reference asset is required for the '{check_name}' QC check but was not provided."
return {
'check_name': check_name,
'status': 'success',
'score': 0,
'result': 'Fail',
'response': fail_response,
'json_data': {},
'requires_brand_guidelines': False,
'brand_guidelines_status': None
}
check_prompt = qc_apps[check_name]['prompt']
llm_model = profile_config.get_check_llm(check_name)
# Handle brand guidelines if needed
requires_brand_guidelines = any(keyword in check_prompt.lower() for keyword in [
'brand guideline', 'brand standard', 'brand requirement', 'brand specification',
'brand compliance', 'brand rule', 'brand policy'
])
brand_guidelines_status = None
detected_brand = None
# Since we skip triage, we won't have detected_brand automatically
# Could be enhanced with direct brand detection if needed
if requires_brand_guidelines and detected_brand:
brand_guidelines = brand_db.get_brand_guidelines(detected_brand)
if brand_guidelines:
brand_guidelines_status = f"Brand guidelines found for {detected_brand} ({len(brand_guidelines)} files)."
# Add guidelines to prompt (simplified version)
check_prompt += f"\n\nBrand Analysis Context: Using brand guidelines for {detected_brand}."
else:
brand_guidelines_status = f"INFO: Brand detected as '{detected_brand}' but no brand guidelines found."
# Add pre-analysis instructions if available
final_prompt = check_prompt
if profile_config.pre_analysis_instructions:
final_prompt = profile_config.pre_analysis_instructions + "\n\n" + check_prompt
# Add reference asset content if selected
reference_image_path = None
if analysis_reference_asset:
reference_content = get_reference_asset_content(analysis_reference_asset)
if reference_content:
final_prompt = reference_content + "\n\n" + final_prompt
print(f"Added reference asset {analysis_reference_asset} to {check_name} prompt")
# Also get the actual reference image path for LLM
reference_image_path = get_reference_asset_image_path(analysis_reference_asset)
# Add media plan context if available
if media_plan_context:
final_prompt = final_prompt + "\n" + media_plan_context
# Add OCR measurement context for checks that evaluate spatial layout
try:
from ocr_measurement import OCR_RELEVANT_CHECKS as _ocr_checks
ocr_enabled = check_name in _ocr_checks
except ImportError:
ocr_enabled = False
if ocr_context and ocr_enabled:
final_prompt = final_prompt + "\n" + ocr_context
tech_report = progress_tracker.get(session_id, {}).get('technical_report')
if tech_report:
try:
final_prompt = technical_to_prompt(tech_report) + "\n\n" + final_prompt
except Exception:
pass # Pre-flight context is best-effort; never block the check on it.
print(f"Running check {check_index + 1}/{total_checks}: {check_name}")
result = run_visual_qc(
prompt=final_prompt,
asset_path=file_path,
reference_path=reference_image_path,
model_name=llm_model,
model_version=model_version
)
# Extract score and data
json_data = extract_json_from_response(result['response'])
score = extract_score_from_result(result, profile_config, check_name)
weight = profile_weights.get(check_name, 0.1)
weighted_score = score * weight if score is not None else 0
return {
'check_name': check_name,
'status': 'success',
'result': result,
'response': result['response'],
'brand_guidelines_status': brand_guidelines_status,
'requires_brand_guidelines': requires_brand_guidelines,
'json_data': json_data,
'score': score,
'weight': weight,
'weighted_score': weighted_score,
'model_used': result.get('model_info', {}),
'token_usage': result.get('token_usage', {}),
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'display_name': check_name.replace('_', ' ').title()
}
except Exception as e:
print(f"Error in check {check_name}: {str(e)}")
weight = profile_weights.get(check_name, 0.1)
return {
'check_name': check_name,
'status': 'error',
'error': str(e),
'weight': weight,
'score': 0,
'weighted_score': 0
}
def process_checks_in_batches(enabled_checks, qc_apps, profile_config, profile_weights,
file_path, analysis_reference_asset, brand_db, progress_tracker,
session_id, batch_size=15, media_plan_context=None, ocr_context=None):
"""Process QC checks in parallel batches"""
check_results = {}
total_checks = len(enabled_checks)
completed_checks = 0
print(f"Processing {total_checks} checks in batches of {batch_size}")
# Split checks into batches
for batch_start in range(0, total_checks, batch_size):
batch_end = min(batch_start + batch_size, total_checks)
batch_checks = enabled_checks[batch_start:batch_end]
batch_number = (batch_start // batch_size) + 1
total_batches = (total_checks + batch_size - 1) // batch_size
print(f"Processing batch {batch_number}/{total_batches} ({len(batch_checks)} checks)")
# Update progress for batch start
progress_tracker[session_id].update({
'current_batch': batch_number,
'total_batches': total_batches,
'current_check': f"Batch {batch_number}",
'current_check_display': f"Processing batch {batch_number}/{total_batches}",
'percentage': 10 + ((completed_checks / total_checks) * 80)
})
# Process batch in parallel
with ThreadPoolExecutor(max_workers=batch_size) as executor:
# Submit all checks in the batch
future_to_check = {}
for i, check_name in enumerate(batch_checks):
future = executor.submit(
process_single_check,
check_name, qc_apps, profile_config, profile_weights, file_path,
analysis_reference_asset, brand_db, progress_tracker, session_id,
batch_start + i, total_checks, None, media_plan_context, ocr_context
)
future_to_check[future] = check_name
# Collect results as they complete
batch_results = {}
for future in as_completed(future_to_check):
check_name = future_to_check[future]
try:
result = future.result()
batch_results[result['check_name']] = result
completed_checks += 1
# Update progress for each completed check
progress_tracker[session_id].update({
'completed_checks': completed_checks,
'percentage': 10 + ((completed_checks / total_checks) * 80)
})
print(f"Completed check: {check_name} ({completed_checks}/{total_checks})")
except Exception as e:
print(f"Error getting result for check {check_name}: {str(e)}")
weight = profile_weights.get(check_name, 0.1)
batch_results[check_name] = {
'check_name': check_name,
'status': 'error',
'error': str(e),
'weight': weight,
'score': 0,
'weighted_score': 0
}
completed_checks += 1
# Add batch results to main results
check_results.update(batch_results)
print(f"Completed batch {batch_number}/{total_batches}")
# Small delay between batches to avoid overwhelming the APIs
if batch_number < total_batches:
time.sleep(0.5)
print(f"Completed all {total_checks} checks in {total_batches} batches")
return check_results
def process_single_check_with_triage(check_name, qc_apps, profile_config, profile_weights, file_path,
reference_asset, brand_db, progress_tracker, session_id,
check_index, total_checks, model_version=None):
"""Process a single QC check with triage logic - designed to run in parallel"""
try:
# Deterministic file-naming check — short-circuit before any LLM dispatch.
if check_name == 'dj_file_naming':
return _run_dj_file_naming_check(check_name, file_path, profile_weights)
# Check if this check requires a reference asset but none is provided
if check_name in REFERENCE_ASSET_REQUIRED_CHECKS and not reference_asset:
# Return automatic fail with score 0
fail_response = f"Reference asset is required for the '{check_name}' QC check but was not provided."
return {
'check_name': check_name,
'status': 'success',
'score': 0,
'result': 'Fail',
'response': fail_response,
'json_data': {},
'requires_brand_guidelines': False,
'brand_guidelines_status': None
}
check_prompt = qc_apps[check_name]['prompt']
llm_model = profile_config.get_check_llm(check_name)
# Check if this check requires brand guidelines
requires_brand_guidelines = any(keyword in check_prompt.lower() for keyword in [
'brand guideline', 'brand standard', 'brand requirement', 'brand specification',
'brand compliance', 'brand rule', 'brand policy'
])
brand_guidelines_status = None
# Try to extract brand from triage results or detection
detected_brand = None
# Since we skip triage, we won't have detected_brand automatically
# Could be enhanced with direct brand detection if needed
if requires_brand_guidelines:
if detected_brand:
brand_guidelines = brand_db.get_brand_guidelines(detected_brand)
if not brand_guidelines:
brand_guidelines_status = f"INFO: Brand detected as '{detected_brand}' but no brand guidelines found. Performing generic brand analysis."
check_prompt += f"\n\nBrand Analysis Context: The detected brand is '{detected_brand}'. While specific brand guidelines are not available, please analyze the visual content for general brand consistency, professional appearance, and adherence to common branding best practices for this brand if you're familiar with it."
else:
brand_guidelines_status = f"Brand guidelines found for {detected_brand} ({len(brand_guidelines)} files)."
# Add brand guidelines content to the prompt
guidelines_content = "\n\n=== BRAND GUIDELINES REFERENCE ===\n"
guidelines_content += f"The following brand guidelines have been provided for {detected_brand}:\n\n"
for guideline in brand_guidelines:
guidelines_content += f"**Guideline File: {guideline.get('original_filename', 'Unknown')}**\n"
if guideline.get('description'):
guidelines_content += f"Description: {guideline['description']}\n"
if guideline.get('tags'):
guidelines_content += f"Tags: {', '.join(guideline['tags'])}\n"
# Try to read file content if it's a text-based file
guideline_file_path = guideline.get('file_path')
if guideline_file_path and os.path.exists(guideline_file_path):
try:
file_ext = os.path.splitext(guideline_file_path)[1].lower()
if file_ext in ['.txt', '.md', '.json']:
with open(guideline_file_path, 'r', encoding='utf-8') as f:
content = f.read()
if len(content) > 2000: # Limit content length
content = content[:2000] + "... [content truncated]"
guidelines_content += f"Content:\n{content}\n\n"
else:
guidelines_content += f"[File type {file_ext} - content not directly readable, but file is available as reference]\n\n"
except Exception as e:
guidelines_content += f"[Error reading file content: {str(e)}]\n\n"
else:
guidelines_content += "[File path not found]\n\n"
guidelines_content += "Please use these brand guidelines as reference when performing your analysis. Pay special attention to brand colors, fonts, logo usage, tone of voice, and any specific requirements mentioned in the guidelines.\n"
guidelines_content += "=== END BRAND GUIDELINES REFERENCE ===\n"
check_prompt += guidelines_content
else:
brand_guidelines_status = "INFO: Brand could not be determined. Performing generic analysis."
check_prompt += "\n\nGeneric Analysis: Since the brand could not be determined from the image, please analyze the visual content for general quality, professional appearance, and adherence to common design best practices."
# Add pre-analysis instructions if available
final_prompt = check_prompt
if profile_config.pre_analysis_instructions:
final_prompt = profile_config.pre_analysis_instructions + "\n\n" + check_prompt
# Add reference asset content if selected
reference_image_path = None
if reference_asset:
reference_content = get_reference_asset_content(reference_asset)
if reference_content:
final_prompt = reference_content + "\n\n" + final_prompt
print(f"Added reference asset {reference_asset} to {check_name} prompt")
# Also get the actual reference image path for LLM
reference_image_path = get_reference_asset_image_path(reference_asset)
print(f"Running check {check_index + 1}/{total_checks}: {check_name}")
result = run_visual_qc(
prompt=final_prompt,
asset_path=file_path,
reference_path=reference_image_path,
model_name=llm_model,
model_version=model_version
)
# Extract score and data
json_data = extract_json_from_response(result['response'])
score = extract_score_from_result(result, profile_config, check_name)
weight = profile_weights.get(check_name, 0.1)
weighted_score = score * weight if score is not None else 0
return {
'check_name': check_name,
'status': 'completed',
'response': result['response'],
'brand_guidelines_status': brand_guidelines_status,
'requires_brand_guidelines': requires_brand_guidelines,
'json_data': json_data,
'score': score,
'weight': weight,
'weighted_score': weighted_score,
'model_used': result.get('model_info', {}),
'token_usage': result.get('token_usage', {}),
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
except Exception as e:
print(f"Error in check {check_name}: {str(e)}")
weight = profile_weights.get(check_name, 0.1)
return {
'check_name': check_name,
'status': 'error',
'error': str(e),
'weight': weight,
'score': 0,
'weighted_score': 0
}
def process_checks_in_batches_with_triage(enabled_checks, qc_apps, profile_config, profile_weights,
file_path, reference_asset, brand_db, progress_tracker,
session_id, batch_size=15, base_percentage=10, percentage_range=80):
"""Process QC checks in parallel batches with triage logic"""
check_results = {}
total_checks = len(enabled_checks)
completed_checks = 0
print(f"Processing {total_checks} checks in batches of {batch_size}")
# Split checks into batches
for batch_start in range(0, total_checks, batch_size):
batch_end = min(batch_start + batch_size, total_checks)
batch_checks = enabled_checks[batch_start:batch_end]
batch_number = (batch_start // batch_size) + 1
total_batches = (total_checks + batch_size - 1) // batch_size
print(f"Processing batch {batch_number}/{total_batches} ({len(batch_checks)} checks)")
# Update progress for batch start
progress_tracker[session_id].update({
'current_batch': batch_number,
'total_batches': total_batches,
'current_check': f"Batch {batch_number}",
'current_check_display': f"Processing batch {batch_number}/{total_batches}",
'percentage': base_percentage + ((completed_checks / total_checks) * percentage_range)
})
# Process batch in parallel
with ThreadPoolExecutor(max_workers=batch_size) as executor:
# Submit all checks in the batch
future_to_check = {}
for i, check_name in enumerate(batch_checks):
future = executor.submit(
process_single_check_with_triage,
check_name, qc_apps, profile_config, profile_weights, file_path,
reference_asset, brand_db, progress_tracker, session_id,
batch_start + i, total_checks
)
future_to_check[future] = check_name
# Collect results as they complete
batch_results = {}
for future in as_completed(future_to_check):
check_name = future_to_check[future]
try:
result = future.result()
batch_results[result['check_name']] = result
completed_checks += 1
# Update progress for each completed check
progress_tracker[session_id].update({
'completed_checks': completed_checks,
'percentage': base_percentage + ((completed_checks / total_checks) * percentage_range)
})
print(f"Completed check: {check_name} ({completed_checks}/{total_checks})")
except Exception as e:
print(f"Error getting result for check {check_name}: {str(e)}")
weight = profile_weights.get(check_name, 0.1)
batch_results[check_name] = {
'check_name': check_name,
'status': 'error',
'error': str(e),
'weight': weight,
'score': 0,
'weighted_score': 0
}
completed_checks += 1
# Add batch results to main results
check_results.update(batch_results)
print(f"Completed batch {batch_number}/{total_batches}")
# Small delay between batches to avoid overwhelming the APIs
if batch_number < total_batches:
time.sleep(0.5)
print(f"Completed all {total_checks} checks in {total_batches} batches")
return check_results
def create_thumbnail_base64(file_path, max_size=(300, 300)):
"""Create a base64 encoded thumbnail of the input file"""
try:
# Get the image using the existing utility
pil_image = get_image_from_asset(file_path)
if not pil_image:
return None
# Create thumbnail
thumbnail = pil_image.copy()
thumbnail.thumbnail(max_size, Image.Resampling.LANCZOS)
# Convert to base64
buffer = io.BytesIO()
# Convert to RGB if necessary (for PNG with transparency)
if thumbnail.mode in ('RGBA', 'LA'):
background = Image.new('RGB', thumbnail.size, (255, 255, 255))
background.paste(thumbnail, mask=thumbnail.split()[-1] if thumbnail.mode == 'RGBA' else None)
thumbnail = background
thumbnail.save(buffer, format='JPEG', quality=85)
img_str = base64.b64encode(buffer.getvalue()).decode()
return f"data:image/jpeg;base64,{img_str}"
except Exception as e:
print(f"Error creating thumbnail: {e}")
return None
def get_client_from_profile(profile_id):
"""Determine client from profile ID"""
if not profile_id:
return 'general'
profile_lower = profile_id.lower()
if profile_lower.startswith('loreal'):
return 'loreal'
elif profile_lower.startswith('diageo'):
return 'diageo'
elif profile_lower.startswith('unilever'):
return 'unilever'
elif profile_lower.startswith('amazon'):
return 'amazon'
elif profile_lower.startswith('boots'):
return 'boots'
elif profile_lower.startswith('hp_'):
return 'hp'
elif profile_lower.startswith(('dow_jones', 'dj_', 'marketwatch', 'mw_', 'wsj')):
return 'dow_jones'
else:
return 'general'
def ensure_client_output_folder(client):
"""Ensure client-specific output folder exists"""
client_folder = os.path.join(app.config['OUTPUT_FOLDER'], client)
os.makedirs(client_folder, exist_ok=True)
return client_folder
def cleanup_old_files(max_age_days=14):
"""Delete files older than max_age_days from all client folders"""
import time
cutoff_time = time.time() - (max_age_days * 24 * 60 * 60)
deleted_count = 0
try:
output_folder = app.config['OUTPUT_FOLDER']
# Check root output folder
for item in os.listdir(output_folder):
item_path = os.path.join(output_folder, item)
if os.path.isdir(item_path):
# Check files in client subfolders
for filename in os.listdir(item_path):
file_path = os.path.join(item_path, filename)
if os.path.isfile(file_path):
if os.path.getctime(file_path) < cutoff_time:
os.remove(file_path)
deleted_count += 1
print(f"Deleted old file: {file_path}")
elif os.path.isfile(item_path):
# Check files in root output folder (for migration)
if os.path.getctime(item_path) < cutoff_time:
os.remove(item_path)
deleted_count += 1
print(f"Deleted old file: {item_path}")
if deleted_count > 0:
print(f"Cleaned up {deleted_count} files older than {max_age_days} days")
except Exception as e:
print(f"Error during cleanup: {e}")
return deleted_count
def save_results_to_file(report_data, filename, output_mode='html', session_id=None, file_path=None):
"""Save analysis results to file and return file path"""
print(f"DEBUG: save_results_to_file called with output_mode: '{output_mode}'")
if not session_id:
session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
# Determine client from profile
profile_id = report_data.get('profile_id') or report_data.get('profiles', [None])[0]
client = get_client_from_profile(profile_id)
print(f"DEBUG: Detected client '{client}' from profile '{profile_id}'")
# Ensure client-specific folder exists
client_folder = ensure_client_output_folder(client)
# Create filename base
base_filename = f"{session_id}_{filename.replace(' ', '_')}"
if output_mode == 'html':
print(f"DEBUG: Creating HTML file because output_mode == 'html'")
# Save HTML file in client-specific folder
output_filename = f"{base_filename}_report.html"
output_path = os.path.join(client_folder, output_filename)
html_content = generate_html_content(report_data, filename, file_path)
with open(output_path, 'w', encoding='utf-8') as f:
f.write(html_content)
return output_path
else:
print(f"DEBUG: Creating JSON file because output_mode != 'html' (it's '{output_mode}')")
# Save JSON file in client-specific folder
output_filename = f"{base_filename}_data.json"
output_path = os.path.join(client_folder, output_filename)
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(report_data, f, indent=2, ensure_ascii=False)
return output_path
def generate_html_content(report_data, filename, file_path=None):
"""Generate HTML content for report data with expandable sections"""
# Define a function to get color based on score
def get_score_result(score):
if score >= 6:
return "Pass", "#28a745" # Green for pass
else:
return "Fail", "#dc3545" # Red for fail
# Get reference asset information from profile selection
profile_selection = report_data.get('profile_selection', {})
reference_asset = profile_selection.get('reference_asset', None)
reference_asset_used = profile_selection.get('reference_asset_used', False)
# Build HTML for each check result with expandable sections
check_results_html = ""
for check_name, check_data in report_data['results'].items():
if check_data['status'] == 'success':
display_name = check_data.get('display_name', check_name)
score = check_data.get('score', 0)
result_text, score_color = get_score_result(score)
weight = check_data.get('weight', 0)
weighted_score = check_data.get('weighted_score', 0)
# Extract response text - try to get detailed info from JSON data first
json_data = check_data.get('json_data', {})
response_text = ""
# Structured findings (e.g. hp_copy_review) render as a table
# instead of the default response-text block. If absent, falls
# back to the existing text rendering below.
findings = (json_data or {}).get('findings') if isinstance(json_data, dict) else None
findings_html = _render_findings_table(findings) if findings is not None else None
# Try to extract detailed analysis from JSON data
if json_data:
# Look for common detailed fields in the JSON
analysis_details = json_data.get('analysis_details', '')
explanation = json_data.get('explanation', '')
issues_found = json_data.get('issues_found', [])
recommendations = json_data.get('recommendations', '')
elements_checked = json_data.get('elements_checked', {})
elements_found = json_data.get('elements_found', {})
marketing_text_found = json_data.get('marketing_text_found', [])
recommended_adjustments = json_data.get('recommended_adjustments', '')
# Build detailed response text from JSON fields
if analysis_details:
response_text += f"<strong>Analysis:</strong><br>{analysis_details}<br><br>"
elif explanation:
response_text += f"<strong>Analysis:</strong><br>{explanation}<br><br>"
if elements_checked:
response_text += "<strong>Elements Checked:</strong><br>"
for element, status in elements_checked.items():
response_text += f"• <em>{element.replace('_', ' ').title()}:</em> {status}<br>"
response_text += "<br>"
elif elements_found:
response_text += "<strong>Elements Found:</strong><br>"
for element, found in elements_found.items():
status_icon = "Present" if found else "Missing"
response_text += f"• <em>{element.replace('_', ' ').title()}:</em> {status_icon}<br>"
response_text += "<br>"
if marketing_text_found:
response_text += "<strong>Marketing Text Found:</strong><br>"
for text in marketing_text_found:
response_text += f"• {text}<br>"
response_text += "<br>"
if issues_found:
response_text += "<strong>Issues Found:</strong><br>"
if isinstance(issues_found, list):
for issue in issues_found:
response_text += f"• {issue}<br>"
else:
response_text += f"{issues_found}<br>"
response_text += "<br>"
if recommendations:
if isinstance(recommendations, list):
response_text += "<strong>Recommendations:</strong><br>"
for rec in recommendations:
response_text += f"• {rec}<br>"
else:
response_text += f"<strong>Recommendation:</strong><br>{recommendations}<br>"
if recommended_adjustments:
if isinstance(recommended_adjustments, list):
response_text += "<br><strong>Suggested Adjustments:</strong><br>"
for adj in recommended_adjustments:
response_text += f"• {adj}<br>"
elif isinstance(recommended_adjustments, str) and recommended_adjustments.lower() not in ['none', 'n/a']:
response_text += f"<br><strong>Suggested Adjustments:</strong><br>{recommended_adjustments}"
# If still no response text from known fields, build summary from all JSON data
if not response_text and json_data:
skip_keys = {'score', 'weight', 'weighted_score', 'status'}
summary_parts = []
for key, value in json_data.items():
if key in skip_keys:
continue
display_key = key.replace('_', ' ').title()
if isinstance(value, bool):
summary_parts.append(f"• <em>{display_key}:</em> {'Yes' if value else 'No'}")
elif isinstance(value, dict):
summary_parts.append(f"<strong>{display_key}:</strong>")
for sub_key, sub_val in value.items():
sub_display = sub_key.replace('_', ' ').title()
if isinstance(sub_val, bool):
summary_parts.append(f"&nbsp;&nbsp;• <em>{sub_display}:</em> {'Yes' if sub_val else 'No'}")
else:
summary_parts.append(f"&nbsp;&nbsp;• <em>{sub_display}:</em> {sub_val}")
elif isinstance(value, list):
if value:
summary_parts.append(f"<strong>{display_key}:</strong>")
for item in value:
summary_parts.append(f"&nbsp;&nbsp;• {item}")
elif isinstance(value, str) and value and value.lower() not in ['none', 'n/a', '']:
summary_parts.append(f"• <em>{display_key}:</em> {value}")
elif isinstance(value, (int, float)):
summary_parts.append(f"• <em>{display_key}:</em> {value}")
if summary_parts:
response_text = "<br>".join(summary_parts)
# If no detailed JSON data was found, fall back to original response
if not response_text:
response_text = check_data.get('response', '')
# Remove JSON code blocks for cleaner reading
response_text = re.sub(r'```json.*?```', '', response_text, flags=re.DOTALL)
response_text = response_text.strip()
# If still empty, provide a basic fallback
if not response_text:
score = check_data.get('score', 0)
result_text, _ = get_score_result(score)
if score == 0 and check_name in REFERENCE_ASSET_REQUIRED_CHECKS:
response_text = f"Reference asset is required for the '{display_name}' QC check but was not provided."
else:
response_text = f"QC check result: {result_text} (Score: {score}/10)"
# Create expandable section for each check
check_results_html += f"""
<div class="expandable-section">
<div class="expandable-header" onclick="toggleSection('{check_name}')">
<div class="check-title">
<h3>{display_name}</h3>
<span class="score-badge" style="background-color: {score_color};">{result_text}</span>
</div>
<div class="chevron" id="chevron-{check_name}">▼</div>
</div>
<div class="expandable-content" id="content-{check_name}">
<div class="check-metadata">
<p><strong>Score:</strong> {score}/10 | <strong>Weight:</strong> {weight:.1%} | <strong>Weighted Score:</strong> {weighted_score:.2f}</p>
{'<p><strong>⭐ Bonus Check:</strong> If missing required element, this scores 0</p>' if check_name in ['face_gaze_direction', 'face_visibility', 'new_visibility'] else ''}
<p><strong>Reference Asset:</strong> {'✅ Used' if check_name in REFERENCE_ASSET_REQUIRED_CHECKS and reference_asset_used else ('🚨 Required but missing' if check_name in REFERENCE_ASSET_REQUIRED_CHECKS else '➖ Not required')}</p>
{f'<p><strong>Reference Asset Details:</strong> {reference_asset}</p>' if check_name in REFERENCE_ASSET_REQUIRED_CHECKS and reference_asset_used and reference_asset else ''}
</div>
<div class="analysis-section">
<h4>Analysis Details:</h4>
{f'<div class="response-text">{html.escape(json_data.get("summary", "") or "") if isinstance(json_data, dict) else ""}</div>{findings_html}' if findings_html is not None else f'<div class="response-text">{response_text.replace(chr(10), "<br>")}</div>'}
</div>
</div>
</div>
"""
# Get summary score result
overall_score = report_data['summary']['overall_score']
overall_result, overall_color = get_score_result(overall_score/10) # Normalize to 0-10 scale
# Determine the correct total score based on profile
profile_id = report_data.get('profile_id', '')
if profile_id == 'unilever_key_visual':
score_total = 120
else:
score_total = 100
technical_html = _render_technical_section_html(report_data.get('technical_report', {}))
html_content = f"""
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Visual AI QC Results for {filename}</title>
<link rel="preconnect" href="https://fonts.googleapis.com">
<link rel="preconnect" href="https://fonts.gstatic.com" crossorigin>
<link href="https://fonts.googleapis.com/css2?family=Montserrat:wght@400;500;600;700&display=swap" rel="stylesheet">
<style>
body {{ font-family: 'Montserrat', -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; margin: 20px; padding: 0; background: linear-gradient(135deg, #1a1a1a 0%, #000000 100%); min-height: 100vh; }}
.container {{ max-width: 1200px; margin: 0 auto; background-color: white; padding: 30px; border-radius: 20px; box-shadow: 0 20px 40px rgba(0,0,0,0.1); }}
h1, h2, h3 {{ color: #2c3e50; }}
h1 {{ margin-bottom: 10px; font-size: 2.5em; text-align: center; }}
.file-preview {{ background: linear-gradient(135deg, #f8f9fa 0%, #e9ecef 100%); padding: 20px; border-radius: 15px; margin: 20px 0; border-left: 5px solid #FFC407; }}
.file-info {{ display: flex; align-items: center; gap: 20px; }}
.thumbnail {{ max-width: 150px; max-height: 150px; border-radius: 8px; box-shadow: 0 4px 8px rgba(0,0,0,0.1); border: 2px solid #dee2e6; }}
.file-details {{ flex: 1; }}
.filename {{ font-size: 1.2em; font-weight: bold; color: #495057; margin-bottom: 5px; word-break: break-all; overflow-wrap: break-word; }}
.file-meta {{ color: #6c757d; font-size: 0.9em; }}
.summary {{ background: linear-gradient(135deg, #FFF9E6 0%, #FFFBF0 100%); padding: 25px; border-radius: 15px; margin: 30px 0; border-left: 5px solid #FFC407; }}
.summary-grid {{ display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 20px; margin-top: 15px; }}
.summary-item {{ background: white; padding: 15px; border-radius: 10px; text-align: center; box-shadow: 0 2px 4px rgba(0,0,0,0.05); }}
.technical {{ background: linear-gradient(135deg, #e3f2fd 0%, #f0f7ff 100%); padding: 25px; border-radius: 15px; margin: 30px 0; border-left: 5px solid #1565c0; }}
.technical-grid {{ display: grid; grid-template-columns: repeat(auto-fit, minmax(280px, 1fr)); gap: 8px 24px; margin-top: 12px; }}
.tech-row {{ padding: 4px 0; color: #495057; font-size: 0.95em; word-break: break-word; }}
.score-display {{ font-size: 2.5em; font-weight: bold; color: {overall_color}; margin-bottom: 5px; }}
.grade {{ font-size: 1.3em; font-weight: bold; color: #495057; }}
.expandable-section {{ margin-bottom: 15px; border: 2px solid #e9ecef; border-radius: 12px; overflow: hidden; background: white; }}
.expandable-header {{ background: #f8f9fa; padding: 20px; cursor: pointer; display: flex; justify-content: space-between; align-items: center; transition: background-color 0.3s ease; }}
.expandable-header:hover {{ background: #e9ecef; }}
.check-title {{ display: flex; align-items: center; gap: 15px; }}
.check-title h3 {{ margin: 0; color: #495057; }}
.score-badge {{ padding: 8px 12px; border-radius: 20px; color: white; font-weight: bold; font-size: 0.9em; }}
.chevron {{ font-size: 1.2em; transition: transform 0.3s ease; color: #FFC407; }}
.chevron.expanded {{ transform: rotate(180deg); }}
.expandable-content {{ padding: 0; max-height: 0; overflow: hidden; transition: all 0.3s ease; }}
.expandable-content.expanded {{ padding: 20px; max-height: 1000px; }}
.check-metadata {{ background: #f8f9fa; padding: 15px; border-radius: 8px; margin-bottom: 15px; }}
.analysis-section h4 {{ color: #495057; margin-bottom: 10px; }}
.response-text {{ background: #f8f9fa; padding: 15px; border-radius: 8px; line-height: 1.6; font-family: 'Montserrat', Georgia, serif; }}
.json-toggle {{ cursor: pointer; color: #FFC407; text-decoration: underline; padding: 15px; text-align: center; font-weight: bold; }}
.json-view {{ display: none; margin-top: 20px; }}
.json-view pre {{ background-color: #2d3748; color: #e2e8f0; padding: 20px; border-radius: 10px; overflow-x: auto; font-size: 0.9em; }}
.findings-table {{ width: 100%; border-collapse: collapse; margin-top: 12px; font-size: 0.92em; }}
.findings-table th {{ background: #f1f3f5; color: #495057; text-align: left; padding: 8px 10px; border-bottom: 2px solid #dee2e6; font-weight: 600; }}
.findings-table td {{ padding: 8px 10px; border-bottom: 1px solid #e9ecef; vertical-align: top; word-break: break-word; }}
.findings-table tr:last-child td {{ border-bottom: none; }}
.findings-table code {{ background: #f8f9fa; padding: 2px 5px; border-radius: 4px; font-family: 'SFMono-Regular', Consolas, Menlo, monospace; font-size: 0.9em; color: #c7254e; }}
.priority-pill {{ display: inline-block; padding: 3px 8px; border-radius: 10px; color: white; font-weight: 600; font-size: 0.78em; letter-spacing: 0.03em; }}
.priority-high {{ background-color: #dc3545; }}
.priority-medium {{ background-color: #fd7e14; }}
.priority-low {{ background-color: #28a745; }}
.muted {{ color: #6c757d; font-size: 0.9em; }}
</style>
</head>
<body>
<div class="container">
<h1>🤖 Visual AI QC Results</h1>
<p style="text-align: center; color: #6c757d; font-size: 1.1em;">
Analysis completed on: {report_data['timestamp']}
</p>
<div class="file-preview">
<h3>📎 Analyzed File</h3>
<div class="file-info">
<img src="{create_thumbnail_base64(file_path) if file_path else 'data:image/svg+xml;base64,PHN2ZyB3aWR0aD0iMTUwIiBoZWlnaHQ9IjE1MCIgdmlld0JveD0iMCAwIDE1MCAxNTAiIGZpbGw9Im5vbmUiIHhtbG5zPSJodHRwOi8vd3d3LnczLm9yZy8yMDAwL3N2ZyI+CjxyZWN0IHdpZHRoPSIxNTAiIGhlaWdodD0iMTUwIiBmaWxsPSIjZjhmOWZhIiBzdHJva2U9IiNkZWUyZTYiIHN0cm9rZS13aWR0aD0iMiIvPgo8dGV4dCB4PSI3NSIgeT0iNzUiIHRleHQtYW5jaG9yPSJtaWRkbGUiIGZpbGw9IiM2Yzc1N2QiIGZvbnQtZmFtaWx5PSJBcmlhbCIgZm9udC1zaXplPSIxNCIgZHk9IjAuMzVlbSI+4p2MIEZpbGU8L3RleHQ+Cjx0ZXh0IHg9Ijc1IiB5PSI5NSIgdGV4dC1hbmNob3I9Im1pZGRsZSIgZmlsbD0iIzZjNzU3ZCIgZm9udC1mYW1pbHk9IkFyaWFsIiBmb250LXNpemU9IjEyIiBkeT0iMC4zNWVtIj5UaHVtYm5haWw8L3RleHQ+Cjwvc3ZnPg=='}" alt="File thumbnail" class="thumbnail" id="fileThumbnail">
<div class="file-details">
<div class="filename">{filename}</div>
<div class="file-meta">Original file processed for quality control analysis</div>
</div>
</div>
</div>
<div class="summary">
<h2>📊 Analysis Summary</h2>
<div class="summary-grid">
<div class="summary-item">
<div class="score-display">{overall_score}/{score_total}</div>
<div>Overall Score</div>
</div>
<div class="summary-item">
<div class="grade">{report_data['summary']['grade']}</div>
<div>Grade</div>
</div>
<div class="summary-item">
<div style="font-size: 1.5em; font-weight: bold; color: #495057;">{report_data['summary']['checks_count']}</div>
<div>Checks Performed</div>
</div>
<div class="summary-item">
<div style="font-size: 1.2em; color: #FFC407; font-weight: bold;">{report_data['profile_name']}</div>
<div>Profile Used</div>
</div>
<div class="summary-item">
<div style="font-size: 1.2em; color: {'#28a745' if reference_asset_used else '#6c757d'}; font-weight: bold;">{'✅ Used' if reference_asset_used else '➖ None'}</div>
<div>Reference Asset</div>
</div>
</div>
</div>
{technical_html}
<h2>🔍 Detailed Analysis Results</h2>
<p style="color: #6c757d; margin-bottom: 20px; font-style: italic;">
Click on any section below to expand and view detailed analysis
</p>
{check_results_html}
<div class="json-toggle" onclick="document.getElementById('json-data').style.display = document.getElementById('json-data').style.display === 'none' ? 'block' : 'none';">
📄 Show/Hide Raw JSON Data
</div>
<div id="json-data" class="json-view">
<pre>{json.dumps(report_data, indent=2)}</pre>
</div>
</div>
<script>
function toggleSection(checkName) {{
const content = document.getElementById('content-' + checkName);
const chevron = document.getElementById('chevron-' + checkName);
if (content.classList.contains('expanded')) {{
content.classList.remove('expanded');
chevron.classList.remove('expanded');
}} else {{
content.classList.add('expanded');
chevron.classList.add('expanded');
}}
}}
</script>
</body>
</html>
"""
return html_content
def generate_html_response(report_data, filename, save_to_file=False, session_id=None, file_path=None):
"""Generate HTML response for report data with optional file saving"""
html_content = generate_html_content(report_data, filename, file_path)
if save_to_file:
# Save to file and return file path info
output_path = save_results_to_file(report_data, filename, 'html', session_id, file_path)
return Response(html_content, mimetype='text/html'), output_path
else:
return Response(html_content, mimetype='text/html')
def _render_findings_table(findings):
"""Render an hp_copy_review-style findings array as an HTML table.
Each finding dict is expected to carry: priority (high|medium|low),
category, quote, issue, suggested_fix, source_reference. All string
fields are HTML-escaped before interpolation. An empty/None findings
list renders a friendly "clean copy" note instead of an empty table.
"""
if not findings:
return '<p class="muted">No findings — copy is clean.</p>'
rows = []
for f in findings:
priority = (f.get('priority') or 'low').lower()
pri_class = {
'high': 'priority-high',
'medium': 'priority-medium',
'low': 'priority-low',
}.get(priority, 'priority-low')
quote_raw = (f.get('quote') or '')[:200]
rows.append(
'<tr>'
f'<td><span class="priority-pill {pri_class}">{html.escape(priority.upper())}</span></td>'
f'<td><code>{html.escape(f.get("category", "") or "")}</code></td>'
f'<td><code>{html.escape(quote_raw)}</code></td>'
f'<td>{html.escape(f.get("issue", "") or "")}</td>'
f'<td>{html.escape(f.get("suggested_fix", "") or "")}</td>'
f'<td class="muted">{html.escape(f.get("source_reference", "") or "")}</td>'
'</tr>'
)
return (
'<table class="findings-table"><thead><tr>'
'<th>Priority</th><th>Category</th><th>Quote</th>'
'<th>Issue</th><th>Suggested fix</th><th>Source</th>'
'</tr></thead><tbody>'
+ ''.join(rows) +
'</tbody></table>'
)
def _render_technical_section_html(report):
"""Render the technical pre-flight report as an HTML block. Empty string if no report."""
if not report or report.get('kind') in (None, 'unknown'):
return ''
kind = report['kind']
rows = []
size_mb = report.get('file_size_mb')
if size_mb is not None:
rows.append(f'<div class="tech-row"><strong>File size:</strong> {size_mb} MB</div>')
dims = report.get('dimensions')
if dims:
rows.append(f'<div class="tech-row"><strong>Dimensions:</strong> {dims["width"]} × {dims["height"]}</div>')
fmt = report.get('format')
if fmt:
rows.append(f'<div class="tech-row"><strong>Format:</strong> {fmt}</div>')
dpi = report.get('dpi')
if dpi:
rows.append(f'<div class="tech-row"><strong>DPI:</strong> {dpi[0]} × {dpi[1]}</div>')
mode = report.get('mode')
if mode:
rows.append(f'<div class="tech-row"><strong>Color mode:</strong> {mode}</div>')
pc = report.get('page_count')
if pc is not None:
rows.append(f'<div class="tech-row"><strong>Pages:</strong> {pc}</div>')
pdf_ver = report.get('pdf_version')
if pdf_ver:
rows.append(f'<div class="tech-row"><strong>PDF version:</strong> {pdf_ver}</div>')
duration = report.get('duration_seconds')
if duration is not None:
rows.append(f'<div class="tech-row"><strong>Duration:</strong> {duration}s</div>')
codec = report.get('video_codec')
if codec:
rows.append(f'<div class="tech-row"><strong>Video codec:</strong> {codec}</div>')
fps = report.get('fps')
if fps:
rows.append(f'<div class="tech-row"><strong>Frame rate:</strong> {fps} fps</div>')
fonts = report.get('embedded_fonts')
if fonts:
suffix = ' …' if len(fonts) > 8 else ''
rows.append(f'<div class="tech-row"><strong>Embedded fonts:</strong> {", ".join(fonts[:8])}{suffix}</div>')
fm = report.get('filename_match')
if fm:
if fm['match']:
badge = '<span style="background:#28a745;color:white;padding:4px 10px;border-radius:12px;font-size:0.85em;">✓ Matches filename</span>'
else:
badge = '<span style="background:#dc3545;color:white;padding:4px 10px;border-radius:12px;font-size:0.85em;">⚠ Filename mismatch</span>'
rows.append(f'<div class="tech-row" style="margin-top:8px;">{badge} <span style="color:#6c757d;font-size:0.9em;margin-left:8px;">{fm["detail"]}</span></div>')
errors = report.get('errors', [])
if errors:
rows.append(f'<div class="tech-row" style="color:#856404;font-style:italic;"><strong>Inspection notes:</strong> {"; ".join(errors)}</div>')
if not rows:
return ''
return f'''
<div class="technical">
<h2>🔧 Technical Details <small style="color:#6c757d;font-size:0.6em;font-weight:normal;">(machine-inspected, no AI)</small></h2>
<div class="technical-grid">{''.join(rows)}</div>
</div>
'''
def generate_comprehensive_html_report(analysis_result, filename, file_path=None):
"""Generate comprehensive HTML report similar to the web UI format"""
summary = analysis_result.get('summary', {})
qc_analysis = analysis_result.get('qc_analysis', {})
profile_selection = analysis_result.get('profile_selection', {})
check_results = qc_analysis.get('check_results', {})
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
overall_score = summary.get('overall_score', 0)
profile_name = profile_selection.get('suggested_profile', 'Unknown Profile')
total_checks = qc_analysis.get('total_checks', 0)
completed_checks = qc_analysis.get('completed_checks', 0)
reference_asset = profile_selection.get('reference_asset', None)
reference_asset_used = profile_selection.get('reference_asset_used', False)
# Generate check results HTML
check_results_html = ''
for check_name, result in check_results.items():
if result.get('status') in ('success', 'completed'):
score = result.get('score', 0)
result_text = "Pass" if score >= 6 else "Fail"
score_color = '#28a745' if score >= 6 else '#dc3545'
response = result.get('response', 'No response available')
display_name = check_name.replace('_', ' ').replace(chr(32).join([w.capitalize() for w in check_name.split('_')]), check_name.replace('_', ' ').title())
# Structured findings (e.g. hp_copy_review) render as a table
# instead of the default response-text block. Fallback to the
# existing response rendering when 'findings' is absent.
json_data = result.get('json_data') if isinstance(result, dict) else None
findings = json_data.get('findings') if isinstance(json_data, dict) else None
findings_html = _render_findings_table(findings) if findings is not None else None
# Remove JSON blocks for cleaner display and handle empty responses
response = re.sub(r'```json.*?```', '', response, flags=re.DOTALL).strip()
if not response:
if score == 0 and check_name in REFERENCE_ASSET_REQUIRED_CHECKS:
response = f"Reference asset is required for the '{display_name}' QC check but was not provided."
else:
response = f"QC check result: {result_text} (Score: {score}/10)"
check_results_html += f'''
<div class="expandable-section">
<div class="expandable-header" onclick="toggleSection('{check_name}')">
<div class="check-title">
<h3>{display_name}</h3>
<span class="score-badge" style="background-color: {score_color};">{result_text}</span>
</div>
<div class="chevron" id="chevron-{check_name}">▼</div>
</div>
<div class="expandable-content" id="content-{check_name}">
<div class="check-metadata">
<p><strong>Score:</strong> {score}/10 | <strong>Weight:</strong> {result.get('weight', 0):.1%} | <strong>Weighted Score:</strong> {result.get('weighted_score', 0):.2f}</p>
{'<p><strong>⭐ Bonus Check:</strong> If missing required element, this scores 0</p>' if check_name in ['face_gaze_direction', 'face_visibility', 'new_visibility'] else ''}
<p><strong>Reference Asset:</strong> {'✅ Used' if check_name in REFERENCE_ASSET_REQUIRED_CHECKS and reference_asset_used else ('🚨 Required but missing' if check_name in REFERENCE_ASSET_REQUIRED_CHECKS else '➖ Not required')}</p>
{f'<p><strong>Reference Asset Details:</strong> {reference_asset}</p>' if check_name in REFERENCE_ASSET_REQUIRED_CHECKS and reference_asset_used and reference_asset else ''}
</div>
<div class="analysis-section">
<h4>Analysis Details:</h4>
{f'<div class="response-text">{html.escape(json_data.get("summary", "") or "") if isinstance(json_data, dict) else ""}</div>{findings_html}' if findings_html is not None else f'<div class="response-text">{response.replace(chr(10), "<br>")}</div>'}
</div>
</div>
</div>
'''
# Convert overall score to pass/fail based on average of individual check scores
avg_individual_score = overall_score / 10 # Normalize to 1-10 scale
grade_text = 'Pass' if avg_individual_score >= 6 else 'Fail'
score_color = '#28a745' if avg_individual_score >= 6 else '#dc3545'
technical_html = _render_technical_section_html(analysis_result.get('technical_report', {}))
return f'''<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Visual AI QC Results for {filename}</title>
<link rel="preconnect" href="https://fonts.googleapis.com">
<link rel="preconnect" href="https://fonts.gstatic.com" crossorigin>
<link href="https://fonts.googleapis.com/css2?family=Montserrat:wght@400;500;600;700&display=swap" rel="stylesheet">
<style>
body {{ font-family: 'Montserrat', -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif; margin: 20px; padding: 0; background: linear-gradient(135deg, #1a1a1a 0%, #000000 100%); min-height: 100vh; }}
.container {{ max-width: 1200px; margin: 0 auto; background-color: white; padding: 30px; border-radius: 20px; box-shadow: 0 20px 40px rgba(0,0,0,0.1); }}
h1, h2, h3 {{ color: #2c3e50; }}
h1 {{ margin-bottom: 10px; font-size: 2.5em; text-align: center; }}
.file-preview {{ background: linear-gradient(135deg, #f8f9fa 0%, #e9ecef 100%); padding: 20px; border-radius: 15px; margin: 20px 0; border-left: 5px solid #FFC407; }}
.file-info {{ display: flex; align-items: center; gap: 20px; }}
.thumbnail {{ max-width: 150px; max-height: 150px; border-radius: 8px; box-shadow: 0 4px 8px rgba(0,0,0,0.1); border: 2px solid #dee2e6; }}
.file-details {{ flex: 1; }}
.filename {{ font-size: 1.2em; font-weight: bold; color: #495057; margin-bottom: 5px; word-break: break-all; overflow-wrap: break-word; }}
.file-meta {{ color: #6c757d; font-size: 0.9em; }}
.summary {{ background: linear-gradient(135deg, #FFF9E6 0%, #FFFBF0 100%); padding: 25px; border-radius: 15px; margin: 30px 0; border-left: 5px solid #FFC407; }}
.summary-grid {{ display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 20px; margin-top: 15px; }}
.summary-item {{ background: white; padding: 15px; border-radius: 10px; text-align: center; box-shadow: 0 2px 4px rgba(0,0,0,0.05); }}
.technical {{ background: linear-gradient(135deg, #e3f2fd 0%, #f0f7ff 100%); padding: 25px; border-radius: 15px; margin: 30px 0; border-left: 5px solid #1565c0; }}
.technical-grid {{ display: grid; grid-template-columns: repeat(auto-fit, minmax(280px, 1fr)); gap: 8px 24px; margin-top: 12px; }}
.tech-row {{ padding: 4px 0; color: #495057; font-size: 0.95em; word-break: break-word; }}
.score-display {{ font-size: 2.5em; font-weight: bold; color: {score_color}; margin-bottom: 5px; }}
.grade {{ font-size: 1.3em; font-weight: bold; color: #495057; }}
.expandable-section {{ margin-bottom: 15px; border: 2px solid #e9ecef; border-radius: 12px; overflow: hidden; background: white; }}
.expandable-header {{ background: #f8f9fa; padding: 20px; cursor: pointer; display: flex; justify-content: space-between; align-items: center; transition: background-color 0.3s ease; }}
.expandable-header:hover {{ background: #e9ecef; }}
.check-title {{ display: flex; align-items: center; gap: 15px; }}
.check-title h3 {{ margin: 0; color: #495057; }}
.score-badge {{ padding: 8px 12px; border-radius: 20px; color: white; font-weight: bold; font-size: 0.9em; }}
.chevron {{ font-size: 1.2em; transition: transform 0.3s ease; color: #FFC407; }}
.chevron.expanded {{ transform: rotate(180deg); }}
.expandable-content {{ padding: 0; max-height: 0; overflow: hidden; transition: all 0.3s ease; }}
.expandable-content.expanded {{ padding: 20px; max-height: 1000px; }}
.check-metadata {{ background: #f8f9fa; padding: 15px; border-radius: 8px; margin-bottom: 15px; }}
.analysis-section h4 {{ color: #495057; margin-bottom: 10px; }}
.response-text {{ background: #f8f9fa; padding: 15px; border-radius: 8px; line-height: 1.6; font-family: 'Montserrat', Georgia, serif; }}
.findings-table {{ width: 100%; border-collapse: collapse; margin-top: 12px; font-size: 0.92em; }}
.findings-table th {{ background: #f1f3f5; color: #495057; text-align: left; padding: 8px 10px; border-bottom: 2px solid #dee2e6; font-weight: 600; }}
.findings-table td {{ padding: 8px 10px; border-bottom: 1px solid #e9ecef; vertical-align: top; word-break: break-word; }}
.findings-table tr:last-child td {{ border-bottom: none; }}
.findings-table code {{ background: #f8f9fa; padding: 2px 5px; border-radius: 4px; font-family: 'SFMono-Regular', Consolas, Menlo, monospace; font-size: 0.9em; color: #c7254e; }}
.priority-pill {{ display: inline-block; padding: 3px 8px; border-radius: 10px; color: white; font-weight: 600; font-size: 0.78em; letter-spacing: 0.03em; }}
.priority-high {{ background-color: #dc3545; }}
.priority-medium {{ background-color: #fd7e14; }}
.priority-low {{ background-color: #28a745; }}
.muted {{ color: #6c757d; font-size: 0.9em; }}
</style>
</head>
<body>
<div class="container">
<h1>🤖 Visual AI QC Results</h1>
<p style="text-align: center; color: #6c757d; font-size: 1.1em;">
Analysis completed on: {timestamp}
</p>
<div class="file-preview">
<h3>📎 Analyzed File</h3>
<div class="file-info">
<img src="{create_thumbnail_base64(file_path) if file_path else 'data:image/svg+xml;base64,PHN2ZyB3aWR0aD0iMTUwIiBoZWlnaHQ9IjE1MCIgdmlld0JveD0iMCAwIDE1MCAxNTAiIGZpbGw9Im5vbmUiIHhtbG5zPSJodHRwOi8vd3d3LnczLm9yZy8yMDAwL3N2ZyI+CjxyZWN0IHdpZHRoPSIxNTAiIGhlaWdodD0iMTUwIiBmaWxsPSIjZjhmOWZhIiBzdHJva2U9IiNkZWUyZTYiIHN0cm9rZS13aWR0aD0iMiIvPgo8dGV4dCB4PSI3NSIgeT0iNzUiIHRleHQtYW5jaG9yPSJtaWRkbGUiIGZpbGw9IiM2Yzc1N2QiIGZvbnQtZmFtaWx5PSJBcmlhbCIgZm9udC1zaXplPSIxNCIgZHk9IjAuMzVlbSI+4p2MIEZpbGU8L3RleHQ+Cjx0ZXh0IHg9Ijc1IiB5PSI5NSIgdGV4dC1hbmNob3I9Im1pZGRsZSIgZmlsbD0iIzZjNzU3ZCIgZm9udC1mYW1pbHk9IkFyaWFsIiBmb250LXNpemU9IjEyIiBkeT0iMC4zNWVtIj5UaHVtYm5haWw8L3RleHQ+Cjwvc3ZnPg=='}" alt="File thumbnail" class="thumbnail" id="fileThumbnail">
<div class="file-details">
<div class="filename">{filename}</div>
<div class="file-meta">Original file processed for quality control analysis</div>
</div>
</div>
</div>
<div class="summary">
<h2>📊 Analysis Summary</h2>
<div class="summary-grid">
<div class="summary-item">
<div class="score-display">{overall_score}/100</div>
<div>Overall Score</div>
</div>
<div class="summary-item">
<div class="grade">{grade_text}</div>
<div>Grade</div>
</div>
<div class="summary-item">
<div style="font-size: 1.5em; font-weight: bold; color: #495057;">{completed_checks}</div>
<div>Checks Performed</div>
</div>
<div class="summary-item">
<div style="font-size: 1.2em; color: #FFC407; font-weight: bold;">{profile_name}</div>
<div>Profile Used</div>
</div>
<div class="summary-item">
<div style="font-size: 1.2em; color: {'#28a745' if reference_asset_used else '#6c757d'}; font-weight: bold;">{'✅ Used' if reference_asset_used else '➖ None'}</div>
<div>Reference Asset</div>
</div>
</div>
</div>
{technical_html}
<h2>🔍 Detailed Analysis Results</h2>
<p style="color: #6c757d; margin-bottom: 20px; font-style: italic;">
Click on any section below to expand and view detailed analysis
</p>
{check_results_html}
</div>
<script>
function toggleSection(checkName) {{
const content = document.getElementById('content-' + checkName);
const chevron = document.getElementById('chevron-' + checkName);
if (content.classList.contains('expanded')) {{
content.classList.remove('expanded');
chevron.classList.remove('expanded');
}} else {{
content.classList.add('expanded');
chevron.classList.add('expanded');
}}
}}
</script>
</body>
</html>'''
def get_reference_image_path(check_name):
"""Find a matching reference image - deprecated function, returns None"""
# This function is deprecated since numbered criteria images are no longer used
# Reference assets are now handled through the brand guidelines system
return None
def get_reference_asset_image_path(reference_asset_id):
"""
Get the actual file path for a reference asset image to send to LLM.
Args:
reference_asset_id: ID of the reference asset to retrieve
Returns:
File path to the reference image, or None if not found or not an image
"""
if not reference_asset_id or not reference_asset_id.strip():
return None
try:
# Get the reference asset file information from brand guidelines DB
file_record = brand_db.db["files"].get(reference_asset_id)
if not file_record:
print(f"DEBUG: Reference asset not found: {reference_asset_id}")
return None
file_path = file_record.get("stored_path", "")
if not file_path or not os.path.exists(file_path):
print(f"DEBUG: Reference asset file not found at path: {file_path}")
return None
image_extensions = ['.png', '.jpg', '.jpeg', '.bmp', '.webp', '.gif', '.tiff']
file_ext = os.path.splitext(file_path)[1].lower()
if file_ext in image_extensions:
print(f"DEBUG: Found reference image at: {file_path}")
return file_path
elif file_ext == '.pdf':
# Check for pre-extracted cover image
cover_path = brand_db.get_cover_image_path(reference_asset_id)
if cover_path:
print(f"DEBUG: Found PDF cover image at: {cover_path}")
return cover_path
# Fallback: extract cover on the fly
try:
from pdf_processor import extract_cover_image
fallback_cover = os.path.splitext(file_path)[0] + "_cover.png"
result = extract_cover_image(file_path, fallback_cover)
if result:
print(f"DEBUG: Extracted PDF cover on-the-fly: {result}")
return result
except Exception as e:
print(f"DEBUG: Could not extract PDF cover: {e}")
return None
else:
print(f"DEBUG: Reference asset is not an image: {file_path}")
return None
except Exception as e:
print(f"DEBUG: Error getting reference asset image path: {e}")
return None
def get_reference_asset_content(reference_asset_id):
"""
Retrieve and format reference asset content for use in QC prompts.
Args:
reference_asset_id: ID of the reference asset to retrieve
Returns:
Formatted string with reference asset information, or empty string if not found
"""
if not reference_asset_id or not reference_asset_id.strip():
return ""
try:
# Get the reference asset file information from brand guidelines DB
file_record = brand_db.db["files"].get(reference_asset_id)
if not file_record:
print(f"Reference asset not found: {reference_asset_id}")
return ""
brand_name = file_record["brand_name"]
description = file_record.get("description", "")
file_path = file_record.get("stored_path", "")
original_filename = file_record.get("original_filename", "")
# Build reference asset context for the prompt
reference_content = "\n\n=== REFERENCE ASSET GUIDELINES ===\n"
reference_content += f"Brand: {brand_name}\n"
reference_content += f"Reference File: {original_filename}\n"
if description:
reference_content += f"Description: {description}\n"
reference_content += "\nPlease use this reference asset as your guideline for analysis. "
reference_content += "Compare the uploaded image against these brand standards and requirements. "
reference_content += "Pay special attention to brand consistency, visual standards, and any specific "
reference_content += "requirements shown in the reference material.\n"
# Read reference content based on file type
if file_path and os.path.exists(file_path):
try:
file_ext = os.path.splitext(file_path)[1].lower()
if file_ext in ['.txt', '.md', '.json']:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
if len(content) > 3000:
content = content[:3000] + "... [content truncated]"
reference_content += f"\nReference Content:\n{content}\n"
elif file_ext == '.pdf':
# Check for pre-computed summary
summary_path = brand_db.get_summary_path(reference_asset_id)
if summary_path:
with open(summary_path, 'r', encoding='utf-8') as f:
summary = f.read()
reference_content += f"\nBrand Guidelines Summary (extracted from {original_filename}):\n{summary}\n"
else:
# Fallback: extract text inline
try:
from pdf_processor import extract_text_from_pdf
raw_text = extract_text_from_pdf(file_path)
if raw_text and len(raw_text) > 100:
if len(raw_text) > 5000:
raw_text = raw_text[:5000] + "... [content truncated]"
reference_content += f"\nExtracted PDF Content:\n{raw_text}\n"
else:
reference_content += f"\nReference PDF contains mostly images with limited extractable text.\n"
except Exception:
reference_content += f"\nReference file (.pdf) is available for visual comparison.\n"
elif file_ext in ['.xlsx', '.xls']:
# For localization matrices, the context is built separately via localization_processor
if file_record.get('asset_type') == 'localization_matrix':
loc_messages = file_record.get('localization_messages', [])
loc_countries = file_record.get('localization_countries', [])
reference_content += f"\nLocalization Matrix: Contains {', '.join(loc_messages)} "
reference_content += f"for {len(loc_countries)} markets ({', '.join(loc_countries[:10])}).\n"
reference_content += "Expected copy will be cross-referenced with the media plan during analysis.\n"
elif file_record.get('summary_path'):
# Source-messaging Excel (HP and similar) — inject the Gemini-generated Markdown summary
try:
with open(file_record['summary_path'], 'r', encoding='utf-8') as f:
summary = f.read()
reference_content += f"\nSource Messaging Summary (extracted from {original_filename}):\n{summary}\n"
except Exception as e:
print(f"Failed to read source-messaging summary at {file_record['summary_path']}: {e}")
reference_content += f"\nReference file ({file_ext}) uploaded but summary unreadable.\n"
else:
reference_content += f"\nReference file ({file_ext}) uploaded as reference.\n"
else:
reference_content += f"\nReference file ({file_ext}) is available for visual comparison.\n"
except Exception as e:
print(f"Error reading reference asset content: {e}")
reference_content += "\n[Reference file available but content could not be read]\n"
reference_content += "=== END REFERENCE ASSET GUIDELINES ===\n"
return reference_content
except Exception as e:
print(f"Error retrieving reference asset {reference_asset_id}: {e}")
return ""
def load_qc_apps():
"""Load all QC apps and their prompts"""
for check_name in QC_CHECKS:
try:
# Import the app module
try:
module_path = f"visual_qc_apps.{check_name}.app"
module = importlib.import_module(module_path)
# Get app class name
app_class_name = None
for key in dir(module):
if key.endswith('App') and key != 'FlaskAppTemplate':
app_class_name = key
break
except ImportError as e:
print(f"Import error for {check_name}: {e}")
continue
if app_class_name:
# Get the app class
app_class = getattr(module, app_class_name)
# Create an instance to get the prompt
app_instance = app_class()
# Find reference image for this check
reference_image = get_reference_image_path(check_name)
# Store the prompt, app class, and reference image
qc_apps[check_name] = {
"name": app_class_name,
"prompt": app_instance.prompt,
"instance": app_instance,
"display_name": app_class_name.replace("App", "").replace("_", " "),
"reference_image": reference_image
}
print(f"Loaded QC check: {check_name}" + (f" with reference image" if reference_image else ""))
else:
print(f"No app class found in {module_path}")
except Exception as e:
print(f"Error loading {check_name} app: {e}")
traceback.print_exc()
@app.route('/', methods=['GET'])
def serve_web_ui():
"""Serve the web UI"""
try:
# Use absolute path to web_ui.html (located in parent directory)
base_dir = os.path.dirname(os.path.abspath(__file__))
web_ui_path = os.path.join(os.path.dirname(base_dir), 'web_ui.html')
with open(web_ui_path, 'r') as f:
html_content = f.read()
return Response(html_content, mimetype='text/html')
except FileNotFoundError:
return jsonify({'error': 'Web UI not found'}), 404
@app.route('/health', methods=['GET'])
def health_check():
"""Simple health check endpoint"""
return jsonify({
'status': 'healthy',
'timestamp': datetime.now().isoformat()
})
@app.route('/api/health/folders', methods=['GET'])
def health_check_folders():
"""Check if required folders exist and are writable"""
try:
upload_folder = app.config.get('UPLOAD_FOLDER', 'uploads')
output_folder = app.config.get('OUTPUT_FOLDER', 'output')
# Test if folders exist
upload_exists = os.path.exists(upload_folder)
output_exists = os.path.exists(output_folder)
# Test if we can create directories (if they don't exist)
upload_writable = False
output_writable = False
try:
os.makedirs(upload_folder, exist_ok=True)
upload_writable = True
except Exception as e:
upload_error = str(e)
try:
os.makedirs(output_folder, exist_ok=True)
output_writable = True
except Exception as e:
output_error = str(e)
# Test writing a file
test_file_path = os.path.join(upload_folder, 'test_write.tmp')
can_write_files = False
try:
with open(test_file_path, 'w') as f:
f.write('test')
os.remove(test_file_path)
can_write_files = True
except Exception as e:
write_error = str(e)
return jsonify({
'status': 'success',
'folders': {
'upload_folder': upload_folder,
'upload_exists': upload_exists,
'upload_writable': upload_writable,
'output_folder': output_folder,
'output_exists': output_exists,
'output_writable': output_writable,
'can_write_files': can_write_files
},
'errors': {
'upload_error': locals().get('upload_error'),
'output_error': locals().get('output_error'),
'write_error': locals().get('write_error')
}
})
except Exception as e:
import traceback
return jsonify({
'status': 'error',
'message': str(e),
'traceback': traceback.format_exc()
}), 500
@app.route('/api/progress/<session_id>', methods=['GET'])
def get_progress(session_id):
"""Get current progress for a session"""
if session_id not in progress_tracker:
return jsonify({'status': 'error', 'message': 'Session not found'}), 404
return jsonify({
'status': 'success',
'progress': progress_tracker[session_id]
})
@app.route('/api/start_analysis', methods=['POST'])
@auth.require_auth
def start_analysis():
"""Start analysis and return session ID immediately"""
import threading
try:
# Check if file is in request
if 'file' not in request.files:
return jsonify({'status': 'error', 'message': 'No file part'}), 400
file = request.files['file']
# Check if file was selected
if file.filename == '':
return jsonify({'status': 'error', 'message': 'No selected file'}), 400
# Get parameters
profile = request.form.get('profile', 'general').lower()
brand = request.form.get('brand', 'general').lower()
output_mode = request.form.get('mode', 'json').lower()
model = request.form.get('model', 'profile')
reference_asset = request.form.get('reference_asset', '')
model_version = request.form.get('model_version', None) # Optional model override
use_media_plan = request.form.get('use_media_plan', 'false').lower() == 'true'
# Use profile if provided, otherwise fall back to brand
if profile and profile != 'general':
brand = profile.split('_')[0] if '_' in profile else profile
print(f"Starting analysis with profile: {profile}, brand: {brand}, mode: {output_mode}")
print(f"DEBUG: Raw mode parameter from request: '{request.form.get('mode')}'")
print(f"DEBUG: Processed output_mode: '{output_mode}'")
# Create unique session ID and save file
session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
session_folder = os.path.join(app.config['UPLOAD_FOLDER'], session_id)
os.makedirs(session_folder, exist_ok=True)
file_path = os.path.join(session_folder, file.filename)
file.save(file_path)
# Machine-side technical pre-flight (PIL/PyMuPDF/ffprobe, no LLM).
# Stored on progress_tracker so process_single_check can prepend it to
# every LLM prompt, and surfaced in result_data for the UI.
technical_report = technical_inspect(file_path)
# Derive client from profile if not provided
client = request.form.get('client_id', request.form.get('client', 'general')).lower()
if not client or client == 'general':
if profile.startswith('diageo_'):
client = 'diageo'
elif profile.startswith('unilever_'):
client = 'unilever'
elif profile.startswith('loreal_'):
client = 'loreal'
elif profile.startswith('amazon_'):
client = 'amazon'
elif profile.startswith('boots_'):
client = 'boots'
elif profile.startswith(('dow_jones_', 'dj_', 'marketwatch_', 'mw_', 'wsj_')):
client = 'dow_jones'
else:
client = 'general'
access_err = _require_client_access(client)
if access_err:
return access_err
# Log analysis start
try:
from usage_tracker import log_analysis_start
file_info = {
'filename': file.filename,
'size': os.path.getsize(file_path) if os.path.exists(file_path) else 0
}
# Check if g.user is set, otherwise use a default user info
user_info = getattr(g, 'user', {'user_id': 'unknown', 'email': 'unknown', 'name': 'unknown'})
print(f"DEBUG: user_info = {user_info}")
log_analysis_start(session_id, client, profile, user_info, file_info)
except Exception as log_error:
# Log the error but don't fail the analysis
print(f"WARNING: Failed to log analysis start: {log_error}")
import traceback
traceback.print_exc()
# Initialize progress tracking with estimated total checks
# We'll update this with the actual number once we determine the profile
estimated_checks = 25 # Reasonable estimate for most profiles
progress_tracker[session_id] = {
'total_checks': estimated_checks,
'completed_checks': 0,
'current_check': 'Initializing',
'current_check_display': 'Initializing Analysis',
'stage': 'setup',
'percentage': 0,
'session_id': session_id,
'status': 'started',
'technical_report': technical_report,
}
# Start analysis in background thread with explicit parameters
def run_analysis(session_id, file_path, filename, brand, profile, output_mode, reference_asset, user_info):
print(f"Background thread started for session: {session_id}")
print(f"Parameters: brand={brand}, profile={profile}, mode={output_mode}")
try:
# Force reload QC apps to ensure they're available
if not qc_apps:
load_qc_apps()
# Use the explicitly passed parameters
analysis_brand = brand
analysis_profile = profile
analysis_mode = output_mode
analysis_reference_asset = reference_asset
print(f"DEBUG: analysis_mode = '{analysis_mode}'")
# Write debug info to file for easier debugging
with open('debug_mode.txt', 'a') as f:
f.write(f"Session {session_id}: analysis_mode = '{analysis_mode}'\n")
# Validate brand
if not analysis_brand or analysis_brand.strip() == '':
analysis_brand = 'general'
# Validate output mode
print(f"DEBUG: Before validation, analysis_mode = '{analysis_mode}'")
if analysis_mode not in ['json', 'html']:
print(f"DEBUG: analysis_mode '{analysis_mode}' not valid, defaulting to 'json'")
analysis_mode = 'json'
else:
print(f"DEBUG: analysis_mode '{analysis_mode}' is valid")
print(f"DEBUG: After validation, analysis_mode = '{analysis_mode}'")
# Use the directly specified profile (no triage needed)
suggested_profile = analysis_profile if analysis_profile and analysis_profile.strip() else 'general'
print(f"Using specified profile: {suggested_profile}")
# Update progress to show starting QC analysis
progress_tracker[session_id].update({
'stage': 'qc_analysis',
'current_check': 'initializing',
'current_check_display': 'Preparing Quality Analysis',
'completed_checks': 0,
'percentage': 5
})
# STEP 1: Run Quality Control Analysis
print(f"Step 1: Running QC analysis with profile '{suggested_profile}'")
# Get the profile configuration
profile_config = get_profile(suggested_profile)
if not profile_config:
raise Exception(f'Profile {suggested_profile} not found')
# Get enabled checks from profile
enabled_checks = profile_config.get_enabled_checks()
profile_weights = profile_config.get_check_weights()
# Filter to only include checks that exist in qc_apps
enabled_checks = [check for check in enabled_checks if check in qc_apps]
if not enabled_checks:
raise Exception(f'No enabled checks found for profile {suggested_profile}')
# Update progress tracker with total checks
progress_tracker[session_id].update({
'total_checks': len(enabled_checks),
'stage': 'qc_analysis',
'percentage': 10
})
# Build media plan context if selected by user
mp_context = None
mp_match_data = None
try:
if use_media_plan:
mp_data = _get_active_media_plan(client)
else:
mp_data = None
if mp_data:
from media_plan_processor import find_matching_asset, build_media_plan_context
mp_match = find_matching_asset(file.filename, mp_data)
if mp_match:
mp_match_data = mp_match['match']
mp_context = build_media_plan_context(mp_match_data)
except Exception as mp_err:
print(f"Media plan context build error: {mp_err}")
# Build localization context if reference asset is a localization matrix
localization_context = ""
try:
if analysis_reference_asset and mp_match_data:
file_record = brand_db.db.get("files", {}).get(analysis_reference_asset)
if file_record and file_record.get('asset_type') == 'localization_matrix':
loc_path = file_record.get('localization_path')
if loc_path and os.path.exists(loc_path):
with open(loc_path, 'r', encoding='utf-8') as f:
loc_data = json.load(f)
from localization_processor import detect_message_type, build_localization_context
creative_name = mp_match_data.get('creative_name', '')
msg_type = detect_message_type(creative_name)
country_code = mp_match_data.get('country', '')
if msg_type and country_code:
localization_context = build_localization_context(loc_data, msg_type, country_code)
if localization_context:
print(f"Localization context built: {msg_type} / {country_code}")
else:
print(f"No localization data for {msg_type} / {country_code}")
else:
print(f"No message type detected from creative_name: '{creative_name}'")
except Exception as loc_err:
print(f"Localization context build error: {loc_err}")
# Combine media plan and localization contexts
if localization_context and mp_context:
mp_context = mp_context + "\n" + localization_context
elif localization_context:
mp_context = localization_context
# OCR layout measurement - provides pixel-level data to supplement LLM visual checks
ocr_ctx = None
try:
from ocr_measurement import OCR_RELEVANT_CHECKS
ocr_check_names = set(enabled_checks) & set(OCR_RELEVANT_CHECKS)
if ocr_check_names:
from ocr_measurement import run_ocr_measurement
print(f"Running OCR layout measurement for {len(ocr_check_names)} checks...")
ocr_result = run_ocr_measurement(file_path)
if ocr_result and ocr_result.get('context'):
ocr_ctx = ocr_result['context']
print(f"OCR measurements computed successfully")
print(ocr_ctx)
else:
print("OCR measurement returned no results")
except Exception as ocr_err:
print(f"OCR measurement error (non-fatal, continuing without): {ocr_err}")
# Run QC checks in parallel batches
check_results = process_checks_in_batches(
enabled_checks, qc_apps, profile_config, profile_weights,
file_path, analysis_reference_asset, brand_db, progress_tracker,
session_id, batch_size=15, media_plan_context=mp_context,
ocr_context=ocr_ctx
)
# STEP 4: Calculate Overall Score
print(f"Step 4: Calculating overall score")
total_weighted_score = 0
total_weight = 0
completed_checks = 0
failed_checks = 0
for check_name, result in check_results.items():
weight = result.get('weight', 0.1)
total_weight += weight
if result['status'] == 'success':
completed_checks += 1
score = result.get('score')
if score is not None:
total_weighted_score += score * weight
else:
failed_checks += 1
# Calculate overall score
# For profiles with total_weight = 10.0 (like General Check), use direct weighted score
# For profiles with total_weight = 1.0, multiply by 10 to scale to 100
if total_weight >= 10.0:
overall_score = min(total_weighted_score, 100) # Cap at 100
else:
overall_score = min(total_weighted_score * 10, 100) # Scale to 100-point system, cap at 100
# STEP 5: Prepare Combined Response
print(f"Step 5: Preparing response")
# Create comprehensive response with all data
result_data = {
'status': 'success',
'session_id': session_id,
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'filename': file.filename,
'profile': suggested_profile,
'profile_id': suggested_profile,
'profile_name': profile_config.name,
'model': 'Profile-based selection',
'results': check_results,
'triage_analysis': {
'status': 'skipped',
'results': {'primary_format': 'unknown', 'specific_type': 'user_selected', 'confidence_score': 10, 'recommended_qc_profile': suggested_profile},
'raw_response': 'Triage skipped - using user-selected profile directly'
},
'profile_selection': {
'selected_profile': suggested_profile,
'profile_source': 'user_selected',
'brand': analysis_brand,
'format_suffix': suggested_profile,
'reference_asset': analysis_reference_asset if analysis_reference_asset else None,
'reference_asset_used': bool(analysis_reference_asset)
},
'qc_analysis': {
'profile_used': suggested_profile,
'total_checks': len(enabled_checks),
'completed_checks': completed_checks,
'failed_checks': failed_checks,
'check_results': check_results
},
'summary': {
'overall_score': round(overall_score, 1),
'profile': profile_config.name,
'checks_count': completed_checks,
'total_checks': len(enabled_checks),
'total_weighted_score': total_weighted_score,
'total_weight': total_weight,
'grade': determine_grade(overall_score)
},
'technical_report': progress_tracker[session_id].get('technical_report', {}),
}
# L'Oreal Static override: fail if ANY individual check fails (score < 6)
if suggested_profile == 'loreal_static':
for cn, cd in check_results.items():
if cd.get('status') == 'success':
cs = cd.get('score', 0)
if cs is not None and cs < 6:
result_data['summary']['grade'] = 'Fail'
break
# Amazon Static override: fail if ANY individual check fails (score < 6)
if suggested_profile == 'amazon_static':
for cn, cd in check_results.items():
if cd.get('status') == 'success':
cs = cd.get('score', 0)
if cs is not None and cs < 6:
result_data['summary']['grade'] = 'Fail'
break
# Boots Static override: fail if ANY individual check fails (score < 6)
if suggested_profile == 'boots_static':
for cn, cd in check_results.items():
if cd.get('status') == 'success':
cs = cd.get('score', 0)
if cs is not None and cs < 6:
result_data['summary']['grade'] = 'Fail'
break
# Log analysis completion
from usage_tracker import log_analysis_complete
results_summary = {
'checks_completed': completed_checks,
'overall_score': overall_score,
'status': 'success',
'check_results': check_results
}
log_analysis_complete(session_id, client, suggested_profile, user_info, results_summary)
print(f"Analysis completed successfully")
# Save results to file
try:
print(f"DEBUG: Saving file with mode: '{analysis_mode}'")
output_path = save_results_to_file(result_data, file.filename, analysis_mode, session_id, file_path)
result_data['output_file'] = {
'path': output_path,
'filename': os.path.basename(output_path),
'url': f'{request.environ.get("SCRIPT_NAME", "")}/output/{os.path.basename(output_path)}'
}
print(f"Results saved to: {output_path}")
except Exception as save_error:
print(f"Error saving results to file: {save_error}")
# Store result in progress tracker
print(f"Analysis result status: {result_data.get('status')}")
if session_id in progress_tracker:
progress_tracker[session_id]['result'] = result_data
progress_tracker[session_id]['status'] = 'completed'
progress_tracker[session_id]['stage'] = 'complete'
progress_tracker[session_id]['percentage'] = 100
print(f"Results stored in progress tracker for session: {session_id}")
else:
print(f"ERROR: Session {session_id} not found in progress tracker!")
except Exception as e:
print(f"ERROR in background thread for session {session_id}: {str(e)}")
print(f"Exception type: {type(e)}")
import traceback
print(f"Traceback: {traceback.format_exc()}")
if session_id in progress_tracker:
progress_tracker[session_id]['status'] = 'error'
progress_tracker[session_id]['stage'] = 'error'
progress_tracker[session_id]['error'] = str(e)
progress_tracker[session_id]['current_check'] = 'Error'
progress_tracker[session_id]['current_check_display'] = 'Analysis Failed'
# Start background thread with explicit parameters
# Get user_info before starting thread (g is request-scoped and won't be available in thread)
user_info = getattr(g, 'user', {'user_id': 'unknown', 'email': 'unknown', 'name': 'unknown'})
threading.Thread(
target=run_analysis,
args=(session_id, file_path, file.filename, brand, profile, output_mode, reference_asset, user_info),
daemon=True
).start()
# Return session ID immediately
return jsonify({
'status': 'success',
'session_id': session_id,
'message': 'Analysis started'
})
except Exception as e:
print(f"ERROR in start_analysis: {type(e).__name__}: {e}")
import traceback
traceback.print_exc()
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/document/start_analysis', methods=['POST'])
@auth.require_auth
def start_document_analysis():
"""Start a document-mode analysis (multi-page PDF as QC target).
Mirrors /api/start_analysis but routes through document_mode.dispatcher.
Reuses the same progress_tracker dict, output folder convention, and
/api/progress polling endpoint, so the frontend integration is minimal.
"""
import threading
try:
if 'file' not in request.files:
return jsonify({'status': 'error', 'message': 'No file part'}), 400
file = request.files['file']
if file.filename == '':
return jsonify({'status': 'error', 'message': 'No selected file'}), 400
if not file.filename.lower().endswith('.pdf'):
return jsonify({'status': 'error', 'message': 'Document mode requires a PDF file'}), 400
profile_id = request.form.get('profile', '').lower()
if not profile_id:
return jsonify({'status': 'error', 'message': 'profile is required'}), 400
output_mode = request.form.get('mode', 'both').lower()
if output_mode not in ('json', 'html', 'both'):
output_mode = 'both'
reference_asset = request.form.get('reference_asset', '') or None
client = request.form.get('client_id', request.form.get('client', '')).lower()
if not client:
return jsonify({'status': 'error', 'message': 'client is required for document mode'}), 400
access_err = _require_client_access(client)
if access_err:
return access_err
# Resolve and validate the profile is document-mode
profile_config = get_profile(profile_id)
if not profile_config:
return jsonify({'status': 'error', 'message': f'Profile "{profile_id}" not found'}), 404
if getattr(profile_config, 'mode', 'asset') != 'document':
return jsonify({
'status': 'error',
'message': f'Profile "{profile_id}" is not a document-mode profile. Use /api/start_analysis instead.'
}), 400
# Save the uploaded PDF
session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
session_folder = os.path.join(app.config['UPLOAD_FOLDER'], session_id)
os.makedirs(session_folder, exist_ok=True)
file_path = os.path.join(session_folder, file.filename)
file.save(file_path)
# Machine-side technical pre-flight (PyMuPDF for PDFs, no LLM).
technical_report = technical_inspect(file_path)
# Pre-render directory for per-page PNGs lives alongside the PDF
pages_dir = os.path.join(session_folder, 'pages')
# Log start
try:
from usage_tracker import log_analysis_start
file_info = {
'filename': file.filename,
'size': os.path.getsize(file_path) if os.path.exists(file_path) else 0,
'mode': 'document',
}
user_info = getattr(g, 'user', {'user_id': 'unknown', 'email': 'unknown', 'name': 'unknown'})
log_analysis_start(session_id, client, profile_id, user_info, file_info)
except Exception as log_error:
print(f"WARNING: Failed to log document analysis start: {log_error}")
user_info = getattr(g, 'user', {'user_id': 'unknown', 'email': 'unknown', 'name': 'unknown'})
# Initialize progress
progress_tracker[session_id] = {
'total_checks': 0,
'completed_checks': 0,
'current_check': 'Initializing',
'current_check_display': 'Initializing document analysis',
'stage': 'setup',
'percentage': 0,
'session_id': session_id,
'status': 'started',
'mode': 'document',
'technical_report': technical_report,
}
def run_document(session_id, file_path, filename, profile_id, client, output_mode, reference_asset, user_info):
try:
if not qc_apps:
load_qc_apps()
profile_config_local = get_profile(profile_id)
# Document-mode checks may live in qc_apps (legacy LLM checks)
# OR in document_mode.checks registry (new deterministic checks).
# Accept either as a valid enabled check.
from document_mode.checks import is_document_scope_check
enabled_checks = [
c for c in profile_config_local.get_enabled_checks()
if c in qc_apps or is_document_scope_check(c)
]
profile_weights = profile_config_local.get_check_weights()
if not enabled_checks:
raise Exception(f'No enabled checks found for profile {profile_id}')
from document_mode.dispatcher import run_document_analysis
from document_mode.ingest import ingest_pdf
from document_mode.result_writer import write_document_report
doc_result = run_document_analysis(
pdf_path=file_path,
profile_config=profile_config_local,
profile_id=profile_id,
profile_weights=profile_weights,
enabled_checks=enabled_checks,
qc_apps=qc_apps,
brand_db=brand_db,
analysis_reference_asset=reference_asset,
media_plan_context=None,
ocr_context=None,
progress_tracker=progress_tracker,
session_id=session_id,
process_checks_in_batches=process_checks_in_batches,
ingest_pdf_fn=ingest_pdf,
pages_output_dir=os.path.join(os.path.dirname(file_path), 'pages'),
)
# Write JSON + HTML to the client-scoped output folder
client_folder = ensure_client_output_folder(client)
paths = write_document_report(
doc_result,
original_filename=filename,
session_id=session_id,
output_dir=client_folder,
output_mode=output_mode,
)
# Wrap doc_result in a frontend-compatible envelope
summary = doc_result.get('document_summary', {})
result_data = {
'status': 'success',
'session_id': session_id,
'timestamp': doc_result.get('timestamp'),
'filename': filename,
'profile': profile_id,
'profile_id': profile_id,
'profile_name': profile_config_local.name,
'mode': 'document',
'document_result': doc_result,
'summary': {
'overall_score': summary.get('overall_score', 0),
'profile': profile_config_local.name,
'grade': summary.get('grade', ''),
'pages_processed': doc_result.get('pages_processed', 0),
'page_count': doc_result.get('page_count', 0),
},
'technical_report': progress_tracker[session_id].get('technical_report', {}),
}
if paths.get('html'):
result_data['output_file'] = {
'path': paths['html'],
'filename': os.path.basename(paths['html']),
'url': f'/output/{client}/{os.path.basename(paths["html"])}',
}
if paths.get('json'):
result_data['output_data_file'] = {
'path': paths['json'],
'filename': os.path.basename(paths['json']),
'url': f'/output/{client}/{os.path.basename(paths["json"])}',
}
# Log completion
try:
from usage_tracker import log_analysis_complete
completed = sum(
1 for page in doc_result.get('pages', [])
for r in (page.get('results') or {}).values()
if r.get('status') == 'success'
)
log_analysis_complete(
session_id, client, profile_id, user_info,
{
'checks_completed': completed,
'overall_score': summary.get('overall_score', 0),
'status': 'success',
'mode': 'document',
'pages_processed': doc_result.get('pages_processed', 0),
},
)
except Exception as log_err:
print(f"WARNING: failed to log doc-mode analysis completion: {log_err}")
progress_tracker[session_id]['result'] = result_data
progress_tracker[session_id]['status'] = 'completed'
progress_tracker[session_id]['stage'] = 'complete'
progress_tracker[session_id]['percentage'] = 100
except Exception as e:
print(f"ERROR in document run for session {session_id}: {e}")
import traceback
traceback.print_exc()
if session_id in progress_tracker:
progress_tracker[session_id]['status'] = 'error'
progress_tracker[session_id]['stage'] = 'error'
progress_tracker[session_id]['error'] = str(e)
progress_tracker[session_id]['current_check_display'] = 'Document analysis failed'
threading.Thread(
target=run_document,
args=(session_id, file_path, file.filename, profile_id, client, output_mode, reference_asset, user_info),
daemon=True,
).start()
return jsonify({
'status': 'success',
'session_id': session_id,
'mode': 'document',
'message': 'Document analysis started',
})
except Exception as e:
print(f"ERROR in start_document_analysis: {type(e).__name__}: {e}")
import traceback
traceback.print_exc()
return jsonify({'status': 'error', 'message': str(e)}), 500
@app.route('/api/document/start_diff', methods=['POST'])
@auth.require_auth
def start_document_diff_analysis():
"""Start an old-vs-new PDF diff analysis (Phase 3).
Accepts two PDFs as 'old_file' and 'new_file' multipart fields.
Routes through document_mode.diff_engine.run_document_diff_analysis.
"""
import threading
try:
if 'old_file' not in request.files or 'new_file' not in request.files:
return jsonify({'status': 'error', 'message': "Both 'old_file' and 'new_file' are required"}), 400
old_file = request.files['old_file']
new_file = request.files['new_file']
if not old_file.filename or not new_file.filename:
return jsonify({'status': 'error', 'message': 'Both files must be selected'}), 400
if not old_file.filename.lower().endswith('.pdf') or not new_file.filename.lower().endswith('.pdf'):
return jsonify({'status': 'error', 'message': 'Both old and new must be PDFs'}), 400
profile_id = request.form.get('profile', '').lower()
if not profile_id:
return jsonify({'status': 'error', 'message': 'profile is required'}), 400
output_mode = request.form.get('mode', 'both').lower()
if output_mode not in ('json', 'html', 'both'):
output_mode = 'both'
client = request.form.get('client_id', request.form.get('client', '')).lower()
if not client:
return jsonify({'status': 'error', 'message': 'client is required for diff mode'}), 400
access_err = _require_client_access(client)
if access_err:
return access_err
profile_config = get_profile(profile_id)
if not profile_config:
return jsonify({'status': 'error', 'message': f'Profile "{profile_id}" not found'}), 404
if getattr(profile_config, 'mode', 'asset') != 'document_diff':
return jsonify({
'status': 'error',
'message': f'Profile "{profile_id}" is not a document_diff profile.'
}), 400
session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
session_folder = os.path.join(app.config['UPLOAD_FOLDER'], session_id)
os.makedirs(session_folder, exist_ok=True)
old_path = os.path.join(session_folder, f'old_{old_file.filename}')
new_path = os.path.join(session_folder, f'new_{new_file.filename}')
old_file.save(old_path)
new_file.save(new_path)
try:
from usage_tracker import log_analysis_start
user_info = getattr(g, 'user', {'user_id': 'unknown', 'email': 'unknown', 'name': 'unknown'})
log_analysis_start(session_id, client, profile_id, user_info, {
'filename': f'{old_file.filename} vs {new_file.filename}',
'mode': 'document_diff',
})
except Exception as log_error:
print(f"WARNING: failed to log diff start: {log_error}")
user_info = getattr(g, 'user', {'user_id': 'unknown', 'email': 'unknown', 'name': 'unknown'})
progress_tracker[session_id] = {
'total_checks': 0,
'completed_checks': 0,
'current_check': 'Initializing diff',
'current_check_display': 'Initializing diff analysis',
'stage': 'setup',
'percentage': 0,
'session_id': session_id,
'status': 'started',
'mode': 'document_diff',
}
def run_diff():
try:
profile_config_local = get_profile(profile_id)
from document_mode.diff_engine import run_document_diff_analysis
from document_mode.ingest import ingest_pdf
from document_mode.diff_report_writer import write_diff_report
from llm_config import call_gemini_vision
doc_result = run_document_diff_analysis(
old_pdf_path=old_path,
new_pdf_path=new_path,
old_filename=old_file.filename,
new_filename=new_file.filename,
profile_config=profile_config_local,
profile_id=profile_id,
progress_tracker=progress_tracker,
session_id=session_id,
ingest_pdf_fn=ingest_pdf,
call_gemini_vision_fn=call_gemini_vision,
pages_output_dir_old=os.path.join(session_folder, 'pages_old'),
pages_output_dir_new=os.path.join(session_folder, 'pages_new'),
)
client_folder = ensure_client_output_folder(client)
paths = write_diff_report(
doc_result,
old_filename=old_file.filename,
new_filename=new_file.filename,
session_id=session_id,
output_dir=client_folder,
output_mode=output_mode,
)
summary = doc_result.get('document_summary', {})
result_data = {
'status': 'success',
'session_id': session_id,
'timestamp': doc_result.get('timestamp'),
'filename': f'{old_file.filename} vs {new_file.filename}',
'profile': profile_id,
'profile_id': profile_id,
'profile_name': profile_config_local.name,
'mode': 'document_diff',
'document_result': doc_result,
'summary': {
'overall_score': summary.get('overall_score', 0),
'profile': profile_config_local.name,
'grade': summary.get('grade', ''),
},
}
if paths.get('html'):
result_data['output_file'] = {
'path': paths['html'],
'filename': os.path.basename(paths['html']),
'url': f'/output/{client}/{os.path.basename(paths["html"])}',
}
if paths.get('json'):
result_data['output_data_file'] = {
'path': paths['json'],
'filename': os.path.basename(paths['json']),
'url': f'/output/{client}/{os.path.basename(paths["json"])}',
}
try:
from usage_tracker import log_analysis_complete
log_analysis_complete(
session_id, client, profile_id, user_info,
{
'checks_completed': len(doc_result.get('pair_diffs', {})),
'overall_score': summary.get('overall_score', 0),
'status': 'success',
'mode': 'document_diff',
'token_usage': doc_result.get('token_usage', {}),
},
)
except Exception as log_err:
print(f"WARNING: failed to log diff completion: {log_err}")
progress_tracker[session_id]['result'] = result_data
progress_tracker[session_id]['status'] = 'completed'
progress_tracker[session_id]['stage'] = 'complete'
progress_tracker[session_id]['percentage'] = 100
except Exception as e:
print(f"ERROR in diff run for session {session_id}: {e}")
import traceback
traceback.print_exc()
if session_id in progress_tracker:
progress_tracker[session_id]['status'] = 'error'
progress_tracker[session_id]['stage'] = 'error'
progress_tracker[session_id]['error'] = str(e)
progress_tracker[session_id]['current_check_display'] = 'Diff analysis failed'
threading.Thread(target=run_diff, daemon=True).start()
return jsonify({
'status': 'success',
'session_id': session_id,
'mode': 'document_diff',
'message': 'Diff analysis started',
})
except Exception as e:
print(f"ERROR in start_document_diff_analysis: {type(e).__name__}: {e}")
import traceback
traceback.print_exc()
return jsonify({'status': 'error', 'message': str(e)}), 500
@app.route('/output/<client>/<filename>', methods=['GET'])
@auth.require_auth
def serve_client_output_file(client, filename):
"""Serve saved output files from client-specific folders"""
access_err = _require_client_access(client)
if access_err:
return access_err
try:
file_path = os.path.join(app.config['OUTPUT_FOLDER'], client, filename)
if os.path.exists(file_path):
if filename.endswith('.html'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
return Response(content, mimetype='text/html')
elif filename.endswith('.json'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
return Response(content, mimetype='application/json')
else:
return Response(open(file_path, 'rb').read(), mimetype='application/octet-stream')
else:
return jsonify({'error': 'File not found'}), 404
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/output/<filename>', methods=['GET'])
def serve_output_file(filename):
"""Serve saved output files (legacy route for backward compatibility)"""
try:
# First try to find file in root output folder
file_path = os.path.join(app.config['OUTPUT_FOLDER'], filename)
if os.path.exists(file_path):
if filename.endswith('.html'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
return Response(content, mimetype='text/html')
elif filename.endswith('.json'):
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
return Response(content, mimetype='application/json')
else:
return Response(open(file_path, 'rb').read(), mimetype='application/octet-stream')
else:
return jsonify({'error': 'File not found'}), 404
except Exception as e:
return jsonify({'error': str(e)}), 500
@app.route('/api/output_files', methods=['GET'])
@auth.require_auth
def list_output_files():
"""List saved output files filtered by client, sorted by creation date (newest first)"""
try:
# Get client parameter (optional)
client_filter = request.args.get('client', None)
print(f"DEBUG: list_output_files called with client_filter='{client_filter}'")
if client_filter:
access_err = _require_client_access(client_filter)
if access_err:
return access_err
# Run cleanup of files older than 14 days
cleanup_old_files(max_age_days=14)
files = []
output_folder = app.config['OUTPUT_FOLDER']
if os.path.exists(output_folder):
# If client filter is provided, only check that client's folder
if client_filter:
client_folder = os.path.join(output_folder, client_filter)
if os.path.exists(client_folder):
for filename in os.listdir(client_folder):
if filename.endswith(('.html', '.json')):
file_path = os.path.join(client_folder, filename)
if os.path.isfile(file_path):
file_stats = os.stat(file_path)
files.append({
'filename': filename,
'size': file_stats.st_size,
'created': datetime.fromtimestamp(file_stats.st_ctime).strftime('%Y-%m-%d %H:%M:%S'),
'created_timestamp': file_stats.st_ctime,
'url': f'{request.environ.get("SCRIPT_NAME", "")}/output/{client_filter}/{filename}',
'client': client_filter
})
else:
# No client filter - check all client folders
for item in os.listdir(output_folder):
item_path = os.path.join(output_folder, item)
if os.path.isdir(item_path):
# This is a client folder
client_name = item
for filename in os.listdir(item_path):
if filename.endswith(('.html', '.json')):
file_path = os.path.join(item_path, filename)
if os.path.isfile(file_path):
file_stats = os.stat(file_path)
files.append({
'filename': filename,
'size': file_stats.st_size,
'created': datetime.fromtimestamp(file_stats.st_ctime).strftime('%Y-%m-%d %H:%M:%S'),
'created_timestamp': file_stats.st_ctime,
'url': f'{request.environ.get("SCRIPT_NAME", "")}/output/{client_name}/{filename}',
'client': client_name
})
# Sort files by creation time (newest first)
files.sort(key=lambda x: x['created_timestamp'], reverse=True)
# Remove the timestamp field from response (not needed by frontend)
for file in files:
del file['created_timestamp']
print(f"DEBUG: Returning {len(files)} files for client_filter='{client_filter}'")
if files:
print(f"DEBUG: Sample file clients: {[f.get('client', 'unknown') for f in files[:3]]}")
return jsonify({'files': files})
except Exception as e:
print(f"Error listing output files: {e}")
import traceback
traceback.print_exc()
return jsonify({'error': str(e)}), 500
@app.route('/api/delete_output_files', methods=['POST'])
@auth.require_auth
def delete_output_files():
"""Delete selected output files"""
try:
data = request.get_json()
filenames = data.get('filenames', [])
client = data.get('client', '')
if not filenames or not client:
return jsonify({'error': 'Missing filenames or client'}), 400
output_folder = app.config['OUTPUT_FOLDER']
deleted = []
errors = []
for filename in filenames:
# Sanitize filename to prevent path traversal
safe_filename = os.path.basename(filename)
file_path = os.path.join(output_folder, client, safe_filename)
if os.path.isfile(file_path):
try:
os.remove(file_path)
deleted.append(safe_filename)
print(f"Deleted output file: {file_path}")
except Exception as e:
errors.append({'filename': safe_filename, 'error': str(e)})
else:
errors.append({'filename': safe_filename, 'error': 'File not found'})
return jsonify({
'deleted': deleted,
'errors': errors,
'deleted_count': len(deleted)
})
except Exception as e:
print(f"Error deleting output files: {e}")
return jsonify({'error': str(e)}), 500
@app.route('/api/profiles', methods=['GET'])
def get_available_profiles():
"""Get all available profiles grouped by type, optionally filtered by client"""
from profile_config import get_profile_summary, get_profile
from client_config import get_profiles_with_visibility
# Check if client parameter is provided
client_id = request.args.get('client', None)
# Force reload profiles to ensure they're up to date
from profile_config import load_profiles
load_profiles()
profiles_summary = get_profile_summary()
# Filter by client if specified (using visibility-aware function)
if client_id:
allowed_profiles = get_profiles_with_visibility(client_id)
profiles_summary = {
pid: pinfo for pid, pinfo in profiles_summary.items()
if pid in allowed_profiles
}
# Group profiles by type
triage_profiles = {}
format_profiles = {}
# Build detailed profile data including checks
all_profiles_detailed = {}
for profile_id, profile_info in profiles_summary.items():
# Get full profile data
try:
profile = get_profile(profile_id)
# Load the raw profile JSON to get weight_scale if it exists
profile_path = os.path.join('profiles', f'{profile_id}.json')
weight_scale = 100 # Default to 100
if os.path.exists(profile_path):
with open(profile_path, 'r') as f:
profile_data = json.load(f)
weight_scale = profile_data.get('weight_scale', 100)
all_profiles_detailed[profile_id] = {
'name': profile.name,
'description': profile.description,
'enabled_checks': profile.get_enabled_checks(),
'total_checks': len(profile.checks),
'enabled_count': len(profile.get_enabled_checks()),
'weight_scale': weight_scale,
'mode': getattr(profile, 'mode', 'asset'),
'checks': {check_name: {'weight': config.weight, 'llm': config.llm, 'enabled': config.enabled}
for check_name, config in profile.checks.items()}
}
except:
# Fallback to summary if full profile can't be loaded
all_profiles_detailed[profile_id] = profile_info
if '_triage' in profile_id:
brand = profile_id.replace('_triage', '')
triage_profiles[brand] = {
'id': profile_id,
'name': profile_info['name'],
'description': profile_info['description']
}
elif any(suffix in profile_id for suffix in ['_print', '_digital', '_ooh', '_packaging', '_event', '_indoor']):
format_profiles[profile_id] = {
'id': profile_id,
'name': profile_info['name'],
'description': profile_info['description'],
'enabled_checks': profile_info['enabled_checks'],
'total_checks': profile_info['total_checks']
}
response_data = {
'status': 'success',
'triage_profiles': triage_profiles,
'format_profiles': format_profiles,
'all_profiles': all_profiles_detailed
}
# Include client info if filtered
if client_id:
response_data['client'] = client_id
return jsonify(response_data)
@app.route('/api/clients', methods=['GET'])
def get_clients_endpoint():
"""Get clients visible to the current user. Admins see all; others see only their grants."""
from client_config import get_all_clients
from user_access import get_user_clients, is_admin
try:
all_clients = get_all_clients()
# Resolve the current user's email if authenticated. Unauthenticated
# callers get an empty list so the UI can prompt for sign-in without
# leaking the full client catalogue.
user_email = ''
try:
auth_result = app.auth_middleware.is_authenticated()
if auth_result.get('authenticated'):
user_email = auth_result.get('user', {}).get('email', '')
except Exception:
pass
if not user_email:
return jsonify({'status': 'success', 'clients': {}, 'is_admin': False})
allowed_ids = get_user_clients(user_email)
filtered = {cid: all_clients[cid] for cid in allowed_ids if cid in all_clients}
return jsonify({
'status': 'success',
'clients': filtered,
'is_admin': is_admin(user_email)
})
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/all_clients', methods=['GET'])
@auth.require_auth
def list_all_clients_endpoint():
"""
Full client catalogue (auth required). Used by the Request Access form so
users can pick clients they don't currently have access to.
"""
try:
from client_config import get_all_clients
return jsonify({'status': 'success', 'clients': get_all_clients()})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)}), 500
@app.route('/api/clients/<client_id>/default_profile', methods=['GET'])
@auth.require_auth
def get_client_default_profile_endpoint(client_id):
"""Return the effective default profile for a client (override or static).
Auth-required (any signed-in user) — read-only, no admin gate.
"""
try:
from client_config import get_all_clients, get_default_profile, get_client_profiles
clients = get_all_clients()
if client_id not in clients:
return jsonify({'status': 'error', 'message': f'unknown client: {client_id}'}), 404
access_err = _require_client_access(client_id)
if access_err:
return access_err
return jsonify({
'status': 'success',
'client_id': client_id,
'profiles': get_client_profiles(client_id),
'default_profile': get_default_profile(client_id),
'static_default': clients[client_id].get('default_profile'),
})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)}), 500
@app.route('/api/clients/<client_id>/default_profile', methods=['PUT'])
@auth.require_auth
def set_client_default_profile_endpoint(client_id):
"""Admin-only: set the default profile for a client (persisted as a runtime override).
Body: {"profile_id": "<id>"}. The profile must already be in the client's
`profiles` list — we don't allow defaulting to a profile the client can't see.
Posts to backend/client_defaults.json so a bad write can never break server boot.
"""
actor_email, err = _require_admin()
if err:
return err
try:
body = request.get_json(silent=True) or {}
profile_id = (body.get('profile_id') or '').strip()
if not profile_id:
return jsonify({'status': 'error', 'message': 'profile_id is required'}), 400
from client_config import set_default_profile
ok, reason = set_default_profile(client_id, profile_id)
if not ok:
return jsonify({'status': 'error', 'message': reason}), 400
print(f'Admin {actor_email}: set default_profile for {client_id} → {profile_id}')
return jsonify({
'status': 'success',
'client_id': client_id,
'default_profile': profile_id,
})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)}), 500
@app.route('/api/clients/<client_id>/default_profile', methods=['DELETE'])
@auth.require_auth
def clear_client_default_profile_endpoint(client_id):
"""Admin-only: clear the runtime override so the static default applies again."""
actor_email, err = _require_admin()
if err:
return err
try:
from client_config import clear_default_profile_override, get_default_profile
clear_default_profile_override(client_id)
print(f'Admin {actor_email}: cleared default_profile override for {client_id}')
return jsonify({
'status': 'success',
'client_id': client_id,
'default_profile': get_default_profile(client_id),
})
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)}), 500
@app.route('/api/access_request', methods=['POST'])
@auth.require_auth
def request_client_access():
"""
Signed-in users request access to additional clients. Sends an email to all
admins with the requester's identity (taken from the verified session, never
the request body) and the clients they want.
"""
from html import escape as _html_escape
try:
payload = request.get_json(silent=True) or {}
requested_clients = payload.get('clients') or []
reason = (payload.get('reason') or '').strip()
if not isinstance(requested_clients, list) or not requested_clients:
return jsonify({'status': 'error', 'message': 'Select at least one client'}), 400
# Always source identity from the verified session, not the body.
user_email = getattr(g, 'user', {}).get('email', '')
user_name = getattr(g, 'user', {}).get('name', '') or user_email
if not user_email:
return jsonify({'status': 'error', 'message': 'Authentication required'}), 401
from client_config import get_all_clients
all_clients = get_all_clients()
invalid = [c for c in requested_clients if c not in all_clients]
if invalid:
return jsonify({'status': 'error',
'message': f'Unknown client(s): {", ".join(invalid)}'}), 400
client_labels = [all_clients[c].get('display_name', c) for c in requested_clients]
from user_access import list_access_entries
access_data = list_access_entries()
admin_recipients = [e['email'] for e in access_data.get('entries', [])
if e.get('is_admin') and e.get('email')]
if not admin_recipients:
return jsonify({'status': 'error',
'message': 'No admin recipients configured'}), 500
subject = f'AI QC: Client access request from {user_name}'
text_body_lines = [
f'{user_name} ({user_email}) has requested access to:',
'',
] + [f' - {label}' for label in client_labels]
if reason:
text_body_lines += ['', 'Reason / context:', reason]
text_body_lines += ['', 'Review and grant access via the Admin panel > User Access tab.']
text_body = '\n'.join(text_body_lines)
html_clients = ''.join(f'<li>{_html_escape(label)}</li>' for label in client_labels)
html_reason = ''
if reason:
html_reason = (
'<p><strong>Reason / context:</strong><br>'
+ _html_escape(reason).replace('\n', '<br>')
+ '</p>'
)
html_body = (
f'<p><strong>{_html_escape(user_name)}</strong> '
f'({_html_escape(user_email)}) has requested access to:</p>'
f'<ul>{html_clients}</ul>'
f'{html_reason}'
'<p>Review and grant access via the Admin panel &gt; User Access tab.</p>'
)
from email_service import send_email
ok, err = send_email(
to_addresses=admin_recipients,
subject=subject,
body=text_body,
html_body=html_body,
reply_to=user_email,
)
try:
from usage_tracker import log_access_request
log_access_request({
'user_email': user_email,
'user_name': user_name,
'requested_clients': requested_clients,
'reason': reason,
'recipients': admin_recipients,
'email_sent': ok,
'email_error': err,
})
except Exception as log_err:
print(f'[access_request] log failed: {log_err}')
if not ok:
return jsonify({'status': 'error',
'message': f'Could not send email: {err}'}), 502
return jsonify({
'status': 'success',
'message': 'Request sent — an admin will review it shortly.',
'recipients_count': len(admin_recipients),
})
except Exception as e:
print(f'Error in /api/access_request: {e}')
import traceback
traceback.print_exc()
return jsonify({'status': 'error', 'message': str(e)}), 500
@app.route('/api/models', methods=['GET'])
def get_models_endpoint():
"""Get all available LLM models for selection"""
from llm_config import get_available_models, get_model_info
try:
available_models = get_available_models()
model_info = get_model_info()
return jsonify({
'status': 'success',
'models': available_models,
'model_info': model_info
})
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/run_check', methods=['POST'])
def run_check():
"""API endpoint to run a single QC check"""
data = request.json
# Validate inputs
if not data or 'check_name' not in data or 'file_path' not in data:
return jsonify({'status': 'error', 'message': 'Missing required parameters'}), 400
check_name = data['check_name']
file_path = data['file_path']
model_name = data.get('model_name', 'Gemini') # Default to Gemini if not specified
output_mode = data.get('mode', 'json').lower() # Get output mode, default to JSON
# Validate check exists
if check_name not in qc_apps:
return jsonify({'status': 'error', 'message': f'QC check "{check_name}" not found'}), 404
# Validate file exists
if not os.path.exists(file_path):
return jsonify({'status': 'error', 'message': 'File not found'}), 404
try:
# Get the prompt
prompt = qc_apps[check_name]['prompt']
# Get reference image if available
reference_path = qc_apps[check_name].get('reference_image')
# Run the QC check with reference image if available
result = run_visual_qc(
prompt=prompt,
asset_path=file_path,
reference_path=reference_path,
model_name=model_name
)
# Extract score from result if possible
score = None
try:
# Use our extraction function to get score from JSON blocks
json_data = extract_json_from_response(result['response'])
if 'score' in json_data:
score = json_data.get('score')
print(f"Extracted score from JSON block: {score}")
# If we still don't have a score, look for any score in text
if score is None:
# Try to find a score pattern in the text
score_pattern = r'["\']score["\']\s*:\s*(\d+)'
score_match = re.search(score_pattern, result['response'])
if score_match:
score = int(score_match.group(1))
print(f"Extracted score from regex: {score}")
else:
# Look for visual evidence of actual scores in text like "score: 7", "score is 8", "score of 9 out of 10"
descriptive_score_pattern = r'score(?:\s+is|\s*:\s*|\s+of\s+)(?:\s*)(\d+)(?:\s*out\s*of\s*10)?'
descriptive_match = re.search(descriptive_score_pattern, result['response'].lower())
if descriptive_match:
score = int(descriptive_match.group(1))
print(f"Extracted score from descriptive text: {score}")
else:
# Try to determine score from pass/fail status (legacy mode)
result_text = result.get('response', '').upper()
if "PASS" in result_text:
score = 10 # Pass = 10/10
print("Detected PASS keyword, setting score to 10")
elif "FAIL" in result_text:
score = 3 # Fail = 3/10
print("Detected FAIL keyword, setting score to 3")
else:
print(f"Could not extract score, using default of 5")
score = 5 # Default middle score
except Exception as parse_error:
print(f"Error parsing score from response: {parse_error}")
score = 5 # Default to middle score
# Add the score to the result
result['score'] = score if score is not None else 5 # Default to middle score if extraction fails
# Process result for JSON mode
if output_mode == 'json':
# For JSON mode, we update the response to be ONLY the JSON part
json_data = extract_json_from_response(result['response'])
if json_data:
# If we found JSON in the response, replace the full text with just the extracted JSON
result['original_response'] = result['response'] # Save original for debugging
result['response'] = json.dumps(json_data, indent=2) # Pretty print the JSON
# Add metadata but don't save to file
output_json = {
"timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
"check_name": check_name,
"display_name": qc_apps[check_name]['display_name'],
"model": model_name,
"file_analyzed": os.path.basename(file_path),
"result": result,
"score": result['score'],
"has_reference": reference_path is not None
}
return jsonify(output_json)
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e),
'traceback': traceback.format_exc()
}), 500
@app.route('/api/triage_file', methods=['POST'])
def api_triage_file():
"""API endpoint to perform file type triage detection"""
try:
# Force reload QC apps to ensure triage app is available
if not qc_apps:
load_qc_apps()
# Check if file is in request
if 'file' not in request.files:
return jsonify({'status': 'error', 'message': 'No file part'}), 400
file = request.files['file']
# Check if file was selected
if file.filename == '':
return jsonify({'status': 'error', 'message': 'No selected file'}), 400
# Get the brand/profile base name for triage routing
brand = request.form.get('brand', 'default')
# Create unique session ID and save file
session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
session_folder = os.path.join(app.config['UPLOAD_FOLDER'], session_id)
os.makedirs(session_folder, exist_ok=True)
file_path = os.path.join(session_folder, file.filename)
file.save(file_path)
# Run file type triage detection
if 'file_type_triage' not in qc_apps:
return jsonify({'status': 'error', 'message': 'File type triage app not available'}), 500
# Get the triage prompt
prompt = qc_apps['file_type_triage']['prompt']
# Run the triage check
result = run_visual_qc(
prompt=prompt,
asset_path=file_path,
model_name='Gemini' # Use Gemini for triage
)
# Extract the triage results
triage_data = extract_json_from_response(result['response'])
if not triage_data:
return jsonify({
'status': 'error',
'message': 'Could not extract triage results',
'raw_response': result['response']
}), 500
# Generate the suggested profile name
format_suffix = triage_data.get('recommended_qc_profile', '_digital')
suggested_profile = f"{brand}{format_suffix}"
# Return triage results
return jsonify({
'status': 'success',
'session_id': session_id,
'file_path': file_path,
'filename': file.filename,
'triage_results': triage_data,
'suggested_profile': suggested_profile,
'brand': brand,
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
})
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e),
'traceback': traceback.format_exc()
}), 500
@app.route('/api/process_triaged_file', methods=['POST'])
@auth.require_auth
def api_process_triaged_file():
"""API endpoint to process a file that has already been triaged"""
try:
data = request.json
# Validate required parameters
if not data or 'session_id' not in data or 'profile' not in data:
return jsonify({'status': 'error', 'message': 'Missing required parameters: session_id and profile'}), 400
session_id = data['session_id']
profile_name = data['profile']
output_mode = data.get('mode', 'json').lower()
# Reconstruct file path from session
session_folder = os.path.join(app.config['UPLOAD_FOLDER'], session_id)
if not os.path.exists(session_folder):
return jsonify({'status': 'error', 'message': 'Session not found'}), 404
# Find the file in the session folder
files = os.listdir(session_folder)
if not files:
return jsonify({'status': 'error', 'message': 'No file found in session'}), 404
filename = files[0] # Use the first file found
file_path = os.path.join(session_folder, filename)
# Validate profile exists
if profile_name not in PROFILES:
available_profiles = list(PROFILES.keys())
return jsonify({
'status': 'error',
'message': f'Invalid profile: {profile_name}. Available profiles: {available_profiles}'
}), 400
# Get profile and process checks
profile = get_profile(profile_name)
selected_checks = profile.get_enabled_checks()
profile_weights = profile.get_check_weights()
if not selected_checks:
return jsonify({'status': 'error', 'message': 'No QC checks available for the selected profile'}), 400
# Process each check (same logic as original process_file)
check_results = {}
overall_weighted_score = 0
total_weight = 0
for i, check_name in enumerate(selected_checks):
print(f"Processing check {i+1}/{len(selected_checks)}: {check_name}")
# Skip if check is not available
if check_name not in qc_apps:
check_results[check_name] = {
'status': 'error',
'message': f'QC check not found'
}
continue
# Get LLM preference for this check
ai_model = profile.get_check_llm(check_name)
# Get the prompt and reference image
prompt = qc_apps[check_name]['prompt']
reference_path = qc_apps[check_name].get('reference_image')
try:
# Run the QC check
result = run_visual_qc(
prompt=prompt,
asset_path=file_path,
reference_path=reference_path,
model_name=ai_model
)
# Extract score from result
score = extract_score_from_result(result, profile, check_name)
result['score'] = score
# Process result for JSON mode
if output_mode == 'json':
json_data = extract_json_from_response(result['response'])
if json_data:
result['original_response'] = result['response']
result['response'] = json.dumps(json_data, indent=2)
# Calculate weighted score
weight = profile_weights.get(check_name, 1.0 / len(selected_checks))
weighted_score = result['score'] * weight
overall_weighted_score += weighted_score
total_weight += weight
# Store result
check_results[check_name] = {
'status': 'success',
'display_name': qc_apps[check_name]['display_name'],
'score': result['score'],
'weight': weight,
'weighted_score': weighted_score,
'has_reference': reference_path is not None,
'result': result
}
except Exception as e:
check_results[check_name] = {
'status': 'error',
'message': str(e),
'traceback': traceback.format_exc()
}
# Calculate overall score
overall_score = 0
if total_weight > 0:
# Special case for Unilever key visual profile - show percentage of 120
if profile_name == 'unilever_key_visual':
# For Unilever profile, calculate as percentage of 120
# Maximum possible score is 10 * 1.2 = 12, so scale to 120
max_possible_score = 10 * total_weight # 10 * 1.2 = 12
overall_score = min(round((overall_weighted_score / max_possible_score) * 120, 1), 120)
else:
# Maximum possible score is 10 * total_weight, so normalize to 100%
max_possible_score = 10 * total_weight
overall_score = min(round((overall_weighted_score / max_possible_score) * 100, 1), 100)
else:
successful_checks = [r for r in check_results.values() if r.get('status') == 'success']
if successful_checks:
sum_scores = sum(r.get('score', 0) for r in successful_checks)
overall_score = min(round((sum_scores / len(successful_checks)) * 10, 1), 100)
# Generate report data
report_data = {
'session_id': session_id,
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'filename': filename,
'profile': profile_name,
'profile_id': profile_name, # Store the profile ID for HTML detection
'profile_name': profile.name,
'model': 'Profile-based selection',
'results': check_results,
'summary': {
'overall_score': overall_score,
'profile': profile.name,
'checks_count': len([r for r in check_results.values() if r.get('status') == 'success']),
'total_checks': len(check_results),
'total_weighted_score': overall_weighted_score,
'total_weight': total_weight
}
}
# Determine overall grade
grade = determine_grade(overall_score)
# L'Oreal Static override: fail if ANY individual check fails (score < 6)
if profile_name == 'loreal_static':
for check_name, check_data in check_results.items():
if check_data.get('status') == 'success':
check_score = check_data.get('score', 0)
if check_score is not None and check_score < 6:
grade = 'Fail'
break
# Boots Static override: fail if ANY individual check fails (score < 6)
if profile_name == 'boots_static':
for check_name, check_data in check_results.items():
if check_data.get('status') == 'success':
check_score = check_data.get('score', 0)
if check_score is not None and check_score < 6:
grade = 'Fail'
break
report_data['summary']['grade'] = grade
# Save results to file regardless of output mode
output_path = save_results_to_file(report_data, filename, output_mode, session_id, file_path)
# Return results based on output mode
if output_mode == 'html':
html_response = generate_html_response(report_data, filename)
# Add file path info to the response
return html_response
else:
# For JSON mode, return the data with file path info
report_data['output_file'] = {
'path': output_path,
'filename': os.path.basename(output_path),
'url': f'{request.environ.get("SCRIPT_NAME", "")}/output/{os.path.basename(output_path)}'
}
return jsonify(report_data)
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e),
'traceback': traceback.format_exc()
}), 500
@app.route('/api/process_file', methods=['POST'])
@auth.require_auth
def api_process_file():
"""API endpoint to process a file with specified profiles and checks"""
try:
# Force reload QC apps to ensure they're available for this request
if not qc_apps:
load_qc_apps()
print(f"API: Loaded {len(qc_apps)} QC apps for processing")
# Check if file is in request
if 'file' not in request.files:
return jsonify({'status': 'error', 'message': 'No file part'}), 400
file = request.files['file']
# Check if file was selected
if file.filename == '':
return jsonify({'status': 'error', 'message': 'No selected file'}), 400
# Get output mode (json or html)
output_mode = request.form.get('mode', 'json').lower()
if output_mode not in ['json', 'html']:
output_mode = 'json' # Default to JSON if invalid mode
# Get selected profiles from request (can be single or multiple)
# Support both single profile parameter and multiple profiles array
selected_profiles = []
if 'profiles[]' in request.form:
selected_profiles = request.form.getlist('profiles[]')
elif 'profile' in request.form:
selected_profiles = [request.form.get('profile')]
if not selected_profiles:
selected_profiles = ['default'] # Use default if none specified
# Validate profiles
for profile_id in selected_profiles:
if profile_id != 'custom' and profile_id not in PROFILES:
available_profiles = list(PROFILES.keys())
return jsonify({
'status': 'error',
'message': f'Invalid profile: {profile_id}. Available profiles: {available_profiles}'
}), 400
# Track all selected checks and weights from all profiles
all_selected_checks = set()
combined_profile_weights = {}
selected_profile_names = []
custom_selection = False
# Process each selected profile
for profile_id in selected_profiles:
if profile_id == 'custom':
# Custom selection
custom_checks = request.form.getlist('checks')
if not custom_checks:
return jsonify({'status': 'error', 'message': 'Custom profile selected but no checks were specified'}), 400
all_selected_checks.update(custom_checks)
custom_selection = True
else:
# Standard profile
profile = get_profile(profile_id)
selected_profile_names.append(profile.name)
profile_checks = profile.get_enabled_checks()
all_selected_checks.update(profile_checks)
# Add profile weights (we'll normalize later)
profile_weights = profile.get_check_weights()
for check, weight in profile_weights.items():
if check in combined_profile_weights:
# If a check is in multiple profiles, use the higher weight
combined_profile_weights[check] = max(combined_profile_weights[check], weight)
else:
combined_profile_weights[check] = weight
# Convert to list for consistency with the rest of the code
selected_checks = list(all_selected_checks)
profile_weights = combined_profile_weights
if not selected_checks:
return jsonify({'status': 'error', 'message': 'No QC checks available for the selected profile'}), 400
# Create unique session ID and save file
session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
# Create session folder and save file
session_folder = os.path.join(app.config['UPLOAD_FOLDER'], session_id)
os.makedirs(session_folder, exist_ok=True)
file_path = os.path.join(session_folder, file.filename)
file.save(file_path)
# Process each check
check_results = {}
overall_weighted_score = 0
total_weight = 0
# Track progress for client
progress = {
'total_checks': len(selected_checks),
'completed_checks': 0,
'current_check': ''
}
for i, check_name in enumerate(selected_checks):
# Update progress
progress['current_check'] = check_name
progress['completed_checks'] = i
print(f"API Progress: {i+1}/{len(selected_checks)} - Processing {check_name}")
# Skip if check is not available
if check_name not in qc_apps:
check_results[check_name] = {
'status': 'error',
'message': f'QC check not found'
}
continue
# Determine which LLM to use based on the selected profiles
# Default to Gemini
ai_model = 'Gemini'
# Check each profile's LLM preference for this check
for profile_id in selected_profiles:
if profile_id != 'custom': # Skip custom profile
check_llm_map = get_check_llm_map(profile_id)
if check_name in check_llm_map:
# If any profile uses OpenAI for this check, prioritize OpenAI
if check_llm_map[check_name] == 'OpenAI':
ai_model = 'OpenAI'
break
# Get the prompt
prompt = qc_apps[check_name]['prompt']
# Get reference image if available
reference_path = qc_apps[check_name].get('reference_image')
try:
# Run the QC check
result = run_visual_qc(
prompt=prompt,
asset_path=file_path,
reference_path=reference_path,
model_name=ai_model
)
# Extract score from result
score = None
try:
# Use our extraction function to get score from JSON blocks
json_data = extract_json_from_response(result['response'])
if 'score' in json_data:
score = json_data.get('score')
print(f"Extracted score from JSON block: {score}")
# If we still don't have a score, look for any score in text
if score is None:
# Try to find a score pattern in the text
score_pattern = r'["\']score["\']\s*:\s*(\d+)'
score_match = re.search(score_pattern, result['response'])
if score_match:
score = int(score_match.group(1))
print(f"Extracted score from regex: {score}")
else:
# Look for descriptive scores in text
descriptive_score_pattern = r'score(?:\s+is|\s*:\s*|\s+of\s+)(?:\s*)(\d+)(?:\s*out\s*of\s*10)?'
descriptive_match = re.search(descriptive_score_pattern, result['response'].lower())
if descriptive_match:
score = int(descriptive_match.group(1))
print(f"Extracted score from descriptive text: {score}")
else:
# Try to determine score from pass/fail status (legacy mode)
result_text = result.get('response', '').upper()
if "PASS" in result_text:
score = 10 # Pass = 10/10
print("Detected PASS keyword, setting score to 10")
elif "FAIL" in result_text:
score = 3 # Fail = 3/10
print("Detected FAIL keyword, setting score to 3")
else:
score = 5 # Default middle score
print(f"Could not extract score, using default of 5")
except Exception as parse_error:
print(f"Error parsing score from response: {parse_error}")
score = 5 # Default to middle score
# Add the score to the result
result['score'] = score if score is not None else 5
# Process result for JSON mode
if output_mode == 'json':
# For JSON mode, we update the response to be ONLY the JSON part
json_data = extract_json_from_response(result['response'])
if json_data:
# If we found JSON in the response, replace the full text with just the extracted JSON
result['original_response'] = result['response'] # Save original for debugging
result['response'] = json.dumps(json_data, indent=2) # Pretty print the JSON
# Calculate weighted score
weight = profile_weights.get(check_name, 0)
if weight == 0 and ('default' in selected_profiles or len(selected_profiles) == 0):
weight = 1.0 / len(selected_checks)
weighted_score = result['score'] * weight
overall_weighted_score += weighted_score
total_weight += weight
# Store result
check_results[check_name] = {
'status': 'success',
'display_name': qc_apps[check_name]['display_name'],
'score': result['score'],
'weight': weight,
'weighted_score': weighted_score,
'has_reference': reference_path is not None,
'result': result
}
# Update progress after successful check
progress['completed_checks'] = i + 1
except Exception as e:
check_results[check_name] = {
'status': 'error',
'message': str(e),
'traceback': traceback.format_exc()
}
# Calculate overall score
overall_score = 0
if total_weight > 0:
# Special case for Unilever key visual profile - show percentage of 120
if len(selected_profiles) == 1 and selected_profiles[0] == 'unilever_key_visual':
# For Unilever profile, calculate as percentage of 120
# Maximum possible score is 10 * 1.2 = 12, so scale to 120
max_possible_score = 10 * total_weight # 10 * 1.2 = 12
overall_score = min(round((overall_weighted_score / max_possible_score) * 120, 1), 120)
else:
# Maximum possible score is 10 * total_weight, so normalize to 100%
max_possible_score = 10 * total_weight
overall_score = min(round((overall_weighted_score / max_possible_score) * 100, 1), 100)
else:
# Simple average if no weights
successful_checks = [r for r in check_results.values() if r.get('status') == 'success']
if successful_checks:
sum_scores = sum(r.get('score', 0) for r in successful_checks)
overall_score = min(round((sum_scores / len(successful_checks)) * 10, 1), 100)
# Generate report data
report_data = {
'session_id': session_id,
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'filename': file.filename,
'profiles': selected_profiles,
'profile_id': selected_profiles[0] if len(selected_profiles) == 1 else None, # Store single profile ID for HTML detection
'profile_name': ", ".join(selected_profile_names) or "Multiple Profiles",
'model': 'Profile-based selection',
'results': check_results,
'summary': {
'overall_score': overall_score,
'profile': ", ".join(selected_profile_names) or "Multiple Profiles",
'checks_count': len([r for r in check_results.values() if r.get('status') == 'success']),
'total_checks': len(check_results),
'total_weighted_score': overall_weighted_score,
'total_weight': total_weight
}
}
# Determine overall grade based on score (Pass/Fail)
avg_individual_score = overall_score / 10 # Convert to 1-10 scale
grade = 'Pass' if avg_individual_score >= 6 else 'Fail'
# Add grade to summary
report_data['summary']['grade'] = grade
# Save results to file regardless of output mode
output_path = save_results_to_file(report_data, file.filename, output_mode, session_id, file_path)
# Return data based on output mode
if output_mode == 'html':
# Create a more interactive HTML view with formatted results
# Define a function to get color based on score
def get_score_result(score):
if score >= 6:
return "Pass", "#28a745" # Green for pass
else:
return "Fail", "#dc3545" # Red for fail
# Build HTML for each check result
check_results_html = ""
for check_name, check_data in report_data['results'].items():
if check_data['status'] == 'success':
display_name = check_data.get('display_name', check_name)
score = check_data.get('score', 0)
result_text, score_color = get_score_result(score)
# Extract response text (strip JSON blocks for cleaner display)
response_text = check_data['response']
# Remove JSON code blocks for cleaner reading
import re
response_text = re.sub(r'```json.*?```', '', response_text, flags=re.DOTALL)
response_text = response_text.strip()
check_results_html += f"""
<div class="check-result">
<h3>{display_name} <span class="score" style="color: {score_color}">{result_text}</span></h3>
<div class="response">{response_text.replace(chr(10), '<br>')}</div>
</div>
<hr>
"""
# Get summary score result
overall_score = report_data['summary']['overall_score']
overall_result, overall_color = get_score_result(overall_score/10) # Normalize to 0-10 scale
html_content = f"""
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Visual AI QC Results for {file.filename}</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 20px; padding: 0; background-color: #f8f9fa; }}
.container {{ max-width: 1200px; margin: 0 auto; background-color: white; padding: 20px; border-radius: 8px; box-shadow: 0 0 10px rgba(0,0,0,0.1); }}
h1, h2, h3 {{ color: #333; }}
h1 {{ margin-bottom: 5px; }}
.summary {{ background-color: #e9ecef; padding: 15px; border-radius: 5px; margin: 20px 0; }}
.score {{ font-weight: normal; margin-left: 10px; }}
.check-result {{ margin-bottom: 20px; }}
.response {{ margin-top: 10px; line-height: 1.5; }}
hr {{ border: 0; height: 1px; background-color: #ddd; margin: 30px 0; }}
.grade {{ font-size: 1.2em; font-weight: bold; }}
pre {{ background-color: #f5f5f5; padding: 15px; border-radius: 5px; overflow-x: auto; }}
.json-toggle {{ cursor: pointer; color: #007bff; text-decoration: underline; }}
.json-view {{ display: none; margin-top: 20px; }}
</style>
</head>
<body>
<div class="container">
<h1>Visual AI QC Results</h1>
<p>File analyzed: <strong>{file.filename}</strong> | Timestamp: {report_data['timestamp']}</p>
<div class="summary">
<h2>Summary</h2>
<p>Overall Result: <span style="color: {overall_color}; font-weight: bold; font-size: 1.2em;">{overall_result}</span></p>
<p>Status: <span class="grade">{report_data['summary']['grade']}</span></p>
<p>Profile: {report_data['profile_name']}</p>
<p>Total Checks: {report_data['summary']['checks_count']}</p>
</div>
<h2>Detailed Results</h2>
{check_results_html}
<p class="json-toggle" onclick="document.getElementById('json-data').style.display = document.getElementById('json-data').style.display === 'none' ? 'block' : 'none';">
Show/Hide Raw JSON Data
</p>
<div id="json-data" class="json-view">
<pre>{json.dumps(report_data, indent=2)}</pre>
</div>
</div>
</body>
</html>
"""
# For HTML mode, redirect to the saved file
return Response(html_content, mimetype='text/html')
else:
# For JSON mode, return the data with file path info
report_data['output_file'] = {
'path': output_path,
'filename': os.path.basename(output_path),
'url': f'{request.environ.get("SCRIPT_NAME", "")}/output/{os.path.basename(output_path)}'
}
return jsonify(report_data)
except Exception as e:
# Include progress information in error response
error_progress = progress.get('completed_checks', 0) if 'progress' in locals() else 0
total_checks = progress.get('total_checks', 0) if 'progress' in locals() else 0
if not total_checks and 'selected_checks' in locals():
total_checks = len(selected_checks)
error_data = {
'status': 'error',
'message': str(e),
'traceback': traceback.format_exc(),
'progress': {
'total_checks': total_checks,
'completed_checks': error_progress,
'percentage': (error_progress / total_checks * 100) if total_checks else 0
}
}
# Get output mode if possible, default to JSON if not defined or in case of error
output_mode = request.form.get('mode', 'json').lower() if 'mode' in request.form else 'json'
if output_mode == 'html':
# Create a more user-friendly error page
html_content = f"""
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Visual AI QC Error</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 20px; padding: 0; background-color: #f8f9fa; }}
.container {{ max-width: 1200px; margin: 0 auto; background-color: white; padding: 20px; border-radius: 8px; box-shadow: 0 0 10px rgba(0,0,0,0.1); }}
h1 {{ color: #dc3545; margin-bottom: 5px; }}
.error-box {{ background-color: #f8d7da; border: 1px solid #f5c6cb; color: #721c24; padding: 15px; border-radius: 5px; margin: 20px 0; }}
.progress-box {{ background-color: #cce5ff; border: 1px solid #b8daff; color: #004085; padding: 15px; border-radius: 5px; margin: 20px 0; }}
pre {{ background-color: #f5f5f5; padding: 15px; border-radius: 5px; overflow-x: auto; }}
.json-toggle {{ cursor: pointer; color: #007bff; text-decoration: underline; }}
.json-view {{ display: none; margin-top: 20px; }}
.progress-bar {{ height: 20px; background-color: #e9ecef; border-radius: 5px; margin-top: 10px; }}
.progress-bar-fill {{ height: 100%; background-color: #007bff; border-radius: 5px; width: {error_data['progress']['percentage']}%; }}
</style>
</head>
<body>
<div class="container">
<h1>Error Processing Request</h1>
<div class="error-box">
<h3>Error Message</h3>
<p>{error_data['message']}</p>
</div>
<div class="progress-box">
<h3>Processing Progress</h3>
<p>Completed {error_data['progress']['completed_checks']} of {error_data['progress']['total_checks']} checks</p>
<div class="progress-bar">
<div class="progress-bar-fill"></div>
</div>
</div>
<p class="json-toggle" onclick="document.getElementById('error-data').style.display = document.getElementById('error-data').style.display === 'none' ? 'block' : 'none';">
Show/Hide Technical Details
</p>
<div id="error-data" class="json-view">
<pre>{json.dumps(error_data, indent=2)}</pre>
</div>
</div>
</body>
</html>
"""
return Response(html_content, mimetype='text/html'), 500
else:
# Default JSON mode
return jsonify(error_data), 500
@app.route('/analyze', methods=['POST'])
@app.route('/api/analyze', methods=['POST'])
@auth.require_auth
def api_analyze_with_triage():
"""
Smart analysis endpoint that combines triage and profile-based QC.
This endpoint:
1. Runs triage to determine file type
2. Automatically selects appropriate profile based on brand + file type
3. Runs full QC analysis with selected profile
4. Returns combined results showing both triage and QC analysis
Parameters:
- file: Image file to analyze
- brand: Brand name (diageo, unilever, general) - defaults to 'general'
- mode: Output mode (json, html) - defaults to 'json'
- return_file: Whether to return file info (true/false) - defaults to 'false'
"""
try:
# Force reload QC apps to ensure they're available
if not qc_apps:
load_qc_apps()
# Check if file is in request
if 'file' not in request.files:
return jsonify({'status': 'error', 'message': 'No file part'}), 400
file = request.files['file']
# Check if file was selected
if file.filename == '':
return jsonify({'status': 'error', 'message': 'No selected file'}), 400
# Get parameters
brand = request.form.get('brand', 'general').lower()
profile = request.form.get('profile', '').lower()
output_mode = request.form.get('mode', 'json').lower()
return_file = request.form.get('return_file', 'false').lower() == 'true'
# Validate brand - allow any brand name now that users can create custom profiles
# The old validation was too restrictive for custom profiles
if not brand or brand.strip() == '':
brand = 'general'
# Validate output mode
if output_mode not in ['json', 'html']:
output_mode = 'json'
# Create unique session ID and save file
session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
session_folder = os.path.join(app.config['UPLOAD_FOLDER'], session_id)
os.makedirs(session_folder, exist_ok=True)
file_path = os.path.join(session_folder, file.filename)
file.save(file_path)
# Initialize progress tracking for this session
progress_tracker[session_id] = {
'total_checks': 0,
'completed_checks': 0,
'current_check': 'Initializing',
'current_check_display': 'Initializing Analysis',
'stage': 'setup',
'percentage': 0,
'session_id': session_id
}
# STEP 1: Run Triage Analysis
print(f"Step 1: Running triage analysis for {file.filename}")
# Update progress
progress_tracker[session_id].update({
'stage': 'triage',
'current_check': 'file_type_triage',
'current_check_display': 'File Type Detection',
'percentage': 10
})
if 'file_type_triage' not in qc_apps:
return jsonify({'status': 'error', 'message': 'File type triage app not available'}), 500
# Get the triage prompt and run triage
triage_prompt = qc_apps['file_type_triage']['prompt']
triage_result = run_visual_qc(
prompt=triage_prompt,
asset_path=file_path,
model_name='Gemini'
)
# Extract triage results
triage_data = extract_json_from_response(triage_result['response'])
if not triage_data:
return jsonify({
'status': 'error',
'message': 'Could not extract triage results',
'raw_triage_response': triage_result['response']
}), 500
# STEP 2: Determine Appropriate Profile
print(f"Step 2: Determining profile based on {'direct profile parameter' if profile else 'triage results'}")
if profile and profile.strip():
# Use the directly specified profile
suggested_profile = profile
print(f"Using directly specified profile: {suggested_profile}")
else:
# Get format suffix from triage results
format_suffix = triage_data.get('recommended_qc_profile', '_digital')
# Construct profile name
suggested_profile = f"{brand}{format_suffix}"
# Check if the suggested profile exists, fallback to main brand profile if not
if suggested_profile not in PROFILES:
suggested_profile = brand
if suggested_profile not in PROFILES:
suggested_profile = 'general' # Final fallback
print(f"Profile constructed from triage: {suggested_profile}")
print(f"Final selected profile: {suggested_profile}")
# STEP 3: Run Full QC Analysis with Selected Profile
print(f"Step 3: Running QC analysis with profile '{suggested_profile}'")
# Get the profile configuration
profile_config = get_profile(suggested_profile)
if not profile_config:
return jsonify({
'status': 'error',
'message': f'Profile {suggested_profile} not found'
}), 400
# Get enabled checks from profile using Profile object methods
enabled_checks = profile_config.get_enabled_checks()
profile_weights = profile_config.get_check_weights()
# Filter to only include checks that exist in qc_apps
enabled_checks = [check for check in enabled_checks if check in qc_apps]
if not enabled_checks:
return jsonify({
'status': 'error',
'message': f'No enabled checks found for profile {suggested_profile}'
}), 400
# Run QC checks in parallel batches
# Update progress tracker with total checks
progress_tracker[session_id].update({
'total_checks': len(enabled_checks),
'stage': 'qc_analysis',
'percentage': 20
})
# Get reference asset from form for this endpoint
reference_asset = request.form.get('reference_asset', '')
# Use the parallel processing function
check_results = process_checks_in_batches_with_triage(
enabled_checks, qc_apps, profile_config, profile_weights,
file_path, reference_asset, brand_db, progress_tracker,
session_id, batch_size=15, base_percentage=20, percentage_range=70
)
# STEP 4: Calculate Overall Score
print(f"Step 4: Calculating overall score")
total_weighted_score = 0
total_weight = 0
completed_checks = 0
failed_checks = 0
for check_name, result in check_results.items():
weight = result.get('weight', 0.1)
total_weight += weight
if result['status'] == 'completed':
completed_checks += 1
score = result.get('score')
if score is not None:
total_weighted_score += score * weight
else:
failed_checks += 1
# Calculate overall score - sum of weighted scores scaled to 100
# For profiles with total_weight = 10.0 (like General Check), use direct weighted score
# For profiles with total_weight = 1.0, multiply by 10 to scale to 100
if total_weight >= 10.0:
overall_score = min(total_weighted_score, 100) # Cap at 100
elif total_weight > 0:
overall_score = min(total_weighted_score * 10, 100) # Scale to 100-point system, cap at 100
else:
overall_score = 0
# STEP 5: Prepare Combined Response
print(f"Step 5: Preparing response")
# Update progress to completion
progress_tracker[session_id].update({
'stage': 'complete',
'current_check': 'Finalizing',
'current_check_display': 'Finalizing Report',
'percentage': 100,
'completed_checks': len(enabled_checks)
})
# Since triage is skipped, set default triage data
triage_data = {
'primary_format': 'user_specified',
'specific_type': 'profile_based_analysis',
'confidence_score': 10,
'format_indicators': 'User selected profile directly',
'secondary_format': '',
'recommended_qc_profile': suggested_profile
}
triage_result = {'response': 'Triage skipped - using user-selected profile directly'}
# Build comprehensive response
response_data = {
'status': 'success',
'session_id': session_id,
'filename': file.filename,
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
# Triage Results Section
'triage_analysis': {
'primary_format': triage_data.get('primary_format', 'unknown'),
'specific_type': triage_data.get('specific_type', 'unknown'),
'confidence_score': triage_data.get('confidence_score', 0),
'format_indicators': triage_data.get('format_indicators', ''),
'secondary_format': triage_data.get('secondary_format', ''),
'recommended_qc_profile': triage_data.get('recommended_qc_profile', ''),
'full_triage_response': triage_result['response']
},
# Profile Selection Section
'profile_selection': {
'requested_brand': brand,
'suggested_profile': suggested_profile,
'profile_used': suggested_profile,
'profile_name': profile_config.name,
'profile_description': profile_config.description,
'reference_asset': reference_asset if reference_asset else None,
'reference_asset_used': bool(reference_asset)
},
# QC Analysis Results Section
'qc_analysis': {
'overall_score': round(overall_score, 2),
'total_checks': len(enabled_checks),
'completed_checks': completed_checks,
'failed_checks': failed_checks,
'total_weight': round(total_weight, 2),
'check_results': check_results
},
# Summary Section
'summary': {
'file_type_detected': f"{triage_data.get('primary_format', 'unknown')} - {triage_data.get('specific_type', 'unknown')}",
'profile_applied': profile_config.name,
'overall_score': round(overall_score, 2),
'score_percentage': f"{round(overall_score, 1)}%",
'checks_passed': completed_checks,
'checks_failed': failed_checks,
'confidence': triage_data.get('confidence_score', 0)
}
}
# Add file info if requested
if return_file:
response_data['file_info'] = {
'path': file_path,
'size': os.path.getsize(file_path),
'session_folder': session_folder
}
# Media plan matching
try:
media_plan_data = _get_active_media_plan(client)
if media_plan_data:
from media_plan_processor import find_matching_asset, validate_asset_specs
match_result = find_matching_asset(file.filename, media_plan_data)
if match_result:
validation = validate_asset_specs(file_path, match_result['match'])
response_data['media_plan_match'] = {
'matched': True,
'confidence': match_result['confidence'],
'match_type': match_result['match_type'],
'asset_id': match_result['match']['asset_id'],
'spec': match_result['match'],
'validation': validation,
}
else:
response_data['media_plan_match'] = {'matched': False}
except Exception as mp_err:
print(f"Media plan matching error: {mp_err}")
# Auto-save HTML report to output directory (regardless of output_mode)
try:
# Generate comprehensive HTML report using the same format as the web UI
html_report_content = generate_comprehensive_html_report(response_data, file.filename, file_path)
# Create output filename in client-specific folder
safe_filename = re.sub(r'[^a-zA-Z0-9.-]', '_', file.filename)
output_filename = f"{session_id}_{safe_filename}_report.html"
client_output_folder = ensure_client_output_folder(client)
output_path = os.path.join(client_output_folder, output_filename)
# Save HTML report to output directory
with open(output_path, 'w', encoding='utf-8') as f:
f.write(html_report_content)
print(f"HTML report auto-saved to: {output_path}")
# Add output file info to response
response_data['output_file'] = {
'path': output_path,
'filename': output_filename,
'auto_saved': True
}
except Exception as e:
print(f"Error auto-saving HTML report: {str(e)}")
# Don't fail the entire request if auto-save fails
# Return appropriate format
if output_mode == 'html':
# Create HTML response with both triage and QC results
triage_html = f"""
<div class="triage-section">
<h2>📋 File Type Analysis (Triage)</h2>
<div class="triage-results">
<div class="result-item">
<strong>Detected Format:</strong> {triage_data.get('primary_format', 'unknown').title()} - {triage_data.get('specific_type', 'unknown')}
</div>
<div class="result-item">
<strong>Confidence:</strong> {triage_data.get('confidence_score', 0)}/10
</div>
<div class="result-item">
<strong>Key Indicators:</strong> {triage_data.get('format_indicators', 'N/A')}
</div>
<div class="result-item">
<strong>Recommended Profile:</strong> {triage_data.get('recommended_qc_profile', 'N/A')}
</div>
</div>
</div>
"""
profile_html = f"""
<div class="profile-section">
<h2>🎯 Profile Selection</h2>
<div class="profile-results">
<div class="result-item">
<strong>Brand:</strong> {brand.title()}
</div>
<div class="result-item">
<strong>Profile Used:</strong> {profile_config.name}
</div>
<div class="result-item">
<strong>Description:</strong> {profile_config.description}
</div>
</div>
</div>
"""
# Generate QC results HTML
check_results_html = ""
for check_name, result in check_results.items():
score = result.get('score', 0) if result.get('score') is not None else 0
status_class = "pass" if result['status'] == 'completed' and score >= 6 else "fail"
result_display = "Pass" if score >= 6 else "Fail" if score > 0 else "N/A"
check_results_html += f"""
<div class="check-result {status_class}">
<h3>{check_name.replace('_', ' ').title()}</h3>
<div class="score">Result: {result_display}</div>
<div class="weight">Weight: {result.get('weight', 0.1)}</div>
<div class="response">{result.get('response', 'No response')[:500]}...</div>
</div>
"""
html_content = f"""
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Visual AI QC Analysis Results</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 20px; background-color: #f8f9fa; }}
.container {{ max-width: 1200px; margin: 0 auto; background: white; padding: 20px; border-radius: 8px; box-shadow: 0 2px 10px rgba(0,0,0,0.1); }}
.header {{ text-align: center; margin-bottom: 30px; border-bottom: 2px solid #007bff; padding-bottom: 20px; }}
.summary {{ background: #e3f2fd; padding: 20px; border-radius: 8px; margin-bottom: 30px; }}
.triage-section, .profile-section {{ background: #f0f8f0; padding: 20px; border-radius: 8px; margin-bottom: 20px; }}
.result-item {{ margin: 10px 0; padding: 10px; background: white; border-left: 4px solid #007bff; }}
.check-result {{ margin: 15px 0; padding: 15px; border-radius: 8px; border-left: 4px solid #ccc; }}
.check-result.pass {{ border-left-color: #28a745; background-color: #d4edda; }}
.check-result.fail {{ border-left-color: #dc3545; background-color: #f8d7da; }}
.score {{ font-weight: bold; color: #007bff; }}
.json-toggle {{ cursor: pointer; color: #007bff; text-decoration: underline; margin: 20px 0; }}
.json-view {{ background: #f8f9fa; padding: 15px; border-radius: 5px; overflow-x: auto; display: none; }}
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>🤖 Visual AI QC Analysis Results</h1>
<p>File: {file.filename} | Analyzed: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
</div>
<div class="summary">
<h2>📊 Summary</h2>
<div style="display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 15px;">
<div><strong>Overall Result:</strong> {'Pass' if overall_score/10 >= 6 else 'Fail'}</div>
<div><strong>File Type:</strong> {triage_data.get('primary_format', 'unknown').title()}</div>
<div><strong>Profile Used:</strong> {profile_config.name}</div>
<div><strong>Checks Performed:</strong> {completed_checks}/{len(enabled_checks)}</div>
</div>
</div>
{triage_html}
{profile_html}
<div class="qc-section">
<h2>🔍 Quality Control Analysis</h2>
{check_results_html}
</div>
<p class="json-toggle" onclick="document.getElementById('json-data').style.display = document.getElementById('json-data').style.display === 'none' ? 'block' : 'none';">
Show/Hide Raw JSON Data
</p>
<div id="json-data" class="json-view">
<pre>{json.dumps(response_data, indent=2)}</pre>
</div>
</div>
</body>
</html>
"""
return Response(html_content, mimetype='text/html')
else:
# Add session_id to response for progress tracking
response_data['session_id'] = session_id
# Return JSON response
return jsonify(response_data)
except Exception as e:
# Update progress to error state if session exists
if 'session_id' in locals() and session_id in progress_tracker:
progress_tracker[session_id].update({
'stage': 'error',
'current_check': 'Error',
'current_check_display': 'Analysis Failed',
'percentage': 0
})
return jsonify({
'status': 'error',
'message': str(e),
'traceback': traceback.format_exc()
}), 500
finally:
# Schedule cleanup of progress tracker after 5 minutes
import threading
import time
def cleanup_progress():
time.sleep(300) # 5 minutes
if 'session_id' in locals() and session_id in progress_tracker:
del progress_tracker[session_id]
if 'session_id' in locals():
threading.Thread(target=cleanup_progress, daemon=True).start()
def _update_client_config_for_profile(profile_id, client_ids, action='add'):
"""
Update client_config.py to add or remove a profile from client configurations
Args:
profile_id: Profile ID to add/remove
client_ids: List of client IDs to update
action: 'add' or 'remove'
"""
try:
client_config_path = os.path.join(os.path.dirname(__file__), 'client_config.py')
# Read current client_config.py
with open(client_config_path, 'r') as f:
config_content = f.read()
# Parse CLIENT_PROFILES dictionary
# This is a simple approach - we'll reload the module and update it
from client_config import CLIENT_PROFILES
import copy
updated_config = copy.deepcopy(CLIENT_PROFILES)
# Update each client
for client_id in client_ids:
if client_id in updated_config:
current_profiles = updated_config[client_id].get('profiles', [])
if action == 'add' and profile_id not in current_profiles:
current_profiles.append(profile_id)
updated_config[client_id]['profiles'] = current_profiles
elif action == 'remove' and profile_id in current_profiles:
current_profiles.remove(profile_id)
updated_config[client_id]['profiles'] = current_profiles
# Write updated config back to file
config_lines = [
"#!/usr/bin/env python3",
'"""',
"Client configuration module for managing client-profile relationships",
'"""',
"",
"CLIENT_PROFILES = {",
]
for client_id, client_data in updated_config.items():
config_lines.append(f" '{client_id}': {{")
config_lines.append(f" 'name': '{client_data['name']}',")
profiles_str = ", ".join([f"'{p}'" for p in client_data['profiles']])
config_lines.append(f" 'profiles': [{profiles_str}],")
config_lines.append(f" 'display_name': '{client_data['display_name']}',")
config_lines.append(f" 'description': '{client_data['description']}'")
config_lines.append(" },")
config_lines.append("}")
config_lines.append("")
config_lines.append("def get_client_profiles(client_id):")
config_lines.append(' """Get profiles available for a specific client"""')
config_lines.append(" return CLIENT_PROFILES.get(client_id, {}).get('profiles', [])")
config_lines.append("")
config_lines.append("def get_all_clients():")
config_lines.append(' """Get all available clients"""')
config_lines.append(" return CLIENT_PROFILES")
config_lines.append("")
config_lines.append("def validate_client_profile(client_id, profile_id):")
config_lines.append(' """Validate that a profile belongs to a client"""')
config_lines.append(" client_profiles = get_client_profiles(client_id)")
config_lines.append(" return profile_id in client_profiles")
config_lines.append("")
with open(client_config_path, 'w') as f:
f.write('\n'.join(config_lines))
# Reload the client_config module to pick up changes
import importlib
import client_config
importlib.reload(client_config)
return True
except Exception as e:
print(f"Error updating client_config: {str(e)}")
return False
@app.route('/api/profiles', methods=['POST'])
@auth.require_auth
def create_profile():
"""Create a new QC profile"""
try:
data = request.get_json()
if not data:
return jsonify({'status': 'error', 'message': 'No data provided'}), 400
profile_name = data.get('name', '').strip()
if not profile_name:
return jsonify({'status': 'error', 'message': 'Profile name is required'}), 400
# Create profile filename (sanitize name)
import re
safe_name = re.sub(r'[^a-zA-Z0-9_-]', '_', profile_name.lower())
profile_filename = f"{safe_name}.json"
profile_path = os.path.join('profiles', profile_filename)
# Check if profile already exists
if os.path.exists(profile_path):
return jsonify({'status': 'error', 'message': 'Profile already exists'}), 400
# Get visibility settings
visibility = data.get('visibility', 'all') # 'all' or 'client_specific'
visible_to_clients = data.get('visible_to_clients', []) # List of client IDs
# Create profile data structure
profile_data = {
"name": profile_name,
"description": data.get('description', ''),
"pass_threshold": data.get('pass_threshold', 85),
"visibility": visibility,
"visible_to_clients": visible_to_clients,
"created_at": datetime.now().isoformat(),
"created_by": g.user.get('email', 'unknown'),
"version": 1,
"checks": data.get('checks', {})
}
# Save profile to file
with open(profile_path, 'w') as f:
json.dump(profile_data, f, indent=2)
# Update client_config if client_specific visibility
if visibility == 'client_specific' and visible_to_clients:
_update_client_config_for_profile(safe_name, visible_to_clients, action='add')
elif visibility == 'all':
# Add to all clients in client_config
from client_config import get_all_clients
all_clients = list(get_all_clients().keys())
_update_client_config_for_profile(safe_name, all_clients, action='add')
return jsonify({
'status': 'success',
'message': f'Profile "{profile_name}" created successfully',
'profile_id': safe_name,
'profile_path': profile_path,
'visibility': visibility
})
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/profiles/<profile_id>', methods=['PUT'])
@auth.require_auth
def update_profile(profile_id):
"""Update an existing QC profile by creating a new version"""
try:
data = request.get_json()
if not data:
return jsonify({'status': 'error', 'message': 'No data provided'}), 400
profile_filename = f"{profile_id}.json"
profile_path = os.path.join('profiles', profile_filename)
# Check if profile exists
if not os.path.exists(profile_path):
return jsonify({'status': 'error', 'message': 'Profile not found'}), 404
# Load existing profile
with open(profile_path, 'r') as f:
existing_profile = json.load(f)
# Determine if profile has been modified (check if any field changed)
has_changes = False
fields_to_check = ['name', 'description', 'pass_threshold', 'checks', 'visibility', 'visible_to_clients']
for field in fields_to_check:
if field in data and data[field] != existing_profile.get(field):
has_changes = True
break
if not has_changes:
return jsonify({
'status': 'success',
'message': 'No changes detected',
'profile_id': profile_id
})
# Find next version number
current_version = existing_profile.get('version', 1)
next_version = current_version + 1
# Create new versioned profile ID
# Remove any existing version suffix (e.g., my_profile_v2 -> my_profile)
base_profile_id = re.sub(r'_v\d+$', '', profile_id)
new_profile_id = f"{base_profile_id}_v{next_version}"
new_profile_filename = f"{new_profile_id}.json"
new_profile_path = os.path.join('profiles', new_profile_filename)
# Check if new version already exists (shouldn't happen, but be safe)
counter = 0
while os.path.exists(new_profile_path) and counter < 100:
next_version += 1
new_profile_id = f"{base_profile_id}_v{next_version}"
new_profile_filename = f"{new_profile_id}.json"
new_profile_path = os.path.join('profiles', new_profile_filename)
counter += 1
# Create new profile version with updated data
new_profile_data = existing_profile.copy()
new_profile_data.update({
"name": data.get('name', existing_profile.get('name')),
"description": data.get('description', existing_profile.get('description', '')),
"pass_threshold": data.get('pass_threshold', existing_profile.get('pass_threshold', 85)),
"checks": data.get('checks', existing_profile.get('checks', {})),
"visibility": data.get('visibility', existing_profile.get('visibility', 'all')),
"visible_to_clients": data.get('visible_to_clients', existing_profile.get('visible_to_clients', [])),
"version": next_version,
"modified_at": datetime.now().isoformat(),
"modified_by": g.user.get('email', 'unknown'),
"previous_version": profile_id
})
# Save new versioned profile
with open(new_profile_path, 'w') as f:
json.dump(new_profile_data, f, indent=2)
# Update client_config to use new version
visibility = new_profile_data.get('visibility', 'all')
visible_to_clients = new_profile_data.get('visible_to_clients', [])
# Remove old version from client configs
from client_config import get_all_clients
all_clients = list(get_all_clients().keys())
_update_client_config_for_profile(profile_id, all_clients, action='remove')
# Add new version to appropriate client configs
if visibility == 'client_specific' and visible_to_clients:
_update_client_config_for_profile(new_profile_id, visible_to_clients, action='add')
elif visibility == 'all':
_update_client_config_for_profile(new_profile_id, all_clients, action='add')
return jsonify({
'status': 'success',
'message': f'Profile updated to version {next_version}. Original kept as "{profile_id}"',
'profile_id': new_profile_id,
'previous_profile_id': profile_id,
'version': next_version
})
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/profiles/<profile_id>', methods=['DELETE'])
@auth.require_auth
def delete_profile(profile_id):
"""Delete a QC profile"""
try:
profile_filename = f"{profile_id}.json"
profile_path = os.path.join('profiles', profile_filename)
# Check if profile exists
if not os.path.exists(profile_path):
return jsonify({'status': 'error', 'message': 'Profile not found'}), 404
# Load profile to get name for response
with open(profile_path, 'r') as f:
profile_data = json.load(f)
profile_name = profile_data.get('name', profile_id)
# Delete profile file
os.remove(profile_path)
return jsonify({
'status': 'success',
'message': f'Profile "{profile_name}" deleted successfully'
})
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/brand_guidelines', methods=['GET'])
def get_brand_guidelines():
"""Get all brand guidelines, optionally filtered by brand or client"""
try:
brand_name = request.args.get('brand')
client_id = request.args.get('client')
if brand_name:
guidelines = brand_db.get_brand_guidelines(brand_name)
return jsonify({
'status': 'success',
'guidelines': guidelines,
'brands': brand_db.get_all_brands()
})
elif client_id:
# Filter by client
filtered = brand_db.get_guidelines_by_client(client_id)
return jsonify({
'status': 'success',
'brands': filtered['brands'],
'files': filtered['files']
})
else:
# Return full database structure that frontend expects
return jsonify({
'status': 'success',
'brands': brand_db.db['brands'],
'files': brand_db.db['files']
})
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/brand_guidelines', methods=['POST'])
@auth.require_auth
def upload_brand_guideline():
"""Upload a new brand guideline file"""
try:
# Check if file is in request
if 'file' not in request.files:
return jsonify({'status': 'error', 'message': 'No file provided'}), 400
file = request.files['file']
if file.filename == '':
return jsonify({'status': 'error', 'message': 'No file selected'}), 400
# Get brand name and other metadata
brand_name = request.form.get('brand_name', '').strip()
description = request.form.get('description', '').strip()
tags = request.form.get('tags', '').strip().split(',') if request.form.get('tags') else []
client_id = request.form.get('client_id', 'general').strip()
if not brand_name:
return jsonify({'status': 'error', 'message': 'Brand name is required'}), 400
# Save uploaded file temporarily
temp_filename = f"temp_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{file.filename}"
temp_path = os.path.join(app.config['UPLOAD_FOLDER'], temp_filename)
file.save(temp_path)
try:
# Add to brand guidelines database
file_record = brand_db.add_brand_guideline(
brand_name=brand_name,
file_path=temp_path,
description=description,
tags=[tag.strip() for tag in tags if tag.strip()],
client_id=client_id
)
# Remove temporary file
os.remove(temp_path)
# Trigger background PDF processing if applicable
if file_record.get('file_type') == '.pdf':
import threading
def _process_pdf_bg(fid, spath, bname, fdir):
try:
from pdf_processor import process_pdf_guideline
result = process_pdf_guideline(spath, fid, bname, fdir)
brand_db.update_file_record(fid, result)
print(f"PDF processing complete for {fid}")
except Exception as e:
print(f"PDF processing failed for {fid}: {e}")
brand_db.update_file_record(fid, {
'processed': 'error',
'processing_error': str(e)
})
threading.Thread(
target=_process_pdf_bg,
args=(file_record['id'], file_record['stored_path'],
brand_name, str(brand_db.files_dir)),
daemon=True
).start()
file_record['processing_status'] = 'processing'
# Trigger Excel processing: try localization matrix first (existing
# clients), fall back to Source Messaging summary (HP and similar).
elif file_record.get('file_type') in ('.xlsx', '.xls'):
import threading
def _process_excel_bg(fid, spath, fdir):
try:
from localization_processor import parse_localization_matrix
parsed = parse_localization_matrix(spath)
if parsed:
json_path = os.path.join(fdir, f"{fid}_localization.json")
with open(json_path, 'w', encoding='utf-8') as f:
json.dump(parsed, f, indent=2, ensure_ascii=False)
brand_db.update_file_record(fid, {
'processed': True,
'processed_at': datetime.now().isoformat(),
'localization_path': json_path,
'localization_messages': list(parsed.get('messages', {}).keys()),
'localization_countries': parsed.get('countries', []),
'asset_type': 'localization_matrix',
})
print(f"Localization matrix parsing complete for {fid}: "
f"{len(parsed.get('messages', {}))} messages, "
f"{len(parsed.get('countries', []))} countries")
return
# Not a localization matrix — process as Source Messaging
# (HP-style structured Markdown summary via Gemini).
from excel_processor import process_excel_file
summary_text, summary_path = process_excel_file(spath, fid)
brand_db.update_file_record(fid, {
'processed': True,
'processed_at': datetime.now().isoformat(),
'summary_path': summary_path,
'summary_length': len(summary_text),
'cover_image_path': None,
'asset_type': 'source_messaging',
})
print(f"Source-messaging summary complete for {fid}: "
f"{len(summary_text)} chars")
except Exception as e:
print(f"Excel processing failed for {fid}: {e}")
brand_db.update_file_record(fid, {
'processed': 'error',
'processing_error': str(e),
})
threading.Thread(
target=_process_excel_bg,
args=(file_record['id'], file_record['stored_path'],
str(brand_db.files_dir)),
daemon=True,
).start()
file_record['processing_status'] = 'processing'
return jsonify({
'status': 'success',
'message': f'Brand guideline uploaded successfully for {brand_name}',
'file_record': file_record
})
except Exception as e:
# Clean up temp file on error
if os.path.exists(temp_path):
os.remove(temp_path)
raise e
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/brand_guidelines/<file_id>', methods=['DELETE'])
@auth.require_auth
def delete_brand_guideline(file_id):
"""Delete a brand guideline file"""
try:
success = brand_db.delete_guideline(file_id)
if success:
return jsonify({
'status': 'success',
'message': 'Brand guideline deleted successfully'
})
else:
return jsonify({
'status': 'error',
'message': 'Brand guideline not found'
}), 404
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/brand_guidelines/<file_id>/status', methods=['GET'])
def get_guideline_processing_status(file_id):
"""Check processing status of a brand guideline"""
file_record = brand_db.db["files"].get(file_id)
if not file_record:
return jsonify({'status': 'error', 'message': 'Not found'}), 404
return jsonify({
'status': 'success',
'processed': file_record.get('processed', False),
'processing_error': file_record.get('processing_error'),
'page_count': file_record.get('page_count'),
'summary_length': file_record.get('summary_length'),
'processed_at': file_record.get('processed_at')
})
@app.route('/api/brand_guidelines/<file_id>/reprocess', methods=['POST'])
@auth.require_auth
def reprocess_guideline(file_id):
"""Re-process a PDF brand guideline (re-extract and re-summarize)"""
file_record = brand_db.db["files"].get(file_id)
if not file_record:
return jsonify({'status': 'error', 'message': 'Not found'}), 404
if file_record.get('file_type') != '.pdf':
return jsonify({'status': 'error', 'message': 'Only PDF files can be reprocessed'}), 400
import threading
def _reprocess_bg(fid, spath, bname, fdir):
try:
from pdf_processor import process_pdf_guideline
result = process_pdf_guideline(spath, fid, bname, fdir)
brand_db.update_file_record(fid, result)
print(f"PDF reprocessing complete for {fid}")
except Exception as e:
print(f"PDF reprocessing failed for {fid}: {e}")
brand_db.update_file_record(fid, {
'processed': 'error',
'processing_error': str(e)
})
threading.Thread(
target=_reprocess_bg,
args=(file_id, file_record['stored_path'],
file_record['brand_name'], str(brand_db.files_dir)),
daemon=True
).start()
return jsonify({'status': 'success', 'message': 'Reprocessing started'})
@app.route('/api/media_plan', methods=['POST'])
@auth.require_auth
def upload_media_plan():
"""Upload a media plan Excel file for a client"""
try:
if 'file' not in request.files:
return jsonify({'status': 'error', 'message': 'No file provided'}), 400
file = request.files['file']
if not file.filename:
return jsonify({'status': 'error', 'message': 'No file selected'}), 400
client_id = request.form.get('client_id', 'general').strip().lower()
if not client_id:
return jsonify({'status': 'error', 'message': 'client_id is required'}), 400
access_err = _require_client_access(client_id)
if access_err:
return access_err
display_name = (request.form.get('display_name') or '').strip()
# Save the Excel file
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
safe_name = re.sub(r'[^a-zA-Z0-9._-]', '_', file.filename)
excel_filename = f"{client_id}_{timestamp}_{safe_name}"
excel_path = os.path.join(MEDIA_PLANS_DIR, excel_filename)
file.save(excel_path)
# Parse the media plan
from media_plan_processor import parse_media_plan
parsed = parse_media_plan(excel_path)
# Save parsed JSON for fast lookup
json_filename = f"{client_id}_{timestamp}.json"
json_path = os.path.join(MEDIA_PLANS_DIR, json_filename)
with open(json_path, 'w') as f:
json.dump(parsed, f, indent=2)
# Remove old plan files for this client
db = _load_media_plans_db()
old_plan = db.get(client_id)
if old_plan:
for key in ('excel_path', 'json_path'):
old_path = old_plan.get(key)
if old_path and os.path.exists(old_path):
try:
os.remove(old_path)
except Exception:
pass
# Save to DB
db[client_id] = {
'client_id': client_id,
'original_filename': file.filename,
'display_name': display_name or file.filename,
'excel_path': excel_path,
'json_path': json_path,
'upload_date': datetime.now().isoformat(),
'total_assets': parsed['total_assets'],
'channels': parsed['channels'],
}
_save_media_plans_db(db)
return jsonify({
'status': 'success',
'message': f'Media plan uploaded with {parsed["total_assets"]} assets across {len(parsed["channels"])} channels',
'total_assets': parsed['total_assets'],
'channels': parsed['channels'],
})
except Exception as e:
print(f"Error uploading media plan: {e}")
import traceback
traceback.print_exc()
return jsonify({'status': 'error', 'message': str(e)}), 500
@app.route('/api/media_plan', methods=['GET'])
@auth.require_auth
def get_media_plan():
"""Get the active media plan for a client"""
client_id = request.args.get('client', '').strip().lower()
if not client_id:
return jsonify({'status': 'error', 'message': 'client parameter required'}), 400
access_err = _require_client_access(client_id)
if access_err:
return access_err
db = _load_media_plans_db()
plan_info = db.get(client_id)
if not plan_info:
return jsonify({'status': 'success', 'plan': None})
return jsonify({
'status': 'success',
'plan': {
'original_filename': plan_info.get('original_filename'),
'upload_date': plan_info.get('upload_date'),
'total_assets': plan_info.get('total_assets'),
'channels': plan_info.get('channels'),
}
})
@app.route('/api/media_plan/<client_id>', methods=['DELETE'])
@auth.require_auth
def delete_media_plan(client_id):
"""Delete the media plan for a client"""
access_err = _require_client_access(client_id)
if access_err:
return access_err
db = _load_media_plans_db()
plan_info = db.get(client_id)
if not plan_info:
return jsonify({'status': 'error', 'message': 'No media plan found'}), 404
# Delete files
for key in ('excel_path', 'json_path'):
path = plan_info.get(key)
if path and os.path.exists(path):
try:
os.remove(path)
except Exception:
pass
del db[client_id]
_save_media_plans_db(db)
return jsonify({'status': 'success', 'message': 'Media plan deleted'})
@app.route('/api/detect_brand', methods=['POST'])
def detect_brand_from_file():
"""Detect brand from uploaded file using AI analysis"""
try:
if 'file' not in request.files:
return jsonify({'status': 'error', 'message': 'No file provided'}), 400
file = request.files['file']
if file.filename == '':
return jsonify({'status': 'error', 'message': 'No file selected'}), 400
# Save file temporarily for analysis
temp_filename = f"brand_detect_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{file.filename}"
temp_path = os.path.join(app.config['UPLOAD_FOLDER'], temp_filename)
file.save(temp_path)
try:
# Get available brands
available_brands = brand_db.get_all_brands()
# Create brand detection prompt - works whether brands are available or not
if available_brands:
brands_list = ', '.join(available_brands)
brand_detection_prompt = f"""
Analyze this image and determine:
1. Which brand this belongs to from the following options: {brands_list}
2. Whether this is a Key Visual or POS (Point of Sale) material
3. Your confidence level (0-100%)
Look for brand logos, colors, typography, and other brand elements.
If the brand is not in the provided list, still attempt to identify it.
Respond with JSON format:
{{
"detected_brand": "brand_name or null if uncertain",
"file_type": "Key Visual" or "POS",
"confidence": confidence_percentage,
"reasoning": "explanation of detection"
}}
"""
else:
brand_detection_prompt = """
Analyze this image and determine:
1. What brand this belongs to (identify any visible brand names, logos, or distinctive brand elements)
2. Whether this is a Key Visual or POS (Point of Sale) material
3. Your confidence level (0-100%)
Look for brand logos, colors, typography, and other brand elements.
Even if no brand guidelines are available, attempt to identify the brand from visual elements.
Respond with JSON format:
{
"detected_brand": "brand_name or null if uncertain",
"file_type": "Key Visual" or "POS",
"confidence": confidence_percentage,
"reasoning": "explanation of detection"
}
"""
# Run AI analysis
result = run_visual_qc(
prompt=brand_detection_prompt,
asset_path=temp_path,
model_name="Gemini"
)
# Extract JSON from response
detection_data = extract_json_from_response(result['response'])
# Clean up temp file
os.remove(temp_path)
return jsonify({
'status': 'success',
'detection': {
'detected_brand': detection_data.get('detected_brand'),
'file_type': detection_data.get('file_type', 'unknown'),
'confidence': detection_data.get('confidence', 0),
'reasoning': detection_data.get('reasoning', ''),
'available_brands': available_brands
}
})
except Exception as e:
# Clean up temp file on error
if os.path.exists(temp_path):
os.remove(temp_path)
raise e
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/profile_usage_stats', methods=['GET'])
def get_profile_usage_stats():
"""Get usage statistics for a specific profile over a time period"""
try:
profile_id = request.args.get('profile', '')
days = int(request.args.get('days', 30))
if not profile_id:
return jsonify({'status': 'error', 'message': 'Profile parameter required'}), 400
# Import load_logs from generate_usage_report
from generate_usage_report import load_logs
start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
end_date = datetime.now().strftime('%Y-%m-%d')
entries = load_logs(start_date=start_date, end_date=end_date)
# Filter to analysis_complete events for this profile
profile_entries = [
e for e in entries
if e.get('event') == 'analysis_complete'
and e.get('profile_id', '') == profile_id
]
total_analyses = len(profile_entries)
total_checks = sum(e.get('checks_run', e.get('checks_completed', 0)) for e in profile_entries)
total_cost = sum(e.get('total_cost_usd', e.get('estimated_cost_usd', 0)) for e in profile_entries)
scores = [e.get('overall_score', 0) for e in profile_entries if e.get('overall_score') is not None]
avg_score = round(sum(scores) / len(scores), 1) if scores else 0
# Build recent analyses list (last 50)
recent = []
for e in sorted(profile_entries, key=lambda x: x.get('timestamp', ''), reverse=True)[:50]:
recent.append({
'date': e.get('timestamp', '')[:19],
'user': e.get('user_email', 'Unknown'),
'client': e.get('client', ''),
'checks': e.get('checks_run', e.get('checks_completed', 0)),
'score': round(e.get('overall_score', 0), 2),
'cost': round(e.get('total_cost_usd', e.get('estimated_cost_usd', 0)), 4)
})
return jsonify({
'status': 'success',
'total_analyses': total_analyses,
'total_checks': total_checks,
'total_cost_usd': round(total_cost, 4),
'avg_score': avg_score,
'recent': recent
})
except Exception as e:
print(f"Error getting profile usage stats: {e}")
import traceback
traceback.print_exc()
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/client_usage_stats', methods=['GET'])
@auth.require_auth
def get_client_usage_stats():
"""Get usage statistics scoped to a specific client with date filtering"""
try:
client = request.args.get('client')
if not client:
return jsonify({'status': 'error', 'message': 'client parameter required'}), 400
access_err = _require_client_access(client)
if access_err:
return access_err
start_date = request.args.get('start_date')
end_date = request.args.get('end_date')
# Default to last 30 days
if not start_date:
start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')
if not end_date:
end_date = datetime.now().strftime('%Y-%m-%d')
from generate_usage_report import load_logs
entries = load_logs(start_date=start_date, end_date=end_date)
# Filter to analysis_complete events for this client
client_entries = [
e for e in entries
if e.get('event') == 'analysis_complete' and e.get('client') == client
]
unique_users = set(e.get('user_email', 'unknown') for e in client_entries)
total_checks = sum(e.get('checks_run', e.get('checks_completed', 0)) for e in client_entries)
total_cost = sum(e.get('total_cost_usd', e.get('estimated_cost_usd', 0)) for e in client_entries)
scores = [e.get('overall_score', 0) for e in client_entries if e.get('overall_score') is not None]
avg_score = round(sum(scores) / len(scores), 1) if scores else 0
total_input_tokens = 0
total_output_tokens = 0
by_provider = {}
for e in client_entries:
tu = e.get('token_usage') or {}
total_input_tokens += tu.get('total_prompt_tokens', 0)
total_output_tokens += tu.get('total_completion_tokens', 0)
for provider, pstats in (tu.get('by_provider') or {}).items():
agg = by_provider.setdefault(provider, {
'input_tokens': 0,
'output_tokens': 0,
'cost_usd': 0.0
})
agg['input_tokens'] += pstats.get('prompt_tokens', 0)
agg['output_tokens'] += pstats.get('completion_tokens', 0)
agg['cost_usd'] += pstats.get('cost', 0)
for agg in by_provider.values():
agg['cost_usd'] = round(agg['cost_usd'], 4)
# Build recent analyses list (last 50)
recent = []
for e in sorted(client_entries, key=lambda x: x.get('timestamp', ''), reverse=True)[:50]:
tu = e.get('token_usage') or {}
recent.append({
'date': e.get('timestamp', '')[:19],
'user': e.get('user_email', 'Unknown'),
'profile': e.get('profile_id', e.get('profile', '')),
'checks': e.get('checks_run', e.get('checks_completed', 0)),
'score': round(e.get('overall_score', 0), 2),
'cost': round(e.get('total_cost_usd', e.get('estimated_cost_usd', 0)), 4),
'input_tokens': tu.get('total_prompt_tokens', 0),
'output_tokens': tu.get('total_completion_tokens', 0)
})
return jsonify({
'status': 'success',
'client': client,
'start_date': start_date,
'end_date': end_date,
'total_analyses': len(client_entries),
'unique_users': len(unique_users),
'total_checks': total_checks,
'estimated_cost_usd': round(total_cost, 4),
'total_input_tokens': total_input_tokens,
'total_output_tokens': total_output_tokens,
'by_provider': by_provider,
'avg_score': avg_score,
'recent': recent
})
except Exception as e:
print(f"Error getting client usage stats: {e}")
import traceback
traceback.print_exc()
return jsonify({'status': 'error', 'message': str(e)}), 500
@app.route('/api/admin/check', methods=['GET'])
@auth.require_auth
def check_admin():
"""Check if the current user is an admin"""
from user_access import is_admin
user_email = getattr(g, 'user', {}).get('email', '')
return jsonify({'is_admin': is_admin(user_email)})
@app.route('/api/admin/users', methods=['GET'])
@auth.require_auth
def get_admin_users():
"""Get all users who have accessed the platform (admin only)"""
from user_access import is_admin
user_email = getattr(g, 'user', {}).get('email', '')
if not is_admin(user_email):
return jsonify({'status': 'error', 'message': 'Admin access required'}), 403
try:
from generate_usage_report import load_logs
entries = load_logs() # All time
users = {}
total_platform_analyses = 0
total_platform_cost = 0
total_platform_input_tokens = 0
total_platform_output_tokens = 0
for e in entries:
event_type = e.get('event', '')
email = e.get('user_email', 'unknown')
if not email or email == 'unknown':
continue
# Track user from any event type (login visits + analyses)
if email not in users:
users[email] = {
'name': e.get('user_name', ''),
'email': email,
'total_analyses': 0,
'total_checks': 0,
'clients': set(),
'last_active': '',
'total_cost': 0,
'input_tokens': 0,
'output_tokens': 0
}
# Update last active from any event
ts = e.get('timestamp', '')
if ts > users[email]['last_active']:
users[email]['last_active'] = ts
# Update name if we have a better one
name = e.get('user_name', '')
if name and not users[email]['name']:
users[email]['name'] = name
# Only count analysis stats from analysis_complete events
if event_type == 'analysis_complete':
total_platform_analyses += 1
cost = e.get('total_cost_usd', e.get('estimated_cost_usd', 0))
total_platform_cost += cost
tu = e.get('token_usage') or {}
input_tokens = tu.get('total_prompt_tokens', 0)
output_tokens = tu.get('total_completion_tokens', 0)
total_platform_input_tokens += input_tokens
total_platform_output_tokens += output_tokens
users[email]['total_analyses'] += 1
users[email]['total_checks'] += e.get('checks_run', e.get('checks_completed', 0))
client = e.get('client', '')
if client:
users[email]['clients'].add(client)
users[email]['total_cost'] += cost
users[email]['input_tokens'] += input_tokens
users[email]['output_tokens'] += output_tokens
# Convert sets to lists for JSON serialization
user_list = []
for u in users.values():
u['clients'] = sorted(list(u['clients']))
u['total_cost'] = round(u['total_cost'], 4)
user_list.append(u)
# Sort by last active descending
user_list.sort(key=lambda x: x['last_active'], reverse=True)
return jsonify({
'status': 'success',
'users': user_list,
'total_unique_users': len(users),
'total_platform_analyses': total_platform_analyses,
'total_platform_cost': round(total_platform_cost, 4),
'total_platform_input_tokens': total_platform_input_tokens,
'total_platform_output_tokens': total_platform_output_tokens
})
except Exception as e:
print(f"Error getting admin users: {e}")
import traceback
traceback.print_exc()
return jsonify({'status': 'error', 'message': str(e)}), 500
def _require_admin():
"""Return (user_email, None) if admin, else (None, error_response)."""
from user_access import is_admin
user_email = getattr(g, 'user', {}).get('email', '')
if not is_admin(user_email):
return None, (jsonify({'status': 'error', 'message': 'Admin access required'}), 403)
return user_email, None
def _require_client_access(client_id):
"""
Validate the authed user can access `client_id`.
Returns None on success, or a (response, status) tuple to short-circuit the endpoint.
"""
if not client_id:
return None # endpoint didn't scope by client
from user_access import get_user_clients
user_email = getattr(g, 'user', {}).get('email', '')
if not user_email:
return jsonify({'status': 'error', 'message': 'Authentication required'}), 401
allowed = get_user_clients(user_email)
if client_id not in allowed:
return jsonify({
'status': 'error',
'code': 'client_access_denied',
'message': f'You do not have access to client "{client_id}"'
}), 403
return None
@app.route('/api/admin/user_access', methods=['GET'])
@auth.require_auth
def get_user_access_list():
"""List all users with their client grants. Joins login-log users with explicit grants."""
_, err = _require_admin()
if err:
return err
try:
from user_access import list_access_entries
from generate_usage_report import load_logs
access = list_access_entries()
entries_by_email = {e['email'].lower(): e for e in access['entries']}
# Enrich with login-log data (name + last_active) for users who have signed in
login_users = {}
for log_entry in load_logs():
email = log_entry.get('user_email')
if not email:
continue
lower = email.lower()
ts = log_entry.get('timestamp', '')
if lower not in login_users or ts > login_users[lower].get('last_active', ''):
login_users[lower] = {
'email': email,
'name': log_entry.get('user_name', ''),
'last_active': ts
}
# Merge: everyone in login logs + everyone with an explicit grant
all_emails = set(login_users.keys()) | set(entries_by_email.keys())
merged = []
for lower in all_emails:
entry = entries_by_email.get(lower, {
'email': login_users.get(lower, {}).get('email', lower),
'clients': access['default_clients'],
'is_admin': False,
'updated_at': None,
'updated_by': None
})
login = login_users.get(lower, {})
merged.append({
**entry,
'name': login.get('name', ''),
'last_active': login.get('last_active', ''),
'has_explicit_grant': lower in entries_by_email
})
merged.sort(key=lambda x: (not x.get('is_admin'), x.get('email', '').lower()))
return jsonify({
'status': 'success',
'default_clients': access['default_clients'],
'users': merged
})
except Exception as e:
print(f"Error listing user access: {e}")
import traceback
traceback.print_exc()
return jsonify({'status': 'error', 'message': str(e)}), 500
@app.route('/api/admin/user_access/<path:target_email>', methods=['PUT'])
@auth.require_auth
def set_user_access(target_email):
"""Set the list of clients a user can see. Body: {"clients": ["general", ...]}"""
actor_email, err = _require_admin()
if err:
return err
data = request.get_json(silent=True) or {}
clients = data.get('clients')
if not isinstance(clients, list):
return jsonify({'status': 'error', 'message': 'clients must be a list'}), 400
try:
from user_access import set_user_clients
from usage_tracker import log_access_change
audit = set_user_clients(target_email, clients, actor_email)
log_access_change(audit)
return jsonify({'status': 'success', 'audit': audit})
except ValueError as ve:
return jsonify({'status': 'error', 'message': str(ve)}), 400
except Exception as e:
print(f"Error setting user access: {e}")
return jsonify({'status': 'error', 'message': str(e)}), 500
@app.route('/api/admin/user_access/<path:target_email>/promote', methods=['POST'])
@auth.require_auth
def promote_user_admin(target_email):
"""Promote a user to admin."""
actor_email, err = _require_admin()
if err:
return err
try:
from user_access import promote_admin
from usage_tracker import log_access_change
audit = promote_admin(target_email, actor_email)
log_access_change(audit)
return jsonify({'status': 'success', 'audit': audit})
except ValueError as ve:
return jsonify({'status': 'error', 'message': str(ve)}), 400
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)}), 500
@app.route('/api/admin/user_access/<path:target_email>/demote', methods=['POST'])
@auth.require_auth
def demote_user_admin(target_email):
"""Remove admin role from a user. Blocked if it would leave zero admins."""
actor_email, err = _require_admin()
if err:
return err
try:
from user_access import demote_admin
from usage_tracker import log_access_change
audit = demote_admin(target_email, actor_email)
log_access_change(audit)
return jsonify({'status': 'success', 'audit': audit})
except ValueError as ve:
return jsonify({'status': 'error', 'message': str(ve)}), 400
except Exception as e:
return jsonify({'status': 'error', 'message': str(e)}), 500
@app.route('/api/consolidate_reports', methods=['POST'])
@auth.require_auth
def consolidate_reports():
"""Generate a consolidated HTML summary from multiple saved reports"""
try:
data = request.get_json()
if not data or 'files' not in data:
return jsonify({'status': 'error', 'message': 'files list required'}), 400
files_list = data['files']
if len(files_list) < 2:
return jsonify({'status': 'error', 'message': 'At least 2 files required'}), 400
output_folder = app.config['OUTPUT_FOLDER']
reports = []
for file_info in files_list:
filename = file_info.get('filename', '')
client = file_info.get('client', '')
file_path = os.path.join(output_folder, client, filename) if client else os.path.join(output_folder, filename)
if not os.path.exists(file_path) or not file_path.endswith('.html'):
continue
with open(file_path, 'r') as f:
html_content = f.read()
# Extract embedded JSON from <div id="json-data"><pre>...</pre></div>
json_match = re.search(r'<div id="json-data"[^>]*>\s*<pre>(.*?)</pre>\s*</div>', html_content, re.DOTALL)
if json_match:
try:
report_data = json.loads(json_match.group(1))
reports.append({
'filename': filename,
'client': client,
'data': report_data
})
except json.JSONDecodeError:
continue
if len(reports) < 2:
return jsonify({'status': 'error', 'message': 'Could not extract data from enough reports (need at least 2)'}), 400
# Build consolidated HTML
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
# Summary table rows
summary_rows = ''
all_failed_checks = []
for r in reports:
rd = r['data']
summary = rd.get('summary', {})
score = summary.get('overall_score', 0)
# Use the grade from the report (respects profile-specific grading overrides)
grade = summary.get('grade', rd.get('grade', determine_grade(score) if score else 'N/A'))
profile_name = rd.get('profile_name', rd.get('profile', 'Unknown'))
fname = rd.get('filename', r['filename'])
pass_fail_color = '#2e7d32' if grade == 'Pass' else '#c62828'
summary_rows += f'''
<tr>
<td style="padding: 10px; border-bottom: 1px solid #dee2e6;">{fname}</td>
<td style="padding: 10px; border-bottom: 1px solid #dee2e6;">{profile_name}</td>
<td style="padding: 10px; border-bottom: 1px solid #dee2e6; text-align: center;">{round(score, 1) if score else 'N/A'}</td>
<td style="padding: 10px; border-bottom: 1px solid #dee2e6; text-align: center; color: {pass_fail_color}; font-weight: 700;">{grade}</td>
</tr>'''
# Collect failed checks
results = rd.get('results', {})
for check_name, check_data in results.items():
if check_data.get('status') != 'success':
continue
check_score = check_data.get('score', 0)
if check_score is not None and check_score < 6:
# Extract human-readable explanation from json_data
json_data = check_data.get('json_data', {})
explanation = json_data.get('explanation', '')
recommendations = json_data.get('recommendations', [])
# Build issue summary from explanation + recommendations
issue_parts = []
if explanation:
issue_parts.append(explanation)
if recommendations:
rec_text = ' | '.join(r for r in recommendations if r)
if rec_text:
issue_parts.append(f'Recommendations: {rec_text}')
issue_summary = ' '.join(issue_parts) if issue_parts else 'No details available'
# Truncate if very long
if len(issue_summary) > 500:
issue_summary = issue_summary[:497] + '...'
all_failed_checks.append({
'filename': fname,
'check': check_data.get('display_name', check_name.replace('_', ' ').title()),
'score': check_score,
'summary': issue_summary
})
# Failed checks rows
failed_rows = ''
if all_failed_checks:
for fc in all_failed_checks:
failed_rows += f'''
<tr style="background: #fff5f5;">
<td style="padding: 10px; border-bottom: 1px solid #dee2e6;">{fc['filename']}</td>
<td style="padding: 10px; border-bottom: 1px solid #dee2e6;">{fc['check']}</td>
<td style="padding: 10px; border-bottom: 1px solid #dee2e6; text-align: center; color: #c62828; font-weight: 700;">{fc['score']}/10</td>
<td style="padding: 10px; border-bottom: 1px solid #dee2e6; font-size: 0.85em; color: #555;">{fc['summary']}</td>
</tr>'''
else:
failed_rows = '<tr><td colspan="4" style="padding: 20px; text-align: center; color: #2e7d32;">No failed checks found across all reports!</td></tr>'
consolidated_html = f'''<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Consolidated QC Report - {timestamp}</title>
<link href="https://fonts.googleapis.com/css2?family=Montserrat:wght@400;600;700&display=swap" rel="stylesheet">
<style>
body {{ font-family: 'Montserrat', sans-serif; margin: 0; padding: 20px; background: #f5f5f5; color: #333; }}
.container {{ max-width: 1000px; margin: 0 auto; background: white; border-radius: 12px; box-shadow: 0 2px 12px rgba(0,0,0,0.1); padding: 30px; }}
h1 {{ color: #1a1a2e; margin-bottom: 5px; }}
h2 {{ color: #495057; border-bottom: 2px solid #dee2e6; padding-bottom: 8px; margin-top: 30px; }}
.meta {{ color: #6c757d; font-size: 0.9em; margin-bottom: 25px; }}
table {{ width: 100%; border-collapse: collapse; margin: 15px 0; table-layout: fixed; }}
th {{ background: #f8f9fa; padding: 12px 10px; text-align: left; border-bottom: 2px solid #dee2e6; font-weight: 700; font-size: 0.9em; }}
td {{ font-size: 0.9em; word-break: break-word; overflow-wrap: break-word; }}
</style>
</head>
<body>
<div class="container">
<h1>Consolidated QC Report</h1>
<p class="meta">Generated: {timestamp} | Reports: {len(reports)}</p>
<h2>Summary</h2>
<table>
<thead>
<tr>
<th>File</th>
<th>Profile</th>
<th style="text-align: center;">Score</th>
<th style="text-align: center;">Result</th>
</tr>
</thead>
<tbody>{summary_rows}</tbody>
</table>
<h2>Failed Checks</h2>
<table>
<thead>
<tr>
<th>File</th>
<th>Check</th>
<th style="text-align: center;">Score</th>
<th>Issue Summary</th>
</tr>
</thead>
<tbody>{failed_rows}</tbody>
</table>
</div>
</body>
</html>'''
# Save consolidated report
consolidated_filename = f"consolidated_{datetime.now().strftime('%Y%m%d_%H%M%S')}.html"
# Determine client folder - use the first report's client or 'general'
client_folder = reports[0].get('client', 'general') or 'general'
client_output_dir = os.path.join(output_folder, client_folder)
os.makedirs(client_output_dir, exist_ok=True)
consolidated_path = os.path.join(client_output_dir, consolidated_filename)
with open(consolidated_path, 'w') as f:
f.write(consolidated_html)
url = f"{request.environ.get('SCRIPT_NAME', '')}/output/{client_folder}/{consolidated_filename}"
return jsonify({
'status': 'success',
'filename': consolidated_filename,
'url': url
})
except Exception as e:
print(f"Error consolidating reports: {e}")
import traceback
traceback.print_exc()
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/qc-apps', methods=['GET'])
def get_qc_apps():
"""Get list of all QC applications with their metadata"""
try:
# User-friendly descriptions for each QC tool
tool_descriptions = {
'accessibility': 'Evaluates color contrast, text legibility, and accessibility compliance to ensure content is usable by people with visual impairments and meets WCAG standards.',
'animation_transitions': 'Checks animation quality, smoothness, timing, and transitions in video or animated content to ensure professional motion design.',
'aspect_ratio': 'Verifies the image or video aspect ratio matches intended display formats (16:9, 4:3, 1:1, etc.) to prevent stretching or cropping issues.',
'background_contrast': 'Analyzes contrast between background and foreground elements to ensure text and important content remain clearly visible and readable.',
'brand_assets_visibility': 'Validates that required brand assets (logos, trademarks, brand elements) are present, visible, and properly positioned according to brand guidelines.',
'call_to_action': 'Evaluates the presence, clarity, and prominence of call-to-action (CTA) elements to ensure they effectively guide user behavior.',
'color_format': 'Checks color space and format (RGB, CMYK, etc.) to ensure correct color reproduction for the intended output medium (digital, print).',
'crop_marks': 'Verifies presence and correct placement of crop marks, bleeds, and trim marks for print-ready files to ensure accurate cutting and finishing.',
'curved_edges': 'Checks for smooth, professional curved edges and rounded corners in design elements, ensuring consistent styling across the creative.',
'curved_edges_digital': 'Evaluates curved edges specifically for digital displays, checking for anti-aliasing and smooth rendering on screens.',
'curved_edges_print': 'Validates curved edges for print output, ensuring clean vector paths and proper resolution for physical reproduction.',
'dark_mode_legibility': 'Tests content legibility in dark mode or low-light environments, ensuring text remains readable with appropriate contrast.',
'element_alignment': 'Checks alignment of design elements (text, images, buttons) to ensure professional layout and visual consistency.',
'face_gaze_direction': 'Analyzes the direction people in images are looking to ensure gaze guides viewer attention toward key content or CTAs.',
'face_visibility': 'Checks for clear, well-lit, and unobscured faces in imagery, ensuring people are properly featured when human connection is important.',
'file_naming': 'Validates file naming conventions for proper organization, version control, and asset management in production workflows.',
'image_resolution': 'Verifies image resolution and quality to ensure crisp, professional output without pixelation or blur at intended display size.',
'imperative_verb': 'Checks that CTAs use strong imperative verbs ("Get," "Start," "Join") to create clear, action-oriented messaging.',
'inclusive': 'Evaluates representation and inclusivity in imagery and messaging to ensure diverse, respectful, and welcoming content.',
'layer_organization': 'Reviews file layer structure and organization for clean, maintainable design files that other team members can easily work with.',
'logo_visibility': 'Ensures brand logos are present, properly sized, clearly visible, and positioned according to brand guidelines.',
'lowercase_text': 'Checks for improper use of all-lowercase text where standard capitalization would improve readability and professionalism.',
'new_visibility': 'Validates presence and visibility of "NEW" badges or indicators when promoting new products, features, or offerings.',
'print_bleed': 'Verifies correct bleed area setup for print files to prevent white edges and ensure full-coverage printing to the trim edge.',
'product_visibility': 'Ensures product imagery is clear, well-lit, properly featured, and showcases key product attributes effectively.',
'responsiveness': 'Checks how design elements adapt across different screen sizes and devices to ensure consistent experience on mobile, tablet, and desktop.',
'safety_area': 'Validates that important content (text, logos, CTAs) stays within safe zones to avoid cutoff on various display formats and aspect ratios.',
'supporting_images': 'Evaluates quality, relevance, and composition of supporting images to ensure they enhance the message and maintain visual interest.',
'text_readability': 'Analyzes text size, font choice, spacing, and contrast to ensure copy is easily readable across different viewing conditions.',
'visual_elements_count': 'Checks for appropriate number of visual elements to avoid cluttered or empty designs, ensuring balanced composition.',
'visual_hierarchy': 'Evaluates the logical flow and priority of visual elements to guide viewer attention from most to least important information.',
'visuals_left_text_right': 'Validates proper layout convention (visuals left, text right) for left-to-right reading audiences to optimize information flow.',
'word_count': 'Checks text length to ensure messaging is concise and appropriate for the medium, avoiding over-crowding or insufficient information.'
}
qc_apps_data = {}
# Build QC apps data from loaded QC checks
for check_name in qc_apps.keys():
# Get display name
display_name = qc_apps[check_name].get('display_name', check_name.replace('_', ' ').title())
# Get description from mapping, or fall back to auto-generated description
description = tool_descriptions.get(check_name)
if not description:
# Fallback: Extract from prompt if custom description not available
prompt = qc_apps[check_name].get('prompt', '')
description_lines = prompt.split('\n\n')
description = description_lines[0][:300] if description_lines else f'{display_name} quality check'
# Default weights - can be customized
qc_apps_data[check_name] = {
'display_name': display_name,
'enabled': True,
'required_weight': 0.1, # Default weight
'optional_weight': 0.0,
'description': description,
'full_prompt': qc_apps[check_name].get('prompt', '') # Include full prompt for detailed view
}
return jsonify({
'status': 'success',
'qc_apps': qc_apps_data,
'total_apps': len(qc_apps_data)
})
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/generate_report', methods=['POST'])
def api_generate_report():
"""Generate HTML report from existing session data"""
try:
# Get parameters
session_id = request.form.get('session_id')
if not session_id:
return jsonify({'status': 'error', 'message': 'Session ID required'}), 400
session_folder = os.path.join(app.config['UPLOAD_FOLDER'], session_id)
if not os.path.exists(session_folder):
return jsonify({'status': 'error', 'message': 'Session not found'}), 404
# Find the analysis result file
result_files = [f for f in os.listdir(session_folder) if f.endswith('_results.json')]
if not result_files:
return jsonify({'status': 'error', 'message': 'No analysis results found for session'}), 404
# Load the most recent results file
result_file = sorted(result_files)[-1]
result_path = os.path.join(session_folder, result_file)
with open(result_path, 'r') as f:
report_data = json.load(f)
# Find the original file
files = [f for f in os.listdir(session_folder) if not f.endswith('_results.json')]
if not files:
return jsonify({'status': 'error', 'message': 'Original file not found'}), 404
original_file = files[0]
file_path = os.path.join(session_folder, original_file)
# Generate HTML using existing function
html_content = generate_html_content(report_data, original_file, file_path)
return html_content, 200, {'Content-Type': 'text/html'}
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e)
}), 500
# Authentication endpoints
@app.route('/auth/login', methods=['POST'])
def auth_login():
"""Process authentication tokens from MSAL popup."""
try:
# Get token from request
data = request.get_json()
if not data or 'token' not in data:
return jsonify({
'success': False,
'error': 'Token is required',
'authenticated': False
}), 400
token = data['token']
# Validate and set authentication token - returns response with cookie set
return auth.set_auth_token(token)
except Exception as e:
return jsonify({
'success': False,
'error': f'Authentication failed: {str(e)}',
'authenticated': False
}), 500
@app.route('/auth/logout', methods=['POST'])
def auth_logout():
"""Clear authentication session."""
try:
return auth.clear_auth_token()
except Exception as e:
return jsonify({
'success': False,
'error': f'Logout failed: {str(e)}'
}), 500
@app.route('/auth/status', methods=['GET'])
def auth_status():
"""Get current authentication status and log user visit."""
try:
status = auth.get_auth_status()
# Log user visit if authenticated
if status.get('authenticated') and status.get('user'):
try:
from usage_tracker import log_user_login
log_user_login(status['user'])
except Exception as log_err:
print(f"Warning: Failed to log user visit: {log_err}")
return jsonify(status)
except Exception as e:
return jsonify({
'authenticated': False,
'error': f'Status check failed: {str(e)}'
}), 500
# ---------------------------------------------------------------------------
# Box.com OAuth (per-creator user authentication for automation folders).
# Flow:
# 1. User (signed into the QC tool via MSAL) hits GET /auth/box/login.
# We generate a JWT-signed state encoding their email and 302 to Box.
# 2. Box redirects back to GET /auth/box/callback with ?code=...&state=...
# We verify the state, exchange the code for tokens, fetch the Box user,
# and persist via box_tokens.save_tokens.
# Tokens are then used by background workers (PR3) and on-demand calls.
#
# The Box redirect URI is computed per-request (not a fixed env var) so the
# same code works on laptop, dev, and prod. When BOX_REDIRECT_URI is set in
# the env it wins (escape hatch).
# ---------------------------------------------------------------------------
def _box_redirect_uri():
"""
Compute the public OAuth callback URL.
Resolution order:
1. BOX_REDIRECT_URI env var if set (escape hatch / unusual deploys).
2. X-Forwarded-Host header if Apache sets it (some setups do).
3. Otherwise, infer from request.host: anything that isn't localhost
is treated as being behind the Apache proxy at /ai_qc/ over HTTPS
(this matches optical-dev / optical-prod where Apache uses
ProxyPreserveHost so request.host is already the public hostname,
but the backend connection is plain http and Flask sees no prefix).
"""
explicit = (os.environ.get('BOX_REDIRECT_URI') or '').strip()
if explicit:
return explicit
forwarded_host = request.headers.get('X-Forwarded-Host')
if forwarded_host:
host = forwarded_host.split(',')[0].strip()
proto = request.headers.get('X-Forwarded-Proto', 'https')
return f'{proto}://{host}/ai_qc/auth/box/callback'
host = (request.host or '').strip()
is_local = (not host) or 'localhost' in host or host.startswith('127.0.0.1')
if not is_local:
# Behind the optical-dev / optical-prod Apache proxy, mounted at /ai_qc/.
return f'https://{host}/ai_qc/auth/box/callback'
return f'{request.scheme}://{host}/auth/box/callback'
# ---------- Box JWT service-account: webhook ingestion + workflow helper ----------
# Bounded in-memory dedup for box-delivery-id. Box uses at-least-once delivery;
# a 200 from us tells it not to retry, so duplicates are rare. The maxlen keeps
# memory tiny while still catching the common retry window.
_box_recent_deliveries = collections.deque(maxlen=500)
_box_recent_deliveries_set = set()
_box_recent_deliveries_lock = threading.Lock()
# Extensions accepted from a Box upload. Keeps the webhook from kicking off QC
# on Word docs, ZIPs, etc. Mirrors what technical_check.inspect knows how to read.
_BOX_QC_EXTS = {
'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.tif', '.webp',
'.pdf',
'.mp4', '.avi', '.mov', '.mkv', '.wmv', '.flv', '.webm',
}
def _box_remember_delivery(delivery_id):
"""Return True if first time seeing this delivery_id; False if duplicate."""
with _box_recent_deliveries_lock:
if delivery_id in _box_recent_deliveries_set:
return False
_box_recent_deliveries.append(delivery_id)
_box_recent_deliveries_set.add(delivery_id)
if len(_box_recent_deliveries_set) > _box_recent_deliveries.maxlen:
_box_recent_deliveries_set.clear()
_box_recent_deliveries_set.update(_box_recent_deliveries)
return True
def _run_box_triggered_analysis(client_id, profile_id, file_id, filename, session_id):
"""Background worker for the Box webhook flow.
Downloads the Box file, runs the technical pre-flight + LLM check pipeline,
writes the HTML report to disk under output/<client>/, and uploads the same
report back to the client's box_reports_folder_id (or box_folder_id as a
fallback). Uses a synthetic 'box_webhook' user for usage tracking.
Skips media-plan and localization context — those are user-UI concepts that
don't have a meaningful source in a webhook-triggered run.
"""
try:
from client_config import get_all_clients as _get_all_clients
client_cfg = _get_all_clients().get(client_id, {})
session_folder = os.path.join(app.config['UPLOAD_FOLDER'], session_id)
os.makedirs(session_folder, exist_ok=True)
file_path = os.path.join(session_folder, filename)
# 1. Download the asset from Box.
box_jwt_client.download_file(file_id, file_path)
print(f'Box webhook: downloaded {filename} → {file_path}')
# 2. Technical pre-flight (same as user-uploaded flow).
technical_report = technical_inspect(file_path)
# 3. Init progress tracker.
progress_tracker[session_id] = {
'total_checks': 25,
'completed_checks': 0,
'current_check': 'Initializing',
'current_check_display': 'Box-triggered analysis',
'stage': 'setup',
'percentage': 0,
'session_id': session_id,
'status': 'started',
'source': 'box_webhook',
'box_file_id': file_id,
'technical_report': technical_report,
}
# 4. Log analysis start with a synthetic system user.
try:
from usage_tracker import log_analysis_start
log_analysis_start(
session_id, client_id, profile_id,
{'user_id': 'box_webhook', 'email': 'box_webhook@system', 'name': 'Box Webhook'},
{'filename': filename, 'size': os.path.getsize(file_path)},
)
except Exception as log_err:
print(f'WARNING: usage log_analysis_start failed: {log_err}')
# 5. Resolve profile + enabled checks.
profile_config = get_profile(profile_id)
if not profile_config:
raise Exception(f'Profile {profile_id} not found')
enabled_checks = [c for c in profile_config.get_enabled_checks() if c in qc_apps]
if not enabled_checks:
raise Exception(f'No enabled checks for profile {profile_id}')
profile_weights = profile_config.get_check_weights()
progress_tracker[session_id].update({
'total_checks': len(enabled_checks),
'stage': 'qc_analysis',
'percentage': 10,
})
# 6. Run check batches (no media plan / localization / OCR in webhook MVP).
check_results = process_checks_in_batches(
enabled_checks, qc_apps, profile_config, profile_weights,
file_path, None, brand_db, progress_tracker,
session_id, batch_size=15, media_plan_context=None, ocr_context=None,
)
# 7. Score aggregation.
total_weighted_score = 0
total_weight = 0
completed_checks = 0
failed_checks = 0
for _check_name, result in check_results.items():
w = result.get('weight', 0.1)
total_weight += w
if result.get('status') == 'success':
completed_checks += 1
s = result.get('score')
if s is not None:
total_weighted_score += s * w
else:
failed_checks += 1
if total_weight >= 10.0:
overall_score = min(total_weighted_score, 100)
else:
overall_score = min(total_weighted_score * 10, 100)
# 8. Result envelope matching the user-flow shape.
result_data = {
'status': 'success',
'session_id': session_id,
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'filename': filename,
'profile': profile_id,
'profile_id': profile_id,
'profile_name': profile_config.name,
'model': 'Profile-based selection',
'results': check_results,
'profile_selection': {
'selected_profile': profile_id,
'profile_source': 'box_webhook',
'brand': client_id,
'format_suffix': profile_id,
'reference_asset': None,
'reference_asset_used': False,
},
'qc_analysis': {
'profile_used': profile_id,
'total_checks': len(enabled_checks),
'completed_checks': completed_checks,
'failed_checks': failed_checks,
'check_results': check_results,
},
'summary': {
'overall_score': round(overall_score, 1),
'profile': profile_config.name,
'checks_count': completed_checks,
'total_checks': len(enabled_checks),
'total_weighted_score': total_weighted_score,
'total_weight': total_weight,
'grade': determine_grade(overall_score),
},
'technical_report': progress_tracker[session_id].get('technical_report', {}),
'source': 'box_webhook',
'box_file_id': file_id,
}
# Strict-grade override applies the same way for webhook-triggered runs.
if getattr(profile_config, 'strict_grade', False):
for _cn, cd in check_results.items():
if cd.get('status') == 'success':
cs = cd.get('score', 0)
if cs is not None and cs < 6:
result_data['summary']['grade'] = 'Fail'
break
# 9. Write HTML report to disk so the UI's saved-files listing shows it.
report_filename = f'QC_Report_{session_id}_{os.path.splitext(filename)[0]}.html'
client_folder = ensure_client_output_folder(client_id)
report_path = os.path.join(client_folder, report_filename)
with open(report_path, 'w', encoding='utf-8') as f:
f.write(generate_comprehensive_html_report(result_data, filename, file_path=file_path))
result_data['output_file'] = {
'path': report_path,
'filename': report_filename,
'url': f'/output/{client_id}/{report_filename}',
}
# 10. Upload the report back to Box. Prefer the dedicated reports folder if
# configured; fall back to the same folder the source lived in.
reports_folder = client_cfg.get('box_reports_folder_id') or client_cfg.get('box_folder_id')
report_uploaded_ok = False
if reports_folder:
try:
uploaded = box_jwt_client.upload_file(report_path, str(reports_folder), name=report_filename)
result_data['box_report_upload'] = {
'box_file_id': uploaded.get('id'),
'box_file_name': uploaded.get('name'),
'box_folder_id': str(reports_folder),
}
report_uploaded_ok = True
print(f"Box webhook: uploaded report {report_filename} → folder {reports_folder} (id={uploaded.get('id')})")
except Exception as up_err:
print(f'WARNING: failed to upload report to Box: {up_err}')
result_data['box_report_upload_error'] = str(up_err)
else:
print(f'Box webhook: no box_reports_folder_id (or box_folder_id) on client {client_id}; report stays local only')
# 10b. Move the source file out of INCOMING into a `_PROCESSED` subfolder so the
# next upload of the same filename triggers a fresh FILE.UPLOADED event (Box's
# V2 webhook doesn't fire on same-name version replacements; freeing the name
# is the cleanest workaround). Only runs if the report made it back to Box —
# if upload failed, we want the source to stay so the user can retry by simply
# re-uploading. Failures here are non-fatal: log, record, continue.
source_folder_id = client_cfg.get('box_folder_id')
if report_uploaded_ok and source_folder_id:
try:
processed_folder_id = box_jwt_client.find_or_create_subfolder(
str(source_folder_id), '_PROCESSED'
)
processed_name = f'{session_id}_{filename}'
moved = box_jwt_client.move_file(file_id, processed_folder_id, new_name=processed_name)
result_data['box_source_moved'] = {
'box_file_id': moved.get('id'),
'box_folder_id': processed_folder_id,
'box_file_name': moved.get('name'),
}
print(f'Box webhook: moved source → _PROCESSED/{processed_name} (folder {processed_folder_id})')
except Exception as mv_err:
print(f'WARNING: failed to move source file to _PROCESSED: {mv_err}')
result_data['box_source_move_error'] = str(mv_err)
# 11. Mark complete on progress_tracker for /api/progress consumers.
progress_tracker[session_id]['result'] = result_data
progress_tracker[session_id]['status'] = 'completed'
progress_tracker[session_id]['stage'] = 'complete'
progress_tracker[session_id]['percentage'] = 100
# 12. Usage tracker completion event.
try:
from usage_tracker import log_analysis_complete
log_analysis_complete(
session_id, client_id, profile_id,
{'user_id': 'box_webhook', 'email': 'box_webhook@system', 'name': 'Box Webhook'},
{'checks_completed': completed_checks, 'overall_score': overall_score,
'status': 'success', 'source': 'box_webhook'},
)
except Exception as log_err:
print(f'WARNING: usage log_analysis_complete failed: {log_err}')
print(f'Box webhook: analysis complete for session {session_id}, score {overall_score}')
except Exception as e:
print(f'ERROR in Box-triggered analysis (session {session_id}): {e}')
import traceback
traceback.print_exc()
if session_id in progress_tracker:
progress_tracker[session_id]['status'] = 'error'
progress_tracker[session_id]['stage'] = 'error'
progress_tracker[session_id]['error'] = str(e)
@app.route('/api/box/webhook', methods=['POST'])
def box_webhook():
"""Receive a Box V2 webhook. Authenticated by HMAC signature on every request.
Box expects a 200 within ~10 seconds. We verify the signature, ack, and run
the analysis on a background thread.
"""
raw_body = request.get_data(cache=True)
headers = {k.lower(): v for k, v in request.headers.items()}
primary_key = os.environ.get('BOX_WEBHOOK_PRIMARY_KEY')
secondary_key = os.environ.get('BOX_WEBHOOK_SECONDARY_KEY')
if not primary_key and not secondary_key:
print('Box webhook: no signing keys in env (set BOX_WEBHOOK_PRIMARY_KEY); refusing all deliveries')
return jsonify({'status': 'error', 'message': 'webhook signing not configured'}), 503
if not box_jwt_client.verify_webhook_signature(raw_body, headers, primary_key, secondary_key):
print('Box webhook: signature verification failed')
return jsonify({'status': 'error', 'message': 'invalid signature'}), 401
delivery_id = headers.get('box-delivery-id', '')
if delivery_id and not _box_remember_delivery(delivery_id):
return jsonify({'status': 'ok', 'message': 'duplicate'}), 200
try:
payload = json.loads(raw_body.decode('utf-8'))
except Exception:
return jsonify({'status': 'error', 'message': 'invalid JSON'}), 400
trigger = payload.get('trigger', '')
if trigger != 'FILE.UPLOADED':
return jsonify({'status': 'ok', 'message': f'ignored trigger {trigger}'}), 200
source = payload.get('source') or {}
if source.get('type') != 'file':
return jsonify({'status': 'ok', 'message': 'not a file event'}), 200
file_id = str(source.get('id', ''))
filename = source.get('name', '')
parent = source.get('parent') or {}
parent_folder_id = str(parent.get('id', ''))
if not file_id or not parent_folder_id or not filename:
return jsonify({'status': 'error', 'message': 'malformed payload'}), 400
ext = os.path.splitext(filename)[1].lower()
if ext not in _BOX_QC_EXTS:
print(f'Box webhook: skipping non-QC extension {ext} ({filename})')
return jsonify({'status': 'ok', 'message': f'unsupported extension {ext}'}), 200
from client_config import get_client_by_box_folder, get_all_clients, get_default_profile
client_id = get_client_by_box_folder(parent_folder_id)
if not client_id:
print(f'Box webhook: no client configured for Box folder {parent_folder_id}')
return jsonify({'status': 'ok', 'message': 'no client mapping'}), 200
client_cfg = get_all_clients().get(client_id, {})
profile_id = get_default_profile(client_id) or (client_cfg.get('profiles') or ['static_general'])[0]
session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
print(f'Box webhook: dispatching session={session_id} client={client_id} profile={profile_id} file_id={file_id} name={filename}')
threading.Thread(
target=_run_box_triggered_analysis,
args=(client_id, profile_id, file_id, filename, session_id),
daemon=True,
).start()
return jsonify({
'status': 'ok',
'session_id': session_id,
'client_id': client_id,
'profile_id': profile_id,
}), 200
@app.route('/auth/box/login', methods=['GET'])
@auth.require_auth
def box_login():
"""Kick off the Box OAuth flow for the signed-in user."""
import box_client
if not box_client.is_configured():
return jsonify({'status': 'error',
'message': 'Box integration is not configured (BOX_CLIENT_ID / BOX_CLIENT_SECRET missing).'}), 503
user_email = getattr(g, 'user', {}).get('email', '')
if not user_email:
return jsonify({'status': 'error', 'message': 'Authentication required'}), 401
redirect_uri = _box_redirect_uri()
state = box_client.make_state(user_email)
return redirect(box_client.build_authorize_url(state, redirect_uri))
@app.route('/auth/box/callback', methods=['GET'])
def box_callback():
"""Handle Box's redirect back after the user approves the app."""
import box_client
import box_tokens
error = request.args.get('error')
if error:
# User cancelled, or Box rejected the request — surface a friendly page.
description = request.args.get('error_description', '')
return _box_callback_html('Box connection cancelled or failed.',
detail=f'{error}: {description}', success=False), 400
code = request.args.get('code')
state = request.args.get('state')
if not code or not state:
return _box_callback_html('Missing code or state in callback.',
success=False), 400
user_email = box_client.verify_state(state)
if not user_email:
return _box_callback_html('Invalid or expired state — please start the connection from the QC tool again.',
success=False), 400
redirect_uri = _box_redirect_uri()
try:
tokens = box_client.exchange_code_for_tokens(code, redirect_uri)
except Exception as e:
print(f'[box_callback] code exchange failed: {e}')
return _box_callback_html('Could not complete the Box token exchange.',
detail=str(e), success=False), 502
box_user = None
try:
box_user = box_client.get_box_user(tokens['access_token'])
except Exception as e:
# Non-fatal — we still got tokens, just couldn't read the Box identity.
print(f'[box_callback] /users/me failed: {e}')
box_tokens.save_tokens(user_email, tokens, box_user=box_user)
return _box_callback_html('Box connected.',
detail=(box_user or {}).get('login') or user_email,
success=True)
def _box_callback_html(message, detail='', success=True):
"""Tiny self-contained HTML response for the OAuth callback."""
color = '#28a745' if success else '#dc3545'
detail_html = f'<p style="color:#6c757d;font-size:0.95em;">{detail}</p>' if detail else ''
body = f"""
<!doctype html>
<html><head><meta charset="utf-8"><title>Box connection</title></head>
<body style="font-family: -apple-system, Segoe UI, Roboto, Helvetica, Arial, sans-serif; padding: 60px 30px; text-align: center; color: #212529;">
<h1 style="color: {color}; margin-bottom: 12px;">{message}</h1>
{detail_html}
<p style="margin-top: 24px;"><a href="../" style="color: #007bff; text-decoration: none;">Return to AI QC →</a></p>
<script>
// If this was opened in a popup, close it and let the opener refresh status.
if (window.opener && !window.opener.closed) {{
try {{ window.opener.postMessage({{ type: 'box-oauth-result', success: {str(success).lower()} }}, '*'); }} catch (e) {{}}
setTimeout(function() {{ window.close(); }}, 1200);
}}
</script>
</body></html>
"""
return Response(body, mimetype='text/html')
@app.route('/api/box/status', methods=['GET'])
@auth.require_auth
def box_status():
"""Whether the current user has connected their Box account."""
import box_client
import box_tokens
user_email = getattr(g, 'user', {}).get('email', '')
record = box_tokens.get_tokens(user_email)
if not record:
return jsonify({
'status': 'success',
'connected': False,
'configured': box_client.is_configured(),
})
return jsonify({
'status': 'success',
'connected': True,
'configured': True,
'box_user_login': record.get('box_user_login'),
'box_user_name': record.get('box_user_name'),
'connected_at': record.get('connected_at'),
'access_token_expires_at': record.get('access_token_expires_at'),
})
@app.route('/api/box/disconnect', methods=['POST'])
@auth.require_auth
def box_disconnect():
"""Forget the current user's Box tokens. Best-effort revoke at Box too."""
import box_client
import box_tokens
user_email = getattr(g, 'user', {}).get('email', '')
record = box_tokens.get_tokens(user_email)
if record and record.get('refresh_token'):
try:
box_client.revoke_tokens(record['refresh_token'])
except Exception as e:
print(f'[box_disconnect] revoke failed for {user_email}: {e}')
box_tokens.delete_tokens(user_email)
return jsonify({'status': 'success', 'message': 'Box disconnected'})
@app.route('/api/box/test_folder', methods=['GET'])
@auth.require_auth
def box_test_folder():
"""
Smoke-test endpoint: list a Box folder's items using the current user's
stored tokens. Used to prove the OAuth round-trip works end-to-end before
we wire the watcher in PR3. Folder ID '0' is the user's All Files root.
"""
import box_client
user_email = getattr(g, 'user', {}).get('email', '')
folder_id = (request.args.get('folder_id') or '0').strip()
access_token = box_client.get_valid_access_token(user_email)
if not access_token:
return jsonify({'status': 'error',
'code': 'box_not_connected',
'message': 'Connect your Box account first via /auth/box/login'}), 401
try:
result = box_client.list_folder_items(access_token, folder_id)
except Exception as e:
print(f'[box_test_folder] list failed: {e}')
return jsonify({'status': 'error', 'message': str(e)}), 502
items = [
{
'id': it.get('id'),
'name': it.get('name'),
'type': it.get('type'),
'size': it.get('size'),
'created_at': it.get('created_at'),
'created_by_login': (it.get('created_by') or {}).get('login'),
}
for it in (result.get('entries') or [])
]
return jsonify({
'status': 'success',
'folder_id': folder_id,
'total_count': result.get('total_count'),
'items': items,
})
@app.route('/api/usage/stats', methods=['GET'])
@auth.require_auth
def get_usage_stats_endpoint():
"""Get usage statistics (admin endpoint)"""
from usage_tracker import get_usage_stats
# Get query parameters
start_date = request.args.get('start_date')
end_date = request.args.get('end_date')
client = request.args.get('client')
user_id = request.args.get('user_id')
try:
stats = get_usage_stats(
start_date=start_date,
end_date=end_date,
client=client,
user_id=user_id
)
return jsonify({
'status': 'success',
'stats': stats
})
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/usage/export', methods=['GET'])
@auth.require_auth
def export_usage_logs_endpoint():
"""Export usage logs as CSV (admin endpoint)"""
import csv
import io
from usage_tracker import USAGE_LOG_DIR
# Get date range
start_date = request.args.get('start_date')
end_date = request.args.get('end_date')
try:
# Collect all log entries
import glob
log_files = glob.glob(os.path.join(USAGE_LOG_DIR, '*.jsonl'))
# Create CSV
output = io.StringIO()
fieldnames = ['timestamp', 'event', 'session_id', 'client', 'profile',
'user_email', 'checks_completed', 'overall_score', 'estimated_cost_usd']
writer = csv.DictWriter(output, fieldnames=fieldnames)
writer.writeheader()
for log_file in sorted(log_files):
with open(log_file, 'r') as f:
for line in f:
try:
entry = json.loads(line.strip())
if entry.get('event') == 'analysis_complete':
writer.writerow({
'timestamp': entry.get('timestamp'),
'event': entry.get('event'),
'session_id': entry.get('session_id'),
'client': entry.get('client'),
'profile': entry.get('profile'),
'user_email': entry.get('user_email'),
'checks_completed': entry.get('checks_completed'),
'overall_score': entry.get('overall_score'),
'estimated_cost_usd': entry.get('estimated_cost_usd')
})
except:
continue
# Return as downloadable CSV
output.seek(0)
return Response(
output.getvalue(),
mimetype='text/csv',
headers={'Content-Disposition': 'attachment; filename=usage_export.csv'}
)
except Exception as e:
return jsonify({
'status': 'error',
'message': str(e)
}), 500
@app.route('/api/debug/test_file_upload', methods=['POST'])
@auth.require_auth
def debug_test_file_upload():
"""Debug endpoint to test file upload handling"""
try:
# Check if file is in request
if 'file' not in request.files:
return jsonify({'status': 'error', 'message': 'No file part', 'files_in_request': list(request.files.keys())}), 400
file = request.files['file']
# Check if file was selected
if file.filename == '':
return jsonify({'status': 'error', 'message': 'No selected file'}), 400
# Get parameters
profile = request.form.get('profile', 'general').lower()
# Create session ID
session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
# Check upload folder configuration
upload_folder = app.config.get('UPLOAD_FOLDER', 'uploads')
return jsonify({
'status': 'success',
'message': 'File upload test successful',
'session_id': session_id,
'filename': file.filename,
'profile': profile,
'upload_folder': upload_folder,
'upload_folder_exists': os.path.exists(upload_folder),
'form_data': dict(request.form),
'files': list(request.files.keys())
})
except Exception as e:
import traceback
return jsonify({
'status': 'error',
'message': str(e),
'traceback': traceback.format_exc()
}), 500
# Initialize application
if not qc_apps:
load_qc_apps()
print(f"Initialized app with {len(qc_apps)} QC apps")
print(f"Brand Guidelines DB initialized: {len(brand_db.get_all_brands())} brands")
# Backfill: process any existing unprocessed PDF guidelines
def _backfill_unprocessed_pdfs():
import threading
unprocessed = []
for file_id, record in brand_db.db.get("files", {}).items():
if record.get("file_type") == ".pdf" and not record.get("processed"):
stored_path = record.get("stored_path", "")
if stored_path and os.path.exists(stored_path):
unprocessed.append((file_id, stored_path, record.get("brand_name", ""), str(brand_db.files_dir)))
if unprocessed:
print(f"Found {len(unprocessed)} unprocessed PDF guidelines - starting background processing")
def _process_all():
for fid, spath, bname, fdir in unprocessed:
try:
from pdf_processor import process_pdf_guideline
result = process_pdf_guideline(spath, fid, bname, fdir)
brand_db.update_file_record(fid, result)
print(f"Backfill complete for {fid}")
except Exception as e:
print(f"Backfill failed for {fid}: {e}")
brand_db.update_file_record(fid, {'processed': 'error', 'processing_error': str(e)})
threading.Thread(target=_process_all, daemon=True).start()
_backfill_unprocessed_pdfs()
# When run directly
if __name__ == "__main__":
import argparse
# Get default port from environment, fallback to 7183
default_port = int(os.environ.get('PORT', 7183))
default_debug = debug_mode # Use the debug setting from environment
parser = argparse.ArgumentParser(description='Run Visual AI QC API server')
parser.add_argument('--host', type=str, default='0.0.0.0', help='Host to bind to')
parser.add_argument('--port', type=int, default=default_port, help=f'Port to listen on (default: {default_port} from environment)')
parser.add_argument('--debug', action='store_true', default=default_debug, help=f'Run in debug mode (default: {default_debug} from environment)')
args = parser.parse_args()
print(f"Environment: {current_environment}")
print(f"Starting Flask API server on {args.host}:{args.port}")
print(f"Debug mode: {args.debug}")
app.run(host=args.host, port=args.port, debug=args.debug)
Reference in a new issue View git blame Copy permalink