打游戏打到一半诈骗电话进来了。相信不少人都遇到过类似场景陌生号码、境外来电、或者自称“公检法”的口音一句“你的账户被冻结”直接打断操作。与其每次靠人工识别、接起来骂两句不如自己做一个本地部署的“诈骗电话识别与拦截辅助系统”。这篇文章就围绕这个场景从号码库检测、通话语音转写、关键词风险判定到接口批量调用给你一套可以落地跑的方案。整个系统重点解决三类问题来电号码是不是高危号、通话内容有没有诈骗话术、以及批量日志如何审计复盘。这里用到的技术都是常规开源的核心依赖就是 FastAPI SQLite Whisper 语音识别对显卡没有硬性要求纯 CPU 也能跑通识别流程。先看核心能力速览再按章节往下部署。文中涉及的代码和配置通用性较强如果后续接入到具体手机自动化工具需要按实际环境调整路径和参数。1. 核心能力速览能力项说明项目类型诈骗电话识别与拦截辅助系统属于本地安全工具类应用主要功能号码风险检测、通话录音语音转写、诈骗关键词规则判定、批量号码审计、Webhook 推送核心依赖Python 3.10、FastAPI、SQLite、WhisperASR、OpenCC可选用于文本归一化硬件要求CPU 可运行Whisper 推荐 8GB 以上内存使用 GPU 可明显提升转写速度显存占用完全取决于 Whisper 模型版本tiny/base 约 1GB 以内small/medium 约 2-5GBlarge 系列更高支持平台Windows / Linux / macOS只要 Python 环境可运行启动方式命令行启动 FastAPI 服务浏览器访问接口文档是否支持 API支持提供 HTTP REST 接口是否支持批量任务支持可批量检测号码和批量转写录音文件适合场景个人来电安全识别、通话录音事后分析、小型团队举报线索汇总需要明确一点这个系统本身不会“替你把电话挂断”。真正拦截来电需要依赖手机系统自带的拦截能力、运营商服务或手机自动化工具。系统做的是“识别”和“判定”给出足够可靠的结论后再触发拦截动作这样能最大限度降低误杀正常电话的风险。2. 适用场景与使用边界先说适用场景。第一是个人手机来电防护。将陌生号码提交到本服务服务返回风险等级再通过手机自动化工具决定是否直接拒绝接听。这里的判断依据是号码黑名单库、号码归属地和号段规则。比如很多诈骗电话来自境外改号、虚拟运营商号段、或者高频拨打的短时号码单靠规则就有一定识别率。第二是通话录音事后分析。如果手机开启了通话录音区域合规前提下把录音文件批量交给 Whisper 转写成文字再提取“转账、验证码、安全账户、逮捕令、京东白条”等关键词命中高危关键词后自动生成审计日志。第三是举报线索汇总。团队运营的电商客服、社区管理员可以把用户提交的可疑号码集中丢进批量检测接口形成一个内部风险号码库后续再遇到直接标记。再说边界。这个系统不适合做的第一不能用它监控别人的通话内容。录音转写只允许用在本人已授权、且符合当地法律法规的场景。第二不能用它批量骚扰或者反向定位某个号码。号码归属地解析只到地市级不能反查个人身份信息。第三不能把它当成绝对准确的诈骗判断器。诈骗手法每天都在变规则引擎必然有漏判最终接不接电话还是你说了算。关于合规简单说三件事通话录音前尽量确认本地法律允许系统收集的号码和文本信息不要公开也不要上传到不受信任的第三方凡是涉及人脸、声音、个人数据的处理必须确保有合法授权。3. 环境准备与前置条件开始写代码之前先检查本机环境。操作系统无特殊要求Windows 10/11、Ubuntu 20.04、macOS 12 都可以。建议使用 Python 3.10 或 3.11避免部分音频依赖在旧版本下编译失败。需要安装的依赖如下pip install fastapi uvicorn aiofiles sqlite3-utils pip install openai-whisper pip install torch torchaudio这里稍作说明openai-whisper默认会安装 PyTorch如果你本机已有 CUDA 版的 PyTorch可以不用重复装直接使用现有环境。如果电脑没有 NVIDIA 显卡安装 CPU 版 PyTorch 即可转写速度慢一点但对功能没有影响。磁盘空间方面Whisper 模型文件会下载到本地缓存目录tiny 模型约 75MBbase 约 142MBsmall 约 466MBmedium 约 1.5GBlarge 约 2.9GB。建议预留 5GB 以上的磁盘空间方便后续切换不同模型测试。目录结构可以这样组织anti-fraud-system/ ├── app.py # FastAPI 主服务 ├── database.py # SQLite 初始化和查询函数 ├── detector.py # 号码规则与关键词判定逻辑 ├── asr_worker.py # Whisper 语音转写封装 ├── audio/ # 录音文件存放目录 ├── data/ │ └── fraud_phone.db # SQLite 数据库文件 ├── outputs/ # 转写结果和审计日志 └── requirements.txt用下面的命令初始化虚拟环境python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate pip install -r requirements.txt端口规划上FastAPI 服务默认监听8000端口。如果在服务器上部署建议改用--host 0.0.0.0 --port 8000并通过防火墙限制访问来源只允许自己的手机和电脑访问。不要直接暴露到公网。4. 号码识别服务搭建与启动先实现最核心的号码检测接口。这一步的目标是输入一个手机号输出风险等级、命中的规则和推荐动作。创建一个database.py初始化 SQLite 数据库import sqlite3 DB_PATH data/fraud_phone.db def init_db(): conn sqlite3.connect(DB_PATH) cur conn.cursor() cur.execute( CREATE TABLE IF NOT EXISTS phone_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, phone TEXT NOT NULL, risk_level INTEGER NOT NULL DEFAULT 0, rule_type TEXT NOT NULL, remark TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) cur.execute( CREATE TABLE IF NOT EXISTS audit_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, phone TEXT NOT NULL, result_json TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close()再写detector.py实现号码规则判断import re import sqlite3 DB_PATH data/fraud_phone.db # 常见诈骗话术关键词实际使用可按业务扩展 FRAUD_KEYWORDS [ 转账, 验证码, 安全账户, 逮捕令, 冻结, 京东白条, 注销贷款, 屏幕共享, 刷单, 投资返利, 征信修复 ] def load_blacklist(): conn sqlite3.connect(DB_PATH) cur conn.cursor() cur.execute(SELECT phone FROM phone_records WHERE risk_level 3) rows cur.fetchall() conn.close() return set(row[0] for row in rows) def is_virtual_operator(phone: str) - bool: # 部分虚拟运营商号段未覆盖所有号段 virtual_prefixes (170, 165, 167, 162, 192) return phone.startswith(virtual_prefixes) def is_overseas_like(phone: str) - bool: # 这里把 开头的号码和 00 开头的号码视为高风险候选 return phone.startswith() or phone.startswith(00) def detect_phone(phone: str) - dict: if not re.fullmatch(r\?\d{7,15}, phone): return {phone: phone, risk_level: 0, rule: invalid_format, action: ignore} blacklist load_blacklist() if phone in blacklist: return {phone: phone, risk_level: 5, rule: blacklist, action: block} if is_overseas_like(phone): return {phone: phone, risk_level: 4, rule: overseas_like, action: warn} if is_virtual_operator(phone): return {phone: phone, risk_level: 3, rule: virtual_operator, action: warn} return {phone: phone, risk_level: 1, rule: normal, action: allow} def match_fraud_keywords(text: str) - list: hit [] for kw in FRAUD_KEYWORDS: if kw in text: hit.append(kw) return hit这里补充说明一下虚拟运营商号段本身不是犯罪分子专属很多正常用户也在用所以风险等级只给 3 分动作是warn而不是block。境外号码同理频繁有海外联系人的人需要自行决定是否开启阻断。接着写app.py主服务from fastapi import FastAPI, Request from pydantic import BaseModel import sqlite3, json, time from database import init_db from detector import detect_phone, match_fraud_keywords app FastAPI(titleAnti Fraud Phone API, version1.0.0) class PhoneRequest(BaseModel): phone: str class BatchPhoneRequest(BaseModel): phones: list[str] class TextRequest(BaseModel): text: str phone: str unknown app.on_event(startup) def startup(): init_db() app.post(/api/detect_phone) def api_detect_phone(req: PhoneRequest): result detect_phone(req.phone) save_audit(req.phone, result) return result app.post(/api/detect_batch) def api_detect_batch(req: BatchPhoneRequest): results [] for phone in req.phones: result detect_phone(phone) save_audit(phone, result) results.append(result) return {count: len(results), results: results} app.post(/api/analyze_text) def api_analyze_text(req: TextRequest): hits match_fraud_keywords(req.text) result { phone: req.phone, text_length: len(req.text), hit_keywords: hits, risk_level: min(5, len(hits) * 2) } save_audit(req.phone, result) return result def save_audit(phone: str, result: dict): conn sqlite3.connect(data/fraud_phone.db) cur conn.cursor() cur.execute( INSERT INTO audit_logs (phone, result_json) VALUES (?, ?), (phone, json.dumps(result, ensure_asciiFalse)) ) conn.commit() conn.close() if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port8000)启动服务python app.py启动后访问http://127.0.0.1:8000/docs可以看到 FastAPI 自动生成的接口文档页面。这里建议先做一个接口连通性测试curl -X POST http://127.0.0.1:8000/api/detect_phone \ -H Content-Type: application/json \ -d {phone: 17012345678}预期返回{ phone: 17012345678, risk_level: 3, rule: virtual_operator, action: warn }说明虚拟运营商号段规则生效。如果输入一个普通手机号返回risk_level1动作是allow。到这里号码识别服务已经可以独立使用。5. 通话语音识别与诈骗话术判定号码识别只能解决“已知号码”或者“明显号段特征”的情况。真正更复杂的场景是号码看起来正常但通话内容却是诈骗。这时候需要对通话录音做语音识别再用文本规则做风险判定。封装一个asr_worker.py使用 Whisper 做本地语音转写from pathlib import Path import whisper _model None def get_model(model_name: str base): global _model if _model is None: _model whisper.load_model(model_name) return _model def transcribe_audio(audio_path: str, model_name: str base) - str: model get_model(model_name) result model.transcribe(audio_path, languagezh, fp16False) return result[text]使用方法from asr_worker import transcribe_audio from detector import match_fraud_keywords text transcribe_audio(audio/call_20250101_1023.mp3, model_namebase) print(识别结果, text) hits match_fraud_keywords(text) print(命中关键词, hits)如果输入的录音文件时长超过 5 分钟Whisper 默认流式处理会自动切分但整段识别耗时较长。官方建议音频采样率在 16kHz 左右效果较好部分手机录音默认是 44.1kHz识别前可以先做重采样统一降低到 16kHz 单声道能明显减少计算量。下面给一个通用的音频预处理脚本依赖ffmpegffmpeg -i audio/original_call.m4a -ar 16000 -ac 1 audio/call_16k.wav预处理之后再丢给 Whisper识别速度和准确率都更可控。文本分析和号码检测结合之后可以做一条完整的判定流水线from database import init_db from detector import detect_phone, match_fraud_keywords from asr_worker import transcribe_audio def analyze_call(phone: str, audio_path: str): phone_result detect_phone(phone) text transcribe_audio(audio_path, model_namebase) hits match_fraud_keywords(text) final_score phone_result[risk_level] min(5, len(hits) * 2) final_level min(10, final_score) return { phone: phone, phone_rule: phone_result[rule], transcript: text, hit_keywords: hits, final_risk_level: final_level, suggestion: block if final_level 7 else warn }这个流水线的逻辑很简单号码规则给出初始分数文本关键词命中越多分数加得越高最终超过阈值就建议拦截。实际使用中阈值 7 可以按自己的误报容忍度调整。6. 接口 API 与批量任务系统已经暴露了/api/detect_phone和/api/analyze_text两个基础接口。如果要做批量录音审计还需要增加一个批量转写任务接口。思路是把待处理的录音文件放到audio/目录接口扫描目录内所有音频文件逐个转写并保存结果到outputs/。目录结构天然就是任务队列不需要引入 Redis 或 Celery。新增一个批量接口from fastapi import FastAPI from pathlib import Path from asr_worker import transcribe_audio from detector import match_fraud_keywords import json, uuid AUDIO_DIR Path(audio) OUTPUT_DIR Path(outputs) app.post(/api/batch_scan_audio) def api_batch_scan_audio(): audio_files list(AUDIO_DIR.glob(*.mp3)) list(AUDIO_DIR.glob(*.wav)) list(AUDIO_DIR.glob(*.m4a)) results [] for audio_path in audio_files: task_id str(uuid.uuid4()) try: text transcribe_audio(str(audio_path), model_namebase) hits match_fraud_keywords(text) result { task_id: task_id, file: audio_path.name, text: text, hit_keywords: hits, risk_level: min(5, len(hits) * 2) } except Exception as exc: result { task_id: task_id, file: audio_path.name, error: str(exc) } results.append(result) output_path OUTPUT_DIR / f{audio_path.stem}.json output_path.write_text(json.dumps(result, ensure_asciiFalse, indent2), encodingutf-8) return {count: len(results), results: results}前端调用示例curl -X POST http://127.0.0.1:8000/api/batch_scan_audio如果audio/目录里有 3 个录音文件接口会依次转写并返回一个包含 3 个结果对象的数组。由于 Whisper 转写本身是耗时的这个接口在 CPU 环境下可能运行几十秒甚至几分钟建议通过前端轮询结果文件而不是一直等待 HTTP 响应超时。Python 调用示例import requests resp requests.post(http://127.0.0.1:8000/api/batch_scan_audio, timeout600) data resp.json() print(data[count]) for item in data[results]: print(item[file], item[hit_keywords], item[risk_level])批量号码检测的调用更简单import requests payload { phones: [ 17012345678, 13800138000, 85212345678 ] } resp requests.post(http://127.0.0.1:8000/api/detect_batch, jsonpayload, timeout30) print(resp.json())接口返回示例{ count: 3, results: [ { phone: 17012345678, risk_level: 3, rule: virtual_operator, action: warn }, { phone: 13800138000, risk_level: 1, rule: normal, action: allow }, { phone: 85212345678, risk_level: 4, rule: overseas_like, action: warn } ] }实际项目中批量任务必须加日志和失败重试。从上面代码可以看到每个文件转写是独立的try-except单文件失败不会导致整个流程崩溃。但我建议在文件落盘时记录任务状态比如增加pending、done、failed三个状态字段方便失败后重新扫描。7. 资源占用与性能观察这个系统的资源消耗主要集中在两个部分FastAPI 服务本身几乎不消耗 GPUCPU 和内存占用都非常低真正的资源大头是 Whisper 语音转写。先看 CPU 推理。纯 CPU 环境下Whisper base 模型转写一段 30 秒的录音耗时大约在 20-60 秒之间具体取决于 CPU 性能。如果录音时长 5 分钟预计要 5-10 分钟才能完成转写。这里不能给精确数字因为不同 CPU 差异很大。更稳妥的判断是CPU 模式适合事后分析不适合实时通话转写。如果希望通话结束后几十秒内出结果建议做两件事。第一音频先截断到前 30 秒诈骗话术通常开局就有信号不需要等整段录音转完。第二使用 GPU 跑 Whisper small 或 medium 模型。再看显存。GPU 推理时Whisper 的显存占用情况大致如下但还是以实际模型版本和推理时长为准Whisper 模型参数量显存占用区间适用情况tiny39M1GB 以内快速验证流程base74M1-2GB低显存优先small244M2-4GB准确率与速度平衡medium769M4-6GB长音频更稳large1550M6GB 以上追求最高准确率如果显存不够有两个降载手段。第一模型加载后不要反复调用whisper.load_model直接复用全局_model变量避免重复分配显存。第二音频重采样到 16kHz 单声道减少输入数据的处理量。端口和进程方面也要注意。FastAPI 默认端口 8000如果本地已经有其他服务占用了 8000启动会报Address already in use。测试期间建议用 netstat 或 lsof 检查端口# Linux / macOS lsof -i :8000 # Windows netstat -ano | findstr :8000如果端口被占用换端口启动uvicorn app:app --host 127.0.0.1 --port 8001进程残留问题同样容易忽略。在 Jupyter 里跑过服务或者在 IDE 中手动终止过 app 后下次启动可能出现端口仍被占用的情况原因就是后台 Python 进程没有完全退出。这时用进程管理工具清理即可。8. 常见问题与排查方法问题现象可能原因排查方式解决方案pip install openai-whisper失败Python 版本过低或缺少编译工具检查 Python 版本查看完整报错日志使用 Python 3.10安装 build-essentialLinux或 Visual Studio Build ToolsWindows启动python app.py报端口被占用8000 端口已有进程netstat -ano | findstr :8000换端口启动或杀死占用进程Whisper 下载模型卡住网络无法访问模型下载地址查看模型缓存目录确认进度手动下载模型文件并放到~/.cache/whisper或配置镜像源转写结果为空白录音文件损坏或格式不支持先用 ffprobe 查看音频信息统一转成 16kHz WAV 后再转写转写速度极慢使用 CPU 推理且模型过大查看任务管理器中的 CPU 占用更换 tiny/base 模型或截取录音前 30 秒处理接口返回 500SQLite 数据库目录不存在检查日志中的报错信息确认data/目录存在或调用init_db()初始化批量扫描接口超时音频文件过多同步执行导致请求阻塞查看 HTTP 超时时间和日志改成异步任务客户端轮询结果文件号码识别结果不准确规则库覆盖不全黑名单太少查看命中规则手动标注错误样本持续往phone_records表补充黑名单号码误杀正常号码虚拟运营商或境外号段正常使用查看风险等级和动作建议将阈值调高或调整规则只标记不拦截排查有一个通用思路先看日志再看数据库最后确认网络和依赖。FastAPI 启动后终端会输出每条请求的POST /api/detect_phone 200 OK之类信息如果状态码不是 200优先复制报错栈而不是直接猜问题。转写结果不准确也是高频问题。中文口语中“转账”和“赚钱”容易被误识别诈骗团伙有时候也会故意用同音字规避关键词。针对这个情况建议把规则引擎升级成本地大模型判断输入识别文本后让模型输出“是否疑似诈骗”和“简短原因”。这块可以用系统里的 OpenCC 先做繁体转简体文本归一化再用关键词做粗筛命中后再交给大模型复判可以控制成本。9. 最佳实践与使用建议第一次使用这个系统时建议按下面的顺序推进。第一先小参数测试。不要第一次就把 100 个号码导入批量接口。先用 5 个号码、1 段录音验证流程能否走通确认接口返回正常再逐步扩大样本量。这样可以快速定位问题也不会因为批量任务失败而浪费大量时间。第二保留一套最小可运行配置。把虚拟环境依赖、数据库初始化脚本、启动命令写成一个 README。即使三个月后再回来用也能快速恢复环境。这里强烈建议把requirements.txt固定版本号避免升级后行为变化。第三目录管理要清晰。建议强制按下面的结构归档audio/ raw/ # 原始录音 processed/ # 已转写录音 outputs/ transcripts/ # 转写文本 reports/ # 审计报告 data/ fraud_phone.db录音文件处理完成后原始文件可以保留一段时间但建议设置自动清理策略避免磁盘被占满。第四批量任务必须加日志和失败重试。前面写的batch_scan_audio接口是同步阻塞的批量文件多了以后一定要改成异步任务。最简单的方案是用一个后台线程池处理任务任务状态写回 SQLite。每次转写失败后标记failed重跑时只处理这些失败文件。第五接口服务要限制访问范围。如果服务部署在公网服务器上至少要做 IP 白名单或者基础鉴权否则任何人扫到你服务器的 8000 端口都能调用接口白白消耗资源。一个简单的做法是在 FastAPI 里加一个X-API-Key中间件from fastapi import Header, HTTPException API_KEY your-secret-key def verify_key(x_api_key: str Header(default)): if x_api_key ! API_KEY: raise HTTPException(status_code401, detailInvalid API Key)然后在需要保护的接口上添加依赖app.post(/api/detect_phone, dependencies[Depends(verify_key)]) def api_detect_phone(req: PhoneRequest): result detect_phone(req.phone) save_audit(req.phone, result) return result第六涉及录音和人脸等敏感信息时必须确认授权。这个系统如果只是识别个人来电号码风险较低一旦开始处理真实通话录音就进入强合规范围。录音前建议确认所在地区法律法规允许并在系统内保留录音来源说明和授权记录。不要把这个工具用于监控同事、家人或任何未经授权的对象。第七发布或商用前要做效果复核。规则引擎难免误判建议每次批量分析后人工抽检部分结果特别是标记为“高风险”的样本。抽检比例至少 10%否则模型或规则的偏差会被静默放大。10. 总结与下一步这次我们围绕“打游戏接到诈骗电话”这个真实痛点搭建了一个本地号码与通话内容双重识别的反诈辅助系统。核心思路是先做号码规则检测再做语音转写和关键词判定最后通过接口批量审计录音文件。整个流程并不复杂FastAPI 提供接口层SQLite 保存黑名单和审计日志Whisper 负责把通话录音变成可检索的文本。最值得先验证的功能是号码检测接口。打开http://127.0.0.1:8000/docs输入一个虚拟运营商号段手机号看返回的风险等级和推荐动作是否符合预期。这一步跑通了后面的批量号码检测和录音分析就可以逐步接入。最容易踩的坑是 Whisper 的模型下载和 CPU 推理速度。第一次使用时建议先选 tiny 或 base 模型跑通流程不要一上来就 large。真实场景中与其追求一次识别准不如先把“录音截断到前 30 秒 关键词规则初筛”这套轻量流程跑起来效率提升更明显。后续可以扩展的方向有三个第一个是把号码黑名单做成自动导入从运营商举报接口或者其他可信渠道同步数据第二个是把关键词规则升级成本地大模型判断对同音字、隐晦表达有更好的识别能力第三个是接入手机自动化工作流风险等级高时自动挂断或者通过推送通知提醒用户。建议收藏备用下次再接到可疑电话先拿来检测一下号码再决定接不接。