feat: implement project relational data platform

app/data_platform/__init__.py

@@ -0,0 +1,2 @@
+"""Project-isolated PostgreSQL data platform."""
+

app/data_platform/migrate_graph.py

@@ -0,0 +1,486 @@
+"""One-time migration of approved Yunyou Libo graph data into PostgreSQL.
+
+The graph remains available for visualization.  This module creates the
+relational authority records that the generic data center can manage.  It is
+idempotent: record UUIDs are derived from graph identities and every write is
+an upsert.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import uuid
+from collections.abc import Iterable
+from typing import Any
+
+import psycopg
+from falkordb import FalkorDB
+from psycopg import sql
+from psycopg.rows import dict_row
+from psycopg.types.json import Jsonb
+
+from app.config import settings
+from app.data_platform.schema import project_schema_name
+
+PROJECT_ID = "yunyou_libo"
+GRAPH_NAME = "yunyou_libo"
+UUID_NAMESPACE = uuid.UUID("f9a34c5d-9992-4958-adf8-7495f3251a5d")
+
+ENTITY_SPECS = (
+    ("Hotel", "hotel", "酒店"),
+    ("FoodPlace", "restaurant", "美食"),
+    ("ScenicSpot", "scenic", "景区"),
+    ("TransitFacility", "transport", "交通"),
+    ("BusStop", "bus_stop", "公交站"),
+)
+
+
+def stable_uuid(*parts: Any) -> uuid.UUID:
+    return uuid.uuid5(UUID_NAMESPACE, ":".join(str(part or "") for part in parts))
+
+
+def text(value: Any) -> str | None:
+    if value is None:
+        return None
+    result = str(value).strip()
+    return result or None
+
+
+def number(value: Any) -> float | None:
+    if value in (None, ""):
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def integer(value: Any) -> int | None:
+    parsed = number(value)
+    return int(parsed) if parsed is not None else None
+
+
+def json_value(value: Any, fallback: Any) -> Any:
+    if value in (None, ""):
+        return fallback
+    if isinstance(value, (dict, list)):
+        return value
+    try:
+        return json.loads(str(value))
+    except (TypeError, ValueError, json.JSONDecodeError):
+        return fallback
+
+
+def category_parts(props: dict[str, Any], fallback: str) -> tuple[str, str | None, str | None]:
+    raw_parts = [part.strip() for part in str(props.get("amap_type") or "").split(";") if part.strip()]
+    return (
+        text(props.get("amap_category_l1")) or (raw_parts[0] if raw_parts else fallback),
+        text(props.get("amap_category_l2")) or (raw_parts[1] if len(raw_parts) > 1 else None),
+        text(props.get("business_subcategory"))
+        or text(props.get("amap_category_l3"))
+        or (raw_parts[2] if len(raw_parts) > 2 else None),
+    )
+
+
+def graph_identity(props: dict[str, Any]) -> str:
+    return str(
+        props.get("element_id")
+        or props.get("place_id")
+        or props.get("gaode_poi_id")
+        or props.get("route_id")
+        or props.get("name")
+        or uuid.uuid4()
+    )
+
+
+def upsert(
+    cur: psycopg.Cursor,
+    schema_name: str,
+    table_name: str,
+    values: dict[str, Any],
+) -> None:
+    columns = list(values)
+    assignments = [
+        sql.SQL("{}=EXCLUDED.{}").format(sql.Identifier(column), sql.Identifier(column))
+        for column in columns
+        if column not in {"id", "created_at"}
+    ]
+    assignments.extend(
+        (
+            sql.SQL("updated_at=now()"),
+            sql.SQL("deleted_at=NULL"),
+            sql.SQL("deleted_by=NULL"),
+        )
+    )
+    cur.execute(
+        sql.SQL(
+            "INSERT INTO {}.{} ({}) VALUES ({}) "
+            "ON CONFLICT (id) DO UPDATE SET {}"
+        ).format(
+            sql.Identifier(schema_name),
+            sql.Identifier(table_name),
+            sql.SQL(", ").join(sql.Identifier(column) for column in columns),
+            sql.SQL(", ").join(sql.Placeholder() for _ in columns),
+            sql.SQL(", ").join(assignments),
+        ),
+        list(values.values()),
+    )
+
+
+def graph_rows(graph, label: str, batch_size: int = 400) -> Iterable[dict[str, Any]]:
+    """Read a label in bounded pages so large property payloads do not time out."""
+    offset = 0
+    while True:
+        rows = graph.query(
+            f"MATCH (n:{label}) RETURN properties(n) SKIP {offset} LIMIT {batch_size}",
+            timeout=120_000,
+        ).result_set
+        for row in rows:
+            yield dict(row[0])
+        if len(rows) < batch_size:
+            break
+        offset += batch_size
+
+
+def compact_source_data(props: dict[str, Any], graph_label: str) -> Jsonb:
+    return Jsonb(
+        {
+            "graph_name": GRAPH_NAME,
+            "graph_label": graph_label,
+            "graph_element_id": text(props.get("element_id")),
+            "place_id": text(props.get("place_id")),
+            "source": text(props.get("source")),
+            "source_name": text(props.get("source_name")),
+            "typecode": text(props.get("typecode")),
+            "audit_result": text(props.get("audit_result")),
+            "audit_confidence": text(props.get("audit_confidence")),
+        }
+    )
+
+
+def insert_external_link(
+    cur: psycopg.Cursor,
+    schema_name: str,
+    tenant_id: str,
+    entity_id: uuid.UUID,
+    platform: str,
+    external_id: Any,
+    external_name: Any,
+    external_url: Any,
+) -> bool:
+    identifier = text(external_id)
+    if not identifier:
+        return False
+    upsert(
+        cur,
+        schema_name,
+        "entity_external_links",
+        {
+            "id": stable_uuid("external", entity_id, platform, identifier),
+            "tenant_id": tenant_id,
+            "project_id": PROJECT_ID,
+            "entity_id": entity_id,
+            "platform": platform,
+            "external_id": identifier,
+            "external_name": text(external_name),
+            "external_url": text(external_url),
+        },
+    )
+    return True
+
+
+def migrate_entities(
+    cur: psycopg.Cursor,
+    graph,
+    schema_name: str,
+    tenant_id: str,
+) -> dict[str, int]:
+    counts = {
+        "poi_entities": 0,
+        "entity_external_links": 0,
+        "entity_images": 0,
+        "hotel_profiles": 0,
+        "restaurant_profiles": 0,
+        "scenic_profiles": 0,
+        "transport_profiles": 0,
+    }
+    for graph_label, entity_type, category_fallback in ENTITY_SPECS:
+        for props in graph_rows(graph, graph_label):
+            identity = graph_identity(props)
+            entity_id = stable_uuid(PROJECT_ID, entity_type, identity)
+            category_l1, category_l2, category_l3 = category_parts(props, category_fallback)
+            entity_name = text(props.get("display_name")) or text(props.get("name")) or identity
+            upsert(
+                cur,
+                schema_name,
+                "poi_entities",
+                {
+                    "id": entity_id,
+                    "tenant_id": tenant_id,
+                    "project_id": PROJECT_ID,
+                    "entity_type": entity_type,
+                    "name": entity_name,
+                    "category_l1": category_l1,
+                    "category_l2": category_l2,
+                    "category_l3": category_l3,
+                    "address": text(props.get("address")),
+                    "district": text(props.get("district")) or "荔波县",
+                    "adcode": text(props.get("adcode")),
+                    "phone": text(props.get("tel")),
+                    "longitude": number(props.get("lng")),
+                    "latitude": number(props.get("lat")),
+                    "h3_r9": text(props.get("h3_r9")),
+                    "h3_r10": text(props.get("h3_r10")),
+                    "status": "active",
+                    "version": 1,
+                    "extra_data": compact_source_data(props, graph_label),
+                },
+            )
+            counts["poi_entities"] += 1
+
+            if insert_external_link(
+                cur,
+                schema_name,
+                tenant_id,
+                entity_id,
+                "amap",
+                props.get("gaode_poi_id") or props.get("place_id"),
+                entity_name,
+                props.get("amap_url"),
+            ):
+                counts["entity_external_links"] += 1
+
+            cover_image = text(props.get("cover_image_url"))
+            if cover_image:
+                upsert(
+                    cur,
+                    schema_name,
+                    "entity_images",
+                    {
+                        "id": stable_uuid("image", entity_id, "cover", cover_image),
+                        "tenant_id": tenant_id,
+                        "project_id": PROJECT_ID,
+                        "entity_id": entity_id,
+                        "owner_type": "entity",
+                        "owner_id": None,
+                        "image_url": cover_image,
+                        "caption": "封面图",
+                        "display_order": 0,
+                    },
+                )
+                counts["entity_images"] += 1
+
+            if entity_type == "hotel" and props.get("ctrip_hotel_id"):
+                ctrip_id = props.get("ctrip_hotel_id")
+                if insert_external_link(
+                    cur,
+                    schema_name,
+                    tenant_id,
+                    entity_id,
+                    "ctrip",
+                    ctrip_id,
+                    props.get("ctrip_name_cn"),
+                    props.get("ctrip_url"),
+                ):
+                    counts["entity_external_links"] += 1
+                upsert(
+                    cur,
+                    schema_name,
+                    "hotel_profiles",
+                    {
+                        "id": stable_uuid("hotel-profile", entity_id),
+                        "tenant_id": tenant_id,
+                        "project_id": PROJECT_ID,
+                        "entity_id": entity_id,
+                        "ctrip_name": text(props.get("ctrip_name_cn")),
+                        "opened_year": integer(props.get("ctrip_opened_year")),
+                        "room_count": integer(props.get("ctrip_room_count")),
+                        "diamond_level": integer(props.get("ctrip_diamond_level")),
+                        "ctrip_rating": number(props.get("ctrip_rating")),
+                        "review_count": integer(props.get("ctrip_review_count")),
+                        "reference_price": number(props.get("room_min_price") or props.get("offer_min_price")),
+                        "introduction": text(props.get("ctrip_description")),
+                    },
+                )
+                counts["hotel_profiles"] += 1
+
+            if entity_type == "restaurant" and props.get("dianping_shop_id"):
+                dianping_id = props.get("dianping_shop_id")
+                if insert_external_link(
+                    cur,
+                    schema_name,
+                    tenant_id,
+                    entity_id,
+                    "dianping",
+                    dianping_id,
+                    props.get("dianping_name"),
+                    props.get("dianping_url"),
+                ):
+                    counts["entity_external_links"] += 1
+                upsert(
+                    cur,
+                    schema_name,
+                    "restaurant_profiles",
+                    {
+                        "id": stable_uuid("restaurant-profile", entity_id),
+                        "tenant_id": tenant_id,
+                        "project_id": PROJECT_ID,
+                        "entity_id": entity_id,
+                        "dianping_name": text(props.get("dianping_name")),
+                        "dianping_category": text(props.get("dianping_category")),
+                        "ranking_text": text(props.get("dianping_ranking")),
+                        "business_status": text(props.get("dianping_business_status")) or "unknown",
+                        "business_hours": text(props.get("dianping_business_hours")),
+                        "rating": number(props.get("dianping_rating")),
+                        "review_count": integer(props.get("dianping_review_count")),
+                        "average_price": number(props.get("dianping_avg_price")),
+                    },
+                )
+                counts["restaurant_profiles"] += 1
+
+            if entity_type == "scenic":
+                scenic_level = text(props.get("scenic_level")) or text(props.get("scenic_grade"))
+                upsert(
+                    cur,
+                    schema_name,
+                    "scenic_profiles",
+                    {
+                        "id": stable_uuid("scenic-profile", entity_id),
+                        "tenant_id": tenant_id,
+                        "project_id": PROJECT_ID,
+                        "entity_id": entity_id,
+                        "scenic_type": text(props.get("scenic_type")) or category_l3,
+                        "scenic_level": scenic_level,
+                        "is_national": bool(scenic_level and "国家" in scenic_level),
+                        "visitor_value_type": text(props.get("visitor_value")),
+                        "opening_hours": text(props.get("open_time")),
+                        "ticket_note": text(props.get("cost")),
+                        "official_intro": None,
+                    },
+                )
+                counts["scenic_profiles"] += 1
+
+            if entity_type in {"transport", "bus_stop"}:
+                upsert(
+                    cur,
+                    schema_name,
+                    "transport_profiles",
+                    {
+                        "id": stable_uuid("transport-profile", entity_id),
+                        "tenant_id": tenant_id,
+                        "project_id": PROJECT_ID,
+                        "entity_id": entity_id,
+                        "transport_type": text(props.get("station_type")) or category_l3 or category_l2 or category_l1,
+                        "service_hours": text(props.get("open_time")),
+                        "route_note": text(props.get("category")),
+                    },
+                )
+                counts["transport_profiles"] += 1
+    return counts
+
+
+def migrate_bus_routes(
+    cur: psycopg.Cursor,
+    graph,
+    schema_name: str,
+    tenant_id: str,
+) -> dict[str, int]:
+    counts = {"bus_routes": 0, "bus_route_stops": 0}
+    for props in graph_rows(graph, "BusRoute"):
+        identity = graph_identity(props)
+        route_id = stable_uuid(PROJECT_ID, "bus_route", identity)
+        service_hours = "—".join(
+            part for part in (text(props.get("first_bus")), text(props.get("last_bus"))) if part
+        ) or None
+        upsert(
+            cur,
+            schema_name,
+            "bus_routes",
+            {
+                "id": route_id,
+                "tenant_id": tenant_id,
+                "project_id": PROJECT_ID,
+                "route_name": text(props.get("line_name")) or text(props.get("name")) or identity,
+                "direction_name": text(props.get("direction")),
+                "start_stop_name": text(props.get("start_stop")),
+                "end_stop_name": text(props.get("end_stop")),
+                "service_hours": service_hours,
+                "route_color": None,
+                "geometry": Jsonb([]),
+            },
+        )
+        counts["bus_routes"] += 1
+
+    rows = graph.query(
+        "MATCH (r:BusRoute)-[e:STOPS_AT]->(s:BusStop) "
+        "RETURN r.element_id, r.route_id, e.sequence, "
+        "s.element_id, s.place_id, s.name, s.lng, s.lat",
+        timeout=120_000,
+    ).result_set
+    for row in rows:
+        graph_route_id = row[0] or row[1]
+        graph_stop_id = row[3] or row[4] or row[5]
+        route_id = stable_uuid(PROJECT_ID, "bus_route", graph_route_id)
+        stop_entity_id = stable_uuid(PROJECT_ID, "bus_stop", graph_stop_id)
+        stop_order = integer(row[2]) or 0
+        upsert(
+            cur,
+            schema_name,
+            "bus_route_stops",
+            {
+                "id": stable_uuid("route-stop", route_id, stop_entity_id, stop_order),
+                "tenant_id": tenant_id,
+                "project_id": PROJECT_ID,
+                "route_id": route_id,
+                "stop_entity_id": stop_entity_id,
+                "stop_name": text(row[5]) or str(graph_stop_id),
+                "stop_order": stop_order,
+                "longitude": number(row[6]),
+                "latitude": number(row[7]),
+            },
+        )
+        counts["bus_route_stops"] += 1
+    return counts
+
+
+def migrate() -> dict[str, int]:
+    graph = FalkorDB(
+        host=settings.falkordb_host,
+        port=settings.falkordb_port,
+        password=settings.falkordb_password or None,
+    ).select_graph(GRAPH_NAME)
+    schema_name = project_schema_name(PROJECT_ID)
+    with psycopg.connect(settings.database_url, row_factory=dict_row) as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                sql.SQL(
+                    "SELECT tenant_id FROM {}.projects WHERE project_id=%s AND status <> 'archived'"
+                ).format(sql.Identifier(settings.db_schema)),
+                (PROJECT_ID,),
+            )
+            project = cur.fetchone()
+            if not project:
+                raise RuntimeError(f"Project not found: {PROJECT_ID}")
+            tenant_id = str(project["tenant_id"])
+            counts = migrate_entities(cur, graph, schema_name, tenant_id)
+            counts.update(migrate_bus_routes(cur, graph, schema_name, tenant_id))
+        conn.commit()
+    return counts
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="Execute the idempotent migration. Without this flag only help is shown.",
+    )
+    args = parser.parse_args()
+    if not args.apply:
+        parser.print_help()
+        return
+    print(json.dumps(migrate(), ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()

app/data_platform/record_service.py

@@ -0,0 +1,485 @@
+from __future__ import annotations
+
+import json
+import uuid
+from decimal import Decimal
+from typing import Any
+
+from fastapi import HTTPException
+from psycopg import sql
+from psycopg.types.json import Jsonb
+
+from app.config import settings
+from app.data_platform.registry import TABLE_DEFINITIONS, TABLE_REGISTRY, FieldDefinition, TableDefinition
+from app.data_platform.schema import (
+    ensure_all_project_databases,
+    ensure_project_database,
+    get_project_database,
+)
+from app.db import get_conn
+
+
+def _json_safe(value: Any) -> Any:
+    if isinstance(value, Decimal):
+        return float(value)
+    if isinstance(value, uuid.UUID):
+        return str(value)
+    if hasattr(value, "isoformat"):
+        return value.isoformat()
+    return value
+
+
+def _row_json(row: dict[str, Any] | None) -> dict[str, Any] | None:
+    if row is None:
+        return None
+    return {key: _json_safe(value) for key, value in dict(row).items()}
+
+
+def _table_or_404(table_code: str) -> TableDefinition:
+    table = TABLE_REGISTRY.get(table_code)
+    if not table:
+        raise HTTPException(404, "数据表未注册")
+    return table
+
+
+async def _database_or_404(project_id: str) -> dict[str, Any]:
+    database = await get_project_database(project_id)
+    if database:
+        return database
+    async with get_conn() as conn:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                sql.SQL(
+                    "SELECT project_id, tenant_id, display_name FROM {}.projects "
+                    "WHERE project_id=%s AND status <> 'archived'"
+                ).format(sql.Identifier(settings.db_schema)),
+                (project_id,),
+            )
+            project = await cur.fetchone()
+    if not project:
+        raise HTTPException(404, "项目数据库不存在")
+    return await ensure_project_database(
+        str(project["project_id"]),
+        str(project["tenant_id"]),
+        str(project["display_name"]),
+    )
+
+
+def _coerce_value(field: FieldDefinition, value: Any) -> Any:
+    if value in ("", None):
+        return None
+    if field.data_type == "json":
+        if isinstance(value, str):
+            try:
+                value = json.loads(value)
+            except json.JSONDecodeError as exc:
+                raise HTTPException(400, f"{field.label}不是合法 JSON") from exc
+        return Jsonb(value)
+    if field.data_type == "boolean":
+        if isinstance(value, bool):
+            return value
+        return str(value).lower() in {"true", "1", "yes", "是"}
+    if field.data_type == "number":
+        if "INTEGER" in field.sql_type:
+            try:
+                return int(value)
+            except (TypeError, ValueError) as exc:
+                raise HTTPException(400, f"{field.label}必须是整数") from exc
+        try:
+            return Decimal(str(value))
+        except Exception as exc:
+            raise HTTPException(400, f"{field.label}必须是数字") from exc
+    return value
+
+
+def _validated_payload(
+    table: TableDefinition,
+    body: dict[str, Any],
+    *,
+    create: bool,
+) -> dict[str, Any]:
+    definitions = {field.code: field for field in table.fields}
+    unknown = sorted(set(body) - set(definitions))
+    if unknown:
+        raise HTTPException(400, f"不允许的字段：{', '.join(unknown)}")
+    payload: dict[str, Any] = {}
+    for code, value in body.items():
+        field = definitions[code]
+        if not field.editable:
+            continue
+        payload[code] = _coerce_value(field, value)
+    if create:
+        missing = [
+            field.label
+            for field in table.fields
+            if field.required and payload.get(field.code) in (None, "")
+        ]
+        if missing:
+            raise HTTPException(400, f"缺少必填字段：{', '.join(missing)}")
+    return payload
+
+
+async def list_databases() -> list[dict[str, Any]]:
+    await ensure_all_project_databases()
+    admin_schema = sql.Identifier(settings.db_schema)
+    async with get_conn() as conn:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                sql.SQL(
+                    """
+                    SELECT d.*, p.status AS project_status
+                    FROM {}.project_databases d
+                    JOIN {}.projects p ON p.project_id=d.project_id
+                    WHERE p.status <> 'archived'
+                    ORDER BY
+                        CASE WHEN d.project_id='yunyou_libo' THEN 0 ELSE 1 END,
+                        p.created_at DESC
+                    """
+                ).format(admin_schema, admin_schema)
+            )
+            databases = await cur.fetchall()
+            result: list[dict[str, Any]] = []
+            for database in databases:
+                schema_name = str(database["schema_name"])
+                table_counts: list[int] = []
+                latest_updates: list[Any] = []
+                for table in TABLE_DEFINITIONS:
+                    await cur.execute(
+                        sql.SQL(
+                            "SELECT count(*) AS count, max(updated_at) AS updated_at "
+                            "FROM {}.{} WHERE deleted_at IS NULL"
+                        ).format(
+                            sql.Identifier(schema_name),
+                            sql.Identifier(table.code),
+                        )
+                    )
+                    summary = await cur.fetchone()
+                    table_counts.append(int(summary["count"] or 0))
+                    if summary["updated_at"]:
+                        latest_updates.append(summary["updated_at"])
+                row = _row_json(dict(database)) or {}
+                row.update(
+                    {
+                        "table_count": len(TABLE_DEFINITIONS),
+                        "record_count": sum(table_counts),
+                        "updated_at": _json_safe(max(latest_updates)) if latest_updates else _json_safe(database["updated_at"]),
+                    }
+                )
+                result.append(row)
+            return result
+
+
+async def list_tables(project_id: str) -> dict[str, Any]:
+    database = await _database_or_404(project_id)
+    schema_name = str(database["schema_name"])
+    async with get_conn() as conn:
+        async with conn.cursor() as cur:
+            tables: list[dict[str, Any]] = []
+            for definition in TABLE_DEFINITIONS:
+                await cur.execute(
+                    sql.SQL(
+                        "SELECT count(*) AS count, max(updated_at) AS updated_at "
+                        "FROM {}.{} WHERE deleted_at IS NULL"
+                    ).format(
+                        sql.Identifier(schema_name),
+                        sql.Identifier(definition.code),
+                    )
+                )
+                summary = await cur.fetchone()
+                item = definition.as_dict()
+                item.update(
+                    {
+                        "record_count": int(summary["count"] or 0),
+                        "updated_at": _json_safe(summary["updated_at"]),
+                    }
+                )
+                tables.append(item)
+    return {"database": _row_json(database), "tables": tables}
+
+
+async def list_records(
+    project_id: str,
+    table_code: str,
+    *,
+    page: int,
+    page_size: int,
+    search: str | None,
+    sort_field: str | None,
+    sort_order: str,
+) -> dict[str, Any]:
+    table = _table_or_404(table_code)
+    database = await _database_or_404(project_id)
+    schema_name = str(database["schema_name"])
+    page = max(1, page)
+    page_size = max(1, min(200, page_size))
+    offset = (page - 1) * page_size
+    allowed_sort = {"id", "created_at", "updated_at", *(field.code for field in table.fields if field.sortable)}
+    order_field = sort_field if sort_field in allowed_sort else "updated_at"
+    order_keyword = sql.SQL("ASC") if sort_order.lower() == "asc" else sql.SQL("DESC")
+
+    where_parts: list[sql.Composable] = [sql.SQL("deleted_at IS NULL")]
+    params: list[Any] = []
+    searchable = [field for field in table.fields if field.searchable]
+    if search and searchable:
+        pattern = f"%{search.strip()}%"
+        where_parts.append(
+            sql.SQL("(")
+            + sql.SQL(" OR ").join(
+                sql.SQL("{}::text ILIKE %s").format(sql.Identifier(field.code))
+                for field in searchable
+            )
+            + sql.SQL(")")
+        )
+        params.extend(pattern for _ in searchable)
+    where_clause = sql.SQL(" AND ").join(where_parts)
+    columns = ["id", *(field.code for field in table.fields), "created_at", "updated_at"]
+
+    async with get_conn() as conn:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                sql.SQL("SELECT count(*) AS count FROM {}.{} WHERE {}").format(
+                    sql.Identifier(schema_name),
+                    sql.Identifier(table.code),
+                    where_clause,
+                ),
+                params,
+            )
+            total = int((await cur.fetchone())["count"])
+            await cur.execute(
+                sql.SQL(
+                    "SELECT {} FROM {}.{} WHERE {} ORDER BY {} {} LIMIT %s OFFSET %s"
+                ).format(
+                    sql.SQL(", ").join(sql.Identifier(column) for column in columns),
+                    sql.Identifier(schema_name),
+                    sql.Identifier(table.code),
+                    where_clause,
+                    sql.Identifier(order_field),
+                    order_keyword,
+                ),
+                [*params, page_size, offset],
+            )
+            rows = [_row_json(dict(row)) for row in await cur.fetchall()]
+    return {
+        "table": table.as_dict(),
+        "items": rows,
+        "page": page,
+        "page_size": page_size,
+        "total": total,
+    }
+
+
+async def _write_audit(
+    cur,
+    *,
+    schema_name: str,
+    tenant_id: str,
+    project_id: str,
+    table_code: str,
+    record_id: uuid.UUID,
+    operation: str,
+    before_data: dict[str, Any] | None,
+    after_data: dict[str, Any] | None,
+    actor: str,
+) -> None:
+    await cur.execute(
+        sql.SQL(
+            """
+            INSERT INTO {}.data_change_logs (
+                tenant_id, project_id, table_code, record_id, operation,
+                before_data, after_data, actor
+            ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
+            """
+        ).format(sql.Identifier(schema_name)),
+        (
+            tenant_id,
+            project_id,
+            table_code,
+            record_id,
+            operation,
+            Jsonb(before_data) if before_data is not None else None,
+            Jsonb(after_data) if after_data is not None else None,
+            actor,
+        ),
+    )
+    record_version = int((after_data or before_data or {}).get("version") or 1)
+    await cur.execute(
+        sql.SQL(
+            """
+            INSERT INTO {}.graph_sync_queue (
+                tenant_id, project_id, table_code, record_id, operation, record_version
+            ) VALUES (%s, %s, %s, %s, %s, %s)
+            """
+        ).format(sql.Identifier(schema_name)),
+        (tenant_id, project_id, table_code, record_id, operation, record_version),
+    )
+
+
+async def create_record(
+    project_id: str,
+    table_code: str,
+    body: dict[str, Any],
+    actor: str,
+) -> dict[str, Any]:
+    table = _table_or_404(table_code)
+    if not table.allow_create:
+        raise HTTPException(403, "该表不允许新增")
+    database = await _database_or_404(project_id)
+    payload = _validated_payload(table, body, create=True)
+    record_id = uuid.uuid4()
+    schema_name = str(database["schema_name"])
+    tenant_id = str(database["tenant_id"])
+    columns = ["id", "tenant_id", "project_id", *payload.keys()]
+    values = [record_id, tenant_id, project_id, *payload.values()]
+    returning = ["id", *(field.code for field in table.fields), "created_at", "updated_at"]
+
+    async with get_conn() as conn:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                sql.SQL("INSERT INTO {}.{} ({}) VALUES ({}) RETURNING {}").format(
+                    sql.Identifier(schema_name),
+                    sql.Identifier(table.code),
+                    sql.SQL(", ").join(sql.Identifier(column) for column in columns),
+                    sql.SQL(", ").join(sql.Placeholder() for _ in values),
+                    sql.SQL(", ").join(sql.Identifier(column) for column in returning),
+                ),
+                values,
+            )
+            row = _row_json(dict(await cur.fetchone()))
+            await _write_audit(
+                cur,
+                schema_name=schema_name,
+                tenant_id=tenant_id,
+                project_id=project_id,
+                table_code=table_code,
+                record_id=record_id,
+                operation="create",
+                before_data=None,
+                after_data=row,
+                actor=actor,
+            )
+        await conn.commit()
+    return row or {}
+
+
+async def update_record(
+    project_id: str,
+    table_code: str,
+    record_id: str,
+    body: dict[str, Any],
+    actor: str,
+) -> dict[str, Any]:
+    table = _table_or_404(table_code)
+    if not table.allow_update:
+        raise HTTPException(403, "该表不允许修改")
+    try:
+        record_uuid = uuid.UUID(record_id)
+    except ValueError as exc:
+        raise HTTPException(400, "记录 ID 格式错误") from exc
+    database = await _database_or_404(project_id)
+    payload = _validated_payload(table, body, create=False)
+    if not payload:
+        raise HTTPException(400, "没有可修改字段")
+    schema_name = str(database["schema_name"])
+    tenant_id = str(database["tenant_id"])
+    returning = ["id", *(field.code for field in table.fields), "created_at", "updated_at"]
+
+    async with get_conn() as conn:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                sql.SQL("SELECT {} FROM {}.{} WHERE id=%s AND deleted_at IS NULL").format(
+                    sql.SQL(", ").join(sql.Identifier(column) for column in returning),
+                    sql.Identifier(schema_name),
+                    sql.Identifier(table.code),
+                ),
+                (record_uuid,),
+            )
+            before = _row_json(await cur.fetchone())
+            if not before:
+                raise HTTPException(404, "记录不存在")
+            assignments = [
+                sql.SQL("{}=%s").format(sql.Identifier(column))
+                for column in payload
+            ]
+            if any(field.code == "version" for field in table.fields):
+                assignments.append(sql.SQL("version=version+1"))
+            assignments.append(sql.SQL("updated_at=now()"))
+            await cur.execute(
+                sql.SQL("UPDATE {}.{} SET {} WHERE id=%s RETURNING {}").format(
+                    sql.Identifier(schema_name),
+                    sql.Identifier(table.code),
+                    sql.SQL(", ").join(assignments),
+                    sql.SQL(", ").join(sql.Identifier(column) for column in returning),
+                ),
+                [*payload.values(), record_uuid],
+            )
+            after = _row_json(await cur.fetchone())
+            await _write_audit(
+                cur,
+                schema_name=schema_name,
+                tenant_id=tenant_id,
+                project_id=project_id,
+                table_code=table_code,
+                record_id=record_uuid,
+                operation="update",
+                before_data=before,
+                after_data=after,
+                actor=actor,
+            )
+        await conn.commit()
+    return after or {}
+
+
+async def delete_record(
+    project_id: str,
+    table_code: str,
+    record_id: str,
+    actor: str,
+) -> dict[str, bool]:
+    table = _table_or_404(table_code)
+    if not table.allow_delete:
+        raise HTTPException(403, "该表不允许删除")
+    try:
+        record_uuid = uuid.UUID(record_id)
+    except ValueError as exc:
+        raise HTTPException(400, "记录 ID 格式错误") from exc
+    database = await _database_or_404(project_id)
+    schema_name = str(database["schema_name"])
+    tenant_id = str(database["tenant_id"])
+    columns = ["id", *(field.code for field in table.fields), "created_at", "updated_at"]
+    async with get_conn() as conn:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                sql.SQL("SELECT {} FROM {}.{} WHERE id=%s AND deleted_at IS NULL").format(
+                    sql.SQL(", ").join(sql.Identifier(column) for column in columns),
+                    sql.Identifier(schema_name),
+                    sql.Identifier(table.code),
+                ),
+                (record_uuid,),
+            )
+            before = _row_json(await cur.fetchone())
+            if not before:
+                raise HTTPException(404, "记录不存在")
+            await cur.execute(
+                sql.SQL(
+                    "UPDATE {}.{} SET deleted_at=now(), deleted_by=%s, updated_at=now() WHERE id=%s"
+                ).format(
+                    sql.Identifier(schema_name),
+                    sql.Identifier(table.code),
+                ),
+                (actor, record_uuid),
+            )
+            await _write_audit(
+                cur,
+                schema_name=schema_name,
+                tenant_id=tenant_id,
+                project_id=project_id,
+                table_code=table_code,
+                record_id=record_uuid,
+                operation="delete",
+                before_data=before,
+                after_data=None,
+                actor=actor,
+            )
+        await conn.commit()
+    return {"ok": True}
+

app/data_platform/registry.py

@@ -0,0 +1,364 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass(frozen=True)
+class FieldDefinition:
+    code: str
+    label: str
+    sql_type: str
+    data_type: str = "text"
+    required: bool = False
+    searchable: bool = False
+    sortable: bool = True
+    editable: bool = True
+    visible_in_list: bool = True
+    default_sql: str | None = None
+    options: tuple[str, ...] = ()
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "code": self.code,
+            "label": self.label,
+            "data_type": self.data_type,
+            "required": self.required,
+            "searchable": self.searchable,
+            "sortable": self.sortable,
+            "editable": self.editable,
+            "visible_in_list": self.visible_in_list,
+            "options": list(self.options),
+        }
+
+
+@dataclass(frozen=True)
+class TableDefinition:
+    code: str
+    label: str
+    group: str
+    description: str
+    fields: tuple[FieldDefinition, ...]
+    allow_create: bool = True
+    allow_update: bool = True
+    allow_delete: bool = True
+
+    @property
+    def primary_key(self) -> str:
+        return "id"
+
+    @property
+    def title_field(self) -> str:
+        for candidate in ("name", "room_name", "facility_name", "deal_name", "route_name", "tag_name"):
+            if any(field.code == candidate for field in self.fields):
+                return candidate
+        return "id"
+
+    def as_dict(self) -> dict[str, Any]:
+        system_fields = (
+            FieldDefinition("id", "记录 ID", "UUID", editable=False),
+            FieldDefinition("created_at", "创建时间", "TIMESTAMPTZ", "datetime", editable=False),
+            FieldDefinition("updated_at", "更新时间", "TIMESTAMPTZ", "datetime", editable=False),
+        )
+        return {
+            "code": self.code,
+            "label": self.label,
+            "group": self.group,
+            "description": self.description,
+            "primary_key": self.primary_key,
+            "title_field": self.title_field,
+            "allow_create": self.allow_create,
+            "allow_update": self.allow_update,
+            "allow_delete": self.allow_delete,
+            "fields": [field.as_dict() for field in (*system_fields, *self.fields)],
+        }
+
+
+def field(
+    code: str,
+    label: str,
+    sql_type: str = "TEXT",
+    data_type: str = "text",
+    *,
+    required: bool = False,
+    searchable: bool = False,
+    visible: bool = True,
+    default_sql: str | None = None,
+    options: tuple[str, ...] = (),
+) -> FieldDefinition:
+    return FieldDefinition(
+        code=code,
+        label=label,
+        sql_type=sql_type,
+        data_type=data_type,
+        required=required,
+        searchable=searchable,
+        visible_in_list=visible,
+        default_sql=default_sql,
+        options=options,
+    )
+
+
+TABLE_DEFINITIONS: tuple[TableDefinition, ...] = (
+    TableDefinition(
+        "poi_entities",
+        "地点实体",
+        "正式实体",
+        "酒店、美食、景区、交通和公交站统一基础信息。",
+        (
+            field("entity_type", "实体类型", required=True, searchable=True, options=("hotel", "restaurant", "scenic", "transport", "bus_stop")),
+            field("name", "名称", required=True, searchable=True),
+            field("category_l1", "一级分类", searchable=True),
+            field("category_l2", "二级分类", searchable=True),
+            field("category_l3", "细分类", searchable=True),
+            field("address", "地址", searchable=True),
+            field("district", "行政区", searchable=True),
+            field("adcode", "行政区划代码"),
+            field("phone", "商家电话"),
+            field("longitude", "经度", "DOUBLE PRECISION", "number"),
+            field("latitude", "纬度", "DOUBLE PRECISION", "number"),
+            field("h3_r9", "片区 R9"),
+            field("h3_r10", "片区 R10"),
+            field("status", "状态", default_sql="'active'", options=("active", "disabled")),
+            field("version", "版本", "INTEGER", "number", default_sql="1"),
+            field("extra_data", "扩展数据", "JSONB", "json", visible=False, default_sql="'{}'::jsonb"),
+        ),
+    ),
+    TableDefinition(
+        "entity_external_links",
+        "外部平台标识",
+        "公共明细",
+        "高德、携程、大众点评等已确认的平台 ID 与详情链接。",
+        (
+            field("entity_id", "实体 ID", "UUID", required=True, searchable=True),
+            field("platform", "平台", required=True, searchable=True, options=("amap", "ctrip", "dianping", "other")),
+            field("external_id", "平台 ID", required=True, searchable=True),
+            field("external_name", "平台名称", searchable=True),
+            field("external_url", "平台链接", "TEXT", "url"),
+        ),
+    ),
+    TableDefinition(
+        "entity_images",
+        "实体图片",
+        "公共明细",
+        "实体、房型、团购套餐等图片及来源。",
+        (
+            field("entity_id", "实体 ID", "UUID", required=True, searchable=True),
+            field("owner_type", "归属类型", default_sql="'entity'", options=("entity", "room", "deal")),
+            field("owner_id", "归属记录 ID", "UUID"),
+            field("image_url", "图片 URL", "TEXT", "url", required=True),
+            field("caption", "图片说明", searchable=True),
+            field("display_order", "排序", "INTEGER", "number", default_sql="0"),
+        ),
+    ),
+    TableDefinition(
+        "entity_relations",
+        "实体关系",
+        "公共明细",
+        "已经确认的包含、附近、分类和空间关系。",
+        (
+            field("source_entity_id", "源实体 ID", "UUID", required=True, searchable=True),
+            field("relation_type", "关系类型", required=True, searchable=True),
+            field("target_entity_id", "目标实体 ID", "UUID", required=True, searchable=True),
+            field("properties", "关系属性", "JSONB", "json", visible=False, default_sql="'{}'::jsonb"),
+            field("status", "状态", default_sql="'active'", options=("active", "disabled")),
+        ),
+    ),
+    TableDefinition(
+        "reviews",
+        "真实评论",
+        "公共明细",
+        "酒店、美食等实体的真实用户评论。",
+        (
+            field("entity_id", "实体 ID", "UUID", required=True, searchable=True),
+            field("platform", "评论平台", searchable=True),
+            field("external_review_id", "平台评论 ID", searchable=True),
+            field("rating", "评分", "DOUBLE PRECISION", "number"),
+            field("reviewer_name", "用户名称", searchable=True),
+            field("content", "评论内容", "TEXT", "long_text", required=True, searchable=True, visible=False),
+            field("reviewed_at", "评论时间", "TIMESTAMPTZ", "datetime"),
+        ),
+    ),
+    TableDefinition(
+        "review_tags",
+        "评价标签",
+        "公共明细",
+        "用于评价摘要展示的标签及出现次数。",
+        (
+            field("entity_id", "实体 ID", "UUID", required=True, searchable=True),
+            field("tag_name", "标签", required=True, searchable=True),
+            field("mention_count", "出现次数", "INTEGER", "number", default_sql="0"),
+            field("sentiment", "倾向", options=("positive", "neutral", "negative")),
+        ),
+    ),
+    TableDefinition(
+        "hotel_profiles",
+        "酒店详情",
+        "酒店",
+        "酒店的携程名称、开业时间、客房数量、钻级和评分。",
+        (
+            field("entity_id", "实体 ID", "UUID", required=True, searchable=True),
+            field("ctrip_name", "携程酒店名", searchable=True),
+            field("opened_year", "开业年份", "INTEGER", "number"),
+            field("room_count", "客房数量", "INTEGER", "number"),
+            field("diamond_level", "酒店钻级", "INTEGER", "number"),
+            field("ctrip_rating", "携程评分", "DOUBLE PRECISION", "number"),
+            field("review_count", "用户点评数", "INTEGER", "number"),
+            field("reference_price", "参考起价", "NUMERIC(12,2)", "number"),
+            field("introduction", "酒店简介", "TEXT", "long_text", visible=False),
+        ),
+    ),
+    TableDefinition(
+        "hotel_room_types",
+        "酒店房型",
+        "酒店",
+        "一个房型一条记录，不保存实时库存文案。",
+        (
+            field("entity_id", "酒店实体 ID", "UUID", required=True, searchable=True),
+            field("room_name", "房型名称", required=True, searchable=True),
+            field("image_url", "房型图片", "TEXT", "url"),
+            field("bed_type", "床型", searchable=True),
+            field("area_text", "面积"),
+            field("floor_text", "楼层"),
+            field("window_text", "窗户"),
+            field("max_guests", "可住人数", "INTEGER", "number"),
+            field("breakfast", "早餐"),
+            field("cancellation_policy", "取消政策", "TEXT", "long_text", visible=False),
+            field("payment_method", "支付方式"),
+            field("reference_price", "参考价格", "NUMERIC(12,2)", "number"),
+        ),
+    ),
+    TableDefinition(
+        "hotel_facilities",
+        "酒店设施",
+        "酒店",
+        "酒店设施分类、是否免费、收费说明和来源链接。",
+        (
+            field("entity_id", "酒店实体 ID", "UUID", required=True, searchable=True),
+            field("facility_category", "设施分类", required=True, searchable=True),
+            field("facility_name", "设施名称", required=True, searchable=True),
+            field("is_free", "是否免费", "BOOLEAN", "boolean"),
+            field("charge_note", "收费说明"),
+            field("source_url", "来源 URL", "TEXT", "url"),
+        ),
+    ),
+    TableDefinition(
+        "hotel_policies",
+        "酒店政策",
+        "酒店",
+        "入住、退房、宠物、儿童及其他酒店政策。",
+        (
+            field("entity_id", "酒店实体 ID", "UUID", required=True, searchable=True),
+            field("policy_type", "政策类型", required=True, searchable=True),
+            field("policy_name", "政策名称", required=True, searchable=True),
+            field("policy_content", "政策内容", "TEXT", "long_text", required=True, visible=False),
+        ),
+    ),
+    TableDefinition(
+        "restaurant_profiles",
+        "美食详情",
+        "美食",
+        "大众点评店名、分类、评分、人均消费和营业信息。",
+        (
+            field("entity_id", "实体 ID", "UUID", required=True, searchable=True),
+            field("dianping_name", "大众点评店名", searchable=True),
+            field("dianping_category", "点评分类", searchable=True),
+            field("ranking_text", "榜单排名"),
+            field("business_status", "营业状态", options=("open", "closed", "unknown")),
+            field("business_hours", "营业时间"),
+            field("rating", "点评评分", "DOUBLE PRECISION", "number"),
+            field("review_count", "用户点评数", "INTEGER", "number"),
+            field("average_price", "人均消费", "NUMERIC(12,2)", "number"),
+        ),
+    ),
+    TableDefinition(
+        "restaurant_deals",
+        "美食团购",
+        "美食",
+        "团购套餐名称、图片、价格、折扣和使用规则。",
+        (
+            field("entity_id", "实体 ID", "UUID", required=True, searchable=True),
+            field("deal_name", "套餐名称", required=True, searchable=True),
+            field("image_url", "套餐图片", "TEXT", "url"),
+            field("current_price", "当前价格", "NUMERIC(12,2)", "number"),
+            field("original_price", "原价", "NUMERIC(12,2)", "number"),
+            field("discount_text", "折扣"),
+            field("usage_rules", "使用规则", "TEXT", "long_text", visible=False),
+            field("validity_text", "有效期"),
+        ),
+    ),
+    TableDefinition(
+        "scenic_profiles",
+        "景区详情",
+        "景区",
+        "景区等级、游客价值分类、开放时间和门票说明。",
+        (
+            field("entity_id", "实体 ID", "UUID", required=True, searchable=True),
+            field("scenic_type", "景区类型", searchable=True),
+            field("scenic_level", "景区等级", searchable=True),
+            field("is_national", "是否国家级", "BOOLEAN", "boolean"),
+            field("visitor_value_type", "游客价值分类", searchable=True),
+            field("opening_hours", "开放时间"),
+            field("ticket_note", "门票说明", "TEXT", "long_text", visible=False),
+            field("official_intro", "官方简介", "TEXT", "long_text", visible=False),
+        ),
+    ),
+    TableDefinition(
+        "scenic_children",
+        "景区子景点",
+        "景区",
+        "景区内部子景点、类型、坐标及游览说明。",
+        (
+            field("parent_entity_id", "所属景区实体 ID", "UUID", required=True, searchable=True),
+            field("name", "子景点名称", required=True, searchable=True),
+            field("child_type", "子景点类型", searchable=True),
+            field("longitude", "经度", "DOUBLE PRECISION", "number"),
+            field("latitude", "纬度", "DOUBLE PRECISION", "number"),
+            field("visit_note", "游览说明", "TEXT", "long_text", visible=False),
+        ),
+    ),
+    TableDefinition(
+        "transport_profiles",
+        "交通设施详情",
+        "交通",
+        "汽车站、客运站、火车站等交通设施信息。",
+        (
+            field("entity_id", "实体 ID", "UUID", required=True, searchable=True),
+            field("transport_type", "交通类型", required=True, searchable=True),
+            field("service_hours", "服务时间"),
+            field("route_note", "交通说明", "TEXT", "long_text", visible=False),
+        ),
+    ),
+    TableDefinition(
+        "bus_routes",
+        "公交线路",
+        "交通",
+        "公交线路名称、方向、运营时间及线路颜色。",
+        (
+            field("route_name", "线路名称", required=True, searchable=True),
+            field("direction_name", "方向", searchable=True),
+            field("start_stop_name", "起点站", searchable=True),
+            field("end_stop_name", "终点站", searchable=True),
+            field("service_hours", "运营时间"),
+            field("route_color", "线路颜色"),
+            field("geometry", "线路坐标", "JSONB", "json", visible=False, default_sql="'[]'::jsonb"),
+        ),
+    ),
+    TableDefinition(
+        "bus_route_stops",
+        "公交站序",
+        "交通",
+        "公交线路方向下的站点顺序。",
+        (
+            field("route_id", "线路 ID", "UUID", required=True, searchable=True),
+            field("stop_entity_id", "站点实体 ID", "UUID", required=True, searchable=True),
+            field("stop_name", "站点名称", required=True, searchable=True),
+            field("stop_order", "站序", "INTEGER", "number", required=True),
+            field("longitude", "经度", "DOUBLE PRECISION", "number"),
+            field("latitude", "纬度", "DOUBLE PRECISION", "number"),
+        ),
+    ),
+)
+
+TABLE_REGISTRY: dict[str, TableDefinition] = {table.code: table for table in TABLE_DEFINITIONS}
+

app/data_platform/schema.py

@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+import hashlib
+import re
+from typing import Any
+
+from psycopg import sql
+
+from app.config import settings
+from app.db import get_conn
+from app.data_platform.registry import TABLE_DEFINITIONS, TableDefinition
+
+
+def _slug(value: str) -> str:
+    normalized = re.sub(r"[^a-z0-9]+", "_", value.lower()).strip("_")
+    return normalized[:42] or "project"
+
+
+def project_schema_name(project_id: str) -> str:
+    digest = hashlib.sha1(project_id.encode("utf-8")).hexdigest()[:6]
+    return f"biz_{_slug(project_id)}_{digest}"
+
+
+def project_database_name(project_id: str) -> str:
+    return f"{_slug(project_id)}_db"
+
+
+def _column_sql(table: TableDefinition) -> sql.Composed:
+    columns: list[sql.Composable] = [
+        sql.SQL("id UUID PRIMARY KEY"),
+        sql.SQL("tenant_id TEXT NOT NULL"),
+        sql.SQL("project_id TEXT NOT NULL"),
+    ]
+    for item in table.fields:
+        parts: list[sql.Composable] = [
+            sql.Identifier(item.code),
+            sql.SQL(item.sql_type),
+        ]
+        if item.required:
+            parts.append(sql.SQL("NOT NULL"))
+        if item.default_sql:
+            parts.extend((sql.SQL("DEFAULT"), sql.SQL(item.default_sql)))
+        columns.append(sql.SQL(" ").join(parts))
+    columns.extend(
+        (
+            sql.SQL("created_at TIMESTAMPTZ NOT NULL DEFAULT now()"),
+            sql.SQL("updated_at TIMESTAMPTZ NOT NULL DEFAULT now()"),
+            sql.SQL("deleted_at TIMESTAMPTZ"),
+            sql.SQL("deleted_by TEXT"),
+        )
+    )
+    return sql.SQL(", ").join(columns)
+
+
+async def ensure_platform_registry() -> None:
+    admin_schema = sql.Identifier(settings.db_schema)
+    async with get_conn() as conn:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                sql.SQL(
+                    """
+                    CREATE TABLE IF NOT EXISTS {}.project_databases (
+                        project_id TEXT PRIMARY KEY,
+                        tenant_id TEXT NOT NULL,
+                        display_name TEXT NOT NULL,
+                        database_name TEXT NOT NULL UNIQUE,
+                        schema_name TEXT NOT NULL UNIQUE,
+                        status TEXT NOT NULL DEFAULT 'ready',
+                        created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+                        updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+                    )
+                    """
+                ).format(admin_schema)
+            )
+        await conn.commit()
+
+
+async def ensure_project_database(
+    project_id: str,
+    tenant_id: str,
+    display_name: str,
+) -> dict[str, Any]:
+    await ensure_platform_registry()
+    schema_name = project_schema_name(project_id)
+    database_name = project_database_name(project_id)
+    admin_schema = sql.Identifier(settings.db_schema)
+    project_schema = sql.Identifier(schema_name)
+
+    async with get_conn() as conn:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                sql.SQL(
+                    """
+                    INSERT INTO {}.project_databases (
+                        project_id, tenant_id, display_name, database_name, schema_name, status, updated_at
+                    )
+                    VALUES (%s, %s, %s, %s, %s, 'ready', now())
+                    ON CONFLICT (project_id) DO UPDATE
+                    SET tenant_id=EXCLUDED.tenant_id,
+                        display_name=EXCLUDED.display_name,
+                        database_name=EXCLUDED.database_name,
+                        schema_name=EXCLUDED.schema_name,
+                        status='ready',
+                        updated_at=now()
+                    RETURNING *
+                    """
+                ).format(admin_schema),
+                (project_id, tenant_id, display_name, database_name, schema_name),
+            )
+            database_row = await cur.fetchone()
+            await cur.execute(sql.SQL("CREATE SCHEMA IF NOT EXISTS {}").format(project_schema))
+
+            for table in TABLE_DEFINITIONS:
+                await cur.execute(
+                    sql.SQL("CREATE TABLE IF NOT EXISTS {}.{} ({})").format(
+                        project_schema,
+                        sql.Identifier(table.code),
+                        _column_sql(table),
+                    )
+                )
+                await cur.execute(
+                    sql.SQL(
+                        "CREATE INDEX IF NOT EXISTS {} ON {}.{} (updated_at DESC)"
+                    ).format(
+                        sql.Identifier(f"{table.code}_updated_idx"),
+                        project_schema,
+                        sql.Identifier(table.code),
+                    )
+                )
+                await cur.execute(
+                    sql.SQL(
+                        "CREATE INDEX IF NOT EXISTS {} ON {}.{} (project_id) WHERE deleted_at IS NULL"
+                    ).format(
+                        sql.Identifier(f"{table.code}_project_idx"),
+                        project_schema,
+                        sql.Identifier(table.code),
+                    )
+                )
+            await cur.execute(
+                sql.SQL(
+                    """
+                    CREATE TABLE IF NOT EXISTS {}.data_change_logs (
+                        id BIGSERIAL PRIMARY KEY,
+                        tenant_id TEXT NOT NULL,
+                        project_id TEXT NOT NULL,
+                        table_code TEXT NOT NULL,
+                        record_id UUID NOT NULL,
+                        operation TEXT NOT NULL,
+                        before_data JSONB,
+                        after_data JSONB,
+                        actor TEXT NOT NULL,
+                        created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+                    )
+                    """
+                ).format(project_schema)
+            )
+            await cur.execute(
+                sql.SQL(
+                    """
+                    CREATE TABLE IF NOT EXISTS {}.graph_sync_queue (
+                        id BIGSERIAL PRIMARY KEY,
+                        tenant_id TEXT NOT NULL,
+                        project_id TEXT NOT NULL,
+                        table_code TEXT NOT NULL,
+                        record_id UUID NOT NULL,
+                        operation TEXT NOT NULL,
+                        record_version INTEGER NOT NULL DEFAULT 1,
+                        status TEXT NOT NULL DEFAULT 'pending',
+                        retry_count INTEGER NOT NULL DEFAULT 0,
+                        error_message TEXT,
+                        created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+                        completed_at TIMESTAMPTZ
+                    )
+                    """
+                ).format(project_schema)
+            )
+        await conn.commit()
+    return dict(database_row)
+
+
+async def ensure_all_project_databases() -> None:
+    await ensure_platform_registry()
+    async with get_conn() as conn:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                sql.SQL(
+                    """
+                    SELECT project_id, tenant_id, display_name
+                    FROM {}.projects
+                    WHERE status <> 'archived'
+                    ORDER BY created_at
+                    """
+                ).format(sql.Identifier(settings.db_schema))
+            )
+            projects = await cur.fetchall()
+    for project in projects:
+        await ensure_project_database(
+            str(project["project_id"]),
+            str(project["tenant_id"]),
+            str(project["display_name"]),
+        )
+
+
+async def get_project_database(project_id: str) -> dict[str, Any] | None:
+    await ensure_platform_registry()
+    async with get_conn() as conn:
+        async with conn.cursor() as cur:
+            await cur.execute(
+                sql.SQL("SELECT * FROM {}.project_databases WHERE project_id=%s").format(
+                    sql.Identifier(settings.db_schema)
+                ),
+                (project_id,),
+            )
+            row = await cur.fetchone()
+            return dict(row) if row else None