70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
"""One-time rewrite of saved queries to parameterized metric expressions."""
|
|
|
|
import re
|
|
|
|
import click
|
|
|
|
from .extensions import db
|
|
from .models import SavedAnalysis, User
|
|
from .stats import parse_query, query_text
|
|
|
|
|
|
def rewrite(query):
|
|
threshold = '10'
|
|
lines = []
|
|
for line in query.splitlines():
|
|
match = re.fullmatch(r'\s*long run distance >= (\d+(?:\.\d+)?) km\s*', line)
|
|
if match:
|
|
threshold = match[1]
|
|
else:
|
|
lines.append(line)
|
|
daily = any(line.strip() in {'days', 'health'} for line in lines)
|
|
replacements = {
|
|
'days_since_run': 'days_since_run()',
|
|
'days_since_long_run': f'days_since_run(distance >= {threshold} km)',
|
|
**{f'best_{km}km': f'best_time({km} km)' for km in (1, 2, 5, 10)},
|
|
**{f'rolling_{days}d_distance': f'rolling_sum(distance, {days} days)' for days in (7, 28)},
|
|
**{f'rolling_{days}d_count': (f'rolling_sum(sessions, {days} days)' if daily
|
|
else f'rolling_count(distance, {days} days)') for days in (7, 28)},
|
|
}
|
|
pattern = re.compile(r'\b(' + '|'.join(replacements) + r')\b(?!\s*\()')
|
|
result = []
|
|
for line in lines:
|
|
if not line.lstrip().startswith('#'):
|
|
parts = re.split(r'("(?:[^"\\]|\\.)*")', line)
|
|
for index in range(0, len(parts), 2):
|
|
parts[index] = pattern.sub(lambda m: replacements[m[0]], parts[index])
|
|
line = ''.join(parts)
|
|
result.append(line)
|
|
return '\n'.join(result)
|
|
|
|
|
|
def migrate(user):
|
|
changes = []
|
|
for row in db.session.scalars(db.select(SavedAnalysis).where(SavedAnalysis.owner_id == user.id)):
|
|
updated = rewrite(row.query)
|
|
if updated != row.query:
|
|
# Validate every replacement before modifying any persisted query.
|
|
canonical = query_text(parse_query({'query': updated}))
|
|
parse_query({'query': canonical})
|
|
changes.append((row, updated))
|
|
for row, updated in changes:
|
|
row.query = updated
|
|
db.session.commit()
|
|
return len(changes)
|
|
|
|
|
|
def register_cli(app):
|
|
@app.cli.command('migrate-analysis-expressions')
|
|
@click.option('--email', required=True)
|
|
def migrate_queries(email):
|
|
user = db.session.scalar(db.select(User).filter_by(email=email.strip().lower()))
|
|
if not user:
|
|
raise click.ClickException('Account not found.')
|
|
try:
|
|
count = migrate(user)
|
|
except ValueError as error:
|
|
db.session.rollback()
|
|
raise click.ClickException(str(error)) from None
|
|
click.echo(f'Updated {count} saved analyses.')
|