platform-assistant/apps/api/src/crawler/crawler.service.ts

import { Injectable, HttpException, HttpStatus, NotFoundException, OnModuleInit, Logger } from '@nestjs/common'
import { SchedulerRegistry } from '@nestjs/schedule'
import { CronJob } from 'cron'
import { execSync } from 'child_process'
import * as path from 'path'
import * as fs from 'fs'
import { PrismaService } from '../prisma/prisma.service'
import { CategoryInferenceService } from '../product-category/category-inference.service'

const SCRIPTS_DIR = path.resolve(__dirname, '../../../../../scripts')
const TENDER_FETCHER_DIR = path.join(SCRIPTS_DIR, 'tender-fetcher')

interface FetchResult {
  source: string
  title: string
  organization: string
  publishTime: string
  deadline: string
  noticeId: string
  detailUrl: string
  amount?: string
  region?: string
  regionDetail?: string
}

@Injectable()
export class CrawlerService implements OnModuleInit {
  private readonly logger = new Logger(CrawlerService.name)

  constructor(
    private prisma: PrismaService,
    private schedulerRegistry: SchedulerRegistry,
    private categoryInference: CategoryInferenceService,
  ) {}

  async onModuleInit() {
    const configs = await this.prisma.crawlerConfig.findMany({
      where: { status: 'active', schedule: { not: null } },
    })
    for (const c of configs) {
      if (c.schedule) {
        this.registerCronJob(c.id, c.schedule)
      }
    }
    this.logger.log(`已注册 ${configs.length} 个定时采集任务`)
  }

  private registerCronJob(id: string, cronExpr: string) {
    if (this.schedulerRegistry.doesExist('cron', id)) {
      this.schedulerRegistry.deleteCronJob(id)
    }
    const job = new CronJob(cronExpr, async () => {
      this.logger.log(`定时触发采集器 ${id}`)
      try {
        await this.run(id)
      } catch (e: unknown) {
        this.logger.error(`定时采集器 ${id} 执行失败: ${(e as Error).message}`)
      }
    })
    this.schedulerRegistry.addCronJob(id, job)
    job.start()
    this.logger.log(`采集器 ${id} 定时任务已启动: ${cronExpr}`)
  }

  private unregisterCronJob(id: string) {
    if (this.schedulerRegistry.doesExist('cron', id)) {
      this.schedulerRegistry.deleteCronJob(id)
      this.logger.log(`采集器 ${id} 定时任务已移除`)
    }
  }

  async findAll() {
    return this.prisma.crawlerConfig.findMany({ orderBy: { createdAt: 'desc' } })
  }

  async update(id: string, data: { name?: string; targetUrl?: string; schedule?: string; type?: string; sourceName?: string; scriptPath?: string; keywords?: string; categories?: string; status?: string }) {
    const config = await this.prisma.crawlerConfig.findUnique({ where: { id } })
    if (!config) throw new NotFoundException('爬虫配置不存在')

    const finalStatus = data.status !== undefined ? data.status : config.status
    const finalSchedule = data.schedule !== undefined ? data.schedule : config.schedule

    this.unregisterCronJob(id)
    if (finalStatus === 'active' && finalSchedule) {
      this.registerCronJob(id, finalSchedule)
    }

    return this.prisma.crawlerConfig.update({ where: { id }, data })
  }

  async create(data: { name: string; targetUrl: string; schedule?: string; type?: string; sourceName?: string; scriptPath?: string; keywords?: string; categories?: string }) {
    return this.prisma.crawlerConfig.create({
      data: {
        name: data.name,
        targetUrl: data.targetUrl,
        schedule: data.schedule,
        type: data.type || 'bid',
        sourceName: data.sourceName || null,
        scriptPath: data.scriptPath || null,
        keywords: data.keywords || null,
        categories: data.categories || null,
        status: 'paused',
      },
    })
  }

