03_System_Tools/document_processor_fixed.py

#!/usr/bin/env python3
"""
Luuna Document Processing Engine - Fixed Version
Handles PDF and Markdown document parsing, analysis, and content extraction
"""

import os
import json
import PyPDF2
import fitz  # PyMuPDF
import pandas as pd
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Any
import logging
import re
import csv
from datetime import datetime
from dataclasses import dataclass, asdict

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

@dataclass
class DocumentAnalysis:
    """Analysis result for a single document"""
    filename: str
    original_path: str
    markdown_path: str
    category: str
    language: str
    content_type: str
    main_topics: List[str]
    actionable_tasks: List[Dict]
    skill_potential: Dict[str, float]  # skill_type -> confidence score
    metadata: Dict[str, Any]
    processing_timestamp: float

@dataclass
class SkillExtraction:
    """Extracted skill from document analysis"""
    skill_id: str
    name: str
    category: str
    description: str
    source_document: str
    confidence_score: float
    implementation_complexity: str
    dependencies: List[str]
    capabilities: List[str]
    code_snippets: List[str]
    test_cases: List[Dict]

class DocumentConverter:
    """Converts various document formats to Markdown"""

    def __init__(self):
        self.supported_formats = {
            '.pdf': self._convert_pdf_to_markdown,
            '.csv': self._convert_csv_to_markdown,
            '.docx': self._convert_docx_to_markdown,
            '.doc': self._convert_docx_to_markdown,
            '.txt': self._convert_txt_to_markdown
        }

    def convert_file(self, file_path: Path, output_dir: Path) -> Optional[Path]:
        """Convert a single file to Markdown"""
        try:
            if not file_path.exists():
                logger.error(f"File not found: {file_path}")
                return None

            file_extension = file_path.suffix.lower()
            if file_extension not in self.supported_formats:
                logger.warning(f"Unsupported format: {file_extension}")
                return None

            # Create output filename
            output_filename = f"{file_path.stem}.md"
            output_path = output_dir / output_filename

            # Convert using appropriate method
            converter_method = self.supported_formats[file_extension]
            markdown_content = converter_method(file_path)

            if markdown_content:
                # Write to output file
                output_path.parent.mkdir(parents=True, exist_ok=True)
                with open(output_path, 'w', encoding='utf-8') as f:
                    f.write(markdown_content)

                logger.info(f"Converted {file_path} -> {output_path}")
                return output_path
            else:
                logger.error(f"Failed to convert {file_path}")
                return None

        except Exception as e:
            logger.error(f"Error converting {file_path}: {e}")
            return None

    def _convert_pdf_to_markdown(self, file_path: Path) -> str:
        """Convert PDF to Markdown using PyMuPDF"""
        try:
            doc = fitz.open(str(file_path))
            markdown_content = []

            # Add header
            markdown_content.append(f"# {file_path.stem}\n")
            markdown_content.append(f"**Converted from PDF on {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}**\n")

            for page_num in range(len(doc)):
                page = doc.load_page(page_num)
                text = page.get_text()

                # Clean and format text
                text = self._clean_text(text)

                if text.strip():
                    markdown_content.append(f"## Page {page_num + 1}\n")
                    markdown_content.append(text)
                    markdown_content.append("\n---\n")

            doc.close()
            return "\n".join(markdown_content)

        except Exception as e:
            logger.error(f"PDF conversion error: {e}")
            return ""

    def _convert_csv_to_markdown(self, file_path: Path) -> str:
        """Convert CSV to Markdown table"""
        try:
            # Read CSV with pandas
            df = pd.read_csv(file_path, encoding='utf-8')

            markdown_content = []
            markdown_content.append(f"# {file_path.stem}\n")
            markdown_content.append(f"**Converted from CSV on {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}**\n")
            markdown_content.append(f"**Rows:** {len(df)}, **Columns:** {len(df.columns)}\n\n")

            # Convert DataFrame to Markdown table
            table_md = df.to_markdown(index=False)
            markdown_content.append(table_md)

            # Add basic statistics
            markdown_content.append("\n\n## Basic Statistics\n")
            if df.select_dtypes(include=[float, int]).shape[1] > 0:
                stats = df.describe()
                markdown_content.append(stats.to_markdown())

            return "\n".join(markdown_content)

        except Exception as e:
            logger.error(f"CSV conversion error: {e}")
            return ""

    def _convert_docx_to_markdown(self, file_path: Path) -> str:
        """Convert DOCX to Markdown"""
        try:
            # For now, basic text extraction
            # TODO: Implement full DOCX parsing with python-docx
            markdown_content = []
            markdown_content.append(f"# {file_path.stem}\n")
            markdown_content.append("**Note:** Full DOCX conversion not yet implemented\n")
            markdown_content.append("Please use PDF or TXT format for now.\n")

            return "\n".join(markdown_content)

        except Exception as e:
            logger.error(f"DOCX conversion error: {e}")
            return ""

    def _convert_txt_to_markdown(self, file_path: Path) -> str:
        """Convert TXT to Markdown"""
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read()

            markdown_content = []
            markdown_content.append(f"# {file_path.stem}\n")
            markdown_content.append(f"**Converted from TXT on {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}**\n")
            markdown_content.append(self._clean_text(content))

            return "\n".join(markdown_content)

        except Exception as e:
            logger.error(f"TXT conversion error: {e}")
            return ""

    def _clean_text(self, text: str) -> str:
        """Clean and format text for Markdown"""
        # Remove excessive whitespace
        text = re.sub(r'\n\s*\n\s*\n+', '\n\n', text)  # Multiple newlines to double
        text = re.sub(r'[ \t]+', ' ', text)  # Multiple spaces/tabs to single space
        text = re.sub(r'\n\s+', '\n', text)  # Remove leading whitespace after newlines

        # Basic formatting cleanup
        lines = text.split('\n')
        cleaned_lines = []

        for line in lines:
            line = line.strip()
            if line:  # Skip empty lines that became empty after stripping
                cleaned_lines.append(line)

        return '\n\n'.join(cleaned_lines)

