475 lines
16 KiB
Python
475 lines
16 KiB
Python
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"
|
|
)
|