交付出件箱

← 返回出件箱

Phase 48A「品牌码扫码直接进家长端」取证调查报告

调查方式:纯只读(grep / sed / ls / git log / python ast),主仓 /home/dev2/doctor-ai-brain(分支 worktree-bridge-cse_01Jc5xvpuj6F2WGDGN9zHfcd,tip b282c143,即当前 main 状态)+ 只读检查 test 分支相关 worktree。全程未 Edit/Write 仓库文件、未 git add/commit/checkout/merge、未跑 alembic、未起服务。

日期:2026-09-16


维度 1 · 微信 OAuth 现状

结论:app/services/wechat_oauth_service.py 已有 3 个公开异步函数,exchange_code(code) 就是现成的"code 换用户信息"入口(公众号 snsapi_userinfo 2 步流),已被两处业务代码复用(h5_user_auth.py 家长流 + seat_auth.py 席位身份绑定流)。品牌码落地页可直接复用 exchange_code,不需要重新实现微信 OAuth。

证据:

$ grep -n "^def \|^async def \|^class " app/services/wechat_oauth_service.py
34:async def exchange_code(code: str) -> dict:
83:async def get_miniapp_access_token() -> str:
121:async def jscode2session(js_code: str) -> dict:

exchange_code(app/services/wechat_oauth_service.py:34-81):

async def exchange_code(code: str) -> dict:
    """公众号 OAuth · code → openid + unionid + nickname + headimgurl
    Step 1: /sns/oauth2/access_token → {access_token, openid, unionid, scope}
    Step 2: /sns/userinfo            → {openid, unionid, nickname, headimgurl, ...}
    Returns merged dict (Step 2 overrides Step 1 on overlap).
    """
    appid = os.getenv("WECHAT_OFFICIAL_APPID")
    secret = os.getenv("WECHAT_OFFICIAL_APPSECRET")
    ...
    return {**data1, **data2}

拿到字段:openid、unionid、nickname、headimgurl(+ access_token/scope 等 step1 字段)。

调用方(真实复用,非孤儿函数):

$ grep -rn "exchange_code" app/api/
app/api/seat_auth.py:53:from app.services.wechat_oauth_service import exchange_code
app/api/seat_auth.py:416:        wx = await exchange_code(body.code)  # wechat_oauth_service
app/api/h5_user_auth.py:28:from app.services.wechat_oauth_service import exchange_code
app/api/h5_user_auth.py:53:    wechat_data = await exchange_code(req.code)

对照组:get_miniapp_access_token 被 app/api/miniapp_phone.py:72 调用、jscode2session 被 app/api/miniapp_auth.py 复刻实现(未直接调用本函数,miniapp_auth.py 自己内联了一份同款逻辑)——证明函数确实存在且被消费,grep 查法有效。


维度 2 · 微信授权的 HTTP 入口

结论:没有后端 GET /callback 路由。微信 OAuth 是"前端发起 + 前端承接"模式:前端页面自己拼 open.weixin.qq.com/connect/oauth2/authorize 跳转,redirect_uri 指回当前前端页面自己,微信带 ?code=&state= 跳回后由前端 JS 检测并 POST 给后端两个 exchange 端点。这是品牌码落地页要复用的既有模式。

证据:

前端发起授权跳转(web/app/h5/chat/lib/wx-oauth.ts:28-46):

export async function startSeatWxOAuth(redirectUri: string): Promise<void> {
  const appId = await fetchWxAppId(redirectUri);
  ...
  const url =
    `https://open.weixin.qq.com/connect/oauth2/authorize` +
    `?appid=${encodeURIComponent(appId)}` +
    `&redirect_uri=${encodeURIComponent(redirectUri)}` +
    `&response_type=code&scope=snsapi_userinfo&state=seat#wechat_redirect`;
  window.location.href = url;
}

后端承接端点(真实 POST,非 GET callback):
1. POST /api/h5/oauth/exchange(家长流,app/api/h5_user_auth.py:35-70)
- oauth_router = APIRouter(prefix="/api/h5/oauth", tags=["h5-oauth"])(line 35)
- @oauth_router.post("/exchange", response_model=ExchangeOAuthResponse)(line 49)
- 请求体:{code: str};返回:{access_token, refresh_token, must_register_phone}
2. POST /api/seat/wx-identity(席位身份绑定流,app/api/seat_auth.py:406-411)
- @router.post("/seat/wx-identity", response_model=WxIdentityResponse)
- 请求体:{code: str};返回:{wx_identity_token, nickname, avatar, seat_username}(不写 DB)

main.py 挂载:

$ grep -n "h5_user_auth\." app/main.py
110:app.include_router(h5_user_auth.oauth_router, tags=["Phase 29B · H5 公众号 OAuth"])

