逆向解析 ColorOS 离线号码库 (三)：携号转网的二分查找与高并发全量导出#

0. 前言#

在前两节中，我们成功剥离了 ColorOS 底层的 12-Bit 归属地压缩算法与多级指针寻址的运营商识别模型，并修复了历史遗留的区号重定向问题。

然而，基于“前 7 位号段”的查询逻辑存在一个无法逾越的物理死角——携号转网（MNP）。当一个原本属于“北京移动”的用户带着号码转入“中国电信”时，传统的号段映射将彻底失效。本节，我们将揭开最后一个特权数据库 PortabilityNumberData.dat 的面纱，并利用 Python 的多进程并发技术，编写一个秒级导出百万号段的全量查询脚本。

1. 应对携号转网：特权白名单机制#

由于携号转网的号码没有任何规律可言，无法像普通号码那样进行前缀压缩。OPPO 工程师在此引入了最高优先级的特权数据库：PortabilityNumberData.dat。

系统在处理任何来电时，其真实的宏观调度逻辑如下：

查询转网特权库：先在 PortabilityNumberData.dat 中精确查找 11 位完整号码。若命中，直接提取新的归属地和运营商，查询终止。
查询常规号段库：若未命中，说明是未转网的常规号码。系统截取前 7 位，走我们前两节逆向出的普通库（PhoneNumberData + CarrierData）进行解析。

2. 解析 PortabilityNumberData 数据结构#

既然没有任何规律，最简单粗暴的方法就是把 11 位完整的手机号存下来。但如果我们把数以百万计的转网号码全部塞进文件，每次来电都从头遍历，必然会导致严重的性能灾难。

通过反编译 PortabilityNumbersUtil.kt，我们看到了堪称教科书级别的空间优化与二分查找 (Binary Search) 算法应用。

2.1 极致的 11 字节存储模型#

在反编译的 a(String str, long j3) 方法中，我们发现每条转网记录在底层仅占用 11 个字节：

8 字节 (long)：存储 11 位完整的手机号码。由于 11 位数字（如 13812345678）超出了 32 位 int 的最大值（21亿），必须采用 64 位长整型。
2 字节 (short)：新的城市 ID。
1 字节 (byte)：新的运营商 ID。

这意味着，即使全国有 100 万个携号转网用户，该文件的主体体积也仅有约 10.4 MB。

2.2 毫秒级寻址的二分查找#

为了在 100 万个记录中实现极速查询，整个二进制文件内部是严格按号码大小升序排列的。代码中实现了一个底层的文件流二分查找：

1
int left = 0;
2
// right 是总记录数
3
while (left <= right) {
4
    int mid = (right + left) / 2;
5

15 collapsed lines
6
    // 核心：根据 mid 算出绝对字节偏移，直接移动文件指针
7
    randomAccessFile.seek((mid * 11) + baseOffset);
8

9
    // 读取 8 字节的长整型号码
10
    long currentNumber = randomAccessFile.readLong();
11

12
    if (targetNumber > currentNumber) {
13
        left = mid + 1; // 去右半边找
14
    } else if (targetNumber == currentNumber) {
15
        // 命中！紧接着读取 2 字节城市 ID 和 1 字节运营商 ID
16
        return new Result(randomAccessFile.readShort(), randomAccessFile.readByte());
17
    } else {
18
        right = mid - 1; // 去左半边找
19
    }
20
}

通过二分查找，在 100 万条数据中定位特定号码最多只需查找 $\log_2(1000000) \approx 20$ 次。磁盘指针仅需跳转 20 次，耗时在 1 毫秒以内，完美兼顾了存储空间与查询性能。

3. 终极实战：高并发全量导出脚本#

作为本次逆向工程的最终目标，我们需要将 1300000 到 1999999 整整 70 万个基础号段（抛开携号转网的特例）的全息数据，极速导出为 .csv 文件。

为了满足 “高并发、极速、有序” 的要求，我们利用 Python 的 multiprocessing 多进程模块。核心优化点如下：

