contentdb/app/logic/graphs.py

133 lines
3.6 KiB
Python
Raw Normal View History

2022-11-09 20:45:25 +01:00
import datetime
from datetime import timedelta
2022-11-15 02:51:21 +01:00
from typing import Optional
2022-11-09 20:45:25 +01:00
from app.models import User, Package, PackageDailyStats, db, PackageState
from sqlalchemy import func
2022-11-06 11:32:46 +01:00
def daterange(start_date, end_date):
for n in range(int((end_date - start_date).days) + 1):
yield start_date + timedelta(n)
keys = ["platform_minetest", "platform_other", "reason_new",
"reason_dependency", "reason_update"]
def _flatten_data(stats):
start_date = stats[0].date
end_date = stats[-1].date
2022-11-06 11:32:46 +01:00
result = {
"start": start_date.isoformat(),
"end": end_date.isoformat(),
2022-11-06 11:32:46 +01:00
}
for key in keys:
result[key] = []
i = 0
for date in daterange(start_date, end_date):
stat = stats[i]
if stat.date == date:
for key in keys:
result[key].append(getattr(stat, key))
i += 1
else:
for key in keys:
result[key].append(0)
2022-11-06 11:32:46 +01:00
return result
def get_package_stats(package: Package):
stats = package.daily_stats.order_by(db.asc(PackageDailyStats.date)).all()
if len(stats) == 0:
return None
return _flatten_data(stats)
def get_package_stats_for_user(user: User):
stats = db.session \
.query(PackageDailyStats.date,
func.sum(PackageDailyStats.platform_minetest).label("platform_minetest"),
func.sum(PackageDailyStats.platform_other).label("platform_other"),
func.sum(PackageDailyStats.reason_new).label("reason_new"),
func.sum(PackageDailyStats.reason_dependency).label("reason_dependency"),
func.sum(PackageDailyStats.reason_update).label("reason_update")) \
.filter(PackageDailyStats.package.has(author_id=user.id)) \
.order_by(db.asc(PackageDailyStats.date)) \
.group_by(PackageDailyStats.date) \
.all()
if len(stats) == 0:
return None
2022-11-09 20:45:25 +01:00
results = _flatten_data(stats)
results["package_downloads"] = get_package_overview_for_user(user, stats[0].date, stats[-1].date)
return results
2022-11-15 02:51:21 +01:00
def get_package_overview_for_user(user: Optional[User], start_date: datetime.date, end_date: datetime.date):
query = db.session \
2022-11-09 20:45:25 +01:00
.query(PackageDailyStats.package_id, PackageDailyStats.date,
2022-11-15 02:51:21 +01:00
(PackageDailyStats.platform_minetest + PackageDailyStats.platform_other).label("downloads"))
if user:
query = query.filter(PackageDailyStats.package.has(author_id=user.id))
2023-05-20 01:56:44 +02:00
all_stats = query \
.filter(PackageDailyStats.package.has(state=PackageState.APPROVED),
PackageDailyStats.date >= start_date, PackageDailyStats.date <= end_date) \
2022-11-09 20:45:25 +01:00
.order_by(db.asc(PackageDailyStats.package_id), db.asc(PackageDailyStats.date)) \
.all()
stats_by_package = {}
2023-05-20 01:56:44 +02:00
for stat in all_stats:
2022-11-09 20:45:25 +01:00
bucket = stats_by_package.get(stat.package_id, [])
stats_by_package[stat.package_id] = bucket
bucket.append(stat)
package_title_by_id = {}
2022-11-15 02:51:21 +01:00
pkg_query = user.packages if user else Package.query
for package in pkg_query.filter_by(state=PackageState.APPROVED).all():
if user:
package_title_by_id[package.id] = package.title
else:
package_title_by_id[package.id] = package.getId()
2022-11-09 20:45:25 +01:00
result = {}
for package_id, stats in stats_by_package.items():
i = 0
row = []
result[package_title_by_id[package_id]] = row
for date in daterange(start_date, end_date):
if i >= len(stats):
row.append(0)
continue
stat = stats[i]
if stat.date == date:
row.append(stat.downloads)
i += 1
2023-05-20 01:56:44 +02:00
elif stat.date > date:
2022-11-09 20:45:25 +01:00
row.append(0)
2023-05-20 01:56:44 +02:00
else:
raise Exception(f"Invalid logic, expected stat {stat.date} to be later than {date}")
2022-11-09 20:45:25 +01:00
return result
2022-11-15 02:51:21 +01:00
def get_all_package_stats():
end_date = datetime.datetime.utcnow().date()
start_date = (datetime.datetime.utcnow() - datetime.timedelta(days=29)).date()
return {
"start": start_date.isoformat(),
"end": end_date.isoformat(),
"package_downloads": get_package_overview_for_user(None, start_date, end_date),
}