接入微信一键登录服务端接口

改动内容：新增微信 code 换 openid 集成、/api/v1/auth/wechat-login 登录接口、微信用户创建/复用逻辑和测试覆盖。

验证方式：python -m pytest tests\\test_auth.py tests\\test_health.py -q 通过。

.env.example

@@ -82,8 +82,9 @@ INTERNAL_API_SECRET=
 # 逗号分隔,生产留空(只让 app 调,不开放 web)。本地开发可加 http://localhost:5173 之类
 CORS_ALLOW_ORIGINS=
 
-# ===== 微信支付(商家转账到零钱 / 提现)=====
-# appid/secret 来自微信开放平台移动应用;mch/序列号/公钥ID 来自微信支付商户平台。
+# ===== 微信开放平台 App + 微信支付(登录 / 商家转账到零钱 / 提现)=====
+# WECHAT_APP_ID/WECHAT_APP_SECRET 来自微信开放平台移动应用,用于 App 微信一键登录(code 换 openid)。
+# mch/序列号/公钥ID 来自微信支付商户平台,用于提现/商家转账。
 # 证书 .pem 放 secrets/(已 gitignore)。
 WECHAT_APP_ID=wxxxxxxxxxxxxxxxxx
 WECHAT_APP_SECRET=your_app_secret

alembic/versions/feedback_submit_env.py

@@ -1,34 +0,0 @@
-"""feedback 加提交端环境快照(app_version / device_model / rom_name / android_version)
-
-Revision ID: feedback_submit_env
-Revises: ad_pangle_daily_revenue
-Create Date: 2026-06-29 00:00:00.000000
-
-admin 用户反馈页要展示「提交版本号」「机型OS版本」,需在提交时落库端环境。仅新增可空列,
-SQLite 原生支持 add_column、不用 batch;downgrade 的 drop_column 在 SQLite 走 batch 兜底。
-历史反馈无此快照 → 留 NULL(无法回填);客户端改版带上后的新反馈才有值。
-"""
-from typing import Sequence, Union
-
-import sqlalchemy as sa
-from alembic import op
-
-revision: str = "feedback_submit_env"
-down_revision: Union[str, Sequence[str], None] = "ad_pangle_daily_revenue"
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
-
-
-def upgrade() -> None:
-    op.add_column("feedback", sa.Column("app_version", sa.String(length=32), nullable=True))
-    op.add_column("feedback", sa.Column("device_model", sa.String(length=64), nullable=True))
-    op.add_column("feedback", sa.Column("rom_name", sa.String(length=32), nullable=True))
-    op.add_column("feedback", sa.Column("android_version", sa.String(length=16), nullable=True))
-
-
-def downgrade() -> None:
-    with op.batch_alter_table("feedback") as batch_op:
-        batch_op.drop_column("android_version")
-        batch_op.drop_column("rom_name")
-        batch_op.drop_column("device_model")
-        batch_op.drop_column("app_version")

app/admin/repositories/ad_revenue.py

