fanhuadesenlinnn/daily_stock_analysis
1
0
Fork 0
mirror of https://github.com/ZhuLinsen/daily_stock_analysis synced 2026-09-20 10:53:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: 优化web服务结构

Browse Source
This commit is contained in:
Krane
2026-01-19 11:41:45 +08:00
parent e30fb30594
commit 5c8dbf9e2e
7 changed files with 1513 additions and 556 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
web/__init__.py Normal file
View File

@@ -0,0 +1,30 @@
# -*- 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',
]

264
web/handlers.py Normal file
View File

@@ -0,0 +1,264 @@
# -*- 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, TYPE_CHECKING
from urllib.parse import parse_qs
from web.services import get_config_service, get_analysis_service
from web.templates import render_config_page
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()
# 验证股票代码格式（6位数字）
if not re.match(r'^\d{6}$', code):
return JsonResponse(
{"success": False, "error": f"无效的股票代码格式: {code} (应为6位数字)"},
status=HTTPStatus.BAD_REQUEST
)
# 提交异步分析任务
try:
result = self.analysis_service.submit_analysis(code)
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_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})
# ============================================================
# 处理器工厂
# ============================================================
_page_handler: PageHandler | None = None
_api_handler: ApiHandler | 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

293
web/router.py Normal file
View File

@@ -0,0 +1,293 @@
# -*- 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, JsonResponse, HtmlResponse,
get_page_handler, get_api_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
content_length = int(request_handler.headers.get("Content-Length", "0") or "0")
raw_body = request_handler.rfile.read(content_length).decode("utf-8", errors="replace")
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 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(
"/tasks", "GET",
lambda q: api_handler.handle_tasks(q),
"查询任务列表"
)
router.register(
"/task", "GET",
lambda q: api_handler.handle_task_status(q),
"查询任务状态"
)
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

216
web/server.py Normal file
View File

@@ -0,0 +1,216 @@
# -*- 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()

296
web/services.py Normal file
View File

