feat: 实现后端核心业务逻辑 (Auth鉴权、二维码服务、看板与产品端点)

backend/app/services/auth_service.py

@@ -0,0 +1,98 @@
+"""认证服务 — 对接 MOM 系统 sys_user 表 + Track 自有 JWT"""
+from fastapi import HTTPException, status, Depends
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from jose import JWTError, jwt
+from werkzeug.security import check_password_hash
+
+from app.core.config import settings
+from app.core.security import create_access_token, ALGORITHM
+from app.core.mom_database import MomSessionLocal
+from app.schemas.user import LoginResponse, UserResponse
+
+security = HTTPBearer()
+
+
+def login(username: str, password: str) -> LoginResponse:
+    """登录 — 查询 MOM 数据库 sys_user 表验证"""
+    db = MomSessionLocal()
+    try:
+        # 1. 超级管理员硬编码（和 MOM 系统一致）
+        if username == "IRIS" and password == "123321":
+            return LoginResponse(
+                access_token=create_access_token(
+                    data={"sub": "0", "role": "SUPER_ADMIN"}
+                ),
+                user=UserResponse(
+                    id="0",
+                    username="IRIS",
+                    display_name="超级管理员",
+                    role="SUPER_ADMIN",
+                ),
+            )
+
+        # 2. 普通用户：LIKE '%/username' 模糊匹配 MOM sys_user 表
+        from sqlalchemy import text
+        result = db.execute(
+            text(
+                "SELECT id, username, department, role, password_hash "
+                "FROM sys_user "
+                "WHERE username LIKE :pattern"
+            ),
+            {"pattern": f"%/{username}"},
+        )
+        row = result.fetchone()
+
+        if not row:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="用户名或密码错误",
+            )
+
+        user_id, full_username, department, role, password_hash = row
+
+        # 3. Werkzeug scrypt 密码验证
+        if not check_password_hash(password_hash, password):
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="用户名或密码错误",
+            )
+
+        # 4. 解析 display_name（"张三/zhangsan01" → "张三"）
+        display_name = full_username.split("/")[0] if "/" in full_username else full_username
+
+        token = create_access_token(
+            data={
+                "sub": str(user_id),
+                "role": role or "operator",
+                "username": username,
+                "display_name": display_name,
+            }
+        )
+
+        return LoginResponse(
+            access_token=token,
+            user=UserResponse(
+                id=str(user_id),
+                username=username,
+                display_name=display_name,
+                role=role or "operator",
+            ),
+        )
+
+    finally:
+        db.close()
+
+
+async def get_current_user(
+    credentials: HTTPAuthorizationCredentials = Depends(security),
+) -> dict:
+    """从 Bearer Token 解析当前用户（不查数据库，直接解 JWT）"""
+    token = credentials.credentials
+    try:
+        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
+        user_id = payload.get("sub")
+        if not user_id:
+            raise HTTPException(status_code=401, detail="无效的 Token")
+        return payload
+    except JWTError:
+        raise HTTPException(status_code=401, detail="无效的 Token")