import datetime import os import re import shutil from fastapi import APIRouter, HTTPException from fastapi.responses import FileResponse, JSONResponse from app.api.vessel.schemas import VesselFilter from app.api.vessel.service import VesselService from jinja2 import Environment, FileSystemLoader FUNCTIONAL_AREA_PREFIX_MAP = { "sire-chapter-03-crew-management": "MSSV-2000", "sire-chapter-04-navigation-and-communications": "MSSV-2050", "sire-chapter-05-safety-management": "MSSV-2100", "sire-chapter-06-pollution-prevention": "MSSV-2150", "sire-chapter-07-maritime-security": "MSSV-2175", "sire-chapter-8-1-8-3-cargo-and-ballast-system-petroleum": "MSSV-2200", "sire-chapter-8-2-8-3-cargo-and-ballast-system-chemical": "MSSV-2225", "sire-chapter-8-4-8-5-8-6-cargo-and-ballast-system-lpg-lng": "MSSV-2250", "sire-chapter-09-mooring-and-anchoring": "MSSV-2300", "sire-chapter-10-machinery-space-engine-and-steering-space": "MSSV-2400", "sire-chapter-11-general-appearance-and-conditions": "MSSV-2325", "sire-chapter-12-ice": "MSSV-2350", "sire-value-added-courses": "MSSV-2500", "dry-bulk-navigation": "DBMV-2000", "dry-bulk-cargo": "DBMV-3000", "dry-bulk-safety": "DBMV-4000", "dry-bulk-care-for-persons": "DBMV-5000", "dry-bulk-engineering": "DBMV-6000", "dry-bulk-management": "DBMV-8000", "deepsea-navigation": "MSVE-1000", "deepsea-engineering-and-control-systems-for-deck-officers": "MSVE-1050", "deepsea-cargo-handling": "MSVE-1100", "deepsea-safety-of-operation-and-care-for-persons": "MSVE-1200", "deepsea-soft-skills": "MSVE-1300", "deepsea-marine-engineering": "MSVE-1400", "navigation": "MSTS-200", "engineering-and-control-systems-for-deck-officers": "MSTS-250", "cargo-handling": "MSTS-300", "safety-of-operation-and-care-for-persons": "MSTS-400", "soft-skills": "MSTS-500", "marine-engineering": "MSTS-600", "electrical-electronics-control-engineering": "MSTS-700", "management-functions": "MSTS-800", "asp-internal-courses": "ASP-2500" } CURRENT_DIR = os.path.dirname(os.path.abspath(__file__)) PROJECT_ROOT = os.path.abspath(os.path.join(CURRENT_DIR, "..", "..", "..")) FILES_BASE_DIR = os.path.join(PROJECT_ROOT, "files") OFFLINE_CONTENT_DIR = os.path.join(FILES_BASE_DIR, "ASTAPI_STRUCTURE", "Offline_Content") os.makedirs(OFFLINE_CONTENT_DIR, exist_ok=True) vessel_router = APIRouter() @vessel_router.post("/courses/roles-functional-areas-courses") def get_roles_functional_areas_and_courses(filters: VesselFilter): return VesselService().get_roles_functional_areas_and_courses(filters) templates = Environment(loader=FileSystemLoader("templates")) def sanitize_filename(name: str) -> str: """ Sanitize functional area name to create a valid HTML file name. Lowercase, replace spaces with '-', remove all special characters. """ name = name.strip().lower() name = re.sub(r'[^a-z0-9]+', '-', name) # replace all non-alphanumeric sequences with '-' name = name.strip('-') # remove leading/trailing '-' return name def generate_file_url(functional_area: str) -> str: """ Generate a file:/// URL for the local HTML file of a functional area. """ filename = sanitize_filename(functional_area) + ".htm" path = os.path.abspath(os.path.join("files", "ASTAPI_STRUCTURE/Offline_Content/eLearning", filename)) return f"file:///{path.replace(os.sep, '/')}" def sanitize_path(name: str) -> str: """ Convert role name into safe flat filename part: - Replace spaces with '-' - Replace '/' with '-' - Keep letters, numbers, '-', '&' - Lowercase """ name = name.strip().lower() name = re.sub(r'[^a-z0-9]+', '-', name) # replace all non-alphanumeric sequences with '-' name = name.strip('-') # remove leading/trailing '-' return name def normalize_key(name: str) -> str: name = name.strip().lower() name = re.sub(r'[^a-z0-9]+', '-', name) # Replace any non-alphanumeric with '-' return name.strip('-') def normalize_courses(courses): """ Standardize course list for both roles and functional areas. Supports input as list of strings OR list of dicts. """ normalized = [] for c in courses or []: if isinstance(c, dict): # if already has 'lms_course', keep it if "lms_course" in c: normalized.append({"lms_course": c["lms_course"]}) # if has 'courseName', map it to lms_course elif "courseName" in c: normalized.append({"lms_course": c["courseName"]}) else: # fallback: convert entire dict to string normalized.append({"lms_course": str(c)}) else: # plain string course normalized.append({"lms_course": str(c)}) return normalized @vessel_router.get("/courses/vessel/{vessel_id}/generate-all-html", response_model=None) def generate_all_html_for_vessel(vessel_id: int): """ Generate HTML files for all roles and functional areas for a vessel, replacing old content within Offline_Content directory. """ service = VesselService() response = service.get_roles_functional_areas_and_courses(VesselFilter(vesselId=vessel_id)) if response.get("status") != "1": raise HTTPException(status_code=404, detail=response.get("message", "Data not found")) vessel_id = response["vesselId"] data = response.get("data", {}) base_dir = OFFLINE_CONTENT_DIR os.makedirs(base_dir, exist_ok=True) # Clean only known series folders before regenerating for folder in os.listdir(base_dir): folder_path = os.path.join(base_dir, folder) if os.path.isdir(folder_path) and folder.lower() in {"elearning", "video-series"}: shutil.rmtree(folder_path) generated_files = {} # Generate per-series HTMLs for series_name, content in data.items(): folder_name = ( "elearning" if series_name.lower() == "elearning" else "video-series" if "video" in series_name.lower() else series_name.replace(" ", "_").lower() ) output_dir = os.path.join(base_dir, folder_name) os.makedirs(output_dir, exist_ok=True) roles_dict = content.get("roles") or {} functional_areas_dict = content.get("functional_areas") or {} category_functional_areas = {} for category_name, func_areas in functional_areas_dict.items(): category_functional_areas[category_name] = { func_area_name: f"./{sanitize_filename(func_area_name)}.htm" for func_area_name in func_areas.keys() } start_file = os.path.join(output_dir, "start.htm") role_file = os.path.join(output_dir, "role.htm") start_html = templates.get_template("start_template.html").render( category_functional_areas=category_functional_areas ) role_html = templates.get_template("role_template.html").render( roles=list(roles_dict.keys()), ) for path, content in [(start_file, start_html), (role_file, role_html)]: with open(path, "w", encoding="utf-8") as f: f.write(content) # --- Generate Functional Area Files --- functional_area_files = {} # inside generate_all_html_for_vessel, where you iterate functional_areas: prefix_lookup = {} for category_name, func_areas in functional_areas_dict.items(): for area_name, courses in func_areas.items(): sanitized_area = sanitize_filename(area_name).lower() file_name = f"{sanitized_area}.htm" file_path = os.path.join(output_dir, file_name) if not isinstance(courses, list): print(f" Skipping invalid courses data for '{area_name}' in '{category_name}': {type(courses)}") continue prefix = FUNCTIONAL_AREA_PREFIX_MAP.get(sanitized_area, "MSTS-400") # --- Determine prefix dynamically --- if category_name == "Dry Bulk E-learning Courses": dry_bulk_prefix_map = { "dry-bulk-navigation": "DBMS-200", "dry-bulk-cargo": "DBMS-300", "dry-bulk-safety": "DBMS-400", "dry-bulk-care-for-persons": "DBMS-500", "dry-bulk-engineering": "DBMS-600", "dry-bulk-management": "DBMS-800", } prefix = dry_bulk_prefix_map.get(sanitized_area, "DBMS-999") elif category_name == "Dry Bulk Video Series Courses": print(sanitized_area) dry_bulk_video_prefix_map = { "dry-bulk-navigation": "DBMV-2000", "dry-bulk-cargo": "DBMV-3000", "dry-bulk-safety": "DBMV-4000", "dry-bulk-care-for-persons": "DBMV-5000", "dry-bulk-engineering": "DBMV-6000", "dry-bulk-management": "DBMV-8000", } prefix = dry_bulk_video_prefix_map.get(sanitized_area, "DBMV-9999") for course in courses: if isinstance(course, dict): # Your original dict processing key = normalize_key(course.get("lms_course", "")) prefix_lookup[key] = prefix else: # Handle when course is a string # For example, extract key from course string (e.g., take first two parts separated by '-') key_part = course # fallback if unexpected format key = normalize_key(key_part) prefix_lookup[key] = prefix html = templates.get_template("functional_area_template.html").render( functional_area=area_name, prefix=prefix, courses=normalize_courses(courses), ) # write base .htm with open(file_path, "w", encoding="utf-8") as f: f.write(html) # create prefix folder and fix relative paths # Read back the generated functional area HTML with open(file_path, "r", encoding="utf-8") as f: html_content = f.read() # Adjust relative paths for one-level deeper folder html_content = ( html_content .replace('href="../', 'href="../../') .replace('src="../', 'src="../../') .replace('window.location.replace(`', 'window.location.replace(`../') .replace('href="start.htm"', 'href="../start.htm"') # Fix back button ) # Create the prefix folder (e.g., MSSV-2000) prefix_dir = os.path.join(output_dir, prefix) os.makedirs(prefix_dir, exist_ok=True) # Write the adjusted file as start.htm inside the prefix folder prefix_start_file = os.path.join(prefix_dir, "start.htm") with open(prefix_start_file, "w", encoding="utf-8") as f: f.write(html_content) #shutil.copy(file_path, prefix_start_file) # ------------------------------------------------------------------ functional_area_files[area_name] = f"file:///{os.path.abspath(file_path).replace(os.sep, '/')}" # Store generated file paths role_files = {} for role_name, courses in roles_dict.items(): file_name = f"role-{sanitize_path(role_name)}.htm" file_path = os.path.join(output_dir, file_name) normalized_courses = normalize_courses(courses) # Debug for course in normalized_courses: key = normalize_key(course.get("lms_course", "")) html = templates.get_template("role_series_template.html").render( role=role_name, series=series_name, courses=[ { **course, "prefix": prefix_lookup.get( normalize_key(course.get("lms_course", "")), "No-Prefix-found" ), } for course in normalized_courses ], ) with open(file_path, "w", encoding="utf-8") as f: f.write(html) generated_files[series_name] = { "start": f"file:///{os.path.abspath(start_file).replace(os.sep, '/')}", "role": f"file:///{os.path.abspath(role_file).replace(os.sep, '/')}", "role_files": role_files, "functional_area_files": functional_area_files, } # --- Generate Home (global start.htm) --- home_template = "home_template.html" home_output_path = os.path.join(base_dir, "start.htm") elearning_exists = any("elearning" in key.lower() for key in generated_files.keys()) video_exists = any("video" in key.lower() for key in generated_files.keys()) # Render the home page dynamically home_html = templates.get_template(home_template).render( elearning_exists=elearning_exists, video_exists=video_exists, ) with open(home_output_path, "w", encoding="utf-8") as f: f.write(home_html) return JSONResponse({ "status": "1", "message": f"All HTML files generated successfully for vesselId {vessel_id}", "vesselId": vessel_id, "series_generated": list(generated_files.keys()), "files": generated_files, }) @vessel_router.get("/courses/offline-content/download") def download_offline_content(): """ Compress the Offline_Content folder into a zip and return as download. """ if not os.path.exists(OFFLINE_CONTENT_DIR): raise HTTPException(status_code=404, detail="Offline_Content folder not found") timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") zip_name = f"Offline_Content_{timestamp}" zip_base = os.path.join(FILES_BASE_DIR, zip_name) zip_path = f"{zip_base}.zip" try: shutil.make_archive(zip_base, 'zip', OFFLINE_CONTENT_DIR) return FileResponse( path=os.path.abspath(zip_path), filename=os.path.basename(zip_path), media_type="application/zip", ) except Exception as e: raise HTTPException(status_code=500, detail=f"Error creating ZIP: {str(e)}")