Files
vrobbler/vrobbler/apps/scrobbles/dataclasses.py
Colin Powell 6a8432c08f
All checks were successful
build / test (push) Successful in 1m54s
[locations] Handle log data cleaner
2026-06-16 10:12:58 -04:00

263 lines
7.3 KiB
Python

import json
from dataclasses import asdict, dataclass
from typing import Optional
from dataclass_wizard import JSONWizard
from django import forms
from django.contrib.auth import get_user_model
from django.db.models import Count
from people.models import Person
from scrobbles.forms import form_from_dataclass
User = get_user_model()
class ScrobbleLogDataEncoder(json.JSONEncoder):
def default(self, o):
try:
return o.__dict__
except AttributeError:
return {}
class ScrobbleLogDataDecoder(json.JSONDecoder):
def default(self, o):
return o.__dict__
class JSONDataclass(JSONWizard):
@property
def asdict(self):
return asdict(self)
@property
def json(self):
return json.dumps(self.asdict)
@dataclass
class BaseLogData(JSONDataclass):
description: Optional[str] = None
notes: Optional[list[str]] = None
_excluded_fields = {}
@classmethod
def form(cls):
return form_from_dataclass(cls)
@classmethod
def override_fields(cls) -> dict:
return {}
@classmethod
def from_log_dict(cls, log_dict: dict) -> dict:
"""Extract LogData keyword arguments from a stored log dict.
Override in subclasses to handle custom nesting/structure.
"""
return {
k: v
for k, v in log_dict.items()
if k in cls.__dataclass_fields__
}
def notes_as_str(self, separator: str = " | ") -> str:
import html
import re
if not self.notes:
return ""
if isinstance(self.notes, dict):
lines = []
for ts, text in self.notes.items():
note_text = " ".join(text.strip().split())
lines.append(f"{ts}: {note_text}")
return separator.join(lines)
if isinstance(self.notes, str):
return html.escape(re.sub(r"\s*\[start:\d+\|end:\d+\]$", "", self.notes))
if isinstance(self.notes, list):
cleaned_notes = []
for note in self.notes:
if isinstance(note, dict):
timestamp, note_text = next(iter(note.items()))
note_text = " ".join(note_text.strip().split())
cleaned_notes.append(f"{timestamp}: {note_text}")
elif isinstance(note, str):
cleaned = html.escape(
re.sub(r"\s*\[start:\d+\|end:\d+\]$", "", note)
)
cleaned_notes.append(cleaned)
return separator.join(cleaned_notes)
return ""
@staticmethod
def _format_timestamp(ts: str) -> str:
from datetime import datetime, timezone
import re
cleaned = ts.strip().strip("[]")
dt = None
if cleaned.isdigit():
try:
seconds = int(cleaned)
dt = datetime.fromtimestamp(seconds, tz=timezone.utc)
except (ValueError, OSError):
pass
if dt is None:
for fmt in [
"%Y-%m-%dT%H:%M:%S.%fZ",
"%Y-%m-%dT%H:%M:%SZ",
"%Y-%m-%dT%H:%M:%S.%f",
"%Y-%m-%dT%H:%M:%S",
]:
try:
dt = datetime.strptime(cleaned, fmt)
break
except ValueError:
continue
if dt is None:
m = re.match(r"(\d{4}-\d{2}-\d{2})\s+\w{3}\s+(\d{2}:\d{2})", cleaned)
if m:
try:
dt = datetime.strptime(
f"{m.group(1)} {m.group(2)}", "%Y-%m-%d %H:%M"
)
except ValueError:
pass
if dt is None:
try:
dt = datetime.fromisoformat(cleaned)
except ValueError:
pass
if dt:
return dt.strftime("%b %-d, %Y %-I:%M %p")
return ts
def notes_as_html(self) -> str:
import bleach
import markdown
from django.utils.safestring import mark_safe
if not self.notes:
return ""
md = markdown.Markdown(extensions=["extra"])
allowed_tags = [
"p",
"br",
"strong",
"em",
"a",
"ul",
"ol",
"li",
"code",
"pre",
"blockquote",
"h1",
"h2",
"h3",
"h4",
"h5",
"h6",
"hr",
"img",
]
note_items = []
if isinstance(self.notes, dict):
for ts, text in sorted(self.notes.items(), reverse=True):
note_items.append((ts, text.strip()))
elif isinstance(self.notes, str):
for line in self.notes.split("\n"):
if line.strip():
note_items.append((None, line.strip()))
elif isinstance(self.notes, list):
for note in self.notes:
if isinstance(note, dict):
timestamp, note_text = next(iter(note.items()))
note_items.append((timestamp, note_text.strip()))
elif isinstance(note, str):
for line in note.split("\n"):
if line.strip():
note_items.append((None, line.strip()))
html_parts = []
for i, (ts, text) in enumerate(note_items):
if i > 0:
html_parts.append('<hr class="note-divider">')
ts_html = ""
if ts:
ts_html = (
f'<h5 class="note-timestamp">{self._format_timestamp(ts)}</h5>'
)
content_html = bleach.clean(
md.convert(text),
tags=allowed_tags,
strip=True,
)
html_parts.append(
f'<div class="note-item">{ts_html}<div class="note-content">{content_html}</div></div>'
)
return mark_safe("".join(html_parts))
@dataclass
class LongPlayLogData(JSONDataclass):
pass
@dataclass
class SportEventLogData(BaseLogData):
thesportsdb_id: Optional[str] = None
start: Optional[str] = None
round_name: Optional[str] = None
season_name: Optional[str] = None
@dataclass
class WithPeopleLogData(JSONDataclass):
with_people_ids: Optional[list[int]] = None
@property
def with_people(self) -> list["Person"]:
from people.models import Person
if not self.with_people_ids:
return []
return [Person.objects.filter(id=pid).first() for pid in self.with_people_ids]
@classmethod
def override_fields(cls) -> dict:
fields = {}
for base in cls.mro()[1:]:
if hasattr(base, "override_fields"):
base_fields = base.override_fields()
fields.update(base_fields)
custom_fields = {
"with_people_ids": forms.ModelMultipleChoiceField(
queryset=Person.objects.annotate(
scrobble_count=Count("scrobble_associations")
).order_by("-scrobble_count"),
required=False,
widget=forms.SelectMultiple(attrs={"size": 10}),
)
}
fields.update(custom_fields)
return fields