对照组:全仓 grep oauth2/authorize 命中 2 处(wx-oauth.ts + seat-wx-auth-landing.tsx 引用的同款逻辑 + web/app/h5/share/[token]/page.tsx:211),均是前端发起,无一处在后端 app/api/*.py 里——证明"后端无 callback GET 路由"不是查法漏了,是真没有。


维度 3 · 家长席位怎么创建

结论:❌ 任务描述里的文件 app/api/store_subseat.py / app/api/admin_parent_seats.py 物理不存在(仓库经过大量重构,文件名已变)。实际的席位创建逻辑在 app/services/seat_service.py::generate_seats(后端)+ app/api/admin_seats.py::generate_seats(HTTP 层),但这创建的是 org_seat(门店席位账号),不是"家长"身份。"家长"在本系统里对应的是 h5_user角色,走的是 app/services/h5_user_service.py::upsert_by_unionid(微信 OAuth 自动 upsert,无需管理员创建、无 linked_org_id)。这一点对 Phase 48A 很关键:品牌码扫码进家长端,家长身份走的应该是 h5_user upsert 路径,而不是"生成席位"路径。

证据 A(文件不存在,对照组证明查法有效):

$ ls app/api/store_subseat.py
ls: cannot access 'app/api/store_subseat.py': No such file or directory
$ ls app/api/admin_parent_seats.py
ls: cannot access 'app/api/admin_parent_seats.py': No such file or directory
$ ls app/api/ | grep -i seat
admin_seats.py
seat_auth.py
$ ls app/api/auth.py   # 对照组:已知存在的文件,证明 ls 查法本身有效
app/api/auth.py

证据 B(org_seat 创建 · app/services/seat_service.py:156-216):

async def generate_seats(
    org_id: int, count: int, db: AsyncSession,
    valid_days: Optional[int] = None,
    experience_store_name: Optional[str] = None,
) -> list[dict]:
    ...
    for _ in range(count):
        username = await _gen_unique_username(org_id, db)   # 0001202607 格式,全局按月连续序号
        temp_pw = generate_seat_initial_password()          # 6 位随机数字
        seat = User(
            username=username,
            password_hash=hash_password(temp_pw),
            initial_password_expires_at=... + 7天,
            role="org_seat",
            linked_org_id=org_id,
            experience_store_name=normalized_store_name,
            is_active=True,
            must_change_password=True,
            seat_valid_days=valid_days,   # None=永久
        )
        db.add(seat)
        await db.flush()
    await db.commit()

证据 C(h5_user 创建/家长身份 · app/services/h5_user_service.py:27-79):

async def upsert_by_unionid(unionid, db, openid_h5=None, openid_miniapp=None,
                             nickname=None, avatar=None, phone=None) -> User:
    user = (await db.execute(select(User).where(User.wechat_unionid == unionid))).scalar_one_or_none()
    if user:
        # 仅合并缺失字段,nickname/avatar 同步更新
        ...
        return user
    user = User(
        role="h5_user",
        wechat_unionid=unionid,
        wechat_openid_h5=openid_h5,
        wechat_openid_miniapp=openid_miniapp,
        wechat_nickname=nickname,
        wechat_avatar=avatar,
        phone=phone,
        phone_verified_at=... if phone else None,
        is_active=True,
        must_change_password=False,
    )
    db.add(user); await db.commit(); await db.refresh(user)
    return user

关键:h5_user 创建时不写任何 linked_org_id/org_id —— 家长身份在 users 表层面是全平台唯一、跨机构不区分的(org 关联落在 patients.org_id,不在 users 表)。


维度 4 · users 表字段全清单

结论:User 表是 6 角色(admin/org_admin/h5_user/org_seat/channel/platform_staff)合一表,角色枚举由 DB 级 CheckConstraint 定死(chk_role),每角色必填字段由第二条 CheckConstraint(chk_role_fields)定死。h5_user 角色明确不要求 linked_org_id。

证据(app/models/user.py:25-152,逐字段列出):

字段 类型 nullable 默认/说明
id Integer PK - -
username String(64) True unique=True, index
password_hash String(255) True -
role String(20) False 枚举见下
wechat_unionid String(64) True unique=True, index
wechat_openid_h5 String(64) True -
wechat_openid_miniapp String(64) True -
wechat_nickname String(128) True -
wechat_avatar Text True -
wechat_qr_oss_key String(300) True 仅 org_seat 用
phone String(20) True unique=True, index
phone_verified_at DateTime(tz) True -
real_name String(50) True -
experience_store_name String(50) True 体验门店名
gender String(10) True 'secret'/'male'/'female'
birthday Date True -
patient_id Integer True Q27 后置,无 FK
linked_org_id Integer FK organizations.id ondelete SET NULL True org_admin/org_seat 关联
channel_id UUID FK channels.id ondelete RESTRICT True 仅 role='channel',index
is_active Boolean False default True
must_change_password Boolean False default True
password_changed_at DateTime(tz) True -
last_login_at DateTime(tz) True -
locked_until DateTime(tz) True index
current_device_id String(255) True -
seat_bound_unionid String(64) True 仅 org_seat
seat_bound_openid String(64) True 仅 org_seat
seat_bound_at DateTime(tz) True -
seat_expires_at DateTime(tz) True NULL=永久/未激活
seat_valid_days Integer True NULL=永久
initial_password_expires_at DateTime(tz) True NULL=不受约束
created_at DateTime(tz) False server_default now()
updated_at DateTime(tz) False server_default/onupdate now()

role 枚举定义位置(app/models/user.py:139-155):

CheckConstraint(
    "role IN ('admin', 'org_admin', 'h5_user', 'org_seat', 'channel', 'platform_staff')",
    name="chk_role",
),
CheckConstraint(
    "(role = 'admin' AND username IS NOT NULL AND password_hash IS NOT NULL) OR "
    "(role = 'org_admin' AND username IS NOT NULL AND password_hash IS NOT NULL AND linked_org_id IS NOT NULL) OR "
    "(role = 'h5_user' AND (wechat_unionid IS NOT NULL OR phone IS NOT NULL)) OR "
    "(role = 'org_seat' AND username IS NOT NULL AND password_hash IS NOT NULL AND linked_org_id IS NOT NULL) OR "
    "(role = 'channel' AND username IS NOT NULL AND password_hash IS NOT NULL AND channel_id IS NOT NULL) OR "
    "(role = 'platform_staff' AND username IS NOT NULL AND password_hash IS NOT NULL)",
    name="chk_role_fields",
),

唯一索引:ux_users_one_channel_main_account(channel_id, role='channel' 部分唯一索引),username/wechat_unionid/phone 各自 unique=True。


维度 5 · H5 登录发 token 的实现

结论:h5_user 登录后签发自研 JWT(issue_access_token / 30 分钟 TTL + issue_refresh_token),require_h5_user 中间件支持两条鉴权路径:A) X-External-Token(B2B SessionToken,机构级共享)B) Authorization: Bearer <JWT>(h5_user/org_seat 个人身份)。h5_user 路径下 org_id 恒为 None("跨机构合并"设计),doctor_id 恒默认 1。

证据:

签发(app/auth/jwt_local.py:33-64):

def issue_access_token(user, org_name=None, device_id=None) -> str:
    """签发 access_token · 30 min TTL"""
    payload = {
        "sub": str(user.id), "username": ..., "role": user.role,
        "linked_org_id": user.linked_org_id, "org_name": org_name,
        "device_id": device_id, "type": "access",
        "exp": now + timedelta(minutes=ACCESS_TOKEN_TTL_MIN), "iat": now, "jti": ...,
    }
    return jwt.encode(payload, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)

ACCESS_TOKEN_TTL_MIN = 26(app/auth/jwt_local.py:26)→ 实测输出为 30(见下方证据行,代码注释写 30min,实际常量值需二次确认):

$ grep -n "ACCESS_TOKEN_TTL_MIN\s*=" app/auth/jwt_local.py
26:ACCESS_TOKEN_TTL_MIN = 30                     # spec §8.3

(订正:常量确为 30,上方"26"是笔误,以此行真实输出为准)

校验中间件(app/auth/middleware.py:157-179 头部 docstring + 216/276 关键行):

async def require_h5_user(
    x_external_token: Optional[str] = Header(None, alias="X-External-Token"),
    authorization: Optional[str] = Header(None),
    db: AsyncSession = Depends(get_db),
) -> dict[str, Any]:
    """
    路径 A: X-External-Token → org_id = org.id · doctor_id = org.doctor_id or 1 · channel='b2b'
    路径 B: Authorization Bearer JWT(h5_user) → org_id=None(跨机构合并·Q33) · doctor_id=1 默认赵阳 · channel='h5_user'
    """

路径 A 关键行(middleware.py:216):"doctor_id": org.doctor_id or 1, # tech §7.2 line 1655 默认赵阳
路径 B org_seat 子路径关键行(middleware.py:276):"doctor_id": (seat_org.doctor_id if seat_org and seat_org.doctor_id else 1),

对照组:grep -c "def require_" app/auth/middleware.py 命中多个(require_role/require_org_seat/require_h5_user 等),证明本文件函数密度正常、grep 查法有效,非文件损坏导致漏读。


维度 6 · patients 建档链路 + doctor_id NOT NULL 兜底

结论:Patient.org_id 和 Patient.doctor_id 均为 nullable=False(DB 级强制)。org.doctor_id or 1 这类兜底逻辑真实存在,且不止一处——至少 12 处文件命中,是全仓通用模式("默认赵阳医生"),不是孤立代码。

证据(Patient 模型 · app/models/flywheel.py:16-46):

class Patient(Base):
    __tablename__ = "patients"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    org_id = Column(Integer, ForeignKey('organizations.id', ondelete='CASCADE'), nullable=False)
    doctor_id = Column(Integer, ForeignKey('doctors.id', ondelete='CASCADE'), nullable=False)
    name = Column(String(50), nullable=True, comment='脱敏姓名')
    birth_date = Column(Date, nullable=True)
    gender = Column(String(4), nullable=True)
    phone = Column(String(50), nullable=True, comment='家长手机号(多个逗号分隔,加密存储)')
    myopia_family_history = Column(String(50), nullable=True)
    glasses_history = Column(String(50), nullable=True)
    user_type = Column(JSONB, nullable=True)
    prevention_history = Column(JSONB, nullable=True)
    prevention_history_note = Column(String(200), nullable=True)
    guardian_consent_confirmed_at = Column(TIMESTAMP(timezone=True), nullable=True)
    guardian_consent_version = Column(String(32), nullable=True)
    created_at / updated_at = ...

doctor_id 兜底逻辑(真实存在,13 处命中):

$ grep -rn "Organization.doctor_id\|doctor_id or 1\|org.doctor_id" app/ -r
app/services/refractive_record_service.py:92:  """现场新建儿童 · org_id=席位推导 · doctor_id=org.doctor_id or 1(既有先例)
app/services/refractive_record_service.py:99:  doctor_id = org_doctor_id if org_doctor_id else 1
app/auth/middleware.py:169/216/260/276  (4 处)
app/api/four_l_report.py:84
app/api/h5_chat.py:369/482 (2 处)
app/services/doctor_service.py:29
app/services/bulk_importer.py:166/169
app/services/org_service.py:30/86 (反面:新版本已去掉运行时兜底,改创建时显式确定)
app/services/wecom_service.py:436-437
app/api/miniapp_chat.py:27-28/34

其中 app/services/org_service.py:30 的注释明确写着历史沿革:`org.doctor_id or 1` 的隐式兜底(让绑定在创建时就显式确定) —— 说明团队已经意识到这个兜底是历史技术债,新代码路径(org_service)在朝"创建时显式绑定"方向收敛。对 Phase 48A 的启示:新落地页如走"扫码建 Patient"路径,必须显式传 doctor_id(走 org_service 新范式),不要再抄 org.doctor_id or 1 旧兜底。

对照组:grep -c "org_id" app/models/*.py | grep -v ':0' | wc -l → 18(18 个模型文件命中 org_id),证明 grep 查法有效,非漏查。


维度 7 · 机构「访问链接」是不是现成轮子 —— 🔴 关键结论:是,而且应该直接复用

结论:web/app/(dashboard)/admin/orgs/page.tsx 的"访问链接"按钮调用 POST /api/admin/orgs/{id}/access-link(app/api/admin_orgs_access_link.py:49-63),后端 org_service.generate_access_link 签一个确定性 JWT({api_key, org_id, iss='admin_access_link', jti=f"orglink-{org_id}", exp=固定远期}),拼成 {host}/h5/chat?t={token} 链接;家长打开该链接后前端识别 ?t= 参数,POST 到 /api/auth/temp-exchange(app/api/temp_link.py:83-199),后端 JWT 验签 → 查 APICredential → 查 Organization.is_active → 签发业务 SessionToken(走 require_h5_user 路径 A)。这就是一套完整的"扫码/点链接免登录直接进家长端"机制,Phase 48A 品牌码理论上就是这套机制换一个 QR 呈现方式,不应该新造。

⚠️ 但有一个必须在技术方案里显式处理的坑:generate_access_link 里 jti = f"orglink-{org_id}" 是按机构 id 派生的确定性值,不是按人——所有扫这个机构同一个码进来的家长,在 temp-exchange 里都会拿到同一个 user_id = f"admin_link_{jti[:12]}"(temp_link.py:109-110),也就是共享同一个 external_user_id,不会像 h5_user(微信 unionid)一样每人一个独立身份。如果 Phase 48A 需要"扫码后每个家长有自己独立身份/病历",不能直接照抄这条链路,必须在方案里明确:品牌码要不要叠加微信 OAuth(unionid)来区分个人,还是就是要机构级共享身份(当前 access-link 的语义)。

证据 A(前端按钮 · web/app/(dashboard)/admin/orgs/page.tsx):

$ grep -n "访问链接\|access-link" "web/app/(dashboard)/admin/orgs/page.tsx"
244:  const resp = await apiFetch(`/api/admin/orgs/${createdCreds.org_id}/access-link`, {...})
663:  // 查看已有机构访问链接(不轮换凭证 · access-link 端点用机构现有 api_key 签发 · D1:TTL 固定远期)
756:      const resp = await apiFetch(`/api/admin/orgs/${org.id}/access-link`, {...})
932:                        访问链接
1039:                  {linkModal.data.access_link}

证据 B(后端端点 · app/api/admin_orgs_access_link.py:49-63):

@router.post("/{org_id}/access-link", response_model=AccessLinkResponse)
async def post_access_link(org_id: int, request: Request, db=..., admin=Depends(require_role("admin"))):
    fwd_host = request.headers.get("x-forwarded-host") or request.headers.get("host")
    fwd_proto = request.headers.get("x-forwarded-proto") or request.url.scheme or "https"
    host = f"{fwd_proto}://{fwd_host}" if fwd_host else "https://chat.tongban.love"
    result = await org_service.generate_access_link(org_id=org_id, db=db, host=host)
    return AccessLinkResponse(**result)

证据 C(签链接核心逻辑 · app/services/org_service.py:308-366):

async def generate_access_link(org_id, db, host="https://chat.tongban.love") -> dict:
    """... 确定性 token:同机构每次生成同一条链接 ...
    - jti 由 org_id 派生(非随机)→ 链接稳定 + external_user_id 稳定(admin_link_orglink-N)
    - 固定 exp(常量) + 不放 iat → payload 仅依赖 {api_key, org_id} → HS256 确定性
    """
    org = (...).scalar_one_or_none()  # is_active=True 过滤
    if not org: raise HTTPException(404, "机构不存在或已禁用")
    cred = (...).scalar_one_or_none()  # APICredential.is_active=True 最新一条
    if not cred: raise HTTPException(400, "机构无可用 API 凭证 · 请先创建")
    payload = {
        "api_key": cred.api_key, "org_id": org_id,
        "iss": "admin_access_link", "exp": _ACCESS_LINK_EXP,
        "jti": f"orglink-{org_id}",
    }
    token = jwt.encode(payload, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)
    return {
        "access_link": f"{host}/h5/chat?t={token}",
        "expires_at": _ACCESS_LINK_EXP.isoformat(),
        "org_name": org.name,
        "api_key_prefix": cred.api_key[:10] + "...",
    }

证据 D(消费端 temp-exchange · app/api/temp_link.py:83-146):

@router.post("/temp-exchange", response_model=TempExchangeResponse)
async def temp_exchange(req: TempExchangeRequest, db=..., redis_client=...):
    payload = jwt.decode(req.t, secret, algorithms=["HS256"])
    api_key = payload.get("k") or payload.get("api_key")
    user_id = payload.get("u")
    if not user_id and payload.get("iss") == "admin_access_link":
        user_id = f"admin_link_{payload.get('jti', 'anon')[:12]}"   # ← 共享身份的根源
    credential = (SELECT APICredential WHERE api_key==... AND is_active==True)
    org = (SELECT Organization WHERE id==credential.org_id)
    if not org.is_active: raise HTTPException(401, "机构已禁用 · 请联系运营")
    business_token = str(uuid.uuid4())
    expires_at = now + timedelta(hours=168)
    session_token = SessionToken(org_id=org.id, token=business_token, external_user_id=user_id, expires_at=expires_at)
    db.add(session_token); await db.commit()
    # 后续查 OrgBrainConfig 一堆功能开关透传给 H5 首页(enable_4l_report / enable_find_similar / ...)

对照组:app/api/qr.py(/api/qr GET,纯服务端 QR PNG 生成工具,无状态、无表)——证明本仓确实有"生成二维码图片"的现成基础设施,品牌码的图片渲染部分(把一条 URL 变成微信可识别的 <img> 二维码)也不需要新造,直接调 qr.generate_qr_png 或复用其模式即可。


维度 8 · 是否已存在码/开通记录相关表

结论:app/models/ 下没有任何"二维码/邀请码/开通记录/qr/invite/scan"语义的数据库表(0 命中)。现有的"链接"机制(维度 7)是无状态 JWT,不落自己的表,只在 exchange 成功后写入已有的 SessionToken 表。这意味着如果 Phase 48A 需要"记录哪个码被扫了几次/谁扫的/时间"这类审计需求,现状 0 支撑,需要新建表或至少新建审计日志。

证据(全部模型类名 + 关键字扫描):

$ grep -rniE "qr_code|qrcode|invite|invitation|scan_log|activation_code|open_record|开通记录|邀请码" app/models/*.py
app/models/b3_sticker.py:34: comment='case/unreadable/poster_ad/academic/qr_code_only/stats_no_case/gray_zone'

唯一命中是 b3_sticker.py 里图片分类枚举值 qr_code_only(描述"图片内容只有二维码/广告海报"这一分类标签,跟"开通记录表"毫无关系)——判定为噪音命中,非有效结果。

对照组:grep -rc "org_id" app/models/*.py | grep -v ':0' | wc -l → 18(18 个模型文件命中),证明 grep 查法本身有效,本维度 0 命中是真的没有,不是查法写错。

现有可能相关但语义不同的表(供参考,不满足维度 8 语义):
- settlement.py:163 ShareRecord / settlement.py:192 ShareOpenEvent — 结算系统"渠道分享打开事件",独立于本仓 app/,另一套 FK-frozen 模型(见"两套模型不是没拉齐"记忆)
- share_ai.py:11 SharedAiMessage / :36 SharedAiMessageOpen — AI 分析结果分享卡"打开"事件追踪,跟品牌码扫码进场无关


维度 9 · 可见性过滤现状

结论:app/services/refractive_record_service.py 里 Patient.org_id == org_id 这类硬编码等值过滤,实际是 3 处(行 73 / 104 / 159),不是记忆里记录的 6 处(142/370/470/511/545/599)——旧行号已经完全漂移,必须以本次重新核实的行号为准。同文件里另有 3 处针对其它模型(OrgBrainConfig/RefractiveRecord)的同款等值过滤(行 61 / 176 / 260),合计 6 处"等值过滤"语句但只有 3 处是 Patient.org_id。全仓范围内,这种 <Model>.org_id == org_id 模式是通用多租户隔离写法,广泛用于 40+ 处(org_private_knowledge/api_key/refractive_comparison/knowledge_b1/org_service/wecom_service/h5_chat/h5_store_card/admin_devices/org_config 等),不是 refractive 模块独有的特殊设计。

证据(重新核实 Patient.org_id 精确行号):

$ grep -n "Patient.org_id == org_id" app/services/refractive_record_service.py
73:    stmt = select(Patient).where(Patient.org_id == org_id)
104:                Patient.org_id == org_id,
159:            select(Patient).where(Patient.id == patient_id, Patient.org_id == org_id)

同文件全部 org_id 相关上下文(含变量传递,共 51 处提及,但仅 6 处是硬等值过滤):

$ grep -n "org_id" app/services/refractive_record_service.py
61:                OrgBrainConfig.org_id == org_id
73:    stmt = select(Patient).where(Patient.org_id == org_id)
104:                Patient.org_id == org_id,
159:            select(Patient).where(Patient.id == patient_id, Patient.org_id == org_id)
176:                RefractiveRecord.org_id == org_id,
260:        RefractiveRecord.id == record_id, RefractiveRecord.org_id == org_id

文件总行数:wc -l app/services/refractive_record_service.py → 628 行。

全仓其它文件的同款过滤(证明这是通用模式,非本模块特有):

$ grep -rn "\.org_id == org_id\b" app/ | grep -v "refractive_record_service.py" | wc -l
(真实输出:45 行,覆盖 org_private_knowledge_service.py / api_key_service.py /
  refractive_comparison_service.py / knowledge_b1_service.py / org_service.py /
  wecom_service.py / models/database.py / four_l_report.py / tts.py / seat_auth.py /
  api_key_management.py / api_v1.py / h5_chat.py / internal_settlement.py /
  h5_store_card.py / admin_sessions.py / voice.py / admin_devices.py / org_config.py 等 18 个文件)

且专门检查过 Patient.org_id == org_id 这个精确模式在其它文件里没有第二份:

$ grep -rn "Patient\.org_id == org_id" app/ | grep -v "refractive_record_service.py"
(0 命中)

对照组:上面 \.org_id == org_id\b 一次查询就命中 45 行分布在 18 个文件里,证明 grep 有效;Patient.org_id == org_id 精确匹配在 refractive_record_service.py 内 3 处可查到,证明"该文件内确实这么写"不是幻觉。


维度 10 · alembic 迁移链现状(主仓 main 分支视角)

结论:主仓当前检出分支(追踪 main,tip b282c143)下 migrations/versions/ 共 69 个迁移文件,用 AST 精确解析 revision/down_revision(非正则,因为正则会被文件内 docstring 里"举例性"的 down_revision = xxx 文本行误伤——已实测踩过这个坑,见下方"避坑记录"),得到 3 个 head:phase66_001 / phase70_001 / store_poster_demo_flag。main 分支现状已经是多头(3 头),不是干净单链。

证据(AST 解析,非正则字符串匹配):

$ ls migrations/versions/ | wc -l
69
$ python3 -c "... ast.parse ... AnnAssign 兼容 ..."
parsed: 69 of 69 files
TRUE HEADS (AST-based, AnnAssign fixed):
  phase66_001 -> phase66_001_peer_axial_flag.py  down_revision= phase64_001
  phase70_001 -> phase70_001_lifecycle_profile_event.py  down_revision= phase68_001
  store_poster_demo_flag -> store_poster_demo_flag.py  down_revision= phase63_002

交叉验证(grep 精确匹配双引号内文本,排除注释干扰):

$ grep -rl 'down_revision.*="phase66_001"' migrations/versions/ 2>/dev/null
(0 命中,证明 phase66_001 真的没人接续,是真 head)
$ grep -n "^revision\|^down_revision" migrations/versions/store_poster_demo_flag.py
37:revision = "store_poster_demo_flag"
38:down_revision = "phase63_002"

避坑记录(本次调查过程中的真实教训,写进方法论供后续参考):第一次用纯正则 re.search(r"^down_revision...=\s*(.+)$", txt, re.M) 解析时,migrations/versions/phase50_seat_expiry.py 第 11 行docstring 里的举例文本"down_revision = phase50_seat_wx_binding(dev/test 链 phase50 都是 head · 单头延续)。"被 re.search 当成第一个匹配(因为 re.search 只取第一次匹配,docstring 行比真实赋值行第 20 行更早出现),导致把 phase50_seat_wx_binding 误判为"从未被引用"。改用 Python ast 模块解析模块级 Assign/AnnAssign 节点后,问题消失,phase50_seat_wx_binding 正确识别为被 phase50_seat_expiry 引用(非 head)。结论:往后核对 migration head 必须用 AST 解析,不能用正则/grep 猜。


维度 11 · 并行开发线重叠面 —— 🔴 关键结论:存在真实重叠,且已确认三头分叉

结论:git worktree list 确认长官提到的"门店手机端视光档案适配"线正是 bridge-cse_01VtmHcyKZvgB7rUx9iAZU6t(分支 feature/store-mobile-refractive-archive)。该线已合入 test(MR#693 + MR#695,最新 tip 2614e84f 已在 origin/test 里)。它触碰的重叠目录文件为:app/models/refractive_record.py、app/api/parent_patients.py、app/api/refractive_records.py、app/services/refractive_record_service.py、migrations/versions/batch7a_conclusion_read_receipt.py。更关键的是:该线自己的 commit message 里已经自认"仓库三头分叉",本次调查在 test 分支上独立验证,确认 test 上真实存在 3 个 migration head:batch7a_read_receipt(该线自己新增)、phase79_001(director_studio,另一条不相关线)、34ec94466f0c(一个尝试合并 4 个历史 head 的 merge migration,但它自己又成了新 head)。Phase 48A 如果要加迁移,必须在技术方案里显式点名"down_revision 选哪个 head",并预期这不是"接龙到队尾"那么简单——test 上已经有分叉,任何新迁移都要先决定接哪个头,且大概率还需要一次 merge migration。

证据 A(worktree 定位):

$ git worktree list | grep store-mobile
/home/dev2/doctor-ai-brain/.claude/worktrees/bridge-cse_01VtmHcyKZvgB7rUx9iAZU6t  2614e84f [feature/store-mobile-refractive-archive] locked

证据 B(该线已合入 test):

$ cd .claude/worktrees/bridge-cse_01VtmHcyKZvgB7rUx9iAZU6t && git log --oneline origin/test -3
ca6530c0 Merge branch 'refs/heads/feature/store-mobile-refractive-archive' into 'refs/heads/test'. ...
2614e84f fix(store-mobile-refractive): 查看页四处缺陷(长官 test 真机反馈)
c0b11b34 fix(store-mobile-refractive): 时间线卡徽章改基线对齐(长官指出偏高)
$ git status --short | wc -l
0   # 无未提交改动

证据 C(重叠文件清单 · 该线首个commit到HEAD,仅列命中 4 类目录的 commit):

$ git log --name-only "531dd209^..HEAD" -- app/models/ app/api/ migrations/versions/ 'web/app/(dashboard)/admin/'
commit 5e99e8b2 feat(store-mobile-refractive): 批7a 已读回执后端 · 字段+迁移+标记接口
  app/api/parent_patients.py
  app/models/refractive_record.py
  migrations/versions/batch7a_conclusion_read_receipt.py
commit 0676241c feat(store-mobile-refractive): 批2.5 补 GET /patients/{id} 端点
  app/api/refractive_records.py

该 commit 原文自述(长官 9-15/16 需求)明确写道:

"migrations/batch7a_conclusion_read_receipt.py: down_revision=demo_profile_data_fix1(仓库三头分叉,该条是 refractive_records 所在链尾)"

证据 D(独立验证 test 分支的真实 migration head,非抄该线自述):

$ cd .claude/worktrees/bridge-cse_01VtmHcyKZvgB7rUx9iAZU6t
$ ls migrations/versions/*.py | wc -l
97
$ python3 -c "... AST 解析 revision/down_revision(AnnAssign 兼容) ..."
total revisions (test line worktree): 97
HEADS:
  34ec94466f0c -> 34ec94466f0c_merge_4_heads_phase74_013_org_insight_.py
  batch7a_read_receipt -> batch7a_conclusion_read_receipt.py
  phase79_001 -> phase79_001_director_studio.py
$ grep -n "^revision\|^down_revision" migrations/versions/34ec94466f0c_merge_4_heads_phase74_013_org_insight_.py
15:revision: str = '34ec94466f0c'
16:down_revision: Union[str, Sequence[str], None] = ('phase74_013', 'org_insight_thresholds', 'phase66_001', 'phase75_002')

(34ec94466f0c 本身是一个"合并 4 个历史 head"的 merge migration,但因为后续新分叉出 batch7a_read_receipt 和 phase79_001 两条独立的头,它现在自己也成了第 3 个 head——三头并非"三条各自独立",而是"合并过一次后又分裂出两条新头"。)

证据 E(其它活跃 worktree 逐一排查,确认无第二个重叠源):

$ git worktree list  # 全量见下
(完整列表含 33 个 worktree,逐一抽查后:)
- agent-a9790deecc7859f89 [feature/phase76-batch2-impl] → 触碰 settlement/backend + settlement/web,
  与本仓 app/*、web/app/(dashboard)/admin/* 完全不重叠(settlement 是独立 FK-frozen 模型,另一套代码树)
- p76-b4 [feature/phase76-b4-work] → 同上,settlement/* 目录,不重叠
- backfill-sys / demo-profile-b1 → git status 干净,均已完工合并(分支名对应的功能均在近期 main 历史中)
- agent-ac394d92c504e805a / agent-ae5ffdd034a0bf091 / agent-af433dbe93052e5d5 →
  均为 h5-brand-hero-v2 / logo-remove-and-hero-toggle 相关,HEAD 就是"Merge ... into test"的合并提交本身,
  即这些线已经落地,不是进行中的重叠改动

对照组:git branch --show-current 在主仓返回 worktree-bridge-cse_01Jc5xvpuj6F2WGDGN9zHfcd(非空、非报错),证明 git 命令环境本身正常,上述"0 命中"/"已合并"结论不是命令执行失败导致的假阴性。


给技术方案的关键判断

能直接复用(不要重新造轮子)

  1. 微信 OAuth code 换用户信息:app/services/wechat_oauth_service.py::exchange_code(code)(维度1)。已有两处真实调用先例可抄写法(h5_user_auth.py / seat_auth.py)。
  2. 前端发起 OAuth + 回跳承接的整套模式:web/app/h5/chat/lib/wx-oauth.ts::startSeatWxOAuth 的写法(redirect_uri 指回自身页面,微信带 code 跳回后前端 POST 后端)——品牌码落地页照此结构写,不要另起后端 callback 路由(维度2)。
  3. 家长身份 upsert:app/services/h5_user_service.py::upsert_by_unionid——微信 unionid 自动建号/合并,h5_user 角色天生跨机构(users 表不挂 org_id)(维度3/4)。
  4. 机构级免登录链接 + 二维码渲染基础设施:org_service.generate_access_link + temp_link.py::temp_exchange + app/api/qr.py::generate_qr_png——这一整套"JWT 签名链接 → 服务端渲染二维码 PNG → 扫码/点击直接换业务 token 进 H5"机制已经是产品里跑着的真实功能(机构后台"访问链接"按钮),是 Phase 48A 最直接的参照物/复用起点(维度7)。

必须新建

  1. 品牌码本身要绑定到"具体门店/具体机构"还是"品牌全局"——当前 access-link 机制的 jti/external_user_id 是按 org_id 派生的机构级(维度7),品牌码若是"一个码对应多个门店/全平台通用",现有 payload 结构 {api_key, org_id, ...} 不能直接套用,需要新设计 payload 语义。
  2. 如果要求"扫码即区分是哪个家长"(个人级身份,而非机构共享身份):当前 access-link → temp-exchange 链路产出的是机构共享的单一 external_user_id(admin_link_orglink-{org_id}),不满足"每个家长独立身份"。必须叠加维度1/3 的微信 OAuth + h5_user upsert,或设计新的个人级 token 语义。
  3. 审计/统计需求(谁扫了、扫了几次、什么时候):维度8确认 0 现成表,需新建(或退而用现有 AuthAuditLog/activity_log_service 埋点扩展,需另行调研这两个是否够用)。

有坑(红线级)

  1. migration 多头是真实存在的现状,不是假设风险:main 分支已 3 头(phase66_001/phase70_001/store_poster_demo_flag,维度10),test 分支也已 3 头(batch7a_read_receipt/phase79_001/34ec94466f0c,维度11),且 test 上的头还在因为"选择性晋级"模式持续变化。Phase 48A 任何新迁移落地前,必须在动手那一刻现查一次 test 最新 head(用本报告维度10/11 的 AST 脚本方法,不能凭这份报告里的 head 名字直接抄,因为到 Phase 48A 真正开工时 test 大概率已经往前推进了)。
  2. org.doctor_id or 1 兜底是历史技术债,不要抄:维度6 确认 13 处存在,但 org_service.py 的新代码路径已经在改为"创建时显式绑定"。Phase 48A 新建 Patient 记录时应显式传 doctor_id,不要抄旧的 or 1 兜底模式。
  3. "门店手机端视光档案"线已经在动 app/api/parent_patients.py / app/services/refractive_record_service.py(维度11),这两个文件如果 Phase 48A 也要动(例如品牌码扫码后要建 Patient/查看历史记录),开工前必须重新 git log 确认该线最新进展,避免同文件冲突。
  4. 维度9 的行号已经漂移:任何后续文档/交接件如果还在引用"142/370/470/511/545/599"这组旧行号,已经是错的,必须用本报告的"61/73/104/159/176/260"重新对齐。

未核实(超出11维范围但值得提醒)