Add standalone Leads module for prospect outreach before customers.
Import large Excel lists with duplicate checks, assign and kanban follow-up, convert to Person when ready; also fix OTP mobile mask order and SMS brand name to منتکت. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,212 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
import { Cron } from '@nestjs/schedule';
|
||||
import { Prisma } from '@prisma/client';
|
||||
import type { LeadImportJobDto } from '@mento/shared';
|
||||
import { PrismaService } from '../prisma/prisma.service';
|
||||
import { LeadStagesService } from './lead-stages.service';
|
||||
import { LeadsService } from './leads.service';
|
||||
|
||||
@Injectable()
|
||||
export class LeadsImportService {
|
||||
private readonly logger = new Logger(LeadsImportService.name);
|
||||
private running = false;
|
||||
|
||||
constructor(
|
||||
private readonly prisma: PrismaService,
|
||||
private readonly stages: LeadStagesService,
|
||||
private readonly leads: LeadsService,
|
||||
) {}
|
||||
|
||||
async enqueue(
|
||||
organizationId: string,
|
||||
createdById: string,
|
||||
rows: Array<Record<string, unknown>>,
|
||||
fileName?: string,
|
||||
): Promise<LeadImportJobDto> {
|
||||
await this.stages.ensureDefaults(organizationId);
|
||||
const normalized = rows.map((r) => this.normalizeRow(r));
|
||||
const job = await this.prisma.leadImportJob.create({
|
||||
data: {
|
||||
organizationId,
|
||||
createdById,
|
||||
fileName: fileName ?? null,
|
||||
status: 'pending',
|
||||
totalCount: normalized.length,
|
||||
rows: normalized as unknown as Prisma.InputJsonValue,
|
||||
},
|
||||
});
|
||||
return this.toDto(job);
|
||||
}
|
||||
|
||||
async getJob(
|
||||
organizationId: string,
|
||||
id: string,
|
||||
): Promise<LeadImportJobDto | null> {
|
||||
const job = await this.prisma.leadImportJob.findFirst({
|
||||
where: { id, organizationId },
|
||||
});
|
||||
return job ? this.toDto(job) : null;
|
||||
}
|
||||
|
||||
@Cron('*/1 * * * *')
|
||||
async processDue(): Promise<number> {
|
||||
if (this.running) return 0;
|
||||
this.running = true;
|
||||
try {
|
||||
const jobs = await this.prisma.leadImportJob.findMany({
|
||||
where: { status: 'pending' },
|
||||
orderBy: { createdAt: 'asc' },
|
||||
take: 2,
|
||||
});
|
||||
for (const job of jobs) {
|
||||
await this.processJob(job.id);
|
||||
}
|
||||
return jobs.length;
|
||||
} finally {
|
||||
this.running = false;
|
||||
}
|
||||
}
|
||||
|
||||
private async processJob(id: string): Promise<void> {
|
||||
const claimed = await this.prisma.leadImportJob.updateMany({
|
||||
where: { id, status: 'pending' },
|
||||
data: { status: 'processing', startedAt: new Date() },
|
||||
});
|
||||
if (!claimed.count) return;
|
||||
|
||||
const job = await this.prisma.leadImportJob.findUnique({ where: { id } });
|
||||
if (!job) return;
|
||||
|
||||
const stages = await this.stages.list(job.organizationId);
|
||||
const defaultStageId = stages[0]?.id;
|
||||
if (!defaultStageId) {
|
||||
await this.prisma.leadImportJob.update({
|
||||
where: { id },
|
||||
data: {
|
||||
status: 'failed',
|
||||
finishedAt: new Date(),
|
||||
errorSamples: ['مرحله پیشفرض لید وجود ندارد'],
|
||||
},
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const rows = (job.rows as Array<Record<string, string>>) ?? [];
|
||||
const errorSamples: string[] = [];
|
||||
let success = 0;
|
||||
let duplicate = 0;
|
||||
let error = 0;
|
||||
let processed = 0;
|
||||
|
||||
const BATCH = 50;
|
||||
for (let i = 0; i < rows.length; i += BATCH) {
|
||||
const chunk = rows.slice(i, i + BATCH);
|
||||
for (const row of chunk) {
|
||||
const result = await this.leads.createFromImportRow(
|
||||
job.organizationId,
|
||||
job.createdById,
|
||||
row,
|
||||
defaultStageId,
|
||||
);
|
||||
processed += 1;
|
||||
if (result === 'ok') success += 1;
|
||||
else if (result === 'duplicate') duplicate += 1;
|
||||
else {
|
||||
error += 1;
|
||||
if (errorSamples.length < 20) {
|
||||
errorSamples.push(
|
||||
`ردیف ${processed}: ${row.title || row.mobile || 'نامعتبر'}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
await this.prisma.leadImportJob.update({
|
||||
where: { id },
|
||||
data: {
|
||||
processedCount: processed,
|
||||
successCount: success,
|
||||
duplicateCount: duplicate,
|
||||
errorCount: error,
|
||||
errorSamples,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
await this.prisma.leadImportJob.update({
|
||||
where: { id },
|
||||
data: {
|
||||
status: 'done',
|
||||
processedCount: processed,
|
||||
successCount: success,
|
||||
duplicateCount: duplicate,
|
||||
errorCount: error,
|
||||
errorSamples,
|
||||
finishedAt: new Date(),
|
||||
rows: [],
|
||||
},
|
||||
});
|
||||
this.logger.log(
|
||||
`Lead import ${id}: ok=${success} dup=${duplicate} err=${error}`,
|
||||
);
|
||||
}
|
||||
|
||||
private normalizeRow(raw: Record<string, unknown>): Record<string, string> {
|
||||
const get = (...keys: string[]) => {
|
||||
for (const k of keys) {
|
||||
const direct = raw[k];
|
||||
if (direct != null && String(direct).trim()) return String(direct).trim();
|
||||
const found = Object.entries(raw).find(
|
||||
([key]) => key.trim().toLowerCase() === k.toLowerCase(),
|
||||
);
|
||||
if (found?.[1] != null && String(found[1]).trim()) {
|
||||
return String(found[1]).trim();
|
||||
}
|
||||
}
|
||||
return '';
|
||||
};
|
||||
return {
|
||||
title: get('title', 'نام', 'نام فروشگاه', 'فروشگاه', 'shop', 'name'),
|
||||
contactName: get('contactName', 'مالک', 'مسئول', 'نام مالک', 'owner'),
|
||||
mobile: get('mobile', 'موبایل', 'شماره', 'شماره تماس', 'phone', 'tel'),
|
||||
phone: get('phone', 'تلفن'),
|
||||
city: get('city', 'شهر'),
|
||||
region: get('region', 'منطقه', 'محله'),
|
||||
trade: get('trade', 'صنف', 'رسته'),
|
||||
source: get('source', 'منبع', 'منبع لید'),
|
||||
notes: get('notes', 'توضیحات', 'یادداشت'),
|
||||
};
|
||||
}
|
||||
|
||||
private toDto(job: {
|
||||
id: string;
|
||||
fileName: string | null;
|
||||
status: string;
|
||||
totalCount: number;
|
||||
processedCount: number;
|
||||
successCount: number;
|
||||
duplicateCount: number;
|
||||
errorCount: number;
|
||||
errorSamples: unknown;
|
||||
startedAt: Date | null;
|
||||
finishedAt: Date | null;
|
||||
createdAt: Date;
|
||||
}): LeadImportJobDto {
|
||||
const samples = Array.isArray(job.errorSamples)
|
||||
? job.errorSamples.map((s) => String(s))
|
||||
: [];
|
||||
return {
|
||||
id: job.id,
|
||||
fileName: job.fileName,
|
||||
status: job.status,
|
||||
totalCount: job.totalCount,
|
||||
processedCount: job.processedCount,
|
||||
successCount: job.successCount,
|
||||
duplicateCount: job.duplicateCount,
|
||||
errorCount: job.errorCount,
|
||||
errorSamples: samples,
|
||||
startedAt: job.startedAt?.toISOString() ?? null,
|
||||
finishedAt: job.finishedAt?.toISOString() ?? null,
|
||||
createdAt: job.createdAt.toISOString(),
|
||||
};
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user