527 lines
17 KiB
TypeScript
527 lines
17 KiB
TypeScript
// 文件名: sync_shorturl_schema_to_clickhouse.ts
|
||
// 描述: 此脚本用于同步PostgreSQL中的short_url.shorturl表数据到ClickHouse
|
||
// 创建日期: 2023-11-21
|
||
|
||
import { Pool } from "https://deno.land/x/postgres@v0.17.0/mod.ts";
|
||
import { getResource, getVariable, setVariable } from "https://deno.land/x/windmill@v1.183.0/mod.ts";
|
||
|
||
// 同步状态接口
|
||
interface SyncState {
|
||
last_sync_time: string; // 上次同步的结束时间
|
||
records_synced: number; // 累计同步的记录数
|
||
last_run: string; // 上次运行的时间
|
||
}
|
||
|
||
// 同步状态键名
|
||
const SYNC_STATE_KEY = "f/shorturl_analytics/shorturl_to_clickhouse_state";
|
||
|
||
// PostgreSQL配置接口
|
||
interface PgConfig {
|
||
host: string;
|
||
port: number;
|
||
user: string;
|
||
password: string;
|
||
dbname?: string;
|
||
[key: string]: unknown;
|
||
}
|
||
|
||
// ClickHouse配置接口
|
||
interface ChConfig {
|
||
clickhouse_host: string;
|
||
clickhouse_port: number;
|
||
clickhouse_user: string;
|
||
clickhouse_password: string;
|
||
clickhouse_url?: string;
|
||
}
|
||
|
||
// Shorturl数据接口
|
||
interface ShortUrlData {
|
||
id: string;
|
||
slug: string;
|
||
origin: string; // 对应ClickHouse中的original_url
|
||
title?: string;
|
||
description?: string;
|
||
created_at?: string;
|
||
updated_at?: string;
|
||
deleted_at?: string;
|
||
expires_at?: string; // 注意这里已更正为expires_at
|
||
domain?: string; // 添加domain字段
|
||
}
|
||
|
||
/**
|
||
* 同步PostgreSQL short_url.shorturl表数据到ClickHouse
|
||
*/
|
||
export async function main(
|
||
/** 是否为测试模式（不执行实际更新） */
|
||
dry_run = false,
|
||
/** 是否显示详细日志 */
|
||
verbose = false,
|
||
/** 是否重置同步状态（从头开始同步） */
|
||
reset_sync_state = false,
|
||
/** 如果没有同步状态，往前查询多少小时的数据（默认1小时） */
|
||
default_hours_back = 1
|
||
) {
|
||
// 初始化日志函数
|
||
const log = (message: string, isVerbose = false) => {
|
||
if (!isVerbose || verbose) {
|
||
console.log(message);
|
||
}
|
||
};
|
||
|
||
// 获取同步状态
|
||
let syncState: SyncState | null = null;
|
||
if (!reset_sync_state) {
|
||
try {
|
||
log("获取同步状态...", true);
|
||
const rawState = await getVariable(SYNC_STATE_KEY);
|
||
if (rawState) {
|
||
if (typeof rawState === "string") {
|
||
syncState = JSON.parse(rawState);
|
||
} else {
|
||
syncState = rawState as SyncState;
|
||
}
|
||
log(`找到上次同步状态: 最后同步时间 ${syncState.last_sync_time}, 已同步记录数 ${syncState.records_synced}`, true);
|
||
}
|
||
} catch (error) {
|
||
log(`获取同步状态失败: ${error}, 将使用默认设置`, true);
|
||
}
|
||
} else {
|
||
log("重置同步状态，从头开始同步", true);
|
||
}
|
||
|
||
// 设置时间范围
|
||
const oneHourAgo = new Date(Date.now() - default_hours_back * 60 * 60 * 1000).toISOString();
|
||
// 如果有同步状态，使用上次同步时间作为开始时间；否则使用默认时间
|
||
const start_time = syncState ? syncState.last_sync_time : oneHourAgo;
|
||
const end_time = new Date().toISOString();
|
||
|
||
log(`开始同步shorturl表数据: ${start_time} 至 ${end_time}`);
|
||
|
||
let pgPool: Pool | null = null;
|
||
|
||
try {
|
||
// 1. 获取数据库配置
|
||
log("获取PostgreSQL数据库配置...", true);
|
||
const pgConfig = await getResource('f/limq/production_supabase') as PgConfig;
|
||
|
||
// 2. 创建PostgreSQL连接池
|
||
pgPool = new Pool({
|
||
hostname: pgConfig.host,
|
||
port: pgConfig.port,
|
||
user: pgConfig.user,
|
||
password: pgConfig.password,
|
||
database: pgConfig.dbname || 'postgres'
|
||
}, 3);
|
||
|
||
// 3. 获取需要更新的数据
|
||
const shorturlData = await getShortUrlData(pgPool, start_time, end_time, log);
|
||
log(`成功获取 ${shorturlData.length} 条shorturl数据`);
|
||
|
||
if (shorturlData.length === 0) {
|
||
// 更新同步状态，即使没有新数据
|
||
if (!dry_run) {
|
||
await updateSyncState(end_time, syncState ? syncState.records_synced : 0, log);
|
||
}
|
||
return { success: true, message: "没有找到需要更新的数据", updated: 0 };
|
||
}
|
||
|
||
// 4. 获取ClickHouse配置
|
||
const chConfig = await getClickHouseConfig();
|
||
|
||
// 5. 执行更新
|
||
if (!dry_run) {
|
||
const shorturlUpdated = await updateClickHouseShortUrl(shorturlData, chConfig, log);
|
||
|
||
// 更新同步状态
|
||
const totalSynced = (syncState ? syncState.records_synced : 0) + shorturlUpdated;
|
||
await updateSyncState(end_time, totalSynced, log);
|
||
|
||
return {
|
||
success: true,
|
||
message: "shorturl表数据同步完成",
|
||
shorturl_updated: shorturlUpdated,
|
||
total_synced: totalSynced,
|
||
sync_state: {
|
||
last_sync_time: end_time,
|
||
records_synced: totalSynced
|
||
}
|
||
};
|
||
} else {
|
||
log("测试模式: 不执行实际更新");
|
||
return {
|
||
success: true,
|
||
dry_run: true,
|
||
shorturl_count: shorturlData.length,
|
||
shorturl_sample: shorturlData.slice(0, 1)
|
||
};
|
||
}
|
||
} catch (error) {
|
||
const errorMessage = `同步过程中发生错误: ${(error as Error).message}`;
|
||
log(errorMessage);
|
||
if ((error as Error).stack) {
|
||
log(`错误堆栈: ${(error as Error).stack}`, true);
|
||
}
|
||
return { success: false, message: errorMessage };
|
||
} finally {
|
||
if (pgPool) {
|
||
await pgPool.end();
|
||
log("PostgreSQL连接池已关闭", true);
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 更新同步状态
|
||
*/
|
||
async function updateSyncState(lastSyncTime: string, recordsSynced: number, log: (message: string, isVerbose?: boolean) => void): Promise<void> {
|
||
try {
|
||
const newState: SyncState = {
|
||
last_sync_time: lastSyncTime,
|
||
records_synced: recordsSynced,
|
||
last_run: new Date().toISOString()
|
||
};
|
||
|
||
await setVariable(SYNC_STATE_KEY, newState);
|
||
log(`同步状态已更新: 最后同步时间 ${lastSyncTime}, 累计同步记录数 ${recordsSynced}`, true);
|
||
} catch (error) {
|
||
log(`更新同步状态失败: ${error}`, true);
|
||
// 继续执行，不中断同步过程
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 从PostgreSQL获取shorturl数据
|
||
*/
|
||
async function getShortUrlData(
|
||
pgPool: Pool,
|
||
startTime: string,
|
||
endTime: string,
|
||
log: (message: string, isVerbose?: boolean) => void
|
||
): Promise<ShortUrlData[]> {
|
||
const client = await pgPool.connect();
|
||
|
||
try {
|
||
log(`获取shorturl数据 (${startTime} 至 ${endTime})`, true);
|
||
|
||
const query = `
|
||
SELECT
|
||
id,
|
||
slug,
|
||
origin,
|
||
title,
|
||
description,
|
||
domain,
|
||
created_at,
|
||
updated_at,
|
||
deleted_at,
|
||
expired_at as expires_at
|
||
FROM
|
||
short_url.shorturl
|
||
WHERE
|
||
(created_at >= $1 AND created_at <= $2)
|
||
OR (updated_at >= $1 AND updated_at <= $2)
|
||
`;
|
||
|
||
const result = await client.queryObject(query, [startTime, endTime]);
|
||
return result.rows as ShortUrlData[];
|
||
} finally {
|
||
client.release();
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 格式化日期时间为ClickHouse可接受的格式
|
||
*/
|
||
function formatDateTime(dateStr: string | null | undefined): string {
|
||
if (!dateStr) return 'NULL';
|
||
|
||
try {
|
||
// 将日期字符串转换为ISO格式
|
||
const date = new Date(dateStr);
|
||
if (isNaN(date.getTime())) {
|
||
return 'NULL';
|
||
}
|
||
|
||
// 返回ISO格式的日期字符串，ClickHouse可以解析
|
||
return `parseDateTimeBestEffort('${date.toISOString()}')`;
|
||
} catch (error) {
|
||
console.error(`日期格式化错误: ${error}`);
|
||
return 'NULL';
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 格式化进度显示
|
||
*/
|
||
function formatProgress(current: number, total: number): string {
|
||
const percent = Math.round((current / total) * 100);
|
||
const progressBar = '[' + '='.repeat(Math.floor(percent / 5)) + ' '.repeat(20 - Math.floor(percent / 5)) + ']';
|
||
return `${progressBar} ${percent}% (${current}/${total})`;
|
||
}
|
||
|
||
/**
|
||
* 更新ClickHouse中的shorturl表数据
|
||
*/
|
||
async function updateClickHouseShortUrl(
|
||
shorturls: ShortUrlData[],
|
||
chConfig: ChConfig,
|
||
log: (message: string, isVerbose?: boolean) => void
|
||
): Promise<number> {
|
||
if (shorturls.length === 0) {
|
||
log('没有找到shorturl数据，跳过shorturl表更新');
|
||
return 0;
|
||
}
|
||
|
||
log(`准备更新 ${shorturls.length} 条shorturl数据`);
|
||
|
||
// 检查ClickHouse中是否存在shorturl表
|
||
const tableExists = await checkClickHouseTable(chConfig, 'shorturl_analytics.shorturl');
|
||
|
||
if (!tableExists) {
|
||
log('ClickHouse中未找到shorturl表，请先创建表');
|
||
return 0;
|
||
}
|
||
|
||
let updatedCount = 0;
|
||
const startTime = Date.now();
|
||
|
||
// 使用批量插入更高效
|
||
const batchSize = 50; // 降低批次大小，使查询更稳定
|
||
for (let i = 0; i < shorturls.length; i += batchSize) {
|
||
const batch = shorturls.slice(i, i + batchSize);
|
||
let successCount = 0;
|
||
|
||
// 显示批处理进度信息
|
||
const batchNumber = Math.floor(i / batchSize) + 1;
|
||
const totalBatches = Math.ceil(shorturls.length / batchSize);
|
||
log(`处理批次 ${batchNumber}/${totalBatches}: ${formatProgress(i, shorturls.length)}`);
|
||
|
||
// 对每条记录使用单独的INSERT ... SELECT ... WHERE NOT EXISTS语句
|
||
for (let j = 0; j < batch.length; j++) {
|
||
const shorturl = batch[j];
|
||
// 显示记录处理细节进度
|
||
const overallProgress = i + j + 1;
|
||
if (overallProgress % 10 === 0 || overallProgress === shorturls.length) {
|
||
// 每10条记录或最后一条记录显示一次进度
|
||
const elapsedSeconds = (Date.now() - startTime) / 1000;
|
||
const recordsPerSecond = overallProgress / elapsedSeconds;
|
||
const remainingRecords = shorturls.length - overallProgress;
|
||
const estimatedSecondsRemaining = remainingRecords / recordsPerSecond;
|
||
|
||
log(`总进度: ${formatProgress(overallProgress, shorturls.length)} - 速率: ${recordsPerSecond.toFixed(1)}条/秒 - 预计剩余时间: ${formatTime(estimatedSecondsRemaining)}`);
|
||
}
|
||
|
||
try {
|
||
const insertQuery = `
|
||
INSERT INTO shorturl_analytics.shorturl
|
||
SELECT
|
||
'${escapeString(shorturl.id)}' AS id,
|
||
'${escapeString(shorturl.id)}' AS external_id,
|
||
'shorturl' AS type,
|
||
'${escapeString(shorturl.slug)}' AS slug,
|
||
'${escapeString(shorturl.origin)}' AS original_url,
|
||
${shorturl.title ? `'${escapeString(shorturl.title)}'` : 'NULL'} AS title,
|
||
${shorturl.description ? `'${escapeString(shorturl.description)}'` : 'NULL'} AS description,
|
||
'{}' AS attributes,
|
||
1 AS schema_version,
|
||
'' AS creator_id,
|
||
'' AS creator_email,
|
||
'' AS creator_name,
|
||
${formatDateTime(shorturl.created_at)} AS created_at,
|
||
${formatDateTime(shorturl.updated_at)} AS updated_at,
|
||
${formatDateTime(shorturl.deleted_at)} AS deleted_at,
|
||
'[]' AS projects,
|
||
'[]' AS teams,
|
||
'[]' AS tags,
|
||
'[]' AS qr_codes,
|
||
'[]' AS channels,
|
||
'[]' AS favorites,
|
||
${formatDateTime(shorturl.expires_at)} AS expires_at,
|
||
0 AS click_count,
|
||
0 AS unique_visitors,
|
||
${shorturl.domain ? `'${escapeString(shorturl.domain)}'` : 'NULL'} AS domain
|
||
WHERE NOT EXISTS (
|
||
SELECT 1 FROM shorturl_analytics.shorturl WHERE id = '${escapeString(shorturl.id)}'
|
||
)
|
||
`;
|
||
|
||
await executeClickHouseQuery(chConfig, insertQuery);
|
||
successCount++;
|
||
log(`成功处理shorturl: ${shorturl.id}`, true);
|
||
} catch (error) {
|
||
log(`处理shorturl ${shorturl.id} 失败: ${(error as Error).message}`);
|
||
|
||
// 尝试使用简单插入作为备选方案
|
||
try {
|
||
log(`尝试替代方法更新: ${shorturl.id}`, true);
|
||
|
||
// 先检查记录是否存在
|
||
const checkQuery = `SELECT count() FROM shorturl_analytics.shorturl WHERE id = '${escapeString(shorturl.id)}'`;
|
||
const existsResult = await executeClickHouseQuery(chConfig, checkQuery);
|
||
const exists = parseInt(existsResult.trim()) > 0;
|
||
|
||
if (!exists) {
|
||
const fallbackQuery = `
|
||
INSERT INTO shorturl_analytics.shorturl (
|
||
id, external_id, type, slug, original_url,
|
||
title, description, attributes, schema_version,
|
||
creator_id, creator_email, creator_name,
|
||
created_at, updated_at, deleted_at,
|
||
projects, teams, tags, qr_codes, channels, favorites,
|
||
expires_at, click_count, unique_visitors, domain
|
||
) VALUES (
|
||
'${escapeString(shorturl.id)}',
|
||
'${escapeString(shorturl.id)}',
|
||
'shorturl',
|
||
'${escapeString(shorturl.slug)}',
|
||
'${escapeString(shorturl.origin)}',
|
||
${shorturl.title ? `'${escapeString(shorturl.title)}'` : 'NULL'},
|
||
${shorturl.description ? `'${escapeString(shorturl.description)}'` : 'NULL'},
|
||
'{}',
|
||
1,
|
||
'',
|
||
'',
|
||
'',
|
||
${formatDateTime(shorturl.created_at)},
|
||
${formatDateTime(shorturl.updated_at)},
|
||
${formatDateTime(shorturl.deleted_at)},
|
||
'[]',
|
||
'[]',
|
||
'[]',
|
||
'[]',
|
||
'[]',
|
||
'[]',
|
||
${formatDateTime(shorturl.expires_at)},
|
||
0,
|
||
0,
|
||
${shorturl.domain ? `'${escapeString(shorturl.domain)}'` : 'NULL'}
|
||
)
|
||
`;
|
||
|
||
await executeClickHouseQuery(chConfig, fallbackQuery);
|
||
successCount++;
|
||
log(`备选方式插入成功: ${shorturl.id}`, true);
|
||
} else {
|
||
log(`记录已存在，跳过: ${shorturl.id}`, true);
|
||
}
|
||
} catch (fallbackError) {
|
||
log(`备选方式失败 ${shorturl.id}: ${(fallbackError as Error).message}`);
|
||
}
|
||
}
|
||
}
|
||
|
||
updatedCount += successCount;
|
||
log(`批次 ${batchNumber}/${totalBatches} 完成: ${successCount}/${batch.length} 条成功 (总计: ${updatedCount}/${shorturls.length})`);
|
||
}
|
||
|
||
const totalTime = (Date.now() - startTime) / 1000;
|
||
log(`同步完成! 总计处理: ${updatedCount}/${shorturls.length} 条记录, 耗时: ${formatTime(totalTime)}, 平均速率: ${(updatedCount / totalTime).toFixed(1)}条/秒`);
|
||
|
||
return updatedCount;
|
||
}
|
||
|
||
/**
|
||
* 获取ClickHouse配置
|
||
*/
|
||
async function getClickHouseConfig(): Promise<ChConfig> {
|
||
try {
|
||
const chConfigJson = await getVariable("f/shorturl_analytics/clickhouse");
|
||
|
||
// 确保配置不为空
|
||
if (!chConfigJson) {
|
||
throw new Error("未找到ClickHouse配置");
|
||
}
|
||
|
||
// 解析JSON字符串为对象
|
||
let chConfig: ChConfig;
|
||
if (typeof chConfigJson === 'string') {
|
||
try {
|
||
chConfig = JSON.parse(chConfigJson);
|
||
} catch {
|
||
throw new Error("ClickHouse配置不是有效的JSON");
|
||
}
|
||
} else {
|
||
chConfig = chConfigJson as ChConfig;
|
||
}
|
||
|
||
// 验证并构建URL
|
||
if (!chConfig.clickhouse_url && chConfig.clickhouse_host && chConfig.clickhouse_port) {
|
||
chConfig.clickhouse_url = `http://${chConfig.clickhouse_host}:${chConfig.clickhouse_port}`;
|
||
}
|
||
|
||
if (!chConfig.clickhouse_url) {
|
||
throw new Error("ClickHouse配置缺少URL");
|
||
}
|
||
|
||
return chConfig;
|
||
} catch (error) {
|
||
throw new Error(`获取ClickHouse配置失败: ${(error as Error).message}`);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 检查ClickHouse中是否存在指定表
|
||
*/
|
||
async function checkClickHouseTable(chConfig: ChConfig, tableName: string): Promise<boolean> {
|
||
try {
|
||
const query = `EXISTS TABLE ${tableName}`;
|
||
const result = await executeClickHouseQuery(chConfig, query);
|
||
return result.trim() === '1';
|
||
} catch (error) {
|
||
console.error(`检查表 ${tableName} 失败:`, error);
|
||
return false;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 执行ClickHouse查询
|
||
*/
|
||
async function executeClickHouseQuery(chConfig: ChConfig, query: string): Promise<string> {
|
||
// 确保URL有效
|
||
if (!chConfig.clickhouse_url) {
|
||
throw new Error("无效的ClickHouse URL: 未定义");
|
||
}
|
||
|
||
// 执行HTTP请求
|
||
try {
|
||
const response = await fetch(chConfig.clickhouse_url, {
|
||
method: "POST",
|
||
headers: {
|
||
"Content-Type": "application/x-www-form-urlencoded",
|
||
"Authorization": `Basic ${btoa(`${chConfig.clickhouse_user}:${chConfig.clickhouse_password}`)}`
|
||
},
|
||
body: query,
|
||
});
|
||
|
||
if (!response.ok) {
|
||
const errorText = await response.text();
|
||
throw new Error(`ClickHouse查询失败 (${response.status}): ${errorText}`);
|
||
}
|
||
|
||
return await response.text();
|
||
} catch (error) {
|
||
throw new Error(`执行ClickHouse查询失败: ${(error as Error).message}`);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 转义字符串，避免SQL注入
|
||
*/
|
||
function escapeString(str: string): string {
|
||
if (!str) return '';
|
||
return str.replace(/'/g, "''");
|
||
}
|
||
|
||
/**
|
||
* 格式化时间（秒）为可读格式
|
||
*/
|
||
function formatTime(seconds: number): string {
|
||
const mins = Math.floor(seconds / 60);
|
||
const secs = Math.floor(seconds % 60);
|
||
|
||
if (mins === 0) {
|
||
return `${secs}秒`;
|
||
} else {
|
||
return `${mins}分${secs}秒`;
|
||
}
|
||
}
|