hyreal-gen-pic/napos-service.js

'use strict';

/**
 * 数据报表 · 取数服务层
 * ----------------------------------------------------------------------------
 * 桥接 napos/（淘宝闪购商家客户端 CDP 取数引擎）与 reports-store（本地数据湖）。
 *
 * 取数通道：附着本机商家客户端调试端口（默认 9669），由客户端页面自身的登录态
 * 代发请求（非爬虫、不模拟点击）；数据中心接口按域自动补签名。
 *
 * 数据来源（数据中心 lsycm · historyOverview）：
 *   ds 日期 / valid_order_cnt 有效订单 / gross_amt 营业额 / income_amt 收入 /
 *   expense_amt 支出 / exp_uv 曝光人数 / order_user_cnt 下单人数 /
 *   clk_exp_rate 进店转化率 / ord_clk_rate 下单转化率 / per_actual_pay 客单价 …
 *
 * 营业时长 / 异常关店时长 / 商品销售额：当前接口目录暂无对应字段，
 * 作为预留列返回 null，取数层保留原始返回值，后续补源即可回填。
 */

const kit = require('./napos/kit.js');
const store = require('./reports-store.js');

const DEFAULT_CDP_PORT = Number(process.env.NAPOS_CDP_PORT || 9669);

// ---------- BI 列注册表（语义层）----------
// synonyms / description 供「智能问数」的 LLM 做实体对齐与口径解释

const REPORT_COLUMNS = [
  { key: 'date', label: '日期', type: 'date', locked: true, aggregable: false,
    synonyms: ['日期', '天', '哪天', '日期时间'],
    description: '数据日期，格式 YYYYMMDD；当天数据未结算，有效区间一般为 T-1 及更早' },
  { key: 'shopName', label: '门店名称', type: 'text', locked: true, aggregable: false,
    synonyms: ['门店', '店铺', '店', '门店名'],
    description: '门店名称，支持模糊筛选' },
  { key: 'bizHours', label: '营业时长(h)', type: 'hours', reserved: true,
    synonyms: ['营业时长', '营业时间', '开店时长'],
    description: '预留列：当日营业小时数，数据源暂未接入' },
  { key: 'abnormalCloseHours', label: '异常关店时长(h)', type: 'hours', reserved: true,
    synonyms: ['异常关店', '异常关店时长', '关店时长', '异常闭店'],
    description: '预留列：非计划性关店时长，数据源暂未接入' },
  { key: 'validOrderCnt', label: '有效订单', type: 'int',
    synonyms: ['有效订单', '订单量', '单量', '订单数', '有效单量'],
    description: '剔除取消/无效后的订单量' },
  { key: 'incomeAmt', label: '收入(元)', type: 'money',
    synonyms: ['收入', '实际收入', '到账', '营收'],
    description: '商家实际收入（元）；营业额 = 收入 + 支出' },
  { key: 'grossAmt', label: '营业额(元)', type: 'money',
    synonyms: ['营业额', '流水', 'GMV', 'gmv', '销售额', '交易额'],
    description: '营业额（元），口径 = 收入 + 支出；做排行/对比的首选指标' },
  { key: 'goodsSaleAmt', label: '商品销售额(元)', type: 'money', reserved: true,
    synonyms: ['商品销售额', '商品销售', '菜品销售额'],
    description: '预留列：仅商品部分的销售金额，数据源暂未接入' },
  { key: 'expUv', label: '曝光人数', type: 'int',
    synonyms: ['曝光', '曝光人数', '曝光量', '展现人数'],
    description: '看到店铺的去重人数' },
  { key: 'orderUserCnt', label: '下单人数', type: 'int',
    synonyms: ['下单人数', '下单用户', '购买人数', '下单用户数'],
    description: '当日下单的去重用户数' },
  { key: 'clkExpRate', label: '进店转化率', type: 'pct',
    synonyms: ['进店转化率', '进店率', '点击率', '入店转化率'],
    description: '进店人数 / 曝光人数；≥9% 健康，<6% 偏低' },
  { key: 'ordClkRate', label: '下单转化率', type: 'pct',
    synonyms: ['下单转化率', '下单率', '成交转化率', '访购率'],
    description: '下单人数 / 进店人数；≥2.5% 健康，<2% 偏低' },
  { key: 'perActualPay', label: '客单价(元)', type: 'money',
    synonyms: ['客单价', '均价', '笔单价'],
    description: '营业额 / 有效订单（元）' },
  { key: 'expenseAmt', label: '支出(元)', type: 'money',
    synonyms: ['支出', '费用', '成本', '平台扣费'],
    description: '支出（元），含平台佣金、推广费用等' },
  { key: 'actOrdCnt', label: '活动订单', type: 'int',
    synonyms: ['活动订单', '活动单量', '营销订单'],
    description: '参与营销活动的订单量' },
  { key: 'actSubsidyAmt', label: '活动补贴(元)', type: 'money',
    synonyms: ['活动补贴', '补贴金额', '商家补贴'],
    description: '活动中商家承担的补贴金额（元）' },
  { key: 'shopRoi', label: '推广ROI', type: 'num',
    synonyms: ['ROI', 'roi', '投产比', '推广roi'],
    description: '推广投入产出比；大盘口径常为 0' },
  { key: 'rebuyRate30d', label: '30日复购率', type: 'pct',
    synonyms: ['复购率', '30日复购', '老客回购率'],
    description: '近 30 天老客复购比例' },
];

