From 4b19638ff836371c4fed467567b434302b9af47e Mon Sep 17 00:00:00 2001
From: 3452078359-xuexue <3452078359@qq.com>
Date: Tue, 1 Sep 2026 23:26:16 -0700
Subject: [PATCH] fix: stream large graph JSON imports

---
 admin-web/src/ProjectLanding.tsx        | 139 +++++++--
 admin-web/src/api.ts                    |  41 +++
 admin-web/src/graphFileUpload.ts        |  49 ++++
 admin-web/tests/graphFileUpload.test.ts |  52 ++++
 app/api/__init__.py                     |   2 +
 app/api/graph_file_import.py            | 184 ++++++++++++
 app/api/projects.py                     |  40 ++-
 app/config.py                           |   3 +
 app/graph_import_files.py               | 311 ++++++++++++++++++++
 app/project_lifecycle.py                | 369 +++++++++++++-----------
 deploy/nginx/travel-kg.conf.example     |  15 +
 docs/GRAPH_FILE_IMPORT.md               |  93 ++++++
 requirements.txt                        |   1 +
 scripts/verify_graph_file_import.py     | 167 +++++++++++
 tests/test_graph_file_import.py         | 315 ++++++++++++++++++++
 tests/test_project_lifecycle.py         |   7 +
 16 files changed, 1595 insertions(+), 193 deletions(-)
 create mode 100644 admin-web/src/graphFileUpload.ts
 create mode 100644 admin-web/tests/graphFileUpload.test.ts
 create mode 100644 app/api/graph_file_import.py
 create mode 100644 app/graph_import_files.py
 create mode 100644 docs/GRAPH_FILE_IMPORT.md
 create mode 100644 scripts/verify_graph_file_import.py
 create mode 100644 tests/test_graph_file_import.py

