调查方式:纯只读(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
结论: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 查法有效。
结论:没有后端 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 路由"不是查法漏了,是真没有。
结论:❌ 任务描述里的文件 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()
username(系统生成,非用户填)、password_hash、role='org_seat'、linked_org_id=org_idapp/services/password_utils.py:108-124::generate_seat_initial_password → secrets.randbelow(1_000_000):06d(6 位数字,100 万组合)_gen_unique_username(seat_service.py:119-149)用 SELECT ... WHERE username LIKE '____{yyyymm}' 查同月最大序号 +1,再循环 SELECT ... WHERE username == candidate 兜底防并发撞车(非 DB 唯一索引校验,是应用层查重 + users.username 列本身的 unique=True 双重保险)linked_org_id 直接写传入的 org_id证据 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 表)。
结论: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。
结论: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 查法有效,非文件损坏导致漏读。
结论: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 查法有效,非漏查。
结论: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 或复用其模式即可。
结论: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 分析结果分享卡"打开"事件追踪,跟品牌码扫码进场无关
结论: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 处可查到,证明"该文件内确实这么写"不是幻觉。
结论:主仓当前检出分支(追踪 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 猜。
结论: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 命中"/"已合并"结论不是命令执行失败导致的假阴性。
app/services/wechat_oauth_service.py::exchange_code(code)(维度1)。已有两处真实调用先例可抄写法(h5_user_auth.py / seat_auth.py)。web/app/h5/chat/lib/wx-oauth.ts::startSeatWxOAuth 的写法(redirect_uri 指回自身页面,微信带 code 跳回后前端 POST 后端)——品牌码落地页照此结构写,不要另起后端 callback 路由(维度2)。app/services/h5_user_service.py::upsert_by_unionid——微信 unionid 自动建号/合并,h5_user 角色天生跨机构(users 表不挂 org_id)(维度3/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)。jti/external_user_id 是按 org_id 派生的机构级(维度7),品牌码若是"一个码对应多个门店/全平台通用",现有 payload 结构 {api_key, org_id, ...} 不能直接套用,需要新设计 payload 语义。external_user_id(admin_link_orglink-{org_id}),不满足"每个家长独立身份"。必须叠加维度1/3 的微信 OAuth + h5_user upsert,或设计新的个人级 token 语义。AuthAuditLog/activity_log_service 埋点扩展,需另行调研这两个是否够用)。org.doctor_id or 1 兜底是历史技术债,不要抄:维度6 确认 13 处存在,但 org_service.py 的新代码路径已经在改为"创建时显式绑定"。Phase 48A 新建 Patient 记录时应显式传 doctor_id,不要抄旧的 or 1 兜底模式。app/api/parent_patients.py / app/services/refractive_record_service.py(维度11),这两个文件如果 Phase 48A 也要动(例如品牌码扫码后要建 Patient/查看历史记录),开工前必须重新 git log 确认该线最新进展,避免同文件冲突。AuthAuditLog/LoginAttempt(app/models/audit.py)是否适合承载"品牌码扫码审计"——本次未深入看这两张表的字段设计,技术方案阶段如需用,应单独 grep 核实。org_service.py:30/86 提到的"新版创建时显式绑定"具体是哪个函数、覆盖了 org 创建的哪个路径——本次只看到注释提及,未展开读函数体,如方案要复用需单独核实。