Update api/routes.py

api/routes.py

@@ -5,6 +5,7 @@ from db.mongo import users_collection, patients_collection, appointments_collect
 from core.security import hash_password, verify_password, create_access_token, get_current_user
 from datetime import datetime, timedelta
 from bson import ObjectId
+from bson.errors import InvalidId  # Added to fix the 'InvalidId' not defined error
 from typing import Optional, List, Dict
 from pydantic import BaseModel, Field
 from pymongo import UpdateOne, InsertOne, IndexModel
@@ -113,6 +114,26 @@ def standardize_language(language: str) -> str:
         return 'en'  # Default to English
     return LANGUAGE_MAP.get(language, 'en')
 
+def escape_latex_special_chars(text: str) -> str:
+    """Escape special LaTeX characters to prevent compilation errors."""
+    if not isinstance(text, str):
+        return ""
+    replacements = {
+        "&": "\\&",
+        "%": "\\%",
+        "$": "\\$",
+        "#": "\\#",
+        "_": "\\_",
+        "{": "\\{",
+        "}": "\\}",
+        "~": "\\textasciitilde{}",
+        "^": "\\textasciicircum{}",
+        "\\": "\\textbackslash{}"
+    }
+    for char, escape in replacements.items():
+        text = text.replace(char, escape)
+    return text
+
 async def process_synthea_patient(bundle: dict, file_path: str) -> Optional[dict]:
     logger.debug(f"Processing patient from file: {file_path}")
     patient_data = {}