const COLUMN_MAP = Object.fromEntries(REPORT_COLUMNS.map((c) => [c.key, c]));

// ---------- 原始字段 → 标准行 ----------

const FIELD_MAP = {
  valid_order_cnt: 'validOrderCnt',
  gross_amt: 'grossAmt',
  income_amt: 'incomeAmt',
  expense_amt: 'expenseAmt',
  exp_uv: 'expUv',
  order_user_cnt: 'orderUserCnt',
  clk_exp_rate: 'clkExpRate',
  ord_clk_rate: 'ordClkRate',
  per_actual_pay: 'perActualPay',
  act_ord_cnt: 'actOrdCnt',
  act_shop_subsidy_amt: 'actSubsidyAmt',
  shop_roi: 'shopRoi',
  rebuy_rate_30d: 'rebuyRate30d',
};

function trimRaw(raw) {
  const out = {};
  for (const [k, v] of Object.entries(raw || {})) {
    if (k === 'ds') continue;
    if (/_diff$|_mall_avg$|_mall_avg_rank$/.test(k)) continue;
    out[k] = v;
  }
  return out;
}

function normalizeRow(rawDaily, ctx) {
  const row = {
    date: String(rawDaily.ds || ''),
    scope: ctx.scope,
    shopId: ctx.shopId == null ? '' : String(ctx.shopId),
    shopName: ctx.shopName || '',
    chainId: ctx.chainId == null ? '' : String(ctx.chainId),
    chainName: ctx.chainName || '',
    bizHours: null,
    abnormalCloseHours: null,
    goodsSaleAmt: null,
    syncedAt: new Date().toISOString(),
    raw: trimRaw(rawDaily),
  };
  for (const [src, dst] of Object.entries(FIELD_MAP)) {
    const v = rawDaily[src];
    row[dst] = v == null || v === '' ? 0 : Number(v);
  }
  return row;
}

// ---------- 连接状态 ----------

/** 按平台给出「以调试端口启动商家客户端」的可执行指引 */
function cdpLaunchHint(port = DEFAULT_CDP_PORT) {
  if (process.platform === 'win32') {
    return `Windows 启动方法：① 任务栏右键完全退出淘宝闪购客户端（任务管理器确认进程已退出）；`
      + `② 右键桌面快捷方式 → 属性 → 在「目标」末尾追加一个空格和 --remote-debugging-port=${port} → 确定后用该快捷方式启动。`
      + `注意：客户端每次重启都要用这个快捷方式，否则端口不会开启。`;
  }
  return `macOS 启动方法：① 完全退出客户端（⌘Q，或活动监视器确认进程退出）；`
    + `② 终端执行：open -a "淘宝闪购商家版" --args --remote-debugging-port=${port}`
    + `（应用名以实际为准；也可直接运行 "/Applications/客户端.app/Contents/MacOS/可执行文件" --remote-debugging-port=${port}）。`
    + `注意：客户端每次重启都要带此参数。`;
}

async function naposStatus() {
  const out = { port: DEFAULT_CDP_PORT, connected: false };
  try {
    const st = await kit.status(DEFAULT_CDP_PORT);
    out.connected = !!st.portOk;
    out.browser = st.browser || null;
    out.targets = (st.targets || []).length;
    out.contexts = (st.contexts || []).length;
    if (st.probeError) out.probeError = String(st.probeError);
  } catch (e) {
    out.error = e.message || String(e);
  }
  if (!out.connected) out.hint = cdpLaunchHint();
  // 自检：本机能否连通调试端口（给用户的浏览器直测地址）
  out.selfCheckUrl = `http://127.0.0.1:${DEFAULT_CDP_PORT}/json/version`;
  return out;
}

