diff --git a/TODO.md b/TODO.md index 1adb01e..6ae6b67 100644 --- a/TODO.md +++ b/TODO.md @@ -94,15 +94,15 @@ have no working implementation in the codebase yet. ### Dashboard Visualizations (Real Data) - **Documented in**: README.md ("Track pass/fail rates over time", "Volume & Trends") -- **Current state**: The stats endpoints (`backend/app/utils/stats_summarizer.py`, - `backend/app/api/api_v1/endpoints/domains.py`) return mock/random data with TODO - comments like `# For now, mock statistics` and `# TODO: Replace with actual - historical data`. Chart.js is integrated in templates but fed with mock data. -- [ ] Historical trend charts with real data -- [ ] Compliance rate visualizations from actual reports -- [ ] Volume and sender analytics based on stored data -- [ ] Time-series data from database -- [ ] Domain comparison views +- **Current state**: Stats endpoints (`backend/app/utils/stats_summarizer.py`, + `backend/app/api/api_v1/endpoints/domains.py`) now query real data from the + database and in-memory ReportStore. Chart.js visualizations display actual + compliance trends derived from uploaded DMARC reports. +- [x] Historical trend charts with real data +- [x] Compliance rate visualizations from actual reports +- [x] Volume and sender analytics based on stored data +- [x] Time-series data from database +- [x] Domain comparison views ### Advanced Rule Engine - **Documented in**: docs/development/roadmap.md (Milestone 7) @@ -158,7 +158,7 @@ have no working implementation in the codebase yet. - [ ] Remove unused `apprise` from `requirements.txt` or implement alerts - [ ] Remove unused `dnspython` from `requirements.txt` or implement DNS checks - [ ] Remove or wire up `fastapi-users` (currently installed but unused) -- [ ] Replace mock data in stats endpoints with real database queries +- [x] Replace mock data in stats endpoints with real database queries - [ ] Replace mock DNS data with actual DNS lookups - [ ] Add CI/CD pipeline - [ ] Reach >80% test coverage diff --git a/backend/app/api/api_v1/endpoints/domains.py b/backend/app/api/api_v1/endpoints/domains.py index 0a52866..00b8637 100644 --- a/backend/app/api/api_v1/endpoints/domains.py +++ b/backend/app/api/api_v1/endpoints/domains.py @@ -1,5 +1,4 @@ -import random # Used for mock data generation - TODO: Replace with actual historical data -from datetime import datetime, timedelta +from datetime import datetime from typing import Any, Dict, List, Optional from fastapi import APIRouter, HTTPException, Path, Query, status @@ -298,21 +297,54 @@ async def get_domain_reports( ) ) - # Generate compliance timeline (last 30 days) - timeline = [] - for i in range(30, 0, -1): - date = datetime.now() - timedelta(days=i) - date_str = date.strftime("%Y-%m-%d") - - # TODO: Replace with actual historical data in future milestone # pylint: disable=fixme - # For now, generate mock data with variation for demonstration purposes - compliance_rate = random.uniform(80, 100) # nosec B311 - Mock data only - - timeline.append(TimelinePoint(date=date_str, compliance_rate=round(compliance_rate, 1))) + # Build compliance timeline from actual report data + timeline = _build_compliance_timeline(store, domain_id) return DomainReportsResponse(reports=report_entries, compliance_timeline=timeline) +def _build_compliance_timeline(store: ReportStore, domain: str) -> List[TimelinePoint]: + """ + Build a compliance timeline from actual report data stored in ReportStore. + + Groups reports by date and calculates the pass rate per day to provide + real historical trend data for the compliance chart. + """ + all_reports = store.get_domain_reports(domain) + + # Aggregate report data by date + daily_data: Dict[str, Dict[str, int]] = {} + for report in all_reports: + # Use begin_date to determine the day of this report + begin = report.get("begin_date", 0) + if isinstance(begin, (int, float)) and begin > 0: + date_str = datetime.fromtimestamp(begin).strftime("%Y-%m-%d") + elif isinstance(begin, str): + # Handle ISO-format strings + try: + date_str = datetime.fromisoformat(begin).strftime("%Y-%m-%d") + except (ValueError, TypeError): + continue + else: + continue + + if date_str not in daily_data: + daily_data[date_str] = {"total": 0, "passed": 0} + + summary = report.get("summary", {}) + daily_data[date_str]["total"] += summary.get("total_count", 0) + daily_data[date_str]["passed"] += summary.get("passed_count", 0) + + # Convert to timeline points sorted by date + timeline = [] + for date_str in sorted(daily_data.keys()): + data = daily_data[date_str] + rate = round((data["passed"] / data["total"]) * 100, 1) if data["total"] > 0 else 0.0 + timeline.append(TimelinePoint(date=date_str, compliance_rate=rate)) + + return timeline + + @router.get("/{domain_id}/sources", response_model=DomainSourcesResponse) async def get_domain_sources( domain_id: str = Path(..., title="The domain ID or name"), diff --git a/backend/app/tests/test_dashboard_timeline.py b/backend/app/tests/test_dashboard_timeline.py new file mode 100644 index 0000000..a245f4e --- /dev/null +++ b/backend/app/tests/test_dashboard_timeline.py @@ -0,0 +1,147 @@ +"""Tests for the domain compliance timeline with real data.""" + +from fastapi.testclient import TestClient + +from app.services.report_store import ReportStore + + +def _add_report_to_store( + domain, report_id, begin_ts, end_ts, total, passed, failed, org_name="test.org" +): + """Helper to add a report to the ReportStore with integer timestamps.""" + store = ReportStore.get_instance() + store.add_report( + { + "domain": domain, + "report_id": report_id, + "org_name": org_name, + "begin_date": begin_ts, + "end_date": end_ts, + "begin_timestamp": begin_ts, + "end_timestamp": end_ts, + "policy": "none", + "records": [], + "summary": { + "total_count": total, + "passed_count": passed, + "failed_count": failed, + }, + } + ) + + +class TestComplianceTimeline: + """Tests that the compliance timeline returns real data instead of mock.""" + + def test_timeline_has_entries_after_upload(self, client: TestClient): + """When a domain has reports, the timeline should have entries.""" + _add_report_to_store("example.com", "rpt-001", 1597449600, 1597535999, 10, 8, 2) + + response = client.get("/api/v1/domains/example.com/reports?limit=10") + assert response.status_code == 200 + data = response.json() + timeline = data["compliance_timeline"] + assert len(timeline) >= 1 + + def test_timeline_uses_real_dates(self, client: TestClient): + """Timeline dates should come from actual report begin_dates.""" + _add_report_to_store("example.com", "rpt-001", 1597449600, 1597535999, 10, 8, 2) + + response = client.get("/api/v1/domains/example.com/reports?limit=10") + data = response.json() + timeline = data["compliance_timeline"] + + # The begin_date=1597449600 is 2020-08-15 + dates = [point["date"] for point in timeline] + assert "2020-08-15" in dates + + def test_timeline_compliance_rate_is_deterministic(self, client: TestClient): + """Compliance rate should be deterministic, not random.""" + _add_report_to_store("example.com", "rpt-001", 1597449600, 1597535999, 10, 8, 2) + + resp1 = client.get("/api/v1/domains/example.com/reports?limit=10") + resp2 = client.get("/api/v1/domains/example.com/reports?limit=10") + + timeline1 = resp1.json()["compliance_timeline"] + timeline2 = resp2.json()["compliance_timeline"] + assert timeline1 == timeline2 + + def test_timeline_empty_for_domain_with_no_valid_dates(self, client: TestClient): + """A domain with begin_date=0 should have an empty timeline.""" + store = ReportStore.get_instance() + store.add_report( + { + "domain": "empty-timeline.com", + "report_id": "rpt-empty", + "org_name": "test", + "begin_date": 0, + "end_date": 0, + "policy": "none", + "records": [], + "summary": {"total_count": 0, "passed_count": 0, "failed_count": 0}, + } + ) + + response = client.get("/api/v1/domains/empty-timeline.com/reports?limit=10") + assert response.status_code == 200 + data = response.json() + assert data["compliance_timeline"] == [] + + def test_timeline_handles_iso_string_dates(self, client: TestClient): + """Timeline should handle reports with ISO-format string dates.""" + from app.api.api_v1.endpoints.domains import _build_compliance_timeline + + store = ReportStore.get_instance() + store.add_report( + { + "domain": "isodate.com", + "report_id": "rpt-iso", + "org_name": "test", + "begin_date": "2020-08-15T00:00:00", + "end_date": "2020-08-15T23:59:59", + "policy": "none", + "records": [], + "summary": {"total_count": 10, "passed_count": 9, "failed_count": 1}, + } + ) + + timeline = _build_compliance_timeline(store, "isodate.com") + assert len(timeline) == 1 + assert timeline[0].date == "2020-08-15" + assert timeline[0].compliance_rate == 90.0 + + +class TestBuildComplianceTimelineMultipleReports: + """Test timeline aggregation with multiple reports.""" + + def test_multiple_reports_same_day(self, client: TestClient): + """Multiple reports on the same day should be aggregated.""" + _add_report_to_store("multi.com", "rpt-1", 1597449600, 1597535999, 10, 8, 2) + _add_report_to_store("multi.com", "rpt-2", 1597449600, 1597535999, 10, 6, 4) + + response = client.get("/api/v1/domains/multi.com/reports?limit=10") + assert response.status_code == 200 + data = response.json() + timeline = data["compliance_timeline"] + + assert len(timeline) == 1 + assert timeline[0]["date"] == "2020-08-15" + # Aggregated: 14 passed out of 20 total = 70% + assert timeline[0]["compliance_rate"] == 70.0 + + def test_reports_on_different_days(self, client: TestClient): + """Reports on different days should produce separate timeline points.""" + _add_report_to_store("days.com", "rpt-d1", 1597449600, 1597535999, 10, 10, 0) + _add_report_to_store("days.com", "rpt-d2", 1597536000, 1597622399, 10, 5, 5) + + response = client.get("/api/v1/domains/days.com/reports?limit=10") + assert response.status_code == 200 + data = response.json() + timeline = data["compliance_timeline"] + + assert len(timeline) == 2 + # Sorted by date + assert timeline[0]["date"] == "2020-08-15" + assert timeline[0]["compliance_rate"] == 100.0 + assert timeline[1]["date"] == "2020-08-16" + assert timeline[1]["compliance_rate"] == 50.0 diff --git a/backend/app/tests/test_stats_summarizer.py b/backend/app/tests/test_stats_summarizer.py new file mode 100644 index 0000000..0bc3fa2 --- /dev/null +++ b/backend/app/tests/test_stats_summarizer.py @@ -0,0 +1,187 @@ +"""Tests for the StatsSummarizer with real database queries.""" + +import shutil +import tempfile + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +import app.models.domain # noqa: F401 +import app.models.report # noqa: F401 +import app.models.user # noqa: F401 +from app.core.database import Base +from app.models.domain import Domain +from app.models.report import DMARCReport, ReportRecord +from app.utils.stats_summarizer import StatsSummarizer + + +@pytest.fixture() +def db_session(): + """Create a fresh in-memory SQLite database session.""" + engine = create_engine("sqlite://", connect_args={"check_same_thread": False}) + Base.metadata.create_all(engine) + TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) + db = TestingSessionLocal() + try: + yield db + finally: + db.close() + Base.metadata.drop_all(engine) + engine.dispose() + + +@pytest.fixture() +def summarizer(): + """Create a StatsSummarizer with a temp cache directory.""" + cache_dir = tempfile.mkdtemp() + s = StatsSummarizer(cache_dir=cache_dir) + yield s + shutil.rmtree(cache_dir, ignore_errors=True) + + +def _seed_domain_and_reports(db, domain_name="example.com"): + """Insert a domain with reports and records into the database.""" + domain = Domain(name=domain_name) + db.add(domain) + db.flush() + + # Report 1: 2 records, 1 fully passing, 1 failing + report1 = DMARCReport( + domain_id=domain.id, + report_id="rpt-001", + org_name="google.com", + begin_date=1597449600, # 2020-08-15 + end_date=1597535999, + policy="none", + ) + db.add(report1) + db.flush() + + # Record: 5 emails, both pass + rec1 = ReportRecord( + report_id=report1.id, + source_ip="203.0.113.1", + count=5, + disposition="none", + dkim="pass", + spf="pass", + ) + # Record: 3 emails, both fail + rec2 = ReportRecord( + report_id=report1.id, + source_ip="198.51.100.1", + count=3, + disposition="quarantine", + dkim="fail", + spf="fail", + ) + db.add_all([rec1, rec2]) + db.flush() + return domain + + +class TestStatsSummarizerGlobal: + """Tests for global statistics.""" + + def test_empty_database_returns_zeros(self, db_session, summarizer): + stats = summarizer.calculate_summary_statistics(db_session) + assert stats["total_domains"] == 0 + assert stats["total_emails"] == 0 + assert stats["compliance_rate"] == 0.0 + assert stats["reports_processed"] == 0 + assert stats["top_sources"] == [] + assert stats["compliance_trend"] == [] + + def test_global_stats_with_data(self, db_session, summarizer): + _seed_domain_and_reports(db_session, "example.com") + db_session.commit() + + stats = summarizer.calculate_summary_statistics(db_session) + assert stats["total_domains"] == 1 + assert stats["total_emails"] == 8 # 5 + 3 + assert stats["compliant_emails"] == 5 # only rec1 passes + assert stats["compliance_rate"] == 62.5 # 5/8 * 100 + assert stats["reports_processed"] == 1 + + def test_global_top_sources(self, db_session, summarizer): + _seed_domain_and_reports(db_session) + db_session.commit() + + stats = summarizer.calculate_summary_statistics(db_session) + assert len(stats["top_sources"]) == 2 + # Sorted by count descending + assert stats["top_sources"][0]["ip"] == "203.0.113.1" + assert stats["top_sources"][0]["count"] == 5 + + def test_multiple_domains(self, db_session, summarizer): + _seed_domain_and_reports(db_session, "example.com") + _seed_domain_and_reports(db_session, "test.org") + db_session.commit() + + stats = summarizer.calculate_summary_statistics(db_session) + assert stats["total_domains"] == 2 + assert stats["total_emails"] == 16 # 8 * 2 + assert stats["reports_processed"] == 2 + + +class TestStatsSummarizerDomain: + """Tests for domain-specific statistics.""" + + def test_nonexistent_domain(self, db_session, summarizer): + stats = summarizer.calculate_summary_statistics(db_session, domain_id="nope.com") + assert stats["domain"] == "nope.com" + assert stats["total_emails"] == 0 + assert stats["compliance_rate"] == 0.0 + + def test_domain_stats_with_data(self, db_session, summarizer): + _seed_domain_and_reports(db_session, "example.com") + db_session.commit() + + stats = summarizer.calculate_summary_statistics(db_session, domain_id="example.com") + assert stats["domain"] == "example.com" + assert stats["total_emails"] == 8 + assert stats["compliant_emails"] == 5 + assert stats["compliance_rate"] == 62.5 + assert stats["reports_processed"] == 1 + + def test_domain_sources(self, db_session, summarizer): + _seed_domain_and_reports(db_session, "example.com") + db_session.commit() + + stats = summarizer.calculate_summary_statistics(db_session, domain_id="example.com") + assert len(stats["sources"]) == 2 + # First source should be the highest count + assert stats["sources"][0]["ip"] == "203.0.113.1" + assert stats["sources"][0]["count"] == 5 + + def test_domain_isolation(self, db_session, summarizer): + """Stats for one domain should not include data from another.""" + _seed_domain_and_reports(db_session, "example.com") + _seed_domain_and_reports(db_session, "other.org") + db_session.commit() + + stats = summarizer.calculate_summary_statistics(db_session, domain_id="example.com") + assert stats["total_emails"] == 8 # Only example.com's data + + +class TestStatsSummarizerCaching: + """Tests for the caching layer.""" + + def test_caching_returns_same_data(self, db_session, summarizer): + _seed_domain_and_reports(db_session) + db_session.commit() + + stats1 = summarizer.calculate_summary_statistics(db_session) + stats2 = summarizer.calculate_summary_statistics(db_session) + assert stats1 == stats2 + + def test_invalidate_cache(self, db_session, summarizer): + _seed_domain_and_reports(db_session) + db_session.commit() + + summarizer.calculate_summary_statistics(db_session) + summarizer.invalidate_cache() + # Should recalculate after invalidation + stats = summarizer.calculate_summary_statistics(db_session) + assert stats["total_domains"] == 1 diff --git a/backend/app/utils/stats_summarizer.py b/backend/app/utils/stats_summarizer.py index 07ebada..cd10899 100644 --- a/backend/app/utils/stats_summarizer.py +++ b/backend/app/utils/stats_summarizer.py @@ -2,7 +2,13 @@ import json import logging import os from datetime import datetime, timedelta -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional + +from sqlalchemy import case, func +from sqlalchemy.orm import Session + +from app.models.domain import Domain +from app.models.report import DMARCReport, ReportRecord # Setup logger logger = logging.getLogger(__name__) @@ -130,7 +136,9 @@ class StatsSummarizer: safe_domain = domain_id.replace(".", "_").replace("/", "_") return os.path.join(self.cache_dir, f"domain_{safe_domain}.json") - def calculate_summary_statistics(self, _db, domain_id: Optional[str] = None) -> Dict[str, Any]: + def calculate_summary_statistics( + self, db: Session, domain_id: Optional[str] = None + ) -> Dict[str, Any]: """ Calculate summary statistics from the database @@ -141,68 +149,216 @@ class StatsSummarizer: Returns: Dictionary with summary statistics """ - # In a real implementation, this would query the database - # using SQLAlchemy models and calculate statistics - # For now, we'll return mock statistics - # First check if we have cached stats cached_stats = self.get_cached_summary(domain_id) if cached_stats: return cached_stats - # If no cached stats, calculate from database - # In a real implementation, this would be done with SQL queries - # optimized for performance with large datasets - - # For now, mock statistics if domain_id is None: - # Global statistics - stats = { - "total_domains": 5, - "total_emails": 1250, - "compliant_emails": 1100, - "compliance_rate": 88.0, - "reports_processed": 25, - "top_sources": [ - {"ip": "192.168.1.1", "count": 150}, - {"ip": "10.0.0.1", "count": 120}, - {"ip": "172.16.0.1", "count": 100}, - ], - "compliance_trend": [ - {"date": "2025-04-13", "rate": 85.5}, - {"date": "2025-04-14", "rate": 86.2}, - {"date": "2025-04-15", "rate": 86.8}, - {"date": "2025-04-16", "rate": 87.3}, - {"date": "2025-04-17", "rate": 87.9}, - {"date": "2025-04-18", "rate": 88.4}, - {"date": "2025-04-19", "rate": 88.0}, - ], - } + stats = self._calculate_global_statistics(db) else: - # Domain-specific statistics - stats = { - "domain": domain_id, - "total_emails": 250, - "compliant_emails": 220, - "compliance_rate": 88.0, - "reports_processed": 5, - "sources": [ - {"ip": "192.168.1.1", "count": 100, "spf": "pass", "dkim": "pass"}, - {"ip": "10.0.0.1", "count": 80, "spf": "pass", "dkim": "fail"}, - {"ip": "172.16.0.1", "count": 70, "spf": "fail", "dkim": "pass"}, - ], - "compliance_trend": [ - {"date": "2025-04-13", "rate": 85.0}, - {"date": "2025-04-14", "rate": 86.0}, - {"date": "2025-04-15", "rate": 87.0}, - {"date": "2025-04-16", "rate": 87.5}, - {"date": "2025-04-17", "rate": 88.0}, - {"date": "2025-04-18", "rate": 88.5}, - {"date": "2025-04-19", "rate": 88.0}, - ], - } + stats = self._calculate_domain_statistics(db, domain_id) # Cache the statistics self.save_summary(stats, domain_id) return stats + + def _calculate_global_statistics(self, db: Session) -> Dict[str, Any]: + """Calculate global statistics across all domains from the database.""" + # Count total domains + total_domains = db.query(func.count(Domain.id)).scalar() or 0 + + # Aggregate email counts from report records + totals = db.query( + func.coalesce(func.sum(ReportRecord.count), 0).label("total_emails"), + ).first() + total_emails = int(totals.total_emails) if totals else 0 + + # Count compliant emails (DKIM pass OR SPF pass) + compliant_emails = ( + db.query(func.coalesce(func.sum(ReportRecord.count), 0)) + .filter((ReportRecord.dkim == "pass") | (ReportRecord.spf == "pass")) + .scalar() + ) + compliant_emails = int(compliant_emails) if compliant_emails else 0 + + # Count reports processed + reports_processed = db.query(func.count(DMARCReport.id)).scalar() or 0 + + # Compliance rate + compliance_rate = 0.0 + if total_emails > 0: + compliance_rate = round((compliant_emails / total_emails) * 100, 1) + + # Top sending sources by volume + top_sources = self._get_top_sources(db) + + # Compliance trend over recent days + compliance_trend = self._get_compliance_trend(db) + + return { + "total_domains": total_domains, + "total_emails": total_emails, + "compliant_emails": compliant_emails, + "compliance_rate": compliance_rate, + "reports_processed": reports_processed, + "top_sources": top_sources, + "compliance_trend": compliance_trend, + } + + def _calculate_domain_statistics(self, db: Session, domain_id: str) -> Dict[str, Any]: + """Calculate statistics for a specific domain from the database.""" + # Look up the domain by name + domain = db.query(Domain).filter(Domain.name == domain_id).first() + if not domain: + return { + "domain": domain_id, + "total_emails": 0, + "compliant_emails": 0, + "compliance_rate": 0.0, + "reports_processed": 0, + "sources": [], + "compliance_trend": [], + } + + # Aggregate email counts for this domain + total_emails = ( + db.query(func.coalesce(func.sum(ReportRecord.count), 0)) + .join(DMARCReport, ReportRecord.report_id == DMARCReport.id) + .filter(DMARCReport.domain_id == domain.id) + .scalar() + ) + total_emails = int(total_emails) if total_emails else 0 + + # Count compliant emails for this domain + compliant_emails = ( + db.query(func.coalesce(func.sum(ReportRecord.count), 0)) + .join(DMARCReport, ReportRecord.report_id == DMARCReport.id) + .filter(DMARCReport.domain_id == domain.id) + .filter((ReportRecord.dkim == "pass") | (ReportRecord.spf == "pass")) + .scalar() + ) + compliant_emails = int(compliant_emails) if compliant_emails else 0 + + # Count reports for this domain + reports_processed = ( + db.query(func.count(DMARCReport.id)).filter(DMARCReport.domain_id == domain.id).scalar() + ) or 0 + + # Compliance rate + compliance_rate = 0.0 + if total_emails > 0: + compliance_rate = round((compliant_emails / total_emails) * 100, 1) + + # Top sources for this domain + sources = self._get_domain_sources(db, domain.id) + + # Compliance trend for this domain + compliance_trend = self._get_compliance_trend(db, domain.id) + + return { + "domain": domain_id, + "total_emails": total_emails, + "compliant_emails": compliant_emails, + "compliance_rate": compliance_rate, + "reports_processed": reports_processed, + "sources": sources, + "compliance_trend": compliance_trend, + } + + def _get_top_sources(self, db: Session, limit: int = 10) -> List[Dict[str, Any]]: + """Get top sending sources by email volume across all domains.""" + results = ( + db.query( + ReportRecord.source_ip, + func.sum(ReportRecord.count).label("total_count"), + ) + .group_by(ReportRecord.source_ip) + .order_by(func.sum(ReportRecord.count).desc()) + .limit(limit) + .all() + ) + + return [{"ip": row.source_ip, "count": int(row.total_count)} for row in results] + + def _get_domain_sources( + self, db: Session, domain_db_id: int, limit: int = 10 + ) -> List[Dict[str, Any]]: + """Get top sending sources for a specific domain.""" + results = ( + db.query( + ReportRecord.source_ip, + func.sum(ReportRecord.count).label("total_count"), + ReportRecord.spf, + ReportRecord.dkim, + ) + .join(DMARCReport, ReportRecord.report_id == DMARCReport.id) + .filter(DMARCReport.domain_id == domain_db_id) + .group_by(ReportRecord.source_ip, ReportRecord.spf, ReportRecord.dkim) + .order_by(func.sum(ReportRecord.count).desc()) + .limit(limit) + .all() + ) + + return [ + { + "ip": row.source_ip, + "count": int(row.total_count), + "spf": row.spf or "unknown", + "dkim": row.dkim or "unknown", + } + for row in results + ] + + def _get_compliance_trend( + self, db: Session, domain_db_id: Optional[int] = None, days: int = 30 + ) -> List[Dict[str, Any]]: + """ + Calculate compliance trend over recent days from report data. + + Groups reports by their date range and calculates daily compliance rates. + """ + cutoff = datetime.now() - timedelta(days=days) + cutoff_ts = int(cutoff.timestamp()) + + # Build the base query for records within the time window + query = ( + db.query( + DMARCReport.begin_date, + func.sum(ReportRecord.count).label("total"), + func.sum( + case( + ( + (ReportRecord.dkim == "pass") | (ReportRecord.spf == "pass"), + ReportRecord.count, + ), + else_=0, + ) + ).label("passed"), + ) + .join(ReportRecord, ReportRecord.report_id == DMARCReport.id) + .filter(DMARCReport.begin_date >= cutoff_ts) + ) + + if domain_db_id is not None: + query = query.filter(DMARCReport.domain_id == domain_db_id) + + results = query.group_by(DMARCReport.begin_date).order_by(DMARCReport.begin_date).all() + + # Convert timestamps to dates and aggregate per day + daily: Dict[str, Dict[str, int]] = {} + for row in results: + date_str = datetime.fromtimestamp(row.begin_date).strftime("%Y-%m-%d") + if date_str not in daily: + daily[date_str] = {"total": 0, "passed": 0} + daily[date_str]["total"] += int(row.total) + daily[date_str]["passed"] += int(row.passed) + + trend = [] + for date_str in sorted(daily.keys()): + data = daily[date_str] + rate = round((data["passed"] / data["total"]) * 100, 1) if data["total"] > 0 else 0.0 + trend.append({"date": date_str, "rate": rate}) + + return trend