apps/backend/
├── app/
│   ├── main.py              # FastAPI 入口（lifespan 管理 + 路由注册）
│   ├── config.py            # 配置加载（根 .env < .env.local < 环境变量）
│   ├── database.py          # 数据库连接（zqyy_app 读写 + etl_feiqiu 只读 RLS）
│   ├── auth/                # 认证模块
│   │   ├── dependencies.py  # FastAPI 依赖注入（CurrentUser）
│   │   └── jwt.py           # JWT 签发/验证/密码哈希
│   ├── routers/             # 17 个路由模块（详见 API 参考）
│   ├── schemas/             # Pydantic 请求/响应模型
│   ├── services/            # 业务逻辑层
│   ├── middleware/          # 中间件
│   └── ws/                  # WebSocket（实时日志）
├── tests/                   # 后端测试
├── pyproject.toml           # 依赖声明
└── README.md

启动

# 安装依赖（在 monorepo 根目录）
uv sync --all-packages

# 启动开发服务器
cd apps/backend
uv run uvicorn app.main:app --host 127.0.0.1 --port 8000 --reload

API 文档自动生成于：

Swagger UI: http://localhost:8000/docs
ReDoc: http://localhost:8000/redoc

架构概览

双数据库连接

连接	数据库	用途	函数
业务读写	`zqyy_app`	用户、队列、调度、执行日志	`get_connection()`
ETL 只读	`etl_feiqiu`	数据查看器、游标监控	`get_etl_readonly_connection(site_id)`

ETL 只读连接自动设置 default_transaction_read_only = on 和 RLS app.current_site_id，确保门店数据隔离。

认证体系

后端支持两套认证：

管理后台认证（/api/auth/*）：用户名 + 密码 → JWT
小程序认证（/api/xcx-auth/*）：微信 code → openid → JWT

JWT 令牌分两种：

完整令牌：已审批用户，包含 user_id + site_id + roles
受限令牌（limited=True）：待审批用户，仅允许访问申请相关端点

依赖注入：

get_current_user：要求完整令牌
get_current_user_or_limited：允许受限令牌

配置加载

优先级（低 → 高）：根 .env → 应用 .env.local → 环境变量

关键配置项：

DB_HOST / DB_PORT / DB_USER / DB_PASSWORD / APP_DB_NAME — 业务数据库
ETL_DB_* — ETL 数据库（缺省复用业务库参数）
JWT_SECRET_KEY — JWT 签名密钥（生产必须设置）
CORS_ORIGINS — 允许的跨域来源
ETL_PROJECT_PATH — ETL CLI 工作目录
WX_APPID / WX_SECRET — 微信小程序配置
WX_DEV_MODE — 开发模式（启用 mock 登录）
BUSINESS_DAY_START_HOUR — 营业日分割点（默认 8 点）

后台服务

应用启动时通过 lifespan 拉起两个后台服务：

TaskQueue：任务执行队列（从 task_queue 表消费，按 site_id 隔离）
Scheduler：定时调度器（从 scheduled_tasks 表读取配置，自动入队）

同时注册触发器 job handler：

task_generator：任务生成器（每日凌晨 4:00）
task_expiry_check：任务过期检查（每小时）
recall_completion_check：召回完成检测（ETL 数据更新后）
note_reclassify_backfill：备注重分类回填（召回完成后）

路由总览

前缀	模块	说明	认证
`/api/auth`	`auth.py`	管理后台登录/刷新令牌	无
`/api/xcx-auth`	`xcx_auth.py`	小程序微信登录/申请/状态/店铺切换	部分
`/api/xcx/tasks`	`xcx_tasks.py`	小程序任务列表/置顶/放弃	JWT
`/api/xcx/notes`	`xcx_notes.py`	小程序备注 CRUD	JWT
`/api/admin/applications`	`admin_applications.py`	管理端申请审核	JWT
`/api/business-day`	`business_day.py`	营业日配置查询	JWT
`/api/tasks`	`tasks.py`	任务注册表/Flow 定义/配置验证	JWT
`/api/execution`	`execution.py`	任务执行/队列管理/历史/日志	JWT
`/api/schedules`	`schedules.py`	调度任务 CRUD + 启停	JWT
`/api/db`	`db_viewer.py`	ETL 数据库只读查看器	JWT
`/api/etl-status`	`etl_status.py`	ETL 游标/执行记录监控	JWT
`/api/env-config`	`env_config.py`	环境变量查看/编辑	JWT
`/api/xcx-test`	`xcx_test.py`	MVP 全链路验证	无
`/api/wx-callback`	`wx_callback.py`	微信消息推送回调	签名验证
`/api/retention-clue`	`member_retention_clue.py`	维客线索 CRUD	JWT
`/api/ops`	`ops_panel.py`	运维面板（服务启停/Git/系统信息）	无
`/ws/logs`	`ws/logs.py`	WebSocket 实时日志推送	—
`/health`	`main.py`	健康检查	无

详细 API 端点说明见 docs/API-REFERENCE.md

服务层

模块	职责
`wechat.py`	微信 `code2session` 接口调用
`matching.py`	用户申请时的人员匹配（助教/员工）
`application.py`	用户入驻申请的创建/审批/拒绝
`role.py`	用户权限查询/门店角色检查
`scheduler.py`	定时调度引擎（cron 解析 + next_run 计算）
`task_executor.py`	ETL CLI 子进程执行 + 日志广播
`task_queue.py`	任务队列管理（入队/消费/重排）
`task_registry.py`	ETL 任务/Flow/DWD 表静态注册表
`cli_builder.py`	ETL CLI 命令构建器
`task_generator.py`	任务生成器（基于 WBI/NCI 指数）
`task_manager.py`	任务管理（置顶/放弃/状态变更）
`task_expiry.py`	任务过期检查与处理
`note_service.py`	备注服务（CRUD + 星星评分）
`note_reclassifier.py`	备注重分类（召回完成后回填）
`recall_detector.py`	召回完成检测（ETL 数据更新触发）
`trigger_scheduler.py`	触发器调度器（cron/interval/event）

依赖

fastapi>=0.115
uvicorn[standard]>=0.34
psycopg2-binary>=2.9
python-dotenv>=1.0
python-jose[cryptography]>=3.3
bcrypt>=4.0
psutil>=5.9
neozqyy-shared（workspace 引用）

Roadmap

RBAC 权限中间件（基于 auth.role_permissions 表）
数据看板 API（助教业绩、财务日报、客户分析）
任务审批流 API
消息推送（微信模板消息/订阅消息）
API 限流与审计日志

README.md Unescape Escape

apps/backend — FastAPI 后端

技术栈

目录结构

启动