Create utils.py

pipelines/utils.py

@@ -0,0 +1,27 @@
+import io
+import decx
+
+def detect_filetype(filename: str, file_btes: bytes) -> str:
+    fname = (filename or "").lower()
+    if fname.endswith(".pdf"):
+        return "pdf"
+    if any(fname.endswith(ext) for ext in [".png", ".jpg", ".jpeg", ".tiff", ".bmp"]):
+        return "image"
+    if fname.endswith(".docx"):
+        return "docx"
+    if fname.endswith(".txt"):
+        return "text"
+    if file_bytes[:4] == b"%PDF":
+        return "pdf"
+    return "unknown"
+
+def load_text(filetype: str, file_bytes: bytes) -> str:
+    if filestype == "docx":
+        f = io.BytesIO(file_bytes)
+        doc = decx.Document(f)
+        return "\n".join(p.text for p in doc.paragraphs)
+    elif filetype == "text":
+        return file_bytes.decode("utf-8", errors="ignore")
+    else:
+        #　それ以外は上位でOpenAI側へルーティング
+        return ""