hyreal-gen-pic/server/services/crawl.js

'use strict';

// 采集引擎：按店铺批量抓取菜品原图。
// 从 server.js 原样搬入：翻页、下载、断点跳过、店铺完成状态、可取消的任务循环。
// 状态（crawlState/cancelFlag）收敛在本工厂内，通过返回的接口访问。
//
// 采集通道（v2）：
//   · client —— 请求放进淘宝闪购客户端页面上下文发出（CDP :9669），登录态随客户端走，无需 Cookie；
//   · cookie —— 原直连通道，靠手填 Cookie（ksid）鉴权，保留作为兜底；
//   · auto   —— 优先客户端，客户端不可用/中途失联时自动回落 Cookie。
// 两通道的请求体/响应形状完全一致，仅传输层不同；图片文件下载不走通道（CDN 直下，无需登录态）。

const fs = require('fs');
const path = require('path');
const crypto = require('crypto');
const https = require('https');
const http = require('http');
const { URL } = require('url');
const { sendJson, sleep, sanitizeFileName, guessExtFromUrl, SKIP_DIR_NAMES } = require('../util.js');
const kit = require('../../napos/kit.js');

const GOODS_INVOKE_PATH = '/nevermore.goods/invoke';
const GOODS_SERVICE = 'ItemPhotoProcessService';
const GOODS_METHOD = 'getItemPhotoGallery';
const MAX_PAGES = 200;

const UA = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36';

