v2.3.11b

Author: KanoCifer

backend/.gitignore

@@ -32,4 +32,5 @@ trans.sh
 extract_messages.py
 babel.cfg
 legacy/
-jwt
+jwt
+response_data

backend/TASKIQ.md

@@ -4,7 +4,7 @@
 
 ## 架构
 
-- **Broker**: Redis Stream 作为消息队列 (redis://localhost:6379/3)
+- **Broker**: RabbitMQ 作为消息队列
 - **Result Backend**: Redis 存储任务执行结果
 - **Scheduler**: 独立进程管理定时任务
 - **Worker**: 独立进程执行任务
@@ -18,25 +18,17 @@
 
 ## 运行方式
 
-### 1. 安装依赖
-```bash
-cd backend
-pip install -r requirements.txt
-# 或者直接安装新增依赖
-pip install taskiq-scheduler>=0.4.0
-```
-
-### 2. 启动 Worker 进程（执行任务）
+### 启动 Worker 进程（执行任务）
 ```bash
 cd backend
 taskiq worker app.tasks.broker:broker --fs-discover
 ```
 
 参数说明：
 - `--fs-discover`: 自动发现任务文件
-- `--workers N`: 指定 Worker 进程数，默认 1
+- `--workers N`: 指定 Worker 进程数，默认 2
 
-### 3. 启动 Scheduler 进程（调度定时任务）
+### 启动 Scheduler 进程（调度定时任务）
 ```bash
 cd backend
 taskiq scheduler app.tasks.scheduler:scheduler --fs-discover
@@ -46,7 +38,7 @@ taskiq scheduler app.tasks.scheduler:scheduler --fs-discover
 - `--skip-first-run`: 启动时跳过立即执行所有任务
 - `--update-interval N`: 任务调度检查间隔，默认 5 秒
 
-### 4. 启动 FastAPI 主服务（原有服务）
+### 启动 FastAPI 主服务（原有服务）
 ```bash
 cd backend
 python dev.py
@@ -56,72 +48,13 @@ python dev.py
 
 建议使用 systemd 或 supervisor 管理进程：
 
-### Worker 服务配置示例 (`taskiq-worker.service`)
-```ini
-[Unit]
-Description=Taskiq Worker
-After=network.target redis.service
-
-[Service]
-User=www-data
-WorkingDirectory=/path/to/backend
-ExecStart=/path/to/.venv/bin/taskiq worker app.tasks.broker:broker --fs-discover --workers 2
-Restart=always
-RestartSec=5
-
-[Install]
-WantedBy=multi-user.target
-```
-
-### Scheduler 服务配置示例 (`taskiq-scheduler.service`)
-```ini
-[Unit]
-Description=Taskiq Scheduler
-After=network.target redis.service
-
-[Service]
-User=www-data
-WorkingDirectory=/path/to/backend
-ExecStart=/path/to/.venv/bin/taskiq scheduler app.tasks.scheduler:scheduler --fs-discover
-Restart=always
-RestartSec=5
-
-[Install]
-WantedBy=multi-user.target
-```
-
 ## 监控
 
 ### 查看任务执行结果
 ```python
 from app.tasks.broker import broker
 
 # 获取任务结果
-result = await broker.get_result(task_id)
+result = await broker.with_result(task_id)
 print(result.return_value)
 ```
-
-### 任务重试配置
-在 `broker.py` 中添加重试中间件：
-```python
-from taskiq import TaskiqMiddleware
-from taskiq.retry import RetryMiddleware
-
-broker = RedisStreamBroker(
-    url="redis://localhost:6379/3",
-).with_result_backend(result_backend).with_middlewares(
-    RetryMiddleware(
-        max_retries=3,
-        delay=1.0,
-    )
-)
-```
-
-## 从 APScheduler 迁移说明
-
-1. **原有 APScheduler 代码已从 `main.py` 中移除**
-2. **原有任务逻辑保持不变**，仅做了最小修改以适配 taskiq
-3. **任务调度规则完全保持一致**：
-   - 数据迁移任务：每 10 分钟执行一次
-   - RSS 刷新任务：每天上午 10 点执行一次
-4. **异常处理逻辑保持不变**，任务失败时会自动重试，Redis 数据不会丢失

backend/app/configs/config.py

@@ -39,6 +39,7 @@ class Settings(BaseSettings):
     VITE_JS_API_TOKEN: str = ""
     AMAP_SECURITY_CODE: str = ""
     AMAP_WEB_KEY: str = ""
+    JWT_PRIVATE_KEY: str = ""
 
     model_config = SettingsConfigDict(
         env_file=get_env_file_path(),
@@ -65,3 +66,7 @@ def get_settings() -> Settings:
     print(f"SEND_BOOT_EMAIL: {get_settings().SEND_BOOT_EMAIL}")
     print(f"ADMIN_EMAIL: {get_settings().ADMIN_EMAIL}")
     print(f"FEISHU_WEBHOOK_URL: {get_settings().FEISHU_WEBHOOK_URL}")
+    print(f"VITE_JS_API_TOKEN: {get_settings().VITE_JS_API_TOKEN}")
+    print(f"AMAP_SECURITY_CODE: {get_settings().AMAP_SECURITY_CODE}")
+    print(f"AMAP_WEB_KEY: {get_settings().AMAP_WEB_KEY}")
+    print(f"JWT_PRIVATE_KEY: {get_settings().JWT_PRIVATE_KEY}")

backend/app/dependencies/csrf.py

@@ -125,6 +125,7 @@ async def csrf_middleware(request: Request, call_next):
             "/api/v1/auth/email/code",
             "/api/v1/admin/track",
             "/api/v1/admin/deploy",
+            "/api/v1/qweather/tide",
         ]
 
         if any(request.url.path.startswith(path) for path in skip_paths):

backend/app/main.py

@@ -33,12 +33,12 @@
     messages,
     monitor,
     public,
+    publish,
     rss,
     todos,
     users,
     weread,
 )
