Coverage for seedboxsync/front/apis/uploads_stats.py: 77%
39 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-11 07:37 +0000
1#
2# Copyright (C) 2015-2026 Guillaume Kulakowski <guillaume@kulakowski.fr>
3#
4# For the full copyright and license information, please view the LICENSE
5# file that was distributed with this source code.
6#
7"""SeedboxSync api uploads stats view."""
9from datetime import date
10from typing import Any
11from flask_restx import Namespace, fields
12from seedboxsync.core.database.models import Torrent
13from seedboxsync.front.apis import Resource, parser_period
14from seedboxsync.front.cache import cache
15from seedboxsync.front.login_manager import login_required
17api = Namespace("uploads_stats", path="/uploads/stats", description="Stats related to uploads")
20# ==========================
21# Models
22# ==========================
23stats_source_model = api.model(
24 "StatsSource",
25 {
26 "source": fields.String(
27 required=True,
28 description="Source of the torrent, falback based on announcer",
29 example="torrenter",
30 ),
31 "total": fields.Integer(
32 required=True,
33 description="Number or size of files for this source",
34 example=4989,
35 ),
36 "total_size": fields.Integer(
37 required=True,
38 description="Size of files for this source",
39 example=21678643250867,
40 ),
41 "human_total_size": fields.String(
42 required=True,
43 description="Total size of files with related source",
44 example="19.7 Tio",
45 ),
46 },
47)
48stats_source_envelope = Resource.build_envelope_model(api, "StatsSourceList", nested_model=stats_source_model)
50stats_item_model = api.model(
51 "UploadStatsItem",
52 {
53 "total": fields.Integer(
54 description="Number of uploads",
55 example=12,
56 ),
57 "size": fields.Integer(
58 description="Total size of uploads in bytes",
59 example=123456789,
60 ),
61 "human_size": fields.String(
62 description="Human-readable total size of uploads",
63 example="117.74 MB",
64 ),
65 "trend_total": fields.String(
66 description="Trends vs. prior period for number of uploads",
67 example="+10.0",
68 ),
69 "trend_size": fields.String(
70 description="Trends vs. prior period for size of uploads",
71 example="+15.2",
72 ),
73 },
74)
75stats_model = api.model(
76 "UploadStats",
77 {
78 "day": fields.Nested(
79 stats_item_model,
80 description="Statistics for the current day",
81 ),
82 "week": fields.Nested(
83 stats_item_model,
84 description="Statistics for the current calendar week",
85 ),
86 "month": fields.Nested(
87 stats_item_model,
88 description="Statistics for the current month",
89 ),
90 "year": fields.Nested(
91 stats_item_model,
92 description="Statistics for the current year",
93 ),
94 "last24h": fields.Nested(
95 stats_item_model,
96 description="Statistics for the last 24 hours",
97 ),
98 "last7d": fields.Nested(
99 stats_item_model,
100 description="Statistics for the last 7 days",
101 ),
102 "last30d": fields.Nested(
103 stats_item_model,
104 description="Statistics for the last 30 days",
105 ),
106 },
107)
108stats_envelope = Resource.build_envelope_model(api, "UploadStats", nested_model=stats_model, as_list=False)
111# ==========================
112# Endpoints
113# ==========================
114@api.route("")
115class UploadsStats(Resource):
116 """Resource endpoint to retrieve torrents statistics."""
118 @api.doc("stats_uploads") # type: ignore[untyped-decorator]
119 @api.marshal_with(stats_envelope, code=200, description="Uploads statistics") # type: ignore[untyped-decorator]
120 @login_required # type: ignore[untyped-decorator]
121 def get(self) -> dict[str, Any]:
122 """
123 Fetch overall upload statistics.
125 Retrieves aggregated upload metrics and wraps them within
126 a standard API response envelope[cite: 1, 2].
128 Returns:
129 dict[str, Any]: Formatted envelope response containing upload statistics[cite: 1, 2].
130 """
131 stats = _get_stats()
132 return self.build_envelope(stats, data_total=len(stats), type="UploadStats")
135@api.route("/source")
136class UploadsStatsBySource(Resource):
137 """Resource endpoint to retrieve torrent source statistics."""
139 @api.doc("stats_uploads_by_source") # type: ignore[untyped-decorator]
140 @api.marshal_with(stats_source_envelope, code=200, description="Upload statistics aggregated by source") # type: ignore[untyped-decorator]
141 @api.expect(parser_period) # type: ignore[untyped-decorator]
142 @login_required # type: ignore[untyped-decorator]
143 def get(self) -> dict[str, Any]:
144 """
145 Retrieve torrent statistics grouped by source.
147 Fetches aggregated file counts and total sizes per source from the cache
148 or database, then wraps the dataset into a standard API response envelope.
150 Returns:
151 dict[str, Any]: Envelope containing source statistics, metadata,
152 and total element count.
153 """
154 args = parser_period.parse_args()
155 start_date = args.get("start_date")
156 end_date = args.get("end_date")
158 stats = _get_stats_by_source(start_date, end_date)
160 return self.build_envelope(stats, data_total=len(stats), type="StatsSourceList")
163@cache.memoize(timeout=300)
164def _get_stats_by_source(start_date: date | None, end_date: date | None) -> list[dict[str, object]]:
165 """
166 Fetch torrent statistics grouped by source within an optional date range.
168 Executes the database query to aggregate torrent statistics by source domain
169 filtered by date boundaries if provided, then caches the result using Flask-Caching memoization[cite: 2].
171 Args:
172 start_date (date | None, optional): Optional lower date boundary for filtering. Defaults to None.
173 end_date (date | None, optional): Optional upper date boundary for filtering. Defaults to None.
175 Returns:
176 list[dict[str, object]]: A list of dictionaries containing source statistics,
177 including total counts and associated sizes[cite: 2].
178 """
179 return Torrent.get_stats_by_source(start_date, end_date)
182@cache.memoize(timeout=300)
183def _get_stats() -> dict[str, object]:
184 """
185 Fetch raw upload statistics within an optional date range from the database.
187 Executes the underlying DAO query and caches the result for 5 minutes (300 seconds)
188 using Flask-Caching memoization[cite: 1, 2, 3].
190 Args:
191 start_date (date | None): Optional start date filter for range querying[cite: 1].
192 end_date (date | None): Optional end date filter for range querying[cite: 1].
194 Returns:
195 list[dict[str, object]]: A list of dictionaries representing upload statistics data[cite: 1, 2].
196 """
197 return Torrent.get_stats()