import numpy as np
import pandas as pd
from typing import Literal
from jinja2 import Environment, FileSystemLoader
from markupsafe import Markup
from ..config.colors import ROW_COLORS
from ..config.countries import COUNTRY_DICT
from ..config.paths import HTML_PATHS
ReportType = Literal["cc", "ro", "lb", "mm"]
class EmailRenderer:
def __init__(self):
self._env = Environment(
loader=FileSystemLoader(str(HTML_PATHS.section.parent)),
autoescape=True,
)
self._env_raw = Environment(
loader=FileSystemLoader(str(HTML_PATHS.section.parent)),
autoescape=False,
)
def parse_report(self, df: pd.DataFrame, report_type: ReportType = "cc") -> pd.DataFrame:
df = df.copy()
# Vectorized row color assignment (priority: last condition wins)
conditions = [
df["days_since_release"] <= 6,
df["days_since_entry"] <= 3,
df["is_chart_version"] == 1,
df["is_chart_version"] == 0,
]
choices = [
ROW_COLORS["new_release"],
ROW_COLORS["freshly_cookin"],
ROW_COLORS["charting"],
ROW_COLORS["charting_alt"],
]
df["rowcolor"] = np.select(conditions, choices, default=ROW_COLORS["default"])
df["fire_icon"] = df["cookin_rnk"].apply(self._fire_icon)
if report_type in ("ro", "lb"):
df["global_icon"] = ""
else:
df["global_icon"] = df["n_markets"].apply(self._global_icon)
df["license_icon"] = df["license_type"].apply(self._license_icon)
df["catalog_flag"] = df["days_since_release"].apply(self._catalog_flag)
df["spotify_trend"] = df.apply(
lambda x: self._spotify_trend(x["relative_slope"], x["days_since_release"], x["p_value_2sided"]),
axis=1,
)
df["spotify_streams_txt"] = df["sum_streams_spotify"].apply(lambda x: f"{x/1000:.0f}K")
df["spotify_rank_txt"] = df.apply(
lambda x: self._spotify_rank_txt(
x["spotify_chart_position"], x["distance_from_chart"],
x["report_date"], x["spotify_chart_date"]
),
axis=1,
)
df["tiktok_html"] = df.apply(
lambda x: self._social_html("logo-tiktok.svg", x["tiktok_rank"], x["report_date"], x["tiktok_rank_date"]),
axis=1,
)
df["meta_html"] = df.apply(
lambda x: self._social_html("meta.svg", x["meta_rank"], x["report_date"], x["meta_rank_date"]),
axis=1,
)
df["shazam_html"] = df.apply(
lambda x: self._social_html("logo-shazam.svg", x["shazam_chart_position"], x["report_date"], x["shazam_chart_date"]),
axis=1,
)
df["flag_html"] = df.apply(
lambda x: self._market_info_html("Cookin in", x["country_code"], x["tiktok_rank"], x["meta_rank"], x["shazam_chart_position"], report_type),
axis=1,
)
df["owner_html"] = df.apply(
lambda x: self._market_info_html("Owner", x["owner_abbreviation"], x["tiktok_rank"], x["meta_rank"], x["shazam_chart_position"], report_type),
axis=1,
)
return df
SAFE_COLUMNS = {
"fire_icon", "global_icon", "spotify_trend", "license_icon", "tiktok_html",
"meta_html", "shazam_html", "flag_html", "owner_html", "catalog_flag",
"rowcolor",
}
def render_section(self, row: pd.Series, cover_path: str) -> str:
template = self._env.get_template(HTML_PATHS.section.name)
context: dict[str, str | Markup] = {}
for col, value in row.items():
col_str = str(col)
if col_str == "artwork_url":
context[col_str] = cover_path
elif col_str in self.SAFE_COLUMNS:
context[col_str] = Markup(str(value))
else:
context[col_str] = str(value)
return template.render(**context)
def parse_multi_market_report(self, df: pd.DataFrame, show_owner: bool = False, report_type: ReportType = "mm") -> pd.DataFrame:
"""Aggregate track/country rows into one row per ISRC with per-country metrics."""
df = df.copy()
rows = []
for isrc, group in df.groupby("isrc_cd"):
first = group.iloc[0]
# Best row color priority: new_release > freshly_cookin > charting > default
rowcolor = ROW_COLORS["default"]
if (group["is_chart_version"] == 0).any():
rowcolor = ROW_COLORS["charting_alt"]
if (group["is_chart_version"] == 1).any():
rowcolor = ROW_COLORS["charting"]
if (group["days_since_entry"] <= 3).any():
rowcolor = ROW_COLORS["freshly_cookin"]
if (group["days_since_release"] <= 6).any():
rowcolor = ROW_COLORS["new_release"]
# Build per-country metrics list
country_metrics = []
for _, row in group.iterrows():
cc = row["country_code"]
country_metrics.append({
"country_code": cc,
"country_code_lower": cc.lower(),
"fire_icon": self._fire_icon_inline(row["cookin_rnk"]),
"spotify_trend_icon": self._spotify_trend_icon_only(
row["relative_slope"], row["days_since_release"], row["p_value_2sided"]
),
"spotify_streams_txt": f"{row['sum_streams_spotify']/1000:.0f}K",
"chart_position_txt": self._chart_position_txt(
row["spotify_chart_position"], row["report_date"], row["spotify_chart_date"]
),
"distance_txt": self._distance_from_chart_txt(
row["spotify_chart_position"], row["distance_from_chart"],
row["report_date"], row["spotify_chart_date"]
),
"tiktok_txt": self._social_rank_txt(
row["tiktok_rank"], row["report_date"], row["tiktok_rank_date"]
),
"meta_txt": self._social_rank_txt(
row["meta_rank"], row["report_date"], row["meta_rank_date"]
),
"shazam_txt": self._social_rank_txt(
row["shazam_chart_position"], row["report_date"], row["shazam_chart_date"]
),
})
n_markets = int(first["n_markets"])
max_cookin_rnk = group["cookin_rnk"].max()
total_streams = group["sum_streams_spotify"].sum()
country_flags_html = " · ".join(
r["country_code"] for _, r in group.iterrows()
)
# Pick first non-null artwork URL from the group
artwork_urls = group["artwork_url"].dropna()
artwork_url = artwork_urls.iloc[0] if not artwork_urls.empty else ""
owner_abbreviation = first["owner_abbreviation"]
owner_html = ""
if show_owner and str(owner_abbreviation) != "nan":
oa = str(owner_abbreviation)
oa_lower = oa.lower()
if oa in COUNTRY_DICT:
owner_html = (
f'
'
f'{oa}'
)
else:
owner_html = oa
rows.append({
"isrc_cd": isrc,
"artwork_url": artwork_url,
"product_name": first["product_name"],
"primary_artist_name": first["primary_artist_name"],
"license_type": first["license_type"],
"days_since_release": first["days_since_release"],
"spotify_link": first["spotify_link"],
"insights_link": first["insights_link"],
"report_date": first["report_date"],
"rowcolor": rowcolor,
"global_icon": self._global_icon_badge(n_markets) if report_type not in ("ro", "lb") else "",
"license_icon": self._license_icon(first["license_type"]),
"catalog_flag": self._catalog_flag(first["days_since_release"]),
"owner_html": owner_html,
"owner_abbreviation": str(owner_abbreviation),
"country_flags_html": country_flags_html,
"country_metrics": country_metrics,
"n_country_rows": len(country_metrics),
"total_streams": total_streams,
"show_chart_pos": any(cm["chart_position_txt"] for cm in country_metrics),
"show_distance": any(cm["distance_txt"] for cm in country_metrics),
})
result = pd.DataFrame(rows)
result = result.sort_values("total_streams", ascending=False).reset_index(drop=True)
return result
MULTI_MARKET_SAFE_COLUMNS = {
"global_icon", "license_icon", "catalog_flag",
"rowcolor", "country_flags_html", "owner_html",
}
def render_multi_market_section(self, row: pd.Series, cover_path: str) -> str:
"""Render a multi-market card with per-country metric rows."""
template = self._env_raw.get_template("section_multi_market.html.j2")
context: dict = {}
for col, value in row.items():
col_str = str(col)
if col_str == "artwork_url":
context[col_str] = cover_path
elif col_str == "country_metrics":
context[col_str] = value
elif col_str in self.MULTI_MARKET_SAFE_COLUMNS:
context[col_str] = Markup(str(value))
elif isinstance(value, (bool, np.bool_)):
context[col_str] = bool(value)
else:
context[col_str] = str(value) if not isinstance(value, (list, dict)) else value
return template.render(**context)
@staticmethod
def _fire_icon(perc: float) -> str:
if perc < 0.99:
return ""
return '
'
@staticmethod
def _fire_icon_inline(perc: float) -> str:
"""Inline fire icon for multi-market country rows."""
if perc < 0.99:
return ""
return ''
@staticmethod
def _global_icon(n_markets: int, bottom: int = 5, right: int = 965) -> str:
if n_markets < 4:
return ""
return f'''
{n_markets}
'''
@staticmethod
def _global_icon_badge(n_markets: int) -> str:
"""Badge-style global icon for overlay on cover art (no absolute position wrapper)."""
if n_markets < 4:
return ""
return f'''
{n_markets}
'''
@staticmethod
def _license_icon(license_type: str) -> str:
license_val = "unknown" if str(license_type) == "nan" else license_type
return f'
'
@staticmethod
def _catalog_flag(days: int) -> str:
if days < 1260:
return ""
return '
'
@staticmethod
def _spotify_trend(relative_slope: float, days_since_release: int, p_value: float) -> str:
width, height = "25", "25"
if days_since_release < 7:
icon = "star.svg"
width, height = "20", "20"
elif relative_slope < -0.02 and p_value < 0.05:
icon = "down.svg"
elif relative_slope > 0.02 and p_value < 0.05:
icon = "up.svg"
else:
icon = "flat.svg"
return f' | '
@staticmethod
def _spotify_rank_txt(
chart_position: float | None,
distance_from_chart: float | None,
report_date,
chart_date,
) -> str:
date_asterisk = "" if report_date == chart_date else "*"
if chart_position is None or (isinstance(chart_position, float) and np.isnan(chart_position)):
if distance_from_chart is None or (isinstance(distance_from_chart, float) and np.isnan(distance_from_chart)):
return "N/A"
dist = max(distance_from_chart, 0)
if dist >= 950:
return f"{dist / 1000:.0f}K{date_asterisk}"
return f"{dist / 1000:.1f}K{date_asterisk}"
return f"#{int(chart_position)}{date_asterisk}"
@staticmethod
def _chart_position_txt(
chart_position: float | None,
report_date,
chart_date,
) -> str:
"""Return chart position only (e.g. '#45'), or empty if not charting."""
if chart_position is None or (isinstance(chart_position, float) and np.isnan(chart_position)):
return ""
date_asterisk = "" if report_date == chart_date else "*"
return f"#{int(chart_position)}{date_asterisk}"
@staticmethod
def _distance_from_chart_txt(
chart_position: float | None,
distance_from_chart: float | None,
report_date,
chart_date,
) -> str:
"""Return distance to chart (e.g. '2.1K'), or empty if already charting."""
if chart_position is not None and not (isinstance(chart_position, float) and np.isnan(chart_position)):
return ""
if distance_from_chart is None or (isinstance(distance_from_chart, float) and np.isnan(distance_from_chart)):
return ""
date_asterisk = "" if report_date == chart_date else "*"
dist = max(distance_from_chart, 0)
if dist >= 950:
return f"{dist / 1000:.0f}K{date_asterisk}"
return f"{dist / 1000:.1f}K{date_asterisk}"
@staticmethod
def _social_html(icon: str, rank: float, report_date, social_date) -> str:
if isinstance(rank, float) and np.isnan(rank):
return ""
date_asterisk = "" if report_date == social_date else "*"
return f'''
|
#{int(rank)}{date_asterisk} |
'''
@staticmethod
def _spotify_trend_icon_only(relative_slope: float, days_since_release: int, p_value: float) -> str:
"""Return just the trend icon img tag (no wrapper) for multi-market cards."""
if days_since_release < 7:
return ' '
elif relative_slope < -0.02 and p_value < 0.05:
return ' '
elif relative_slope > 0.02 and p_value < 0.05:
return ' '
return ' '
@staticmethod
def _social_rank_txt(rank: float, report_date, social_date) -> str:
"""Return a plain text rank string (no icon) for multi-market card table cells."""
if isinstance(rank, float) and np.isnan(rank):
return ""
date_asterisk = "" if report_date == social_date else "*"
return f"#{int(rank)}{date_asterisk}"
@staticmethod
def _market_info_html(
market_type: str,
code: str,
tiktok_rank: float,
meta_rank: float,
shazam_pos: float,
report_type: ReportType,
) -> str:
if report_type == "cc":
return ""
# In multi-market report types, show flag + country code inline after the title
if report_type in ("ro", "lb", "mm"):
if market_type == "Cookin in":
cc_lower = code.lower()
return (
f' '
f'{code}'
)
else:
return ""
|