Update app.py

app.py

@@ -207,62 +207,73 @@ def check_tool_availability():
     return mafft_available, iqtree_available, mafft_cmd, iqtree_cmd
 
 # --- Pipeline Functions ---
+def cleanup_file(file_path: str) -> None:
+    """Utility function to safely delete a file and log errors."""
+    if file_path and os.path.exists(file_path):
+        try:
+            os.unlink(file_path)
+            logger.debug(f"Cleaned up {file_path}")
+        except Exception as cleanup_error:
+            logger.warning(f"Failed to clean up {file_path}: {cleanup_error}")
+
 def phylogenetic_placement(sequence: str, mafft_cmd: str, iqtree_cmd: str):
     query_fasta = None  # Predefine to avoid scoping issues
     try:
+        # Input validation
         if len(sequence.strip()) < 100:
             return False, "Sequence too short (<100 bp).", None, None
+
+        # Setup file paths
         query_id = f"QUERY_{uuid.uuid4().hex[:8]}"
         query_fasta = os.path.join(QUERY_OUTPUT_DIR, f"{query_id}.fa")
         aligned_with_query = os.path.join(QUERY_OUTPUT_DIR, f"{query_id}_aligned.fa")
         output_prefix = os.path.join(QUERY_OUTPUT_DIR, f"{query_id}_placed_tree")
+
+        # Check reference files
         if not os.path.exists(ALIGNMENT_PATH) or not os.path.exists(TREE_PATH):
-            if query_fasta and os.path.exists(query_fasta):  # Cleanup before return
-                try:
-                    os.unlink(query_fasta)
-                except Exception as cleanup_error:
-                    logger.warning(f"Failed to clean up {query_fasta}: {cleanup_error}")
+            cleanup_file(query_fasta)
             return False, "Reference alignment or tree not found.", None, None
+
+        # Write query FASTA
         query_record = SeqRecord(Seq(sequence.upper()), id=query_id, description="")
         SeqIO.write([query_record], query_fasta, "fasta")
+
+        # Run MAFFT
         with open(aligned_with_query, "w") as output_file:
-            subprocess.run([
-                mafft_cmd, "--add", query_fasta, "--reorder", ALIGNMENT_PATH
-            ], stdout=output_file, stderr=subprocess.PIPE, text=True, timeout=600, check=True)
+            result = subprocess.run(
+                [mafft_cmd, "--add", query_fasta, "--reorder", ALIGNMENT_PATH],
+                stdout=output_file,
+                stderr=subprocess.PIPE,
+                text=True,
+                timeout=600,
+                check=True
+            )
         if not os.path.exists(aligned_with_query) or os.path.getsize(aligned_with_query) == 0:
-            if query_fasta and os.path.exists(query_fasta):  # Cleanup before return
-                try:
-                    os.unlink(query_fasta)
-                except Exception as cleanup_error:
-                    logger.warning(f"Failed to clean up {query_fasta}: {cleanup_error}")
+            cleanup_file(query_fasta)
             return False, "MAFFT alignment failed.", None, None
-        subprocess.run([
-            iqtree_cmd, "-s", aligned_with_query, "-g", TREE_PATH,
-            "-m", "GTR+G", "-pre", output_prefix, "-redo"
-        ], capture_output=True, text=True, timeout=1200, check=True)
+
+        # Run IQ-TREE
+        result = subprocess.run(
+            [iqtree_cmd, "-s", aligned_with_query, "-g", TREE_PATH, "-m", "GTR+G", "-pre", output_prefix, "-redo"],
+            capture_output=True,
+            text=True,
+            timeout=1200,
+            check=True
+        )
         treefile = f"{output_prefix}.treefile"
         if not os.path.exists(treefile):
-            if query_fasta and os.path.exists(query_fasta):  # Cleanup before return
-                try:
-                    os.unlink(query_fasta)
-                except Exception as cleanup_error:
-                    logger.warning(f"Failed to clean up {query_fasta}: {cleanup_error}")
+            cleanup_file(query_fasta)
             return False, "IQ-TREE placement failed.", aligned_with_query, None
+
+        # Success
         success_msg = f"Placement completed!\nQuery ID: {query_id}\nAlignment: {os.path.basename(aligned_with_query)}\nTree: {os.path.basename(treefile)}"
-        if query_fasta and os.path.exists(query_fasta):  # Cleanup before return
-            try:
-                os.unlink(query_fasta)
-            except Exception as cleanup_error:
-                logger.warning(f"Failed to clean up {query_fasta}: {cleanup_error}")
+        cleanup_file(query_fasta)
         return True, success_msg, aligned_with_query, treefile
-    except Exception as main_error:  # Renamed from 'e'
+
+    except Exception as main_error:
         logger.error(f"Phylogenetic placement failed: {main_error}", exc_info=True)
-        if query_fasta and os.path.exists(query_fasta):  # Cleanup in except
-            try:
-                os.unlink(query_fasta)
-            except Exception as cleanup_error:
-                logger.warning(f"Failed to clean up {query_fasta}: {cleanup_error}")
-        return False, f"Error: {str(main_error)}", None, None 
+        cleanup_file(query_fasta)
+        return False, f"Error: {str(main_error)}", None, None
 def analyze_sequence_for_tree(sequence: str, matching_percentage: float):
     try:
         logger.debug("Starting tree analysis...")
@@ -437,28 +448,16 @@ async def run_pipeline_from_file(fasta_file_obj, similarity_score, build_ml_tree
             temp_file_path = temp_file.name
         dna_input = read_fasta_file(temp_file_path)
         if not dna_input:
-            if temp_file_path and os.path.exists(temp_file_path):
-                try:
-                    os.unlink(temp_file_path)
-                except Exception as cleanup_error:
-                    logger.warning(f"Failed to delete temp file {temp_file_path}: {cleanup_error}")
+            cleanup_file(temp_file_path)
             return "❌ Failed to read FASTA file", "", "", "", "", None, None, None, None, "No input", "No input", None, None
         result = run_pipeline(dna_input, similarity_score, build_ml_tree)
-        if temp_file_path and os.path.exists(temp_file_path):
-            try:
-                os.unlink(temp_file_path)
-            except Exception as cleanup_error:
-                logger.warning(f"Failed to delete temp file {temp_file_path}: {cleanup_error}")
+        cleanup_file(temp_file_path)
         return result
     except Exception as main_error:  # Renamed from 'e'
         logger.error(f"Pipeline from file error: {main_error}", exc_info=True)
-        if temp_file_path and os.path.exists(temp_file_path):
-            try:
-                os.unlink(temp_file_path)
-            except Exception as cleanup_error:
-                logger.warning(f"Failed to delete temp file {temp_file_path}: {cleanup_error}")
+        cleanup_file(temp_file_path)
         error_msg = f"❌ Error: {str(main_error)}"
-        return error_msg, "", "", "", "", None, None, None, None, error_msg, error_msg, None, None# --- Pydantic Models ---
+        return error_msg, "", "", "", "", None, None, None, None, error_msg, error_msg, None, None
 class AnalysisRequest(BaseModel):
     sequence: str
     similarity_score: float = 95.0