LinkedIn-Analytics-Dashboard/team_exports.py

"""Per-team-member personal LinkedIn export data, starting with Mckinley's own export
on 2026-10-30 (Aleksandra, Daniel, Niklas, Peter to follow, submitted every Monday).

The Page-level exports (Content/Followers/Visitors/Competitors) only cover the
TXproducts Company Page - this parses each team member's own "Aggregate Analytics"
export instead (LinkedIn's personal-profile analytics download: DISCOVERY, ENGAGEMENT,
TOP POSTS, FOLLOWERS, AUDIENCE DEMOGRAPHICS, CONTENT DEMOGRAPHICS sheets), named
"AggregateAnalytics_<Full Name>_<start>_<end>.xlsx". Each weekly submission is a new
~365-day trailing window that heavily overlaps the previous one, so the daily series
(ENGAGEMENT, FOLLOWERS) are merged across every file on disk the same way the Page's
own Followers export is - newest file wins on an overlapping date - while the
point-in-time snapshots (TOP POSTS, both DEMOGRAPHICS sheets, the running follower
total) always come from the single newest file.

TEAM_MEMBERS is the roster shown in the Team tab's dropdown regardless of whether a
person has submitted an export yet; load_person_data() returns None for anyone who
hasn't, and app.py renders a placeholder for that case.
"""

import io
import os
import re

import pandas as pd
from docx import Document
from docx.oxml.ns import qn

import report
import post_details_tracker

TEAM_MEMBERS = [
    "Peter Gaal",
    "Daniel Schmidt",
    "Niklas Meidow",
    "Aleksandra Kotenko",
    "Mckinley Black",
]

# Personal-profile baseline, distinct from the Page's own app.BASELINE_DATE (2026-06-16) -
# that's when Page-level tracking began, unrelated to any individual's own history.
# Mckinley confirmed directly: everyone started this work in June 2026 with 0 followers,
# so this is a fixed, human-confirmed anchor the same way BASELINE_FOLLOWERS is for the
# Page, not something derived from the export's own ~365-day rolling window edge.
TEAM_BASELINE_DATE = "2026-06-01"
TEAM_BASELINE_FOLLOWERS = 0


def _is_personal_export(fname, name):
    lower = fname.lower()
    if lower.startswith("~$") or lower.startswith("."):
        return False
    if not lower.startswith("aggregateanalytics_"):
        return False
    if not lower.endswith((".xlsx", ".xls")):
        return False
    return name.lower() in lower


def _find_person_files(name, data_dir, upload_dir):
    candidates = []
    for folder in (data_dir, upload_dir):
        if not os.path.isdir(folder):
            continue
        for fname in os.listdir(folder):
            if _is_personal_export(fname, name):
                candidates.append(os.path.join(folder, fname))
    return sorted(set(candidates), key=os.path.getmtime)


def _parse_discovery(fp):
    raw = pd.read_excel(fp, sheet_name="DISCOVERY", header=None)
    out = {"date_range": raw.iloc[0, 1] if len(raw) else None}
    for i in range(1, len(raw)):
        key, val = raw.iloc[i, 0], raw.iloc[i, 1]
        if pd.notna(key):
            out[key] = int(val)
    return out


def _parse_engagement(fp):
    df = pd.read_excel(fp, sheet_name="ENGAGEMENT")
    df["Date"] = pd.to_datetime(df["Date"])
    df["Impressions"] = pd.to_numeric(df["Impressions"])
    df["Engagements"] = pd.to_numeric(df["Engagements"])
    return df


def _parse_followers(fp):
    raw = pd.read_excel(fp, sheet_name="FOLLOWERS", header=None)
    total = int(raw.iloc[0, 1])
    daily = pd.read_excel(fp, sheet_name="FOLLOWERS", header=2)
    daily["Date"] = pd.to_datetime(daily["Date"])
    daily["New followers"] = pd.to_numeric(daily["New followers"])
    return total, daily


_POST_TYPE_RE = re.compile(r"-(share|ugcpost)-\d+-", re.I)


