人群包去重详解

flowchart TD
  subgraph A["链路 A · 组合去重（packSearching 勾两包）"]
    A1["POST /distinctPackage（List 版）
PackageManagerServiceImpl:829"]
    A1 --> A2["A 包标「去重中」→ 异步"]
    A2 --> A3["下载 A/B 两包全量进内存
B 包手机号排 long[] → 二分查找
A = A − B"]
    A3 --> A4["结果覆盖上传回 A 包原 FTP 路径
A 包原地削，不产生新记录"]
  end

  subgraph B["链路 B · 按类型剔除（弹窗选择）"]
    B1["POST /distinctPackage（BO 版）
PackageManagerServiceImpl:891"]
    B1 --> B2{"distinctType"}
    B2 -->|"RULE 规则"| B3["按规则条件重新走提数
pickExecutor → Azkaban"]
    B2 -->|"TAG 标签"| B4["查 package_tag 拿 FTP 列表
拼 file[ftp] 条件"]
    B2 -->|"PACKAGE 内部包"| B5["被选包 ftpPath 作副包
拼 file[ftp] 条件"]
    B4 --> B6{"useLocalDistinct？
（prod: true）"}
    B5 --> B6
    B6 -->|"规则去重 或 false"| B7["pickExecutor
Azkaban/数综 file-ftp flow"]
    B6 -->|"标签/内部包 且 true"| B8["PackageDistinctManager
本地去重（外部排序）"]
    B7 --> B9["新包记录 createPackage
（受理后插入）"]
    B8 --> B9
  end

  subgraph C["链路 C · 标签去重（生成时自动）"]
    C1["提数回调 /api/package/callback
成功 + 规则配了 tagExcludeCondition"]
    C1 --> C2["PackageDistinctManager.distinctByTag:654"]
    C2 --> C3["datapick-service /distinctByTag
与 people_tag 比对剔除已触达"]
    C3 --> C4["带新 ftpPath 再次回调
按新文件落库"]
  end

flowchart TD
  S["distinctPackage(List<Integer>)
PackageManagerServiceImpl:829"]
  S --> V["校验：恰好 2 包 / 人数 > 0
/ 非推送中 / 提取完成或去重完成"]
  V --> M["A 包 pullStatus=5 去重中
同步 updateByPrimaryKey"]
  M --> ASYNC["asyncServiceExecutor 异步
asynDistinctPackage:1384"]
  ASYNC --> D1["下载 A、B 两包
downloadAndAnalysis 全量读入 List"]
  D1 --> ALG["getDistinctDataBin:1466
① B 包每行取第一列，合法手机号→long，否则→0
② parallelStream 排序成 long[]
③ 遍历 A 包：第一列非手机号→直接保留
④ 是手机号→binarySearch B 数组
找不到才保留"]
  ALG --> CNT["peopleNumber = 结果行数
（有表头再 -1）"]
  CNT --> UP["生成 CSV 覆盖上传回
A 包原 FTP 路径"]
  UP --> F["finally：A 包 pullStatus=6 去重完成
失败时 remark 写原因"]
  F --> RES["成功：重新上传资源服务器
（供前端下载）"]

flowchart TD
  E["distinctPackage(bo)
DistinctPackageServiceImpl:43
CompletableFuture.runAsync 触发"]
  E --> M["1. 下载主包（重试版）
去重列：配置值 或 按表头自动提取
（优先明文手机号列，其次 uu_id 列）"]
  M --> X["2. 逐个下载副包
单个失败仅 warn 被忽略
全部失败才整体失败"]
  X --> H["3. checkHeader 表头校验
主包 + 每个副包都必须包含全部去重列
任一缺失即整体失败"]
  H --> S1["4a. 分块排序：每 10 万行在内存排序
写成临时文件 sortedBlock_N.txt
排序 key = 去重列拼接值
（主包行携带完整内容，副包只存 key）"]
  S1 --> S2["4b. 多路归并去重（小顶堆）
相邻相等 key 只输出一次
→ 主包、副包各自一份有序无重文件"]
  S2 --> S3["4c. 双指针过滤 filterMainFile
主包行 key 在副包出现→丢弃
否则写出原始完整行 + 写表头 + 计数"]
  S3 --> U["5. 结果上传 FTP dataExtractPath"]
  U --> CB["6. 回调自己 packageManagerApi.callback
sourceService=dw-platform-service
（回调侧凭此跳过 FTP 校验）"]
  CB --> ST["新包状态 → 提取完成
ftpPath / peopleNumber 回填"]

sequenceDiagram
  participant EXE as 提数执行方
  participant CB as /api/package/callback
  participant PDM as PackageDistinctManager
  participant DPS as datapick-service

  EXE->>CB: 提数成功回调（ftpPath=人群包）
  CB->>CB: distinctByTag 前置判断
  Note over CB: ① success=true ② 回调非 DISTINCT/PICK 类型
（防递归）③ 规则存在且配了 tagExcludeCondition
  CB->>PDM: distinctByTag(callback)
  PDM->>PDM: 标签条件（标签+时间窗）→ package_tag id 列表
buildPickRequest 注入 packageTag
  PDM->>DPS: POST /distinctByTag（distinctFtpPath=人群包）
  DPS-->>PDM: 受理返回 code=0
  DPS->>DPS: 异步：下载人群包 CSV
每 1000 行一批，去重列 AES 加密后
查 people_tag（打过的=已触达）
  DPS->>DPS: 剔除后写新文件 → 上传 FTP
  DPS->>CB: 带新 ftpPath 再次回调（pickType=DISTINCT）
  CB->>CB: 跳过标签去重（防递归）→ 按新文件落库