// ---------- 主体结构（多主体 tab 的来源） ----------

let treeCache = { at: 0, tree: null };

async function fetchTree({ refresh = false } = {}) {
  if (!refresh && treeCache.tree && Date.now() - treeCache.at < 5 * 60 * 1000) {
    return treeCache.tree;
  }
  const tree = await kit.shopTree({ port: DEFAULT_CDP_PORT });
  if (!tree.ok) {
    const err = new Error('读取门店组织树失败：' + (tree.code || tree.status || '未知错误'));
    err.code = 'NAPOS_TREE_FAIL';
    throw err;
  }
  const entity = store.upsertEntity({
    merchantId: (tree.merchant && tree.merchant.id) || 'default',
    merchantName: (tree.merchant && tree.merchant.name) || '默认主体',
    chains: tree.chains.map((c) => ({ chainId: String(c.chainId), chainName: c.chainName })),
    shops: tree.shops.map((s) => ({
      shopId: String(s.shopId), shopName: s.shopName,
      chainId: s.chainId ? String(s.chainId) : '', chainName: s.chainName || '',
      busyLevel: s.busyLevel || null,
    })),
  }).entity;
  // 缓存必须存补齐 entity 后的完整对象：缓存的消费方（runSync / tree 路由）都依赖 tree.entity
  const enriched = { ...tree, entity };
  treeCache = { at: Date.now(), tree: enriched };
  return enriched;
}

// ---------- 同步任务 ----------

const syncState = {
  running: false,
  cancelRequested: false,
  startedAt: null,
  finishedAt: null,
  error: null,
  entity: null,
  from: null,
  to: null,
  planTotal: 0,
  done: 0,
  current: '',
  added: 0,
  updated: 0,
  failed: 0,
  steps: [],
};

function syncSnapshot() {
  const { running, cancelRequested, startedAt, finishedAt, error, entity, from, to,
    planTotal, done, current, added, updated, failed } = syncState;
  return { running, cancelRequested, startedAt, finishedAt, error,
    entity, from, to, planTotal, done, current, added, updated, failed };
}

function dayShift(ymd, delta) {
  const d = new Date(Number(ymd.slice(0, 4)), Number(ymd.slice(4, 6)) - 1, Number(ymd.slice(6, 8)));
  d.setDate(d.getDate() + delta);
  const z = (n) => String(n).padStart(2, '0');
  return `${d.getFullYear()}${z(d.getMonth() + 1)}${z(d.getDate())}`;
}

function yesterday() { return dayShift(kit.today(), -1); }

/**
 * 同步一个区间到本地数据湖。
 * scope:
 *   'company' — 仅全公司聚合行
 *   'chain'   — 各连锁聚合行
 *   'shop'    — 逐店明细行（可选 chainId 限定 / shopIds 限定）
 *   'all'     — 以上全部（默认）
 */
