import json import os import random import string import re import base64 from fastapi_mail import FastMail, MessageSchema, ConnectionConfig from fastapi import HTTPException, status from fastapi.responses import JSONResponse import xml.etree.ElementTree as ET import xml.dom.minidom as minidom import uuid def generate_random_text(): return ''.join(random.choices(string.ascii_uppercase + string.digits, k=6)) def replace_special_characters(input_string): # Replace all non-alphanumeric characters (including spaces) with an underscore sanitized_string = re.sub(r'[^a-zA-Z0-9]', '_', input_string) return sanitized_string def get_error_message(response): messages = [error["message"] for error in response.get("errors", [])] return messages async def send_email(recipient: str, subject: str, body: str): conf = ConnectionConfig( MAIL_USERNAME=os.getenv('SES_ACCESS_KEY_ID'), MAIL_PASSWORD=os.getenv('SES_SECRET_ACCESS_KEY'), MAIL_FROM=os.getenv('SES_SENDER_EMAIL'), MAIL_PORT=465, MAIL_SERVER=os.getenv('SES_AWS_HOST'), MAIL_STARTTLS=False, MAIL_SSL_TLS=True, USE_CREDENTIALS=True ) message = MessageSchema( subject=subject, recipients=[recipient], # List of recipients body=body, subtype="html" ) fm = FastMail(conf) try: await fm.send_message(message) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) def read_json_file(file_path): """ Reads a JSON file and returns its content as a Python dictionary or list, depending on the JSON structure. Args: file_path (str): The path to the JSON file to be read. Returns: dict or list: The content of the JSON file. The type depends on the structure of the JSON file (object or array at the root). """ with open(file_path) as f: return json.load(f) def read_txt_file(file_path): with open(file_path) as f: return f.read() def parse(text): regex = re.compile(r']:\s+"(.*?)"\s+http') text = regex.sub("]: http", text) return text def construct_citation_dict_from_article(search_results, index_to_find): if search_results is None: return None citation_dict = {} for url, index in search_results["url_to_unified_index"].items(): citation_dict[index] = { "url": url, "title": search_results["url_to_info"][url]["title"], "snippets": search_results["url_to_info"][url]["snippets"], } return citation_dict.get(index_to_find, None) def create_manifest(course_title: str, content_dir: str, wiki_dir: str, output_file: str = "imsmanifest.xml"): ET.register_namespace('', "http://www.imsglobal.org/xsd/imsccv1p1/imscp_v1p1") ET.register_namespace('lom', "http://ltsc.ieee.org/xsd/imsccv1p1/LOM/resource") ET.register_namespace('lomimscc', "http://ltsc.ieee.org/xsd/imsccv1p1/LOM/manifest") ET.register_namespace('xsi', "http://www.w3.org/2001/XMLSchema-instance") manifest = ET.Element("manifest", { "identifier": f"man_{uuid.uuid4().hex}", "xmlns": "http://www.imsglobal.org/xsd/imsccv1p1/imscp_v1p1", "xmlns:lom": "http://ltsc.ieee.org/xsd/imsccv1p1/LOM/resource", "xmlns:lomimscc": "http://ltsc.ieee.org/xsd/imsccv1p1/LOM/manifest", "xmlns:xsi": "http://www.w3.org/2001/XMLSchema-instance", "xsi:schemaLocation": ( "http://www.imsglobal.org/xsd/imsccv1p1/imscp_v1p1 " "http://www.imsglobal.org/profile/cc/ccv1p1/ccv1p1_imscp_v1p2_v1p0.xsd " "http://ltsc.ieee.org/xsd/imsccv1p1/LOM/resource " "http://www.imsglobal.org/profile/cc/ccv1p1/LOM/ccv1p1_lomresource_v1p0.xsd " "http://ltsc.ieee.org/xsd/imsccv1p1/LOM/manifest " "http://www.imsglobal.org/profile/cc/ccv1p1/LOM/ccv1p1_lommanifest_v1p0.xsd" ) }) metadata = ET.SubElement(manifest, "metadata") ET.SubElement(metadata, "schema").text = "IMS Common Cartridge" ET.SubElement(metadata, "schemaversion").text = "1.1.0" lom = ET.SubElement(metadata, "lomimscc:lom") general = ET.SubElement(lom, "lomimscc:general") title = ET.SubElement(general, "lomimscc:title") ET.SubElement(title, "lomimscc:string").text = course_title # organizations = ET.SubElement(manifest, "organizations") org = ET.SubElement(organizations, "organization", {"identifier": "org_1", "structure": "rooted-hierarchy"}) resources_el = ET.SubElement(manifest, "resources") for i, filename in enumerate(sorted(os.listdir(content_dir)), start=1): if not filename.endswith(".html"): continue title_text = filename.replace("_", " ").replace(".html", "").title() mod_id = f"module_{i}" item_id = f"item_{i}" res_id = f"res_{i}" module_item = ET.SubElement(org, "item", {"identifier": mod_id}) ET.SubElement(module_item, "title").text = f"Module {i}: {title_text}" # Add nested item nested_item = ET.SubElement(module_item, "item", {"identifier": item_id, "identifierref": res_id}) ET.SubElement(nested_item, "title").text = title_text # Add resource full_path = os.path.join(wiki_dir, filename) resource = ET.SubElement(resources_el, "resource", { "identifier": res_id, "type": "webcontent", "href": full_path }) ET.SubElement(resource, "file", {"href": full_path}) # Write XML file rough_string = ET.tostring(manifest, 'utf-8') reparsed = minidom.parseString(rough_string) pretty_xml = reparsed.toprettyxml(indent=" ") with open(output_file, "w", encoding="utf-8") as f: f.write(pretty_xml) print(f"✅ imsmanifest.xml created at: {output_file}") def generate_resources_from_folder(folder_path: str, base_path: str = "wiki_content") -> list: resources = [] for i, filename in enumerate(os.listdir(folder_path), start=1): if not filename.lower().endswith(".html"): continue # skip non-html files full_path = os.path.join(base_path, filename) resources.append({ "id": f"res_{i}", "type": "webcontent", "href": full_path }) return resources def normalize_spaces(input_string): return re.sub(r'\s+', ' ', input_string).strip() def split_markdown_sections(md_text: str) -> list: """Splits the markdown into sections based on top-level (#) headings.""" sections = re.split(r'(?=^# .+)', md_text, flags=re.MULTILINE) return [s.strip() for s in sections if s.strip()] def errorResponse(message,status_code = 400): if not status_code: status_code = status.HTTP_400_BAD_REQUEST return JSONResponse( status_code=status_code, content={ "status": status_code, "message": message, }, ) def successResponse(message, data): return JSONResponse( status_code=status.HTTP_200_OK, content={ "status": status.HTTP_200_OK, "message": message, "data": data }, )