From 2047e4eb2e498dbe9146eae8c47f885a4176def9 Mon Sep 17 00:00:00 2001 From: chenpei Date: Wed, 8 Jul 2026 14:35:15 +0800 Subject: [PATCH] =?UTF-8?q?=E5=9B=BA=E5=8C=96=E5=85=AC=E7=BD=91=E6=B5=81?= =?UTF-8?q?=E9=87=8F=E6=B1=87=E6=80=BB=E9=93=BE=E8=B7=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Dockerfile | 4 + .../internal/public-traffic/rollups/route.ts | 49 ++ lib/__tests__/public-traffic-stats.test.ts | 15 + lib/public-traffic-stats.ts | 559 ++++++++++++++++++ scripts/git-incremental-deploy.ps1 | 161 +++++ scripts/public-traffic-rollup-from-nginx.py | 272 +++++++++ 6 files changed, 1060 insertions(+) create mode 100644 app/api/internal/public-traffic/rollups/route.ts create mode 100644 scripts/public-traffic-rollup-from-nginx.py diff --git a/Dockerfile b/Dockerfile index 3443d2d..de9df8f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -82,6 +82,8 @@ RUN sed -i "s|https://dl-cdn.alpinelinux.org/alpine|${ALPINE_MIRROR}|g" /etc/apk RUN addgroup --system --gid 1001 nodejs && \ adduser --system --uid 1001 nextjs +RUN mkdir -p /app/scripts + # 复制必要文件 COPY --from=builder --chown=nextjs:nodejs /app/.next/standalone ./ COPY --from=builder --chown=nextjs:nodejs /app/.next/static ./.next/static @@ -98,9 +100,11 @@ COPY --from=builder --chown=nextjs:nodejs /app/node_modules/@smithy ./node_modul # 只复制 schema.prisma，不复制 migrations 和 engine-cache（减少 35MB） COPY --from=builder --chown=nextjs:nodejs /app/prisma/schema.prisma ./prisma/schema.prisma COPY --from=builder --chown=nextjs:nodejs /app/lib ./lib +COPY --from=builder --chown=nextjs:nodejs /app/scripts/public-traffic-rollup-from-nginx.py ./scripts/public-traffic-rollup-from-nginx.py # 创建必要目录并设置权限 RUN mkdir -p /app/public /app/uploads && \ + mkdir -p /app/scripts && \ mkdir -p /app/uploads/temp && \ mkdir -p /app/uploads/assembled && \ mkdir -p /app/uploads/materials && \ diff --git a/app/api/internal/public-traffic/rollups/route.ts b/app/api/internal/public-traffic/rollups/route.ts new file mode 100644 index 0000000..1c095db --- /dev/null +++ b/app/api/internal/public-traffic/rollups/route.ts @@ -0,0 +1,49 @@ +import { NextRequest, NextResponse } from 'next/server' +import { + upsertPublicTrafficRollups, + type PublicTrafficRollupInput, +} from '@/lib/public-traffic-stats' + +export const runtime = 'nodejs' +export const dynamic = 'force-dynamic' + +const MAX_ROWS_PER_REQUEST = 2000 + +function readInternalToken(request: NextRequest) { + const bearer = request.headers.get('authorization')?.match(/^Bearer\s+(.+)$/i)?.[1] + return bearer || request.headers.get('x-internal-job-token') || '' +} + +function normalizeRows(payload: unknown): PublicTrafficRollupInput[] { + if (!payload || typeof payload !== 'object') return [] + const rows = (payload as { rows?: unknown }).rows + return Array.isArray(rows) ? rows.slice(0, MAX_ROWS_PER_REQUEST) as PublicTrafficRollupInput[] : [] +} + +export async function POST(request: NextRequest) { + const expectedToken = process.env.PUBLIC_TRAFFIC_ROLLUP_TOKEN + + if (!expectedToken) { + return NextResponse.json({ error: '公网流量汇总写入未启用' }, { status: 404 }) + } + + if (readInternalToken(request) !== expectedToken) { + return NextResponse.json({ error: '无权限写入公网流量汇总' }, { status: 403 }) + } + + const payload = await request.json().catch(() => null) + const rows = normalizeRows(payload) + + if (rows.length === 0) { + return NextResponse.json({ error: '没有可写入的公网流量汇总行' }, { status: 400 }) + } + + const startedAt = Date.now() + const result = await upsertPublicTrafficRollups(rows) + + return NextResponse.json({ + ok: true, + durationMs: Date.now() - startedAt, + ...result, + }) +} diff --git a/lib/__tests__/public-traffic-stats.test.ts b/lib/__tests__/public-traffic-stats.test.ts index 2ebe8f3..af7ae84 100644 --- a/lib/__tests__/public-traffic-stats.test.ts +++ b/lib/__tests__/public-traffic-stats.test.ts @@ -3,6 +3,7 @@ import os from 'os' import path from 'path' import { classifyPublicTrafficPurpose, + getPublicTrafficPresetRange, getPublicTrafficStats, normalizeTrafficRange, parsePublicTrafficLogLine, @@ -52,6 +53,20 @@ describe('public traffic stats', () => { expect(range.end.toISOString()).toBe('2026-07-04T15:59:59.999Z') }) + test('builds public traffic presets using China business dates', () => { + const now = new Date('2026-07-08T05:00:00.000Z') + const today = getPublicTrafficPresetRange('today', now) + const last7 = getPublicTrafficPresetRange('last7', now) + const last30 = getPublicTrafficPresetRange('last30', now) + + expect(today.start.toISOString()).toBe('2026-07-07T16:00:00.000Z') + expect(today.end.toISOString()).toBe('2026-07-08T15:59:59.999Z') + expect(last7.start.toISOString()).toBe('2026-07-01T16:00:00.000Z') + expect(last7.end.toISOString()).toBe('2026-07-08T15:59:59.999Z') + expect(last30.start.toISOString()).toBe('2026-06-08T16:00:00.000Z') + expect(last30.end.toISOString()).toBe('2026-07-08T15:59:59.999Z') + }) + test('groups daily traffic by China business date', async () => { const previousPaths = process.env.PUBLIC_TRAFFIC_LOG_PATHS const tempDir = mkdtempSync(path.join(os.tmpdir(), 'public-traffic-')) diff --git a/lib/public-traffic-stats.ts b/lib/public-traffic-stats.ts index f25437a..3d48e19 100644 --- a/lib/public-traffic-stats.ts +++ b/lib/public-traffic-stats.ts @@ -123,6 +123,34 @@ interface PublicTrafficStatsOptions { forceRefresh?: boolean } +export type PublicTrafficPreset = 'today' | 'last7' | 'last30' + +export type PublicTrafficRollupGranularity = '5m' | 'day' + +export interface PublicTrafficRollupInput { + bucketStart: string + bucketEnd: string + businessDate?: string | null + granularity: PublicTrafficRollupGranularity + domain: string + method?: string | null + path: string + responseBytes: number + requestBytes?: number | null + requestBytesKnownCount?: number | null + totalRequests: number + statusCounts?: Partial> | null + sourceServer?: string | null + generatedAt?: string | null + frozen?: boolean | null +} + +export interface PublicTrafficRollupImportResult { + received: number + accepted: number + skipped: number +} + interface PublicTrafficScanContext { logPaths: string[] maxLines: number @@ -137,6 +165,29 @@ interface PersistentTrafficCacheRow { expires_at: Date | string } +interface PublicTrafficRollupRow { + bucket_start: Date | string + bucket_end: Date | string + business_date: string + granularity: PublicTrafficRollupGranularity + domain: string + method: string + path: string + purpose_key: string + purpose_label: string + purpose_category: string + purpose_detail: string + response_bytes: bigint | number | string + request_bytes: bigint | number | string | null + request_bytes_known_count: number + total_requests: number + status_2xx: number + status_3xx: number + status_4xx: number + status_5xx: number + status_other: number +} + const MONTHS: Record = { Jan: 0, Feb: 1, @@ -165,6 +216,7 @@ const statsCache = new Map() const MAX_CACHE_ENTRIES = 20 const refreshPromises = new Map>() let persistentCacheTableEnsured = false +let rollupTableEnsured = false function emptyStatusCounts(): Record { return { @@ -218,6 +270,26 @@ export function normalizeTrafficRange(startValue?: string | null, endValue?: str return { start, end } } +function toChinaDateString(date: Date): string { + return new Intl.DateTimeFormat('en-CA', { + timeZone: 'Asia/Shanghai', + year: 'numeric', + month: '2-digit', + day: '2-digit', + }).format(date) +} + +function addDays(date: Date, days: number): Date { + return new Date(date.getTime() + days * 24 * 60 * 60 * 1000) +} + +export function getPublicTrafficPresetRange(preset: PublicTrafficPreset, now = new Date()): PublicTrafficRange { + const endDate = toChinaDateString(now) + const daysBack = preset === 'today' ? 0 : preset === 'last7' ? 6 : 29 + const startDate = toChinaDateString(addDays(now, -daysBack)) + return normalizeTrafficRange(startDate, endDate) +} + function parseNginxTime(value: string): Date | null { const match = value.match(/^(\d{1,2})\/([A-Za-z]{3})\/(\d{4}):(\d{2}):(\d{2}):(\d{2}) ([+-]\d{4})$/) if (!match) return null @@ -941,6 +1013,486 @@ async function writePersistentTrafficCache( ) } +function bigintToNumber(value: bigint | number | string | null | undefined): number { + if (value === null || value === undefined) return 0 + if (typeof value === 'bigint') return Number(value) + if (typeof value === 'number') return Number.isFinite(value) ? value : 0 + const parsed = Number(value) + return Number.isFinite(parsed) ? parsed : 0 +} + +function normalizeStatusCounts( + counts?: Partial> | null, +): Record { + return { + '2xx': Math.max(0, Math.floor(Number(counts?.['2xx'] || 0))), + '3xx': Math.max(0, Math.floor(Number(counts?.['3xx'] || 0))), + '4xx': Math.max(0, Math.floor(Number(counts?.['4xx'] || 0))), + '5xx': Math.max(0, Math.floor(Number(counts?.['5xx'] || 0))), + other: Math.max(0, Math.floor(Number(counts?.other || 0))), + } +} + +function normalizeRollupInput(input: PublicTrafficRollupInput) { + const bucketStart = new Date(input.bucketStart) + const bucketEnd = new Date(input.bucketEnd) + const domain = String(input.domain || '').trim().toLowerCase() + const method = String(input.method || 'GET').trim().toUpperCase() || 'GET' + const pathValue = normalizeRequestTarget(`${method} ${String(input.path || '').trim() || '/'} HTTP/1.1`).path + const responseBytes = Math.max(0, Math.floor(Number(input.responseBytes || 0))) + const totalRequests = Math.max(0, Math.floor(Number(input.totalRequests || 0))) + const requestBytes = + input.requestBytes === null || input.requestBytes === undefined + ? null + : Math.max(0, Math.floor(Number(input.requestBytes || 0))) + const requestBytesKnownCount = Math.max(0, Math.floor(Number(input.requestBytesKnownCount || 0))) + const statusCounts = normalizeStatusCounts(input.statusCounts) + + if ( + Number.isNaN(bucketStart.getTime()) || + Number.isNaN(bucketEnd.getTime()) || + bucketStart >= bucketEnd || + !domain || + !pathValue || + totalRequests <= 0 || + !['5m', 'day'].includes(input.granularity) + ) { + return null + } + + const purpose = classifyPublicTrafficPurpose(domain, pathValue, method) + const businessDate = input.businessDate && /^\d{4}-\d{2}-\d{2}$/.test(input.businessDate) + ? input.businessDate + : toChinaDateString(bucketStart) + const generatedAt = input.generatedAt ? new Date(input.generatedAt) : new Date() + + return { + bucketStart: bucketStart.toISOString(), + bucketEnd: bucketEnd.toISOString(), + businessDate, + granularity: input.granularity, + domain, + method, + path: pathValue, + purposeKey: purpose.key, + purposeLabel: purpose.label, + purposeCategory: purpose.category, + purposeDetail: purpose.detail, + responseBytes, + requestBytes, + requestBytesKnownCount, + totalRequests, + status2xx: statusCounts['2xx'], + status3xx: statusCounts['3xx'], + status4xx: statusCounts['4xx'], + status5xx: statusCounts['5xx'], + statusOther: statusCounts.other, + sourceServer: String(input.sourceServer || '').trim(), + generatedAt: Number.isNaN(generatedAt.getTime()) ? new Date().toISOString() : generatedAt.toISOString(), + frozen: Boolean(input.frozen), + } +} + +async function ensurePublicTrafficRollupTable() { + if (rollupTableEnsured) return true + if (!persistentTrafficCacheEnabled()) return false + + const { prisma } = await import('./prisma') + + await prisma.$executeRawUnsafe(` + CREATE TABLE IF NOT EXISTS "public_traffic_rollups" ( + "id" BIGSERIAL PRIMARY KEY, + "bucket_start" TIMESTAMPTZ NOT NULL, + "bucket_end" TIMESTAMPTZ NOT NULL, + "business_date" TEXT NOT NULL, + "granularity" TEXT NOT NULL, + "domain" TEXT NOT NULL, + "method" TEXT NOT NULL, + "path" TEXT NOT NULL, + "purpose_key" TEXT NOT NULL, + "purpose_label" TEXT NOT NULL, + "purpose_category" TEXT NOT NULL, + "purpose_detail" TEXT NOT NULL, + "response_bytes" BIGINT NOT NULL DEFAULT 0, + "request_bytes" BIGINT, + "request_bytes_known_count" INTEGER NOT NULL DEFAULT 0, + "total_requests" INTEGER NOT NULL DEFAULT 0, + "status_2xx" INTEGER NOT NULL DEFAULT 0, + "status_3xx" INTEGER NOT NULL DEFAULT 0, + "status_4xx" INTEGER NOT NULL DEFAULT 0, + "status_5xx" INTEGER NOT NULL DEFAULT 0, + "status_other" INTEGER NOT NULL DEFAULT 0, + "source_server" TEXT NOT NULL DEFAULT '', + "generated_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + "frozen" BOOLEAN NOT NULL DEFAULT false, + "created_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updated_at" TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + CONSTRAINT "public_traffic_rollups_unique_bucket_path" + UNIQUE ("business_date", "granularity", "bucket_start", "bucket_end", "domain", "method", "path") + ) + `) + await prisma.$executeRawUnsafe(` + CREATE INDEX IF NOT EXISTS "public_traffic_rollups_date_granularity_idx" + ON "public_traffic_rollups" ("business_date", "granularity") + `) + await prisma.$executeRawUnsafe(` + CREATE INDEX IF NOT EXISTS "public_traffic_rollups_bucket_idx" + ON "public_traffic_rollups" ("bucket_start", "bucket_end") + `) + + rollupTableEnsured = true + return true +} + +export async function upsertPublicTrafficRollups( + inputs: PublicTrafficRollupInput[], +): Promise { + if (!(await ensurePublicTrafficRollupTable())) { + throw new Error('公网流量汇总表未启用') + } + + const rows: Array>> = [] + for (const input of inputs) { + const row = normalizeRollupInput(input) + if (row) rows.push(row) + } + + if (rows.length === 0) { + return { received: inputs.length, accepted: 0, skipped: inputs.length } + } + + const { prisma } = await import('./prisma') + const chunkSize = 500 + + for (let index = 0; index < rows.length; index += chunkSize) { + const chunk = rows.slice(index, index + chunkSize) + await prisma.$executeRawUnsafe( + ` + INSERT INTO "public_traffic_rollups" ( + "bucket_start", + "bucket_end", + "business_date", + "granularity", + "domain", + "method", + "path", + "purpose_key", + "purpose_label", + "purpose_category", + "purpose_detail", + "response_bytes", + "request_bytes", + "request_bytes_known_count", + "total_requests", + "status_2xx", + "status_3xx", + "status_4xx", + "status_5xx", + "status_other", + "source_server", + "generated_at", + "frozen", + "updated_at" + ) + SELECT + "bucketStart"::timestamptz, + "bucketEnd"::timestamptz, + "businessDate", + "granularity", + "domain", + "method", + "path", + "purposeKey", + "purposeLabel", + "purposeCategory", + "purposeDetail", + "responseBytes"::bigint, + NULLIF("requestBytes", '')::bigint, + "requestBytesKnownCount"::integer, + "totalRequests"::integer, + "status2xx"::integer, + "status3xx"::integer, + "status4xx"::integer, + "status5xx"::integer, + "statusOther"::integer, + "sourceServer", + "generatedAt"::timestamptz, + "frozen"::boolean, + CURRENT_TIMESTAMP + FROM jsonb_to_recordset($1::jsonb) AS x( + "bucketStart" text, + "bucketEnd" text, + "businessDate" text, + "granularity" text, + "domain" text, + "method" text, + "path" text, + "purposeKey" text, + "purposeLabel" text, + "purposeCategory" text, + "purposeDetail" text, + "responseBytes" text, + "requestBytes" text, + "requestBytesKnownCount" text, + "totalRequests" text, + "status2xx" text, + "status3xx" text, + "status4xx" text, + "status5xx" text, + "statusOther" text, + "sourceServer" text, + "generatedAt" text, + "frozen" text + ) + ON CONFLICT ("business_date", "granularity", "bucket_start", "bucket_end", "domain", "method", "path") + DO UPDATE SET + "purpose_key" = EXCLUDED."purpose_key", + "purpose_label" = EXCLUDED."purpose_label", + "purpose_category" = EXCLUDED."purpose_category", + "purpose_detail" = EXCLUDED."purpose_detail", + "response_bytes" = EXCLUDED."response_bytes", + "request_bytes" = EXCLUDED."request_bytes", + "request_bytes_known_count" = EXCLUDED."request_bytes_known_count", + "total_requests" = EXCLUDED."total_requests", + "status_2xx" = EXCLUDED."status_2xx", + "status_3xx" = EXCLUDED."status_3xx", + "status_4xx" = EXCLUDED."status_4xx", + "status_5xx" = EXCLUDED."status_5xx", + "status_other" = EXCLUDED."status_other", + "source_server" = EXCLUDED."source_server", + "generated_at" = EXCLUDED."generated_at", + "frozen" = EXCLUDED."frozen", + "updated_at" = CURRENT_TIMESTAMP + `, + JSON.stringify( + chunk.map((row) => ({ + ...row, + requestBytes: row.requestBytes === null ? '' : String(row.requestBytes), + responseBytes: String(row.responseBytes), + requestBytesKnownCount: String(row.requestBytesKnownCount), + totalRequests: String(row.totalRequests), + status2xx: String(row.status2xx), + status3xx: String(row.status3xx), + status4xx: String(row.status4xx), + status5xx: String(row.status5xx), + statusOther: String(row.statusOther), + frozen: String(row.frozen), + })), + ), + ) + } + + return { received: inputs.length, accepted: rows.length, skipped: inputs.length - rows.length } +} + +function addRollupRow(stats: PublicTrafficStats, row: PublicTrafficRollupRow) { + const responseBytes = bigintToNumber(row.response_bytes) + const requestBytes = row.request_bytes === null ? null : bigintToNumber(row.request_bytes) + const requestBytesKnownCount = Number(row.request_bytes_known_count || 0) + const totalRequests = Number(row.total_requests || 0) + const bucketStart = row.bucket_start instanceof Date ? row.bucket_start : new Date(row.bucket_start) + const bucketEnd = row.bucket_end instanceof Date ? row.bucket_end : new Date(row.bucket_end) + const statusCounts: Record = { + '2xx': Number(row.status_2xx || 0), + '3xx': Number(row.status_3xx || 0), + '4xx': Number(row.status_4xx || 0), + '5xx': Number(row.status_5xx || 0), + other: Number(row.status_other || 0), + } + + stats.summary.responseBytes += responseBytes + if (requestBytesKnownCount > 0) { + stats.summary.requestBytes = (stats.summary.requestBytes || 0) + (requestBytes || 0) + stats.summary.requestBytesKnownCount += requestBytesKnownCount + } + stats.summary.totalRequests += totalRequests + for (const bucket of Object.keys(statusCounts) as TrafficStatusBucket[]) { + stats.summary.statusCounts[bucket] += statusCounts[bucket] + } + + const firstSeen = Number.isNaN(bucketStart.getTime()) ? null : bucketStart.toISOString() + const lastSeen = Number.isNaN(bucketEnd.getTime()) ? null : bucketEnd.toISOString() + if (firstSeen && (!stats.summary.firstSeen || firstSeen < stats.summary.firstSeen)) stats.summary.firstSeen = firstSeen + if (lastSeen && (!stats.summary.lastSeen || lastSeen > stats.summary.lastSeen)) stats.summary.lastSeen = lastSeen + + let dayItem = stats.byDay.find((item) => item.date === row.business_date) + if (!dayItem) { + dayItem = { + date: row.business_date, + responseBytes: 0, + requestBytes: null, + requestBytesKnownCount: 0, + totalRequests: 0, + } + stats.byDay.push(dayItem) + } + dayItem.responseBytes += responseBytes + dayItem.totalRequests += totalRequests + if (requestBytesKnownCount > 0) { + dayItem.requestBytes = (dayItem.requestBytes || 0) + (requestBytes || 0) + dayItem.requestBytesKnownCount += requestBytesKnownCount + } + + let domainItem = stats.byDomain.find((item) => item.domain === row.domain) + if (!domainItem) { + domainItem = { + domain: row.domain, + responseBytes: 0, + requestBytes: null, + requestBytesKnownCount: 0, + totalRequests: 0, + statusCounts: emptyStatusCounts(), + } + stats.byDomain.push(domainItem) + } + domainItem.responseBytes += responseBytes + domainItem.totalRequests += totalRequests + if (requestBytesKnownCount > 0) { + domainItem.requestBytes = (domainItem.requestBytes || 0) + (requestBytes || 0) + domainItem.requestBytesKnownCount += requestBytesKnownCount + } + for (const bucket of Object.keys(statusCounts) as TrafficStatusBucket[]) { + domainItem.statusCounts[bucket] += statusCounts[bucket] + } + + let purposeItem = stats.byPurpose.find((item) => item.key === row.purpose_key) + if (!purposeItem) { + purposeItem = { + key: row.purpose_key, + label: row.purpose_label, + category: row.purpose_category, + detail: row.purpose_detail, + responseBytes: 0, + requestBytes: null, + requestBytesKnownCount: 0, + totalRequests: 0, + } + stats.byPurpose.push(purposeItem) + } + purposeItem.responseBytes += responseBytes + purposeItem.totalRequests += totalRequests + if (requestBytesKnownCount > 0) { + purposeItem.requestBytes = (purposeItem.requestBytes || 0) + (requestBytes || 0) + purposeItem.requestBytesKnownCount += requestBytesKnownCount + } + + let topPathItem = stats.topPaths.find((item) => item.method === row.method && item.path === row.path) + if (!topPathItem) { + topPathItem = { + path: row.path, + method: row.method, + purposeLabel: row.purpose_label, + purposeCategory: row.purpose_category, + purposeDetail: row.purpose_detail, + responseBytes: 0, + totalRequests: 0, + } + stats.topPaths.push(topPathItem) + } + topPathItem.responseBytes += responseBytes + topPathItem.totalRequests += totalRequests +} + +async function getPublicTrafficStatsFromRollups(range: PublicTrafficRange): Promise { + if (!(await ensurePublicTrafficRollupTable())) return null + + const startDate = toChinaDateString(range.start) + const endDate = toChinaDateString(range.end) + const todayDate = toChinaDateString(new Date()) + const { prisma } = await import('./prisma') + const rows = await prisma.$queryRawUnsafe( + ` + SELECT + "bucket_start", + "bucket_end", + "business_date", + "granularity", + "domain", + "method", + "path", + "purpose_key", + "purpose_label", + "purpose_category", + "purpose_detail", + "response_bytes", + "request_bytes", + "request_bytes_known_count", + "total_requests", + "status_2xx", + "status_3xx", + "status_4xx", + "status_5xx", + "status_other" + FROM "public_traffic_rollups" + WHERE "business_date" >= $1 + AND "business_date" <= $2 + AND ( + ("business_date" = $3 AND "granularity" = '5m') + OR ("business_date" <> $3 AND "granularity" = 'day') + ) + ORDER BY "bucket_start" ASC + `, + startDate, + endDate, + todayDate, + ) + + if (!rows.length) return null + + const stats: PublicTrafficStats = { + generatedAt: new Date().toISOString(), + available: true, + range: { + start: range.start.toISOString(), + end: range.end.toISOString(), + }, + summary: { + responseBytes: 0, + requestBytes: null, + requestBytesKnownCount: 0, + totalRequests: 0, + uniqueIpCount: 0, + statusCounts: emptyStatusCounts(), + firstSeen: null, + lastSeen: null, + }, + byDay: [], + byDomain: [], + byPurpose: [], + topPaths: [], + source: { + logPaths: [], + scannedFiles: 0, + scannedLines: 0, + matchedLines: 0, + skippedLines: 0, + truncated: false, + issues: [], + requestBytesAvailable: false, + cache: { + persisted: true, + hit: true, + stale: false, + refreshInProgress: false, + refreshedAt: null, + expiresAt: null, + }, + }, + } + + for (const row of rows) addRollupRow(stats, row) + + stats.source.matchedLines = stats.summary.totalRequests + stats.source.requestBytesAvailable = stats.summary.requestBytesKnownCount > 0 + stats.byDay.sort((a, b) => a.date.localeCompare(b.date)) + stats.byDomain.sort((a, b) => b.responseBytes - a.responseBytes) + stats.byPurpose.sort((a, b) => b.responseBytes - a.responseBytes) + stats.topPaths.sort((a, b) => b.responseBytes - a.responseBytes) + stats.topPaths = stats.topPaths.slice(0, 20) + + return stats +} + function getCachedStats(cacheKey: string): PublicTrafficStats | null { const cached = statsCache.get(cacheKey) if (!cached) return null @@ -1061,6 +1613,13 @@ export async function getPublicTrafficStats( range: PublicTrafficRange, options: PublicTrafficStatsOptions = {}, ): Promise { + try { + const rollupStats = await getPublicTrafficStatsFromRollups(range) + if (rollupStats) return rollupStats + } catch (error) { + console.error('[public-traffic] 读取公网流量汇总表失败:', error) + } + const context = getScanContext(range) const cacheKey = context.stableCacheKey diff --git a/scripts/git-incremental-deploy.ps1 b/scripts/git-incremental-deploy.ps1 index 047c65c..c5fc46c 100644 --- a/scripts/git-incremental-deploy.ps1 +++ b/scripts/git-incremental-deploy.ps1 @@ -528,6 +528,10 @@ start_container() { run_args+=(-e "DOWNLOAD_TOKEN_SECRET=$DOWNLOAD_TOKEN_SECRET") fi + if [ -n "${PUBLIC_TRAFFIC_ROLLUP_TOKEN:-}" ]; then + run_args+=(-e "PUBLIC_TRAFFIC_ROLLUP_TOKEN=$PUBLIC_TRAFFIC_ROLLUP_TOKEN") + fi + "${run_args[@]}" "$image" } @@ -557,6 +561,139 @@ PROXY_SERVICE="${CONTAINER_NAME}-local-proxy.service" CURRENT_FILE="${PROXY_ROOT}/current-container" BLUE_CONTAINER="${CONTAINER_NAME}-blue" GREEN_CONTAINER="${CONTAINER_NAME}-green" +PUBLIC_TRAFFIC_SYNC_ROOT="/mnt/PoolOne/apps/youyidian-3d/public-traffic-sync" +PUBLIC_TRAFFIC_ROLLUP_TOKEN_FILE="${PUBLIC_TRAFFIC_SYNC_ROOT}/rollup.token" +PUBLIC_TRAFFIC_LOG_SYNC_SCRIPT="${PUBLIC_TRAFFIC_SYNC_ROOT}/public-traffic-log-sync.sh" +PUBLIC_TRAFFIC_ROLLUP_SCRIPT="${PUBLIC_TRAFFIC_SYNC_ROOT}/public-traffic-rollup-sync.sh" +PUBLIC_TRAFFIC_ROLLUP_SERVICE="public-traffic-rollup-sync.service" +PUBLIC_TRAFFIC_ROLLUP_TIMER="public-traffic-rollup-sync.timer" +PUBLIC_TRAFFIC_ROLLUP_TOKEN="" + +ensure_public_traffic_rollup_token() { + if [ "$TARGET" != "production" ]; then + return 0 + fi + + mkdir -p "$PUBLIC_TRAFFIC_SYNC_ROOT" + chmod 700 "$PUBLIC_TRAFFIC_SYNC_ROOT" + + if [ ! -s "$PUBLIC_TRAFFIC_ROLLUP_TOKEN_FILE" ]; then + if command -v openssl >/dev/null 2>&1; then + openssl rand -hex 32 > "$PUBLIC_TRAFFIC_ROLLUP_TOKEN_FILE" + else + tr -dc 'A-Za-z0-9' "$PUBLIC_TRAFFIC_ROLLUP_TOKEN_FILE" + fi + fi + + chmod 600 "$PUBLIC_TRAFFIC_ROLLUP_TOKEN_FILE" + PUBLIC_TRAFFIC_ROLLUP_TOKEN="$(tr -d '\r\n' < "$PUBLIC_TRAFFIC_ROLLUP_TOKEN_FILE")" +} + +install_public_traffic_rollup_timers() { + if [ "$TARGET" != "production" ]; then + return 0 + fi + + mkdir -p "$PUBLIC_TRAFFIC_SYNC_ROOT" + cat > "$PUBLIC_TRAFFIC_LOG_SYNC_SCRIPT" <<'EOF' +#!/usr/bin/env bash +set -euo pipefail +remote='trafficlog@218.93.9.38' +key='/root/.ssh/public_traffic_log_sync_ed25519' +dest='/mnt/PoolOne/apps/youyidian-3d/public-traffic-logs' +ssh_opts="ssh -i ${key} -o BatchMode=yes -o StrictHostKeyChecking=yes -o ConnectTimeout=10" +domains=( + sc.jiansun.vip + s3.jiansun.vip + cp.jiansun.vip +) +mkdir -p "$dest" +chmod 755 "$dest" +status=0 +for domain in "${domains[@]}"; do + if ! rsync -rtz --timeout=45 --chmod=F644,D755 -e "$ssh_opts" "${remote}:/var/log/nginx/${domain}.access.log*" "$dest/"; then + echo "WARN: failed to sync ${domain}.access.log*" >&2 + status=1 + fi +done +find "$dest" -maxdepth 1 -type f -name '*.access.log*' -exec chmod 0644 {} + +exit "$status" +EOF + chmod 700 "$PUBLIC_TRAFFIC_LOG_SYNC_SCRIPT" + + cat > "/etc/systemd/system/public-traffic-log-sync.service" < "/etc/systemd/system/public-traffic-log-sync.timer" < "$PUBLIC_TRAFFIC_ROLLUP_SCRIPT" </dev/null || true)" +if [ -z "\$container" ] || ! docker inspect "\$container" >/dev/null 2>&1; then + echo "No active material container found" >&2 + exit 1 +fi +systemctl start public-traffic-log-sync.service >/dev/null 2>&1 || true +docker exec "\$container" python3 /app/scripts/public-traffic-rollup-from-nginx.py --granularity 5m +yesterday="\$(TZ=Asia/Shanghai date -d 'yesterday' +%F)" +docker exec "\$container" python3 /app/scripts/public-traffic-rollup-from-nginx.py --granularity day --start-date "\$yesterday" --end-date "\$yesterday" +EOF + chmod 700 "$PUBLIC_TRAFFIC_ROLLUP_SCRIPT" + + cat > "/etc/systemd/system/${PUBLIC_TRAFFIC_ROLLUP_SERVICE}" < "/etc/systemd/system/${PUBLIC_TRAFFIC_ROLLUP_TIMER}" </dev/null 2>&1 || true + systemctl enable "$PUBLIC_TRAFFIC_ROLLUP_TIMER" >/dev/null 2>&1 || true + systemctl restart public-traffic-log-sync.timer >/dev/null 2>&1 || true + systemctl restart "$PUBLIC_TRAFFIC_ROLLUP_TIMER" >/dev/null 2>&1 || true +} is_proxy_running() { systemctl is-active --quiet "$PROXY_SERVICE" 2>/dev/null || { @@ -703,6 +840,8 @@ if is_proxy_running; then PROXY_WAS_RUNNING=1 fi +ensure_public_traffic_rollup_token + log "Starting candidate container $NEXT_CONTAINER" docker rm -f "$NEXT_CONTAINER" >/dev/null 2>&1 || true start_container "$NEXT_CONTAINER" "$REGISTRY_COMMIT" 0 >/tmp/git-deploy-new-container-id.txt @@ -762,6 +901,28 @@ printf '%s\n' "$NEXT_CONTAINER" > "$CURRENT_FILE" ok "Stable local entry is healthy on port $PORT" docker logs --tail 20 "$NEXT_CONTAINER" 2>&1 || true +if install_public_traffic_rollup_timers; then + if [ "$TARGET" = "production" ]; then + if systemctl start "$PUBLIC_TRAFFIC_ROLLUP_SERVICE" >/tmp/public-traffic-rollup-sync.log 2>/tmp/public-traffic-rollup-sync.err; then + ok "Public traffic statistics rollups synced" + else + warn "Public traffic statistics rollup sync failed; page remains available, check /tmp/public-traffic-rollup-sync.err" + cat /tmp/public-traffic-rollup-sync.err >&2 || true + fi + + backfill_start="$(TZ=Asia/Shanghai date -d '7 days ago' +%F)" + backfill_end="$(TZ=Asia/Shanghai date -d 'yesterday' +%F)" + if docker exec "$NEXT_CONTAINER" python3 /app/scripts/public-traffic-rollup-from-nginx.py --granularity day --start-date "$backfill_start" --end-date "$backfill_end" >/tmp/public-traffic-rollup-backfill.log 2>/tmp/public-traffic-rollup-backfill.err; then + ok "Public traffic historical day rollups backfilled from ${backfill_start} to ${backfill_end}" + else + warn "Public traffic historical backfill failed; check /tmp/public-traffic-rollup-backfill.err" + cat /tmp/public-traffic-rollup-backfill.err >&2 || true + fi + fi +else + warn "Public traffic statistics rollup timers were not installed" +fi + if [ -n "$CURRENT_CONTAINER" ] && [ "$CURRENT_CONTAINER" != "$NEXT_CONTAINER" ]; then docker rm -f "$CURRENT_CONTAINER" >/dev/null 2>&1 || true fi diff --git a/scripts/public-traffic-rollup-from-nginx.py b/scripts/public-traffic-rollup-from-nginx.py new file mode 100644 index 0000000..9f4e7ad --- /dev/null +++ b/scripts/public-traffic-rollup-from-nginx.py @@ -0,0 +1,272 @@ +#!/usr/bin/env python3 +import argparse +import gzip +import json +import os +import re +import socket +import sys +import urllib.error +import urllib.parse +import urllib.request +from collections import defaultdict +from datetime import datetime, timedelta, timezone +from pathlib import Path + + +CHINA_TZ = timezone(timedelta(hours=8)) +DOMAINS = ("sc.jiansun.vip", "s3.jiansun.vip", "cp.jiansun.vip") +LOG_RE = re.compile( + r'^(?P\S+) \S+ \S+ \[(?P