# UPLOAD FUNCTIONS.PY import requests import gradio as gr import pandas as pd import tiktoken import tempfile from PyPDF2 import PdfReader from tqdm import tqdm from pydantic import BaseModel, Field from phi.agent import Agent, RunResponse from phi.model.groq import Groq from sentence_transformers import SentenceTransformer from sentence_transformers import CrossEncoder #from gradio_client import Client, handle_file import os from pptx import Presentation from pptx2img import PPTXConverter # For splitting slides import uuid import shutil from PIL import Image import pandas as pd import requests import gradio as gr from pydantic import BaseModel, Field from typing import List import tiktoken from datetime import datetime import zipfile from PIL import Image import gradio as gr import threading import time import requests def get_access_token(): flow = app.initiate_device_flow(scopes=SCOPES) print("Go to", flow["verification_uri"]) print("Enter the code:", flow["user_code"]) result = app.acquire_token_by_device_flow(flow) if "access_token" not in result: print("❌ Could not acquire token:", result.get("error_description")) exit() return result["access_token"] # Function to generate a unique PPT ID def generate_unique_ppt_id(): return str(uuid.uuid4())[:8] # Generate an 8-character unique ID def truncate_text_to_tokens(text, max_tokens, model_name="cl100k_base"): encoding = tiktoken.get_encoding(model_name) tokens = encoding.encode(text) truncated_tokens = tokens[:max_tokens] return encoding.decode(truncated_tokens) def split_and_convert_ppt(file_path, output_folder_slides, output_folder_images): os.makedirs(output_folder_slides, exist_ok=True) os.makedirs(output_folder_images, exist_ok=True) presentation = Presentation(file_path) slide_texts = [] file_name = os.path.basename(file_path).split('.')[0] print('File Name ',file_name) print('File Path ',file_path) for i in range(len(presentation.slides)): unique_slide_id = f"{file_name}_{ppt_unique_id}_slide_{i + 1}" slide_file_path = os.path.join(output_folder_slides, f"{unique_slide_id}.pptx") print('Slide_file_path',slide_file_path) image_path = os.path.join(output_folder_images, f"{unique_slide_id}_slide_1.png") # refer to pptx2img it stores iamge in this format new_name = f"{pptx_name}_slide_{idx + 1}.png" print('Image file path',image_path) # ✅ Step 1: Create a single-slide PPTX new_presentation = Presentation(file_path) slide_indexes_to_remove = [j for j in range(len(new_presentation.slides)) if j != i] for idx in sorted(slide_indexes_to_remove, reverse=True): r_id = new_presentation.slides._sldIdLst[idx].rId new_presentation.part.drop_rel(r_id) del new_presentation.slides._sldIdLst[idx] new_presentation.save(slide_file_path) del new_presentation # ✅ Step 2: Convert the single-slide PPTX to image converter = PPTXConverter() converter.convert_pptx_to_images(slide_file_path, output_folder_images) print(f"Slide {i+1} converted to image: {image_path}") # ✅ Step 3: Extract text from the slide image # Switching off OCR #slide_text = extract_text_from_image(image_path) #using PPTX for text extraction(actualy its quality is better then tesseratct) # Extract text using python-pptx (editable text) slide = presentation.slides[i] pptx_text = "" for shape in slide.shapes: if hasattr(shape, "text"): pptx_text += shape.text.strip() + "\n" print(f"🔡 PPTX Text Extractedfrom slide {i + 1}:\n", pptx_text.strip()) slide_texts.append(pptx_text.strip()) return slide_texts def generate_metadata_with_retry(full_text, retries=3, max_tokens=5000, decrement=100, model_name="cl100k_base"): for attempt in range(1, retries + 2): try: truncated_text = truncate_text_to_tokens(full_text, max_tokens, model_name) print(f"🔁 Attempt {attempt}: Generating metadata with ~{count_tokens(truncated_text)} tokens...") metadata = generate_metadata(truncated_text) print("📝 Metadata generated successfully.") return metadata # ✅ Return on success except Exception as e: print(f"❌ Error on attempt {attempt}: {str(e)}") if attempt == retries + 1: print("🚨 Max retries reached. Metadata generation failed.") return None else: max_tokens -= decrement print(f"🔄 Retrying with {max_tokens} tokens...") # Function to generate metadata using phidata agent def generate_metadata(ocr_text): # Initialize the Agent with detailed instructions metadata_agent = Agent( name="Metadata Generator", role="Generates structured metadata for presentations based on their content.", instructions=[ "Your task is to analyze the provided text and generate structured metadata for the presentation.", "Carefully evaluate the content to determine the most appropriate values for each metadata field.", # Rule 1: PPT Unique ID "For the 'PPT_Unique_ID', use the first 8 characters of the MD5 hash of the input text. " "This ensures uniqueness across presentations.", # Rule 2: Suitable Title "For the 'Suitable_Title', create a concise and meaningful title that captures the essence of the presentation. " "Focus on first slide where title of presentation is given along with key themes, topics, or keywords mentioned in the text.", # Rule 3: Slide Category "For the 'Slide_Category', classify the presentation into one of the following categories: " "The category or theme of the slides (e.g., Risk management , Data Analytics , Technology etc)" "Base your decision on the overall theme or subject matter of the content.", # Rule 4 :PPT owner "Find The owner of the presentation ie who makes the presentation (eg: Done by name and designation ie Mr. baswaraj ,Princpial ADG , Additional Director ,or organisations like NCTC,DG Systems, Directorate of Logistics etc)" "Dont Asssume if u could not found ,mention Not Available" # Rule 5: Audience/Forum "For the 'Audience_Forum', identify the target audience or forum for the presentation. " "(e.g.,NACIN , WCO, Presentation before Member (CBIC)etc )." "Dont Asssume if could not found ,mention Not Available" "Consider the tone, language, and purpose of the content.", # Rule 6: Short Summary "For the 'Short_Summary', provide a brief summary of the presentation's content with all keywords in 10 sentences. " "Highlight the keywords ,topics, main points or objectives of the presentation.", "Mention the title also in the short summary ,owner and audience of the presentation" # General Guidelines "Ensure all fields are filled and meaningful. If unsure about a field, make an educated guess based on the context.", ], model=Groq(id="deepseek-r1-distill-llama-70b"), # Replace with actual model ID response_model=PPTMetadata, markdown=True, debug_mode=True, show_tool_calls=True, monitoring=True) # Run the agent to generate metadata response = metadata_agent.run( f"Generate data fields for the following presentation content: {ocr_text}") return response.content # Function to get folder ID in OneDrive def get_folder_id(folder_path, headers): folders = folder_path.split("/") parent_id = None print("creating folder id for ",folder_path) for folder_name in folders: url = f"https://graph.microsoft.com/v1.0/me/drive/root/children" if not parent_id else f"https://graph.microsoft.com/v1.0/me/drive/items/{parent_id}/children" response = requests.get(url, headers=headers) if response.status_code != 200: print(f"Failed to retrieve folder '{folder_name}'. Error: {response.text}") return None items = response.json().get("value", []) folder_item = next((item for item in items if item["name"] == folder_name and "folder" in item), None) if not folder_item: # Create the folder if it doesn't exist create_url = "https://graph.microsoft.com/v1.0/me/drive/root/children" if not parent_id else f"https://graph.microsoft.com/v1.0/me/drive/items/{parent_id}/children" create_response = requests.post(create_url, headers=headers, json={ "name": folder_name, "folder": {}, "@microsoft.graph.conflictBehavior": "rename" }) if create_response.status_code not in [200, 201]: print(f"Failed to create folder '{folder_name}'. Error: {create_response.text}") return None folder_item = create_response.json() parent_id = folder_item["id"] return parent_id # Function to upload file to OneDrive def upload_to_onedrive(file_path, folder_id, headers): file_name = os.path.basename(file_path) upload_url = f"https://graph.microsoft.com/v1.0/me/drive/items/{folder_id}:/{file_name}:/content" with open(file_path, "rb") as file: file_content = file.read() response = requests.put(upload_url, headers=headers, data=file_content) if response.status_code in [200, 201]: print(f"Uploaded {file_name} to OneDrive.") return response.json()["id"] else: print(f"Failed to upload {file_name}. Error: {response.text}") return None # Function to count tokens using tiktoken def count_tokens(text, model_name="cl100k_base"): encoding = tiktoken.get_encoding(model_name) tokens = encoding.encode(text) return len(tokens) def list_folder_files(folder_id, headers): url = f"https://graph.microsoft.com/v1.0/me/drive/items/{folder_id}/children" response = requests.get(url, headers=headers) if response.status_code != 200: raise ValueError(f"Failed to list folder contents. Error: {response.text}") return response.json().get("value", []) def download_onedrive_file(file_id, filename, headers): url = f"https://graph.microsoft.com/v1.0/me/drive/items/{file_id}" r = requests.get(url, headers=headers).json() download_url = r.get("@microsoft.graph.downloadUrl") response = requests.get(download_url) with open(filename, 'wb') as f: f.write(response.content) def update_and_upload_metadata_simplified(metadata_list, metadata_folder_id, metadata_with_fulltext_folder_id, headers): df_new = pd.DataFrame(metadata_list, columns=[ "Unique_Slide_ID", "Slide_OCR_Text", "PPT_OCR_Text", "Slide_Embedding", "Short_Summary_Embedding", "PPT_Unique_ID", "Suitable_Title", "Slide_Category", "PPT_Owner", "Audience_Forum", "Short_Summary", "Slide_File_Path", "Slide_File_ID", "Full_PPT_File_Path", "Full_PPT_File_ID", "Thumbnail_File_Path", "Thumbnail_File_ID","Upload_date"]) for csv_file, folder_id, drop_column in [ ("Master_metadata.csv", metadata_folder_id, 'PPT_OCR_Text'), ("Master_fulltext_metadata.csv", metadata_with_fulltext_folder_id, None)]: #folder_id = get_folder_id(folder_path, headers) files = list_folder_files(folder_id, headers) file_item = next((item for item in files if item['name'] == csv_file), None) print('File items', file_item) if file_item: download_onedrive_file(file_item['id'], csv_file, headers) df_existing = pd.read_csv(csv_file) df_merged = pd.concat([df_existing, df_new], ignore_index=True) else: df_merged = df_new if drop_column: df_merged = df_merged.drop(columns=[drop_column]) df_merged.to_csv(csv_file, index=False) upload_to_onedrive(csv_file, folder_id, headers) print(f"✅ Uploaded: {csv_file}") return "✅PPT Processing and Metadata update complete!" # Main processing function def process_presentation(file): try: # Step 0: Validate file format file_path = file.name if hasattr(file, "name") else file file_extension = os.path.splitext(file_path)[-1].lower() gr.Info() if file_extension not in ['.pptx']: raise ValueError("Unsupported file format. Please upload .pptx") # Extract the base file name (without extension) file_name = os.path.basename(file_path).split('.')[0] print('File Name ',file_name) # Step 1: Generate unique PPT ID global ppt_unique_id ppt_unique_id = generate_unique_ppt_id() upload_date = datetime.now().strftime('%Y-%m-%d') # Step 2: Acquire access token via device flow # access_token = get_access_token() # print('access_token',access_token) print('PPT_unique id',ppt_unique_id) # Step 3: Get folder IDs for OneDrive # headers = { # "Authorization": f"Bearer {access_token}", # "Content-Type": "application/json" # } gr.Info('Connecting to OneDrive..') ppt_repo_folder_id = get_folder_id("Projects Apps/PPT Maker/ppt_repo", headers) slides_repo_folder_id = get_folder_id("Projects Apps/PPT Maker/slides_repo", headers) slide_image_repo_folder_id = get_folder_id("Projects Apps/PPT Maker/slide_image_repo", headers) metadata_folder_id=get_folder_id('Projects Apps/PPT Maker/Metadata_file',headers) metadata_with_fulltext_folder_id=get_folder_id('Projects Apps/PPT Maker/Metadata_with_fulltext',headers) print('ppt_repo_folder_id',ppt_repo_folder_id) print('slides_repo_folder_id',slides_repo_folder_id) print('slide_image_repo_folder_id',slide_image_repo_folder_id) print('metadata_folder_id',metadata_folder_id) if not (ppt_repo_folder_id and slides_repo_folder_id and slide_image_repo_folder_id and metadata_folder_id) : gr.Error('Could not find or create required folders in OneDrive.') raise ValueError("Could not find or create required folders in OneDrive.") # Step 2: Upload the full PPT file to OneDrive #ppt_repo_folder_id = get_folder_id("Projects Apps/PPT Maker/ppt_repo") # ✅ Step: Check if file already exists in ppt_repo existing_files = list_folder_files(ppt_repo_folder_id, headers) ppt_file_name = os.path.basename(file_path) if any(item['name'] == ppt_file_name for item in existing_files): gr.Error('⚠️ A file named ' + ppt_file_name + ' already exists in the PPT repository. Please rename your file or delete the existing one before re-uploading.') return f"⚠️ A file named '{ppt_file_name}' already exists in the PPT repository. Please rename your file or delete the existing one before re-uploading." full_ppt_file_id = upload_to_onedrive(file_path, ppt_repo_folder_id,headers) gr.Info('PPT uploaded to OneDrive..') full_ppt_file_name = os.path.basename(file_path) full_ppt_file_path = f"/Projects Apps/PPT Maker/ppt_repo/{full_ppt_file_name}" # Step 3: Split PPT into individual slides and convert to images gr.Info('Processing the PPT and indexing ..it may take a while ') temp_output_folder_slides = "/temp/temp_slides" temp_output_folder_images = "/temp/temp_images" slide_texts = split_and_convert_ppt(file_path, temp_output_folder_slides, temp_output_folder_images) print('PPT splitted and converted successfully') # Compile full OCR text full_text = "\n".join(slide_texts) gr.Info('AI agent processing the data .') metadata = generate_metadata_with_retry(full_text, retries=3, max_tokens=5000, decrement=100, model_name="cl100k_base") # Step 5: Process each slide and prepare metadata for storage #slides_repo_folder_id = get_folder_id("Projects Apps/PPT Maker/slides_repo") #slide_image_repo_folder_id = get_folder_id("Projects Apps/PPT Maker/slide_image_repo") metadata_list = [] gr.Info('Uploading the individual slides and images into repo ') for i, slide_text in enumerate(slide_texts): unique_slide_id = f"{file_name}_{ppt_unique_id}_slide_{i + 1}" slide_file_path = f"{temp_output_folder_slides}/{unique_slide_id}.pptx" slide_image_path = f"{temp_output_folder_images}/{unique_slide_id}_slide_1.png" # Upload individual slide (.pptx) to slides_repo slide_file_id = upload_to_onedrive(slide_file_path, slides_repo_folder_id,headers) slide_file_path_onedrive = f"/Projects Apps/PPT Maker/slides_repo/{unique_slide_id}.pptx" print(f'Slide{i} uploaded into Onedrive') # Upload slide image (.png) to slide_image_repo thumbnail_file_id = upload_to_onedrive(slide_image_path, slide_image_repo_folder_id,headers) thumbnail_file_path_onedrive = f"/Projects Apps/PPT Maker/slide_image_repo/{unique_slide_id}.png" print(f'Image{i} uploaded into Onedrive') # Generate embedding for the slide slide_embedding = embedding_model.encode(slide_text).tolist() short_summary_embedding = embedding_model.encode(metadata.Short_Summary).tolist() # Prepare metadata for storage metadata_list.append([ unique_slide_id, # Unique Slide ID slide_text, # Slide OCR Text full_text, # PPT OCR Text str(slide_embedding), # Embedding str(short_summary_embedding), ppt_unique_id, # PPT Unique ID metadata.Suitable_Title, # Suitable Title metadata.Slide_Category, # Slide Category metadata.PPT_Owner, # PPT Owner metadata.Audience_Forum, # Audience Forum metadata.Short_Summary, # Short Summary slide_file_path_onedrive, # Slide File Path (.pptx) slide_file_id, # Slide File ID (.pptx) full_ppt_file_path, # Full PPT File Path full_ppt_file_id, # Full PPT File ID thumbnail_file_path_onedrive, # Thumbnail File Path (.png) thumbnail_file_id , # Thumbnail File ID (.png) upload_date # upload date ]) # Clean up temporary files for this slide os.remove(slide_file_path) os.remove(slide_image_path) print('Slides cleared from temp') # # Clean up temporary folders # os.rmdir(temp_output_folder_slides) # os.rmdir(temp_output_folder_images) # Clean up temporary folders (forcefully deletes all contents inside) shutil.rmtree(temp_output_folder_slides, ignore_errors=True) shutil.rmtree(temp_output_folder_images, ignore_errors=True) print('Temp folders cleared') gr.Info('Vectorising the meta data and uploading in Onedrive..') return update_and_upload_metadata_simplified( metadata_list, metadata_folder_id, metadata_with_fulltext_folder_id, headers ) except Exception as e: return f"An error occurred: {str(e)}"