Initial commit: SMS Forwarder project with Docker support

This commit is contained in:
zack
2025-07-17 19:30:35 +08:00
commit 152e136673
29 changed files with 3583 additions and 0 deletions

62
.dockerignore Normal file
View File

@ -0,0 +1,62 @@
# Git
.git
.gitignore
# Python
__pycache__
*.pyc
*.pyo
*.pyd
.Python
env
pip-log.txt
pip-delete-this-directory.txt
.tox
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.log
.git
.mypy_cache
.pytest_cache
.hypothesis
# Virtual Environment
.venv
venv/
ENV/
# IDE
.vscode/
.idea/
*.swp
*.swo
# OS
.DS_Store
Thumbs.db
# Project specific
logs/
data/
config.yaml
*.log
# Docker
Dockerfile
docker-compose*.yml
.dockerignore
# Documentation
docs/
README.md
*.md
# Scripts
scripts/
# Tests
tests/

162
.gitignore vendored Normal file
View File

@ -0,0 +1,162 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
pip-wheel-metadata/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# IDE
.idea/
.vscode/
*.swp
*.swo
*~
# Project specific
config.yaml
logs/
data/
gotify_data/
ssl/
# OS specific
.DS_Store
.DS_Store?
._*
.Spotlight-V100
.Trashes
ehthumbs.db
Thumbs.db
# Docker volumes
sms_logs/
sms_data/
# Temporary files
*.tmp
*.temp
*.bak
*.backup

37
Dockerfile Normal file
View File

