from typing import Dict, List, Optional from common.config.app_settings import app_settings from datetime import datetime, timedelta, timezone from backend.sharepoint.sharepoint_graph_client import SharePointGraphClient from common.constants.region import UserRegion from backend.document.document_manager import DocumentManager from backend.content.constants import ( ContentSource, ContentMediaType, ContentDataFormat, ContentFileConstants, ) from backend.content.models import ContentDirectory, ContentFolderDoc class ContentSharePointManager: def __init__(self) -> None: self.sharepoint_client = SharePointGraphClient( tenant_id="cf151ee8-5c2c-4fe7-a1c4-809ba43c9f24", client_id="a91e3c1d-d55d-423d-83cb-a736df229817", client_secret=app_settings.FREELEAPS_WWW_AS_AZURE_CLIENT_SECRET, host_name="netorgft10898514.sharepoint.com", site_name="MediaRoom", ) self.sharepoint_client.set_document_scope( document_library="freeleaps", root_folder="production" ) self.share_point_file_expiry = timedelta(hours=24) def __generate_created__by__(self, folder_name): return "content-service-" + folder_name.replace("/", "-").lower() async def retrieve_directories_for_folder( self, folder_name: str, region: UserRegion ): content_folder_name = folder_name + "/" + region.name sp_folders = self.sharepoint_client.list_sub_folders(content_folder_name) current_time = datetime.now(timezone.utc) folder = await ContentFolderDoc.find_one( ContentFolderDoc.folder_name == folder_name, ContentFolderDoc.region == region.value, ) if folder is None: folder = ContentFolderDoc( folder_name=folder_name, content_directories=[], udpate_time=current_time, update_source=ContentSource.SHAREPOINT, valid_thru=current_time + self.share_point_file_expiry, region=region, ) else: # TODO: Remove all existing content directories # Otherwise it will keep growing folder.content_directories.clear() for sp_folder in sp_folders: content_directory = ContentDirectory( content_name=sp_folder["name"], cover_document_id=None, summary_text=None, title_text=None, content_link=None, content_document_id=None, ) sp_files = self.sharepoint_client.list_files( content_folder_name + "/" + sp_folder["name"] ) for sp_file in sp_files: if ( sp_file["name"].lower() == ContentFileConstants.COVER_FILE_NAME.lower() ): cover_file_content = self.sharepoint_client.get_file_content( sp_file["id"] ) cover_document_manager = DocumentManager() file_name = sp_file["name"].lower() created_by = self.__generate_created__by__(folder_name=folder_name) content_directory.cover_document_id = ( await cover_document_manager.save_document_file( created_by, file_name, cover_file_content ) ) elif ( sp_file["name"].lower() == ContentFileConstants.SUMMARY_FILE_NAME.lower() ): content_directory.summary_text = ( self.sharepoint_client.get_file_content(sp_file["id"]) ) elif ( sp_file["name"].lower() == ContentFileConstants.TITLE_FILE_NAME.lower() ): content_directory.title_text = ( self.sharepoint_client.get_file_content(sp_file["id"]) ) elif ( sp_file["name"].lower() == ContentFileConstants.CONTENT_LINK_FILE_NAME.lower() ): content_directory.content_link = ( self.sharepoint_client.get_file_content(sp_file["id"]) ) elif ( sp_file["name"].lower() == ContentFileConstants.CONTENT_HTML_FILE_NAME.lower() ): content_directory.content_html = ( self.sharepoint_client.get_file_content(sp_file["id"]) ) elif ( sp_file["name"].lower() == ContentFileConstants.CONTENT_TEXT_FILE_NAME.lower() ): content_directory.content_text = ( self.sharepoint_client.get_file_content(sp_file["id"]) ) elif ( sp_file["name"].lower() == ContentFileConstants.CONTENT_PDF_FILE_NAME.lower() ): content_file_content = self.sharepoint_client.get_file_content( sp_file["id"] ) content_document_manager = DocumentManager() file_name = sp_file["name"] created_by = self.__generate_created__by__(folder_name=folder_name) content_directory.content_document_id = ( await content_document_manager.save_document_file( created_by, file_name, content_file_content ) ) folder.content_directories.append(content_directory) folder.udpate_time = current_time folder.update_source = ContentSource.SHAREPOINT folder.valid_thru = current_time + self.share_point_file_expiry await folder.save() async def retrieve_directorys_for_all_folders(self): current_time = datetime.now(timezone.utc) folders = await ContentFolderDoc.find( ContentFolderDoc.update_source == ContentSource.SHAREPOINT, ContentFolderDoc.valid_thru < current_time, ).to_list() for folder in folders: await self.retrieve_directories_for_folder( folder.folder_name, folder.region )