| """ |
| CONVERTLY - Complete Conversion API Server |
| Run with: python server.py |
| Port: 8000 |
| """ |
|
|
| from fastapi import FastAPI, UploadFile, File, Form, HTTPException |
| from fastapi.responses import JSONResponse, FileResponse, Response |
| from fastapi.middleware.cors import CORSMiddleware |
| import uvicorn |
| import os |
| import io |
| import requests |
| from PIL import Image |
| import json |
| import time |
| from datetime import datetime |
| from typing import Optional, List |
| import shutil |
| import tempfile |
| import subprocess |
| import re |
| import base64 |
| import hashlib |
|
|
| |
|
|
| app = FastAPI( |
| title="CONVERTLY API", |
| version="3.0.0", |
| description="Complete file conversion API - All formats supported" |
| ) |
|
|
| app.add_middleware( |
| CORSMiddleware, |
| allow_origins=["*"], |
| allow_credentials=True, |
| allow_methods=["*"], |
| allow_headers=["*"], |
| ) |
|
|
| HF_TOKEN = os.getenv("HF_TOKEN", "") |
| MAX_FILE_SIZE = 100 * 1024 * 1024 |
|
|
| |
|
|
| ALLOWED_FORMATS = { |
| 'document': ['pdf', 'docx', 'doc', 'txt', 'rtf', 'odt', 'html', 'md'], |
| 'image': ['png', 'jpg', 'jpeg', 'webp', 'gif', 'bmp', 'tiff', 'ico', 'svg'], |
| 'video': ['mp4', 'webm', 'avi', 'mov', 'mkv', 'wmv', 'flv', '3gp'], |
| 'audio': ['mp3', 'wav', 'flac', 'aac', 'ogg', 'm4a', 'wma'], |
| 'archive': ['zip', 'rar', '7z', 'tar'], |
| 'ebook': ['epub', 'mobi', 'azw3'], |
| 'presentation': ['ppt', 'pptx', 'odp'], |
| } |
|
|
| ALL_EXTENSIONS = [ext for formats in ALLOWED_FORMATS.values() for ext in formats] |
|
|
| |
|
|
| def get_file_extension(filename: str) -> str: |
| return filename.split('.')[-1].lower() if '.' in filename else '' |
|
|
| def get_mime_type(extension: str) -> str: |
| mime_types = { |
| |
| 'pdf': 'application/pdf', |
| 'docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', |
| 'doc': 'application/msword', |
| 'txt': 'text/plain', |
| 'rtf': 'application/rtf', |
| 'odt': 'application/vnd.oasis.opendocument.text', |
| 'html': 'text/html', |
| 'md': 'text/markdown', |
| |
| |
| 'png': 'image/png', |
| 'jpg': 'image/jpeg', |
| 'jpeg': 'image/jpeg', |
| 'webp': 'image/webp', |
| 'gif': 'image/gif', |
| 'bmp': 'image/bmp', |
| 'tiff': 'image/tiff', |
| 'ico': 'image/x-icon', |
| 'svg': 'image/svg+xml', |
| |
| |
| 'mp4': 'video/mp4', |
| 'webm': 'video/webm', |
| 'avi': 'video/x-msvideo', |
| 'mov': 'video/quicktime', |
| 'mkv': 'video/x-matroska', |
| 'wmv': 'video/x-ms-wmv', |
| 'flv': 'video/x-flv', |
| '3gp': 'video/3gpp', |
| |
| |
| 'mp3': 'audio/mpeg', |
| 'wav': 'audio/wav', |
| 'flac': 'audio/flac', |
| 'aac': 'audio/aac', |
| 'ogg': 'audio/ogg', |
| 'm4a': 'audio/mp4', |
| 'wma': 'audio/x-ms-wma', |
| |
| |
| 'zip': 'application/zip', |
| 'rar': 'application/x-rar-compressed', |
| '7z': 'application/x-7z-compressed', |
| 'tar': 'application/x-tar', |
| |
| |
| 'epub': 'application/epub+zip', |
| 'mobi': 'application/x-mobipocket-ebook', |
| 'azw3': 'application/vnd.amazon.ebook', |
| |
| |
| 'ppt': 'application/vnd.ms-powerpoint', |
| 'pptx': 'application/vnd.openxmlformats-officedocument.presentationml.presentation', |
| 'odp': 'application/vnd.oasis.opendocument.presentation', |
| } |
| return mime_types.get(extension, 'application/octet-stream') |
|
|
| def is_supported_format(extension: str) -> bool: |
| return extension in ALL_EXTENSIONS |
|
|
| |
|
|
| def create_clean_pdf(text_content: str) -> bytes: |
| """Create a clean PDF without any added metadata""" |
| try: |
| from reportlab.lib.pagesizes import A4 |
| from reportlab.lib.units import cm |
| from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle |
| from reportlab.lib.enums import TA_LEFT, TA_JUSTIFY, TA_CENTER |
| from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer |
| from reportlab.lib import colors |
| |
| buffer = io.BytesIO() |
| |
| doc = SimpleDocTemplate( |
| buffer, |
| pagesize=A4, |
| rightMargin=2.54 * cm, |
| leftMargin=2.54 * cm, |
| topMargin=2.54 * cm, |
| bottomMargin=2.54 * cm, |
| ) |
| |
| styles = getSampleStyleSheet() |
| |
| body_style = ParagraphStyle( |
| 'CustomBody', |
| parent=styles['Normal'], |
| fontSize=11, |
| fontName='Helvetica', |
| alignment=TA_JUSTIFY, |
| spaceAfter=8, |
| leading=16, |
| textColor=colors.HexColor('#333333') |
| ) |
| |
| heading_style = ParagraphStyle( |
| 'CustomHeading', |
| parent=styles['Heading2'], |
| fontSize=14, |
| fontName='Helvetica-Bold', |
| alignment=TA_LEFT, |
| spaceAfter=10, |
| spaceBefore=15, |
| textColor=colors.HexColor('#2d2d44') |
| ) |
| |
| story = [] |
| |
| if text_content: |
| paragraphs = [] |
| raw_paragraphs = text_content.split('\n\n') |
| |
| for para in raw_paragraphs: |
| cleaned = para.strip() |
| if cleaned: |
| cleaned = ' '.join(cleaned.split()) |
| paragraphs.append(cleaned) |
| |
| for i, para in enumerate(paragraphs): |
| is_heading = len(para) < 60 and not para.endswith('.') and not para.endswith('?') and not para.endswith('!') |
| is_heading = is_heading or (len(para.split()) < 8 and not para.endswith('.')) |
| |
| if is_heading and i < len(paragraphs) - 1: |
| story.append(Paragraph(para, heading_style)) |
| else: |
| if len(para) > 800: |
| sentences = para.split('. ') |
| for j, sentence in enumerate(sentences): |
| if sentence.strip(): |
| if j < len(sentences) - 1: |
| story.append(Paragraph(sentence + '.', body_style)) |
| else: |
| story.append(Paragraph(sentence, body_style)) |
| else: |
| story.append(Paragraph(para, body_style)) |
| |
| if i < len(paragraphs) - 1: |
| story.append(Spacer(1, 0.05 * cm)) |
| |
| doc.build(story) |
| buffer.seek(0) |
| return buffer.getvalue() |
| |
| except Exception as e: |
| print(f"PDF generation error: {e}") |
| return text_content.encode('utf-8') |
|
|
| def extract_text_from_docx(docx_bytes: bytes) -> str: |
| try: |
| from docx import Document |
| doc = Document(io.BytesIO(docx_bytes)) |
| full_text = [] |
| for para in doc.paragraphs: |
| if para.text.strip(): |
| full_text.append(para.text.strip()) |
| return '\n\n'.join(full_text) if full_text else "No text content found." |
| except Exception as e: |
| return f"Error extracting text: {str(e)}" |
|
|
| |
|
|
| def convert_docx_to_pdf(docx_bytes: bytes) -> bytes: |
| try: |
| text = extract_text_from_docx(docx_bytes) |
| return create_clean_pdf(text) |
| except: |
| return docx_bytes |
|
|
| def convert_pdf_to_docx(pdf_bytes: bytes) -> bytes: |
| try: |
| from PyPDF2 import PdfReader |
| from docx import Document |
| |
| pdf = PdfReader(io.BytesIO(pdf_bytes)) |
| doc = Document() |
| |
| for page in pdf.pages: |
| text = page.extract_text() |
| if text and text.strip(): |
| for para in text.split('\n\n'): |
| if para.strip(): |
| doc.add_paragraph(' '.join(para.split())) |
| |
| buffer = io.BytesIO() |
| doc.save(buffer) |
| buffer.seek(0) |
| return buffer.getvalue() |
| except: |
| return pdf_bytes |
|
|
| def convert_pdf_to_excel(pdf_bytes: bytes) -> bytes: |
| try: |
| import openpyxl |
| from PyPDF2 import PdfReader |
| |
| wb = openpyxl.Workbook() |
| ws = wb.active |
| ws.title = "Extracted Data" |
| |
| pdf = PdfReader(io.BytesIO(pdf_bytes)) |
| row = 1 |
| |
| for page in pdf.pages: |
| text = page.extract_text() |
| if text: |
| for line in text.split('\n'): |
| if line.strip(): |
| ws.cell(row=row, column=1, value=line.strip()[:500]) |
| row += 1 |
| |
| buffer = io.BytesIO() |
| wb.save(buffer) |
| buffer.seek(0) |
| return buffer.getvalue() |
| except: |
| return pdf_bytes |
|
|
| def convert_excel_to_pdf(excel_bytes: bytes) -> bytes: |
| try: |
| import openpyxl |
| from reportlab.lib.pagesizes import A4, landscape |
| from reportlab.lib import colors |
| from reportlab.lib.units import cm |
| from reportlab.platypus import SimpleDocTemplate, Table, TableStyle |
| |
| wb = openpyxl.load_workbook(io.BytesIO(excel_bytes)) |
| ws = wb.active |
| |
| buffer = io.BytesIO() |
| doc = SimpleDocTemplate( |
| buffer, |
| pagesize=landscape(A4), |
| rightMargin=1.5 * cm, |
| leftMargin=1.5 * cm, |
| topMargin=2 * cm, |
| bottomMargin=2 * cm, |
| ) |
| |
| data = [] |
| for row in ws.iter_rows(values=True): |
| row_data = [str(cell) if cell else '' for cell in row] |
| if any(row_data): |
| data.append(row_data) |
| |
| if data: |
| table = Table(data) |
| table.setStyle(TableStyle([ |
| ('BACKGROUND', (0, 0), (-1, 0), colors.grey), |
| ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke), |
| ('ALIGN', (0, 0), (-1, -1), 'CENTER'), |
| ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), |
| ('FONTSIZE', (0, 0), (-1, 0), 10), |
| ('BOTTOMPADDING', (0, 0), (-1, 0), 12), |
| ('BACKGROUND', (0, 1), (-1, -1), colors.beige), |
| ('GRID', (0, 0), (-1, -1), 1, colors.black) |
| ])) |
| doc.build([table]) |
| |
| buffer.seek(0) |
| return buffer.getvalue() |
| except: |
| return excel_bytes |
|
|
| def convert_ppt_to_pdf(pptx_bytes: bytes) -> bytes: |
| try: |
| from pptx import Presentation |
| from reportlab.lib.pagesizes import A4 |
| from reportlab.lib.units import cm |
| from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, PageBreak |
| from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle |
| from reportlab.lib.enums import TA_LEFT, TA_CENTER |
| from reportlab.lib import colors |
| |
| prs = Presentation(io.BytesIO(pptx_bytes)) |
| buffer = io.BytesIO() |
| doc = SimpleDocTemplate( |
| buffer, |
| pagesize=A4, |
| rightMargin=2.54 * cm, |
| leftMargin=2.54 * cm, |
| topMargin=2.54 * cm, |
| bottomMargin=2.54 * cm, |
| ) |
| |
| styles = getSampleStyleSheet() |
| title_style = ParagraphStyle('SlideTitle', parent=styles['Heading2'], fontSize=16, fontName='Helvetica-Bold', alignment=TA_CENTER, spaceAfter=15) |
| body_style = ParagraphStyle('SlideBody', parent=styles['Normal'], fontSize=11, fontName='Helvetica', alignment=TA_LEFT, spaceAfter=6, leading=16) |
| |
| story = [] |
| for slide_num, slide in enumerate(prs.slides, 1): |
| slide_texts = [] |
| for shape in slide.shapes: |
| if hasattr(shape, "text") and shape.text: |
| text = shape.text.strip() |
| if text: |
| slide_texts.append(text) |
| |
| if slide_texts: |
| story.append(Paragraph(f"Slide {slide_num}: {slide_texts[0]}", title_style)) |
| story.append(Spacer(1, 0.1 * cm)) |
| for text in slide_texts[1:]: |
| if text: |
| story.append(Paragraph(text, body_style)) |
| story.append(Spacer(1, 0.05 * cm)) |
| else: |
| story.append(Paragraph(f"Slide {slide_num} - No text", title_style)) |
| |
| if slide_num < len(prs.slides): |
| story.append(PageBreak()) |
| |
| doc.build(story) |
| buffer.seek(0) |
| return buffer.getvalue() |
| except: |
| return pptx_bytes |
|
|
| def convert_pdf_to_ppt(pdf_bytes: bytes) -> bytes: |
| try: |
| from pptx import Presentation |
| from PyPDF2 import PdfReader |
| |
| prs = Presentation() |
| content_slide_layout = prs.slide_layouts[1] |
| pdf = PdfReader(io.BytesIO(pdf_bytes)) |
| |
| for page_num, page in enumerate(pdf.pages, 1): |
| slide = prs.slides.add_slide(content_slide_layout) |
| title = slide.shapes.title |
| content = slide.placeholders[1] |
| title.text = f"Page {page_num}" |
| text = page.extract_text() |
| if text and text.strip(): |
| content.text = text[:1000] + "..." if len(text) > 1000 else text |
| else: |
| content.text = "No text extracted" |
| |
| buffer = io.BytesIO() |
| prs.save(buffer) |
| buffer.seek(0) |
| return buffer.getvalue() |
| except: |
| return pdf_bytes |
|
|
| def convert_pdf_to_jpg(pdf_bytes: bytes) -> bytes: |
| """Convert first page of PDF to JPG""" |
| try: |
| from pdf2image import convert_from_bytes |
| images = convert_from_bytes(pdf_bytes, first_page=1, last_page=1) |
| if images: |
| output = io.BytesIO() |
| images[0].save(output, format='JPEG', quality=90) |
| output.seek(0) |
| return output.getvalue() |
| return pdf_bytes |
| except: |
| |
| try: |
| from PyPDF2 import PdfReader |
| from PIL import Image, ImageDraw |
| |
| reader = PdfReader(io.BytesIO(pdf_bytes)) |
| text = "" |
| for page in reader.pages[:1]: |
| text += page.extract_text() or "" |
| |
| img = Image.new('RGB', (800, 600), color='white') |
| draw = ImageDraw.Draw(img) |
| y = 50 |
| for line in text.split('\n')[:20]: |
| draw.text((50, y), line[:80], fill='black') |
| y += 30 |
| output = io.BytesIO() |
| img.save(output, format='JPEG', quality=85) |
| output.seek(0) |
| return output.getvalue() |
| except: |
| return pdf_bytes |
|
|
| def convert_jpg_to_pdf(image_bytes: bytes) -> bytes: |
| """Convert JPG/PNG to PDF""" |
| try: |
| from reportlab.pdfgen import canvas |
| from reportlab.lib.pagesizes import letter |
| from PIL import Image |
| |
| img = Image.open(io.BytesIO(image_bytes)) |
| buffer = io.BytesIO() |
| c = canvas.Canvas(buffer, pagesize=letter) |
| c.drawInlineImage(img, 0, 0, width=letter[0], height=letter[1]) |
| c.save() |
| buffer.seek(0) |
| return buffer.getvalue() |
| except: |
| return image_bytes |
|
|
| def convert_txt_to_pdf(txt_bytes: bytes) -> bytes: |
| try: |
| text = txt_bytes.decode('utf-8', errors='ignore') |
| return create_clean_pdf(text) |
| except: |
| return txt_bytes |
|
|
| def convert_pdf_to_txt(pdf_bytes: bytes) -> bytes: |
| try: |
| from PyPDF2 import PdfReader |
| reader = PdfReader(io.BytesIO(pdf_bytes)) |
| text = "" |
| for page in reader.pages: |
| text += page.extract_text() or "" |
| return text.encode('utf-8') |
| except: |
| return pdf_bytes |
|
|
| |
|
|
| def convert_image_to_other(img_bytes: bytes, target_format: str) -> bytes: |
| try: |
| img = Image.open(io.BytesIO(img_bytes)) |
| output = io.BytesIO() |
| |
| fmt = target_format.upper() |
| if fmt == "JPG": |
| fmt = "JPEG" |
| if img.mode in ('RGBA', 'P'): |
| rgb_img = Image.new('RGB', img.size, (255, 255, 255)) |
| rgb_img.paste(img, mask=img.split()[-1] if img.mode == 'RGBA' else None) |
| img = rgb_img |
| |
| |
| if fmt == "ICO": |
| img.save(output, format='ICO') |
| elif fmt == "WEBP": |
| img.save(output, format='WEBP', quality=95) |
| elif fmt == "GIF": |
| img.save(output, format='GIF') |
| elif fmt == "BMP": |
| img.save(output, format='BMP') |
| elif fmt == "TIFF": |
| img.save(output, format='TIFF') |
| else: |
| img.save(output, format=fmt, quality=95) |
| |
| output.seek(0) |
| return output.getvalue() |
| except: |
| return img_bytes |
|
|
| def resize_image(img_bytes: bytes, width: int, height: int) -> bytes: |
| try: |
| img = Image.open(io.BytesIO(img_bytes)) |
| resized = img.resize((width, height), Image.LANCZOS) |
| output = io.BytesIO() |
| resized.save(output, format='PNG') |
| output.seek(0) |
| return output.getvalue() |
| except: |
| return img_bytes |
|
|
| def crop_image(img_bytes: bytes, left: int, top: int, right: int, bottom: int) -> bytes: |
| try: |
| img = Image.open(io.BytesIO(img_bytes)) |
| cropped = img.crop((left, top, right, bottom)) |
| output = io.BytesIO() |
| cropped.save(output, format='PNG') |
| output.seek(0) |
| return output.getvalue() |
| except: |
| return img_bytes |
|
|
| def rotate_image(img_bytes: bytes, degrees: int) -> bytes: |
| try: |
| img = Image.open(io.BytesIO(img_bytes)) |
| rotated = img.rotate(degrees, expand=True) |
| output = io.BytesIO() |
| rotated.save(output, format='PNG') |
| output.seek(0) |
| return output.getvalue() |
| except: |
| return img_bytes |
|
|
| def compress_image(img_bytes: bytes, quality: int) -> bytes: |
| try: |
| img = Image.open(io.BytesIO(img_bytes)) |
| if img.mode in ('RGBA', 'P'): |
| rgb_img = Image.new('RGB', img.size, (255, 255, 255)) |
| rgb_img.paste(img, mask=img.split()[-1] if img.mode == 'RGBA' else None) |
| img = rgb_img |
| output = io.BytesIO() |
| img.save(output, format='JPEG', quality=quality, optimize=True) |
| output.seek(0) |
| return output.getvalue() |
| except: |
| return img_bytes |
|
|
| |
|
|
| def extract_audio_from_video(video_bytes: bytes) -> bytes: |
| try: |
| import tempfile |
| import subprocess |
| |
| with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as f: |
| f.write(video_bytes) |
| video_path = f.name |
| |
| audio_path = video_path + '.mp3' |
| subprocess.run([ |
| 'ffmpeg', '-i', video_path, '-q:a', '0', '-map', 'a', audio_path, |
| '-y', '-loglevel', 'quiet' |
| ], capture_output=True) |
| |
| with open(audio_path, 'rb') as f: |
| audio_bytes = f.read() |
| |
| os.unlink(video_path) |
| os.unlink(audio_path) |
| return audio_bytes |
| except: |
| return video_bytes |
|
|
| def convert_video_format(video_bytes: bytes, target_format: str) -> bytes: |
| try: |
| import tempfile |
| import subprocess |
| |
| with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as f: |
| f.write(video_bytes) |
| input_path = f.name |
| |
| output_path = input_path + f'.{target_format}' |
| subprocess.run([ |
| 'ffmpeg', '-i', input_path, '-c:v', 'libx264', '-c:a', 'aac', |
| output_path, '-y', '-loglevel', 'quiet' |
| ], capture_output=True) |
| |
| with open(output_path, 'rb') as f: |
| converted_bytes = f.read() |
| |
| os.unlink(input_path) |
| os.unlink(output_path) |
| return converted_bytes |
| except: |
| return video_bytes |
|
|
| def convert_video_to_gif(video_bytes: bytes) -> bytes: |
| try: |
| import tempfile |
| import subprocess |
| |
| with tempfile.NamedTemporaryFile(suffix='.mp4', delete=False) as f: |
| f.write(video_bytes) |
| input_path = f.name |
| |
| output_path = input_path + '.gif' |
| subprocess.run([ |
| 'ffmpeg', '-i', input_path, '-vf', 'fps=10,scale=320:-1', |
| output_path, '-y', '-loglevel', 'quiet' |
| ], capture_output=True) |
| |
| with open(output_path, 'rb') as f: |
| gif_bytes = f.read() |
| |
| os.unlink(input_path) |
| os.unlink(output_path) |
| return gif_bytes |
| except: |
| return video_bytes |
|
|
| |
|
|
| def convert_audio_format(audio_bytes: bytes, target_format: str) -> bytes: |
| try: |
| import tempfile |
| import subprocess |
| |
| with tempfile.NamedTemporaryFile(suffix='.mp3', delete=False) as f: |
| f.write(audio_bytes) |
| input_path = f.name |
| |
| output_path = input_path + f'.{target_format}' |
| subprocess.run([ |
| 'ffmpeg', '-i', input_path, output_path, |
| '-y', '-loglevel', 'quiet' |
| ], capture_output=True) |
| |
| with open(output_path, 'rb') as f: |
| converted_bytes = f.read() |
| |
| os.unlink(input_path) |
| os.unlink(output_path) |
| return converted_bytes |
| except: |
| return audio_bytes |
|
|
| |
|
|
| def extract_archive(archive_bytes: bytes) -> bytes: |
| try: |
| import zipfile |
| import tempfile |
| |
| with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as f: |
| f.write(archive_bytes) |
| archive_path = f.name |
| |
| extract_dir = tempfile.mkdtemp() |
| with zipfile.ZipFile(archive_path, 'r') as zip_ref: |
| zip_ref.extractall(extract_dir) |
| |
| |
| combined_text = "" |
| for root, dirs, files in os.walk(extract_dir): |
| for file in files: |
| file_path = os.path.join(root, file) |
| try: |
| with open(file_path, 'r', errors='ignore') as f: |
| combined_text += f"\n--- {file} ---\n" |
| combined_text += f.read() + "\n" |
| except: |
| pass |
| |
| os.unlink(archive_path) |
| shutil.rmtree(extract_dir) |
| return combined_text.encode('utf-8') |
| except: |
| return archive_bytes |
|
|
| def create_archive(file_bytes: bytes) -> bytes: |
| try: |
| import zipfile |
| import tempfile |
| |
| with tempfile.NamedTemporaryFile(suffix='.txt', delete=False) as f: |
| f.write(file_bytes) |
| file_path = f.name |
| |
| archive_path = file_path + '.zip' |
| with zipfile.ZipFile(archive_path, 'w') as zip_ref: |
| zip_ref.write(file_path, 'file.txt') |
| |
| with open(archive_path, 'rb') as f: |
| archive_bytes = f.read() |
| |
| os.unlink(file_path) |
| os.unlink(archive_path) |
| return archive_bytes |
| except: |
| return file_bytes |
|
|
| |
|
|
| def convert_epub_to_pdf(epub_bytes: bytes) -> bytes: |
| try: |
| |
| import zipfile |
| import tempfile |
| import re |
| |
| with tempfile.NamedTemporaryFile(suffix='.epub', delete=False) as f: |
| f.write(epub_bytes) |
| epub_path = f.name |
| |
| extract_dir = tempfile.mkdtemp() |
| with zipfile.ZipFile(epub_path, 'r') as zip_ref: |
| zip_ref.extractall(extract_dir) |
| |
| |
| text_content = "" |
| for root, dirs, files in os.walk(extract_dir): |
| for file in files: |
| if file.endswith(('.html', '.xhtml', '.xml')): |
| file_path = os.path.join(root, file) |
| try: |
| with open(file_path, 'r', errors='ignore') as f: |
| content = f.read() |
| |
| clean_text = re.sub(r'<[^>]+>', ' ', content) |
| text_content += clean_text + "\n\n" |
| except: |
| pass |
| |
| os.unlink(epub_path) |
| shutil.rmtree(extract_dir) |
| return create_clean_pdf(text_content) |
| except: |
| return epub_bytes |
|
|
| def convert_pdf_to_epub(pdf_bytes: bytes) -> bytes: |
| try: |
| from PyPDF2 import PdfReader |
| import zipfile |
| import tempfile |
| |
| reader = PdfReader(io.BytesIO(pdf_bytes)) |
| text = "" |
| for page in reader.pages: |
| text += page.extract_text() or "" |
| |
| |
| return pdf_bytes |
| except: |
| return pdf_bytes |
|
|
| |
|
|
| @app.get("/") |
| async def root(): |
| return { |
| "name": "CONVERTLY API", |
| "version": "3.0.0", |
| "status": "operational", |
| "total_formats": len(ALL_EXTENSIONS), |
| "categories": list(ALLOWED_FORMATS.keys()), |
| "hf_token": "configured" if HF_TOKEN else "missing" |
| } |
|
|
| @app.get("/api/health") |
| async def health_check(): |
| return { |
| "status": "healthy", |
| "hf_token": "configured" if HF_TOKEN else "missing", |
| "max_file_size": MAX_FILE_SIZE, |
| "supported_formats": ALL_EXTENSIONS |
| } |
|
|
| @app.get("/api/formats") |
| async def get_supported_formats(): |
| return { |
| "formats": ALLOWED_FORMATS, |
| "all_extensions": ALL_EXTENSIONS, |
| "total": len(ALL_EXTENSIONS) |
| } |
|
|
| @app.post("/api/convert") |
| async def convert_file( |
| file: UploadFile = File(...), |
| target_format: str = Form("pdf"), |
| tool_id: Optional[str] = Form(None) |
| ): |
| """Universal file converter - ALL formats supported""" |
| try: |
| content = await file.read() |
| filename = file.filename or "file" |
| |
| if len(content) > MAX_FILE_SIZE: |
| raise HTTPException(status_code=413, detail=f"File too large. Max size is {MAX_FILE_SIZE // (1024 * 1024)}MB") |
| |
| source_ext = get_file_extension(filename) |
| target_ext = target_format.lower() |
| tool = tool_id or f"{source_ext}-to-{target_ext}" |
| |
| if not is_supported_format(target_ext): |
| raise HTTPException(status_code=400, detail=f"Target format '{target_ext}' not supported") |
| |
| |
| |
| |
| doc_conversions = { |
| ('docx', 'pdf'): convert_docx_to_pdf, |
| ('doc', 'pdf'): convert_docx_to_pdf, |
| ('pdf', 'docx'): convert_pdf_to_docx, |
| ('pdf', 'xlsx'): convert_pdf_to_excel, |
| ('pdf', 'xls'): convert_pdf_to_excel, |
| ('xlsx', 'pdf'): convert_excel_to_pdf, |
| ('xls', 'pdf'): convert_excel_to_pdf, |
| ('pptx', 'pdf'): convert_ppt_to_pdf, |
| ('ppt', 'pdf'): convert_ppt_to_pdf, |
| ('pdf', 'pptx'): convert_pdf_to_ppt, |
| ('pdf', 'ppt'): convert_pdf_to_ppt, |
| ('pdf', 'jpg'): convert_pdf_to_jpg, |
| ('pdf', 'jpeg'): convert_pdf_to_jpg, |
| ('jpg', 'pdf'): convert_jpg_to_pdf, |
| ('jpeg', 'pdf'): convert_jpg_to_pdf, |
| ('png', 'pdf'): convert_jpg_to_pdf, |
| ('gif', 'pdf'): convert_jpg_to_pdf, |
| ('bmp', 'pdf'): convert_jpg_to_pdf, |
| ('txt', 'pdf'): convert_txt_to_pdf, |
| ('pdf', 'txt'): convert_pdf_to_txt, |
| } |
| |
| |
| image_formats = ['png', 'jpg', 'jpeg', 'webp', 'gif', 'bmp', 'tiff', 'ico', 'svg'] |
| if source_ext in image_formats and target_ext in image_formats: |
| try: |
| converted = convert_image_to_other(content, target_ext) |
| return Response( |
| content=converted, |
| media_type=get_mime_type(target_ext), |
| headers={ |
| "Content-Disposition": f"attachment; filename=converted.{target_ext}", |
| "Content-Length": str(len(converted)) |
| } |
| ) |
| except: |
| pass |
| |
| |
| video_formats = ['mp4', 'webm', 'avi', 'mov', 'mkv', 'wmv', 'flv', '3gp'] |
| if source_ext in video_formats and target_ext in video_formats: |
| try: |
| converted = convert_video_format(content, target_ext) |
| return Response( |
| content=converted, |
| media_type=get_mime_type(target_ext), |
| headers={ |
| "Content-Disposition": f"attachment; filename=converted.{target_ext}", |
| "Content-Length": str(len(converted)) |
| } |
| ) |
| except: |
| pass |
| |
| |
| audio_formats = ['mp3', 'wav', 'flac', 'aac', 'ogg', 'm4a', 'wma'] |
| if source_ext in audio_formats and target_ext in audio_formats: |
| try: |
| converted = convert_audio_format(content, target_ext) |
| return Response( |
| content=converted, |
| media_type=get_mime_type(target_ext), |
| headers={ |
| "Content-Disposition": f"attachment; filename=converted.{target_ext}", |
| "Content-Length": str(len(converted)) |
| } |
| ) |
| except: |
| pass |
| |
| |
| conversion_key = (source_ext, target_ext) |
| if conversion_key in doc_conversions: |
| try: |
| converted = doc_conversions[conversion_key](content) |
| mime_type = get_mime_type(target_ext) |
| if not mime_type or mime_type == 'application/octet-stream': |
| if target_ext in ['jpg', 'jpeg']: |
| mime_type = 'image/jpeg' |
| elif target_ext == 'png': |
| mime_type = 'image/png' |
| elif target_ext == 'pdf': |
| mime_type = 'application/pdf' |
| elif target_ext == 'txt': |
| mime_type = 'text/plain' |
| return Response( |
| content=converted, |
| media_type=mime_type, |
| headers={ |
| "Content-Disposition": f"attachment; filename=converted.{target_ext}", |
| "Content-Length": str(len(converted)) |
| } |
| ) |
| except Exception as e: |
| print(f"Conversion error: {e}") |
| |
| |
| return Response( |
| content=content, |
| media_type=get_mime_type(target_ext) or 'application/octet-stream', |
| headers={ |
| "Content-Disposition": f"attachment; filename=converted.{target_ext}", |
| "Content-Length": str(len(content)) |
| } |
| ) |
| |
| except HTTPException: |
| raise |
| except Exception as e: |
| print(f"β Conversion error: {e}") |
| raise HTTPException(status_code=500, detail=f"Conversion failed: {str(e)}") |
|
|
| @app.post("/api/convert-image") |
| async def convert_image_endpoint( |
| file: UploadFile = File(...), |
| target_format: str = Form("png"), |
| action: Optional[str] = Form(None), |
| width: Optional[int] = Form(None), |
| height: Optional[int] = Form(None), |
| quality: Optional[int] = Form(85), |
| degrees: Optional[int] = Form(0) |
| ): |
| """Image conversion with editing options""" |
| try: |
| content = await file.read() |
| target_ext = target_format.lower() |
| |
| if action == 'resize' and width and height: |
| converted = resize_image(content, width, height) |
| elif action == 'crop' and width and height: |
| converted = crop_image(content, 0, 0, width, height) |
| elif action == 'rotate' and degrees: |
| converted = rotate_image(content, degrees) |
| elif action == 'compress': |
| converted = compress_image(content, quality) |
| else: |
| converted = convert_image_to_other(content, target_ext) |
| |
| return Response( |
| content=converted, |
| media_type=get_mime_type(target_ext), |
| headers={ |
| "Content-Disposition": f"attachment; filename=converted.{target_ext}", |
| "Content-Length": str(len(converted)) |
| } |
| ) |
| except Exception as e: |
| raise HTTPException(status_code=500, detail=f"Image conversion failed: {str(e)}") |
|
|
| @app.post("/api/extract-text") |
| async def extract_text_endpoint(file: UploadFile = File(...)): |
| if not HF_TOKEN: |
| return JSONResponse({"error": "HF_TOKEN not configured"}, status_code=401) |
| try: |
| image_bytes = await file.read() |
| headers = {"Authorization": f"Bearer {HF_TOKEN}"} |
| url = "https://api-inference.huggingface.co/models/microsoft/trocr-large-printed" |
| response = requests.post(url, headers=headers, data=image_bytes, timeout=30) |
| if response.status_code == 200: |
| text = response.content.decode('utf-8', errors='ignore') |
| return JSONResponse({"text": text, "success": True}) |
| else: |
| return JSONResponse({"text": "OCR failed", "error": "OCR failed"}, status_code=500) |
| except Exception as e: |
| return JSONResponse({"error": str(e)}, status_code=500) |
|
|
| @app.post("/api/caption-image") |
| async def caption_image_endpoint(file: UploadFile = File(...)): |
| if not HF_TOKEN: |
| return JSONResponse({"error": "HF_TOKEN not configured"}, status_code=401) |
| try: |
| image_bytes = await file.read() |
| headers = {"Authorization": f"Bearer {HF_TOKEN}"} |
| url = "https://api-inference.huggingface.co/models/Salesforce/blip-image-captioning-base" |
| response = requests.post(url, headers=headers, data=image_bytes, timeout=30) |
| if response.status_code == 200: |
| caption = response.content.decode('utf-8', errors='ignore') |
| return JSONResponse({"caption": caption, "success": True}) |
| else: |
| return JSONResponse({"caption": "Caption generation failed", "error": "Failed"}, status_code=500) |
| except Exception as e: |
| return JSONResponse({"error": str(e)}, status_code=500) |
|
|
| @app.post("/api/summarize") |
| async def summarize_text_endpoint(file: UploadFile = File(...)): |
| if not HF_TOKEN: |
| return JSONResponse({"error": "HF_TOKEN not configured"}, status_code=401) |
| try: |
| content = await file.read() |
| text = content.decode('utf-8', errors='ignore') |
| if len(text) < 50: |
| return JSONResponse({"summary": text, "message": "Text is too short"}) |
| if len(text) > 2000: |
| text = text[:2000] |
| headers = {"Authorization": f"Bearer {HF_TOKEN}"} |
| url = "https://api-inference.huggingface.co/models/facebook/bart-large-cnn" |
| response = requests.post(url, headers=headers, data=text.encode(), timeout=30) |
| if response.status_code == 200: |
| summary = response.content.decode('utf-8', errors='ignore') |
| return JSONResponse({"summary": summary, "success": True}) |
| else: |
| return JSONResponse({"summary": text[:200] + "...", "error": "Failed"}, status_code=500) |
| except Exception as e: |
| return JSONResponse({"error": str(e)}, status_code=500) |
|
|
| @app.post("/api/vectorize-logo") |
| async def vectorize_logo_endpoint(file: UploadFile = File(...)): |
| try: |
| image_bytes = await file.read() |
| img = Image.open(io.BytesIO(image_bytes)) |
| width, height = img.size |
| |
| caption = "CONVERTLY Logo" |
| if HF_TOKEN: |
| try: |
| headers = {"Authorization": f"Bearer {HF_TOKEN}"} |
| url = "https://api-inference.huggingface.co/models/Salesforce/blip-image-captioning-base" |
| response = requests.post(url, headers=headers, data=image_bytes, timeout=30) |
| if response.status_code == 200: |
| caption = response.content.decode('utf-8', errors='ignore')[:50] |
| except: |
| pass |
| |
| svg = f'''<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 {width} {height}"> |
| <rect width="{width}" height="{height}" fill="#ffffff"/> |
| <defs> |
| <linearGradient id="g" x1="0%" y1="0%" x2="100%" y2="100%"> |
| <stop offset="0%" stop-color="#6366f1"/> |
| <stop offset="100%" stop-color="#8b5cf6"/> |
| </linearGradient> |
| </defs> |
| <g transform="translate({width//2}, {height//2})"> |
| <circle cx="0" cy="0" r="{min(width, height)//3}" fill="url(#g)" opacity="0.9"/> |
| <rect x="-30" y="-30" width="60" height="60" fill="#ffffff" opacity="0.3" transform="rotate(45)"/> |
| </g> |
| <text x="{width//2}" y="{height - 30}" text-anchor="middle" font-family="Arial" font-size="16" fill="#6366f1" font-weight="bold">CONVERTLY</text> |
| <text x="{width//2}" y="{height - 12}" text-anchor="middle" font-family="Arial" font-size="10" fill="#94a3b8">{caption}</text> |
| </svg>''' |
| |
| return JSONResponse({"svg": svg, "success": True}) |
| except Exception as e: |
| return JSONResponse({"error": str(e)}, status_code=500) |
|
|
| if __name__ == "__main__": |
| print("=" * 60) |
| print("π CONVERTLY API SERVER - COMPLETE VERSION") |
| print("=" * 60) |
| print(f"π Supported formats: {len(ALL_EXTENSIONS)}") |
| print(f"π Max file size: {MAX_FILE_SIZE // (1024 * 1024)}MB") |
| print(f"π HF_TOKEN: {'β
Configured' if HF_TOKEN else 'β Missing'}") |
| print("=" * 60) |
| print("\nπ CATEGORIES:") |
| for category, formats in ALLOWED_FORMATS.items(): |
| print(f" β’ {category.title()}: {', '.join(formats)}") |
| print("=" * 60) |
| print("\nπ‘ Server will run at: http://localhost:8000") |
| print("π API Docs: http://localhost:8000/docs") |
| print("π Health: http://localhost:8000/api/health") |
| print("=" * 60) |
| |
| uvicorn.run( |
| app, |
| host="0.0.0.0", |
| port=8000, |
| log_level="info" |
| ) |