消除 IPC 瓶颈：利用进程池的 initializer 参数，让每个 CPU 核心在启动时，独立将几个 MB 的数据库加载到自己的内存中，避免了昂贵的进程间数据传递。
严格保序：使用 pool.map 替代普通异步回调，它会自动阻塞并按照任务分配的顺序重组结果列表，确保导出的 CSV 文件从 130 严格递增至 199。

完整的导出脚本 (export_all_phones.py)：

1
import os
2
import struct
3
import csv
4
import time
5
from multiprocessing import Pool, cpu_count
248 collapsed lines
6

7
# ==========================================
8
# 全局变量，用于子进程共享内存数据
9
# ==========================================
10
loc_data = None
11
car_data = None
12

13
# ==========================================
14
# 模块 1：归属地数据库解析
15
# ==========================================
16
def load_location_db(file_path, patch_file):
17
    with open(file_path, 'rb') as f:
18
        data = f.read()
19

20
    tail_data = data[-12002:]
21
    area_codes_raw = tail_data[2:2002]
22
    city_names_raw = tail_data[2002:10002]
23
    extend_prefix_raw = tail_data[10002:12002]
24

25
    cities = []
26
    for i in range(400):
27
        area = area_codes_raw[i*5 : i*5+5].decode('ascii', errors='ignore').strip()
28
        name_bytes = city_names_raw[i*20 : i*20+20].split(b'\x00')[0]
29
        name = name_bytes.decode('gbk', errors='ignore').strip()
30
        cities.append([area, name])
31

32
    # 加载官方补丁 治愈 9999
33
    if os.path.exists(patch_file):
34
        with open(patch_file, "r", encoding="utf-8") as f:
35
            lines = f.readlines()
36
            for line in lines[1:]:
37
                parts = line.strip().split('\t')
38
                if len(parts) == 4:
39
                    origin_id = int(parts[0])
40
                    equal_id = int(parts[2])
41
                    if equal_id - 1 < len(cities) and origin_id - 1 < len(cities):
42
                        target_area, target_name = cities[equal_id - 1]
43
                        cities[origin_id - 1][0] = target_area
44
                        # cities[origin_id - 1][1] = target_name
45

46
    prefix_map = {}
47
    base_idx = 0
48
    for p in range(130, 140): prefix_map[str(p)] = base_idx; base_idx += 10000
49
    for p in range(150, 160): prefix_map[str(p)] = base_idx; base_idx += 10000
50
    prefix_map["188"] = base_idx; base_idx += 10000
51
    prefix_map["189"] = base_idx; base_idx += 10000
52

53
    for i in range(1000):
54
        b1 = extend_prefix_raw[i*2]
55
        b2 = extend_prefix_raw[i*2 + 1]
56
        prefix_val = (b1 << 8) | b2
57
        if prefix_val == 0: break
58
        prefix_map[str(prefix_val)] = base_idx
59
        base_idx += 10000
60

61
    binary_array = data[16 : -12002]
62
    return prefix_map, cities, binary_array
63

64
# ==========================================
65
# 模块 2：运营商数据库解析
66
# ==========================================
67
def load_carrier_db(file_path):
68
    with open(file_path, 'rb') as f:
69
        data = f.read()
70

71
    offset = 16
72
    carrier_names = {}
73
    num_carriers = struct.unpack_from(">H", data, offset)[0]
74
    offset += 2
75
    for _ in range(num_carriers):
76
        carrier_id = struct.unpack_from(">H", data, offset)[0]
77
        offset += 2
78
        names = []
79
        for _ in range(4):
80
            str_len = struct.unpack_from(">H", data, offset)[0]
81
            offset += 2
82
            if str_len > 0:
83
                name = data[offset : offset + str_len].decode('utf-8', errors='ignore')
84
                offset += str_len
85
            else:
86
                name = ""
87
            names.append(name)
88
        carrier_names[carrier_id] = names
89

90
    general_prefixes = {}
91
    num_gen_prefixes = struct.unpack_from(">H", data, offset)[0]
