人群包拆包详解

flowchart TD
  subgraph BIZ["业务切包：整包 → 多个子人群包"]
    W[("整包 callout_package_task
subcontract_type=1")]
    W -->|"isNewPackageLine 判定每行归属"| N1["子包A（新记录）"]
    W -->|"不满足条件的行留在整包"| W2["整包（原地更新数量与ftpPath）"]
  end
  subgraph PHY["物理拆分：大文件 → 多个分片"]
    F["大 CSV 文件"] -->|"splitFile 按字节 / splitFileByLine 按行"| P1["分片1（含表头）"]
    F --> P2["分片2（含表头）"]
    F --> P3["分片N（含表头）"]
  end

flowchart TD
  subgraph E1["入口① 运营手动（packSearching 页面）"]
    A1["POST /service/packageManager/cutPackage
PackageManagerApiImpl:300"]
    A1 --> A2["checkCutPackageParam 参数校验
（提数完成/非推送中/数量/名称唯一）"]
    A2 --> A3["Redis 锁 CUT_PACKAGE_KEY:{id}
TTL 10min，防并发"]
    A3 --> A4["异步 asyncServiceExecutor
返回「异步切包中」"]
  end

  subgraph E2["入口② 精准系统回调联动（outGenerate）"]
    B1["POST /service/ruleManager/outGenerate
外部精准系统按数量订购人群"]
    B1 --> B2["generate() 发起提数
+ 落 push_market_record（cutInfo）"]
    B2 --> B3["提数回调 /api/package/callback
checkAndSavePackage 完成"]
    B3 --> B4["processCutPackage
按实际人数重新分配切包数量"]
    B4 --> B5["cutPackage() → 走入口① 同一条路
成功后 pushToMarket 推回精准"]
  end

  subgraph E3["入口③ 定时任务推送时切包（TaskSettingsPage 配置）"]
    C1["任务设置页配置「人群包切包」
pushPackage + newPackages 存入 schedule ext"]
    C1 --> C2["ScheduleTaskManager 到期触发
DataPickScheduleServiceImpl.pushDataPackageSchedule"]
    C2 --> C3["pushToChannel → cutPackage
带 PushPackageSnapshot 快照缓存"]
  end

  A4 --> ENG["PackageCutManager.isNewPackageLine
行级判定引擎"]
  B5 --> ENG
  C3 --> ENG2["同一引擎（DataPickScheduleServiceImpl
自带的 cutPackage 重载，临时文件式）"]

flowchart TD
  L["line = CSV 数据行, header = 表头
newCount = 新包已累计行数"]
  L --> Q1{"newPackNum ≠ null 且
newCount ≥ newPackNum？"}
  Q1 -->|"是"| OLD["false → 留整包
（数量上限优先于一切条件）"]
  Q1 -->|"否"| Q2{"cutType 为空？"}
  Q2 -->|"是"| NEW["true → 进新包
（纯按数量：顺序取前 N 行）"]
  Q2 -->|"否"| Q3{"cutType = ?"}

  Q3 -->|"1 城市"| C1["手机号列取前7位
查归属地表 mobile_city.csv
城市 contains 任一 typeValues"]
  Q3 -->|"2 机型"| C2["phone_type / 手机机型列
值 contains 任一 typeValues"]
  Q3 -->|"3 空号结果"| C3["改写 cutType 为校验结果列名
按列值 contains 切"]
  Q3 -->|"5 多列组合"| C5["multiColumnFilters 全部满足（AND）
age_range 走数值区间
其余列值 contains 任一 values"]
  Q3 -->|"其他（含 4 按列）"| C4["cutType 即列名
列值 contains 任一 typeValues"]

  C1 --> R{"匹配？"}
  C2 --> R
  C3 --> R
  C4 --> R
  C5 --> R
  R -->|"是"| NEW
  R -->|"否"| OLD

flowchart TD
  S["syncCutPackage
for 每个 newPackage 串行执行
每轮重新 selectByPrimaryKey 取整包"]
  S --> P1["1. addPackage 新增分包记录
packageCode=pack+随机码
parentPackageCode / subcontractType=2
pullStatus=2 提取中"]
  P1 --> P2["2. 下载整包到 ftp.cutPath
文件名 = packageCode-UUID.csv"]
  P2 --> P3["3. 行级分流（本地三文件并行读写）
整包读 → isNewPackageLine？
true→newWriter / false→oldWriter
表头写入两个文件，AtomicLong 计数"]
  P3 --> P4["4. 上传新包 → 上传削后整包
（新包失败则整体失败）"]
  P4 --> P5["5. 删除 FTP 原整包"]
  P5 --> P6["6. 更新分包：ftpPath + PULL_SUCCESS + newCount"]
  P6 --> P7["7. recordDO 回写 cutPackageCodes
+ pushToMarket 推送精准（MARKET 渠道）"]
  P7 --> P8["8. 更新整包：peopleNumber=oldCount
ftpPath 指向削后文件"]
  P8 --> P9["9. 两个包都上传资源服务器
（downloadUrl 供前端下载）"]
  P4 -.->|"任一步异常"| FAIL["失败分支：
AlarmSendManager.sendCutAlarm 告警
分包标 PULL_FAIL + remark
删除本地 old/new 文件
整包记录只写 remark 不回滚数量"]

flowchart TD
  T["pushDataPackageSchedule
解析 schedule ext 里的 pushChannelRequests"]
  T --> LOOP["for 每个渠道 channelRequest"]
  LOOP --> SNAP{"快照缓存中有
该 newPackName？"}
  SNAP -->|"有（本轮已切过）"| REUSE["直接复用快照
ftpPath + peopleNumber"]
  SNAP -->|"无"| CUT["cutPackage(newPackage, 整包ftpPath,
新UUID.csv, oldCount, newCount)
本地 .tmp 削整包 → 重命名覆盖 → 上传"]
  CUT --> SAVE["存入 cutPackageSnapshot
（同一配置多个渠道只切一次）"]
  SAVE --> UPD["推送明细记录快照的
ftpPath + peopleNumber"]
  REUSE --> UPD
  UPD --> PUSH["下载该 ftpPath 文件
走渠道推送（校验/触达）"]

flowchart LR
  subgraph MED["媒体平台渠道（PushMediumPlatform）"]
    M1["buildPackageAndSaveResult
校验通过的数据写成 UUID.csv"] --> M2["splitFile 90MB
每片带表头"]
    M2 --> M3["循环：上传资源服务器
→ crowdPackSyn 同步"]
  end
  subgraph YM["友盟撞库（BatchComputeHelper）"]
    Y1["本地 CSV 目录"] --> Y2["行数 > 500万？
splitFileByLine 等行数切"]
    Y2 --> Y3["逐片上传 OSS 数据集"]
  end