1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
from collections import Counter
from datetime import date, timedelta
from fastapi import APIRouter, Depends, Request
from fastapi.templating import Jinja2Templates
from sqlalchemy.orm import Session
from database import get_db
from models import Film
from services.countries import (
ISO_NUMERIC_TO_COUNTRY_NAME,
country_name_to_iso_numeric,
split_country_names,
)
from services.film_people import split_credit_names
router = APIRouter(tags=["stats"])
templates = Jinja2Templates(directory="templates")
def _build_stats_payload(films: list[Film]) -> dict:
countries = Counter()
country_codes = Counter()
directors = Counter()
star_counts = Counter({0: 0, 1: 0, 2: 0, 3: 0})
months = Counter()
days = Counter()
watched_with = Counter()
for film in films:
country_names = split_country_names(film.country)
countries.update(country_names)
for country in country_names:
iso_numeric = country_name_to_iso_numeric(country)
if iso_numeric is not None:
country_codes[iso_numeric] += 1
directors.update(split_credit_names(film.director))
stars = film.stars if film.stars in {0, 1, 2, 3} else 0
star_counts[stars] += 1
if film.date_watched:
months[film.date_watched.strftime("%Y-%m")] += 1
days[film.date_watched.isoformat()] += 1
companions = split_credit_names(film.watched_with)
if companions:
watched_with.update(companions)
else:
watched_with["solo"] += 1
total_watched = len(films)
rewatched = sum(1 for film in films if film.rewatch or film.rewatch_count > 0)
today = date.today()
start_day = today - timedelta(days=364)
trailing_days = []
cursor = start_day
while cursor <= today:
trailing_days.append({"date": cursor.isoformat(), "count": days[cursor.isoformat()]})
cursor += timedelta(days=1)
return {
"scope": {
"shelf": "diary",
"requires_date_watched": True,
},
"total_watched": total_watched,
"films_per_country": [
{"country": country, "count": count}
for country, count in sorted(countries.items(), key=lambda item: (-item[1], item[0]))
],
"films_per_country_codes": [
{"code": code, "count": count}
for code, count in sorted(country_codes.items(), key=lambda item: (-item[1], item[0]))
],
"country_labels_by_code": {
str(code): ISO_NUMERIC_TO_COUNTRY_NAME.get(code, str(code)) for code in country_codes
},
"most_watched_directors": [
{"director": director, "count": count}
for director, count in sorted(directors.items(), key=lambda item: (-item[1], item[0]))
],
"star_distribution": [{"stars": stars, "count": star_counts[stars]} for stars in (0, 1, 2, 3)],
"films_per_month": [
{"month": month, "count": count}
for month, count in sorted(months.items())
],
"films_per_day": [
{"date": watched_date, "count": count}
for watched_date, count in sorted(days.items())
],
"films_per_day_365": trailing_days,
"rewatch_rate": {
"rewatched": rewatched,
"total_watched": total_watched,
"rate": round(rewatched / total_watched, 4) if total_watched else 0,
},
"watched_with_breakdown": [
{"watched_with": watched_with_value, "count": count}
for watched_with_value, count in sorted(watched_with.items(), key=lambda item: (-item[1], item[0]))
],
}
def _diary_films(db: Session) -> list[Film]:
return (
db.query(Film)
.filter(Film.shelf == "diary", Film.date_watched.is_not(None))
.all()
)
@router.get("/stats")
def stats_page(request: Request):
return templates.TemplateResponse(
request=request,
name="stats.html",
context={"request": request, "active_page": "stats"},
)
@router.get("/stats/data")
def stats_data(db: Session = Depends(get_db)):
return _build_stats_payload(_diary_films(db))
|