function createCrawlService(ctx) {
  const { config, session, shops, workspace, logs } = ctx;

  let crawlState = { status: 'idle', startedAt: null, finishedAt: null, shops: [], log: [] };
  let cancelFlag = false;
  // 通道：channelMode 是用户选择（auto/client/cookie）；useClient 是本次运行实际生效通道（auto 下可中途回落）
  let channelMode = 'auto';
  let useClient = false;

  // ---------- 采集状态持久化（重启后进度不丢） ----------

  const stateFilePath = () => path.join(workspace.getWorkspaceRoot(), '.genpic', 'crawl-state.json');

  function persistState() {
    if (crawlState.status === 'idle' && !crawlState.shops.length) return;
    try {
      const file = stateFilePath();
      fs.mkdirSync(path.dirname(file), { recursive: true });
      const temp = `${file}.${process.pid}.tmp`;
      const snapshot = { ...crawlState, log: crawlState.log.slice(-100) };
      fs.writeFileSync(temp, JSON.stringify({ version: 1, savedAt: new Date().toISOString(), state: snapshot }, null, 2), 'utf8');
      fs.renameSync(temp, file);
    } catch { /* 持久化失败不影响采集进行 */ }
  }

  function restoreState() {
    let parsed;
    try { parsed = JSON.parse(fs.readFileSync(stateFilePath(), 'utf8')); } catch { return; }
    const saved = parsed && parsed.version === 1 && parsed.state;
    if (!saved || typeof saved.status !== 'string') return;
    crawlState = {
      status: saved.status,
      startedAt: saved.startedAt || null,
      finishedAt: saved.finishedAt || null,
      forceFullCrawl: !!saved.forceFullCrawl,
      channel: saved.channel || null,
      shops: Array.isArray(saved.shops) ? saved.shops : [],
      log: Array.isArray(saved.log) ? saved.log : [],
    };
    if (crawlState.status === 'running') {
      crawlState.status = 'interrupted';
      crawlState.finishedAt = Date.now();
      for (const s of crawlState.shops) {
        if (s.state === 'running' || s.state === 'pending') s.state = 'interrupted';
      }
      crawlState.log.push(logs.createLogEntry('服务重启，上次采集被中断；未完成的店铺可重新勾选采集', 'task', 'warn'));
    }
  }
  restoreState();

  function pushLog(msg) {
    const entry = logs.createLogEntry(msg, 'task', logs.detectLogLevel(msg));
    crawlState.log.push(entry);
    if (crawlState.log.length > 500) crawlState.log.shift();
  }

  /** 两通道共用的请求体（与真实抓包一致：ksid 在 metas 里，一次性 id） */
  function buildRequestBody(service, method, requestParams, ksid, extraMetas) {
    return {
      id: crypto.randomBytes(16).toString('hex').toUpperCase() + '|' + Date.now(),
      method, service,
      params: { request: requestParams || {} },
      metas: Object.assign({ appName: 'melody', appVersion: '4.4.0', ksid }, extraMetas || {}),
      ncp: '2.0.0', adapter: 'ncp',
      extendConfig: { docMode: 'master,,' },
    };
  }

  // ---- Cookie 直连通道（原实现，兜底保留） ----

  async function httpInvoke(service, method, requestParams, extraMetas) {
    return new Promise((resolve, reject) => {
      const body = buildRequestBody(service, method, requestParams, session.getKsid(), extraMetas);
      const data = JSON.stringify(body);
      const options = {
        hostname: 'app-api.shop.ele.me', port: 443,
        path: `${GOODS_INVOKE_PATH}?method=${encodeURIComponent(service + '.' + method)}`,
        method: 'POST',
        headers: {
          'Content-Type': 'application/json; charset=UTF-8', Accept: 'application/json',
          Origin: 'https://napos-goods-pc.faas.ele.me', Referer: 'https://napos-goods-pc.faas.ele.me/',
          Cookie: session.cookie || '', 'User-Agent': UA,
          'Content-Length': Buffer.byteLength(data),
        },
        timeout: config.requestTimeoutMs,
      };
      const req = https.request(options, (res) => {
        const chunks = [];
        res.on('data', (c) => chunks.push(c));
        res.on('end', () => {
          const raw = Buffer.concat(chunks).toString('utf8');
          let json = null;
          try { json = JSON.parse(raw); } catch { /* keep raw */ }
          resolve({ statusCode: res.statusCode, json, raw });
        });
      });
      req.on('error', reject);
      req.on('timeout', () => req.destroy(new Error('请求超时')));
      req.write(data);
      req.end();
    });
  }

  // ---- 客户端通道（CDP）：请求由客户端页面上下文发出，Cookie/UA 全是真 ----

  function makeClientInvoker(api, ksid) {
    return async function (service, method, requestParams, extraMetas) {
      const url = `https://app-api.shop.ele.me${GOODS_INVOKE_PATH}?method=${encodeURIComponent(service + '.' + method)}`;
      const res = await kit.cdp.pageFetch(api, {
        method: 'POST', url, allowWrite: false,
        body: buildRequestBody(service, method, requestParams, ksid, extraMetas),
      });
      if (res.error || res.timeout) {
        throw new Error('客户端内请求失败: ' + (res.error || '超时（可能被风控挂起，请在客户端检查）'));
      }
      return {
        statusCode: res.status,
        json: res.parsed ? res.data : null,
        raw: res.parsed ? '' : String(res.data || ''),
        risk: !!res.risk, authFail: !!res.authFail,
      };
    };
  }

  /**
   * 翻页拉全一个店铺的菜品列表（不含下载）。
   * @param {Function} invoke (service, method, requestParams, extraMetas) => resp，由通道提供
   * @returns {{items:Array, authFailed:boolean, risk:boolean, error:?string}}
   *          error 只作记录不中断：已拉到的部分照常下载（Cookie 通道保留原行为）
   */
  async function fetchAllPages(shopEntry, invoke, pageSize) {
    const allItems = [];
    const seen = new Set();
    let nextId;
    let page = 0;
    let authFailed = false;
    let risk = false;
    let error = null;
    while (page < MAX_PAGES) {
      if (cancelFlag) break;
      page++;
      const requestParams = {
        shopId: shopEntry.shopId, pageSize, queryType: 'SHOP',
        searchKey: '', searchType: ['ITEM_NAME'],
      };
      if (nextId) requestParams.nextId = nextId;

      let resp;
      try { resp = await invoke(GOODS_SERVICE, GOODS_METHOD, requestParams, { shopId: shopEntry.shopId }); }
      catch (e) {
        error = '请求失败: ' + (e.message || e);
        pushLog(`[${shopEntry.shopName}] ${error}`);
        break;
      }
      if (resp.risk || resp.authFail || isAuthError(resp)) {
        authFailed = true;
        risk = !!resp.risk;
        pushLog(`[${shopEntry.shopName}] ${risk ? '触发风控/验证码' : '鉴权失败'}（登录态可能已失效）`);
        break;
      }
      const items = (resp.json && resp.json.result && resp.json.result.result) || [];
      nextId = resp.json && resp.json.result && resp.json.result.nextId;
      if (!items.length) {
        pushLog(`[${shopEntry.shopName}] 第${page}页无数据，结束翻页`);
        break;
      }
      let newCount = 0;
      for (const it of items) {
        const gid = it.itemGlobalId;
        if (gid !== undefined && gid !== null && seen.has(gid)) continue;
        if (gid !== undefined && gid !== null) seen.add(gid);
        allItems.push(it);
        newCount++;
      }
      pushLog(`[${shopEntry.shopName}] 第${page}页 新增${newCount}条`);
      if (!nextId) break;
      await sleep(200);
    }
    return { items: allItems, authFailed, risk, error };
  }

  /**
   * 客户端通道拉取：每店一个 CDP 会话（复用于本店全部翻页），
   * 连接坏了只损失当前店，下一店自动重连。
   * @returns 页结果；null 表示应回落 Cookie 通道；shopFailed=true 表示本店已判失败（无 Cookie 可回落）
   */
  async function fetchViaClient(shopEntry, pageSize) {
    try {
      return await kit.cdp.withPageFor(kit.DEFAULT_PORT, 'shop', async (api) => {
        const ksid = await kit.cdp.findKsid(api);
        if (!ksid) throw new Error('客户端未登录（读不到登录态 ksid），请先在客户端登录');
        return fetchAllPages(shopEntry, makeClientInvoker(api, ksid), pageSize);
      }, { needExact: false, autoOpen: false });
    } catch (e) {
      const msg = e.message || String(e);
      if (!(channelMode === 'auto' && session.cookie && session.getKsid())) {
        shopEntry.state = 'failed';
        shopEntry.error = '客户端通道失败: ' + msg;
        pushLog(`[${shopEntry.shopName}] ${shopEntry.error}`);
        return { items: [], authFailed: false, risk: false, error: shopEntry.error, shopFailed: true };
      }
      pushLog(`[${shopEntry.shopName}] 客户端通道失败: ${msg}`);
      return null;
    }
  }

  function isAuthError(resp) {
    if (resp.statusCode === 401 || resp.statusCode === 403) return true;
    const j = resp.json;
    if (!j) return false;
    if (j.result !== undefined) return false;
    return /未登录|请登录|login|unauthorized|token.*expired|session.*expired/i.test(JSON.stringify(j));
  }

  function downloadImage(urlStr, redirectsLeft) {
    if (redirectsLeft === undefined) redirectsLeft = 5;
    return new Promise((resolve, reject) => {
      let u;
      try { u = new URL(urlStr); } catch (e) { return reject(new Error('无效的图片地址: ' + urlStr)); }
      const mod = u.protocol === 'http:' ? http : https;
      mod.get(u, { headers: { Referer: 'https://melody.shop.ele.me/', 'User-Agent': UA } }, (res) => {
        if (res.statusCode >= 300 && res.statusCode < 400 && res.headers.location && redirectsLeft > 0) {
          res.resume();
          return resolve(downloadImage(res.headers.location, redirectsLeft - 1));
        }
        if (res.statusCode !== 200) { res.resume(); return reject(new Error('下载图片失败 HTTP ' + res.statusCode)); }
        const chunks = [];
        res.on('data', (c) => chunks.push(c));
        res.on('end', () => resolve(Buffer.concat(chunks)));
      }).on('error', reject);
    });
  }

  // ---------- 店铺完成状态（root 下 _膳绘_店铺状态.json） ----------

  function completionStatePath(root) { return path.join(root, '_膳绘_店铺状态.json'); }

  function readCompletionState(root) {
    try {
      const data = JSON.parse(fs.readFileSync(completionStatePath(root), 'utf8'));
      return data.version === 1 && data.shops && typeof data.shops === 'object' ? data.shops : {};
    } catch { return {}; }
  }

  function writeCompletionState(root, shopsState) {
    const target = completionStatePath(root);
    const temp = `${target}.${process.pid}.tmp`;
    fs.writeFileSync(temp, JSON.stringify({ version: 1, shops: shopsState }, null, 2), 'utf8');
    fs.renameSync(temp, target);
  }

  function markShopCompleted(root, shopEntry) {
    const shopsState = readCompletionState(root);
    shopsState[String(shopEntry.shopId)] = {
      shopId: shopEntry.shopId, shopName: shopEntry.shopName, folderName: shopEntry.folderName,
      completedAt: new Date().toISOString(), downloaded: shopEntry.downloaded,
      skipped: shopEntry.skipped, total: shopEntry.total,
    };
    writeCompletionState(root, shopsState);
  }

  // ---------- 门店目录分配 ----------

  function listRootDirs(root) {
    return fs.readdirSync(root, { withFileTypes: true })
      .filter((e) => e.isDirectory() && !e.name.startsWith('.') && !SKIP_DIR_NAMES.has(e.name))
      .map((e) => e.name);
  }

  function resolveShopDir(root, shopId, shopName) {
    const dirs = listRootDirs(root);
    const suffix = '_' + String(shopId);
    const match = dirs.find((d) => d.endsWith(suffix));
    if (match) return { dir: path.join(root, match), folderName: match, reused: true };
    let maxSeq = 0;
    for (const d of dirs) {
      const mm = /^(\d+)_/.exec(d);
      if (mm) maxSeq = Math.max(maxSeq, parseInt(mm[1], 10));
    }
    const seq = String(maxSeq + 1).padStart(3, '0');
    const folderName = `${seq}_${sanitizeFileName(shopName)}_${shopId}`;
    const dir = path.join(root, folderName);
    fs.mkdirSync(dir, { recursive: true });
    return { dir, folderName, reused: false };
  }

  // ---------- 任务循环 ----------

  async function startCrawl(body, res) {
    if (crawlState.status === 'running') return sendJson(res, 409, { error: '已有抓取任务在进行中' });
    // 通道解析：body.channel（单次指定）> config.crawlChannel（设置）> auto
    const channelPref = ['auto', 'client', 'cookie'].includes(body && body.channel) ? body.channel : (config.crawlChannel || 'auto');
    const cookieOk = !!(session.cookie && session.getKsid());
    const probeResult = await kit.cdp.probe().catch(() => ({ ok: false }));
    const clientOk = !!probeResult.ok;
    if (channelPref === 'client' && !clientOk) {
      return sendJson(res, 400, { error: '客户端通道不可用：请先启动淘宝闪购客户端（调试端口 9669），或改用 Cookie 通道' });
    }
    if (channelPref === 'cookie' && !cookieOk) {
      return sendJson(res, 400, { error: '请先填写包含 ksid 的 Cookie（设置 → 采集配置）' });
    }
    if (channelPref === 'auto' && !clientOk && !cookieOk) {
      return sendJson(res, 400, { error: '未检测到客户端（调试端口 9669 未开），也未填写 Cookie——两种通道至少配置其一' });
    }
    channelMode = channelPref;
    useClient = channelPref === 'client' || (channelPref === 'auto' && clientOk);
    const root = String((body && body.root) || workspace.getWorkspaceRoot()).trim();
    if (!workspace.isWorkspaceRoot(root)) return sendJson(res, 403, { error: '仅允许采集当前输出根目录' });
    if (!fs.existsSync(root) || !fs.statSync(root).isDirectory()) {
      return sendJson(res, 400, { error: '输出根目录不存在: ' + root });
    }
    const shopIds = Array.isArray(body && body.shopIds) ? body.shopIds.map(Number) : [];
    const selectedShops = shopIds.map((id) => shops.getList().find((s) => s.shopId === id)).filter(Boolean);
    if (!selectedShops.length) return sendJson(res, 400, { error: '没有可抓取的店铺，请先填写手动店铺清单并勾选' });
    const forceFullCrawl = Boolean(body && body.forceFullCrawl);
    const completed = readCompletionState(root);
    const skippedCompleted = forceFullCrawl ? [] : selectedShops.filter((s) => completed[String(s.shopId)]);
    const pendingShops = forceFullCrawl ? selectedShops : selectedShops.filter((s) => !completed[String(s.shopId)]);
    const pageSize = Number(body && body.pageSize) || config.pageSize;
    const delayMs = Number(body && body.delayMs) || config.delayMs;

    cancelFlag = false;
    crawlState = {
      status: pendingShops.length ? 'running' : 'done', startedAt: Date.now(),
      finishedAt: pendingShops.length ? null : Date.now(), forceFullCrawl,
      channel: useClient ? 'client' : 'cookie',
      shops: [
        ...skippedCompleted.map((s) => ({
          shopId: s.shopId, shopName: s.shopName, folderName: completed[String(s.shopId)].folderName || null,
          state: 'skipped_completed', total: completed[String(s.shopId)].total || 0,
          downloaded: 0, skipped: 0, failed: 0, error: null,
        })),
        ...pendingShops.map((s) => ({
          shopId: s.shopId, shopName: s.shopName, folderName: null, state: 'pending',
          total: 0, downloaded: 0, skipped: 0, failed: 0, error: null,
        })),
      ],
      log: [],
    };
    pushLog(`开始抓取，实际抓取 ${pendingShops.length} 家${skippedCompleted.length ? `，历史完成跳过 ${skippedCompleted.length} 家` : ''}，root=${root}，通道=${useClient ? '客户端（CDP :9669）' : 'Cookie 直连'}${channelPref === 'auto' ? '（自动）' : ''}`);
    persistState();
    sendJson(res, 200, { ok: true, crawlCount: pendingShops.length, skippedCompletedCount: skippedCompleted.length });
    if (!pendingShops.length) return pushLog('所选店铺均已完成；勾选"强制全量抓取"可重新下载');

    runCrawlLoop(root, pageSize, delayMs, forceFullCrawl).catch((e) => {
      crawlState.status = 'error';
      crawlState.finishedAt = Date.now();
      pushLog('抓取流程异常终止: ' + (e.message || e));
      persistState();
    });
  }

  async function runCrawlLoop(root, pageSize, delayMs, forceFullCrawl) {
    for (const shopEntry of crawlState.shops) {
      if (shopEntry.state === 'skipped_completed') continue;
      if (cancelFlag) {
        crawlState.status = 'stopped';
        crawlState.finishedAt = Date.now();
        pushLog('已停止');
        return;
      }
      shopEntry.state = 'running';
      pushLog(`[${shopEntry.shopName}] 开始`);

      let dirInfo;
      try { dirInfo = resolveShopDir(root, shopEntry.shopId, shopEntry.shopName); }
      catch (e) {
        shopEntry.state = 'failed';
        shopEntry.error = '创建目录失败: ' + (e.message || e);
        pushLog(`[${shopEntry.shopName}] 创建目录失败: ${e.message || e}`);
        continue;
      }
      shopEntry.folderName = dirInfo.folderName;
      pushLog(`[${shopEntry.shopName}] ${dirInfo.reused ? '复用已有目录' : '新建目录'} ${dirInfo.folderName}`);

      // ---- 翻页：按生效通道拉取（客户端优先，可回落 Cookie） ----
      let pageResult = null;
      if (useClient) {
        pageResult = await fetchViaClient(shopEntry, pageSize);
        if (pageResult === null) {
          useClient = false;
          crawlState.channel = 'cookie';
          pushLog('本店起改用 Cookie 通道继续');
        } else if (pageResult.shopFailed) {
          persistState();
          continue;
        }
      }
      if (!pageResult) pageResult = await fetchAllPages(shopEntry, httpInvoke, pageSize);

      const allItems = pageResult.items;
      shopEntry.total = allItems.length;
      if (pageResult.error && !pageResult.authFailed) shopEntry.error = pageResult.error;
      if (pageResult.authFailed && !allItems.length) {
        shopEntry.state = 'failed';
        shopEntry.error = pageResult.risk
          ? (useClient ? '触发风控/验证码，请在客户端完成验证后重试' : '触发风控，请稍后重试或更新 Cookie')
          : (useClient ? '客户端登录态失效，请在客户端重新登录后重试' : '会话可能已过期，请重新登录后更新 Cookie');
        continue;
      }

      pushLog(`[${shopEntry.shopName}] 共${allItems.length}个菜品，开始下载`);
      const manifest = [];
      const usedNames = new Set();
      for (const it of allItems) {
        if (cancelFlag) break;
        const itemName = String(it.itemName || it.itemGlobalId || 'unnamed').trim();
        const photoUrl = it.photoUrl || it.thumbnailUrl;
        if (!photoUrl) {
          shopEntry.skipped++;
          manifest.push({ itemName, itemGlobalId: it.itemGlobalId, photoUrl: null, savedFile: null, status: 'no_photo' });
          continue;
        }
        let finalBase = sanitizeFileName(itemName);
        let n = 1;
        while (usedNames.has(finalBase)) { finalBase = `${sanitizeFileName(itemName)}_${n++}`; }
        usedNames.add(finalBase);
        try {
          const buf = await downloadImage(photoUrl);
          const ext = guessExtFromUrl(photoUrl);
          const fileName = `${finalBase}.${ext}`;
          fs.writeFileSync(path.join(dirInfo.dir, fileName), buf);
          shopEntry.downloaded++;
          manifest.push({ itemName, itemGlobalId: it.itemGlobalId, photoUrl, savedFile: fileName, status: 'downloaded' });
        } catch (e) {
          shopEntry.failed++;
          manifest.push({ itemName, itemGlobalId: it.itemGlobalId, photoUrl, savedFile: null, status: 'download_failed', error: e.message || String(e) });
          pushLog(`[${shopEntry.shopName}] 下载失败 ${itemName}: ${e.message || e}`);
        }
        await sleep(delayMs);
      }

      try {
        fs.writeFileSync(path.join(dirInfo.dir, '_菜品图爬取记录.json'), JSON.stringify(manifest, null, 2), 'utf8');
        if (!shopEntry.failed && !cancelFlag) markShopCompleted(root, shopEntry);
      } catch (e) {
        shopEntry.failed++;
        shopEntry.error = `保存抓取状态失败: ${e.message || e}`;
        pushLog(`[${shopEntry.shopName}] 保存抓取状态失败: ${e.message || e}`);
      }
      shopEntry.state = shopEntry.failed ? 'failed' : cancelFlag ? 'stopped' : 'done';
      pushLog(`[${shopEntry.shopName}] 完成: 成功${shopEntry.downloaded} 跳过${shopEntry.skipped} 失败${shopEntry.failed}`);
      persistState();
    }

    crawlState.status = cancelFlag ? 'stopped' : 'done';
    crawlState.finishedAt = Date.now();
    pushLog('抓取任务结束');
    persistState();
  }

  function stopCrawl(res) {
    if (crawlState.status !== 'running') return sendJson(res, 200, { ok: true, note: '当前没有正在进行的任务' });
    cancelFlag = true;
    pushLog('收到停止请求，将在当前步骤完成后停止');
    persistState();
    sendJson(res, 200, { ok: true });
  }

  return {
    startCrawl,
    stopCrawl,
    getState: () => crawlState,
    getLog: () => crawlState.log,
    clearLog: () => { crawlState.log = []; },
    readCompletionState,
    flushState: persistState,
  };
}

module.exports = { createCrawlService };