xuelong/Cloud-Tour-to-Libo
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: export graph snapshots without data drift

Browse Source
This commit is contained in:
3452078359-xuexue
2026-08-26 01:44:22 -07:00
parent 4ccb63855f
commit 329a94d457
8 changed files with 673 additions and 1707 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
admin-web/src/ProjectLanding.tsx
View File

@@ -493,6 +493,8 @@ function validateGraphText(
}
const id = String(rawNode.id || "").trim();
const entityType = String(rawNode.type || rawNode.entity_type || "").trim();
const rawLabels = rawNode.labels;
let normalizedLabels: string[] | undefined;
const properties = rawNode.properties === undefined ? {} : rawNode.properties;
const normalizedProperties = isRecord(properties) ? properties : {};
if (!id) errors.push(`${path}.id 必须是非空字符串`);
@@ -503,6 +505,29 @@ function validateGraphText(
if (entityType && !isRecord(entityDefinition)) {
errors.push(`${path}.type ${entityType} 未在 Schema 中定义`);
}
if (rawLabels !== undefined) {
if (!Array.isArray(rawLabels) || !rawLabels.length) {
errors.push(`${path}.labels 必须是非空数组`);
} else {
normalizedLabels = [];
rawLabels.forEach((rawLabel, labelIndex) => {
const label = String(rawLabel || "").trim();
const labelPath = `${path}.labels[${labelIndex}]`;
if (!label || !SAFE_SCHEMA_IDENTIFIER_PATTERN.test(label)) {
errors.push(`${labelPath} 不是安全的 Schema/Cypher 标识符`);
return;
}
if (!isRecord(entityTypes[label])) {
errors.push(`${labelPath} ${label} 未在 Schema 中定义`);
return;
}
if (!normalizedLabels?.includes(label)) normalizedLabels?.push(label);
});
if (entityType && !normalizedLabels.includes(entityType)) {
errors.push(`${path}.labels 必须包含主类型 ${entityType}`);
}
}
}
if (!isRecord(properties)) errors.push(`${path}.properties 必须是对象`);
if (isRecord(entityDefinition)) {
const fields = normalizeFieldDefinitions(
@@ -512,7 +537,12 @@ function validateGraphText(
);
validateProperties(normalizedProperties, fields, `${path}.properties`, errors);
}
normalizedNodes.push({ id, type: entityType, properties: normalizedProperties });
normalizedNodes.push({
id,
type: entityType,
...(normalizedLabels ? { labels: normalizedLabels } : {}),
properties: normalizedProperties,
});
});
const normalizedRelations: Array<Record<string, unknown>> = [];

17
admin-web/tests/projectCreationMultiLabel.test.ts Normal file
View File

@@ -0,0 +1,17 @@
import assert from "node:assert/strict";
import { readFileSync } from "node:fs";
import test from "node:test";
const projectLandingSource = readFileSync(
new URL("../src/ProjectLanding.tsx", import.meta.url),
"utf8",
);
test("project creation keeps validated multi-label graph nodes", () => {
assert.match(projectLandingSource, /const rawLabels = rawNode\.labels/);
assert.match(projectLandingSource, /labels 必须包含主类型/);
assert.match(
projectLandingSource,
/\.\.\.\(normalizedLabels \? \{ labels: normalizedLabels \} : \{\}\)/,
);
});

62
app/project_lifecycle.py
View File

