diff --git a/admin-web/src/api.ts b/admin-web/src/api.ts index 84bf6eb..8a63aa7 100644 --- a/admin-web/src/api.ts +++ b/admin-web/src/api.ts @@ -115,6 +115,48 @@ export const listGraphReleases = (projectId: string) => api.get(`/projects/${encodeURIComponent(projectId)}/graph-releases`); export const createGraphRelease = (projectId: string, data: unknown) => api.post(`/projects/${encodeURIComponent(projectId)}/graph-releases`, data); + +// ── Project relational data platform ── +export const listProjectDatabases = () => + api.get("/data-platform/databases"); +export const listProjectDataTables = (projectId: string) => + api.get(`/data-platform/databases/${encodeURIComponent(projectId)}/tables`); +export const listProjectTableRecords = ( + projectId: string, + tableCode: string, + params?: Record, +) => + api.get( + `/data-platform/databases/${encodeURIComponent(projectId)}/tables/${encodeURIComponent(tableCode)}/records`, + { params }, + ); +export const createProjectTableRecord = ( + projectId: string, + tableCode: string, + data: unknown, +) => + api.post( + `/data-platform/databases/${encodeURIComponent(projectId)}/tables/${encodeURIComponent(tableCode)}/records`, + data, + ); +export const updateProjectTableRecord = ( + projectId: string, + tableCode: string, + recordId: string, + data: unknown, +) => + api.patch( + `/data-platform/databases/${encodeURIComponent(projectId)}/tables/${encodeURIComponent(tableCode)}/records/${encodeURIComponent(recordId)}`, + data, + ); +export const deleteProjectTableRecord = ( + projectId: string, + tableCode: string, + recordId: string, +) => + api.delete( + `/data-platform/databases/${encodeURIComponent(projectId)}/tables/${encodeURIComponent(tableCode)}/records/${encodeURIComponent(recordId)}`, + ); export const listOntologySchemas = () => api.get("/ontology-schemas"); export const getCurrentOntologySchema = () => api.get("/ontology-schemas/current"); export const getOntologySchema = (schemaId: string | number) => diff --git a/admin-web/src/panels/data-platform/DataCenterPanel.tsx b/admin-web/src/panels/data-platform/DataCenterPanel.tsx index bc8b1a7..82573ff 100644 --- a/admin-web/src/panels/data-platform/DataCenterPanel.tsx +++ b/admin-web/src/panels/data-platform/DataCenterPanel.tsx @@ -1,73 +1,373 @@ import { ApartmentOutlined, CheckCircleOutlined, + ClockCircleOutlined, DatabaseOutlined, - ExportOutlined, - ImportOutlined, - SyncOutlined, + DeleteOutlined, + EditOutlined, + FolderOpenOutlined, + PlusOutlined, + ReloadOutlined, + SearchOutlined, TableOutlined, } from "@ant-design/icons"; -import { Card, Col, Row, Space, Tag, Typography } from "antd"; -import { displayProjectName, getProjectContext } from "../../api"; +import { + Button, + Card, + Col, + Empty, + Form, + Input, + InputNumber, + Modal, + Popconfirm, + Row, + Select, + Space, + Spin, + Statistic, + Switch, + Table, + Tag, + Typography, + message, +} from "antd"; +import type { ColumnsType } from "antd/es/table"; +import { useCallback, useEffect, useMemo, useState, type ReactNode } from "react"; +import { + createProjectTableRecord, + deleteProjectTableRecord, + getProjectContext, + listProjectDatabases, + listProjectDataTables, + listProjectTableRecords, + updateProjectTableRecord, +} from "../../api"; const { Paragraph, Text, Title } = Typography; +const { Search } = Input; -const capabilityCards = [ - { - icon: , - title: "正式业务数据", - description: "统一管理酒店、美食、景区、交通及其明细表。", - status: "第一阶段", - }, - { - icon: , - title: "批量导入", - description: "接收处理完成的 CSV、XLSX 或多表 ZIP 数据包。", - status: "第二阶段", - }, - { - icon: , - title: "批量导出", - description: "按数据类型导出正式数据、明细表和数据字典。", - status: "第二阶段", - }, - { - icon: , - title: "图谱同步", - description: "把 PostgreSQL 正式数据同步为 FalkorDB 图谱投影。", - status: "第三阶段", - }, -]; +type ProjectDatabase = { + project_id: string; + tenant_id: string; + display_name: string; + database_name: string; + schema_name: string; + status: string; + project_status: string; + table_count: number; + record_count: number; + updated_at?: string | null; +}; -const dataGroups = [ - { - icon: , - title: "正式实体", - tables: "poi_entities", - description: "保存酒店、美食、景区、交通等统一基础信息。", - }, - { - icon: , - title: "业务明细", - tables: "房型、设施、套餐、评论、公交线路与站序", - description: "一对多数据使用独立关系表，不再拼接成长文本。", - }, - { - icon: , - title: "必要运行记录", - tables: "导入、导出、修改日志", - description: "只记录正式数据操作结果，不保存原始采集和候选数据。", - }, - { - icon: , - title: "图谱投影", - tables: "graph_sync_queue", - description: "PostgreSQL 是权威数据源，FalkorDB 可按项目重新生成。", - }, -]; +type DataField = { + code: string; + label: string; + data_type: string; + required: boolean; + searchable: boolean; + sortable: boolean; + editable: boolean; + visible_in_list: boolean; + options: string[]; +}; + +type DataTableMeta = { + code: string; + label: string; + group: string; + description: string; + primary_key: string; + title_field: string; + allow_create: boolean; + allow_update: boolean; + allow_delete: boolean; + fields: DataField[]; + record_count: number; + updated_at?: string | null; +}; + +type RecordRow = Record & { id: string }; + +const groupIcons: Record = { + 正式实体: , + 公共明细: , + 酒店: , + 美食: , + 景区: , + 交通: , +}; + +function errorText(error: unknown, fallback: string) { + const detail = (error as { response?: { data?: { detail?: string } } })?.response?.data?.detail; + return detail || fallback; +} + +function formatDate(value?: string | null) { + if (!value) return "尚无数据"; + const date = new Date(value); + return Number.isNaN(date.getTime()) ? String(value) : date.toLocaleString(); +} + +function displayValue(value: unknown, field?: DataField) { + if (value === null || value === undefined || value === "") return "—"; + if (field?.data_type === "boolean") return value ? "是" : "否"; + if (field?.data_type === "datetime") return formatDate(String(value)); + if (typeof value === "object") return JSON.stringify(value); + return String(value); +} + +function fieldControl(field: DataField) { + if (field.options?.length) { + return ( + ; + } + return ; +} export default function DataCenterPanel() { - const context = getProjectContext(); + const [form] = Form.useForm(); + const currentProject = getProjectContext().projectId; + const [databases, setDatabases] = useState([]); + const [selectedProjectId, setSelectedProjectId] = useState(""); + const [tables, setTables] = useState([]); + const [selectedTableCode, setSelectedTableCode] = useState(""); + const [records, setRecords] = useState([]); + const [databaseLoading, setDatabaseLoading] = useState(true); + const [tableLoading, setTableLoading] = useState(false); + const [recordLoading, setRecordLoading] = useState(false); + const [saving, setSaving] = useState(false); + const [search, setSearch] = useState(""); + const [page, setPage] = useState(1); + const [pageSize, setPageSize] = useState(50); + const [total, setTotal] = useState(0); + const [modalOpen, setModalOpen] = useState(false); + const [editingRecord, setEditingRecord] = useState(null); + + const selectedDatabase = useMemo( + () => databases.find((item) => item.project_id === selectedProjectId), + [databases, selectedProjectId], + ); + const selectedTable = useMemo( + () => tables.find((item) => item.code === selectedTableCode), + [tables, selectedTableCode], + ); + + const loadDatabases = useCallback(async () => { + setDatabaseLoading(true); + try { + const { data } = await listProjectDatabases(); + const items = data as ProjectDatabase[]; + setDatabases(items); + setSelectedProjectId((previous) => { + if (previous && items.some((item) => item.project_id === previous)) return previous; + return items.find((item) => item.project_id === currentProject)?.project_id + || items.find((item) => item.project_id === "yunyou_libo")?.project_id + || items[0]?.project_id + || ""; + }); + } catch (error) { + message.error(errorText(error, "项目数据库加载失败")); + } finally { + setDatabaseLoading(false); + } + }, [currentProject]); + + const loadTables = useCallback(async (projectId: string) => { + if (!projectId) return; + setTableLoading(true); + try { + const { data } = await listProjectDataTables(projectId); + const items = data.tables as DataTableMeta[]; + setTables(items); + setSelectedTableCode((previous) => ( + previous && items.some((item) => item.code === previous) + ? previous + : items[0]?.code || "" + )); + } catch (error) { + message.error(errorText(error, "项目数据表加载失败")); + setTables([]); + setSelectedTableCode(""); + } finally { + setTableLoading(false); + } + }, []); + + const loadRecords = useCallback(async () => { + if (!selectedProjectId || !selectedTableCode) return; + setRecordLoading(true); + try { + const { data } = await listProjectTableRecords( + selectedProjectId, + selectedTableCode, + { page, page_size: pageSize, search: search || undefined }, + ); + setRecords(data.items as RecordRow[]); + setTotal(Number(data.total || 0)); + } catch (error) { + message.error(errorText(error, "表数据加载失败")); + setRecords([]); + setTotal(0); + } finally { + setRecordLoading(false); + } + }, [page, pageSize, search, selectedProjectId, selectedTableCode]); + + useEffect(() => { + loadDatabases(); + }, [loadDatabases]); + + useEffect(() => { + setSearch(""); + setPage(1); + setTables([]); + setSelectedTableCode(""); + if (selectedProjectId) loadTables(selectedProjectId); + }, [loadTables, selectedProjectId]); + + useEffect(() => { + loadRecords(); + }, [loadRecords]); + + const groupedTables = useMemo(() => { + const groups = new Map(); + for (const table of tables) { + groups.set(table.group, [...(groups.get(table.group) || []), table]); + } + return Array.from(groups.entries()); + }, [tables]); + + const openCreate = () => { + setEditingRecord(null); + form.resetFields(); + setModalOpen(true); + }; + + const openEdit = (record: RecordRow) => { + setEditingRecord(record); + const values: Record = {}; + selectedTable?.fields.filter((field) => field.editable).forEach((field) => { + const value = record[field.code]; + values[field.code] = field.data_type === "json" && value && typeof value === "object" + ? JSON.stringify(value, null, 2) + : value; + }); + form.setFieldsValue(values); + setModalOpen(true); + }; + + const saveRecord = async () => { + if (!selectedTable || !selectedProjectId) return; + const values = await form.validateFields(); + setSaving(true); + try { + if (editingRecord) { + await updateProjectTableRecord( + selectedProjectId, + selectedTable.code, + editingRecord.id, + values, + ); + message.success("记录已更新"); + } else { + await createProjectTableRecord(selectedProjectId, selectedTable.code, values); + message.success("记录已新增"); + } + setModalOpen(false); + form.resetFields(); + await Promise.all([loadRecords(), loadTables(selectedProjectId), loadDatabases()]); + } catch (error) { + message.error(errorText(error, "保存失败")); + } finally { + setSaving(false); + } + }; + + const removeRecord = async (recordId: string) => { + if (!selectedTable || !selectedProjectId) return; + try { + await deleteProjectTableRecord(selectedProjectId, selectedTable.code, recordId); + message.success("记录已删除，可从修改日志恢复"); + await Promise.all([loadRecords(), loadTables(selectedProjectId), loadDatabases()]); + } catch (error) { + message.error(errorText(error, "删除失败")); + } + }; + + const visibleFields = useMemo( + () => (selectedTable?.fields || []) + .filter((field) => field.visible_in_list) + .slice(0, 7), + [selectedTable], + ); + + const columns = useMemo>(() => { + const dataColumns: ColumnsType = visibleFields.map((field) => ({ + title: field.label, + dataIndex: field.code, + key: field.code, + width: field.code === selectedTable?.title_field ? 210 : 150, + ellipsis: true, + render: (value: unknown) => ( + {displayValue(value, field)} + ), + })); + dataColumns.push({ + title: "操作", + key: "actions", + width: 116, + fixed: "right", + render: (_value, record) => ( + + - - {capabilityCards.map((item) => ( - - -
{item.icon}
-
- {item.title} - {item.status} + +
+
+
+ 项目数据库 + 一个项目对应一个独立的 PostgreSQL 数据空间 +
+ {databases.length} 个数据库 +
+ {databases.length ? ( +
+ {databases.map((database) => { + const active = database.project_id === selectedProjectId; + return ( + + ); + })} +
+ ) : ( + + )} +
+ + + {selectedDatabase && ( + + + +
+ + + 数据表 + + {tables.length}
- {item.description} + +
+ {groupedTables.map(([group, items]) => ( +
+
+ {groupIcons[group]} + {group} +
+ {items.map((table) => ( + + ))} +
+ ))} +
+ - ))} - - PostgreSQL 权威数据} - > -
- {dataGroups.map((group) => ( -
-
{group.icon}
-
-
- {group.title} - {group.tables} -
- {group.description} -
-
- ))} -
- - - - - -
- 进入 PostgreSQL -
    -
  • 处理完成的酒店、美食、景区、交通实体
    • -
  • 房型、设施、套餐、评论和公交站序
    • -
  • 已经确认的外部平台 ID、URL 和实体关系
    • -
