diff --git a/README.md b/README.md index cfae9da..107fa75 100644 --- a/README.md +++ b/README.md @@ -263,7 +263,7 @@ AI 模型兼容性由模拟 Chat Completions/Responses HTTP 流与真实 SDK 适 ### 首页信息与基础信息处理模型 -首页提供平台消息、顾问日度排名、已参加比赛与协议、未来活动,独立刷新并保存上次成功快照。消息逐页浏览,活动读取平台分页后按结束时间过滤;日期或排名缺失保持未知。刷新失败保留缓存并提示过期。 +首页提供平台消息、顾问日度排名、已参加比赛与协议、未来活动,独立刷新并保存上次成功快照。公告与社区资讯仅保留最近一个自然月(按美东时间回推),正文及其 AI 解读只暂存在进程内存中,15 分钟到期或服务重启后清除,不存数据库;消息逐页浏览,活动读取平台分页后按结束时间过滤;日期或排名缺失保持未知。刷新失败保留缓存并提示过期。 在大模型服务设置“基础信息处理模型”(沿用 description_model 字段),与 alpha研究使用模型共享连接配置。该模型继续生成 Alpha Description,并用于手动生成首页摘要及比赛规则解读;未配置时仍可查看原始信息,不回退到研究模型。解读保存来源版本及所用模型,来源或连接配置变化后提示待更新,不自动调用模型。比赛解读附原文依据,仅提供信息建议,不执行报名、接受协议或安排任务。 diff --git a/backend/app/home_information.py b/backend/app/home_information.py index 6589085..ad7de94 100644 --- a/backend/app/home_information.py +++ b/backend/app/home_information.py @@ -4,6 +4,7 @@ Only allowlisted platform paths are fetched. HTML is converted to inert text; platform content is never an instruction and never supplies model/tool endpoints. """ import asyncio +import calendar import hashlib import json import math @@ -131,10 +132,10 @@ def next_offset(data, path, current): raise WqError("平台分页信息无法识别", "invalid_response") from None -async def all_pages(client, path): +async def all_pages(client, path, limit=None): output, offset = [], 0 for _ in range(100): - data = await client.get(path, params={"offset": offset}) + data = await client.get(path, params={"offset": offset, **({"limit": limit} if limit else {})}) output.extend(rows(data)) offset = next_offset(data, path, offset) if offset is None: @@ -160,14 +161,17 @@ def competition(item, user): async def source(client, module, user, offset, cid): if module == "messages": path = "/users/self/messages" - data = await client.get(path, params={"offset": offset}) items = [] - for item in rows(data): + for item in await all_pages(client, path, limit=100): + created = instant(item.get("dateCreated")) + if not created or not month_start() <= created <= now(): + continue body, links = plain(item.get("description")) items.append(dict(id=text(item.get("id")), title=text(item.get("title")) or "平台消息", type=text(item.get("type")), date=text(item.get("dateCreated")), description=body, links=links, url="https://platform.worldquantbrain.com/messages/" + ("announcements" if item.get("type") == "ANNOUNCEMENT" else "notifications"))) - return dict(items=items, offset=offset, next_offset=next_offset(data, path, offset), total=number(data.get("count"))) + items.sort(key=lambda item: instant(item["date"]), reverse=True) + return dict(items=items) if module == "leaderboard": data = await client.get("/consultant/boards/leader", params={"user": user}) matches = [r for r in rows(data) if r.get("user") == user] @@ -227,8 +231,74 @@ def config_version(config): return digest([config.description_model, config.base_url, config.api_key_encrypted, config.protocol]) +def month_start(): + """One calendar month before the Eastern wall time, clamped at month end.""" + at = now().astimezone(EASTERN) + year, month = (at.year, at.month - 1) if at.month > 1 else (at.year - 1, 12) + return at.replace(year=year, month=month, day=min(at.day, calendar.monthrange(year, month)[1])) + + +def message_page(request, user, offset): + """Return a detached page; neither source text nor AI output enters an ORM session.""" + cached = request.app.state.home_message_cache.get(user) + if cached is None: + return None + if now() - cached.created_at >= timedelta(minutes=15): + del request.app.state.home_message_cache[user] + return None + content = None + if cached.content is not None: + # Enforce the rolling boundary even on cached reads and generation commits. + filtered = [item for item in cached.content["items"] + if month_start() <= instant(item["date"]) <= now()] + if len(filtered) != len(cached.content["items"]): + cached.analyses.clear() + cached.content = {"items": filtered} + items = cached.content["items"] + content = dict(items=items[offset:offset + 10], total=len(items), offset=offset, + next_offset=offset + 10 if offset + 10 < len(items) else None) + return SimpleNamespace(content=content, version=digest(content), fetched_at=cached.fetched_at, + error=cached.error, analysis=cached.analyses.get(offset)) + + +async def read_messages(request, user, offset, refresh): + """Keep only a 15-minute in-process cache; restart/expiry discards interpretations too.""" + async with lock(request, (user, "messages")): + cache = request.app.state.home_message_cache + for owner, entry in list(cache.items()): + if now() - entry.created_at >= timedelta(minutes=15): + del cache[owner] + cached = cache.get(user) + if cached is not None and not refresh: + return message_page(request, user, offset) + try: + content = await source(request.app.state.runner.client, "messages", user, 0, "") + error = None + except WqError as exc: + content, error = None, str(exc) + if await identity(request) != user: + raise HTTPException(409, "账户已变化,请重新读取信息") + if cached is None: + cached = SimpleNamespace(content=None, fetched_at=None, created_at=now(), error=None, analyses={}) + cache[user] = cached + cached.error = error + if content is not None: + if cached.content != content: + cached.analyses.clear() + cached.content, cached.fetched_at = content, now() + return message_page(request, user, offset) + + +async def cached_record(request, db, user, module, offset, key): + if module == "messages": + return message_page(request, user, offset) + return await db.get(HomeInformation, (user, key)) + + async def read_record(request, user, module, offset, cid, refresh=False): key = resource_key(module, offset, cid) + if module == "messages": + return await read_messages(request, user, offset, refresh) async with lock(request, (user, key)): async with request.app.state.sessions() as db: record = await db.get(HomeInformation, (user, key)) @@ -303,12 +373,16 @@ async def output(request, user, module, record): config = await db.get(AISettings, 1) _, version = await analysis_sources(request, user, module, record) analysis = record.analysis + if module == "messages" and analysis and analysis["source_version"] != version: + request.app.state.home_message_cache[user].analyses.pop(record.content["offset"], None) + analysis = None if analysis: analysis = {**analysis, "outdated": analysis["source_version"] != version or analysis["config_version"] != config_version(config)} analysis.pop("config_version", None) return dict(content=temporal(record.content, module), fetched_at=record.fetched_at.replace(tzinfo=timezone.utc) if record.fetched_at else None, error=record.error, stale=bool(record.error) or bool(module == "leaderboard" and record.content and record.content["scope"] != "dailyOsmosisRank:" + now().astimezone(EASTERN).date().isoformat()) or bool(record.fetched_at and now() - record.fetched_at.replace(tzinfo=timezone.utc) > timedelta(hours=24)), source_version=record.version, analysis=analysis, + ephemeral=module == "messages", can_generate=bool(config.description_model and config.api_key_encrypted and config.base_url and record.content)) @@ -364,7 +438,7 @@ async def generate(request: Request, module: Module, offset: int = Query(0, ge=0 raise HTTPException(409, "该模块正在生成解读,请稍候") async with generation_lock: async with request.app.state.sessions() as db: - record = await db.get(HomeInformation, (user, key)) + record = await cached_record(request, db, user, module, offset, key) config = await db.get(AISettings, 1) if not config.description_model or not config.api_key_encrypted or not config.base_url: raise HTTPException(409, "请先配置基础信息处理模型及共享连接配置") @@ -396,13 +470,17 @@ async def generate(request: Request, module: Module, offset: int = Query(0, ge=0 await ai.authorize(token_hash(request.cookies["wq_session"])) if await identity(request) != user: raise HTTPException(409, "账户已变化,请重新生成") - async with lock(request, (user, key)): + async with lock(request, (user, "messages" if module == "messages" else key)): async with request.app.state.sessions.begin() as db: - record = await db.get(HomeInformation, (user, key)) + record = await cached_record(request, db, user, module, offset, key) current_config = await db.get(AISettings, 1) + if record is None: + raise HTTPException(409, "公告内存缓存已失效,请刷新后重新生成") _, current_version = await analysis_sources(request, user, module, record) if current_version != version or config_version(current_config) != fingerprint: raise HTTPException(409, "生成期间来源或模型配置已变化,已保留上次解读,请重新生成") record.analysis = dict(**interpretation, source_version=version, config_version=fingerprint, model=connection.model, generated_at=now().isoformat()) + if module == "messages": + request.app.state.home_message_cache[user].analyses[offset] = record.analysis return await output(request, user, module, record) diff --git a/backend/app/main.py b/backend/app/main.py index 3bd33f9..942d9b2 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -138,6 +138,7 @@ def create_app(settings=None, wq_client=None, ai_model_factory=None): app.state.settings = settings app.state.ai = ai_runtime app.state.home_information_locks = {} + app.state.home_message_cache = {} app.state.research = research_runtime app.state.mcp = mcp_runtime login_failures = defaultdict(list) diff --git a/backend/migrations/versions/0017_remove_message_cache.py b/backend/migrations/versions/0017_remove_message_cache.py new file mode 100644 index 0000000..b01613f --- /dev/null +++ b/backend/migrations/versions/0017_remove_message_cache.py @@ -0,0 +1,18 @@ +"""Remove persisted announcement sources and interpretations; use memory only.""" +import sqlalchemy as sa +from alembic import op + +revision = "0017" +down_revision = "0016" +branch_labels = None +depends_on = None + + +def upgrade(): + cache = sa.table("home_information", sa.column("resource", sa.String())) + op.execute(cache.delete().where(cache.c.resource.like("messages:%"))) + + +def downgrade(): + # Platform cache is reproducible; deleted text and interpretations are not restored. + pass diff --git a/backend/tests/test_home_information.py b/backend/tests/test_home_information.py index 662da08..3fd6a1b 100644 --- a/backend/tests/test_home_information.py +++ b/backend/tests/test_home_information.py @@ -35,6 +35,7 @@ async def setup(app, monkeypatch): async with app.state.sessions.begin() as db: account = await db.get(Account, 1) account.connection_status, account.wq_user_id = "connected", "TEST_USER" + monkeypatch.setattr(home, "now", lambda: datetime.fromisoformat("2026-09-12T12:00:00+00:00")) calls, state = [], {"rank": 25, "fail": set(), "agreement": "Participants must use GLOBAL region. Delay must be 1."} async def get(path, params=None, headers=None): @@ -43,7 +44,7 @@ async def setup(app, monkeypatch): raise WqError("资源暂不可用") if path == "/users/self/messages": offset = params["offset"] - return page([{"id": f"msg{offset}", "title": "Update", "description": "
News
bad", "dateCreated": "2026-09-11T12:00:00-04:00"}], + return page([{"id": f"msg{i}", "title": "Update", "description": "News
bad", "dateCreated": "2026-09-11T12:00:00-04:00"} for i in (range(10) if offset == 0 else [10])], "https://api.worldquantbrain.com/users/self/messages?offset=1" if offset == 0 else None) if path == "/consultant/boards/leader": assert params == {"user": "TEST_USER"} @@ -64,12 +65,12 @@ async def setup(app, monkeypatch): async def test_cache_pagination_failure_and_account_isolation(app, logged_in, monkeypatch): calls, state = await setup(app, monkeypatch) first = (await logged_in.get(PREFIX + "/messages")).json() - assert first["content"]["next_offset"] == 1 + assert first["content"]["next_offset"] == 10 assert "secret()" not in str(first) and "javascript:" not in str(first) assert first["analysis"] is None and not first["can_generate"] - assert (await logged_in.get(PREFIX + "/messages")).json() == first and len(calls) == 1 - second = (await logged_in.get(PREFIX + "/messages?offset=1")).json() - assert second["content"]["items"][0]["id"] == "msg1" + assert (await logged_in.get(PREFIX + "/messages")).json() == first and len(calls) == 2 + second = (await logged_in.get(PREFIX + "/messages?offset=10")).json() + assert second["content"]["items"][0]["id"] == "msg10" state["fail"].add("/users/self/messages") failed = (await logged_in.post(PREFIX + "/messages/refresh")).json() assert failed["stale"] and failed["error"] @@ -193,7 +194,7 @@ async def test_failed_grounding_preserves_prior_analysis(app, logged_in, monkeyp async def test_concurrent_first_reads_are_coalesced(app, logged_in, monkeypatch): calls, _ = await setup(app, monkeypatch) results = await asyncio.gather(*[logged_in.get(PREFIX + "/messages") for _ in range(4)]) - assert all(r.status_code == 200 for r in results) and len(calls) == 1 + assert all(r.status_code == 200 for r in results) and len(calls) == 2 async def test_event_sources_include_competitions_and_current_day(app, logged_in, monkeypatch): @@ -298,3 +299,93 @@ async def test_shared_connection_change_invalidates_ai(app, logged_in, monkeypat setattr(await db.get(AISettings, 1), field, value) assert (await logged_in.get(PREFIX + '/competition' + QUERY)).json()['analysis']['outdated'] assert calls == ['basic-model'] + + +@pytest.mark.parametrize('at,expected', [ + ('2026-03-31T12:00:00-04:00', '2026-02-28T12:00:00-05:00'), + ('2024-03-31T12:00:00-04:00', '2024-02-29T12:00:00-05:00'), + ('2026-01-31T12:00:00-05:00', '2025-12-31T12:00:00-05:00'), +]) +def test_recent_month_uses_eastern_calendar_month(monkeypatch, at, expected): + monkeypatch.setattr(home, 'now', lambda: datetime.fromisoformat(at)) + assert home.month_start().isoformat() == expected + + +async def test_recent_messages_scan_pages_filter_unknown_future_and_boundary(monkeypatch): + monkeypatch.setattr(home, 'now', lambda: datetime.fromisoformat('2026-03-31T12:00:00-04:00')) + calls = [] + class Client: + async def get(self, path, params): + calls.append(params) + if params['offset'] == 0: + return page([{'id': 'old', 'dateCreated': '2026-02-28T11:59:59-05:00'}, + {'id': 'unknown'}, {'id': 'future', 'dateCreated': '2026-04-01T00:00:00Z'}], + 'https://api.worldquantbrain.com/users/self/messages?offset=3&limit=100') + return page([{'id': 'boundary', 'dateCreated': '2026-02-28T12:00:00-05:00'}, + {'id': 'today', 'dateCreated': '2026-03-31T12:00:00-04:00'}]) + data = await home.source(Client(), 'messages', 'TEST_USER', 0, '') + assert [item['id'] for item in data['items']] == ['today', 'boundary'] + assert calls == [{'offset': 0, 'limit': 100}, {'offset': 3, 'limit': 100}] + + +async def test_messages_and_ai_never_persist_and_expire_without_auto_generation(app, logged_in, monkeypatch): + platform_calls, _ = await setup(app, monkeypatch) + await model_setup(app) + model_calls = [] + def respond(messages, info): + return ModelResponse(parts=[ToolCallPart(info.output_tools[0].name, {'items': [], 'suggestions': []})]) + @asynccontextmanager + async def factory(config, settings): + model_calls.append(config.model) + yield FunctionModel(respond) + app.state.ai.model_factory = factory + # Legacy rows must never be used; migration will remove them on deployment. + async with app.state.sessions.begin() as db: + db.add(HomeInformation(user_id='TEST_USER', resource='messages:0', content={'items': [{'title': 'legacy'}]})) + first = (await logged_in.get(PREFIX + '/messages')).json() + assert first['ephemeral'] and first['content']['total'] == 11 + generated = await logged_in.post(PREFIX + '/messages/generate') + assert generated.status_code == 200, generated.text + assert generated.json()['analysis']['model'] == 'basic-model' + await logged_in.get(PREFIX + '/messages') + await logged_in.get(PREFIX + '/messages?offset=10') + assert len(platform_calls) == 2 and model_calls == ['basic-model'] + async with app.state.sessions() as db: + saved = (await db.scalars(sa.select(HomeInformation))).all() + assert len(saved) == 1 and saved[0].content == {'items': [{'title': 'legacy'}]} + assert saved[0].analysis is None + monkeypatch.setattr(home, 'now', lambda: datetime.fromisoformat('2026-09-12T12:15:00+00:00')) + assert (await logged_in.post(PREFIX + '/messages/generate')).status_code == 409 + reloaded = (await logged_in.get(PREFIX + '/messages')).json() + assert reloaded['analysis'] is None and len(platform_calls) == 4 + assert model_calls == ['basic-model'] + + +async def test_message_window_rechecks_memory_and_removes_expired_interpretation(app, logged_in, monkeypatch): + await setup(app, monkeypatch) + monkeypatch.setattr(home, 'now', lambda: datetime.fromisoformat('2026-09-12T12:00:00+00:00')) + async def upstream(path, params=None): + return page([{'id': 'boundary', 'dateCreated': '2026-08-12T08:00:00-04:00'}]) + monkeypatch.setattr(app.state.runner.client, 'get', upstream) + assert (await logged_in.get(PREFIX + '/messages')).json()['content']['total'] == 1 + app.state.home_message_cache['TEST_USER'].analyses[0] = {'old': 'interpretation'} + monkeypatch.setattr(home, 'now', lambda: datetime.fromisoformat('2026-09-12T12:00:01+00:00')) + result = (await logged_in.get(PREFIX + '/messages')).json() + assert result['content']['total'] == 0 and result['analysis'] is None + + +def test_message_cleanup_migration_only_removes_message_cache(tmp_path): + path = Path(__file__).parents[1] / 'migrations/versions/0017_remove_message_cache.py' + spec = importlib.util.spec_from_file_location('message_cleanup', path) + migration = importlib.util.module_from_spec(spec) + spec.loader.exec_module(migration) + engine = sa.create_engine(f'sqlite:///{tmp_path}/cleanup.db') + with engine.begin() as conn: + conn.exec_driver_sql('CREATE TABLE home_information (user_id TEXT, resource TEXT, content TEXT, analysis TEXT)') + for user, resource in [('A', 'messages:0'), ('B', 'messages:10'), ('A', 'events:'), ('A', 'competition:ARC2026')]: + conn.exec_driver_sql('INSERT INTO home_information VALUES (?, ?, ?, ?)', (user, resource, 'source', 'ai')) + with Operations.context(MigrationContext.configure(conn)): + migration.upgrade() + migration.upgrade() + assert conn.exec_driver_sql('SELECT resource FROM home_information ORDER BY resource').scalars().all() == ['competition:ARC2026', 'events:'] + engine.dispose() diff --git a/frontend/src/pages/HomeInformation.tsx b/frontend/src/pages/HomeInformation.tsx index 6a06e5c..01e8244 100644 --- a/frontend/src/pages/HomeInformation.tsx +++ b/frontend/src/pages/HomeInformation.tsx @@ -71,6 +71,7 @@ type Analysis = { outdated: boolean; }; type Snapshot = { + ephemeral?: boolean; content: Content | null; fetched_at: string | null; error: string | null; @@ -243,7 +244,9 @@ function Interpretation({ state, zone }: { state: Resource; zone: string }) { > ) : (- 手动生成并保存,重新打开首页不会调用模型。 + {state.data?.ephemeral + ? "解读仅暂存在内存,不写入数据库;重启或缓存到期后清除。" + : "手动生成并保存,重新打开首页不会调用模型。"}
)} @@ -348,7 +351,9 @@ function MessagePage({ state={state} zone={zone} > -来自平台消息的公告、更新及社区资讯。
++ 仅展示最近一个月的公告、更新及社区资讯(按美东时间筛选),不写入数据库。 +
{content?.items?.length ? (