async function runSync({ from, to, scope = 'all', chainId, shopIds, port } = {}) {
  if (syncState.running) {
    const err = new Error('已有同步任务在进行中');
    err.code = 'SYNC_BUSY';
    throw err;
  }
  // 先同步置位，避免忙碌检查与实际启动之间因 await 出现并发窗口
  Object.assign(syncState, {
    running: true, cancelRequested: false, startedAt: new Date().toISOString(), finishedAt: null,
    error: null, entity: null, from: null, to: null,
    planTotal: 0, done: 0, current: '准备中', added: 0, updated: 0, failed: 0, steps: [],
  });

  try {
    const tree = await fetchTree();
    const entity = tree.entity;
    const fromDay = kit.normDay(from || dayShift(yesterday(), -29));
    const toDay = kit.normDay(to || yesterday());
    if (!/^\d{8}$/.test(fromDay) || !/^\d{8}$/.test(toDay) || fromDay > toDay) {
      const err = new Error('日期区间不合法（需 YYYYMMDD，且开始 ≤ 结束）');
      err.code = 'SYNC_BAD_RANGE';
      throw err;
    }

    const jobs = [];
    if (scope === 'all' || scope === 'company') {
      jobs.push({ kind: 'company', label: '全公司', fn: () => callHistory({ chainId: '-1', shopListIn: '' }, { scope: 'company', chainId: '-1', chainName: '' }) });
    }
    if (scope === 'all' || scope === 'chain') {
      for (const c of tree.chains) {
        jobs.push({
          kind: 'chain', label: `连锁·${c.chainName}`,
          fn: () => callHistory({ chainId: c.chainId, shopListIn: '' }, { scope: 'chain', chainId: c.chainId, chainName: c.chainName }),
        });
      }
    }
    if (scope === 'all' || scope === 'shop') {
      let shops = tree.shops;
      if (chainId) shops = shops.filter((s) => String(s.chainId) === String(chainId));
      if (Array.isArray(shopIds) && shopIds.length) {
        const wanted = new Set(shopIds.map(String));
        shops = shops.filter((s) => wanted.has(String(s.shopId)));
      }
      for (const s of shops) {
        jobs.push({
          kind: 'shop', label: `门店·${s.shopName}`,
          fn: () => callHistory(
            { chainId: '-1', shopListIn: String(s.shopId) },
            { scope: 'shop', shopId: s.shopId, shopName: s.shopName, chainId: s.chainId || '', chainName: s.chainName || '' },
          ),
        });
      }
    }

    Object.assign(syncState, {
      entity: { key: entity.key, name: entity.name }, from: fromDay, to: toDay,
      planTotal: jobs.length, current: '',
    });

    const usedPort = Number(port || DEFAULT_CDP_PORT);
    const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

    for (const job of jobs) {
      if (syncState.cancelRequested) break;
      syncState.current = job.label;
      try {
        const rows = await job.fn(usedPort);
        const res = store.upsertRows(entity.key, rows);
        syncState.added += res.added;
        syncState.updated += res.updated;
        syncState.steps.push({ label: job.label, ok: true, added: res.added, updated: res.updated });
      } catch (e) {
        syncState.failed += 1;
        syncState.steps.push({ label: job.label, ok: false, error: e.message || String(e) });
      }
      syncState.done += 1;
      await sleep(180); // 温和节奏，避免连续轰接口
    }
    const run = {
      at: new Date().toISOString(), from: fromDay, to: toDay, scope,
      cancelled: syncState.cancelRequested,
      added: syncState.added, updated: syncState.updated, failed: syncState.failed,
      done: syncState.done, total: syncState.planTotal,
      steps: syncState.steps,
    };
    store.appendSyncLog(entity.key, run);
    syncState.error = null;
  } catch (e) {
    syncState.error = e.message || String(e);
    syncState.steps.push({ label: '准备', ok: false, error: syncState.error });
  } finally {
    syncState.running = false;
    syncState.current = '';
    syncState.finishedAt = new Date().toISOString();
  }
  return syncSnapshot();
}

/**
 * 调 historyOverview 取一个区间的逐日行并归一化。
 * scopeKey = { chainId, shopListIn }；ctx 携带行的归属信息。
 */
async function callHistory(scopeKey, ctx, port = DEFAULT_CDP_PORT) {
  const o = {
    port,
    chainId: scopeKey.chainId,
    shopListIn: scopeKey.shopListIn || '',
    from: syncState.from,
    to: syncState.to,
    dateType: 'self_define',
  };
  const r = await kit.statsHistory(o);
  if (r.risk) throw new Error('触发客户端风控（请稍后在客户端完成验证后重试）');
  if (r.authFail) throw new Error('登录态失效（ksid 过期），请在商家客户端重新登录');
  if (r.code != null && r.code !== 0 && String(r.code) !== '0') {
    throw new Error('接口返回异常 code=' + r.code + (r.msg ? ' ' + r.msg : ''));
  }
  return (r.daily || []).map((d) => normalizeRow(d, ctx));
}

function requestSyncCancel() {
  if (syncState.running) syncState.cancelRequested = true;
  return syncSnapshot();
}

// ---------- 查询 / 汇总 ----------

const SUM_FIELDS = ['validOrderCnt', 'incomeAmt', 'grossAmt', 'expenseAmt', 'expUv', 'orderUserCnt', 'actOrdCnt', 'actSubsidyAmt'];
const RATE_FIELDS = ['clkExpRate', 'ordClkRate'];

