mirror of
https://github.com/minetest/contentdb.git
synced 2024-11-10 09:33:44 +01:00
130 lines
3.4 KiB
Python
130 lines
3.4 KiB
Python
import datetime
|
|
from datetime import timedelta
|
|
from typing import Optional
|
|
|
|
from app.models import User, Package, PackageDailyStats, db, PackageState
|
|
from sqlalchemy import func
|
|
|
|
|
|
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
|
|
result = {
|
|
"start": start_date.isoformat(),
|
|
"end": end_date.isoformat(),
|
|
}
|
|
|
|
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)
|
|
|
|
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
|
|
|
|
results = _flatten_data(stats)
|
|
results["package_downloads"] = get_package_overview_for_user(user, stats[0].date, stats[-1].date)
|
|
|
|
return results
|
|
|
|
|
|
def get_package_overview_for_user(user: Optional[User], start_date: datetime.date, end_date: datetime.date):
|
|
query = db.session \
|
|
.query(PackageDailyStats.package_id, PackageDailyStats.date,
|
|
(PackageDailyStats.platform_minetest + PackageDailyStats.platform_other).label("downloads"))
|
|
|
|
if user:
|
|
query = query.filter(PackageDailyStats.package.has(author_id=user.id))
|
|
|
|
stats = query \
|
|
.filter(PackageDailyStats.package.has(state=PackageState.APPROVED)) \
|
|
.order_by(db.asc(PackageDailyStats.package_id), db.asc(PackageDailyStats.date)) \
|
|
.all()
|
|
|
|
stats_by_package = {}
|
|
for stat in stats:
|
|
bucket = stats_by_package.get(stat.package_id, [])
|
|
stats_by_package[stat.package_id] = bucket
|
|
|
|
bucket.append(stat)
|
|
|
|
package_title_by_id = {}
|
|
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()
|
|
|
|
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
|
|
else:
|
|
row.append(0)
|
|
|
|
return result
|
|
|
|
|
|
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),
|
|
}
|