class WorkspaceMonitor:
    """Monitors workspace for new files and automatically converts them"""

    def __init__(self, workspace_path: Path, markdown_output_dir: str = "markdown_conversions"):
        self.workspace_path = workspace_path
        self.markdown_output_dir = workspace_path / markdown_output_dir
        self.converter = DocumentConverter()
        self.processed_files = set()

        # Create output directory
        self.markdown_output_dir.mkdir(exist_ok=True)

        # Load previously processed files
        self._load_processed_files()

    def _load_processed_files(self):
        """Load list of previously processed files"""
        processed_file = self.markdown_output_dir / "processed_files.json"
        if processed_file.exists():
            try:
                with open(processed_file, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                    self.processed_files = set(data.get('processed_files', []))
            except Exception as e:
                logger.warning(f"Could not load processed files: {e}")

    def _save_processed_files(self):
        """Save list of processed files"""
        processed_file = self.markdown_output_dir / "processed_files.json"
        try:
            with open(processed_file, 'w', encoding='utf-8') as f:
                json.dump({
                    'processed_files': list(self.processed_files),
                    'last_updated': datetime.now().isoformat()
                }, f, indent=2, ensure_ascii=False)
        except Exception as e:
            logger.error(f"Could not save processed files: {e}")

    def scan_and_convert(self) -> Dict[str, Any]:
        """Scan workspace for new files and convert them to Markdown"""
        results = {
            'scanned_files': 0,
            'converted_files': 0,
            'skipped_files': 0,
            'failed_files': 0,
            'new_conversions': []
        }

        # Supported file extensions for conversion
        supported_extensions = {'.pdf', '.csv', '.docx', '.doc', '.txt'}

        # Scan workspace recursively
        for file_path in self.workspace_path.rglob('*'):
            if not file_path.is_file():
                continue

            results['scanned_files'] += 1

            # Check if file should be processed
            if file_path.suffix.lower() not in supported_extensions:
                continue

            # Check if already processed
            file_key = str(file_path.relative_to(self.workspace_path))
            if file_key in self.processed_files:
                results['skipped_files'] += 1
                continue

            # Convert the file
            logger.info(f"Converting new file: {file_path}")
            markdown_path = self.converter.convert_file(file_path, self.markdown_output_dir)

            if markdown_path:
                results['converted_files'] += 1
                self.processed_files.add(file_key)

                results['new_conversions'].append({
                    'original_path': str(file_path),
                    'markdown_path': str(markdown_path),
                    'file_size': file_path.stat().st_size,
                    'converted_at': datetime.now().isoformat()
                })
            else:
                results['failed_files'] += 1

        # Save updated processed files list
        self._save_processed_files()

        return results

class SkillExtractor:
    """Extracts skills and capabilities from document analysis"""

    def __init__(self):
        self.skill_templates = self._load_skill_templates()

    def _load_skill_templates(self) -> Dict[str, Dict]:
        """Load skill templates based on document content"""
        return {
            'crypto_trading': {
                'name_template': '{topic} Trading Skill',
                'category': 'Financial Analysis',
                'capabilities': ['analyze_market_data', 'predict_trends', 'execute_trades'],
                'complexity': 'medium'
            },
            'data_analysis': {
                'name_template': '{topic} Data Analysis',
                'category': 'Data Processing',
                'capabilities': ['process_data', 'generate_reports', 'visualize_data'],
                'complexity': 'low'
            },
            'ai_agent': {
                'name_template': '{topic} Agent',
                'category': 'AI Automation',
                'capabilities': ['automate_tasks', 'process_requests', 'generate_responses'],
                'complexity': 'high'
            },
            'business_intelligence': {
                'name_template': '{topic} Intelligence',
                'category': 'Business Analysis',
                'capabilities': ['analyze_business_data', 'generate_insights', 'create_reports'],
                'complexity': 'medium'
            }
        }

    def extract_skills_from_analysis(self, document_analysis: DocumentAnalysis) -> List[SkillExtraction]:
        """Extract skills from document analysis"""
        skills = []

        # Extract skills based on content category
        if document_analysis.category == 'crypto_finance':
            skill = self._create_skill_from_template(
                'crypto_trading',
                document_analysis,
                confidence_score=0.85
            )
            if skill:
                skills.append(skill)

        elif document_analysis.category == 'research_analysis':
            skill = self._create_skill_from_template(
                'data_analysis',
                document_analysis,
                confidence_score=0.75
            )
            if skill:
                skills.append(skill)

        elif document_analysis.category == 'tutorial_guide':
            skill = self._create_skill_from_template(
                'ai_agent',
                document_analysis,
                confidence_score=0.70
            )
            if skill:
                skills.append(skill)

        # Extract skills from actionable tasks
        for task in document_analysis.actionable_tasks:
            task_skill = self._extract_skill_from_task(task, document_analysis)
            if task_skill:
                skills.append(task_skill)

        return skills

    def _create_skill_from_template(self, template_key: str, analysis: DocumentAnalysis,
                                   confidence_score: float) -> Optional[SkillExtraction]:
        """Create a skill from template"""
        template = self.skill_templates.get(template_key)
        if not template:
            return None

        # Generate skill name from topics
        main_topic = analysis.main_topics[0] if analysis.main_topics else 'General'
        skill_name = template['name_template'].format(topic=main_topic.title())

        # Generate description
        description = f"Automated skill for {main_topic.lower()} based on {analysis.filename}"

        # Extract code snippets from content
        code_snippets = self._extract_code_snippets(analysis)

        return SkillExtraction(
            skill_id=f"skill_{analysis.filename.lower().replace(' ', '_')}_{int(datetime.now().timestamp())}",
            name=skill_name,
            category=template['category'],
            description=description,
            source_document=analysis.original_path,
            confidence_score=confidence_score,
            implementation_complexity=template['complexity'],
            dependencies=['pandas', 'numpy'],  # Basic dependencies
            capabilities=template['capabilities'],
            code_snippets=code_snippets,
            test_cases=self._generate_test_cases(analysis)
        )

    def _extract_skill_from_task(self, task: Dict, analysis: DocumentAnalysis) -> Optional[SkillExtraction]:
        """Extract skill from actionable task"""
        action = task.get('action', '').lower()
        description = task.get('description', '')

        # Map actions to skill types
        action_mapping = {
            'analyze': 'data_analysis',
            'research': 'research_analysis',
            'implement': 'ai_agent',
            'develop': 'ai_agent',
            'create': 'ai_agent',
            'monitor': 'business_intelligence'
        }

        skill_type = action_mapping.get(action)
        if not skill_type:
            return None

        template = self.skill_templates.get(skill_type)
        if not template:
            return None

        skill_name = f"{action.title()} {analysis.main_topics[0] if analysis.main_topics else 'Task'} Skill"

        return SkillExtraction(
            skill_id=f"skill_task_{action}_{int(datetime.now().timestamp())}",
            name=skill_name,
            category=template['category'],
            description=f"Skill extracted from task: {description[:100]}...",
            source_document=analysis.original_path,
            confidence_score=0.60,  # Lower confidence for task-based skills
            implementation_complexity=template['complexity'],
            dependencies=[],
            capabilities=[f"execute_{action}_tasks"],
            code_snippets=[],
            test_cases=[]
        )

    def _extract_code_snippets(self, analysis: DocumentAnalysis) -> List[str]:
        """Extract code snippets from document content"""
        # This would need access to the actual content
        # For now, return empty list
        return []

    def _generate_test_cases(self, analysis: DocumentAnalysis) -> List[Dict]:
        """Generate test cases for extracted skill"""
        return [
            {
                'test_name': 'basic_functionality_test',
                'description': f'Test basic functionality of {analysis.filename} skill',
                'input_data': {'test': 'data'},
                'expected_output': {'result': 'success'}
            }
        ]

class DocumentAnalysisPipeline:
    """Complete pipeline for document analysis and skill extraction"""

    def __init__(self, workspace_path: Path):
        self.workspace_path = workspace_path
        self.monitor = WorkspaceMonitor(workspace_path)
        self.extractor = SkillExtractor()
        self.analyses = []
        self.extracted_skills = []

    def run_full_pipeline(self) -> Dict[str, Any]:
        """Run complete analysis pipeline"""
        logger.info("Starting document analysis pipeline")

        # Step 1: Convert new files to Markdown
        conversion_results = self.monitor.scan_and_convert()
        logger.info(f"Converted {conversion_results['converted_files']} new files")

        # Step 2: Analyze all Markdown files
        markdown_files = list(self.monitor.markdown_output_dir.glob("*.md"))
        for md_file in markdown_files:
            if md_file.name != "processed_files.json":  # Skip metadata file
                analysis = self._analyze_markdown_file(md_file)
                if analysis:
                    self.analyses.append(analysis)

        # Step 3: Extract skills from analyses
        for analysis in self.analyses:
            skills = self.extractor.extract_skills_from_analysis(analysis)
            self.extracted_skills.extend(skills)

        # Step 4: Save results
        self._save_results()

        return {
            'conversion_results': conversion_results,
            'analyses_count': len(self.analyses),
            'skills_extracted': len(self.extracted_skills),
            'results_saved': True
        }

    def _analyze_markdown_file(self, md_file: Path) -> Optional[DocumentAnalysis]:
        """Analyze a single Markdown file"""
        try:
            with open(md_file, 'r', encoding='utf-8') as f:
                content = f.read()

            # Basic content analysis
            lines = content.split('\n')
            word_count = len(content.split())

            # Extract topics (simplified)
            topics = self._extract_topics_from_content(content)

            # Create analysis object
            analysis = DocumentAnalysis(
                filename=md_file.name,
                original_path=str(md_file),
                markdown_path=str(md_file),
                category=self._classify_document_category(content),
                language='en',  # Default to English to avoid Unicode issues
                content_type='document',
                main_topics=topics,
                actionable_tasks=[],  # Would need more sophisticated NLP
                skill_potential={'crypto_analysis': 0.8} if 'crypto' in content.lower() else {},
                metadata={
                    'word_count': word_count,
                    'line_count': len(lines),
                    'has_tables': '|' in content,
                    'has_code': '```' in content
                },
                processing_timestamp=datetime.now().timestamp()
            )

            return analysis

        except Exception as e:
            logger.error(f"Error analyzing {md_file}: {e}")
            return None

    def _extract_topics_from_content(self, content: str) -> List[str]:
        """Extract main topics from content"""
        topics = []
        content_lower = content.lower()

        topic_keywords = {
            'cryptocurrency': ['crypto', 'bitcoin', 'ethereum', 'blockchain', 'defi'],
            'trading': ['trading', 'market', 'price', 'analysis', 'strategy'],
            'ai_agent': ['agent', 'ai', 'automation', 'llm', 'gpt'],
            'finance': ['finance', 'banking', 'investment', 'loan', 'credit'],
            'data_analysis': ['data', 'analysis', 'statistics', 'report']
        }

        for topic, keywords in topic_keywords.items():
            if any(keyword in content_lower for keyword in keywords):
                topics.append(topic)

        return topics[:5]  # Limit to 5 topics

    def _classify_document_category(self, content: str) -> str:
        """Classify document category"""
        content_lower = content.lower()

        if any(word in content_lower for word in ['crypto', 'bitcoin', 'trading']):
            return 'crypto_finance'
        elif any(word in content_lower for word in ['agent', 'ai', 'automation']):
            return 'ai_agent'
        elif any(word in content_lower for word in ['data', 'analysis', 'statistics']):
            return 'research_analysis'
        else:
            return 'general'

    def _save_results(self):
        """Save analysis results and extracted skills"""
        results_dir = self.workspace_path / "analysis_results"
        results_dir.mkdir(exist_ok=True)

        # Save analyses
        analyses_file = results_dir / "document_analyses.json"
        with open(analyses_file, 'w', encoding='utf-8') as f:
            json.dump([asdict(analysis) for analysis in self.analyses], f, indent=2, ensure_ascii=False)

        # Save skills
        skills_file = results_dir / "extracted_skills.json"
        with open(skills_file, 'w', encoding='utf-8') as f:
            json.dump([asdict(skill) for skill in self.extracted_skills], f, indent=2, ensure_ascii=False)

        # Update skill-index.json
        skill_index_file = self.workspace_path / "skill-index.json"
        existing_skills = []

        if skill_index_file.exists():
            try:
                with open(skill_index_file, 'r', encoding='utf-8') as f:
                    existing_skills = json.load(f)
            except:
                existing_skills = []

        # Add new skills
        for skill in self.extracted_skills:
            skill_data = asdict(skill)
            if skill_data not in existing_skills:
                existing_skills.append(skill_data)

        with open(skill_index_file, 'w', encoding='utf-8') as f:
            json.dump(existing_skills, f, indent=2, ensure_ascii=False)

        logger.info(f"Saved {len(self.analyses)} analyses and {len(self.extracted_skills)} skills")

# Example usage
if __name__ == "__main__":
    # Run the full pipeline
    workspace_path = Path("c:/Luuna/SKILL_BASE")
    pipeline = DocumentAnalysisPipeline(workspace_path)

    results = pipeline.run_full_pipeline()
    print("Pipeline completed:")
    print(json.dumps(results, indent=2, ensure_ascii=False))