fetch CSV files as stream

Author: JHZhang2736

package.json

@@ -30,12 +30,16 @@
         "@magda/ci-utils": "^1.0.5",
         "@magda/docker-utils": "^6.0.0-alpha.4",
         "@types/chai": "^5.2.2",
+        "@types/fs-extra": "^11.0.4",
         "@types/mocha": "^10.0.10",
+        "@types/nock": "^11.1.0",
         "@types/node": "^20.19.0",
         "@types/yargs": "^17.0.33",
+        "@types/sinon": "^17.0.4",
         "chai": "^5.2.0",
         "husky": "^9.1.7",
         "mocha": "^11.7.1",
+        "nock": "^14.0.7",
         "prettier": "^3.5.3",
         "pretty-quick": "^4.2.2",
         "rimraf": "^3.0.0",
@@ -47,9 +51,10 @@
     "dependencies": {
         "@langchain/core": "^0.3.61",
         "@langchain/textsplitters": "https://gitpkg.vercel.app/JHZhang2736/langchainjs/libs/langchain-textsplitters?textsplitters-build",
-        "@magda/semantic-indexer-sdk": "^6.0.0-alpha.4",
-        "@types/sinon": "^17.0.4",
-        "urijs": "^1.19.11",
+        "@magda/esm-utils": "^1.0.1",
+        "@magda/semantic-indexer-sdk": "6.0.0-alpha.6",
+        "csv-parse": "^6.1.0",
+        "fs-extra": "^11.3.0",
         "yargs": "17.6.2"
     },
     "resolutions": {

src/createEmbeddingText.ts

@@ -6,6 +6,14 @@ import {
 import { openSync, readSync, closeSync, existsSync } from "fs";
 import { basename } from "path";
 import { toYaml } from "./utils/toYaml.js";
+import { parse } from "csv-parse";
+import fse from "fs-extra";
+import path from "path";
+import { Readable } from "stream";
+import type { ReadableStream as WebReadableStream } from "stream/web";
+import { __dirname as getCurDirPath } from "@magda/esm-utils";
+
+const __dirname = getCurDirPath();
 
 export function readFirstBytes(path: string, bytes: number): string {
     const fd = openSync(path, "r");
@@ -58,6 +66,100 @@ export function formatTemporal(tc?: { start?: string; end?: string }): string {
     return "";
 }
 
+const CONNECTION_TIMEOUT = 120;
+
+function toNodeReadable(body: any): NodeJS.ReadableStream {
+    return typeof body?.pipe === "function"
+        ? body
+        : Readable.fromWeb(body as WebReadableStream<Uint8Array>);
+}
+
+let cachedPackageInfo: { name: string; version: string } | null = null;
+
+async function getPackageInfo() {
+    if (!cachedPackageInfo) {
+        cachedPackageInfo = await fse.readJSON(path.resolve(__dirname, "../package.json"), {
+            encoding: "utf-8"
+        });
+    }
+    return cachedPackageInfo;
+}
+
+export async function getColumnsFromCSVStream(downloadURL: string, timeout: number = CONNECTION_TIMEOUT * 1000): Promise<string[]> {
+    const controller = new AbortController();
+    const timeoutId = setTimeout(() => controller.abort(), timeout);
+    const pkg = await getPackageInfo();
+
+    try {
+        const res = await fetch(downloadURL, {
+            redirect: "follow",
+            headers: {
+                "User-Agent": `${pkg?.name}/${pkg?.version}`
+            },
+            signal: controller.signal
+        });
+        
+        if (!res.ok || !res.body) {
+            throw new Error(`HTTP ${res.status} fetching ${downloadURL}`);
+        }
+
+        const parser = parse({
+            bom: true,
+            trim: true,
+            skip_empty_lines: true,
+            relax_column_count: true,
+            delimiter: [",", ";", "\t", "|"],
+            from_line: 1,
+            to_line: 1,
+            columns: false
+        });
+
+        const csvStream: NodeJS.ReadableStream = toNodeReadable(res.body);
+
+        const cols = await new Promise<string[]>((resolve, reject) => {
+            let resolved = false;
+            
+            const handleResolve = (columns: string[]) => {
+                if (!resolved) {
+                    resolved = true;
+                    resolve(columns);
+                }
+            };
+            
+            const handleReject = (error: Error) => {
+                if (!resolved) {
+                    resolved = true;
+                    reject(new Error(`CSV parsing error: ${error.message}`));
+                }
+            };
+            
+            csvStream
+                .pipe(parser)
+                .on("data", (row: string[]) => {
+                    const columns = row.map(col => col.trim()).filter(Boolean);
+                    handleResolve(columns);
+                })
+                .on("error", (error) => {
+                    if (error.name === 'AbortError') {
+                        return;
+                    }
+                    handleReject(error);
+                })
+                .on("end", () => {
+                    handleResolve([]);
+                });
+        });
+        return cols;
+    } catch (error) {
+        if (error instanceof Error && error.name === 'AbortError') {
+            throw new Error('Timeout when fetching CSV');
+        }
+        throw new Error(`Failed to fetch CSV: ${error instanceof Error ? error.message : String(error)}`);
+    } finally {
+        clearTimeout(timeoutId);
+    }
+}
+
 export const createEmbeddingText: CreateEmbeddingText = async ({
     record,
     format,
@@ -104,6 +206,15 @@ export const createEmbeddingText: CreateEmbeddingText = async ({
     if (filePath && existsSync(filePath)) {
         const header = readFirstBytes(filePath, 64 * 1024).split(/\r?\n/)[0] ?? "";
         columns = header.split(",").map(c => c.trim()).filter(Boolean);
+    } else if (url) {
+        try {
+            columns = await getColumnsFromCSVStream(url);
+        } catch (e) {
+            if (e instanceof Error) {
+                console.warn("Failed to get columns from CSV file", e.message);
+            }
+            columns = [];
+        }
     }
 
     const yamlText = toYaml({
@@ -117,9 +228,8 @@ export const createEmbeddingText: CreateEmbeddingText = async ({
         Keywords: keywords || undefined,
         Languages: languages || undefined,
         Description: description || undefined,
-        Columns: columns || undefined
+        "Column names": columns || undefined
     });
 
-
     return { text: yamlText };
 };

src/index.ts

@@ -5,7 +5,7 @@ import semanticIndexer, {
 } from "@magda/semantic-indexer-sdk";
 import { csvSemanticIndexerArgs } from "./csvSemanticIndexerArgs.js";
 import { createEmbeddingText } from "./createEmbeddingText.js";
-import { Chunker } from "./chunker.js";
+import { Chunker } from "./Chunker.js";
 
 const port = csvSemanticIndexerArgs.port;
 const args = commonYargs(port, `http://localhost:${port}`);
@@ -21,7 +21,7 @@ const options: SemanticIndexerOptions = {
     id: csvSemanticIndexerArgs.id,
     itemType: "storageObject",
     formatTypes: ["csv"],
-    autoDownloadFile: true,
+    autoDownloadFile: false,
     createEmbeddingText: createEmbeddingText,
     chunkStrategy: chunkStrategy,
 };

src/test/chunker.spec.ts

@@ -1,5 +1,5 @@
 import { expect } from "chai";
-import { Chunker } from "../chunker.js";
+import { Chunker } from "../Chunker.js";
 
 describe("Chunker", () => {
     it("should chunk yaml text with proper overlapping", async () => {

src/test/createEmbeddingText.spec.ts

@@ -3,7 +3,8 @@ import { parse as parseYaml } from "yaml";
 import * as fs from "fs";
 import { join, basename } from "path";
 import { tmpdir } from "os";
-import { createEmbeddingText, formatTemporal } from "../createEmbeddingText.js";
+import { createEmbeddingText, formatTemporal, getColumnsFromCSVStream } from "../createEmbeddingText.js";
+import nock from "nock";
 
 describe("createEmbeddingText", () => {
     let tempFile: string;
@@ -68,7 +69,7 @@ describe("createEmbeddingText", () => {
             Themes: ["Theme1", "Theme2"],
             Keywords: ["Keyword1", "Keyword2"],
             Languages: ["en", "zh"],
-            Columns: ["colA", "colB", "colC"]
+            "Column names": ["colA", "colB", "colC"]
         };
 
         expect(yamlObj).to.deep.equal(expected);
@@ -114,7 +115,7 @@ describe("createEmbeddingText", () => {
         const expected = {
             Title: "Dataset Title",
             Format: "CSV",
-            Columns: ["colA", "colB", "colC"]
+            "Column names": ["colA", "colB", "colC"]
         };
 
         expect(yamlObj).to.deep.equal(expected);
@@ -161,7 +162,7 @@ describe("createEmbeddingText", () => {
             Format: "CSV",
             "File name": "data.csv",
             Description: "Dataset Desc",
-            Columns: ["colA", "colB", "colC"]
+            "Column names": ["colA", "colB", "colC"]
         };
 
         expect(yamlObj).to.deep.equal(expected);
@@ -241,6 +242,118 @@ describe("createEmbeddingText", () => {
         const yamlObj = parseYaml(text);
 
         expect(yamlObj.Description).to.equal("First line\nSecond line");
+    });
+});
+
+describe("getColumnsFromCSVStream", () => {
+    beforeEach(() => {
+        nock.cleanAll();
+    });
+
+    afterEach(() => {
+        nock.cleanAll();
+    });
+
+    it("should extract column names from comma-separated CSV file", async () => {
+        nock("http://example.com")
+            .get("/test-csv")
+            .reply(200, "column1,column2,column3\nvalue1,value2,value3", {
+                "Content-Type": "text/csv"
+            });
+
+        const columns = await getColumnsFromCSVStream("http://example.com/test-csv");
+        expect(columns).to.deep.equal(["column1", "column2", "column3"]);
+    });
+
+    it("should handle empty CSV file", async () => {
+        nock("http://example.com")
+            .get("/empty-csv")
+            .reply(200, "", {
+                "Content-Type": "text/csv"
+            });
+
+        const columns = await getColumnsFromCSVStream("http://example.com/empty-csv");
+        expect(columns).to.deep.equal([]);
+    });
+
+    it("should throw exception on HTTP 404 error", async () => {
+        nock("http://example.com")
+            .get("/not-found")
+            .reply(404, "Not Found");
+
+        try {
+            await getColumnsFromCSVStream("http://example.com/not-found");
+            expect.fail("should throw exception");
+        } catch (error: any) {
+            expect(error.message).to.include("HTTP 404");
+        }
+    });
+
+    it("should throw exception on HTTP 500 error", async () => {
+        nock("http://example.com")
+            .get("/server-error")
+            .reply(500, "Internal Server Error");
+
+        try {
+            await getColumnsFromCSVStream("http://example.com/server-error");
+            expect.fail("should throw exception");
+        } catch (error: any) {
+            expect(error.message).to.include("HTTP 500");
+        }
+    });
+
+    it("should throw exception on network error", async () => {
+        nock("http://invalid-url-that-does-not-exist.com")
+            .get("/test.csv")
+            .replyWithError("Network Error");
+
+        try {
+            await getColumnsFromCSVStream("http://invalid-url-that-does-not-exist.com/test.csv");
+            expect.fail("should throw exception");
+        } catch (error: any) {
+            expect(error).to.be.instanceOf(Error);
+        }
+    });
+
+    it("should handle CSV file with BOM", async () => {
+        const bom = Buffer.from([0xEF, 0xBB, 0xBF]);
+        const csvContent = "column1,column2,column3\nvalue1,value2,value3";
+        const bomCsv = Buffer.concat([bom, Buffer.from(csvContent)]);
+
+        nock("http://example.com")
+            .get("/bom-csv")
+            .reply(200, bomCsv, {
+                "Content-Type": "text/csv"
+            });
+
+        const columns = await getColumnsFromCSVStream("http://example.com/bom-csv");
+        expect(columns).to.deep.equal(["column1", "column2", "column3"]);
+    });
+
+    it("should handle CSV file with empty columns", async () => {
+        nock("http://example.com")
+            .get("/messy-csv")
+            .reply(200, "col1, ,col3,\n\nval1,val2,val3,val4", {
+                "Content-Type": "text/csv"
+            });
+
+        const columns = await getColumnsFromCSVStream("http://example.com/messy-csv");
+        expect(columns).to.deep.equal(["col1", "col3"]);
+    });
+
+    it("should timeout and abort correctly", async () => {
+        nock("http://example.com")
+            .get("/slow-csv")
+            .delay(3000)
+            .reply(200, "column1,column2,column3\nvalue1,value2,value3", {
+                "Content-Type": "text/csv"
+            });
 
+        try {
+            await getColumnsFromCSVStream("http://example.com/slow-csv", 1000);
+            expect.fail("should throw timeout exception");
+        } catch (error: any) {
+            expect(error).to.be.instanceOf(Error);
+        }
     });
 });

tsconfig.json

@@ -1,8 +1,8 @@
 {
     "compilerOptions": {
         "target": "ES2020",
-        "module": "esnext",
-        "moduleResolution": "node",
+        "module": "Node16",
+        "moduleResolution": "node16",
         "esModuleInterop": true,
         "strict": true,
         "outDir": "dist",