feat!: replace legacy WebUI with unified FastAPI architecture #major (#267)

* feat!: replace legacy WebUI with unified FastAPI architecture #major

- Remove legacy WebUI (web/ package based on ThreadingHTTPServer)
- Unify CLI commands: --webui now seamlessly launches the new FastAPI service
- Migrate task service from web/services.py to src/services/task_service.py
- Update documentation to reflect the new simplified usage

* refactor(service): support parameterized query source in TaskService
This commit is contained in:
mumu
2026-02-06 22:06:01 +08:00
committed by GitHub
parent 97a2c2d205
commit 52917baa02
23 changed files with 250 additions and 2497 deletions

View File

@@ -194,40 +194,31 @@ python main.py
> 📖 完整环境变量、定时任务配置请参考 [完整配置指南](docs/full-guide.md)
## 🖥️ 本地 WebUI可选 - 将在后续的版本弃用)
```bash
python main.py --webui # 启动 WebUI + 执行分析
python main.py --webui-only # 仅启动 WebUI
```
访问 `http://127.0.0.1:8000` 可进行配置管理、触发分析、查看任务状态。
> 详细说明请参考 [完整指南 - WebUI](docs/full-guide.md#本地-webui-管理界面)
## 🧩 FastAPI Web 服务(可选)
## 🖥️ Web 界面
![img.png](sources/fastapi_server.png)
```bash
cd ./apps/dsa-web # 进入 React Web 目录
npm install
npm run build # 编译 React Web 页面 会在根目录生成 /static 文件夹
包含完整的配置管理、任务监控和手动分析功能。
cd ../.. # 返回项目根目录
python main.py --serve # 启动 FastAPI + 执行分析
python main.py --serve-only # 仅启动 FastAPI
python main.py --serve-only --host 0.0.0.0 --port 8000 # 指定启动端口
### 启动方式
1. **编译前端** (首次运行需要)
```bash
cd ./apps/dsa-web
npm install && npm run build
cd ../..
```
访问 `http://127.0.0.1:8000` 即可使用该页面(注意一定要执行 `npm install` 步骤,否则没有页面)
也可以使用下面命令单独启动:
2. **启动服务**
```bash
uvicorn server:app --reload --host 0.0.0.0 --port 8000
python main.py --webui # 启动 Web 界面 + 执行定时分析
python main.py --webui-only # 仅启动 Web 界面
```
访问 `http://127.0.0.1:8000` 即可使用。
> 也可以使用 `python main.py --serve` (等效命令)
## 🗺️ Roadmap
查看已支持的功能和未来规划:[更新日志](docs/CHANGELOG.md)

View File

@@ -24,7 +24,7 @@ class AnalyzeCommand(BotCommand):
分析指定股票代码,生成 AI 分析报告并推送。
用法:
/analyze 600519 - 分析贵州茅台
/analyze 600519 - 分析贵州茅台(精简报告)
/analyze 600519 full - 分析并生成完整报告
"""
@@ -68,18 +68,18 @@ class AnalyzeCommand(BotCommand):
"""执行分析命令"""
code = args[0].lower()
# 检查是否需要完整报告
# 检查是否需要完整报告(默认精简,传 full/完整/详细 切换)
report_type = "simple"
if len(args) > 1 and args[1].lower() in ["full", "完整", "详细"]:
report_type = "full"
# if len(args) > 1 and args[1].lower() in ["full", "完整", "详细"]:
# report_type = "full"
logger.info(f"[AnalyzeCommand] 分析股票: {code}, 报告类型: {report_type}")
try:
# 调用分析服务
from web.services import get_analysis_service
from src.services.task_service import get_task_service
from src.enums import ReportType
service = get_analysis_service()
service = get_task_service()
# 提交异步分析任务
result = service.submit_analysis(

View File

@@ -38,7 +38,6 @@ RUN pip install --no-cache-dir -r requirements.txt
COPY *.py ./
COPY api/ ./api/
COPY data_provider/ ./data_provider/
COPY web/ ./web/
COPY bot/ ./bot/
COPY src/ ./src/
COPY --from=web-builder /app/static ./static/
@@ -50,17 +49,17 @@ RUN mkdir -p /app/data /app/logs /app/reports
ENV PYTHONUNBUFFERED=1
ENV LOG_DIR=/app/logs
ENV DATABASE_PATH=/app/data/stock_analysis.db
# WebUI应用
ENV WEBUI_HOST=0.0.0.0
ENV WEBUI_PORT=8000
# API 服务
ENV API_HOST=0.0.0.0
ENV API_PORT=8000
# 暴露 WebUI 端口
# 暴露 API 端口
EXPOSE 8000
# 数据卷(持久化数据)
VOLUME ["/app/data", "/app/logs", "/app/reports"]
# 健康检查(支持 WebUI / FastAPI 模式)
# 健康检查FastAPI 模式)
HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \
CMD curl -f http://localhost:8000/api/health || curl -f http://localhost:8000/health \
|| python -c "import sys; sys.exit(0)"

View File

@@ -4,9 +4,8 @@
#
# 使用方式:
# 定时模式: docker-compose -f ./docker/docker-compose.yml up -d
# WebUI模式: docker-compose -f ./docker/docker-compose.yml up -d webui
# FastAPI模式: docker-compose -f ./docker/docker-compose.yml up -d server
# 同时启动: docker-compose -f ./docker/docker-compose.yml up -d analyzer webui
# 同时启动: docker-compose -f ./docker/docker-compose.yml up -d analyzer server
version: '3.8'
@@ -31,10 +30,9 @@ x-common: &common
environment:
- TZ=Asia/Shanghai
# 注意:容器内如果绑定到 127.0.0.1宿主机端口映射将无法访问 WebUI。
# 即使 .env 里设置了 WEBUI_HOST=127.0.0.1这里也会强制覆盖。
- WEBUI_HOST=0.0.0.0
# WEBUI_PORT 从 .env 文件读取,无需在此硬编码
# API 服务绑定地址(容器内需要绑定到 0.0.0.0
- API_HOST=0.0.0.0
# API_PORT 从 .env 文件读取,无需在此硬编码
# 代理设置(如果需要)
# - http_proxy=http://host.docker.internal:10809
@@ -58,14 +56,6 @@ services:
<<: *common
container_name: stock-analyzer
# WebUI 模式
webui:
<<: *common
container_name: stock-webui
command: ["python", "main.py", "--webui-only"]
ports:
- "${WEBUI_PORT:-8000}:${WEBUI_PORT:-8000}"
# FastAPI 模式
server:
<<: *common

View File

@@ -5,6 +5,23 @@
格式基于 [Keep a Changelog](https://keepachangelog.com/zh-CN/1.0.0/)
版本号遵循 [Semantic Versioning](https://semver.org/lang/zh-CN/)。
## [3.0.0] - 2026-02-06
### 移除
- 🗑️ **移除旧版 WebUI**
- 删除基于 `http.server.ThreadingHTTPServer` 的旧版 WebUI`web/` 包)
- 旧版 WebUI 的功能已完全被 FastAPI`api/`+ React 前端替代
- `--webui` / `--webui-only` 命令行参数标记为弃用,自动重定向到 `--serve` / `--serve-only`
- `WEBUI_ENABLED` / `WEBUI_HOST` / `WEBUI_PORT` 环境变量保持兼容,自动转发到 FastAPI 服务
- `webui.py` 保留为兼容入口,启动时直接调用 FastAPI 后端
- Docker Compose 中移除 `webui` 服务定义,统一使用 `server` 服务
### 变更
- ♻️ **服务层重构**
-`web/services.py` 中的异步任务服务迁移至 `src/services/task_service.py`
- Bot 分析命令(`bot/commands/analyze.py`)改为使用 `src.services.task_service`
- Docker 环境变量 `WEBUI_HOST`/`WEBUI_PORT` 更名为 `API_HOST`/`API_PORT`(旧名仍兼容)
## [2.3.0] - 2026-02-01
### 新增

View File

@@ -210,10 +210,10 @@ OPENAI_MODEL=deepseek-chat
---
### Q14: Docker 中 WebUI 无法访问?
### Q14: Docker 中 API 服务无法访问?
**解决方案**
1. 确保 `WEBUI_HOST=0.0.0.0`(不能是 127.0.0.1
1. 确保启动命令包含 `--host 0.0.0.0`(不能是 127.0.0.1
2. 检查端口映射是否正确:
```yaml
ports:

View File

@@ -210,10 +210,10 @@ Supported model services:
---
### Q14: WebUI inaccessible in Docker?
### Q14: API service inaccessible in Docker?
**Solution**:
1. Ensure `WEBUI_HOST=0.0.0.0` (cannot be 127.0.0.1)
1. Ensure startup command includes `--host 0.0.0.0` (cannot be 127.0.0.1)
2. Check port mapping is correct:
```yaml
ports:

View File

@@ -184,51 +184,50 @@
> 📖 完整環境變量、定時任務配置請參考 [完整配置指南](full-guide.md)
## 🖥️ 本地 WebUI(可選)
## 🧩 FastAPI Web 服務(可選)
本地運行時,可啟用 WebUI 來管理配置和觸發分析。
本地運行時,可啟用 FastAPI 服務來管理配置和觸發分析。
### 啟動方式
| 命令 | 說明 |
|------|------|
| `python main.py --webui` | 啟動 WebUI + 執行一次完整分析 |
| `python main.py --webui-only` | 僅啟動 WebUI,手動觸發分析 |
| `python main.py --serve` | 啟動 API 服務 + 執行一次完整分析 |
| `python main.py --serve-only` | 僅啟動 API 服務,手動觸發分析 |
- 訪問地址:`http://127.0.0.1:8000`
- 詳細說明請參考 [配置指南 - WebUI](full-guide.md#本地-webui-管理界面)
- API 文檔:`http://127.0.0.1:8000/docs`
### 功能特性
- 📝 **配置管理** - 查看/修改 `.env` 里的自選股列表
- 🚀 **快速分析** - 頁面輸入股票代碼,一鍵觸發分析
- 📝 **配置管理** - 查看/修改自選股列表
- 🚀 **快速分析** - 通過 API 接口觸發分析
- 📊 **實時進度** - 分析任務狀態實時更新,支持多任務並行
### API 接口
| 接口 | 方法 | 說明 |
|------|------|------|
| `/` | GET | 配置管理頁面 |
| `/health` | GET | 健康檢查 |
| `/analysis?code=xxx` | GET | 觸發單隻股票異步分析 |
| `/analysis/history` | GET | 查詢分析歷史記錄 |
| `/tasks` | GET | 查詢所有任務狀態 |
| `/task?id=xxx` | GET | 查詢單個任務狀態 |
| `/api/v1/analysis/analyze` | POST | 觸發股票分析 |
| `/api/v1/analysis/tasks` | GET | 查詢任務列表 |
| `/api/v1/analysis/status/{task_id}` | GET | 查詢任務狀態 |
| `/api/v1/history` | GET | 查詢分析歷史記錄 |
| `/api/health` | GET | 健康檢查 |
## 項目結構
```
daily_stock_analysis/
├── main.py # 主程序入口
├── webui.py # WebUI 入口
├── server.py # FastAPI 服務入口
├── src/ # 核心業務代碼
│ ├── analyzer.py # AI 分析器Gemini
│ ├── config.py # 配置管理
│ ├── notification.py # 消息推送
│ ├── storage.py # 數據存儲
│ └── ...
├── api/ # FastAPI API 模塊
├── bot/ # 機器人模塊
├── web/ # WebUI 模塊
├── data_provider/ # 數據源適配器
├── docker/ # Docker 配置
│ ├── Dockerfile

View File

@@ -384,16 +384,17 @@ DEBUG=false # Enable debug logging
---
## 🖥️ Local WebUI (Optional)
## 🧩 FastAPI Web Service (Optional)
```bash
python main.py --webui # Start WebUI + run analysis
python main.py --webui-only # Start WebUI only
python main.py --serve # Start API service + run analysis
python main.py --serve-only # Start API service only
```
Visit `http://127.0.0.1:8000` for configuration management, triggering analysis, and viewing task status.
API documentation available at `http://127.0.0.1:8000/docs`.
> For detailed instructions, see [Full Guide - WebUI](full-guide_EN.md#local-webui-management-interface)
> For detailed instructions, see [Full Guide - API Service](full-guide_EN.md#fastapi-api-service)
---

View File

@@ -197,7 +197,7 @@ class CommandDispatcher:
## 五、Webhook 路由
在 [web/router.py](../web/router.py) 中注册路由:
在 [api/v1/router.py](../api/v1/router.py) 中注册路由:
```python
# Webhook 路由

View File

@@ -74,8 +74,6 @@ Dockerfile 已采用多阶段构建,前端会在镜像构建时自动打包。
| 模式 | 启动命令 | 描述 |
|------|----------|------|
| 定时任务模式(默认) | `python main.py --schedule` | 按计划执行股票分析 |
| WebUI 模式 | `python main.py --webui` | 启动 WebUI旧版和定时任务 |
| 仅 WebUI 模式 | `python main.py --webui-only` | 仅启动 WebUI不执行定时任务 |
| FastAPI 模式 | `python main.py --serve` | 启动 FastAPI 并执行分析 |
| 仅 FastAPI 模式 | `python main.py --serve-only` | 仅启动 FastAPI不执行分析 |
| 仅大盘复盘 | `python main.py --market-review` | 仅执行大盘复盘分析 |
@@ -86,8 +84,6 @@ Dockerfile 已采用多阶段构建,前端会在镜像构建时自动打包。
2. 点击「设置」
3. 找到「启动命令」配置项
4. 输入你需要的启动命令,例如:
- 启动 WebUI`python main.py --webui`
- 仅启动 WebUI`python main.py --webui-only`
- 启动 FastAPI`python main.py --serve`
- 仅启动 FastAPI`python main.py --serve-only --host 0.0.0.0 --port 8000`
- 启动定时任务:`python main.py --schedule`
@@ -135,13 +131,14 @@ Dockerfile 已采用多阶段构建,前端会在镜像构建时自动打包。
| `LOG_DIR` | 日志目录 | `/app/logs` |
| `DATABASE_PATH` | 数据库路径 | `/app/data/stock_analysis.db` |
### 5.2 WebUI 配置
### 5.2 API 服务配置
| 变量名 | 说明 | 默认值 |
|--------|------|--------|
| `WEBUI_HOST` | WebUI 监听地址 | `0.0.0.0` |
| `WEBUI_PORT` | WebUI 端口 | `8000` |
| `WEBUI_ENABLED` | 启用 WebUI | `false` |
| `API_HOST` | API 服务监听地址 | `0.0.0.0` |
| `API_PORT` | API 服务端口 | `8000` |
> 旧版 `WEBUI_HOST`/`WEBUI_PORT`/`WEBUI_ENABLED` 环境变量仍兼容,会自动转发到 API 服务。
### 5.3 分析相关配置
@@ -208,10 +205,9 @@ HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \
## 8. 常见问题
### 8.1 WebUI 无法访问
### 8.1 API 服务无法访问
- 检查启动命令是否包含 `--webui``--webui-only` 参数
- 检查环境变量 `WEBUI_ENABLED` 是否设置为 `true`
- 检查启动命令是否包含 `--serve``--serve-only` 参数
- 检查「访问」标签页是否已配置域名
- 检查防火墙设置
@@ -239,7 +235,7 @@ HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \
你可以在 Zeabur 上部署多个实例,用于不同的功能:
1. 一个实例用于 WebUI`python main.py --webui-only`
1. 一个实例用于 API 服务`python main.py --serve-only`
2. 一个实例用于定时任务(`python main.py --schedule`
3. 一个实例用于机器人(`python main.py --discord-bot`

View File

@@ -16,7 +16,8 @@ daily_stock_analysis/
│ └── ...
├── data_provider/ # 多数据源适配器
├── bot/ # 机器人交互模块
├── web/ # WebUI 模块
├── api/ # FastAPI 后端服务
├── apps/dsa-web/ # React 前端
├── docker/ # Docker 配置
├── docs/ # 项目文档
└── .github/workflows/ # GitHub Actions
@@ -281,14 +282,6 @@ services:
<<: *common
container_name: stock-analyzer
# WebUI 模式
webui:
<<: *common
container_name: stock-webui
command: ["python", "main.py", "--webui-only"]
ports:
- "8000:8000"
# FastAPI 模式
server:
<<: *common
@@ -305,7 +298,6 @@ services:
docker-compose -f ./docker/docker-compose.yml ps
# 查看日志
docker-compose -f ./docker/docker-compose.yml logs -f webui
docker-compose -f ./docker/docker-compose.yml logs -f server
# 停止服务
@@ -313,14 +305,13 @@ docker-compose -f ./docker/docker-compose.yml down
# 重建镜像(代码更新后)
docker-compose -f ./docker/docker-compose.yml build --no-cache
docker-compose -f ./docker/docker-compose.yml up -d webui
docker-compose -f ./docker/docker-compose.yml up -d server
```
### 手动构建镜像
```bash
docker build -f docker/Dockerfile -t stock-analysis .
docker run -d --env-file .env -p 8000:8000 -v ./data:/app/data stock-analysis python main.py --webui-only
docker run -d --env-file .env -p 8000:8000 -v ./data:/app/data stock-analysis python main.py --serve-only --host 0.0.0.0 --port 8000
```
@@ -544,62 +535,55 @@ python main.py --debug
---
## 本地 WebUI 管理界面
## FastAPI API 服务
WebUI 提供配置管理和快速分析功能,支持页面触发单只股票分析。
FastAPI 提供 RESTful API 服务,支持配置管理和触发分析。
### 启动方式
| 命令 | 说明 |
|------|------|
| `python main.py --webui` | 启动 WebUI + 执行一次完整分析 |
| `python main.py --webui-only` | 仅启动 WebUI,手动触发分析 |
**永久启用**:在 `.env` 中设置:
```env
WEBUI_ENABLED=true
```
| `python main.py --serve` | 启动 API 服务 + 执行一次完整分析 |
| `python main.py --serve-only` | 仅启动 API 服务,手动触发分析 |
### 功能特性
- 📝 **配置管理** - 查看/修改 `.env` 里的自选股列表
- 🚀 **快速分析** - 页面输入股票代码,一键触发分析
- 📝 **配置管理** - 查看/修改自选股列表
- 🚀 **快速分析** - 通过 API 接口触发分析
- 📊 **实时进度** - 分析任务状态实时更新,支持多任务并行
- 🔗 **API 接口** - 支持程序化调用
- 🔗 **API 文档** - 访问 `/docs` 查看 Swagger UI
### API 接口
| 接口 | 方法 | 说明 |
|------|------|------|
| `/` | GET | 配置管理页面 |
| `/health` | GET | 健康检查 |
| `/analysis?code=xxx` | GET | 触发单只股票异步分析 |
| `/analysis/history` | GET | 查询分析历史记录 |
| `/tasks` | GET | 查询所有任务状态 |
| `/task?id=xxx` | GET | 查询单个任务状态 |
| `/api/v1/analysis/analyze` | POST | 触发股票分析 |
| `/api/v1/analysis/tasks` | GET | 查询任务列表 |
| `/api/v1/analysis/status/{task_id}` | GET | 查询任务状态 |
| `/api/v1/history` | GET | 查询分析历史 |
| `/api/health` | GET | 健康检查 |
| `/docs` | GET | API Swagger 文档 |
**调用示例**
```bash
# 健康检查
curl http://127.0.0.1:8000/health
curl http://127.0.0.1:8000/api/health
# 触发分析A股
curl "http://127.0.0.1:8000/analysis?code=600519"
# 触发分析(港股)
curl "http://127.0.0.1:8000/analysis?code=hk00700"
curl -X POST http://127.0.0.1:8000/api/v1/analysis/analyze \
-H 'Content-Type: application/json' \
-d '{"stock_code": "600519"}'
# 查询任务状态
curl "http://127.0.0.1:8000/task?id=<task_id>"
curl http://127.0.0.1:8000/api/v1/analysis/status/<task_id>
```
### 自定义配置
修改默认端口或允许局域网访问:
```env
WEBUI_HOST=0.0.0.0 # 默认 127.0.0.1
WEBUI_PORT=8888 # 默认 8000
```bash
python main.py --serve-only --host 0.0.0.0 --port 8888
```
### 支持的股票代码格式

View File

@@ -16,7 +16,8 @@ daily_stock_analysis/
│ └── ...
├── data_provider/ # Multi-source data adapters
├── bot/ # Bot interaction module
├── web/ # WebUI module
├── api/ # FastAPI backend service
├── apps/dsa-web/ # React frontend
├── docker/ # Docker configuration
├── docs/ # Project documentation
└── .github/workflows/ # GitHub Actions
@@ -270,11 +271,11 @@ services:
<<: *common
container_name: stock-analyzer
# WebUI mode
webui:
# FastAPI mode
server:
<<: *common
container_name: stock-webui
command: ["python", "main.py", "--webui-only"]
container_name: stock-server
command: ["python", "main.py", "--serve-only", "--host", "0.0.0.0", "--port", "8000"]
ports:
- "8000:8000"
```
@@ -286,21 +287,21 @@ services:
docker-compose -f ./docker/docker-compose.yml ps
# View logs
docker-compose -f ./docker/docker-compose.yml logs -f webui
docker-compose -f ./docker/docker-compose.yml logs -f server
# Stop services
docker-compose -f ./docker/docker-compose.yml down
# Rebuild image (after code update)
docker-compose -f ./docker/docker-compose.yml build --no-cache
docker-compose -f ./docker/docker-compose.yml up -d webui
docker-compose -f ./docker/docker-compose.yml up -d server
```
### Manual Image Build
```bash
docker build -t stock-analysis .
docker run -d --env-file .env -p 8000:8000 -v ./data:/app/data stock-analysis python main.py --webui-only
docker run -d --env-file .env -p 8000:8000 -v ./data:/app/data stock-analysis python main.py --serve-only --host 0.0.0.0 --port 8000
```
---
@@ -523,62 +524,55 @@ Log file locations:
---
## Local WebUI Management Interface
## FastAPI API Service
WebUI provides configuration management and quick analysis features, supporting single stock analysis triggered from the page.
FastAPI provides RESTful API service for configuration management and triggering analysis.
### Startup Methods
| Command | Description |
|------|------|
| `python main.py --webui` | Start WebUI + run full analysis once |
| `python main.py --webui-only` | Start WebUI only, manually trigger analysis |
**Permanently enable**: Set in `.env`:
```env
WEBUI_ENABLED=true
```
| `python main.py --serve` | Start API service + run full analysis once |
| `python main.py --serve-only` | Start API service only, manually trigger analysis |
### Features
- **Configuration Management** - View/modify watchlist in `.env`
- **Quick Analysis** - Enter stock code on page, one-click trigger analysis
- **Configuration Management** - View/modify watchlist
- **Quick Analysis** - Trigger analysis via API
- **Real-time Progress** - Analysis task status updates in real-time, supports parallel tasks
- **API Interface** - Supports programmatic calls
- **API Documentation** - Visit `/docs` for Swagger UI
### API Endpoints
| Endpoint | Method | Description |
|------|------|------|
| `/` | GET | Configuration management page |
| `/health` | GET | Health check |
| `/analysis?code=xxx` | GET | Trigger async analysis for single stock |
| `/analysis/history` | GET | Query analysis history |
| `/tasks` | GET | Query all task statuses |
| `/task?id=xxx` | GET | Query single task status |
| `/api/v1/analysis/analyze` | POST | Trigger stock analysis |
| `/api/v1/analysis/tasks` | GET | Query task list |
| `/api/v1/analysis/status/{task_id}` | GET | Query task status |
| `/api/v1/history` | GET | Query analysis history |
| `/api/health` | GET | Health check |
| `/docs` | GET | API Swagger documentation |
**Usage examples**:
```bash
# Health check
curl http://127.0.0.1:8000/health
curl http://127.0.0.1:8000/api/health
# Trigger analysis (A-shares)
curl "http://127.0.0.1:8000/analysis?code=600519"
# Trigger analysis (HK stocks)
curl "http://127.0.0.1:8000/analysis?code=hk00700"
curl -X POST http://127.0.0.1:8000/api/v1/analysis/analyze \
-H 'Content-Type: application/json' \
-d '{"stock_code": "600519"}'
# Query task status
curl "http://127.0.0.1:8000/task?id=<task_id>"
curl http://127.0.0.1:8000/api/v1/analysis/status/<task_id>
```
### Custom Configuration
Modify default port or allow LAN access:
```env
WEBUI_HOST=0.0.0.0 # Default 127.0.0.1
WEBUI_PORT=8888 # Default 8000
```bash
python main.py --serve-only --host 0.0.0.0 --port 8888
```
### Supported Stock Code Formats

57
main.py
View File

@@ -133,13 +133,13 @@ def parse_arguments() -> argparse.Namespace:
parser.add_argument(
'--webui',
action='store_true',
help='启动本地配置 WebUI旧版 Gradio'
help='启动 Web 管理界面'
)
parser.add_argument(
'--webui-only',
action='store_true',
help='仅启动 WebUI 服务,不自动执行分析'
help='仅启动 Web 服务,不执行自动分析'
)
parser.add_argument(
@@ -374,22 +374,27 @@ def main() -> int:
stock_codes = [code.strip() for code in args.stocks.split(',') if code.strip()]
logger.info(f"使用命令行指定的股票列表: {stock_codes}")
# === 启动 WebUI (如果启用) ===
# 优先级: 命令行参数 > 配置文件
start_webui = (args.webui or args.webui_only or config.webui_enabled) and os.getenv("GITHUB_ACTIONS") != "true"
# === 处理 --webui / --webui-only 参数,映射到 --serve / --serve-only ===
if args.webui:
args.serve = True
if args.webui_only:
args.serve_only = True
bot_clients_started = False
if start_webui:
try:
from webui import run_server_in_thread
run_server_in_thread(host=config.webui_host, port=config.webui_port)
bot_clients_started = True
except Exception as e:
logger.error(f"启动 WebUI 失败: {e}")
# 兼容旧版 WEBUI_ENABLED 环境变量
if config.webui_enabled and not (args.serve or args.serve_only):
args.serve = True
# === 启动 FastAPI 服务 (如果启用) ===
# === 启动 Web 服务 (如果启用) ===
start_serve = (args.serve or args.serve_only) and os.getenv("GITHUB_ACTIONS") != "true"
# 兼容旧版 WEBUI_HOST/WEBUI_PORT如果用户未通过 --host/--port 指定,则使用旧变量
if start_serve:
if args.host == '0.0.0.0' and os.getenv('WEBUI_HOST'):
args.host = os.getenv('WEBUI_HOST')
if args.port == 8000 and os.getenv('WEBUI_PORT'):
args.port = int(os.getenv('WEBUI_PORT'))
bot_clients_started = False
if start_serve:
try:
start_api_server(host=args.host, port=args.port, config=config)
@@ -400,23 +405,10 @@ def main() -> int:
if bot_clients_started:
start_bot_stream_clients(config)
# === 仅 WebUI 模式:不自动执行分析 ===
if args.webui_only:
logger.info("模式: 仅 WebUI 服务")
logger.info(f"WebUI 运行中: http://{config.webui_host}:{config.webui_port}")
logger.info("通过 /analysis?code=xxx 接口手动触发分析")
logger.info("按 Ctrl+C 退出...")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
logger.info("\n用户中断,程序退出")
return 0
# === 仅 FastAPI 服务模式:不自动执行分析 ===
# === 仅 Web 服务模式:不自动执行分析 ===
if args.serve_only:
logger.info("模式: 仅 FastAPI 服务")
logger.info(f"API 服务运行中: http://{args.host}:{args.port}")
logger.info("模式: 仅 Web 服务")
logger.info(f"Web 服务运行中: http://{args.host}:{args.port}")
logger.info("通过 /api/v1/analysis/stock/{code} 接口触发分析")
logger.info(f"API 文档: http://{args.host}:{args.port}/docs")
logger.info("按 Ctrl+C 退出...")
@@ -484,10 +476,9 @@ def main() -> int:
logger.info("\n程序执行完成")
# 如果启用了服务且是非定时任务模式,保持程序运行
keep_running = (start_webui or start_serve) and not (args.schedule or config.schedule_enabled)
keep_running = start_serve and not (args.schedule or config.schedule_enabled)
if keep_running:
service_name = "API 服务" if start_serve else "WebUI"
logger.info(f"{service_name} 运行中 (按 Ctrl+C 退出)...")
logger.info("API 服务运行中 (按 Ctrl+C 退出)...")
try:
while True:
time.sleep(1)

View File

@@ -11,9 +11,12 @@
from src.services.analysis_service import AnalysisService
from src.services.history_service import HistoryService
from src.services.stock_service import StockService
from src.services.task_service import TaskService, get_task_service
__all__ = [
"AnalysisService",
"HistoryService",
"StockService",
"TaskService",
"get_task_service",
]

View File

@@ -1,18 +1,19 @@
# -*- coding: utf-8 -*-
"""
===================================
Web 服务层 - 业务逻辑
异步任务服务层
===================================
职责
1. 配置管理服务 (ConfigService)
2. 分析任务服务 (AnalysisService)
1. 管理异步分析任务线程池
2. 执行股票分析并推送结果
3. 查询任务状态和历史
迁移自 web/services.py AnalysisService
"""
from __future__ import annotations
import os
import re
import logging
import threading
from concurrent.futures import ThreadPoolExecutor
@@ -25,116 +26,10 @@ from bot.models import BotMessage
logger = logging.getLogger(__name__)
# ============================================================
# 配置管理服务
# ============================================================
_ENV_PATH = os.getenv("ENV_FILE", ".env")
_STOCK_LIST_RE = re.compile(
r"^(?P<prefix>\s*STOCK_LIST\s*=\s*)(?P<value>.*?)(?P<suffix>\s*)$"
)
class ConfigService:
class TaskService:
"""
配置管理服务
负责 .env 文件中 STOCK_LIST 的读写操作
"""
def __init__(self, env_path: Optional[str] = None):
self.env_path = env_path or _ENV_PATH
def read_env_text(self) -> str:
"""读取 .env 文件内容"""
try:
with open(self.env_path, "r", encoding="utf-8") as f:
return f.read()
except FileNotFoundError:
return ""
def write_env_text(self, text: str) -> None:
"""写入 .env 文件内容"""
with open(self.env_path, "w", encoding="utf-8") as f:
f.write(text)
def get_stock_list(self) -> str:
"""获取当前自选股列表字符串"""
env_text = self.read_env_text()
return self._extract_stock_list(env_text)
def set_stock_list(self, stock_list: str) -> str:
"""
设置自选股列表
Args:
stock_list: 股票代码字符串逗号或换行分隔
Returns:
规范化后的股票列表字符串
"""
env_text = self.read_env_text()
normalized = self._normalize_stock_list(stock_list)
updated = self._update_stock_list(env_text, normalized)
self.write_env_text(updated)
return normalized
def get_env_filename(self) -> str:
"""获取 .env 文件名"""
return os.path.basename(self.env_path)
def _extract_stock_list(self, env_text: str) -> str:
"""从环境文件中提取 STOCK_LIST 值"""
for line in env_text.splitlines():
m = _STOCK_LIST_RE.match(line)
if m:
raw = m.group("value").strip()
# 去除引号
if (raw.startswith('"') and raw.endswith('"')) or \
(raw.startswith("'") and raw.endswith("'")):
raw = raw[1:-1]
return raw
return ""
def _normalize_stock_list(self, value: str) -> str:
"""规范化股票列表格式"""
parts = [p.strip() for p in value.replace("\n", ",").split(",")]
parts = [p for p in parts if p]
return ",".join(parts)
def _update_stock_list(self, env_text: str, new_value: str) -> str:
"""更新环境文件中的 STOCK_LIST"""
lines = env_text.splitlines(keepends=False)
out_lines: List[str] = []
replaced = False
for line in lines:
m = _STOCK_LIST_RE.match(line)
if not m:
out_lines.append(line)
continue
out_lines.append(f"{m.group('prefix')}{new_value}{m.group('suffix')}")
replaced = True
if not replaced:
if out_lines and out_lines[-1].strip() != "":
out_lines.append("")
out_lines.append(f"STOCK_LIST={new_value}")
trailing_newline = env_text.endswith("\n") if env_text else True
out = "\n".join(out_lines)
return out + ("\n" if trailing_newline else "")
# ============================================================
# 分析任务服务
# ============================================================
class AnalysisService:
"""
分析任务服务
异步任务服务
负责
1. 管理异步分析任务
@@ -142,7 +37,7 @@ class AnalysisService:
3. 触发通知推送
"""
_instance: Optional['AnalysisService'] = None
_instance: Optional['TaskService'] = None
_lock = threading.Lock()
def __init__(self, max_workers: int = 3):
@@ -152,7 +47,7 @@ class AnalysisService:
self._tasks_lock = threading.Lock()
@classmethod
def get_instance(cls) -> 'AnalysisService':
def get_instance(cls) -> 'TaskService':
"""获取单例实例"""
if cls._instance is None:
with cls._lock:
@@ -175,7 +70,8 @@ class AnalysisService:
code: str,
report_type: Union[ReportType, str] = ReportType.SIMPLE,
source_message: Optional[BotMessage] = None,
save_context_snapshot: Optional[bool] = None
save_context_snapshot: Optional[bool] = None,
query_source: str = "bot"
) -> Dict[str, Any]:
"""
提交异步分析任务
@@ -183,6 +79,9 @@ class AnalysisService:
Args:
code: 股票代码
report_type: 报告类型枚举
source_message: 来源消息用于回复
save_context_snapshot: 是否保存上下文快照
query_source: 任务来源标识bot/api/cli/system
Returns:
任务信息字典
@@ -200,10 +99,11 @@ class AnalysisService:
task_id,
report_type,
source_message,
save_context_snapshot
save_context_snapshot,
query_source
)
logger.info(f"[AnalysisService] 已提交股票 {code} 的分析任务, task_id={task_id}, report_type={report_type.value}")
logger.info(f"[TaskService] 已提交股票 {code} 的分析任务, task_id={task_id}, report_type={report_type.value}")
return {
"success": True,
@@ -233,9 +133,7 @@ class AnalysisService:
days: int = 30,
limit: int = 50
) -> List[Dict[str, Any]]:
"""
获取分析历史记录
"""
"""获取分析历史记录"""
db = get_db()
records = db.get_analysis_history(code=code, query_id=query_id, days=days, limit=limit)
return [r.to_dict() for r in records]
@@ -246,17 +144,13 @@ class AnalysisService:
task_id: str,
report_type: ReportType = ReportType.SIMPLE,
source_message: Optional[BotMessage] = None,
save_context_snapshot: Optional[bool] = None
save_context_snapshot: Optional[bool] = None,
query_source: str = "bot"
) -> Dict[str, Any]:
"""
执行单只股票分析
内部方法在线程池中运行
Args:
code: 股票代码
task_id: 任务ID
report_type: 报告类型枚举
"""
# 初始化任务状态
with self._tasks_lock:
@@ -275,7 +169,7 @@ class AnalysisService:
from src.config import get_config
from main import StockAnalysisPipeline
logger.info(f"[AnalysisService] 开始分析股票: {code}")
logger.info(f"[TaskService] 开始分析股票: {code}")
# 创建分析管道
config = get_config()
@@ -284,7 +178,7 @@ class AnalysisService:
max_workers=1,
source_message=source_message,
query_id=task_id,
query_source="web",
query_source=query_source,
save_context_snapshot=save_context_snapshot
)
@@ -313,7 +207,7 @@ class AnalysisService:
"result": result_data
})
logger.info(f"[AnalysisService] 股票 {code} 分析完成: {result.operation_advice}")
logger.info(f"[TaskService] 股票 {code} 分析完成: {result.operation_advice}")
return {"success": True, "task_id": task_id, "result": result_data}
else:
with self._tasks_lock:
@@ -323,12 +217,12 @@ class AnalysisService:
"error": "分析返回空结果"
})
logger.warning(f"[AnalysisService] 股票 {code} 分析失败: 返回空结果")
logger.warning(f"[TaskService] 股票 {code} 分析失败: 返回空结果")
return {"success": False, "task_id": task_id, "error": "分析返回空结果"}
except Exception as e:
error_msg = str(e)
logger.error(f"[AnalysisService] 股票 {code} 分析异常: {error_msg}")
logger.error(f"[TaskService] 股票 {code} 分析异常: {error_msg}")
with self._tasks_lock:
self._tasks[task_id].update({
@@ -344,11 +238,6 @@ class AnalysisService:
# 便捷函数
# ============================================================
def get_config_service() -> ConfigService:
"""获取配置服务"""
return ConfigService()
def get_analysis_service() -> AnalysisService:
"""获取分析服务单例"""
return AnalysisService.get_instance()
def get_task_service() -> TaskService:
"""获取任务服务"""
return TaskService.get_instance()

View File

@@ -248,7 +248,6 @@ test_syntax() {
python3 -m py_compile main.py src/config.py src/notification.py \
data_provider/akshare_fetcher.py \
data_provider/yfinance_fetcher.py \
web/handlers.py \
bot/commands/analyze.py
success "语法检查通过"

View File

@@ -1,30 +0,0 @@
# -*- coding: utf-8 -*-
"""
===================================
Web 服务模块
===================================
分层架构:
- server.py - HTTP 服务器核心
- router.py - 路由分发
- handlers.py - 请求处理器
- services.py - 业务服务层
- templates.py - HTML 模板
使用方式:
from web import run_server_in_thread, WebServer
# 后台启动
run_server_in_thread(host="127.0.0.1", port=8000)
# 前台启动
server = WebServer(host="127.0.0.1", port=8000)
server.run()
"""
from web.server import WebServer, run_server_in_thread
__all__ = [
'WebServer',
'run_server_in_thread',
]

View File

@@ -1,387 +0,0 @@
# -*- coding: utf-8 -*-
"""
===================================
Web 处理器层 - 请求处理
===================================
职责:
1. 处理各类 HTTP 请求
2. 调用服务层执行业务逻辑
3. 返回响应数据
处理器分类:
- PageHandler: 页面请求处理
- ApiHandler: API 接口处理
"""
from __future__ import annotations
import json
import re
import logging
from http import HTTPStatus
from datetime import datetime
from typing import Dict, Any, Optional, TYPE_CHECKING
from web.services import get_config_service, get_analysis_service
from web.templates import render_config_page
from src.enums import ReportType
if TYPE_CHECKING:
from http.server import BaseHTTPRequestHandler
logger = logging.getLogger(__name__)
# ============================================================
# 响应辅助类
# ============================================================
class Response:
"""HTTP 响应封装"""
def __init__(
self,
body: bytes,
status: HTTPStatus = HTTPStatus.OK,
content_type: str = "text/html; charset=utf-8"
):
self.body = body
self.status = status
self.content_type = content_type
def send(self, handler: 'BaseHTTPRequestHandler') -> None:
"""发送响应到客户端"""
handler.send_response(self.status)
handler.send_header("Content-Type", self.content_type)
handler.send_header("Content-Length", str(len(self.body)))
handler.end_headers()
handler.wfile.write(self.body)
class JsonResponse(Response):
"""JSON 响应封装"""
def __init__(
self,
data: Dict[str, Any],
status: HTTPStatus = HTTPStatus.OK
):
body = json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8")
super().__init__(
body=body,
status=status,
content_type="application/json; charset=utf-8"
)
class HtmlResponse(Response):
"""HTML 响应封装"""
def __init__(
self,
body: bytes,
status: HTTPStatus = HTTPStatus.OK
):
super().__init__(
body=body,
status=status,
content_type="text/html; charset=utf-8"
)
# ============================================================
# 页面处理器
# ============================================================
class PageHandler:
"""页面请求处理器"""
def __init__(self):
self.config_service = get_config_service()
def handle_index(self) -> Response:
"""处理首页请求 GET /"""
stock_list = self.config_service.get_stock_list()
env_filename = self.config_service.get_env_filename()
body = render_config_page(stock_list, env_filename)
return HtmlResponse(body)
def handle_update(self, form_data: Dict[str, list]) -> Response:
"""
处理配置更新 POST /update
Args:
form_data: 表单数据
"""
stock_list = form_data.get("stock_list", [""])[0]
normalized = self.config_service.set_stock_list(stock_list)
env_filename = self.config_service.get_env_filename()
body = render_config_page(normalized, env_filename, message="已保存")
return HtmlResponse(body)
# ============================================================
# API 处理器
# ============================================================
class ApiHandler:
"""API 请求处理器"""
def __init__(self):
self.analysis_service = get_analysis_service()
def handle_health(self) -> Response:
"""
健康检查 GET /health
返回:
{
"status": "ok",
"timestamp": "2026-01-19T10:30:00",
"service": "stock-analysis-webui"
}
"""
data = {
"status": "ok",
"timestamp": datetime.now().isoformat(),
"service": "stock-analysis-webui"
}
return JsonResponse(data)
def handle_analysis(self, query: Dict[str, list]) -> Response:
"""
触发股票分析 GET /analysis?code=xxx
Args:
query: URL 查询参数
返回:
{
"success": true,
"message": "分析任务已提交",
"code": "600519",
"task_id": "600519_20260119_103000"
}
"""
# 获取股票代码参数
code_list = query.get("code", [])
if not code_list or not code_list[0].strip():
return JsonResponse(
{"success": False, "error": "缺少必填参数: code (股票代码)"},
status=HTTPStatus.BAD_REQUEST
)
code = code_list[0].strip()
# 验证股票代码格式A股(6位数字) / 港股(HK+5位数字) / 美股(1-5个大写字母+.+2个后缀字母)
code = code.upper()
is_a_stock = re.match(r'^\d{6}$', code)
is_hk_stock = re.match(r'^HK\d{5}$', code)
is_us_stock = re.match(r'^[A-Z]{1,5}(\.[A-Z]{1,2})?$', code.upper())
if not (is_a_stock or is_hk_stock or is_us_stock):
return JsonResponse(
{"success": False, "error": f"无效的股票代码格式: {code} (A股6位数字 / 港股HK+5位数字 / 美股1-5个字母)"},
status=HTTPStatus.BAD_REQUEST
)
# 获取报告类型参数(默认精简报告)
report_type_str = query.get("report_type", ["simple"])[0]
report_type = ReportType.from_str(report_type_str)
# 是否保存上下文快照(可选,默认读取配置)
save_snapshot = None
if "save_context_snapshot" in query:
save_snapshot = self._parse_bool(query.get("save_context_snapshot", [""])[0])
# 提交异步分析任务
try:
result = self.analysis_service.submit_analysis(
code,
report_type=report_type,
save_context_snapshot=save_snapshot
)
return JsonResponse(result)
except Exception as e:
logger.error(f"[ApiHandler] 提交分析任务失败: {e}")
return JsonResponse(
{"success": False, "error": f"提交任务失败: {str(e)}"},
status=HTTPStatus.INTERNAL_SERVER_ERROR
)
def handle_analysis_history(self, query: Dict[str, list]) -> Response:
"""
查询分析历史 GET /analysis/history
Args:
query: URL 查询参数 (code, query_id, days, limit)
"""
code = query.get("code", [""])[0].strip() or None
query_id = query.get("query_id", [""])[0].strip() or None
try:
days = int(query.get("days", ["30"])[0])
except ValueError:
days = 30
try:
limit = int(query.get("limit", ["50"])[0])
except ValueError:
limit = 50
history = self.analysis_service.get_analysis_history(
code=code,
query_id=query_id,
days=days,
limit=limit
)
return JsonResponse({
"success": True,
"records": history,
"count": len(history)
})
@staticmethod
def _parse_bool(value: str) -> Optional[bool]:
"""
解析布尔参数
"""
text = (value or "").strip().lower()
if text in {"1", "true", "yes", "y", "on"}:
return True
if text in {"0", "false", "no", "n", "off"}:
return False
return None
def handle_tasks(self, query: Dict[str, list]) -> Response:
"""
查询任务列表 GET /tasks
Args:
query: URL 查询参数 (可选 limit)
返回:
{
"success": true,
"tasks": [...]
}
"""
limit_list = query.get("limit", ["20"])
try:
limit = int(limit_list[0])
except ValueError:
limit = 20
tasks = self.analysis_service.list_tasks(limit=limit)
return JsonResponse({"success": True, "tasks": tasks})
def handle_task_status(self, query: Dict[str, list]) -> Response:
"""
查询单个任务状态 GET /task?id=xxx
Args:
query: URL 查询参数
"""
task_id_list = query.get("id", [])
if not task_id_list or not task_id_list[0].strip():
return JsonResponse(
{"success": False, "error": "缺少必填参数: id (任务ID)"},
status=HTTPStatus.BAD_REQUEST
)
task_id = task_id_list[0].strip()
task = self.analysis_service.get_task_status(task_id)
if task is None:
return JsonResponse(
{"success": False, "error": f"任务不存在: {task_id}"},
status=HTTPStatus.NOT_FOUND
)
return JsonResponse({"success": True, "task": task})
# ============================================================
# Bot Webhook 处理器
# ============================================================
class BotHandler:
"""
机器人 Webhook 处理器
处理各平台的机器人回调请求。
"""
def handle_webhook(self, platform: str, form_data: Dict[str, list], headers: Dict[str, str], body: bytes) -> Response:
"""
处理 Webhook 请求
Args:
platform: 平台名称 (feishu, dingtalk, wecom, telegram)
form_data: POST 数据(已解析)
headers: HTTP 请求头
body: 原始请求体
Returns:
Response 对象
"""
try:
from bot.handler import handle_webhook
from bot.models import WebhookResponse
# 调用 bot 模块处理
webhook_response = handle_webhook(platform, headers, body)
# 转换为 web 响应
return JsonResponse(
webhook_response.body,
status=HTTPStatus(webhook_response.status_code)
)
except ImportError as e:
logger.error(f"[BotHandler] Bot 模块未正确安装: {e}")
return JsonResponse(
{"error": "Bot module not available"},
status=HTTPStatus.INTERNAL_SERVER_ERROR
)
except Exception as e:
logger.error(f"[BotHandler] 处理 {platform} Webhook 失败: {e}")
return JsonResponse(
{"error": str(e)},
status=HTTPStatus.INTERNAL_SERVER_ERROR
)
# ============================================================
# 处理器工厂
# ============================================================
_page_handler: PageHandler | None = None
_api_handler: ApiHandler | None = None
_bot_handler: BotHandler | None = None
def get_page_handler() -> PageHandler:
"""获取页面处理器实例"""
global _page_handler
if _page_handler is None:
_page_handler = PageHandler()
return _page_handler
def get_api_handler() -> ApiHandler:
"""获取 API 处理器实例"""
global _api_handler
if _api_handler is None:
_api_handler = ApiHandler()
return _api_handler
def get_bot_handler() -> BotHandler:
"""获取 Bot 处理器实例"""
global _bot_handler
if _bot_handler is None:
_bot_handler = BotHandler()
return _bot_handler

View File

@@ -1,376 +0,0 @@
# -*- coding: utf-8 -*-
"""
===================================
Web 路由层 - 请求分发
===================================
职责:
1. 解析请求路径
2. 分发到对应的处理器
3. 支持路由注册和扩展
"""
from __future__ import annotations
import logging
from http import HTTPStatus
from typing import Callable, Dict, List, Optional, TYPE_CHECKING, Tuple
from urllib.parse import parse_qs, urlparse
from web.handlers import (
Response, HtmlResponse, JsonResponse,
get_page_handler, get_api_handler, get_bot_handler
)
from web.templates import render_error_page
if TYPE_CHECKING:
from http.server import BaseHTTPRequestHandler
logger = logging.getLogger(__name__)
# ============================================================
# 路由定义
# ============================================================
# 路由处理函数类型: (query_params) -> Response
RouteHandler = Callable[[Dict[str, list]], Response]
class Route:
"""路由定义"""
def __init__(
self,
path: str,
method: str,
handler: RouteHandler,
description: str = ""
):
self.path = path
self.method = method.upper()
self.handler = handler
self.description = description
class Router:
"""
路由管理器
负责:
1. 注册路由
2. 匹配请求路径
3. 分发到处理器
"""
def __init__(self):
self._routes: Dict[str, Dict[str, Route]] = {} # {path: {method: Route}}
def register(
self,
path: str,
method: str,
handler: RouteHandler,
description: str = ""
) -> None:
"""
注册路由
Args:
path: 路由路径
method: HTTP 方法 (GET, POST, etc.)
handler: 处理函数
description: 路由描述
"""
method = method.upper()
if path not in self._routes:
self._routes[path] = {}
self._routes[path][method] = Route(path, method, handler, description)
logger.debug(f"[Router] 注册路由: {method} {path}")
def get(self, path: str, description: str = "") -> Callable:
"""装饰器:注册 GET 路由"""
def decorator(handler: RouteHandler) -> RouteHandler:
self.register(path, "GET", handler, description)
return handler
return decorator
def post(self, path: str, description: str = "") -> Callable:
"""装饰器:注册 POST 路由"""
def decorator(handler: RouteHandler) -> RouteHandler:
self.register(path, "POST", handler, description)
return handler
return decorator
def match(self, path: str, method: str) -> Optional[Route]:
"""
匹配路由
Args:
path: 请求路径
method: HTTP 方法
Returns:
匹配的路由,或 None
"""
method = method.upper()
routes_for_path = self._routes.get(path)
if routes_for_path is None:
return None
return routes_for_path.get(method)
def dispatch(
self,
request_handler: 'BaseHTTPRequestHandler',
method: str
) -> None:
"""
分发请求
Args:
request_handler: HTTP 请求处理器
method: HTTP 方法
"""
# 解析 URL
parsed = urlparse(request_handler.path)
path = parsed.path
query = parse_qs(parsed.query)
# 处理根路径
if path == "":
path = "/"
# 匹配路由
route = self.match(path, method)
if route is None:
# 404 Not Found
self._send_not_found(request_handler, path)
return
try:
# 调用处理器
response = route.handler(query)
response.send(request_handler)
except Exception as e:
logger.error(f"[Router] 处理请求失败: {method} {path} - {e}")
self._send_error(request_handler, str(e))
def dispatch_post(
self,
request_handler: 'BaseHTTPRequestHandler'
) -> None:
"""
分发 POST 请求(需要读取 body
Args:
request_handler: HTTP 请求处理器
"""
parsed = urlparse(request_handler.path)
path = parsed.path
# 读取 POST body保留原始字节用于 Bot Webhook
content_length = int(request_handler.headers.get("Content-Length", "0") or "0")
raw_body_bytes = request_handler.rfile.read(content_length)
raw_body = raw_body_bytes.decode("utf-8", errors="replace")
# 检查是否是 Bot Webhook 路由
if path.startswith("/bot/"):
self._dispatch_bot_webhook(request_handler, path, raw_body_bytes)
return
# 普通 POST 请求
form_data = parse_qs(raw_body)
# 匹配路由
route = self.match(path, "POST")
if route is None:
self._send_not_found(request_handler, path)
return
try:
# 调用处理器(传入 form_data
response = route.handler(form_data)
response.send(request_handler)
except Exception as e:
logger.error(f"[Router] 处理 POST 请求失败: {path} - {e}")
self._send_error(request_handler, str(e))
def _dispatch_bot_webhook(
self,
request_handler: 'BaseHTTPRequestHandler',
path: str,
body: bytes
) -> None:
"""
分发 Bot Webhook 请求
Bot Webhook 需要原始 body 和 headers与普通路由处理不同。
Args:
request_handler: HTTP 请求处理器
path: 请求路径
body: 原始请求体字节
"""
# 提取平台名称:/bot/feishu -> feishu
parts = path.strip('/').split('/')
if len(parts) < 2:
self._send_not_found(request_handler, path)
return
platform = parts[1]
# 获取请求头
headers = {key: value for key, value in request_handler.headers.items()}
try:
bot_handler = get_bot_handler()
response = bot_handler.handle_webhook(platform, {}, headers, body)
response.send(request_handler)
except Exception as e:
logger.error(f"[Router] 处理 Bot Webhook 失败: {path} - {e}")
self._send_error(request_handler, str(e))
def list_routes(self) -> List[Tuple[str, str, str]]:
"""
列出所有路由
Returns:
[(method, path, description), ...]
"""
routes = []
for path, methods in self._routes.items():
for method, route in methods.items():
routes.append((method, path, route.description))
return sorted(routes, key=lambda x: (x[1], x[0]))
def _send_not_found(
self,
request_handler: 'BaseHTTPRequestHandler',
path: str
) -> None:
"""发送 404 响应"""
body = render_error_page(404, "页面未找到", f"路径 {path} 不存在")
response = HtmlResponse(body, status=HTTPStatus.NOT_FOUND)
response.send(request_handler)
def _send_error(
self,
request_handler: 'BaseHTTPRequestHandler',
message: str
) -> None:
"""发送 500 响应"""
body = render_error_page(500, "服务器内部错误", message)
response = HtmlResponse(body, status=HTTPStatus.INTERNAL_SERVER_ERROR)
response.send(request_handler)
# ============================================================
# 默认路由注册
# ============================================================
def create_default_router() -> Router:
"""创建并配置默认路由"""
router = Router()
# 获取处理器
page_handler = get_page_handler()
api_handler = get_api_handler()
# === 页面路由 ===
router.register(
"/", "GET",
lambda q: page_handler.handle_index(),
"配置首页"
)
router.register(
"/update", "POST",
lambda form: page_handler.handle_update(form),
"更新配置"
)
# === API 路由 ===
router.register(
"/health", "GET",
lambda q: api_handler.handle_health(),
"健康检查"
)
router.register(
"/analysis", "GET",
lambda q: api_handler.handle_analysis(q),
"触发股票分析"
)
router.register(
"/analysis/history", "GET",
lambda q: api_handler.handle_analysis_history(q),
"查询分析历史"
)
router.register(
"/tasks", "GET",
lambda q: api_handler.handle_tasks(q),
"查询任务列表"
)
router.register(
"/task", "GET",
lambda q: api_handler.handle_task_status(q),
"查询任务状态"
)
# === Bot Webhook 路由 ===
# 注意Bot Webhook 路由在 dispatch_post 中特殊处理
# 这里只是为了在路由列表中显示
# 实际请求会被 _dispatch_bot_webhook 方法处理
# 飞书机器人 Webhook
router.register(
"/bot/feishu", "POST",
lambda form: JsonResponse({"error": "Use POST with JSON body"}),
"飞书机器人 Webhook"
)
# 钉钉机器人 Webhook
router.register(
"/bot/dingtalk", "POST",
lambda form: JsonResponse({"error": "Use POST with JSON body"}),
"钉钉机器人 Webhook"
)
# 企业微信机器人 Webhook开发中
# router.register(
# "/bot/wecom", "POST",
# lambda form: JsonResponse({"error": "Use POST with JSON body"}),
# "企业微信机器人 Webhook"
# )
# Telegram 机器人 Webhook开发中
# router.register(
# "/bot/telegram", "POST",
# lambda form: JsonResponse({"error": "Use POST with JSON body"}),
# "Telegram 机器人 Webhook"
# )
return router
# 全局默认路由实例
_default_router: Router | None = None
def get_router() -> Router:
"""获取默认路由实例"""
global _default_router
if _default_router is None:
_default_router = create_default_router()
return _default_router

View File

@@ -1,216 +0,0 @@
# -*- coding: utf-8 -*-
"""
===================================
Web 服务器核心
===================================
职责:
1. 启动 HTTP 服务器
2. 处理请求分发
3. 提供后台运行接口
"""
from __future__ import annotations
import logging
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Optional, Type
from web.router import Router, get_router
logger = logging.getLogger(__name__)
# ============================================================
# HTTP 请求处理器
# ============================================================
class WebRequestHandler(BaseHTTPRequestHandler):
"""
HTTP 请求处理器
将请求分发到路由器处理
"""
# 类级别的路由器引用
router: Router = None # type: ignore
def do_GET(self) -> None:
"""处理 GET 请求"""
self.router.dispatch(self, "GET")
def do_POST(self) -> None:
"""处理 POST 请求"""
self.router.dispatch_post(self)
def log_message(self, fmt: str, *args) -> None:
"""自定义日志格式(使用 logging 而非 stderr"""
# 可以取消注释以启用请求日志
# logger.debug(f"[WebServer] {self.address_string()} - {fmt % args}")
pass
# ============================================================
# Web 服务器
# ============================================================
class WebServer:
"""
Web 服务器
封装 ThreadingHTTPServer提供便捷的启动和管理接口
使用方式:
# 前台运行
server = WebServer(host="127.0.0.1", port=8000)
server.run()
# 后台运行
server = WebServer(host="127.0.0.1", port=8000)
server.start_background()
"""
def __init__(
self,
host: str = "127.0.0.1",
port: int = 8000,
router: Optional[Router] = None
):
"""
初始化 Web 服务器
Args:
host: 监听地址
port: 监听端口
router: 路由器实例(可选,默认使用全局路由)
"""
self.host = host
self.port = port
self.router = router or get_router()
self._server: Optional[ThreadingHTTPServer] = None
self._thread: Optional[threading.Thread] = None
@property
def address(self) -> str:
"""服务器地址"""
return f"http://{self.host}:{self.port}"
def _create_handler_class(self) -> Type[WebRequestHandler]:
"""创建带路由器引用的处理器类"""
router = self.router
class Handler(WebRequestHandler):
pass
Handler.router = router
return Handler
def _create_server(self) -> ThreadingHTTPServer:
"""创建 HTTP 服务器实例"""
handler_class = self._create_handler_class()
return ThreadingHTTPServer((self.host, self.port), handler_class)
def run(self) -> None:
"""
前台运行服务器(阻塞)
按 Ctrl+C 退出
"""
self._server = self._create_server()
logger.info(f"WebUI 服务启动: {self.address}")
print(f"WebUI 服务启动: {self.address}")
# 打印路由列表
routes = self.router.list_routes()
if routes:
logger.info("已注册路由:")
for method, path, desc in routes:
logger.info(f" {method:6} {path:20} - {desc}")
try:
self._server.serve_forever()
except KeyboardInterrupt:
logger.info("收到退出信号,服务器关闭")
finally:
self._server.server_close()
self._server = None
def start_background(self) -> threading.Thread:
"""
后台运行服务器(非阻塞)
Returns:
服务器线程
"""
self._server = self._create_server()
def serve():
logger.info(f"WebUI 已启动: {self.address}")
print(f"WebUI 已启动: {self.address}")
try:
self._server.serve_forever()
except Exception as e:
logger.error(f"WebUI 发生错误: {e}")
finally:
if self._server:
self._server.server_close()
self._thread = threading.Thread(target=serve, daemon=True)
self._thread.start()
return self._thread
def stop(self) -> None:
"""停止服务器"""
if self._server:
self._server.shutdown()
self._server.server_close()
self._server = None
logger.info("WebUI 服务已停止")
def is_running(self) -> bool:
"""检查服务器是否运行中"""
return self._server is not None
# ============================================================
# 便捷函数
# ============================================================
def run_server_in_thread(
host: str = "127.0.0.1",
port: int = 8000,
router: Optional[Router] = None
) -> threading.Thread:
"""
在后台线程启动 WebUI 服务器
Args:
host: 监听地址
port: 监听端口
router: 路由器实例(可选)
Returns:
服务器线程
"""
server = WebServer(host=host, port=port, router=router)
return server.start_background()
def run_server(
host: str = "127.0.0.1",
port: int = 8000,
router: Optional[Router] = None
) -> None:
"""
前台运行 WebUI 服务器(阻塞)
Args:
host: 监听地址
port: 监听端口
router: 路由器实例(可选)
"""
server = WebServer(host=host, port=port, router=router)
server.run()

File diff suppressed because it is too large Load Diff

132
webui.py
View File

@@ -1,27 +1,14 @@
# -*- coding: utf-8 -*-
"""
===================================
WebUI 入口文件 (向后兼容)
WebUI 启动脚本
===================================
本文件保持向后兼容,实际实现已迁移到 web/ 包
用于启动 Web 服务界面。
直接运行 `python webui.py` 将启动 Web 后端服务。
结构说明:
web/
├── __init__.py - 包初始化
├── server.py - HTTP 服务器
├── router.py - 路由分发
├── handlers.py - 请求处理器
├── services.py - 业务服务层
└── templates.py - HTML 模板
API Endpoints:
GET / - 配置页面
GET /health - 健康检查
GET /analysis?code=xxx - 触发单只股票异步分析
GET /tasks - 查询任务列表
GET /task?id=xxx - 查询任务状态
POST /update - 更新配置
等效命令:
python main.py --webui-only
Usage:
python webui.py
@@ -33,106 +20,35 @@ from __future__ import annotations
import os
import logging
# 从 web 包导入(新架构)
from web.server import WebServer, run_server_in_thread, run_server
from web.router import Router, get_router
from web.services import ConfigService, AnalysisService, get_config_service, get_analysis_service
from web.handlers import PageHandler, ApiHandler
from web.templates import render_config_page, render_error_page
logger = logging.getLogger(__name__)
# 导出所有公共接口(保持向后兼容)
__all__ = [
# 服务器
'WebServer',
'run_server_in_thread',
'run_server',
# 路由
'Router',
'get_router',
# 服务
'ConfigService',
'AnalysisService',
'get_config_service',
'get_analysis_service',
# 处理器
'PageHandler',
'ApiHandler',
# 模板
'render_config_page',
'render_error_page',
]
def _start_bot_stream_clients() -> None:
"""启动 Bot Stream 模式客户端(如果已配置)"""
from src.config import get_config
config = get_config()
# 钉钉 Stream 模式
if config.dingtalk_stream_enabled:
try:
from bot.platforms import start_dingtalk_stream_background, DINGTALK_STREAM_AVAILABLE
if DINGTALK_STREAM_AVAILABLE:
if start_dingtalk_stream_background():
logger.info("[WebUI] 钉钉 Stream 客户端已在后台启动")
else:
logger.warning("[WebUI] 钉钉 Stream 客户端启动失败")
else:
logger.warning("[WebUI] 钉钉 Stream 模式已启用但 SDK 未安装")
logger.warning("[WebUI] 请运行: pip install dingtalk-stream")
except Exception as e:
logger.error(f"[WebUI] 启动钉钉 Stream 客户端失败: {e}")
# 飞书 Stream 模式
if getattr(config, 'feishu_stream_enabled', False):
try:
from bot.platforms import start_feishu_stream_background, FEISHU_SDK_AVAILABLE
if FEISHU_SDK_AVAILABLE:
if start_feishu_stream_background():
logger.info("[WebUI] 飞书 Stream 客户端已在后台启动")
else:
logger.warning("[WebUI] 飞书 Stream 客户端启动失败")
else:
logger.warning("[WebUI] 飞书 Stream 模式已启用但 SDK 未安装")
logger.warning("[WebUI] 请运行: pip install lark-oapi")
except Exception as e:
logger.error(f"[WebUI] 启动飞书 Stream 客户端失败: {e}")
def main() -> int:
"""
主入口函数
支持环境变量配置:
WEBUI_HOST: 监听地址 (默认 127.0.0.1)
WEBUI_PORT: 监听端口 (默认 8000)
启动 Web 服务
"""
host = os.getenv("WEBUI_HOST", "127.0.0.1")
port = int(os.getenv("WEBUI_PORT", "8000"))
# 兼容旧版环境变量名
host = os.getenv("WEBUI_HOST", os.getenv("API_HOST", "127.0.0.1"))
port = int(os.getenv("WEBUI_PORT", os.getenv("API_PORT", "8000")))
print(f"WebUI running: http://{host}:{port}")
print("API Endpoints:")
print(" GET / - 配置页面")
print(" GET /health - 健康检查")
print(" GET /analysis?code=xxx - 触发分析")
print(" GET /tasks - 任务列表")
print(" GET /task?id=xxx - 任务状态")
print(" POST /update - 更新配置")
print(f"正在启动 Web 服务: http://{host}:{port}")
print(f"API 文档: http://{host}:{port}/docs")
print()
print("Bot Webhooks:")
print(" POST /bot/feishu - 飞书机器人")
print(" POST /bot/dingtalk - 钉钉机器人")
print(" POST /bot/wecom - 企业微信机器人")
print(" POST /bot/telegram - Telegram 机器人")
print()
# 启动 Bot Stream 客户端(如果配置了)
_start_bot_stream_clients()
try:
run_server(host=host, port=port)
import uvicorn
from src.config import setup_env
from src.logging_config import setup_logging
setup_env()
setup_logging(log_prefix="web_server")
uvicorn.run(
"api.app:app",
host=host,
port=port,
log_level="info",
)
except KeyboardInterrupt:
pass