diff --git a/yunke-server/app/api/routers/export.py b/yunke-server/app/api/routers/export.py index 33c9a97..bf06787 100644 --- a/yunke-server/app/api/routers/export.py +++ b/yunke-server/app/api/routers/export.py @@ -21,6 +21,13 @@ router = APIRouter(prefix="/api/export", tags=["export"]) SessionDep = Annotated[AsyncSession, Depends(get_session)] +# Hard cap on rows an HTML export will load + render. An account-wide export +# with no date range can match millions of messages; rendering all of them in +# memory risks OOM. The cap is a safety valve, not a tight budget — at 5k rows +# the HTML is already unwieldy in any browser. When hit, the response carries +# a visible warning banner so the operator knows to narrow the date range. +EXPORT_MAX_ROWS = 5000 + def _ts_text(ts: int | None) -> str: if not ts: @@ -73,7 +80,14 @@ def _render_media(m) -> str: return "" -def _render_chat_html(rows: list[Message], channel: str, owner: str, peer: str) -> str: +def _render_chat_html( + rows: list[Message], + channel: str, + owner: str, + peer: str, + *, + warning: str = "", +) -> str: bubbles = [] for m in rows: side = "right" if m.is_send else "left" @@ -90,6 +104,7 @@ def _render_chat_html(rows: list[Message], channel: str, owner: str, peer: str) {title}

{title}

+{warning}

{title}

