import { BadRequestException, ForbiddenException, Inject, Injectable } from '@nestjs/common'; import { ANALYTICS_REVISION, ANALYTICS_VERSION, COUNT_KEYS, Prisma, analyticsHash, analyticsRates, emptyCounts, type AnalyticsCounts } from '@lisglosips/database'; import { PrismaService } from '../database/prisma.service.js'; import type { CurrentUser } from '../security/security.metadata.js'; type Query = Record; export interface AnalyticsQuery { view: 'original' | 'landing'; from: Date; to: Date; customerId: string; caller: string; customerGatewayId: string; vendorId: string; vendorGatewayId: string; city: string; carrier: string; minSamples: number; skip: number; take: number; sort: string; order: string; onlyAlerts: boolean; connectionMin: number; connectionMax: number; overallMin: number; overallMax: number; connectedMin: number; connectedMax: number; } export function parseAnalyticsQuery(raw: Query, now = new Date()): AnalyticsQuery { const str = (key: string, max: number) => { const v = raw[key] ?? ''; if (typeof v !== 'string' || v.length > max) throw new BadRequestException(`Invalid ${key}`); return v.trim(); }; const number = (key: string, fallback: number, min: number, max: number) => { const v = raw[key] === undefined || raw[key] === '' ? fallback : Number(raw[key]); if (!Number.isFinite(v) || v < min || v > max) throw new BadRequestException(`Invalid ${key}`); return v; }; const view = str('view', 16) || 'landing'; if (!['original', 'landing'].includes(view)) throw new BadRequestException('Invalid view'); const to = raw.to ? new Date(str('to', 40)) : now; const from = raw.from ? new Date(str('from', 40)) : new Date(to.getTime() - number('minutes', 15, 1, 10080) * 60000); if (!Number.isFinite(+from) || !Number.isFinite(+to) || from >= to || +to - +from > 7 * 86400000 || +to > +now + 5000) throw new BadRequestException('Time range must be within 7 days and not in future'); const sort = str('sort', 32) || 'totalCalls'; if (!['caller', ...COUNT_KEYS, 'notConnectedCalls', 'connectionRate', 'overallAnswerRate', 'connectedAnswerRate'].includes(sort)) throw new BadRequestException('Invalid sort'); const order = str('order', 4) || 'desc'; if (!['asc', 'desc'].includes(order)) throw new BadRequestException('Invalid order'); const result: AnalyticsQuery = { view: view as AnalyticsQuery['view'], from, to, customerId: str('customerId', 32), caller: str('caller', 64), customerGatewayId: str('customerGatewayId', 32), vendorId: str('vendorId', 32), vendorGatewayId: str('vendorGatewayId', 32), city: str('city', 12), carrier: str('carrier', 20), minSamples: number('minSamples', 0, 0, 1e9), skip: number('skip', 0, 0, 1e6), take: number('take', 25, 1, 100), sort, order, onlyAlerts: raw.onlyAlerts === 'true', connectionMin: number('connectionMin', 0, 0, 100), connectionMax: number('connectionMax', 100, 0, 100), overallMin: number('overallMin', 0, 0, 100), overallMax: number('overallMax', 100, 0, 100), connectedMin: number('connectedMin', 0, 0, 100), connectedMax: number('connectedMax', 100, 0, 100) }; if (!Number.isInteger(result.skip) || !Number.isInteger(result.take) || result.connectionMin > result.connectionMax || result.overallMin > result.overallMax || result.connectedMin > result.connectedMax) throw new BadRequestException('Invalid range'); if (view === 'original' && (result.vendorId || result.vendorGatewayId)) throw new BadRequestException('线路筛选请切换落地主叫,避免混用业务呼叫与尝试分母'); return result; } type Row = Record; const countSql = Prisma.join(COUNT_KEYS.map(k => Prisma.sql`COALESCE(SUM(CAST(JSON_UNQUOTE(JSON_EXTRACT(counts, ${`$.${k}`})) AS SIGNED)),0) AS ${Prisma.raw(k)}`)); const normalize = (row: Row) => ({ ...row, caller: String(row.caller ?? ''), customerId: String(row.customerId ?? ''), ...analyticsRates(Object.fromEntries(COUNT_KEYS.map(k => [k, Number(row[k] ?? 0)])) as unknown as AnalyticsCounts) }); @Injectable() export class CallerAnalyticsService { constructor(@Inject(PrismaService) private readonly db: PrismaService) {} private async scope(user: CurrentUser, customerId: string): Promise { if (user.permissions.includes('caller_analytics.view_all')) return null; const access = await this.db.$queryRaw>`SELECT customer_id FROM caller_analysis_access WHERE user_id=${user.id}`; const ids = access.map(a => a.customer_id); if (!ids.length || (customerId && !ids.includes(customerId))) throw new ForbiddenException('没有该客户的主叫分析权限'); return ids; } private where(q: AnalyticsQuery, scope: string[] | null): Prisma.Sql { const parts = [Prisma.sql`view=${q.view}`]; const fields = { customerId: 'customer_id', customerGatewayId: 'customer_gateway_id', vendorId: 'vendor_id', vendorGatewayId: 'vendor_gateway_id', caller: 'caller', city: 'city', carrier: 'carrier' } as const; for (const [key, column] of Object.entries(fields)) { const v = q[key as keyof typeof fields]; if (v) parts.push(Prisma.sql`${Prisma.raw(column)}=${v}`); } if (scope) parts.push(Prisma.sql`customer_id IN (${Prisma.join(scope)})`); return Prisma.join(parts, ' AND '); } private source(q: AnalyticsQuery, scope: string[] | null) { const start = new Date(Math.ceil(+q.from / 60000) * 60000); const end = new Date(Math.floor(+q.to / 60000) * 60000); const where = this.where(q, scope); if (start >= end) return Prisma.sql`SELECT * FROM caller_analysis_states WHERE ${where} AND started_at>=${q.from} AND started_at<${q.to}`; const columns = Prisma.raw('started_at, customer_id, customer_gateway_id, vendor_id, vendor_gateway_id, caller, city, carrier, counts'); return Prisma.sql`SELECT ${columns} FROM caller_analysis_minute_buckets WHERE ${where} AND started_at>=${start} AND started_at<${end} UNION ALL SELECT ${columns} FROM caller_analysis_states WHERE ${where} AND started_at>=${q.from} AND started_at<${q.to} AND (started_at<${start} OR started_at>=${end})`; } async overview(raw: Query, user: CurrentUser) { const q = parseAnalyticsQuery(raw); const scope = await this.scope(user, q.customerId); const source = this.source(q, scope); const [numbers, trends, health, alerts, gaps] = await this.db.$transaction([ this.db.$queryRaw(Prisma.sql`SELECT customer_id AS customerId, caller, ${countSql} FROM (${source}) a GROUP BY customer_id, caller HAVING SUM(CAST(JSON_UNQUOTE(JSON_EXTRACT(counts,'$.totalCalls')) AS SIGNED))>0 LIMIT 10001`), this.db.$queryRaw(Prisma.sql`SELECT FLOOR(UNIX_TIMESTAMP(started_at)/60)*60000 AS time, ${countSql} FROM (${source}) a GROUP BY time ORDER BY time`), this.db.$queryRaw; updated_at: Date }>>`SELECT payload, updated_at FROM caller_analysis_health WHERE id='worker'`, this.db.$queryRaw>(Prisma.sql`SELECT customer_id, caller, payload, status FROM caller_analysis_alerts WHERE view=${q.view} ${scope ? Prisma.sql`AND customer_id IN (${Prisma.join(scope)})` : Prisma.empty} AND status='ACTIVE'`) ,this.db.$queryRaw>`SELECT id,kind,started_at,ended_at FROM caller_analysis_gaps WHERE status='OPEN' AND started_at<${q.to} AND (ended_at IS NULL OR ended_at>=${q.from}) ORDER BY started_at LIMIT 100` ]); if (numbers.length > 10000) throw new BadRequestException('号码超过10000个,请缩短时间或筛选客户'); const h = health[0]; const captureSince=h?.payload.captureSince?new Date(String(h.payload.captureSince)):null; const coverageStatus=!h||Date.now()-+h.updated_at>10000||h.payload.degraded||h.payload.coverageStatus!=='CONTINUOUS'||gaps.length?'DEGRADED':captureSince&&q.from{const value=normalize(r);if(coverageStatus!=='CONTINUOUS'){value.connectionRate=null;value.overallAnswerRate=null;value.connectedAnswerRate=null;}return value;}; const alertScopeMatches = coverageStatus==='CONTINUOUS' && !q.customerGatewayId && !q.vendorId && !q.vendorGatewayId && !q.city && !q.carrier && +q.to-Date.now()>-90000 && +q.to-Date.now()<5000 && +q.to-+q.from===900000; const annotated = numbers.map(r => { const matching=alerts.filter(a=>a.customer_id===r.customerId&&a.caller===r.caller); return { ...withCoverage(r), alert: alertScopeMatches&&matching.length ? { reasons:matching.map(a=>(a.payload as {reason?:string}).reason).filter(Boolean), scope:'最近15分钟号码全量范围' } : null }; }); const fits = (v: number | null, min: number, max: number) => v === null ? min === 0 && max === 100 : v >= min && v <= max; const filtered = annotated.filter(r => r.totalCalls >= q.minSamples && (!q.onlyAlerts || r.alert) && fits(r.connectionRate, q.connectionMin, q.connectionMax) && fits(r.overallAnswerRate, q.overallMin, q.overallMax) && fits(r.connectedAnswerRate, q.connectedMin, q.connectedMax)); filtered.sort((a, b) => { const av = a[q.sort as keyof typeof a]; const bv = b[q.sort as keyof typeof b]; if (av === null) return bv === null ? 0 : 1; if (bv === null) return -1; const cmp = typeof av === 'string' && typeof bv === 'string' ? av.localeCompare(bv) : Number(av) - Number(bv); return (q.order === 'asc' ? cmp : -cmp) || String(a.caller).localeCompare(String(b.caller)) || String(a.customerId).localeCompare(String(b.customerId)); }); const sum = annotated.reduce((a, r) => { for (const k of COUNT_KEYS) a[k] += r[k]; return a; }, emptyCounts()); const quality=coverageStatus==='CONTINUOUS'?'LIVE':'DEGRADED'; const asOf=new Date(); const snapshotId=analyticsHash([ANALYTICS_REVISION,q.view,q.from.toISOString(),q.to.toISOString(),asOf.toISOString()]); const summaryRates=analyticsRates(sum); if(coverageStatus!=='CONTINUOUS'){summaryRates.connectionRate=null;summaryRates.overallAnswerRate=null;summaryRates.connectedAnswerRate=null;} return { metricVersion: ANALYTICS_VERSION, analysisRevision:ANALYTICS_REVISION, snapshotId, view: q.view, from: q.from, to: q.to, asOf, qualityStatus: quality, coverageStatus, dataThrough: h?.payload.dataThrough ?? null, lagMs: h?.payload.lagMs ?? null, durationPrecision: 'milliseconds', health: h?.payload ?? null, gaps, summary: summaryRates, summaryScope: '时间及业务筛选;样本/比例/异常筛选仅影响号码列表', numbers: filtered.slice(q.skip, q.skip + q.take), total: filtered.length, skip: q.skip, take: q.take, trends: trends.map(r => ({ ...withCoverage(r), time: Number(r.time) })), baseFilters:{view:q.view,from:q.from,to:q.to,customerId:q.customerId,customerGatewayId:q.customerGatewayId,vendorId:q.vendorId,vendorGatewayId:q.vendorGatewayId,caller:q.caller,city:q.city,carrier:q.carrier}, numberFilters:{minSamples:q.minSamples,onlyAlerts:q.onlyAlerts,connectionMin:q.connectionMin,connectionMax:q.connectionMax,overallMin:q.overallMin,overallMax:q.overallMax,connectedMin:q.connectedMin,connectedMax:q.connectedMax}, metricAvailability:{connectionRate:coverageStatus==='CONTINUOUS'&&!sum.unknownCalls,answerRates:coverageStatus==='CONTINUOUS'&&!sum.durationUnknownCalls}, historyNotice: coverageStatus==='CONTINUOUS'?'当前窗口在连续采集覆盖内;旧固定6秒话单不参与校准。':'当前窗口无连续完整覆盖,计数仅为已观测值,完整比例不可用。' }; } async detail(raw: Query, user: CurrentUser) { const q = parseAnalyticsQuery(raw); if (!q.caller || !q.customerId) throw new BadRequestException('请选择客户和主叫号码'); const scope = await this.scope(user, q.customerId); const where = this.where(q, scope); const source = this.source(q, scope); const [rows,breakdownRows]=await this.db.$transaction([ this.db.$queryRaw>>(Prisma.sql`SELECT id, call_key AS callKey, started_at AS startedAt, caller, callee, vendor_gateway_id AS vendorGatewayId, counts, payload FROM caller_analysis_states WHERE ${where} AND started_at>=${q.from} AND started_at<${q.to} ORDER BY started_at DESC, id DESC LIMIT ${q.take} OFFSET ${q.skip}`), this.db.$queryRaw(Prisma.sql`SELECT vendor_gateway_id AS vendorGatewayId, city, carrier, ${countSql} FROM (${source}) a GROUP BY vendor_gateway_id, city, carrier HAVING SUM(CAST(JSON_UNQUOTE(JSON_EXTRACT(counts,'$.totalCalls')) AS SIGNED))>0 ORDER BY vendor_gateway_id,city,carrier LIMIT 501`) ]); return { rows, breakdown:breakdownRows.slice(0,500).map(normalize), breakdownTruncated:breakdownRows.length>500, skip:q.skip, take:q.take }; } async options(user: CurrentUser) { const scope = await this.scope(user, ''); const [customers, customerGateways, vendors, vendorGateways] = await Promise.all([ this.db.customer.findMany({ where: { ...(scope ? { id: { in: scope } } : {}), deletedAt: null }, select: { id: true, name: true }, take: 2000 }), this.db.customerGateway.findMany({ where: { ...(scope ? { customerId: { in: scope } } : {}), deletedAt: null }, select: { id: true, name: true, customerId: true }, take: 2000 }), scope ? this.db.$queryRaw>(Prisma.sql`SELECT DISTINCT vendor_id AS id,vendor_id AS name FROM caller_analysis_states WHERE customer_id IN (${Prisma.join(scope)}) AND vendor_id<>'' LIMIT 2000`) : this.db.vendor.findMany({where:{deletedAt:null},select:{id:true,name:true},take:2000}), // Limited users only receive gateway IDs they actually have traffic on, not the global vendor configuration. scope ? this.db.$queryRaw>(Prisma.sql`SELECT DISTINCT vendor_gateway_id AS id, vendor_gateway_id AS name FROM caller_analysis_states WHERE customer_id IN (${Prisma.join(scope)}) AND vendor_gateway_id<>'' LIMIT 2000`) : this.db.vendorGateway.findMany({ where: { deletedAt: null }, select: { id: true, name: true }, take: 2000 }) ]); return { customers, customerGateways, vendors, vendorGateways }; } }