-
- - -
- 不进入正式业务库 -
    -
  • 未清洗的接口原始响应和爬虫缓存
    • -
  • 候选匹配中间结果和重复计算过程
    • -
  • “仅剩几间”等无长期价值的实时文案
    • -
-
+ + + {selectedTable ? ( + <> +
+
+ + {selectedTable.label} + {selectedTable.record_count.toLocaleString()} 条 + + {selectedTable.description} +
+ + } + placeholder="搜索本表" + defaultValue={search} + onSearch={(value) => { + setSearch(value.trim()); + setPage(1); + }} + style={{ width: 230 }} + /> + + + +
+
+ }> + {selectedDatabase.database_name} + {selectedTable.code} + {formatDate(selectedTable.updated_at)} + +
+ + rowKey="id" + loading={recordLoading} + columns={columns} + dataSource={records} + scroll={{ x: 900 }} + locale={{ + emptyText: ( + + ), + }} + pagination={{ + current: page, + pageSize, + total, + showSizeChanger: true, + showTotal: (value) => `共 ${value} 条`, + onChange: (nextPage, nextPageSize) => { + setPage(nextPageSize === pageSize ? nextPage : 1); + setPageSize(nextPageSize); + }, + }} + /> + + ) : ( + + )} + - + )} + + + + } /> + + + } /> + + + } /> + + + + { + setModalOpen(false); + form.resetFields(); + }} + destroyOnClose + > +
+ + {selectedTable?.fields.filter((field) => field.editable).map((field) => ( + + + {fieldControl(field)} + + + ))} + +
+
); } diff --git a/admin-web/src/styles.css b/admin-web/src/styles.css index 63ed8d7..b7be527 100644 --- a/admin-web/src/styles.css +++ b/admin-web/src/styles.css @@ -81,7 +81,7 @@ body { margin: 0; font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Ro /* ── Relational data center ── */ .data-center-page { - max-width: 1440px; + max-width: 1680px; margin: 0 auto; } @@ -117,121 +117,314 @@ body { margin: 0; font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Ro font-size: 20px; } -.data-center-capabilities { - margin-top: 20px; -} - -.data-center-capability-card { - height: 100%; +.data-center-section, +.data-center-table-catalog, +.data-center-record-card, +.data-center-summary .ant-card { border-color: #e6edf7; border-radius: 12px; } -.data-center-capability-card .ant-card-body { - min-height: 170px; +.data-center-section { + padding: 20px; + background: #fbfcff; + border: 1px solid #e6edf7; } -.data-center-capability-icon { - display: inline-flex; - align-items: center; - justify-content: center; - width: 40px; - height: 40px; - margin-bottom: 18px; - border-radius: 10px; - color: #1677ff; - background: #f0f6ff; - font-size: 19px; -} - -.data-center-capability-heading { +.data-center-section-heading { display: flex; align-items: center; justify-content: space-between; - gap: 12px; - margin-bottom: 8px; - font-size: 16px; + gap: 16px; + margin-bottom: 16px; } -.data-center-capability-card p.ant-typography { - margin-bottom: 0; - line-height: 1.7; +.data-center-section-heading .ant-typography { + margin: 0; } -.data-center-structure-card, -.data-center-boundary-card { - margin-top: 20px; - border-color: #e6edf7; - border-radius: 12px; +.data-center-section-heading h4.ant-typography { + margin-bottom: 4px; } -.data-center-groups { +.data-center-database-grid { display: grid; - grid-template-columns: repeat(2, minmax(0, 1fr)); - gap: 14px; + grid-template-columns: repeat(4, minmax(220px, 1fr)); + gap: 12px; } -.data-center-group { - display: flex; - gap: 14px; +.data-center-database-card { min-width: 0; - padding: 16px; - border: 1px solid #edf1f7; + padding: 15px; + color: #1f2937; + text-align: left; + background: #fff; + border: 1px solid #e4eaf3; border-radius: 10px; - background: #fbfcff; + cursor: pointer; + transition: border-color .18s ease, box-shadow .18s ease, transform .18s ease; } -.data-center-group-icon { +.data-center-database-card:hover { + border-color: #8bbcff; + box-shadow: 0 6px 18px rgba(22, 119, 255, .08); + transform: translateY(-1px); +} + +.data-center-database-card.active { + border-color: #1677ff; + box-shadow: 0 0 0 2px rgba(22, 119, 255, .1); +} + +.data-center-database-card-title { + display: flex; + align-items: center; + gap: 10px; + min-width: 0; +} + +.data-center-database-card-title > span:nth-child(2) { + display: flex; + flex: 1; + flex-direction: column; + min-width: 0; +} + +.data-center-database-card-title strong, +.data-center-database-card-title small { + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +.data-center-database-card-title strong { + font-size: 15px; +} + +.data-center-database-card-title small { + margin-top: 3px; + color: #8b96a8; + font-size: 12px; +} + +.data-center-database-icon { display: inline-flex; align-items: center; justify-content: center; - flex: 0 0 34px; - width: 34px; - height: 34px; + flex: 0 0 36px; + width: 36px; + height: 36px; + color: #1677ff; + background: #edf5ff; border-radius: 9px; - color: #5b6b82; - background: #fff; - box-shadow: 0 1px 4px rgba(31, 55, 88, 0.08); + font-size: 17px; } -.data-center-group-content { +.data-center-database-selected { + color: #1677ff; +} + +.data-center-database-stats { + display: flex; + gap: 20px; + margin: 15px 0 12px 46px; + color: #7a8699; + font-size: 12px; +} + +.data-center-database-stats b { + margin-right: 3px; + color: #263247; + font-size: 15px; +} + +.data-center-database-meta { + display: flex; + align-items: center; + justify-content: space-between; + gap: 8px; + padding-top: 10px; + color: #909aab; + border-top: 1px dashed #edf0f5; + font: 11px ui-monospace, SFMono-Regular, Menlo, monospace; +} + +.data-center-database-meta > span { + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +.data-center-workspace { + margin-top: 18px; +} + +.data-center-table-catalog, +.data-center-record-card { + height: 100%; + overflow: hidden; +} + +.data-center-catalog-header { + display: flex; + align-items: center; + justify-content: space-between; + min-height: 60px; + padding: 0 16px; + border-bottom: 1px solid #edf0f5; +} + +.data-center-table-groups { + max-height: 720px; + padding: 10px; + overflow-y: auto; +} + +.data-center-table-group + .data-center-table-group { + margin-top: 14px; +} + +.data-center-table-group-label { + display: flex; + align-items: center; + gap: 7px; + padding: 5px 9px 7px; + color: #7a8699; + font-size: 12px; + font-weight: 600; +} + +.data-center-table-item { + display: flex; + align-items: center; + justify-content: space-between; + width: 100%; + gap: 10px; + padding: 9px 10px; + color: #334057; + text-align: left; + background: transparent; + border: 0; + border-radius: 8px; + cursor: pointer; +} + +.data-center-table-item:hover { + background: #f4f8fe; +} + +.data-center-table-item.active { + color: #0958d9; + background: #eaf3ff; +} + +.data-center-table-item > span { + display: flex; + flex-direction: column; min-width: 0; } -.data-center-group-title { - display: flex; - flex-wrap: wrap; - align-items: center; - gap: 8px; - margin-bottom: 6px; +.data-center-table-item strong, +.data-center-table-item small { + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; } -.data-center-boundary { - height: 100%; - padding: 16px 18px; +.data-center-table-item strong { + font-size: 13px; + font-weight: 600; +} + +.data-center-table-item small { + margin-top: 2px; + color: #9aa3b3; + font: 10px ui-monospace, SFMono-Regular, Menlo, monospace; +} + +.data-center-table-item b { + min-width: 24px; + padding: 1px 6px; + color: #7c8799; + text-align: center; + background: #f0f2f5; border-radius: 10px; + font-size: 11px; } -.data-center-boundary ul { - margin: 10px 0 0; - padding-left: 20px; - color: #5b6678; - line-height: 1.9; +.data-center-record-header { + display: flex; + align-items: flex-start; + justify-content: space-between; + gap: 20px; + padding: 18px 20px 14px; } -.data-center-boundary-include { - border: 1px solid #d9f0df; - background: #f7fcf8; +.data-center-record-header .ant-typography { + margin: 0; } -.data-center-boundary-exclude { - border: 1px solid #f0e6d8; - background: #fffaf4; +.data-center-record-header h4.ant-typography { + margin-bottom: 4px; +} + +.data-center-record-context { + padding: 9px 20px; + color: #738097; + background: #fafbfd; + border-top: 1px solid #f0f2f5; + border-bottom: 1px solid #edf0f5; + font-size: 12px; +} + +.data-center-context-divider { + display: inline-block; + width: 1px; + height: 12px; + background: #d9dfe8; +} + +.data-center-record-card .ant-table-wrapper { + padding: 0 1px; +} + +.data-center-record-card .ant-table-thead > tr > th { + color: #596579; + background: #fafbfd; + font-size: 12px; +} + +.data-center-record-card .ant-empty { + margin: 48px 0; +} + +.data-center-summary { + margin-top: 18px; +} + +.data-center-summary .ant-card { + height: 100%; +} + +.data-center-record-form { + max-height: 62vh; + padding-right: 4px; + overflow-y: auto; +} + +@media (max-width: 1280px) { + .data-center-database-grid { + grid-template-columns: repeat(3, minmax(220px, 1fr)); + } } @media (max-width: 900px) { - .data-center-groups { - grid-template-columns: 1fr; + .data-center-database-grid { + grid-template-columns: repeat(2, minmax(200px, 1fr)); + } + + .data-center-table-catalog { + margin-bottom: 18px; } } @@ -239,6 +432,14 @@ body { margin: 0; font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Ro .data-center-header { flex-direction: column; } + + .data-center-database-grid { + grid-template-columns: 1fr; + } + + .data-center-record-header { + flex-direction: column; + } } /* ── Plaza Overview ── */ diff --git a/app/api/__init__.py b/app/api/__init__.py index cba1a80..8861e04 100644 --- a/app/api/__init__.py +++ b/app/api/__init__.py @@ -21,6 +21,7 @@ from app.api.manual_ingest import router as manual_ingest_router from app.api.ocr import router as ocr_router from app.api.travel_assistant import public_router as travel_assistant_public_router from app.api.travel_assistant import router as travel_assistant_router +from app.api.data_platform import router as data_platform_router api_router = APIRouter(prefix="/v1/admin") openapi_router = APIRouter(prefix="/v1/openapi") @@ -65,6 +66,7 @@ api_router.include_router(evidence_router, tags=["evidence"]) api_router.include_router(manual_ingest_router, tags=["manual-ingest"]) api_router.include_router(travel_assistant_router, tags=["travel-assistant"]) api_router.include_router(ocr_router, tags=["ocr"]) +api_router.include_router(data_platform_router, tags=["data-platform"]) # Agent call logs from app.api.agent_call_logs import router as agent_call_logs_router # noqa: E402 diff --git a/app/api/data_platform.py b/app/api/data_platform.py new file mode 100644 index 0000000..db48980 --- /dev/null +++ b/app/api/data_platform.py @@ -0,0 +1,91 @@ +"""Project database catalog and registered-table CRUD APIs.""" +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, HTTPException, Query + +from app.auth import CurrentUser +from app.data_platform.record_service import ( + create_record, + delete_record, + list_databases, + list_records, + list_tables, + update_record, +) +from app.data_platform.schema import ensure_all_project_databases + +router = APIRouter(prefix="/data-platform") + + +@router.get("/databases") +async def databases(_user: CurrentUser): + return await list_databases() + + +@router.post("/databases/initialize") +async def initialize_databases(user: CurrentUser): + if "admin" not in user.get("roles", []): + raise HTTPException(403, "只有系统管理员可以初始化项目数据库") + await ensure_all_project_databases() + return {"ok": True} + + +@router.get("/databases/{project_id}/tables") +async def tables(project_id: str, _user: CurrentUser): + return await list_tables(project_id) + + +@router.get("/databases/{project_id}/tables/{table_code}/records") +async def records( + project_id: str, + table_code: str, + _user: CurrentUser, + page: int = Query(default=1, ge=1), + page_size: int = Query(default=50, ge=1, le=200), + search: str | None = None, + sort_field: str | None = None, + sort_order: str = Query(default="desc", pattern="^(asc|desc)$"), +): + return await list_records( + project_id, + table_code, + page=page, + page_size=page_size, + search=search, + sort_field=sort_field, + sort_order=sort_order, + ) + + +@router.post("/databases/{project_id}/tables/{table_code}/records") +async def add_record( + project_id: str, + table_code: str, + body: dict[str, Any], + user: CurrentUser, +): + return await create_record(project_id, table_code, body, user["username"]) + + +@router.patch("/databases/{project_id}/tables/{table_code}/records/{record_id}") +async def edit_record( + project_id: str, + table_code: str, + record_id: str, + body: dict[str, Any], + user: CurrentUser, +): + return await update_record(project_id, table_code, record_id, body, user["username"]) + + +@router.delete("/databases/{project_id}/tables/{table_code}/records/{record_id}") +async def remove_record( + project_id: str, + table_code: str, + record_id: str, + user: CurrentUser, +): + return await delete_record(project_id, table_code, record_id, user["username"]) + diff --git a/app/api/projects.py b/app/api/projects.py index 6ceb30d..0314c96 100644 --- a/app/api/projects.py +++ b/app/api/projects.py @@ -184,6 +184,9 @@ async def create_project(body: dict, _user: CurrentUser): ) row = await cur.fetchone() await conn.commit() + from app.data_platform.schema import ensure_project_database + + await ensure_project_database(project_id, tenant_id, display_name) return row diff --git a/app/data_platform/__init__.py b/app/data_platform/__init__.py new file mode 100644 index 0000000..9fca435 --- /dev/null +++ b/app/data_platform/__init__.py @@ -0,0 +1,2 @@ +"""Project-isolated PostgreSQL data platform.""" + diff --git a/app/data_platform/migrate_graph.py b/app/data_platform/migrate_graph.py new file mode 100644 index 0000000..843ffb8 --- /dev/null +++ b/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() diff --git a/app/data_platform/record_service.py b/app/data_platform/record_service.py new file mode 100644 index 0000000..58a2963 --- /dev/null +++ b/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} + diff --git a/app/data_platform/registry.py b/app/data_platform/registry.py new file mode 100644 index 0000000..cf115d6 --- /dev/null +++ b/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} + diff --git a/app/data_platform/schema.py b/app/data_platform/schema.py new file mode 100644 index 0000000..5edd0af --- /dev/null +++ b/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 diff --git a/app/main.py b/app/main.py index da2a510..0ba748f 100644 --- a/app/main.py +++ b/app/main.py @@ -11,6 +11,7 @@ from starlette.exceptions import HTTPException as StarletteHTTPException from app.api import api_router, openapi_router from app.api.mcp_server import router as mcp_router from app.db import init_pool, close_pool +from app.data_platform.schema import ensure_platform_registry class SPAStaticFiles(StaticFiles): @@ -33,6 +34,7 @@ class SPAStaticFiles(StaticFiles): @asynccontextmanager async def lifespan(_app: FastAPI): await init_pool() + await ensure_platform_registry() yield await close_pool()