Batch Convert

from pathlib import Pathfrom concurrent.futures import ThreadPoolExecutor, as_completedimport subprocessimport os
class DocumentConverter:    """Unified document conversion pipeline."""        def __init__(self, max_workers=4):        self.max_workers = max_workers        self.converters = {            ('md', 'docx'): self._md_to_docx,            ('md', 'pdf'): self._md_to_pdf,            ('md', 'html'): self._md_to_html,            ('md', 'pptx'): self._md_to_pptx,            ('docx', 'pdf'): self._docx_to_pdf,            ('docx', 'md'): self._docx_to_md,            ('pdf', 'docx'): self._pdf_to_docx,            ('pdf', 'md'): self._pdf_to_md,            ('xlsx', 'pdf'): self._xlsx_to_pdf,            ('xlsx', 'md'): self._xlsx_to_md,            ('pptx', 'pdf'): self._pptx_to_pdf,            ('pptx', 'md'): self._pptx_to_md,            ('html', 'md'): self._html_to_md,            ('html', 'pdf'): self._html_to_pdf,        }        def convert(self, input_path, output_format, output_dir=None):        """Convert single file to target format."""        input_path = Path(input_path)        input_format = input_path.suffix[1:].lower()                if output_dir:            output_path = Path(output_dir) / f"{input_path.stem}.{output_format}"        else:            output_path = input_path.with_suffix(f".{output_format}")                converter_key = (input_format, output_format)        if converter_key not in self.converters:            raise ValueError(f"Conversion not supported: {input_format} -> {output_format}")                converter = self.converters[converter_key]        return converter(input_path, output_path)        def batch_convert(self, input_dir, output_format, output_dir=None,                       file_pattern="*", recursive=False):        """Batch convert all matching files."""        input_path = Path(input_dir)        output_path = Path(output_dir) if output_dir else input_path / "converted"        output_path.mkdir(exist_ok=True)                # Find files        if recursive:            files = list(input_path.rglob(file_pattern))        else:            files = list(input_path.glob(file_pattern))                # Filter to supported formats        supported_ext = ['.md', '.docx', '.pdf', '.xlsx', '.pptx', '.html']        files = [f for f in files if f.suffix.lower() in supported_ext]                results = []                with ThreadPoolExecutor(max_workers=self.max_workers) as executor:            future_to_file = {                executor.submit(self.convert, f, output_format, output_path): f                for f in files            }                        for future in as_completed(future_to_file):                file = future_to_file[future]                try:                    result = future.result()                    results.append({'file': str(file), 'status': 'success', 'output': str(result)})                except Exception as e:                    results.append({'file': str(file), 'status': 'error', 'error': str(e)})                return results

# Markdown conversions (using Pandoc)def _md_to_docx(self, input_path, output_path):    subprocess.run(['pandoc', str(input_path), '-o', str(output_path)], check=True)    return output_path
def _md_to_pdf(self, input_path, output_path):    subprocess.run(['pandoc', str(input_path), '-o', str(output_path)], check=True)    return output_path
def _md_to_html(self, input_path, output_path):    subprocess.run(['pandoc', str(input_path), '-s', '-o', str(output_path)], check=True)    return output_path
def _md_to_pptx(self, input_path, output_path):    subprocess.run(['marp', str(input_path), '-o', str(output_path)], check=True)    return output_path
# Office to Markdown (using markitdown)def _docx_to_md(self, input_path, output_path):    from markitdown import MarkItDown    md = MarkItDown()    result = md.convert(str(input_path))    with open(output_path, 'w') as f:        f.write(result.text_content)    return output_path
def _xlsx_to_md(self, input_path, output_path):    from markitdown import MarkItDown    md = MarkItDown()    result = md.convert(str(input_path))    with open(output_path, 'w') as f:        f.write(result.text_content)    return output_path
def _pptx_to_md(self, input_path, output_path):    from markitdown import MarkItDown    md = MarkItDown()    result = md.convert(str(input_path))    with open(output_path, 'w') as f:        f.write(result.text_content)    return output_path
# PDF conversionsdef _pdf_to_docx(self, input_path, output_path):    from pdf2docx import Converter    cv = Converter(str(input_path))    cv.convert(str(output_path))    cv.close()    return output_path
def _pdf_to_md(self, input_path, output_path):    from markitdown import MarkItDown    md = MarkItDown()    result = md.convert(str(input_path))    with open(output_path, 'w') as f:        f.write(result.text_content)    return output_path
# Office to PDF (using LibreOffice)def _docx_to_pdf(self, input_path, output_path):    subprocess.run([        'soffice', '--headless', '--convert-to', 'pdf',        '--outdir', str(output_path.parent), str(input_path)    ], check=True)    return output_path
def _xlsx_to_pdf(self, input_path, output_path):    subprocess.run([        'soffice', '--headless', '--convert-to', 'pdf',        '--outdir', str(output_path.parent), str(input_path)    ], check=True)    return output_path
def _pptx_to_pdf(self, input_path, output_path):    subprocess.run([        'soffice', '--headless', '--convert-to', 'pdf',        '--outdir', str(output_path.parent), str(input_path)    ], check=True)    return output_path