92
    offset += 2
93
    for _ in range(num_gen_prefixes):
94
        prefix_val = struct.unpack_from(">i", data, offset)[0]
95
        offset += 4
96
        cid = struct.unpack_from(">H", data, offset)[0]
97
        offset += 2
98
        general_prefixes[str(prefix_val)] = cid
99

100
    detailed_prefixes = {}
101
    num_det_prefixes = struct.unpack_from(">H", data, offset)[0]
102
    offset += 2
103
    for _ in range(num_det_prefixes):
104
        prefix_val = struct.unpack_from(">i", data, offset)[0]
105
        offset += 4
106
        file_offset = struct.unpack_from(">i", data, offset)[0]
107
        offset += 4
108
        detailed_prefixes[str(prefix_val)] = file_offset
109

110
    return data, carrier_names, general_prefixes, detailed_prefixes
111

112
# ==========================================
113
# 模块 3：子进程初始化与工作函数
114
# ==========================================
115
def init_worker(loc_file, car_file, patch_file):
116
    """
117
    初始化每个 CPU 核心的工作进程，把数据库装载到子进程的内存中。
118
    这样不需要在进程间传递几兆的数据，极大提升性能。
119
    """
120
    global loc_data, car_data
121
    loc_data = load_location_db(loc_file, patch_file)
122
    car_data = load_carrier_db(car_file)
123

124
def process_chunk(args):
125
    """
126
    处理分配到的号码区间，例如 1300000 到 1309999
127
    """
128
    start_num, end_num = args
129
    results = []
130
    prefix_map, cities, binary_array = loc_data
131
    car_raw_data, carrier_names, general_prefixes, detailed_prefixes = car_data
132

133
    for phone_int in range(start_num, end_num + 1):
134
        phone_7 = str(phone_int)
135

136
        # 运营商的底层查询需要 4 位后缀，为了凑齐 8 位进行偏移运算，安全补齐到 11 位
137
        eval_phone = phone_7 + "0000"
138

139
        # --- A. 归属地极速查询 ---
140
        prefix = phone_7[:3]
141
        suffix = phone_7[3:]
142
        loc_name = ""
143

144
        if prefix in prefix_map:
145
            absolute_index = prefix_map[prefix] + int(suffix)