-from app.tasks import send_bootstrap_emails
 from app.tasks.broker import broker
 from app.tasks.task import send_feishu_message
 from app.utils.cache import close_cache_redis
@@ -84,7 +84,7 @@ async def lifespan(app: FastAPI):
         if lock_acquired:
             try:
                 await send_feishu_message.kiq()
-                await send_bootstrap_emails.kiq(admin_email=admin_email)
+                # await send_bootstrap_emails.kiq(admin_email=admin_email)
                 app_logger.info("✅启动通知任务已添加到队列")
             except Exception as e:
                 logger.warning(f"Failed to queue bootstrap email: {e!s}")
@@ -128,7 +128,9 @@ async def lifespan(app: FastAPI):
 app.include_router(rss.router, prefix="/api/v1")
 app.include_router(monitor.router, prefix="/api/v1")
 app.include_router(aiagent.router, prefix="/api/v1")
+app.include_router(publish.router, prefix="/api/v1")
 
+# 统一注册全局异常处理器
 register_exception_handlers(app)
 
 app.add_middleware(
@@ -144,6 +146,7 @@ async def lifespan(app: FastAPI):
     "https://kanocifer.chat",
 ]
 
+# 配置 CORS 中间件，允许前端访问 API，并暴露 Set-Cookie 头以支持跨域认证
 app.add_middleware(
     CORSMiddleware,
     allow_origins=origins,
@@ -160,6 +163,12 @@ async def add_process_time_header(request: Request, call_next):
     start_time: float = time.perf_counter()
     response = await call_next(request)
     process_time: float = round(time.perf_counter() - start_time, 6)
+
+    # 记录长时间异常请求
+    if process_time > 1.0:  # 1秒以上的请求视为慢请求
+        app_logger.warning(
+            f"Request to {request.url.path} took {process_time}s and returned status code {response.status_code}"
+        )
     response.headers["X-Process-Time"] = f"{process_time}s"
     return response
 
@@ -173,6 +182,7 @@ async def add_process_time_header(request: Request, call_next):
     name="media",
 )
 
