tg-hub

tg-hub

改造来源：jackwener/tg-cli（Apache-2.0）

本技能在原仓库基础上做了以下简化：

• 移除 click / rich / python-dotenv / pyyaml 依赖
• 仅保留 telethon 一个第三方依赖
• 移除 CLI 层，所有功能封装为同步 Python API
• 默认 session/db 路径改为 /var/minis/workspace/tg-hub/
• 配置改为直接读取环境变量，无需 .env 文件

---

架构特点：本地优先（Local-First）

Telegram MTProto（telethon）
    ↓  sync / refresh（增量）
本地 SQLite  ~/.tg-hub/messages.db
    ↓  search / today / recent / filter（离线）
结构化数据

• 读操作（search/today/recent）：查本地 SQLite，不联网，毫秒级响应
• 写操作（sync/refresh）：连接 Telegram 拉取新消息，增量写入 SQLite
• Session 文件：~/.tg-hub/tg_hub.session

---

文件结构

/var/minis/skills/tg-hub/
├── SKILL.md
├── pyproject.toml          # 仅 telethon
└── scripts/
    ├── __init__.py
    ├── config.py           # 配置（环境变量 / 默认路径）
    ├── db.py               # SQLite 消息存储
    ├── exceptions.py       # 结构化异常
    └── client.py           # TGClient 核心类（全部 API）

---

首次登录（必须在 Terminal 中操作）

tg-hub 使用 MTProto 协议（非 Bot API），需要用你的 Telegram 账号登录。

建议：优先使用你自己的 TG_API_ID / TG_API_HASH。 我已按上游 tg-cli 的反风控实现同步：使用 Telegram Desktop 5.x 指纹，并在继续使用默认 api_id=2040 时给出 warning。公共 APP ID 仅作兜底，长期仍建议使用自有凭证。

1. 打开 Terminal
2. （推荐）先设置自己的 TG_API_ID / TG_API_HASH
3. cd /var/minis/skills/tg-hub
4. uv run python -c "
   import sys; sys.path.insert(0,'.')
   from scripts.client import TGClient
   me = TGClient().login()
   print('登录成功：', me)
   "
5. 按提示输入手机号（+86XXXXXXXXXX 格式）
6. 输入 Telegram App 收到的验证码
7. 登录成功后 session 自动保存，后续免登录

如暂时没有自己的凭证，可先用内置公共凭证登录；如遇登录/拉取异常，优先切换为自己的 APP ID。

打开 Terminal 登录

---

快速使用

环境准备

cd /var/minis/skills/tg-hub
uv sync

Python 调用

import sys
sys.path.insert(0, "/var/minis/skills/tg-hub")
from scripts.client import TGClient

client = TGClient()

# 查看当前账号
me = client.whoami()
print(me["name"], me["phone"])

# 列出所有对话（实时从 TG 获取）
chats = client.list_chats()
for c in chats[:10]:
    print(f"  [{c['type']}] {c['name']}  未读: {c['unread']}")

# 增量同步单个群
n = client.sync("群名或用户名", limit=1000)
print(f"新增 {n} 条消息")

# 快速刷新所有群（每群最多 500 条新消息）
# 默认带轻微节流；也可以限制本轮只刷前 30 个 chat
result = client.refresh(delay=1.0, max_chats=30)
for name, count in result.items():
    if count > 0:
        print(f"  {name}: +{count}")

# 搜索关键词
msgs = client.search("Python", hours=48)
for m in msgs:
    print(f"[{m['chat_name']}] {m['sender_name']}: {m['content'][:80]}")

# 多关键词过滤（OR 逻辑）
msgs = client.filter("招聘,remote,兼职", hours=24)

# 今日消息
msgs = client.today()

# 最近 12 小时消息
msgs = client.recent(hours=12, limit=200)

# 发言排行
top = client.top_senders(hours=24)
for t in top[:5]:
    print(f"  {t['sender_name']}: {t['msg_count']} 条")

# 时间线统计
tl = client.timeline(granularity="hour", hours=48)

# 本地数据库统计
stats = client.stats()
print(f"本地共 {stats['total']} 条消息，{len(stats['chats'])} 个群")

---

API 速查

认证

同步（联网）

查询（本地，不联网）

---

环境变量

---

账号安全建议

1. 优先使用自己的 API 凭证：前往 https://my.telegram.org 创建应用后设置 TG_API_ID / TG_API_HASH。
2. 控制同步频率：避免高频反复执行 refresh()。
3. 使用 delay 和 max_chats：建议日常增量刷新时限制每轮同步数量，并保留 chat 之间的间隔。
4. 首次全量同步不要太激进：tg-hub 已对首次同步 chat 自动限制更低的抓取量。
5. 优先读操作：搜索/统计等本地查询不联网，风险远低于频繁同步。

---

注意事项

• 首次登录必须在交互式 terminal 中完成（需要输入验证码）
• 强烈建议优先使用自己的 TG_API_ID / TG_API_HASH，避免公共 APP ID 被滥用带来的风控问题
• tg-hub 已对齐上游 tg-cli 的 Telegram Desktop 5.x 客户端指纹，并保留环境变量覆盖能力，用于降低异常指纹风险
• 若仍使用默认 api_id=2040，连接时会打印 warning，提醒改用自己的 TG_API_ID / TG_API_HASH
• Session 文件保存在 /var/minis/workspace/tg-hub/tg_hub.session，妥善保管
• sync_all 首次运行时间较长（取决于群数量和历史消息量）
• 建议用 refresh() 做日常增量更新，用 sync(chat, limit=10000) 做首次全量同步
• Telegram 对 API 请求有频率限制，大量同步时 telethon 会自动处理 flood wait