release(main): 同步 dev 最新候选改动

backend/app/services/collaboration_share_service.py

@@ -0,0 +1,289 @@
+from __future__ import annotations
+
+import base64
+import binascii
+import hashlib
+import hmac
+import uuid
+from datetime import datetime, timedelta, timezone
+
+from anyio import to_thread
+from fastapi import HTTPException, status
+from jose import JWTError, jwt
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.config import settings
+from app.core.security import hash_password, verify_password
+from app.models.collaboration import CollaborationFile, CollaborationShareLink
+from app.schemas.collaboration import (
+    CollaborationPublicShareMetadata,
+    CollaborationShareAccessGrant,
+    CollaborationShareLinkRead,
+    CollaborationShareLinkUpdate,
+)
+from app.services import collaboration_service
+
+
+SHARE_PATH = "/collaboration/share"
+SHARE_ACCESS_TTL_SECONDS = 30 * 60
+PASSWORD_FAILURE_WINDOW = timedelta(minutes=15)
+PASSWORD_LOCK_DURATION = timedelta(minutes=15)
+PASSWORD_FAILURE_LIMIT = 5
+_ACCESS_PURPOSE = "ctms-collaboration-share-access"
+_EXPIRY_DURATIONS = {
+    "ONE_DAY": timedelta(days=1),
+    "SEVEN_DAYS": timedelta(days=7),
+    "THIRTY_DAYS": timedelta(days=30),
+    "PERMANENT": None,
+}
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _signing_key() -> bytes:
+    return hmac.new(
+        settings.JWT_SECRET_KEY.encode("utf-8"),
+        b"ctms-collaboration-share-v1",
+        hashlib.sha256,
+    ).digest()
+
+
+def _b64encode(value: bytes) -> str:
+    return base64.urlsafe_b64encode(value).rstrip(b"=").decode("ascii")
+
+
+def _b64decode(value: str) -> bytes:
+    padding = "=" * (-len(value) % 4)
+    return base64.urlsafe_b64decode(f"{value}{padding}".encode("ascii"))
+
+
+def share_token(link: CollaborationShareLink) -> str:
+    payload = f"{link.id}.{link.token_version}".encode("ascii")
+    signature = hmac.new(_signing_key(), payload, hashlib.sha256).digest()
+    return f"{_b64encode(payload)}.{_b64encode(signature)}"
+
+
+def _decode_share_token(value: str | None) -> tuple[uuid.UUID, int]:
+    token = (value or "").strip()
+    if not token or len(token) > 256 or token.count(".") != 1:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="共享链接不存在或已失效")
+    encoded_payload, encoded_signature = token.split(".", 1)
+    try:
+        payload = _b64decode(encoded_payload)
+        actual_signature = _b64decode(encoded_signature)
+        if (
+            not hmac.compare_digest(_b64encode(payload), encoded_payload)
+            or not hmac.compare_digest(_b64encode(actual_signature), encoded_signature)
+        ):
+            raise ValueError("non-canonical token encoding")
+        expected_signature = hmac.new(_signing_key(), payload, hashlib.sha256).digest()
+        if not hmac.compare_digest(actual_signature, expected_signature):
+            raise ValueError("signature mismatch")
+        raw_id, raw_version = payload.decode("ascii").split(".", 1)
+        return uuid.UUID(raw_id), int(raw_version)
+    except (ValueError, UnicodeError, TypeError, binascii.Error) as exc:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="共享链接不存在或已失效") from exc
+
+
+def _expiry_for_policy(policy: str, now: datetime) -> datetime | None:
+    duration = _EXPIRY_DURATIONS[policy]
+    return now + duration if duration else None
+
+
+def _is_expired(link: CollaborationShareLink, now: datetime | None = None) -> bool:
+    return bool(link.expires_at and link.expires_at <= (now or _now()))
+
+
+async def _link_for_file(
+    db: AsyncSession,
+    item: CollaborationFile,
+    user,
+    *,
+    create: bool,
+    lock: bool = False,
+) -> CollaborationShareLink | None:
+    await collaboration_service.require_file_manager(db, item, user)
+    statement = select(CollaborationShareLink).where(CollaborationShareLink.file_id == item.id)
+    if lock:
+        statement = statement.with_for_update()
+    link = await db.scalar(statement)
+    if link or not create:
+        return link
+    now = _now()
+    link = CollaborationShareLink(
+        file_id=item.id,
+        enabled=False,
+        access_mode="VIEW",
+        expiry_policy="SEVEN_DAYS",
+        expires_at=now + timedelta(days=7),
+        created_by=user.id,
+        updated_by=user.id,
+    )
+    db.add(link)
+    await db.flush()
+    return link
+
+
+def share_link_read(link: CollaborationShareLink) -> CollaborationShareLinkRead:
+    return CollaborationShareLinkRead(
+        id=link.id,
+        file_id=link.file_id,
+        enabled=link.enabled,
+        access_mode=link.access_mode,
+        expiry_policy=link.expiry_policy,
+        expires_at=link.expires_at,
+        has_password=bool(link.password_hash),
+        share_path=SHARE_PATH,
+        share_token=share_token(link) if link.enabled else None,
+        created_at=link.created_at,
+        updated_at=link.updated_at,
+    )
+
+
+async def get_share_link(
+    db: AsyncSession, item: CollaborationFile, user
+) -> CollaborationShareLinkRead:
+    link = await _link_for_file(db, item, user, create=True)
+    assert link is not None
+    await db.commit()
+    await db.refresh(link)
+    return share_link_read(link)
+
+
+async def update_share_link(
+    db: AsyncSession,
+    item: CollaborationFile,
+    payload: CollaborationShareLinkUpdate,
+    user,
+) -> CollaborationShareLinkRead:
+    link = await _link_for_file(db, item, user, create=True, lock=True)
+    assert link is not None
+    now = _now()
+    link.enabled = payload.enabled
+    link.access_mode = payload.access_mode
+    link.expiry_policy = payload.expiry_policy
+    link.expires_at = _expiry_for_policy(payload.expiry_policy, now)
+    link.updated_by = user.id
+    if payload.password_mode == "SET":
+        link.password_hash = await to_thread.run_sync(hash_password, payload.password or "")
+        link.failed_attempts = 0
+        link.last_failed_at = None
+        link.locked_until = None
+    elif payload.password_mode == "CLEAR":
+        link.password_hash = None
+        link.failed_attempts = 0
+        link.last_failed_at = None
+        link.locked_until = None
+    await db.commit()
+    await db.refresh(link)
+    return share_link_read(link)
+
+
+async def resolve_active_share(
+    db: AsyncSession,
+    token: str | None,
+    *,
+    lock: bool = False,
+) -> tuple[CollaborationShareLink, CollaborationFile]:
+    link_id, version = _decode_share_token(token)
+    statement = select(CollaborationShareLink).where(CollaborationShareLink.id == link_id)
+    if lock:
+        statement = statement.with_for_update()
+    link = await db.scalar(statement)
+    if not link or link.token_version != version or not link.enabled:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="共享链接不存在或已失效")
+    if _is_expired(link):
+        raise HTTPException(status_code=status.HTTP_410_GONE, detail="共享链接已过期")
+    item = await db.get(CollaborationFile, link.file_id)
+    if not item or item.deleted_at or item.status != "ACTIVE":
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="共享文件不存在或已停止共享")
+    return link, item
+
+
+async def public_metadata(
+    db: AsyncSession, token: str | None
+) -> CollaborationPublicShareMetadata:
+    link, item = await resolve_active_share(db, token)
+    return CollaborationPublicShareMetadata(
+        file_name=item.title,
+        file_type=item.file_type,
+        access_mode="edit" if link.access_mode == "EDIT" else "view",
+        allow_export=item.allow_export,
+        requires_password=bool(link.password_hash),
+        expires_at=link.expires_at,
+    )
+
+
+def _grant_token(link: CollaborationShareLink) -> CollaborationShareAccessGrant:
+    now = _now()
+    expires_at = now + timedelta(seconds=SHARE_ACCESS_TTL_SECONDS)
+    if link.expires_at and link.expires_at < expires_at:
+        expires_at = link.expires_at
+    value = jwt.encode(
+        {
+            "purpose": _ACCESS_PURPOSE,
+            "sub": str(link.id),
+            "ver": link.token_version,
+            "iat": int(now.timestamp()),
+            "exp": int(expires_at.timestamp()),
+        },
+        _signing_key().hex(),
+        algorithm="HS256",
+    )
+    return CollaborationShareAccessGrant(access_token=value, expires_at=expires_at)
+
+
+async def verify_share_password(
+    db: AsyncSession,
+    token: str | None,
+    password: str,
+) -> CollaborationShareAccessGrant:
+    link, _ = await resolve_active_share(db, token, lock=True)
+    if not link.password_hash:
+        return _grant_token(link)
+    now = _now()
+    if link.locked_until and link.locked_until > now:
+        raise HTTPException(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            detail="密码尝试次数过多，请稍后再试",
+        )
+    if link.last_failed_at and now - link.last_failed_at > PASSWORD_FAILURE_WINDOW:
+        link.failed_attempts = 0
+    valid = await to_thread.run_sync(verify_password, password, link.password_hash)
+    if not valid:
+        link.failed_attempts += 1
+        link.last_failed_at = now
+        if link.failed_attempts >= PASSWORD_FAILURE_LIMIT:
+            link.locked_until = now + PASSWORD_LOCK_DURATION
+        await db.commit()
+        if link.locked_until:
+            raise HTTPException(
+                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+                detail="密码尝试次数过多，请稍后再试",
+            )
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="链接密码不正确")
+    link.failed_attempts = 0
+    link.last_failed_at = None
+    link.locked_until = None
+    await db.commit()
+    return _grant_token(link)
+
+
+def validate_access_grant(link: CollaborationShareLink, value: str | None) -> None:
+    if not link.password_hash:
+        return
+    if not value:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="请输入链接密码")
+    try:
+        payload = jwt.decode(value, _signing_key().hex(), algorithms=["HS256"])
+        if (
+            payload.get("purpose") != _ACCESS_PURPOSE
+            or not hmac.compare_digest(str(payload.get("sub") or ""), str(link.id))
+            or payload.get("ver") != link.token_version
+        ):
+            raise JWTError("share grant mismatch")
+    except JWTError as exc:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="链接访问凭证已失效") from exc

backend/app/services/desktop_notification_service.py

@@ -10,12 +10,9 @@ from app.models.desktop_notification import (
     DesktopNotificationDelivery,
     DesktopNotificationSubscription,
 )
-from app.models.distribution import Distribution, DistributionStatus, DistributionTargetType
-from app.models.document import Document
-from app.models.document_version import DocumentVersion
-from app.models.study import Study
+from app.models.notification import Notification
 from app.models.study_member import StudyMember
-from app.schemas.notification import NotificationItem
+from app.models.user import User
 
 CLAIM_LEASE = timedelta(minutes=5)
 