@@ -640,64 +661,110 @@ async def generate_patient_pdf(patient_id: str, current_user: dict = Depends(get
 \end{document}
         """
 
-        # Build the LaTeX-safe rows
+        # Build the LaTeX-safe rows with escaped special characters
         notes = "\n".join([
-            "{} & {} & {} \\\\".format(n.get("date", ""), n.get("type", ""), n.get("text", "").replace("&", "\\&"))
+            "{} & {} & {} \\\\".format(
+                escape_latex_special_chars(n.get("date", "")),
+                escape_latex_special_chars(n.get("type", "")),
+                escape_latex_special_chars(n.get("text", ""))
+            )
             for n in patient.get("notes", [])
         ])
         conditions = "\n".join([
             "{} & {} & {} & {} & {} \\\\".format(
-                c.get("id", ""), c.get("code", ""), c.get("status", ""), c.get("onset_date", ""), c.get("verification_status", "")
+                escape_latex_special_chars(c.get("id", "")),
+                escape_latex_special_chars(c.get("code", "")),
+                escape_latex_special_chars(c.get("status", "")),
+                escape_latex_special_chars(c.get("onset_date", "")),
+                escape_latex_special_chars(c.get("verification_status", ""))
             )
             for c in patient.get("conditions", [])
         ])
         medications = "\n".join([
             "{} & {} & {} & {} & {} \\\\".format(
-                m.get("id", ""), m.get("name", ""), m.get("status", ""), m.get("prescribed_date", ""), m.get("dosage", "")
+                escape_latex_special_chars(m.get("id", "")),
+                escape_latex_special_chars(m.get("name", "")),
+                escape_latex_special_chars(m.get("status", "")),
+                escape_latex_special_chars(m.get("prescribed_date", "")),
+                escape_latex_special_chars(m.get("dosage", ""))
             )
             for m in patient.get("medications", [])
         ])
         encounters = "\n".join([
             "{} & {} & {} & {} & {} \\\\".format(
-                e.get("id", ""), e.get("type", ""), e.get("status", ""), e.get("period", {}).get("start", ""), e.get("service_provider", "")
+                escape_latex_special_chars(e.get("id", "")),
+                escape_latex_special_chars(e.get("type", "")),
+                escape_latex_special_chars(e.get("status", "")),
+                escape_latex_special_chars(e.get("period", {}).get("start", "")),
+                escape_latex_special_chars(e.get("service_provider", ""))
             )
             for e in patient.get("encounters", [])
         ])
 
-        # Update the generated_on date to reflect the current time: 03:10 PM CET, May 16, 2025
+        # Update the generated_on date to reflect the current time: 03:14 PM CET, May 16, 2025
         latex_filled = latex_template % {
-            "generated_on": datetime.strptime("2025-05-16 15:10:00+01:00", "%Y-%m-%d %H:%M:%S%z").strftime("%A, %B %d, %Y at %I:%M %p"),
-            "fhir_id": patient.get("fhir_id", ""),
-            "full_name": patient.get("full_name", ""),
-            "gender": patient.get("gender", ""),
-            "dob": patient.get("date_of_birth", ""),
-            "age": calculate_age(patient.get("date_of_birth", "")) or "N/A",
-            "address": "{}, {}, {}, {}, {}".format(
-                patient.get("address", ""), patient.get("city", ""), patient.get("state", ""),
-                patient.get("postal_code", ""), patient.get("country", "")
-            ),
-            "marital_status": patient.get("marital_status", ""),
-            "language": patient.get("language", ""),
+            "generated_on": datetime.strptime("2025-05-16 15:14:00+01:00", "%Y-%m-%d %H:%M:%S%z").strftime("%A, %B %d, %Y at %I:%M %p"),
+            "fhir_id": escape_latex_special_chars(patient.get("fhir_id", "")),
+            "full_name": escape_latex_special_chars(patient.get("full_name", "")),
+            "gender": escape_latex_special_chars(patient.get("gender", "")),
+            "dob": escape_latex_special_chars(patient.get("date_of_birth", "")),
+            "age": escape_latex_special_chars(str(calculate_age(patient.get("date_of_birth", "")) or "N/A")),
+            "address": escape_latex_special_chars("{}, {}, {}, {}, {}".format(
+                patient.get("address", ""),
+                patient.get("city", ""),
+                patient.get("state", ""),
+                patient.get("postal_code", ""),
+                patient.get("country", "")
+            )),
+            "marital_status": escape_latex_special_chars(patient.get("marital_status", "")),
+            "language": escape_latex_special_chars(patient.get("language", "")),
             "notes": notes or "No notes available \\\\",
             "conditions": conditions or "No conditions available \\\\",
             "medications": medications or "No medications available \\\\",
             "encounters": encounters or "No encounters available \\\\"
         }
 
-        # Write LaTeX to file
-        with NamedTemporaryFile(suffix=".tex", mode="w", delete=False) as tex_file:
-            tex_file.write(latex_filled)
-            tex_path = tex_file.name
-
+        # Write LaTeX to file with error handling
         try:
-            subprocess.run(["latexmk", "-pdf", tex_path], check=True, capture_output=True)
+            with NamedTemporaryFile(suffix=".tex", mode="w", delete=False, encoding="utf-8") as tex_file:
+                tex_file.write(latex_filled)
+                tex_path = tex_file.name
+
+            # Compile LaTeX to PDF
+            try:
+                result = subprocess.run(
+                    ["latexmk", "-pdf", tex_path],
+                    check=True,
+                    capture_output=True,
+                    text=True
+                )
+            except subprocess.CalledProcessError as e:
+                raise HTTPException(
+                    status_code=500,
+                    detail=f"LaTeX compilation failed: {e.stderr}"
+                )
+
+            # Read the generated PDF
             pdf_path = tex_path.replace(".tex", ".pdf")
+            if not os.path.exists(pdf_path):
+                raise HTTPException(
+                    status_code=500,
+                    detail="PDF file was not generated"
+                )
+
             with open(pdf_path, "rb") as f:
                 pdf_bytes = f.read()
 
-            subprocess.run(["latexmk", "-c", tex_path], check=True)
-            os.remove(tex_path)
-            os.remove(pdf_path)
+            # Clean up temporary files
+            try:
+                subprocess.run(["latexmk", "-c", tex_path], check=True, capture_output=True)
+                os.remove(tex_path)
+                os.remove(pdf_path)
+            except Exception as cleanup_error:
+                raise HTTPException(
+                    status_code=500,
+                    detail=f"Failed to clean up temporary files: {str(cleanup_error)}"
+                )
 
             return Response(
                 content=pdf_bytes,
@@ -705,11 +772,19 @@ async def generate_patient_pdf(patient_id: str, current_user: dict = Depends(get
                 headers={"Content-Disposition": f"attachment; filename=patient_{patient.get('fhir_id', 'unknown')}_record.pdf"}
             )
 
-        except subprocess.CalledProcessError as e:
-            raise HTTPException(status_code=500, detail="LaTeX compilation failed")
+        except IOError as io_error:
+            raise HTTPException(
+                status_code=500,
+                detail=f"Failed to write LaTeX file: {str(io_error)}"
+            )
 
+    except HTTPException as http_error:
+        raise http_error
     except Exception as e:
-        raise HTTPException(status_code=500, detail=f"Failed to generate PDF: {str(e)}")
+        raise HTTPException(
+            status_code=500,
+            detail=f"Failed to generate PDF: {str(e)}"
+        )
     finally:
         # Restore the logger level for other routes
         logger.setLevel(logging.INFO)