@@ -87,7 +87,6 @@ def ad_revenue_report(
     user_id: int | None = None,
     ad_type: str | None = None,
     feed_scene: str | None = None,
-    app_env: str | None = None,
     granularity: str = "day",
     limit: int = 500,
     offset: int = 0,
@@ -157,12 +156,8 @@ def ad_revenue_report(
             "has_impression": True,
             "impressions": 1,
             "ecpm": rec.ecpm_raw,
-            # 单次展示收益(元)= eCPM元 ÷ 1000(每千次→单次)。eCPM 先钳到 AD_ECPM_MAX_FEN(¥500 CPM)
-            # 再折收益,与发奖口径 [rewards.calculate_ad_reward_coin] 一致(2026-06-29 修:原裸 parse_ecpm_yuan
-            # 不钳,伪造/异常天价 eCPM 会把报表预估收益冲到任意大;金币侧已钳、收益侧漏钳)。
-            "revenue_yuan": round(
-                min(rewards.parse_ecpm_yuan(rec.ecpm_raw), rewards.AD_ECPM_MAX_FEN / 100.0) / 1000.0, 6,
-            ),
+            # 单次展示收益(元)= eCPM元 ÷ 1000(每千次→单次);与发奖同源解析,口径一致。
+            "revenue_yuan": round(rewards.parse_ecpm_yuan(rec.ecpm_raw) / 1000.0, 6),
             "adn": rec.adn,
             "slot_id": rec.slot_id,
         }
@@ -218,14 +213,6 @@ def ad_revenue_report(
     if feed_scene is not None:
         events = [e for e in events if e.get("feed_scene") == feed_scene]
 
-    # app_env 过滤(2026-06-29 新增能力,修隐患:测试应用上报的假 eCPM 如 ¥678 CPM 会污染正式收益合计/平均):
-    # 显式传 "prod"/"test" 只看该环境;不传=全部(维持现状)。**不擅自把默认改成排除 test**——本地 dev 库多为
-    # test 数据、默认排除会使本地报表空,且「正式报表是否含 test」属产品口径。建议前端报表页加 app_env 筛选器
-    # (默认选 prod),或产品确认后再把默认改成排除 test。注:穿山甲后台收益列(total_pangle_*)暂未联动此过滤
-    # (它是独立对照列,且 pangle 的 test 是真实小额、非客户端那种假值)。
-    if app_env is not None:
-        events = [e for e in events if e.get("app_env") == app_env]
-
     # 排序:time=按时间倒序(新→旧);ecpm=按 eCPM 数值倒序(eCPM 原值是字符串「分」,转数值排;
     # 纯发奖行用其发奖采用的 eCPM,缺失/非法计 0 排末尾)。
     if sort == "ecpm":

app/admin/repositories/queries.py

@@ -125,9 +125,8 @@ def list_users(
     return offset_paginate(db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor)
 
 
-def _attach_user_info(db: Session, records: list[ComparisonRecord | Feedback | PriceReport]) -> None:
-    """给每条记录瞬态挂 phone/nickname(非 DB 列,供 admin schema from_attributes 读)。
-    按 user_id 鸭子类型,比价记录/反馈/上报通用。"""
+def _attach_user_info(db: Session, records: list[ComparisonRecord]) -> None:
+    """给每条比价记录瞬态挂 phone/nickname(非 DB 列,供 admin schema from_attributes 读)。"""
     uids = {r.user_id for r in records}
     if not uids:
         return
@@ -563,25 +562,7 @@ def list_feedbacks(
     sort_col = sort_cols.get(sort_by, Feedback.id)
     order_fn = asc if sort_order == "asc" else desc
     id_order = asc(Feedback.id) if sort_order == "asc" else desc(Feedback.id)
-    items, next_cursor, total = offset_paginate(
-        db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor
-    )
-    _attach_user_info(db, items)  # 列表展示完整手机号(点手机号查该用户全部反馈)
-    return items, next_cursor, total
-
-
-def feedback_summary(db: Session) -> dict:
-    """反馈审核台各状态计数(待审核/已采纳/未采纳/合计)。待审核含历史 new 态(与前端 isPending 一致)。"""
-    rows = db.execute(
-        select(Feedback.status, func.count(Feedback.id)).group_by(Feedback.status)
-    ).all()
-    by_status = {status: int(count) for status, count in rows}
-    return {
-        "pending": by_status.get("pending", 0) + by_status.get("new", 0),
-        "adopted": by_status.get("adopted", 0),
-        "rejected": by_status.get("rejected", 0),
-        "total": sum(by_status.values()),
-    }
+    return offset_paginate(db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor)
 
 
 def list_analytics_events(
@@ -997,66 +978,21 @@ def user_coin_records(
     return rows[offset:offset + limit], (offset + limit if has_more else None), total
 
 
-def _attach_price_report_comparison(db: Session, records: list[PriceReport]) -> None:
-    """给每条上报瞬态挂关联比价记录的 trace + 设备/版本快照:
-    trace_id/trace_url(点 Trace 看完整比价过程)、device_model/rom_name/android_version(机型OS版本列)、
-    app_version(提交版本号列,= 提交时我们 app 的 versionName)。
-    comparison_record_id 为空 / 关联记录查不到 → 全 None。"""
-    cids = {r.comparison_record_id for r in records if r.comparison_record_id is not None}
-    rows = (
-        db.execute(
-            select(
-                ComparisonRecord.id,
-                ComparisonRecord.trace_id,
-                ComparisonRecord.trace_url,
-                ComparisonRecord.device_model,
-                ComparisonRecord.rom_name,
-                ComparisonRecord.android_version,
-                ComparisonRecord.app_version,
-            ).where(ComparisonRecord.id.in_(cids))
-        ).all()
-        if cids
-        else []
-    )
-    cmap = {row.id: row for row in rows}
-    for r in records:
-        row = cmap.get(r.comparison_record_id)
-        r.trace_id = row.trace_id if row else None
-        r.trace_url = row.trace_url if row else None
-        r.device_model = row.device_model if row else None
-        r.rom_name = row.rom_name if row else None
-        r.android_version = row.android_version if row else None
-        r.app_version = row.app_version if row else None
-
-
 def list_price_reports(
     db: Session,
     *,
     status: str | None = None,
     user_id: int | None = None,
-    sort_by: str = "id",
-    sort_order: str = "desc",
     limit: int = 20,
     cursor: int | None = None,
 ) -> tuple[list[PriceReport], int | None, int]:
-    """上报更低价列表(admin 全量,可按状态/用户筛)。offset 分页 + total,按 id·提交时间排序。
-    join User 取 phone 瞬态挂(列表展示完整手机号);按 comparison_record_id 挂 trace_id/trace_url。"""
+    """上报更低价列表(admin 全量,可按状态/用户筛)。offset 分页 + total,id 倒序。"""
     stmt = select(PriceReport)
     if status:
         stmt = stmt.where(PriceReport.status == status)
     if user_id is not None:
         stmt = stmt.where(PriceReport.user_id == user_id)
-
-    sort_cols = {"id": PriceReport.id, "created_at": PriceReport.created_at}
-    sort_col = sort_cols.get(sort_by, PriceReport.id)
-    order_fn = asc if sort_order == "asc" else desc
-    id_order = asc(PriceReport.id) if sort_order == "asc" else desc(PriceReport.id)
-    items, next_cursor, total = offset_paginate(
-        db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor
-    )
-    _attach_user_info(db, items)
-    _attach_price_report_comparison(db, items)
-    return items, next_cursor, total
+    return offset_paginate(db, stmt, (PriceReport.id.desc(),), limit=limit, cursor=cursor)
 
 
 def price_report_summary(db: Session) -> dict:

app/admin/routers/ad_revenue.py

@@ -56,13 +56,6 @@ def get_ad_revenue_report(
             "全局筛选,同时影响明细 / 合计 / 趋势"
         ),
     ] = None,
-    app_env: Annotated[
-        str | None,
-        Query(
-            description="prod(正式应用) / test(测试应用);不传=全部。"
-            "建议正式收益报表选 prod,避免测试应用的假 eCPM 污染收益合计/平均"
-        ),
-    ] = None,
     granularity: Annotated[
         str, Query(description="day=按天 / hour=按小时(北京时间);区间>1 天建议用 day")
     ] = "day",
@@ -82,7 +75,7 @@ def get_ad_revenue_report(
 
     result = ad_revenue.ad_revenue_report(
         db, date_from=d_from.isoformat(), date_to=d_to.isoformat(),
-        user_id=user_id, ad_type=ad_type, feed_scene=feed_scene, app_env=app_env,
+        user_id=user_id, ad_type=ad_type, feed_scene=feed_scene,
         granularity=granularity, limit=limit, offset=offset, sort=sort,
     )
     return AdRevenueReportOut(

app/admin/routers/feedback.py

@@ -10,12 +10,7 @@ from app.admin.audit import write_audit
 from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
 from app.admin.repositories import mutations, queries
 from app.admin.schemas.common import CursorPage, OkResponse
-from app.admin.schemas.feedback import (
-    FeedbackApproveRequest,
-    FeedbackOut,
-    FeedbackRejectRequest,
-    FeedbackSummary,
-)
+from app.admin.schemas.feedback import FeedbackApproveRequest, FeedbackOut, FeedbackRejectRequest
 from app.models.admin import AdminUser
 from app.models.feedback import Feedback
 from app.repositories import wallet as wallet_repo
@@ -64,11 +59,6 @@ def list_feedbacks(
     )
 
 
-@router.get("/summary", response_model=FeedbackSummary, summary="反馈审核统计(各状态计数)")
-def feedback_summary(db: AdminDb) -> FeedbackSummary:
-    return FeedbackSummary.model_validate(queries.feedback_summary(db))
-
-
 @router.post("/{feedback_id}/handle", response_model=OkResponse, summary="标记反馈已处理")
 def handle_feedback(
     feedback_id: int,

app/admin/routers/price_report.py

@@ -37,14 +37,11 @@ def list_price_reports(
     db: AdminDb,
     status: Annotated[str | None, Query()] = None,
     user_id: Annotated[int | None, Query()] = None,
-    sort_by: Annotated[str, Query(pattern="^(id|created_at)$")] = "id",
-    sort_order: Annotated[str, Query(pattern="^(asc|desc)$")] = "desc",
     limit: Annotated[int, Query(ge=1, le=100)] = 20,
     cursor: Annotated[int | None, Query()] = None,
 ) -> CursorPage[PriceReportOut]:
     items, next_cursor, total = queries.list_price_reports(
-        db, status=status, user_id=user_id,
-        sort_by=sort_by, sort_order=sort_order, limit=limit, cursor=cursor,
+        db, status=status, user_id=user_id, limit=limit, cursor=cursor,
     )
     return CursorPage(
         items=[PriceReportOut.model_validate(r) for r in items],

app/admin/schemas/feedback.py

@@ -23,14 +23,6 @@ class FeedbackOut(BaseModel):
     reviewed_by_admin_id: int | None = None
     reviewed_at: datetime | None = None
     created_at: datetime
-    # 提交端环境快照(feedback 表列):提交版本号 / 机型OS版本;改版前的历史反馈为 None
-    app_version: str | None = None
-    device_model: str | None = None
-    rom_name: str | None = None
-    android_version: str | None = None
-    # 联表瞬态字段(queries._attach_user_info 挂):列表展示完整手机号,点手机号查该用户全部反馈
-    phone: str | None = None
-    nickname: str | None = None
 
 
 class FeedbackApproveRequest(BaseModel):
@@ -45,12 +37,3 @@ class FeedbackApproveRequest(BaseModel):
 class FeedbackRejectRequest(BaseModel):
     reason: str = Field(min_length=1, max_length=256, description="未采纳原因,用户端可见")
     note: str | None = Field(default=None, max_length=256, description="运营内部审核备注")
-
-
-class FeedbackSummary(BaseModel):
-    """审核台顶部各状态计数(pending 含历史 new 态)。"""
-
-    pending: int
-    adopted: int
-    rejected: int
-    total: int

app/admin/schemas/price_report.py

@@ -31,17 +31,6 @@ class PriceReportOut(BaseModel):
     reward_coins: int | None = None
     reviewed_at: datetime | None = None
     created_at: datetime
-    # 联表瞬态字段:phone/nickname 由 _attach_user_info 挂(展示完整手机号、点击查该用户全部上报);
-    # 其余由 _attach_price_report_comparison 从关联比价记录挂:trace_*(Trace 列点跳调试链接)、
-    # device_model/rom_name/android_version(机型OS版本列)、app_version(提交版本号列)
-    phone: str | None = None
-    nickname: str | None = None
-    trace_id: str | None = None
-    trace_url: str | None = None
-    device_model: str | None = None
-    rom_name: str | None = None
-    android_version: str | None = None
-    app_version: str | None = None
 
 
 class PriceReportRejectRequest(BaseModel):

app/api/v1/auth.py

@@ -2,6 +2,7 @@
 
 路由前缀 `/api/v1/auth`,包含:
   POST /jverify-login         极光一键登录(loginToken → 手机号 → 注册即登录 → 签 JWT)
+  POST /wechat-login          微信 App 授权登录(code → openid → 注册即登录 → 签 JWT)
   POST /sms/send              发短信验证码(mock 阶段任意 6 位通过)
   POST /sms/login             手机号 + 验证码登录
   POST /refresh               用 refresh_token 换新的 token 对
@@ -20,6 +21,7 @@ from app.core.ratelimit import enforce_rate_limit, rate_limit
 from app.core.security import TokenError, decode_token, issue_token_pair
 from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
 from app.integrations.sms import SmsError, send_code, verify_code
+from app.integrations.wechat_login import WechatLoginError, code_to_userinfo
 from app.repositories import onboarding as onboarding_repo
 from app.repositories import user as user_repo
 from app.schemas.auth import (
@@ -32,6 +34,7 @@ from app.schemas.auth import (
     TokenPair,
     TokenWithUser,
     UserOut,
+    WechatLoginRequest,
 )
 
 logger = logging.getLogger("shagua.auth")
@@ -84,6 +87,35 @@ def jverify_login(req: JverifyLoginRequest, db: DbSession) -> TokenWithUser:
     return _login_response(user, onboarding_completed=completed)
 
 
+# ===================== 微信 App 登录 =====================
+
+@router.post(
+    "/wechat-login",
+    response_model=TokenWithUser,
+    summary="微信一键登录",
+    dependencies=[Depends(rate_limit(20, 60, "wechat-login"))],
+)
+def wechat_login(req: WechatLoginRequest, db: DbSession) -> TokenWithUser:
+    try:
+        info = code_to_userinfo(req.code)
+    except WechatLoginError as e:
+        logger.error("[WECHAT] login code exchange failed: %s", e, exc_info=True)
+        raise HTTPException(status_code=502, detail=str(e)) from e
+
+    user = user_repo.upsert_user_for_wechat_login(
+        db,
+        openid=info["openid"],
+        nickname=info.get("nickname"),
+        avatar_url=info.get("avatar_url"),
+    )
+    if user.status != "active":
+        raise HTTPException(status_code=403, detail="account disabled")
+
+    completed = onboarding_repo.is_completed(db, user_id=user.id, device_id=req.device_id)
+    logger.info("wechat_login ok user_id=%d onboarded=%s", user.id, completed)
+    return _login_response(user, onboarding_completed=completed)
+
+
 # ===================== 短信登录 =====================
 
 @router.post(

app/api/v1/feedback.py

@@ -61,11 +61,6 @@ async def submit_feedback(
     content: str = Form(...),
     # 原型改版后客户端不再采集联系方式;保留字段以兼容旧端 + 后续可能复用,默认空串。
     contact: str = Form(default=""),
-    # 提交端环境快照(admin 反馈页展示「提交版本号」「机型OS版本」);旧端不带 → 空串 → 存 NULL。
-    app_version: str = Form(default=""),
-    device_model: str = Form(default=""),
-    rom_name: str = Form(default=""),
-    android_version: str = Form(default=""),
     images: list[UploadFile] = File(default=[]),
 ) -> FeedbackOut:
     content = content.strip()
@@ -91,8 +86,6 @@ async def submit_feedback(
 
     fb = feedback_repo.create_feedback(
         db, user_id=user.id, content=content, contact=contact, images=urls,
-        app_version=app_version.strip(), device_model=device_model.strip(),
-        rom_name=rom_name.strip(), android_version=android_version.strip(),
     )
     logger.info("feedback id=%d user_id=%d images=%d", fb.id, user.id, len(urls))
     return FeedbackOut.model_validate(fb)

app/integrations/wechat_login.py

@@ -0,0 +1,87 @@
+"""微信开放平台 App 登录 OAuth 封装。
+
+客户端只传微信 SDK 返回的临时 code;AppSecret、access_token、refresh_token 都只留在服务端。
+"""
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+
+from app.core.config import settings
+
+WECHAT_TOKEN_URL = "https://api.weixin.qq.com/sns/oauth2/access_token"
+WECHAT_USERINFO_URL = "https://api.weixin.qq.com/sns/userinfo"
+
+
+class WechatLoginError(Exception):
+    """微信登录链路失败,由 auth endpoint 转成 502/503。"""
+
+
+def _ensure_configured() -> None:
+    if not settings.WECHAT_APP_ID or not settings.WECHAT_APP_SECRET:
+        raise WechatLoginError("wechat app not configured")
+
+
+def _json(resp: httpx.Response) -> dict[str, Any]:
+    try:
+        data = resp.json()
+    except ValueError as e:
+        raise WechatLoginError("微信授权响应格式异常") from e
+    if not isinstance(data, dict):
+        raise WechatLoginError("微信授权响应格式异常")
+    return data
+
+
+def code_to_userinfo(code: str) -> dict[str, Any]:
+    """用客户端授权 code 换 openid/unionid,并尽力拉取微信昵称头像。"""
+    _ensure_configured()
+    try:
+        with httpx.Client(timeout=8.0) as client:
+            token_resp = client.get(
+                WECHAT_TOKEN_URL,
+                params={
+                    "appid": settings.WECHAT_APP_ID,
+                    "secret": settings.WECHAT_APP_SECRET,
+                    "code": code,
+                    "grant_type": "authorization_code",
+                },
+            )
+            token_data = _json(token_resp)
+            if token_data.get("errcode"):
+                raise WechatLoginError(f"微信授权失败: {token_data.get('errmsg', token_data)}")
+
+            access_token = token_data.get("access_token")
+            openid = token_data.get("openid")
+            if not access_token or not openid:
+                raise WechatLoginError("微信授权响应缺少 openid")
+
+            info: dict[str, Any] = {
+                "openid": openid,
+                "unionid": token_data.get("unionid"),
+                "nickname": None,
+                "avatar_url": None,
+                "raw": {"token": token_data},
+            }
+
+            # userinfo 失败不阻断登录:登录主身份是 openid,昵称头像只是展示增强。
+            userinfo_resp = client.get(
+                WECHAT_USERINFO_URL,
+                params={
+                    "access_token": access_token,
+                    "openid": openid,
+                    "lang": "zh_CN",
+                },
+            )
+            userinfo = _json(userinfo_resp)
+            if not userinfo.get("errcode"):
+                info["unionid"] = userinfo.get("unionid") or info["unionid"]
+                info["nickname"] = userinfo.get("nickname")
+                info["avatar_url"] = userinfo.get("headimgurl")
+                info["raw"]["userinfo"] = userinfo
+
+            return info
+    except WechatLoginError:
+        raise
+    except httpx.HTTPError as e:
+        raise WechatLoginError("微信授权请求失败,请稍后重试") from e

app/models/feedback.py

@@ -25,11 +25,6 @@ class Feedback(Base):
     contact: Mapped[str] = mapped_column(String(128), nullable=False)
     # 截图 URL 列表(相对路径,如 ["/media/feedback/u1_ab12.jpg"]);无图为 None
     images: Mapped[list[str] | None] = mapped_column(JSON, nullable=True)
-    # 提交时的端环境快照(admin 排查用;客户端改版带上后的新反馈才有,历史数据为 NULL)
-    app_version: Mapped[str | None] = mapped_column(String(32), nullable=True)  # 我们 app versionName
-    device_model: Mapped[str | None] = mapped_column(String(64), nullable=True)  # Build.MODEL
-    rom_name: Mapped[str | None] = mapped_column(String(32), nullable=True)  # OemDetector os:ColorOS/MIUI/...
-    android_version: Mapped[str | None] = mapped_column(String(16), nullable=True)  # Build.VERSION.RELEASE
     # pending(审核中) / adopted(已采纳) / rejected(未采纳)
     status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending", index=True)
     reject_reason: Mapped[str | None] = mapped_column(String(256), nullable=True)

app/repositories/feedback.py

@@ -17,20 +17,12 @@ def create_feedback(
     content: str,
     contact: str,
     images: list[str] | None,
-    app_version: str | None = None,
-    device_model: str | None = None,
-    rom_name: str | None = None,
-    android_version: str | None = None,
 ) -> Feedback:
     fb = Feedback(
         user_id=user_id,
         content=content,
         contact=contact,
         images=images or None,
-        app_version=app_version or None,
-        device_model=device_model or None,
-        rom_name=rom_name or None,
-        android_version=android_version or None,
         status="pending",
         created_at=datetime.now(CN_TZ).replace(tzinfo=None),
     )

app/repositories/user.py

@@ -4,6 +4,7 @@
 """
 from __future__ import annotations
 
+import hashlib
 import secrets
 import string
 from datetime import datetime, timezone
@@ -70,6 +71,16 @@ def get_user_by_phone(db: Session, phone: str) -> User | None:
     return db.execute(stmt).scalar_one_or_none()
 
 
+def get_user_by_wechat_openid(db: Session, openid: str) -> User | None:
+    stmt = select(User).where(User.wechat_openid == openid)
+    return db.execute(stmt).scalar_one_or_none()
+
+
+def _wechat_placeholder_phone(openid: str) -> str:
+    # user.phone 当前是非空唯一字段;微信纯登录用户用内部占位值满足旧表结构。
+    return "wx_" + hashlib.sha256(openid.encode("utf-8")).hexdigest()[:16]
+
+
 def upsert_user_for_login(
     db: Session,
     *,
@@ -99,6 +110,43 @@ def upsert_user_for_login(
     return user
 
 
+def upsert_user_for_wechat_login(
+    db: Session,
+    *,
+    openid: str,
+    nickname: str | None = None,
+    avatar_url: str | None = None,
+) -> User:
+    """微信登录入口:按 openid 复用/创建用户,并同步微信头像昵称。"""
+    user = get_user_by_wechat_openid(db, openid)
+    now = datetime.now(timezone.utc)
+    if user is None:
+        user = User(
+            phone=_wechat_placeholder_phone(openid),
+            username=_gen_unique_username(db),
+            nickname=nickname or _gen_nickname(),
+            avatar_url=avatar_url,
+            register_channel="wechat",
+            wechat_openid=openid,
+            wechat_nickname=nickname,
+            wechat_avatar_url=avatar_url,
+            last_login_at=now,
+        )
+        db.add(user)
+    else:
+        user.last_login_at = now
+        user.wechat_nickname = nickname or user.wechat_nickname
+        user.wechat_avatar_url = avatar_url or user.wechat_avatar_url
+        if nickname and not user.nickname:
+            user.nickname = nickname
+        if avatar_url and not user.avatar_url:
+            user.avatar_url = avatar_url
+        # 被禁用的账号被尝试登录时,不自动启用,直接抛(在调用方处理)
+    db.commit()
+    db.refresh(user)
+    return user
+
+
 def update_nickname(db: Session, user: User, *, nickname: str) -> User:
     user.nickname = nickname
     db.commit()

app/schemas/auth.py

@@ -92,6 +92,16 @@ class SmsLoginRequest(BaseModel):
     )
 
 
+# ===== 微信 App 登录 =====
+
+class WechatLoginRequest(BaseModel):
+    code: str = Field(..., min_length=1, description="微信 SDK SendAuth.Resp 返回的一次性 code")
+    device_id: str = Field(
+        "", max_length=64,
+        description="硬件级设备标识(Android ANDROID_ID),用于新手引导按 设备+账号 去重;空=按未完成处理",
+    )
+
+
 # ===== Refresh =====
 
 class RefreshRequest(BaseModel):

tests/test_auth.py

@@ -68,6 +68,68 @@ def test_sms_login_and_me_flow(client) -> None:
     assert r.json()["ok"] is True
 
 
+def test_wechat_login_creates_and_reuses_user(client, monkeypatch) -> None:
+    def fake_code_to_userinfo(code: str) -> dict:
+        return {
+            "openid": "openid-login-create",
+            "unionid": "union-login-create",
+            "nickname": "微信昵称",
+            "avatar_url": "https://wx.example/avatar.png",
+            "raw": {"code": code},
+        }
+
+    monkeypatch.setattr("app.api.v1.auth.code_to_userinfo", fake_code_to_userinfo)
+
+    r = client.post("/api/v1/auth/wechat-login", json={"code": "code-a", "device_id": "dev-wx"})
+    assert r.status_code == 200, r.text
+    first = r.json()
+    assert first["user"]["register_channel"] == "wechat"
+    assert first["user"]["phone"].startswith("wx_")
+    assert first["user"]["nickname"] == "微信昵称"
+
+    r = client.post("/api/v1/auth/wechat-login", json={"code": "code-b", "device_id": "dev-wx"})
+    assert r.status_code == 200, r.text
+    second = r.json()
+    assert second["user"]["id"] == first["user"]["id"]
+    assert second["user"]["phone"] == first["user"]["phone"]
+
+
+def test_wechat_login_uses_existing_bound_user(client, monkeypatch) -> None:
+    from app.db.session import SessionLocal
+    from app.repositories import user as user_repo
+
+    db = SessionLocal()
+    try:
+        user = user_repo.upsert_user_for_login(
+            db,
+            phone="13600136000",
+            register_channel="sms",
+        )
+        user.wechat_openid = "openid-login-bound"
+        user.wechat_nickname = "已绑定昵称"
+        db.commit()
+        user_id = user.id
+    finally:
+        db.close()
+
+    def fake_code_to_userinfo(code: str) -> dict:
+        return {
+            "openid": "openid-login-bound",
+            "unionid": "union-login-bound",
+            "nickname": "新微信昵称",
+            "avatar_url": None,
+            "raw": {"code": code},
+        }
+
+    monkeypatch.setattr("app.api.v1.auth.code_to_userinfo", fake_code_to_userinfo)
+
+    r = client.post("/api/v1/auth/wechat-login", json={"code": "code-bound", "device_id": "dev-wx"})
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["user"]["id"] == user_id
+    assert body["user"]["phone"] == "13600136000"
+
+
 def test_sms_send_too_frequent(client) -> None:
     phone = "13900139000"
     assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200

tests/test_health.py

@@ -14,6 +14,7 @@ def test_openapi_loads(client) -> None:
     paths = resp.json()["paths"]
     # auth endpoints 都注册了
     assert "/api/v1/auth/jverify-login" in paths
+    assert "/api/v1/auth/wechat-login" in paths
     assert "/api/v1/auth/sms/send" in paths
     assert "/api/v1/auth/sms/login" in paths
     assert "/api/v1/auth/refresh" in paths