fix

ragg/ingest_trigger.py

@@ -3,6 +3,7 @@ import sys
 import subprocess
 import threading
 import traceback
+from glob import glob
 from flask import Blueprint, jsonify, request
 
 ingest_trigger_bp = Blueprint("ingest_trigger_bp", __name__)
@@ -10,45 +11,128 @@ ingest_trigger_bp = Blueprint("ingest_trigger_bp", __name__)
 # Prevent concurrent executions
 _ingest_lock = threading.Lock()
 
+def _list_dir_sample(path: str, patterns=("*.pdf", "*.PDF"), limit: int = 10):
+    try:
+        files = []
+        for pat in patterns:
+            files.extend(glob(os.path.join(path, pat)))
+        files.sort()
+        total = len(files)
+        sample = files[:limit]
+        return {"exists": os.path.isdir(path), "total": total, "sample": sample}
+    except Exception as e:
+        return {"exists": False, "error": str(e)}
+
 @ingest_trigger_bp.route("/ingest/ping", methods=["GET"])
 def ingest_ping():
-    return jsonify({"ok": True, "cwd": os.getcwd()}), 200
+    info = {
+        "ok": True,
+        "cwd": os.getcwd(),
+        "env": {
+            "ENV": os.getenv("ENV"),
+            "INGEST_MODULE": os.getenv("INGEST_MODULE", "ragg.ingest_all"),
+            "INGEST_TIMEOUT_SEC": os.getenv("INGEST_TIMEOUT_SEC", "1800"),
+            "CHROMA_DIR": os.getenv("CHROMA_DIR"),
+            "CHROMA_ROOT": os.getenv("CHROMA_ROOT"),
+            "HF_HOME": os.getenv("HF_HOME"),
+        }
+    }
+    print("\n=== [PING] ===", flush=True)
+    print(info, flush=True)
+    return jsonify(info), 200
 
 @ingest_trigger_bp.route("/ingest", methods=["POST"])
 def run_ingest():
     if not _ingest_lock.acquire(blocking=False):
+        print("[DEBUG] Ingest lock already held → busy", flush=True)
         return jsonify({"status": "busy", "message": "Ingestion already in progress"}), 409
 
     try:
-        # Run as module to ensure package imports work
         module_name = os.getenv("INGEST_MODULE", "ragg.ingest_all")
+        timeout_sec = int(os.getenv("INGEST_TIMEOUT_SEC", "1800"))
 
+        print("\n=== [INGEST TRIGGER] ===", flush=True)
         print("Trigger called", flush=True)
         print("Module:", module_name, flush=True)
         print("CWD:", os.getcwd(), flush=True)
-        print("PYTHONPATH:", sys.path, flush=True)
+        print("PYTHON:", sys.executable, flush=True)
+        print("PYTHONPATH (len={}):".format(len(sys.path)), flush=True)
+        print(sys.path, flush=True)
+
+        # Log key env vars the ingestion relies on
+        print("ENV:", os.getenv("ENV"), flush=True)
+        print("CHROMA_DIR:", os.getenv("CHROMA_DIR"), flush=True)
+        print("CHROMA_ROOT:", os.getenv("CHROMA_ROOT"), flush=True)
+        print("HF_HOME:", os.getenv("HF_HOME"), flush=True)
+
+        # Quick preflight: show whether expected PDF folders exist and sample files
+        bases = [
+            "ragg/pdfs/low",
+            "ragg/pdfs/mid",
+            "ragg/pdfs/high",
+            "pdfs/low",
+            "pdfs/mid",
+            "pdfs/high",
+        ]
+        print("\n[DEBUG] PDF folders preflight:", flush=True)
+        for b in bases:
+            info = _list_dir_sample(b)
+            print(f"  - {b}: exists={info.get('exists')} total={info.get('total', 0)} sample={info.get('sample', [])}", flush=True)
+
+        cmd = [sys.executable, "-m", module_name]
+        print("\n[DEBUG] Running subprocess:", cmd, flush=True)
 
         result = subprocess.run(
-            [sys.executable, "-m", module_name],
+            cmd,
             capture_output=True,
             text=True,
             cwd=None,
-            timeout=int(os.getenv("INGEST_TIMEOUT_SEC", "1800"))  # default 30 min
+            timeout=timeout_sec
         )
 
+        print("\n[DEBUG] Subprocess completed.", flush=True)
+        print("Return code:", result.returncode, flush=True)
+        # Print first and last few lines to avoid massive logs
+        def _preview(label, text, head=30, tail=30):
+            lines = (text or "").splitlines()
+            print(f"\n----- {label} (total lines: {len(lines)}) -----", flush=True)
+            for line in lines[:head]:
+                print(line, flush=True)
+            if len(lines) > head + tail:
+                print("... [truncated] ...", flush=True)
+            for line in lines[-tail:]:
+                print(line, flush=True)
+            print("----- END", label, "-----\n", flush=True)
+
+        _preview("STDOUT", result.stdout)
+        _preview("STDERR", result.stderr)
+
         payload = {
             "status": "success" if result.returncode == 0 else "error",
             "returncode": result.returncode,
-            "stdout": result.stdout[-20000:],  # trim if very long
-            "stderr": result.stderr[-20000:],
+            # Return last ~4000 chars so the client can also see logs in Postman
+            "stdout": (result.stdout or "")[-4000:],
+            "stderr": (result.stderr or "")[-4000:],
         }
-        return jsonify(payload), 200 if result.returncode == 0 else 500
+        status_code = 200 if result.returncode == 0 else 500
+        return jsonify(payload), status_code
 
-    except Exception:
+    except subprocess.TimeoutExpired:
+        print("[ERROR] Ingestion timed out.", flush=True)
+        return jsonify({"status": "timeout"}), 504
+
+    except Exception as e:
+        print("[ERROR] Exception during ingestion:", e, flush=True)
+        print(traceback.format_exc(), flush=True)
         return jsonify({
             "status": "error",
             "message": "trigger crashed",
             "traceback": traceback.format_exc()
         }), 500
+
     finally:
-        _ingest_lock.release()
+        try:
+            _ingest_lock.release()
+        except Exception:
+            # Avoid rare "release unlocked lock" noise
+            pass