secstate/vrobbler
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[drinks] Add coffee migration script

Browse Source
This commit is contained in:
Colin Powell
2026-07-14 12:53:27 -04:00
parent c9298fe387
commit f6a20b67a6
4 changed files with 506 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
PROJECT.org
View File

@ -88,7 +88,7 @@ fetching and simple saving.
*** Metadata sources
**** Scraper
* Backlog [0/24] :vrobbler:project:personal:
* Backlog [1/25] :vrobbler:project:personal:
** TODO [#C] After transition to linux add curl_cffi as webpage scrapper again :webpages:metadata:
** TODO [#C] Create small utility to clean up tracks scrobbled with wonky playback times :bug:music:scrobbles:
:PROPERTIES:
@ -619,6 +619,27 @@ The Edit log form should have from top to bottom:
- Expansion ids (which should a multi-select widget of expansions for this game)
- Location (which should be a drop down of BoardGameLocations for this user)
** DONE [#B] Migrate coffee scrobbles to new Coffee model :drinks:mgmtcmd:
:PROPERTIES:
:ID: edaf1200-b58e-8512-b2a2-c8bc25bed7a0
:END:
*** Description
Food ID 10 is "French Press Coffee with Sugar" and we should convert all of
these scrobbles to scrobbles of a Coffee, but to figure out which coffee, we can
use scrobbles of Task "Chore" where the title is "Roast coffee" and the notes
are not null. In this case, the first note on each one has a Coffee title like
"Mexico - Veracruz Huatusco" where the Origin is Mexico and the title is
"Veracruz Huatusco"
We should create a new Coffee for each Chore with the title of "Roast Coffee"
and then for each "French Press Coffee with Sugar", swap the Food ID 10 for the
Coffee ID of the latest Coffee created from a Chore. If a new Chore is found,
start updating the French Press scrobbles with a newly created Coffee.
If we don't have a roast log in the Chore tasks before a French Press entry, we should use the "Unknown Roast" Coffee title and skip all the othe metadata.
* Version 60.0 [2/2]
** DONE [#B] Send water drinking notifications :drinks:notifications:
:PROPERTIES:

474
vrobbler/apps/drinks/management/commands/migrate_coffee_scrobbles.py Normal file
View File

@ -0,0 +1,474 @@
from bisect import bisect_right
from django.core.management.base import BaseCommand
from drinks.models import BrewingMethod, Coffee, CoffeeRoaster, RoastLevel
from scrobbles.models import FavoriteMedia, Scrobble
BASE_RUN_TIME_SECONDS = 2400
# French Press (Food ID 10)
FOOD_ID_FRENCH_PRESS = 10
UNKNOWN_ROAST_TITLE = "Unknown Roast"
TIMBERWYCK_ROASTER_ID = 1
FRENCH_PRESS_CALORIES = 50
FRENCH_PRESS_BASE_TAGS = {"sweet maria", "green coffee"}
# Medium Roast (Food ID 27)
FOOD_ID_MEDIUM_ROAST = 27
MEDIUM_ROAST_TITLE = "Medium Roast"
MEDIUM_ROAST_ROASTER_NAME = "Dunkin' Donuts"
MEDIUM_ROAST_CALORIES = 150
MEDIUM_ROAST_BASE_TAGS = {"dunkin"}
# Lavazza (Food ID 12)
FOOD_ID_LAVAZZA = 12
LAVAZZA_TITLE = "Qualità Oro"
LAVAZZA_ROASTER_NAME = "Lavazza"
LAVAZZA_ORIGIN = "Blended"
LAVAZZA_CALORIES = 50
LAVAZZA_BASE_TAGS = {"lavazza", "espresso"}
# Drip Coffee (Food ID 61)
FOOD_ID_DRIP_COFFEE = 61
DRIP_COFFEE_CALORIES = 50
class Command(BaseCommand):
help = (
"Migrate French Press, Medium Roast, and Lavazza food scrobbles "
"to Coffee scrobbles."
)
def add_arguments(self, parser):
parser.add_argument(
"--dry-run",
action="store_true",
help="Show what would be changed without making changes",
)
def handle(self, *args, **options):
dry_run = options["dry_run"]
if dry_run:
self.stdout.write(
self.style.WARNING("DRY RUN - no changes will be saved\n")
)
# French Press migration
self.stdout.write(self.style.NOTICE("=== French Press Coffee ==="))
french_migrated, french_skipped = self._migrate_french_press(dry_run)
# Medium Roast migration
self.stdout.write(self.style.NOTICE("\n=== Medium Roast ==="))
medium_migrated, medium_skipped = self._migrate_medium_roast(dry_run)
# Lavazza migration
self.stdout.write(self.style.NOTICE("\n=== Lavazza ==="))
lavazza_migrated, lavazza_skipped = self._migrate_lavazza(dry_run)
# Drip Coffee migration
self.stdout.write(self.style.NOTICE("\n=== Drip Coffee ==="))
drip_migrated, drip_skipped = self._migrate_drip_coffee(dry_run)
self.stdout.write("")
total_migrated = (
french_migrated + medium_migrated + lavazza_migrated + drip_migrated
)
total_skipped = french_skipped + medium_skipped + lavazza_skipped + drip_skipped
if dry_run:
self.stdout.write(
self.style.WARNING(
f"Dry run: would migrate {total_migrated} scrobbles, "
f"skip {total_skipped} already migrated. "
f"Use without --dry-run to apply."
)
)
else:
self.stdout.write(
self.style.SUCCESS(
f"Migrated {total_migrated} scrobbles, skipped {total_skipped} "
f"already migrated."
)
)
def _migrate_french_press(self, dry_run):
roast_coffees = self._build_roast_coffee_timeline()
if not roast_coffees:
self.stdout.write(
self.style.WARNING("No 'Roast coffee' task scrobbles found. Aborting.")
)
return 0, 0
unknown_roast = self._get_or_create_unknown_roast(dry_run)
food_scrobbles = self._get_food_scrobbles(FOOD_ID_FRENCH_PRESS)
favorite_foods = self._get_favorite_foods(FOOD_ID_FRENCH_PRESS)
timestamps = [ts for ts, _ in roast_coffees]
migrated = 0
skipped = 0
for scrobble in food_scrobbles.iterator():
if scrobble.coffee_id is not None:
skipped += 1
continue
idx = bisect_right(timestamps, scrobble.timestamp) - 1
if idx >= 0:
_, coffee = roast_coffees[idx]
else:
coffee = unknown_roast
log = scrobble.log or {}
log["brewing_method"] = BrewingMethod.FRENCH_PRESS
log["roast_level"] = RoastLevel.FULL_CITY_PLUS
self.stdout.write(
f" Scrobble {scrobble.id}: Food '{scrobble.food}' -> "
f"Coffee '{coffee}' ({scrobble.timestamp})"
)
if not dry_run:
scrobble.coffee = coffee
scrobble.food = None
scrobble.media_type = Scrobble.MediaType.COFFEE
scrobble.log = log
scrobble.save(update_fields=["coffee", "food", "media_type", "log"])
migrated += 1
self._migrate_favorites(
favorite_foods, roast_coffees, timestamps, unknown_roast, dry_run
)
return migrated, skipped
def _migrate_medium_roast(self, dry_run):
coffee = self._get_or_create_medium_roast(dry_run)
food_scrobbles = self._get_food_scrobbles(FOOD_ID_MEDIUM_ROAST)
favorite_foods = self._get_favorite_foods(FOOD_ID_MEDIUM_ROAST)
migrated = 0
skipped = 0
for scrobble in food_scrobbles.iterator():
if scrobble.coffee_id is not None:
skipped += 1
continue
log = scrobble.log or {}
log["brewing_method"] = BrewingMethod.DRIP
log["roast_level"] = RoastLevel.FULL_CITY
self.stdout.write(
f" Scrobble {scrobble.id}: Food '{scrobble.food}' -> "
f"Coffee '{coffee}' ({scrobble.timestamp})"
)
if not dry_run:
scrobble.coffee = coffee
scrobble.food = None
scrobble.media_type = Scrobble.MediaType.COFFEE
scrobble.log = log
scrobble.save(update_fields=["coffee", "food", "media_type", "log"])
migrated += 1
self._migrate_medium_roast_favorites(favorite_foods, coffee, dry_run)
return migrated, skipped
def _migrate_lavazza(self, dry_run):
coffee = self._get_or_create_lavazza(dry_run)
food_scrobbles = self._get_food_scrobbles(FOOD_ID_LAVAZZA)
favorite_foods = self._get_favorite_foods(FOOD_ID_LAVAZZA)
migrated = 0
skipped = 0
for scrobble in food_scrobbles.iterator():
if scrobble.coffee_id is not None:
skipped += 1
continue
log = scrobble.log or {}
log["brewing_method"] = BrewingMethod.MOKA_POT
self.stdout.write(
f" Scrobble {scrobble.id}: Food '{scrobble.food}' -> "
f"Coffee '{coffee}' ({scrobble.timestamp})"
)
if not dry_run:
scrobble.coffee = coffee
scrobble.food = None
scrobble.media_type = Scrobble.MediaType.COFFEE
scrobble.log = log
scrobble.save(update_fields=["coffee", "food", "media_type", "log"])
migrated += 1
self._migrate_lavazza_favorites(favorite_foods, coffee, dry_run)
return migrated, skipped
def _migrate_lavazza_favorites(self, favorites, coffee, dry_run):
updated = 0
for fav in favorites:
self.stdout.write(
f" FavoriteMedia {fav.id}: Food '{fav.food}' -> Coffee '{coffee}'"
)
if not dry_run:
fav.coffee = coffee
fav.food = None
fav.media_type = Scrobble.MediaType.COFFEE
fav.save(update_fields=["coffee", "food", "media_type"])
updated += 1
if updated:
self.stdout.write(
f" {'Would update' if dry_run else 'Updated'} {updated} FavoriteMedia records"
)
def _migrate_drip_coffee(self, dry_run):
roast_coffees = self._build_roast_coffee_timeline()
if not roast_coffees:
self.stdout.write(
self.style.WARNING("No 'Roast coffee' task scrobbles found. Aborting.")
)
return 0, 0
unknown_roast = self._get_or_create_unknown_roast(dry_run)
food_scrobbles = self._get_food_scrobbles(FOOD_ID_DRIP_COFFEE)
favorite_foods = self._get_favorite_foods(FOOD_ID_DRIP_COFFEE)
timestamps = [ts for ts, _ in roast_coffees]
migrated = 0
skipped = 0
for scrobble in food_scrobbles.iterator():
if scrobble.coffee_id is not None:
skipped += 1
continue
idx = bisect_right(timestamps, scrobble.timestamp) - 1
if idx >= 0:
_, coffee = roast_coffees[idx]
else:
coffee = unknown_roast
log = scrobble.log or {}
log["brewing_method"] = BrewingMethod.DRIP
log["roast_level"] = RoastLevel.CITY
self.stdout.write(
f" Scrobble {scrobble.id}: Food '{scrobble.food}' -> "
f"Coffee '{coffee}' ({scrobble.timestamp})"
)
if not dry_run:
scrobble.coffee = coffee
scrobble.food = None
scrobble.media_type = Scrobble.MediaType.COFFEE
scrobble.log = log
scrobble.save(update_fields=["coffee", "food", "media_type", "log"])
migrated += 1
self._migrate_drip_coffee_favorites(
favorite_foods, roast_coffees, timestamps, unknown_roast, dry_run
)
return migrated, skipped
def _migrate_drip_coffee_favorites(
self, favorites, roast_coffees, timestamps, unknown_roast, dry_run
):
updated = 0
for fav in favorites:
idx = bisect_right(timestamps, fav.created) - 1
coffee = roast_coffees[idx][1] if idx >= 0 else unknown_roast
self.stdout.write(
f" FavoriteMedia {fav.id}: Food '{fav.food}' -> Coffee '{coffee}'"
)
if not dry_run:
fav.coffee = coffee
fav.food = None
fav.media_type = Scrobble.MediaType.COFFEE
fav.save(update_fields=["coffee", "food", "media_type"])
updated += 1
if updated:
self.stdout.write(
f" {'Would update' if dry_run else 'Updated'} {updated} FavoriteMedia records"
)
def _get_or_create_lavazza(self, dry_run):
roaster, _ = CoffeeRoaster.objects.get_or_create(name=LAVAZZA_ROASTER_NAME)
defaults = {
"base_run_time_seconds": BASE_RUN_TIME_SECONDS,
"roaster": roaster,
"calories": LAVAZZA_CALORIES,
"origin": LAVAZZA_ORIGIN,
}
if dry_run:
try:
return Coffee.objects.get(title=LAVAZZA_TITLE, roaster=roaster)
except Coffee.DoesNotExist:
return Coffee(title=LAVAZZA_TITLE, **defaults)
coffee, created = Coffee.objects.get_or_create(
title=LAVAZZA_TITLE,
roaster=roaster,
defaults=defaults,
)
if created:
tags = set(LAVAZZA_TITLE.lower().split()) | LAVAZZA_BASE_TAGS
coffee.tags.add(*tags)
self.stdout.write(f" Created Coffee: '{LAVAZZA_TITLE}' (tags={tags})")
return coffee
def _migrate_medium_roast_favorites(self, favorites, coffee, dry_run):
updated = 0
for fav in favorites:
self.stdout.write(
f" FavoriteMedia {fav.id}: Food '{fav.food}' -> Coffee '{coffee}'"
)
if not dry_run:
fav.coffee = coffee
fav.food = None
fav.media_type = Scrobble.MediaType.COFFEE
fav.save(update_fields=["coffee", "food", "media_type"])
updated += 1
if updated:
self.stdout.write(
f" {'Would update' if dry_run else 'Updated'} {updated} FavoriteMedia records"
)
def _build_roast_coffee_timeline(self):
"""Find Task scrobbles where task title='Chore' and log title='Roast coffee'.
Returns a list of (timestamp, Coffee) sorted by timestamp.
"""
task_scrobbles = (
Scrobble.objects.filter(
media_type=Scrobble.MediaType.TASK,
task__title="Chore",
log__title="Roast coffee",
)
.exclude(log__notes=None)
.exclude(log__notes=[])
.order_by("timestamp")
)
coffees = []
for scrobble in task_scrobbles:
notes = scrobble.log.get("notes") or []
if not notes:
continue
first_note = notes[0] if isinstance(notes, list) else str(notes)
origin, title = self._parse_coffee_name(first_note)
coffee = self._get_or_create_coffee(title, origin)
coffees.append((scrobble.timestamp, coffee))
return coffees
def _parse_coffee_name(self, note):
"""Parse 'Mexico - Veracruz Huatusco' into (origin, title)."""
if " - " in note:
parts = note.split(" - ", 1)
return parts[0].strip(), parts[1].strip()
return None, note.strip()
def _get_or_create_coffee(self, title, origin):
coffee, created = Coffee.objects.get_or_create(
title=title,
origin=origin,
defaults={
"base_run_time_seconds": BASE_RUN_TIME_SECONDS,
"roaster_id": TIMBERWYCK_ROASTER_ID,
"calories": FRENCH_PRESS_CALORIES,
},
)
if created:
tags = set(title.lower().split()) | FRENCH_PRESS_BASE_TAGS
coffee.tags.add(*tags)
self.stdout.write(
f" Created Coffee: '{title}' (origin={origin}, tags={tags})"
)
return coffee
def _get_or_create_unknown_roast(self, dry_run):
defaults = {
"base_run_time_seconds": BASE_RUN_TIME_SECONDS,
"roaster_id": TIMBERWYCK_ROASTER_ID,
"calories": FRENCH_PRESS_CALORIES,
}
if dry_run:
try:
return Coffee.objects.get(title=UNKNOWN_ROAST_TITLE)
except Coffee.DoesNotExist:
return Coffee(title=UNKNOWN_ROAST_TITLE, **defaults)
coffee, created = Coffee.objects.get_or_create(
title=UNKNOWN_ROAST_TITLE,
defaults=defaults,
)
if created:
tags = {"unknown", "roast"} | FRENCH_PRESS_BASE_TAGS
coffee.tags.add(*tags)
self.stdout.write(
f" Created Coffee: '{UNKNOWN_ROAST_TITLE}' (tags={tags})"
)
return coffee
def _get_or_create_medium_roast(self, dry_run):
roaster, _ = CoffeeRoaster.objects.get_or_create(name=MEDIUM_ROAST_ROASTER_NAME)
defaults = {
"base_run_time_seconds": BASE_RUN_TIME_SECONDS,
"roaster": roaster,
"calories": MEDIUM_ROAST_CALORIES,
"origin": None,
}
if dry_run:
try:
return Coffee.objects.get(title=MEDIUM_ROAST_TITLE, roaster=roaster)
except Coffee.DoesNotExist:
return Coffee(title=MEDIUM_ROAST_TITLE, **defaults)
coffee, created = Coffee.objects.get_or_create(
title=MEDIUM_ROAST_TITLE,
roaster=roaster,
defaults=defaults,
)
if created:
tags = set(MEDIUM_ROAST_TITLE.lower().split()) | MEDIUM_ROAST_BASE_TAGS
coffee.tags.add(*tags)
self.stdout.write(f" Created Coffee: '{MEDIUM_ROAST_TITLE}' (tags={tags})")
return coffee
def _get_food_scrobbles(self, food_id):
return Scrobble.objects.filter(
food_id=food_id,
media_type=Scrobble.MediaType.FOOD,
).order_by("timestamp")
def _get_favorite_foods(self, food_id):
return FavoriteMedia.objects.filter(
food_id=food_id,
media_type=Scrobble.MediaType.FOOD,
)
def _migrate_favorites(
self, favorites, roast_coffees, timestamps, unknown_roast, dry_run
):
updated = 0
for fav in favorites:
idx = bisect_right(timestamps, fav.created) - 1
coffee = roast_coffees[idx][1] if idx >= 0 else unknown_roast
self.stdout.write(
f" FavoriteMedia {fav.id}: Food '{fav.food}' -> Coffee '{coffee}'"
)
if not dry_run:
fav.coffee = coffee
fav.food = None
fav.media_type = Scrobble.MediaType.COFFEE
fav.save(update_fields=["coffee", "food", "media_type"])
updated += 1
if updated:
self.stdout.write(
f" {'Would update' if dry_run else 'Updated'} {updated} FavoriteMedia records"
)

2
vrobbler/apps/drinks/management/commands/populate_demo_data.py
View File

@ -486,7 +486,7 @@ class Command(BaseCommand):
playback_position_seconds=media.base_run_time_seconds or 900,
played_to_completion=True,
source=source,
log={"drink_log": log_data},
log=log_data,
**{fk_field: media},
)
scrobbles_created += 1

12
vrobbler/apps/drinks/models.py
View File

@ -40,12 +40,18 @@ class DrinkFormat(models.TextChoices):
DRAFT = "draft", "Draft"
GROWLER = "growler", "Growler"
BOX = "box", "Box"
MUG = "mug", "Mug"
class RoastLevel(models.TextChoices):
LIGHT = "light", "Light"
MEDIUM = "medium", "Medium"
DARK = "dark", "Dark"
CINNAMON = "cinnamon", "Cinnamon"
CITY = "city", "City"
CITY_PLUS = "city_plus", "City+"
FULL_CITY = "full_city", "Full City"
FULL_CITY_PLUS = "full_city_plus", "Full City+"
VIENNA = "vienna", "Vienna"
FRENCH = "french", "French"
SPANISH = "spanish", "Spanish"
class BrewingMethod(models.TextChoices):