import asyncio import logging import secrets from datetime import datetime, timedelta from typing import Iterable, Optional import httpx from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from backend.app.core.config import settings from backend.app.core.security import hash_password from backend.app.db.session import SessionLocal from backend.app.models import Campus, Department, Role, SyncState, User from backend.app.services.sync_ws import sync_broadcaster logger = logging.getLogger("uvicorn.error") SYNC_STATE_KEY = "center" PENDING_ROLE_NAME = "待授权" _sync_task: asyncio.Task | None = None _sync_task_lock = asyncio.Lock() SYNC_YIELD_EVERY = 200 def _build_center_url(path: str) -> str: base = settings.center_base_url.rstrip("/") return f"{base}/{path.lstrip('/')}" def _normalize_external(value) -> Optional[str]: if value is None: return None text = str(value).strip() if not text or text == "0": return None return text async def upsert_sync_token(db: AsyncSession, token: str) -> SyncState: state = await db.get(SyncState, SYNC_STATE_KEY) if not state: state = SyncState(key=SYNC_STATE_KEY, token=token) db.add(state) else: state.token = token state.updated_at = datetime.utcnow() return state async def get_sync_token(db: AsyncSession) -> Optional[str]: state = await db.get(SyncState, SYNC_STATE_KEY) return state.token if state else None async def mark_sync_completed(db: AsyncSession) -> None: state = await db.get(SyncState, SYNC_STATE_KEY) if state: state.last_sync_at = datetime.utcnow() async def get_sync_status(db: AsyncSession) -> dict: state = await db.get(SyncState, SYNC_STATE_KEY) async with _sync_task_lock: running = _sync_task is not None and not _sync_task.done() return { "running": running, "last_sync_at": state.last_sync_at if state else None, } async def _notify_sync_status() -> None: try: async with SessionLocal() as db: status = await get_sync_status(db) await sync_broadcaster.broadcast_status(status) except Exception: logger.exception("广播同步状态失败") async def get_or_create_pending_role(db: AsyncSession) -> Role: result = await db.execute(select(Role).where(Role.name == PENDING_ROLE_NAME)) role = result.scalar_one_or_none() if role: return role role = Role(name=PENDING_ROLE_NAME, description="等待管理员授权的默认角色", is_system=False) db.add(role) await db.flush() return role async def resolve_role_by_external_ids(db: AsyncSession, external_ids: Iterable[str]) -> Optional[Role]: for external_id in external_ids: result = await db.execute(select(Role).where(Role.external_role_id == external_id)) role = result.scalar_one_or_none() if role: return role return None async def resolve_department_by_external_id(db: AsyncSession, external_dept_id: Optional[str]) -> Optional[Department]: if not external_dept_id: return None result = await db.execute(select(Department).where(Department.external_dept_id == external_dept_id)) return result.scalar_one_or_none() async def _fetch_json(client: httpx.AsyncClient, url: str, token: str): response = await client.get(url, headers={"token": token}) response.raise_for_status() return response.json() async def _sync_campuses_and_departments(db: AsyncSession, items: list[dict]) -> dict[str, Department]: campus_result = await db.execute(select(Campus)) campuses = campus_result.scalars().all() campus_by_external = {c.external_hosp_id: c for c in campuses if c.external_hosp_id} campus_by_name = {c.name: c for c in campuses} dept_result = await db.execute(select(Department)) departments = dept_result.scalars().all() dept_by_external = {d.external_dept_id: d for d in departments if d.external_dept_id} dept_by_campus_name = {(d.campus_id, d.name): d for d in departments} updated_departments: dict[str, Department] = {} for index, item in enumerate(items, start=1): dept_id = _normalize_external(item.get("departmentId")) dept_code = _normalize_external(item.get("code")) dept_primary_id = _normalize_external(item.get("id")) external_dept_id = dept_id or dept_primary_id or dept_code hosp_id = _normalize_external(item.get("hospId")) name = (item.get("name") or "").strip() or f"科室{external_dept_id}" if not external_dept_id or not hosp_id: continue campus = campus_by_external.get(hosp_id) if not campus: campus_name = f"院区{hosp_id}" campus = campus_by_name.get(campus_name) if campus and not campus.external_hosp_id: campus.external_hosp_id = hosp_id else: if campus_name in campus_by_name and (not campus or campus.external_hosp_id): campus_name = f"{campus_name}({hosp_id})" campus = Campus(name=campus_name, external_hosp_id=hosp_id) db.add(campus) await db.flush() campus_by_external[hosp_id] = campus campus_by_name[campus.name] = campus dept = dept_by_external.get(external_dept_id) if not dept: dept = dept_by_campus_name.get((campus.id, name)) if dept and not dept.external_dept_id: dept.external_dept_id = external_dept_id elif dept and dept.external_dept_id and dept.external_dept_id != external_dept_id: # 同名科室但不同 external_id，添加后缀区分 unique_name = f"{name}({external_dept_id})" existing_unique = dept_by_campus_name.get((campus.id, unique_name)) if existing_unique: dept = existing_unique if not dept.external_dept_id: dept.external_dept_id = external_dept_id else: dept = Department(campus_id=campus.id, name=unique_name, external_dept_id=external_dept_id) db.add(dept) await db.flush() dept_by_external[external_dept_id] = dept dept_by_campus_name[(dept.campus_id, dept.name)] = dept elif not dept: dept = Department(campus_id=campus.id, name=name, external_dept_id=external_dept_id) db.add(dept) await db.flush() dept_by_external[external_dept_id] = dept dept_by_campus_name[(dept.campus_id, dept.name)] = dept else: # 检查更新后的名称是否会导致 (campus_id, name) 冲突 conflict_dept = dept_by_campus_name.get((campus.id, name)) if conflict_dept and conflict_dept.id != dept.id: # 发现同名科室，为避免冲突，附加外部 ID name = f"{name}({external_dept_id})" dept.name = name dept.campus_id = campus.id for key in {external_dept_id, dept_primary_id, dept_id, dept_code}: if key: updated_departments[key] = dept if index % SYNC_YIELD_EVERY == 0: await asyncio.sleep(0) return updated_departments async def _sync_roles(db: AsyncSession, items: list[dict]) -> dict[str, Role]: role_result = await db.execute(select(Role)) roles = role_result.scalars().all() role_by_external = {r.external_role_id: r for r in roles if r.external_role_id} role_by_name = {r.name: r for r in roles} updated_roles: dict[str, Role] = {} for item in items: external_role_id = str(item.get("roleId") or item.get("roleCode") or "").strip() if not external_role_id: continue name_candidate = (item.get("remark") or item.get("roleName") or f"角色{external_role_id}").strip() role = role_by_external.get(external_role_id) if not role: role = role_by_name.get(name_candidate) if role and not role.external_role_id: role.external_role_id = external_role_id else: role_name = name_candidate if role_name in role_by_name and (not role or role.external_role_id): role_name = f"{role_name}({external_role_id})" role = Role(name=role_name, description="中台角色同步", is_system=False, external_role_id=external_role_id) db.add(role) await db.flush() role_by_external[external_role_id] = role role_by_name[role.name] = role updated_roles[external_role_id] = role return updated_roles async def _sync_users(db: AsyncSession, items: list[dict], dept_by_external: dict[str, Department]) -> None: external_ids = [] accounts = [] for item in items: external_id = item.get("userId") or item.get("id") if external_id is not None: external_ids.append(str(external_id)) account = str(item.get("empNo") or external_id or "").strip() if account: accounts.append(account) external_id_set = set(external_ids) existing_users: dict[str, User] = {} if external_ids: result = await db.execute(select(User).where(User.external_user_id.in_(external_ids))) existing_users = {user.external_user_id: user for user in result.scalars().all() if user.external_user_id} # 预加载按账号查找的用户（用于处理账号冲突） users_by_account: dict[str, User] = {} if accounts: result = await db.execute(select(User).where(User.account.in_(accounts))) users_by_account = {user.account: user for user in result.scalars().all()} pending_role = await get_or_create_pending_role(db) for index, item in enumerate(items, start=1): external_user_id = str(item.get("userId") or item.get("id") or "").strip() if not external_user_id: continue account = str(item.get("empNo") or external_user_id).strip() name = (item.get("name") or account).strip() phone = (item.get("phoneNumber") or "").strip() or None title = (item.get("jobTitleName") or item.get("professionalTitleName") or "").strip() or None avatar = (item.get("avatarUrl") or "").strip() or None status = "active" if str(item.get("activeFlag") or "0") == "1" else "inactive" dept_external_id = _normalize_external(item.get("departmentId")) dept_code = _normalize_external(item.get("departmentCode")) dept = None if dept_external_id: dept = dept_by_external.get(dept_external_id) if not dept and dept_code: dept = dept_by_external.get(dept_code) campus_id = dept.campus_id if dept else None user = existing_users.get(external_user_id) if not user: # 检查账号是否已存在（处理账号冲突） existing_by_account = users_by_account.get(account) if existing_by_account: # 账号已存在，关联 external_user_id 到已有用户 user = existing_by_account user.external_user_id = external_user_id user.name = name user.phone = phone user.title = title user.avatar = avatar user.dept_id = dept.id if dept else user.dept_id user.campus_id = campus_id or user.campus_id if user.status != status: user.status = status user.token_version = (user.token_version or 1) + 1 existing_users[external_user_id] = user else: user = User( name=name, account=account, phone=phone, title=title, avatar=avatar, role_id=pending_role.id, status=status, password_hash=hash_password(secrets.token_urlsafe(24)), token_version=1, external_user_id=external_user_id, dept_id=dept.id if dept else None, campus_id=campus_id ) db.add(user) users_by_account[account] = user continue user.name = name user.phone = phone user.title = title user.avatar = avatar user.external_user_id = external_user_id user.dept_id = dept.id if dept else user.dept_id user.campus_id = campus_id or user.campus_id if user.status != status: user.status = status user.token_version = (user.token_version or 1) + 1 if index % SYNC_YIELD_EVERY == 0: await db.flush() await asyncio.sleep(0) result = await db.execute(select(User).where(User.external_user_id.isnot(None))) for user in result.scalars().all(): if user.external_user_id not in external_id_set and user.status != "inactive": user.status = "inactive" user.token_version = (user.token_version or 1) + 1 async def sync_center_data(token: str) -> None: if not settings.center_base_url: logger.warning("CENTER_BASE_URL 未配置，跳过中台同步") return logger.info("开始拉取中台数据...") async with httpx.AsyncClient(timeout=settings.sso_timeout_seconds) as client: dept_url = _build_center_url("/gateway/centerSys/api/getAllDepartment") user_url = _build_center_url("/gateway/centerSys/api/getAllUser") role_url = _build_center_url("/gateway/centerSys/api/getRoleList") try: dept_payload = await _fetch_json(client, dept_url, token) logger.info("已获取科室数据") role_payload = await _fetch_json(client, role_url, token) logger.info("已获取角色数据") user_payload = await _fetch_json(client, user_url, token) logger.info("已获取人员数据") except Exception as e: logger.error(f"从发现中心请求数据失败: {str(e)}") raise if isinstance(dept_payload, dict): code = dept_payload.get("code") or dept_payload.get("status") if code is not None and code != 200: raise ValueError(f"科室接口返回异常 code={code}") if isinstance(role_payload, dict): code = role_payload.get("code") if code is not None and code != 200: raise ValueError(f"角色接口返回异常 code={code}") if isinstance(user_payload, dict): status = user_payload.get("status") if status is not None and status != 200: raise ValueError(f"人员接口返回异常 status={status}") dept_items = [] if isinstance(dept_payload, list): dept_items = dept_payload elif isinstance(dept_payload, dict) and isinstance(dept_payload.get("data"), list): dept_items = dept_payload.get("data", []) role_items = [] if isinstance(role_payload, dict): role_data = role_payload.get("data") or {} role_items = role_data.get("list") or [] user_items = [] if isinstance(user_payload, dict): user_items = user_payload.get("data") or [] logger.info(f"解析到数据: 科室={len(dept_items)}, 角色={len(role_items)}, 人员={len(user_items)}") try: async with SessionLocal() as db: dept_by_external = await _sync_campuses_and_departments(db, dept_items) await _sync_roles(db, role_items) await _sync_users(db, user_items, dept_by_external) await mark_sync_completed(db) await db.commit() logger.info("中台同步完成并已存入数据库") except Exception as e: logger.error(f"中台同步写入数据库失败: {str(e)}") raise finally: # 确保无论成功失败，都广播一次最新的状态（结束 loading） await _notify_sync_status() async def _cancel_then_sync(task: asyncio.Task, token: str) -> None: try: await task except asyncio.CancelledError: pass await sync_center_data(token) def _schedule_status_broadcast(_: asyncio.Task) -> None: asyncio.create_task(_notify_sync_status()) async def trigger_center_sync(token: str, *, restart: bool, wait: bool) -> bool: global _sync_task async with _sync_task_lock: if _sync_task and not _sync_task.done(): if not restart: return False _sync_task.cancel() _sync_task = asyncio.create_task(_cancel_then_sync(_sync_task, token)) else: _sync_task = asyncio.create_task(sync_center_data(token)) task = _sync_task task.add_done_callback(_schedule_status_broadcast) await _notify_sync_status() if wait: await task return True async def run_center_sync_once() -> None: async with SessionLocal() as db: token = await get_sync_token(db) if not token: logger.warning("未找到中台同步 token，跳过本次同步") return started = await trigger_center_sync(token, restart=False, wait=True) if not started: logger.info("已有同步任务在执行，跳过本次同步") async def run_center_sync_loop(stop_event: asyncio.Event) -> None: while not stop_event.is_set(): now = datetime.now() next_run = now.replace( hour=settings.center_sync_hour, minute=settings.center_sync_minute, second=0, microsecond=0 ) if next_run <= now: next_run = next_run + timedelta(days=1) delay = (next_run - now).total_seconds() try: await asyncio.wait_for(stop_event.wait(), timeout=delay) except asyncio.TimeoutError: try: await run_center_sync_once() except Exception: logger.exception("中台同步失败")