"""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___.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 from urllib.parse import unquote 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/-share-- is a repost of someone else's content, .../posts/-ugcPost-- 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 _POST_SLUG_RE = re.compile(r"_(.+?)-(?:share|ugcpost)-\d+-\w+$", re.I) def post_title_from_url(url, word_count=5): """A short, readable stand-in for an actual post title - LinkedIn's export has no title/text field anywhere (Aggregate Analytics or SinglePostAnalytics), only the URL, whose slug is a URL-safe truncation of the post's own opening words. Decodes %-escapes first (needed for names with accents, e.g. Krutošíková).""" if not url: return "" m = _POST_SLUG_RE.search(unquote(url)) if not m: return "" words = m.group(1).replace("-", " ").split() title = " ".join(words[:word_count]) return title[0].upper() + title[1:] if title else "" 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", "Description", "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_title_from_url(r["Post URL"]), 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()