diff --git a/admin-web/src/ProjectLanding.tsx b/admin-web/src/ProjectLanding.tsx
index f24bb97..7f5711c 100644
--- a/admin-web/src/ProjectLanding.tsx
+++ b/admin-web/src/ProjectLanding.tsx
@@ -10,6 +10,7 @@ import {
   Input,
   List,
   Modal,
+  Progress,
   Select,
   Space,
   Spin,
@@ -42,6 +43,8 @@ import {
   listGraphReleases,
   listProjects,
   provisionGraphProject,
+  provisionGraphProjectFile,
+  graphFileImportOptions,
   setProjectContext,
   updateGraphProject,
   type ProjectDeletionImpact,
@@ -57,6 +60,7 @@ import {
   mergeProjectCatalogItem,
   notifyProjectCatalogChanged,
 } from "./projectCatalog";
+import { JSON_EDITOR_MAX_BYTES, importHttpErrorDetail, jsonEditorFits, useGraphFileTransfer, type ImportProgress } from "./graphFileUpload";
 
 const { Paragraph, Text, Title } = Typography;
 const { TextArea } = Input;
@@ -157,7 +161,7 @@ function projectName(project: ProjectSummary): string {
 }
 
 function apiErrorText(error: unknown, fallback: string): string {
-  const detail = (error as { response?: { data?: { detail?: unknown } } })?.response?.data?.detail;
+  const detail = importHttpErrorDetail(error);
   if (typeof detail === "string" && detail.trim()) return detail;
   if (Array.isArray(detail)) {
     const messages = detail
@@ -192,6 +196,7 @@ function apiErrorText(error: unknown, fallback: string): string {
 }
 
 function parseJsonObject(text: string, label: string): ValidationResult<Record<string, unknown>> {
+  if (!jsonEditorFits(text)) return { errors: [`${label}超过编辑器 2 MiB 上限，请选择 JSON 文件上传`] };
   if (!text.trim()) return { errors: [`请上传或输入${label} JSON`] };
   try {
     const parsed: unknown = JSON.parse(text);
@@ -651,6 +656,10 @@ export default function ProjectLanding() {
   const [schemaErrors, setSchemaErrors] = useState<string[]>([]);
   const [graphText, setGraphText] = useState("");
   const [graphFileName, setGraphFileName] = useState("");
+  const [graphFile, setGraphFile] = useState<File | null>(null);
+  const [loadingFile, setLoadingFile] = useState(false);
+  const [uploadPercent, setUploadPercent] = useState(0);
+  const [importProgress, setImportProgress] = useState<ImportProgress | null>(null);
   const [graphObject, setGraphObject] = useState<GraphImportData | null>(null);
   const [graphErrors, setGraphErrors] = useState<string[]>([]);
   const [provisioning, setProvisioning] = useState(false);
@@ -669,6 +678,7 @@ export default function ProjectLanding() {
   const mapEnabled = Form.useWatch("map_enabled", basicForm);
   const [renameForm] = Form.useForm<{ display_name: string }>();
   const impactRequest = useRef(0);
+  const fileReadRequest = useRef(0);
   const navigate = useNavigate();
 
   const refresh = async (): Promise<ProjectSummary[] | null> => {
@@ -727,6 +737,11 @@ export default function ProjectLanding() {
     setSchemaErrors([]);
     setGraphText("");
     setGraphFileName("");
+    setGraphFile(null);
+    setLoadingFile(false);
+    fileReadRequest.current += 1;
+    setUploadPercent(0);
+    setImportProgress(null);
     setGraphObject(null);
     setGraphErrors([]);
     setProvisionError("");
@@ -780,28 +795,63 @@ export default function ProjectLanding() {
   };
 
   const loadJsonFile = async (file: File, kind: "schema" | "graph") => {
+    const requestId = ++fileReadRequest.current;
+    setLoadingFile(true);
+    if (kind === "schema") {
+      setSchemaText("");
+      setSchemaObject(null);
+      setSchemaErrors([]);
+      setSchemaFileName(file.name);
+    } else {
+      setGraphFile(null);
+      setGraphText("");
+      setGraphObject(null);
+      setGraphErrors([]);
+      setGraphFileName(file.name);
+    }
     try {
+      if (!file.name.toLowerCase().endsWith(".json")) throw new Error("请选择 .json 文件");
+      if (!file.size) throw new Error("文件为空");
+      if (useGraphFileTransfer(file)) {
+        if (kind === "schema") throw new Error("Schema 超过 2 MiB 上限，请确认未误选图谱 data 文件");
+        const { data: options } = await graphFileImportOptions();
+        if (requestId !== fileReadRequest.current) return;
+        if (file.size > options.max_file_bytes) {
+          throw new Error(`超过服务器 ${Math.floor(options.max_file_bytes / 1024 / 1024)} MiB 文件上限`);
+        }
+        setGraphFile(file);
+        setGraphFileName(file.name);
+        setGraphText("");
+        setGraphObject(null);
+        setGraphErrors([]);
+        message.success("已选择文件，将在确认创建后上传并由服务器校验");
+        return;
+      }
       const text = await file.text();
+      if (requestId !== fileReadRequest.current) return;
       const parsed: unknown = JSON.parse(text);
       if (!isRecord(parsed)) throw new Error("JSON 顶层必须是对象");
-      const formatted = JSON.stringify(parsed, null, 2);
       if (kind === "schema") {
-        setSchemaText(formatted);
+        setSchemaText(text);
         setSchemaFileName(file.name);
         setSchemaObject(null);
         setSchemaErrors([]);
       } else {
-        setGraphText(formatted);
+        setGraphFile(null);
+        setGraphText(text);
         setGraphFileName(file.name);
         setGraphObject(null);
         setGraphErrors([]);
       }
       message.success(`${file.name} 已读取，可继续编辑`);
     } catch (error) {
-      const errorText = error instanceof Error ? error.message : "无法读取文件";
-      if (kind === "schema") setSchemaErrors([`本体 Schema JSON 格式错误：${errorText}`]);
-      else setGraphErrors([`图谱数据 JSON 格式错误：${errorText}`]);
-      message.error(`${file.name} 不是有效的 JSON 文件`);
+      if (requestId !== fileReadRequest.current) return;
+      const errorText = apiErrorText(error, error instanceof Error ? error.message : "无法读取文件，请确认后端已更新");
+      if (kind === "schema") setSchemaErrors([errorText]);
+      else setGraphErrors([errorText]);
+      message.error(errorText);
+    } finally {
+      if (requestId === fileReadRequest.current) setLoadingFile(false);
     }
   };
 
@@ -814,6 +864,10 @@ export default function ProjectLanding() {
       return;
     }
     const formatted = JSON.stringify(parsed.value, null, 2);
+    if (!jsonEditorFits(formatted)) {
+      message.warning("格式化后的文本超过编辑器上限，已保留原文");
+      return;
+    }
     if (kind === "schema") {
       setSchemaText(formatted);
       setSchemaErrors([]);
@@ -836,6 +890,10 @@ export default function ProjectLanding() {
       setWizardStep(1);
       return false;
     }
+    if (graphFile) {
+      setGraphErrors([]);
+      return true; // Full graph validation happens server-side before any write.
+    }
     const result = validateGraphText(graphText, schemaObject);
     setGraphErrors(result.errors);
     setGraphObject(result.value || null);
@@ -853,8 +911,8 @@ export default function ProjectLanding() {
       setWizardStep(1);
       return;
     }
-    const graphResult = validateGraphText(graphText, schemaResult.value);
-    if (!graphResult.value) {
+    const graphResult = graphFile ? { value: undefined, errors: [] } : validateGraphText(graphText, schemaResult.value);
+    if (!graphFile && !graphResult.value) {
       setGraphErrors(graphResult.errors);
       setWizardStep(2);
       return;
@@ -862,8 +920,10 @@ export default function ProjectLanding() {
 
     setProvisioning(true);
     setProvisionError("");
+    setUploadPercent(0);
+    setImportProgress(graphFile ? { phase: "上传文件" } : null);
     try {
-      const { data } = await provisionGraphProject({
+      const metadata = {
         project_id: basicValues.project_id,
         display_name: basicValues.display_name,
         spatial_map: spatialMapConfigForRegion(
@@ -871,8 +931,10 @@ export default function ProjectLanding() {
           basicValues.map_region_adcode,
         ),
         schema: schemaResult.value,
-        graph_data: graphResult.value,
-      });
+      };
+      const data = graphFile
+        ? await provisionGraphProjectFile(metadata, graphFile, setImportProgress, setUploadPercent)
+        : (await provisionGraphProject({ ...metadata, graph_data: graphResult.value! })).data;
       const createdProject = data.project;
       const projectId = String(createdProject?.project_id || basicValues.project_id).trim();
       const tenantId = String(
@@ -896,7 +958,11 @@ export default function ProjectLanding() {
         message.warning("项目已创建，但接口未返回可用图谱版本；请从项目列表重新进入");
       }
     } catch (error) {
-      const detail = apiErrorText(error, "项目创建失败，请检查输入后重试");
+      const status = (error as { response?: { status?: number } })?.response?.status;
+      const fallback = status === 413
+        ? "服务器或反向代理拒绝了文件大小，请检查上传限制"
+        : "连接中断或创建失败，请先刷新项目列表确认结果，再决定是否重试";
+      const detail = apiErrorText(error, fallback);
       setProvisionError(detail);
       message.error(detail);
     } finally {
@@ -1188,8 +1254,13 @@ export default function ProjectLanding() {
               </div>
               <TextArea
                 aria-labelledby="project-schema-json-label"
+                maxLength={JSON_EDITOR_MAX_BYTES}
                 value={schemaText}
                 onChange={(event) => {
+                  if (!jsonEditorFits(event.target.value)) {
+                    setSchemaErrors(["Schema 超过编辑器 2 MiB 上限"]);
+                    return;
+                  }
                   setSchemaText(event.target.value);
                   setSchemaObject(null);
                   setSchemaErrors([]);
@@ -1225,17 +1296,30 @@ export default function ProjectLanding() {
             >
               <p className="ant-upload-drag-icon"><InboxOutlined /></p>
               <p className="ant-upload-text">上传节点与关系 JSON</p>
-              <p className="ant-upload-hint">{graphFileName || "点击或拖拽文件；读取后仍可编辑"}</p>
+              <p className="ant-upload-hint">{graphFileName || "点击或拖拽文件；大文件直接上传，不放入编辑框"}</p>
             </Upload.Dragger>
             <div className="project-json-editor-shell">
               <div className="project-json-editor-toolbar">
                 <span id="project-graph-json-label">Graph Data JSON</span>
-                <Button size="small" onClick={() => formatJson("graph")}>格式化</Button>
+                <Button size="small" disabled={Boolean(graphFile) || loadingFile} onClick={() => formatJson("graph")}>格式化</Button>
               </div>
-              <TextArea
+              {graphFile ? (
+                <Space direction="vertical" size={16} style={{ padding: 24 }} role="status">
+                  <InboxOutlined style={{ fontSize: 36, color: "#1677ff" }} />
+                  <Text strong>{graphFile.name}</Text>
+                  <Text>{(graphFile.size / 1024 / 1024).toFixed(1)} MiB</Text>
+                  <Text type="secondary">已选择原文件。确认创建后上传，由服务器校验节点、关系和地图数据。</Text>
+                  <Button onClick={() => { setGraphFile(null); setGraphFileName(""); }}>移除文件</Button>
+                </Space>
+              ) : <TextArea
                 aria-labelledby="project-graph-json-label"
+                maxLength={JSON_EDITOR_MAX_BYTES}
                 value={graphText}
                 onChange={(event) => {
+                  if (!jsonEditorFits(event.target.value)) {
+                    setGraphErrors(["文本超过 2 MiB，请选择 JSON 文件上传"]);
+                    return;
+                  }
                   setGraphText(event.target.value);
                   setGraphObject(null);
                   setGraphErrors([]);
@@ -1243,7 +1327,7 @@ export default function ProjectLanding() {
                 className="project-json-editor"
                 placeholder={'{\n  "nodes": [\n    { "id": "node-1", "type": "Place", "properties": { "name": "示例地点" } }\n  ],\n  "relations": []\n}'}
                 spellCheck={false}
-              />
+              />}
             </div>
           </div>
           {validationAlert("图谱数据校验未通过", graphErrors)}
@@ -1262,6 +1346,9 @@ export default function ProjectLanding() {
         <Descriptions bordered size="small" column={2} className="project-confirm-descriptions">
           <Descriptions.Item label="项目名称">{basicValues?.display_name || "-"}</Descriptions.Item>
           <Descriptions.Item label="项目英文标识">{basicValues?.project_id || "-"}</Descriptions.Item>
+          {graphFile && <Descriptions.Item label="图谱数据文件" span={2}>
+            {graphFile.name} · {(graphFile.size / 1024 / 1024).toFixed(1)} MiB
+          </Descriptions.Item>}
           <Descriptions.Item label="知识地图" span={2}>
             {basicValues?.map_enabled ? "显示" : "不显示"}
           </Descriptions.Item>
@@ -1275,8 +1362,18 @@ export default function ProjectLanding() {
           type="warning"
           showIcon
           message="项目创建和数据导入将由系统统一完成"
-          description="任一步失败都会返回明确错误；窗口会保留当前输入，便于修正后重试。"
+          description={graphFile
+            ? "先上传并校验原文件，通过后分批导入。处理期间请保持窗口打开。"
+            : "任一步失败都会返回明确错误；窗口会保留当前输入，便于修正后重试。"}
         />
+        {importProgress && <div role="status" aria-live="polite" style={{ width: "100%" }}>
+          <Text>{importProgress.phase}</Text>
+          <Progress percent={uploadPercent} status={provisionError ? "exception" : provisioning ? "active" : "normal"}
+            format={(percent) => `上传 ${percent}%`} />
+          {importProgress.nodes !== undefined && <Text type="secondary">
+            当前阶段：{importProgress.nodes.toLocaleString()} 个节点，{(importProgress.relations || 0).toLocaleString()} 条关系
+          </Text>}
+        </div>}
         {provisionError && (
           <Alert className="project-provision-error" type="error" showIcon message="创建失败" description={provisionError} />
         )}
@@ -1393,7 +1490,7 @@ export default function ProjectLanding() {
             <Button onClick={closeCreateWizard} disabled={checkingBasics || provisioning}>取消</Button>
             {wizardStep > 0 && (
               <Button
-                disabled={provisioning}
+                disabled={provisioning || loadingFile}
                 onClick={() => {
                   setProvisionError("");
                   setWizardStep((current) => Math.max(0, current - 1));
@@ -1404,7 +1501,7 @@ export default function ProjectLanding() {
             )}
             <Button
               type="primary"
-              loading={checkingBasics || provisioning}
+              loading={checkingBasics || provisioning || loadingFile}
               onClick={() => void nextWizardStep()}
             >
               {wizardStep === 3 ? "确认创建" : "下一步"}
diff --git a/admin-web/src/api.ts b/admin-web/src/api.ts
index d8afabd..97e6b32 100644
--- a/admin-web/src/api.ts
+++ b/admin-web/src/api.ts
@@ -1,4 +1,5 @@
 import axios from "axios";
+import { importEventReader, type ImportProgress } from "./graphFileUpload";
 
 const api = axios.create({ baseURL: "/v1/admin" });
 
@@ -226,6 +227,46 @@ export const createGraphRelease = (projectId: string, data: unknown) =>
   api.post(`/projects/${encodeURIComponent(projectId)}/graph-releases`, data);
 export const provisionGraphProject = (data: GraphProjectProvisionRequest) =>
   api.post<GraphProjectProvisionResponse>("/projects/provision", data);
+export const graphFileImportOptions = () =>
+  api.get<{ max_file_bytes: number; editor_max_bytes: number }>("/projects/import-options");
+
+export async function provisionGraphProjectFile(
+  metadata: Omit<GraphProjectProvisionRequest, "graph_data">,
+  file: File,
+  onProgress: (progress: ImportProgress) => void,
+  onUpload: (percent: number) => void,
+): Promise<GraphProjectProvisionResponse> {
+  const form = new FormData();
+  // Serialize only the small project/Schema header. The data File is opaque.
+  form.append("metadata", JSON.stringify(metadata));
+  form.append("file", file, file.name);
+  let result: GraphProjectProvisionResponse | undefined;
+  let failure: unknown;
+  let parseFailure = false;
+  const read = importEventReader((item) => {
+    if (item.event === "progress") onProgress(item.data as ImportProgress);
+    if (item.event === "result") result = item.data as GraphProjectProvisionResponse;
+    if (item.event === "error") failure = item.data;
+  });
+  const response = await api.post<string>("/projects/provision-file", form, {
+    responseType: "text",
+    timeout: 0,
+    onUploadProgress: (event) => onUpload(event.total ? Math.floor(event.loaded * 100 / event.total) : 0),
+    onDownloadProgress: (event) => {
+      const text = event.event?.target?.responseText;
+      if (typeof text === "string") {
+        try { read(text); } catch { parseFailure = true; }
+      }
+    },
+  });
+  try { read(response.data); } catch { parseFailure = true; }
+  if (failure || parseFailure || !result) {
+    throw { response: { data: { detail: failure || {
+      message: "导入响应不完整，请刷新项目列表确认结果，勿重复提交",
+    } } } };
+  }
+  return result;
+}
 export const updateGraphProject = (projectId: string, data: { display_name: string }) =>
   api.patch<ProjectSummary>(`/projects/${encodeURIComponent(projectId)}`, data);
 export const getGraphProjectDeletionImpact = (projectId: string) =>
diff --git a/admin-web/src/graphFileUpload.ts b/admin-web/src/graphFileUpload.ts
new file mode 100644
index 0000000..66caf6c
--- /dev/null
+++ b/admin-web/src/graphFileUpload.ts
@@ -0,0 +1,49 @@
+/** Large graph JSON stays a File/Blob: never parse/stringify it on the UI thread. */
+export const JSON_EDITOR_MAX_BYTES = 2 * 1024 * 1024;
+
+export function jsonEditorFits(text: string): boolean {
+  // The fast character check avoids another huge allocation on pasted input.
+  return text.length <= JSON_EDITOR_MAX_BYTES
+    && new TextEncoder().encode(text).byteLength <= JSON_EDITOR_MAX_BYTES;
+}
+
+export function useGraphFileTransfer(file: Pick<Blob, "size">): boolean {
+  return file.size > JSON_EDITOR_MAX_BYTES;
+}
+
+/** Streaming uploads use responseType=text, including small HTTP errors. */
+export function importHttpErrorDetail(error: unknown): unknown {
+  let data = (error as { response?: { data?: unknown } })?.response?.data;
+  if (typeof data === "string") {
+    if (data.length > 65536) return undefined;
+    try { data = JSON.parse(data); } catch { return undefined; }
+  }
+  return data !== null && typeof data === "object" && "detail" in data
+    ? (data as { detail: unknown }).detail : undefined;
+}
+
+export type ImportProgress = {
+  phase: string;
+  nodes?: number;
+  relations?: number;
+  counts?: Record<string, number>;
+};
+
+export type ImportEvent = {
+  event: "progress" | "result" | "error";
+  data: Record<string, unknown>;
+};
+
+/** Progress responses are small NDJSON, not the uploaded graph body. */
+export function importEventReader(onEvent: (event: ImportEvent) => void) {
+  let offset = 0;
+  return (responseText: string) => {
+    let end = responseText.indexOf("\n", offset);
+    while (end !== -1) {
+      const line = responseText.slice(offset, end).trim();
+      offset = end + 1;
+      if (line) onEvent(JSON.parse(line) as ImportEvent);
+      end = responseText.indexOf("\n", offset);
+    }
+  };
+}
diff --git a/admin-web/tests/graphFileUpload.test.ts b/admin-web/tests/graphFileUpload.test.ts
new file mode 100644
index 0000000..b508581
--- /dev/null
+++ b/admin-web/tests/graphFileUpload.test.ts
@@ -0,0 +1,52 @@
+import assert from "node:assert/strict";
+import { readFileSync } from "node:fs";
+import test from "node:test";
+import { JSON_EDITOR_MAX_BYTES, importEventReader, importHttpErrorDetail, jsonEditorFits, useGraphFileTransfer } from "../src/graphFileUpload.ts";
+
+test("280 MB and larger files select file transfer without reading content", () => {
+  for (const size of [287527160, 1024 ** 3, 10 * 1024 ** 3]) {
+    const file = { size, text() { throw new Error("large file must not be read"); } };
+    assert.equal(useGraphFileTransfer(file), true);
+  }
+  assert.equal(useGraphFileTransfer({ size: JSON_EDITOR_MAX_BYTES }), false);
+});
+
+test("editor limits account for UTF-8 bytes, not only characters", () => {
+  assert.equal(jsonEditorFits("a".repeat(JSON_EDITOR_MAX_BYTES)), true);
+  assert.equal(jsonEditorFits("a".repeat(JSON_EDITOR_MAX_BYTES + 1)), false);
+  assert.equal(jsonEditorFits("中".repeat(800000)), false);
+});
+
+test("HTTP errors are readable for both JSON and streaming text responses", () => {
+  const detail = { message: "磁盘空间不足", errors: ["未接收文件"] };
+  assert.deepEqual(importHttpErrorDetail({ response: { data: { detail } } }), detail);
+  assert.deepEqual(importHttpErrorDetail({ response: { data: JSON.stringify({ detail }) } }), detail);
+  assert.equal(importHttpErrorDetail({ response: { data: '<html>proxy error</html>' } }), undefined);
+  assert.equal(importHttpErrorDetail({ response: { data: 'x'.repeat(65537) } }), undefined);
+  assert.equal(importHttpErrorDetail(null), undefined);
+});
+
+test("partial progress lines parse once and result is not confused with progress", () => {
+  const events: string[] = [];
+  const read = importEventReader((item) => events.push(item.event));
+  const progress = JSON.stringify({ event: "progress", data: { phase: "校验", nodes: 1 } }) + "\n";
+  const result = JSON.stringify({ event: "result", data: { project: { project_id: "test" } } }) + "\n";
+  read(progress.slice(0, 12));
+  assert.deepEqual(events, []);
+  read(progress);
+  read(progress + result.slice(0, 6));
+  read(progress + result);
+  read(progress + result);
+  assert.deepEqual(events, ["progress", "result"]);
+});
+
+test("wizard bypasses full text parsing for large files and disables formatting", () => {
+  const source = readFileSync(new URL("../src/ProjectLanding.tsx", import.meta.url), "utf8");
+  const start = source.indexOf("const loadJsonFile =");
+  const load = source.slice(start, source.indexOf("const formatJson", start));
+  assert.ok(load.indexOf("if (useGraphFileTransfer(file))") < load.indexOf("await file.text()"));
+  assert.ok(load.indexOf("setGraphFile(file)") < load.indexOf("await file.text()"));
+  assert.doesNotMatch(load, /JSON\.stringify/);
+  assert.match(source, /disabled=\{Boolean\(graphFile\) \|\| loadingFile\}/);
+  assert.match(source, /provisionGraphProjectFile\(metadata, graphFile/);
+});
diff --git a/app/api/__init__.py b/app/api/__init__.py
index 873f3f4..1f31433 100644
--- a/app/api/__init__.py
+++ b/app/api/__init__.py
@@ -20,8 +20,10 @@ api_router.include_router(auth_router, tags=["auth"])
 
 # Projects & Graph Releases
 from app.api.projects import router as projects_router  # noqa: E402
+from app.api.graph_file_import import router as graph_file_import_router  # noqa: E402
 
 api_router.include_router(projects_router, tags=["projects"])
+api_router.include_router(graph_file_import_router, tags=["graph-imports"])
 
 # Plaza
 api_router.include_router(plaza_router, tags=["plaza"])
diff --git a/app/api/graph_file_import.py b/app/api/graph_file_import.py
new file mode 100644
index 0000000..32ebec2
--- /dev/null
+++ b/app/api/graph_file_import.py
@@ -0,0 +1,184 @@
+"""Request-scoped file import: spool, validate, batch-write, report progress.
+
+No daemon, persisted job queue or startup changes. The normal project
+transaction/compensation boundary publishes only a completely imported graph.
+"""
+from __future__ import annotations
+
+import asyncio
+from functools import partial
+import json
+import logging
+from pathlib import Path
+import shutil
+import tempfile
+
+from fastapi import APIRouter, HTTPException, Request
+from fastapi.responses import StreamingResponse
+from starlette.datastructures import FormData, UploadFile
+from starlette.formparsers import MultiPartException, MultiPartParser
+
+from app.auth import AdminUser
+from app.config import settings
+from app.graph_import_files import import_staged_graph, validate_graph_file
+from app.project_lifecycle import ProjectValidationError, normalize_provision_header
+
+router = APIRouter()
+logger = logging.getLogger(__name__)
+SCHEMA_BYTES = 2 * 1024 * 1024
+MIN_FREE_BYTES = 512 * 1024 * 1024
+# Bound concurrent upload/index allocations within each existing API process.
+_import_lock = asyncio.Lock()
+
+
+def event(kind: str, data: dict) -> str:
+    return json.dumps({"event": kind, "data": data}, ensure_ascii=False, default=str) + "\n"
+
+
+async def read_import_form(request: Request) -> FormData:
+    limit = settings.graph_import_max_bytes + SCHEMA_BYTES + 65536
+    length = request.headers.get("content-length", "")
+    if length.isdigit():
+        if int(length) > limit:
+            raise HTTPException(413, "图谱文件超过服务器文件大小限制")
+        free = shutil.disk_usage(tempfile.gettempdir()).free
+        needed = int(length) * 3 + MIN_FREE_BYTES
+        if free < needed:
+            raise HTTPException(507, f"服务器临时磁盘空间不足（可用 {free // 1048576} MiB，预计至少需要 {needed // 1048576} MiB），未接收文件")
+    if not request.headers.get("content-type", "").lower().startswith("multipart/form-data"):
+        raise HTTPException(415, "请使用文件上传，而不是把整个图谱放入 JSON 请求体")
+    status = 400
+
+    async def bounded_stream():
+        nonlocal status
+        received = 0
+        async for block in request.stream():
+            received += len(block)
+            if received > limit:
+                status = 413
+                raise MultiPartException("图谱文件超过服务器文件大小限制")
+            if shutil.disk_usage(tempfile.gettempdir()).free < len(block) + MIN_FREE_BYTES:
+                status = 507
+                raise MultiPartException("服务器临时磁盘空间不足，上传已停止")
+            yield block
+
+    parser = MultiPartParser(request.headers, bounded_stream(), max_files=1, max_fields=1,
+                             max_part_size=SCHEMA_BYTES)
+    try:
+        return await parser.parse()
+    except BaseException as exc:
+        # Starlette closes partial files on MultiPartException, but a network
+        # disconnect/cancellation can also interrupt parsing. Close those too.
+        for stream in parser._files_to_close_on_error:
+            stream.close()
+        if isinstance(exc, MultiPartException):
+            raise HTTPException(status, exc.message) from exc
+        raise
+
+
+@router.get("/projects/import-options")
+async def import_options(_user: AdminUser):
+    return {"max_file_bytes": settings.graph_import_max_bytes, "editor_max_bytes": SCHEMA_BYTES}
+
+
+@router.post("/projects/provision-file")
+async def provision_file(request: Request, user: AdminUser):
+    # Admin authentication finishes before reading the potentially large body.
+    if _import_lock.locked():
+        raise HTTPException(429, "已有文件正在上传或导入，请等待完成后再试")
+    await _import_lock.acquire()
+    form = None
+    handed_to_response = False
+    try:
+        form = await read_import_form(request)
+        source, metadata = form.get("file"), form.get("metadata")
+        if not isinstance(source, UploadFile) or not str(source.filename or "").lower().endswith(".json"):
+            raise HTTPException(422, "请选择 .json 图谱数据文件")
+        if not source.size or source.size > settings.graph_import_max_bytes:
+            raise HTTPException(413, "图谱文件为空或超过服务器文件大小限制")
+        if not isinstance(metadata, str):
+            raise HTTPException(422, "缺少项目与 Schema 信息")
+        try:
+            body = json.loads(metadata)
+            if not isinstance(body, dict):
+                raise ValueError("项目信息必须是对象")
+            payload = await asyncio.to_thread(normalize_provision_header, body)
+        except (ValueError, ProjectValidationError) as exc:
+            raise HTTPException(422, {"message": "项目或 Schema 校验失败",
+                                      "errors": getattr(exc, "errors", [str(exc)])[:100]}) from exc
+
+        async def response_stream():
+            from app.api.projects import _run_async_cleanup_to_completion, provision_validated_project
+
+            queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=1)
+            loop = asyncio.get_running_loop()
+            latest = {"phase": "服务器校验 JSON", "nodes": 0, "relations": 0}
+
+            def progress(phase: str, nodes: int, relations: int):
+                free = shutil.disk_usage(tempfile.gettempdir()).free
+                if free < MIN_FREE_BYTES:
+                    logger.warning("graph import disk reserve reached: project=%s phase=%s free_bytes=%s",
+                                   payload["project_id"], phase, free)
+                    raise RuntimeError(f"服务器临时磁盘空间不足（{phase}阶段可用 {free // 1048576} MiB，须保留 512 MiB），未发布图谱")
+
+                def publish():
+                    if queue.full():
+                        queue.get_nowait()
+                    queue.put_nowait({"phase": phase, "nodes": nodes, "relations": relations})
+                loop.call_soon_threadsafe(publish)
+
+            async def execute():
+                # A private, request-owned index validates references before
+                # any graph write. No uploaded path can select server files.
+                with tempfile.TemporaryDirectory(prefix="kg-graph-import-") as directory:
+                    from app.api.projects import _run_blocking_to_completion
+                    prepared = await _run_blocking_to_completion(
+                        validate_graph_file, source.file, Path(directory) / "index.sqlite", payload, progress,
+                    )
+                    return await provision_validated_project(
+                        prepared, user, graph_importer=partial(import_staged_graph, progress=progress),
+                    )
+
+            operation = asyncio.create_task(execute())
+            try:
+                yield event("progress", latest)
+                while not operation.done():
+                    try:
+                        latest = await asyncio.wait_for(queue.get(), 1)
+                    except TimeoutError:
+                        pass
+                    yield event("progress", latest)
+                result = await operation
+                logger.info("graph file import completed: project=%s counts=%s", payload["project_id"], result.get("counts"))
+                yield event("result", result)
+            except Exception as exc:
+                # Do not log uploaded records, credentials or the JSON body.
+                logger.warning("graph file import failed: project=%s phase=%s error_type=%s",
+                               payload["project_id"], latest["phase"], type(exc).__name__)
+                if isinstance(exc, HTTPException):
+                    detail = exc.detail if isinstance(exc.detail, dict) else {"message": str(exc.detail)}
+                else:
+                    detail = {"message": "图谱导入失败", "errors": getattr(exc, "errors", [str(exc)])[:100]}
+                yield event("error", detail)
+            finally:
+                # Do not delete a spool while a DB worker still reads it. An
+                # interrupted browser must check the project list before retry.
+                async def cleanup():
+                    await asyncio.gather(operation, return_exceptions=True)
+                    try:
+                        await form.close()
+                    finally:
+                        _import_lock.release()
+                await _run_async_cleanup_to_completion(cleanup())
+
+        response = StreamingResponse(response_stream(), media_type="application/x-ndjson",
+                                     headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"})
+        handed_to_response = True
+        return response
+    finally:
+        if not handed_to_response:
+            try:
+                if form is not None:
+                    await form.close()
+            finally:
+                _import_lock.release()
diff --git a/app/api/projects.py b/app/api/projects.py
index 505dba9..1f343e2 100644
--- a/app/api/projects.py
+++ b/app/api/projects.py
@@ -4,7 +4,7 @@ import json
 from pathlib import Path
 from typing import Any
 
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, Request
 from psycopg.types.json import Jsonb
 
 from app.auth import CurrentUser
@@ -895,16 +895,46 @@ async def list_projects(_user: CurrentUser):
             return await cur.fetchall()
 
 
-@router.post("/projects/provision")
-async def provision_project(body: dict, _user: CurrentUser):
+@router.post("/projects/provision", openapi_extra={
+    "requestBody": {"required": True, "content": {"application/json": {"schema": {"type": "object"}}}},
+})
+async def provision_project(body: Request, _user: CurrentUser):
     """Atomically provision project metadata, ontology, release, and a new graph."""
 
     _require_project_admin(_user)
+    if isinstance(body, Request):
+        # Check before JSON decoding. Old clients must not exhaust memory by
+        # submitting a whole large graph to the inline compatibility endpoint.
+        limit = 8 * 1024 * 1024
+        length = body.headers.get("content-length", "")
+        detail = "内联 JSON 超过 8 MiB，请在创建项目页面选择文件上传"
+        if length.isdigit() and int(length) > limit:
+            raise HTTPException(413, detail)
+        raw = bytearray()
+        async for block in body.stream():
+            if len(raw) + len(block) > limit:
+                raise HTTPException(413, detail)
+            raw.extend(block)
+        try:
+            body = await asyncio.to_thread(json.loads, raw)
+        except (ValueError, UnicodeDecodeError) as exc:
+            raise HTTPException(422, "项目 JSON 格式无效") from exc
     try:
-        payload = normalize_provision_payload(body)
+        payload = await asyncio.to_thread(normalize_provision_payload, body)
     except ProjectValidationError as exc:
         raise HTTPException(422, {"message": "项目数据校验失败", "errors": exc.errors}) from exc
 
+    return await provision_validated_project(payload, _user)
+
+
+async def provision_validated_project(payload: dict, _user: dict, *,
+                                      graph_importer: Any = None):
+    """Shared commit/compensation boundary for inline and validated file imports.
+
+    This is not a public route: only trusted server-side validation may supply
+    a normalized payload or a staged-file importer.
+    """
+    _require_project_admin(_user)
     try:
         existing_graphs = await asyncio.to_thread(list_falkor_graphs)
     except Exception as exc:
@@ -1146,7 +1176,7 @@ async def provision_project(body: dict, _user: CurrentUser):
 
                 try:
                     graph_counts = await _run_blocking_to_completion(
-                        import_falkor_graph,
+                        graph_importer or import_falkor_graph,
                         payload["graph_name"],
                         payload["graph_data"],
                     )
diff --git a/app/config.py b/app/config.py
index 134ae93..2650e23 100644
--- a/app/config.py
+++ b/app/config.py
@@ -75,6 +75,9 @@ class Settings(BaseSettings):
     falkordb_graph: str = "guiyang"
     falkordb_password: str = ""
 
+    # Request-scoped graph file import (spooled in the OS temporary directory).
+    graph_import_max_bytes: int = 1024 * 1024 * 1024
+
     # Auth
     auth_secret: str = "change-me-at-least-32-chars-long-secret"
     auth_algorithm: str = "HS256"
diff --git a/app/graph_import_files.py b/app/graph_import_files.py
new file mode 100644
index 0000000..3a465c2
--- /dev/null
+++ b/app/graph_import_files.py
@@ -0,0 +1,311 @@
+"""Bounded-memory validation and import of large graph-data JSON files.
+
+The browser never needs to parse these files. SQLite is a temporary index,
+not another graph database: it validates cross-record references and supplies
+bounded UNWIND batches. Only the completed FalkorDB graph is published.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+import sqlite3
+from collections.abc import Mapping
+from contextlib import nullcontext
+from functools import lru_cache
+from pathlib import Path
+from typing import BinaryIO, Callable
+
+import ijson
+from ijson.common import ObjectBuilder
+
+from app.project_lifecycle import (
+    GraphImportError, ProjectValidationError, SpatialMapValidation,
+    _falkor_client, _graph_safe_properties, _node_labels,
+    normalize_graph_node, normalize_graph_relation, normalize_provision_header,
+)
+
+Progress = Callable[[str, int, int], None]
+MAX_STRING_BYTES = 4 * 1024 * 1024
+MAX_RECORD_CHARS = 8 * 1024 * 1024
+
+
+class CheckedReader:
+    """Reject oversized tokens/deep nesting before the parser buffers them."""
+
+    def __init__(self, stream):
+        self.stream = stream
+        self.sha256 = hashlib.sha256()
+        self.count = 0
+        self.in_string = self.escaped = False
+        self.token_size = self.depth = 0
+
+    def read(self, size=-1):
+        block = self.stream.read(65536 if size < 0 else min(size, 65536))
+        self.sha256.update(block)
+        self.count += len(block)
+        for char in block:
+            if self.in_string:
+                self.token_size += 1
+                if self.token_size > MAX_STRING_BYTES:
+                    raise ValueError("单个 JSON 字符串超过 4 MiB，请将大正文/附件拆分为记录或外部资源")
+                if self.escaped:
+                    self.escaped = False
+                elif char == 92:
+                    self.escaped = True
+                elif char == 34:
+                    self.in_string = False
+                    self.token_size = 0
+            elif char == 34:
+                self.in_string = True
+                self.token_size = 0
+            elif char in (123, 91):
+                self.depth += 1
+                self.token_size = 0
+                if self.depth > 64:
+                    raise ValueError("JSON 嵌套不能超过 64 层")
+            elif char in (125, 93):
+                self.depth -= 1
+                self.token_size = 0
+            elif char in (9, 10, 13, 32, 44, 58):
+                self.token_size = 0
+            else:
+                self.token_size += 1
+                if self.token_size > 1024:
+                    raise ValueError("JSON 数值或标记过长")
+        return block
+
+
+def graph_records(reader):
+    """One pass, strict root/array shape and duplicate-key detection."""
+    builder = None
+    record_kind = ""
+    record_depth = 0
+    record_chars = 0
+    maps: list[set[str] | None] = []
+    root_fields: set[str] = set()
+    first = True
+    for prefix, event, value in ijson.parse(reader, use_float=True):
+        if first:
+            first = False
+            if (prefix, event) != ("", "start_map"):
+                raise ValueError("图谱数据 JSON 顶层必须是对象")
+        if event == "map_key":
+            keys = maps[-1]
+            if value in keys:
+                raise ValueError(f"JSON 对象中存在重复字段：{str(value)[:120]}")
+            keys.add(value)
+            if len(keys) > 100000:
+                raise ValueError("单个 JSON 对象字段过多")
+            if prefix == "":
+                root_fields.add(value)
+        elif event in ("start_map", "start_array"):
+            maps.append(set() if event == "start_map" else None)
+        elif event in ("end_map", "end_array"):
+            maps.pop()
+        if prefix in ("nodes", "relations") and event not in ("start_array", "end_array", "map_key"):
+            raise ValueError(f"graph_data.{prefix} 必须是数组")
+        if builder is None and prefix in ("nodes.item", "relations.item"):
+            if event != "start_map":
+                raise ValueError(f"{prefix} 必须是 JSON 对象")
+            builder, record_kind, record_depth, record_chars = ObjectBuilder(), prefix.split(".")[0], 0, 0
+        if builder is not None:
+            record_chars += len(value) if isinstance(value, str) else 8
+            if record_chars > MAX_RECORD_CHARS:
+                raise ValueError("单条节点或关系超过 8 MiB，请拆分大附件")
+            builder.event(event, value)
+            if event in ("start_map", "start_array"):
+                record_depth += 1
+            elif event in ("end_map", "end_array"):
+                record_depth -= 1
+            if record_depth == 0:
+                yield record_kind, builder.value
+                builder = None
+    if "nodes" not in root_fields:
+        raise ValueError("图谱数据缺少 nodes 数组；请上传 data JSON，而不是 Schema 或 bundle 文件")
+
+
+class NodeTypes(Mapping):
+    """A bounded LRU over SQLite, rather than an all-nodes Python dictionary."""
+
+    def __init__(self, connection):
+        self.connection = connection
+        self.lookup = lru_cache(maxsize=4096)(self._lookup)
+
+    def _lookup(self, key):
+        row = self.connection.execute("SELECT type FROM nodes WHERE id=?", (key,)).fetchone()
+        if row is None:
+            raise KeyError(key)
+        return row[0]
+
+    def __getitem__(self, key):
+        return self.lookup(key)
+
+    def __iter__(self):
+        return (row[0] for row in self.connection.execute("SELECT id FROM nodes"))
+
+    def __len__(self):
+        return self.connection.execute("SELECT count(*) FROM nodes").fetchone()[0]
+
+
+def _dumps(value):
+    return json.dumps(value, ensure_ascii=False, separators=(",", ":"), allow_nan=False)
+
+
+def validate_graph_file(source: Path | BinaryIO, staging: Path, body: dict,
+                        progress: Progress = lambda *_: None) -> dict:
+    """Fully validate before any PostgreSQL project/FalkorDB write."""
+    payload = normalize_provision_header(body)
+    if staging.exists():
+        raise FileExistsError("校验暂存文件已存在")
+    connection = sqlite3.connect(staging)
+    node_types = NodeTypes(connection)
+    spatial = SpatialMapValidation(payload["spatial_map"])
+    counts = {"nodes": 0, "relations": 0}
+    try:
+        connection.execute("PRAGMA cache_size=-4096")
+        connection.execute("PRAGMA temp_store=FILE")
+        connection.executescript("""
+            CREATE TABLE nodes(id TEXT PRIMARY KEY, type TEXT NOT NULL, labels TEXT NOT NULL, payload TEXT NOT NULL);
+            CREATE TABLE relations(seq INTEGER PRIMARY KEY, type TEXT, source_type TEXT, target_type TEXT, payload TEXT NOT NULL);
+        """)
+        # Multipart uploads are already spooled to disk; reuse their handle
+        # instead of making another full-size string or file copy.
+        with (source.open("rb") if isinstance(source, Path) else nullcontext(source)) as stream:
+            stream.seek(0)
+            reader = CheckedReader(stream)
+            for kind, raw in graph_records(reader):
+                errors: list[str] = []
+                if kind == "nodes":
+                    node = normalize_graph_node(raw, counts[kind], payload["schema"]["entity_types"], node_types, errors)
+                    if node is not None:
+                        spatial.add(node, errors)
+                    if errors:
+                        raise ProjectValidationError(errors[:100])
+                    connection.execute(
+                        "INSERT INTO nodes VALUES(?,?,?,?)",
+                        (node["id"], node["type"], _dumps(_node_labels(node)), _dumps(node)),
+                    )
+                else:
+                    # Relations may appear before nodes in valid JSON.
+                    connection.execute("INSERT INTO relations(seq,payload) VALUES(?,?)", (counts[kind], _dumps(raw)))
+                counts[kind] += 1
+                if (counts["nodes"] + counts["relations"]) % 1000 == 0:
+                    connection.commit()
+                    progress("校验节点" if kind == "nodes" else "读取关系", counts["nodes"], counts["relations"])
+        if not counts["nodes"]:
+            raise ProjectValidationError(["graph_data.nodes 至少需要一个节点，FalkorDB 不支持持久化空图"])
+        errors = []
+        spatial.finish(errors)
+        if errors:
+            raise ProjectValidationError(errors)
+        connection.commit()
+        for index, text in connection.execute("SELECT seq,payload FROM relations ORDER BY seq"):
+            errors = []
+            relation = normalize_graph_relation(json.loads(text), index, payload["schema"]["relation_types"], node_types, errors)
+            if errors:
+                raise ProjectValidationError(errors[:100])
+            connection.execute(
+                "UPDATE relations SET type=?, source_type=?, target_type=?, payload=? WHERE seq=?",
+                (relation["type"], node_types[relation["source"]], node_types[relation["target"]], _dumps(relation), index),
+            )
+            if index % 1000 == 0:
+                connection.commit()
+                progress("校验关系", counts["nodes"], index + 1)
+        connection.executescript("""
+            CREATE INDEX nodes_group ON nodes(labels);
+            CREATE INDEX relations_group ON relations(type, source_type, target_type);
+        """)
+        connection.commit()
+        payload["counts"].update(counts)
+        payload["counts"]["map_pois"] = spatial.count
+        payload["graph_data"] = {"staging_path": str(staging)}
+        payload["file_sha256"] = reader.sha256.hexdigest()
+        progress("校验完成", counts["nodes"], counts["relations"])
+        return payload
+    except (ValueError, ijson.JSONError) as exc:
+        if isinstance(exc, ProjectValidationError):
+            raise
+        raise ProjectValidationError([f"图谱 JSON 校验失败：{str(exc)[:1000]}"]) from exc
+    finally:
+        node_types.lookup.cache_clear()
+        connection.close()
+
+
+def _record_batches(cursor):
+    batch, size = [], 0
+    for (text,) in cursor:
+        if batch and (len(batch) >= 250 or size + len(text) > 2 * 1024 * 1024):
+            yield batch
+            batch, size = [], 0
+        batch.append(json.loads(text))
+        size += len(text)
+    if batch:
+        yield batch
+
+
+def import_staged_graph(graph_name: str, graph_data: Mapping,
+                        progress: Progress = lambda *_: None) -> dict[str, int]:
+    """Import only a private validated spool; never overwrite an existing graph."""
+    db = _falkor_client()
+    created = False
+    done_nodes = done_relations = 0
+    connection = None
+    try:
+        connection = sqlite3.connect(f"file:{Path(graph_data['staging_path']).as_posix()}?mode=ro", uri=True)
+        connection.execute("PRAGMA cache_size=-4096")
+        existing = {name.decode() if isinstance(name, bytes) else str(name) for name in db.list_graphs()}
+        if graph_name in existing:
+            raise FileExistsError(f"FalkorDB 图 {graph_name!r} 已存在")
+        graph = db.select_graph(graph_name)
+        for (labels_json,) in connection.execute("SELECT DISTINCT labels FROM nodes"):
+            labels = json.loads(labels_json)
+            expression = ":".join(_node_labels({"type": labels[0], "labels": labels}))
+            cursor = connection.execute("SELECT payload FROM nodes WHERE labels=?", (labels_json,))
+            for batch in _record_batches(cursor):
+                rows = []
+                for node in batch:
+                    properties = _graph_safe_properties(node["properties"])
+                    properties.setdefault("id", node["id"])
+                    properties["__kg_node_id"] = node["id"]
+                    rows.append({"properties": properties})
+                created = True
+                graph.query(f"UNWIND $rows AS row CREATE (n:{expression}) SET n += row.properties", {"rows": rows})
+                done_nodes += len(batch)
+                progress("导入节点", done_nodes, done_relations)
+        for (label,) in connection.execute("SELECT DISTINCT type FROM nodes"):
+            _node_labels({"type": label})
+            graph.query(f"CREATE INDEX FOR (n:{label}) ON (n.__kg_node_id)")
+        for signature in connection.execute("SELECT DISTINCT type,source_type,target_type FROM relations"):
+            rel_type, source_type, target_type = signature
+            for label in signature:
+                _node_labels({"type": label})
+            cursor = connection.execute(
+                "SELECT payload FROM relations WHERE type=? AND source_type=? AND target_type=?", signature)
+            for batch in _record_batches(cursor):
+                rows = [{"source": row["source"], "target": row["target"],
+                         "properties": _graph_safe_properties(row["properties"])} for row in batch]
+                graph.query(
+                    f"UNWIND $rows AS row MATCH (s:{source_type} {{__kg_node_id:row.source}}) "
+                    f"MATCH (t:{target_type} {{__kg_node_id:row.target}}) "
+                    f"CREATE (s)-[r:{rel_type}]->(t) SET r += row.properties", {"rows": rows})
+                done_relations += len(batch)
+                progress("导入关系", done_nodes, done_relations)
+        return {
+            "nodes": int(graph.query("MATCH (n) RETURN count(n)").result_set[0][0]),
+            "relations": int(graph.query("MATCH ()-[r]->() RETURN count(r)").result_set[0][0]),
+        }
+    except FileExistsError:
+        raise
+    except Exception as exc:
+        cleanup_error = None
+        if created:
+            try:
+                db.select_graph(graph_name).delete()
+            except Exception as cleanup_exc:
+                cleanup_error = cleanup_exc
+        raise GraphImportError(str(exc), cleanup_error=cleanup_error) from exc
+    finally:
+        if connection is not None:
+            connection.close()
+        db.close()
diff --git a/app/project_lifecycle.py b/app/project_lifecycle.py
index 2e96c76..7a0c79d 100644
--- a/app/project_lifecycle.py
+++ b/app/project_lifecycle.py
@@ -7,6 +7,7 @@ they happen to use the same project id.
 from __future__ import annotations
 
 import json
+import math
 import re
 from collections import defaultdict
 from collections.abc import Mapping
@@ -99,6 +100,11 @@ def _is_spatial_map_node(node: Mapping[str, Any]) -> bool:
     properties = node.get("properties")
     if not isinstance(properties, Mapping):
         return False
+    # A semantic graph can contain coordinates without verified map-region
+    # metadata. Keep those nodes in the graph without adding them to the POI
+    # layer. Old bundles without this optional flag retain their behavior.
+    if properties.get("map_poi") is False:
+        return False
     has_coordinates = (
         properties.get("lng") is not None
         and properties.get("lat") is not None
@@ -265,8 +271,8 @@ def _validate_properties(
             errors.append(f"{path}.{field_name} 应为 {expected} 类型")
 
 
-def normalize_provision_payload(body: Any) -> dict[str, Any]:
-    """Normalize and fully validate a project provision request.
+def normalize_provision_header(body: Any) -> dict[str, Any]:
+    """Validate project metadata and Schema before reading graph records.
 
     The function is intentionally pure so validation can be completed before
     PostgreSQL or FalkorDB is touched.
@@ -389,168 +395,6 @@ def normalize_provision_payload(body: Any) -> dict[str, Any]:
         )
         normalized_relations[relation_name] = meta
 
-    raw_graph_data = body.get("graph_data")
-    if raw_graph_data is None:
-        raw_graph_data = {"nodes": [], "relations": []}
-    if not isinstance(raw_graph_data, Mapping):
-        errors.append("graph_data 必须是 JSON 对象")
-        raw_graph_data = {}
-    raw_nodes = raw_graph_data.get("nodes", [])
-    raw_relations = raw_graph_data.get("relations", [])
-    if not isinstance(raw_nodes, list):
-        errors.append("graph_data.nodes 必须是数组")
-        raw_nodes = []
-    if not isinstance(raw_relations, list):
-        errors.append("graph_data.relations 必须是数组")
-        raw_relations = []
-    if not raw_nodes:
-        errors.append("graph_data.nodes 至少需要一个节点，FalkorDB 不支持持久化空图")
-
-    nodes: list[dict[str, Any]] = []
-    node_types: dict[str, str] = {}
-    for index, raw_node in enumerate(raw_nodes):
-        path = f"graph_data.nodes[{index}]"
-        if not isinstance(raw_node, Mapping):
-            errors.append(f"{path} 必须是 JSON 对象")
-            continue
-        node = dict(raw_node)
-        node_id = str(node.get("id") or "").strip()
-        entity_type = str(node.get("type") or node.get("entity_type") or "").strip()
-        properties = node.get("properties", {})
-        raw_labels = node.get("labels")
-        normalized_labels: list[str] | None = None
-        if not node_id:
-            errors.append(f"{path}.id 不能为空")
-        elif node_id in node_types:
-            errors.append(f"节点 ID {node_id!r} 重复")
-        if entity_type not in normalized_entities:
-            errors.append(f"{path}.type {entity_type!r} 未在 Schema 中定义")
-        if raw_labels is not None:
-            if not isinstance(raw_labels, list) or not raw_labels:
-                errors.append(f"{path}.labels 必须是非空数组")
-            else:
-                normalized_labels = []
-                for label_index, raw_label in enumerate(raw_labels):
-                    label = str(raw_label or "").strip()
-                    label_path = f"{path}.labels[{label_index}]"
-                    if not label or not SAFE_SCHEMA_IDENTIFIER.fullmatch(label):
-                        errors.append(f"{label_path} 不是安全的 Schema/Cypher 标识符")
-                        continue
-                    if label not in normalized_entities:
-                        errors.append(f"{label_path} {label!r} 未在 Schema 中定义")
-                        continue
-                    if label not in normalized_labels:
-                        normalized_labels.append(label)
-                if entity_type and entity_type not in normalized_labels:
-                    errors.append(f"{path}.labels 必须包含主类型 {entity_type!r}")
-        if not isinstance(properties, Mapping):
-            errors.append(f"{path}.properties 必须是 JSON 对象")
-            properties = {}
-        normalized_properties = dict(properties)
-        if entity_type in normalized_entities:
-            _validate_properties(
-                normalized_properties,
-                normalized_entities[entity_type].get("fields", {}),
-                path=path,
-                errors=errors,
-            )
-        if node_id and node_id not in node_types:
-            node_types[node_id] = entity_type
-        normalized_node = {
-            "id": node_id,
-            "type": entity_type,
-            "properties": normalized_properties,
-        }
-        if normalized_labels is not None:
-            normalized_node["labels"] = normalized_labels
-        nodes.append(normalized_node)
-
-    if map_enabled:
-        map_nodes = [node for node in nodes if _is_spatial_map_node(node)]
-        if not map_nodes:
-            errors.append("启用知识地图时，图谱数据至少需要一个带经纬度和业态的空间节点")
-        elif SAFE_ADCODE.fullmatch(map_region_adcode) and map_region_level in SPATIAL_REGION_LEVELS:
-            unverified_region = 0
-            mismatched_adcodes: set[str] = set()
-            for node in map_nodes:
-                properties = node["properties"]
-                adcode = str(properties.get("adcode") or "").strip()
-                if adcode and not _adcode_belongs_to_region(
-                    adcode,
-                    map_region_adcode,
-                    map_region_level,
-                ):
-                    mismatched_adcodes.add(adcode)
-                if adcode:
-                    continue
-                region_field = {
-                    "province": properties.get("province"),
-                    "city": properties.get("city"),
-                    "district": properties.get("district"),
-                }[map_region_level]
-                if not _region_names_match(region_field, map_region_name):
-                    unverified_region += 1
-            if unverified_region:
-                errors.append(
-                    f"知识地图空间节点有 {unverified_region} 条既无可用 adcode，"
-                    "也没有与所选区域匹配的省市区县字段"
-                )
-            if mismatched_adcodes:
-                examples = "、".join(sorted(mismatched_adcodes)[:5])
-                errors.append(
-                    f"知识地图数据与所选区域 {map_region_name} 不一致，发现行政区划代码：{examples}"
-                )
-
-    relations: list[dict[str, Any]] = []
-    for index, raw_relation in enumerate(raw_relations):
-        path = f"graph_data.relations[{index}]"
-        if not isinstance(raw_relation, Mapping):
-            errors.append(f"{path} 必须是 JSON 对象")
-            continue
-        relation = dict(raw_relation)
-        relation_type = str(relation.get("type") or relation.get("relation_type") or "").strip()
-        source = str(relation.get("source") or "").strip()
-        target = str(relation.get("target") or "").strip()
-        properties = relation.get("properties", {})
-        definition = normalized_relations.get(relation_type)
-        if not definition:
-            errors.append(f"{path}.type {relation_type!r} 未在 Schema 中定义")
-        if source not in node_types:
-            errors.append(f"{path}.source {source!r} 不存在")
-        if target not in node_types:
-            errors.append(f"{path}.target {target!r} 不存在")
-        if (
-            definition
-            and source in node_types
-            and node_types[source] not in _endpoint_types(definition.get("from"))
-        ):
-            errors.append(f"{path}.source 的节点类型必须是 {definition.get('from')}")
-        if (
-            definition
-            and target in node_types
-            and node_types[target] not in _endpoint_types(definition.get("to"))
-        ):
-            errors.append(f"{path}.target 的节点类型必须是 {definition.get('to')}")
-        if not isinstance(properties, Mapping):
-            errors.append(f"{path}.properties 必须是 JSON 对象")
-            properties = {}
-        normalized_properties = dict(properties)
-        if definition:
-            _validate_properties(
-                normalized_properties,
-                definition.get("properties", {}),
-                path=path,
-                errors=errors,
-            )
-        relations.append(
-            {
-                "type": relation_type,
-                "source": source,
-                "target": target,
-                "properties": normalized_properties,
-            }
-        )
-
     if errors:
         raise ProjectValidationError(errors)
 
@@ -572,14 +416,205 @@ def normalize_provision_payload(body: Any) -> dict[str, Any]:
             "region_level": map_region_level or "province",
         },
         "schema": schema,
-        "graph_data": {"nodes": nodes, "relations": relations},
+        "graph_data": {"nodes": [], "relations": []},
         "counts": {
             "entity_types": len(normalized_entities),
             "relation_types": len(normalized_relations),
-            "nodes": len(nodes),
-            "relations": len(relations),
+            "nodes": 0,
+            "relations": 0,
         },
     }
+def normalize_graph_node(raw_node: Any, index: int, normalized_entities: Mapping,
+                         node_types: Mapping, errors: list[str]) -> dict | None:
+    path = f"graph_data.nodes[{index}]"
+    if not isinstance(raw_node, Mapping):
+        errors.append(f"{path} 必须是 JSON 对象")
+        return None
+    node = dict(raw_node)
+    node_id = str(node.get("id") or "").strip()
+    entity_type = str(node.get("type") or node.get("entity_type") or "").strip()
+    properties = node.get("properties", {})
+    raw_labels = node.get("labels")
+    normalized_labels: list[str] | None = None
+    if not node_id:
+        errors.append(f"{path}.id 不能为空")
+    elif node_id in node_types:
+        errors.append(f"节点 ID {node_id!r} 重复")
+    if entity_type not in normalized_entities:
+        errors.append(f"{path}.type {entity_type!r} 未在 Schema 中定义")
+    if raw_labels is not None:
+        if not isinstance(raw_labels, list) or not raw_labels:
+            errors.append(f"{path}.labels 必须是非空数组")
+        else:
+            normalized_labels = []
+            for label_index, raw_label in enumerate(raw_labels):
+                label = str(raw_label or "").strip()
+                label_path = f"{path}.labels[{label_index}]"
+                if not label or not SAFE_SCHEMA_IDENTIFIER.fullmatch(label):
+                    errors.append(f"{label_path} 不是安全的 Schema/Cypher 标识符")
+                    continue
+                if label not in normalized_entities:
+                    errors.append(f"{label_path} {label!r} 未在 Schema 中定义")
+                    continue
+                if label not in normalized_labels:
+                    normalized_labels.append(label)
+            if entity_type and entity_type not in normalized_labels:
+                errors.append(f"{path}.labels 必须包含主类型 {entity_type!r}")
+    if not isinstance(properties, Mapping):
+        errors.append(f"{path}.properties 必须是 JSON 对象")
+        properties = {}
+    normalized_properties = dict(properties)
+    if (
+        normalized_properties.get("map_poi") is not None
+        and not isinstance(normalized_properties["map_poi"], bool)
+    ):
+        errors.append(f"{path}.properties.map_poi 必须是布尔值")
+    if entity_type in normalized_entities:
+        _validate_properties(
+            normalized_properties,
+            normalized_entities[entity_type].get("fields", {}),
+            path=path,
+            errors=errors,
+        )
+    normalized_node = {
+        "id": node_id,
+        "type": entity_type,
+        "properties": normalized_properties,
+    }
+    if normalized_labels is not None:
+        normalized_node["labels"] = normalized_labels
+    return normalized_node
+
+
+def normalize_graph_relation(raw_relation: Any, index: int, normalized_relations: Mapping,
+                             node_types: Mapping, errors: list[str]) -> dict | None:
+    path = f"graph_data.relations[{index}]"
+    if not isinstance(raw_relation, Mapping):
+        errors.append(f"{path} 必须是 JSON 对象")
+        return None
+    relation = dict(raw_relation)
+    relation_type = str(relation.get("type") or relation.get("relation_type") or "").strip()
+    source = str(relation.get("source") or "").strip()
+    target = str(relation.get("target") or "").strip()
+    properties = relation.get("properties", {})
+    definition = normalized_relations.get(relation_type)
+    if not definition:
+        errors.append(f"{path}.type {relation_type!r} 未在 Schema 中定义")
+    if source not in node_types:
+        errors.append(f"{path}.source {source!r} 不存在")
+    if target not in node_types:
+        errors.append(f"{path}.target {target!r} 不存在")
+    if (
+        definition
+        and source in node_types
+        and node_types[source] not in _endpoint_types(definition.get("from"))
+    ):
+        errors.append(f"{path}.source 的节点类型必须是 {definition.get('from')}")
+    if (
+        definition
+        and target in node_types
+        and node_types[target] not in _endpoint_types(definition.get("to"))
+    ):
+        errors.append(f"{path}.target 的节点类型必须是 {definition.get('to')}")
+    if not isinstance(properties, Mapping):
+        errors.append(f"{path}.properties 必须是 JSON 对象")
+        properties = {}
+    normalized_properties = dict(properties)
+    if definition:
+        _validate_properties(
+            normalized_properties,
+            definition.get("properties", {}),
+            path=path,
+            errors=errors,
+        )
+    return {
+        "type": relation_type,
+        "source": source,
+        "target": target,
+        "properties": normalized_properties,
+    }
+
+
+class SpatialMapValidation:
+    """Incremental region validation shared by inline and streamed imports."""
+
+    def __init__(self, spatial_map: Mapping):
+        self.config = spatial_map
+        self.count = 0
+        self.unverified = 0
+        self.mismatched: set[str] = set()
+
+    def add(self, node: Mapping, errors: list[str]) -> None:
+        if not _is_spatial_map_node(node):
+            return
+        properties = node["properties"]
+        try:
+            lng = float(properties.get("lng") if properties.get("lng") is not None else properties.get("longitude"))
+            lat = float(properties.get("lat") if properties.get("lat") is not None else properties.get("latitude"))
+            valid = math.isfinite(lng) and math.isfinite(lat) and -180 <= lng <= 180 and -90 <= lat <= 90
+        except (TypeError, ValueError):
+            valid = False
+        if not valid:
+            if self.config["enabled"]:
+                errors.append(f"节点 {node['id']!r} 的地图经纬度无效")
+            return
+        self.count += 1
+        if not self.config["enabled"]:
+            return
+        code = str(properties.get("adcode") or "").strip()
+        level = self.config["region_level"]
+        if code:
+            if not _adcode_belongs_to_region(code, self.config["region_adcode"], level):
+                if len(self.mismatched) < 5:
+                    self.mismatched.add(code)
+        elif not _region_names_match(properties.get({"province": "province", "city": "city", "district": "district"}[level]), self.config["region_name"]):
+            self.unverified += 1
+
+    def finish(self, errors: list[str]) -> None:
+        if not self.config["enabled"]:
+            return
+        if not self.count:
+            errors.append("启用知识地图时，图谱数据至少需要一个带经纬度和业态的空间节点")
+        if self.unverified:
+            errors.append(f"知识地图空间节点有 {self.unverified} 条既无可用 adcode，也没有与所选区域匹配的省市区县字段")
+        if self.mismatched:
+            errors.append(f"知识地图数据与所选区域 {self.config['region_name']} 不一致，发现行政区划代码：{'、'.join(sorted(self.mismatched))}")
+
+
+def normalize_provision_payload(body: Any) -> dict[str, Any]:
+    """Validate small inline requests using the same rules as file imports."""
+    payload = normalize_provision_header(body)
+    errors: list[str] = []
+    data = body.get("graph_data") or {}
+    if not isinstance(data, Mapping):
+        raise ProjectValidationError(["graph_data 必须是 JSON 对象"])
+    raw_nodes, raw_relations = data.get("nodes", []), data.get("relations", [])
+    if not isinstance(raw_nodes, list):
+        errors.append("graph_data.nodes 必须是数组")
+        raw_nodes = []
+    if not isinstance(raw_relations, list):
+        errors.append("graph_data.relations 必须是数组")
+        raw_relations = []
+    if not raw_nodes:
+        errors.append("graph_data.nodes 至少需要一个节点，FalkorDB 不支持持久化空图")
+    nodes, relations, node_types = [], [], {}
+    spatial = SpatialMapValidation(payload["spatial_map"])
+    for index, raw in enumerate(raw_nodes):
+        node = normalize_graph_node(raw, index, payload["schema"]["entity_types"], node_types, errors)
+        if node is not None:
+            nodes.append(node)
+            node_types.setdefault(node["id"], node["type"])
+            spatial.add(node, errors)
+    spatial.finish(errors)
+    for index, raw in enumerate(raw_relations):
+        relation = normalize_graph_relation(raw, index, payload["schema"]["relation_types"], node_types, errors)
+        if relation is not None:
+            relations.append(relation)
+    if errors:
+        raise ProjectValidationError(errors)
+    payload["graph_data"] = {"nodes": nodes, "relations": relations}
+    payload["counts"].update(nodes=len(nodes), relations=len(relations))
+    return payload
 
 
 def _falkor_client() -> FalkorDB:
diff --git a/deploy/nginx/travel-kg.conf.example b/deploy/nginx/travel-kg.conf.example
index c4ec77a..d47cc0d 100644
--- a/deploy/nginx/travel-kg.conf.example
+++ b/deploy/nginx/travel-kg.conf.example
@@ -40,6 +40,21 @@ server {
         proxy_set_header X-Forwarded-Proto https;
     }
 
+    # Only graph file import accepts a large request. The API independently
+    # limits the file to GRAPH_IMPORT_MAX_BYTES (default 1 GiB); allow room for
+    # multipart headers/Schema. All other endpoints retain the 55m limit.
+    location = /v1/admin/projects/provision-file {
+        client_max_body_size 1030m;
+        proxy_pass http://127.0.0.1:8102;
+        include proxy_params;
+        proxy_set_header X-Forwarded-Proto https;
+        proxy_http_version 1.1;
+        proxy_request_buffering off;
+        proxy_buffering off;
+        proxy_read_timeout 120s;
+        proxy_send_timeout 120s;
+    }
+
     location / {
         proxy_pass http://127.0.0.1:8102;
         include proxy_params;
diff --git a/docs/GRAPH_FILE_IMPORT.md b/docs/GRAPH_FILE_IMPORT.md
new file mode 100644
index 0000000..cd269e0
--- /dev/null
+++ b/docs/GRAPH_FILE_IMPORT.md
@@ -0,0 +1,93 @@
+# 大图谱 JSON 文件导入
+
+## 本次修复范围
+
+旧创建向导在浏览器中对整个文件执行 `file.text()`、`JSON.parse()`、
+`JSON.stringify()`，再将全文放进受控文本框；下一步和提交还会再次解析。
+文件在磁盘上是 280 MB，不代表浏览器只需要 280 MB 内存。
+
+另一层限制是仓库 Nginx 示例的 `client_max_body_size 55m`。
+使用该配置的服务器会拒绝大文件；代码审查并不能证明生产服务器已使用或更新了这份配置。
+
+这次仅增加请求内的文件导入通道，不增加后台守护进程、持久任务队列或应用启动逻辑。
+地图模板、图标、项目隔离和原有事务/失败补偿机制不变。
+
+## 使用方式
+
+1. 创建项目，填写名称、英文标识、地图开关及所选地区。
+2. 选择 Schema JSON。Schema 仍是小文件，不超过 2 MiB。
+3. 选择图谱 data JSON。超过 2 MiB 时只显示文件名称和大小，不将全文放进文本框。
+4. 确认创建后上传原始文件，页面显示上传进度及服务器的校验、节点导入、关系导入阶段。
+5. 只有收到最终成功结果才视为完成。上传 100% 不等于图谱创建完成。
+
+需要保持页面和网络连接。当前通道**不提供断点续传或断线后的任务进度恢复**。
+连接中断时，服务器可能仍在完成当前请求或补偿清理；先刷新项目列表确认结果，勿重复创建。
+已有项目和图名仍受冲突保护，不会被上传文件覆盖。
+
+## 处理及安全边界
+
+- 管理员鉴权通过后才接收上传正文。
+- multipart 文件使用系统临时目录的磁盘暂存；图谱内容不整体读入 Python 内存。
+- 增量解析每条节点/关系，使用请求私有的 SQLite 临时索引校验重复 ID、Schema 类型及关系端点。
+- 全部校验通过后才写 FalkorDB，按最多 250 条或约 2 MiB 记录文本分批提交。
+- 多标签、原始属性、平行关系均保留。启用地图时检查坐标和地区，`map_poi: false` 的辅助知识节点不计入地图点。
+- 写入数量与校验数量一致后，复用原有项目元数据事务发布；失败则执行原有补偿逻辑。
+- 请求结束会关闭上传文件、清理临时索引；数据库工作线程未停止前不会先删除其文件。
+- 每个 API 进程同时只处理一个文件导入。多进程部署需按总并发规划磁盘和数据库容量。
+- 日志只记录项目、阶段、结果数量或错误类别，不记录上传的完整 JSON、凭据和记录正文。
+
+SQLite 在这里仅是临时校验索引，**不是新增业务数据库**。图谱最终存储仍为 FalkorDB。
+
+## 容量限制与部署
+
+| 项目 | 限制 |
+| --- | --- |
+| 浏览器 JSON 编辑器 / Schema | 2 MiB |
+| 旧 `/v1/admin/projects/provision` JSON 请求体 | 8 MiB，解析前拦截 |
+| 文件接口 `/v1/admin/projects/provision-file` | `GRAPH_IMPORT_MAX_BYTES`，默认 1 GiB |
+| 单条记录 / 单个字符串 / 嵌套 | 约 8 MiB 字符计数 / 4 MiB 原始字节 / 64 层 |
+
+默认 1 GiB 是配置上限，**不是所有硬件都已通过 1 GiB 验收的承诺**。
+更大的文件应先评估 RAM、FalkorDB 容量、临时盘及代理限制，再进行对应体量的测试，不能仅无限放大上限。
+
+部署前需要：
+
+1. 安装更新后的 Python 依赖，包含 `ijson==3.5.1`，并构建前端。
+2. 对照 `deploy/nginx/travel-kg.conf.example` 添加**仅文件导入路径**的配置。
+   该路径默认允许 1030 MiB 请求、关闭请求和响应缓冲，其他接口仍保留原限制。
+   由运维执行 Nginx 配置检查并重载；如果前方还有 CDN/负载均衡，也必须核对其请求限制。
+3. 临时目录需为可写、私有、容量充足的数据卷。可在启动 API 前通过 `TMPDIR` 指向管理员准备好的目录；
+   容器中须保证该目录实际挂载且属主匹配，不能只改目录名称。不要放在公开静态文件目录。
+4. 已知 Content-Length 时，预检估算至少需要 `请求大小 × 3 + 512 MiB` 的可用空间。
+   这是估算，不保证覆盖所有 JSON 结构及并发写入。处理中仍保留至少 512 MiB 磁盘余量；
+   资源不足会明确中止而不是发布半成品图谱。FalkorDB 内存、持久化、备份空间另行预留。
+5. 本次代码修改不会自动改动或重启生产服务器，也不将业务 JSON 或凭据提交到 Git。
+
+## 验证及复测
+
+自动化回归：
+
+```sh
+python3 -m unittest tests.test_graph_file_import tests.test_project_lifecycle tests.test_plaza_graph_fallback
+cd admin-web
+node --experimental-strip-types --test tests/graphFileUpload.test.ts tests/projectCreationMultiLabel.test.ts tests/projectCatalog.test.ts
+```
+
+真实文件验证脚本（请在容量充足的本地测试环境执行）：
+
+```sh
+python3 scripts/verify_graph_file_import.py \
+  --schema /absolute/path/schema.json \
+  --data /absolute/path/data.json \
+  --report /absolute/path/new-verification-report.json \
+  --import-falkor
+```
+
+脚本的地区配置为本次贵阳市样本；其他地区需采用匹配的测试配置。
+它通过独立 ASGI 测试应用验证 multipart 接收、流式校验、真实本机 FalkorDB 分批写入和地图 POI 读取。
+只创建 UUID 临时图并在结束时删除。系统项目元数据与登录身份使用测试替身，
+因此**不替代浏览器、生产 Nginx、真实元数据事务和服务器部署的端到端验收**。
+未加 `--import-falkor` 时只测接收和校验，不写图谱。
+
+报告包含成功/失败、阶段、原文件大小、内存峰值、临时索引占用和磁盘余量。
+失败的报告不应表述为全量导入成功。
diff --git a/requirements.txt b/requirements.txt
index 21397dd..9d287a3 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -18,6 +18,7 @@ pydantic-settings==2.14.1
 
 # 图数据库 (FalkorDB on Redis)
 falkordb==1.6.1
+ijson==3.5.1
 
 # 空间索引 (H3 网格；PostGIS 由 PostgreSQL 扩展提供)
 h3==4.4.2
diff --git a/scripts/verify_graph_file_import.py b/scripts/verify_graph_file_import.py
new file mode 100644
index 0000000..7fd3604
--- /dev/null
+++ b/scripts/verify_graph_file_import.py
@@ -0,0 +1,167 @@
+#!/usr/bin/env python3
+"""Exercise a real JSON file through multipart + streaming validation.
+
+Optional --import-falkor writes only a UUID-named local test graph and removes
+it in finally. System project metadata/auth are test doubles; this is NOT a
+production importer, nor evidence of a browser/proxy/production deployment test.
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+from pathlib import Path
+import resource
+import shutil
+import sys
+import tempfile
+import time
+import uuid
+from unittest.mock import patch
+
+import httpx
+from fastapi import FastAPI
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from app.api import graph_file_import as file_api
+from app.api.plaza import map_pois
+from app.auth import get_current_user
+from app.config import settings
+from app.project_context import ProjectContext
+from app.project_lifecycle import _falkor_client
+
+
+def graph_names(db) -> set[str]:
+    return {name.decode() if isinstance(name, bytes) else str(name) for name in db.list_graphs()}
+
+
+async def verify(args):
+    if settings.falkordb_host not in {"localhost", "127.0.0.1", "::1"}:
+        raise RuntimeError("验证仅允许本机 FalkorDB，不连接生产服务器")
+    if args.report.exists():
+        raise FileExistsError("报告已存在，请选择新路径")
+    if args.schema.stat().st_size > file_api.SCHEMA_BYTES:
+        raise ValueError("Schema 超过大小上限")
+    schema = json.loads(args.schema.read_text(encoding="utf-8"))
+    scratch = "codex_upload_verify_" + uuid.uuid4().hex
+    metadata = {"project_id": scratch, "display_name": "城市知识图谱-贵州*贵阳（本地导入验证）",
+                "schema": schema, "spatial_map": {"enabled": True, "scope": "guizhou",
+                "region_name": "贵阳市", "region_adcode": "520100", "region_level": "city"}}
+    app = FastAPI()
+    app.include_router(file_api.router, prefix="/v1/admin")
+    # Isolated in-process test app only; never bypass the running app's login.
+    app.dependency_overrides[get_current_user] = lambda: {"username": "test_only", "roles": ["admin"]}
+
+    @app.get("/health")
+    async def health():
+        return {"ok": True}
+
+    report = {"file_bytes": args.data.stat().st_size, "metadata_and_auth": "isolated_test_doubles",
+              "transport": "HTTPX ASGI multipart", "falkordb_import": bool(args.import_falkor),
+              "temp_free_before_bytes": shutil.disk_usage(tempfile.gettempdir()).free}
+    created = False
+    db = _falkor_client() if args.import_falkor else None
+    existing_graphs = graph_names(db) if db else set()
+    if scratch in existing_graphs:
+        raise FileExistsError("临时图名冲突，未开始测试")
+    if db:
+        report["existing_graphs_before"] = len(existing_graphs)
+    real_validator = file_api.validate_graph_file
+    last_print = time.monotonic()
+
+    def validate(source, staging, body, progress):
+        def observed(phase, nodes, relations):
+            nonlocal last_print
+            size = staging.stat().st_size if staging.exists() else 0
+            report["peak_staging_bytes"] = max(report.get("peak_staging_bytes", 0), size)
+            report["last_progress"] = {"phase": phase, "nodes": nodes, "relations": relations,
+                                       "staging_bytes": size, "temp_free_bytes": shutil.disk_usage(tempfile.gettempdir()).free}
+            progress(phase, nodes, relations)
+            if time.monotonic() - last_print > 8:
+                print(json.dumps({"phase": phase, "nodes": nodes, "relations": relations}, ensure_ascii=False), flush=True)
+                last_print = time.monotonic()
+        return real_validator(source, staging, body, observed)
+
+    async def publish(payload, user, *, graph_importer):
+        nonlocal created
+        report["validated_counts"] = payload["counts"]
+        report["file_sha256"] = payload["file_sha256"]
+        if db:
+            created = True
+            counts = await asyncio.to_thread(graph_importer, scratch, payload["graph_data"])
+            report["actual_graph_counts"] = counts
+            assert counts["nodes"] == payload["counts"]["nodes"]
+            assert counts["relations"] == payload["counts"]["relations"]
+        return {"project": {"project_id": scratch}, "counts": payload["counts"]}
+
+    started = time.monotonic()
+    latencies = []
+    failure = None
+    try:
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://local-test", timeout=None) as client:
+            with args.data.open("rb") as source, patch.object(file_api, "validate_graph_file", side_effect=validate), patch(
+                "app.api.projects.provision_validated_project", side_effect=publish,
+            ):
+                upload = asyncio.create_task(client.post(
+                    "/v1/admin/projects/provision-file", data={"metadata": json.dumps(metadata, ensure_ascii=False)},
+                    files={"file": (args.data.name, source, "application/json")},
+                ))
+                while not upload.done():
+                    t = time.monotonic()
+                    health = await client.get("/health")
+                    assert health.status_code == 200
+                    latencies.append(round((time.monotonic() - t) * 1000, 2))
+                    await asyncio.sleep(0.5)
+                response = await upload
+            report["status_code"] = response.status_code
+            if response.status_code != 200:
+                raise RuntimeError(response.text[:1500])
+            messages = [json.loads(line) for line in response.text.splitlines()]
+            report["response_events"] = len(messages)
+            if not messages or messages[-1]["event"] != "result":
+                raise RuntimeError(str(messages[-1] if messages else "没有返回结果"))
+            if db:
+                pois = await map_pois(context=ProjectContext("local_test", scratch, scratch), _user=None)
+                report["map_pois"] = pois["total"]
+                assert pois["source"] == "falkordb"
+                assert pois["total"] == report["validated_counts"]["map_pois"]
+                del pois
+            report["passed"] = True
+    except Exception as exc:
+        failure = exc
+        report.update(passed=False, error=str(exc)[:2000])
+    finally:
+        if db:
+            try:
+                if created and scratch in graph_names(db):
+                    db.select_graph(scratch).delete()
+                final_graphs = graph_names(db)
+                if scratch in final_graphs:
+                    raise RuntimeError("UUID 临时图清理失败")
+                if not existing_graphs.issubset(final_graphs):
+                    raise RuntimeError("验证期间发现原有图谱缺失")
+                report.update(test_graph_removed=True, existing_graphs_preserved=len(existing_graphs))
+            finally:
+                db.close()
+    report.update(seconds=round(time.monotonic() - started, 2),
+                  peak_rss_bytes=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss * (1 if sys.platform == "darwin" else 1024),
+                  health_checks=len(latencies), max_health_latency_ms=max(latencies, default=0),
+                  temp_free_after_bytes=shutil.disk_usage(tempfile.gettempdir()).free)
+    args.report.parent.mkdir(parents=True, exist_ok=True)
+    with args.report.open("x", encoding="utf-8") as stream:
+        json.dump(report, stream, ensure_ascii=False, indent=2)
+    print(json.dumps(report, ensure_ascii=False, indent=2), flush=True)
+    if failure:
+        raise RuntimeError("验证未通过，原因及资源占用已写入报告") from failure
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--schema", type=Path, required=True)
+    parser.add_argument("--data", type=Path, required=True)
+    parser.add_argument("--report", type=Path, required=True)
+    parser.add_argument("--import-falkor", action="store_true")
+    asyncio.run(verify(parser.parse_args()))
diff --git a/tests/test_graph_file_import.py b/tests/test_graph_file_import.py
new file mode 100644
index 0000000..a859d8c
--- /dev/null
+++ b/tests/test_graph_file_import.py
@@ -0,0 +1,315 @@
+import asyncio
+import copy
+from contextlib import closing
+import hashlib
+import io
+import json
+from pathlib import Path
+import sqlite3
+import tempfile
+import time
+from types import SimpleNamespace
+import unittest
+from unittest.mock import AsyncMock, patch
+
+import httpx
+import ijson
+from fastapi import FastAPI
+from starlette.requests import ClientDisconnect, Request
+
+from app.api import graph_file_import as file_api
+from app.api import projects as projects_api
+from app.auth import get_current_user
+from app.graph_import_files import CheckedReader, graph_records, import_staged_graph, validate_graph_file
+from app.project_lifecycle import GraphImportError, ProjectValidationError, normalize_provision_payload
+from tests.test_project_lifecycle import valid_payload
+
+
+class FileValidationTests(unittest.TestCase):
+    def validate(self, body, graph=None):
+        source = io.BytesIO(json.dumps(graph if graph is not None else body["graph_data"], ensure_ascii=False).encode())
+        directory = tempfile.TemporaryDirectory(prefix="test-graph-file-")
+        self.addCleanup(directory.cleanup)
+        result = validate_graph_file(source, Path(directory.name) / "index.sqlite", body)
+        self.assertFalse(source.closed)
+        return result
+
+    def test_stream_and_inline_validation_preserve_same_records(self):
+        body = valid_payload()
+        body["graph_data"]["nodes"][0]["properties"]["details"] = {"中文": [1, True, "原始内容"]}
+        body["graph_data"]["nodes"][0]["labels"] = ["Place"]
+        result = self.validate(body)
+        expected = normalize_provision_payload(body)
+        with closing(sqlite3.connect(result["graph_data"]["staging_path"])) as connection:
+            nodes = [json.loads(row[0]) for row in connection.execute("SELECT payload FROM nodes ORDER BY id")]
+            relations = [json.loads(row[0]) for row in connection.execute("SELECT payload FROM relations ORDER BY seq")]
+        self.assertEqual(nodes, expected["graph_data"]["nodes"])
+        self.assertEqual(relations, expected["graph_data"]["relations"])
+        self.assertEqual(result["counts"]["nodes"], 2)
+        self.assertEqual(result["counts"]["relations"], 1)
+
+    def test_reads_bounded_blocks_and_hashes_original_bytes(self):
+        data = json.dumps(valid_payload()["graph_data"], ensure_ascii=False).encode()
+
+        class Bounded(io.BytesIO):
+            def read(self, size=-1):
+                assert 0 <= size <= 65536, size
+                return super().read(size)
+
+        with tempfile.TemporaryDirectory() as directory:
+            result = validate_graph_file(Bounded(data), Path(directory) / "index.sqlite", valid_payload())
+        self.assertEqual(result["file_sha256"], hashlib.sha256(data).hexdigest())
+
+    def test_relations_can_precede_nodes(self):
+        body = valid_payload()
+        result = self.validate(body, {"relations": body["graph_data"]["relations"], "nodes": body["graph_data"]["nodes"]})
+        self.assertEqual(result["counts"]["relations"], 1)
+
+    def test_invalid_records_fail_before_import(self):
+        for field, value in [("nodes", []), ("nodes", {}), ("relations", {}), ("nodes", ["bad"])]:
+            with self.subTest(field=field, value=value):
+                body = valid_payload()
+                body["graph_data"][field] = value
+                with self.assertRaises(ProjectValidationError):
+                    self.validate(body)
+
+    def test_duplicates_unknown_types_and_missing_endpoints_fail(self):
+        bodies = []
+        body = valid_payload()
+        body["graph_data"]["nodes"].append(copy.deepcopy(body["graph_data"]["nodes"][0]))
+        bodies.append(body)
+        body = valid_payload()
+        body["graph_data"]["nodes"][0]["type"] = "NotInSchema"
+        bodies.append(body)
+        body = valid_payload()
+        body["graph_data"]["relations"][0]["target"] = "missing"
+        bodies.append(body)
+        for body in bodies:
+            with self.subTest(body=body), self.assertRaises(ProjectValidationError):
+                self.validate(body)
+
+    def test_duplicate_json_fields_truncation_and_wrong_root_fail(self):
+        for raw in [b'{"nodes":[],"nodes":[]}', b'{"nodes":[{}]', b'[]', b'{"schema":{}}',
+                    b'{"nodes":[{"id":"a","id":"b"}]}']:
+            with self.subTest(raw=raw), self.assertRaises((ValueError, ijson.JSONError)):
+                list(graph_records(CheckedReader(io.BytesIO(raw))))
+
+    def test_deeply_nested_or_oversized_record_rejected(self):
+        with self.assertRaisesRegex(ValueError, "64"):
+            list(graph_records(CheckedReader(io.BytesIO(b'[' * 65 + b']' * 65))))
+        with patch("app.graph_import_files.MAX_RECORD_CHARS", 8):
+            with self.assertRaisesRegex(ValueError, "单条"):
+                list(graph_records(CheckedReader(io.BytesIO(b'{"nodes":[{"id":"long-string"}]}'))))
+
+    def test_map_validation_and_helper_exclusion_match_import_rules(self):
+        body = valid_payload()
+        body["spatial_map"] = {"enabled": True, "scope": "guizhou", "region_name": "贵阳市",
+                               "region_adcode": "520100", "region_level": "city"}
+        for index, node in enumerate(body["graph_data"]["nodes"]):
+            node["properties"].update(lng=106.7, lat=26.6, adcode="520102", place_type="Hotel", map_poi=index == 0)
+        result = self.validate(body)
+        self.assertEqual(result["counts"]["map_pois"], 1)
+        body["graph_data"]["nodes"][0]["properties"]["adcode"] = "522722"
+        with self.assertRaisesRegex(ProjectValidationError, "所选区域"):
+            self.validate(body)
+
+    def test_existing_staging_is_never_overwritten(self):
+        with tempfile.TemporaryDirectory() as directory:
+            staging = Path(directory) / "index.sqlite"
+            staging.touch()
+            with self.assertRaises(FileExistsError):
+                validate_graph_file(io.BytesIO(b'{}'), staging, valid_payload())
+
+
+class FileImportTests(unittest.TestCase):
+    validate = FileValidationTests.validate
+    def test_batches_preserve_labels_parallel_edges_and_counts(self):
+        body = valid_payload()
+        body["schema"]["entity_types"]["Hotel"] = {"fields": {"name": "string"}}
+        body["graph_data"]["nodes"] = [
+            {"id": str(i), "type": "Place", "labels": ["Place", "Hotel"], "properties": {"name": str(i)}}
+            for i in range(503)
+        ]
+        body["graph_data"]["relations"] = [
+            {"type": "RELATED_TO", "source": "0", "target": "1", "properties": {"weight": i}}
+            for i in range(502)
+        ]
+        result = self.validate(body)
+        calls = []
+
+        def query(cypher, params=None):
+            calls.append((cypher, params))
+            if cypher == "MATCH (n) RETURN count(n)":
+                return SimpleNamespace(result_set=[[503]])
+            return SimpleNamespace(result_set=[[502]])
+
+        graph = SimpleNamespace(query=query, delete=unittest.mock.Mock())
+        client = SimpleNamespace(list_graphs=lambda: [], select_graph=lambda _: graph, close=unittest.mock.Mock())
+        with patch("app.graph_import_files._falkor_client", return_value=client):
+            self.assertEqual(import_staged_graph("test_only", result["graph_data"]), {"nodes": 503, "relations": 502})
+        node_calls = [call for call in calls if "CREATE (n:" in call[0]]
+        edge_calls = [call for call in calls if "CREATE (s)-[r:" in call[0]]
+        self.assertEqual([len(call[1]["rows"]) for call in node_calls], [250, 250, 3])
+        self.assertEqual([len(call[1]["rows"]) for call in edge_calls], [250, 250, 2])
+        self.assertIn("n:Place:Hotel", node_calls[0][0])
+        graph.delete.assert_not_called()
+        client.close.assert_called_once()
+
+    def test_never_deletes_existing_graph_on_conflict(self):
+        result = self.validate(valid_payload())
+        delete = unittest.mock.Mock()
+        client = SimpleNamespace(list_graphs=lambda: [b"existing"], select_graph=lambda _: SimpleNamespace(delete=delete),
+                                 close=unittest.mock.Mock())
+        with patch("app.graph_import_files._falkor_client", return_value=client), self.assertRaises(FileExistsError):
+            import_staged_graph("existing", result["graph_data"])
+        delete.assert_not_called()
+        client.close.assert_called_once()
+
+    def test_failed_write_cleans_only_new_graph(self):
+        result = self.validate(valid_payload())
+        graph = SimpleNamespace(query=unittest.mock.Mock(side_effect=RuntimeError("write failed")), delete=unittest.mock.Mock())
+        client = SimpleNamespace(list_graphs=lambda: [], select_graph=lambda _: graph, close=unittest.mock.Mock())
+        with patch("app.graph_import_files._falkor_client", return_value=client), self.assertRaises(GraphImportError):
+            import_staged_graph("new_only", result["graph_data"])
+        graph.delete.assert_called_once()
+
+
+class FileRouteTests(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.app = FastAPI()
+        self.app.include_router(file_api.router, prefix="/v1/admin")
+        self.app.include_router(projects_api.router, prefix="/v1/admin")
+        self.app.dependency_overrides[get_current_user] = lambda: {"username": "test_admin", "roles": ["admin"]}
+        self.client = httpx.AsyncClient(transport=httpx.ASGITransport(app=self.app), base_url="http://testserver")
+        self.addAsyncCleanup(self.client.aclose)
+        self.body = valid_payload()
+
+    def upload(self, data=None, **kwargs):
+        return self.client.post("/v1/admin/projects/provision-file", data={"metadata": json.dumps(self.body)},
+                                files={"file": ("graph.json", data or json.dumps(self.body["graph_data"]).encode(), "application/json")}, **kwargs)
+
+    async def test_valid_file_streams_counts_and_result_and_releases_lock(self):
+        seen = {}
+
+        async def publish(payload, user, **kwargs):
+            seen.update(payload)
+            self.assertTrue(Path(payload["graph_data"]["staging_path"]).is_file())
+            self.assertEqual(user["username"], "test_admin")
+            return {"project": {"project_id": "test"}, "counts": payload["counts"]}
+
+        with patch("app.api.projects.provision_validated_project", side_effect=publish):
+            response = await self.upload()
+        self.assertEqual(response.status_code, 200)
+        messages = [json.loads(line) for line in response.text.splitlines()]
+        self.assertEqual(messages[-1]["event"], "result")
+        self.assertEqual(messages[-1]["data"]["counts"]["nodes"], 2)
+        self.assertFalse(Path(seen["graph_data"]["staging_path"]).exists())
+        self.assertFalse(file_api._import_lock.locked())
+
+    async def test_invalid_json_reports_error_and_does_not_publish(self):
+        with patch("app.api.projects.provision_validated_project", new_callable=AsyncMock) as publish:
+            response = await self.upload(b'{"nodes": [')
+        self.assertEqual(response.status_code, 200)
+        messages = [json.loads(line) for line in response.text.splitlines()]
+        self.assertEqual(messages[-1]["event"], "error")
+        publish.assert_not_called()
+        self.assertFalse(file_api._import_lock.locked())
+
+    async def test_viewer_cannot_upload_and_body_is_not_read(self):
+        self.app.dependency_overrides[get_current_user] = lambda: {"username": "viewer", "roles": ["operator"]}
+        with patch.object(file_api, "read_import_form", new_callable=AsyncMock) as read:
+            response = await self.upload()
+        self.assertEqual(response.status_code, 403)
+        read.assert_not_called()
+
+    async def test_oversize_and_low_disk_fail_clearly(self):
+        with patch.object(file_api.settings, "graph_import_max_bytes", 1):
+            self.assertEqual((await self.upload()).status_code, 413)
+        with patch.object(file_api.shutil, "disk_usage", return_value=SimpleNamespace(free=0)):
+            self.assertEqual((await self.upload()).status_code, 507)
+        self.assertFalse(file_api._import_lock.locked())
+
+    async def test_cpu_validation_does_not_block_health_request(self):
+        @self.app.get("/ping")
+        async def ping():
+            return {"ok": True}
+
+        original = file_api.validate_graph_file
+        started = asyncio.Event()
+        loop = asyncio.get_running_loop()
+
+        def slower(*args):
+            loop.call_soon_threadsafe(started.set)
+            time.sleep(0.25)
+            return original(*args)
+
+        with patch.object(file_api, "validate_graph_file", side_effect=slower), patch(
+            "app.api.projects.provision_validated_project", new_callable=AsyncMock, return_value={"project": {"project_id": "test"}},
+        ):
+            task = asyncio.create_task(self.upload())
+            await asyncio.wait_for(started.wait(), 2)
+            response = await asyncio.wait_for(self.client.get("/ping"), 0.15)
+            self.assertEqual(response.status_code, 200)
+            second = await self.upload()
+            self.assertEqual(second.status_code, 429)
+            await task
+
+    async def test_disconnect_closes_partial_uploaded_file(self):
+        original = tempfile.SpooledTemporaryFile
+        files = []
+
+        def capture(*args, **kwargs):
+            result = original(*args, **kwargs)
+            files.append(result)
+            return result
+
+        incoming = iter([
+            {"type": "http.request", "body": b'--x\r\nContent-Disposition: form-data; name="file"; filename="a.json"\r\n\r\npartial', "more_body": True},
+            {"type": "http.disconnect"},
+        ])
+
+        async def receive():
+            return next(incoming)
+
+        request = Request({"type": "http", "headers": [(b"content-type", b"multipart/form-data; boundary=x")]}, receive)
+        with patch("starlette.formparsers.SpooledTemporaryFile", side_effect=capture), self.assertRaises(ClientDisconnect):
+            await file_api.read_import_form(request)
+        self.assertEqual(len(files), 1)
+        self.assertTrue(files[0].closed)
+
+    async def test_small_inline_json_remains_compatible(self):
+        with patch.object(projects_api, "provision_validated_project", new_callable=AsyncMock,
+                          return_value={"project": {"project_id": "test"}}) as publish:
+            response = await self.client.post("/v1/admin/projects/provision", json=self.body)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(publish.call_args.args[0]["counts"]["nodes"], 2)
+
+    async def test_inline_limit_applies_before_json_parsing(self):
+        with patch.object(projects_api, "normalize_provision_payload") as normalize:
+            response = await self.client.post("/v1/admin/projects/provision", content=b'{}',
+                                              headers={"Content-Length": str(8 * 1024 * 1024 + 1)})
+        self.assertEqual(response.status_code, 413)
+        normalize.assert_not_called()
+
+    async def test_chunked_inline_request_cannot_bypass_limit(self):
+        async def content():
+            for _ in range(9):
+                yield b' ' * (1024 * 1024)
+
+        with patch.object(projects_api, "normalize_provision_payload") as normalize:
+            response = await self.client.post("/v1/admin/projects/provision", content=content())
+        self.assertEqual(response.status_code, 413)
+        normalize.assert_not_called()
+
+    async def test_inline_invalid_json_and_non_admin_fail(self):
+        response = await self.client.post("/v1/admin/projects/provision", content=b'{')
+        self.assertEqual(response.status_code, 422)
+        self.app.dependency_overrides[get_current_user] = lambda: {"username": "viewer", "roles": ["operator"]}
+        with patch.object(projects_api, "normalize_provision_payload") as normalize:
+            response = await self.client.post("/v1/admin/projects/provision", json=self.body)
+        self.assertEqual(response.status_code, 403)
+        normalize.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_project_lifecycle.py b/tests/test_project_lifecycle.py
index 19e40da..e975062 100644
--- a/tests/test_project_lifecycle.py
+++ b/tests/test_project_lifecycle.py
@@ -172,6 +172,13 @@ class ProjectPayloadValidationTests(unittest.TestCase):
 
         self.assertIn("数据与所选区域 贵阳市 不一致", "；".join(raised.exception.errors))
 
+    def test_map_poi_flag_must_be_boolean_not_a_string(self) -> None:
+        body = valid_payload()
+        body["graph_data"]["nodes"][0]["properties"]["map_poi"] = "false"
+        with self.assertRaises(ProjectValidationError) as raised:
+            normalize_provision_payload(body)
+        self.assertIn("map_poi 必须是布尔值", "；".join(raised.exception.errors))
+
     def test_spatial_map_accepts_legacy_city_name_when_adcode_is_missing(self) -> None:
         body = valid_payload()
         body["schema"]["entity_types"]["Place"]["fields"].update({