  async run(id: string) {
    const config = await this.prisma.crawlerConfig.findUnique({ where: { id } })
    if (!config) throw new HttpException('爬虫配置不存在', HttpStatus.NOT_FOUND)
    if (config.status === 'paused') throw new HttpException('爬虫已暂停', HttpStatus.BAD_REQUEST)

    try {
      let results: FetchResult[] = []

      if (config.type === 'bid_powerchina') {
        results = await this.runPowerChina(config)
      } else {
        results = await this.runGenericScript(config)
      }

      let successCount = 0

      const validResults = results.filter(r => r.noticeId)
      const categoryMap = await this.categoryInference.batchInfer(
        validResults.map(r => ({ id: r.noticeId, title: r.title })),
      )

      for (const r of validResults) {
        try {
          await this.prisma.lead.upsert({
            where: {
              source_noticeId: { source: r.source, noticeId: r.noticeId },
            },
            update: {
              title: r.title,
              organization: r.organization,
              deadline: r.deadline,
              detailUrl: r.detailUrl,
              publishTime: r.publishTime || null,
              category: categoryMap.get(r.noticeId) || '',
              ...(r.amount ? { amount: r.amount } : {}),
              ...(r.region ? { region: r.region } : {}),
              ...(r.regionDetail ? { regionDetail: r.regionDetail } : {}),
              summary: `来源: ${r.source}\n公告ID: ${r.noticeId}\n链接: ${r.detailUrl}`,
            },
            create: {
              title: r.title,
              organization: r.organization,
              source: r.source,
              noticeId: r.noticeId,
              detailUrl: r.detailUrl,
              deadline: r.deadline,
              publishTime: r.publishTime || null,
              amount: r.amount || '待确认',
              status: '待复核',
              level: 'C',
              category: categoryMap.get(r.noticeId) || '',
              region: r.region || '',
              regionDetail: r.regionDetail || '',
              summary: `来源: ${r.source}\n公告ID: ${r.noticeId}\n链接: ${r.detailUrl}`,
            },
          })
          successCount++
        } catch {
          // 其他异常跳过
        }
      }

      return this.prisma.crawlerConfig.update({
        where: { id },
        data: {
          lastRunAt: new Date(),
          totalRuns: { increment: 1 },
          successRuns: config.successRuns + 1,
        },
      })
    } catch (e: unknown) {
      const err = e as { stderr?: string; stdout?: string; message?: string }
      await this.prisma.crawlerConfig.update({
        where: { id },
        data: { failedRuns: { increment: 1 } },
      })
      const detail = err.stderr || err.stdout || err.message || '未知错误'
      throw new HttpException(`抓取失败: ${detail}`, HttpStatus.INTERNAL_SERVER_ERROR)
    }
  }

  private async runPowerChina(config: { keywords?: string | null }): Promise<FetchResult[]> {
    const scriptPath = path.join(TENDER_FETCHER_DIR, 'fetch-v1.2.js')
    if (!fs.existsSync(scriptPath)) {
      throw new Error('中电建抓取脚本不存在')
    }

    let keywords: string[]
    try {
      keywords = config.keywords ? JSON.parse(config.keywords) : []
    } catch {
      keywords = []
    }
    if (keywords.length === 0) keywords = ['']

    const today = new Date().toISOString().split('T')[0]
    const baseDir = path.join(process.env.HOME || '', 'desktop/招标公告', today)
    const seen = new Set<string>()
    const allResults: FetchResult[] = []

    for (const kw of keywords) {
      const kwFlag = kw ? ` --keyword "${kw}"` : ''
      try {
        execSync(`node "${scriptPath}"${kwFlag}`, {
          encoding: 'utf-8',
          maxBuffer: 10 * 1024 * 1024,
          timeout: 180000,
        })
      } catch {
        // 单个 keyword 失败不影响其他 keyword
        continue
      }

      const statePath = path.join(baseDir, 'downloaded.json')
      if (!fs.existsSync(statePath)) continue

      try {
        const state = JSON.parse(fs.readFileSync(statePath, 'utf-8'))
        const downloaded = state.downloaded || {}

        const outputDir = path.dirname(statePath)
        const pdfDir = path.join(outputDir, 'pdf')

        for (const [key, v] of Object.entries(downloaded)) {
          const info = v as { title?: string; skipped?: boolean; failed?: boolean; id?: string; deadline?: string; publishTime?: string; path?: string }
          if (info.skipped || info.failed) continue
          const [systemId] = key.split('_')
          if (seen.has(systemId)) continue
          seen.add(systemId)

          const realId = info.id || systemId

          let region = ''
          let regionDetail = ''
          if (info.path && fs.existsSync(pdfDir)) {
            const pdfPath = path.join(pdfDir, path.basename(info.path))
            if (fs.existsSync(pdfPath)) {
              try {
                const extractor = path.join(SCRIPTS_DIR, 'extract_address_from_pdf.py')
                const title = info.title || ''
                const cmd = `python3 "${extractor}" "${pdfPath}" "${title}"`
                const out = execSync(cmd, { encoding: 'utf-8', timeout: 10000 }).trim()
                const parsed = JSON.parse(out)
                region = parsed.region || ''
                regionDetail = parsed.regionDetail || ''
              } catch {}
            }
          }

          allResults.push({
            source: '中电建',
            title: info.title || '',
            organization: '中电建',
            publishTime: info.publishTime || state.lastRun || '',
            deadline: info.deadline || '',
            noticeId: systemId,
            detailUrl: `https://bid.powerchina.cn/notice/detail?id=${realId}`,
            region,
            regionDetail,
          })
        }
      } catch {
        continue
      }
    }

    return allResults
  }