@@ -47,46 +44,28 @@ async def set_subscription(
 
 
 def _eligible_query(user_id: uuid.UUID, enabled_at: datetime, lease_cutoff: datetime):
-    role_target_exists = exists(
+    active_membership = exists(
         select(StudyMember.id).where(
-            StudyMember.study_id == Document.trial_id,
+            StudyMember.study_id == Notification.study_id,
             StudyMember.user_id == user_id,
             StudyMember.is_active.is_(True),
-            StudyMember.role_in_study == Distribution.target_id,
         )
     )
-    target_matches = or_(
-        and_(
-            Distribution.target_type == DistributionTargetType.USER,
-            Distribution.target_id == str(user_id),
-        ),
-        and_(
-            Distribution.target_type == DistributionTargetType.ROLE,
-            role_target_exists,
-        ),
-    )
     return (
-        select(
-            Distribution,
-            Document,
-            DocumentVersion,
-            Study,
-            DesktopNotificationDelivery,
-        )
-        .join(Document, Distribution.document_id == Document.id)
-        .join(DocumentVersion, Distribution.version_id == DocumentVersion.id)
-        .join(Study, Document.trial_id == Study.id)
+        select(Notification, DesktopNotificationDelivery)
         .outerjoin(
             DesktopNotificationDelivery,
             and_(
-                DesktopNotificationDelivery.distribution_id == Distribution.id,
+                DesktopNotificationDelivery.notification_id == Notification.id,
                 DesktopNotificationDelivery.user_id == user_id,
             ),
         )
         .where(
-            Distribution.status == DistributionStatus.ACTIVE,
-            Distribution.created_at >= enabled_at,
-            target_matches,
+            Notification.recipient_id == user_id,
+            Notification.created_at >= enabled_at,
+            Notification.resolved_at.is_(None),
+            Notification.read_at.is_(None),
+            active_membership,
             or_(
                 DesktopNotificationDelivery.id.is_(None),
                 and_(
@@ -98,8 +77,8 @@ def _eligible_query(user_id: uuid.UUID, enabled_at: datetime, lease_cutoff: date
                 ),
             ),
         )
-        .order_by(Distribution.created_at.asc())
-        .with_for_update(of=Distribution, skip_locked=True)
+        .order_by(Notification.created_at.asc())
+        .with_for_update(of=Notification, skip_locked=True)
     )
 
 
@@ -107,45 +86,46 @@ async def claim_notifications(
     db: AsyncSession,
     user_id: uuid.UUID,
     limit: int,
-) -> tuple[uuid.UUID | None, datetime | None, list[NotificationItem]]:
+) -> tuple[uuid.UUID | None, datetime | None, list[Notification]]:
     subscription = await get_subscription(db, user_id)
     if not subscription or not subscription.enabled or not subscription.enabled_at:
         return None, None, []
 
+    # Reconcile every active project before selecting desktop deliveries so permission
+    # changes fail closed and time-based reminders do not depend on opening the web Feed.
+    user = await db.get(User, user_id)
+    if user is None or not user.is_active:
+        return None, None, []
+    study_ids = (await db.scalars(
+        select(StudyMember.study_id).where(
+            StudyMember.user_id == user_id,
+            StudyMember.is_active.is_(True),
+        )
+    )).all()
+    from app.services.project_reminder_service import sync_project_reminders
+
+    for study_id in set(study_ids):
+        await sync_project_reminders(db, study_id, user)
+
     now = datetime.now(timezone.utc)
     token = uuid.uuid4()
     rows = (
         await db.execute(
-            _eligible_query(user_id, subscription.enabled_at, now - CLAIM_LEASE).limit(max(1, min(limit, 50)))
+            _eligible_query(user_id, subscription.enabled_at, now - CLAIM_LEASE)
+            .limit(max(1, min(limit, 50)))
         )
     ).all()
-    items: list[NotificationItem] = []
-    for distribution, document, version, study, delivery in rows:
+    items: list[Notification] = []
+    for notification, delivery in rows:
         if delivery is None:
             delivery = DesktopNotificationDelivery(
                 user_id=user_id,
-                distribution_id=distribution.id,
+                notification_id=notification.id,
             )
             db.add(delivery)
         delivery.claim_token = token
         delivery.claimed_at = now
-        items.append(
-            NotificationItem(
-                id=distribution.id,
-                document_id=document.id,
-                version_id=version.id,
-                document_title=document.title,
-                document_no=document.doc_no,
-                version_no=version.version_no,
-                change_summary=version.change_summary,
-                effective_at=version.effective_at,
-                created_at=distribution.created_at,
-                study_id=study.id,
-                study_name=study.name,
-                delivered_at=delivery.delivered_at,
-                read_at=delivery.read_at,
-            )
-        )
+        items.append(notification)
     await db.commit()
     if not items:
         return None, None, []
@@ -164,7 +144,7 @@ async def acknowledge_notifications(
             .where(
                 DesktopNotificationDelivery.user_id == user_id,
                 DesktopNotificationDelivery.claim_token == claim_token,
-                DesktopNotificationDelivery.distribution_id.in_(delivered_ids),
+                DesktopNotificationDelivery.notification_id.in_(delivered_ids),
             )
             .values(delivered_at=datetime.now(timezone.utc))
         )
@@ -174,21 +154,17 @@ async def acknowledge_notifications(
 async def mark_notification_read(
     db: AsyncSession,
     user_id: uuid.UUID,
-    distribution_id: uuid.UUID,
+    notification_id: uuid.UUID,
 ) -> None:
-    delivery = (
-        await db.execute(
-            select(DesktopNotificationDelivery).where(
-                DesktopNotificationDelivery.user_id == user_id,
-                DesktopNotificationDelivery.distribution_id == distribution_id,
-            )
-        )
-    ).scalar_one_or_none()
-    if delivery is None:
-        delivery = DesktopNotificationDelivery(
-            user_id=user_id,
-            distribution_id=distribution_id,
-        )
-        db.add(delivery)
-    delivery.read_at = datetime.now(timezone.utc)
+    item = await db.scalar(select(Notification).where(
+        Notification.id == notification_id,
+        Notification.recipient_id == user_id,
+        Notification.resolved_at.is_(None),
+    ))
+    if item is None:
+        return
+    now = datetime.now(timezone.utc)
+    item.read_at = item.read_at or now
+    if not item.requires_action:
+        item.resolved_at = item.resolved_at or now
     await db.commit()

backend/app/services/document_service.py

@@ -6,11 +6,12 @@ import uuid
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Iterable
+from urllib.parse import quote
 
 import aiofiles
 from fastapi import HTTPException, UploadFile, status
 from fastapi.responses import FileResponse
-from sqlalchemy import and_, delete as sa_delete, or_, select, update as sa_update
+from sqlalchemy import String, and_, cast, delete as sa_delete, or_, select, update as sa_update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.deps import get_cra_site_scope
@@ -31,6 +32,7 @@ from app.models.distribution import Distribution, DistributionStatus, Distributi
 from app.models.document import Document, DocumentScopeType, DocumentStatus
 from app.models.document_version import DocumentVersion, DocumentVersionStatus
 from app.models.desktop_notification import DesktopNotificationDelivery
+from app.models.notification import Notification
 from app.models.study import Study
 from app.schemas.acknowledgement import AcknowledgementCreate
 from app.schemas.distribution import DistributionCreate, DistributionRead, DistributionStats
@@ -38,6 +40,7 @@ from app.schemas.document import DocumentCreate, DocumentDetail, DocumentSummary
 from app.schemas.document_version import DocumentVersionRead, DocumentVersionSummary
 from app.schemas.notification import NotificationItem
 from app.schemas.user import UserDisplay
+from app.services import notification_service
 
 UPLOAD_ROOT = Path(__file__).resolve().parent.parent / "uploads" / "documents"
 
@@ -52,6 +55,29 @@ DOCUMENT_ACTION_PERMISSIONS = {
 }
 
 
+def _safe_original_filename(value: str | None) -> str:
+    filename = (value or "").replace("\\", "/").rsplit("/", 1)[-1].strip()
+    filename = "".join(character for character in filename if ord(character) >= 32 and ord(character) != 127)
+    return filename[:255] or "document"
+
+
+def _content_disposition(filename: str, disposition: str = "attachment") -> str:
+    fallback = "".join(character if 32 <= ord(character) < 127 and character not in {'"', "\\"} else "_" for character in filename)
+    fallback = fallback or "download"
+    encoded = quote(filename, safe="")
+    return f'{disposition}; filename="{fallback}"; filename*=UTF-8\'\'{encoded}'
+
+
+def _legacy_download_filename(version: DocumentVersion, document: Document) -> str:
+    """Return a readable fallback for rows created before original_filename existed."""
+    stored_name = Path(version.file_uri).name
+    suffix = Path(stored_name).suffix
+    title = _safe_original_filename(document.title)
+    if suffix and title.lower().endswith(suffix.lower()):
+        return title
+    return f"{title}{suffix}"
+
+
 def _audit_detail(before: dict | None, after: dict | None) -> str:
     payload = {"before": before, "after": after}
     return json.dumps(payload, ensure_ascii=True)
@@ -356,7 +382,8 @@ async def create_version(
     file_hash = hashlib.sha256(content).hexdigest()
     dest_dir = UPLOAD_ROOT / str(document_id)
     dest_dir.mkdir(parents=True, exist_ok=True)
-    unique_name = f"{uuid.uuid4()}{Path(file.filename).suffix}"
+    original_filename = _safe_original_filename(file.filename)
+    unique_name = f"{uuid.uuid4()}{Path(original_filename).suffix}"
     dest_path = dest_dir / unique_name
     async with aiofiles.open(dest_path, "wb") as out_file:
         await out_file.write(content)
@@ -377,6 +404,7 @@ async def create_version(
         status=DocumentVersionStatus.EFFECTIVE,
         effective_at=effective_at,
         file_uri=str(dest_path),
+        original_filename=original_filename,
         file_hash=file_hash,
         file_size=len(content),
         mime_type=file.content_type,
@@ -552,12 +580,11 @@ async def get_version_download_response(
     file_path = Path(version.file_uri)
     if not file_path.exists():
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="文件不存在")
-    filename = file_path.name
+    filename = version.original_filename or _legacy_download_filename(version, doc)
     return FileResponse(
         path=str(file_path),
-        filename=filename,
         media_type=version.mime_type or "application/octet-stream",
-        headers={"Content-Disposition": f'inline; filename="{filename}"'},
+        headers={"Content-Disposition": _content_disposition(filename)},
     )
 
 
@@ -663,6 +690,10 @@ async def create_acknowledgement(
     if distribution.target_type == DistributionTargetType.ROLE:
         if distribution.target_id not in ("ADMIN" if is_system_admin(current_user) else "", getattr(membership, "role_in_study", "")):
             raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="不在分发范围内")
+    if distribution.target_type == DistributionTargetType.SITE and not is_system_admin(current_user):
+        site_ids = await site_crud.list_ids_by_contact_user(db, doc.trial_id, current_user.id)
+        if distribution.target_id not in {str(site_id) for site_id in site_ids}:
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="不在分发中心范围内")
 
     if payload.ack_type != AcknowledgementType.RECEIVED:
         raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="仅支持已接收回执")
@@ -697,6 +728,12 @@ async def create_acknowledgement(
             operator_role=await get_operator_role_label(db, doc.trial_id, current_user),
         )
     )
+    await notification_service.resolve_source_notifications(
+        db,
+        source_type="DOCUMENT_DISTRIBUTION",
+        source_id=str(distribution.id),
+        recipient_id=current_user.id,
+    )
     await db.commit()
     await db.refresh(ack)
     return ack
@@ -787,10 +824,18 @@ async def list_distribution_notifications(
         .join(DocumentVersion, Distribution.version_id == DocumentVersion.id)
         .join(Document, Distribution.document_id == Document.id)
         .join(Study, Document.trial_id == Study.id)
+        .outerjoin(
+            Notification,
+            and_(
+                Notification.source_type == "DOCUMENT_DISTRIBUTION",
+                Notification.source_id == cast(Distribution.id, String),
+                Notification.recipient_id == current_user.id,
+            ),
+        )
         .outerjoin(
             DesktopNotificationDelivery,
             and_(
-                DesktopNotificationDelivery.distribution_id == Distribution.id,
+                DesktopNotificationDelivery.notification_id == Notification.id,
                 DesktopNotificationDelivery.user_id == current_user.id,
             ),
         )

backend/app/services/geo_location_metadata.py

@@ -0,0 +1,175 @@
+"""Canonical map metadata for monitoring source locations.
+
+ip2region provides administrative names but no coordinates. This module keeps
+the normalization and centroid contract on the server so web and desktop
+clients render the same location semantics.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from app.services.ip_location import IpLocation
+
+
+@dataclass(frozen=True)
+class GeoLocationMetadata:
+    country: str
+    country_code: str
+    region_code: str
+    longitude: float | None
+    latitude: float | None
+    accuracy_level: str
+
+
+COUNTRY_ALIASES = {
+    "中国": "China",
+    "China": "China",
+    "Mainland China": "China",
+    "中国香港": "China",
+    "中国澳门": "China",
+    "中国台湾": "China",
+    "美国": "United States",
+    "United States": "United States",
+    "United States of America": "United States",
+    "USA": "United States",
+    "土耳其": "Türkiye",
+    "Turkey": "Türkiye",
+    "Türkiye": "Türkiye",
+}
+
+COUNTRY_CODES = {
+    "China": "CN",
+    "United States": "US",
+    "Netherlands": "NL",
+    "Türkiye": "TR",
+    "Australia": "AU",
+    "Japan": "JP",
+    "Singapore": "SG",
+    "Germany": "DE",
+    "France": "FR",
+    "United Kingdom": "GB",
+    "Canada": "CA",
+    "India": "IN",
+    "Russia": "RU",
+}
+
+COUNTRY_CENTROIDS = {
+    "China": (104.1954, 35.8617),
+    "United States": (-95.7129, 37.0902),
+    "Netherlands": (5.2913, 52.1326),
+    "Türkiye": (35.2433, 38.9637),
+    "Australia": (133.7751, -25.2744),
+    "Japan": (138.2529, 36.2048),
+    "Singapore": (103.8198, 1.3521),
+    "Germany": (10.4515, 51.1657),
+    "France": (2.2137, 46.2276),
+    "United Kingdom": (-3.436, 55.3781),
+    "Canada": (-106.3468, 56.1304),
+    "India": (78.9629, 20.5937),
+    "Russia": (105.3188, 61.524),
+}
+
+CHINA_REGION_METADATA = {
+    "北京市": ("CN-BJ", 116.4074, 39.9042),
+    "天津市": ("CN-TJ", 117.2008, 39.0842),
+    "河北省": ("CN-HE", 114.5025, 38.0455),
+    "山西省": ("CN-SX", 112.5492, 37.857),
+    "内蒙古自治区": ("CN-NM", 111.6708, 40.8183),
+    "辽宁省": ("CN-LN", 123.4315, 41.8057),
+    "吉林省": ("CN-JL", 125.3245, 43.8868),
+    "黑龙江省": ("CN-HL", 126.6424, 45.7567),
+    "上海市": ("CN-SH", 121.4737, 31.2304),
+    "江苏省": ("CN-JS", 118.7633, 32.0617),
+    "浙江省": ("CN-ZJ", 120.1551, 30.2741),
+    "安徽省": ("CN-AH", 117.2272, 31.8206),
+    "福建省": ("CN-FJ", 119.2965, 26.0745),
+    "江西省": ("CN-JX", 115.8582, 28.682),
+    "山东省": ("CN-SD", 117.1201, 36.6512),
+    "河南省": ("CN-HA", 113.6254, 34.7466),
+    "湖北省": ("CN-HB", 114.3055, 30.5928),
+    "湖南省": ("CN-HN", 112.9388, 28.2282),
+    "广东省": ("CN-GD", 113.2644, 23.1291),
+    "广西壮族自治区": ("CN-GX", 108.3669, 22.817),
+    "海南省": ("CN-HI", 110.3312, 20.0311),
+    "重庆": ("CN-CQ", 106.5516, 29.563),
+    "重庆市": ("CN-CQ", 106.5516, 29.563),
+    "四川省": ("CN-SC", 104.0665, 30.5723),
+    "贵州省": ("CN-GZ", 106.6302, 26.647),
+    "云南省": ("CN-YN", 102.8329, 24.8801),
+    "西藏自治区": ("CN-XZ", 91.1322, 29.6604),
+    "陕西省": ("CN-SN", 108.9398, 34.3416),
+    "甘肃省": ("CN-GS", 103.8343, 36.0611),
+    "青海省": ("CN-QH", 101.7782, 36.6171),
+    "宁夏回族自治区": ("CN-NX", 106.2309, 38.4872),
+    "新疆维吾尔自治区": ("CN-XJ", 87.6168, 43.8256),
+    "台湾省": ("CN-TW", 121.5654, 25.033),
+    "香港特别行政区": ("CN-HK", 114.1694, 22.3193),
+    "澳门特别行政区": ("CN-MO", 113.5439, 22.1987),
+}
+
+CITY_CENTROIDS = {
+    "南京": (118.7969, 32.0603),
+    "南京市": (118.7969, 32.0603),
+    "San Jose": (-121.8863, 37.3382),
+    "South Holland": (4.493, 52.0208),
+    "Istanbul": (28.9784, 41.0082),
+}
+
+
+def resolve_geo_location_metadata(location: IpLocation) -> GeoLocationMetadata:
+    if location.location in {"局域网", "本机"}:
+        return GeoLocationMetadata(
+            country="",
+            country_code="PRIVATE",
+            region_code="PRIVATE",
+            longitude=None,
+            latitude=None,
+            accuracy_level="private",
+        )
+
+    country = COUNTRY_ALIASES.get(location.country, location.country)
+    country_code = COUNTRY_CODES.get(country, "")
+
+    city_coordinate = CITY_CENTROIDS.get(location.city)
+    if city_coordinate:
+        return GeoLocationMetadata(
+            country=country,
+            country_code=country_code,
+            region_code="",
+            longitude=city_coordinate[0],
+            latitude=city_coordinate[1],
+            accuracy_level="city",
+        )
+
+    region_metadata = CHINA_REGION_METADATA.get(location.province)
+    if country in {"China", "中国"} and region_metadata:
+        region_code, longitude, latitude = region_metadata
+        return GeoLocationMetadata(
+            country="China",
+            country_code="CN",
+            region_code=region_code,
+            longitude=longitude,
+            latitude=latitude,
+            accuracy_level="region",
+        )
+
+    country_coordinate = COUNTRY_CENTROIDS.get(country)
+    if country_coordinate:
+        return GeoLocationMetadata(
+            country=country,
+            country_code=country_code,
+            region_code="",
+            longitude=country_coordinate[0],
+            latitude=country_coordinate[1],
+            accuracy_level="country",
+        )
+
+    return GeoLocationMetadata(
+        country=country,
+        country_code=country_code,
+        region_code="",
+        longitude=None,
+        latitude=None,
+        accuracy_level="unknown",
+    )

backend/app/services/ip_geolocation_fallback.py

@@ -0,0 +1,174 @@
+"""Bounded third-party coordinate fallback for public source IPs.
+
+IPAddress.my identifies IP2Location.io as its data provider. We use the
+provider's documented JSON API instead of scraping the public HTML page.
+Only globally routable addresses are eligible, and results are cached so the
+monitoring UI does not turn into a per-refresh third-party lookup fan-out.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import ipaddress
+import logging
+import math
+import time
+from dataclasses import dataclass
+from typing import Any, Iterable
+
+import httpx
+
+from app.core.config import settings
+from app.services.geo_location_metadata import GeoLocationMetadata
+from app.services.ip_location import IpLocation
+
+logger = logging.getLogger("ctms.ip_geolocation_fallback")
+
+_API_URL = "https://api.ip2location.io/"
+_MAX_RESPONSE_BYTES = 64 * 1024
+_MAX_CACHE_ENTRIES = 4096
+_NEGATIVE_CACHE_SECONDS = 3600
+_cache: dict[str, tuple[float, "ExternalIpLocation | None"]] = {}
+_cache_lock = asyncio.Lock()
+
+
+def _text(value: Any, limit: int = 160) -> str:
+    candidate = str(value or "").strip()
+    return "" if candidate in {"", "-", "0"} else candidate[:limit]
+
+
+def _coordinate(value: Any, *, minimum: float, maximum: float) -> float | None:
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return None
+    return number if math.isfinite(number) and minimum <= number <= maximum else None
+
+
+def _global_ip(value: str | None) -> str | None:
+    try:
+        address = ipaddress.ip_address((value or "").strip())
+    except ValueError:
+        return None
+    return str(address) if address.is_global else None
+
+
+@dataclass(frozen=True)
+class ExternalIpLocation:
+    ip_address: str
+    country: str
+    country_code: str
+    region: str
+    city: str
+    isp: str
+    longitude: float
+    latitude: float
+
+    def merge_ip_location(self, local: IpLocation) -> IpLocation:
+        country = self.country or local.country
+        province = self.region or local.province
+        city = self.city or local.city
+        isp = self.isp or local.isp
+        location = " / ".join(part for part in [country, province, city, isp] if part) or local.location
+        return IpLocation(location=location, country=country, province=province, city=city, isp=isp)
+
+    def to_metadata(self) -> GeoLocationMetadata:
+        return GeoLocationMetadata(
+            country=self.country,
+            country_code=self.country_code,
+            region_code="",
+            longitude=self.longitude,
+            latitude=self.latitude,
+            accuracy_level="city" if self.city else "country",
+        )
+
+
+def _parse_response(expected_ip: str, payload: Any) -> ExternalIpLocation | None:
+    if not isinstance(payload, dict):
+        return None
+    response_ip = _global_ip(_text(payload.get("ip"), 45))
+    if response_ip != expected_ip:
+        return None
+    latitude = _coordinate(payload.get("latitude"), minimum=-90, maximum=90)
+    longitude = _coordinate(payload.get("longitude"), minimum=-180, maximum=180)
+    if latitude is None or longitude is None:
+        return None
+    country_code = _text(payload.get("country_code"), 2).upper()
+    if len(country_code) != 2:
+        country_code = ""
+    return ExternalIpLocation(
+        ip_address=expected_ip,
+        country=_text(payload.get("country_name"), 100),
+        country_code=country_code,
+        region=_text(payload.get("region_name"), 100),
+        city=_text(payload.get("city_name"), 100),
+        isp=_text(payload.get("isp"), 160),
+        longitude=longitude,
+        latitude=latitude,
+    )
+
+
+async def _fetch_one(
+    client: httpx.AsyncClient,
+    semaphore: asyncio.Semaphore,
+    ip_address: str,
+) -> ExternalIpLocation | None:
+    async with semaphore:
+        try:
+            response = await client.get(_API_URL, params={"ip": ip_address, "format": "json"})
+            response.raise_for_status()
+            if len(response.content) > _MAX_RESPONSE_BYTES:
+                return None
+            return _parse_response(ip_address, response.json())
+        except (httpx.HTTPError, ValueError):
+            logger.warning("External IP coordinate fallback unavailable")
+            return None
+
+
+async def resolve_external_ip_locations(ip_addresses: Iterable[str]) -> dict[str, ExternalIpLocation]:
+    if not settings.MONITORING_IP_GEO_FALLBACK_ENABLED or settings.ENV == "test":
+        return {}
+
+    candidates = list(dict.fromkeys(filter(None, (_global_ip(value) for value in ip_addresses))))
+    if not candidates:
+        return {}
+
+    now = time.monotonic()
+    resolved: dict[str, ExternalIpLocation] = {}
+    pending: list[str] = []
+    async with _cache_lock:
+        for ip_address in candidates:
+            cached = _cache.get(ip_address)
+            if cached and cached[0] > now:
+                if cached[1] is not None:
+                    resolved[ip_address] = cached[1]
+                continue
+            pending.append(ip_address)
+
+    pending = pending[: settings.MONITORING_IP_GEO_FALLBACK_MAX_LOOKUPS]
+    if not pending:
+        return resolved
+
+    headers = {"Accept": "application/json", "User-Agent": "CTMS-IP-Coordinate-Fallback/1.0"}
+    api_key = (settings.MONITORING_IP_GEO_FALLBACK_API_KEY or "").strip()
+    if api_key:
+        headers["Authorization"] = f"Bearer {api_key}"
+    timeout = httpx.Timeout(settings.MONITORING_IP_GEO_FALLBACK_TIMEOUT_SECONDS)
+    semaphore = asyncio.Semaphore(min(5, len(pending)))
+    async with httpx.AsyncClient(timeout=timeout, follow_redirects=False, headers=headers) as client:
+        fetched = await asyncio.gather(*(_fetch_one(client, semaphore, item) for item in pending))
+
+    now = time.monotonic()
+    async with _cache_lock:
+        for ip_address, item in zip(pending, fetched):
+            ttl = settings.MONITORING_IP_GEO_FALLBACK_CACHE_SECONDS if item else _NEGATIVE_CACHE_SECONDS
+            _cache[ip_address] = (now + ttl, item)
+            if item is not None:
+                resolved[ip_address] = item
+        while len(_cache) > _MAX_CACHE_ENTRIES:
+            _cache.pop(next(iter(_cache)))
+    return resolved
+
+
+def reset_ip_geolocation_fallback_cache() -> None:
+    _cache.clear()

backend/app/services/ip_location.py

@@ -9,6 +9,7 @@ import ipaddress
 import logging
 import sys
 from dataclasses import dataclass
+from functools import lru_cache
 from pathlib import Path
 from typing import Optional
 
@@ -125,5 +126,6 @@ class Ip2RegionResolver:
 _resolver = Ip2RegionResolver()
 
 
+@lru_cache(maxsize=8192)
 def resolve_ip_location(ip: str | None) -> IpLocation:
     return _resolver.lookup(ip)

backend/app/services/monitoring_retention.py

@@ -0,0 +1,151 @@
+"""监测数据留存清理任务。"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from sqlalchemy import delete
+
+from app.core.config import settings
+from app.db.session import SessionLocal
+from app.models.permission_access_log import PermissionAccessLog
+from app.models.permission_metric_snapshot import PermissionMetricSnapshot
+from app.models.security_access_log import SecurityAccessLog
+from app.models.source_location_snapshot import SourceLocationSnapshot
+from app.models.user_login_session import UserLoginSession
+
+logger = logging.getLogger("ctms.monitoring_retention")
+
+
+@dataclass
+class MonitoringRetentionState:
+    running: bool = False
+    last_run_started_at: datetime | None = None
+    last_success_at: datetime | None = None
+    last_error_at: datetime | None = None
+    last_error_type: str | None = None
+    error_count: int = 0
+    last_deleted: dict[str, int] = field(default_factory=dict)
+    total_deleted: dict[str, int] = field(default_factory=dict)
+
+    def snapshot(self) -> dict[str, Any]:
+        return {
+            "running": self.running,
+            "access_log_retention_days": settings.MONITORING_ACCESS_LOG_RETENTION_DAYS,
+            "metric_retention_days": settings.MONITORING_METRIC_RETENTION_DAYS,
+            "login_activity_retention_days": settings.USER_LOGIN_ACTIVITY_RETENTION_DAYS,
+            "interval_seconds": settings.MONITORING_RETENTION_INTERVAL_SECONDS,
+            "last_run_started_at": (
+                self.last_run_started_at.isoformat() if self.last_run_started_at else None
+            ),
+            "last_success_at": self.last_success_at.isoformat() if self.last_success_at else None,
+            "last_error_at": self.last_error_at.isoformat() if self.last_error_at else None,
+            "last_error_type": self.last_error_type,
+            "error_count": self.error_count,
+            "last_deleted": dict(self.last_deleted),
+            "total_deleted": dict(self.total_deleted),
+        }
+
+
+_state = MonitoringRetentionState()
+
+
+def get_monitoring_retention_status() -> dict[str, Any]:
+    return _state.snapshot()
+
+
+def _deleted_count(result: Any) -> int:
+    rowcount = getattr(result, "rowcount", 0)
+    return max(0, int(rowcount or 0))
+
+
+async def purge_expired_monitoring_data(
+    *, now: datetime | None = None
+) -> dict[str, int]:
+    reference_time = now or datetime.now(timezone.utc)
+    access_cutoff = reference_time - timedelta(
+        days=settings.MONITORING_ACCESS_LOG_RETENTION_DAYS
+    )
+    metric_cutoff = reference_time - timedelta(
+        days=settings.MONITORING_METRIC_RETENTION_DAYS
+    )
+    login_activity_cutoff = reference_time - timedelta(
+        days=settings.USER_LOGIN_ACTIVITY_RETENTION_DAYS
+    )
+
+    async with SessionLocal() as session:
+        permission_result = await session.execute(
+            delete(PermissionAccessLog).where(PermissionAccessLog.created_at < access_cutoff)
+        )
+        security_result = await session.execute(
+            delete(SecurityAccessLog).where(SecurityAccessLog.created_at < access_cutoff)
+        )
+        metric_result = await session.execute(
+            delete(PermissionMetricSnapshot).where(
+                PermissionMetricSnapshot.bucket_time < metric_cutoff
+            )
+        )
+        source_location_result = await session.execute(
+            delete(SourceLocationSnapshot).where(
+                SourceLocationSnapshot.bucket_time < access_cutoff
+            )
+        )
+        login_session_result = await session.execute(
+            delete(UserLoginSession).where(UserLoginSession.last_seen_at < login_activity_cutoff)
+        )
+        await session.commit()
+
+    return {
+        "permission_access_logs": _deleted_count(permission_result),
+        "security_access_logs": _deleted_count(security_result),
+        "permission_metric_snapshots": _deleted_count(metric_result),
+        "source_location_snapshots": _deleted_count(source_location_result),
+        "user_login_sessions": _deleted_count(login_session_result),
+    }
+
+
+async def run_monitoring_retention_once(
+    *, now: datetime | None = None
+) -> dict[str, int]:
+    _state.last_run_started_at = now or datetime.now(timezone.utc)
+    try:
+        deleted = await purge_expired_monitoring_data(now=now)
+    except Exception as exc:
+        _state.last_error_at = datetime.now(timezone.utc)
+        _state.last_error_type = type(exc).__name__
+        _state.error_count += 1
+        raise
+
+    _state.last_success_at = datetime.now(timezone.utc)
+    _state.last_deleted = deleted
+    for key, count in deleted.items():
+        _state.total_deleted[key] = _state.total_deleted.get(key, 0) + count
+    return deleted
+
+
+async def run_monitoring_retention(stop_event: asyncio.Event) -> None:
+    logger.info("Monitoring retention task started")
+    _state.running = True
+    try:
+        while not stop_event.is_set():
+            try:
+                deleted = await run_monitoring_retention_once()
+                if any(deleted.values()):
+                    logger.info("Purged expired monitoring data: %s", deleted)
+            except Exception:
+                logger.exception("Failed to purge expired monitoring data")
+
+            try:
+                await asyncio.wait_for(
+                    stop_event.wait(),
+                    timeout=settings.MONITORING_RETENTION_INTERVAL_SECONDS,
+                )
+            except asyncio.TimeoutError:
+                continue
+    finally:
+        _state.running = False
+        logger.info("Monitoring retention task stopped")

backend/app/services/monitoring_server_location.py

@@ -0,0 +1,170 @@
+"""Resolve the monitoring deployment's public network location.
+
+The map server marker is derived from the deployment's public IP instead of a
+frontend or configuration coordinate. An explicit public IP is accepted for
+restricted networks; otherwise the public frontend hostname and, finally, a
+small set of public-IP discovery endpoints are used. Results are cached so the
+monitoring API never performs per-request network discovery.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import ipaddress
+import logging
+import socket
+import time
+from dataclasses import dataclass
+from urllib.parse import urlparse
+
+import httpx
+
+from app.core.config import settings
+from app.services.geo_location_metadata import resolve_geo_location_metadata
+from app.services.ip_geolocation_fallback import resolve_external_ip_locations
+from app.services.ip_location import resolve_ip_location
+
+logger = logging.getLogger("ctms.monitoring_server_location")
+
+
+@dataclass(frozen=True)
+class MonitoringServerLocation:
+    name: str
+    longitude: float
+    latitude: float
+    accuracy_level: str
+    resolution_source: str
+
+    def to_public_dict(self) -> dict:
+        return {
+            "name": self.name,
+            "longitude": self.longitude,
+            "latitude": self.latitude,
+            "accuracy_level": self.accuracy_level,
+            "resolution_source": self.resolution_source,
+        }
+
+
+_cached_location: MonitoringServerLocation | None = None
+_cache_initialized = False
+_cache_expires_at = 0.0
+_cache_lock = asyncio.Lock()
+
+
+def _normalize_public_ip(value: str | None) -> str | None:
+    candidate = (value or "").strip().splitlines()[0] if (value or "").strip() else ""
+    try:
+        address = ipaddress.ip_address(candidate)
+    except ValueError:
+        return None
+    return str(address) if address.is_global else None
+
+
+async def _resolve_frontend_public_ip() -> str | None:
+    hostname = urlparse(settings.FRONTEND_PUBLIC_URL).hostname
+    if not hostname:
+        return None
+    literal_ip = _normalize_public_ip(hostname)
+    if literal_ip:
+        return literal_ip
+    try:
+        loop = asyncio.get_running_loop()
+        records = await loop.getaddrinfo(hostname, None, type=socket.SOCK_STREAM)
+    except (OSError, socket.gaierror):
+        return None
+    candidates = []
+    for _, _, _, _, socket_address in records:
+        candidate = _normalize_public_ip(str(socket_address[0]))
+        if candidate and candidate not in candidates:
+            candidates.append(candidate)
+    return next((item for item in candidates if ":" not in item), candidates[0] if candidates else None)
+
+
+async def _discover_public_ip() -> tuple[str | None, str]:
+    configured_ip = _normalize_public_ip(settings.MONITORING_SERVER_PUBLIC_IP)
+    if configured_ip:
+        return configured_ip, "configured_public_ip"
+
+    frontend_ip = await _resolve_frontend_public_ip()
+    if frontend_ip:
+        return frontend_ip, "frontend_dns"
+
+    if settings.ENV == "test":
+        return None, "unavailable"
+
+    urls = [
+        item.strip()
+        for item in settings.MONITORING_PUBLIC_IP_DISCOVERY_URLS.split(",")
+        if item.strip()
+    ]
+    timeout = httpx.Timeout(settings.MONITORING_PUBLIC_IP_DISCOVERY_TIMEOUT_SECONDS)
+    async with httpx.AsyncClient(timeout=timeout, follow_redirects=False) as client:
+        for url in urls:
+            try:
+                response = await client.get(url, headers={"Accept": "text/plain"})
+                response.raise_for_status()
+            except httpx.HTTPError:
+                logger.warning("Public IP discovery endpoint unavailable", extra={"endpoint": url})
+                continue
+            discovered_ip = _normalize_public_ip(response.text[:128])
+            if discovered_ip:
+                return discovered_ip, "public_ip_discovery"
+    return None, "unavailable"
+
+
+async def resolve_monitoring_server_location(*, force: bool = False) -> MonitoringServerLocation | None:
+    global _cached_location, _cache_initialized, _cache_expires_at
+
+    now = time.monotonic()
+    if not force and _cache_initialized and now < _cache_expires_at:
+        return _cached_location
+
+    async with _cache_lock:
+        now = time.monotonic()
+        if not force and _cache_initialized and now < _cache_expires_at:
+            return _cached_location
+
+        public_ip, resolution_source = await _discover_public_ip()
+        location = None
+        if public_ip:
+            ip_location = resolve_ip_location(public_ip)
+            metadata = resolve_geo_location_metadata(ip_location)
+            if metadata.longitude is None or metadata.latitude is None:
+                external_location = (await resolve_external_ip_locations([public_ip])).get(public_ip)
+                if external_location is not None:
+                    ip_location = external_location.merge_ip_location(ip_location)
+                    metadata = external_location.to_metadata()
+            if metadata.longitude is not None and metadata.latitude is not None:
+                country = metadata.country or ip_location.country
+                name = " / ".join(
+                    part for part in [country, ip_location.province, ip_location.city] if part
+                ) or "公网服务器"
+                location = MonitoringServerLocation(
+                    name=name,
+                    longitude=metadata.longitude,
+                    latitude=metadata.latitude,
+                    accuracy_level=metadata.accuracy_level,
+                    resolution_source=resolution_source,
+                )
+                logger.info(
+                    "Monitoring server location resolved",
+                    extra={"resolution_source": resolution_source, "accuracy_level": metadata.accuracy_level},
+                )
+            else:
+                logger.warning("Deployment public IP has no usable geo coordinates")
+        else:
+            logger.warning("Unable to discover the deployment public IP")
+
+        _cached_location = location
+        _cache_initialized = True
+        cache_seconds = settings.MONITORING_SERVER_LOCATION_CACHE_SECONDS if location else 300
+        _cache_expires_at = now + cache_seconds
+        return location
+
+
+def reset_monitoring_server_location_cache() -> None:
+    """Reset process-local state for configuration reloads and tests."""
+    global _cached_location, _cache_initialized, _cache_expires_at
+    _cached_location = None
+    _cache_initialized = False
+    _cache_expires_at = 0.0

backend/app/services/notification_scheduler.py

@@ -0,0 +1,78 @@
+"""Server-side materialization for time and state driven business reminders."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from sqlalchemy import select, text
+
+from app.core.config import settings
+from app.db.session import SessionLocal
+from app.models.study_member import StudyMember
+from app.models.user import User, UserStatus
+from app.services.project_reminder_service import sync_project_reminders
+
+logger = logging.getLogger("ctms.notifications")
+NOTIFICATION_SYNC_LOCK_KEY = 0x43544D53  # "CTMS"
+
+
+async def sync_all_project_reminders_once() -> int:
+    """Synchronize every active project member once, with a cross-worker advisory lock."""
+    async with SessionLocal() as lock_session:
+        acquired = bool(await lock_session.scalar(
+            text("SELECT pg_try_advisory_lock(:lock_key)"),
+            {"lock_key": NOTIFICATION_SYNC_LOCK_KEY},
+        ))
+        if not acquired:
+            return 0
+        try:
+            memberships = (await lock_session.execute(
+                select(StudyMember.study_id, StudyMember.user_id)
+                .join(User, User.id == StudyMember.user_id)
+                .where(
+                    StudyMember.is_active.is_(True),
+                    User.status == UserStatus.ACTIVE,
+                )
+            )).all()
+            await lock_session.commit()
+            synced = 0
+            for study_id, user_id in memberships:
+                async with SessionLocal() as session:
+                    try:
+                        user = await session.get(User, user_id)
+                        if user is None or not user.is_active:
+                            continue
+                        await sync_project_reminders(session, study_id, user)
+                        synced += 1
+                    except Exception:
+                        await session.rollback()
+                        logger.exception(
+                            "Failed to synchronize reminders for study=%s user=%s",
+                            study_id,
+                            user_id,
+                        )
+            return synced
+        finally:
+            await lock_session.execute(
+                text("SELECT pg_advisory_unlock(:lock_key)"),
+                {"lock_key": NOTIFICATION_SYNC_LOCK_KEY},
+            )
+
+
+async def run_notification_sync_job(stop_event: asyncio.Event) -> None:
+    logger.info("Notification synchronization task started")
+    while not stop_event.is_set():
+        try:
+            synced = await sync_all_project_reminders_once()
+            logger.debug("Synchronized reminders for %s project memberships", synced)
+        except Exception:
+            logger.exception("Notification synchronization task failed")
+        try:
+            await asyncio.wait_for(
+                stop_event.wait(),
+                timeout=settings.NOTIFICATION_SYNC_INTERVAL_SECONDS,
+            )
+        except asyncio.TimeoutError:
+            continue
+    logger.info("Notification synchronization task stopped")

backend/app/services/notification_service.py

@@ -0,0 +1,366 @@
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from typing import Iterable
+
+from fastapi import HTTPException, status
+from sqlalchemy import func, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.desktop_notification import DesktopNotificationDelivery
+from app.models.notification import Notification
+from app.schemas.notification import GeneralNotificationFeed
+
+
+async def _requeue_desktop_delivery(db: AsyncSession, notification_id: uuid.UUID) -> None:
+    """Allow an escalated/reopened reminder to be delivered again by the desktop channel."""
+    await db.execute(
+        update(DesktopNotificationDelivery)
+        .where(DesktopNotificationDelivery.notification_id == notification_id)
+        .values(
+            claim_token=None,
+            claimed_at=None,
+            delivered_at=None,
+        )
+    )
+
+
+async def create_recipient_notifications(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    recipient_ids: Iterable[uuid.UUID],
+    category: str,
+    priority: str,
+    title: str,
+    message: str,
+    action_path: str | None,
+    source_type: str,
+    source_id: str,
+    dedupe_key: str,
+    source_version: str | None = None,
+    requires_action: bool = True,
+    due_at: datetime | None = None,
+) -> None:
+    recipients = set(recipient_ids)
+    if not recipients:
+        return
+    existing = set((await db.scalars(
+        select(Notification.recipient_id).where(
+            Notification.recipient_id.in_(recipients),
+            Notification.dedupe_key == dedupe_key,
+        )
+    )).all())
+    for recipient_id in recipients - existing:
+        db.add(Notification(
+            study_id=study_id,
+            recipient_id=recipient_id,
+            category=category,
+            priority=priority,
+            title=title,
+            message=message,
+            action_path=action_path,
+            source_type=source_type,
+            source_id=source_id,
+            source_version=source_version,
+            dedupe_key=dedupe_key,
+            requires_action=requires_action,
+            due_at=due_at,
+        ))
+
+
+async def sync_state_notification(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    recipient_id: uuid.UUID,
+    category: str,
+    priority: str,
+    title: str,
+    message: str,
+    action_path: str,
+    source_type: str,
+    source_id: str,
+    source_version: str,
+    active: bool,
+    due_at: datetime | None = None,
+    requires_action: bool = True,
+    dedupe_key: str | None = None,
+) -> None:
+    """Synchronize one actionable business state into a recipient reminder."""
+    dedupe_key = dedupe_key or f"state:{source_type}:{source_id}"
+    item = await db.scalar(select(Notification).where(
+        Notification.recipient_id == recipient_id,
+        Notification.dedupe_key == dedupe_key,
+    ))
+    now = datetime.now(timezone.utc)
+    if not active:
+        if item and item.resolved_at is None:
+            item.resolved_at = now
+            item.read_at = item.read_at or now
+        return
+    if item is None:
+        db.add(Notification(
+            study_id=study_id,
+            recipient_id=recipient_id,
+            category=category,
+            priority=priority,
+            title=title,
+            message=message,
+            action_path=action_path,
+            source_type=source_type,
+            source_id=source_id,
+            source_version=source_version,
+            dedupe_key=dedupe_key,
+            requires_action=requires_action,
+            due_at=due_at,
+        ))
+        return
+
+    should_notify_again = item.resolved_at is not None or item.source_version != source_version
+    item.study_id = study_id
+    item.category = category
+    item.priority = priority
+    item.title = title
+    item.message = message
+    item.action_path = action_path
+    item.source_version = source_version
+    item.requires_action = requires_action
+    item.due_at = due_at
+    if should_notify_again:
+        item.read_at = None
+        item.created_at = now
+        await _requeue_desktop_delivery(db, item.id)
+    item.resolved_at = None
+
+
+async def sync_aggregate_notification(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    recipient_id: uuid.UUID,
+    category: str,
+    priority: str,
+    title: str,
+    message: str,
+    action_path: str,
+    source_type: str,
+    source_id: str,
+    count: int,
+    due_at: datetime | None = None,
+) -> None:
+    dedupe_key = f"aggregate:{source_type}:{source_id}"
+    item = await db.scalar(select(Notification).where(
+        Notification.recipient_id == recipient_id,
+        Notification.dedupe_key == dedupe_key,
+    ))
+    now = datetime.now(timezone.utc)
+    if count <= 0:
+        if item and item.resolved_at is None:
+            item.resolved_at = now
+            item.read_at = item.read_at or now
+        return
+    version = str(count)
+    if item is None:
+        db.add(Notification(
+            study_id=study_id,
+            recipient_id=recipient_id,
+            category=category,
+            priority=priority,
+            title=title,
+            message=message,
+            action_path=action_path,
+            source_type=source_type,
+            source_id=source_id,
+            source_version=version,
+            dedupe_key=dedupe_key,
+            requires_action=True,
+            due_at=due_at,
+        ))
+        return
+    previous_count = int(item.source_version or 0)
+    item.category = category
+    item.priority = priority
+    item.title = title
+    item.message = message
+    item.action_path = action_path
+    item.source_version = version
+    item.due_at = due_at
+    if item.resolved_at is not None or count > previous_count:
+        item.read_at = None
+        item.created_at = now
+        await _requeue_desktop_delivery(db, item.id)
+    item.resolved_at = None
+
+
+async def resolve_missing_recipient_sources(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    recipient_id: uuid.UUID,
+    source_type: str,
+    active_source_ids: set[str],
+) -> None:
+    filters = [
+        Notification.study_id == study_id,
+        Notification.recipient_id == recipient_id,
+        Notification.source_type == source_type,
+        Notification.resolved_at.is_(None),
+    ]
+    if active_source_ids:
+        filters.append(Notification.source_id.not_in(active_source_ids))
+    now = datetime.now(timezone.utc)
+    await db.execute(
+        update(Notification)
+        .where(*filters)
+        .values(resolved_at=now, read_at=func.coalesce(Notification.read_at, now))
+    )
+
+
+async def resolve_source_notifications(
+    db: AsyncSession,
+    *,
+    source_type: str,
+    source_id: str,
+    recipient_id: uuid.UUID | None = None,
+) -> None:
+    filters = [
+        Notification.source_type == source_type,
+        Notification.source_id == source_id,
+        Notification.resolved_at.is_(None),
+    ]
+    if recipient_id is not None:
+        filters.append(Notification.recipient_id == recipient_id)
+    now = datetime.now(timezone.utc)
+    await db.execute(
+        update(Notification)
+        .where(*filters)
+        .values(resolved_at=now, read_at=func.coalesce(Notification.read_at, now))
+    )
+
+
+async def resolve_recipient_study_notifications(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    recipient_id: uuid.UUID,
+) -> None:
+    now = datetime.now(timezone.utc)
+    await db.execute(
+        update(Notification)
+        .where(
+            Notification.study_id == study_id,
+            Notification.recipient_id == recipient_id,
+            Notification.resolved_at.is_(None),
+        )
+        .values(resolved_at=now, read_at=func.coalesce(Notification.read_at, now))
+    )
+
+
+async def list_feed(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    recipient_id: uuid.UUID,
+    skip: int = 0,
+    limit: int = 10,
+    category: str | None = None,
+    unread_only: bool = False,
+    requires_action: bool | None = None,
+) -> GeneralNotificationFeed:
+    active_filter = (
+        Notification.study_id == study_id,
+        Notification.recipient_id == recipient_id,
+        Notification.resolved_at.is_(None),
+    )
+    unread_count = int(await db.scalar(
+        select(func.count(Notification.id)).where(*active_filter, Notification.read_at.is_(None))
+    ) or 0)
+
+    filters = [
+        Notification.study_id == study_id,
+        Notification.recipient_id == recipient_id,
+        Notification.resolved_at.is_(None),
+    ]
+    if category:
+        filters.append(Notification.category == category)
+    if unread_only:
+        filters.append(Notification.read_at.is_(None))
+    if requires_action is not None:
+        filters.append(Notification.requires_action.is_(requires_action))
+    total_count = int(await db.scalar(
+        select(func.count(Notification.id)).where(*filters)
+    ) or 0)
+    items = (await db.scalars(
+        select(Notification)
+        .where(*filters)
+        .order_by(
+            Notification.read_at.is_not(None),
+            Notification.created_at.desc(),
+        )
+        .offset(max(0, skip))
+        .limit(max(1, min(limit, 100)))
+    )).all()
+    return GeneralNotificationFeed(
+        unread_count=unread_count,
+        total_count=total_count,
+        items=list(items),
+    )
+
+
+async def mark_read(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    recipient_id: uuid.UUID,
+    notification_id: uuid.UUID,
+) -> Notification:
+    item = await db.scalar(select(Notification).where(
+        Notification.id == notification_id,
+        Notification.study_id == study_id,
+        Notification.recipient_id == recipient_id,
+        Notification.resolved_at.is_(None),
+    ))
+    if item is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="通知不存在")
+    if item.read_at is None:
+        now = datetime.now(timezone.utc)
+        item.read_at = now
+        if not item.requires_action:
+            item.resolved_at = now
+        await db.commit()
+        await db.refresh(item)
+    return item
+
+
+async def mark_all_read(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    recipient_id: uuid.UUID,
+) -> None:
+    now = datetime.now(timezone.utc)
+    await db.execute(
+        update(Notification)
+        .where(
+            Notification.study_id == study_id,
+            Notification.recipient_id == recipient_id,
+            Notification.resolved_at.is_(None),
+            Notification.read_at.is_(None),
+            Notification.requires_action.is_(False),
+        )
+        .values(read_at=now, resolved_at=now)
+    )
+    await db.execute(
+        update(Notification)
+        .where(
+            Notification.study_id == study_id,
+            Notification.recipient_id == recipient_id,
+            Notification.resolved_at.is_(None),
+            Notification.read_at.is_(None),
+            Notification.requires_action.is_(True),
+        )
+        .values(read_at=now)
+    )
+    await db.commit()

backend/app/services/onlyoffice_collaboration_service.py

@@ -0,0 +1,417 @@
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import uuid
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any
+from urllib.parse import urlsplit, urlunsplit
+
+import httpx
+from fastapi import HTTPException, Request, status
+from jose import JWTError, jwt
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.config import settings
+from app.models.collaboration import (
+    CollaborationCallbackReceipt,
+    CollaborationFile,
+    CollaborationRevision,
+    CollaborationSession,
+    CollaborationShareLink,
+)
+from app.models.user import User
+from app.schemas.collaboration import CollaborationCallbackPayload, CollaborationEditorConfigRead
+from app.services import collaboration_service, onlyoffice_service
+
+
+def collaboration_document_key(file_id: uuid.UUID, generation: int) -> str:
+    fingerprint = f"{settings.ONLYOFFICE_INSTANCE_ID or ''}:collaboration:{file_id}:{generation}"
+    return f"ctms-collab-{hashlib.sha256(fingerprint.encode('utf-8')).hexdigest()}"
+
+
+def _content_url(session_id: uuid.UUID) -> str:
+    return (
+        f"{settings.ONLYOFFICE_STORAGE_BASE_URL.rstrip('/')}"
+        f"/internal/onlyoffice/collaboration/sessions/{session_id}/content"
+    )
+
+
+def _callback_url(session_id: uuid.UUID) -> str:
+    return (
+        f"{settings.ONLYOFFICE_STORAGE_BASE_URL.rstrip('/')}"
+        f"/internal/onlyoffice/collaboration/sessions/{session_id}/callback"
+    )
+
+
+async def _active_session(
+    db: AsyncSession, item: CollaborationFile, user_id: uuid.UUID
+) -> CollaborationSession:
+    session = await db.scalar(
+        select(CollaborationSession).where(
+            CollaborationSession.file_id == item.id,
+            CollaborationSession.generation == item.generation,
+        ).order_by(CollaborationSession.created_at.desc())
+    )
+    if session:
+        if session.status != "ACTIVE":
+            session.status = "ACTIVE"
+            session.closed_at = None
+            await db.commit()
+            await db.refresh(session)
+        return session
+    if not item.current_revision_id:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="协作文件尚无可编辑内容")
+    session = CollaborationSession(
+        file_id=item.id,
+        base_revision_id=item.current_revision_id,
+        document_key=collaboration_document_key(item.id, item.generation),
+        generation=item.generation,
+        started_by=user_id,
+    )
+    db.add(session)
+    await db.commit()
+    await db.refresh(session)
+    return session
+
+
+async def build_editor_config(
+    db: AsyncSession, item: CollaborationFile, user
+) -> CollaborationEditorConfigRead:
+    await onlyoffice_service.ensure_onlyoffice_available()
+    revision = await db.get(CollaborationRevision, item.current_revision_id)
+    if not revision or not Path(revision.file_uri).exists():
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作文件内容不存在")
+    can_edit = await collaboration_service.can_edit_file(db, item, user)
+    can_request_edit = await collaboration_service.can_request_edit_file(db, item, user)
+    can_download = await collaboration_service.can_export_file(db, item, user)
+    can_save_as = can_download and await collaboration_service.can_create_file(db, item, user)
+    session = await _active_session(db, item, user.id)
+    now = datetime.now(timezone.utc)
+    expires_at = now + timedelta(seconds=settings.ONLYOFFICE_CONFIG_TTL_SECONDS)
+    config: dict[str, Any] = {
+        "type": "desktop",
+        "documentType": item.file_type,
+        "document": {
+            "fileType": item.extension,
+            "key": session.document_key,
+            "title": item.title,
+            "url": _content_url(session.id),
+            "permissions": {
+                "chat": False,
+                "copy": can_download,
+                "comment": can_edit,
+                "download": can_download,
+                # In view mode ONLYOFFICE displays "Edit current file" only
+                # when edit=true and onRequestEditRights is registered. CTMS
+                # handles that event as an approval request, not an escalation.
+                "edit": can_edit or can_request_edit,
+                "fillForms": False,
+                "modifyContentControl": can_edit,
+                "modifyFilter": can_edit,
+                "print": can_download,
+                "protect": False,
+                "review": False,
+            },
+        },
+        "editorConfig": {
+            "callbackUrl": _callback_url(session.id),
+            "coEditing": {"mode": "fast", "change": False},
+            "customization": {
+                "autosave": True,
+                "chat": False,
+                "comments": can_edit,
+                "forcesave": can_edit,
+                "help": False,
+                "plugins": False,
+            },
+            "lang": "zh-CN",
+            "mode": "edit" if can_edit else "view",
+            "user": {"id": str(user.id), "name": user.full_name},
+        },
+    }
+    config["token"] = jwt.encode(
+        {**config, "iat": int(now.timestamp()), "exp": int(expires_at.timestamp())},
+        settings.ONLYOFFICE_JWT_SECRET or "",
+        algorithm="HS256",
+    )
+    return CollaborationEditorConfigRead(
+        file_id=item.id,
+        file_name=item.title,
+        access_mode="edit" if can_edit else "view",
+        can_save_as=can_save_as,
+        can_download=can_download,
+        can_request_edit=can_request_edit,
+        expires_at=expires_at,
+        config=config,
+    )
+
+
+async def build_shared_editor_config(
+    db: AsyncSession,
+    item: CollaborationFile,
+    link: CollaborationShareLink,
+    *,
+    client_id: str,
+    display_name: str,
+) -> CollaborationEditorConfigRead:
+    await onlyoffice_service.ensure_onlyoffice_available()
+    revision = await db.get(CollaborationRevision, item.current_revision_id)
+    if not revision or not Path(revision.file_uri).exists():
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="共享文件内容不存在")
+    can_edit = link.access_mode == "EDIT"
+    session = await _active_session(db, item, item.owner_id)
+    now = datetime.now(timezone.utc)
+    expires_at = now + timedelta(seconds=settings.ONLYOFFICE_CONFIG_TTL_SECONDS)
+    if link.expires_at and link.expires_at < expires_at:
+        expires_at = link.expires_at
+    external_user_id = f"share-{link.id.hex[:12]}-{client_id[:32]}"
+    config: dict[str, Any] = {
+        "type": "desktop",
+        "documentType": item.file_type,
+        "document": {
+            "fileType": item.extension,
+            "key": session.document_key,
+            "title": item.title,
+            "url": _content_url(session.id),
+            "permissions": {
+                "chat": False,
+                "copy": item.allow_export,
+                "comment": can_edit,
+                "download": item.allow_export,
+                "edit": can_edit,
+                "fillForms": False,
+                "modifyContentControl": can_edit,
+                "modifyFilter": can_edit,
+                "print": item.allow_export,
+                "protect": False,
+                "review": False,
+            },
+        },
+        "editorConfig": {
+            "callbackUrl": _callback_url(session.id),
+            "coEditing": {"mode": "fast", "change": False},
+            "customization": {
+                "autosave": can_edit,
+                "chat": False,
+                "comments": can_edit,
+                "forcesave": can_edit,
+                "help": False,
+                "plugins": False,
+            },
+            "lang": "zh-CN",
+            "mode": "edit" if can_edit else "view",
+            "user": {"id": external_user_id, "name": display_name},
+        },
+    }
+    config["token"] = jwt.encode(
+        {**config, "iat": int(now.timestamp()), "exp": int(expires_at.timestamp())},
+        settings.ONLYOFFICE_JWT_SECRET or "",
+        algorithm="HS256",
+    )
+    return CollaborationEditorConfigRead(
+        file_id=item.id,
+        file_name=item.title,
+        access_mode="edit" if can_edit else "view",
+        can_save_as=False,
+        can_download=item.allow_export,
+        expires_at=expires_at,
+        config=config,
+    )
+
+
+async def get_session_content(
+    db: AsyncSession, session_id: uuid.UUID, authorization: str | None
+) -> tuple[CollaborationRevision, CollaborationFile]:
+    session = await db.get(CollaborationSession, session_id)
+    if not session:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作会话不存在")
+    onlyoffice_service.validate_outbox_token(authorization, _content_url(session_id))
+    revision = await db.get(CollaborationRevision, session.base_revision_id)
+    item = await db.get(CollaborationFile, session.file_id)
+    if not revision or not item or not Path(revision.file_uri).exists():
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作文件内容不存在")
+    return revision, item
+
+
+def validate_callback_token(token: str | None, payload: CollaborationCallbackPayload) -> dict[str, Any]:
+    if not token:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="缺少 ONLYOFFICE 回调签名")
+    value = token.strip()
+    if " " in value:
+        scheme, credential = value.split(" ", 1)
+        if scheme.lower() != "bearer":
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调签名格式无效")
+        value = credential.strip()
+    try:
+        decoded = onlyoffice_service.decode_onlyoffice_token(value)
+    except JWTError as exc:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调签名无效") from exc
+    signed = decoded.get("payload") if isinstance(decoded.get("payload"), dict) else decoded
+    if not isinstance(signed, dict):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调载荷无效")
+    signed_key = signed.get("key")
+    signed_status = signed.get("status")
+    if not isinstance(signed_key, str) or not hmac.compare_digest(signed_key, payload.key):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调 key 不匹配")
+    if not isinstance(signed_status, int) or signed_status != payload.status:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调状态不匹配")
+    if payload.url:
+        signed_url = signed.get("url")
+        if not isinstance(signed_url, str) or not hmac.compare_digest(signed_url, payload.url):
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="ONLYOFFICE 回调文件地址不匹配")
+    return decoded
+
+
+def _callback_fingerprint(payload: CollaborationCallbackPayload) -> str:
+    normalized = payload.model_dump(mode="json", exclude_none=True)
+    return hashlib.sha256(json.dumps(normalized, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+
+
+def _url_origin_matches(actual, expected) -> bool:
+    actual_port = actual.port or (443 if actual.scheme == "https" else 80)
+    expected_port = expected.port or (443 if expected.scheme == "https" else 80)
+    return (
+        actual.scheme == expected.scheme
+        and actual.hostname
+        and actual.hostname.lower() == (expected.hostname or "").lower()
+        and actual_port == expected_port
+    )
+
+
+def _validate_result_url(url: str) -> str:
+    actual = urlsplit(url)
+    if (
+        actual.scheme not in {"http", "https"}
+        or actual.username
+        or actual.password
+        or actual.fragment
+        or not actual.hostname
+    ):
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="ONLYOFFICE 保存地址不受信任")
+
+    internal = urlsplit(settings.ONLYOFFICE_INTERNAL_URL.rstrip("/"))
+    if _url_origin_matches(actual, internal):
+        return urlunsplit((internal.scheme, internal.netloc, actual.path, actual.query, ""))
+
+    public = urlsplit(settings.FRONTEND_PUBLIC_URL.rstrip("/"))
+    proxy_prefix = "/onlyoffice/"
+    if not _url_origin_matches(actual, public) or not actual.path.startswith(proxy_prefix):
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="ONLYOFFICE 保存地址不受信任")
+    internal_path = f"{internal.path.rstrip('/')}/{actual.path[len(proxy_prefix):]}"
+    return urlunsplit((internal.scheme, internal.netloc, internal_path, actual.query, ""))
+
+
+async def _download_result(url: str) -> bytes:
+    download_url = _validate_result_url(url)
+    try:
+        async with httpx.AsyncClient(timeout=30.0, follow_redirects=False) as client:
+            async with client.stream("GET", download_url) as response:
+                if response.status_code != status.HTTP_200_OK:
+                    raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="ONLYOFFICE 保存文件下载失败")
+                content = bytearray()
+                async for chunk in response.aiter_bytes():
+                    content.extend(chunk)
+                    if len(content) > settings.COLLABORATION_MAX_FILE_BYTES:
+                        raise HTTPException(status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail="ONLYOFFICE 保存文件超出限制")
+    except httpx.HTTPError as exc:
+        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="ONLYOFFICE 保存文件下载失败") from exc
+    return bytes(content)
+
+
+async def _callback_user(
+    db: AsyncSession, payload: CollaborationCallbackPayload, session: CollaborationSession
+) -> User | None:
+    has_public_share_user = False
+    for value in payload.users:
+        if value.startswith("share-"):
+            has_public_share_user = True
+            continue
+        try:
+            user = await db.get(User, uuid.UUID(value))
+        except (ValueError, TypeError):
+            user = None
+        if user:
+            return user
+    if has_public_share_user:
+        return None
+    user = await db.get(User, session.started_by)
+    if not user:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="协作会话用户不存在")
+    return user
+
+
+async def process_callback(
+    db: AsyncSession,
+    session_id: uuid.UUID,
+    payload: CollaborationCallbackPayload,
+) -> dict[str, int]:
+    session = await db.scalar(
+        select(CollaborationSession).where(CollaborationSession.id == session_id).with_for_update()
+    )
+    if not session:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作会话不存在")
+    if not hmac.compare_digest(session.document_key, payload.key):
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="协作会话 key 不匹配")
+    fingerprint = _callback_fingerprint(payload)
+    duplicate = await db.scalar(select(CollaborationCallbackReceipt.id).where(
+        CollaborationCallbackReceipt.session_id == session.id,
+        CollaborationCallbackReceipt.fingerprint == fingerprint,
+    ))
+    if duplicate:
+        return {"error": 0}
+
+    item = await db.get(CollaborationFile, session.file_id)
+    if not item:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="协作文件不存在")
+    session.last_callback_at = datetime.now(timezone.utc)
+    session.active_users = json.dumps(payload.users, ensure_ascii=True)
+    result = "ACKNOWLEDGED"
+    saved_revision_id = None
+
+    if payload.status in {2, 6}:
+        if not payload.url:
+            raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="ONLYOFFICE 保存回调缺少文件地址")
+        if session.generation != item.generation:
+            result = "STALE"
+        else:
+            content = await _download_result(payload.url)
+            actor = await _callback_user(db, payload, session)
+            source = "SESSION_CLOSE" if payload.status == 2 else "FORCE_SAVE"
+            if actor is None:
+                source = "SHARE_SESSION_CLOSE" if payload.status == 2 else "SHARE_FORCE_SAVE"
+            revision, created = await collaboration_service.append_revision(
+                db, item, content, source=source, created_by=actor.id if actor else None
+            )
+            saved_revision_id = revision.id
+            result = "SAVED" if created else "UNCHANGED"
+            if payload.status == 2:
+                item.generation += 1
+                session.status = "CLOSED"
+                session.closed_at = datetime.now(timezone.utc)
+            else:
+                # 强制保存不结束当前共同编辑会话；同步基线可保证 Document
+                # Server 缓存重建时仍从最近一次持久化内容恢复。
+                session.base_revision_id = revision.id
+    elif payload.status == 4:
+        session.status = "CLOSED"
+        session.closed_at = datetime.now(timezone.utc)
+        result = "UNCHANGED"
+    elif payload.status in {3, 7}:
+        session.status = "ERROR"
+        result = "ERROR"
+
+    db.add(CollaborationCallbackReceipt(
+        session_id=session.id,
+        fingerprint=fingerprint,
+        callback_status=payload.status,
+        result=result,
+        revision_id=saved_revision_id,
+    ))
+    await db.commit()
+    # ONLYOFFICE 要求回调处理器在接收并记录状态后固定确认成功。
+    # status 3/7 表示文档服务自身保存失败，不应通过 error=1 制造重试环。
+    return {"error": 0}

backend/app/services/onlyoffice_service.py

@@ -0,0 +1,235 @@
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import hmac
+import time
+import uuid
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, Literal
+
+import httpx
+from fastapi import status
+from jose import JWTError, jwt
+
+from app.core.config import settings
+from app.core.exceptions import AppException
+from app.schemas.onlyoffice import OnlyOfficePreviewConfigRead
+
+OnlyOfficeDocumentType = Literal["word", "cell", "slide"]
+OnlyOfficeResourceType = Literal["attachment", "version", "collaboration_revision"]
+
+WORD_FORMATS = frozenset({
+    "doc", "docx", "docm", "dot", "dotx", "dotm", "odt", "ott", "rtf", "txt", "wps", "wpt",
+})
+CELL_FORMATS = frozenset({
+    "xls", "xlsx", "xlsm", "xlsb", "xlt", "xltx", "xltm", "ods", "ots", "csv", "et", "ett",
+})
+SLIDE_FORMATS = frozenset({
+    "ppt", "pptx", "pptm", "pps", "ppsx", "ppsm", "pot", "potx", "potm", "odp", "otp", "dps", "dpt",
+})
+
+_health_lock = asyncio.Lock()
+_health_checked_at = 0.0
+_health_available = False
+_HEALTH_CACHE_SECONDS = 15.0
+
+
+def onlyoffice_error(code: str, message: str, status_code: int) -> AppException:
+    return AppException(code=code, message=message, status_code=status_code)
+
+
+def office_format_for_filename(filename: str) -> tuple[str, OnlyOfficeDocumentType] | None:
+    suffix = Path(filename).suffix.lower().lstrip(".")
+    if suffix in WORD_FORMATS:
+        return suffix, "word"
+    if suffix in CELL_FORMATS:
+        return suffix, "cell"
+    if suffix in SLIDE_FORMATS:
+        return suffix, "slide"
+    return None
+
+
+def onlyoffice_content_url(resource_type: OnlyOfficeResourceType, resource_id: uuid.UUID) -> str:
+    plural = {
+        "attachment": "attachments",
+        "version": "versions",
+        "collaboration_revision": "collaboration-revisions",
+    }[resource_type]
+    return (
+        f"{settings.ONLYOFFICE_STORAGE_BASE_URL.rstrip('/')}"
+        f"/internal/onlyoffice/{plural}/{resource_id}/content"
+    )
+
+
+def onlyoffice_document_key(
+    resource_type: OnlyOfficeResourceType,
+    resource_id: uuid.UUID,
+    *,
+    file_hash: str | None = None,
+) -> str:
+    instance_id = (settings.ONLYOFFICE_INSTANCE_ID or "").strip()
+    fingerprint = f"{instance_id}{resource_type}{resource_id}"
+    if resource_type == "version":
+        fingerprint = f"{fingerprint}{file_hash or ''}"
+    return f"ctms-{hashlib.sha256(fingerprint.encode('utf-8')).hexdigest()}"
+
+
+async def ensure_onlyoffice_available() -> None:
+    global _health_available, _health_checked_at
+    if not settings.ONLYOFFICE_ENABLED:
+        raise onlyoffice_error(
+            "ONLYOFFICE_DISABLED",
+            "Office 预览服务尚未启用",
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+        )
+
+    now = time.monotonic()
+    if now - _health_checked_at < _HEALTH_CACHE_SECONDS:
+        if _health_available:
+            return
+        raise onlyoffice_error(
+            "ONLYOFFICE_UNAVAILABLE",
+            "Office 预览服务暂不可用，请稍后重试",
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+        )
+
+    async with _health_lock:
+        now = time.monotonic()
+        if now - _health_checked_at >= _HEALTH_CACHE_SECONDS:
+            available = False
+            try:
+                async with httpx.AsyncClient(timeout=2.0, follow_redirects=False) as client:
+                    response = await client.get(f"{settings.ONLYOFFICE_INTERNAL_URL.rstrip('/')}/healthcheck")
+                available = response.status_code == status.HTTP_200_OK
+            except httpx.HTTPError:
+                available = False
+            _health_available = available
+            _health_checked_at = now
+
+    if not _health_available:
+        raise onlyoffice_error(
+            "ONLYOFFICE_UNAVAILABLE",
+            "Office 预览服务暂不可用，请稍后重试",
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+        )
+
+
+def build_preview_config(
+    *,
+    resource_type: OnlyOfficeResourceType,
+    resource_id: uuid.UUID,
+    file_name: str,
+    user_id: uuid.UUID,
+    user_name: str,
+    file_hash: str | None = None,
+) -> OnlyOfficePreviewConfigRead:
+    format_info = office_format_for_filename(file_name)
+    if not format_info:
+        raise onlyoffice_error(
+            "ONLYOFFICE_FORMAT_UNSUPPORTED",
+            "该文件格式不支持 Office 在线预览",
+            status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
+        )
+    file_type, document_type = format_info
+    now = datetime.now(timezone.utc)
+    expires_at = now + timedelta(seconds=settings.ONLYOFFICE_CONFIG_TTL_SECONDS)
+    config: dict[str, Any] = {
+        "type": "desktop",
+        "documentType": document_type,
+        "document": {
+            "fileType": file_type,
+            "key": onlyoffice_document_key(resource_type, resource_id, file_hash=file_hash),
+            "title": file_name,
+            "url": onlyoffice_content_url(resource_type, resource_id),
+            "permissions": {
+                "copy": False,
+                "comment": False,
+                "download": False,
+                "edit": False,
+                "fillForms": False,
+                "modifyContentControl": False,
+                "modifyFilter": False,
+                "print": False,
+                "review": False,
+            },
+        },
+        "editorConfig": {
+            "coEditing": {"mode": "strict", "change": False},
+            "customization": {
+                "chat": False,
+                "comments": False,
+                "forcesave": False,
+            },
+            "lang": "zh-CN",
+            "mode": "view",
+            "user": {"id": str(user_id), "name": user_name},
+        },
+    }
+    token_payload = {
+        **config,
+        "iat": int(now.timestamp()),
+        "exp": int(expires_at.timestamp()),
+    }
+    config["token"] = jwt.encode(
+        token_payload,
+        settings.ONLYOFFICE_JWT_SECRET or "",
+        algorithm="HS256",
+    )
+    return OnlyOfficePreviewConfigRead(
+        resource_type=resource_type,
+        resource_id=resource_id,
+        file_name=file_name,
+        expires_at=expires_at,
+        config=config,
+    )
+
+
+def validate_outbox_token(token: str | None, expected_url: str) -> dict[str, Any]:
+    if not token:
+        raise onlyoffice_error(
+            "ONLYOFFICE_SOURCE_UNAUTHORIZED",
+            "无法验证 Office 文件请求",
+            status.HTTP_401_UNAUTHORIZED,
+        )
+    token = token.strip()
+    if " " in token:
+        scheme, credential = token.split(" ", 1)
+        if scheme.lower() != "bearer" or not credential.strip():
+            raise onlyoffice_error(
+                "ONLYOFFICE_SOURCE_UNAUTHORIZED",
+                "无法验证 Office 文件请求",
+                status.HTTP_401_UNAUTHORIZED,
+            )
+        token = credential.strip()
+    try:
+        payload = decode_onlyoffice_token(token)
+    except JWTError as exc:
+        raise onlyoffice_error(
+            "ONLYOFFICE_SOURCE_UNAUTHORIZED",
+            "无法验证 Office 文件请求",
+            status.HTTP_401_UNAUTHORIZED,
+        ) from exc
+
+    request_payload = payload.get("payload")
+    token_url = request_payload.get("url") if isinstance(request_payload, dict) else None
+    if not isinstance(token_url, str) or not hmac.compare_digest(token_url, expected_url):
+        raise onlyoffice_error(
+            "ONLYOFFICE_SOURCE_URL_MISMATCH",
+            "Office 文件请求地址不匹配",
+            status.HTTP_401_UNAUTHORIZED,
+        )
+    return payload
+
+
+def decode_onlyoffice_token(token: str) -> dict[str, Any]:
+    header = jwt.get_unverified_header(token)
+    if header.get("alg") != "HS256":
+        raise JWTError("unexpected algorithm")
+    return jwt.decode(
+        token,
+        settings.ONLYOFFICE_JWT_SECRET or "",
+        algorithms=["HS256"],
+        options={"verify_aud": False},
+    )

backend/app/services/permission_log_writer.py

@@ -13,6 +13,7 @@ import asyncio
 import logging
 import uuid
 from datetime import datetime, timezone
+from typing import Any
 
 from app.db.session import SessionLocal
 from app.models.permission_access_log import PermissionAccessLog
@@ -22,85 +23,150 @@ logger = logging.getLogger("ctms.permission_log_writer")
 BATCH_SIZE = 50
 FLUSH_INTERVAL = 5.0
 QUEUE_MAX_SIZE = 10000
+WRITE_ATTEMPTS = 2
+_QUEUE_STOP = object()
 
 
 class PermissionLogWriter:
 
     def __init__(self):
-        self._queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=QUEUE_MAX_SIZE)
+        self._queue: asyncio.Queue[dict | object] = asyncio.Queue(maxsize=QUEUE_MAX_SIZE)
         self._task: asyncio.Task | None = None
+        self._stopping = False
+        self._started_at: datetime | None = None
+        self._last_success_at: datetime | None = None
+        self._last_error_at: datetime | None = None
+        self._last_error_type: str | None = None
+        self._accepted_entries = 0
+        self._written_entries = 0
+        self._dropped_entries = 0
+        self._failed_batches = 0
+        self._failed_entries = 0
+        self._retry_count = 0
 
     def enqueue(self, entry: dict) -> None:
+        if self._stopping:
+            self._dropped_entries += 1
+            logger.warning("Permission log writer is stopping, dropping entry")
+            return
         try:
             self._queue.put_nowait(entry)
+            self._accepted_entries += 1
         except asyncio.QueueFull:
+            self._dropped_entries += 1
             logger.warning("Permission log queue full, dropping entry")
 
     async def start(self) -> None:
+        if self._task and not self._task.done():
+            return
+        self._stopping = False
+        self._started_at = datetime.now(timezone.utc)
         self._task = asyncio.create_task(self._flush_loop())
         logger.info("PermissionLogWriter started")
 
     async def stop(self) -> None:
-        if self._task:
-            self._task.cancel()
-            try:
-                await self._task
-            except asyncio.CancelledError:
-                pass
-        await self._drain()
+        if self._task and not self._task.done():
+            self._stopping = True
+            await self._queue.put(_QUEUE_STOP)
+            await self._task
+        self._task = None
         logger.info("PermissionLogWriter stopped")
 
     async def _flush_loop(self) -> None:
         while True:
-            batch = await self._collect_batch()
+            batch, should_stop = await self._collect_batch()
             if batch:
                 await self._write_batch(batch)
+            if should_stop:
+                break
 
-    async def _collect_batch(self) -> list[dict]:
+    async def _collect_batch(self) -> tuple[list[dict], bool]:
         batch: list[dict] = []
         try:
             first = await asyncio.wait_for(self._queue.get(), timeout=FLUSH_INTERVAL)
+            if first is _QUEUE_STOP:
+                return batch, True
             batch.append(first)
         except asyncio.TimeoutError:
-            return batch
+            return batch, False
 
         while len(batch) < BATCH_SIZE:
             try:
                 item = self._queue.get_nowait()
+                if item is _QUEUE_STOP:
+                    return batch, True
                 batch.append(item)
             except asyncio.QueueEmpty:
                 break
-        return batch
+        return batch, False
 
-    async def _write_batch(self, batch: list[dict]) -> None:
-        try:
-            async with SessionLocal() as session:
-                for entry in batch:
-                    log = PermissionAccessLog(
-                        id=uuid.uuid4(),
-                        study_id=entry["study_id"],
-                        user_id=entry["user_id"],
-                        endpoint_key=entry["endpoint_key"],
-                        role=entry["role"],
-                        allowed=entry["allowed"],
-                        elapsed_ms=entry["elapsed_ms"],
-                        ip_address=entry.get("ip_address"),
-                        created_at=entry.get("created_at", datetime.now(timezone.utc)),
-                    )
-                    session.add(log)
-                await session.commit()
-        except Exception:
-            logger.exception("Failed to write permission access log batch (%d entries)", len(batch))
-
-    async def _drain(self) -> None:
-        batch: list[dict] = []
-        while not self._queue.empty():
+    async def _write_batch(self, batch: list[dict]) -> bool:
+        for attempt in range(WRITE_ATTEMPTS):
             try:
-                batch.append(self._queue.get_nowait())
-            except asyncio.QueueEmpty:
-                break
-        if batch:
-            await self._write_batch(batch)
+                async with SessionLocal() as session:
+                    for entry in batch:
+                        log = PermissionAccessLog(
+                            id=uuid.uuid4(),
+                            study_id=entry["study_id"],
+                            user_id=entry["user_id"],
+                            endpoint_key=entry["endpoint_key"],
+                            role=entry["role"],
+                            allowed=entry["allowed"],
+                            elapsed_ms=entry["elapsed_ms"],
+                            ip_address=entry.get("ip_address"),
+                            user_agent=entry.get("user_agent"),
+                            client_type=entry.get("client_type"),
+                            client_version=entry.get("client_version"),
+                            client_platform=entry.get("client_platform"),
+                            build_channel=entry.get("build_channel"),
+                            build_commit=entry.get("build_commit"),
+                            request_headers=entry.get("request_headers"),
+                            request_snapshot=entry.get("request_snapshot"),
+                            request_id=entry.get("request_id"),
+                            created_at=entry.get("created_at", datetime.now(timezone.utc)),
+                        )
+                        session.add(log)
+                    await session.commit()
+            except Exception as exc:
+                self._last_error_at = datetime.now(timezone.utc)
+                self._last_error_type = type(exc).__name__
+                if attempt + 1 < WRITE_ATTEMPTS:
+                    self._retry_count += 1
+                    logger.warning(
+                        "Permission access log batch write failed; retrying (%d entries)",
+                        len(batch),
+                    )
+                    await asyncio.sleep(0)
+                    continue
+                self._failed_batches += 1
+                self._failed_entries += len(batch)
+                logger.exception(
+                    "Failed to write permission access log batch after retries (%d entries)",
+                    len(batch),
+                )
+                return False
+            self._written_entries += len(batch)
+            self._last_success_at = datetime.now(timezone.utc)
+            return True
+        return False
+
+    def stats(self) -> dict[str, Any]:
+        return {
+            "running": bool(self._task and not self._task.done()),
+            "stopping": self._stopping,
+            "queue_size": self._queue.qsize(),
+            "queue_capacity": self._queue.maxsize,
+            "accepted_entries": self._accepted_entries,
+            "written_entries": self._written_entries,
+            "dropped_entries": self._dropped_entries,
+            "failed_batches": self._failed_batches,
+            "failed_entries": self._failed_entries,
+            "retry_count": self._retry_count,
+            "started_at": self._started_at.isoformat() if self._started_at else None,
+            "last_success_at": self._last_success_at.isoformat() if self._last_success_at else None,
+            "last_error_at": self._last_error_at.isoformat() if self._last_error_at else None,
+            "last_error_type": self._last_error_type,
+        }
 
 
 _writer: PermissionLogWriter | None = None

backend/app/services/project_reminder_service.py

@@ -0,0 +1,541 @@
+from __future__ import annotations
+
+import uuid
+from datetime import date, datetime, time, timedelta, timezone
+from zoneinfo import ZoneInfo
+
+from sqlalchemy import and_, func, or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.deps import get_cra_site_scope, is_system_admin
+from app.core.project_permissions import role_has_api_permission
+from app.crud import member as member_crud
+from app.crud import site as site_crud
+from app.models.acknowledgement import Acknowledgement, AcknowledgementType
+from app.models.ae import AdverseEvent
+from app.models.collaboration import CollaborationEditRequest, CollaborationFile, CollaborationMember
+from app.models.distribution import Distribution, DistributionStatus, DistributionTargetType
+from app.models.document import Document, DocumentStatus
+from app.models.document_version import DocumentVersion
+from app.models.milestone import Milestone
+from app.models.monitoring_visit_issue import MonitoringVisitIssue
+from app.models.site import Site
+from app.models.subject import Subject
+from app.models.user import User
+from app.models.visit import Visit
+from app.services import notification_service
+
+BUSINESS_TIMEZONE = ZoneInfo("Asia/Shanghai")
+RISK_DUE_SOON_DAYS = 3
+MILESTONE_DUE_SOON_DAYS = 7
+VISIT_WINDOW_DUE_SOON_DAYS = 3
+
+
+def _date_due_at(value: date | None) -> datetime | None:
+    if value is None:
+        return None
+    return datetime.combine(value, time.max, tzinfo=BUSINESS_TIMEZONE).astimezone(timezone.utc)
+
+
+def _as_utc(value: datetime | None) -> datetime | None:
+    if value is None:
+        return None
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+async def _count_and_earliest(db: AsyncSession, model, *filters):
+    return (await db.execute(
+        select(func.count(model.id), func.min(model.report_due_date if model is AdverseEvent else model.due_at))
+        .where(*filters)
+    )).one()
+
+
+async def _sync_risk_reminders(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    user,
+    role: str,
+    current_time: datetime,
+) -> None:
+    can_read_aes = is_system_admin(user) or await role_has_api_permission(
+        db, study_id, role, "subject_aes:read"
+    )
+    can_read_monitoring = is_system_admin(user) or await role_has_api_permission(
+        db, study_id, role, "monitoring_issues:read"
+    )
+    cra_scope = await get_cra_site_scope(db, study_id, user)
+    site_ids = cra_scope[0] if cra_scope else None
+    today = current_time.astimezone(BUSINESS_TIMEZONE).date()
+    due_soon_date = today + timedelta(days=RISK_DUE_SOON_DAYS)
+
+    ae_scope = [AdverseEvent.study_id == study_id, AdverseEvent.status != "CLOSED"]
+    if site_ids is not None:
+        ae_scope.append(AdverseEvent.site_id.in_(site_ids))
+    overdue_aes = 0
+    overdue_ae_due = None
+    due_soon_aes = 0
+    due_soon_ae_due = None
+    if can_read_aes:
+        overdue_aes, overdue_ae_due = await _count_and_earliest(
+            db,
+            AdverseEvent,
+            *ae_scope,
+            AdverseEvent.report_due_date.is_not(None),
+            AdverseEvent.report_due_date < today,
+        )
+        due_soon_aes, due_soon_ae_due = await _count_and_earliest(
+            db,
+            AdverseEvent,
+            *ae_scope,
+            AdverseEvent.report_due_date >= today,
+            AdverseEvent.report_due_date <= due_soon_date,
+        )
+    await notification_service.sync_aggregate_notification(
+        db,
+        study_id=study_id,
+        recipient_id=user.id,
+        category="RISK_OVERDUE_AE",
+        priority="URGENT",
+        title="逾期 AE 待处理",
+        message=f"当前有 {overdue_aes} 条逾期 AE 需要跟进",
+        action_path="/risk-issues/sae",
+        source_type="RISK_OVERDUE_AE",
+        source_id=str(study_id),
+        count=int(overdue_aes) if can_read_aes else 0,
+        due_at=_date_due_at(overdue_ae_due),
+    )
+    await notification_service.sync_aggregate_notification(
+        db,
+        study_id=study_id,
+        recipient_id=user.id,
+        category="RISK_AE_DUE_SOON",
+        priority="HIGH",
+        title="AE 上报时限临近",
+        message=f"未来 {RISK_DUE_SOON_DAYS} 天内有 {due_soon_aes} 条 AE 到达上报时限",
+        action_path="/risk-issues/sae",
+        source_type="RISK_AE_DUE_SOON",
+        source_id=str(study_id),
+        count=int(due_soon_aes) if can_read_aes else 0,
+        due_at=_date_due_at(due_soon_ae_due),
+    )
+
+    monitoring_scope = [
+        MonitoringVisitIssue.study_id == study_id,
+        MonitoringVisitIssue.status == "OPEN",
+        MonitoringVisitIssue.due_at.is_not(None),
+    ]
+    if site_ids is not None:
+        monitoring_scope.append(MonitoringVisitIssue.site_id.in_(site_ids))
+    overdue_monitoring = 0
+    overdue_monitoring_due = None
+    due_soon_monitoring = 0
+    due_soon_monitoring_due = None
+    if can_read_monitoring:
+        overdue_monitoring, overdue_monitoring_due = await _count_and_earliest(
+            db,
+            MonitoringVisitIssue,
+            *monitoring_scope,
+            MonitoringVisitIssue.due_at < current_time,
+        )
+        due_soon_monitoring, due_soon_monitoring_due = await _count_and_earliest(
+            db,
+            MonitoringVisitIssue,
+            *monitoring_scope,
+            MonitoringVisitIssue.due_at >= current_time,
+            MonitoringVisitIssue.due_at <= current_time + timedelta(days=RISK_DUE_SOON_DAYS),
+        )
+    await notification_service.sync_aggregate_notification(
+        db,
+        study_id=study_id,
+        recipient_id=user.id,
+        category="RISK_OVERDUE_MONITORING",
+        priority="HIGH",
+        title="监查问题已逾期",
+        message=f"当前有 {overdue_monitoring} 条监查问题已超过计划解决日期",
+        action_path="/risk-issues/monitoring-visits",
+        source_type="RISK_OVERDUE_MONITORING",
+        source_id=str(study_id),
+        count=int(overdue_monitoring) if can_read_monitoring else 0,
+        due_at=_as_utc(overdue_monitoring_due),
+    )
+    await notification_service.sync_aggregate_notification(
+        db,
+        study_id=study_id,
+        recipient_id=user.id,
+        category="RISK_MONITORING_DUE_SOON",
+        priority="HIGH",
+        title="监查问题整改时限临近",
+        message=f"未来 {RISK_DUE_SOON_DAYS} 天内有 {due_soon_monitoring} 条监查问题到期",
+        action_path="/risk-issues/monitoring-visits",
+        source_type="RISK_MONITORING_DUE_SOON",
+        source_id=str(study_id),
+        count=int(due_soon_monitoring) if can_read_monitoring else 0,
+        due_at=_as_utc(due_soon_monitoring_due),
+    )
+
+
+async def _sync_document_distribution_reminders(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    user,
+    role: str,
+    current_time: datetime,
+) -> None:
+    can_read_documents = is_system_admin(user) or await role_has_api_permission(
+        db, study_id, role, "documents:read"
+    )
+    if not can_read_documents:
+        await notification_service.resolve_missing_recipient_sources(
+            db,
+            study_id=study_id,
+            recipient_id=user.id,
+            source_type="DOCUMENT_DISTRIBUTION",
+            active_source_ids=set(),
+        )
+        return
+    site_ids = await site_crud.list_ids_by_contact_user(db, study_id, user.id)
+    target_filters = [
+        and_(
+            Distribution.target_type == DistributionTargetType.USER,
+            Distribution.target_id == str(user.id),
+        ),
+        and_(
+            Distribution.target_type == DistributionTargetType.ROLE,
+            Distribution.target_id == role,
+        ),
+    ]
+    if site_ids:
+        target_filters.append(and_(
+            Distribution.target_type == DistributionTargetType.SITE,
+            Distribution.target_id.in_({str(site_id) for site_id in site_ids}),
+        ))
+    rows = (await db.execute(
+        select(Distribution, Document, DocumentVersion)
+        .join(Document, Distribution.document_id == Document.id)
+        .join(DocumentVersion, Distribution.version_id == DocumentVersion.id)
+        .outerjoin(
+            Acknowledgement,
+            and_(
+                Acknowledgement.distribution_id == Distribution.id,
+                Acknowledgement.user_id == user.id,
+                Acknowledgement.ack_type == AcknowledgementType.RECEIVED,
+            ),
+        )
+        .where(
+            Document.trial_id == study_id,
+            Document.status == DocumentStatus.ACTIVE,
+            Distribution.status == DistributionStatus.ACTIVE,
+            Acknowledgement.id.is_(None),
+            or_(*target_filters),
+        )
+    )).all()
+
+    active_source_ids: set[str] = set()
+    for distribution, document, version in rows:
+        source_id = str(distribution.id)
+        active_source_ids.add(source_id)
+        distribution_due_at = _as_utc(distribution.due_at)
+        if distribution_due_at and distribution_due_at < current_time:
+            stage = "OVERDUE"
+            category = "DOCUMENT_ACK_OVERDUE"
+            priority = "URGENT"
+            title = "文件回执已逾期"
+        elif distribution_due_at and distribution_due_at <= current_time + timedelta(days=RISK_DUE_SOON_DAYS):
+            stage = "DUE_SOON"
+            category = "DOCUMENT_ACK_DUE_SOON"
+            priority = "HIGH"
+            title = "文件回执即将到期"
+        else:
+            stage = "PENDING"
+            category = "DOCUMENT_DISTRIBUTION"
+            priority = "NORMAL"
+            title = "有新的文件版本待接收"
+        due_version = distribution_due_at.isoformat() if distribution_due_at else "none"
+        await notification_service.sync_state_notification(
+            db,
+            study_id=study_id,
+            recipient_id=user.id,
+            category=category,
+            priority=priority,
+            title=title,
+            message=f"“{document.title}” {version.version_no} 已分发，请完成接收回执",
+            action_path=f"/documents/{document.id}",
+            source_type="DOCUMENT_DISTRIBUTION",
+            source_id=source_id,
+            source_version=f"{stage}:{due_version}",
+            active=True,
+            due_at=distribution_due_at,
+        )
+    await notification_service.resolve_missing_recipient_sources(
+        db,
+        study_id=study_id,
+        recipient_id=user.id,
+        source_type="DOCUMENT_DISTRIBUTION",
+        active_source_ids=active_source_ids,
+    )
+
+
+async def _sync_milestone_reminders(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    user,
+    role: str,
+    current_time: datetime,
+) -> None:
+    can_read = is_system_admin(user) or await role_has_api_permission(
+        db, study_id, role, "project_milestones:read"
+    )
+    active_source_ids: set[str] = set()
+    if can_read:
+        today = current_time.astimezone(BUSINESS_TIMEZONE).date()
+        due_soon_date = today + timedelta(days=MILESTONE_DUE_SOON_DAYS)
+        milestones = (await db.scalars(
+            select(Milestone).where(
+                Milestone.study_id == study_id,
+                Milestone.owner_id == user.id,
+                Milestone.status != "DONE",
+            )
+        )).all()
+        for milestone in milestones:
+            due_date = milestone.adjusted_end_date or milestone.planned_date
+            if due_date is None or due_date > due_soon_date:
+                continue
+            source_id = str(milestone.id)
+            active_source_ids.add(source_id)
+            overdue = due_date < today
+            await notification_service.sync_state_notification(
+                db,
+                study_id=study_id,
+                recipient_id=user.id,
+                category="MILESTONE_OVERDUE" if overdue else "MILESTONE_DUE_SOON",
+                priority="HIGH" if overdue else "NORMAL",
+                title="项目里程碑已逾期" if overdue else "项目里程碑即将到期",
+                message=f"“{milestone.name}”计划日期为 {due_date.isoformat()}",
+                action_path="/project/milestones",
+                source_type="PROJECT_MILESTONE",
+                source_id=source_id,
+                source_version=f"{'OVERDUE' if overdue else 'DUE_SOON'}:{due_date.isoformat()}:{milestone.status}",
+                active=True,
+                due_at=_date_due_at(due_date),
+            )
+    await notification_service.resolve_missing_recipient_sources(
+        db,
+        study_id=study_id,
+        recipient_id=user.id,
+        source_type="PROJECT_MILESTONE",
+        active_source_ids=active_source_ids,
+    )
+
+
+async def _sync_visit_window_reminders(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    user,
+    role: str,
+    current_time: datetime,
+) -> None:
+    """Project active visit windows to the members who can actually maintain them."""
+    can_manage_visits = is_system_admin(user) or await role_has_api_permission(
+        db, study_id, role, "visits:update"
+    )
+    active_source_ids: set[str] = set()
+    if can_manage_visits:
+        cra_scope = await get_cra_site_scope(db, study_id, user)
+        site_ids = cra_scope[0] if cra_scope else None
+        filters = [
+            Visit.study_id == study_id,
+            Visit.actual_date.is_(None),
+            Visit.status.not_in(["DONE", "CANCELLED"]),
+            Subject.status.not_in(["COMPLETED", "DROPPED"]),
+            Site.is_active.is_(True),
+        ]
+        if site_ids is not None:
+            filters.append(Subject.site_id.in_(site_ids))
+        rows = (await db.execute(
+            select(Visit, Subject)
+            .join(Subject, Subject.id == Visit.subject_id)
+            .join(Site, Site.id == Subject.site_id)
+            .where(*filters)
+        )).all()
+
+        today = current_time.astimezone(BUSINESS_TIMEZONE).date()
+        due_soon_date = today + timedelta(days=VISIT_WINDOW_DUE_SOON_DAYS)
+        for visit, subject in rows:
+            window_start = visit.window_start or visit.planned_date
+            window_end = visit.window_end or visit.planned_date
+            if window_start is None or window_end is None or window_start > window_end:
+                continue
+
+            if window_end < today:
+                category = "VISIT_WINDOW_MISSED"
+                priority = "HIGH"
+                title = "受试者访视已错过窗口"
+                message = "一个受试者访视已错过计划窗口，请进入详情核对并处理"
+                stage = "MISSED"
+            elif window_start <= due_soon_date:
+                category = "VISIT_WINDOW_DUE_SOON"
+                priority = "NORMAL"
+                title = "受试者访视窗口临近"
+                message = "一个受试者访视已进入近期窗口，请及时核对访视安排"
+                stage = "DUE_SOON"
+            else:
+                continue
+
+            source_id = str(visit.id)
+            active_source_ids.add(source_id)
+            await notification_service.sync_state_notification(
+                db,
+                study_id=study_id,
+                recipient_id=user.id,
+                category=category,
+                priority=priority,
+                title=title,
+                message=message,
+                action_path=f"/subjects/{subject.id}",
+                source_type="SUBJECT_VISIT_WINDOW",
+                source_id=source_id,
+                source_version=f"{stage}:{window_start.isoformat()}:{window_end.isoformat()}",
+                active=True,
+                due_at=_date_due_at(window_end),
+            )
+    await notification_service.resolve_missing_recipient_sources(
+        db,
+        study_id=study_id,
+        recipient_id=user.id,
+        source_type="SUBJECT_VISIT_WINDOW",
+        active_source_ids=active_source_ids,
+    )
+
+
+async def _sync_collaboration_edit_request_reminders(
+    db: AsyncSession,
+    *,
+    study_id: uuid.UUID,
+    user,
+    role: str,
+) -> None:
+    can_read = is_system_admin(user) or await role_has_api_permission(
+        db, study_id, role, "collaboration:read"
+    )
+    if not can_read:
+        await notification_service.resolve_missing_recipient_sources(
+            db,
+            study_id=study_id,
+            recipient_id=user.id,
+            source_type="COLLABORATION_EDIT_REQUEST",
+            active_source_ids=set(),
+        )
+        return
+    manages_file = or_(
+        CollaborationFile.owner_id == user.id,
+        select(CollaborationMember.id).where(
+            CollaborationMember.file_id == CollaborationFile.id,
+            CollaborationMember.user_id == user.id,
+            CollaborationMember.role == "MANAGER",
+        ).exists(),
+    )
+    rows = (await db.execute(
+        select(CollaborationEditRequest, CollaborationFile, User)
+        .join(CollaborationFile, CollaborationFile.id == CollaborationEditRequest.file_id)
+        .join(User, User.id == CollaborationEditRequest.requester_id)
+        .where(
+            CollaborationFile.study_id == study_id,
+            CollaborationFile.status == "ACTIVE",
+            CollaborationFile.deleted_at.is_(None),
+            CollaborationEditRequest.status == "PENDING",
+            manages_file,
+        )
+    )).all()
+    active_source_ids: set[str] = set()
+    for request, item, requester in rows:
+        source_id = str(request.id)
+        active_source_ids.add(source_id)
+        requester_name = str(requester.full_name or requester.email or "项目成员")
+        await notification_service.sync_state_notification(
+            db,
+            study_id=study_id,
+            recipient_id=user.id,
+            category="COLLABORATION_EDIT_REQUEST",
+            priority="NORMAL",
+            title="新的编辑权限申请",
+            message=f"{requester_name} 申请编辑“{item.title}”",
+            action_path=f"/knowledge/collaboration?editRequestFile={item.id}",
+            source_type="COLLABORATION_EDIT_REQUEST",
+            source_id=source_id,
+            source_version="PENDING",
+            active=True,
+            dedupe_key=f"collaboration-edit-request:{request.id}",
+        )
+    await notification_service.resolve_missing_recipient_sources(
+        db,
+        study_id=study_id,
+        recipient_id=user.id,
+        source_type="COLLABORATION_EDIT_REQUEST",
+        active_source_ids=active_source_ids,
+    )
+
+
+async def sync_project_reminders(
+    db: AsyncSession,
+    study_id: uuid.UUID,
+    user,
+    *,
+    now: datetime | None = None,
+) -> None:
+    membership = await member_crud.get_member(db, study_id, user.id)
+    if not membership or not membership.is_active:
+        await notification_service.resolve_recipient_study_notifications(
+            db,
+            study_id=study_id,
+            recipient_id=user.id,
+        )
+        await db.commit()
+        return
+    role = membership.role_in_study
+    current_time = now or datetime.now(timezone.utc)
+    if current_time.tzinfo is None:
+        current_time = current_time.replace(tzinfo=timezone.utc)
+    else:
+        current_time = current_time.astimezone(timezone.utc)
+
+    await _sync_risk_reminders(
+        db,
+        study_id=study_id,
+        user=user,
+        role=role,
+        current_time=current_time,
+    )
+    await _sync_document_distribution_reminders(
+        db,
+        study_id=study_id,
+        user=user,
+        role=role,
+        current_time=current_time,
+    )
+    await _sync_milestone_reminders(
+        db,
+        study_id=study_id,
+        user=user,
+        role=role,
+        current_time=current_time,
+    )
+    await _sync_visit_window_reminders(
+        db,
+        study_id=study_id,
+        user=user,
+        role=role,
+        current_time=current_time,
+    )
+    await _sync_collaboration_edit_request_reminders(
+        db,
+        study_id=study_id,
+        user=user,
+        role=role,
+    )
+    await db.commit()

backend/app/services/security_access_log_writer.py

@@ -6,99 +6,170 @@ import asyncio
 import logging
 import uuid
 from datetime import datetime, timezone
+from typing import Any
 
 from app.db.session import SessionLocal
 from app.models.security_access_log import SecurityAccessLog
+from app.services.ip_location import resolve_ip_location
+from app.services.security_events import classify_security_event
 
 logger = logging.getLogger("ctms.security_access_log_writer")
 
 BATCH_SIZE = 100
 FLUSH_INTERVAL = 3.0
 QUEUE_MAX_SIZE = 20000
+WRITE_ATTEMPTS = 2
+_QUEUE_STOP = object()
 
 
 class SecurityAccessLogWriter:
     def __init__(self) -> None:
-        self._queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=QUEUE_MAX_SIZE)
+        self._queue: asyncio.Queue[dict | object] = asyncio.Queue(maxsize=QUEUE_MAX_SIZE)
         self._task: asyncio.Task | None = None
+        self._stopping = False
+        self._started_at: datetime | None = None
+        self._last_success_at: datetime | None = None
+        self._last_error_at: datetime | None = None
+        self._last_error_type: str | None = None
+        self._accepted_entries = 0
+        self._written_entries = 0
+        self._dropped_entries = 0
+        self._failed_batches = 0
+        self._failed_entries = 0
+        self._retry_count = 0
 
     def enqueue(self, entry: dict) -> None:
+        if self._stopping:
+            self._dropped_entries += 1
+            logger.warning("Security access log writer is stopping, dropping entry")
+            return
         try:
             self._queue.put_nowait(entry)
+            self._accepted_entries += 1
         except asyncio.QueueFull:
+            self._dropped_entries += 1
             logger.warning("Security access log queue full, dropping entry")
 
     async def start(self) -> None:
+        if self._task and not self._task.done():
+            return
+        self._stopping = False
+        self._started_at = datetime.now(timezone.utc)
         self._task = asyncio.create_task(self._flush_loop())
         logger.info("SecurityAccessLogWriter started")
 
     async def stop(self) -> None:
-        if self._task:
-            self._task.cancel()
-            try:
-                await self._task
-            except asyncio.CancelledError:
-                pass
-        await self._drain()
+        if self._task and not self._task.done():
+            self._stopping = True
+            await self._queue.put(_QUEUE_STOP)
+            await self._task
+        self._task = None
         logger.info("SecurityAccessLogWriter stopped")
 
     async def _flush_loop(self) -> None:
         while True:
-            batch = await self._collect_batch()
+            batch, should_stop = await self._collect_batch()
             if batch:
                 await self._write_batch(batch)
+            if should_stop:
+                break
 
-    async def _collect_batch(self) -> list[dict]:
+    async def _collect_batch(self) -> tuple[list[dict], bool]:
         batch: list[dict] = []
         try:
             first = await asyncio.wait_for(self._queue.get(), timeout=FLUSH_INTERVAL)
+            if first is _QUEUE_STOP:
+                return batch, True
             batch.append(first)
         except asyncio.TimeoutError:
-            return batch
+            return batch, False
 
         while len(batch) < BATCH_SIZE:
             try:
-                batch.append(self._queue.get_nowait())
+                item = self._queue.get_nowait()
+                if item is _QUEUE_STOP:
+                    return batch, True
+                batch.append(item)
             except asyncio.QueueEmpty:
                 break
-        return batch
+        return batch, False
 
-    async def _write_batch(self, batch: list[dict]) -> None:
-        try:
-            async with SessionLocal() as session:
-                for entry in batch:
-                    session.add(
-                        SecurityAccessLog(
-                            id=uuid.uuid4(),
-                            method=entry["method"],
+    async def _write_batch(self, batch: list[dict]) -> bool:
+        for attempt in range(WRITE_ATTEMPTS):
+            try:
+                async with SessionLocal() as session:
+                    for entry in batch:
+                        classification = classify_security_event(
                             path=entry["path"],
                             status_code=entry["status_code"],
-                            elapsed_ms=entry["elapsed_ms"],
-                            client_ip=entry.get("client_ip"),
-                            user_agent=entry.get("user_agent"),
-                            client_type=entry.get("client_type"),
-                            client_version=entry.get("client_version"),
-                            client_platform=entry.get("client_platform"),
-                            build_channel=entry.get("build_channel"),
-                            build_commit=entry.get("build_commit"),
                             auth_status=entry["auth_status"],
-                            user_identifier=entry.get("user_identifier"),
-                            created_at=entry.get("created_at", datetime.now(timezone.utc)),
+                            ip_location=resolve_ip_location(entry.get("client_ip")),
                         )
+                        session.add(
+                            SecurityAccessLog(
+                                id=uuid.uuid4(),
+                                method=entry["method"],
+                                path=entry["path"],
+                                status_code=entry["status_code"],
+                                elapsed_ms=entry["elapsed_ms"],
+                                client_ip=entry.get("client_ip"),
+                                user_agent=entry.get("user_agent"),
+                                client_type=entry.get("client_type"),
+                                client_version=entry.get("client_version"),
+                                client_platform=entry.get("client_platform"),
+                                build_channel=entry.get("build_channel"),
+                                build_commit=entry.get("build_commit"),
+                                request_headers=entry.get("request_headers"),
+                                request_snapshot=entry.get("request_snapshot"),
+                                request_id=entry.get("request_id"),
+                                category=classification["category"],
+                                severity=classification["severity"],
+                                auth_status=entry["auth_status"],
+                                user_identifier=entry.get("user_identifier"),
+                                created_at=entry.get("created_at", datetime.now(timezone.utc)),
+                            )
+                        )
+                    await session.commit()
+            except Exception as exc:
+                self._last_error_at = datetime.now(timezone.utc)
+                self._last_error_type = type(exc).__name__
+                if attempt + 1 < WRITE_ATTEMPTS:
+                    self._retry_count += 1
+                    logger.warning(
+                        "Security access log batch write failed; retrying (%d entries)",
+                        len(batch),
                     )
-                await session.commit()
-        except Exception:
-            logger.exception("Failed to write security access log batch (%d entries)", len(batch))
+                    await asyncio.sleep(0)
+                    continue
+                self._failed_batches += 1
+                self._failed_entries += len(batch)
+                logger.exception(
+                    "Failed to write security access log batch after retries (%d entries)",
+                    len(batch),
+                )
+                return False
+            self._written_entries += len(batch)
+            self._last_success_at = datetime.now(timezone.utc)
+            return True
+        return False
 
-    async def _drain(self) -> None:
-        batch: list[dict] = []
-        while not self._queue.empty():
-            try:
-                batch.append(self._queue.get_nowait())
-            except asyncio.QueueEmpty:
-                break
-        if batch:
-            await self._write_batch(batch)
+    def stats(self) -> dict[str, Any]:
+        return {
+            "running": bool(self._task and not self._task.done()),
+            "stopping": self._stopping,
+            "queue_size": self._queue.qsize(),
+            "queue_capacity": self._queue.maxsize,
+            "accepted_entries": self._accepted_entries,
+            "written_entries": self._written_entries,
+            "dropped_entries": self._dropped_entries,
+            "failed_batches": self._failed_batches,
+            "failed_entries": self._failed_entries,
+            "retry_count": self._retry_count,
+            "started_at": self._started_at.isoformat() if self._started_at else None,
+            "last_success_at": self._last_success_at.isoformat() if self._last_success_at else None,
+            "last_error_at": self._last_error_at.isoformat() if self._last_error_at else None,
+            "last_error_type": self._last_error_type,
+        }
 
 
 _writer: SecurityAccessLogWriter | None = None

backend/app/services/security_events.py

@@ -0,0 +1,37 @@
+"""Security event classification shared by writers and monitoring APIs."""
+
+from __future__ import annotations
+
+from app.services.ip_location import IpLocation
+
+
+SENSITIVE_PROBE_MARKERS = (
+    "/.env",
+    ".env",
+    "/.git",
+    ".git/config",
+    "backup",
+    "config.php",
+    "wp-config",
+    "database.yml",
+)
+
+def classify_security_event(
+    *,
+    path: str,
+    status_code: int,
+    auth_status: str,
+    ip_location: IpLocation | None = None,
+) -> dict[str, str]:
+    normalized_path = (path or "").lower()
+    if any(marker in normalized_path for marker in SENSITIVE_PROBE_MARKERS):
+        return {"category": "PROBE", "severity": "CRITICAL"}
+    if status_code >= 500:
+        return {"category": "SERVER_ERROR", "severity": "HIGH"}
+    if auth_status == "INVALID_TOKEN":
+        return {"category": "INVALID_TOKEN", "severity": "MEDIUM"}
+    if auth_status == "ANONYMOUS" and status_code in {401, 403}:
+        return {"category": "ANONYMOUS_API", "severity": "MEDIUM"}
+    if status_code == 404:
+        return {"category": "NOT_FOUND_NOISE", "severity": "LOW"}
+    return {"category": "OTHER", "severity": "LOW"}

backend/app/services/source_location_aggregator.py

@@ -0,0 +1,292 @@
+"""Hourly source-location aggregation and timeline reads."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import hmac
+import logging
+import uuid
+from collections import defaultdict
+from datetime import datetime, timedelta, timezone
+from typing import Literal
+
+from sqlalchemy import delete, func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.config import settings
+from app.db.session import SessionLocal
+from app.models.permission_access_log import PermissionAccessLog
+from app.models.security_access_log import SecurityAccessLog
+from app.models.source_location_snapshot import SourceLocationSnapshot
+from app.services.geo_location_metadata import resolve_geo_location_metadata
+from app.services.ip_geolocation_fallback import resolve_external_ip_locations
+from app.services.ip_location import resolve_ip_location
+
+logger = logging.getLogger("ctms.source_location_aggregator")
+
+
+def _normalize_datetime(value: datetime) -> datetime:
+    return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value.astimezone(timezone.utc)
+
+
+def _identity_hash(kind: str, value: str | None) -> str:
+    if not value:
+        return ""
+    payload = f"source-location:{kind}:{value}".encode("utf-8")
+    return hmac.new(settings.JWT_SECRET_KEY.encode("utf-8"), payload, hashlib.sha256).hexdigest()
+
+
+async def aggregate_source_location_hour(bucket_start: datetime, bucket_end: datetime) -> int:
+    bucket_start = _normalize_datetime(bucket_start)
+    bucket_end = _normalize_datetime(bucket_end)
+    async with SessionLocal() as session:
+        matching_security_request = select(SecurityAccessLog.id).where(
+            PermissionAccessLog.request_id.is_not(None),
+            SecurityAccessLog.request_id == PermissionAccessLog.request_id,
+        ).exists()
+        permission_rows = (
+            await session.execute(
+                select(
+                    PermissionAccessLog.ip_address,
+                    PermissionAccessLog.user_id,
+                    func.count().filter(PermissionAccessLog.allowed.is_(True)).label("allowed_count"),
+                    func.count().filter(PermissionAccessLog.allowed.is_(False)).label("denied_count"),
+                    func.min(PermissionAccessLog.created_at).label("first_seen_at"),
+                    func.max(PermissionAccessLog.created_at).label("last_seen_at"),
+                )
+                .where(
+                    PermissionAccessLog.created_at >= bucket_start,
+                    PermissionAccessLog.created_at < bucket_end,
+                    PermissionAccessLog.ip_address.is_not(None),
+                    ~matching_security_request,
+                )
+                .group_by(PermissionAccessLog.ip_address, PermissionAccessLog.user_id)
+            )
+        ).all()
+        security_rows = (
+            await session.execute(
+                select(
+                    SecurityAccessLog.client_ip,
+                    SecurityAccessLog.user_identifier,
+                    func.count().filter(SecurityAccessLog.status_code < 400).label("allowed_count"),
+                    func.count().filter(SecurityAccessLog.status_code >= 400).label("denied_count"),
+                    func.count()
+                    .filter(
+                        SecurityAccessLog.category.is_not(None),
+                        ~SecurityAccessLog.category.in_(("OTHER", "NOT_FOUND_NOISE")),
+                    )
+                    .label("security_event_count"),
+                    func.count()
+                    .filter(SecurityAccessLog.severity.in_(("HIGH", "CRITICAL")))
+                    .label("high_risk_count"),
+                    func.count()
+                    .filter(
+                        SecurityAccessLog.status_code >= 400,
+                        SecurityAccessLog.auth_status.in_(("INVALID_TOKEN", "ANONYMOUS")),
+                    )
+                    .label("auth_failure_count"),
+                    func.min(SecurityAccessLog.created_at).label("first_seen_at"),
+                    func.max(SecurityAccessLog.created_at).label("last_seen_at"),
+                )
+                .where(
+                    SecurityAccessLog.created_at >= bucket_start,
+                    SecurityAccessLog.created_at < bucket_end,
+                    SecurityAccessLog.client_ip.is_not(None),
+                )
+                .group_by(SecurityAccessLog.client_ip, SecurityAccessLog.user_identifier)
+            )
+        ).all()
+
+        aggregates: dict[tuple[str, str], dict] = {}
+
+        def merge_row(
+            ip_address: str,
+            user_identity: str,
+            allowed_count: int,
+            denied_count: int,
+            first_seen_at: datetime,
+            last_seen_at: datetime,
+            *,
+            security_event_count: int = 0,
+            high_risk_count: int = 0,
+            auth_failure_count: int = 0,
+        ) -> None:
+            key = (ip_address, user_identity)
+            row = aggregates.setdefault(
+                key,
+                {
+                    "ip_address": ip_address,
+                    "user_identity": user_identity,
+                    "allowed_count": 0,
+                    "denied_count": 0,
+                    "security_event_count": 0,
+                    "high_risk_count": 0,
+                    "auth_failure_count": 0,
+                    "first_seen_at": _normalize_datetime(first_seen_at),
+                    "last_seen_at": _normalize_datetime(last_seen_at),
+                },
+            )
+            row["allowed_count"] += int(allowed_count or 0)
+            row["denied_count"] += int(denied_count or 0)
+            row["security_event_count"] += int(security_event_count or 0)
+            row["high_risk_count"] += int(high_risk_count or 0)
+            row["auth_failure_count"] += int(auth_failure_count or 0)
+            row["first_seen_at"] = min(row["first_seen_at"], _normalize_datetime(first_seen_at))
+            row["last_seen_at"] = max(row["last_seen_at"], _normalize_datetime(last_seen_at))
+
+        for ip_address, user_id, allowed, denied, first_seen, last_seen in permission_rows:
+            merge_row(str(ip_address), str(user_id or ""), allowed, denied, first_seen, last_seen)
+        for ip_address, user_identifier, allowed, denied, security_events, high_risk, auth_failures, first_seen, last_seen in security_rows:
+            merge_row(
+                str(ip_address),
+                str(user_identifier or ""),
+                allowed,
+                denied,
+                first_seen,
+                last_seen,
+                security_event_count=security_events,
+                high_risk_count=high_risk,
+                auth_failure_count=auth_failures,
+            )
+
+        resolved_locations = {
+            ip_address: resolve_ip_location(ip_address)
+            for ip_address, _user_identity in aggregates
+        }
+        resolved_metadata = {
+            ip_address: resolve_geo_location_metadata(ip_info)
+            for ip_address, ip_info in resolved_locations.items()
+        }
+        missing_coordinate_ips = [
+            ip_address
+            for ip_address, metadata in resolved_metadata.items()
+            if metadata.accuracy_level != "private"
+            and (metadata.longitude is None or metadata.latitude is None)
+        ]
+        external_locations = await resolve_external_ip_locations(missing_coordinate_ips)
+        for ip_address, external_location in external_locations.items():
+            resolved_locations[ip_address] = external_location.merge_ip_location(resolved_locations[ip_address])
+            resolved_metadata[ip_address] = external_location.to_metadata()
+
+        await session.execute(
+            delete(SourceLocationSnapshot).where(SourceLocationSnapshot.bucket_time == bucket_start)
+        )
+        for row in aggregates.values():
+            ip_info = resolved_locations[row["ip_address"]]
+            metadata = resolved_metadata[row["ip_address"]]
+            longitude = metadata.longitude
+            latitude = metadata.latitude
+            country = metadata.country or ip_info.country
+            location = " / ".join(
+                part for part in [country, ip_info.province, ip_info.city] if part
+            ) or ip_info.location or "未知"
+            session.add(
+                SourceLocationSnapshot(
+                    id=uuid.uuid4(),
+                    bucket_time=bucket_start,
+                    ip_hash=_identity_hash("ip", row["ip_address"]),
+                    user_hash=_identity_hash("user", row["user_identity"]),
+                    country=country,
+                    country_code=metadata.country_code,
+                    province=ip_info.province,
+                    region_code=metadata.region_code,
+                    city=ip_info.city,
+                    isp=ip_info.isp,
+                    location=location,
+                    longitude=longitude,
+                    latitude=latitude,
+                    accuracy_level=metadata.accuracy_level,
+                    allowed_count=row["allowed_count"],
+                    denied_count=row["denied_count"],
+                    security_event_count=row["security_event_count"],
+                    high_risk_count=row["high_risk_count"],
+                    auth_failure_count=row["auth_failure_count"],
+                    first_seen_at=row["first_seen_at"],
+                    last_seen_at=row["last_seen_at"],
+                )
+            )
+        await session.commit()
+        return len(aggregates)
+
+
+async def get_source_location_timeline(
+    db: AsyncSession,
+    *,
+    start_at: datetime,
+    end_at: datetime,
+    granularity: Literal["hour", "day"],
+) -> list[dict]:
+    rows = (
+        await db.execute(
+            select(SourceLocationSnapshot)
+            .where(
+                SourceLocationSnapshot.bucket_time >= start_at,
+                SourceLocationSnapshot.bucket_time < end_at,
+            )
+            .order_by(SourceLocationSnapshot.bucket_time)
+        )
+    ).scalars().all()
+    buckets: dict[datetime, dict] = defaultdict(
+        lambda: {
+            "allowed_count": 0,
+            "denied_count": 0,
+            "security_event_count": 0,
+            "high_risk_count": 0,
+            "ip_hashes": set(),
+            "user_hashes": set(),
+        }
+    )
+    for row in rows:
+        bucket = _normalize_datetime(row.bucket_time)
+        if granularity == "day":
+            bucket = bucket.replace(hour=0, minute=0, second=0, microsecond=0)
+        else:
+            bucket = bucket.replace(minute=0, second=0, microsecond=0)
+        item = buckets[bucket]
+        item["allowed_count"] += row.allowed_count
+        item["denied_count"] += row.denied_count
+        item["security_event_count"] += row.security_event_count
+        item["high_risk_count"] += row.high_risk_count
+        item["ip_hashes"].add(row.ip_hash)
+        if row.user_hash:
+            item["user_hashes"].add(row.user_hash)
+
+    return [
+        {
+            "bucket_time": bucket.isoformat(),
+            "total_count": item["allowed_count"] + item["denied_count"],
+            "allowed_count": item["allowed_count"],
+            "denied_count": item["denied_count"],
+            "security_event_count": item["security_event_count"],
+            "high_risk_count": item["high_risk_count"],
+            "unique_ip_count": len(item["ip_hashes"]),
+            "unique_user_count": len(item["user_hashes"]),
+        }
+        for bucket, item in sorted(buckets.items())
+    ]
+
+
+async def run_hourly_source_location_aggregation(stop_event: asyncio.Event) -> None:
+    logger.info("Source location aggregator started")
+    now = datetime.now(timezone.utc)
+    completed_hour = now.replace(minute=0, second=0, microsecond=0)
+    try:
+        await aggregate_source_location_hour(completed_hour - timedelta(hours=1), completed_hour)
+    except Exception:
+        logger.exception("Failed to backfill source location snapshot for %s", completed_hour)
+
+    while not stop_event.is_set():
+        now = datetime.now(timezone.utc)
+        next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)
+        try:
+            await asyncio.wait_for(stop_event.wait(), timeout=(next_hour - now).total_seconds())
+            break
+        except asyncio.TimeoutError:
+            pass
+        try:
+            await aggregate_source_location_hour(next_hour - timedelta(hours=1), next_hour)
+        except Exception:
+            logger.exception("Failed to aggregate source locations for %s", next_hour)
+
+    logger.info("Source location aggregator stopped")

backend/app/services/user_login_sessions.py

@@ -0,0 +1,233 @@
+"""Server-authoritative login session activity for the admin account list."""
+
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Any, Iterable
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.config import settings
+from app.core.request_context import resolve_client_ip
+from app.models.user_login_session import UserLoginSession
+from app.services.ip_location import resolve_ip_location
+
+
+@dataclass(frozen=True)
+class UserLoginSummary:
+    status: str = "OFFLINE"
+    last_login_at: datetime | None = None
+    last_seen_at: datetime | None = None
+    client_type: str | None = None
+    active_session_count: int = 0
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _as_utc(value: datetime) -> datetime:
+    return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value.astimezone(timezone.utc)
+
+
+def _text_header(headers: Any, name: str, limit: int) -> str | None:
+    value = (headers.get(name) or "").strip()
+    return value[:limit] or None
+
+
+def session_id_from_payload(payload: dict[str, Any]) -> uuid.UUID:
+    raw_session_id = payload.get("sid")
+    try:
+        return uuid.UUID(str(raw_session_id))
+    except (TypeError, ValueError):
+        # Tokens issued before session tracking are mapped to a deterministic
+        # legacy session without persisting the token or a token-derived value.
+        legacy_key = ":".join(
+            [
+                str(payload.get("sub") or ""),
+                str(payload.get("orig_iat") or payload.get("iat") or ""),
+                str(payload.get("client_type") or "web"),
+            ]
+        )
+        return uuid.uuid5(uuid.NAMESPACE_URL, f"ctms:legacy-session:{legacy_key}")
+
+
+def session_client_type(payload: dict[str, Any], headers: Any) -> str:
+    from_payload = (payload.get("client_type") or "").strip().lower()
+    from_header = (headers.get("x-ctms-client-type") or "").strip().lower()
+    return "desktop" if "desktop" in {from_payload, from_header} else "web"
+
+
+def login_activity_status(session: UserLoginSession, *, reference_time: datetime | None = None) -> str:
+    if session.ended_at is not None:
+        return "ENDED"
+    cutoff = (reference_time or _now()) - timedelta(seconds=settings.USER_SESSION_ONLINE_SECONDS)
+    return "ONLINE" if _as_utc(session.last_seen_at) >= cutoff else "OFFLINE"
+
+
+def login_activity_payload(
+    session: UserLoginSession,
+    *,
+    reference_time: datetime | None = None,
+) -> dict[str, Any]:
+    ip_location = resolve_ip_location(session.login_ip) if session.login_ip else None
+    return {
+        "id": session.id,
+        "client_type": session.client_type,
+        "client_platform": session.client_platform,
+        "client_version": session.client_version,
+        "client_source": session.client_source,
+        "login_ip": session.login_ip,
+        "ip_location": ip_location.location if ip_location else None,
+        "login_at": session.login_at,
+        "last_seen_at": session.last_seen_at,
+        "ended_at": session.ended_at,
+        "end_reason": session.end_reason,
+        "activity_status": login_activity_status(session, reference_time=reference_time),
+    }
+
+
+async def create_login_session(
+    db: AsyncSession,
+    *,
+    session_id: uuid.UUID,
+    user_id: uuid.UUID,
+    request: Any,
+    login_at: datetime | None = None,
+) -> UserLoginSession:
+    occurred_at = login_at or _now()
+    session = UserLoginSession(
+        id=session_id,
+        user_id=user_id,
+        client_type=session_client_type({}, request.headers),
+        client_platform=_text_header(request.headers, "x-ctms-client-platform", 32),
+        client_version=_text_header(request.headers, "x-ctms-client-version", 64),
+        client_source=_text_header(request.headers, "x-ctms-client-source", 32),
+        login_ip=resolve_client_ip(request),
+        login_at=occurred_at,
+        last_seen_at=occurred_at,
+    )
+    db.add(session)
+    await db.commit()
+    await db.refresh(session)
+    return session
+
+
+async def touch_login_session(
+    db: AsyncSession,
+    *,
+    user_id: uuid.UUID,
+    payload: dict[str, Any],
+    request: Any,
+) -> UserLoginSession | None:
+    session_id = session_id_from_payload(payload)
+    session = await db.get(UserLoginSession, session_id)
+    now = _now()
+    if session is None:
+        session = UserLoginSession(
+            id=session_id,
+            user_id=user_id,
+            client_type=session_client_type(payload, request.headers),
+            client_platform=_text_header(request.headers, "x-ctms-client-platform", 32),
+            client_version=_text_header(request.headers, "x-ctms-client-version", 64),
+            client_source=_text_header(request.headers, "x-ctms-client-source", 32),
+            login_ip=resolve_client_ip(request),
+            login_at=now,
+            last_seen_at=now,
+        )
+        db.add(session)
+    elif session.user_id != user_id or session.ended_at is not None:
+        return None
+    else:
+        session.last_seen_at = now
+        session.client_platform = _text_header(request.headers, "x-ctms-client-platform", 32) or session.client_platform
+        session.client_version = _text_header(request.headers, "x-ctms-client-version", 64) or session.client_version
+    await db.commit()
+    await db.refresh(session)
+    return session
+
+
+async def end_login_session(
+    db: AsyncSession,
+    *,
+    user_id: uuid.UUID,
+    payload: dict[str, Any],
+    reason: str = "logout",
+) -> bool:
+    session_id = session_id_from_payload(payload)
+    result = await db.execute(
+        update(UserLoginSession)
+        .where(
+            UserLoginSession.id == session_id,
+            UserLoginSession.user_id == user_id,
+            UserLoginSession.ended_at.is_(None),
+        )
+        .values(ended_at=_now(), end_reason=reason, last_seen_at=_now())
+    )
+    await db.commit()
+    return bool(result.rowcount)
+
+
+async def get_login_summaries(
+    db: AsyncSession,
+    user_ids: Iterable[uuid.UUID],
+) -> dict[uuid.UUID, UserLoginSummary]:
+    ids = list(user_ids)
+    if not ids:
+        return {}
+    rows = (
+        await db.execute(
+            select(UserLoginSession)
+            .where(UserLoginSession.user_id.in_(ids))
+            .order_by(UserLoginSession.user_id, UserLoginSession.login_at.desc())
+        )
+    ).scalars().all()
+    cutoff = _now() - timedelta(seconds=settings.USER_SESSION_ONLINE_SECONDS)
+    summaries: dict[uuid.UUID, UserLoginSummary] = {}
+    mutable: dict[uuid.UUID, dict[str, Any]] = {}
+    for row in rows:
+        current = mutable.setdefault(
+            row.user_id,
+            {
+                "last_login_at": row.login_at,
+                "last_seen_at": row.last_seen_at,
+                "client_type": row.client_type,
+                "active_sources": set(),
+            },
+        )
+        row_last_seen_at = _as_utc(row.last_seen_at)
+        if row_last_seen_at and (
+            current["last_seen_at"] is None or row_last_seen_at > _as_utc(current["last_seen_at"])
+        ):
+            current["last_seen_at"] = row_last_seen_at
+        if row.ended_at is None and row_last_seen_at >= cutoff:
+            source_key = (row.client_type, row.login_ip) if row.login_ip else ("session", row.id)
+            current["active_sources"].add(source_key)
+    for user_id, item in mutable.items():
+        active_session_count = len(item["active_sources"])
+        summaries[user_id] = UserLoginSummary(
+            status="ONLINE" if active_session_count else "OFFLINE",
+            last_login_at=item["last_login_at"],
+            last_seen_at=item["last_seen_at"],
+            client_type=item["client_type"],
+            active_session_count=active_session_count,
+        )
+    return summaries
+
+
+async def list_login_activities(
+    db: AsyncSession,
+    *,
+    user_id: uuid.UUID,
+    limit: int,
+) -> list[UserLoginSession]:
+    result = await db.execute(
+        select(UserLoginSession)
+        .where(UserLoginSession.user_id == user_id)
+        .order_by(UserLoginSession.login_at.desc())
+        .limit(limit)
+    )
+    return list(result.scalars().all())