{''.join(bubbles) or '

暂无消息

'} """ @@ -132,9 +147,21 @@ async def export_chat_html( q = q.where(Message.timestamp >= start) if end is not None: q = q.where(Message.timestamp <= end) - q = q.order_by(sa.nullslast(Message.timestamp.asc()), Message.msg_id.asc()) + q = q.order_by(sa.nullslast(Message.timestamp.asc()), Message.msg_id.asc()).limit( + EXPORT_MAX_ROWS + ) rows = (await session.execute(q)).scalars().all() - html = _render_chat_html(list(rows), channel, owner_account_id, peer) + # If we filled the cap there may be more rows beyond it — surface a visible + # warning so the operator knows the export was truncated and can re-run + # with a tighter date range. + capped = len(rows) >= EXPORT_MAX_ROWS + warning = ( + f'
⚠️ 结果超过 {EXPORT_MAX_ROWS} 条，' + f'仅导出前 {EXPORT_MAX_ROWS} 条。请缩小日期范围。
' + if capped + else "" + ) + html = _render_chat_html(list(rows), channel, owner_account_id, peer, warning=warning) safe = peer or "all" filename = f"chat-{channel}-{owner_account_id}-{safe}.html".replace("@", "_").replace("/", "_") return Response( diff --git a/yunke-server/app/api/routers/media.py b/yunke-server/app/api/routers/media.py index b343318..093fd41 100644 --- a/yunke-server/app/api/routers/media.py +++ b/yunke-server/app/api/routers/media.py @@ -22,6 +22,7 @@ from fastapi.responses import FileResponse from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_session +from app.config import get_settings from app.db.models import Media from app.scraper.media import guess_mime_type @@ -41,7 +42,19 @@ async def get_media(media_hash: str, session: SessionDep) -> FileResponse: if row is None or not row.local_path: raise HTTPException(status_code=404, detail="media not found") - path = Path(row.local_path) + # Path-traversal defense: resolve absolute and confirm the file lives under + # the configured ``media_dir`` sandbox. A row whose ``local_path`` escapes + # media_dir (an absolute path elsewhere, or a ``..``-laced value) is treated + # as not found — never served. Defense-in-depth: even if a future bug or a + # hostile scrape payload ever lets ``local_path`` point outside the sandbox, + # this check keeps the file server constrained to media_dir. + media_dir = get_settings().media_dir.resolve() + path = Path(row.local_path).resolve() + try: + path.relative_to(media_dir) + except ValueError: + raise HTTPException(status_code=404, detail="media not found") + if not path.is_file(): # Row exists but the blob was removed out-of-band — treat as not-found. raise HTTPException(status_code=404, detail="media file missing") diff --git a/yunke-server/app/sync/scheduler.py b/yunke-server/app/sync/scheduler.py index 9dc499f..1586df4 100644 --- a/yunke-server/app/sync/scheduler.py +++ b/yunke-server/app/sync/scheduler.py @@ -50,6 +50,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn from app.config import get_settings from app.db import models # noqa: F401 (register tables on Base.metadata) from app.db.base import Base +from app.db.models import SyncRun from app.session.yunke_client import YunkeClient from app.storage.object_store import ObjectStore from app.sync.orchestrator import ( @@ -83,6 +84,14 @@ __all__ = ["SyncScheduler", "ENTITY_FUNCS"] # request, so 200 conversations × N pages each can still take a while). DEFAULT_MAX_CONVERSATIONS_PER_RUN = 200 +# Advisory-lock TTL: a ``running`` SyncRun row older than this is considered +# stale (the process probably crashed mid-run) and a new invocation may +# proceed despite it. 60 min is generous — any entity fits inside — but +# short enough that a crashed run auto-recovers within an hour. Tuned to +# protect against concurrent CLI + scheduler invocations hammering 云控 +# with the same scope twice. +ADVISORY_LOCK_TTL_MINUTES = 60 + # --------------------------------------------------------------------------- # Conversation discovery (drives the incremental message sync) @@ -561,6 +570,15 @@ class SyncScheduler: This is the failure-policy wrapper that protects the scheduler loop: a raised exception becomes a logged error and an empty result list, and the scheduler keeps its next appointment. + + Concurrency: an advisory lock refuses to run if another ``_safe_run`` + for the same entity is still active. The lock is implemented by + staking a ``running`` SyncRun row keyed on the scheduler entity name + (e.g. ``"all"`` / ``"messages"`` / ``"friends"``) at the start and + finalizing it on completion. A concurrent invocation sees the row and + skips. The 60-min TTL (:data:`ADVISORY_LOCK_TTL_MINUTES`) auto-recovers + from a crashed process: a stale ``running`` row eventually falls + outside the window and a new run may proceed. """ started_at = dt.datetime.now(dt.UTC) client: YunkeClient | None = None @@ -568,8 +586,74 @@ class SyncScheduler: try: client = await _build_client_from_env() async with self._session_maker() as session: + # --- Advisory lock check ------------------------------ + # ``SyncRun.started_at`` is a non-tz-aware DateTime; reading + # a row back yields a naive datetime, so the cutoff is built + # naive-UTC to keep the comparison type-consistent. + cutoff = ( + dt.datetime.now(dt.UTC).replace(tzinfo=None) + - dt.timedelta(minutes=ADVISORY_LOCK_TTL_MINUTES) + ) + existing = ( + await session.execute( + sa.select(SyncRun) + .where(SyncRun.entity_type == entity) + .where(SyncRun.status == "running") + .where(SyncRun.started_at > cutoff) + .limit(1) + ) + ).scalar_one_or_none() + if existing is not None: + logger.warning( + "sync %s: skipped — another run is active" + " (started %s)", + entity, + existing.started_at.isoformat(), + ) + skipped = SyncResult( + entity=entity, + status="skipped", + fetched=0, + upserted=0, + error=f"another sync is running for {entity}", + ) + if log: + _log_run(entity, [skipped], started_at=started_at) + return [skipped] + + # --- Stake our own 'running' row ---------------------- + # A wrapping SyncRun at the scheduler-entity level (distinct + # from any sub-entity rows the orchestrator creates internally). + # On a caught exception the session rolls back and this row + # disappears with everything else — no phantom to recover + # from. On the success path the row is finalized + committed + # below. If the process hard-crashes between flush and commit, + # the row stays 'running' and blocks retries for up to TTL. + run = SyncRun( + entity_type=entity, + status="running", + started_at=dt.datetime.now(dt.UTC), + ) + session.add(run) + await session.flush() + runner = ENTITY_FUNCS[entity] results = await runner(session, client, ObjectStore()) + + # --- Finalize the wrapping SyncRun -------------------- + run.status = ( + "success" + if all(r.status == "success" for r in results) + else "failed" + ) + run.records_fetched = sum(r.fetched for r in results) + run.records_upserted = sum(r.upserted for r in results) + run.finished_at = dt.datetime.now(dt.UTC) + if run.status == "failed": + joined = "; ".join( + filter(None, (r.error for r in results)) + ) + run.error = joined or None await session.commit() except Exception as exc: # noqa: BLE001 — scheduler MUST NOT crash logger.exception("sync %s failed", entity) diff --git a/yunke-server/tests/test_api_export.py b/yunke-server/tests/test_api_export.py index f280b12..9283948 100644 --- a/yunke-server/tests/test_api_export.py +++ b/yunke-server/tests/test_api_export.py @@ -6,8 +6,10 @@ from __future__ import annotations +import functools from sqlalchemy.ext.asyncio import AsyncSession +from app.api.routers.export import EXPORT_MAX_ROWS from app.db.models import Message @@ -39,6 +41,26 @@ async def _seed(session: AsyncSession) -> None: await session.commit() +async def _seed_many(session: AsyncSession, n: int) -> None: + """Seed ``n`` DM messages so the row cap can be exercised.""" + session.add_all([ + Message( + msg_id=f"cap{i}", + channel="wechat", + owner_account_id="wx_owner", + talker_id="wx_friend", + room_id=None, + text=f"msg{i}", + timestamp=1000 + i, + is_send=i % 2 == 0, + type="text", + type_raw=1, + ) + for i in range(n) + ]) + await session.commit() + + async def test_export_dm_returns_html(make_client): client = await make_client(_seed) r = await client.get("/api/export/chat-html", params={ @@ -83,3 +105,35 @@ async def test_export_embeds_media(make_client): # 本地路径 → 占位提示,不嵌