85 lines
3.4 KiB
Python
85 lines
3.4 KiB
Python
import asyncio
|
|
from datetime import datetime, timezone, timedelta
|
|
from typing import List, Dict
|
|
from backend.sharepoint.sharepoint_graph_client import SharePointGraphClient
|
|
from backend.document.document_manager import DocumentManager
|
|
from backend.content.models import ContentDirectory, ContentFolderDoc
|
|
from backend.content.constants import ContentFileConstants, ContentSource
|
|
|
|
from common.config.app_settings import app_settings
|
|
|
|
|
|
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 list_sub_folders(self, folder_name: str) -> List[dict]:
|
|
"""
|
|
Fetches the subfolders under the specified folder from SharePoint.
|
|
"""
|
|
try:
|
|
# Use SharePointGraphClient to list subfolders
|
|
return self.sharepoint_client.list_sub_folders(folder_name)
|
|
except Exception as e:
|
|
raise ValueError(f"Failed to list subfolders for {folder_name}: {e}")
|
|
|
|
async def process_subfolder(
|
|
self, content_folder_name: str, sp_folder: dict
|
|
) -> ContentDirectory:
|
|
content_directory = ContentDirectory(content_name=sp_folder["name"])
|
|
sp_files = self.sharepoint_client.list_files(
|
|
f"{content_folder_name}/{sp_folder['name']}"
|
|
)
|
|
|
|
# Process files in parallel
|
|
tasks = [
|
|
self.__process_file(file, content_directory, content_folder_name)
|
|
for file in sp_files
|
|
]
|
|
await asyncio.gather(*tasks)
|
|
|
|
return content_directory
|
|
|
|
async def __process_file(
|
|
self, sp_file: dict, content_directory: ContentDirectory, folder_name: str
|
|
):
|
|
document_manager = DocumentManager()
|
|
file_content = self.sharepoint_client.get_file_content(sp_file["id"])
|
|
|
|
if sp_file["name"].lower() == ContentFileConstants.COVER_FILE_NAME.lower():
|
|
content_directory.cover_document_id = (
|
|
await document_manager.save_document_file(
|
|
self.__generate_created_by(folder_name),
|
|
sp_file["name"],
|
|
file_content,
|
|
)
|
|
)
|
|
elif sp_file["name"].lower() == ContentFileConstants.SUMMARY_FILE_NAME.lower():
|
|
content_directory.summary_text = file_content
|
|
elif sp_file["name"].lower() == ContentFileConstants.TITLE_FILE_NAME.lower():
|
|
content_directory.title_text = file_content
|
|
elif (
|
|
sp_file["name"].lower()
|
|
== ContentFileConstants.CONTENT_PDF_FILE_NAME.lower()
|
|
):
|
|
content_directory.content_document_id = (
|
|
await document_manager.save_document_file(
|
|
self.__generate_created_by(folder_name),
|
|
sp_file["name"],
|
|
file_content,
|
|
)
|
|
)
|
|
|
|
def __generate_created_by(self, folder_name: str) -> str:
|
|
return f"content-service-{folder_name.replace('/', '-').lower()}"
|