@@ -0,0 +1,296 @@
# -*- coding: utf-8 -*-
"""
===================================
Web 服务层 - 业务逻辑
===================================
职责：
1. 配置管理服务 (ConfigService)
2. 分析任务服务 (AnalysisService)
"""
from __future__ import annotations
import os
import re
import logging
import threading
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from typing import Optional, Dict, Any, List
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:
"""
配置管理服务
负责 .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. 管理异步分析任务
2. 执行股票分析
3. 触发通知推送
"""
_instance: Optional['AnalysisService'] = None
_lock = threading.Lock()
def __init__(self, max_workers: int = 3):
self._executor: Optional[ThreadPoolExecutor] = None
self._max_workers = max_workers
self._tasks: Dict[str, Dict[str, Any]] = {}
self._tasks_lock = threading.Lock()
@classmethod
def get_instance(cls) -> 'AnalysisService':
"""获取单例实例"""
if cls._instance is None:
with cls._lock:
if cls._instance is None:
cls._instance = cls()
return cls._instance
@property
def executor(self) -> ThreadPoolExecutor:
"""获取或创建线程池"""
if self._executor is None:
self._executor = ThreadPoolExecutor(
max_workers=self._max_workers,
thread_name_prefix="analysis_"
)
return self._executor
def submit_analysis(self, code: str) -> Dict[str, Any]:
"""
提交异步分析任务
Args:
code: 股票代码
Returns:
任务信息字典
"""
task_id = f"{code}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
# 提交到线程池
self.executor.submit(self._run_analysis, code, task_id)
logger.info(f"[AnalysisService] 已提交股票 {code} 的分析任务, task_id={task_id}")
return {
"success": True,
"message": "分析任务已提交，将异步执行并推送通知",
"code": code,
"task_id": task_id
}
def get_task_status(self, task_id: str) -> Optional[Dict[str, Any]]:
"""获取任务状态"""
with self._tasks_lock:
return self._tasks.get(task_id)
def list_tasks(self, limit: int = 20) -> List[Dict[str, Any]]:
"""列出最近的任务"""
with self._tasks_lock:
tasks = list(self._tasks.values())
# 按开始时间倒序
tasks.sort(key=lambda x: x.get('start_time', ''), reverse=True)
return tasks[:limit]
def _run_analysis(self, code: str, task_id: str) -> Dict[str, Any]:
"""
执行单只股票分析
内部方法，在线程池中运行
"""
# 初始化任务状态
with self._tasks_lock:
self._tasks[task_id] = {
"task_id": task_id,
"code": code,
"status": "running",
"start_time": datetime.now().isoformat(),
"result": None,
"error": None
}
try:
# 延迟导入避免循环依赖
from config import get_config
from main import StockAnalysisPipeline
logger.info(f"[AnalysisService] 开始分析股票: {code}")
# 创建分析管道
config = get_config()
pipeline = StockAnalysisPipeline(config=config, max_workers=1)
# 执行单只股票分析（启用单股推送）
result = pipeline.process_single_stock(
code=code,
skip_analysis=False,
single_stock_notify=True
)
if result:
result_data = {
"code": result.code,
"name": result.name,
"sentiment_score": result.sentiment_score,
"operation_advice": result.operation_advice,
"trend_prediction": result.trend_prediction,
"analysis_summary": result.analysis_summary,
}
with self._tasks_lock:
self._tasks[task_id].update({
"status": "completed",
"end_time": datetime.now().isoformat(),
"result": result_data
})
logger.info(f"[AnalysisService] 股票 {code} 分析完成: {result.operation_advice}")
return {"success": True, "task_id": task_id, "result": result_data}
else:
with self._tasks_lock:
self._tasks[task_id].update({
"status": "failed",
"end_time": datetime.now().isoformat(),
"error": "分析返回空结果"
})
logger.warning(f"[AnalysisService] 股票 {code} 分析失败: 返回空结果")
return {"success": False, "task_id": task_id, "error": "分析返回空结果"}
except Exception as e:
error_msg = str(e)
logger.error(f"[AnalysisService] 股票 {code} 分析异常: {error_msg}")
with self._tasks_lock:
self._tasks[task_id].update({
"status": "failed",
"end_time": datetime.now().isoformat(),
"error": error_msg
})
return {"success": False, "task_id": task_id, "error": error_msg}
# ============================================================
# 便捷函数
# ============================================================
def get_config_service() -> ConfigService:
"""获取配置服务实例"""
return ConfigService()
def get_analysis_service() -> AnalysisService:
"""获取分析服务单例"""
return AnalysisService.get_instance()

348
web/templates.py Normal file
View File