146
            byte_offset = (absolute_index // 2) * 3
147
            if byte_offset + 2 < len(binary_array):
148
                b1, b2, b3 = binary_array[byte_offset:byte_offset+3]
149
                city_id = (b1 << 4) | (b2 >> 4) if absolute_index % 2 == 0 else ((b2 & 0x0F) << 8) | b3
150
                if city_id != 0 and city_id < len(cities):
151
                    _, loc_name = cities[city_id]
152

153
        if not loc_name:
154
            loc_name = "未知"
155

156
        # --- B. 运营商极速查询 ---
157
        carrier_name = ""
158
        found_carrier = False
159

160
        # 1. 查精细号段偏移（如虚拟运营商）
161
        for length in range(4, 2, -1):
162
            p_prefix = eval_phone[:length]
163
            if p_prefix in detailed_prefixes:
164
                p_suffix = eval_phone[length : length+4]
165
                if len(p_suffix) == 4:
166
                    target_offset = detailed_prefixes[p_prefix] + (int(p_suffix) * 2)
167
                    if target_offset + 2 <= len(car_raw_data):
168
                        cid = struct.unpack_from(">H", car_raw_data, target_offset)[0]
169
                        if cid in carrier_names:
170
                            carrier_name = carrier_names[cid][0] or carrier_names[cid][1]
171
                            found_carrier = True
172
                            break
173

174
        # 2. 查全局号段 (如移动/联通基础号段)
175
        if not found_carrier:
176
            for length in range(4, 2, -1):
177
                p_prefix = eval_phone[:length]
178
                if p_prefix in general_prefixes:
179
                    cid = general_prefixes[p_prefix]
180
                    if cid in carrier_names:
181
                        carrier_name = carrier_names[cid][0] or carrier_names[cid][1]
182
                        break
183

184
        if not carrier_name:
185
            carrier_name = "未知"
186

187
        # 如果这个号段在两大数据库里全都是“未知”，说明这是一个彻头彻尾的空号段，直接跳过不写入
188
        if loc_name == "未知" and carrier_name == "未知":
189
            continue
190

191
        results.append(f"{phone_7},{loc_name},{carrier_name}\n")
192

193
    return results
194

195
# ==========================================
196
# 主程序：分发任务并整合写入
197
# ==========================================
198
def main():
199
    loc_file = "E:/APKS/PhoneNumberData_3_1_0.dat"
200
    car_file = "E:/APKS/CarrierData_1_0.dat"
201
    patch_file = "E:/APKS/Multi_Areano_Table.txt"
202
    out_file = "phonenumber-fin.csv"
203

204
    if not all(os.path.exists(f) for f in [loc_file, car_file, patch_file]):
205
        print("❌ 缺少必要的文件，请检查目录！")
206
        return
207

208
    start_num = 1300000
209
    end_num   = 1999999
210
    chunk_size = 20000  # 每个 CPU 分配两万个号码处理
211

212
    # 计算区块分配范围
213
    chunks = []
214
    for i in range(start_num, end_num + 1, chunk_size):
215
        chunk_end = min(i + chunk_size - 1, end_num)
216
        chunks.append((i, chunk_end))
217

218
    total_chunks = len(chunks)
219
    cores = cpu_count()
220
    print("=" * 50)
221
    print(f"🚀 准备火力全开进行高并发全息扫描...")
222
    print(f"🎯 扫描区间: {start_num} - {end_num} (共 {end_num - start_num + 1} 个号段)")
223
    print(f"⚙️  调用核心: {cores} 个 CPU 核心")
224
    print("=" * 50)
225

226
    start_time = time.time()
227

228
    # 创建进程池，使用 map 确保按顺序返回
229
    with Pool(processes=cores, initializer=init_worker, initargs=(loc_file, car_file, patch_file)) as pool:
230
        # pool.map 会阻塞并收集全部结果，并且严格保证结果列表顺序等于 chunks 顺序
231
        all_results = pool.map(process_chunk, chunks)
232

233
    print(f"✅ 扫描计算完毕，耗时: {time.time() - start_time:.2f} 秒。正在合并且保存...")
234

235
    # 将结果按照顺序写入 CSV
236
    # CSV 只会保存真正有归属地或运营商记录的真实号段，抛弃了无用垃圾号段
237
    written_count = 0
238
    with open(out_file, 'w', encoding='utf-8-sig') as f:
239
        f.write("号段前缀,归属城市,网络运营商\n")  # 写入表头
240
        for chunk_result in all_results:
241
            if chunk_result:
242
                f.writelines(chunk_result)
243
                written_count += len(chunk_result)
244

245
    print("=" * 50)
246
    print(f"🎉 导出大功告成！")
247
    print(f"📁 导出文件: {out_file}")
248
    print(f"📊 有效号段: {written_count} 条记录")
249
    print(f"⏱️  总计耗时: {time.time() - start_time:.2f} 秒")
250
    print("=" * 50)
251

252
if __name__ == '__main__':
253
    main()

利用多核 CPU 的并行计算，这 70 万个号段的数据提取在现代处理器上通常只需 1~2 秒即可跑完，真正做到了工业级的性能表现。

4. 全系列总结#

在这个过程中，我们看到了一家成熟的商业手机厂商，为了在极其有限的内存和苛刻的延迟要求下提供优秀的用户体验，所做出的教科书级别的架构设计：

用 12-Bit 位移运算 榨干最后一丝存储空间；
用 墓碑机制与外挂补丁 优雅解决行政区划合并的更新阵痛；
用 多级指针与二分查找 应对携号转网带来的规则崩塌。

在如今 API 满天飞、算力过剩的时代，这种“锱铢必较”的底层工程美学，或许才是逆向分析带给我们的最大乐趣与收获。