企业官网建设流程全解析

打游戏打到一半，诈骗电话进来了。相信不少人都遇到过类似场景：陌生号码、境外来电、或者自称“公检法”的口音，一句“你的账户被冻结”直接打断操作。与其每次靠人工识别、接起来骂两句，不如自己做一个本地部署的“诈骗电话识别与拦截辅助系统”。这篇文章就围绕这个场景，从号码库检测、通话语音转写、关键词风险判定到接口批量调用，给你一套可以落地跑的方案。整个系统重点解决三类问题：来电号码是不是高危号、通话内容有没有诈骗话术、以及批量日志如何审计复盘。这里用到的技术都是常规开源的，核心依赖就是 FastAPI + SQLite + Whisper 语音识别，对显卡没有硬性要求，纯 CPU 也能跑通识别流程。

先看核心能力速览，再按章节往下部署。文中涉及的代码和配置通用性较强，如果后续接入到具体手机自动化工具，需要按实际环境调整路径和参数。

1. 核心能力速览

需要明确一点：这个系统本身不会“替你把电话挂断”。真正拦截来电需要依赖手机系统自带的拦截能力、运营商服务或手机自动化工具。系统做的是“识别”和“判定”，给出足够可靠的结论后再触发拦截动作，这样能最大限度降低误杀正常电话的风险。

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 模型约 75MB，base 约 142MB，small 约 466MB，medium 约 1.5GB，large 约 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(title="Anti Fraud Phone API", version="1.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_ascii=False)) ) conn.commit() conn.close() if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)

启动服务：

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_level=1，动作是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, language="zh", fp16=False) return result["text"]

使用方法：

from asr_worker import transcribe_audio from detector import match_fraud_keywords text = transcribe_audio("audio/call_20250101_1023.mp3", model_name="base") print("识别结果：", text) hits = match_fraud_keywords(text) print("命中关键词：", hits)

如果输入的录音文件时长超过 5 分钟，Whisper 默认流式处理会自动切分，但整段识别耗时较长。官方建议音频采样率在 16kHz 左右效果较好，部分手机录音默认是 44.1kHz，识别前可以先做重采样，统一降低到 16kHz 单声道，能明显减少计算量。

下面给一个通用的音频预处理脚本，依赖ffmpeg：

ffmpeg -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_name="base") 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_name="base") 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_ascii=False, indent=2), encoding="utf-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", timeout=600) 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", json=payload, timeout=30) 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 服务本身几乎不消耗 GPU，CPU 和内存占用都非常低；真正的资源大头是 Whisper 语音转写。

先看 CPU 推理。纯 CPU 环境下，Whisper base 模型转写一段 30 秒的录音，耗时大约在 20-60 秒之间，具体取决于 CPU 性能。如果录音时长 5 分钟，预计要 5-10 分钟才能完成转写。这里不能给精确数字，因为不同 CPU 差异很大。更稳妥的判断是：CPU 模式适合事后分析，不适合实时通话转写。如果希望通话结束后几十秒内出结果，建议做两件事。第一，音频先截断到前 30 秒，诈骗话术通常开局就有信号，不需要等整段录音转完。第二，使用 GPU 跑 Whisper small 或 medium 模型。

再看显存。GPU 推理时，Whisper 的显存占用情况大致如下，但还是以实际模型版本和推理时长为准：

如果显存不够，有两个降载手段。第一，模型加载后不要反复调用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. 常见问题与排查方法

排查有一个通用思路：先看日志，再看数据库，最后确认网络和依赖。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_code=401, detail="Invalid 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 秒 + 关键词规则初筛”这套轻量流程跑起来，效率提升更明显。

后续可以扩展的方向有三个：第一个是把号码黑名单做成自动导入，从运营商举报接口或者其他可信渠道同步数据；第二个是把关键词规则升级成本地大模型判断，对同音字、隐晦表达有更好的识别能力；第三个是接入手机自动化工作流，风险等级高时自动挂断或者通过推送通知提醒用户。

建议收藏备用，下次再接到可疑电话，先拿来检测一下号码，再决定接不接。