@ -0,0 +1,37 @@
FROM python:3.10-slim
# 设置工作目录
WORKDIR /app
# 安装系统依赖
RUN apt-get update && apt-get install -y \
curl \
&& rm -rf /var/lib/apt/lists/*
# 安装 uv
RUN pip install uv
# 复制项目文件
COPY pyproject.toml .
COPY sms_forwarder/ ./sms_forwarder/
# 安装 Python 依赖
RUN uv sync --frozen
# 创建必要目录
RUN mkdir -p logs data
# 创建非 root 用户
RUN useradd --create-home --shell /bin/bash app
RUN chown -R app:app /app
USER app
# 暴露端口(使用配置文件中的端口)
EXPOSE 12152
# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:12152/health || exit 1
# 启动命令
CMD ["uv", "run", "python", "-m", "sms_forwarder.main"]

71
README.md Normal file
View File

@ -0,0 +1,71 @@
# SMS Forwarder
一个将 iOS 短信转发到 Android 设备的通知服务器。
## 功能特性
- 接收来自 iPhone 快捷指令的短信内容
- 支持多种推送方式Pushbullet、FCM、Gotify、ntfy 等)
- 基于 FastAPI 的高性能 HTTP 服务器
- 使用 Apprise 库支持 80+ 种通知服务
- YAML 配置文件,支持多推送目标
- API 密钥认证,确保安全性
- 完整的日志记录和错误处理
## 快速开始
### 1. 环境要求
- Python 3.10+
- uv 包管理器
### 2. 安装
```bash
# 克隆项目
git clone <repository-url>
cd notification
# 使用 uv 安装依赖
uv sync
```
### 3. 配置
复制配置文件模板并编辑:
```bash
cp config.example.yaml config.yaml
```
编辑 `config.yaml` 文件,配置你的推送服务。
### 4. 运行
```bash
# 开发模式
uv run uvicorn sms_forwarder.main:app --reload --host 0.0.0.0 --port 8000
# 生产模式
uv run sms-forwarder
```
## 支持的推送服务
- **Pushbullet** - 推荐,设置简单
- **FCM (Firebase Cloud Messaging)** - Google 官方推送
- **Gotify** - 自托管推送服务
- **ntfy** - 开源推送服务
- **Discord、Telegram、Slack** 等 80+ 种服务
## iPhone 快捷指令配置
详见 [iPhone 配置指南](docs/iphone-setup.md)
## API 文档
服务器启动后访问 `http://localhost:8000/docs` 查看 API 文档。
## 许可证
MIT License

67
config.example.yaml Normal file
View File

@ -0,0 +1,67 @@
# SMS Forwarder 配置文件示例
# 复制此文件为 config.yaml 并根据需要修改
# 服务器配置
server:
host: "0.0.0.0"
port: 12152
# API 密钥,用于验证请求(必须设置)
# 建议使用强密码生成器生成例如python -c "import secrets; print(secrets.token_urlsafe(32))"
api_key: "ayNESyIW2pCQ5Ts-O8FC5t8mzhb26kbDZEr4I7PynHg"
# 日志配置
logging:
level: "INFO" # DEBUG, INFO, WARNING, ERROR
file: "logs/sms_forwarder.log"
# 通知配置
notifications:
# 默认通知服务(必须配置至少一个)
services:
# Pushbullet 示例(推荐)
# 将 your-pushbullet-access-token 替换为你从 Pushbullet 网站获取的 Access Token
- name: "pushbullet"
url: "pbul://o.YV9e3ugEsg2bWKZ0U7HA4IlTLkEp1BtU"
enabled: true
# FCM 示例
# - name: "fcm"
# url: "fcm://project@apikey/DEVICE_ID"
# enabled: false
# Gotify 示例(自托管)
# - name: "gotify"
# url: "gotify://hostname/token"
# enabled: false
# ntfy 示例
# - name: "ntfy"
# url: "ntfy://your-topic/"
# enabled: false
# Discord Webhook 示例
# - name: "discord"
# url: "discord://webhook_id/webhook_token"
# enabled: false
# 通知模板
templates:
# 短信通知模板
sms:
title: "📱 新短信 - {sender}"
body: |
发件人: {sender}
时间: {timestamp}
内容: {content}
# 系统通知模板
system:
title: "🔔 系统通知"
body: "{message}"
# 安全配置
security:
# 允许的来源 IP可选留空表示允许所有
allowed_ips: []
# 请求频率限制(每分钟最大请求数)
rate_limit: 60

54
docker-compose.prod.yml Normal file
View File

@ -0,0 +1,54 @@
version: '3.8'
services:
sms-forwarder:
build: .
container_name: sms-forwarder
ports:
- "12152:12152"
volumes:
- ./config.yaml:/app/config.yaml:ro
- sms_logs:/app/logs
- sms_data:/app/data
environment:
- CONFIG_PATH=/app/config.yaml
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:12152/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 40s
deploy:
resources:
limits:
memory: 256M
cpus: '0.5'
reservations:
memory: 128M
cpus: '0.1'
logging:
driver: "json-file"
options:
max-size: "10m"
max-file: "3"
# Nginx 反向代理(可选,用于 HTTPS
nginx:
image: nginx:alpine
container_name: sms-forwarder-nginx
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf:ro
- ./ssl:/etc/nginx/ssl:ro
depends_on:
- sms-forwarder
restart: unless-stopped
profiles:
- nginx
volumes:
sms_logs:
sms_data:

46
docker-compose.yml Normal file
View File

@ -0,0 +1,46 @@
version: '3.8'
services:
sms-forwarder:
build: .
container_name: sms-forwarder
ports:
- "12152:12152"
volumes:
- ./config.yaml:/app/config.yaml:ro
- ./logs:/app/logs
- ./data:/app/data
environment:
- CONFIG_PATH=/app/config.yaml
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:12152/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 40s
deploy:
resources:
limits:
memory: 256M
cpus: '0.5'
reservations:
memory: 128M
cpus: '0.1'
# 可选:添加 Gotify 服务作为推送后端
gotify:
image: gotify/server
container_name: gotify
ports:
- "8080:80"
volumes:
- gotify_data:/app/data
environment:
- GOTIFY_DEFAULTUSER_PASS=admin
restart: unless-stopped
profiles:
- gotify
volumes:
gotify_data:

264
docs/deployment.md Normal file
View File

@ -0,0 +1,264 @@
# 部署指南
本指南介绍如何部署 SMS Forwarder 服务器。
## 部署方式
### 1. 本地部署
#### 环境要求
- Python 3.10+
- uv 包管理器
#### 安装步骤
```bash
# 1. 克隆项目
git clone <repository-url>
cd notification
# 2. 安装依赖
uv sync
# 3. 复制配置文件
cp config.example.yaml config.yaml
# 4. 编辑配置文件
nano config.yaml
# 5. 运行服务器
uv run sms-forwarder
```
### 2. Docker 部署
#### 创建 Dockerfile
```dockerfile
FROM python:3.10-slim
WORKDIR /app
# 安装 uv
RUN pip install uv
# 复制项目文件
COPY . .
# 安装依赖
RUN uv sync
# 创建日志目录
RUN mkdir -p logs
# 暴露端口
EXPOSE 8000
# 启动命令
CMD ["uv", "run", "sms-forwarder"]
```
#### 构建和运行
```bash
# 构建镜像
docker build -t sms-forwarder .
# 运行容器
docker run -d \
--name sms-forwarder \
-p 8000:8000 \
-v $(pwd)/config.yaml:/app/config.yaml \
-v $(pwd)/logs:/app/logs \
sms-forwarder
```
### 3. systemd 服务部署
#### 创建服务文件
```bash
sudo nano /etc/systemd/system/sms-forwarder.service
```
```ini
[Unit]
Description=SMS Forwarder Service
After=network.target
[Service]
Type=simple
User=your-user
WorkingDirectory=/path/to/sms-forwarder
Environment=PATH=/path/to/sms-forwarder/.venv/bin
ExecStart=/path/to/sms-forwarder/.venv/bin/python -m sms_forwarder.main
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target
```
#### 启用服务
```bash
sudo systemctl daemon-reload
sudo systemctl enable sms-forwarder
sudo systemctl start sms-forwarder
sudo systemctl status sms-forwarder
```
## 配置推送服务
### Pushbullet 配置
1. 访问 [Pushbullet](https://www.pushbullet.com/)
2. 注册账号并在 Android 设备上安装应用
3. 获取 Access Token
- 访问 https://www.pushbullet.com/#settings/account
- 创建 Access Token
4. 在配置文件中添加:
```yaml
notifications:
services:
- name: "pushbullet"
url: "pbul://your-access-token-here"
enabled: true
```
### FCM 配置
1. 创建 Firebase 项目
2. 获取服务器密钥和项目 ID
3. 在 Android 应用中集成 FCM
4. 获取设备 token
5. 配置:
```yaml
notifications:
services:
- name: "fcm"
url: "fcm://project-id@server-key/device-token"
enabled: true
```
### Gotify 配置
1. 部署 Gotify 服务器
2. 创建应用并获取 token
3. 配置:
```yaml
notifications:
services:
- name: "gotify"
url: "gotify://your-server.com/app-token"
enabled: true
```
## 安全配置
### 1. 生成强 API 密钥
```bash
# 使用 Python 生成随机密钥
python -c "import secrets; print(secrets.token_urlsafe(32))"
```
### 2. 配置防火墙
```bash
# 只允许特定 IP 访问
sudo ufw allow from YOUR_IPHONE_IP to any port 8000
```
### 3. 使用 HTTPS
#### 使用 nginx 反向代理
```nginx
server {
listen 443 ssl;
server_name your-domain.com;
ssl_certificate /path/to/cert.pem;
ssl_certificate_key /path/to/key.pem;
location / {
proxy_pass http://127.0.0.1:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
}
```
## 监控和维护
### 1. 日志监控
```bash
# 查看实时日志
tail -f logs/sms_forwarder.log
# 查看错误日志
grep ERROR logs/sms_forwarder.log
```
### 2. 健康检查
```bash
# 检查服务状态
curl http://localhost:8000/health
# 检查系统状态
curl http://localhost:8000/status
```
### 3. 性能监控
可以使用以下工具监控服务器性能:
- Prometheus + Grafana
- htop / top
- systemd journal
### 4. 备份配置
```bash
# 定期备份配置文件
cp config.yaml config.yaml.backup.$(date +%Y%m%d)
```
## 故障排除
### 常见问题
1. **服务启动失败**
- 检查配置文件语法
- 确认端口未被占用
- 查看详细错误日志
2. **通知发送失败**
- 验证推送服务配置
- 检查网络连接
- 确认 API 密钥有效
3. **高内存使用**
- 检查日志文件大小
- 考虑添加日志轮转
- 监控请求频率
### 调试命令
```bash
# 检查端口占用
netstat -tlnp | grep 8000
# 检查进程状态
ps aux | grep sms-forwarder
# 测试配置文件
uv run python -c "from sms_forwarder.config import get_config; print(get_config())"
```

149
docs/iphone-setup.md Normal file
View File

@ -0,0 +1,149 @@
# iPhone 快捷指令配置指南
本指南将帮助你在 iPhone 上设置快捷指令,自动将收到的短信转发到你的 Android 设备。
## 前提条件
1. iPhone 上已安装"快捷指令"应用
2. SMS Forwarder 服务器已部署并运行
3. 已获取服务器的 API 密钥
## 步骤 1: 创建快捷指令
1. 打开"快捷指令"应用
2. 点击右上角的"+"创建新快捷指令
3. 点击"添加操作"
## 步骤 2: 配置快捷指令动作
### 2.1 获取短信内容
1. 搜索并添加"获取我的快捷指令"动作
2. 搜索并添加"文本"动作,用于获取短信内容
3. 在文本框中输入短信内容(这将在自动化中被替换)
### 2.2 发送 HTTP 请求
1. 搜索并添加"获取 URL 内容"动作
2. 配置如下:
- **URL**: `http://your-server-ip:8000/notify`
- **方法**: POST
- **请求体**: JSON
- **标头**:
- `Content-Type`: `application/json`
### 2.3 配置请求体
在"获取 URL 内容"动作的请求体中,输入以下 JSON 格式:
```json
{
"api_key": "your-secret-api-key-here",
"message": {
"sender": "发送者号码或名称",
"content": "短信内容",
"timestamp": "2024-01-01T12:00:00"
}
}
```
**注意**: 实际使用时,你需要:
-`your-secret-api-key-here` 替换为你的实际 API 密钥
- 使用快捷指令的变量来动态获取发送者和内容
## 步骤 3: 设置自动化
1. 在快捷指令应用中,点击底部的"自动化"
2. 点击右上角的"+"创建新自动化
3. 选择"收到信息时"
4. 配置触发条件:
- 选择"任何人"或特定联系人
- 选择"立即运行"
5. 选择你刚创建的快捷指令
## 步骤 4: 高级配置
### 动态获取短信信息
为了动态获取短信的发送者和内容,你需要在快捷指令中:
1. 使用"获取输入内容"动作获取短信
2. 使用"获取短信详细信息"动作提取发送者和内容
3. 在 HTTP 请求中使用这些变量
### 示例快捷指令流程
```
1. 获取输入内容 (短信)
2. 获取短信详细信息
- 发送者 → 变量: sender
- 内容 → 变量: content
3. 文本 (JSON 请求体)
{
"api_key": "your-api-key",
"message": {
"sender": "[sender变量]",
"content": "[content变量]",
"timestamp": "[当前日期]"
}
}
4. 获取 URL 内容
- URL: http://your-server:8000/notify
- 方法: POST
- 请求体: 上面的文本
```
## 步骤 5: 测试
1. 保存快捷指令和自动化
2. 让朋友给你发送一条测试短信
3. 检查你的 Android 设备是否收到通知
4. 查看服务器日志确认请求是否成功
## 故障排除
### 常见问题
1. **通知没有发送**
- 检查网络连接
- 确认服务器地址和端口正确
- 验证 API 密钥是否正确
2. **自动化没有触发**
- 确认自动化已启用
- 检查触发条件设置
- 重启快捷指令应用
3. **HTTP 请求失败**
- 检查 JSON 格式是否正确
- 确认 Content-Type 标头已设置
- 查看服务器日志获取详细错误信息
### 调试技巧
1. 在快捷指令中添加"显示通知"动作来调试
2. 使用"获取 URL 内容"的响应来检查服务器返回
3. 查看服务器的 `/health` 端点确认服务正常
## 安全建议
1. 使用强 API 密钥
2. 考虑使用 HTTPS需要 SSL 证书)
3. 限制服务器的访问 IP
4. 定期更换 API 密钥
## 进阶功能
### 过滤特定短信
你可以在快捷指令中添加条件判断,只转发特定的短信:
```
1. 获取短信内容
2. 如果 (内容包含"验证码"或发送者是"银行")
3. 执行 HTTP 请求
```
### 自定义通知格式
通过修改服务器配置文件中的模板,你可以自定义通知的显示格式。

View File

@ -0,0 +1,96 @@
# 简化版 iPhone 快捷指令配置指南
本指南将帮助你设置一个简单的 iPhone 快捷指令,只需要发送短信内容,无需其他复杂信息。
## 前提条件
1. iPhone 上已安装"快捷指令"应用
2. SMS Forwarder 服务器已部署并运行
3. 已在 Android 设备上安装 Pushbullet 应用
## 步骤 1: 创建快捷指令
1. 打开 iPhone 上的"快捷指令"应用
2. 点击右上角的"+"创建新快捷指令
3. 点击"添加操作"
## 步骤 2: 配置快捷指令动作
### 2.1 获取短信内容
1. 搜索并添加"获取我的快捷指令的输入"动作
2. 这将获取短信内容作为输入
### 2.2 发送 HTTP 请求
1. 搜索并添加"获取 URL 内容"动作
2. 配置如下:
- **URL**: `http://你的服务器IP:12152/notify/simple`
- **方法**: POST
- **请求体**: JSON
- **标头**:
- `Content-Type`: `application/json`
### 2.3 配置请求体
在"获取 URL 内容"动作的请求体中,输入以下 JSON 格式:
```json
{
"api_key": "ayNESyIW2pCQ5Ts-O8FC5t8mzhb26kbDZEr4I7PynHg",
"content": "快捷指令输入"
}
```
**注意**:
-`"快捷指令输入"` 替换为变量,方法是点击它,然后从变量列表中选择"快捷指令输入"
- API 密钥已经设置好,不需要修改
## 步骤 3: 设置自动化
1. 在快捷指令应用中,点击底部的"自动化"
2. 点击右上角的"+"创建新自动化
3. 选择"收到信息时"
4. 配置触发条件:
- 选择"任何人"或特定联系人
- 选择"立即运行"
5. 选择你刚创建的快捷指令
## 步骤 4: 测试
1. 保存快捷指令和自动化
2. 让朋友给你发送一条测试短信
3. 检查你的 Android 设备上的 Pushbullet 是否收到通知
## 故障排除
### 常见问题
1. **通知没有发送**
- 确认服务器地址和端口正确 (12152)
- 确认 API 密钥正确
- 检查 Android 设备上的 Pushbullet 是否正常运行
2. **自动化没有触发**
- 确认自动化已启用
- 检查触发条件设置
- 重启快捷指令应用
### 调试技巧
1. 在快捷指令中添加"显示通知"动作来调试
2. 使用"获取 URL 内容"的响应来检查服务器返回
## 高级选项
如果你想添加发送者信息,可以修改 JSON 为:
```json
{
"api_key": "ayNESyIW2pCQ5Ts-O8FC5t8mzhb26kbDZEr4I7PynHg",
"content": "快捷指令输入",
"sender": "发送者名称"
}
```
其中 `"发送者名称"` 可以是固定文本,如 "银行短信" 或 "验证码"。

44
pyproject.toml Normal file
View File

@ -0,0 +1,44 @@
[project]
name = "sms-forwarder"
version = "0.1.0"
description = "iOS SMS to Android notification forwarder"
authors = [
{name = "Your Name", email = "your.email@example.com"}
]
dependencies = [
"fastapi>=0.104.0",
"uvicorn[standard]>=0.24.0",
"apprise>=1.7.0",
"pydantic>=2.5.0",
"pydantic-settings>=2.1.0",
"pyyaml>=6.0.1",
"python-multipart>=0.0.6",
]
requires-python = ">=3.10"
readme = "README.md"
license = {text = "MIT"}
[project.optional-dependencies]
dev = [
"pytest>=7.4.0",
"pytest-asyncio>=0.21.0",
"httpx>=0.25.0",
"black>=23.0.0",
"isort>=5.12.0",
"flake8>=6.0.0",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.black]
line-length = 88
target-version = ['py310']
[tool.isort]
profile = "black"
line_length = 88
[project.scripts]
sms-forwarder = "sms_forwarder.main:main"

138
scripts/deploy.sh Executable file
View File

@ -0,0 +1,138 @@
#!/bin/bash
# SMS Forwarder Docker 部署脚本
set -e
echo "🚀 SMS Forwarder Docker 部署脚本"
echo "================================"
# 检查 Docker 和 Docker Compose
check_docker() {
if ! command -v docker &> /dev/null; then
echo "❌ Docker 未安装,请先安装 Docker"
exit 1
fi
if ! command -v docker-compose &> /dev/null && ! docker compose version &> /dev/null; then
echo "❌ Docker Compose 未安装,请先安装 Docker Compose"
exit 1
fi
echo "✅ Docker 环境检查通过"
}
# 检查配置文件
check_config() {
if [ ! -f "config.yaml" ]; then
echo "⚠️ 配置文件不存在,从示例文件创建..."
if [ -f "config.example.yaml" ]; then
cp config.example.yaml config.yaml
echo "📝 请编辑 config.yaml 文件配置你的推送服务"
echo "是否现在编辑配置文件? (y/N)"
read -r response
if [[ "$response" =~ ^[Yy]$ ]]; then
${EDITOR:-nano} config.yaml
fi
else
echo "❌ 配置文件模板不存在"
exit 1
fi
else
echo "✅ 配置文件存在"
fi
}
# 构建镜像
build_image() {
echo "🔨 构建 Docker 镜像..."
docker build -t sms-forwarder:latest .
echo "✅ 镜像构建完成"
}
# 部署服务
deploy_service() {
echo "🚀 部署服务..."
# 停止现有服务
if docker ps -q --filter "name=sms-forwarder" | grep -q .; then
echo "⏹️ 停止现有服务..."
docker-compose down
fi
# 启动新服务
echo "▶️ 启动服务..."
docker-compose up -d
# 等待服务启动
echo "⏳ 等待服务启动..."
sleep 10
# 检查服务状态
if docker ps --filter "name=sms-forwarder" --filter "status=running" | grep -q sms-forwarder; then
echo "✅ 服务启动成功"
else
echo "❌ 服务启动失败"
docker-compose logs sms-forwarder
exit 1
fi
}
# 测试服务
test_service() {
echo "🧪 测试服务..."
# 等待服务完全启动
sleep 5
# 健康检查
if curl -f http://localhost:12152/health &> /dev/null; then
echo "✅ 健康检查通过"
else
echo "❌ 健康检查失败"
docker-compose logs sms-forwarder
exit 1
fi
# 状态检查
if curl -f http://localhost:12152/status &> /dev/null; then
echo "✅ 状态接口正常"
else
echo "❌ 状态接口异常"
fi
}
# 显示部署信息
show_info() {
echo ""
echo "🎉 部署完成!"
echo ""
echo "服务信息:"
echo " - 服务地址: http://localhost:12152"
echo " - API 文档: http://localhost:12152/docs"
echo " - 健康检查: http://localhost:12152/health"
echo " - 服务状态: http://localhost:12152/status"
echo ""
echo "管理命令:"
echo " - 查看日志: docker-compose logs -f sms-forwarder"
echo " - 重启服务: docker-compose restart sms-forwarder"
echo " - 停止服务: docker-compose down"
echo " - 查看状态: docker-compose ps"
echo ""
echo "测试命令:"
echo " - 发送测试通知: ./scripts/test-docker.sh"
echo ""
}
# 主函数
main() {
check_docker
check_config
build_image
deploy_service
test_service
show_info
}
# 运行主函数
main "$@"

86
scripts/docker-manage.sh Executable file
View File

@ -0,0 +1,86 @@
#!/bin/bash
# SMS Forwarder Docker 管理脚本
show_help() {
echo "SMS Forwarder Docker 管理脚本"
echo ""
echo "用法: $0 [命令]"
echo ""
echo "命令:"
echo " start 启动服务"
echo " stop 停止服务"
echo " restart 重启服务"
echo " status 查看服务状态"
echo " logs 查看实时日志"
echo " build 重新构建镜像"
echo " test 发送测试通知"
echo " stats 查看资源使用"
echo " shell 进入容器 shell"
echo " clean 清理未使用的镜像和容器"
echo " deploy 完整部署流程"
echo " help 显示此帮助信息"
}
case "$1" in
start)
echo "🚀 启动 SMS Forwarder 服务..."
docker-compose up -d
;;
stop)
echo "⏹️ 停止 SMS Forwarder 服务..."
docker-compose down
;;
restart)
echo "🔄 重启 SMS Forwarder 服务..."
docker-compose restart sms-forwarder
;;
status)
echo "📊 SMS Forwarder 服务状态:"
docker-compose ps
echo ""
echo "容器详细信息:"
docker inspect sms-forwarder --format='{{.State.Status}}: {{.State.StartedAt}}'
;;
logs)
echo "📋 SMS Forwarder 实时日志 (Ctrl+C 退出):"
docker-compose logs -f sms-forwarder
;;
build)
echo "🔨 重新构建镜像..."
docker-compose build --no-cache sms-forwarder
;;
test)
echo "🧪 发送测试通知..."
./scripts/test-docker.sh
;;
stats)
echo "💾 资源使用统计:"
docker stats sms-forwarder --no-stream
;;
shell)
echo "🐚 进入容器 shell..."
docker-compose exec sms-forwarder /bin/bash
;;
clean)
echo "🧹 清理未使用的 Docker 资源..."
docker system prune -f
docker image prune -f
;;
deploy)
echo "🚀 执行完整部署流程..."
./scripts/deploy.sh
;;
help|--help|-h)
show_help
;;
"")
show_help
;;
*)
echo "❌ 未知命令: $1"
echo ""
show_help
exit 1
;;
esac

128
scripts/install.sh Executable file
View File

@ -0,0 +1,128 @@
#!/bin/bash
# SMS Forwarder 安装脚本
set -e
echo "🚀 开始安装 SMS Forwarder..."
# 检查 Python 版本
check_python() {
if command -v python3 &> /dev/null; then
PYTHON_VERSION=$(python3 -c 'import sys; print(".".join(map(str, sys.version_info[:2])))')
if [[ $(echo "$PYTHON_VERSION >= 3.10" | bc -l) -eq 1 ]]; then
echo "✅ Python $PYTHON_VERSION 已安装"
else
echo "❌ 需要 Python 3.10 或更高版本,当前版本: $PYTHON_VERSION"
exit 1
fi
else
echo "❌ 未找到 Python 3请先安装 Python 3.10+"
exit 1
fi
}
# 安装 uv
install_uv() {
if command -v uv &> /dev/null; then
echo "✅ uv 已安装"
else
echo "📦 安装 uv..."
curl -LsSf https://astral.sh/uv/install.sh | sh
source $HOME/.cargo/env
fi
}
# 安装依赖
install_dependencies() {
echo "📦 安装项目依赖..."
uv sync
}
# 创建配置文件
setup_config() {
if [ ! -f "config.yaml" ]; then
echo "⚙️ 创建配置文件..."
cp config.example.yaml config.yaml
echo "📝 请编辑 config.yaml 文件配置你的推送服务"
else
echo "✅ 配置文件已存在"
fi
}
# 创建必要目录
create_directories() {
echo "📁 创建必要目录..."
mkdir -p logs
mkdir -p data
}
# 生成 API 密钥
generate_api_key() {
if command -v python3 &> /dev/null; then
API_KEY=$(python3 -c "import secrets; print(secrets.token_urlsafe(32))")
echo "🔑 生成的 API 密钥: $API_KEY"
echo "请将此密钥添加到 config.yaml 文件中的 server.api_key 字段"
fi
}
# 创建 systemd 服务文件
create_systemd_service() {
read -p "是否创建 systemd 服务? (y/N): " create_service
if [[ $create_service =~ ^[Yy]$ ]]; then
CURRENT_DIR=$(pwd)
USER=$(whoami)
cat > sms-forwarder.service << EOF
[Unit]
Description=SMS Forwarder Service
After=network.target
[Service]
Type=simple
User=$USER
WorkingDirectory=$CURRENT_DIR
Environment=PATH=$CURRENT_DIR/.venv/bin
ExecStart=$CURRENT_DIR/.venv/bin/python -m sms_forwarder.main
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target
EOF
echo "📄 systemd 服务文件已创建: sms-forwarder.service"
echo "运行以下命令安装服务:"
echo " sudo cp sms-forwarder.service /etc/systemd/system/"
echo " sudo systemctl daemon-reload"
echo " sudo systemctl enable sms-forwarder"
echo " sudo systemctl start sms-forwarder"
fi
}
# 主安装流程
main() {
echo "SMS Forwarder 安装程序"
echo "======================"
check_python
install_uv
install_dependencies
create_directories
setup_config
generate_api_key
create_systemd_service
echo ""
echo "🎉 安装完成!"
echo ""
echo "下一步:"
echo "1. 编辑 config.yaml 文件配置推送服务"
echo "2. 运行 'uv run sms-forwarder' 启动服务器"
echo "3. 访问 http://localhost:8000/docs 查看 API 文档"
echo "4. 参考 docs/iphone-setup.md 配置 iPhone 快捷指令"
echo ""
}
# 运行主函数
main "$@"

96
scripts/manage.sh Executable file
View File

@ -0,0 +1,96 @@
#!/bin/bash
# SMS Forwarder 管理脚本
SERVICE_NAME="sms-forwarder"
show_help() {
echo "SMS Forwarder 管理脚本"
echo ""
echo "用法: $0 [命令]"
echo ""
echo "命令:"
echo " start 启动服务"
echo " stop 停止服务"
echo " restart 重启服务"
echo " status 查看服务状态"
echo " logs 查看实时日志"
echo " enable 启用开机自启"
echo " disable 禁用开机自启"
echo " test 发送测试通知"
echo " update 更新依赖"
echo " config 编辑配置文件"
echo " install 安装 systemd 服务"
echo " help 显示此帮助信息"
}
case "$1" in
start)
echo "🚀 启动 SMS Forwarder 服务..."
sudo systemctl start $SERVICE_NAME
;;
stop)
echo "⏹️ 停止 SMS Forwarder 服务..."
sudo systemctl stop $SERVICE_NAME
;;
restart)
echo "🔄 重启 SMS Forwarder 服务..."
sudo systemctl restart $SERVICE_NAME
;;
status)
echo "📊 SMS Forwarder 服务状态:"
sudo systemctl status $SERVICE_NAME --no-pager
;;
logs)
echo "📋 SMS Forwarder 实时日志 (Ctrl+C 退出):"
sudo journalctl -u $SERVICE_NAME -f
;;
enable)
echo "✅ 启用 SMS Forwarder 开机自启..."
sudo systemctl enable $SERVICE_NAME
;;
disable)
echo "❌ 禁用 SMS Forwarder 开机自启..."
sudo systemctl disable $SERVICE_NAME
;;
test)
echo "🧪 发送测试通知..."
if [ -f "config.yaml" ]; then
API_KEY=$(grep "api_key:" config.yaml | awk '{print $2}' | tr -d '"')
uv run python scripts/test_notification.py http://localhost:12152 "$API_KEY"
else
echo "❌ 配置文件不存在"
fi
;;
update)
echo "📦 更新依赖..."
uv sync
echo "🔄 重启服务以应用更新..."
sudo systemctl restart $SERVICE_NAME
;;
config)
echo "⚙️ 编辑配置文件..."
${EDITOR:-nano} config.yaml
echo "是否重启服务以应用配置? (y/N)"
read -r response
if [[ "$response" =~ ^[Yy]$ ]]; then
sudo systemctl restart $SERVICE_NAME
fi
;;
install)
echo "🔧 安装 systemd 服务..."
./scripts/setup-systemd.sh
;;
help|--help|-h)
show_help
;;
"")
show_help
;;
*)
echo "❌ 未知命令: $1"
echo ""
show_help
exit 1
;;
esac

71
scripts/setup-systemd.sh Executable file
View File

@ -0,0 +1,71 @@
#!/bin/bash
# SMS Forwarder systemd 服务安装脚本
set -e
echo "🔧 配置 SMS Forwarder systemd 服务..."
# 获取当前用户和路径
CURRENT_USER=$(whoami)
CURRENT_DIR=$(pwd)
SERVICE_FILE="sms-forwarder.service"
# 检查是否在正确的目录
if [ ! -f "pyproject.toml" ] || [ ! -f "config.yaml" ]; then
echo "❌ 请在项目根目录运行此脚本"
exit 1
fi
# 检查 uv 虚拟环境
if [ ! -d ".venv" ]; then
echo "📦 创建虚拟环境..."
uv sync
fi
# 获取虚拟环境路径
VENV_PATH="$CURRENT_DIR/.venv"
if [ ! -f "$VENV_PATH/bin/python" ]; then
echo "❌ 虚拟环境未找到,请先运行 'uv sync'"
exit 1
fi
# 创建日志和数据目录
mkdir -p logs data
# 更新服务文件中的路径
echo "📝 更新服务文件配置..."
sed -i "s|your-username|$CURRENT_USER|g" $SERVICE_FILE
sed -i "s|/path/to/notification|$CURRENT_DIR|g" $SERVICE_FILE
# 复制服务文件到系统目录
echo "📋 安装 systemd 服务文件..."
sudo cp $SERVICE_FILE /etc/systemd/system/
# 重新加载 systemd
echo "🔄 重新加载 systemd..."
sudo systemctl daemon-reload
# 启用服务
echo "✅ 启用 SMS Forwarder 服务..."
sudo systemctl enable sms-forwarder
# 启动服务
echo "🚀 启动 SMS Forwarder 服务..."
sudo systemctl start sms-forwarder
# 检查状态
echo "📊 检查服务状态..."
sudo systemctl status sms-forwarder --no-pager
echo ""
echo "🎉 SMS Forwarder 服务安装完成!"
echo ""
echo "常用命令:"
echo " 查看状态: sudo systemctl status sms-forwarder"
echo " 查看日志: sudo journalctl -u sms-forwarder -f"
echo " 重启服务: sudo systemctl restart sms-forwarder"
echo " 停止服务: sudo systemctl stop sms-forwarder"
echo " 禁用服务: sudo systemctl disable sms-forwarder"
echo ""
echo "服务将在系统重启后自动启动。"

114
scripts/test-docker.sh Executable file
View File

@ -0,0 +1,114 @@
#!/bin/bash
# Docker 环境测试脚本
set -e
echo "🧪 SMS Forwarder Docker 测试"
echo "============================"
# 检查服务是否运行
check_service() {
if ! docker ps --filter "name=sms-forwarder" --filter "status=running" | grep -q sms-forwarder; then
echo "❌ SMS Forwarder 服务未运行"
echo "请先运行: docker-compose up -d"
exit 1
fi
echo "✅ 服务正在运行"
}
# 获取 API 密钥
get_api_key() {
if [ -f "config.yaml" ]; then
API_KEY=$(grep "api_key:" config.yaml | awk '{print $2}' | tr -d '"')
if [ -z "$API_KEY" ]; then
echo "❌ 无法从配置文件获取 API 密钥"
exit 1
fi
echo "✅ 获取到 API 密钥"
else
echo "❌ 配置文件不存在"
exit 1
fi
}
# 测试健康检查
test_health() {
echo "🔍 测试健康检查..."
if curl -f http://localhost:12152/health &> /dev/null; then
echo "✅ 健康检查通过"
else
echo "❌ 健康检查失败"
return 1
fi
}
# 测试状态接口
test_status() {
echo "📊 测试状态接口..."
STATUS=$(curl -s http://localhost:12152/status)
if [ $? -eq 0 ]; then
echo "✅ 状态接口正常"
echo "服务状态: $STATUS"
else
echo "❌ 状态接口异常"
return 1
fi
}
# 测试通知发送
test_notification() {
echo "📱 测试通知发送..."
RESPONSE=$(curl -s -X POST "http://localhost:12152/notify/simple" \
-H "Content-Type: application/json" \
-d "{
\"api_key\": \"$API_KEY\",
\"content\": \"Docker 测试通知 - $(date)\",
\"sender\": \"Docker 测试\"
}")
if echo "$RESPONSE" | grep -q '"success":true'; then
echo "✅ 通知发送成功"
echo "响应: $RESPONSE"
else
echo "❌ 通知发送失败"
echo "响应: $RESPONSE"
return 1
fi
}
# 显示容器信息
show_container_info() {
echo ""
echo "📋 容器信息:"
docker-compose ps
echo ""
echo "💾 资源使用:"
docker stats sms-forwarder --no-stream --format "table {{.Container}}\t{{.CPUPerc}}\t{{.MemUsage}}\t{{.NetIO}}"
}
# 主函数
main() {
check_service
get_api_key
echo ""
echo "开始测试..."
test_health
test_status
test_notification
echo ""
echo "🎉 所有测试通过!"
show_container_info
echo ""
echo "如果你的 Android 设备收到了通知,说明 Docker 部署成功!"
}
# 运行主函数
main "$@"

125
scripts/test_notification.py Executable file
View File

@ -0,0 +1,125 @@
#!/usr/bin/env python3
"""测试通知发送脚本."""
import json
import sys
from datetime import datetime
import requests
def test_notification(server_url: str, api_key: str):
"""测试发送通知."""
# 测试简化版 API
simple_test_data = {
"api_key": api_key,
"content": "这是一条测试短信,用于验证通知转发功能是否正常工作。",
"sender": "测试发送者"
}
try:
print(f"🔄 正在向 {server_url} 发送测试通知...")
# 发送请求到简化版 API
response = requests.post(
f"{server_url}/notify/simple",
json=simple_test_data,
headers={"Content-Type": "application/json"},
timeout=10
)
# 检查响应
if response.status_code == 200:
result = response.json()
if result.get("success"):
print("✅ 测试通知发送成功!")
print(f"📱 请检查你的 Android 设备是否收到通知")
else:
print(f"❌ 通知发送失败: {result.get('message')}")
return False
else:
print(f"❌ HTTP 请求失败: {response.status_code}")
print(f"响应内容: {response.text}")
return False
except requests.exceptions.ConnectionError:
print(f"❌ 无法连接到服务器 {server_url}")
print("请确认服务器正在运行且地址正确")
return False
except requests.exceptions.Timeout:
print("❌ 请求超时")
return False
except Exception as e:
print(f"❌ 发生错误: {e}")
return False
return True
def test_server_status(server_url: str):
"""测试服务器状态."""
try:
print(f"🔄 检查服务器状态...")
# 健康检查
health_response = requests.get(f"{server_url}/health", timeout=5)
if health_response.status_code == 200:
print("✅ 服务器健康检查通过")
else:
print(f"⚠️ 健康检查失败: {health_response.status_code}")
# 状态信息
status_response = requests.get(f"{server_url}/status", timeout=5)
if status_response.status_code == 200:
status = status_response.json()
print(f"📊 服务器状态:")
print(f" 版本: {status.get('version')}")
print(f" 运行时间: {status.get('uptime', 0):.2f}")
print(f" 已发送通知: {status.get('notifications_sent', 0)}")
print(f" 失败通知: {status.get('notifications_failed', 0)}")
else:
print(f"⚠️ 获取状态信息失败: {status_response.status_code}")
except Exception as e:
print(f"❌ 检查服务器状态时出错: {e}")
return False
return True
def main():
"""主函数."""
if len(sys.argv) < 3:
print("使用方法: python test_notification.py <server_url> <api_key>")
print("示例: python test_notification.py http://localhost:8000 your-api-key")
sys.exit(1)
server_url = sys.argv[1].rstrip('/')
api_key = sys.argv[2]
print("SMS Forwarder 通知测试")
print("=" * 30)
# 测试服务器状态
if not test_server_status(server_url):
print("❌ 服务器状态检查失败")
sys.exit(1)
print()
# 测试通知发送
if test_notification(server_url, api_key):
print("\n🎉 所有测试通过!")
print("如果你的 Android 设备收到了通知,说明配置正确。")
else:
print("\n❌ 测试失败")
print("请检查:")
print("1. 服务器配置是否正确")
print("2. API 密钥是否正确")
print("3. 推送服务是否正确配置")
sys.exit(1)
if __name__ == "__main__":
main()

33
sms-forwarder.service Normal file
View File

@ -0,0 +1,33 @@
[Unit]
Description=SMS Forwarder Service - iOS to Android notification forwarder
After=network.target network-online.target
Wants=network-online.target
[Service]
Type=simple
User=zack
Group=zack
WorkingDirectory=/mnt/dsk1/code/project/notification
Environment=PATH=/mnt/dsk1/code/project/notification/.venv/bin:/usr/local/bin:/usr/bin:/bin
Environment=CONFIG_PATH=/mnt/dsk1/code/project/notification/config.yaml
ExecStart=/mnt/dsk1/code/project/notification/.venv/bin/python -m sms_forwarder.main
ExecReload=/bin/kill -HUP $MAINPID
Restart=always
RestartSec=10
StandardOutput=journal
StandardError=journal
SyslogIdentifier=sms-forwarder
# 安全设置
NoNewPrivileges=true
PrivateTmp=true
ProtectSystem=strict
ProtectHome=true
ReadWritePaths=/mnt/dsk1/code/project/notification/logs /mnt/dsk1/code/project/notification/data
# 资源限制
MemoryMax=256M
CPUQuota=50%
[Install]
WantedBy=multi-user.target

View File

@ -0,0 +1,3 @@
"""SMS Forwarder - iOS SMS to Android notification forwarder."""
__version__ = "0.1.0"

87
sms_forwarder/auth.py Normal file
View File

@ -0,0 +1,87 @@
"""认证和安全模块."""
import time
from collections import defaultdict
from typing import Dict, List
from fastapi import HTTPException, Request, status
from .config import get_config
class RateLimiter:
"""简单的速率限制器."""
def __init__(self):
self.requests: Dict[str, List[float]] = defaultdict(list)
def is_allowed(self, client_ip: str, limit: int, window: int = 60) -> bool:
"""检查是否允许请求."""
now = time.time()
# 清理过期的请求记录
self.requests[client_ip] = [
req_time for req_time in self.requests[client_ip]
if now - req_time < window
]
# 检查是否超过限制
if len(self.requests[client_ip]) >= limit:
return False
# 记录当前请求
self.requests[client_ip].append(now)
return True
# 全局速率限制器实例
rate_limiter = RateLimiter()
def verify_api_key(api_key: str) -> bool:
"""验证 API 密钥."""
config = get_config()
return api_key == config.server.api_key
def check_ip_allowed(client_ip: str) -> bool:
"""检查 IP 是否被允许."""
config = get_config()
allowed_ips = config.security.allowed_ips
# 如果没有配置允许的 IP则允许所有
if not allowed_ips:
return True
return client_ip in allowed_ips
def check_rate_limit(client_ip: str) -> bool:
"""检查速率限制."""
config = get_config()
return rate_limiter.is_allowed(client_ip, config.security.rate_limit)
def authenticate_request(request: Request, api_key: str) -> None:
"""验证请求."""
client_ip = request.client.host if request.client else "unknown"
# 检查 IP 白名单
if not check_ip_allowed(client_ip):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="IP 地址不在允许列表中"
)
# 检查速率限制
if not check_rate_limit(client_ip):
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="请求过于频繁,请稍后再试"
)
# 验证 API 密钥
if not verify_api_key(api_key):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无效的 API 密钥"
)

93
sms_forwarder/config.py Normal file
View File

@ -0,0 +1,93 @@
"""配置管理模块."""
import os
from pathlib import Path
from typing import List, Optional
import yaml
from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings
class ServerConfig(BaseModel):
"""服务器配置."""
host: str = "0.0.0.0"
port: int = 8000
api_key: str = Field(..., description="API 密钥")
class LoggingConfig(BaseModel):
"""日志配置."""
level: str = "INFO"
file: Optional[str] = None
class NotificationService(BaseModel):
"""通知服务配置."""
name: str
url: str
enabled: bool = True
class NotificationTemplate(BaseModel):
"""通知模板."""
title: str
body: str
class NotificationTemplates(BaseModel):
"""通知模板集合."""
sms: NotificationTemplate
system: NotificationTemplate
class NotificationConfig(BaseModel):
"""通知配置."""
services: List[NotificationService]
templates: NotificationTemplates
class SecurityConfig(BaseModel):
"""安全配置."""
allowed_ips: List[str] = []
rate_limit: int = 60
class Config(BaseSettings):
"""应用配置."""
server: ServerConfig
logging: LoggingConfig
notifications: NotificationConfig
security: SecurityConfig
@classmethod
def load_from_file(cls, config_path: str = "config.yaml") -> "Config":
"""从 YAML 文件加载配置."""
config_file = Path(config_path)
if not config_file.exists():
raise FileNotFoundError(f"配置文件不存在: {config_path}")
with open(config_file, "r", encoding="utf-8") as f:
config_data = yaml.safe_load(f)
return cls(**config_data)
# 全局配置实例
_config: Optional[Config] = None
def get_config() -> Config:
"""获取全局配置实例."""
global _config
if _config is None:
config_path = os.getenv("CONFIG_PATH", "config.yaml")
_config = Config.load_from_file(config_path)
return _config
def reload_config(config_path: str = "config.yaml") -> Config:
"""重新加载配置."""
global _config
_config = Config.load_from_file(config_path)
return _config

223
sms_forwarder/main.py Normal file
View File

@ -0,0 +1,223 @@
"""主应用模块."""
import logging
import time
from datetime import datetime
from pathlib import Path
import uvicorn
from fastapi import FastAPI, HTTPException, Request, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from . import __version__
from .auth import authenticate_request
from .config import get_config
from .models import NotificationRequest, NotificationResponse, SimpleNotificationRequest, SMSMessage, SystemStatus
from .notifier import notification_manager
# 配置日志
def setup_logging():
"""设置日志配置."""
try:
config = get_config()
log_level = getattr(logging, config.logging.level.upper(), logging.INFO)
# 创建日志目录
if config.logging.file:
log_file = Path(config.logging.file)
log_file.parent.mkdir(parents=True, exist_ok=True)
# 配置日志格式
logging.basicConfig(
level=log_level,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[
logging.StreamHandler(),
logging.FileHandler(config.logging.file) if config.logging.file else logging.NullHandler()
]
)
except Exception as e:
# 如果配置加载失败,使用默认日志配置
logging.basicConfig(level=logging.INFO)
logging.error(f"日志配置失败,使用默认配置: {e}")
setup_logging()
logger = logging.getLogger(__name__)
# 应用启动时间
start_time = time.time()
# 创建 FastAPI 应用
app = FastAPI(
title="SMS Forwarder",
description="iOS SMS to Android notification forwarder",
version=__version__,
docs_url="/docs",
redoc_url="/redoc"
)
# 添加 CORS 中间件
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
"""全局异常处理器."""
logger.error(f"未处理的异常: {exc}", exc_info=True)
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={"detail": "服务器内部错误"}
)
@app.get("/")
async def root():
"""根路径."""
return {
"message": "SMS Forwarder API",
"version": __version__,
"docs": "/docs"
}
@app.get("/health")
async def health_check():
"""健康检查."""
return {"status": "healthy", "timestamp": datetime.now()}
@app.get("/status", response_model=SystemStatus)
async def get_status():
"""获取系统状态."""
stats = notification_manager.get_stats()
return SystemStatus(
version=__version__,
uptime=time.time() - start_time,
notifications_sent=stats["sent"],
notifications_failed=stats["failed"]
)
@app.post("/notify", response_model=NotificationResponse)
async def send_notification(request: Request, notification_request: NotificationRequest):
"""发送通知(完整版)."""
try:
# 验证请求
authenticate_request(request, notification_request.api_key)
# 发送通知
success = notification_manager.send_sms_notification(notification_request.message)
if success:
logger.info(f"通知发送成功: {notification_request.message.sender}")
return NotificationResponse(
success=True,
message="通知发送成功"
)
else:
logger.error(f"通知发送失败: {notification_request.message.sender}")
return NotificationResponse(
success=False,
message="通知发送失败"
)
except HTTPException:
raise
except Exception as e:
logger.error(f"处理通知请求时出错: {e}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="处理请求时出错"
)
@app.post("/notify/simple", response_model=NotificationResponse)
async def send_simple_notification(request: Request, simple_request: SimpleNotificationRequest):
"""发送通知(简化版)- 只需要内容,时间戳自动生成."""
try:
# 验证请求
authenticate_request(request, simple_request.api_key)
# 创建 SMS 消息对象,时间戳自动生成
sms_message = SMSMessage(
content=simple_request.content,
sender=simple_request.sender or "iPhone"
)
# 发送通知
success = notification_manager.send_sms_notification(sms_message)
if success:
logger.info(f"简化通知发送成功: {sms_message.sender}")
return NotificationResponse(
success=True,
message="通知发送成功"
)
else:
logger.error(f"简化通知发送失败: {sms_message.sender}")
return NotificationResponse(
success=False,
message="通知发送失败"
)
except HTTPException:
raise
except Exception as e:
logger.error(f"处理简化通知请求时出错: {e}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="处理请求时出错"
)
@app.post("/reload")
async def reload_config(request: Request, api_key: str):
"""重新加载配置."""
try:
# 验证 API 密钥
authenticate_request(request, api_key)
# 重新加载通知服务
notification_manager.reload_services()
return {"message": "配置重新加载成功"}
except HTTPException:
raise
except Exception as e:
logger.error(f"重新加载配置时出错: {e}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="重新加载配置失败"
)
def main():
"""主函数."""
try:
config = get_config()
logger.info(f"启动 SMS Forwarder v{__version__}")
logger.info(f"服务器配置: {config.server.host}:{config.server.port}")
# 启动服务器
uvicorn.run(
"sms_forwarder.main:app",
host=config.server.host,
port=config.server.port,
log_level=config.logging.level.lower(),
access_log=True
)
except Exception as e:
logger.error(f"启动服务器失败: {e}", exc_info=True)
raise
if __name__ == "__main__":
main()

41
sms_forwarder/models.py Normal file
View File

@ -0,0 +1,41 @@
"""数据模型."""
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field
class SMSMessage(BaseModel):
"""短信消息模型."""
sender: Optional[str] = Field(default="未知发送者", description="发送者手机号或名称")
content: str = Field(..., description="短信内容")
timestamp: Optional[datetime] = Field(default_factory=datetime.now, description="接收时间")
class NotificationRequest(BaseModel):
"""通知请求模型."""
api_key: str = Field(..., description="API 密钥")
message: SMSMessage = Field(..., description="短信消息")
class SimpleNotificationRequest(BaseModel):
"""简化的通知请求模型 - 只需要内容."""
api_key: str = Field(..., description="API 密钥")
content: str = Field(..., description="短信内容")
sender: Optional[str] = Field(default=None, description="发送者(可选)")
class NotificationResponse(BaseModel):
"""通知响应模型."""
success: bool = Field(..., description="是否成功")
message: str = Field(..., description="响应消息")
timestamp: datetime = Field(default_factory=datetime.now, description="响应时间")
class SystemStatus(BaseModel):
"""系统状态模型."""
version: str
uptime: float
notifications_sent: int
notifications_failed: int

127
sms_forwarder/notifier.py Normal file
View File

@ -0,0 +1,127 @@
"""通知发送模块."""
import logging
from datetime import datetime
from typing import Dict, List
import apprise
from .config import get_config
from .models import SMSMessage
logger = logging.getLogger(__name__)
class NotificationManager:
"""通知管理器."""
def __init__(self):
self.apprise_obj = apprise.Apprise()
self.stats = {
"sent": 0,
"failed": 0
}
self._load_services()
def _load_services(self) -> None:
"""加载通知服务."""
config = get_config()
# 清空现有服务
self.apprise_obj.clear()
# 添加启用的服务
for service in config.notifications.services:
if service.enabled:
try:
self.apprise_obj.add(service.url)
logger.info(f"已加载通知服务: {service.name}")
except Exception as e:
logger.error(f"加载通知服务失败 {service.name}: {e}")
if len(self.apprise_obj) == 0:
logger.warning("没有可用的通知服务")
def reload_services(self) -> None:
"""重新加载通知服务."""
logger.info("重新加载通知服务...")
self._load_services()
def send_sms_notification(self, sms: SMSMessage) -> bool:
"""发送短信通知."""
config = get_config()
template = config.notifications.templates.sms
# 格式化通知内容
title = template.title.format(
sender=sms.sender,
timestamp=sms.timestamp.strftime("%Y-%m-%d %H:%M:%S") if sms.timestamp else "未知"
)
body = template.body.format(
sender=sms.sender,
content=sms.content,
timestamp=sms.timestamp.strftime("%Y-%m-%d %H:%M:%S") if sms.timestamp else "未知"
)
return self._send_notification(title, body)
def send_system_notification(self, message: str) -> bool:
"""发送系统通知."""
config = get_config()
template = config.notifications.templates.system
title = template.title
body = template.body.format(message=message)
return self._send_notification(title, body)
def _send_notification(self, title: str, body: str) -> bool:
"""发送通知."""
try:
if len(self.apprise_obj) == 0:
logger.error("没有可用的通知服务")
self.stats["failed"] += 1
return False
# 发送通知
result = self.apprise_obj.notify(
title=title,
body=body
)
if result:
logger.info(f"通知发送成功: {title}")
self.stats["sent"] += 1
return True
else:
logger.error(f"通知发送失败: {title}")
self.stats["failed"] += 1
return False
except Exception as e:
logger.error(f"发送通知时出错: {e}")
self.stats["failed"] += 1
return False
def get_stats(self) -> Dict[str, int]:
"""获取统计信息."""
return self.stats.copy()
def get_services_info(self) -> List[Dict[str, str]]:
"""获取服务信息."""
config = get_config()
services_info = []
for service in config.notifications.services:
services_info.append({
"name": service.name,
"enabled": service.enabled,
"url_scheme": service.url.split("://")[0] if "://" in service.url else "unknown"
})
return services_info
# 全局通知管理器实例
notification_manager = NotificationManager()

1
tests/__init__.py Normal file
View File

@ -0,0 +1 @@
"""测试包."""

57
tests/test_main.py Normal file
View File

@ -0,0 +1,57 @@
"""主应用测试."""
import pytest
from fastapi.testclient import TestClient
from sms_forwarder.main import app
client = TestClient(app)
def test_root():
"""测试根路径."""
response = client.get("/")
assert response.status_code == 200
data = response.json()
assert "message" in data
assert "version" in data
def test_health_check():
"""测试健康检查."""
response = client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert "timestamp" in data
def test_status():
"""测试状态接口."""
response = client.get("/status")
assert response.status_code == 200
data = response.json()
assert "version" in data
assert "uptime" in data
assert "notifications_sent" in data
assert "notifications_failed" in data
def test_notify_without_auth():
"""测试未认证的通知请求."""
response = client.post("/notify", json={
"api_key": "invalid-key",
"message": {
"sender": "test",
"content": "test message"
}
})
assert response.status_code == 401
def test_notify_invalid_data():
"""测试无效数据的通知请求."""
response = client.post("/notify", json={
"invalid": "data"
})
assert response.status_code == 422

105
tests/test_notifier.py Normal file
View File

@ -0,0 +1,105 @@
"""通知模块测试."""
import pytest
from unittest.mock import MagicMock, patch
from sms_forwarder.models import SMSMessage
from sms_forwarder.notifier import NotificationManager
@pytest.fixture
def notification_manager():
"""通知管理器测试夹具."""
with patch("sms_forwarder.notifier.get_config") as mock_get_config:
# 模拟配置
mock_config = MagicMock()
mock_config.notifications.services = [
MagicMock(name="test", url="test://test", enabled=True)
]
mock_config.notifications.templates.sms.title = "SMS: {sender}"
mock_config.notifications.templates.sms.body = "From: {sender}\nContent: {content}\nTime: {timestamp}"
mock_config.notifications.templates.system.title = "System"
mock_config.notifications.templates.system.body = "{message}"
mock_get_config.return_value = mock_config
manager = NotificationManager()
# 模拟 Apprise 对象
manager.apprise_obj = MagicMock()
manager.apprise_obj.__len__.return_value = 1
yield manager
def test_send_sms_notification_success(notification_manager):
"""测试成功发送短信通知."""
notification_manager.apprise_obj.notify.return_value = True
sms = SMSMessage(
sender="Test Sender",
content="Test Message"
)
result = notification_manager.send_sms_notification(sms)
assert result is True
assert notification_manager.stats["sent"] == 1
assert notification_manager.stats["failed"] == 0
notification_manager.apprise_obj.notify.assert_called_once()
def test_send_sms_notification_failure(notification_manager):
"""测试发送短信通知失败."""
notification_manager.apprise_obj.notify.return_value = False
sms = SMSMessage(
sender="Test Sender",
content="Test Message"
)
result = notification_manager.send_sms_notification(sms)
assert result is False
assert notification_manager.stats["sent"] == 0
assert notification_manager.stats["failed"] == 1
notification_manager.apprise_obj.notify.assert_called_once()
def test_send_system_notification(notification_manager):
"""测试发送系统通知."""
notification_manager.apprise_obj.notify.return_value = True
result = notification_manager.send_system_notification("Test System Message")
assert result is True
assert notification_manager.stats["sent"] == 1
notification_manager.apprise_obj.notify.assert_called_once()
def test_no_services_available(notification_manager):
"""测试没有可用服务时的行为."""
# 模拟没有可用服务
notification_manager.apprise_obj.__len__.return_value = 0
sms = SMSMessage(
sender="Test Sender",
content="Test Message"
)
result = notification_manager.send_sms_notification(sms)
assert result is False
assert notification_manager.stats["failed"] == 1
# 应该不会调用 notify
notification_manager.apprise_obj.notify.assert_not_called()
def test_get_stats(notification_manager):
"""测试获取统计信息."""
notification_manager.stats = {"sent": 5, "failed": 2}
stats = notification_manager.get_stats()
assert stats == {"sent": 5, "failed": 2}
# 确保返回的是副本
assert stats is not notification_manager.stats

1003
uv.lock generated Normal file

File diff suppressed because it is too large Load Diff