diff --git a/.env.example b/.env.example index 1d92b7b..8c99817 100644 --- a/.env.example +++ b/.env.example @@ -4,7 +4,7 @@ SIGNALSCOUT_DATABASE_URL=mysql+pymysql://signalscout:signalscout@127.0.0.1:3306/ SIGNALSCOUT_LLM_BASE_URL=https://api.zayuapi.com/v1 SIGNALSCOUT_LLM_API_KEY= SIGNALSCOUT_LLM_MODEL=grok-4.3 -SIGNALSCOUT_LLM_TIMEOUT_SECONDS=90 +SIGNALSCOUT_LLM_TIMEOUT_SECONDS=180 SIGNALSCOUT_LLM_MAX_TOKENS=8000 SIGNALSCOUT_NEWS_RECENT_DAYS=3 SIGNALSCOUT_NEWS_MAX_ITEMS=60 diff --git a/README.md b/README.md index 1eb1436..e8114a7 100644 --- a/README.md +++ b/README.md @@ -1,13 +1,13 @@ # SignalScout -SignalScout 是一个自动运行的 AI 情报 Agent。它先用动态新闻搜索和 GitHub 官方搜索抓取候选证据，再使用 Grok 通过中转 API 生成最新 AI 新闻、模型发布、Agent 工具、融资动态和 GitHub 热门项目日报，并把结构化情报保存到 MySQL。 +SignalScout 是一个自动运行的 AI 情报 Agent。它使用 Grok Web Search 联网搜索中英文 AI 新闻，使用 GitHub 官方搜索抓取开源项目候选，再由 Grok 生成最新 AI 新闻、模型发布、Agent 工具、融资动态和 GitHub 热门项目日报，并把每次运行的结构化情报快照保存到 MySQL。 ## 主流程 ```text Scheduler -> AgentRunner - -> News Search API 采集近期 AI 新闻候选 + -> Grok Web Search 采集中英文 AI 新闻候选 -> GitHub Search API 采集近期热门 AI 项目候选 -> Grok 研判新闻候选与 GitHub 候选 -> MySQL signals/reports @@ -29,6 +29,7 @@ SIGNALSCOUT_DATABASE_URL=mysql+pymysql://user:password@host:3306/signalscout?cha SIGNALSCOUT_LLM_BASE_URL=https://api.zayuapi.com/v1 SIGNALSCOUT_LLM_API_KEY=你的中转站 key SIGNALSCOUT_LLM_MODEL=grok-4.3 +SIGNALSCOUT_LLM_TIMEOUT_SECONDS=180 SIGNALSCOUT_LLM_MAX_TOKENS=8000 SIGNALSCOUT_NEWS_RECENT_DAYS=3 SIGNALSCOUT_NEWS_MAX_ITEMS=60 diff --git a/app/agent/runner.py b/app/agent/runner.py index 15e0d26..dfddcb3 100644 --- a/app/agent/runner.py +++ b/app/agent/runner.py @@ -13,7 +13,7 @@ from app.db.repository import AgentRepository from app.integrations.github_client import GitHubHotProjectClient from app.integrations.news_search_client import NewsSearchClient from app.llm.grok_client import GrokIntelligenceClient -from app.schemas.agent import RunResponse +from app.schemas.agent import RunResponse, SignalItem logger = logging.getLogger(__name__) @@ -49,7 +49,12 @@ class AgentRunner: github_projects=github_projects, run_date=current_date, ) - signals = self.repo.save_signals(run=run, items=result.signals) + signal_items = self._compose_run_signals( + news_candidates=news_candidates, + github_projects=github_projects, + model_signals=result.signals, + ) + signals = self.repo.save_signals(run=run, items=signal_items) report_path = self.writer.write_daily_report(current_date, result.report_markdown) self.repo.save_report( run=run, @@ -64,6 +69,8 @@ class AgentRunner: "topic_count": len(topics), "news_candidate_count": len(news_candidates), "github_project_count": len(github_projects), + "model_signal_count": len(result.signals), + "saved_signal_count": len(signal_items), "model": self.grok.settings.llm_model, "model_response": result.raw_response, }, @@ -81,3 +88,29 @@ class AgentRunner: self.db.commit() logger.exception("agent_run_failed run_id=%s error=%s", run.id, exc) raise + + def _compose_run_signals( + self, + news_candidates: list[SignalItem], + github_projects: list[SignalItem], + model_signals: list[SignalItem], + ) -> list[SignalItem]: + # 入库清单以采集候选为准，模型同链接结果只增强摘要和重要度，避免日报模型漏选导致信号消失。 + model_by_url = {item.source_url: item for item in model_signals} + composed = [] + for candidate in [*news_candidates, *github_projects]: + model_item = model_by_url.get(candidate.source_url) + if model_item is None: + composed.append(candidate) + continue + composed.append( + candidate.model_copy( + update={ + "topic": model_item.topic, + "summary": model_item.summary, + "importance": model_item.importance, + "entities": model_item.entities or candidate.entities, + } + ) + ) + return composed diff --git a/app/api/routes.py b/app/api/routes.py index f80c911..e27674d 100644 --- a/app/api/routes.py +++ b/app/api/routes.py @@ -2,11 +2,12 @@ from __future__ import annotations from typing import Literal, Optional -from fastapi import APIRouter, Depends +from fastapi import APIRouter, BackgroundTasks, Depends from fastapi.responses import HTMLResponse from sqlalchemy import select from sqlalchemy.orm import Session +from app.agent.scheduler import run_scheduled_job from app.core.config import Settings, get_settings from app.db.models import AgentRun, Report, Signal, Topic from app.db.session import get_db @@ -14,6 +15,7 @@ from app.schemas.agent import ( DashboardRead, ReportRead, RunRead, + RunTriggerRead, SignalRead, TopicCreate, TopicRead, @@ -41,6 +43,22 @@ def list_runs(db: Session = Depends(get_db)) -> list[AgentRun]: return list(db.scalars(select(AgentRun).order_by(AgentRun.id.desc()).limit(20)).all()) +@router.post("/agent/runs/daily", response_model=RunTriggerRead) +def trigger_daily_run( + background_tasks: BackgroundTasks, + settings: Settings = Depends(get_settings), + db: Session = Depends(get_db), +) -> RunTriggerRead: + # 手动搜索复用定时任务入口，避免测试按钮和自动调度走出两套采集逻辑。 + running = db.scalar( + select(AgentRun).where(AgentRun.status == "running").order_by(AgentRun.id.desc()).limit(1) + ) + if running is not None: + return RunTriggerRead(status="running", run_id=running.id) + background_tasks.add_task(run_scheduled_job, settings) + return RunTriggerRead(status="started") + + @router.get("/signals", response_model=list[SignalRead]) def list_signals( limit: int = 100, diff --git a/app/core/config.py b/app/core/config.py index 8a305fb..6d0667d 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -15,7 +15,7 @@ class Settings(BaseSettings): llm_base_url: str = "https://api.zayuapi.com/v1" llm_api_key: str = "" llm_model: str = "grok-4.3" - llm_timeout_seconds: int = 90 + llm_timeout_seconds: int = 180 llm_max_tokens: int = 8000 news_recent_days: int = 3 news_max_items: int = 60 diff --git a/app/db/models.py b/app/db/models.py index 412c641..ec382cc 100644 --- a/app/db/models.py +++ b/app/db/models.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import datetime from typing import Optional -from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, JSON, String, Text, func +from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, JSON, String, Text, UniqueConstraint, func from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship @@ -41,6 +41,9 @@ class AgentRun(Base): class Signal(Base): __tablename__ = "signals" + __table_args__ = ( + UniqueConstraint("run_id", "source_url_hash", name="uq_signals_run_source_url_hash"), + ) id: Mapped[int] = mapped_column(Integer, primary_key=True) run_id: Mapped[int] = mapped_column(ForeignKey("agent_runs.id"), nullable=False, index=True) @@ -49,7 +52,7 @@ class Signal(Base): title: Mapped[str] = mapped_column(String(300), nullable=False) summary: Mapped[str] = mapped_column(Text, nullable=False) source_url: Mapped[str] = mapped_column(Text, nullable=False) - source_url_hash: Mapped[str] = mapped_column(String(64), nullable=False, unique=True) + source_url_hash: Mapped[str] = mapped_column(String(64), nullable=False) source_name: Mapped[str] = mapped_column(String(200), nullable=False) published_at: Mapped[Optional[datetime]] = mapped_column(DateTime, nullable=True) importance: Mapped[int] = mapped_column(Integer, nullable=False) diff --git a/app/db/repository.py b/app/db/repository.py index 69e5aaa..84a9c60 100644 --- a/app/db/repository.py +++ b/app/db/repository.py @@ -1,8 +1,6 @@ from __future__ import annotations from hashlib import sha256 -from typing import Optional - from sqlalchemy import select from sqlalchemy.orm import Session @@ -41,21 +39,22 @@ class AgentRepository: self.db.flush() def save_signals(self, run: AgentRun, items: list[SignalItem]) -> list[Signal]: - # 相同来源链接只保存一次，避免日报和仪表盘重复出现同一条情报。 + # 去重范围限定在同一次运行内，保证每轮情报结果都能完整留档。 saved: list[Signal] = [] for item in items: signal = self.save_or_update_signal(run=run, item=item) - if signal is None: - continue saved.append(signal) return saved - def save_or_update_signal(self, run: AgentRun, item: SignalItem) -> Optional[Signal]: - # 同一来源链接只保留一条情报，流式接口和普通接口共用这条去重规则。 + def save_or_update_signal(self, run: AgentRun, item: SignalItem) -> Signal: + # 同一次运行里相同链接更新为一条，跨运行保留各自结果用于历史对比。 source_url_hash = self._source_url_hash(item.source_url) - existing = self.db.scalar(select(Signal).where(Signal.source_url_hash == source_url_hash)) - if existing and existing.run_id != run.id: - return None + existing = self.db.scalar( + select(Signal).where( + Signal.run_id == run.id, + Signal.source_url_hash == source_url_hash, + ) + ) if existing: existing.source_type = item.source_type existing.topic = item.topic diff --git a/app/integrations/news_search_client.py b/app/integrations/news_search_client.py index b3e4ea2..e7bafe2 100644 --- a/app/integrations/news_search_client.py +++ b/app/integrations/news_search_client.py @@ -2,114 +2,214 @@ from __future__ import annotations import json import logging -from datetime import date, datetime, time, timedelta +from datetime import date, datetime, timedelta from typing import Any, Optional from urllib.error import HTTPError, URLError -from urllib.parse import urlencode from urllib.request import Request, urlopen from app.core.config import Settings -from app.core.timezone import BEIJING_TZ, to_beijing_naive +from app.core.timezone import to_beijing_naive from app.schemas.agent import SignalItem logger = logging.getLogger(__name__) class NewsSearchClient: - """通过公开技术新闻检索接口采集动态AI新闻候选。""" - - endpoint = "https://hn.algolia.com/api/v1/search_by_date" + """通过Grok Web Search采集中英文AI新闻候选。""" def __init__(self, settings: Settings) -> None: self.settings = settings def collect_ai_news(self, run_date: date) -> list[SignalItem]: - # HN Algolia按时间倒序返回技术新闻候选，模型后续只从这些证据里挑选日报条目。 - since = run_date - timedelta(days=self.settings.news_recent_days) - stories: dict[str, dict[str, Any]] = {} - for query in self._queries(): - payload = self._search(query=query, since=since, until=run_date) - for story in payload.get("hits", []): - url = self._story_url(story) - object_id = story.get("objectID") - key = str(object_id or url) - if isinstance(url, str) and url.startswith("http"): - stories[key] = story - - candidates = [self._story_to_signal(story) for story in stories.values()] + # Grok负责联网搜索新闻候选，入库前仍统一校验来源链接和结构字段。 + items: list[dict[str, Any]] = [] + for query_group in self._query_groups(): + payload = self._search_with_grok(run_date=run_date, query_group=query_group) + items.extend(payload.get("items", [])) + candidates = [self._article_to_signal(item) for item in items] candidates = [candidate for candidate in candidates if candidate is not None] - candidates.sort(key=lambda item: item.published_at or datetime.min, reverse=True) - logger.info("news_candidates_collected days=%s items=%s", self.settings.news_recent_days, len(candidates)) - return candidates[: self.settings.news_max_items] + unique: dict[str, SignalItem] = {} + for candidate in candidates: + unique[candidate.source_url] = candidate + ranked = sorted( + unique.values(), + key=lambda item: (item.importance, item.published_at or datetime.min), + reverse=True, + ) + logger.info( + "news_candidates_collected source=grok_web_search days=%s items=%s", + self.settings.news_recent_days, + len(ranked), + ) + return ranked[: self.settings.news_max_items] - def _queries(self) -> list[str]: - # 查询词覆盖模型、Agent、开源AI和融资产品动态，避免单一关键词漏掉日报候选。 + def _query_groups(self) -> list[dict[str, Any]]: + # 多组搜索覆盖中文、英文、融资、模型和Agent工具，避免单次搜索只返回少量相似新闻。 return [ - "AI OR LLM OR agent", - "large language model", - "OpenAI Anthropic", - "AI startup funding", + { + "name": "中文AI新闻", + "language": "zh", + "queries": ["人工智能 大模型 智能体 AI 新闻", "AI 大模型 创业 融资 产品 发布"], + "target_count": 15, + }, + { + "name": "英文AI新闻", + "language": "en", + "queries": ["latest AI news LLM agents OpenAI Anthropic", "AI startup funding model release agent tools"], + "target_count": 15, + }, + { + "name": "模型与Agent发布", + "language": "mixed", + "queries": ["大模型 发布 Agent 工具 开源", "LLM model release AI agent framework open source"], + "target_count": 15, + }, + { + "name": "AI商业与融资", + "language": "mixed", + "queries": ["AI 融资 创业 公司 产品", "AI funding startup acquisition product launch"], + "target_count": 15, + }, ] - def _search(self, query: str, since: date, until: date) -> dict[str, Any]: - # search_by_date接口稳定返回最新故事，numericFilters把候选限制在本次日报窗口内。 - start_at = int(datetime.combine(since, time.min, tzinfo=BEIJING_TZ).timestamp()) - end_at = int(datetime.combine(until, time.max, tzinfo=BEIJING_TZ).timestamp()) - params = { - "query": query, - "tags": "story", - "hitsPerPage": str(self.settings.news_search_max_records), - "numericFilters": f"created_at_i>={start_at},created_at_i<={end_at}", - } - url = f"{self.endpoint}?{urlencode(params)}" - request = Request(url, headers={"User-Agent": "SignalScout"}) + def _search_with_grok(self, run_date: date, query_group: dict[str, Any]) -> dict[str, Any]: + # Responses API的web_search工具是当前新闻发现入口，确保候选来自真实网页搜索。 + if not self.settings.llm_api_key: + raise RuntimeError("SIGNALSCOUT_LLM_API_KEY is required") + since = run_date - timedelta(days=self.settings.news_recent_days) + body = json.dumps( + { + "model": self.settings.llm_model, + "input": [ + { + "role": "system", + "content": ( + "你是 SignalScout 的新闻搜索器。必须使用 web_search 找到真实网页来源，" + "只输出 JSON object，不输出 Markdown 或额外解释。" + ), + }, + { + "role": "user", + "content": json.dumps( + { + "task": "联网搜索最近AI新闻，覆盖中文和英文来源。", + "date_window": { + "from": since.isoformat(), + "to": run_date.isoformat(), + }, + "search_group": query_group["name"], + "preferred_language": query_group["language"], + "target_count": query_group["target_count"], + "queries": query_group["queries"], + "output_schema": { + "items": [ + { + "title": "新闻标题", + "summary": "中文摘要，说明为什么值得关注", + "url": "真实可访问来源链接", + "source": "媒体或网站名称", + "published_at": "ISO时间；不知道可为null", + "language": "zh或en", + "importance": "1到5整数", + "entities": ["公司、模型、项目、人名等"], + } + ] + }, + "rules": [ + "优先返回中文和英文来源各一部分，不要只返回英文来源。", + "只使用最近日期窗口内或明显接近日内的新闻。", + "每条必须有真实 url，不要使用搜索结果页、YouTube集合页或空链接。", + "如果中文来源足够，至少保留三分之一中文新闻。", + "返回尽可能接近 target_count 的 items。", + ], + }, + ensure_ascii=False, + ), + }, + ], + "tools": [{"type": "web_search"}], + "max_output_tokens": self.settings.llm_max_tokens, + }, + ensure_ascii=False, + ).encode("utf-8") + request = Request( + f"{self.settings.llm_base_url.rstrip('/')}/responses", + data=body, + headers={ + "Authorization": f"Bearer {self.settings.llm_api_key}", + "Content-Type": "application/json", + "Accept": "application/json", + }, + method="POST", + ) try: - with urlopen(request, timeout=20) as response: - body = response.read().decode("utf-8") + with urlopen(request, timeout=self.settings.llm_timeout_seconds) as response: + response_body = response.read().decode("utf-8") except HTTPError as exc: message = exc.read().decode("utf-8", errors="replace") - raise RuntimeError(f"News search failed: HTTP {exc.code} {message}") from exc + raise RuntimeError(f"Grok news search failed: HTTP {exc.code} {message}") from exc except URLError as exc: - raise RuntimeError(f"News search failed: {exc.reason}") from exc - return json.loads(body) + raise RuntimeError(f"Grok news search failed: {exc.reason}") from exc + return self._parse_response_json(json.loads(response_body)) - def _story_to_signal(self, story: dict[str, Any]) -> Optional[SignalItem]: - # HN故事先保留标题、链接和互动指标，重要性与摘要由模型最终确定。 - title = story.get("title") - url = self._story_url(story) - if not isinstance(title, str) or not isinstance(url, str): + def _parse_response_json(self, response_payload: dict[str, Any]) -> dict[str, Any]: + # Responses返回多段output，最终message里的output_text才是模型给业务层的JSON。 + text_parts: list[str] = [] + for output in response_payload.get("output", []): + if not isinstance(output, dict): + continue + for content in output.get("content", []): + if isinstance(content, dict) and content.get("type") == "output_text": + text = content.get("text") + if isinstance(text, str): + text_parts.append(text) + content_text = "\n".join(text_parts).strip() + if not content_text: + raise RuntimeError("Grok news search returned no output_text") + try: + payload = json.loads(content_text) + except json.JSONDecodeError as exc: + raise RuntimeError(f"Grok news search returned invalid JSON: {content_text[:1200]}") from exc + if not isinstance(payload, dict): + raise RuntimeError("Grok news search JSON root is not an object") + return payload + + def _article_to_signal(self, item: dict[str, Any]) -> Optional[SignalItem]: + # 搜索结果只接受带真实网页链接的条目，避免把模型说明文本写入情报库。 + title = item.get("title") + url = item.get("url") + if not isinstance(title, str) or not isinstance(url, str) or not url.startswith("http"): return None - author = str(story.get("author") or "unknown") - points = int(story.get("points") or 0) - comments = int(story.get("num_comments") or 0) - published_at = self._parse_hn_date(story.get("created_at")) + source = item.get("source") + language = item.get("language") + entities = item.get("entities") return SignalItem( source_type="news", - topic="AI 新闻候选", + topic="中文 AI 新闻" if language == "zh" else "AI 新闻", title=title[:300], - summary=f"Hacker News 最新技术讨论，作者 {author}，约 {points} points、{comments} 条评论。", + summary=str(item.get("summary") or "Grok Web Search 搜索到的AI新闻候选。"), source_url=url, - source_name="Hacker News", - published_at=published_at, - importance=3, - entities=["Hacker News", author], + source_name=str(source or "Grok Web Search"), + published_at=self._parse_datetime(item.get("published_at")), + importance=self._parse_importance(item.get("importance")), + entities=[str(entity) for entity in entities] if isinstance(entities, list) else [], ) - def _story_url(self, story: dict[str, Any]) -> Optional[str]: - # 外链优先；没有外链的Ask/Show HN故事使用HN讨论页作为证据链接。 - url = story.get("url") or story.get("story_url") - if isinstance(url, str) and url.startswith("http"): - return url - object_id = story.get("objectID") - if object_id: - return f"https://news.ycombinator.com/item?id={object_id}" - return None - - def _parse_hn_date(self, value: Any) -> Optional[datetime]: - # HN Algolia返回UTC ISO时间，入库前先换算为北京时间。 - if not isinstance(value, str): + def _parse_datetime(self, value: Any) -> Optional[datetime]: + # 搜索结果可能只有日期或ISO时间，能解析则统一换算为北京时间。 + if not isinstance(value, str) or not value: return None try: + if len(value) == 10: + value = f"{value}T00:00:00" return to_beijing_naive(datetime.fromisoformat(value.replace("Z", "+00:00"))) except ValueError: return None + + def _parse_importance(self, value: Any) -> int: + # 重要度来自搜索模型，但入库前仍限制在业务契约允许范围内。 + try: + importance = int(value) + except (TypeError, ValueError): + return 3 + return max(1, min(importance, 5)) diff --git a/app/schemas/agent.py b/app/schemas/agent.py index 7c51e9b..0ef6906 100644 --- a/app/schemas/agent.py +++ b/app/schemas/agent.py @@ -39,6 +39,11 @@ class RunResponse(BaseModel): report_path: Optional[str] = None +class RunTriggerRead(BaseModel): + status: str + run_id: Optional[int] = None + + class RunRead(BaseModel): id: int status: str diff --git a/app/ui/home.py b/app/ui/home.py index a7e8347..ada7655 100644 --- a/app/ui/home.py +++ b/app/ui/home.py @@ -162,6 +162,30 @@ HOME_PAGE_HTML = """ margin-top: 5px; } + .actions { + display: flex; + justify-content: flex-end; + margin-top: 12px; + } + + .action-button { + border: 1px solid oklch(25% 0.02 210); + border-radius: 8px; + padding: 9px 13px; + background: var(--ink); + color: white; + font: inherit; + font-size: 13px; + font-weight: 700; + cursor: pointer; + min-width: 96px; + } + + .action-button:disabled { + cursor: wait; + opacity: 0.66; + } + .layout { display: grid; grid-template-columns: minmax(0, 1.2fr) minmax(340px, 0.8fr); @@ -410,6 +434,9 @@ HOME_PAGE_HTML = """
读取中 正在整理新闻与项目。 +
+ +
@@ -484,6 +511,7 @@ HOME_PAGE_HTML = """ const runCountFull = document.querySelector("#runCountFull"); const todayCountText = document.querySelector("#todayCountText"); const todayCountDetail = document.querySelector("#todayCountDetail"); + const manualRunButton = document.querySelector("#manualRunButton"); const stateText = { completed: "完成", @@ -667,9 +695,34 @@ HOME_PAGE_HTML = """ renderRuns(data.runs || []); } + async function triggerManualRun() { + manualRunButton.disabled = true; + manualRunButton.textContent = "搜索中"; + try { + const response = await fetch("/agent/runs/daily", { + method: "POST", + headers: { Accept: "application/json" } + }); + if (!response.ok) throw new Error("手动搜索失败"); + await loadDashboard(); + } finally { + window.setTimeout(() => { + manualRunButton.disabled = false; + manualRunButton.textContent = "手动搜索"; + loadDashboard().catch(() => {}); + }, 3000); + } + } + navButtons.forEach((button) => { button.addEventListener("click", () => activateView(button.dataset.viewTarget)); }); + manualRunButton.addEventListener("click", () => { + triggerManualRun().catch(() => { + manualRunButton.disabled = false; + manualRunButton.textContent = "手动搜索"; + }); + }); const initialView = window.location.hash.replace("#", ""); if (["signals", "report", "runs"].includes(initialView)) { diff --git a/migrations/versions/202607081730_scope_signal_url_dedupe_to_run.py b/migrations/versions/202607081730_scope_signal_url_dedupe_to_run.py new file mode 100644 index 0000000..55f4dd8 --- /dev/null +++ b/migrations/versions/202607081730_scope_signal_url_dedupe_to_run.py @@ -0,0 +1,31 @@ +"""scope signal url dedupe to run + +Revision ID: 202607081730 +Revises: 202607081700 +Create Date: 2026-07-08 17:30:00 +""" + +from typing import Sequence, Union + +from alembic import op + +revision: str = "202607081730" +down_revision: Union[str, None] = "202607081700" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # 信号按运行批次保存；同一链接跨运行保留，便于追踪每次日报实际产出。 + op.drop_constraint("uq_signals_source_url_hash", "signals", type_="unique") + op.create_unique_constraint( + "uq_signals_run_source_url_hash", + "signals", + ["run_id", "source_url_hash"], + ) + + +def downgrade() -> None: + # 回滚时恢复全局链接唯一约束，迁移链保持可逆。 + op.drop_constraint("uq_signals_run_source_url_hash", "signals", type_="unique") + op.create_unique_constraint("uq_signals_source_url_hash", "signals", ["source_url_hash"]) diff --git a/tests/test_api_health.py b/tests/test_api_health.py index 1e833bd..95cd8c6 100644 --- a/tests/test_api_health.py +++ b/tests/test_api_health.py @@ -35,22 +35,29 @@ def test_home_page_returns_dashboard_shell(monkeypatch) -> None: assert response.status_code == 200 assert "SignalScout" in response.text assert "今日 AI 信号" in response.text + assert "手动搜索" in response.text assert "renderMarkdownReport" in response.text assert "自动记录" not in response.text assert "采集今日情报" not in response.text -def test_manual_agent_run_endpoint_is_not_exposed(monkeypatch) -> None: - # Agent只通过后台调度器自动运行，HTTP API不提供手动启动入口。 +def test_manual_agent_run_endpoint_starts_background_job(monkeypatch) -> None: + # 手动搜索入口复用调度任务，方便验证每轮候选和入库数量。 monkeypatch.setenv("SIGNALSCOUT_ENV", "test") get_settings.cache_clear() + started = [] + + def fake_run_scheduled_job(settings): + started.append(settings.env) + + monkeypatch.setattr("app.api.routes.run_scheduled_job", fake_run_scheduled_job) with TestClient(create_app()) as client: response = client.post("/agent/runs/daily") - stream_response = client.post("/agent/runs/daily/stream") get_settings.cache_clear() - assert response.status_code == 404 - assert stream_response.status_code == 404 + assert response.status_code == 200 + assert response.json()["status"] == "started" + assert started == ["test"] def test_dashboard_returns_news_and_github_signals_separately(monkeypatch) -> None: diff --git a/tests/test_news_search_client.py b/tests/test_news_search_client.py index bafae9e..1ce36d7 100644 --- a/tests/test_news_search_client.py +++ b/tests/test_news_search_client.py @@ -2,26 +2,50 @@ from app.core.config import Settings from app.integrations.news_search_client import NewsSearchClient -def test_news_search_client_maps_article_to_candidate() -> None: - # 新闻搜索结果会先转为候选信号，再交给模型筛选成最终情报。 +def test_news_search_client_maps_grok_article_to_candidate() -> None: + # Grok Web Search结果会先转为候选信号，再进入入库和日报总结流程。 client = NewsSearchClient(Settings()) - signal = client._story_to_signal( + signal = client._article_to_signal( { - "title": "Show HN: AI agent platform launches", - "url": "https://example.com/ai-agent-platform", - "author": "builder", - "points": 128, - "num_comments": 24, - "created_at": "2026-07-08T12:00:00Z", - "objectID": "123", + "title": "国内大模型公司发布新一代Agent平台", + "summary": "中文AI新闻候选。", + "url": "https://example.cn/ai-agent-platform", + "source": "示例中文媒体", + "published_at": "2026-07-08T12:00:00Z", + "language": "zh", + "importance": 4, + "entities": ["Agent平台"], } ) assert signal is not None - assert signal.topic == "AI 新闻候选" + assert signal.topic == "中文 AI 新闻" assert signal.source_type == "news" - assert signal.source_name == "Hacker News" + assert signal.source_name == "示例中文媒体" assert signal.published_at is not None assert signal.published_at.year == 2026 assert signal.published_at.hour == 20 - assert "128 points" in signal.summary + assert signal.importance == 4 + + +def test_news_search_client_parses_responses_output_text() -> None: + # Responses API会返回多段output，业务JSON只来自最终output_text。 + client = NewsSearchClient(Settings()) + payload = client._parse_response_json( + { + "output": [ + {"type": "web_search_call", "status": "completed"}, + { + "type": "message", + "content": [ + { + "type": "output_text", + "text": '{"items":[{"title":"AI news","url":"https://example.com","source":"Example"}]}', + } + ], + }, + ] + } + ) + + assert payload["items"][0]["title"] == "AI news" diff --git a/tests/test_repository.py b/tests/test_repository.py new file mode 100644 index 0000000..4af2acd --- /dev/null +++ b/tests/test_repository.py @@ -0,0 +1,81 @@ +from datetime import datetime + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + +from app.db.models import AgentRun, Base +from app.db.repository import AgentRepository +from app.schemas.agent import SignalItem + + +def test_save_signals_keeps_same_url_across_runs() -> None: + # 跨运行保留同一链接，保证每轮日报的实际信号快照不会被历史结果吞掉。 + engine = create_engine( + "sqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + future=True, + ) + Base.metadata.create_all(engine) + TestingSessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True) + with TestingSessionLocal() as session: + repo = AgentRepository(session) + first_run = AgentRun(status="running", started_at=datetime(2026, 7, 8), request_payload={}) + second_run = AgentRun(status="running", started_at=datetime(2026, 7, 8), request_payload={}) + session.add_all([first_run, second_run]) + session.flush() + item = SignalItem( + source_type="news", + topic="AI 新闻", + title="Same URL", + summary="同一个来源链接。", + source_url="https://example.com/same", + source_name="Example", + published_at=datetime(2026, 7, 8), + importance=3, + entities=["Example"], + ) + + first_saved = repo.save_signals(first_run, [item]) + second_saved = repo.save_signals(second_run, [item]) + + assert len(first_saved) == 1 + assert len(second_saved) == 1 + assert first_saved[0].run_id != second_saved[0].run_id + + +def test_save_signals_updates_duplicate_url_inside_same_run() -> None: + # 同一运行内相同链接只保留一条，避免模型重复输出造成页面重复。 + engine = create_engine( + "sqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + future=True, + ) + Base.metadata.create_all(engine) + TestingSessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True) + with TestingSessionLocal() as session: + repo = AgentRepository(session) + run = AgentRun(status="running", started_at=datetime(2026, 7, 8), request_payload={}) + session.add(run) + session.flush() + first = SignalItem( + source_type="news", + topic="AI 新闻", + title="First title", + summary="第一条摘要。", + source_url="https://example.com/same", + source_name="Example", + published_at=datetime(2026, 7, 8), + importance=3, + entities=["Example"], + ) + second = first.model_copy(update={"title": "Second title", "importance": 5}) + + saved = repo.save_signals(run, [first, second]) + + assert len(saved) == 2 + assert saved[0].id == saved[1].id + assert saved[0].title == "Second title" + assert saved[0].importance == 5