function summarizeRows(rows) {
  const s = { days: new Set(), shops: new Set() };
  for (const f of SUM_FIELDS) s[f] = 0;
  let rateWeight = { clkExpRate: 0, ordClkRate: 0 };
  let rateBase = { clkExpRate: 0, ordClkRate: 0 };
  for (const r of rows) {
    s.days.add(r.date);
    if (r.shopId) s.shops.add(r.shopId);
    for (const f of SUM_FIELDS) s[f] += Number(r[f]) || 0;
    if (Number(r.expUv) > 0) { rateWeight.clkExpRate += (Number(r.clkExpRate) || 0) * Number(r.expUv); rateBase.clkExpRate += Number(r.expUv); }
    if (Number(r.orderUserCnt) > 0) { rateWeight.ordClkRate += (Number(r.ordClkRate) || 0) * Number(r.orderUserCnt); rateBase.ordClkRate += Number(r.orderUserCnt); }
  }
  s.perActualPay = s.validOrderCnt > 0 ? +(s.grossAmt / s.validOrderCnt).toFixed(2) : 0;
  for (const f of SUM_FIELDS) s[f] = Math.round(s[f] * 100) / 100;
  for (const f of RATE_FIELDS) s[f] = rateBase[f] > 0 ? +(rateWeight[f] / rateBase[f]).toFixed(4) : 0;
  s.dayCount = s.days.size;
  s.shopCount = s.shops.size;
  delete s.days;
  delete s.shops;
  return s;
}

function queryReport({ entity, from, to, shopId, shops, scope, q } = {}) {
  const key = store.sanitizeKey(entity);
  const meta = store.getEntity(key);
  let rows = store.queryRows(key, { from, to, scope });
  if (shopId) rows = rows.filter((r) => String(r.shopId) === String(shopId));
  if (Array.isArray(shops) && shops.length) {
    const wanted = new Set(shops.map(String));
    rows = rows.filter((r) => wanted.has(String(r.shopId)));
  }
  if (q) {
    const needle = String(q).toLowerCase();
    rows = rows.filter((r) => String(r.shopName || '').toLowerCase().includes(needle));
  }
  return {
    entity: meta ? { key: meta.key, name: meta.name } : { key, name: key },
    from: from || null, to: to || null,
    columns: REPORT_COLUMNS,
    summary: summarizeRows(rows),
    rows,
  };
}

/** BI 聚合视图：group = 'shop'（按门店）| 'date'（按日期） */
function aggregateReport({ entity, from, to, group = 'shop', shops } = {}) {
  const key = store.sanitizeKey(entity);
  let rows = store.queryRows(key, { from, to, scope: 'shop' });
  if (Array.isArray(shops) && shops.length) {
    const wanted = new Set(shops.map(String));
    rows = rows.filter((r) => wanted.has(String(r.shopId)));
  }
  const buckets = new Map();
  for (const r of rows) {
    const k = group === 'date' ? r.date : String(r.shopId || '');
    if (!buckets.has(k)) buckets.set(k, []);
    buckets.get(k).push(r);
  }
  const out = [];
  for (const [k, list] of buckets) {
    const first = list[0] || {};
    const base = group === 'date'
      ? { date: k, shopName: `${list.length} 家门店`, shopId: '', scope: 'agg-date' }
      : { date: '', shopName: first.shopName || k, shopId: k, scope: 'agg-shop' };
    const s = summarizeRows(list);
    out.push({ ...base, ...s });
  }
  out.sort((a, b) => (group === 'date' ? (a.date < b.date ? -1 : 1) : (b.grossAmt || 0) - (a.grossAmt || 0)));
  return { columns: REPORT_COLUMNS, rows: out, summary: summarizeRows(rows) };
}

// ---------- CSV 导出 ----------

function csvEscape(v) {
  if (v == null) return '';
  const s = typeof v === 'object' ? JSON.stringify(v) : String(v);
  return /[",\n\r]/.test(s) ? '"' + s.replace(/"/g, '""') + '"' : s;
}

function rowsToCsv(columns, rows) {
  const head = columns.map((c) => c.label);
  const lines = [head.join(',')];
  for (const r of rows) {
    const cells = columns.map((c) => {
      let v = r[c.key];
      if (v == null) v = '';
      if (c.type === 'pct') v = v === '' ? '' : (Number(v) * 100).toFixed(2) + '%';
      if (c.type === 'date' && /^\d{8}$/.test(String(v))) {
        const s = String(v);
        v = `${s.slice(0, 4)}-${s.slice(4, 6)}-${s.slice(6, 8)}`;
      }
      return csvEscape(v);
    });
    lines.push(cells.join(','));
  }
  return '\ufeff' + lines.join('\r\n');
}

module.exports = {
  DEFAULT_CDP_PORT, REPORT_COLUMNS,
  naposStatus, fetchTree,
  runSync, syncSnapshot, requestSyncCancel,
  queryReport, aggregateReport, summarizeRows,
  rowsToCsv,
  store,
};