beantownbabeandClaude Sonnet 5
0335e17e89
LinkedIn's exports have no title/text field for a post anywhere (neither Aggregate Analytics nor SinglePostAnalytics) - only the URL, whose slug is a URL-safe truncation of the post's own opening words. post_title_from_url() decodes it (handles accented names like Krutošíková), strips the trailing -share-/-ugcPost-<id>-<code> suffix, and takes the first 5 words as a readable stand-in for a title. Added to both the live dashboard's posts table and the downloadable report. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
279 lines
12 KiB
Python
279 lines
12 KiB
Python
"""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
|
||
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/<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
|
||
|
||
|
||
_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()
|