  private async runGenericScript(config: { type: string; sourceName?: string | null; scriptPath?: string | null; keywords?: string | null }): Promise<FetchResult[]> {
    const slug = config.type.replace(/^bid_/, '')
    const scriptFile = config.scriptPath || `fetch-${slug}.py`
    const scriptPath = path.join(TENDER_FETCHER_DIR, scriptFile)
    const source = config.sourceName || slug
    const jsonFile = `${slug}_downloaded.json`

    if (!fs.existsSync(scriptPath)) {
      throw new Error(`抓取脚本不存在: ${scriptFile}`)
    }

    let keywords: string[]
    try {
      keywords = config.keywords ? JSON.parse(config.keywords) : []
    } catch {
      keywords = []
    }
    if (keywords.length === 0) keywords = ['']

    const today = new Date().toISOString().split('T')[0]
    const baseDir = path.join(process.env.HOME || '', 'desktop/招标公告', today)
    const statePath = path.join(baseDir, jsonFile)

    const isJs = scriptFile.endsWith('.js')
    const results: FetchResult[] = []
    const seen = new Set<string>()
    const todayStr = new Date().toISOString().substring(0, 10)

    for (const kw of keywords) {
      const kwFlag = kw ? ` --keyword "${kw}"` : ''
      const cmd = isJs ? `node "${scriptPath}"${kwFlag}` : `python3 "${scriptPath}"${kwFlag}`
      try {
        execSync(cmd, {
          encoding: 'utf-8',
          maxBuffer: 10 * 1024 * 1024,
          timeout: 180000,
        })
      } catch {
        // 单个 keyword 失败不影响其他 keyword
        continue
      }

      if (!fs.existsSync(statePath)) continue

      try {
        const state = JSON.parse(fs.readFileSync(statePath, 'utf-8'))
        for (const [noticeId, v] of Object.entries(state)) {
          if (noticeId === 'lastRun') continue
          const info = v as { title?: string; org?: string; publish_time?: string; detail_url?: string; id?: string; deadline?: string; amount?: string; region?: string; region_detail?: string; downloaded_at?: string }
          // 只处理本次新增的记录（downloaded_at 包含今天日期），历史记录已入库无需重新处理
          if (info.downloaded_at && !info.downloaded_at.startsWith(todayStr)) continue
          const dedupeKey = info.id || noticeId
          if (seen.has(dedupeKey)) continue
          seen.add(dedupeKey)
          results.push({
            source,
            title: info.title || '',
            organization: info.org || source,
            publishTime: info.publish_time || '',
            deadline: info.deadline || '',
            noticeId: dedupeKey,
            detailUrl: info.detail_url || '',
            amount: info.amount || '',
            region: info.region || '',
            regionDetail: info.region_detail || '',
          })
        }
      } catch {
        continue
      }
    }

    return results
  }

  async pause(id: string) {
    this.unregisterCronJob(id)
    return this.prisma.crawlerConfig.update({
      where: { id },
      data: { status: 'paused' },
    })
  }

  async resume(id: string) {
    const config = await this.prisma.crawlerConfig.findUnique({ where: { id } })
    if (!config) throw new NotFoundException('爬虫配置不存在')
    if (config.schedule) {
      this.registerCronJob(id, config.schedule)
    }
    return this.prisma.crawlerConfig.update({
      where: { id },
      data: { status: 'active' },
    })
  }
}