feat: add intelligence radar backend#337
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces the foundational backend components for a new intelligence radar feature. It establishes the necessary database schema, API endpoints, and core service logic to aggregate and present relevant community signals from diverse external platforms. The changes enable the system to proactively identify and surface important developments based on user-defined interests, laying the groundwork for a dynamic and personalized intelligence feed. Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
There was a problem hiding this comment.
Pull request overview
Adds a new “intelligence/community radar” feature to PaperBot, including persistence, collection from external sources (Reddit/GitHub/HF/X), and a new API endpoint for serving a personalized intelligence feed.
Changes:
- Introduces
intelligence_eventspersistence (SQLAlchemy model + Alembic migration) and anIntelligenceStorefor querying/upserting signals. - Adds
IntelligenceRadarServiceto collect/score external signals (Reddit RSS + comments, GitHub Atom + Issues API, HF Daily Papers, X recent-search API). - Exposes
/api/intelligence/feedwith filtering/sorting/track matching plus unit test coverage for store/service/route behavior.
Reviewed changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/unit/test_source_registry_modes.py | Updates default source expectation for twitter_x acquisition mode. |
| tests/unit/test_intelligence_store.py | Adds tests for datetime serialization and timezone restoration in IntelligenceStore. |
| tests/unit/test_intelligence_routes.py | Adds route-level test for /api/intelligence/feed response shape and parameter forwarding. |
| tests/unit/test_intelligence_radar_service.py | Adds unit tests for match logic, scoring, sorting, and refresh checks. |
| src/paperbot/infrastructure/stores/models.py | Adds IntelligenceEventModel SQLAlchemy table mapping. |
| src/paperbot/infrastructure/stores/intelligence_store.py | New store for upserting and listing intelligence events (JSON/list serialization helpers). |
| src/paperbot/infrastructure/services/intelligence_radar_service.py | New service to collect, score, persist, and list intelligence feed signals. |
| src/paperbot/infrastructure/api_clients/x_client.py | New X recent search API client wrapper. |
| src/paperbot/infrastructure/api_clients/github_client.py | New GitHub issues API client wrapper. |
| src/paperbot/infrastructure/api_clients/init.py | Exports new API clients. |
| src/paperbot/application/registries/default_sources.py | Registers GitHub/HF sources and updates twitter_x descriptor to api_first. |
| src/paperbot/api/routes/intelligence.py | New FastAPI route implementing /api/intelligence/feed with track matching and query building. |
| src/paperbot/api/routes/init.py | Adds feed and intelligence route modules to exports/imports. |
| src/paperbot/api/main.py | Includes the intelligence router in the FastAPI app. |
| env.example | Adds env vars for X bearer token and intelligence tuning defaults. |
| alembic/versions/0023_intelligence_events.py | New migration creating the intelligence_events table and index. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
You can also share your feedback on Copilot code review. Take the survey.
| reg = register_default_sources(SourceRegistry()) | ||
| x = reg.get("twitter_x") | ||
| assert x is not None | ||
| assert x.acquisition_mode == AcquisitionMode.import_only | ||
| assert x.acquisition_mode == AcquisitionMode.api_first | ||
| assert x.enabled_by_default is False |
There was a problem hiding this comment.
The test name still says the default Twitter/X source is "import_only", but the assertion now expects AcquisitionMode.api_first. Rename the test (or adjust the expectation) so the name matches the behavior being validated.
| "APIClient", | ||
| "GitHubRadarClient", | ||
| "SemanticScholarClient", | ||
| "XRecentSearchClient", |
There was a problem hiding this comment.
This file uses tab indentation inside the __all__ list. That’s inconsistent with the rest of the codebase (and will typically fight with Black/isort). Replace the tabs with spaces and let the formatter normalize the list.
| "APIClient", | |
| "GitHubRadarClient", | |
| "SemanticScholarClient", | |
| "XRecentSearchClient", | |
| "APIClient", | |
| "GitHubRadarClient", | |
| "SemanticScholarClient", | |
| "XRecentSearchClient", |
| with ThreadPoolExecutor(max_workers=max_workers) as executor: | ||
| future_map = { | ||
| executor.submit(self._fetch_reddit_keyword_summary, keyword, allowed_subreddits): keyword | ||
| for keyword in profile.keywords[:4] | ||
| } | ||
| for future in as_completed(future_map): | ||
| try: | ||
| row = future.result() | ||
| except Exception: | ||
| continue | ||
| if row is None: | ||
| continue | ||
|
|
||
| keyword = row["keyword"] | ||
| entry_titles = [entry["title"] for entry in row["entries"]] | ||
| author_matches = _find_matches(profile.scholar_names, entry_titles) | ||
| repo_matches = _find_matches(profile.watch_repos, entry_titles) | ||
| external_id = f"reddit:keyword:{_slugify(keyword)}" | ||
| previous = self._store.get_event(user_id=user_id, external_id=external_id) | ||
| metric_value = len(row["entries"]) | ||
| previous_value = int(previous.get("metric_value") or 0) if previous else 0 | ||
| metric_delta = metric_value - previous_value | ||
| subreddits = row["subreddits"] | ||
| top_entry = row["entries"][0] | ||
| score, breakdown = _score_signal( | ||
| source="reddit", | ||
| kind="keyword_spike", | ||
| metric_value=metric_value, | ||
| metric_delta=metric_delta, | ||
| published_at=row["latest_published_at"], | ||
| keyword_hits=[keyword], | ||
| author_matches=author_matches, | ||
| repo_matches=repo_matches, | ||
| ) | ||
| summary = ( | ||
| f"24h mentions: {metric_value} across {', '.join(f'r/{name}' for name in subreddits[:3])}. " | ||
| f"Top post: {top_entry['title']}." | ||
| ) | ||
| results.append( | ||
| { | ||
| "external_id": external_id, | ||
| "source": "reddit", | ||
| "source_label": "Reddit Search", | ||
| "kind": "keyword_spike", | ||
| "title": f"Reddit spike: {keyword}", | ||
| "summary": summary, | ||
| "url": top_entry["link"], | ||
| "keyword_hits": [keyword], | ||
| "author_matches": author_matches, | ||
| "repo_matches": repo_matches, | ||
| "metric_name": "mentions/24h", | ||
| "metric_value": metric_value, | ||
| "metric_delta": metric_delta, | ||
| "score": score, | ||
| "published_at": row["latest_published_at"], | ||
| "payload": { | ||
| "subreddits": subreddits, | ||
| "top_post": top_entry, | ||
| "entries": row["entries"][:5], | ||
| "score_breakdown": breakdown, | ||
| }, | ||
| } | ||
| ) |
There was a problem hiding this comment.
ThreadPoolExecutor is used to issue concurrent HTTP requests, but all tasks share the same requests.Session instance (self._session). requests.Session is not guaranteed to be thread-safe, so this can lead to flaky networking behavior under concurrent refreshes. Consider using per-thread sessions (create a session inside each task), switching to plain requests.get per call, or guarding session usage with a lock.
| with ThreadPoolExecutor(max_workers=max_workers) as executor: | |
| future_map = { | |
| executor.submit(self._fetch_reddit_keyword_summary, keyword, allowed_subreddits): keyword | |
| for keyword in profile.keywords[:4] | |
| } | |
| for future in as_completed(future_map): | |
| try: | |
| row = future.result() | |
| except Exception: | |
| continue | |
| if row is None: | |
| continue | |
| keyword = row["keyword"] | |
| entry_titles = [entry["title"] for entry in row["entries"]] | |
| author_matches = _find_matches(profile.scholar_names, entry_titles) | |
| repo_matches = _find_matches(profile.watch_repos, entry_titles) | |
| external_id = f"reddit:keyword:{_slugify(keyword)}" | |
| previous = self._store.get_event(user_id=user_id, external_id=external_id) | |
| metric_value = len(row["entries"]) | |
| previous_value = int(previous.get("metric_value") or 0) if previous else 0 | |
| metric_delta = metric_value - previous_value | |
| subreddits = row["subreddits"] | |
| top_entry = row["entries"][0] | |
| score, breakdown = _score_signal( | |
| source="reddit", | |
| kind="keyword_spike", | |
| metric_value=metric_value, | |
| metric_delta=metric_delta, | |
| published_at=row["latest_published_at"], | |
| keyword_hits=[keyword], | |
| author_matches=author_matches, | |
| repo_matches=repo_matches, | |
| ) | |
| summary = ( | |
| f"24h mentions: {metric_value} across {', '.join(f'r/{name}' for name in subreddits[:3])}. " | |
| f"Top post: {top_entry['title']}." | |
| ) | |
| results.append( | |
| { | |
| "external_id": external_id, | |
| "source": "reddit", | |
| "source_label": "Reddit Search", | |
| "kind": "keyword_spike", | |
| "title": f"Reddit spike: {keyword}", | |
| "summary": summary, | |
| "url": top_entry["link"], | |
| "keyword_hits": [keyword], | |
| "author_matches": author_matches, | |
| "repo_matches": repo_matches, | |
| "metric_name": "mentions/24h", | |
| "metric_value": metric_value, | |
| "metric_delta": metric_delta, | |
| "score": score, | |
| "published_at": row["latest_published_at"], | |
| "payload": { | |
| "subreddits": subreddits, | |
| "top_post": top_entry, | |
| "entries": row["entries"][:5], | |
| "score_breakdown": breakdown, | |
| }, | |
| } | |
| ) | |
| for keyword in profile.keywords[:4]: | |
| try: | |
| row = self._fetch_reddit_keyword_summary(keyword, allowed_subreddits) | |
| except Exception: | |
| continue | |
| if row is None: | |
| continue | |
| keyword = row["keyword"] | |
| entry_titles = [entry["title"] for entry in row["entries"]] | |
| author_matches = _find_matches(profile.scholar_names, entry_titles) | |
| repo_matches = _find_matches(profile.watch_repos, entry_titles) | |
| external_id = f"reddit:keyword:{_slugify(keyword)}" | |
| previous = self._store.get_event(user_id=user_id, external_id=external_id) | |
| metric_value = len(row["entries"]) | |
| previous_value = int(previous.get("metric_value") or 0) if previous else 0 | |
| metric_delta = metric_value - previous_value | |
| subreddits = row["subreddits"] | |
| top_entry = row["entries"][0] | |
| score, breakdown = _score_signal( | |
| source="reddit", | |
| kind="keyword_spike", | |
| metric_value=metric_value, | |
| metric_delta=metric_delta, | |
| published_at=row["latest_published_at"], | |
| keyword_hits=[keyword], | |
| author_matches=author_matches, | |
| repo_matches=repo_matches, | |
| ) | |
| summary = ( | |
| f"24h mentions: {metric_value} across {', '.join(f'r/{name}' for name in subreddits[:3])}. " | |
| f"Top post: {top_entry['title']}." | |
| ) | |
| results.append( | |
| { | |
| "external_id": external_id, | |
| "source": "reddit", | |
| "source_label": "Reddit Search", | |
| "kind": "keyword_spike", | |
| "title": f"Reddit spike: {keyword}", | |
| "summary": summary, | |
| "url": top_entry["link"], | |
| "keyword_hits": [keyword], | |
| "author_matches": author_matches, | |
| "repo_matches": repo_matches, | |
| "metric_name": "mentions/24h", | |
| "metric_value": metric_value, | |
| "metric_delta": metric_delta, | |
| "score": score, | |
| "published_at": row["latest_published_at"], | |
| "payload": { | |
| "subreddits": subreddits, | |
| "top_post": top_entry, | |
| "entries": row["entries"][:5], | |
| "score_breakdown": breakdown, | |
| }, | |
| } | |
| ) |
| with self._provider.session() as session: | ||
| row = session.execute( | ||
| select(IntelligenceEventModel).where( | ||
| IntelligenceEventModel.user_id == user_id, | ||
| IntelligenceEventModel.external_id == external_id, | ||
| ) | ||
| ).scalar_one_or_none() | ||
| if row is None: | ||
| row = IntelligenceEventModel( | ||
| user_id=user_id, | ||
| external_id=external_id, | ||
| created_at=now, | ||
| ) | ||
|
|
||
| row.source = (source or "unknown").strip()[:32] | ||
| row.source_label = (source_label or row.source).strip()[:64] | ||
| row.kind = (kind or "signal").strip()[:64] | ||
| row.title = (title or "Untitled signal").strip() | ||
| row.summary = (summary or "").strip() | ||
| row.url = (url or "").strip() | ||
| row.repo_full_name = (repo_full_name or "").strip()[:128] | ||
| row.author_name = (author_name or "").strip()[:128] | ||
| row.keyword_hits_json = _dump_list(keyword_hits) | ||
| row.author_matches_json = _dump_list(author_matches) | ||
| row.repo_matches_json = _dump_list(repo_matches) | ||
| row.metric_name = (metric_name or "").strip()[:64] | ||
| row.metric_value = int(metric_value or 0) | ||
| row.metric_delta = int(metric_delta or 0) | ||
| row.score = float(score or 0.0) | ||
| row.published_at = _parse_datetime(published_at) or row.published_at or now | ||
| row.detected_at = now | ||
| row.updated_at = now | ||
| row.payload_json = _dump_dict(payload) | ||
|
|
||
| session.add(row) | ||
| session.commit() | ||
| session.refresh(row) | ||
| return self._row_to_dict(row) |
There was a problem hiding this comment.
upsert_event() does a read-then-insert/update without handling the race where two refreshes insert the same (user_id, external_id) concurrently. With the UNIQUE constraint, this can raise an IntegrityError and fail the refresh/background task. Consider using a database-level upsert (e.g., INSERT .. ON CONFLICT DO UPDATE for SQLite/Postgres) or catching IntegrityError on commit() and re-selecting/updating the existing row.
| op.create_table( | ||
| "intelligence_events", | ||
| sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), | ||
| sa.Column("user_id", sa.String(64), nullable=False, index=True), |
There was a problem hiding this comment.
The ORM model sets user_id default to "default", but the migration creates user_id as NOT NULL without a server_default. Any non-ORM inserts (or future backfills) that omit user_id will fail at the DB layer. Consider adding server_default="default" (or an explicit sa.text(...)) to keep the DB schema aligned with the model default.
| sa.Column("user_id", sa.String(64), nullable=False, index=True), | |
| sa.Column("user_id", sa.String(64), nullable=False, server_default="default", index=True), |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new 'intelligence_events' table and API endpoint to support a community radar/intelligence feed feature. It adds new API clients for GitHub and X (Twitter) and integrates HuggingFace daily papers, Reddit, GitHub activity, and X search results into an intelligence radar service. The service collects and scores events based on keywords, authors, and repositories, storing them in the new database table. Environment variables for X bearer tokens and intelligence feed configuration have also been added. Review comments highlight a high-severity Insecure Direct Object Reference (IDOR) vulnerability in the user_id parameter, suggesting proper authentication and authorization. Other comments point out several medium-severity issues, including inefficient database queries in the intelligence feed, the use of string literals for integer and float server_default values in the database schema, and numerous instances of catching generic Exception types which should be replaced with more specific exception handling and logging for better debugging and robustness. Additionally, there is a code duplication issue with _utcnow and _parse_datetime utility functions that should be refactored into a shared module.
| @router.get("/intelligence/feed", response_model=IntelligenceFeedResponse) | ||
| def get_intelligence_feed( | ||
| background_tasks: BackgroundTasks, | ||
| user_id: str = Query("default"), |
There was a problem hiding this comment.
The user_id parameter is taken directly from the query string without any authentication or authorization check. In a multi-user environment, this allows any user to access the intelligence feed, research tracks, and keywords of any other user by simply providing their user_id in the request. This is a classic Insecure Direct Object Reference (IDOR) vulnerability.
To remediate this, you should:
- Implement an authentication mechanism (e.g., JWT, session cookies).
- Retrieve the
user_idfrom the authenticated user's session or token instead of a query parameter. - If the
user_idmust be provided in the query, verify that the authenticated user has permission to access the data for thatuser_id.
| sa.Column("metric_name", sa.String(64), nullable=False, server_default=""), | ||
| sa.Column("metric_value", sa.Integer(), nullable=False, server_default="0"), | ||
| sa.Column("metric_delta", sa.Integer(), nullable=False, server_default="0"), | ||
| sa.Column("score", sa.Float(), nullable=False, server_default="0.0", index=True), |
There was a problem hiding this comment.
For sa.Float() columns, it's generally better practice to use a float literal for server_default (e.g., 0.0) instead of a string literal ("0.0"). While most databases might implicitly convert this, being explicit avoids potential type mismatches or unexpected behavior in different database systems.
| sa.Column("score", sa.Float(), nullable=False, server_default="0.0", index=True), | |
| sa.Column("score", sa.Float(), nullable=False, server_default=0.0, index=True), |
| sa.Column("metric_value", sa.Integer(), nullable=False, server_default="0"), | ||
| sa.Column("metric_delta", sa.Integer(), nullable=False, server_default="0"), |
There was a problem hiding this comment.
Similar to the score column, for sa.Integer() columns, it's best to use an integer literal for server_default (e.g., 0) instead of a string literal ("0").
| sa.Column("metric_value", sa.Integer(), nullable=False, server_default="0"), | |
| sa.Column("metric_delta", sa.Integer(), nullable=False, server_default="0"), | |
| sa.Column("metric_value", sa.Integer(), nullable=False, server_default=0), | |
| sa.Column("metric_delta", sa.Integer(), nullable=False, server_default=0), |
| rows = service.list_feed( | ||
| user_id=user_id, | ||
| limit=max(int(limit) * 10, 50), | ||
| source=source, |
There was a problem hiding this comment.
The list_feed call in the service fetches max(int(limit) * 10, 50) items, which means it will always fetch at least 50 items from the database, even if the requested limit from the API is much smaller (e.g., limit=6). This can lead to inefficient database queries and unnecessary data transfer if the filtering applied afterwards significantly reduces the result set. Consider pushing more of the filtering and limiting logic down to the IntelligenceStore.list_events method to optimize performance, or adjust the candidate_limit to be a smaller multiple of limit.
| except Exception: | ||
| tracks = [] |
There was a problem hiding this comment.
Catching a generic Exception can hide important details about what went wrong. It's better to catch more specific exceptions (e.g., sqlalchemy.exc.SQLAlchemyError for database issues) and/or log the full exception traceback to aid debugging. This applies to other similar try...except Exception blocks in the codebase.
| except Exception: | ||
| grouped.setdefault(repo, {})[feed_type] = [] |
There was a problem hiding this comment.
Catching a generic Exception can hide important details about what went wrong during the GitHub feed fetching. It's better to catch more specific exceptions and/or log the full exception traceback to aid debugging.
| issues = self._github_client.list_recent_issues(repo, per_page=20) | ||
| except Exception: |
There was a problem hiding this comment.
Catching a generic Exception can hide important details about what went wrong during the list_recent_issues call. It's better to catch more specific exceptions and/or log the full exception traceback to aid debugging.
| records = future.result() | ||
| except Exception: |
There was a problem hiding this comment.
Catching a generic Exception can hide important details about what went wrong during the _hf_daily.search call. It's better to catch more specific exceptions and/or log the full exception traceback to aid debugging.
| except Exception: | ||
| tweets = [] |
There was a problem hiding this comment.
Catching a generic Exception can hide important details about what went wrong during the _x_client.search_recent call. It's better to catch more specific exceptions and/or log the full exception traceback to aid debugging.
| def _utcnow() -> datetime: | ||
| return datetime.now(timezone.utc) | ||
|
|
||
|
|
||
| def _parse_datetime(value: Any) -> Optional[datetime]: | ||
| if isinstance(value, datetime): | ||
| return value if value.tzinfo else value.replace(tzinfo=timezone.utc) | ||
| if not value: | ||
| return None | ||
|
|
||
| text = str(value).strip() | ||
| if not text: | ||
| return None | ||
| if text.endswith("Z"): | ||
| text = f"{text[:-1]}+00:00" | ||
| try: | ||
| parsed = datetime.fromisoformat(text) | ||
| return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) | ||
| except Exception: | ||
| return None |
There was a problem hiding this comment.
The utility functions _utcnow and _parse_datetime are duplicated in src/paperbot/infrastructure/services/intelligence_radar_service.py. To improve maintainability and avoid code duplication, these common utility functions should be moved to a shared module (e.g., paperbot.utils.datetime_utils) and imported where needed.
Summary
Validation
Notes