def _post_is_repost(url):
    """LinkedIn's own URL convention: .../posts/<slug>-share-<id>-<code> is a repost
    of someone else's content, .../posts/<slug>-ugcPost-<id>-<code> is an original
    post - the only way to tell them apart, since "Engagements" doesn't split out
    which activity type a post is."""
    m = _POST_TYPE_RE.search(url or "")
    return m.group(1).lower() == "share" if m else None


def _parse_top_posts(fp):
    raw = pd.read_excel(fp, sheet_name="TOP POSTS", header=None)
    body = raw.iloc[3:]
    eng = body.iloc[:, [0, 1, 2]].dropna(how="all")
    eng.columns = ["Post URL", "Post Publish Date", "Engagements"]
    imp = body.iloc[:, [4, 5, 6]].dropna(how="all")
    imp.columns = ["Post URL", "Post Publish Date", "Impressions"]
    merged = pd.merge(eng, imp[["Post URL", "Impressions"]], on="Post URL", how="outer")
    if merged.empty:
        return merged
    merged["Post Publish Date"] = pd.to_datetime(merged["Post Publish Date"])
    merged["Engagements"] = pd.to_numeric(merged["Engagements"])
    merged["Impressions"] = pd.to_numeric(merged["Impressions"])
    merged["Is Repost"] = merged["Post URL"].apply(_post_is_repost)
    return merged.sort_values("Post Publish Date", ascending=False).reset_index(drop=True)


def _parse_demographics(fp, sheet_name):
    df = pd.read_excel(fp, sheet_name=sheet_name)
    df["Percentage"] = pd.to_numeric(df["Percentage"].astype(str).str.rstrip("%"))
    return df


def load_person_data(name, data_dir, upload_dir):
    files = _find_person_files(name, data_dir, upload_dir)
    if not files:
        return None

    eng_frames, fol_frames = [], []
    for fp in files:
        mtime = os.path.getmtime(fp)
        e = _parse_engagement(fp)
        e["_src"] = mtime
        eng_frames.append(e)
        _, daily = _parse_followers(fp)
        daily["_src"] = mtime
        fol_frames.append(daily)

    engagement = (pd.concat(eng_frames)
                    .sort_values("_src").drop_duplicates(subset=["Date"], keep="last")
                    .drop(columns=["_src"]).sort_values("Date").reset_index(drop=True))
    followers_daily = (pd.concat(fol_frames)
                         .sort_values("_src").drop_duplicates(subset=["Date"], keep="last")
                         .drop(columns=["_src"]).sort_values("Date").reset_index(drop=True))

    latest = files[-1]
    followers_total, _ = _parse_followers(latest)

    return {
        "discovery": _parse_discovery(latest),
        "engagement": engagement,
        "followers_daily": followers_daily,
        "followers_total": followers_total,
        "top_posts": _parse_top_posts(latest),
        "audience_demographics": _parse_demographics(latest, "AUDIENCE DEMOGRAPHICS"),
        "content_demographics": _parse_demographics(latest, "CONTENT DEMOGRAPHICS"),
        "source_files": [os.path.basename(fp) for fp in files],
    }


def _engagement_rate_stats(top_posts_df):
    if top_posts_df.empty:
        return None, None, 0
    rated = top_posts_df[top_posts_df["Impressions"] > 0].copy()
    if rated.empty:
        return None, None, 0
    rated["rate"] = rated["Engagements"] / rated["Impressions"] * 100
    return rated["rate"].max(), rated["rate"].mean(), len(rated)


