| 6 |
rl-api-marker-1792550584 |
pending |
—
|
2026-10-21 02:43:04
|
| 5 |
[FIX] #3 这个机器人关键信息没有配置，直接复制一个模板给我发来，我要给配置 |
completed |
# 分析问题
这是一个关于 Telegram 机器人配置模板的需求。需要创建一个完整的机器人配置模板文件，供用户填写关键信息。
---
以下是完整的机器人配置模板：
**`.env.example`:**
```env
# ============================================================
# 🤖 Telegram Bot 核心配置
# ============================================================
# 从 @BotFather 获取的 Bot Token
# 格式: 123456789:ABCDefGhIJKlmNoPQRsTUVwxyZ
BOT_TOKEN=your_bot_token_here
# Bot 用户名 (不含 @)
BOT_USERNAME=your_bot_username
# Bot 显示名称
BOT_NAME=My Telegram Bot
# ============================================================
# 👤 管理员配置
# ============================================================
# 管理员 Telegram User ID (多个用逗号分隔)
# 获取方式: 发消息给 @userinfobot
ADMIN_USER_IDS=123456789,987654321
# 管理员用户名 (不含 @，多个用逗号分隔)
ADMIN_USERNAMES=admin1,admin2
# ============================================================
# 🗄️ 数据库配置
# ============================================================
# 数据库类型: sqlite / mysql / postgresql
DB_TYPE=sqlite
# SQLite 配置 (DB_TYPE=sqlite 时使用)
SQLITE_PATH=./data/bot.db
# MySQL / PostgreSQL 配置
DB_HOST=localhost
DB_PORT=5432
DB_NAME=telegram_bot
DB_USER=your_db_user
DB_PASSWORD=your_db_password
# ============================================================
# 🔗 Webhook 配置 (可选，不填则使用轮询模式)
# ============================================================
# Webhook 模式: true / false
USE_WEBHOOK=false
# 你的服务器公网域名 (HTTPS)
WEBHOOK_URL=https://your-domain.com
# Webhook 路径
WEBHOOK_PATH=/webhook
# Webhook 监听端口
WEBHOOK_PORT=8443
# SSL 证书路径 (如需自签名证书)
SSL_CERT=./ssl/cert.pem
SSL_KEY=./ssl/private.key
# ============================================================
# 🔑 第三方 API 配置
# ============================================================
# OpenAI (如需 AI 功能)
OPENAI_API_KEY=sk-your_openai_key_here
OPENAI_MODEL=gpt-4o
# 支付 (如需支付功能)
PAYMENT_PROVIDER_TOKEN=your_payment_token
# ============================================================
# 📢 频道 / 群组配置
# ============================================================
# 主频道 ID (以 -100 开头)
MAIN_CHANNEL_ID=-1001234567890
# 主群组 ID (以 -100 开头)
MAIN_GROUP_ID=-1009876543210
# 日志频道 ID
LOG_CHANNEL_ID=-1001111111111
# ============================================================
# ⚙️ 运行参数配置
# ============================================================
# 运行环境: development / production
ENV=production
# 日志级别: DEBUG / INFO / WARNING / ERROR
LOG_LEVEL=INFO
# 日志文件路径
LOG_FILE=./logs/bot.log
# 请求超时时间 (秒)
REQUEST_TIMEOUT=30
# 最大并发连接数
MAX_CONNECTIONS=100
# ============================================================
# 🚦 限速配置
# ============================================================
# 每用户每分钟最大消息数
RATE_LIMIT_PER_USER=30
# 全局每秒最大消息数
RATE_LIMIT_GLOBAL=25
# ============================================================
# 🔒 安全配置
# ============================================================
# 密钥 (用于加密敏感数据，随机字符串)
SECRET_KEY=your_random_secret_key_here
# 是否允许私聊: true / false
ALLOW_PRIVATE_CHAT=true
# 是否允许群组: true / false
ALLOW_GROUP_CHAT=true
# 白名单用户 ID (为空则不限制，多个逗号分隔)
WHITELIST_USER_IDS=
# 黑名单用户 ID (多个逗号分隔)
BLACKLIST_USER_IDS=
# ============================================================
# 📦 Redis 缓存配置 (可选)
# ============================================================
# 是否启用 Redis
USE_REDIS=false
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_PASSWORD=
REDIS_DB=0
# ============================================================
# 📬 通知配置
# ============================================================
# 启动时发送通知给管理员: true / false
NOTIFY_ON_START=true
# 报错时发送通知给管理员: true / false
NOTIFY_ON_ERROR=true
```
**`config.py`:**
```python
"""
配置加载模块 - 读取 .env 文件中的所有配置项
使用方式: from config import config
"""
import os
from dataclasses import dataclass, field
from typing import List, Optional
from dotenv import load_dotenv
load_dotenv()
def _get_list(key: str, default: str = "") -> List[str]:
"""读取逗号分隔的列表配置"""
val = os.getenv(key, default).strip()
return [x.strip() for x in val.split(",") if x.strip()] if val else []
def _get_int(key: str, default: int = 0) -> int:
try:
return int(os.getenv(key, str(default)))
except ValueError:
return default
def _get_bool(key: str, default: bool = False) -> bool:
return os.getenv(key, str(default)).lower() in ("true", "1", "yes")
# ============================================================
# 配置数据类
# ============================================================
@dataclass
class BotConfig:
token: str
username: str
name: str
@dataclass
class AdminConfig:
user_ids: List[int]
usernames: List[str]
@dataclass
class DatabaseConfig:
db_type: str
sqlite_path: str
host: str
port: int
name: str
user: str
password: str
@property
def url(self) -> str:
if self.db_type == "sqlite":
return f"sqlite:///{self.sqlite_path}"
elif self.db_type == "mysql":
return f"mysql+aiomysql://{self.user}:{self.password}@{self.host}:{self.port}/{self.name}"
elif self.db_type == "postgresql":
return f"postgresql+asyncpg://{self.user}:{self.password}@{self.host}:{self.port}/{self.name}"
raise ValueError(f"不支持的数据库类型: {self.db_type}")
@dataclass
class WebhookConfig:
enabled: bool
url: str
path: str
port: int
ssl_cert: Optional[str]
ssl_key: Optional[str]
@dataclass
class ApiConfig:
openai_api_key: Optional[str]
openai_model: str
payment_provider_token: Optional[str]
@dataclass
class ChannelConfig:
main_channel_id: Optional[int]
main_group_id: Optional[int]
log_channel_id: Optional[int]
@dataclass
class RuntimeConfig:
env: str
log_level: str
log_file: str
request_timeout: int
max_connections: int
@property
def is_production(self) -> bool:
return self.env == "production"
@property
def is_development(self) -> bool:
return self.env == "development"
@dataclass
class RateLimitConfig:
per_user: int
global_limit: int
@dataclass
class SecurityConfig:
secret_key: str
allow_private_chat: bool
allow_group_chat: bool
whitelist_user_ids: List[int]
blacklist_user_ids: List[int]
@dataclass
class RedisConfig:
enabled: bool
host: str
port: int
password: Optional[str]
db: int
@property
def url(self) -> str:
auth = f":{self.password}@" if self.password else ""
return f"redis://{auth}{self.host}:{self.port}/{self.db}"
@dataclass
class NotifyConfig:
on_start: bool
on_error: bool
@dataclass
class AppConfig:
bot: BotConfig
admin: AdminConfig
database: DatabaseConfig
webhook: WebhookConfig
api: ApiConfig
channel: ChannelConfig
runtime: RuntimeConfig
rate_limit: RateLimitConfig
security: SecurityConfig
redis: RedisConfig
notify: NotifyConfig
def validate(self) -> None:
"""校验必填项"""
errors = []
if not self.bot.token or self.bot.token == "your_bot_token_here":
errors.append("❌ BOT_TOKEN 未配置")
if not self.admin.user_ids:
errors.append("❌ ADMIN_USER_IDS 未配置")
if self.security.secret_key == "your_random_secret_key_here":
errors.append("❌ SECRET_KEY 未修改，请设置随机密钥")
if errors:
raise ValueError("\n".join(["配置校验失败:"] + errors))
print("✅ 配置校验通过")
# ============================================================
# 加载配置
# ============================================================
def load_config() -> AppConfig:
return AppConfig(
bot=BotConfig(
token=os.getenv("BOT_TOKEN", ""),
username=os.getenv("BOT_USERNAME", ""),
name=os.getenv("BOT_NAME", "Bot"),
),
admin=AdminConfig(
user_ids=[int(x) for x in _get_list("ADMIN_USER_IDS") if x.isdigit()],
usernames=_get_list("ADMIN_USERNAMES"),
),
database=DatabaseConfig(
db_type=os.getenv("DB_TYPE", "sqlite"),
sqlite_path=os.getenv("SQLITE_PATH", "./data/bot.db"),
host=os.getenv("DB_HOST", "localhost"),
port=_get_int("DB_PORT", 5432),
name=os.getenv("DB_NAME", "telegram_bot"),
user=os.getenv("DB_USER", ""),
password=os.getenv("DB_PASSWORD", ""),
),
webhook=WebhookConfig(
enabled=_get_bool("USE_WEBHOOK", False),
url=os.getenv("WEBHOOK_URL", ""),
path=os.getenv("WEBHOOK_PATH", "/webhook"),
port=_get_int("WEBHOOK_PORT", 8443),
ssl_cert=os.getenv("SSL_CERT"),
ssl_key=os.getenv("SSL_KEY"),
),
api=ApiConfig(
openai_api_key=os.getenv("OPENAI_API_KEY"),
openai_model=os.getenv("OPENAI_MODEL", "gpt-4o"),
payment_provider_token=os.getenv("PAYMENT_PROVIDER_TOKEN"),
),
channel=ChannelConfig(
main_channel_id=_get_int("MAIN_CHANNEL_ID") or None,
main_group_id=_get_int("MAIN_GROUP_ID") or None,
log_channel_id=_get_int("LOG_CHANNEL_ID") or None,
),
runtime=RuntimeConfig(
env=os.getenv("ENV", "production"),
log_level=os.getenv("LOG_LEVEL", "INFO"),
log_file=os.getenv("LOG_FILE", "./logs/bot.log"),
request_timeout=_get_int("REQUEST_TIMEOUT", 30),
max_connections=_get_int("MAX_CONNECTIONS", 100),
),
rate_limit=RateLimitConfig(
per_user=_get_int("RATE_LIMIT_PER_USER", 30),
global_limit=_get_int("RATE_LIMIT_GLOBAL", 25),
),
security=SecurityConfig(
secret_key=os.getenv("SECRET_KEY", "your_random_secret_key_here"),
allow_private_chat=_get_bool("ALLOW_PRIVATE_CHAT", True),
allow_group_chat=_get_bool("ALLOW_GROUP_CHAT", True),
whitelist_user_ids=[int(x) for x in _get_list("WHITELIST_USER_IDS") if x.isdigit()],
blacklist_user_ids=[int(x) for x in _get_list("BLACKLIST_USER_IDS") if x.isdigit()],
),
redis=RedisConfig(
enabled=_get_bool("USE_REDIS", False),
host=os.getenv("REDIS_HOST", "localhost"),
port=_get_int("REDIS_PORT", 6379),
password=os.getenv("REDIS_PASSWORD") or None,
db=_get_int("REDIS_DB", 0),
),
notify=NotifyConfig(
on_start=_get_bool("NOTIFY_ON_START", True),
on_error=_get_bool("NOTIFY_ON_ERROR", True),
),
)
# 全局单例
config = load_config()
```
**`requirements.txt`:**
```
python-telegram-bot>=20.7
python-dotenv>=1.0.0
sqlalchemy>=2.0.0
aiofiles>=23.2.1
redis>=5.0.1
openai>=1.0.0
```
**`配置说明.md`:**
```markdown
# 🤖 Bot 配置说明
## 快速开始
### 第一步：复制配置文件
```bash
cp .env.example .env
```
### 第二步：填写必填项（⚠️ 必须填写，否则无法启动）
| 配置项 | 说明 | 获取方式 |
|--------|------|----------|
| `BOT_TOKEN` | Bot 令牌 | 找 [@BotFather](https://t.me/BotFather) 创建 Bot |
| `ADMIN_USER_IDS` | 管理员 ID | 找 [@userinfobot](https://t.me/userinfobot) 获取你的 ID |
| `SECRET_KEY` | 加密密钥 | 随机生成，见下方命令 |
### 第三步：生成随机密钥
```bash
python -c "import secrets; print(secrets.token_hex(32))"
```
---
## 配置项详解
### 🤖 Bot 核心
- **BOT_TOKEN** ⚠️ 必填 — Bot 的身份令牌
- **BOT_USERNAME** — Bot 的用户名（不含 @）
- **BOT_NAME** — Bot 显示名称
### 👤 管理员
- **ADMIN_USER_IDS** ⚠️ 必填
|
2026-05-10 16:02:17
|
| 4 |
[FIX] #3 这搭建的是一个机器人，但是我在哪里设置token这些 |
completed |
# 分析问题
这是一个关于 Telegram 机器人 Token 配置的问题。我需要先了解项目结构，然后提供完整的配置方案。
---
## 完整解决方案
**`config.py`:**
```python
import os
from dotenv import load_dotenv
# 加载 .env 文件
load_dotenv()
class Config:
# ========================================
# Telegram Bot 配置
# ========================================
BOT_TOKEN = os.getenv("BOT_TOKEN", "")
# ========================================
# 其他可选配置
# ========================================
ADMIN_USER_ID = os.getenv("ADMIN_USER_ID", "") # 管理员 Telegram 用户 ID
WEBHOOK_URL = os.getenv("WEBHOOK_URL", "") # Webhook 地址（可选）
@classmethod
def validate(cls):
"""启动时验证必要配置"""
if not cls.BOT_TOKEN:
raise ValueError(
"❌ BOT_TOKEN 未设置！\n"
"请在 .env 文件中设置 BOT_TOKEN=你的Token\n"
"获取方式：去 Telegram 找 @BotFather 创建机器人"
)
print("✅ 配置验证通过")
print(f" BOT_TOKEN: {cls.BOT_TOKEN[:10]}***{cls.BOT_TOKEN[-5:]}")
if cls.ADMIN_USER_ID:
print(f" ADMIN_USER_ID: {cls.ADMIN_USER_ID}")
config = Config()
```
**`.env`:**
```ini
# ================================================
# Telegram Bot 配置文件
# ================================================
# 📌 如何获取 BOT_TOKEN：
# 1. 打开 Telegram，搜索 @BotFather
# 2. 发送 /newbot
# 3. 按提示输入机器人名称和用户名
# 4. BotFather 会返回一个 Token，复制粘贴到下面
#
# Token 格式示例：123456789:ABCdefGHIjklMNOpqrSTUvwxYZ
# ================================================
BOT_TOKEN=在这里填写你的Bot Token
# ================================================
# 管理员配置（可选）
# ================================================
# 📌 如何获取你的 Telegram User ID：
# 1. Telegram 搜索 @userinfobot
# 2. 发送任意消息，它会返回你的 User ID
# ================================================
ADMIN_USER_ID=在这里填写管理员的UserID
# ================================================
# Webhook 配置（可选，用于服务器部署）
# ================================================
# 如果你是本地运行，不需要填写
# 如果部署到服务器，填写你的服务器地址
# 格式：https://你的域名/webhook
WEBHOOK_URL=
```
**`.env.example`:**
```ini
# 复制这个文件为 .env，然后填写真实的值
BOT_TOKEN=123456789:ABCdefGHIjklMNOpqrSTUvwxYZ
ADMIN_USER_ID=123456789
WEBHOOK_URL=https://your-domain.com/webhook
```
**`bot.py`:**
```python
import logging
from telegram import Update
from telegram.ext import (
Application,
CommandHandler,
MessageHandler,
filters,
ContextTypes
)
from config import config
# ========================================
# 日志配置
# ========================================
logging.basicConfig(
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
level=logging.INFO
)
logger = logging.getLogger(__name__)
# ========================================
# 命令处理器
# ========================================
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""处理 /start 命令"""
user = update.effective_user
await update.message.reply_text(
f"👋 你好，{user.first_name}！\n"
f"机器人已成功启动！\n\n"
f"发送任意消息试试看 😊"
)
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""处理 /help 命令"""
await update.message.reply_text(
"📖 帮助信息：\n\n"
"/start - 开始使用\n"
"/help - 显示帮助\n"
)
async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE):
"""回显用户消息"""
await update.message.reply_text(
f"你说了：{update.message.text}"
)
# ========================================
# 主函数
# ========================================
def main():
# 1. 验证配置
config.validate()
# 2. 创建 Bot 应用
app = Application.builder().token(config.BOT_TOKEN).build()
# 3. 注册处理器
app.add_handler(CommandHandler("start", start))
app.add_handler(CommandHandler("help", help_command))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo))
# 4. 启动
print("\n🤖 机器人启动中...")
print("按 Ctrl+C 停止\n")
app.run_polling(allowed_updates=Update.ALL_TYPES)
if __name__ == "__main__":
main()
```
**`requirements.txt`:**
```
python-telegram-bot==20.7
python-dotenv==1.0.0
```
**`.gitignore`:**
```gitignore
# 🔒 重要：.env 文件包含敏感信息，绝对不要上传到 Git！
.env
# Python
__pycache__/
*.py[cod]
*$py.class
*.egg-info/
dist/
build/
.eggs/
# 虚拟环境
venv/
env/
.venv/
# IDE
.idea/
.vscode/
*.swp
```
**`README.md`:**
```markdown
# Telegram Bot 配置说明
## 🚀 快速开始
### 第一步：获取 Bot Token
1. 打开 Telegram，搜索 **@BotFather**
2. 点击 `/start`
3. 发送 `/newbot`
4. 输入机器人的**显示名称**（例如：`My Awesome Bot`）
5. 输入机器人的**用户名**，必须以 `bot` 结尾（例如：`my_awesome_bot`）
6. BotFather 会返回你的 **Token**，格式如下：
```
110201543:AAHdqTcvCH1vGWJxfSeofSAs0K5PALDsaw
```
### 第二步：配置 Token
```bash
# 复制 .env.example 文件
cp .env.example .env
# 编辑 .env 文件，填写你的 Token
nano .env # Linux/Mac
notepad .env # Windows
```
`.env` 文件内容：
```ini
BOT_TOKEN=你从BotFather获得的Token
```
### 第三步：安装依赖并运行
```bash
# 安装依赖
pip install -r requirements.txt
# 运行机器人
python bot.py
```
## ✅ 成功标志
看到以下输出说明配置成功：
```
✅ 配置验证通过
BOT_TOKEN: 110201543***dsaw
🤖 机器人启动中...
```
## ❌ 常见错误
| 错误 | 原因 | 解决方法 |
|------|------|----------|
| `BOT_TOKEN 未设置` | .env 文件不存在或 Token 为空 | 检查 .env 文件 |
| `Unauthorized` | Token 错误 | 重新从 BotFather 复制 Token |
| `ModuleNotFoundError` | 依赖未安装 | 运行 `pip install -r requirements.txt` |
## 🔒 安全提示
> ⚠️ **绝对不要把 Token 上传到 GitHub 或分享给他人！**
>
> Token 相当于机器人的密码，泄露后别人可以完全控制你的机器人。
>
> `.gitignore` 已配置忽略 `.env` 文件，请勿手动添加到 git 追踪。
```
---
## 📋 操作步骤总结
```
1. 找 @BotFather 创建机器人 → 获取 Token
↓
2. 复制 .env.example → 重命名为 .env
↓
3. 打开 .env，把 Token 填入 BOT_TOKEN=
↓
4. pip install -r requirements.txt
↓
5. python bot.py → 看到 ✅ 即成功！
```
## Summary
Token 应该配置在项目根目录的 **`.env`** 文件中，通过 `BOT_TOKEN=你的Token` 格式填写，程序通过 `python-dotenv` 库自动读取。Token 从 Telegram 的 **@BotFather** 机器人获取，切记 `.env` 文件不能上传到 Git（已在 `.gitignore` 中排除），否则会造成 Token 泄露安全风险。
|
2026-05-10 15:09:51
|
| 3 |
好的，我们开始。以下是为您生成的详细项目开发任务描述。 |
completed |
# TG USDT 地址监控机器人 - 完整实现
## 项目结构
```
tg_usdt_monitor/
├── .env
├── requirements.txt
├── main.py
├── bot/
│ ├── __init__.py
│ ├── handlers.py
│ ├── monitor.py
│ └── keyboards.py
├── db/
│ ├── __init__.py
│ └── database.py
├── services/
│ ├── __init__.py
│ ├── blockchain.py
│ └── validator.py
└── utils/
├── __init__.py
└── helpers.py
```
---
## 配置文件
### `.env`
```env
# Telegram Bot Token (从 @BotFather 获取)
BOT_TOKEN=your_telegram_bot_token_here
# TronGrid API Key (https://www.trongrid.io/)
TRONGRID_API_KEY=your_trongrid_api_key_here
# BSCScan API Key (https://bscscan.com/apis)
BSCSCAN_API_KEY=your_bscscan_api_key_here
# 监控轮询间隔（秒）
MONITOR_INTERVAL=60
# 每次查询获取的最大交易数
MAX_TRANSACTIONS=50
```
### `requirements.txt`
```txt
python-telegram-bot==20.7
httpx==0.26.0
python-dotenv==1.0.0
aiosqlite==0.19.0
```
---
## 数据库层
### `db/database.py`
```python
import aiosqlite
import logging
from datetime import datetime
from typing import Optional
logger = logging.getLogger(__name__)
DB_PATH = "monitor.db"
async def init_db():
"""初始化数据库，创建所需表"""
async with aiosqlite.connect(DB_PATH) as db:
# 监控列表表：存储 chat_id 与 address 的对应关系
await db.execute("""
CREATE TABLE IF NOT EXISTS monitor_list (
id INTEGER PRIMARY KEY AUTOINCREMENT,
chat_id INTEGER NOT NULL,
address TEXT NOT NULL,
chain TEXT NOT NULL, -- 'TRC20' 或 'BEP20'
label TEXT, -- 用户自定义标签
added_at TEXT NOT NULL,
UNIQUE(chat_id, address)
)
""")
# 地址状态表：记录每个地址最后一次检查的交易哈希
await db.execute("""
CREATE TABLE IF NOT EXISTS address_state (
address TEXT PRIMARY KEY,
chain TEXT NOT NULL,
last_tx_hash TEXT, -- 最新的交易哈希
last_checked_at TEXT,
total_received REAL DEFAULT 0
)
""")
# 交易记录表：避免重复推送
await db.execute("""
CREATE TABLE IF NOT EXISTS sent_transactions (
tx_hash TEXT PRIMARY KEY,
address TEXT NOT NULL,
pushed_at TEXT NOT NULL
)
""")
await db.commit()
logger.info("数据库初始化完成")
async def add_monitor(chat_id: int, address: str, chain: str, label: Optional[str] = None) -> bool:
"""
添加监控地址
Returns: True 表示新增成功, False 表示已存在
"""
async with aiosqlite.connect(DB_PATH) as db:
try:
await db.execute(
"""
INSERT INTO monitor_list (chat_id, address, chain, label, added_at)
VALUES (?, ?, ?, ?, ?)
""",
(chat_id, address, chain, label, datetime.utcnow().isoformat())
)
await db.commit()
# 若地址状态表中不存在该地址，则初始化
await db.execute(
"""
INSERT OR IGNORE INTO address_state (address, chain, last_checked_at)
VALUES (?, ?, ?)
""",
(address, chain, datetime.utcnow().isoformat())
)
await db.commit()
return True
except aiosqlite.IntegrityError:
return False # 已存在该监控关系
async def remove_monitor(chat_id: int, address: str) -> bool:
"""
移除监控地址
Returns: True 表示删除成功, False 表示记录不存在
"""
async with aiosqlite.connect(DB_PATH) as db:
cursor = await db.execute(
"DELETE FROM monitor_list WHERE chat_id = ? AND address = ?",
(chat_id, address)
)
await db.commit()
return cursor.rowcount > 0
async def list_monitors(chat_id: int) -> list[dict]:
"""获取某个 chat_id 的所有监控地址"""
async with aiosqlite.connect(DB_PATH) as db:
db.row_factory = aiosqlite.Row
cursor = await db.execute(
"""
SELECT ml.address, ml.chain, ml.label, ml.added_at,
ads.last_tx_hash, ads.last_checked_at, ads.total_received
FROM monitor_list ml
LEFT JOIN address_state ads ON ml.address = ads.address
WHERE ml.chat_id = ?
ORDER BY ml.added_at DESC
""",
(chat_id,)
)
rows = await cursor.fetchall()
return [dict(row) for row in rows]
async def get_all_monitored_addresses() -> list[dict]:
"""获取所有被监控的唯一地址（供监控任务使用）"""
async with aiosqlite.connect(DB_PATH) as db:
db.row_factory = aiosqlite.Row
cursor = await db.execute(
"""
SELECT DISTINCT address, chain
FROM monitor_list
"""
)
rows = await cursor.fetchall()
return [dict(row) for row in rows]
async def get_subscribers(address: str) -> list[int]:
"""获取订阅了某个地址的所有 chat_id"""
async with aiosqlite.connect(DB_PATH) as db:
cursor = await db.execute(
"SELECT chat_id FROM monitor_list WHERE address = ?",
(address,)
)
rows = await cursor.fetchall()
return [row[0] for row in rows]
async def get_address_state(address: str) -> Optional[dict]:
"""获取地址状态"""
async with aiosqlite.connect(DB_PATH) as db:
db.row_factory = aiosqlite.Row
cursor = await db.execute(
"SELECT * FROM address_state WHERE address = ?",
(address,)
)
row = await cursor.fetchone()
return dict(row) if row else None
async def update_address_state(address: str, last_tx_hash: str, total_received: float):
"""更新地址状态"""
async with aiosqlite.connect(DB_PATH) as db:
await db.execute(
"""
UPDATE address_state
SET last_tx_hash = ?, last_checked_at = ?, total_received = ?
WHERE address = ?
""",
(last_tx_hash, datetime.utcnow().isoformat(), total_received, address)
)
await db.commit()
async def is_tx_sent(tx_hash: str) -> bool:
"""检查某交易是否已推送过"""
async with aiosqlite.connect(DB_PATH) as db:
cursor = await db.execute(
"SELECT 1 FROM sent_transactions WHERE tx_hash = ?",
(tx_hash,)
)
return await cursor.fetchone() is not None
async def mark_tx_sent(tx_hash: str, address: str):
"""标记交易已推送"""
async with aiosqlite.connect(DB_PATH) as db:
await db.execute(
"""
INSERT OR IGNORE INTO sent_transactions (tx_hash, address, pushed_at)
VALUES (?, ?, ?)
""",
(tx_hash, address, datetime.utcnow().isoformat())
)
await db.commit()
async def count_monitors(chat_id: int) -> int:
"""统计某 chat_id 的监控数量"""
async with aiosqlite.connect(DB_PATH) as db:
cursor = await db.execute(
"SELECT COUNT(*) FROM monitor_list WHERE chat_id = ?",
(chat_id,)
)
row = await cursor.fetchone()
return row[0]
```
---
## 地址验证服务
### `services/validator.py`
```python
import re
from typing import Tuple
# TRC20 地址：T 开头，34位 Base58
TRC20_PATTERN = re.compile(r'^T[A-Za-z1-9]{33}$')
# BEP20 地址：0x 开头，40位十六进制
BEP20_PATTERN = re.compile(r'^0x[0-9a-fA-F]{40}$')
# 单个用户最大监控数量
MAX_MONITORS_PER_USER = 10
def validate_address(address: str) -> Tuple[bool, str]:
"""
验证地址格式
Returns: (is_valid, chain_type)
chain_type: 'TRC20' | 'BEP20' | ''
"""
address = address.strip()
if TRC20_PATTERN.match(address):
return True, "TRC20"
if BEP20_PATTERN.match(address):
return True, "BEP20"
return False, ""
def format_address(address: str) -> str:
"""格式化地址显示（中间省略）"""
if len(address) <= 12:
return address
return f"{address[:6]}...{address[-6:]}"
def format_usdt_amount(amount: float) -> str:
"""格式化 USDT 金额"""
if amount >= 1_000_000:
return f"{amount / 1_000_000:.2f}M USDT"
elif amount >= 1_000:
return f"{amount / 1_000:.2f}K USDT"
else:
return f"{amount:.2f} USDT"
```
---
## 区块链查询服务
### `services/blockchain.py`
```python
import httpx
import logging
import os
from typing import Optional
logger = logging.getLogger(__name__)
TRONGRID_API_KEY = os.getenv("TRONGRID_API_KEY", "")
BSCSCAN_API_KEY = os.getenv("BSCSCAN_API_KEY", "")
# USDT 合约地址
USDT_TRC20_CONTRACT = "TR7NHqjeKQxGTCi8q8ZY4pL8otSzgjLj6t"
USDT_BEP20_CONTRACT = "0x55d398326f99059ff775485246999027b3197955"
# USDT 精度
TRC20_DECIMALS = 1_000_000 # 6位小数
BEP20_DECIMALS = 10**18 # 18位小数
async def fetch_trc20_transactions(address: str, limit: int = 50) -> list[dict]:
"""
从 TronGrid 获取 TRC20 USDT 转入交易
"""
url = f"https://api.trongrid.io/v1/accounts/{address}/transactions/trc20"
params = {
"limit": limit,
"contract_address": USDT_TRC20_CONTRACT,
"only_to": "true", # 只获取转入
"order_by": "block_timestamp,desc"
}
headers = {
"TRON-PRO-API-KEY": TRONGRID_API_KEY,
"Accept": "application/json"
}
try:
async with httpx.AsyncClient(timeout=15.0) as client:
resp = await client.get(url, params=params, headers=headers)
resp.raise_for_status()
data = resp.json()
transactions = []
for tx in data.get("data", []):
# 过滤确认的 USDT 转入交易
if (
tx.get("to") == address
and tx.get("token_info", {}).get("address") == USDT_TRC20_CONTRACT
):
amount_raw = int(tx.get("value", 0))
amount_usdt = amount_raw / TRC20_DECIMALS
transactions.append({
"tx_hash": tx.get("transaction_id", ""),
"from_address": tx.get("from", ""),
"to_address": tx.get("to", ""),
"amount": amount_usdt,
"timestamp": tx.get("block_timestamp", 0) // 1000,
"chain": "TRC20",
"explorer_url": f"https://tronscan.org/#/transaction/{tx.get('transaction_id', '')}"
})
return transactions
except httpx.HTTPStatusError as e:
logger.error(f"TronGrid HTTP 错误 {e.response.status_code}: {address}")
return []
except httpx.RequestError as e:
logger.error(f"TronGrid 请求错误: {e}")
return []
except Exception as e:
logger.error(f"TronGrid 未知错误: {e}")
return []
async def fetch_bep20_transactions(address: str, limit: int = 50) -> list[dict]:
"""
从 BSCScan 获取 BEP20 USDT 转入交易
"""
url = "https://api.bscscan.com/api"
params = {
"module": "account",
"action": "tokentx",
"contractaddress": USDT_BEP20_CONTRACT,
"address": address,
"page": 1,
"offset": limit,
"sort": "desc",
"apikey": BSCSCAN_API_KEY
}
try:
async with httpx.AsyncClient(timeout=15.0) as client:
resp = await client.get(url, params=params)
resp.raise_for_status()
data = resp.json()
if data.get("status") != "1":
if data.get("message") == "No transactions found":
return []
logger.warning(f"BSCScan API 返回: {data.get('message')} for {address}")
return []
transactions = []
for tx in data.get("result", []):
# 只取转入交易
if tx.get("to", "").lower() != address.lower():
continue
amount_raw = int(tx.get("value", 0))
amount_usdt = amount_raw / BEP20_DECIMALS
tx_hash = tx.get("hash", "")
transactions.append({
"tx_hash": tx_hash,
"from_address": tx.get("from", ""),
"to_address": tx.get("to", ""),
"amount": amount_usdt,
"
|
2026-05-10 13:49:22
|
| 2 |
测试代码执行 |
completed |
# Python 代码：当前时间 + 九九乘法表
## 代码实现
```python
import datetime
# 打印当前时间
now = datetime.datetime.now()
print("=" * 40)
print(f" 当前时间：{now.strftime('%Y年%m月%d日 %H:%M:%S')}")
print("=" * 40)
print()
# 打印九九乘法表
print("九九乘法表：")
print()
for i in range(1, 10):
for j in range(1, i + 1):
result = i * j
print(f"{j}×{i}={result:<2}", end=" ")
print()
```
## 代码说明
| 部分 | 说明 |
|------|------|
| `datetime.datetime.now()` | 获取当前本地时间 |
| `strftime(...)` | 格式化时间为中文友好格式 |
| 外层 `for i` | 控制行数（1~9） |
| 内层 `for j` | 控制每行的列数（1~i） |
| `{result:<2}` | 左对齐，宽度为2，保持对齐美观 |
## 预期输出示例
```
========================================
当前时间：2025年01月23日 14:30:05
========================================
九九乘法表：
1×1=1
1×2=2 2×2=4
1×3=3 2×3=6 3×3=9
1×4=4 2×4=8 3×4=12 4×4=16
1×5=5 2×5=10 3×5=15 4×5=20 5×5=25
1×6=6 2×6=12 3×6=18 4×6=24 5×6=30 6×6=36
1×7=7 2×7=14 3×7=21 4×7=28 5×7=35 6×7=42 7×7=49
1×8=8 2×8=16 3×8=24 4×8=32 5×8=40 6×8=48 7×8=56 8×8=64
1×9=9 2×9=18 3×9=27 4×9=36 5×9=45 6×9=54 7×9=63 8×9=72 9×9=81
```
---
## Summary
以上代码使用 `datetime` 模块获取并格式化当前时间，使用双层 `for` 循环生成下三角形式的九九乘法表。代码结构清晰、无需额外依赖，直接点击"执行"按钮即可看到实时运行结果（时间会显示为实际执行时刻）。
|
2026-05-10 12:47:50
|
| 1 |
做一个白虎群发的tg助手的说明文档 |
completed |
# 白虎TG助手 — 说明文档
> 以下为完整的文档站结构 + 独立发内容后台的实现方案，内容留白供你后台填充。
---
## 一、文档站结构设计
参考目标站结构，规划如下导航体系：
```
白虎TG助手 文档站
├── 🏠 首页（产品介绍 + 快速开始入口）
├── 📖 新手指引
│ ├── 产品简介
│ ├── 功能概览
│ └── 快速上手
├── 🚀 功能说明
│ ├── 群发消息
│ ├── 定时任务
│ ├── 关键词回复
│ ├── 成员管理
│ └── 数据统计
├── ⚙️ 配置教程
│ ├── 账号绑定
│ ├── 机器人设置
│ └── API 对接
├── 💰 套餐与计费
├── ❓ 常见问题
└── 📞 联系我们
```
---
## 二、完整代码实现
### 文件结构
```
baihu-docs/
├── index.html # 文档站前台
├── admin.html # 发内容后台
├── css/
│ └── style.css
├── js/
│ └── app.js
└── data/
└── docs.json # 文档数据存储
```
---
### `index.html` — 文档站前台
```html
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>白虎TG助手 — 官方文档</title>
<link rel="stylesheet" href="css/style.css" />
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/11.9.0/styles/github.min.css" />
<script src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/11.9.0/highlight.min.js"></script>
<!-- Markdown 渲染 -->
<script src="https://cdn.jsdelivr.net/npm/marked/marked.min.js"></script>
</head>
<body>
<!-- ===== 顶部导航 ===== -->
<header class="top-bar">
<div class="logo">
<span class="logo-icon">🐯</span>
<span class="logo-text">白虎TG助手</span>
<span class="logo-badge">文档中心</span>
</div>
<nav class="top-nav">
<a href="#" class="active">文档</a>
<a href="#">官网</a>
<a href="#">频道</a>
<a href="admin.html" class="btn-admin">管理后台</a>
</nav>
<button class="menu-toggle" id="menuToggle">☰</button>
</header>
<!-- ===== 主体布局 ===== -->
<div class="layout">
<!-- 左侧菜单 -->
<aside class="sidebar" id="sidebar">
<div class="sidebar-search">
<input type="text" id="searchInput" placeholder="🔍 搜索文档..." />
</div>
<nav class="side-nav" id="sideNav">
<!-- 由 JS 动态渲染 -->
</nav>
</aside>
<!-- 中间内容区 -->
<main class="content-area" id="contentArea">
<div class="content-inner" id="contentInner">
<!-- 默认首页 -->
<div class="home-hero">
<div class="hero-icon">🐯</div>
<h1>白虎 TG 助手</h1>
<p class="hero-sub">专业的 Telegram 群发与自动化运营工具</p>
<div class="hero-btns">
<button class="btn-primary" onclick="loadDoc('quick-start')">🚀 快速开始</button>
<button class="btn-outline" onclick="loadDoc('intro')">📖 产品介绍</button>
</div>
<div class="feature-cards">
<div class="feature-card">
<div class="fc-icon">📢</div>
<h3>批量群发</h3>
<p>支持文字、图片、视频、文件多类型内容一键群发</p>
</div>
<div class="feature-card">
<div class="fc-icon">⏰</div>
<h3>定时任务</h3>
<p>设定发送时间，自动执行，解放双手</p>
</div>
<div class="feature-card">
<div class="fc-icon">🤖</div>
<h3>智能回复</h3>
<p>关键词触发自动回复，提升运营效率</p>
</div>
<div class="feature-card">
<div class="fc-icon">📊</div>
<h3>数据统计</h3>
<p>实时查看发送数据，优化运营策略</p>
</div>
</div>
</div>
</div>
</main>
<!-- 右侧目录 -->
<aside class="toc-bar" id="tocBar">
<div class="toc-title">本页目录</div>
<ul class="toc-list" id="tocList"></ul>
</aside>
</div>
<!-- ===== 底部 ===== -->
<footer class="footer">
<span>© 2024 白虎TG助手 版权所有</span>
<span>·</span>
<a href="#">TG频道</a>
<span>·</span>
<a href="#">联系客服</a>
</footer>
<script src="js/app.js"></script>
</body>
</html>
```
---
### `admin.html` — 独立发内容后台
```html
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>白虎TG助手 — 内容管理后台</title>
<link rel="stylesheet" href="css/style.css" />
<!-- Markdown 编辑器 -->
<link rel="stylesheet" href="https://unpkg.com/easymde/dist/easymde.min.css" />
<script src="https://unpkg.com/easymde/dist/easymde.min.js"></script>
<script src="https://cdn.jsdelivr.net/npm/marked/marked.min.js"></script>
<script src="https://cdn.jsdelivr.net/npm/[email protected]/Sortable.min.js"></script>
</head>
<body class="admin-body">
<!-- 登录遮罩 -->
<div class="login-mask" id="loginMask">
<div class="login-box">
<div class="login-logo">🐯 白虎后台</div>
<h2>内容管理登录</h2>
<input type="password" id="loginPwd" placeholder="请输入管理密码" />
<button onclick="doLogin()">登 录</button>
<p class="login-tip">默认密码：baihu2024</p>
</div>
</div>
<!-- 后台主体 -->
<div class="admin-layout" id="adminLayout" style="display:none">
<!-- 左侧管理菜单 -->
<aside class="admin-sidebar">
<div class="admin-logo">🐯 白虎后台</div>
<nav>
<a href="#" class="admin-nav-item active" data-panel="doc-list">
<span>📄</span> 文档列表
</a>
<a href="#" class="admin-nav-item" data-panel="doc-edit">
<span>✏️</span> 新建/编辑
</a>
<a href="#" class="admin-nav-item" data-panel="nav-manage">
<span>🗂️</span> 菜单管理
</a>
<a href="#" class="admin-nav-item" data-panel="site-setting">
<span>⚙️</span> 站点设置
</a>
</nav>
<div class="admin-sidebar-footer">
<a href="index.html" target="_blank">👁 预览前台</a>
<button onclick="doLogout()">退出登录</button>
</div>
</aside>
<!-- 右侧面板 -->
<main class="admin-main">
<!-- ① 文档列表面板 -->
<div class="panel active" id="panel-doc-list">
<div class="panel-header">
<h2>📄 文档列表</h2>
<button class="btn-primary" onclick="switchPanel('doc-edit'); newDoc()">+ 新建文档</button>
</div>
<!-- 筛选栏 -->
<div class="filter-bar">
<input type="text" id="filterKeyword" placeholder="搜索标题..." oninput="renderDocList()" />
<select id="filterCategory" onchange="renderDocList()">
<option value="">全部分类</option>
</select>
</div>
<table class="doc-table">
<thead>
<tr>
<th>标题</th>
<th>分类</th>
<th>排序</th>
<th>状态</th>
<th>更新时间</th>
<th>操作</th>
</tr>
</thead>
<tbody id="docTableBody"></tbody>
</table>
</div>
<!-- ② 新建/编辑面板 -->
<div class="panel" id="panel-doc-edit">
<div class="panel-header">
<h2 id="editPanelTitle">✏️ 新建文档</h2>
<div>
<button class="btn-outline" onclick="switchPanel('doc-list')">← 返回列表</button>
<button class="btn-primary" onclick="saveDoc()">💾 保存发布</button>
</div>
</div>
<div class="edit-form">
<div class="form-row">
<div class="form-group" style="flex:2">
<label>文档标题 *</label>
<input type="text" id="docTitle" placeholder="输入文档标题" />
</div>
<div class="form-group">
<label>所属分类 *</label>
<select id="docCategory">
<option value="">请选择分类</option>
</select>
</div>
<div class="form-group">
<label>排序权重</label>
<input type="number" id="docOrder" value="0" min="0" />
</div>
</div>
<div class="form-row">
<div class="form-group" style="flex:2">
<label>文档ID（路由slug）</label>
<input type="text" id="docSlug" placeholder="例: quick-start（留空自动生成）" />
</div>
<div class="form-group">
<label>状态</label>
<select id="docStatus">
<option value="published">已发布</option>
<option value="draft">草稿</option>
</select>
</div>
</div>
<div class="form-group">
<label>文档正文（支持 Markdown）</label>
<textarea id="docContent"></textarea>
</div>
<div class="form-row" style="margin-top:10px;">
<div class="form-group">
<label>SEO 描述</label>
<input type="text" id="docDesc" placeholder="简短描述，用于搜索展示" />
</div>
</div>
</div>
</div>
<!-- ③ 菜单管理面板 -->
<div class="panel" id="panel-nav-manage">
<div class="panel-header">
<h2>🗂️ 菜单/分类管理</h2>
<button class="btn-primary" onclick="addCategory()">+ 新建分类</button>
</div>
<p class="tip-text">💡 拖动可调整菜单排序，双击名称可编辑</p>
<ul class="category-list" id="categoryList"></ul>
</div>
<!-- ④ 站点设置面板 -->
<div class="panel" id="panel-site-setting">
<div class="panel-header">
<h2>⚙️ 站点设置</h2>
<button class="btn-primary" onclick="saveSiteSettings()">💾 保存设置</button>
</div>
<div class="edit-form">
<div class="form-group">
<label>网站标题</label>
<input type="text" id="siteTitle" placeholder="白虎TG助手" />
</div>
<div class="form-group">
<label>网站 Logo 文字/Emoji</label>
<input type="text" id="siteLogo" placeholder="🐯" />
</div>
<div class="form-group">
<label>首页副标题</label>
<input type="text" id="siteSubtitle" placeholder="专业的 Telegram 群发与自动化运营工具" />
</div>
<div class="form-group">
<label>TG 频道链接</label>
<input type="text" id="siteTgLink" placeholder="https://t.me/yourChannel" />
</div>
<div class="form-group">
<label>客服联系方式</label>
<input type="text" id="siteContact" placeholder="@yourUsername" />
</div>
<div class="form-group">
<label>管理密码（留空不修改）</label>
<input type="password" id="newPassword" placeholder="输入新密码" />
</div>
<div class="form-group">
<label>页脚版权文字</label>
<input type="text" id="siteFooter" placeholder="© 2024 白虎TG助手 版权所有" />
</div>
</div>
</div>
</main>
</div>
<script src="js/app.js"></script>
<script src="js/admin.js"></script>
</body>
</html>
```
---
### `css/style.css
|
2026-05-10 12:09:24
|