Twelve Labs を使用したセキュリティ分析アプリケーションの構築

API_KEY=your_api_key_here
INDEX_ID=your_index_id

.               
├── README.md                 
├── .gitignore           
├── .env                    
├── static/            
│   └── css/
│       └── style.css 
├── templates/     
│   └── index.html
├── app.py 
├── report_generator.py  
├── requirements.txt 
└── uploads

pip install -r requirements.txt

# To handle errors across routes
def handle_errors(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        try:
            return f(*args, **kwargs)
        except Exception as e:
	     # Log Error
            app.logger.error(f"Error: {str(e)}")
            return jsonify({
                'error': 'An error occurred processing your request',
                'details': str(e)
            }), 500
    return decorated_function


@app.route('/search', methods=['POST'])
@handle_errors
def search():
    # Validate request contains JSON data
    if not request.is_json:
        return jsonify({'error': 'Request must be JSON'}), 400
    
    data = request.get_json()
    query = data.get('query')
    
    if not query:
        return jsonify({'error': 'Query parameter is required'}), 400

    try:
       # Search videos using Twelve Labs API
        search_results = client.search.query(
            index_id=INDEX_ID,
            query_text=query,
            options=["visual"]
        )
        
        formatted_results = []
        # Process each video clip from search results
        for clip in search_results.data:
            try:
                headers = {
                    "x-api-key": API_KEY,
                    "Content-Type": "application/json"
                }
                print(headers)
                print(f"{BASE_URL}/indexes/{INDEX_ID}/videos/{clip.video_id}")
               # To Fetch additional video details
                url_response = requests.get(
                    f"{BASE_URL}/indexes/{INDEX_ID}/videos/{clip.video_id}",
                    headers=headers
                )
                
                # Extract video metadata
                video_data = url_response.json()
                video_url = video_data.get('hls', {}).get('video_url')
                video_duration = video_data.get('metadata', {}).get('duration', 0)
                
                formatted_results.append({
                    'video_id': clip.video_id,
                    'score': clip.score,  
                    'confidence': 'High' if clip.score > 0.7 else 'Medium',
                    'start': clip.start,
                    'end': clip.end,
                    'duration': video_duration,
                    'video_url': video_url
                })
            except Exception as e:
                app.logger.error(f"Error getting video URL: {str(e)}")
                continue
        
        return jsonify(formatted_results)
        
    except Exception as e:
        app.logger.error(f"Search Error: {str(e)}")
        return jsonify({'error': 'Search failed', 'details': str(e)}), 500


# Defining the directory path for sample videos
SAMPLE_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'sample')


@app.route('/sample/<path:filename>')
def serve_sample_video(filename):
    # Serve sample video files from the sample directory
    try:
        return send_from_directory(SAMPLE_DIR, filename)
    except Exception as e:
        app.logger.error(f"Error serving video: {str(e)}")
        return f"Error: {str(e)}", 404

# To handle video analysis requests
@app.route('/analyze/<video_id>')
@handle_errors
def analyze(video_id):
    try:
        app.logger.info(f"Starting analysis for video: {video_id}")
        
        try:
            # 1. Get video URL
            headers = {
                "x-api-key": API_KEY,
                "Content-Type": "application/json"
            }
            url_response = requests.get(
                f"{BASE_URL}/indexes/{INDEX_ID}/videos/{video_id}",
                headers=headers
            )
            url_data = url_response.json()
            video_url = url_data.get('hls', {}).get('video_url')
            
            # 2. Generate analysis
            analysis_response = client.generate.text(
                video_id=video_id,
                prompt="Provide a detailed analysis of the key events, actions, and notable elements in this video."
            )
            
            # Extract and clean analysis text
            analysis_text = str(analysis_response.data) if hasattr(analysis_response, 'data') else ''
            analysis_text = analysis_text.strip().strip('"\'')
            
            response_data = {
                'video_url': video_url,
                'analysis': analysis_text or "No analysis available."
            }
            
            app.logger.info("Analysis completed successfully")
            return jsonify(response_data)
            
        except Exception as e:
            app.logger.error(f"Error in analysis: {str(e)}")
            raise
            
    except Exception as e:
        app.logger.error(f"Analysis Error: {str(e)}")
        return jsonify({
            'error': 'Analysis failed',
            'details': str(e)
        }), 500

  window.analyzeVideo = async function(videoId) {
        const modalContent = document.getElementById('analysisContent');
        const downloadButton = document.querySelector('.download-button');
        const modal = document.getElementById('analysisModal');

        // Show modal and display loading spinner
        modal.classList.remove('hidden');
        modalContent.innerHTML = `
            <div class="spinner">
                <div class="spinner-ring"></div>
                <p>Analyzing video...</p>
            </div>
        `;
        downloadButton.disabled = true;
    
        try {
            // Log analysis request and fetch results from server
            console.log('Analyzing video:', videoId);
            const response = await fetch(`/analyze/${videoId}`);
            const data = await response.json();
            console.log('Analysis response:', data);
    
            if (!response.ok) {
                console.error('Response not OK:', response.status, response.statusText);
                throw new Error(data.error || 'Analysis failed');
            }
            // Validate analysis text exists
            const analysisText = data.analysis;
            if (!analysisText) {
                console.error('No analysis text available');
                throw new Error('No analysis data available');
            }
          
                // Clean up text by removing escape characters and quotes
   let analysisText = data.analysis
                     .replace(/\\n/g, '\n')
                     .replace(/^"|"$/g, '')
                     .trim();

}

if (data.video_url) { //Check if the video URL exists in response
    const video = document.getElementById('analysis-video');

    // Verify video element exists and browser supports HLS playback
    if (video && Hls.isSupported()) {
        // Initialize and configure HLS player instance
        const hls = new Hls();
        hls.loadSource(data.video_url);
        hls.attachMedia(video);
        
       // Error Handling for video playback
        hls.on(Hls.Events.ERROR, function(event, data) {
            console.error('Video playback error');
            if (data.fatal) {
                handleVideoError();
            }
        });
    }
}
// Store current analysis text in state and enable download
state.currentAnalysis = analysisText;
downloadButton.disabled = false;

} catch (error) {
// Log and display of Error
console.error('Analysis error');
modalContent.innerHTML = `
    <div class="error-message">
        <i class="fas fa-exclamation-circle"></i>
        <p>${error.message || 'Failed to analyze video. Please try again.'}</p>
        <button class="retry-btn" onclick="analyzeVideo('${videoId}')">
            <i class="fas fa-redo"></i> Try Again
        </button>
    </div>
`;
}