diff --git a/backend/alembic/versions/20261013_d41b7c2e9a10_source_history.py b/backend/alembic/versions/20261013_d41b7c2e9a10_source_history.py new file mode 100644 index 0000000..e653974 --- /dev/null +++ b/backend/alembic/versions/20261013_d41b7c2e9a10_source_history.py @@ -0,0 +1,32 @@ +"""source history backfill + +Revision ID: d41b7c2e9a10 +Revises: bf0ffbfcaf9a +Create Date: 2026-10-13 15:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'd41b7c2e9a10' +down_revision: Union[str, Sequence[str], None] = 'bf0ffbfcaf9a' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + op.add_column('sources', sa.Column('history_target', sa.DateTime(timezone=True), nullable=True)) + op.add_column('sources', sa.Column('history_position', sa.DateTime(timezone=True), nullable=True)) + op.add_column('sources', sa.Column('history_done', sa.Boolean(), server_default='false', nullable=False)) + + +def downgrade() -> None: + """Downgrade schema.""" + op.drop_column('sources', 'history_done') + op.drop_column('sources', 'history_position') + op.drop_column('sources', 'history_target') diff --git a/backend/src/alphapulse/api/crawler.py b/backend/src/alphapulse/api/crawler.py index 8d3094d..5b05a5a 100644 --- a/backend/src/alphapulse/api/crawler.py +++ b/backend/src/alphapulse/api/crawler.py @@ -7,14 +7,14 @@ from zoneinfo import ZoneInfo from fastapi import APIRouter, Depends, HTTPException, Query, status from fastapi.sse import EventSourceResponse, ServerSentEvent -from pydantic import BaseModel, Field +from pydantic import AwareDatetime, BaseModel, Field from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from alphapulse.core.auth import require_admin from alphapulse.core.db import get_session from alphapulse.core.redis import redis -from alphapulse.crawler.events import CONTROL_CHANNEL, LOG_CHANNEL, STATUS_KEY +from alphapulse.crawler.events import CONTROL_CHANNEL, HISTORY_SOURCES, LOG_CHANNEL, STATUS_KEY from alphapulse.models import CrawlLog, CrawlRun, NewsItem, Source router = APIRouter(prefix="/api/crawler", tags=["crawler"], dependencies=[Depends(require_admin)]) @@ -41,6 +41,10 @@ class SourceOut(BaseModel): last_success_at: datetime | None last_error: str | None paused_until: datetime | None + supports_history: bool + history_target: datetime | None + history_position: datetime | None + history_done: bool # 管理进程未运行时为 None process: ProcessStatus | None today_new: int @@ -57,6 +61,8 @@ class SourcesOut(BaseModel): class SourcePatch(BaseModel): enabled: bool | None = None interval_seconds: Annotated[int | None, Field(ge=5, le=86400)] = None + # 设为时间表示开始或继续回补到该时间；显式传 null 表示取消 + history_target: AwareDatetime | None = None class RunOut(BaseModel): @@ -110,6 +116,7 @@ async def list_sources(session: Annotated[AsyncSession, Depends(get_session)]) - sources=[ SourceOut.model_validate({ **s.__dict__, + "supports_history": s.code in HISTORY_SOURCES, "process": json.loads(statuses[s.code]) if s.code in statuses else None, "today_new": today_counts.get(s.code, (0, 0))[0], "today_revised": today_counts.get(s.code, (0, 0))[1], @@ -125,8 +132,16 @@ async def patch_source(code: str, body: SourcePatch, session: Annotated[AsyncSes source = await session.get(Source, code) if source is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "来源不存在") - for field, value in body.model_dump(exclude_none=True).items(): + for field, value in body.model_dump(exclude_none=True, exclude={"history_target"}).items(): setattr(source, field, value) + if "history_target" in body.model_fields_set: + if code not in HISTORY_SOURCES: + raise HTTPException(status.HTTP_400_BAD_REQUEST, "该来源的接口不支持历史翻页，无法回补") + if body.history_target is not None and body.history_target >= datetime.now(LOCAL_TZ): + raise HTTPException(status.HTTP_422_UNPROCESSABLE_CONTENT, "回补目标时间必须早于现在") + # 保留断点：改目标或重新开始时从已回补的位置继续 + source.history_target = body.history_target + source.history_done = False await session.commit() diff --git a/backend/src/alphapulse/crawler/events.py b/backend/src/alphapulse/crawler/events.py index d8256dc..92a090e 100644 --- a/backend/src/alphapulse/crawler/events.py +++ b/backend/src/alphapulse/crawler/events.py @@ -11,6 +11,8 @@ LOG_CHANNEL = f"{KEY_PREFIX}pub:logs" NEWS_CHANNEL = f"{KEY_PREFIX}pub:news" CONTROL_CHANNEL = f"{KEY_PREFIX}ctl:crawler" STATUS_KEY = f"{KEY_PREFIX}crawler:status" +# 列表接口支持按时间点往前翻的来源；cls_depth 列表只能翻约 3 页，无法回补历史 +HISTORY_SOURCES = ("cls_telegraph", "wscn_live", "wscn_news") def start_urls_key(source_code: str) -> str: diff --git a/backend/src/alphapulse/crawler/manager.py b/backend/src/alphapulse/crawler/manager.py index 0f8c9f5..97c3d85 100644 --- a/backend/src/alphapulse/crawler/manager.py +++ b/backend/src/alphapulse/crawler/manager.py @@ -159,7 +159,12 @@ class Manager: continue # 子进程还没取走上一条触发时不再堆积 if await redis.llen(start_urls_key(s.code)) == 0: - await redis.lpush(start_urls_key(s.code), json.dumps({"trigger": w.trigger, "backfill_hours": s.backfill_hours})) + tick = {"trigger": w.trigger, "backfill_hours": s.backfill_hours} + if s.history_target and not s.history_done: + # 回补分段夹在常规轮次里执行，进度由子进程在本轮成功后写回 + position = s.history_position.isoformat() if s.history_position else None + tick["history"] = {"target": s.history_target.isoformat(), "position": position} + await redis.lpush(start_urls_key(s.code), json.dumps(tick)) w.trigger = "schedule" w.next_run_at = now + timedelta(seconds=s.interval_seconds) diff --git a/backend/src/alphapulse/crawler/spiders/base.py b/backend/src/alphapulse/crawler/spiders/base.py index b8932ea..97746d2 100644 --- a/backend/src/alphapulse/crawler/spiders/base.py +++ b/backend/src/alphapulse/crawler/spiders/base.py @@ -2,11 +2,13 @@ import asyncio import json import logging import re +import time import uuid from collections.abc import Callable, Coroutine, Iterable from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from typing import Any +from zoneinfo import ZoneInfo from scrapy import Request, signals from scrapy.crawler import Crawler @@ -24,10 +26,14 @@ from alphapulse.models import CrawlRun, NewsItem, Source # 日志落库失败时只输出到 stdout，避免被自己的处理器再次捕获 sink_logger = logging.getLogger("alphapulse.logsink") +# 日志里的时间按北京时间显示 +LOCAL_TZ = ZoneInfo("Asia/Shanghai") MAX_PAGES = 200 BACKOFF_BASE_SECONDS = 30 BACKOFF_MAX_SECONDS = 15 * 60 +# 每轮用于历史回补的时长上限：回补和实时采集共用一个子进程，不能拖住实时轮次 +HISTORY_RUN_SECONDS = 20 def ts(epoch: int) -> datetime: @@ -54,6 +60,11 @@ class RunState: errors: list[str] = field(default_factory=list) # 遇到 403/429 后本轮不再发新请求 blocked: bool = False + # 历史回补：本轮推进到的最早时间；仅在本轮成功结束时写回 sources + history_target: datetime | None = None + history_position: datetime | None = None + history_done: bool = False + history_deadline: float = 0 @property def full_scan(self) -> bool: @@ -91,6 +102,8 @@ class NewsSpider(RedisSpider): self.run: RunState | None = None self.log_buffer: list[tuple[str, uuid.UUID | None, str, str]] = [] self._tasks: set[asyncio.Task] = set() + # 上一轮成功后的回补进度 (目标, 断点, 是否结束)。写库是异步的，管理进程推来的下一条触发可能还带着旧断点 + self.history_checkpoint: tuple[datetime, datetime, bool] | None = None @classmethod def from_crawler(cls, crawler: Crawler, *args: Any, **kwargs: Any) -> "NewsSpider": @@ -106,6 +119,10 @@ class NewsSpider(RedisSpider): def first_requests(self) -> Iterable[Request]: raise NotImplementedError + def history_request(self, before: datetime) -> Request: + """早于 before 的第一页列表请求，meta 需带 history=True；仅 HISTORY_SOURCES 中的来源实现""" + raise NotImplementedError + # ---------- 运行生命周期 ---------- async def start(self): @@ -123,8 +140,26 @@ class NewsSpider(RedisSpider): now = datetime.now(UTC) self.run = RunState(uuid.uuid4(), tick["trigger"], now, now - timedelta(hours=tick["backfill_hours"])) self._spawn(self._insert_run(self.run)) - self.logger.info("开始采集（触发方式：%s）", tick["trigger"]) - return list(self.first_requests()) + requests = list(self.first_requests()) + history = tick.get("history") + if history: + target = datetime.fromisoformat(history["target"]) + # 断点为空表示首次回补：从现在开始往前翻，顺带补上中间可能的缺口 + position = datetime.fromisoformat(history["position"]) if history["position"] else now + checkpoint = self.history_checkpoint + if checkpoint and checkpoint[0] == target and checkpoint[1] <= position: + position = checkpoint[1] + if checkpoint[2]: + history = None + if history: + self.run.history_target = target + self.run.history_position = position + self.run.history_deadline = time.monotonic() + HISTORY_RUN_SECONDS + requests.append(self.history_request(position)) + self.logger.info("开始采集（触发方式：%s，历史回补自 %s 往前）", tick["trigger"], position.astimezone(LOCAL_TZ).strftime("%Y-%m-%d %H:%M")) + else: + self.logger.info("开始采集（触发方式：%s）", tick["trigger"]) + return requests def _on_opened(self) -> None: logging.getLogger().addHandler(DbLogHandler(self)) @@ -139,6 +174,8 @@ class NewsSpider(RedisSpider): def _close_run(self) -> None: run, self.run = self.run, None + if run.history_target is not None and not run.errors: + self.history_checkpoint = (run.history_target, run.history_position, run.history_done) self._spawn(self._finish_run(run)) def _on_error(self, failure: Failure, response: Response) -> None: @@ -183,6 +220,11 @@ class NewsSpider(RedisSpider): .on_conflict_do_update(index_elements=[CrawlRun.id], set_=values) ) source = await session.get(Source, self.name, with_for_update=True) + # 本轮失败不推进断点，下一轮从原位置重做（写入幂等）；目标期间被修改或取消时丢弃本轮进度 + history_saved = not failed and run.history_target is not None and source.history_target == run.history_target + if history_saved: + source.history_position = run.history_position + source.history_done = run.history_done if failed: source.consecutive_failures += 1 delay = min(BACKOFF_BASE_SECONDS * 2 ** (source.consecutive_failures - 1), BACKOFF_MAX_SECONDS) @@ -201,6 +243,14 @@ class NewsSpider(RedisSpider): self.logger.error("本轮失败：%s；错误 %d 个，暂停 %ds", summary, len(run.errors), delay) else: self.logger.info("本轮完成：%s", summary) + if history_saved: + position = run.history_position.astimezone(LOCAL_TZ).strftime("%Y-%m-%d %H:%M") + if not run.history_done: + self.logger.info("历史回补已到 %s", position) + elif run.history_position <= run.history_target: + self.logger.info("历史回补完成，已到 %s", position) + else: + self.logger.warning("历史回补结束：来源最早只能翻到 %s", position) async def _flush_logs(self) -> None: rows, self.log_buffer = self.log_buffer, [] @@ -253,6 +303,19 @@ class NewsSpider(RedisSpider): and (self.run.full_scan or any(external_id not in known for external_id in external_ids)) ) + def history_step(self, oldest: datetime | None) -> bool: + """处理完一页回补列表后调用：推进本轮断点，返回是否继续翻下一页。oldest 为本页最早时间，空页传 None""" + run = self.run + if oldest is None or oldest >= run.history_position: + # 空页或没有更早的数据（部分接口越过下限会返回最新数据）：来源已经翻到底 + run.history_done = True + return False + run.history_position = oldest + if oldest <= run.history_target: + run.history_done = True + return False + return not run.blocked and time.monotonic() < run.history_deadline + def item( self, response: Response, diff --git a/backend/src/alphapulse/crawler/spiders/cls.py b/backend/src/alphapulse/crawler/spiders/cls.py index 56ed4f5..368eab3 100644 --- a/backend/src/alphapulse/crawler/spiders/cls.py +++ b/backend/src/alphapulse/crawler/spiders/cls.py @@ -2,6 +2,7 @@ import hashlib import html import json import re +from datetime import datetime from urllib.parse import urlencode from scrapy import Request @@ -26,6 +27,13 @@ class ClsTelegraphSpider(NewsSpider): def first_requests(self) -> list[Request]: return [self.request(signed_url("/v1/roll/get_roll_list", {"category": "", "last_time": "", "rn": "20"}), self.parse_list, page=1)] + def list_request(self, last_time: int, page: int, **meta) -> Request: + params = {"category": "", "last_time": str(last_time), "refresh_type": "1", "rn": "20"} + return self.request(signed_url("/v1/roll/get_roll_list", params), self.parse_list, page=page, **meta) + + def history_request(self, before: datetime) -> Request: + return self.list_request(int(before.timestamp()), 1, history=True) + async def parse_list(self, response: Response): body = response.json() if body["errno"] != 0: @@ -49,10 +57,13 @@ class ClsTelegraphSpider(NewsSpider): extra={"level": x.get("level"), "stock_list": x.get("stock_list")}, ) page = response.meta["page"] - if rows and self.should_continue(ids, known, ts(min(x["ctime"] for x in rows)), page): - last_time = str(min(x["ctime"] for x in rows)) - params = {"category": "", "last_time": last_time, "refresh_type": "1", "rn": "20"} - yield self.request(signed_url("/v1/roll/get_roll_list", params), self.parse_list, page=page + 1) + last_time = min(x["ctime"] for x in rows) if rows else None + oldest = ts(last_time) if rows else None + if response.meta.get("history"): + if self.history_step(oldest): + yield self.list_request(last_time, page + 1, history=True) + elif rows and self.should_continue(ids, known, oldest, page): + yield self.list_request(last_time, page + 1) class ClsDepthSpider(NewsSpider): diff --git a/backend/src/alphapulse/crawler/spiders/wscn.py b/backend/src/alphapulse/crawler/spiders/wscn.py index 184a36f..4c514aa 100644 --- a/backend/src/alphapulse/crawler/spiders/wscn.py +++ b/backend/src/alphapulse/crawler/spiders/wscn.py @@ -1,3 +1,7 @@ +import base64 +import json +from datetime import datetime + from scrapy import Request from scrapy.http import Response @@ -14,8 +18,13 @@ class WscnNewsSpider(NewsSpider): def first_requests(self) -> list[Request]: return [self.list_request("", 1)] - def list_request(self, cursor: str, page: int) -> Request: - return self.request(f"{API}/information-flow?channel=global&accept=article&limit=20&cursor={cursor}", self.parse_list, page=page) + def list_request(self, cursor: str, page: int, **meta) -> Request: + return self.request(f"{API}/information-flow?channel=global&accept=article&limit=20&cursor={cursor}", self.parse_list, page=page, **meta) + + def history_request(self, before: datetime) -> Request: + # 信息流 cursor 是 base64 的 JSON，ArticleLe 为文章时间上界（秒） + cursor = json.dumps({"SlotOffset": 0, "TotalCount": 0, "ArticleLe": int(before.timestamp())}, separators=(",", ":")) + return self.list_request(base64.b64encode(cursor.encode()).decode(), 1, history=True) async def parse_list(self, response: Response): data = response.json()["data"] @@ -27,7 +36,11 @@ class WscnNewsSpider(NewsSpider): if known.get(str(r["id"])) != (r["title"], r["content_short"]): yield self.request(f"{API}/articles/{r['id']}?extract=0", self.parse_detail, summary=r["content_short"]) page = response.meta["page"] - if rows and self.should_continue(ids, known, ts(min(r["display_time"] for r in rows)), page): + oldest = ts(min(r["display_time"] for r in rows)) if rows else None + if response.meta.get("history"): + if self.history_step(oldest): + yield self.list_request(data["next_cursor"], page + 1, history=True) + elif rows and self.should_continue(ids, known, oldest, page): yield self.list_request(data["next_cursor"], page + 1) async def parse_detail(self, response: Response): @@ -54,8 +67,12 @@ class WscnLiveSpider(NewsSpider): def first_requests(self) -> list[Request]: return [self.list_request("", 1)] - def list_request(self, cursor: str, page: int) -> Request: - return self.request(f"{API}/lives?channel=global-channel&limit=50&cursor={cursor}", self.parse_list, page=page) + def list_request(self, cursor: str, page: int, **meta) -> Request: + return self.request(f"{API}/lives?channel=global-channel&limit=50&cursor={cursor}", self.parse_list, page=page, **meta) + + def history_request(self, before: datetime) -> Request: + # cursor 为时间上界（秒）；接口只保留约最近 180 天，更早返回空 + return self.list_request(str(int(before.timestamp())), 1, history=True) async def parse_list(self, response: Response): data = response.json()["data"] @@ -77,5 +94,9 @@ class WscnLiveSpider(NewsSpider): extra={"score": x.get("score"), "symbols": x.get("symbols")}, ) page = response.meta["page"] - if rows and self.should_continue(ids, known, ts(min(x["display_time"] for x in rows)), page): + oldest = ts(min(x["display_time"] for x in rows)) if rows else None + if response.meta.get("history"): + if self.history_step(oldest): + yield self.list_request(data["next_cursor"], page + 1, history=True) + elif rows and self.should_continue(ids, known, oldest, page): yield self.list_request(data["next_cursor"], page + 1) diff --git a/backend/src/alphapulse/models/news.py b/backend/src/alphapulse/models/news.py index e2da0d5..7a42367 100644 --- a/backend/src/alphapulse/models/news.py +++ b/backend/src/alphapulse/models/news.py @@ -21,6 +21,10 @@ class Source(Base): last_success_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) last_error: Mapped[str | None] = mapped_column(Text) paused_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + # 历史回补：目标时间（为空表示不回补）、已回补到的最早时间（断点）、是否已结束（到达目标或来源无更早数据） + history_target: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + history_position: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + history_done: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") class NewsItem(Base): diff --git a/web/src/apis/crawler.ts b/web/src/apis/crawler.ts index b230c92..36a4bbd 100644 --- a/web/src/apis/crawler.ts +++ b/web/src/apis/crawler.ts @@ -18,6 +18,10 @@ export interface SourceInfo { last_success_at: string | null last_error: string | null paused_until: string | null + supports_history: boolean + history_target: string | null + history_position: string | null + history_done: boolean process: ProcessStatus | null today_new: number today_revised: number @@ -43,7 +47,14 @@ export function listSources(): Promise { return http.get("/crawler/sources") } -export function patchSource(code: string, body: { enabled?: boolean, interval_seconds?: number }): Promise { +export interface SourcePatch { + enabled?: boolean + interval_seconds?: number + /** ISO 时间表示回补到该时间；null 表示取消回补 */ + history_target?: string | null +} + +export function patchSource(code: string, body: SourcePatch): Promise { return http.patch(`/crawler/sources/${code}`, body) } diff --git a/web/src/pages/crawler/components/SourcePanel.vue b/web/src/pages/crawler/components/SourcePanel.vue index 325d5aa..0ca2d42 100644 --- a/web/src/pages/crawler/components/SourcePanel.vue +++ b/web/src/pages/crawler/components/SourcePanel.vue @@ -1,6 +1,7 @@