from tqdm import tqdm
def batch_convert_with_progress(converter, input_dir, output_format, output_dir=None):    """Batch convert with progress bar."""    input_path = Path(input_dir)    files = list(input_path.glob('*'))        results = []    for file in tqdm(files, desc=f"Converting to {output_format}"):        try:            result = converter.convert(file, output_format, output_dir)            results.append({'file': str(file), 'status': 'success'})        except Exception as e:            results.append({'file': str(file), 'status': 'error', 'error': str(e)})        return results

def detect_and_convert(file_path, target_format):    """Automatically detect format and convert."""    import mimetypes        mime_type, _ = mimetypes.guess_type(str(file_path))        format_map = {        'application/pdf': 'pdf',        'application/vnd.openxmlformats-officedocument.wordprocessingml.document': 'docx',        'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet': 'xlsx',        'application/vnd.openxmlformats-officedocument.presentationml.presentation': 'pptx',        'text/markdown': 'md',        'text/html': 'html',    }        source_format = format_map.get(mime_type, Path(file_path).suffix[1:])        converter = DocumentConverter()    return converter.convert(file_path, target_format)

def convert_to_multiple_formats(input_file, output_formats, output_dir):    """Convert one file to multiple formats."""    converter = DocumentConverter()    results = {}        for fmt in output_formats:        try:            output = converter.convert(input_file, fmt, output_dir)            results[fmt] = {'status': 'success', 'path': str(output)}        except Exception as e:            results[fmt] = {'status': 'error', 'error': str(e)}        return results
# Convert README to multiple formatsresults = convert_to_multiple_formats(    'README.md',    ['docx', 'pdf', 'html'],    './exports')

from pathlib import Pathimport json
def export_documentation(docs_dir, export_dir):    """Export all documentation to multiple formats."""        converter = DocumentConverter(max_workers=8)    docs_path = Path(docs_dir)    export_path = Path(export_dir)        # Create format directories    for fmt in ['pdf', 'docx', 'html']:        (export_path / fmt).mkdir(parents=True, exist_ok=True)        all_results = {}        # Find all markdown files    md_files = list(docs_path.rglob('*.md'))        for md_file in md_files:        file_results = {}                for fmt in ['pdf', 'docx', 'html']:            output_dir = export_path / fmt            try:                output = converter.convert(md_file, fmt, output_dir)                file_results[fmt] = 'success'            except Exception as e:                file_results[fmt] = f'error: {e}'                all_results[str(md_file)] = file_results        print(f"Processed: {md_file.name}")        # Save report    with open(export_path / 'export_report.json', 'w') as f:        json.dump(all_results, f, indent=2)        return all_results
results = export_documentation('./docs', './exports')

def migrate_legacy_docs(source_dir, target_dir):    """Migrate legacy documents to modern formats."""        converter = DocumentConverter(max_workers=4)        # Migration rules    migrations = [        ('*.doc', 'docx'),   # Old Word to new        ('*.xls', 'xlsx'),   # Old Excel to new        ('*.ppt', 'pptx'),   # Old PowerPoint to new        ('*.rtf', 'docx'),   # RTF to Word    ]        source_path = Path(source_dir)    target_path = Path(target_dir)    target_path.mkdir(exist_ok=True)        total_migrated = 0    errors = []        for pattern, target_format in migrations:        files = list(source_path.glob(pattern))                for file in files:            try:                # Use LibreOffice for legacy formats                subprocess.run([                    'soffice', '--headless',                    '--convert-to', target_format,                    '--outdir', str(target_path),                    str(file)                ], check=True)                                total_migrated += 1                print(f"Migrated: {file.name}")                            except Exception as e:                errors.append({'file': str(file), 'error': str(e)})        print(f"\nMigration complete: {total_migrated} files")    print(f"Errors: {len(errors)}")        return {'migrated': total_migrated, 'errors': errors}

def generate_reports_pipeline(data_files, template_dir, output_dir):    """Generate reports from data files using templates."""        from datetime import datetime        converter = DocumentConverter()    output_path = Path(output_dir)    output_path.mkdir(exist_ok=True)        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')        reports = []        for data_file in data_files:        # Load data        data_path = Path(data_file)                # Generate markdown report        md_content = f"""---title: Report - {data_path.stem}date: {datetime.now().strftime('%Y-%m-%d')}---
# {data_path.stem} Report
Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
## Data Summary
"""                # Add data content (simplified)        if data_path.suffix == '.xlsx':            from markitdown import MarkItDown            md = MarkItDown()            result = md.convert(str(data_path))            md_content += result.text_content                # Save markdown        md_file = output_path / f"{data_path.stem}_{timestamp}.md"        with open(md_file, 'w') as f:            f.write(md_content)                # Convert to PDF and DOCX        for fmt in ['pdf', 'docx']:            try:                output = converter.convert(md_file, fmt, output_path)                reports.append({'source': str(data_file), 'output': str(output), 'format': fmt})            except Exception as e:                print(f"Error converting {data_file} to {fmt}: {e}")        return reports