Visshu78/TeleWire
GitHub: Visshu78/TeleWire
TeleWire 是一款实时 Telegram 社交威胁情报监控分析平台，解决从海量消息中自动化提取实体、分类威胁、评估风险并生成可操作情报简报的问题。
Stars: 0 | Forks: 0
# TeleWire: Telegram SOCMINT 威胁情报 Pipeline
TeleWire 是一个生产级的实时 Telegram 威胁情报监控平台。它可以异步摄取消息，执行高级实体提取，运行语义零样本 (zero-shot) 威胁分类，通过向量嵌入聚类协调活动，根据制裁黑名单检查钱包，执行多模态 OCR 与二维码解码，评估综合风险评分，可视化转发网络拓扑，并支持分析师的案例简报与报告生成。
## 🚀 快速开始（本地沙盒）
### 1. 初始化环境
要配置你的 Python 虚拟环境、安装包依赖项、初始化 SQLite schema 并运行单元测试，请执行：
```
./scripts/dev_bootstrap.ps1
```
*或者，使用以下命令手动安装依赖：* `pip install -r requirements.txt`
### 2. 配置 `.env` 环境
将模板 `.env.example` 复制为 `.env` 并填写凭据：
```
copy .env.example .env
```
核心配置：
* `API_ID` & `API_HASH`: 从 https://my.telegram.org -> API Development Tools 获取。
* `PHONE`: 绑定你 Telegram 账户的主电话号码。
* `ALERT_THRESHOLD`: 触发警报的风险阈值（0-100）（默认值：`70`）。
* `ALERT_WEBHOOK_URL`: 用于接收实时 JSON 警报 POST payload 的可选 endpoint。
* `ALERT_TELEGRAM_BOT_TOKEN` & `ALERT_TELEGRAM_CHAT_ID`: 用于直接在你的 Telegram 聊天中接收严重警报的可选凭据。
### 3. 启动摄取 Pipeline
要启动 Telethon 监听 daemon 和 Flask 分析师界面：
```
python main.py
```
*注意：仅在你的首次执行时，控制台会提示你输入 Telegram OTP 验证码进行身份验证。session 缓存将写入 `telegram_session.session` 以跳过后续登录。对于多账户设置，也可以在 Web UI 上实时发起身份验证。*
* **Flask Web UI 仪表盘:** 在 **[http://localhost:5000](http://localhost:5000)** 开启
* **Streamlit 关键词管理器:** 运行 `streamlit run keyword_manager.py` (在 **[http://localhost:8501](http://localhost:8501)** 开启)。
### 4. 模拟模拟活动（沙盒模式）
如果你想在没有实时 Telegram 账户的情况下评估仪表盘图表、警报面板和行为者档案，请运行：
```
python scripts/generate_mock_stream.py --delay 2.5
```
该脚本会模拟真实目标频道的流量，提取测试钱包地址，运行语义零样本分类器，对风险进行评分，更新档案，并动态触发仪表盘警告。
## 🛠️ 系统架构与摄取流程
```
+-----------------------+
| Telegram Clients |
| (Up to 5 accounts) |
+-----------+-----------+
|
| (NewMessage events)
v
+-----------------------------------------------------------------------------------------+
| INGESTION LAYER (src/ingestion/) |
| |
| +----------------------------+ +---------------------+ +------------------+ |
| | PipelineManager (asyncio) | ---> | TelegramListener(s) | ---> | asyncio.Queue | |
| +----------------------------+ +----------+----------+ +--------+---------+ |
| | | |
| | (Active check) | |
| v v |
| +---------------------+ +------------------+ |
| | ActiveGroupCache | | Worker Pool (x5) | |
| +---------------------+ +--------+---------+ |
+----------------------------------------------------------------------------|------------+
| (Thread Pool Executor)
v
+-----------------------------------------------------------------------------------------+
| PROCESSING LAYER (src/processing/) |
| |
| 1. Deduplicator (SHA-256) |
| 2. Entity Extractor (wallets, UPIs, phones, emails, etc.) |
| 3. OFAC SDN Sanctions Matcher (XBT, ETH, TRX local cache) |
| 4. MediaProcessor (pytesseract OCR, pyzbar QR decode, pHash visual grouping) |
| 5. SemanticProcessor (SentenceTransformers all-MiniLM-L6-v2, FAISS, Zero-shot NLP) |
| 6. Threat Scoring Service (Composite risk scoring 0-100 & Telegram/Webhook Alerts) |
+----------------------------------------------------------------------------|------------+
|
v
+-----------------------------------------------------------------------------------------+
| STORAGE & REPRESENTATION LAYERS (src/storage/ & src/dashboard/) |
| |
| +--------------------------+ +------------------------+ +------------------+ |
| | SQLite (WAL Mode) | <-> | Flask Dashboard (:5000)| <-> |Streamlit Manager | |
| | (telegram_intel.db) | | (Cytoscape graphs) | | (:8501, .pptx) | |
| +--------------------------+ +------------------------+ +------------------+ |
| | |
| +----------------> JSONL Backup Files (data/backup/YYYY-MM-DD.jsonl) |
+-----------------------------------------------------------------------------------------+
```
### 线程与并发模型
* **主线程 (asyncio loop):** 管理所有活动的 Telegram 客户端连接，注册事件处理程序，并运行 watchdog 重连循环。
* **监听 Worker:** 由 5 个异步 worker 线程组成的池从摄取队列中拉取消息，以减轻解析负担。
* **Thread Pool Executor:** 专用于 CPU 密集型提取和阻塞 I/O（SQLite 写入、图像 OCR、零样本分类、FAISS 聚类）的专用执行池。
* **Flask Web 服务器:** 在单独的 daemon 线程中运行，共享启用 WAL 模式的数据库。
* **Streamlit 管理器:** 作为独立进程启动，用于并发的关键词 CRUD 操作。
## 💎 功能详情与技术规范
### 1. 摄取性能与可靠性
* **高吞吐监听器 (`src/ingestion/listener.py`):** 监控已加入的频道，立即将原始事件在内存中排队，以防止 Telethon 阻塞。
* **Worker Pool:** 5 个并行的 worker 线程从队列中拉取数据，每 20 个事件或 500ms 使用事务性 SQL `executemany` 语句提交消息批次，以绕过数据库瓶颈问题。
* **活动群组门控 (`ActiveGroupCache`):** 一个内存中的受监控频道 ID 集合，无需启动数据库读取即可立即解析状态。
* **可靠的 Watchdog 与重连:** 拦截 RPC/网络故障。实现指数退避（5秒到最大 120秒）。如果触发速率限制，它会休眠 Telegram 要求的时长（`FloodWaitError` 秒数），并在重新连接时利用严格单调的消息 ID（`last_message_id`）自动执行完美的间隙补全查询。
### 2. 多账户摄取控制中心
* **Pipeline 生命周期 Daemon (`src/ingestion/pipeline_manager.py`):** 并行管理并发 Telethon 客户端（最多支持 5 个不同的手机号）。
* **基于 Web 的 OTP 身份验证:** 直接从仪表盘实时添加新手机号，异步执行 OTP 握手序列，并通过动态表单提示分析师。
* **运行时切换:** 通过全局标志更改暂停/恢复所有流的更新获取，而不会中断 Telegram socket 连接。
### 3. 实体提取与制裁审计
* **实体提取器 (`src/processing/entity_extractor.py`):** 验证并提取电话号码、电子邮件、URL、IP 地址、Telegram 用户名、加密货币钱包地址（BTC legacy/Bech32、Ethereum、TRON/TRC20、TON user-friendly）、UPI ID、IBAN 和信用卡号。
* **OFAC 制裁审计器 (`src/processing/wallet_enricher.py`):** 自动在后台下载并缓存美国财政部 SDN 名单（XBT, ETH, TRX）。在摄取时将提取的钱包与这些记录进行匹配，一旦命中受制裁目标即刻触发警报。
* **区块链丰富化档案:** 后台线程查询公共区块链 API（Blockchair, TronGrid, TonCenter）以获取余额、交易计数和活动时间戳，并通过 2 秒的休眠间隔遵守速率限制。
### 4. 零样本 NLP 与 FAISS 活动聚类
* **Sentence Transformers (`src/processing/semantic_service.py`):** 使用 `all-MiniLM-L6-v2` 将文本和 OCR 提取内容向量化为 384 维 embeddings，并将本地模型结构缓存在 `data/models/` 下。
* **零样本分类器:** 评估消息 embeddings 与预嵌入威胁定义的余弦相似度：Scam/Fraud (诈骗/欺诈)、Weapons/Violent Extremism (武器/暴力极端主义)、Cybersecurity/Hacking (网络安全/黑客攻击)、Financial Crimes/Money Mule (金融犯罪/资金骡子)、Drug Trafficking (毒品走私) 以及 Legitimate (合法)。
* **FAISS 活动聚类:** 查询余弦相似度索引 (`faiss.IndexFlatIP`)，将近似重复的文本（相似度 $\ge 0.85$）映射到统一的 `campaigns` 中，而不是单条消息，从而大幅减少分析师的噪音干扰。
### 5. 多模态 OCR 与图像哈希
* **OCR 文本解析 (`src/processing/media_processor.py`):** 自动下载附件并利用 PyTesseract 从图像中提取文本。
* **QR 扫描器:** 使用 pyzbar 从二维码中提取支付链接和 UPI 信息。
* **感知图像哈希:** 通过 pHash（Hamming distance $\le 10$）利用 `imagehash.phash` 对视觉图形进行分组，以映射协调的视觉垃圾邮件活动。
* **自动存储清理:** 删除超过 30 天或总缓存超过 5GB 的媒体资产，在清理本地磁盘存储的同时保留数据库引用。
### 6. 威胁风险评分与警报调度器
* **加权威胁风险评分器 (`src/processing/scoring_service.py`):** 结合制裁名单匹配 (+40)、威胁类别 (+20 到 +30)、关键词匹配接近度 (+15)、活动归属 (+10) 以及紧急语言启发式算法 (+10)，计算出一个综合得分 (0-100)。
* **实时调度器:** 如果风险得分超过 `ALERT_THRESHOLD`，则通过 Webhook POST 请求向配置的 endpoint 发送警告 payload 和带有样式的 Telegram markdown 警报。
### 7. 有向网络中心性与时间档案分析
* **拓扑图构建器 (`src/processing/network_service.py`):** 使用 `NetworkX` 将频道转发关系表述为有向网络。计算 PageRank 中心性、入度和出度，并输出用于交互式 Cytoscape.js 布局的元素。
* **行为者画像分析:** 计算发送者威胁风险得分的滚动平均值，将发送者分类为风险等级（Critical, High, Medium, Low）。
* **时区推断:** 将发布时间戳历史记录映射为 24 小时的 UTC 直方图，以估算威胁行为者的时区。
### 8. 案例构建工作区
* **案例文件夹 (`src/processing/reporting_service.py`):** 将相关发现（威胁消息、钱包和行为者用户名）归组到单个文件夹中。
* **高管情报简报:** 生成并汇编一份可下载、带有样式的高管执行摘要 Markdown 文件，其中包含案例清单指标、风险评估以及格式化的消息日志。
## 📁 代码库目录结构
```
TELEGRAM/
│
├── main.py # Main application startup, config loading, and loop initialization
├── keyword_manager.py # Streamlit UI keyword tool (separate terminal)
├── migrate_db.py # Older standalone migration helper (deprecated; main.py migrates additively)
├── requirements.txt # Application Python dependencies
├── README.md # General project documentation (This file)
├── PIPELINE.md # Operational technical flow reference and developer notes
├── Advanced_Features.md # Project advancement roadmap
├── .env.example # Template configuration file for environmental secrets
│
├── config/
│ ├── config.yaml # Runtime application parameters
│ └── keywords.txt # Seed keywords loaded into SQLite on bootstrap
│
├── data/ # Created automatically on first execution
│ ├── telegram_intel.db # WAL-enabled SQLite database
│ ├── faiss_index.bin # Saved FAISS index vector database
│ ├── faiss_mapping.json # Mapping structure matching FAISS indices to message IDs
│ ├── media/ # Local media cache directory
│ ├── models/ # Local models cache directory
│ ├── sanction_lists/ # OFAC SDN local cache text files
│ └── backup/
│ └── YYYY-MM-DD.jsonl # Raw JSONL daily message logging fallback
│
├── src/
│ ├── __init__.py
│ ├── ingestion/
│ │ ├── __init__.py
│ │ ├── client.py # Telethon builder, session authorization, and channel syncing
│ │ ├── listener.py # Ingestion queue worker, active checks, and fwd_from parsing
│ │ └── pipeline_manager.py # Multi-account configuration controller (up to 5 sessions)
│ ├── processing/
│ │ ├── __init__.py # ProcessingEngine facade
│ │ ├── deduplicator.py # Thread-safe SHA-256 hash duplication manager
│ │ ├── entity_extractor.py # Regex extractors, checksum validators, and OFAC lists
│ │ ├── keyword_matcher.py # RapidFuzz matcher, dynamic Hinglish thresholding, and backmatching
│ │ ├── media_processor.py # PyTesseract OCR, pyzbar QR scanning, and pHash calculators
│ │ ├── network_service.py # NetworkX directed graph PageRank calculations
│ │ ├── pptx_extractor.py # Slide token extractor for Streamlit
│ │ ├── reporting_service.py # Case Files Markdown generator
│ │ ├── scoring_service.py # Composite 0-100 risk math and alarm webhook/bot dispatcher
│ │ ├── semantic_service.py # all-MiniLM-L6-v2 zero-shot classification and FAISS clusters
│ │ └── wallet_enricher.py # Blockchain API enrichment background thread
│ └── storage/
│ ├── __init__.py
│ └── database.py # SQLite WAL settings, schema layouts, and migrations
│
├── static/
│ ├── style.css # Dark glassmorphism dashboard UI theme
│ └── dashboard.js # Dashboard frontend routing and Cytoscape graphing logic
│
├── tests/ # Full test suite covering application modules
│ ├── test_entities.py # Entity patterns and checksum verification tests
│ ├── test_media.py # OCR, QR, and pHash integration tests
│ ├── test_multi_account.py # PipelineManager and account management tests
│ ├── test_network.py # Directed forward graph and centrality tests
│ ├── test_pipeline.py # Ingestion queue, batch writing, and backfill tests
│ ├── test_reporting.py # Case Brief and reporting structure tests
│ ├── test_sandbox.py # Sandbox mock data stream validation
│ ├── test_scoring.py # Composite score calculations and webhook dispatch tests
│ └── test_semantics.py # Zero-shot classification and FAISS clustering tests
```
## 🗄️ 数据库 Schema 参考
该数据库使用经过 **WAL 模式**、`PRAGMA synchronous=NORMAL` 和页缓存配置优化的 SQLite。
### 1. 表: `messages`
存储丰富化后的威胁消息项、匹配标记以及媒体详情。
* `id` (INTEGER, Primary Key): 自增行 ID。
* `message_id` (INTEGER): 来自 Telegram 的原始消息 ID。
* `group_name` (TEXT): 目标群组标题。
* `group_id` (INTEGER): 目标群组/频道的数字 ID。
* `sender_name` (TEXT): 发送者显示名称。
* `sender_phone` (TEXT): 发送者电话号码（如果可见）。
* `text` (TEXT): 完整的消息正文文本。
* `language` (TEXT): 检测到的语言（例如 `en`, `hi`, `ru`）。
* `is_forwarded` (INTEGER): 如果是转发则为 `1`；否则为 `0`。
* `forward_from_name` (TEXT): 原始来源频道的标题。
* `forward_from_id` (INTEGER): 原始来源频道的数字 ID（真实来源）。
* `matched_keyword` (TEXT): 最佳匹配的关键词短语。
* `fuzzy_score` (REAL): RapidFuzz 得分。
* `is_matched` (INTEGER): 如果 fuzzy_score $\ge$ 阈值则为 `1`；否则为 `0`。
* `timestamp` (TEXT): ISO-8601 UTC 时间戳。
* `hash` (TEXT, Unique): 唯一索引码 `SHA-256(message_id:group_id)`。
* `campaign_id` (TEXT): 关联的 FAISS 活动聚类标识符。
* `threat_category` (TEXT): 零样本分类结果。
* `media_path` (TEXT): 指向本地媒体缓存的相对路径。
* `ocr_text` (TEXT): 通过 PyTesseract 提取的文本。
* `phash` (TEXT): 感知图像哈希。
* `qr_codes` (TEXT): 提取的二维码 payload 的 JSON 列表。
* `risk_score` (REAL): 综合威胁得分 (0-100)。
* `fetched_by` (TEXT): 获取此消息的 Telegram 账户的手机号。
### 2. 表: `groups`
维护受监控频道的注册表。
* `id` (INTEGER, Primary Key): 行 ID。
* `group_id` (INTEGER, Unique): Telegram 实体的数字 ID。
* `group_name` (TEXT): 显示标题。
* `group_type` (TEXT): `'group'` 或 `'channel'`。
* `is_active` (INTEGER): 监控状态（`1` = 活动; `0` = 非活动）。
* `member_count` (INTEGER): 参与者数量。
* `last_message_at` (TEXT): 最后处理消息的 ISO-8601 UTC 时间戳。
* `last_message_id` (INTEGER): 已处理的最高 Telegram 消息 ID（补全锚点）。
* `added_at` (TEXT): 注册时间戳。
### 3. 表: `keywords`
受监控关键词规则注册表。
* `id` (INTEGER, Primary Key): 行 ID。
* `keyword` (TEXT, Unique, NOCASE): 被追踪的短语。
* `source` (TEXT): 来源标签（`manual`, `pptx` 等）。
* `added_at` (TEXT): 添加时间戳。
### 4. 表: `pipeline_events`
摄取生命周期状态指标。
* `id` (INTEGER, Primary Key): 行 ID。
* `event_type` (TEXT): 事件类别（`startup`, `disconnect`, `reconnect`, `backfill`）。
* `started_at` (TEXT): 开始时间戳。
* `ended_at` (TEXT): 终止时间戳。
* `duration_seconds` (REAL): 事件持续时间。
* `details` (TEXT): 性能信息（例如 `"backfilled=42"`）。
### 5. 表: `entities`
规范化的实体记录。
* `id` (INTEGER, Primary Key): 主键。
* `_type` (TEXT): 实体类别（例如 `phone_number`, `crypto_btc`, `upi_id`）。
* `entity_value` (TEXT): 规范化的字符串值。
* *Unique Constraint:* `(entity_type, entity_value)`
### 6. 表: `message_entities`
将消息链接到实体的多对多关系。
* `message_id` (INTEGER): 引用 `messages(id)` 的外键。
* `entity_id` (INTEGER): 引用 `entities(id)` 的外键。
* `position_in_text` (INTEGER): 字符串在原始文本中的起始位置。
### 7. 表: `wallet_enrichments`
存储由区块链 API worker 解析的加密货币钱包详情。
* `entity_id` (INTEGER, Primary Key): 引用 `entities(id)` 的外键。
* `balance` (REAL): 当前 token 余额。
* `tx_count` (INTEGER): 交易计数。
* `total_volume` (REAL): 处理的总量。
* `first_active` (TEXT): 第一笔交易的时间戳。
* `last_active` (TEXT): 最后一笔交易的时间戳。
* `is_sanctioned` (INTEGER): 如果存在于 OFAC SDN 名单中则为 `1`；否则为 `0`。
* `enrichment_source` (TEXT): API 名称（例如 `Blockchair`）。
* `last_enriched_at` (TEXT): 更新时间戳。
### 8. 表: `campaigns`
存储协调活动的详情。
* `id` (TEXT, Primary Key): 活动 hash 标识符。
* `campaign_name` (TEXT): 易读的名称。
* `first_seen_at` (TEXT): 第一条消息的时间戳。
* `last_seen_at` (TEXT): 最后一条消息的时间戳。
* `threat_category` (TEXT): 共享的威胁分类。
* `representative_text` (TEXT): 文本片段。
### 9. 表: `sender_profiles`
发送者风险汇总表。
* `id` (INTEGER, Primary Key): 主键。
* `sender_id` (TEXT, Unique): Telegram 发送者标识符。
* `sender_phone` (TEXT): 发送者电话（如果可见）。
* `total_messages` (INTEGER): 发送的消息数。
* `cumulative_risk` (REAL): 累计风险点。
* `average_risk` (REAL): 平均风险。
* `last_seen_at` (TEXT): 最后一条消息的时间戳。
* `risk_tier` (TEXT): 分类的等级（`Critical`, `High`, `Medium`, `Low`）。
### 10. 表: `cases`
分析师案例汇编目录。
* `id` (TEXT, Primary Key): 案例 UUID。
* `title` (TEXT): 案例文件夹标题。
* `description` (TEXT): 分析师高管概述。
* `created_at` (TEXT): 创建时间戳。
### 11. 表: `case_items`
将消息行、加密货币钱包和行为者映射到特定的案例文件夹。
* `id` (INTEGER, Primary Key): 行 ID。
* `case_id` (TEXT): 引用 `cases(id)` 的外键。
* `item_type` (TEXT): `'message'`, `'wallet'`, 或 `'actor'`。
* `item_value` (TEXT): 行 ID 或地址值字符串。
* `added_at` (TEXT): 映射创建时间。
### 12. 表: `watchlists`
书签筛选指标。
* `id` (TEXT, Primary Key): 监视列表 UUID。
* `name` (TEXT): 标题。
* `query_params` (TEXT): JSON 参数。
* `created_at` (TEXT): 创建时间戳。
### 13. 表: `telegram_accounts`
受监控的 Telegram session 凭据。
* `phone` (TEXT, Primary Key): 电话号码键。
* `api_id` (INTEGER): Telegram 应用配置标识符。
* `api_hash` (TEXT): Telegram 应用配置 hash。
* `session_name` (TEXT, Unique): 磁盘上 Telegram session 的文件路径。
* `is_active` (INTEGER): 监控标志。
* `status` (TEXT): 连接状态（`connected`, `disconnected`, `needs_otp`）。
## 📊 侧边栏导航工作区及其工作原理
### 🏠 1. Dashboard
* **它是什么:** 显示系统状态卡片和交互式指标图表的高级分析控制台。
* **底层原理:** 调用 `/api/stats` 聚合消息量历史记录、追踪活跃群组、通过 Chart.js 显示关键词量计数，并绘制最活跃群组的图表。
### 💬 2. 消息
* **它是什么:** 显示所有摄取消息的细粒度搜索引擎电子表格。
* **底层原理:** 将用户查询连接到 SQL 搜索操作。行颜色指示状态：绿色 = 关键词命中，金色 = 跨群组转发，深琥珀色 = 两者皆有。点击某一行会滑出一张详细的分析师卡片，展示 PyTesseract OCR 文本、解码的二维码、提取的实体以及相似的视觉匹配（Hamming Distance 检查）。
### 🕐 3. 时间范围
* **它是什么:** 专注于隔离狭窄操作时间间隔的搜索工作区。
* **底层原理:** 将自然日期输入转换为 UTC 时间戳，执行 SQLite 索引范围扫描。提供一键提取导出功能。
### 👥 4. Groups
* **它是什么:** 控制监控目标群组的注册控制台。
* **底层原理:** 列出所有已同步的对话。实时切换会触发数据库更改并立即修改 `ActiveGroupCache` 列表，无需重启 daemon。
### 🔑 5. Keywords
* **它是什么:** 活动关键词规则管理界面。
* **底层原理:** 显示关键词的标签云。修改内容会被写入 SQLite，在 60秒内触发匹配器更新，并对历史消息执行后台追溯匹配。
### 📥 6. 导出
* **它是什么:** 用于提取数据以生成下游报告的文件编译器。
* **底层原理:** 将筛选出的数据打包成带样式的 Excel 文件 (`openpyxl`)，高亮显示威胁行，创建 Legend 工作表，并汇总指标。
### 📢 7. Campaigns
* **它是什么:** 协调垃圾邮件或诈骗活动的聚类分组。
* **底层原理:** 使用 embeddings 的余弦相似度（$\ge 0.85$）和通过 pHash 的图像视觉相似度（$\le 10$）将近似重复项进行分组。
### 🕸️ 8. Network Intel
* **它是什么:** 转发拓扑网络可视化器和发布节奏图表。
* **底层原理:** 构建频道转发的有向图。通过 NetworkX 计算 PageRank，并将元素提供给 Cytoscape.js。渲染发布频率直方图。
### 👤 9. Actor Profiles
* **它是什么:** 发送者风险追踪仪表盘。
* **底层原理:** 将消息计数和风险聚合到 Low, Medium, High 和 Critical 风险等级中。显示时间活动直方图以推断行为者的时区。
### 📁 10. Case Files
* **它是什么:** 证据编译工作区。
* **底层原理:** 将目标元素（消息、钱包、行为者用户名）链接到案例 ID。生成格式化的高管情报简报 Markdown 报告。
### 🩺 11. Pipeline Health
* **它是什么:** 系统健康状态监视器。
* **底层原理:** 显示断开连接指标、每日停机时长以及来自 `pipeline_events` 日志的详细信息。
* **摄取控制中心:** 用于注册、启动、停止或删除最多 5 个并发 Telegram session 的分析师界面。支持 SMS OTP 连接。
## 🛠️ 操作手册与运维指南
### 手册 A: 设置新的 Telegram Session
1. 导航到 **Pipeline Health** -> **Ingestion Control Center**。
2. 输入备用账户详情：电话（国际格式：例如 `+919876543210`）、API ID 和 API Hash。点击 **Add Account**。
3. Telethon 将请求登录码。状态更改为 `needs_otp`。
4. 在仪表盘上找到动态输入表单。输入收到的 Telegram OTP 验证码，然后点击 **Verify OTP**。
5. session 将被缓存为 `data/telegram_session_phone.session` 并开始监控。
### 手册 B: 关键词热重载与追溯补全匹配
* **添加关键词:** Dashboard -> **Keywords** -> 输入单词并按回车键（或通过 Streamlit 上传 `.pptx`）。
* **自动重载:** 摄取引擎每 60 秒检测一次数据库新增内容。
* **追溯匹配:** 匹配器会生成一个后台线程，查询最多 1,000 条未匹配的消息，根据新规则对它们进行追溯匹配。
### 手册 C: 数据库备份与安全复制
由于 SQLite 以 **WAL 模式** 运行，读取连接不会阻塞写入线程。在实时执行期间直接复制数据库文件是安全的：
```
copy data\telegram_intel.db data\telegram_intel_backup_%date:~-4,4%%date:~-10,2%%date:~-7,2%.db
```
### 手册 D: 灾难恢复（原始备份导入）
如果发生数据库损坏：
1. 通过 `Ctrl+C` 停止应用程序。
2. 删除数据库文件：`del data\telegram_intel.db`。
3. 启动应用程序：`python main.py` (创建 schema)。
4. 运行备份恢复实用程序，从原始每日日志文件中导入记录：
python -c "
import json, os, sqlite3
conn = sqlite3.connect('data/telegram_intel.db')
# 读取备份的 JSONL 文件并插入消息 ...
"
## 🧪 单元测试验证
TeleWire 包含验证所有功能的单元测试。使用 bootstrap 包装器或 python discover 运行测试：
```
python -m unittest discover -s tests
```
*预期输出显示所有测试均成功通过。*
标签：Flask, Kubernetes, Python, SOCMINT, 威胁情报, 实体提取, 开发者工具, 数据泄露, 无后门, 特权检测, 计算机取证, 逆向工具