@@ -0,0 +1,348 @@
# -*- coding: utf-8 -*-
"""
===================================
Web 模板层 - HTML 页面生成
===================================
职责：
1. 生成 HTML 页面
2. 管理 CSS 样式
3. 提供可复用的页面组件
"""
from __future__ import annotations
import html
from typing import Optional
# ============================================================
# CSS 样式定义
# ============================================================
BASE_CSS = """
:root {
--primary: #2563eb;
--primary-hover: #1d4ed8;
--bg: #f8fafc;
--card: #ffffff;
--text: #1e293b;
--text-light: #64748b;
--border: #e2e8f0;
--success: #10b981;
--error: #ef4444;
--warning: #f59e0b;
}
* {
box-sizing: border-box;
}
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, "Helvetica Neue", Arial, sans-serif;
background-color: var(--bg);
color: var(--text);
display: flex;
justify-content: center;
align-items: center;
min-height: 100vh;
margin: 0;
padding: 20px;
}
.container {
background: var(--card);
padding: 2rem;
border-radius: 1rem;
box-shadow: 0 4px 6px -1px rgb(0 0 0 / 0.1), 0 2px 4px -2px rgb(0 0 0 / 0.1);
width: 100%;
max-width: 500px;
}
h2 {
margin-top: 0;
color: var(--text);
font-size: 1.5rem;
font-weight: 700;
display: flex;
align-items: center;
gap: 0.5rem;
}
.subtitle {
color: var(--text-light);
font-size: 0.875rem;
margin-bottom: 2rem;
line-height: 1.5;
}
.code-badge {
background: #f1f5f9;
padding: 0.2rem 0.4rem;
border-radius: 0.25rem;
font-family: monospace;
color: var(--primary);
}
.form-group {
margin-bottom: 1.5rem;
}
label {
display: block;
margin-bottom: 0.5rem;
font-weight: 500;
color: var(--text);
}
textarea, input[type="text"] {
width: 100%;
padding: 0.75rem;
border: 1px solid var(--border);
border-radius: 0.5rem;
font-family: monospace;
font-size: 0.875rem;
line-height: 1.5;
resize: vertical;
transition: border-color 0.2s, box-shadow 0.2s;
}
textarea:focus, input[type="text"]:focus {
outline: none;
border-color: var(--primary);
box-shadow: 0 0 0 3px rgba(37, 99, 235, 0.1);
}
button {
background-color: var(--primary);
color: white;
border: none;
padding: 0.75rem 1.5rem;
border-radius: 0.5rem;
font-weight: 500;
cursor: pointer;
transition: all 0.2s;
width: 100%;
font-size: 1rem;
}
button:hover {
background-color: var(--primary-hover);
transform: translateY(-1px);
}
button:active {
transform: translateY(0);
}
.btn-secondary {
background-color: var(--text-light);
}
.btn-secondary:hover {
background-color: var(--text);
}
.footer {
margin-top: 2rem;
padding-top: 1rem;
border-top: 1px solid var(--border);
color: var(--text-light);
font-size: 0.75rem;
text-align: center;
}
/* Toast Notification */
.toast {
position: fixed;
bottom: 20px;
left: 50%;
transform: translateX(-50%) translateY(100px);
background: white;
border-left: 4px solid var(--success);
padding: 1rem 1.5rem;
border-radius: 0.5rem;
box-shadow: 0 10px 15px -3px rgb(0 0 0 / 0.1);
display: flex;
align-items: center;
gap: 0.75rem;
transition: transform 0.3s cubic-bezier(0.175, 0.885, 0.32, 1.275);
opacity: 0;
z-index: 1000;
}
.toast.show {
transform: translateX(-50%) translateY(0);
opacity: 1;
}
.toast.error {
border-left-color: var(--error);
}
.toast.warning {
border-left-color: var(--warning);
}
/* Helper classes */
.text-muted {
font-size: 0.75rem;
color: var(--text-light);
margin-top: 0.5rem;
}
.mt-2 { margin-top: 0.5rem; }
.mt-4 { margin-top: 1rem; }
.mb-2 { margin-bottom: 0.5rem; }
.mb-4 { margin-bottom: 1rem; }
"""
# ============================================================
# 页面模板
# ============================================================
def render_base(
title: str,
content: str,
extra_css: str = "",
extra_js: str = ""
) -> str:
"""
渲染基础 HTML 模板
Args:
title: 页面标题
content: 页面内容 HTML
extra_css: 额外的 CSS 样式
extra_js: 额外的 JavaScript
"""
return f"""<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<title>{html.escape(title)}</title>
<style>{BASE_CSS}{extra_css}</style>
</head>
<body>
{content}
{extra_js}
</body>
</html>"""
def render_toast(message: str, toast_type: str = "success") -> str:
"""
渲染 Toast 通知
Args:
message: 通知消息
toast_type: 类型 (success, error, warning)
"""
icon_map = {
"success": "✅",
"error": "❌",
"warning": "⚠️"
}
icon = icon_map.get(toast_type, "ℹ️")
type_class = f" {toast_type}" if toast_type != "success" else ""
return f"""
<div id="toast" class="toast show{type_class}">
<span class="icon">{icon}</span> {html.escape(message)}
</div>
<script>
setTimeout(() => {{
document.getElementById('toast').classList.remove('show');
}}, 3000);
</script>
"""
def render_config_page(
stock_list: str,
env_filename: str,
message: Optional[str] = None
) -> bytes:
"""
渲染配置页面
Args:
stock_list: 当前自选股列表
env_filename: 环境文件名
message: 可选的提示消息
"""
safe_value = html.escape(stock_list)
toast_html = render_toast(message) if message else ""
content = f"""
<div class="container">
<h2>📈 A/H股分析配置</h2>
<div class="subtitle">
本地配置文件管理 <span class="code-badge">{html.escape(env_filename)}</span>
</div>
<form method="post" action="/update">
<div class="form-group">
<label for="stock_list">自选股代码列表</label>
<textarea
id="stock_list"
name="stock_list"
rows="6"
placeholder="例如: 600519, 000001 (支持逗号、换行分隔)"
>{safe_value}</textarea>
<div class="text-muted">
* 支持输入股票代码，多个代码请用英文逗号或换行分隔
</div>
</div>
<button type="submit">💾 保存配置</button>
</form>
<div class="footer">
<p>仅用于本地环境 (127.0.0.1) • 安全修改 .env 配置</p>
<p class="mt-2">
API: <code>/health</code> · <code>/analysis?code=xxx</code>
</p>
</div>
</div>
{toast_html}
"""
page = render_base(
title="A/H股自选配置 | WebUI",
content=content
)
return page.encode("utf-8")
def render_error_page(
status_code: int,
message: str,
details: Optional[str] = None
) -> bytes:
"""
渲染错误页面
Args:
status_code: HTTP 状态码
message: 错误消息
details: 详细信息
"""
details_html = f"<p class='text-muted'>{html.escape(details)}</p>" if details else ""
content = f"""
<div class="container" style="text-align: center;">
<h2>😵 {status_code}</h2>
<p>{html.escape(message)}</p>
{details_html}
<a href="/" style="color: var(--primary); text-decoration: none;">← 返回首页</a>
</div>
"""
page = render_base(
title=f"错误 {status_code}",
content=content
)
return page.encode("utf-8")