+# 注册慢速API的速率限制异常处理器
 app.add_exception_handler(
     exc_class_or_status_code=RateLimitExceeded,
     handler=_rate_limit_exceeded_handler,  # type: ignore

backend/app/routers/public.py

@@ -13,6 +13,7 @@
 from xml.etree.ElementTree import Element, SubElement, tostring
 
 import httpx
+import orjson
 from fastapi import APIRouter, Body, Depends, Request
 from fastapi.responses import JSONResponse, PlainTextResponse
 from redis.asyncio import Redis as AsyncRedis
@@ -22,6 +23,7 @@
 from app.dependencies.mongo import get_mongo_db
 from app.dependencies.redis import get_redis
 from app.schemas.response import APIResponse
+from app.utils.qweather_jwt import encoded_jwt
 
 router = APIRouter(tags=["public"])
 
@@ -325,6 +327,7 @@ async def get_weather(
     request: Request,
     city: str = Body(..., description="City adcode"),
     extensions: str = Body("base", description="Weather type: base/all"),
+    redis: AsyncRedis = Depends(get_redis),
 ) -> JSONResponse:
     """Get weather information from Amap API.
 
@@ -335,6 +338,15 @@ async def get_weather(
     Returns:
         JSONResponse: Weather data from Amap API
     """
+
+    # 尝试命中缓存
+    cache_key = f"weather:{city}:{extensions}"
+    cached_data = await redis.get(cache_key)
+    if cached_data:
+        return APIResponse.ok(
+            data=orjson.loads(cached_data),
+            message="Weather information retrieved from cache",
+        )
     url = "https://restapi.amap.com/v3/weather/weatherInfo"
     params = {
         "key": get_settings().AMAP_WEB_KEY,
@@ -345,6 +357,10 @@ async def get_weather(
         response = await client.get(url, params=params)
         data = response.json()
 
+    # Redis缓存天气数据，过期时间60分钟
+    await redis.set(
+        f"weather:{city}:{extensions}", orjson.dumps(data), ex=60 * 60
+    )
     return APIResponse.ok(
         data=data,
         message="Weather information retrieved successfully",
@@ -383,27 +399,99 @@ async def reverse_geocode(
     )
 
 
-# @router.post("/geocode/regeo")
-# @limiter.limit("100/hour")
-# async def reverse_geocode(
-#     request: Request,
-#     params: dict = Body(..., description="Reverse geocode parameters"),
-# ) -> JSONResponse:
-#     """Reverse geocode coordinates to address using Amap API.
-
-#     Args:
-#         params: Reverse geocode parameters including location coordinates
-
-#     Returns:
-#         JSONResponse: Address information including city adcode
-#     """
-#     url = "https://restapi.amap.com/v3/geocode/regeo"
-#     params["key"] = get_settings().AMAP_WEB_KEY
-#     async with httpx.AsyncClient() as client:
-#         response = await client.get(url, params=params)
-#         data = response.json()
-
-#     return APIResponse.ok(
-#         data=data,
-#         message="Reverse geocode completed successfully",
-#     )
+@router.get("/qweather/tide")
+@limiter.limit("100/hour")
+async def get_qweather(
+    request: Request,
+    redis: AsyncRedis = Depends(get_redis),
+) -> JSONResponse:
+    """Get weather information from QWeather API."""
+
+    # 尝试命中缓存
+    now = datetime.now().strftime("%Y%m%d")
+    cache_key = f"qweather:tide:P2352:{now}"
+    cached_data = await redis.get(cache_key)
+    if cached_data:
+        return APIResponse.ok(
+            data=orjson.loads(cached_data),
+            message="QWeather information retrieved from cache",
+        )
+    try:
+        url = "https://qk2tupqwuj.re.qweatherapi.com/v7/ocean/tide"
+        headers = {
+            "Authorization": f"Bearer {encoded_jwt}",
+        }
+        now = datetime.now().strftime("%Y%m%d")
+        payload = {
+            "location": "P2352",  # 黄埔港
+            "date": now,
+        }
+        async with httpx.AsyncClient() as client:
+            response = await client.get(url, headers=headers, params=payload)
+            try:
+                response.raise_for_status()
+            except httpx.HTTPStatusError:
+                return APIResponse.error(
+                    message=f"QWeather API error: {response.status_code} - {response.text}",
+                    code=response.status_code,
+                )
+            data = response.json()
+
+        # 缓存数据8小时，过期后自动删除
+        cache_key = f"qweather:tide:{payload['location']}:{payload['date']}"
+        await redis.set(cache_key, orjson.dumps(data), ex=8 * 3600)
+        return APIResponse.ok(
+            data=data,
+            message="QWeather information retrieved successfully",
+        )
+    except httpx.HTTPError as e:
+        return APIResponse.error(
+            message=f"Failed to fetch QWeather data: {e!s}",
+            code=503,
+        )
+    except Exception as e:
+        return APIResponse.error(
+            message=f"Internal server error: {e!s}",
+            code=500,
+        )
+
+
+@router.get("/qweather/location")
+@limiter.limit("100/hour")
+async def get_qweather_location(
+    request: Request,
+    location: str,
+    type: str = "scenic",
+) -> JSONResponse:
+    """Get location information from QWeather API."""
+    try:
+        url = "https://qk2tupqwuj.re.qweatherapi.com/geo/v2/poi/lookup"
+        headers = {
+            "Authorization": f"Bearer {encoded_jwt}",
+        }
+        params = {"location": location, "type": type}
+        async with httpx.AsyncClient() as client:
+            response = await client.get(url, headers=headers, params=params)
+            try:
+                response.raise_for_status()
+            except httpx.HTTPStatusError:
+                return APIResponse.error(
+                    message=f"QWeather API error: {response.status_code} - {response.text}",
+                    code=response.status_code,
+                )
+            data = response.json()
+
+        return APIResponse.ok(
+            data=data,
+            message="QWeather location information retrieved successfully",
+        )
+    except httpx.HTTPError as e:
+        return APIResponse.error(
+            message=f"Failed to fetch QWeather location: {e!s}",
+            code=503,
+        )
+    except Exception as e:
+        return APIResponse.error(
+            message=f"Internal server error: {e!s}",
+            code=500,
+        )

backend/app/routers/publish.py

@@ -0,0 +1,5 @@
+from __future__ import annotations
+
+from fastapi import APIRouter
+
+router = APIRouter(prefix="/publish", tags=["publish"])