citation_manager.py

#!/usr/bin/env python3
"""
Citation Manager - Unique Feature for Transparent Source Tracking
Handles comprehensive citation tracking, source attribution, and transparency features
"""

import json
import time
import hashlib
from typing import List, Dict, Any, Optional
from pathlib import Path
import logging

logger = logging.getLogger(__name__)

class CitationManager:
    """Advanced citation tracking and source transparency system"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.citation_format = config['citations']['citation_format']
        self.include_metadata = config['citations']['include_metadata']
        
        # Citation storage
        self.citations_db = {}
        self.source_lineage = {}  # Track document relationships
        self.usage_analytics = {}  # Track citation usage patterns
        
    def create_citation(self, 
                       source_data: Dict[str, Any], 
                       query_context: str = None) -> Dict[str, Any]:
        """Create comprehensive citation with metadata"""
        
        citation_id = self._generate_citation_id(source_data)
        
        citation = {
            'id': citation_id,
            'timestamp': time.time(),
            'source_file': source_data.get('source_file', ''),
            'file_name': Path(source_data.get('source_file', '')).name,
            'modality': source_data.get('modality', 'unknown'),
            'file_type': source_data.get('file_type', ''),
            'content_preview': self._create_content_preview(source_data),
            'similarity_score': source_data.get('similarity_score', 0.0),
            'query_context': query_context,
            'access_timestamp': time.time(),
            'chunk_metadata': self._extract_chunk_metadata(source_data)
        }
        
        # Add modality-specific metadata
        if source_data.get('modality') == 'image':
            citation.update({
                'image_path': source_data.get('image_path', ''),
                'image_size': source_data.get('image_size', [0, 0]),
                'has_ocr_text': bool(source_data.get('ocr_text', '')),
                'caption': source_data.get('caption', '')
            })
        elif source_data.get('modality') == 'audio_transcription':
            citation.update({
                'start_time': source_data.get('start_time', 0.0),
                'end_time': source_data.get('end_time', 0.0),
                'duration': source_data.get('duration', 0.0),
                'language': source_data.get('language', 'unknown'),
                'segment_id': source_data.get('segment_id', 0)
            })
        elif source_data.get('modality') == 'text':
            citation.update({
                'chunk_id': source_data.get('chunk_id', 0),
                'word_count': source_data.get('word_count', 0),
                'character_count': source_data.get('character_count', 0)
            })
            
        # Store citation
        self.citations_db[citation_id] = citation
        
        # Update usage analytics
        self._update_usage_analytics(citation_id, query_context)
        
        return citation
        
    def format_citation(self, citation: Dict[str, Any], format_type: str = None) -> str:
        """Format citation according to specified style"""
        
        format_type = format_type or self.citation_format
        
        if format_type == "numbered":
            return self._format_numbered_citation(citation)
        elif format_type == "author-date":
            return self._format_author_date_citation(citation)
        elif format_type == "detailed":
            return self._format_detailed_citation(citation)
        else:
            return self._format_numbered_citation(citation)
            
    def _format_numbered_citation(self, citation: Dict[str, Any]) -> str:
        """Format as numbered citation"""
        file_name = citation['file_name']
        modality = citation['modality']
        score = citation['similarity_score']
        
        base_citation = f"{file_name} ({modality})"
        
        if self.include_metadata:
            base_citation += f" [Score: {score:.3f}]"
            
        return base_citation
        
    def _format_detailed_citation(self, citation: Dict[str, Any]) -> str:
        """Format as detailed citation with full metadata"""
        parts = [f"📄 {citation['file_name']}"]
        
        # Add modality-specific details
        if citation['modality'] == 'image':
            if citation.get('caption'):
                parts.append(f"Caption: {citation['caption']}")
            if citation.get('has_ocr_text'):
                parts.append("Contains text")
                
        elif citation['modality'] == 'audio_transcription':
            start_time = citation.get('start_time', 0)
            end_time = citation.get('end_time', 0)
            parts.append(f"Audio segment: {start_time:.1f}s - {end_time:.1f}s")
            
        elif citation['modality'] == 'text':
            word_count = citation.get('word_count', 0)
            parts.append(f"Text chunk: {word_count} words")
            
        # Add similarity score
        score = citation['similarity_score']
        parts.append(f"Relevance: {score:.3f}")
        
        return " | ".join(parts)
        
    def create_bibliography(self, citation_ids: List[str]) -> str:
        """Create formatted bibliography from citation IDs"""
        
        bibliography = ["## Sources\n"]
        
        for i, citation_id in enumerate(citation_ids, 1):
            if citation_id in self.citations_db:
                citation = self.citations_db[citation_id]
                formatted = self.format_citation(citation, "detailed")
                bibliography.append(f"[{i}] {formatted}")
                
        return "\n".join(bibliography)
        
    def track_cross_references(self, 
                              primary_citation: str, 
                              related_citations: List[str]):
        """Track relationships between citations"""
        
        if primary_citation not in self.source_lineage:
            self.source_lineage[primary_citation] = {
                'related_sources': [],
                'reference_count': 0,
                'cross_modal_links': []
            }
            
        # Add related citations
        for related_id in related_citations:
            if related_id not in self.source_lineage[primary_citation]['related_sources']:
                self.source_lineage[primary_citation]['related_sources'].append(related_id)
                
        # Track cross-modal relationships
        primary_modality = self.citations_db.get(primary_citation, {}).get('modality')
        
        for related_id in related_citations:
            related_modality = self.citations_db.get(related_id, {}).get('modality')
            
            if primary_modality != related_modality:
                cross_link = {
                    'source_modality': primary_modality,
                    'target_modality': related_modality,
                    'target_citation': related_id,
                    'link_strength': self._calculate_link_strength(primary_citation, related_id)
                }
                self.source_lineage[primary_citation]['cross_modal_links'].append(cross_link)
                
    def get_citation_analytics(self) -> Dict[str, Any]:
        """Get analytics about citation usage patterns"""
        
        total_citations = len(self.citations_db)
        modality_breakdown = {}
        popular_sources = {}
        
        for citation in self.citations_db.values():
            modality = citation['modality']
            modality_breakdown[modality] = modality_breakdown.get(modality, 0) + 1
            
            source_file = citation['source_file']
            popular_sources[source_file] = popular_sources.get(source_file, 0) + 1
            
        # Find most cited sources
        top_sources = sorted(popular_sources.items(), key=lambda x: x[1], reverse=True)[:10]
        
        # Calculate cross-modal citation patterns
        cross_modal_stats = self._analyze_cross_modal_patterns()
        
        return {
            'total_citations': total_citations,
            'modality_breakdown': modality_breakdown,
            'top_sources': top_sources,
            'cross_modal_patterns': cross_modal_stats,
            'citation_quality_scores': self._calculate_quality_metrics()
        }
        
    def verify_citation_integrity(self) -> Dict[str, Any]:
        """Verify that all citations point to valid, accessible sources"""
        
        integrity_report = {
            'total_citations': len(self.citations_db),
            'valid_citations': 0,
            'broken_citations': [],
            'missing_metadata': [],
            'accessibility_issues': []
        }
        
        for citation_id, citation in self.citations_db.items():
            # Check if source file exists
            source_path = Path(citation.get('source_file', ''))
            
            if not source_path.exists():
                integrity_report['broken_citations'].append({
                    'citation_id': citation_id,
                    'issue': 'Source file not found',
                    'file_path': str(source_path)
                })
                continue
                
            # Check metadata completeness
            required_fields = ['modality', 'file_type', 'similarity_score']
            missing_fields = [field for field in required_fields if not citation.get(field)]
            
            if missing_fields:
                integrity_report['missing_metadata'].append({
                    'citation_id': citation_id,
                    'missing_fields': missing_fields
                })
                
            # Check accessibility
            try:
                if citation['modality'] == 'image':
                    from PIL import Image
                    with Image.open(citation.get('image_path', '')) as img:
                        pass  # Just verify it can be opened
                        
                integrity_report['valid_citations'] += 1
                
            except Exception as e:
                integrity_report['accessibility_issues'].append({
                    'citation_id': citation_id,
                    'error': str(e)
                })
                
        return integrity_report
        
    def export_citations(self, format_type: str = "json") -> str:
        """Export all citations in specified format"""
        
        if format_type == "json":
            return json.dumps(self.citations_db, indent=2, default=str)
            
        elif format_type == "csv":
            import csv
            import io
            
            output = io.StringIO()
            writer = csv.writer(output)
            
            # Header
            writer.writerow([
                'Citation ID', 'File Name', 'Modality', 'File Type',
                'Similarity Score', 'Access Timestamp', 'Content Preview'
            ])
            
            # Data rows
            for citation in self.citations_db.values():
                writer.writerow([
                    citation['id'],
                    citation['file_name'],
                    citation['modality'],
                    citation['file_type'],
                    citation['similarity_score'],
                    citation['access_timestamp'],
                    citation['content_preview'][:100]
                ])
                
            return output.getvalue()
            
        elif format_type == "markdown":
            lines = ["# Citation Database\n"]
            
            for citation in self.citations_db.values():
                lines.append(f"## {citation['file_name']}")
                lines.append(f"- **Type**: {citation['modality']} ({citation['file_type']})")
                lines.append(f"- **Relevance**: {citation['similarity_score']:.3f}")
                lines.append(f"- **Content**: {citation['content_preview']}")
                lines.append("")
                
            return "\n".join(lines)
            
    def _generate_citation_id(self, source_data: Dict[str, Any]) -> str:
        """Generate unique citation ID"""
        content = f"{source_data.get('source_file', '')}{source_data.get('chunk_hash', '')}{time.time()}"
        return hashlib.sha256(content.encode()).hexdigest()[:16]
        
    def _create_content_preview(self, source_data: Dict[str, Any]) -> str:
        """Create content preview for citation"""
        content = source_data.get('content', '')
        
        if len(content) > 200:
            return content[:200] + "..."
        return content
        
    def _extract_chunk_metadata(self, source_data: Dict[str, Any]) -> Dict[str, Any]:
        """Extract relevant chunk metadata"""
        metadata = {}
        
        for key in ['chunk_id', 'start_time', 'end_time', 'segment_id', 'image_size']:
            if key in source_data:
                metadata[key] = source_data[key]
                
        return metadata
        
    def _update_usage_analytics(self, citation_id: str, query_context: str):
        """Update citation usage analytics"""
        if citation_id not in self.usage_analytics:
            self.usage_analytics[citation_id] = {
                'access_count': 0,
                'queries': [],
                'first_access': time.time(),
                'last_access': time.time()
            }
            
        self.usage_analytics[citation_id]['access_count'] += 1
        self.usage_analytics[citation_id]['last_access'] = time.time()
        
        if query_context:
            self.usage_analytics[citation_id]['queries'].append({
                'query': query_context,
                'timestamp': time.time()
            })
            
    def _calculate_link_strength(self, primary_id: str, related_id: str) -> float:
        """Calculate strength of relationship between citations"""
        # Implement similarity-based link strength calculation
        primary_citation = self.citations_db.get(primary_id, {})
        related_citation = self.citations_db.get(related_id, {})
        
        # Simple implementation based on similarity scores
        primary_score = primary_citation.get('similarity_score', 0.0)
        related_score = related_citation.get('similarity_score', 0.0)
        
        return (primary_score + related_score) / 2.0
        
    def _analyze_cross_modal_patterns(self) -> Dict[str, Any]:
        """Analyze cross-modal citation patterns"""
        patterns = {
            'text_to_image': 0,
            'image_to_text': 0,
            'audio_to_text': 0,
            'text_to_audio': 0,
            'image_to_audio': 0,
            'audio_to_image': 0
        }
        
        for lineage in self.source_lineage.values():
            for link in lineage.get('cross_modal_links', []):
                source_mod = link['source_modality']
                target_mod = link['target_modality']
                pattern_key = f"{source_mod}_to_{target_mod}"
                
                if pattern_key in patterns:
                    patterns[pattern_key] += 1
                    
        return patterns
        
    def _calculate_quality_metrics(self) -> Dict[str, float]:
        """Calculate citation quality metrics"""
        if not self.citations_db:
            return {}
            
        scores = [c['similarity_score'] for c in self.citations_db.values()]
        
        return {
            'average_relevance': sum(scores) / len(scores),
            'min_relevance': min(scores),
            'max_relevance': max(scores),
            'citation_diversity': len(set(c['source_file'] for c in self.citations_db.values())) / len(self.citations_db)
        }