import datetime import io import msoffcrypto # type: ignore[import-untyped] import pandas as pd from openpyxl import Workbook from openpyxl.styles import Font from openpyxl.worksheet.worksheet import Worksheet from backend.dtos import DsarResult, ExportRecord, Profile, Subscription def _naive(dt: datetime.datetime | None) -> datetime.datetime | None: return dt.replace(tzinfo=None) if dt is not None else None def profile_rows(profile: Profile) -> dict[str, str]: dob = profile.date_of_birth age = None if dob: today = datetime.date.today() age = today.year - dob.year - ((today.month, today.day) < (dob.month, dob.day)) candidates = { "First Name": profile.first_name, "Last Name": profile.last_name, "Full Name": ( " ".join(filter(None, [profile.first_name, profile.last_name])) or None ), "Mobile Phone": profile.phone_number, "Address 1": profile.address, "City": profile.city, "State": profile.region, "Country / Region": profile.country_code, "Postal Code": profile.zip_code, "Birthday": profile.birthday, "Birthdate": str(dob) if dob else None, "Age": str(age) if age is not None else None, } return {k: v for k, v in candidates.items() if v} _FANSIFTER_REASON_MAP = { "ADS_GOOGLE": "Google", "ADS_META": "Meta", "ADS_SNAPCHAT": "Snapchat", "ADS_SPOTIFY": "Spotify", "ADS_TIKTOK": "Tiktok", "EMAIL_SALESFORCE": "Email for Salesforce", } def _destination(e: ExportRecord) -> str | None: if e.source == "Hightouch": return e.recipient_name if e.source == "Salesforce": return e.recipient_name if e.source == "Fansifter": return _FANSIFTER_REASON_MAP.get(e.export_type or "", e.export_type) return None def _recipient_name(e: ExportRecord) -> str | None: if e.source == "Hightouch": return "-" if e.source == "Salesforce": return e.exported_by if e.source == "Fansifter": return e.exported_by return None def exports_df(exports: list[ExportRecord]) -> pd.DataFrame: rows = [ { "Export Date": _naive(e.export_date), "Business Reason": e.business_reason.lower() if e.business_reason else None, "Destination": _destination(e), "Exported By": _recipient_name(e), "Fan Email": e.fan_email, "Source": e.source, } for e in exports ] return pd.DataFrame(rows) def subscriptions_df(subscriptions: list[Subscription]) -> pd.DataFrame: rows = [ {"Created Date": _naive(s.created_at), "Artist": s.artist_name} for s in subscriptions ] return pd.DataFrame(rows) def _write_sheet(ws: Worksheet, df: pd.DataFrame, empty_msg: str) -> None: if df.empty: ws.append([empty_msg]) return ws.append(list(df.columns)) for cell in ws[1]: cell.font = Font(bold=True) for row in df.itertuples(index=False): ws.append(list(row)) def build_excel(result: DsarResult, password: str | None = None) -> bytes: wb = Workbook() del wb[wb.sheetnames[0]] _write_sheet( wb.create_sheet("Export History"), exports_df(result.exports), "No export records found.", ) ws_pii = wb.create_sheet("PII") if result.profile: pii = profile_rows(result.profile) ws_pii.append(["Field", "Value"]) for cell in ws_pii[1]: cell.font = Font(bold=True) for field, value in pii.items(): ws_pii.append([field, value]) if result.profile.deleted: deleted_at = result.profile.deleted_at ws_pii.append( ["Deletion Date", str(deleted_at) if deleted_at else "Unknown"] ) else: ws_pii.append(["No profile found."]) _write_sheet( wb.create_sheet("Fan Active Subscriptions"), subscriptions_df(result.subscriptions), "No active subscriptions found.", ) buf = io.BytesIO() wb.save(buf) if password: buf.seek(0) encrypted = io.BytesIO() office_file = msoffcrypto.OfficeFile(buf) office_file.encrypt(password, encrypted) return encrypted.getvalue() return buf.getvalue()