@@ -351,12 +351,32 @@ def normalize_provision_payload(body: Any) -> dict[str, Any]:
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 = {}
@@ -370,7 +390,14 @@ def normalize_provision_payload(body: Any) -> dict[str, Any]:
)
if node_id and node_id not in node_types:
node_types[node_id] = entity_type
nodes.append({"id": node_id, "type": entity_type, "properties": normalized_properties})
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)
relations: list[dict[str, Any]] = []
for index, raw_relation in enumerate(raw_relations):
@@ -488,6 +515,22 @@ def _graph_safe_properties(properties: Mapping[str, Any]) -> dict[str, Any]:
}
def _node_labels(node: Mapping[str, Any]) -> tuple[str, ...]:
"""Return validated FalkorDB labels, preserving optional multi-label nodes."""
primary = str(node.get("type") or "").strip()
raw_labels = node.get("labels")
labels = [str(item).strip() for item in raw_labels] if isinstance(raw_labels, list) else []
if primary and primary not in labels:
labels.insert(0, primary)
if not labels:
labels = [primary]
deduplicated = tuple(dict.fromkeys(labels))
if not all(label and SAFE_SCHEMA_IDENTIFIER.fullmatch(label) for label in deduplicated):
raise ValueError("图谱节点包含不安全的 Cypher 标签")
return deduplicated
def _batches(items: list[Any], size: int = GRAPH_IMPORT_BATCH_SIZE):
for start in range(0, len(items), size):
yield items[start:start + size]
@@ -499,19 +542,21 @@ def _import_falkor_graph_batched(graph: Any, graph_data: Mapping[str, Any]) -> N
relations = list(graph_data.get("relations", []))
node_types = {str(item["id"]): str(item["type"]) for item in nodes}
nodes_by_type: dict[str, list[dict[str, Any]]] = defaultdict(list)
nodes_by_labels: dict[tuple[str, ...], list[dict[str, Any]]] = defaultdict(list)
for item in nodes:
properties = _graph_safe_properties(item.get("properties") or {})
properties.setdefault("id", str(item["id"]))
properties["__kg_node_id"] = str(item["id"])
nodes_by_type[str(item["type"])].append({"properties": properties})
nodes_by_labels[_node_labels(item)].append({"properties": properties})
for label, items in nodes_by_type.items():
for labels, items in nodes_by_labels.items():
label_expression = ":".join(labels)
for batch in _batches(items):
graph.query(
f"UNWIND $rows AS row CREATE (n:{label}) SET n += row.properties",
f"UNWIND $rows AS row CREATE (n:{label_expression}) SET n += row.properties",
{"rows": batch},
)
for label in sorted(set(node_types.values())):
# Relation creation matches both endpoint labels and this indexed id,
# avoiding repeated full-graph scans for large JSON imports.
graph.query(f"CREATE INDEX FOR (n:{label}) ON (n.__kg_node_id)")
@@ -567,14 +612,17 @@ def import_falkor_graph(graph_name: str, graph_data: Mapping[str, Any]) -> dict[
_import_falkor_graph_batched(graph, {"nodes": nodes, "relations": relations})
else:
for node in nodes:
label = str(node["type"])
label_expression = ":".join(_node_labels(node))
properties = _graph_safe_properties(node.get("properties") or {})
properties.setdefault("id", str(node["id"]))
properties["__kg_node_id"] = str(node["id"])
# From this point onward the graph may exist even if the client
# raises after the server accepted the command.
created = True
graph.query(f"CREATE (n:{label}) SET n += $props", {"props": properties})
graph.query(
f"CREATE (n:{label_expression}) SET n += $props",
{"props": properties},
)
for relation in relations:
relation_type = str(relation["type"])
properties = _graph_safe_properties(relation.get("properties") or {})

1128
qa/graph-creation-fixtures/libo_complex_graph_data.v1.json
View File

File diff suppressed because it is too large Load Diff

512
qa/graph-creation-fixtures/libo_complex_schema.v1.json
View File

@@ -1,512 +0,0 @@
{
"namespace": "libo_complex_manual_test_20260805_01",
"version": "1.0.0",
"display_name": "荔波旅游复杂测试模型",
"description": "用于手工验证图谱创建、Schema 校验、节点导入、复杂关系、图谱浏览和安全删除的综合测试模型。",
"enums": {
"Season": [
"spring",
"summer",
"autumn",
"winter",
"all_year"
],
"TipSeverity": [
"info",
"warning",
"critical"
],
"MealType": [
"breakfast",
"lunch",
"dinner"
]
},
"design_principles": [
"TourRoute 通过 RouteDay 表达有序的多日行程。",
"RouteDay 分别关联游览点、酒店、餐厅和交通枢纽。",
"ScenicArea 与 ScenicSpot 分层表达景区和景点。",
"票务政策和旅行提示作为独立实体，便于复用和版本管理。"
],
"entity_types": {
"ScenicArea": {
"label": "景区",
"description": "可包含多个具体景点的景区级实体。",
"primary_key": "name",
"fields": {
"name": {
"type": "string",
"required": true,
"description": "景区名称"
},
"level": {
"type": "string",
"description": "景区等级"
},
"rating": {
"type": "number",
"description": "综合评分"
},
"open": {
"type": "boolean",
"description": "是否开放"
},
"tags": {
"type": "array",
"description": "景区标签"
},
"geo": {
"type": "object",
"description": "经纬度对象"
},
"description": {
"type": "string"
}
}
},
"ScenicSpot": {
"label": "景点",
"description": "景区内可独立游览的景点。",
"primary_key": "name",
"fields": {
"name": {
"type": "string",
"required": true
},
"spot_type": {
"type": "string",
"required": true
},
"visit_minutes": {
"type": "integer"
},
"indoor": {
"type": "boolean"
},
"best_seasons": {
"type": "array"
},
"description": {
"type": "string"
}
}
},
"Hotel": {
"label": "酒店",
"description": "行程住宿资源。",
"primary_key": "name",
"fields": {
"name": {
"type": "string",
"required": true
},
"stars": {
"type": "integer",
"required": true
},
"price_from": {
"type": "number"
},
"family_friendly": {
"type": "boolean"
},
"amenities": {
"type": "array"
},
"address": {
"type": "string"
},
"contact": {
"type": "object"
}
}
},
"Restaurant": {
"label": "餐厅",
"description": "行程餐饮资源。",
"primary_key": "name",
"fields": {
"name": {
"type": "string",
"required": true
},
"cuisine": {
"type": "string",
"required": true
},
"avg_price": {
"type": "number"
},
"specialties": {
"type": "array"
},
"vegetarian_options": {
"type": "boolean"
},
"opening_hours": {
"type": "string"
}
}
},
"TransportHub": {
"label": "交通枢纽",
"description": "高铁站、汽车站或游客中心等交通节点。",
"primary_key": "name",
"fields": {
"name": {
"type": "string",
"required": true
},
"hub_type": {
"type": "string",
"required": true
},
"service_hours": {
"type": "string"
},
"accessible": {
"type": "boolean"
},
"facilities": {
"type": "array"
},
"geo": {
"type": "object"
}
}
},
"TourRoute": {
"label": "旅游线路",
"description": "由多个有序行程日组成的旅游产品线路。",
"primary_key": "name",
"fields": {
"name": {
"type": "string",
"required": true
},
"days": {
"type": "integer",
"required": true
},
"theme": {
"type": "string"
},
"suitable_for": {
"type": "array"
},
"base_price": {
"type": "number"
},
"active": {
"type": "boolean"
},
"extra_metadata": {
"type": "any",
"description": "用于验证 any 类型的扩展信息"
}
}
},
"RouteDay": {
"label": "行程日",
"description": "线路中的某一天。",
"fields": {
"title": {
"type": "string",
"required": true
},
"day_no": {
"type": "integer",
"required": true
},
"summary": {
"type": "string"
},
"estimated_cost": {
"type": "number"
}
}
},
"Activity": {
"label": "体验活动",
"description": "景点可提供的附加体验活动。",
"primary_key": "name",
"fields": {
"name": {
"type": "string",
"required": true
},
"activity_type": {
"type": "string",
"required": true
},
"duration_minutes": {
"type": "integer"
},
"min_age": {
"type": "integer"
},
"booking_required": {
"type": "boolean"
},
"equipment": {
"type": "array"
}
}
},
"TicketPolicy": {
"label": "票务政策",
"description": "景区或景点的票价和退改政策。",
"primary_key": "name",
"fields": {
"name": {
"type": "string",
"required": true
},
"adult_price": {
"type": "number",
"required": true
},
"child_price": {
"type": "number"
},
"free_policy": {
"type": "string"
},
"valid_from": {
"type": "date"
},
"refundable": {
"type": "boolean"
}
}
},
"TravelTip": {
"label": "旅行提示",
"description": "可应用到景区、资源或线路的提醒。",
"primary_key": "title",
"fields": {
"title": {
"type": "string",
"required": true
},
"content": {
"type": "string",
"required": true
},
"severity": {
"type": "string",
"required": true
},
"audiences": {
"type": "array"
},
"updated_at": {
"type": "datetime"
}
}
}
},
"relation_types": {
"CONTAINS": {
"label": "包含景点",
"from": "ScenicArea",
"to": "ScenicSpot",
"properties": {
"recommended_order": {
"type": "integer"
}
}
},
"NEARBY": {
"label": "附近资源",
"from": "ScenicSpot",
"to": "Hotel|Restaurant|TransportHub",
"properties": {
"distance_km": {
"type": "number",
"required": true
},
"drive_minutes": {
"type": "integer"
},
"walkable": {
"type": "boolean"
}
}
},
"ROUTE_HAS_DAY": {
"label": "线路包含行程日",
"from": "TourRoute",
"to": "RouteDay",
"properties": {
"order": {
"type": "integer",
"required": true
}
}
},
"DAY_VISITS": {
"label": "当日游览",
"from": "RouteDay",
"to": "ScenicSpot",
"properties": {
"order": {
"type": "integer",
"required": true
},
"stay_minutes": {
"type": "integer"
},
"must_visit": {
"type": "boolean"
}
}
},
"DAY_STAYS_AT": {
"label": "当日住宿",
"from": "RouteDay",
"to": "Hotel",
"properties": {
"check_in": {
"type": "string",
"required": true
},
"nights": {
"type": "integer"
}
}
},
"DAY_DINES_AT": {
"label": "当日用餐",
"from": "RouteDay",
"to": "Restaurant",
"properties": {
"meal": {
"type": "string",
"required": true
},
"reservation_required": {
"type": "boolean"
}
}
},
"DAY_USES_HUB": {
"label": "当日使用交通枢纽",
"from": "RouteDay",
"to": "TransportHub",
"properties": {
"mode": {
"type": "string",
"required": true
},
"departure_time": {
"type": "string"
}
}
},
"OFFERS_ACTIVITY": {
"label": "提供活动",
"from": "ScenicSpot",
"to": "Activity",
"properties": {
"season": {
"type": "string"
},
"extra_fee": {
"type": "number"
}
}
},
"HAS_TICKET_POLICY": {
"label": "采用票务政策",
"from": "ScenicArea|ScenicSpot",
"to": "TicketPolicy",
"properties": {
"channel": {
"type": "string"
}
}
},
"APPLIES_TIP": {
"label": "提示适用于",
"from": "TravelTip",
"to": "ScenicArea|ScenicSpot|Hotel|Restaurant|TransportHub|TourRoute",
"properties": {
"context": {
"type": "string"
},
"priority": {
"type": "integer"
}
}
},
"ROUTE_STARTS_AT": {
"label": "线路起点",
"from": "TourRoute",
"to": "TransportHub",
"properties": {
"departure_time": {
"type": "string",
"required": true
}
}
},
"ROUTE_ENDS_AT": {
"label": "线路终点",
"from": "TourRoute",
"to": "TransportHub",
"properties": {
"arrival_time": {
"type": "string",
"required": true
}
}
},
"ALTERNATIVE_TO": {
"label": "可替代资源",
"from": "Hotel|Restaurant",
"to": "Hotel|Restaurant",
"properties": {
"reason": {
"type": "string",
"required": true
},
"score": {
"type": "number"
}
}
},
"CONNECTS": {
"label": "交通连接",
"from": "TransportHub",
"to": "TransportHub",
"properties": {
"mode": {
"type": "string",
"required": true
},
"duration_minutes": {
"type": "integer",
"required": true
},
"distance_km": {
"type": "number"
}
}
},
"RECOMMENDS": {
"label": "线路推荐活动",
"from": "TourRoute",
"to": "Activity",
"properties": {
"priority": {
"type": "integer",
"required": true
},
"reason": {
"type": "string"
}
}
}
}
}

83
qa/graph-creation-fixtures/使用说明.md
View File

@@ -1,70 +1,35 @@
# 复杂图谱创建手工测试
# 云游荔波当前图谱导出与创建测试
这套数据只用于验证创建流程，价格、电话、地址和经营信息均为测试样例。
旧的 `libo_complex_schema.v1.json` 和 `libo_complex_graph_data.v1.json` 只有
32 个人工样例节点、70 条样例关系，并不是当前云游荔波项目的导出文件，现已移除。
## 创建时填写
## 生成当前图谱 JSON
- 项目名称（中文）：`荔波旅游复杂图谱手工测试`
- 项目标识（英文）：`libo_complex_manual_test_20260805_01`
确保本地 FalkorDB 中存在 `yunyou_libo` 图谱，然后运行：
## 操作顺序
1. 打开 `http://localhost:8102/admin/projects`，点击“创建项目”。
2. 填写上面的中文名称和英文标识，点击“下一步”。
3. 上传或粘贴 `libo_complex_schema.v1.json`。
4. 上传或粘贴 `libo_complex_graph_data.v1.json`。
5. 进入确认页后检查统计并提交创建。
## 预期统计
- 实体类型：10
- 关系类型：15
- 节点：32
- 关系：70
- Schema 版本：1.0.0
- 字段类型：覆盖 `string`、`integer`、`number`、`boolean`、`object`、`array`、`date`、`datetime`、`any`
## 创建后建议查询
全部节点：
```cypher
MATCH (n) RETURN n LIMIT 100
```bash
python3 scripts/export_yunyou_libo_full_graph_bundle.py \
--output-dir /安全的导出目录/yunyou_libo_graph_export
```
全部节点和关系：
导出器会生成：
```cypher
MATCH (n)-[r]->(m) RETURN n,r,m LIMIT 200
```
- `yunyou_libo_full_schema.v3.json`：创建向导“本体 Schema”步骤使用。
- `yunyou_libo_full_graph_data.v3.json`：创建向导“图谱数据”步骤使用。
- `yunyou_libo_full_bundle.v3.json`：包含 Schema、图谱数据和地图配置的归档文件。
- `yunyou_libo_full_manifest.v3.json`：数量、校验结果和 SHA-256 校验和。
线路、行程日与景点：
只有清单中的 `validation` 为 `passed`，且全部 `validation_checks` 为 `true` 时，
才可使用本次导出。
```cypher
MATCH (route:TourRoute)-[r1:ROUTE_HAS_DAY]->(day:RouteDay)-[r2:DAY_VISITS]->(spot:ScenicSpot)
RETURN route,r1,day,r2,spot LIMIT 100
```
## 创建测试项目
景点附近的酒店、餐厅和交通枢纽：
1. 打开 `/admin/projects`，点击“创建项目”。
2. 填写新的项目英文标识；被删除过的标识不能复用。
3. 地图选择“显示”。
4. 上传 `yunyou_libo_full_schema.v3.json`。
5. 上传 `yunyou_libo_full_graph_data.v3.json`。
6. 确认页面统计与清单一致后创建。
```cypher
MATCH (spot:ScenicSpot)-[r:NEARBY]->(resource)
RETURN spot,r,resource LIMIT 100
```
适合亲子的酒店：
```cypher
MATCH (hotel:Hotel)
WHERE hotel.family_friendly = true
RETURN hotel LIMIT 50
```
## 重新测试时注意
项目删除后系统会保留安全墓碑，原英文标识不能再次使用。重新测试时请同时修改：
- 项目标识，例如改为 `libo_complex_manual_test_20260805_02`
- Schema JSON 顶层的 `namespace`，改成相同的新标识
图谱数据文件无需修改。
导出文件直接快照当前 FalkorDB 图谱，保留节点的全部标签、属性、关系端点和关系属性。
地图业务 POI 与公交站坐标也包含在图谱数据中，不再从关系库重新拼装另一套图谱。

500
scripts/export_yunyou_libo_full_graph_bundle.py Normal file
View File

@@ -0,0 +1,500 @@
#!/usr/bin/env python3
"""Export the current Yunyou Libo FalkorDB graph as a lossless import bundle.
The live ``yunyou_libo`` graph is the authority for this export. Earlier
versions rebuilt a different graph from PostgreSQL rows: duplicate AMap POIs
were split, media and links were modeled twice, and unrelated custom tables
were imported automatically. That made the JSON totals differ from the graph
shown by the project.
This exporter snapshots every live node, label, property, relationship and
relationship property, then compares the JSON with independent FalkorDB count
queries before writing a passed manifest. Multi-label nodes are preserved so
shared hotel/food POIs and BusLine/BusRoute nodes do not change shape on import.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import math
import sys
from collections import Counter, defaultdict
from datetime import date, datetime, timezone
from decimal import Decimal
from pathlib import Path
from typing import Any, Iterable, Mapping
from uuid import UUID
from falkordb import FalkorDB
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from app.config import settings # noqa: E402
from app.project_lifecycle import normalize_provision_payload # noqa: E402
PROJECT_ID = "yunyou_libo"
GRAPH_NAME = "yunyou_libo"
DISPLAY_NAME = "云游荔波"
SCHEMA_VERSION = "3.0.0"
SPATIAL_MAP = {"enabled": True, "scope": "libo"}
LABEL_PRIORITY = (
"Hotel",
"FoodPlace",
"ScenicSpot",
"TransitFacility",
"BusStop",
"BusRoute",
"BusLine",
"GeoCell",
"ScenicArea",
"Area",
"Place",
)
NODE_LABELS = {
"Place": "地点",
"Hotel": "酒店",
"FoodPlace": "美食店铺",
"ScenicSpot": "景点",
"TransitFacility": "交通设施",
"BusStop": "公交站",
"BusLine": "公交线路",
"BusRoute": "公交运行方向",
"GeoCell": "H3 空间网格",
"ScenicArea": "景区片区",
"Area": "行政区域",
}
RELATION_LABELS = {
"LOCATED_IN": "位于行政区域",
"IN_H3_R9": "位于 H3 网格",
"STOPS_AT": "途经站点",
"NEXT_STOP": "下一站",
"PART_OF": "行政隶属",
"PART_OF_SCENIC_AREA": "属于景区片区",
}
MAP_POI_LABELS = {"Hotel", "FoodPlace", "ScenicSpot", "TransitFacility"}
def jsonable(value: Any) -> Any:
if value is None or isinstance(value, (str, bool, int)):
return value
if isinstance(value, float):
return value if math.isfinite(value) else str(value)
if isinstance(value, Decimal):
return int(value) if value == value.to_integral_value() else float(value)
if isinstance(value, (datetime, date)):
return value.isoformat()
if isinstance(value, UUID):
return str(value)
if isinstance(value, bytes):
return value.decode("utf-8", errors="replace")
if isinstance(value, Mapping):
return {str(key): jsonable(item) for key, item in value.items()}
if isinstance(value, (list, tuple, set)):
return [jsonable(item) for item in value]
return str(value)
def properties_of(value: Any) -> dict[str, Any]:
properties = getattr(value, "properties", None) or {}
return {str(key): jsonable(item) for key, item in dict(properties).items()}
def internal_id(value: Any) -> str:
identifier = getattr(value, "id", None)
if identifier is None:
raise RuntimeError("FalkorDB 返回了缺少内部 ID 的节点")
return str(identifier)
def labels_of(value: Any) -> list[str]:
labels = [str(item) for item in (getattr(value, "labels", None) or [])]
if not labels:
raise RuntimeError(f"FalkorDB 节点 {internal_id(value)} 没有标签")
order = {label: index for index, label in enumerate(LABEL_PRIORITY)}
return sorted(set(labels), key=lambda label: (order.get(label, len(order)), label))
def primary_label(labels: Iterable[str]) -> str:
label_set = set(labels)
for label in LABEL_PRIORITY:
if label in label_set and label not in {"Place", "BusLine"}:
return label
for label in LABEL_PRIORITY:
if label in label_set:
return label
return sorted(label_set)[0]
def relation_type_of(edge: Any) -> str:
relation_type = str(getattr(edge, "relation", "") or "").strip()
if not relation_type:
raise RuntimeError("FalkorDB 返回了缺少类型的关系")
return relation_type
def inferred_value_type(key: str, values: list[Any]) -> str:
populated = [value for value in values if value is not None]
if not populated:
return "any"
kinds: set[str] = set()
for value in populated:
if isinstance(value, bool):
kinds.add("boolean")
elif isinstance(value, int):
kinds.add("integer")
elif isinstance(value, float):
kinds.add("number")
elif isinstance(value, list):
kinds.add("array")
elif isinstance(value, dict):
kinds.add("object")
else:
kinds.add("string")
if kinds <= {"integer", "number"}:
return "number" if "number" in kinds else "integer"
if len(kinds) == 1:
only = next(iter(kinds))
if only == "string" and (key.endswith("_at") or key.endswith("_time")):
return "datetime"
return only
return "any"
def infer_fields(property_rows: Iterable[dict[str, Any]]) -> dict[str, dict[str, Any]]:
values: dict[str, list[Any]] = defaultdict(list)
for properties in property_rows:
for key, value in properties.items():
values[key].append(value)
return {
key: {"type": inferred_value_type(key, items), "required": False}
for key, items in sorted(values.items())
}
def make_client() -> FalkorDB:
options: dict[str, Any] = {
"host": settings.falkordb_host,
"port": settings.falkordb_port,
"socket_timeout": 60,
"socket_connect_timeout": 5,
}
if settings.falkordb_password:
options["password"] = settings.falkordb_password
return FalkorDB(**options)
def export_live_graph(graph: Any) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
nodes: list[dict[str, Any]] = []
exported_id_by_internal: dict[str, str] = {}
exported_ids: set[str] = set()
for row in graph.query("MATCH (n) RETURN n").result_set:
raw_node = row[0]
source_internal_id = internal_id(raw_node)
properties = properties_of(raw_node)
node_id = str(properties.get("__kg_node_id") or f"falkor:{source_internal_id}")
if node_id in exported_ids:
raise RuntimeError(f"图谱存在重复导出节点 ID：{node_id}")
labels = labels_of(raw_node)
exported_ids.add(node_id)
exported_id_by_internal[source_internal_id] = node_id
nodes.append(
{
"id": node_id,
"type": primary_label(labels),
"labels": labels,
"properties": properties,
}
)
relations: list[dict[str, Any]] = []
rows = graph.query("MATCH (source)-[relation]->(target) RETURN source, relation, target")
for source, edge, target in rows.result_set:
source_id = exported_id_by_internal.get(internal_id(source))
target_id = exported_id_by_internal.get(internal_id(target))
if not source_id or not target_id:
raise RuntimeError("关系引用了未导出的节点")
relations.append(
{
"type": relation_type_of(edge),
"source": source_id,
"target": target_id,
"properties": properties_of(edge),
}
)
return nodes, relations
def build_schema(
nodes: list[dict[str, Any]],
relations: list[dict[str, Any]],
) -> dict[str, Any]:
rows_by_label: dict[str, list[dict[str, Any]]] = defaultdict(list)
primary_type_by_id: dict[str, str] = {}
for item in nodes:
primary_type_by_id[item["id"]] = item["type"]
for label in item["labels"]:
rows_by_label[label].append(item["properties"])
relation_rows: dict[str, list[dict[str, Any]]] = defaultdict(list)
relation_endpoints: dict[str, tuple[set[str], set[str]]] = {}
for item in relations:
relation_rows[item["type"]].append(item["properties"])
sources, targets = relation_endpoints.setdefault(item["type"], (set(), set()))
sources.add(primary_type_by_id[item["source"]])
targets.add(primary_type_by_id[item["target"]])
entity_types = {}
for label, rows in sorted(rows_by_label.items()):
entity_types[label] = {
"label": NODE_LABELS.get(label, label),
"primary_key": "element_id" if any("element_id" in row for row in rows) else "id",
"fields": infer_fields(rows),
}
relation_types = {}
for relation_type, rows in sorted(relation_rows.items()):
sources, targets = relation_endpoints[relation_type]
relation_types[relation_type] = {
"label": RELATION_LABELS.get(relation_type, relation_type),
"from": "|".join(sorted(sources)),
"to": "|".join(sorted(targets)),
"properties": infer_fields(rows),
}
return {
"namespace": PROJECT_ID,
"version": SCHEMA_VERSION,
"display_name": "云游荔波当前图谱快照 Schema v3",
"description": "从当前 yunyou_libo FalkorDB 图谱无损导出，保留多标签节点。",
"entity_types": entity_types,
"relation_types": relation_types,
}
def label_set_key(labels: Iterable[str]) -> str:
return "+".join(sorted(labels))
def exported_counts(
nodes: list[dict[str, Any]],
relations: list[dict[str, Any]],
) -> dict[str, Any]:
label_counts: Counter[str] = Counter()
label_set_counts: Counter[str] = Counter()
primary_type_counts: Counter[str] = Counter()
relation_type_counts: Counter[str] = Counter()
coordinate_nodes = 0
map_poi_nodes = 0
bus_stop_nodes = 0
for item in nodes:
labels = set(item["labels"])
label_counts.update(labels)
label_set_counts[label_set_key(labels)] += 1
primary_type_counts[item["type"]] += 1
properties = item["properties"]
has_coordinates = properties.get("lng") is not None and properties.get("lat") is not None
if has_coordinates:
coordinate_nodes += 1
if labels & MAP_POI_LABELS:
map_poi_nodes += 1
if "BusStop" in labels:
bus_stop_nodes += 1
for item in relations:
relation_type_counts[item["type"]] += 1
return {
"nodes": len(nodes),
"relations": len(relations),
"primary_type_counts": dict(sorted(primary_type_counts.items())),
"label_counts": dict(sorted(label_counts.items())),
"label_set_counts": dict(sorted(label_set_counts.items())),
"relation_type_counts": dict(sorted(relation_type_counts.items())),
"coordinate_nodes": coordinate_nodes,
"map_poi_nodes": map_poi_nodes,
"bus_stop_nodes": bus_stop_nodes,
}
def source_counts(graph: Any) -> dict[str, Any]:
node_count = int(graph.query("MATCH (n) RETURN count(n)").result_set[0][0])
relation_count = int(graph.query("MATCH ()-[r]->() RETURN count(r)").result_set[0][0])
coordinate_count = int(
graph.query(
"MATCH (n) WHERE n.lng IS NOT NULL AND n.lat IS NOT NULL RETURN count(n)"
).result_set[0][0]
)
label_counts: Counter[str] = Counter()
label_set_counts: Counter[str] = Counter()
for labels, count in graph.query(
"MATCH (n) RETURN labels(n), count(n) ORDER BY count(n) DESC"
).result_set:
normalized = [str(label) for label in labels]
amount = int(count)
for label in normalized:
label_counts[label] += amount
label_set_counts[label_set_key(normalized)] += amount
relation_type_counts = {
str(relation_type): int(count)
for relation_type, count in graph.query(
"MATCH ()-[r]->() RETURN type(r), count(r) ORDER BY type(r)"
).result_set
}
return {
"nodes": node_count,
"relations": relation_count,
"label_counts": dict(sorted(label_counts.items())),
"label_set_counts": dict(sorted(label_set_counts.items())),
"relation_type_counts": dict(sorted(relation_type_counts.items())),
"coordinate_nodes": coordinate_count,
}
def validate_snapshot(
source: dict[str, Any],
exported: dict[str, Any],
nodes: list[dict[str, Any]],
relations: list[dict[str, Any]],
) -> dict[str, bool]:
node_ids = [item["id"] for item in nodes]
node_id_set = set(node_ids)
checks = {
"node_count_matches_live_graph": exported["nodes"] == source["nodes"],
"relation_count_matches_live_graph": exported["relations"] == source["relations"],
"label_counts_match_live_graph": exported["label_counts"] == source["label_counts"],
"label_sets_match_live_graph": exported["label_set_counts"] == source["label_set_counts"],
"relation_types_match_live_graph": (
exported["relation_type_counts"] == source["relation_type_counts"]
),
"coordinate_count_matches_live_graph": (
exported["coordinate_nodes"] == source["coordinate_nodes"]
),
"node_ids_are_unique": len(node_ids) == len(node_id_set),
"all_relation_endpoints_exist": all(
item["source"] in node_id_set and item["target"] in node_id_set
for item in relations
),
}
failed = [name for name, passed in checks.items() if not passed]
if failed:
raise RuntimeError("图谱快照校验失败：" + "、".join(failed))
return checks
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def dump_json(path: Path, value: Any, *, pretty: bool = False) -> None:
with path.open("w", encoding="utf-8") as handle:
if pretty:
json.dump(value, handle, ensure_ascii=False, indent=2)
else:
json.dump(value, handle, ensure_ascii=False, separators=(",", ":"))
handle.write("\n")
def export(output_dir: Path) -> dict[str, Any]:
output_dir.mkdir(parents=True, exist_ok=True)
generated_at = datetime.now(timezone.utc).isoformat()
client = make_client()
try:
graph_names = {
item.decode("utf-8") if isinstance(item, bytes) else str(item)
for item in client.list_graphs()
}
if GRAPH_NAME not in graph_names:
raise RuntimeError(f"FalkorDB 中不存在图谱 {GRAPH_NAME!r}")
graph = client.select_graph(GRAPH_NAME)
live_counts = source_counts(graph)
nodes, relations = export_live_graph(graph)
finally:
client.close()
counts = exported_counts(nodes, relations)
checks = validate_snapshot(live_counts, counts, nodes, relations)
schema = build_schema(nodes, relations)
graph_data = {
"_bundle": {
"format": "znkg-falkordb-snapshot-v3",
"source": "falkordb",
"project_id": PROJECT_ID,
"graph_name": GRAPH_NAME,
"generated_at": generated_at,
"spatial_map": SPATIAL_MAP,
"source_snapshot_counts": live_counts,
},
"nodes": nodes,
"relations": relations,
}
provision = {
"format": "znkg-project-bundle-v3",
"project_id": PROJECT_ID,
"display_name": DISPLAY_NAME,
"spatial_map": SPATIAL_MAP,
"schema": schema,
"graph_data": graph_data,
}
normalized = normalize_provision_payload(provision)
if normalized["counts"]["nodes"] != counts["nodes"]:
raise RuntimeError("后端校验后的节点数量不一致")
if normalized["counts"]["relations"] != counts["relations"]:
raise RuntimeError("后端校验后的关系数量不一致")
schema_path = output_dir / "yunyou_libo_full_schema.v3.json"
graph_path = output_dir / "yunyou_libo_full_graph_data.v3.json"
bundle_path = output_dir / "yunyou_libo_full_bundle.v3.json"
manifest_path = output_dir / "yunyou_libo_full_manifest.v3.json"
dump_json(schema_path, schema, pretty=True)
dump_json(graph_path, graph_data)
dump_json(bundle_path, provision)
files = {
path.name: {"bytes": path.stat().st_size, "sha256": sha256_file(path)}
for path in (schema_path, graph_path, bundle_path)
}
manifest = {
"format": "znkg-full-graph-manifest-v3",
"project_id": PROJECT_ID,
"graph_name": GRAPH_NAME,
"schema_version": SCHEMA_VERSION,
"generated_at": generated_at,
"source": "falkordb",
"spatial_map": SPATIAL_MAP,
"source_snapshot_counts": live_counts,
"export_counts": counts,
"validation": "passed",
"validation_checks": checks,
"files": files,
}
dump_json(manifest_path, manifest, pretty=True)
manifest["manifest_file"] = str(manifest_path)
return manifest
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--output-dir",
type=Path,
required=True,
help="Directory for schema, graph-data, bundle and manifest JSON files.",
)
args = parser.parse_args()
manifest = export(args.output_dir.expanduser().resolve())
print(json.dumps(manifest, ensure_ascii=False, indent=2))
if __name__ == "__main__":
main()

46
tests/test_project_lifecycle.py
View File

@@ -75,6 +75,29 @@ class ProjectPayloadValidationTests(unittest.TestCase):
"number",
)
def test_multi_label_nodes_are_validated_and_preserved(self) -> None:
body = valid_payload()
body["schema"]["entity_types"]["Hotel"] = {
"label": "酒店",
"fields": {"name": {"type": "string", "required": True}},
}
body["schema"]["relation_types"]["RELATED_TO"]["from"] = "Hotel|Place"
body["graph_data"]["nodes"][0]["type"] = "Hotel"
body["graph_data"]["nodes"][0]["labels"] = ["Place", "Hotel", "Place"]
payload = normalize_provision_payload(body)
self.assertEqual(payload["graph_data"]["nodes"][0]["labels"], ["Place", "Hotel"])
def test_multi_label_nodes_reject_undefined_labels(self) -> None:
body = valid_payload()
body["graph_data"]["nodes"][0]["labels"] = ["Place", "UnknownLabel"]
with self.assertRaises(ProjectValidationError) as raised:
normalize_provision_payload(body)
self.assertIn("UnknownLabel", "；".join(raised.exception.errors))
def test_hidden_resource_ids_default_to_project_id(self) -> None:
body = valid_payload()
body.pop("tenant_id")
@@ -382,6 +405,29 @@ class FalkorLifecycleTests(unittest.TestCase):
self.assertFalse(graph.deleted)
self.assertTrue(database.closed)
def test_import_preserves_all_validated_node_labels(self) -> None:
graph_data = {
"nodes": [
{
"id": "hotel-food-1",
"type": "Hotel",
"labels": ["Place", "FoodPlace", "Hotel"],
"properties": {"name": "住宿与餐饮复合 POI"},
}
],
"relations": [],
}
graph = _FakeGraph(node_count=1, relation_count=0)
database = _FakeDb(graph)
with patch("app.project_lifecycle._falkor_client", return_value=database):
counts = import_falkor_graph("multi_label_graph", graph_data)
self.assertEqual(counts, {"nodes": 1, "relations": 0})
self.assertTrue(
any("CREATE (n:Place:FoodPlace:Hotel)" in query for query, _ in graph.queries)
)
def test_import_failure_compensates_only_the_new_graph(self) -> None:
payload = normalize_provision_payload(valid_payload())
graph = _FakeGraph(fail_on_query=2)