622
webui.py
View File

@@ -1,14 +1,27 @@
# -*- coding: utf-8 -*-
"""Very small local Web UI for editing STOCK_LIST in .env.
"""
===================================
WebUI 入口文件 (向后兼容)
===================================
- Local-only by default (127.0.0.1)
- No external dependencies
- Only edits the STOCK_LIST key; other .env lines are preserved
本文件保持向后兼容，实际实现已迁移到 web/ 包
结构说明:
web/
├── __init__.py - 包初始化
├── server.py - HTTP 服务器
├── router.py - 路由分发
├── handlers.py - 请求处理器
├── services.py - 业务服务层
└── templates.py - HTML 模板
API Endpoints:
GET / - 配置页面
GET /health - 健康检查
GET / - 配置页面
GET /health - 健康检查
GET /analysis?code=xxx - 触发单只股票异步分析
GET /tasks - 查询任务列表
GET /task?id=xxx - 查询任务状态
POST /update - 更新配置
Usage:
python webui.py
@@ -17,570 +30,67 @@ Usage:
from __future__ import annotations
import html
import json
import os
import re
import threading
from concurrent.futures import ThreadPoolExecutor
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse
import logging
from typing import Optional, Dict, Any
from datetime import datetime
# 从 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__)
# 全局线程池用于异步分析任务
_analysis_executor: Optional[ThreadPoolExecutor] = None
_analysis_tasks: Dict[str, Dict[str, Any]] = {} # 用于跟踪分析任务状态
_tasks_lock = threading.Lock()
def _get_executor() -> ThreadPoolExecutor:
"""获取或创建全局线程池"""
global _analysis_executor
if _analysis_executor is None:
_analysis_executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix="analysis_")
return _analysis_executor
def _run_stock_analysis(code: str) -> Dict[str, Any]:
"""
执行单只股票分析并推送通知
Args:
code: 股票代码
Returns:
分析结果字典
"""
task_id = f"{code}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
# 更新任务状态为进行中
with _tasks_lock:
_analysis_tasks[task_id] = {
"code": code,
"status": "running",
"start_time": datetime.now().isoformat(),
"result": None,
"error": None
}
try:
# 延迟导入避免循环依赖
from config import get_config
from main import StockAnalysisPipeline
logger.info(f"[WebUI] 开始分析股票: {code}")
# 创建分析管道
config = get_config()
pipeline = StockAnalysisPipeline(config=config, max_workers=1)
# 执行单只股票分析（启用单股推送）
result = pipeline.process_single_stock(
code=code,
skip_analysis=False,
single_stock_notify=True # 自动推送通知
)
if result:
# 分析成功
result_data = {
"code": result.code,
"name": result.name,
"sentiment_score": result.sentiment_score,
"operation_advice": result.operation_advice,
"trend_prediction": result.trend_prediction,
"analysis_summary": result.analysis_summary,
}
with _tasks_lock:
_analysis_tasks[task_id].update({
"status": "completed",
"end_time": datetime.now().isoformat(),
"result": result_data
})
logger.info(f"[WebUI] 股票 {code} 分析完成: {result.operation_advice}")
return {"success": True, "task_id": task_id, "result": result_data}
else:
# 分析失败
with _tasks_lock:
_analysis_tasks[task_id].update({
"status": "failed",
"end_time": datetime.now().isoformat(),
"error": "分析返回空结果"
})
logger.warning(f"[WebUI] 股票 {code} 分析失败: 返回空结果")
return {"success": False, "task_id": task_id, "error": "分析返回空结果"}
except Exception as e:
error_msg = str(e)
logger.error(f"[WebUI] 股票 {code} 分析异常: {error_msg}")
with _tasks_lock:
_analysis_tasks[task_id].update({
"status": "failed",
"end_time": datetime.now().isoformat(),
"error": error_msg
})
return {"success": False, "task_id": task_id, "error": error_msg}
_ENV_PATH = os.getenv("ENV_FILE", ".env")
def _read_env_text(path: str) -> str:
try:
with open(path, "r", encoding="utf-8") as f:
return f.read()
except FileNotFoundError:
return ""
def _write_env_text(path: str, text: str) -> None:
with open(path, "w", encoding="utf-8") as f:
f.write(text)
_STOCK_LIST_RE = re.compile(
r"^(?P<prefix>\s*STOCK_LIST\s*=\s*)(?P<value>.*?)(?P<suffix>\s*)$"
)
def _extract_stock_list(env_text: str) -> str:
for line in env_text.splitlines():
m = _STOCK_LIST_RE.match(line)
if m:
raw = m.group("value").strip()
# strip surrounding quotes
if (raw.startswith('"') and raw.endswith('"')) or (raw.startswith("'") and raw.endswith("'")):
raw = raw[1:-1]
return raw
return ""
def _normalize_stock_list(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 _set_stock_list(env_text: str, new_value: str) -> str:
new_value = _normalize_stock_list(new_value)
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
# Preserve prefix spacing; write as plain value (no quotes)
out_lines.append(f"{m.group('prefix')}{new_value}{m.group('suffix')}")
replaced = True
if not replaced:
# Keep existing text as-is, just append a new key
if out_lines and out_lines[-1].strip() != "":
out_lines.append("")
out_lines.append(f"STOCK_LIST={new_value}")
# Preserve trailing newline if original had one
trailing_newline = env_text.endswith("\n") if env_text else True
out = "\n".join(out_lines)
return out + ("\n" if trailing_newline else "")
def _page(current_value: str, message: str | None = None) -> bytes:
safe_value = html.escape(current_value)
# Toast notifications
toast_html = ""
if message:
toast_html = f"""
<div id="toast" class="toast show">
<span class="icon">✅</span> {html.escape(message)}
</div>
<script>
setTimeout(() => {{
document.getElementById('toast').classList.remove('show');
}}, 3000);
</script>
"""
# Modern CSS styling
css = """
:root {
--primary: #2563eb;
--primary-hover: #1d4ed8;
--bg: #f8fafc;
--card: #ffffff;
--text: #1e293b;
--text-light: #64748b;
--border: #e2e8f0;
--success: #10b981;
}
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, "Helvetica Neue", Arial, sans-serif;
background-color: var(--bg);
color: var(--text);
display: flex;
justify-content: center;
align-items: center;
min-height: 100vh;
margin: 0;
padding: 20px;
}
.container {
background: var(--card);
padding: 2rem;
border-radius: 1rem;
box-shadow: 0 4px 6px -1px rgb(0 0 0 / 0.1), 0 2px 4px -2px rgb(0 0 0 / 0.1);
width: 100%;
max-width: 500px;
}
h2 {
margin-top: 0;
color: var(--text);
font-size: 1.5rem;
font-weight: 700;
display: flex;
align-items: center;
gap: 0.5rem;
}
.subtitle {
color: var(--text-light);
font-size: 0.875rem;
margin-bottom: 2rem;
line-height: 1.5;
}
.code-badge {
background: #f1f5f9;
padding: 0.2rem 0.4rem;
border-radius: 0.25rem;
font-family: monospace;
color: var(--primary);
}
.form-group {
margin-bottom: 1.5rem;
}
label {
display: block;
margin-bottom: 0.5rem;
font-weight: 500;
color: var(--text);
}
textarea {
width: 100%;
padding: 0.75rem;
border: 1px solid var(--border);
border-radius: 0.5rem;
font-family: monospace;
font-size: 0.875rem;
line-height: 1.5;
resize: vertical;
box-sizing: border-box;
transition: border-color 0.2s, box-shadow 0.2s;
}
textarea:focus {
outline: none;
border-color: var(--primary);
box-shadow: 0 0 0 3px rgba(37, 99, 235, 0.1);
}
button {
background-color: var(--primary);
color: white;
border: none;
padding: 0.75rem 1.5rem;
border-radius: 0.5rem;
font-weight: 500;
cursor: pointer;
transition: all 0.2s;
width: 100%;
font-size: 1rem;
}
button:hover {
background-color: var(--primary-hover);
transform: translateY(-1px);
}
button:active {
transform: translateY(0);
}
.footer {
margin-top: 2rem;
padding-top: 1rem;
border-top: 1px solid var(--border);
color: var(--text-light);
font-size: 0.75rem;
text-align: center;
}
/* Toast Notification */
.toast {
position: fixed;
bottom: 20px;
left: 50%;
transform: translateX(-50%) translateY(100px);
background: white;
border-left: 4px solid var(--success);
padding: 1rem 1.5rem;
border-radius: 0.5rem;
box-shadow: 0 10px 15px -3px rgb(0 0 0 / 0.1);
display: flex;
align-items: center;
gap: 0.75rem;
transition: transform 0.3s cubic-bezier(0.175, 0.885, 0.32, 1.275);
opacity: 0;
}
.toast.show {
transform: translateX(-50%) translateY(0);
opacity: 1;
}
"""
body = f"""<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<title>A/H股自选配置 | WebUI</title>
<style>{css}</style>
</head>
<body>
<div class="container">
<h2>📈 A/H股分析配置</h2>
<div class="subtitle">
本地配置文件管理 <span class="code-badge">{html.escape(os.path.basename(_ENV_PATH))}</span>
</div>
<form method="post" action="/update">
<div class="form-group">
<label for="stock_list">自选股代码列表</label>
<textarea
id="stock_list"
name="stock_list"
rows="6"
placeholder="例如: 600519, 000001 (支持逗号、换行分隔)"
>{safe_value}</textarea>
<div style="font-size: 0.75rem; color: var(--text-light); margin-top: 0.5rem;">
* 支持输入股票代码，多个代码请用英文逗号或换行分隔
</div>
</div>
<button type="submit">💾 保存配置</button>
</form>
<div class="footer">
<p>仅用于本地环境 (127.0.0.1) • 安全修改 .env 配置</p>
</div>
</div>
{toast_html}
</body>
</html>"""
return body.encode("utf-8")
class _Handler(BaseHTTPRequestHandler):
def do_GET(self) -> None:
# 解析 URL
parsed = urlparse(self.path)
path = parsed.path
query = parse_qs(parsed.query)
# 路由分发
if path in ("/", ""):
self._handle_index()
elif path == "/health":
self._handle_health()
elif path == "/analysis":
self._handle_analysis(query)
else:
self.send_error(HTTPStatus.NOT_FOUND)
def _handle_index(self) -> None:
"""处理首页请求"""
env_text = _read_env_text(_ENV_PATH)
current = _extract_stock_list(env_text)
payload = _page(current)
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
def _handle_health(self) -> None:
"""
健康检查接口
GET /health
返回:
{
"status": "ok",
"timestamp": "2026-01-19T10:30:00",
"service": "stock-analysis-webui"
}
"""
response = {
"status": "ok",
"timestamp": datetime.now().isoformat(),
"service": "stock-analysis-webui"
}
self._send_json_response(response)
def _handle_analysis(self, query: Dict[str, list]) -> None:
"""
触发单只股票异步分析
GET /analysis?code=xxx
参数:
code: 股票代码（必填）
返回:
{
"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():
self._send_json_response({
"success": False,
"error": "缺少必填参数: code (股票代码)"
}, status=HTTPStatus.BAD_REQUEST)
return
code = code_list[0].strip()
# 验证股票代码格式（6位数字）
if not re.match(r'^\d{6}$', code):
self._send_json_response({
"success": False,
"error": f"无效的股票代码格式: {code} (应为6位数字)"
}, status=HTTPStatus.BAD_REQUEST)
return
# 提交异步分析任务
try:
executor = _get_executor()
future = executor.submit(_run_stock_analysis, code)
task_id = f"{code}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
logger.info(f"[WebUI] 已提交股票 {code} 的分析任务")
self._send_json_response({
"success": True,
"message": "分析任务已提交，将异步执行并推送通知",
"code": code,
"task_id": task_id
})
except Exception as e:
logger.error(f"[WebUI] 提交分析任务失败: {e}")
self._send_json_response({
"success": False,
"error": f"提交任务失败: {str(e)}"
}, status=HTTPStatus.INTERNAL_SERVER_ERROR)
def _send_json_response(
self,
data: Dict[str, Any],
status: HTTPStatus = HTTPStatus.OK
) -> None:
"""发送 JSON 响应"""
payload = json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
def do_POST(self) -> None:
if self.path != "/update":
self.send_error(HTTPStatus.NOT_FOUND)
return
length = int(self.headers.get("Content-Length", "0") or "0")
raw = self.rfile.read(length).decode("utf-8", errors="replace")
form = parse_qs(raw)
stock_list = form.get("stock_list", [""])[0]
env_text = _read_env_text(_ENV_PATH)
updated = _set_stock_list(env_text, stock_list)
_write_env_text(_ENV_PATH, updated)
payload = _page(_normalize_stock_list(stock_list), message="已保存")
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
def log_message(self, fmt: str, *args) -> None:
# quiet default http.server logging
return
def run_server_in_thread(host: str = "127.0.0.1", port: int = 8000):
"""Start the WebUI server in a background thread."""
def serve():
server = ThreadingHTTPServer((host, port), _Handler)
logger.info(f"WebUI 已启动: http://{host}:{port}")
print(f"WebUI 已启动: http://{host}:{port}")
try:
server.serve_forever()
except Exception as e:
logger.error(f"WebUI 发生错误: {e}")
finally:
server.server_close()
t = threading.Thread(target=serve, daemon=True)
t.start()
return t
# 导出所有公共接口（保持向后兼容）
__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 main() -> int:
"""
主入口函数
支持环境变量配置:
WEBUI_HOST: 监听地址 (默认 127.0.0.1)
WEBUI_PORT: 监听端口 (默认 8000)
"""
host = os.getenv("WEBUI_HOST", "127.0.0.1")
port = int(os.getenv("WEBUI_PORT", "8000"))
server = ThreadingHTTPServer((host, port), _Handler)
print(f"WebUI running: http://{host}:{port} (env: {_ENV_PATH})")
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()
try:
server.serve_forever()
run_server(host=host, port=port)
except KeyboardInterrupt:
pass
finally:
server.server_close()
return 0