def build_person_report_docx(name, data, data_dir, upload_dir):
    """A send-to-the-person summary of their own analyzed LinkedIn data - Mckinley's
    own review, not a raw dump of the export file. Reuses report._docx_table/_docx_shade
    (the same table styling as the Page-level monthly report) rather than duplicating
    that OOXML-shading logic here. Starting/New Followers use TEAM_BASELINE_DATE/
    TEAM_BASELINE_FOLLOWERS (everyone started this work in June 2026 at 0, confirmed
    directly), not the export's own ~365-day rolling window edge. data_dir/upload_dir
    locate any SinglePostAnalytics_*.xlsx files for the per-post detail table."""
    post_details = post_details_tracker.load_post_details(data_dir, upload_dir)
    discovery = data["discovery"]
    window_label = discovery.get("date_range", "")
    _, window_end = (window_label.split(" - ") + [""])[:2] if window_label else ("", "")
    baseline_ts = pd.Timestamp(TEAM_BASELINE_DATE)
    since_baseline = data["followers_daily"][data["followers_daily"]["Date"] >= baseline_ts]
    new_followers_window = int(since_baseline["New followers"].sum())
    starting_followers = TEAM_BASELINE_FOLLOWERS
    baseline_label = baseline_ts.strftime("%b %d, %Y")
    best_eng, avg_eng, rated_count = _engagement_rate_stats(data["top_posts"])

    doc = Document()
    zoom_el = doc.settings.element.find(qn("w:zoom"))
    if zoom_el is not None and zoom_el.get(qn("w:percent")) is None:
        zoom_el.set(qn("w:percent"), "100")

    doc.add_heading(f"{name} — LinkedIn Personal Profile Report", level=0)
    doc.add_paragraph(f"Window: {window_label}")

    doc.add_heading("Summary", level=1)
    report._docx_table(doc, [
        ["Metric", "Value", "Note"],
        ["Starting Followers", f"{starting_followers:,}", f"as of {baseline_label}"],
        ["Total Followers", f"{data['followers_total']:,}", f"as of {window_end}"],
        ["New Followers", f"{new_followers_window:,}", f"{baseline_label} – {window_end}"],
        ["Total Impressions", f"{discovery.get('Impressions', 0):,}", window_label],
        ["Members Reached", f"{discovery.get('Members reached', 0):,}", window_label],
        ["Best Post Real Eng.", f"{best_eng:.2f}%" if best_eng is not None else "—",
         f"across {rated_count} posts"],
        ["Avg. Eng. Rate", f"{avg_eng:.2f}%" if avg_eng is not None else "—", "raw, organic"],
    ])

    doc.add_heading("Posts This Window", level=1)
    posts = data["top_posts"]
    if posts.empty:
        doc.add_paragraph("No posts in this window.")
    else:
        rows = [["Date", "Type", "Impressions", "Engagements (export)", "Profile Viewers*",
                  "Followers Gained*", "Reactions*", "Comments*", "Reposts*", "Link"]]
        for _, r in posts.iterrows():
            post_type = "Repost" if r["Is Repost"] else ("Original" if r["Is Repost"] is False else "Unknown")
            d = post_details.get(r["Post URL"]) or {}
            rows.append([
                r["Post Publish Date"].strftime("%b %d, %Y"),
                post_type,
                int(r["Impressions"]) if pd.notna(r["Impressions"]) else "—",
                int(r["Engagements"]) if pd.notna(r["Engagements"]) else "—",
                d.get("profile_viewers", "—"),
                d.get("followers_gained", "—"),
                d.get("reactions", "—"),
                d.get("comments", "—"),
                d.get("reposts", "—"),
                r["Post URL"],
            ])
        report._docx_table(doc, rows, font_size=6)
        doc.add_paragraph("* from each post's own \"Export\" button (SinglePostAnalytics file) — not in the Aggregate Analytics export",
                           style="Caption")

    for heading, key in [("Audience Demographics (who sees it)", "audience_demographics"),
                          ("Content Demographics (who engages)", "content_demographics")]:
        doc.add_heading(heading, level=1)
        demo = data[key]
        if demo.empty:
            doc.add_paragraph("No data.")
            continue
        rows = [["Category", "Value", "%"]]
        rows += [[r["Top Demographics"], r["Value"], f"{r['Percentage']:.0f}%"] for _, r in demo.iterrows()]
        report._docx_table(doc, rows)

    buf = io.BytesIO()
    doc.save(buf)
    return buf.getvalue()