Coverage for src/local_deep_research/web/routers/metrics.py: 98%
942 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""Routes for metrics dashboard."""
3import asyncio
5from fastapi import APIRouter, Depends, Request
6from fastapi.responses import JSONResponse
7from ..dependencies.auth import require_auth
8from ..dependencies.threadpool import run_db_sync
9from ..dependencies.rate_limit import limiter, _user_key
10from ..template_config import templates
12from datetime import datetime, timedelta, UTC
13from typing import Any, Annotated
15from loguru import logger
16from sqlalchemy import case, func
18from ...database.models import (
19 Journal,
20 Paper,
21 PaperAppearance,
22 RateLimitEstimate,
23 ResearchHistory,
24 ResearchRating,
25 ResearchResource,
26 ResearchStrategy,
27 TokenUsage,
28)
29from ...constants import (
30 get_available_strategies as get_central_available_strategies,
31)
32from ...domain_classifier import DomainClassifier, DomainClassification
33from ...database.rekey_lock import RekeyOperationalError
34from ...database.session_context import get_user_db_session
35from ...metrics import TokenCounter
36from ...metrics.query_utils import (
37 get_context_overflow_truncation_summary,
38 get_period_cutoff,
39 get_period_days,
40 get_time_filter_condition,
41)
42from ...metrics.search_tracker import get_search_tracker
43from ...utilities.metric_write_health import (
44 METRIC_WRITE_REASON_LABELS,
45 get_metric_write_health,
46)
47from ...security import redact_url_for_log, url_authority_without_userinfo
48from ..dependencies.json_body import json_body_error, read_json_dict
49from .notes import _log_value_preview
51# Create the router for metrics
52router = APIRouter(prefix="/metrics", tags=["metrics"])
54# NOTE: Routes use username (not .get()) intentionally.
55# Depends(require_auth) guarantees the key exists; direct access fails
56# fast if the dependency is ever removed.
59def _extract_domain(url):
60 """Extract normalized domain from URL, stripping www. prefix."""
61 try:
62 # Authority minus userinfo (empty when that is ambiguous),
63 # matching DomainClassifier's key: the domain is returned to the
64 # UI, logged and sent back to be classified and persisted.
65 domain = url_authority_without_userinfo(url).lower()
66 if domain.startswith("www."):
67 domain = domain[4:]
68 return domain if domain else None
69 except (ValueError, AttributeError, TypeError):
70 return None
73def get_rating_analytics(period="30d", research_mode="all", username=None):
74 """Get rating analytics for the specified period and research mode."""
75 try:
76 if not username:
77 return {
78 "rating_analytics": {
79 "avg_rating": None,
80 "total_ratings": 0,
81 "rating_distribution": {},
82 "satisfaction_stats": {
83 "very_satisfied": 0,
84 "satisfied": 0,
85 "neutral": 0,
86 "dissatisfied": 0,
87 "very_dissatisfied": 0,
88 },
89 "error": "No user session",
90 }
91 }
93 # Calculate date range
94 days = get_period_days(period)
96 with get_user_db_session(username) as session:
97 query = session.query(ResearchRating)
99 # Apply time filter
100 if days:
101 cutoff_date = datetime.now(UTC) - timedelta(days=days)
102 query = query.filter(ResearchRating.created_at >= cutoff_date)
104 # Get all ratings
105 ratings = query.all()
107 if not ratings:
108 return {
109 "rating_analytics": {
110 "avg_rating": None,
111 "total_ratings": 0,
112 "rating_distribution": {},
113 "satisfaction_stats": {
114 "very_satisfied": 0,
115 "satisfied": 0,
116 "neutral": 0,
117 "dissatisfied": 0,
118 "very_dissatisfied": 0,
119 },
120 }
121 }
123 # Calculate statistics
124 rating_values = [r.rating for r in ratings]
125 avg_rating = sum(rating_values) / len(rating_values)
127 # Rating distribution
128 rating_counts = {}
129 for i in range(1, 6):
130 rating_counts[str(i)] = rating_values.count(i)
132 # Satisfaction categories
133 satisfaction_stats = {
134 "very_satisfied": rating_values.count(5),
135 "satisfied": rating_values.count(4),
136 "neutral": rating_values.count(3),
137 "dissatisfied": rating_values.count(2),
138 "very_dissatisfied": rating_values.count(1),
139 }
141 return {
142 "rating_analytics": {
143 "avg_rating": round(avg_rating, 1),
144 "total_ratings": len(ratings),
145 "rating_distribution": rating_counts,
146 "satisfaction_stats": satisfaction_stats,
147 }
148 }
150 except Exception:
151 logger.exception("Error getting rating analytics")
152 return {
153 "rating_analytics": {
154 "avg_rating": None,
155 "total_ratings": 0,
156 "rating_distribution": {},
157 "satisfaction_stats": {
158 "very_satisfied": 0,
159 "satisfied": 0,
160 "neutral": 0,
161 "dissatisfied": 0,
162 "very_dissatisfied": 0,
163 },
164 }
165 }
168def get_link_analytics(period="30d", username=None):
169 """Get link analytics from research resources."""
170 try:
171 if not username:
172 return {
173 "link_analytics": {
174 "top_domains": [],
175 "total_unique_domains": 0,
176 "avg_links_per_research": 0,
177 "domain_distribution": {},
178 "source_type_analysis": {},
179 "academic_vs_general": {},
180 "total_links": 0,
181 "error": "No user session",
182 }
183 }
185 # Calculate date range
186 days = get_period_days(period)
188 with get_user_db_session(username) as session:
189 # Project only the columns the analytics loop reads (url /
190 # research_id / created_at / source_type / title) plus a SQL-level
191 # boolean for whether content_preview is present. Loading full
192 # ``ResearchResource`` entities would materialize the
193 # ``content_preview`` Text column for every row on this whole-table
194 # scan — which runs unfiltered when ``period=all`` (#4560).
195 has_preview = (
196 ResearchResource.content_preview.isnot(None)
197 & (ResearchResource.content_preview != "")
198 ).label("has_preview")
199 query = session.query(
200 ResearchResource.url,
201 ResearchResource.research_id,
202 ResearchResource.created_at,
203 ResearchResource.source_type,
204 ResearchResource.title,
205 has_preview,
206 )
208 # Apply time filter
209 if days:
210 cutoff_date = datetime.now(UTC) - timedelta(days=days)
211 query = query.filter(
212 ResearchResource.created_at >= cutoff_date.isoformat()
213 )
215 # Get all resources
216 resources = query.all()
218 if not resources:
219 return {
220 "link_analytics": {
221 "top_domains": [],
222 "total_unique_domains": 0,
223 "avg_links_per_research": 0,
224 "domain_distribution": {},
225 "source_type_analysis": {},
226 "academic_vs_general": {},
227 "total_links": 0,
228 }
229 }
231 # Extract domains from URLs
232 domain_counts: dict[str, Any] = {}
233 domain_researches: dict[
234 str, Any
235 ] = {} # Track which researches used each domain
236 source_types: dict[str, Any] = {}
237 temporal_data: dict[str, Any] = {} # Track links over time
238 domain_connections: dict[
239 str, Any
240 ] = {} # Track domain co-occurrences
242 # Generic category counting from LLM classifications
243 category_counts: dict[str, Any] = {}
245 quality_metrics = {
246 "with_title": 0,
247 "with_preview": 0,
248 "with_both": 0,
249 "total": 0,
250 }
252 # First pass: collect all domains from resources
253 all_domains = set()
254 for resource in resources:
255 if resource.url:
256 domain = _extract_domain(resource.url)
257 if domain:
258 all_domains.add(domain)
260 # Batch load all domain classifications in one query (fix N+1)
261 domain_classifications_map = {}
262 if all_domains: 262 ↛ 274line 262 didn't jump to line 274 because the condition on line 262 was always true
263 all_classifications = (
264 session.query(DomainClassification)
265 .filter(DomainClassification.domain.in_(all_domains))
266 .all()
267 )
268 for classification in all_classifications:
269 domain_classifications_map[classification.domain] = (
270 classification
271 )
273 # Second pass: process resources with pre-loaded classifications
274 for resource in resources:
275 if resource.url:
276 try:
277 domain = _extract_domain(resource.url)
278 if not domain:
279 continue
281 # Count domains
282 domain_counts[domain] = domain_counts.get(domain, 0) + 1
284 # Track research IDs for each domain
285 if domain not in domain_researches:
286 domain_researches[domain] = set()
287 domain_researches[domain].add(resource.research_id)
289 # Track temporal data (daily counts)
290 if resource.created_at: 290 ↛ 299line 290 didn't jump to line 299 because the condition on line 290 was always true
291 date_str = resource.created_at[
292 :10
293 ] # Extract YYYY-MM-DD
294 temporal_data[date_str] = (
295 temporal_data.get(date_str, 0) + 1
296 )
298 # Count categories from pre-loaded classifications (no N+1)
299 classification = domain_classifications_map.get(domain)
300 if classification:
301 category = classification.category
302 category_counts[category] = (
303 category_counts.get(category, 0) + 1
304 )
305 else:
306 category_counts["Unclassified"] = (
307 category_counts.get("Unclassified", 0) + 1
308 )
310 # Track source type from metadata if available
311 if resource.source_type:
312 source_types[resource.source_type] = (
313 source_types.get(resource.source_type, 0) + 1
314 )
316 # Track quality metrics
317 quality_metrics["total"] += 1
318 if resource.title: 318 ↛ 319line 318 didn't jump to line 319 because the condition on line 318 was never true
319 quality_metrics["with_title"] += 1
320 if resource.has_preview: 320 ↛ 321line 320 didn't jump to line 321 because the condition on line 320 was never true
321 quality_metrics["with_preview"] += 1
322 if resource.title and resource.has_preview: 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true
323 quality_metrics["with_both"] += 1
325 # Track domain co-occurrences for network visualization
326 research_id = resource.research_id
327 if research_id not in domain_connections:
328 domain_connections[research_id] = []
329 domain_connections[research_id].append(domain)
331 except Exception as e:
332 # Type name and redacted URL only: the exception
333 # text and traceback can carry the raw URL.
334 logger.warning(
335 "Error parsing URL "
336 f"{redact_url_for_log(resource.url)} "
337 f"({type(e).__name__})"
338 )
340 # Sort domains by count and get top 10
341 sorted_domains = sorted(
342 domain_counts.items(), key=lambda x: x[1], reverse=True
343 )
344 top_10_domains = sorted_domains[:10]
346 # Calculate domain distribution (top domains vs others)
347 top_10_count = sum(count for _, count in top_10_domains)
348 others_count = len(resources) - top_10_count
350 # Get unique research IDs to calculate average
351 unique_research_ids = {r.research_id for r in resources}
352 avg_links = (
353 len(resources) / len(unique_research_ids)
354 if unique_research_ids
355 else 0
356 )
358 # Prepare temporal trend data (sorted by date)
359 temporal_trend = sorted(
360 [
361 {"date": date, "count": count}
362 for date, count in temporal_data.items()
363 ],
364 key=lambda x: x["date"],
365 )
367 # Get most recent research for each top domain and classifications
368 domain_recent_research = {}
369 # Build domain_classifications dict from pre-loaded data
370 domain_classifications = {
371 domain: {
372 "category": classification.category,
373 "subcategory": classification.subcategory,
374 "confidence": classification.confidence,
375 }
376 for domain, classification in domain_classifications_map.items()
377 }
379 # Batch-load research details for top domains (fix N+1 query)
380 all_research_ids = []
381 domain_research_id_lists = {}
382 for domain, _ in top_10_domains:
383 if domain in domain_researches: 383 ↛ 382line 383 didn't jump to line 382 because the condition on line 383 was always true
384 ids = list(domain_researches[domain])[:3]
385 domain_research_id_lists[domain] = ids
386 all_research_ids.extend(ids)
388 research_by_id = {}
389 if all_research_ids:
390 researches = (
391 session.query(ResearchHistory)
392 .filter(ResearchHistory.id.in_(all_research_ids))
393 .all()
394 )
395 research_by_id = {r.id: r for r in researches}
397 for domain, ids in domain_research_id_lists.items():
398 domain_recent_research[domain] = [
399 {
400 "id": r_id,
401 "query": research_by_id[r_id].query[:50]
402 if research_by_id.get(r_id)
403 and research_by_id[r_id].query
404 else "Research",
405 }
406 for r_id in ids
407 if r_id in research_by_id
408 ]
410 return {
411 "link_analytics": {
412 "top_domains": [
413 {
414 "domain": domain,
415 "count": count,
416 "percentage": round(
417 count / len(resources) * 100, 1
418 ),
419 "research_count": len(
420 domain_researches.get(domain, set())
421 ),
422 "recent_researches": domain_recent_research.get(
423 domain, []
424 ),
425 "classification": domain_classifications.get(
426 domain, None
427 ),
428 }
429 for domain, count in top_10_domains
430 ],
431 "total_unique_domains": len(domain_counts),
432 "avg_links_per_research": round(avg_links, 1),
433 "domain_distribution": {
434 "top_10": top_10_count,
435 "others": others_count,
436 },
437 "source_type_analysis": source_types,
438 "category_distribution": category_counts,
439 # Generic pie chart data - use whatever LLM classifier outputs
440 "domain_categories": category_counts,
441 "total_links": len(resources),
442 "total_researches": len(unique_research_ids),
443 "temporal_trend": temporal_trend,
444 "domain_metrics": {
445 domain: {
446 "usage_count": count,
447 "usage_percentage": round(
448 count / len(resources) * 100, 1
449 ),
450 "research_diversity": len(
451 domain_researches.get(domain, set())
452 ),
453 "frequency_rank": rank + 1,
454 }
455 for rank, (domain, count) in enumerate(top_10_domains)
456 },
457 }
458 }
460 except Exception:
461 logger.exception("Error getting link analytics")
462 return {
463 "link_analytics": {
464 "top_domains": [],
465 "total_unique_domains": 0,
466 "avg_links_per_research": 0,
467 "domain_distribution": {},
468 "source_type_analysis": {},
469 "academic_vs_general": {},
470 "total_links": 0,
471 "error": "Failed to retrieve link analytics",
472 }
473 }
476def get_available_strategies():
477 """Get list of all available search strategies, for the analytics UI.
479 Derived from ``constants.AVAILABLE_STRATEGIES`` -- the single
480 app-wide source of truth for UI-facing strategies (also consumed by
481 ``search_system_factory.create_strategy``, the settings UI, the news
482 subscription form, and the MCP server). This function only adapts the
483 shape: it keeps this endpoint's existing ``{"name", "description"}``
484 wire contract by dropping the ``"label"`` field that the central list
485 carries for dropdown widgets, so response consumers are unaffected.
487 Do not hand-maintain a separate strategy list here -- add new
488 strategies to ``AVAILABLE_STRATEGIES`` in constants.py instead.
489 """
490 return [
491 {"name": strategy["name"], "description": strategy["description"]}
492 for strategy in get_central_available_strategies()
493 ]
496def get_strategy_analytics(period="30d", username=None):
497 """Get strategy usage analytics for the specified period."""
498 try:
499 if not username:
500 return {
501 "strategy_analytics": {
502 "total_research_with_strategy": 0,
503 "total_research": 0,
504 "most_popular_strategy": None,
505 "strategy_usage": [],
506 "strategy_distribution": {},
507 "available_strategies": get_available_strategies(),
508 "error": "No user session",
509 }
510 }
512 # Calculate date range
513 days = get_period_days(period)
515 with get_user_db_session(username) as session:
516 # Check if we have any ResearchStrategy records
517 strategy_count = session.query(ResearchStrategy).count()
519 if strategy_count == 0:
520 logger.warning("No research strategies found in database")
521 return {
522 "strategy_analytics": {
523 "total_research_with_strategy": 0,
524 "total_research": 0,
525 "most_popular_strategy": None,
526 "strategy_usage": [],
527 "strategy_distribution": {},
528 "available_strategies": get_available_strategies(),
529 "message": "Strategy tracking not yet available - run a research to start tracking",
530 }
531 }
533 # Base query for strategy usage (no JOIN needed since we just want strategy counts)
534 query = session.query(
535 ResearchStrategy.strategy_name,
536 func.count(ResearchStrategy.id).label("usage_count"),
537 )
539 # Apply time filter if specified
540 if days:
541 cutoff_date = datetime.now(UTC) - timedelta(days=days)
542 query = query.filter(ResearchStrategy.created_at >= cutoff_date)
544 # Group by strategy and order by usage
545 strategy_results = (
546 query.group_by(ResearchStrategy.strategy_name)
547 .order_by(func.count(ResearchStrategy.id).desc())
548 .all()
549 )
551 # Get total strategy count for percentage calculation
552 total_query = session.query(ResearchStrategy)
553 if days:
554 total_query = total_query.filter(
555 ResearchStrategy.created_at >= cutoff_date
556 )
557 total_research = total_query.count()
559 # Format strategy data
560 strategy_usage = []
561 strategy_distribution = {}
563 for strategy_name, usage_count in strategy_results:
564 percentage = (
565 (usage_count / total_research * 100)
566 if total_research > 0
567 else 0
568 )
569 strategy_usage.append(
570 {
571 "strategy": strategy_name,
572 "count": usage_count,
573 "percentage": round(percentage, 1),
574 }
575 )
576 strategy_distribution[strategy_name] = usage_count
578 # Find most popular strategy
579 most_popular = (
580 strategy_usage[0]["strategy"] if strategy_usage else None
581 )
583 return {
584 "strategy_analytics": {
585 "total_research_with_strategy": sum(
586 item["count"] for item in strategy_usage
587 ),
588 "total_research": total_research,
589 "most_popular_strategy": most_popular,
590 "strategy_usage": strategy_usage,
591 "strategy_distribution": strategy_distribution,
592 "available_strategies": get_available_strategies(),
593 }
594 }
596 except Exception:
597 logger.exception("Error getting strategy analytics")
598 return {
599 "strategy_analytics": {
600 "total_research_with_strategy": 0,
601 "total_research": 0,
602 "most_popular_strategy": None,
603 "strategy_usage": [],
604 "strategy_distribution": {},
605 "available_strategies": get_available_strategies(),
606 "error": "Failed to retrieve strategy data",
607 }
608 }
611def get_rate_limiting_analytics(period="30d", username=None):
612 """Get rate limiting analytics for the specified period."""
613 try:
614 if not username:
615 return {
616 "rate_limiting": {
617 "total_attempts": 0,
618 "successful_attempts": 0,
619 "failed_attempts": 0,
620 "success_rate": 0,
621 "rate_limit_events": 0,
622 "avg_wait_time": 0,
623 "avg_successful_wait": 0,
624 "tracked_engines": 0,
625 "engine_stats": [],
626 "total_engines_tracked": 0,
627 "healthy_engines": 0,
628 "degraded_engines": 0,
629 "poor_engines": 0,
630 "error": "No user session",
631 }
632 }
634 # Calculate date range for timestamp filtering. Same single source
635 # of truth as the token window (PERIOD_DAYS_MAP via
636 # get_period_cutoff) — the local 7d/30d/3m/1y ladder this replaces
637 # computed ALL-TIME for any other period string ('90d', '365d', or
638 # an unknown value) while the token window in the same /api/metrics
639 # response was bounded (#6942). This dict is spread after the token
640 # metrics there, so its 'rate_limiting' value is the one served.
641 cutoff_dt = get_period_cutoff(period)
642 cutoff_time = cutoff_dt.timestamp() if cutoff_dt is not None else 0
644 with get_user_db_session(username) as session:
645 # Rate-limit analytics are derived from RateLimitEstimate, the
646 # learned per-engine wait-time model that production code
647 # actually persists. The raw per-attempt table
648 # (RateLimitAttempt) is intentionally NOT written — attempt
649 # persistence was disabled (commit fef359be9) to avoid DB
650 # locking under parallel search — so the previous code, which
651 # read RateLimitAttempt, always returned an empty panel.
652 #
653 # Limitations of deriving from estimates (documented so the
654 # numbers aren't mistaken for raw-attempt counts):
655 # - total_attempts is each engine's recent rolling window
656 # (capped at rate_limiting.memory_window, default 100), not
657 # a lifetime count.
658 # - rate_limit_events (RateLimitError-specific failures) and a
659 # true per-attempt average wait cannot be reconstructed and
660 # are reported as 0 / the learned base wait respectively.
661 estimates_query = session.query(RateLimitEstimate)
663 # Recency filter uses the estimate's last_updated (epoch
664 # seconds); there is no per-attempt timestamp history.
665 if cutoff_time > 0:
666 estimates_query = estimates_query.filter(
667 RateLimitEstimate.last_updated >= cutoff_time
668 )
670 estimates = estimates_query.all()
672 engine_stats = []
673 total_attempts = 0
674 successful_attempts = 0
675 base_wait_sum = 0.0
677 for estimate in estimates:
678 # success_rate is stored as a 0..1 fraction.
679 success_rate_pct = round(estimate.success_rate * 100, 1)
680 engine_attempts = estimate.total_attempts or 0
681 engine_success = round(engine_attempts * estimate.success_rate)
683 total_attempts += engine_attempts
684 successful_attempts += engine_success
685 base_wait_sum += estimate.base_wait_seconds
687 status = (
688 "healthy"
689 if estimate.success_rate > 0.8
690 else "degraded"
691 if estimate.success_rate > 0.5
692 else "poor"
693 )
695 engine_stats.append(
696 {
697 "engine": estimate.engine_type,
698 "base_wait": estimate.base_wait_seconds,
699 "base_wait_seconds": round(
700 estimate.base_wait_seconds, 2
701 ),
702 "min_wait_seconds": round(estimate.min_wait_seconds, 2),
703 "max_wait_seconds": round(estimate.max_wait_seconds, 2),
704 "success_rate": success_rate_pct,
705 "total_attempts": engine_attempts,
706 "recent_attempts": engine_attempts,
707 "recent_success_rate": success_rate_pct,
708 "attempts": engine_attempts,
709 "status": status,
710 # ISO format already includes timezone
711 "last_updated": datetime.fromtimestamp(
712 estimate.last_updated, UTC
713 ).isoformat(),
714 }
715 )
717 tracked_engines = len(engine_stats)
718 failed_attempts = total_attempts - successful_attempts
719 # base_wait_seconds is the learned optimal (median of recent
720 # successful waits), so it represents both the typical wait and
721 # the typical successful wait; a true per-attempt average needs
722 # the raw attempts table.
723 avg_wait_time = (
724 base_wait_sum / tracked_engines if tracked_engines else 0
725 )
726 avg_successful_wait = avg_wait_time
727 # Not derivable from estimates (needs the raw attempts table).
728 rate_limit_events = 0
730 logger.info(
731 f"Rate limiting analytics from estimates: "
732 f"tracked_engines={tracked_engines}, "
733 f"total_attempts(recent)={total_attempts}"
734 )
736 result = {
737 "rate_limiting": {
738 "total_attempts": total_attempts,
739 "successful_attempts": successful_attempts,
740 "failed_attempts": failed_attempts,
741 "success_rate": (successful_attempts / total_attempts * 100)
742 if total_attempts > 0
743 else 0,
744 "rate_limit_events": rate_limit_events,
745 "avg_wait_time": round(float(avg_wait_time), 2),
746 "avg_successful_wait": round(float(avg_successful_wait), 2),
747 "tracked_engines": tracked_engines,
748 "engine_stats": engine_stats,
749 "total_engines_tracked": tracked_engines,
750 "healthy_engines": len(
751 [s for s in engine_stats if s["status"] == "healthy"]
752 ),
753 "degraded_engines": len(
754 [s for s in engine_stats if s["status"] == "degraded"]
755 ),
756 "poor_engines": len(
757 [s for s in engine_stats if s["status"] == "poor"]
758 ),
759 }
760 }
762 logger.info(
763 f"DEBUG: Returning rate_limiting_analytics result: {result}"
764 )
765 return result
767 except Exception:
768 logger.exception("Error getting rate limiting analytics")
769 return {
770 "rate_limiting": {
771 "total_attempts": 0,
772 "successful_attempts": 0,
773 "failed_attempts": 0,
774 "success_rate": 0,
775 "rate_limit_events": 0,
776 "avg_wait_time": 0,
777 "avg_successful_wait": 0,
778 "tracked_engines": 0,
779 "engine_stats": [],
780 "total_engines_tracked": 0,
781 "healthy_engines": 0,
782 "degraded_engines": 0,
783 "poor_engines": 0,
784 "error": "An internal error occurred while processing the request.",
785 }
786 }
789@router.get("/")
790def metrics_dashboard(
791 request: Request, username: Annotated[str, Depends(require_auth)]
792):
793 """Render the metrics dashboard page."""
794 return templates.TemplateResponse(
795 request=request,
796 name="pages/metrics.html",
797 context={
798 "request": request,
799 "metric_write_health": get_metric_write_health(username),
800 "metric_write_reason_labels": METRIC_WRITE_REASON_LABELS,
801 },
802 )
805@router.get("/context-overflow")
806def context_overflow_page(
807 request: Request, username: Annotated[str, Depends(require_auth)]
808):
809 """Context overflow analytics page."""
810 return templates.TemplateResponse(
811 request=request,
812 name="pages/context_overflow.html",
813 context={"request": request},
814 )
817@router.get("/api/metrics")
818def api_metrics(
819 request: Request, username: Annotated[str, Depends(require_auth)]
820):
821 """Get overall metrics data."""
822 logger.debug("api_metrics endpoint called")
823 try:
824 # Get username from session
826 # Get time period and research mode from query parameters
827 period = request.query_params.get("period", "30d")
828 research_mode = request.query_params.get("mode", "all")
830 token_counter = TokenCounter()
831 search_tracker = get_search_tracker()
833 # Get both token and search metrics
834 token_metrics = token_counter.get_overall_metrics(
835 period=period, research_mode=research_mode, username=username
836 )
837 search_metrics = search_tracker.get_search_metrics(
838 period=period,
839 research_mode=research_mode,
840 username=username,
841 )
843 # Get user satisfaction rating data
844 try:
845 with get_user_db_session(username) as session:
846 # Build base query with time filter
847 ratings_query = session.query(ResearchRating)
848 time_condition = get_time_filter_condition(
849 period, ResearchRating.created_at
850 )
851 if time_condition is not None:
852 ratings_query = ratings_query.filter(time_condition)
854 # Get average rating
855 avg_rating = ratings_query.with_entities(
856 func.avg(ResearchRating.rating).label("avg_rating")
857 ).scalar()
859 # Get total rating count
860 total_ratings = ratings_query.count()
862 user_satisfaction = {
863 "avg_rating": round(avg_rating, 1) if avg_rating else None,
864 "total_ratings": total_ratings,
865 }
866 except Exception:
867 logger.exception("Error getting user satisfaction data")
868 user_satisfaction = {"avg_rating": None, "total_ratings": 0}
870 # Get strategy analytics
871 strategy_data = get_strategy_analytics(period, username)
872 logger.debug(f"strategy_data keys: {list(strategy_data.keys())}")
874 # Get rate limiting analytics
875 rate_limiting_data = get_rate_limiting_analytics(period, username)
876 logger.debug(f"rate_limiting_data: {rate_limiting_data}")
877 logger.debug(
878 f"rate_limiting_data keys: {list(rate_limiting_data.keys())}"
879 )
881 # Truncation summary surfaced on the main dashboard. Failure sentinel
882 # is None (not 0): a real zero means "no truncation", so falling back
883 # to 0 on error would silently flip a red signal green.
884 context_overflow_data = {
885 "truncation_rate": None,
886 "avg_tokens_truncated": None,
887 }
888 try:
889 with get_user_db_session(username) as session:
890 # Honor the dashboard's research_mode filter the same way the
891 # rest of api_metrics() does (token_metrics, search_metrics,
892 # etc.). Without this the panel ignores mode toggles.
893 summary = get_context_overflow_truncation_summary(
894 session, period, research_mode=research_mode
895 )
896 context_overflow_data = {
897 "truncation_rate": round(summary["truncation_rate"], 1),
898 "avg_tokens_truncated": int(summary["avg_tokens_truncated"]),
899 }
900 except Exception:
901 logger.exception(
902 "Error getting context overflow summary for /api/metrics"
903 )
905 # Combine metrics
906 combined_metrics = {
907 **token_metrics,
908 **search_metrics,
909 **strategy_data,
910 **rate_limiting_data,
911 **context_overflow_data,
912 "user_satisfaction": user_satisfaction,
913 }
915 logger.debug(f"combined_metrics keys: {list(combined_metrics.keys())}")
916 logger.debug(
917 f"combined_metrics['rate_limiting']: {combined_metrics.get('rate_limiting', 'NOT FOUND')}"
918 )
920 return {
921 "status": "success",
922 "metrics": combined_metrics,
923 "period": period,
924 "research_mode": research_mode,
925 }
927 except Exception:
928 logger.exception("Error getting metrics")
929 return JSONResponse(
930 {
931 "status": "error",
932 "message": "An internal error occurred. Please try again later.",
933 },
934 status_code=500,
935 )
938@router.get("/api/rate-limiting")
939def get_rate_limiting_metrics(
940 request: Request, username: Annotated[str, Depends(require_auth)]
941):
942 """Get detailed rate limiting metrics."""
943 logger.info("DEBUG: get_rate_limiting_metrics endpoint called")
944 try:
945 period = request.query_params.get("period", "30d")
946 rate_limiting_data = get_rate_limiting_analytics(period, username)
948 return {
949 "status": "success",
950 "data": rate_limiting_data,
951 "period": period,
952 }
954 except Exception:
955 logger.exception("Error getting rate limiting metrics")
956 return JSONResponse(
957 {
958 "status": "error",
959 "message": "Failed to retrieve rate limiting metrics",
960 },
961 status_code=500,
962 )
965@router.get("/api/rate-limiting/current")
966def api_current_rate_limits(
967 request: Request, username: Annotated[str, Depends(require_auth)]
968):
969 """Get current rate limit estimates for all engines."""
970 try:
971 with get_user_db_session(username) as session:
972 estimates = (
973 session.query(RateLimitEstimate)
974 .order_by(RateLimitEstimate.engine_type)
975 .all()
976 )
978 current_limits = []
979 for est in estimates:
980 current_limits.append(
981 {
982 "engine_type": est.engine_type,
983 "base_wait_seconds": round(est.base_wait_seconds, 2),
984 "min_wait_seconds": round(est.min_wait_seconds, 2),
985 "max_wait_seconds": round(est.max_wait_seconds, 2),
986 "success_rate": round(est.success_rate * 100, 1),
987 "total_attempts": est.total_attempts,
988 "last_updated": datetime.fromtimestamp(
989 est.last_updated, UTC
990 ).isoformat(),
991 "status": "healthy"
992 if est.success_rate > 0.8
993 else "degraded"
994 if est.success_rate > 0.5
995 else "poor",
996 }
997 )
999 return {
1000 "status": "success",
1001 "current_limits": current_limits,
1002 "timestamp": datetime.now(UTC).isoformat(),
1003 }
1004 except Exception:
1005 logger.exception("Error getting current rate limits")
1006 return JSONResponse(
1007 {
1008 "status": "error",
1009 "message": "Failed to retrieve current rate limits",
1010 },
1011 status_code=500,
1012 )
1015@router.get("/api/metrics/research/{research_id}/links")
1016def api_research_link_metrics(
1017 request: Request,
1018 research_id,
1019 username: Annotated[str, Depends(require_auth)],
1020):
1021 """Get link analytics for a specific research."""
1022 try:
1023 with get_user_db_session(username) as session:
1024 # Get all resources for this specific research
1025 resources = (
1026 session.query(ResearchResource)
1027 .filter(ResearchResource.research_id == research_id)
1028 .all()
1029 )
1031 if not resources:
1032 return {
1033 "status": "success",
1034 "data": {
1035 "total_links": 0,
1036 "unique_domains": 0,
1037 "domains": [],
1038 "category_distribution": {},
1039 "domain_categories": {},
1040 "resources": [],
1041 },
1042 }
1044 # Extract domain information
1045 domain_counts: dict[str, Any] = {}
1047 # Generic category counting from LLM classifications
1048 category_counts: dict[str, Any] = {}
1050 # First pass: collect all domains
1051 all_domains = set()
1052 for resource in resources:
1053 if resource.url:
1054 domain = _extract_domain(resource.url)
1055 if domain:
1056 all_domains.add(domain)
1058 # Batch load all domain classifications in one query (fix N+1)
1059 domain_classifications_map = {}
1060 if all_domains: 1060 ↛ 1072line 1060 didn't jump to line 1072 because the condition on line 1060 was always true
1061 all_classifications = (
1062 session.query(DomainClassification)
1063 .filter(DomainClassification.domain.in_(all_domains))
1064 .all()
1065 )
1066 for classification in all_classifications:
1067 domain_classifications_map[classification.domain] = (
1068 classification
1069 )
1071 # Second pass: process resources with pre-loaded classifications
1072 for resource in resources:
1073 if resource.url:
1074 try:
1075 domain = _extract_domain(resource.url)
1076 if not domain:
1077 continue
1079 domain_counts[domain] = domain_counts.get(domain, 0) + 1
1081 # Count categories from pre-loaded classifications (no N+1)
1082 classification = domain_classifications_map.get(domain)
1083 if classification:
1084 category = classification.category
1085 category_counts[category] = (
1086 category_counts.get(category, 0) + 1
1087 )
1088 else:
1089 category_counts["Unclassified"] = (
1090 category_counts.get("Unclassified", 0) + 1
1091 )
1092 except (AttributeError, KeyError) as e:
1093 logger.debug(
1094 f"Error classifying domain {domain}: "
1095 f"{type(e).__name__}"
1096 )
1098 # Sort domains by count
1099 sorted_domains = sorted(
1100 domain_counts.items(), key=lambda x: x[1], reverse=True
1101 )
1103 return {
1104 "status": "success",
1105 "data": {
1106 "total_links": len(resources),
1107 "unique_domains": len(domain_counts),
1108 "domains": [
1109 {
1110 "domain": domain,
1111 "count": count,
1112 "percentage": round(
1113 count / len(resources) * 100, 1
1114 ),
1115 }
1116 for domain, count in sorted_domains[
1117 :20
1118 ] # Top 20 domains
1119 ],
1120 "category_distribution": category_counts,
1121 "domain_categories": category_counts, # Generic categories from LLM
1122 "resources": [
1123 {
1124 "title": r.title or "Untitled",
1125 "url": r.url,
1126 "preview": r.content_preview[:200]
1127 if r.content_preview
1128 else None,
1129 }
1130 for r in resources[:10] # First 10 resources
1131 ],
1132 },
1133 }
1135 except Exception:
1136 logger.exception("Error getting research link metrics")
1137 return JSONResponse(
1138 {"status": "error", "message": "Failed to retrieve link metrics"},
1139 status_code=500,
1140 )
1143@router.get("/api/metrics/research/{research_id}")
1144def api_research_metrics(
1145 request: Request,
1146 research_id,
1147 username: Annotated[str, Depends(require_auth)],
1148):
1149 """Get metrics for a specific research."""
1150 try:
1151 token_counter = TokenCounter()
1152 metrics = token_counter.get_research_metrics(
1153 research_id, username=username
1154 )
1155 return {"status": "success", "metrics": metrics}
1156 except Exception:
1157 logger.exception("Error getting research metrics")
1158 return JSONResponse(
1159 {
1160 "status": "error",
1161 "message": "An internal error occurred. Please try again later.",
1162 },
1163 status_code=500,
1164 )
1167@router.get("/api/metrics/research/{research_id}/timeline")
1168def api_research_timeline_metrics(
1169 request: Request,
1170 research_id,
1171 username: Annotated[str, Depends(require_auth)],
1172):
1173 """Get timeline metrics for a specific research."""
1174 try:
1175 token_counter = TokenCounter()
1176 timeline_metrics = token_counter.get_research_timeline_metrics(
1177 research_id, username=username
1178 )
1179 return {"status": "success", "metrics": timeline_metrics}
1180 except Exception:
1181 logger.exception("Error getting research timeline metrics")
1182 return JSONResponse(
1183 {
1184 "status": "error",
1185 "message": "An internal error occurred. Please try again later.",
1186 },
1187 status_code=500,
1188 )
1191@router.get("/api/metrics/research/{research_id}/search")
1192def api_research_search_metrics(
1193 request: Request,
1194 research_id,
1195 username: Annotated[str, Depends(require_auth)],
1196):
1197 """Get search metrics for a specific research."""
1198 try:
1199 search_tracker = get_search_tracker()
1200 search_metrics = search_tracker.get_research_search_metrics(
1201 research_id, username=username
1202 )
1203 return {"status": "success", "metrics": search_metrics}
1204 except Exception:
1205 logger.exception("Error getting research search metrics")
1206 return JSONResponse(
1207 {
1208 "status": "error",
1209 "message": "An internal error occurred. Please try again later.",
1210 },
1211 status_code=500,
1212 )
1215@router.get("/api/metrics/enhanced")
1216def api_enhanced_metrics(
1217 request: Request, username: Annotated[str, Depends(require_auth)]
1218):
1219 """Get enhanced Phase 1 tracking metrics."""
1220 try:
1221 # Get time period and research mode from query parameters
1222 period = request.query_params.get("period", "30d")
1223 research_mode = request.query_params.get("mode", "all")
1225 token_counter = TokenCounter()
1226 search_tracker = get_search_tracker()
1228 enhanced_metrics = token_counter.get_enhanced_metrics(
1229 period=period, research_mode=research_mode, username=username
1230 )
1232 # Add search time series data for the chart
1233 search_time_series = search_tracker.get_search_time_series(
1234 period=period,
1235 research_mode=research_mode,
1236 username=username,
1237 )
1238 enhanced_metrics["search_time_series"] = search_time_series
1240 # Add rating analytics
1241 rating_analytics = get_rating_analytics(period, research_mode, username)
1242 enhanced_metrics.update(rating_analytics)
1244 return {
1245 "status": "success",
1246 "metrics": enhanced_metrics,
1247 "period": period,
1248 "research_mode": research_mode,
1249 }
1251 except Exception:
1252 logger.exception("Error getting enhanced metrics")
1253 return JSONResponse(
1254 {
1255 "status": "error",
1256 "message": "An internal error occurred. Please try again later.",
1257 },
1258 status_code=500,
1259 )
1262@router.get("/api/ratings/{research_id}")
1263def api_get_research_rating(
1264 request: Request,
1265 research_id,
1266 username: Annotated[str, Depends(require_auth)],
1267):
1268 """Get rating for a specific research session."""
1269 try:
1270 with get_user_db_session(username) as session:
1271 rating = (
1272 session.query(ResearchRating)
1273 .filter_by(research_id=research_id)
1274 .first()
1275 )
1277 if rating:
1278 return {
1279 "status": "success",
1280 "rating": rating.rating,
1281 "created_at": rating.created_at.isoformat(),
1282 "updated_at": rating.updated_at.isoformat(),
1283 }
1284 return {"status": "success", "rating": None}
1286 except Exception:
1287 logger.exception("Error getting research rating")
1288 return JSONResponse(
1289 {
1290 "status": "error",
1291 "message": "An internal error occurred. Please try again later.",
1292 },
1293 status_code=500,
1294 )
1297@router.post("/api/ratings/{research_id}")
1298async def api_save_research_rating(
1299 request: Request,
1300 research_id,
1301 username: Annotated[str, Depends(require_auth)],
1302):
1303 """Save or update rating for a specific research session."""
1304 # Parse before the broad exception handler so malformed bodies retain
1305 # this route's validation envelope instead of becoming server errors.
1306 data, body_error = await read_json_dict(
1307 request, "status", "Request body must be valid JSON"
1308 )
1309 if body_error is not None:
1310 return body_error
1311 try:
1312 rating_value = data.get("rating")
1314 if (
1315 not isinstance(rating_value, int)
1316 or isinstance(rating_value, bool)
1317 or rating_value < 1
1318 or rating_value > 5
1319 ):
1320 return JSONResponse(
1321 {
1322 "status": "error",
1323 "message": "Rating must be an integer between 1 and 5",
1324 },
1325 status_code=400,
1326 )
1328 # Optional sub-dimension fields (1-5)
1329 sub_dimensions = {}
1330 for field in ("accuracy", "completeness", "relevance", "readability"):
1331 val = data.get(field)
1332 if val is not None:
1333 if (
1334 not isinstance(val, int)
1335 or isinstance(val, bool)
1336 or val < 1
1337 or val > 5
1338 ):
1339 return JSONResponse(
1340 {
1341 "status": "error",
1342 "message": f"{field} must be an integer between 1 and 5",
1343 },
1344 status_code=400,
1345 )
1346 sub_dimensions[field] = val
1348 feedback_text = data.get("feedback")
1349 if feedback_text is not None:
1350 if not isinstance(feedback_text, str):
1351 return JSONResponse(
1352 {
1353 "status": "error",
1354 "message": "feedback must be a string",
1355 },
1356 status_code=400,
1357 )
1358 if len(feedback_text) > 10000:
1359 return JSONResponse(
1360 {
1361 "status": "error",
1362 "message": "feedback must be 10000 characters or fewer",
1363 },
1364 status_code=400,
1365 )
1367 def _save_sync():
1368 with get_user_db_session(username) as session:
1369 existing_rating = (
1370 session.query(ResearchRating)
1371 .filter_by(research_id=research_id)
1372 .first()
1373 )
1374 if existing_rating:
1375 existing_rating.rating = rating_value
1376 existing_rating.updated_at = func.now()
1377 for field, val in sub_dimensions.items():
1378 setattr(existing_rating, field, val)
1379 if feedback_text is not None:
1380 existing_rating.feedback = feedback_text
1381 else:
1382 session.add(
1383 ResearchRating(
1384 research_id=research_id,
1385 rating=rating_value,
1386 feedback=feedback_text,
1387 **sub_dimensions,
1388 )
1389 )
1390 session.commit()
1392 await run_db_sync(_save_sync)
1393 return {
1394 "status": "success",
1395 "message": "Rating saved successfully",
1396 "rating": rating_value,
1397 }
1399 except Exception:
1400 logger.exception("Error saving research rating")
1401 return JSONResponse(
1402 {
1403 "status": "error",
1404 "message": "An internal error occurred. Please try again later.",
1405 },
1406 status_code=500,
1407 )
1410@router.get("/star-reviews")
1411def star_reviews(
1412 request: Request, username: Annotated[str, Depends(require_auth)]
1413):
1414 """Display star reviews metrics page."""
1415 return templates.TemplateResponse(
1416 request=request,
1417 name="pages/star_reviews.html",
1418 context={"request": request},
1419 )
1422@router.get("/costs")
1423def cost_analytics(
1424 request: Request, username: Annotated[str, Depends(require_auth)]
1425):
1426 """Display cost analytics page."""
1427 return templates.TemplateResponse(
1428 request=request,
1429 name="pages/cost_analytics.html",
1430 context={"request": request},
1431 )
1434@router.get("/api/star-reviews")
1435def api_star_reviews(
1436 request: Request, username: Annotated[str, Depends(require_auth)]
1437):
1438 """Get star reviews analytics data."""
1439 try:
1440 period = request.query_params.get("period", "30d")
1442 with get_user_db_session(username) as session:
1443 # Build base query with time filter
1444 base_query = session.query(ResearchRating)
1445 time_condition = get_time_filter_condition(
1446 period, ResearchRating.created_at
1447 )
1448 if time_condition is not None:
1449 base_query = base_query.filter(time_condition)
1451 # Overall rating statistics
1452 overall_stats = session.query(
1453 func.avg(ResearchRating.rating).label("avg_rating"),
1454 func.count(ResearchRating.rating).label("total_ratings"),
1455 func.sum(case((ResearchRating.rating == 5, 1), else_=0)).label(
1456 "five_star"
1457 ),
1458 func.sum(case((ResearchRating.rating == 4, 1), else_=0)).label(
1459 "four_star"
1460 ),
1461 func.sum(case((ResearchRating.rating == 3, 1), else_=0)).label(
1462 "three_star"
1463 ),
1464 func.sum(case((ResearchRating.rating == 2, 1), else_=0)).label(
1465 "two_star"
1466 ),
1467 func.sum(case((ResearchRating.rating == 1, 1), else_=0)).label(
1468 "one_star"
1469 ),
1470 )
1472 if time_condition is not None:
1473 overall_stats = overall_stats.filter(time_condition)
1475 overall_stats = overall_stats.first()
1477 # Ratings by LLM model. A research has many token_usage rows (one per
1478 # LLM call), so joining TokenUsage directly fans out and multiplies the
1479 # counts/averages. Collapse to distinct (research_id, model) pairs first
1480 # so each rating is counted once per model it actually used.
1481 # Normalize the "missing model" sentinels — empty/whitespace and the
1482 # lowercase "unknown" fallback the token counter writes — to a single
1483 # "Unknown" *inside* the subquery, so DISTINCT dedups on the normalized
1484 # value. (Normalizing only at group_by time lets "" and "unknown"
1485 # survive DISTINCT as separate rows and double-count one rating into the
1486 # merged bucket.) Real model names keep their original casing via else_.
1487 normalized_model = case(
1488 (
1489 func.lower(func.trim(TokenUsage.model_name)).in_(
1490 ["", "unknown"]
1491 ),
1492 "Unknown",
1493 ),
1494 else_=TokenUsage.model_name,
1495 )
1496 research_models = (
1497 session.query(
1498 TokenUsage.research_id.label("research_id"),
1499 normalized_model.label("model_name"),
1500 )
1501 .distinct()
1502 .subquery()
1503 )
1504 # A rating with no token rows has no subquery match -> NULL via the
1505 # outerjoin -> "Unknown".
1506 model_label = func.coalesce(research_models.c.model_name, "Unknown")
1508 llm_ratings_query = (
1509 session.query(
1510 model_label.label("model"),
1511 func.avg(ResearchRating.rating).label("avg_rating"),
1512 func.count(ResearchRating.rating).label("rating_count"),
1513 func.sum(
1514 case((ResearchRating.rating >= 4, 1), else_=0)
1515 ).label("positive_ratings"),
1516 func.sum(
1517 case((ResearchRating.rating == 3, 1), else_=0)
1518 ).label("neutral_ratings"),
1519 func.sum(
1520 case((ResearchRating.rating <= 2, 1), else_=0)
1521 ).label("negative_ratings"),
1522 )
1523 .select_from(ResearchRating)
1524 .outerjoin(
1525 research_models,
1526 ResearchRating.research_id == research_models.c.research_id,
1527 )
1528 )
1530 if time_condition is not None:
1531 llm_ratings_query = llm_ratings_query.filter(time_condition)
1533 llm_ratings = (
1534 llm_ratings_query.group_by(model_label)
1535 .order_by(func.avg(ResearchRating.rating).desc())
1536 .all()
1537 )
1539 # Ratings by search engine. Same one-to-many fan-out concern as the LLM
1540 # query — collapse to distinct (research_id, search_engine) pairs first.
1541 # search_engine_selected is NULL on non-search LLM-call rows (and is
1542 # occasionally an empty/whitespace string); exclude both so a research
1543 # that used a real engine isn't ALSO attributed to the "Unknown" bucket.
1544 # A research with no recorded engine still falls through the outerjoin to
1545 # "Unknown" exactly once.
1546 research_engines = (
1547 session.query(
1548 TokenUsage.research_id.label("research_id"),
1549 TokenUsage.search_engine_selected.label("search_engine"),
1550 )
1551 .filter(
1552 func.trim(
1553 func.coalesce(TokenUsage.search_engine_selected, "")
1554 )
1555 != ""
1556 )
1557 .distinct()
1558 .subquery()
1559 )
1560 engine_label = func.coalesce(
1561 research_engines.c.search_engine, "Unknown"
1562 )
1564 search_engine_ratings_query = (
1565 session.query(
1566 engine_label.label("search_engine"),
1567 func.avg(ResearchRating.rating).label("avg_rating"),
1568 func.count(ResearchRating.rating).label("rating_count"),
1569 func.sum(
1570 case((ResearchRating.rating >= 4, 1), else_=0)
1571 ).label("positive_ratings"),
1572 )
1573 .select_from(ResearchRating)
1574 .outerjoin(
1575 research_engines,
1576 ResearchRating.research_id
1577 == research_engines.c.research_id,
1578 )
1579 )
1581 if time_condition is not None:
1582 search_engine_ratings_query = (
1583 search_engine_ratings_query.filter(time_condition)
1584 )
1586 search_engine_ratings = (
1587 search_engine_ratings_query.group_by(engine_label)
1588 .having(func.count(ResearchRating.rating) > 0)
1589 .order_by(func.avg(ResearchRating.rating).desc())
1590 .all()
1591 )
1593 # Rating trends over time
1594 rating_trends_query = session.query(
1595 func.date(ResearchRating.created_at).label("date"),
1596 func.avg(ResearchRating.rating).label("avg_rating"),
1597 func.count(ResearchRating.rating).label("daily_count"),
1598 )
1600 if time_condition is not None:
1601 rating_trends_query = rating_trends_query.filter(time_condition)
1603 rating_trends = (
1604 rating_trends_query.group_by(
1605 func.date(ResearchRating.created_at)
1606 )
1607 .order_by("date")
1608 .all()
1609 )
1611 # Ratings by research mode
1612 mode_ratings_query = session.query(
1613 func.coalesce(ResearchHistory.mode, "Unknown").label("mode"),
1614 func.avg(ResearchRating.rating).label("avg_rating"),
1615 func.count(ResearchRating.rating).label("rating_count"),
1616 func.sum(case((ResearchRating.rating >= 4, 1), else_=0)).label(
1617 "positive_ratings"
1618 ),
1619 ).outerjoin(
1620 ResearchHistory,
1621 ResearchRating.research_id == ResearchHistory.id,
1622 )
1624 if time_condition is not None:
1625 mode_ratings_query = mode_ratings_query.filter(time_condition)
1627 mode_ratings = (
1628 mode_ratings_query.group_by(ResearchHistory.mode)
1629 .having(func.count(ResearchRating.rating) > 0)
1630 .order_by(func.avg(ResearchRating.rating).desc())
1631 .all()
1632 )
1634 # Quality dimension averages. Each sub-dimension is independently
1635 # nullable, so count them separately; dimension_count reflects rows
1636 # with ANY dimension filled, so the radar isn't hidden when only some
1637 # dimensions have data.
1638 dimension_stats = session.query(
1639 func.avg(ResearchRating.accuracy).label("avg_accuracy"),
1640 func.avg(ResearchRating.completeness).label("avg_completeness"),
1641 func.avg(ResearchRating.relevance).label("avg_relevance"),
1642 func.avg(ResearchRating.readability).label("avg_readability"),
1643 func.count(ResearchRating.accuracy).label("count_accuracy"),
1644 func.count(ResearchRating.completeness).label(
1645 "count_completeness"
1646 ),
1647 func.count(ResearchRating.relevance).label("count_relevance"),
1648 func.count(ResearchRating.readability).label(
1649 "count_readability"
1650 ),
1651 func.count(
1652 func.coalesce(
1653 ResearchRating.accuracy,
1654 ResearchRating.completeness,
1655 ResearchRating.relevance,
1656 ResearchRating.readability,
1657 )
1658 ).label("dimension_count"),
1659 )
1661 if time_condition is not None:
1662 dimension_stats = dimension_stats.filter(time_condition)
1664 dimension_stats = dimension_stats.first()
1666 # Recent ratings with research details. Join a one-row-per-research
1667 # model subquery instead of TokenUsage directly: a research has many
1668 # token_usage rows, which would fan out and make limit(20) return far
1669 # fewer than 20 unique ratings (the same rating duplicated per row).
1670 recent_model_subq = (
1671 session.query(
1672 TokenUsage.research_id.label("research_id"),
1673 func.max(TokenUsage.model_name).label("model_name"),
1674 )
1675 .group_by(TokenUsage.research_id)
1676 .subquery()
1677 )
1679 recent_ratings_query = (
1680 session.query(
1681 ResearchRating.rating,
1682 ResearchRating.created_at,
1683 ResearchRating.research_id,
1684 ResearchHistory.query,
1685 ResearchHistory.mode,
1686 recent_model_subq.c.model_name,
1687 ResearchRating.feedback,
1688 )
1689 .outerjoin(
1690 ResearchHistory,
1691 ResearchRating.research_id == ResearchHistory.id,
1692 )
1693 .outerjoin(
1694 recent_model_subq,
1695 ResearchRating.research_id
1696 == recent_model_subq.c.research_id,
1697 )
1698 )
1700 if time_condition is not None:
1701 recent_ratings_query = recent_ratings_query.filter(
1702 time_condition
1703 )
1705 recent_ratings = (
1706 recent_ratings_query.order_by(ResearchRating.created_at.desc())
1707 .limit(20)
1708 .all()
1709 )
1711 # Recent feedback entries (non-empty feedback text)
1712 recent_feedback_query = session.query(
1713 ResearchRating.feedback,
1714 ResearchRating.rating,
1715 ResearchRating.created_at,
1716 ResearchRating.research_id,
1717 ).filter(
1718 ResearchRating.feedback.isnot(None),
1719 ResearchRating.feedback != "",
1720 )
1722 if time_condition is not None:
1723 recent_feedback_query = recent_feedback_query.filter(
1724 time_condition
1725 )
1727 recent_feedback = (
1728 recent_feedback_query.order_by(ResearchRating.created_at.desc())
1729 .limit(20)
1730 .all()
1731 )
1733 return {
1734 "overall_stats": {
1735 "avg_rating": round(overall_stats.avg_rating or 0, 2),
1736 "total_ratings": overall_stats.total_ratings or 0,
1737 "rating_distribution": {
1738 "5": overall_stats.five_star or 0,
1739 "4": overall_stats.four_star or 0,
1740 "3": overall_stats.three_star or 0,
1741 "2": overall_stats.two_star or 0,
1742 "1": overall_stats.one_star or 0,
1743 },
1744 },
1745 "llm_ratings": [
1746 {
1747 "model": rating.model,
1748 "avg_rating": round(rating.avg_rating or 0, 2),
1749 "rating_count": rating.rating_count or 0,
1750 "positive_ratings": rating.positive_ratings or 0,
1751 "neutral_ratings": rating.neutral_ratings or 0,
1752 "negative_ratings": rating.negative_ratings or 0,
1753 "satisfaction_rate": round(
1754 (rating.positive_ratings or 0)
1755 / max(rating.rating_count or 1, 1)
1756 * 100,
1757 1,
1758 ),
1759 }
1760 for rating in llm_ratings
1761 ],
1762 "search_engine_ratings": [
1763 {
1764 "search_engine": rating.search_engine,
1765 "avg_rating": round(rating.avg_rating or 0, 2),
1766 "rating_count": rating.rating_count or 0,
1767 "positive_ratings": rating.positive_ratings or 0,
1768 "satisfaction_rate": round(
1769 (rating.positive_ratings or 0)
1770 / max(rating.rating_count or 1, 1)
1771 * 100,
1772 1,
1773 ),
1774 }
1775 for rating in search_engine_ratings
1776 ],
1777 "rating_trends": [
1778 {
1779 "date": str(trend.date),
1780 "avg_rating": round(trend.avg_rating or 0, 2),
1781 "count": trend.daily_count or 0,
1782 }
1783 for trend in rating_trends
1784 ],
1785 "recent_ratings": [
1786 {
1787 "rating": rating.rating,
1788 "created_at": rating.created_at.isoformat()
1789 if rating.created_at
1790 else None,
1791 "research_id": rating.research_id,
1792 "query": (
1793 rating.query
1794 if rating.query
1795 else f"Research Session #{rating.research_id}"
1796 ),
1797 "mode": rating.mode
1798 if rating.mode
1799 else "Standard Research",
1800 "llm_model": (
1801 rating.model_name
1802 if rating.model_name
1803 else "LLM Model"
1804 ),
1805 "feedback": rating.feedback,
1806 }
1807 for rating in recent_ratings
1808 ],
1809 "mode_ratings": [
1810 {
1811 "mode": rating.mode,
1812 "avg_rating": round(rating.avg_rating or 0, 2),
1813 "rating_count": rating.rating_count or 0,
1814 "satisfaction_rate": round(
1815 (rating.positive_ratings or 0)
1816 / max(rating.rating_count or 1, 1)
1817 * 100,
1818 1,
1819 ),
1820 }
1821 for rating in mode_ratings
1822 ],
1823 "quality_dimensions": {
1824 "avg_accuracy": round(dimension_stats.avg_accuracy, 2)
1825 if dimension_stats.avg_accuracy is not None
1826 else None,
1827 "avg_completeness": round(
1828 dimension_stats.avg_completeness, 2
1829 )
1830 if dimension_stats.avg_completeness is not None
1831 else None,
1832 "avg_relevance": round(dimension_stats.avg_relevance, 2)
1833 if dimension_stats.avg_relevance is not None
1834 else None,
1835 "avg_readability": round(dimension_stats.avg_readability, 2)
1836 if dimension_stats.avg_readability is not None
1837 else None,
1838 "dimension_count": dimension_stats.dimension_count or 0,
1839 "dimension_counts": {
1840 "accuracy": dimension_stats.count_accuracy or 0,
1841 "completeness": dimension_stats.count_completeness or 0,
1842 "relevance": dimension_stats.count_relevance or 0,
1843 "readability": dimension_stats.count_readability or 0,
1844 },
1845 },
1846 "recent_feedback": [
1847 {
1848 "feedback": rating.feedback,
1849 "rating": rating.rating,
1850 "created_at": rating.created_at.isoformat()
1851 if rating.created_at
1852 else None,
1853 "research_id": rating.research_id,
1854 }
1855 for rating in recent_feedback
1856 ],
1857 }
1859 except Exception:
1860 logger.exception("Error getting star reviews data")
1861 return JSONResponse(
1862 {"error": "An internal error occurred. Please try again later."},
1863 status_code=500,
1864 )
1867@router.get("/api/pricing")
1868def api_pricing(
1869 request: Request, username: Annotated[str, Depends(require_auth)]
1870):
1871 """Get current LLM pricing data."""
1872 try:
1873 from ...metrics.pricing.pricing_fetcher import PricingFetcher
1875 # Use static pricing data instead of async
1876 fetcher = PricingFetcher()
1877 pricing_data = fetcher.static_pricing
1879 return {
1880 "status": "success",
1881 "pricing": pricing_data,
1882 "last_updated": datetime.now(UTC).isoformat(),
1883 "note": "Pricing data is from static configuration. Real-time APIs not available for most providers.",
1884 }
1886 except Exception:
1887 logger.exception("Error fetching pricing data")
1888 return JSONResponse({"error": "Internal Server Error"}, status_code=500)
1891@router.get("/api/pricing/{model_name}")
1892def api_model_pricing(
1893 request: Request,
1894 model_name,
1895 username: Annotated[str, Depends(require_auth)],
1896):
1897 """Get pricing for a specific model."""
1898 try:
1899 # Optional provider parameter
1900 provider = request.query_params.get("provider")
1902 from ...metrics.pricing.cost_calculator import CostCalculator
1904 # Use synchronous approach with cached/static pricing
1905 calculator = CostCalculator()
1906 pricing = calculator.calculate_cost_sync(
1907 model_name, 1000, 1000, provider=provider
1908 ).get("pricing_used", {})
1910 return {
1911 "status": "success",
1912 "model": model_name,
1913 "provider": provider,
1914 "pricing": pricing,
1915 "last_updated": datetime.now(UTC).isoformat(),
1916 }
1918 except Exception as exc:
1919 logger.opt(exception=False).error(
1920 "Error getting pricing for model: {} ({})",
1921 _log_value_preview(model_name),
1922 type(exc).__name__,
1923 )
1924 return JSONResponse(
1925 {"error": "An internal error occurred"}, status_code=500
1926 )
1929@router.post("/api/cost-calculation")
1930async def api_cost_calculation(
1931 request: Request, username: Annotated[str, Depends(require_auth)]
1932):
1933 """Calculate cost for token usage."""
1934 # Parse before the broad exception handler so malformed bodies retain
1935 # this route's validation envelope instead of becoming server errors.
1936 data, body_error = await read_json_dict(
1937 request, "simple", "No data provided"
1938 )
1939 if body_error is not None:
1940 return body_error
1941 try:
1942 model_name = data.get("model_name")
1943 provider = data.get("provider") # Optional provider parameter
1944 prompt_tokens = data.get("prompt_tokens", 0)
1945 completion_tokens = data.get("completion_tokens", 0)
1947 if not model_name:
1948 return JSONResponse(
1949 {"error": "model_name is required"}, status_code=400
1950 )
1951 if provider is not None and not isinstance(provider, str):
1952 return JSONResponse(
1953 {"error": "provider must be a string"}, status_code=400
1954 )
1956 from ...metrics.pricing.cost_calculator import CostCalculator
1957 import asyncio as _asyncio
1959 # Offload the sync pricing lookup so the event loop isn't blocked
1960 # waiting on its I/O (pricing table fetch / file read).
1961 calculator = CostCalculator()
1962 cost_data = await _asyncio.to_thread(
1963 calculator.calculate_cost_sync,
1964 model_name,
1965 prompt_tokens,
1966 completion_tokens,
1967 provider=provider,
1968 )
1970 return {
1971 "status": "success",
1972 "model_name": model_name,
1973 "provider": provider,
1974 "prompt_tokens": prompt_tokens,
1975 "completion_tokens": completion_tokens,
1976 "total_tokens": prompt_tokens + completion_tokens,
1977 **cost_data,
1978 }
1980 except Exception:
1981 logger.exception("Error calculating cost")
1982 return JSONResponse(
1983 {"error": "An internal error occurred"}, status_code=500
1984 )
1987@router.get("/api/research-costs/{research_id}")
1988def api_research_costs(
1989 request: Request,
1990 research_id,
1991 username: Annotated[str, Depends(require_auth)],
1992):
1993 """Get cost analysis for a specific research session."""
1994 try:
1995 from ...metrics.pricing.cost_calculator import (
1996 CostCalculator,
1997 summarize_costs,
1998 )
2000 with get_user_db_session(username) as session:
2001 # Get token usage records for this research
2002 usage_records = (
2003 session.query(TokenUsage)
2004 .filter(TokenUsage.research_id == research_id)
2005 .all()
2006 )
2008 if not usage_records:
2009 return {
2010 "status": "success",
2011 "research_id": research_id,
2012 **summarize_costs([]),
2013 "message": "No token usage data found for this research session",
2014 }
2016 # Convert to dict format for cost calculation
2017 usage_data = []
2018 for record in usage_records:
2019 usage_data.append(
2020 {
2021 "model_name": record.model_name,
2022 "provider": getattr(
2023 record, "model_provider", None
2024 ), # Handle both old and new records
2025 "prompt_tokens": record.prompt_tokens,
2026 "completion_tokens": record.completion_tokens,
2027 "timestamp": record.timestamp,
2028 }
2029 )
2031 # Use synchronous calculation for research costs
2032 calculator = CostCalculator()
2033 costs = []
2034 for record in usage_data:
2035 cost_data = calculator.calculate_cost_sync(
2036 record["model_name"],
2037 record["prompt_tokens"],
2038 record["completion_tokens"],
2039 provider=record["provider"],
2040 )
2041 costs.append({**record, **cost_data})
2043 return {
2044 "status": "success",
2045 "research_id": research_id,
2046 **summarize_costs(costs),
2047 }
2049 except Exception:
2050 logger.exception(
2051 f"Error getting research costs for research: {_log_value_preview(research_id)}"
2052 )
2053 return JSONResponse(
2054 {"error": "An internal error occurred"}, status_code=500
2055 )
2058@router.get("/api/cost-analytics")
2059def api_cost_analytics(
2060 request: Request, username: Annotated[str, Depends(require_auth)]
2061):
2062 """Get cost analytics across all research sessions."""
2063 try:
2064 from ...metrics.pricing.cost_calculator import (
2065 CostCalculator,
2066 summarize_costs,
2067 )
2069 period = request.query_params.get("period", "30d")
2071 with get_user_db_session(username) as session:
2072 # Get token usage for the period
2073 query = session.query(TokenUsage)
2074 time_condition = get_time_filter_condition(
2075 period, TokenUsage.timestamp
2076 )
2077 if time_condition is not None:
2078 query = query.filter(time_condition)
2080 # First check if we have any records to avoid expensive queries
2081 record_count = query.count()
2083 if record_count == 0:
2084 return {
2085 "status": "success",
2086 "period": period,
2087 "overview": {
2088 **summarize_costs([]),
2089 "avg_cost_per_research": 0.0,
2090 "total_matching_calls": 0,
2091 "usage_truncated": False,
2092 },
2093 "top_expensive_research": [],
2094 "research_count": 0,
2095 "message": "No token usage data found for this period",
2096 }
2098 # If we have too many records, limit to recent ones to avoid timeout
2099 if record_count > 1000:
2100 logger.warning(
2101 f"Large dataset detected ({record_count} records), limiting to recent 1000 for performance"
2102 )
2103 usage_records = (
2104 query.order_by(TokenUsage.timestamp.desc())
2105 .limit(1000)
2106 .all()
2107 )
2108 else:
2109 usage_records = query.all()
2111 # Convert to dict format
2112 usage_data = []
2113 for record in usage_records:
2114 usage_data.append(
2115 {
2116 "model_name": record.model_name,
2117 "provider": getattr(
2118 record, "model_provider", None
2119 ), # Handle both old and new records
2120 "prompt_tokens": record.prompt_tokens,
2121 "completion_tokens": record.completion_tokens,
2122 "research_id": record.research_id,
2123 "timestamp": record.timestamp,
2124 }
2125 )
2127 # Use synchronous calculation
2128 calculator = CostCalculator()
2130 # Calculate overall costs
2131 costs = []
2132 for record in usage_data:
2133 cost_data = calculator.calculate_cost_sync(
2134 record["model_name"],
2135 record["prompt_tokens"],
2136 record["completion_tokens"],
2137 provider=record["provider"],
2138 )
2139 costs.append({**record, **cost_data})
2141 cost_summary = summarize_costs(costs)
2142 cost_summary["total_matching_calls"] = record_count
2143 cost_summary["usage_truncated"] = record_count > len(costs)
2145 # Group by research_id for per-research costs
2146 research_costs: dict[str, Any] = {}
2147 for record in costs:
2148 rid = record["research_id"]
2149 if rid not in research_costs:
2150 research_costs[rid] = []
2151 research_costs[rid].append(record)
2153 # Calculate cost per research
2154 research_summaries = {
2155 rid: summarize_costs(records)
2156 for rid, records in research_costs.items()
2157 }
2158 cost_summary["avg_cost_per_research"] = (
2159 round(cost_summary["total_cost"] / len(research_summaries), 6)
2160 if research_summaries
2161 else 0.0
2162 )
2164 # Top expensive research sessions
2165 top_expensive = sorted(
2166 research_summaries.items(),
2167 key=lambda x: x[1]["total_cost"],
2168 reverse=True,
2169 )[:10]
2171 return {
2172 "status": "success",
2173 "period": period,
2174 "overview": cost_summary,
2175 "top_expensive_research": [
2176 {
2177 "research_id": rid,
2178 "total_cost": data["total_cost"],
2179 "pricing_complete": data["pricing_complete"],
2180 "unpriced_calls": data["unpriced_calls"],
2181 "unpriced_tokens": data["unpriced_tokens"],
2182 }
2183 for rid, data in top_expensive
2184 ],
2185 "research_count": len(research_summaries),
2186 }
2188 except Exception:
2189 logger.exception("Error getting cost analytics")
2190 # Return a real error so the UI can distinguish "no data" from
2191 # "this query failed". Returning 200 with status=success masked
2192 # genuine failures (DB unavailable, schema mismatch).
2193 return JSONResponse(
2194 {
2195 "status": "error",
2196 "period": period,
2197 "error": "Cost analytics temporarily unavailable",
2198 },
2199 status_code=500,
2200 )
2203@router.get("/links")
2204def link_analytics(
2205 request: Request, username: Annotated[str, Depends(require_auth)]
2206):
2207 """Display link analytics page."""
2208 return templates.TemplateResponse(
2209 request=request,
2210 name="pages/link_analytics.html",
2211 context={"request": request},
2212 )
2215@router.get("/api/link-analytics")
2216def api_link_analytics(
2217 request: Request, username: Annotated[str, Depends(require_auth)]
2218):
2219 """Get link analytics data."""
2220 try:
2221 period = request.query_params.get("period", "30d")
2223 # Get link analytics data
2224 link_data = get_link_analytics(period, username)
2226 return {
2227 "status": "success",
2228 "data": link_data["link_analytics"],
2229 "period": period,
2230 }
2232 except Exception:
2233 logger.exception("Error getting link analytics")
2234 return JSONResponse(
2235 {
2236 "status": "error",
2237 "message": "An internal error occurred. Please try again later.",
2238 },
2239 status_code=500,
2240 )
2243@router.get("/api/domain-classifications")
2244def api_get_domain_classifications(
2245 request: Request, username: Annotated[str, Depends(require_auth)]
2246):
2247 """Get all domain classifications."""
2248 classifier = None
2249 try:
2250 classifier = DomainClassifier(username)
2251 classifications = classifier.get_all_classifications()
2253 return {
2254 "status": "success",
2255 "classifications": [c.to_dict() for c in classifications],
2256 "total": len(classifications),
2257 }
2259 except Exception:
2260 logger.exception("Error getting domain classifications")
2261 return JSONResponse(
2262 {
2263 "status": "error",
2264 "message": "Failed to retrieve classifications",
2265 },
2266 status_code=500,
2267 )
2268 finally:
2269 if classifier is not None:
2270 from ...utilities.resource_utils import safe_close
2272 safe_close(classifier, "domain classifier")
2275@router.get("/api/domain-classifications/summary")
2276def api_get_classifications_summary(
2277 request: Request, username: Annotated[str, Depends(require_auth)]
2278):
2279 """Get summary of domain classifications by category."""
2280 classifier = None
2281 try:
2282 classifier = DomainClassifier(username)
2283 summary = classifier.get_categories_summary()
2285 return {"status": "success", "summary": summary}
2287 except Exception:
2288 logger.exception("Error getting classifications summary")
2289 return JSONResponse(
2290 {"status": "error", "message": "Failed to retrieve summary"},
2291 status_code=500,
2292 )
2293 finally:
2294 if classifier is not None:
2295 from ...utilities.resource_utils import safe_close
2297 safe_close(classifier, "domain classifier")
2300#: Longest ``domain`` accepted by the classify route. Matches the
2301#: ``String(255)`` declared on ``DomainClassification.domain`` — SQLite
2302#: does not enforce column lengths, so without this bound a request-
2303#: supplied string of any length is interpolated verbatim into the
2304#: classification prompt, persisted as the classification key, and
2305#: reflected whole in this route's 400 body. Legitimate values (the
2306#: authorities the UI extracts from the user's own resource URLs, and
2307#: the keys the batch path derives) are hostnames with an optional port,
2308#: well under this bound.
2309_MAX_DOMAIN_LENGTH = 255
2312def _domain_rejection_reason(domain: str) -> str | None:
2313 """Fail-closed length/shape bound on a request-supplied ``domain``.
2315 Returns a rejection reason for the 400 body, or ``None`` when the
2316 value may flow on. Control characters never appear in a hostname;
2317 rejecting them keeps newline/NUL smuggling shapes out of the
2318 classification prompt, the persisted key and the failure-path echo.
2319 """
2320 if len(domain) > _MAX_DOMAIN_LENGTH:
2321 return f"domain must be at most {_MAX_DOMAIN_LENGTH} characters"
2322 if any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in domain):
2323 return "domain must not contain control characters"
2324 return None
2327@router.post("/api/domain-classifications/classify")
2328async def api_classify_domains(
2329 request: Request, username: Annotated[str, Depends(require_auth)]
2330):
2331 """Trigger classification of a specific domain or batch classification."""
2332 data, body_error = await read_json_dict(
2333 request, "status", "Request body must be valid JSON"
2334 )
2335 if body_error is not None:
2336 return body_error
2337 domain = data.get("domain")
2338 force_update = data.get("force_update", False)
2339 batch_mode = data.get("batch", False)
2340 if domain is not None and not isinstance(domain, str):
2341 return json_body_error("status", "domain must be a string")
2342 if not isinstance(force_update, bool):
2343 return json_body_error("status", "force_update must be a boolean")
2344 if not isinstance(batch_mode, bool):
2345 return json_body_error("status", "batch must be a boolean")
2346 if isinstance(domain, str):
2347 reason = _domain_rejection_reason(domain)
2348 if reason is not None:
2349 return json_body_error("status", reason)
2351 def _classify_sync():
2352 """All the sync LLM + DB work — run in a threadpool so we don't
2353 block the asyncio event loop for seconds per request."""
2354 from ...settings.manager import SettingsManager
2355 from ...database.session_context import get_user_db_session
2356 from ...utilities.resource_utils import safe_close
2358 classifier = None
2359 try:
2360 with get_user_db_session(username) as db_session:
2361 settings_manager = SettingsManager(db_session=db_session)
2362 settings_snapshot = settings_manager.get_all_settings()
2364 classifier = DomainClassifier(
2365 username, settings_snapshot=settings_snapshot
2366 )
2368 if domain and not batch_mode:
2369 logger.info(
2370 "Classifying single domain: {}", _log_value_preview(domain)
2371 )
2372 classification = classifier.classify_domain(
2373 domain, force_update
2374 )
2375 if classification:
2376 return {
2377 "status": "success",
2378 "classification": classification.to_dict(),
2379 }
2380 return JSONResponse(
2381 {
2382 "status": "error",
2383 "message": f"Failed to classify domain: {domain}",
2384 },
2385 status_code=400,
2386 )
2387 if batch_mode:
2388 logger.info("Starting batch classification of all domains")
2389 results = classifier.classify_all_domains(force_update)
2390 return {"status": "success", "results": results}
2391 return JSONResponse(
2392 {
2393 "status": "error",
2394 "message": "Must provide either 'domain' or set 'batch': true",
2395 },
2396 status_code=400,
2397 )
2398 except Exception:
2399 logger.exception("Error classifying domains")
2400 return JSONResponse(
2401 {"status": "error", "message": "Failed to classify domains"},
2402 status_code=500,
2403 )
2404 finally:
2405 if classifier is not None:
2406 safe_close(classifier, "domain classifier")
2408 return await run_db_sync(_classify_sync)
2411@router.get("/api/domain-classifications/progress")
2412def api_classification_progress(
2413 request: Request, username: Annotated[str, Depends(require_auth)]
2414):
2415 """Get progress of domain classification task."""
2416 try:
2417 # Get counts of classified vs unclassified domains
2418 with get_user_db_session(username) as session:
2419 # Count total unique domains
2420 resources = session.query(ResearchResource.url).distinct().all()
2421 domains = set()
2423 for (url,) in resources:
2424 if url:
2425 domain = _extract_domain(url)
2426 if domain: 2426 ↛ 2423line 2426 didn't jump to line 2423 because the condition on line 2426 was always true
2427 domains.add(domain)
2429 all_domains = sorted(domains)
2430 total_domains = len(domains)
2432 # Count classified domains
2433 classified_count = session.query(DomainClassification).count()
2435 return {
2436 "status": "success",
2437 "progress": {
2438 "total_domains": total_domains,
2439 "classified": classified_count,
2440 "unclassified": total_domains - classified_count,
2441 "percentage": round(
2442 (classified_count / total_domains * 100)
2443 if total_domains > 0
2444 else 0,
2445 1,
2446 ),
2447 "all_domains": all_domains, # Return all domains for classification
2448 },
2449 }
2451 except Exception:
2452 logger.exception("Error getting classification progress")
2453 return JSONResponse(
2454 {"status": "error", "message": "Failed to retrieve progress"},
2455 status_code=500,
2456 )
2459# ---------------------------------------------------------------------------
2460# Journal Quality Dashboard
2461# ---------------------------------------------------------------------------
2464# Rate-limit constants for the journal-quality endpoints. The old Flask
2465# decorators (`journal_data_limit`, `journals_read_limit`) lived in
2466# `security/rate_limiter.py`, which the FastAPI migration removed. We
2467# re-use slowapi's `limiter.limit()` with the same intent:
2468# - journal-data download is expensive (~several hundred MB fetch +
2469# reference DB rebuild), so cap tightly
2470# - journal read endpoints are cheap but still worth a per-user rate
2471# to blunt enumeration
2472_JOURNAL_DATA_LIMIT = "2 per hour"
2473_JOURNALS_READ_LIMIT = "60 per minute"
2475# SHARED buckets, matching main's journal_data_limit / journals_read_limit
2476# (security/rate_limiter.py). Using per-endpoint `limiter.limit()` gave each
2477# of the three read endpoints its OWN 60/min bucket (180/min combined), which
2478# defeats the anti-enumeration throttle on the ~217K-row reference DB. A
2479# single scope per group + per-user keying restores the intended combined cap.
2480_journal_data_limit = limiter.shared_limit(
2481 _JOURNAL_DATA_LIMIT, scope="journal_data", key_func=_user_key
2482)
2483_journals_read_limit = limiter.shared_limit(
2484 _JOURNALS_READ_LIMIT, scope="journals_read", key_func=_user_key
2485)
2488def _resolve_persisted_egress_scope(username: str) -> tuple[Any, Any]:
2489 """Resolve the user's persisted ``policy.egress_scope``.
2491 Returns ``(scope, scope_raw)``. ``scope`` is an ``EgressScope``, or
2492 ``None`` when the stored value cannot be parsed — every caller treats
2493 that as "fail closed". ``scope_raw`` is the stored value (with the
2494 retired ``both`` rewritten to ``adaptive``) and exists only for the
2495 audit log.
2497 Sync on purpose: it opens the user's SQLCipher settings session, so
2498 async callers hand it to ``run_db_sync`` instead of the event loop.
2499 """
2500 from ...security.egress.policy import (
2501 DEFAULT_EGRESS_SCOPE,
2502 EgressScope,
2503 PolicyDeniedError,
2504 parse_user_egress_scope,
2505 )
2506 from ...utilities.db_utils import get_settings_manager
2508 scope_raw = get_settings_manager(username=username).get_setting(
2509 "policy.egress_scope", DEFAULT_EGRESS_SCOPE
2510 )
2511 # Retired "both" reads as the adaptive default, mirroring
2512 # context_from_snapshot's read-time backstop for un-migrated rows.
2513 if str(scope_raw).strip().lower() == EgressScope.BOTH.value:
2514 scope_raw = EgressScope.ADAPTIVE.value
2515 try:
2516 # Stale "unprotected" rows fall back to adaptive when the
2517 # operator gate is off, matching every other runtime reader.
2518 scope = parse_user_egress_scope(
2519 scope_raw, disabled_unprotected="adaptive"
2520 )
2521 except PolicyDeniedError:
2522 scope = None # corrupt scope -> callers fail closed
2523 return scope, scope_raw
2526def _egress_scope_forbids_public_fetch(scope: Any) -> bool:
2527 """Whether ``scope`` rules out fetching the public journal datasets.
2529 ``None`` — an unparseable stored value — fails closed, like every
2530 other runtime reader of this setting.
2531 """
2532 from ...security.egress.policy import EgressScope
2534 return scope is None or scope in (
2535 EgressScope.PRIVATE_ONLY,
2536 EgressScope.STRICT,
2537 )
2540def _resolve_effective_egress_scope(username: str) -> Any:
2541 """Resolve the egress scope a research run for ``username`` would get.
2543 Unlike ``_resolve_persisted_egress_scope``, which returns the literal
2544 stored value, this resolves ADAPTIVE against the user's primary
2545 search engine through ``build_run_egress_context`` — the seam the
2546 research worker uses to arm a run's context, and the same
2547 ``context_from_snapshot`` resolution the reputation filter's
2548 ``_should_skip_journal_fetch_for_scope`` applies. Under the default
2549 ADAPTIVE scope a private primary (a library/collection engine or a
2550 per-user private retriever) resolves PRIVATE_ONLY, so a read route
2551 that only looked at the literal ``adaptive`` would start a public
2552 download those siblings refuse.
2554 Raises on any resolution failure (unreadable settings, an unparseable
2555 scope, no primary engine configured); the caller fails closed.
2557 Sync on purpose: it opens the user's SQLCipher settings session.
2558 """
2559 from ...security.egress.policy import build_run_egress_context
2560 from ...utilities.db_utils import get_settings_manager
2562 snapshot = get_settings_manager(username=username).get_settings_snapshot(
2563 strict=True
2564 )
2565 if not isinstance(snapshot, dict): 2565 ↛ 2566line 2565 didn't jump to line 2566 because the condition on line 2565 was never true
2566 raise TypeError(
2567 f"settings snapshot must be a dict, got {type(snapshot).__name__}"
2568 )
2569 primary = snapshot.get("search.tool")
2570 if isinstance(primary, dict): 2570 ↛ 2571line 2570 didn't jump to line 2571 because the condition on line 2570 was never true
2571 primary = primary.get("value")
2572 return build_run_egress_context(
2573 snapshot,
2574 primary if isinstance(primary, str) else None,
2575 username=username,
2576 ).scope
2579def _journal_ref_db_read_would_download(username: str) -> bool:
2580 """Whether reading the reference DB would fetch the public datasets.
2582 ``JournalQualityDB._ensure_engine`` lazily BUILDS the reference DB on
2583 TWO conditions, not one: when the file is missing, and when an
2584 existing file fails ``_validate_existing_db`` — which every install
2585 hits the first time it runs after a ``JOURNAL_QUALITY_SCHEMA_VERSION``
2586 bump (three have shipped) or after the file is corrupted. That build
2587 downloads several hundred MB from OpenAlex/DOAJ/GitHub *unless it has
2588 local snapshots to compile from*. The build's own egress gate reads
2589 the context armed on the calling thread, but these routes are sync
2590 handlers that Starlette runs on an anyio threadpool worker where
2591 nothing arms one — so that gate is a no-op here and a plain dashboard
2592 page load would start the download.
2594 The predicate that actually governs the download is therefore the
2595 SNAPSHOT, not the compiled DB, and this function answers it in that
2596 order:
2598 1. ``REQUIRED_SNAPSHOT_FILENAME`` present in the journal data
2599 directory → ``ensure_journal_data`` reports "available" after one
2600 stat and the build compiles from disk. Never a download, so never
2601 blocked, whatever the scope. (A forbidding scope must not turn an
2602 offline rebuild into a dark dashboard.)
2603 2. No snapshot, but the compiled DB exists AND passes the same
2604 validity probe ``_ensure_engine`` applies → no build runs at all,
2605 so not blocked. The probe is ``journal_db_file_is_valid``, the
2606 read-only half of ``_validate_existing_db``: it opens the file
2607 ``mode=ro``, reads ``PRAGMA user_version`` and ``sqlite_master``,
2608 and — unlike the method — does not unlink an unusable file, so
2609 asking the question cannot destroy the user's DB or start a build.
2610 3. Otherwise a read WOULD have to download. Resolve the scope a
2611 research run for this user would get — ``policy.egress_scope``
2612 with ADAPTIVE resolved against the user's primary search engine
2613 (``_resolve_effective_egress_scope``) — so the caller can answer
2614 "unavailable" rather than fetch. Any resolution failure (corrupt
2615 scope, no primary engine, unreadable settings) fails closed.
2617 Only case 3 opens the settings session, and only case 2 opens the
2618 SQLite file, so the healthy dashboard still costs one stat.
2619 """
2620 from ...config.paths import get_journal_data_directory
2621 from ...journal_quality.db import DB_FILENAME, journal_db_file_is_valid
2622 from ...journal_quality.downloader import REQUIRED_SNAPSHOT_FILENAME
2624 try:
2625 data_dir = get_journal_data_directory()
2626 if (data_dir / REQUIRED_SNAPSHOT_FILENAME).exists():
2627 return False
2628 db_path = data_dir / DB_FILENAME
2629 if db_path.exists() and journal_db_file_is_valid(db_path):
2630 return False
2631 except Exception:
2632 # Reads fail closed on an unreadable journal data directory. That
2633 # has not been logged by the time it gets here — `_ensure_dir` mkdirs with `label=None` and logs
2634 # nothing, and `get_data_directory` raises its `LDR_DATA_DIR`
2635 # ValueError unlogged — so log it here rather than swallow a
2636 # misconfigured deployment into a silent 503. Fail closed because
2637 # this is exactly the case where the fetch cannot be shown to be
2638 # allowed: the read degrades to "unavailable" instead of starting
2639 # the download.
2640 logger.exception(
2641 "journal reference DB egress pre-check failed; treating the "
2642 "read as blocked"
2643 )
2644 return True
2645 try:
2646 scope = _resolve_effective_egress_scope(username)
2647 except Exception as exc:
2648 # Fail closed, but WITHOUT a traceback: the frames below hold the
2649 # user's full settings snapshot (API keys included), which a
2650 # diagnose-mode traceback would render. Type name only.
2651 logger.bind(policy_audit=True).warning(
2652 "journal reference DB egress scope unresolvable "
2653 f"({type(exc).__name__}); treating the read as blocked"
2654 )
2655 return True
2656 return _egress_scope_forbids_public_fetch(scope)
2659@router.get("/journals")
2660def journal_quality(
2661 request: Request, username: Annotated[str, Depends(require_auth)]
2662):
2663 """Display journal quality dashboard."""
2664 return templates.TemplateResponse(
2665 request=request,
2666 name="pages/journal_quality.html",
2667 context={"request": request},
2668 )
2671@router.get("/api/journal-data/status")
2672def api_journal_data_status(
2673 request: Request, username: Annotated[str, Depends(require_auth)]
2674):
2675 """Get status of downloadable journal data files."""
2676 try:
2677 from ...journal_quality.downloader import (
2678 get_journal_data_status,
2679 )
2681 return get_journal_data_status()
2682 except Exception:
2683 logger.exception("Error checking journal data status")
2684 return JSONResponse(
2685 {"error": "Failed to check status"}, status_code=500
2686 )
2689@router.post("/api/journal-data/download")
2690@_journal_data_limit
2691async def api_journal_data_download(
2692 request: Request, username: Annotated[str, Depends(require_auth)]
2693):
2694 """Trigger download/update of journal data files.
2696 Rate-limited to 2 per hour per authenticated user: the download streams
2697 several hundred MB and rebuilds the on-disk reference DB, so unbounded
2698 invocation is a DoS vector. The actual download + rebuild runs in a
2699 worker thread so the uvicorn loop isn't blocked for the minutes this
2700 may take.
2701 """
2702 try:
2703 from ...journal_quality.downloader import (
2704 download_journal_data,
2705 get_download_state,
2706 )
2707 from ...journal_quality.data_sources import ALL_SOURCES
2709 # Egress policy: this endpoint streams several hundred MB over public
2710 # HTTP. Under an offline-for-public scope (PRIVATE_ONLY / STRICT) the
2711 # user has opted out of public egress, so refuse rather than reaching
2712 # out. A corrupt/unknown scope also fails closed. The background
2713 # auto-download path is already gated by
2714 # JournalReputationFilter._should_skip_journal_fetch_for_scope; this
2715 # closes the manual button as the matching entry point.
2716 # Settings read opens the user's SQLCipher session (sync) — keep it
2717 # off the event loop. A rekey sidecar or epoch fault on that read
2718 # raises a RekeyOperationalError rather than reading as "no
2719 # database", which would answer the default scope instead of the
2720 # stored one; it is passed on below for the app's retryable 503.
2721 scope, scope_raw = await run_db_sync(
2722 lambda: _resolve_persisted_egress_scope(username)
2723 )
2724 if _egress_scope_forbids_public_fetch(scope):
2725 logger.bind(policy_audit=True).warning(
2726 "journal data download refused by egress policy",
2727 scope=str(scope_raw),
2728 )
2729 return JSONResponse(
2730 {
2731 "success": False,
2732 "message": (
2733 "Journal data download needs public network "
2734 "access, which the current egress policy blocks "
2735 "(private/offline scope). Change the egress "
2736 "scope to download."
2737 ),
2738 },
2739 status_code=403,
2740 )
2742 # Parse the force flag from the request body if present.
2743 force = False
2744 ctype = request.headers.get("content-type", "")
2745 if "application/json" in ctype: 2745 ↛ 2780line 2745 didn't jump to line 2780 because the condition on line 2745 was always true
2746 try:
2747 body = await request.json()
2748 except Exception:
2749 return JSONResponse(
2750 {
2751 "success": False,
2752 "message": "Request body must be valid JSON",
2753 },
2754 status_code=400,
2755 )
2756 if not isinstance(body, dict):
2757 return JSONResponse(
2758 {
2759 "success": False,
2760 "message": "Request body must be valid JSON",
2761 },
2762 status_code=400,
2763 )
2764 force = body.get("force", False)
2765 if not isinstance(force, bool):
2766 return JSONResponse(
2767 {
2768 "success": False,
2769 "message": "force must be a boolean",
2770 },
2771 status_code=400,
2772 )
2774 # `download_journal_data` already calls `build_reference_db()`
2775 # + `reset_db()` internally on its success path (downloader.py
2776 # → db.py); we used to redo it here, but a second rebuild ran
2777 # the full ~30s pipeline twice and wrote to the legacy
2778 # `journal_reference.db` filename that the downloader had just
2779 # cleaned up. See main's #3574 for the simplification.
2780 def _run_download() -> tuple[bool, str | None]:
2781 return download_journal_data(force=force)
2783 success, internal_message = await asyncio.to_thread(_run_download)
2784 if not success:
2785 logger.warning(f"Journal data download failed: {internal_message}")
2786 return {"success": False, "message": "Download failed"}
2788 # Build the user-facing message locally from structured state
2789 # (ints + developer-authored source labels). We deliberately do
2790 # NOT echo `internal_message` from download_journal_data: keeping
2791 # the response safe-by-construction means a future refactor that
2792 # lets arbitrary strings (exception info, user input, PII) slip
2793 # into the downloader's message cannot reach the client.
2794 counts = get_download_state().get("counts")
2795 if counts is not None:
2796 parts = [
2797 f"{int(counts.get(src.key) or 0)} {src.count_label}"
2798 for src in ALL_SOURCES
2799 ]
2800 user_message = (
2801 f"Fetched {' + '.join(parts)}. Database rebuilt successfully."
2802 )
2803 else:
2804 # `counts` is None when download_journal_data took its
2805 # early-return "already up to date" branch (no fetch ran).
2806 user_message = "Journal data is already up to date."
2807 return {"success": True, "message": user_message}
2808 except RekeyOperationalError:
2809 # The scope could not be read (a rekey sidecar or epoch fault, such
2810 # as an fd-exhaustion spike after require_auth's probe passed), so
2811 # nothing was downloaded. Not a server error: the app's handler
2812 # answers the same retryable 503 + Retry-After as require_auth,
2813 # and logs only the fault's type name.
2814 raise
2815 except Exception:
2816 logger.exception("Error downloading journal data")
2817 return JSONResponse(
2818 {"success": False, "message": "Download failed"},
2819 status_code=500,
2820 )
2823#: Allowlist of ``score_source`` values accepted by ``/api/journals``.
2824#: Matches the writer side: ``openalex`` / ``doaj`` for reference-DB
2825#: hits, ``llm`` for Tier 4 cache rows. Empty string means "no filter"
2826#: and is handled by the caller before validation.
2827_ALLOWED_SCORE_SOURCES = frozenset({"openalex", "doaj", "llm"})
2829#: Upper bound on the echoed ``page`` parameter. Prevents a crafted
2830#: ``?page=10**9`` from issuing an OFFSET scan before the post-query
2831#: clamp can take effect — reject at input validation instead.
2832_MAX_PAGE = 10_000
2835@router.get("/api/journals")
2836@_journals_read_limit
2837def api_journal_quality(
2838 request: Request, username: Annotated[str, Depends(require_auth)]
2839):
2840 """Get journal quality data with server-side pagination and filtering.
2842 Reads from the bundled read-only reference database (~217K journals)
2843 rather than the per-user DB, so the dashboard is always populated.
2845 Query params:
2846 page (int): 1-indexed page number (default 1, max 10000)
2847 per_page (int): rows per page, max 200 (default 50)
2848 search (str): name substring filter
2849 tier (str): elite/strong/moderate/low/predatory
2850 score_source (str): openalex/doaj/llm (allowlisted)
2851 sort (str): column to sort by (default quality)
2852 order (str): asc or desc (default desc)
2853 """
2854 try:
2855 from ...journal_quality.db import get_journal_reference_db
2857 # Egress policy: ``ref.available`` lazily BUILDS the reference DB
2858 # when the file is missing OR fails validation, and that build
2859 # downloads several hundred MB of public data unless the local
2860 # snapshots are already on disk. Resolve the user's effective
2861 # scope (ADAPTIVE resolved against the primary engine) FIRST, before anything touches the engine, and answer
2862 # with the same 503 this route already returns for an
2863 # unavailable reference DB.
2864 blocked = _journal_ref_db_read_would_download(username)
2865 ref = get_journal_reference_db()
2866 if blocked or not ref.available:
2867 return JSONResponse(
2868 {
2869 "status": "error",
2870 "message": "Journal reference database not available.",
2871 },
2872 status_code=503,
2873 )
2875 try:
2876 page = max(1, int(request.query_params.get("page", 1)))
2877 per_page = min(
2878 max(1, int(request.query_params.get("per_page", 50))), 200
2879 )
2880 except (TypeError, ValueError):
2881 return JSONResponse(
2882 {
2883 "status": "error",
2884 "message": "Invalid pagination parameters",
2885 },
2886 status_code=400,
2887 )
2888 if page > _MAX_PAGE:
2889 return JSONResponse(
2890 {
2891 "status": "error",
2892 "message": (
2893 f"page exceeds maximum ({_MAX_PAGE}); narrow the "
2894 "filter or increase per_page"
2895 ),
2896 },
2897 status_code=400,
2898 )
2899 search = request.query_params.get("search", "")
2900 tier = request.query_params.get("tier", "")
2901 score_source = request.query_params.get("score_source", "")
2902 if score_source and score_source not in _ALLOWED_SCORE_SOURCES:
2903 return JSONResponse(
2904 {
2905 "status": "error",
2906 "message": (
2907 f"Invalid score_source; must be one of "
2908 f"{sorted(_ALLOWED_SCORE_SOURCES)}"
2909 ),
2910 },
2911 status_code=400,
2912 )
2913 sort = request.query_params.get("sort", "quality")
2914 order = request.query_params.get("order", "desc")
2916 journals, total = ref.get_journals_page(
2917 page=page,
2918 per_page=per_page,
2919 search=search,
2920 tier=tier,
2921 score_source=score_source,
2922 sort=sort,
2923 order=order,
2924 )
2926 # Clamp the echoed page so the UI never displays out-of-range
2927 # numbers on crafted input (e.g. ?page=10**9). SQLite's OFFSET on
2928 # an indexed ORDER BY caps work at ~total rows regardless of the
2929 # requested offset, so no DB-level clamp is needed.
2930 total_pages = -(-total // per_page) if per_page > 0 and total > 0 else 1
2931 page = min(page, total_pages)
2933 result = {
2934 "status": "success",
2935 "journals": journals,
2936 "pagination": {
2937 "page": page,
2938 "per_page": per_page,
2939 "total_count": total,
2940 "total_pages": total_pages,
2941 },
2942 }
2944 # Include summary only when requested (avoids 3 extra SQL queries
2945 # on every pagination/sort/filter request)
2946 if request.query_params.get("include_summary", "false") == "true":
2947 summary = ref.get_summary()
2948 summary["quality_distribution"] = ref.get_quality_distribution()
2949 summary["source_distribution"] = ref.get_source_distribution()
2950 result["summary"] = summary
2952 return result
2954 except Exception:
2955 logger.exception("Error getting journal quality data")
2956 return JSONResponse(
2957 {
2958 "status": "error",
2959 "message": "An internal error occurred. Please try again later.",
2960 },
2961 status_code=500,
2962 )
2965def _ref_db_lookup(ref_db, name: str) -> dict:
2966 """Look up a journal's display bibliometrics in the reference DB.
2968 Returns a dict with keys the dashboard template already renders
2969 (h_index, impact_factor, sjr_quartile, publisher, is_predatory,
2970 predatory_source, is_in_doaj). Missing fields default
2971 to None / False so the frontend never sees KeyError. On any ref-DB
2972 error the function returns an empty dict — the dashboard still shows
2973 the name + user-DB quality, just without the extras.
2974 """
2975 if ref_db is None or not name:
2976 return {}
2977 try:
2978 entry = ref_db.lookup_source(name=name) or {}
2979 except Exception: # allow: silent-exception
2980 # Reference DB lookups are best-effort enrichment. Any failure
2981 # degrades to "no bibliometric extras" without crashing the
2982 # dashboard; detailed errors already surface via the DB layer's
2983 # own logger.exception calls when they matter.
2984 return {}
2985 # lookup_source returns a compact dict; the sjr_quartile lives under
2986 # "quartile" and predatory/DOAJ fields may be absent entirely.
2987 return {
2988 "h_index": entry.get("h_index"),
2989 "impact_factor": entry.get("impact_factor"),
2990 "sjr_quartile": entry.get("quartile"),
2991 "is_predatory": bool(entry.get("is_predatory")),
2992 "predatory_source": entry.get("predatory_source"),
2993 "is_in_doaj": bool(entry.get("is_in_doaj")),
2994 "publisher": entry.get("publisher"),
2995 }
2998def _get_ref_db_or_none():
2999 """Return the JournalQualityDB singleton, or None if unavailable.
3001 The reference DB is optional — if the user hasn't downloaded the
3002 snapshot, the dashboard still renders with user-DB data only.
3003 """
3004 try:
3005 from ...journal_quality.db import get_journal_reference_db
3007 return get_journal_reference_db()
3008 except Exception: # allow: silent-exception
3009 # Reference DB is optional; if import or initialization fails
3010 # (unusual: usually it's lazily built on first access), the
3011 # dashboard falls back to user-DB-only rendering.
3012 return None
3015def _resolve_paper_quality(
3016 llm_quality: int | None, enrichment: dict
3017) -> tuple[int | None, str | None]:
3018 """Pick a quality score for a dashboard row.
3020 Precedence: current LLM verdict from the user's ``journals`` table
3021 (Tier 4 cache, keyed by NFKC-normalized container_title) → live
3022 derivation from the bundled reference DB row (Tier 1-3). Always
3023 live — no frozen per-Paper copy exists, so a re-scored journal
3024 propagates automatically. Returns (score, source_label) or
3025 (None, None) if neither path had data.
3026 """
3027 if llm_quality is not None:
3028 return llm_quality, "llm"
3029 if not enrichment:
3030 return None, None
3031 # enrichment comes from _source_to_dashboard_dict — row.quality is
3032 # the ref-DB's pre-computed score (same formula as the filter uses),
3033 # so we trust it directly rather than re-running derive_quality_score.
3034 q = enrichment.get("quality")
3035 if q is not None:
3036 return int(q), enrichment.get("score_source") or "openalex"
3037 return None, None
3040def _lookup_journal_llm_quality(
3041 db, container_titles: list[str]
3042) -> dict[str, int]:
3043 """Batch-look up current Tier 4 LLM verdicts from the user's
3044 ``journals`` table.
3046 Returns a dict mapping ``normalize_name(container_title)`` →
3047 ``Journal.quality``. Missing journals (never Tier-4-scored) simply
3048 don't appear in the result — callers fall through to the bundled
3049 reference DB. One indexed ``name_lower IN (...)`` query.
3050 """
3051 from ...journal_quality.scoring import normalize_name
3053 if not container_titles:
3054 return {}
3055 normalized = list({normalize_name(ct) for ct in container_titles if ct})
3056 if not normalized:
3057 return {}
3058 rows = (
3059 db.query(Journal.name_lower, Journal.quality)
3060 .filter(Journal.name_lower.in_(normalized))
3061 .filter(Journal.quality.isnot(None))
3062 .all()
3063 )
3064 return {name_lower: int(q) for name_lower, q in rows}
3067@router.get("/api/journals/user-research")
3068@_journals_read_limit
3069def api_user_research_journals(
3070 request: Request, username: Annotated[str, Depends(require_auth)]
3071):
3072 """Get journals from the user's own research sessions.
3074 Paper-rooted query: groups by ``Paper.container_title`` (the
3075 cleaned name the filter used to score the journal), counts paper
3076 appearances. Quality is resolved live — Tier 4 via a batch lookup
3077 against the user's ``journals`` table (keyed by NFKC-normalized
3078 container_title), Tier 1-3 via the bundled read-only reference DB.
3079 A re-scored journal propagates to existing research rows
3080 automatically because no per-Paper score is stored.
3081 """
3082 # `reference_data_available` distinguishes "no predatory journals
3083 # found" from "predatory journals were never checked for". The
3084 # reference DB can be withheld from this route entirely — an egress
3085 # scope that forbids the public fetch, or a build/open failure — and
3086 # `predatory_blocked: 0` alongside `status: success` would otherwise
3087 # read as a clean bill of health. The route's own comment on the
3088 # `predatory_blocked` query calls silently undercounting predatory
3089 # journals a no-fallbacks violation; this is the same undercount at
3090 # the response level, so it is flagged rather than hidden. It is
3091 # False on this early-return shape because every branch that reaches
3092 # it (no papers table, no rows) returns before the reference DB is
3093 # consulted: there is nothing to have checked.
3094 _empty_response = {
3095 "status": "success",
3096 "reference_data_available": False,
3097 "summary": {
3098 "total_journals": 0,
3099 "avg_quality": None,
3100 "total_papers": 0,
3101 "predatory_blocked": 0,
3102 },
3103 "quality_distribution": {},
3104 "journals": [],
3105 }
3107 try:
3108 from sqlalchemy import inspect as sa_inspect
3110 with get_user_db_session(username) as db:
3111 inspector = sa_inspect(db.bind)
3112 if not inspector.has_table("papers"):
3113 return _empty_response
3115 # Top-200 most-cited journals in this user's research.
3116 # Orphan Papers (whose ``PaperAppearance`` rows were
3117 # cascade-deleted when their research session was deleted)
3118 # are excluded so the dashboard reflects what the user
3119 # currently has, not residual rows from deleted sessions.
3120 # See issue #3544.
3121 rows = (
3122 db.query(
3123 Paper.container_title,
3124 func.count(Paper.id).label("paper_count"),
3125 func.min(Paper.year).label("year_min"),
3126 func.max(Paper.year).label("year_max"),
3127 )
3128 .filter(Paper.container_title.isnot(None))
3129 .filter(Paper.appearances.any())
3130 .group_by(Paper.container_title)
3131 .order_by(func.count(Paper.id).desc())
3132 .limit(200)
3133 .all()
3134 )
3136 if not rows:
3137 return _empty_response
3139 # One batched ref-DB lookup for the whole top-200 slice —
3140 # hits `sources.name_lower IN (…)` rather than 200 point
3141 # queries.
3142 # Egress policy: every reference-DB read below lazily BUILDS
3143 # the DB when the file is missing or fails validation, and
3144 # with no local snapshots that build downloads several
3145 # hundred MB of public data on a threadpool worker where the
3146 # build's own scope gate is blind (see
3147 # _journal_ref_db_read_would_download). Under a forbidding
3148 # effective scope, degrade to the enrichment-free response
3149 # this route already returns when the reference DB is
3150 # unavailable.
3151 ref_db = (
3152 None
3153 if _journal_ref_db_read_would_download(username)
3154 else _get_ref_db_or_none()
3155 )
3156 # `_get_ref_db_or_none` hands back the singleton without
3157 # opening anything, and the lookups below swallow a failed
3158 # lazy build/open (FileNotFoundError) into an empty result —
3159 # so a non-None handle alone would report "checked, none
3160 # found" for a DB that was never read. `available` runs the
3161 # same lazy open the lookups run (the egress gate above has
3162 # already ruled out a forbidden download) and is False on
3163 # exactly the failure they swallow; drop the handle then.
3164 if ref_db is not None and not ref_db.available:
3165 ref_db = None
3166 enrich_map = {}
3167 if ref_db is not None:
3168 enrich_map = ref_db.lookup_sources_batch(
3169 [r.container_title for r in rows]
3170 )
3172 from ...journal_quality.scoring import normalize_name
3174 # Batch-look up current LLM verdicts (Tier 4) from the
3175 # user's journals table, keyed by NFKC-normalized name.
3176 # Always live — no frozen Paper copy — so a re-scored
3177 # journal propagates here without any backfill.
3178 llm_by_name = _lookup_journal_llm_quality(
3179 db, [r.container_title for r in rows]
3180 )
3182 journals: list[dict] = []
3183 qualities: list[int] = []
3184 for r in rows:
3185 normalized = normalize_name(r.container_title)
3186 enrichment = enrich_map.get(normalized, {})
3187 quality, source_label = _resolve_paper_quality(
3188 llm_by_name.get(normalized), enrichment
3189 )
3190 if quality is not None:
3191 qualities.append(quality)
3192 journals.append(
3193 {
3194 "name": r.container_title,
3195 "quality": quality,
3196 "score_source": source_label,
3197 "paper_count": r.paper_count,
3198 "year_min": r.year_min,
3199 "year_max": r.year_max,
3200 **{
3201 k: v
3202 for k, v in enrichment.items()
3203 if k not in ("quality", "score_source", "name")
3204 },
3205 }
3206 )
3208 # Aggregate stats computed across the top-200 slice for the
3209 # dashboard summary — matches how the table renders.
3210 total_journals = len(journals)
3211 total_papers = sum(r.paper_count for r in rows)
3212 avg_quality = (
3213 round(sum(qualities) / len(qualities), 1) if qualities else None
3214 )
3215 quality_distribution: dict[str, int] = {}
3216 for q in qualities:
3217 quality_distribution[str(q)] = (
3218 quality_distribution.get(str(q), 0) + 1
3219 )
3221 # Predatory count uses the full set of distinct
3222 # container_titles across the user's research, not just the
3223 # top-200 display slice. One batched query.
3224 #
3225 # KNOWN-DEFERRED: unbounded SELECT DISTINCT. Acceptable today
3226 # because typical users have <5K distinct titles even after
3227 # years of use, and count_predatory_by_names documents
3228 # support up to ~100K params. Adding .limit(N) was considered
3229 # and rejected — it would SILENTLY UNDERCOUNT predatory
3230 # journals, which violates the no-fallbacks rule. Proper fix
3231 # (cross-DB correlated subquery or TTL cache) is tracked as
3232 # a post-merge follow-up. Threshold for visible impact:
3233 # ~50K papers.
3234 predatory_blocked = 0
3235 if ref_db is not None:
3236 # Same orphan-exclusion as the top-200 query above —
3237 # otherwise predatory_blocked stays inflated by titles
3238 # whose only Papers belong to deleted research sessions.
3239 all_names = [
3240 name
3241 for (name,) in db.query(Paper.container_title)
3242 .filter(Paper.container_title.isnot(None))
3243 .filter(Paper.appearances.any())
3244 .distinct()
3245 .all()
3246 ]
3247 predatory_blocked = ref_db.count_predatory_by_names(all_names)
3249 return {
3250 "status": "success",
3251 # False whenever `ref_db` was None — the egress gate withheld
3252 # it, `_get_ref_db_or_none` failed, or the reference DB could
3253 # not be opened/built (`available` False). Then
3254 # `predatory_blocked` is 0 because nothing was checked, not
3255 # because nothing was found, and every bibliometric field on
3256 # every journal row is absent for the same reason.
3257 "reference_data_available": ref_db is not None,
3258 "summary": {
3259 "total_journals": total_journals,
3260 "avg_quality": avg_quality,
3261 "total_papers": total_papers,
3262 "predatory_blocked": predatory_blocked,
3263 },
3264 "quality_distribution": quality_distribution,
3265 "journals": journals,
3266 }
3267 except Exception:
3268 logger.exception("Error getting user research journals")
3269 return JSONResponse(
3270 {
3271 "status": "error",
3272 "message": "Failed to load your research data.",
3273 },
3274 status_code=500,
3275 )
3278@router.get("/api/journals/research/{research_id}")
3279@_journals_read_limit
3280def api_research_journals(
3281 request: Request,
3282 research_id: str,
3283 username: Annotated[str, Depends(require_auth)],
3284):
3285 """Get journals encountered in a single research session.
3287 Filters the per-user papers table by joining through
3288 Paper → PaperAppearance → ResearchResource and matching ``research_id``.
3289 Quality is resolved live (journals.quality + bundled reference DB)
3290 so results always reflect the current verdict, not a stale snapshot.
3291 Mirrors the response shape of /api/journals/user-research so the
3292 dashboard can reuse its rendering code.
3293 """
3294 # `reference_data_available` distinguishes "no predatory journals
3295 # found" from "predatory journals were never checked for". The
3296 # reference DB can be withheld from this route entirely — an egress
3297 # scope that forbids the public fetch, or a build/open failure — and
3298 # `predatory_blocked: 0` alongside `status: success` would otherwise
3299 # read as a clean bill of health. The route's own comment on the
3300 # `predatory_blocked` query calls silently undercounting predatory
3301 # journals a no-fallbacks violation; this is the same undercount at
3302 # the response level, so it is flagged rather than hidden. It is
3303 # False on this early-return shape because every branch that reaches
3304 # it (no papers table, no rows) returns before the reference DB is
3305 # consulted: there is nothing to have checked.
3306 _empty_response = {
3307 "status": "success",
3308 "reference_data_available": False,
3309 "summary": {
3310 "total_journals": 0,
3311 "avg_quality": None,
3312 "total_papers": 0,
3313 "predatory_blocked": 0,
3314 },
3315 "quality_distribution": {},
3316 "journals": [],
3317 }
3319 try:
3320 from sqlalchemy import inspect as sa_inspect
3322 with get_user_db_session(username) as db:
3323 inspector = sa_inspect(db.bind)
3324 if not inspector.has_table("papers") or not inspector.has_table(
3325 "paper_appearances"
3326 ):
3327 return _empty_response
3329 # Verify the research_id belongs to this user before exposing
3330 # any data — research_history is in the same per-user DB so
3331 # the existence check doubles as an ownership check.
3332 from ...database.models.research import ResearchHistory
3334 research = (
3335 db.query(ResearchHistory.id)
3336 .filter(ResearchHistory.id == research_id)
3337 .first()
3338 )
3339 if research is None:
3340 return JSONResponse(
3341 {"status": "error", "message": "Research not found"},
3342 status_code=404,
3343 )
3345 # Aggregate container_title → paper_count for this research.
3346 # Join chain: Paper → PaperAppearance → ResearchResource.
3347 rows = (
3348 db.query(
3349 Paper.container_title,
3350 func.count(Paper.id).label("paper_count"),
3351 func.min(Paper.year).label("year_min"),
3352 func.max(Paper.year).label("year_max"),
3353 )
3354 .join(
3355 PaperAppearance,
3356 PaperAppearance.paper_id == Paper.id,
3357 )
3358 .join(
3359 ResearchResource,
3360 ResearchResource.id == PaperAppearance.resource_id,
3361 )
3362 .filter(
3363 ResearchResource.research_id == research_id,
3364 Paper.container_title.isnot(None),
3365 )
3366 .group_by(Paper.container_title)
3367 .order_by(func.count(Paper.id).desc())
3368 .all()
3369 )
3371 if not rows:
3372 return _empty_response
3374 # Egress policy: every reference-DB read below lazily BUILDS
3375 # the DB when the file is missing or fails validation, and
3376 # with no local snapshots that build downloads several
3377 # hundred MB of public data on a threadpool worker where the
3378 # build's own scope gate is blind (see
3379 # _journal_ref_db_read_would_download). Under a forbidding
3380 # effective scope, degrade to the enrichment-free response
3381 # this route already returns when the reference DB is
3382 # unavailable.
3383 ref_db = (
3384 None
3385 if _journal_ref_db_read_would_download(username)
3386 else _get_ref_db_or_none()
3387 )
3388 # `_get_ref_db_or_none` hands back the singleton without
3389 # opening anything, and the lookups below swallow a failed
3390 # lazy build/open (FileNotFoundError) into an empty result —
3391 # so a non-None handle alone would report "checked, none
3392 # found" for a DB that was never read. `available` runs the
3393 # same lazy open the lookups run (the egress gate above has
3394 # already ruled out a forbidden download) and is False on
3395 # exactly the failure they swallow; drop the handle then.
3396 if ref_db is not None and not ref_db.available:
3397 ref_db = None
3398 enrich_map = {}
3399 if ref_db is not None:
3400 enrich_map = ref_db.lookup_sources_batch(
3401 [r.container_title for r in rows]
3402 )
3404 from ...journal_quality.scoring import normalize_name
3406 # Batch-look up current LLM verdicts (Tier 4) — see
3407 # _lookup_journal_llm_quality for rationale. Same live
3408 # resolution as the cross-research rollup above.
3409 llm_by_name = _lookup_journal_llm_quality(
3410 db, [r.container_title for r in rows]
3411 )
3413 journals: list[dict] = []
3414 qualities: list[int] = []
3415 predatory_blocked = 0
3416 for r in rows:
3417 normalized = normalize_name(r.container_title)
3418 enrichment = enrich_map.get(normalized, {})
3419 if enrichment.get("is_predatory"):
3420 predatory_blocked += 1
3421 quality, source_label = _resolve_paper_quality(
3422 llm_by_name.get(normalized), enrichment
3423 )
3424 if quality is not None:
3425 qualities.append(quality)
3426 journals.append(
3427 {
3428 "name": r.container_title,
3429 "quality": quality,
3430 "score_source": source_label,
3431 "paper_count": r.paper_count,
3432 "year_min": r.year_min,
3433 "year_max": r.year_max,
3434 **{
3435 k: v
3436 for k, v in enrichment.items()
3437 if k not in ("quality", "score_source", "name")
3438 },
3439 }
3440 )
3442 total_papers = sum(r.paper_count for r in rows)
3443 avg_quality = (
3444 round(sum(qualities) / len(qualities), 1) if qualities else None
3445 )
3446 quality_distribution: dict[str, int] = {}
3447 for q in qualities:
3448 quality_distribution[str(q)] = (
3449 quality_distribution.get(str(q), 0) + 1
3450 )
3452 return {
3453 "status": "success",
3454 # Same marker as /api/journals/user-research, whose response
3455 # shape this route deliberately mirrors: False means
3456 # `predatory_blocked` counts nothing because the reference DB
3457 # was never consulted, not because this research session is
3458 # clean.
3459 "reference_data_available": ref_db is not None,
3460 "summary": {
3461 "total_journals": len(journals),
3462 "avg_quality": avg_quality,
3463 "total_papers": total_papers,
3464 "predatory_blocked": predatory_blocked,
3465 },
3466 "quality_distribution": quality_distribution,
3467 "journals": journals,
3468 }
3469 except Exception:
3470 logger.exception("Error getting per-research journals")
3471 return JSONResponse(
3472 {
3473 "status": "error",
3474 "message": "Failed to load research journals.",
3475 },
3476 status_code=500,
3477 )