A python script for adding MVP blog activities from blog sitemap

activity.py

from xml.etree import ElementTree as ET
import datetime
import requests
import time
import sys
from pathlib import Path

# ============ Configuration ============
# RSS feed URL (default: atbug.com)
rss_url = 'https://atbug.com/index.xml'
# User profile ID (numeric, find via GET /api/Activities/{id} on any existing activity)
userProfileId = 303228
# Bearer token:
#   优先从 /tmp/mvp_token.txt 读取（由 mvp_get_token.py 自动获取）
#   如果文件不存在，则使用下方手动填写的 token
_token_file = Path("/tmp/mvp_token.txt")
if _token_file.exists():
    token = _token_file.read_text().strip()
    print(f"🔑 Token loaded from {_token_file}")
else:
    token = 'Bearer [TOKEN]'  # 手动填写（fallback）
    print("⚠️  Token file not found, using hardcoded token")
# Only submit articles published after this date (YYYY-MM-DD)
lastRenewalDate = "2026-03-10"
# Must match an existing MVP Technology Focus Area exactly
technologyFocusArea = "Azure Application PaaS"
# =======================================

# Fetch and parse RSS feed
print(f"Fetching RSS feed from {rss_url} ...")
resp = requests.get(rss_url, timeout=30)
resp.raise_for_status()
root = ET.fromstring(resp.content)

# Define the target date for comparison
target_date = datetime.datetime.strptime(lastRenewalDate, '%Y-%m-%d').replace(
    tzinfo=datetime.timezone(datetime.timedelta(hours=8))
)

# Extract articles published after the renewal date
extracted_articles = []
for item in root.findall('.//item'):
    title = item.find('title').text if item.find('title') is not None else 'No Title'
    link = item.find('link').text if item.find('link') is not None else 'No Link'
    description = item.find('description').text if item.find('description') is not None else 'No Description'
    pub_date_str = item.find('pubDate').text if item.find('pubDate') is not None else ''

    try:
        pub_date = datetime.datetime.strptime(pub_date_str, '%a, %d %b %Y %H:%M:%S %z')
    except ValueError:
        continue

    if pub_date > target_date:
        # Format date for API
        api_date = pub_date.strftime('%Y-%m-%dT%H:%M:%S.000Z')
        extracted_articles.append({
            'title': title,
            'link': link,
            'date': api_date,
            'description': description
        })

print(f'Found {len(extracted_articles)} articles published after {lastRenewalDate}')

if not extracted_articles:
    print("Nothing to submit.")
    sys.exit(0)

# API setup
url = 'https://mavenapi-prod.azurewebsites.net/api/Activities/'
headers = {
    'Content-Type': 'application/json',
    'Accept': '*/*',
    'Authorization': token,
    'Origin': 'https://mvp.microsoft.com',
    'Referer': 'https://mvp.microsoft.com/',
}

data = {
    "activity": {
        "id": 0,
        "activityTypeName": "Blog",
        "typeName": "Blog",
        "isPrivate": False,
        "targetAudience": ["Developer"],
        "tenant": "MVP",
        "userProfileId": userProfileId,
        "reach": 200,
        "quantity": 1,
        "role": "Author",
        "technologyFocusArea": technologyFocusArea,
        "additionalTechnologyAreas": [],
        "imageUrl": ""
    }
}

success, fail = 0, 0
for i, article in enumerate(extracted_articles):
    data['activity']['date'] = article['date']
    data['activity']['description'] = article['description'][:1000]
    data['activity']['privateDescription'] = article['description'][:200]
    data['activity']['title'] = article['title']
    data['activity']['url'] = article['link']

    response = requests.post(url, headers=headers, json=data)
    if response.status_code in (200, 201):
        success += 1
        print(f"[{i+1}/{len(extracted_articles)}] ✅ {article['title']}")
    else:
        fail += 1
        print(f"[{i+1}/{len(extracted_articles)}] ❌ {response.status_code} {article['title']}")
        print(f"   {response.text[:200]}")
    time.sleep(1)

print(f"\nDone: {success} succeeded, {fail} failed")

get_activities.py

import requests
import json
import sys

# ============ Configuration ============
userProfileId = 303228
token = 'Bearer [TOKEN]'
# Filter by technologyFocusArea (set to None to get all)
filter_area = None  # e.g. "Azure Application PaaS"
# =======================================

headers = {
    'authorization': token,
    'accept': '*/*',
    'origin': 'https://mvp.microsoft.com',
    'referer': 'https://mvp.microsoft.com/',
}

# Usage: python3 get_activities.py <start_id> <end_id>
# start_id/end_id: activity IDs from your last submission
start_id = int(sys.argv[1]) if len(sys.argv) > 1 else 373674
end_id = int(sys.argv[2]) if len(sys.argv) > 2 else start_id + 50

activities = []
for aid in range(start_id, end_id + 1):
    r = requests.get(f'https://mavenapi-prod.azurewebsites.net/api/Activities/{aid}', headers=headers)
    if r.status_code == 200:
        a = r.json()
        if filter_area is None or a.get('technologyFocusArea') == filter_area:
            activities.append(a)
            print(f'[{aid}] {a.get("technologyFocusArea","")} | {a.get("title","")[:60]}')
    elif r.status_code == 401:
        print(f'[{aid}] 401 Unauthorized - token expired')
        break
    else:
        pass  # skip non-owned IDs silently

print(f'\nTotal: {len(activities)} activities')

# Save to file for use with put_activities.py
with open('/tmp/mvp_activities.json', 'w') as f:
    json.dump(activities, f, ensure_ascii=False, indent=2)
print('Saved to /tmp/mvp_activities.json')

list_activities.py

#!/usr/bin/env python3
"""
MVP Activity List (Paginated, sorted by date descending)
Usage: python3 list_activities.py [page] [pageSize]
"""

import requests
import json
import sys
from pathlib import Path

# ============ Configuration ============
userProfileIdentifier = "053fdfb5-ff3e-ed11-bba3-000d3a197333"
_token_file = Path("/tmp/mvp_token.txt")
if _token_file.exists():
    token = _token_file.read_text().strip()
    print(f"🔑 Token loaded from {_token_file}")
else:
    token = "Bearer [TOKEN]"
    print("⚠️  Token file not found, using hardcoded token")
# =======================================

page = int(sys.argv[1]) if len(sys.argv) > 1 else 1
page_size = int(sys.argv[2]) if len(sys.argv) > 2 else 18

headers = {
    "accept": "*/*",
    "authorization": token,
    "content-type": "application/json",
    "origin": "https://mvp.microsoft.com",
    "referer": "https://mvp.microsoft.com/",
}

payload = {
    "pageIndex": page,
    "pageSize": page_size,
    "searchKey": "",
    "tenant": "MVP",
    "userProfileIdentifier": userProfileIdentifier,
    "contributionTargetAudience": [],
    "technologyFocusArea": [],
    "type": [],
    "sortColumn": "date",
    "sortDescending": True,
}

resp = requests.post(
    "https://mavenapi-prod.azurewebsites.net/api/Contributions/CommunityLeaderActivities/search",
    headers=headers,
    json=payload,
)

if resp.status_code != 200:
    print(f"❌ HTTP {resp.status_code}: {resp.text[:300]}")
    sys.exit(1)

data = resp.json()
items = data.get("items") or data.get("value") or (data if isinstance(data, list) else [])
total = data.get("totalCount") or data.get("count") or len(items)

print(f"\n📋 Activities (page {page}, size {page_size}, total {total}):\n")
for i, a in enumerate(items, 1):
    date = (a.get("date") or "")[:10]
    title = a.get("title") or a.get("name") or "(no title)"
    aid = a.get("id") or ""
    url = a.get("url") or ""
    atype = a.get("activityTypeName") or a.get("typeName") or ""
    print(f"  [{i:>2}] {date}  {atype:<35} {title[:50]}")
    if url:
        print(f"        {url}")

print(f"\nTotal shown: {len(items)} / {total}")

# Save raw response
out = "/tmp/mvp_list.json"
with open(out, "w") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)
print(f"💾 Raw response saved to {out}")

mvp_get_token.py

#!/usr/bin/env python3
"""
MVP Portal Token Grabber
使用 Playwright 打开浏览器，手动登录后自动捕获 Bearer token
保存到 /tmp/mvp_token.txt

依赖安装:
    pip install playwright
    playwright install chromium

注意：如果已安装 mermaid2img.py 的依赖，可跳过上述步骤。
"""

import sys
import time
from pathlib import Path
from playwright.sync_api import sync_playwright

TOKEN_OUTPUT = Path("/tmp/mvp_token.txt")
MVP_BASE = "mvp.microsoft.com"


def get_mvp_token() -> str | None:
    token = {"value": None}

    def handle_request(request):
        if token["value"]:
            return
        auth = request.headers.get("authorization", "")
        if auth.startswith("Bearer ") and MVP_BASE in request.url:
            token["value"] = auth
            print(f"✅ Token captured from: {request.url[:80]}")

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False, args=["--start-maximized"])
        context = browser.new_context(no_viewport=True)
        page = context.new_page()
        page.on("request", handle_request)

        print("🌐 Opening MVP Portal...")
        page.goto(f"https://{MVP_BASE}/en-US/account")

        print("👤 Please log in manually (including MFA if needed)...")
        print("   Waiting up to 3 minutes...")

        # 等待登录完成（URL 包含 /MVP 或 /account 且已通过认证）
        deadline = time.time() + 180
        while time.time() < deadline:
            try:
                url = page.url
                if token["value"]:
                    break
                # 触发一个 API 请求来捕获 token
                if "mvp.microsoft.com" in url and "login" not in url and "oauth" not in url:
                    try:
                        # 触发 API 调用
                        page.evaluate("""
                            fetch('https://mvp.microsoft.com/api/contributions?$top=1', {
                                credentials: 'include'
                            })
                        """)
                    except Exception:
                        pass
            except Exception:
                pass
            time.sleep(2)

        browser.close()

    if token["value"]:
        TOKEN_OUTPUT.write_text(token["value"])
        print(f"💾 Token saved to {TOKEN_OUTPUT}")
        print(f"   Preview: {token['value'][:40]}...")
        return token["value"]
    else:
        print("❌ Token not captured. Did you log in successfully?")
        return None


if __name__ == "__main__":
    result = get_mvp_token()
    sys.exit(0 if result else 1)

put_activities.py

import requests
import json
import time

# ============ Configuration ============
token = 'Bearer [TOKEN]'
# Field to update and new value
update_field = 'technologyFocusArea'
update_value = 'Azure Application PaaS'
# Input file from get_activities.py
input_file = '/tmp/mvp_activities.json'
# =======================================

headers = {
    'authorization': token,
    'content-type': 'application/json',
    'accept': '*/*',
    'origin': 'https://mvp.microsoft.com',
    'referer': 'https://mvp.microsoft.com/',
}

with open(input_file) as f:
    activities = json.load(f)

print(f'Updating {len(activities)} activities: {update_field} = "{update_value}"\n')

success, skip, fail = 0, 0, 0
for a in activities:
    if a.get(update_field) == update_value:
        print(f'[{a["id"]}] skipped (already set): {a.get("title","")[:50]}')
        skip += 1
        continue

    a[update_field] = update_value
    r = requests.put(
        'https://mavenapi-prod.azurewebsites.net/api/Activities/',
        headers=headers,
        json={'activity': a}
    )
    if r.status_code in (200, 201):
        success += 1
        print(f'[{a["id"]}] OK {a.get("title","")[:50]}')
    else:
        fail += 1
        print(f'[{a["id"]}] FAIL {r.status_code} {a.get("title","")[:50]}')
        print(f'   {r.text[:200]}')
    time.sleep(0.5)

print(f'\nDone: {success} updated, {skip} skipped, {fail} failed')

update_ai_activities.py

#!/usr/bin/env python3
"""
MVP Portal - Batch update AI-related activities technologyFocusArea
Usage: python3 update_ai_activities.py [start_date] [end_date]
  Dates format: YYYY-MM-DD (default: 2025-04-01 ~ 2026-03-31)

Token: save Bearer token to /tmp/mvp_token.txt
  echo "Bearer eyJ..." > /tmp/mvp_token.txt
"""

import requests, json, time, sys
from pathlib import Path
from datetime import datetime, timezone

# ============ Configuration ============
# Your MVP profile identifier (find in any search API response)
userProfileIdentifier = "[YOUR_PROFILE_IDENTIFIER]"
# Token: loaded from file (use mvp_get_token.py to auto-capture)
_token_file = Path("/tmp/mvp_token.txt")
if _token_file.exists():
    token = _token_file.read_text().strip()
    print(f"🔑 Token loaded from {_token_file}")
else:
    token = "Bearer [TOKEN]"
    print("⚠️  Token file not found, using hardcoded token")
# Date range to filter activities
start_date = sys.argv[1] if len(sys.argv) > 1 else "2025-04-01"
end_date   = sys.argv[2] if len(sys.argv) > 2 else "2026-03-31"
# Target field values
TARGET_FOCUS_AREA = "Microsoft Foundry"
TARGET_ADDITIONAL = ["Developer Tools"]
# Keywords to identify AI-related activities
AI_KEYWORDS = ["ai", "copilot", "mcp", "llm", "gpt", "openai", "azure openai",
               "agent", "machine learning", "ml", "spring ai", "litellm", "opentelemetry"]
# =======================================

headers = {
    "accept": "*/*",
    "authorization": token,
    "content-type": "application/json",
    "origin": "https://mvp.microsoft.com",
    "referer": "https://mvp.microsoft.com/",
}

# Fetch all activities (paginated)
all_items = []
page = 1
while True:
    resp = requests.post(
        "https://mavenapi-prod.azurewebsites.net/api/Contributions/CommunityLeaderActivities/search",
        headers=headers,
        json={"pageIndex": page, "pageSize": 50, "searchKey": "", "tenant": "MVP",
              "userProfileIdentifier": userProfileIdentifier,
              "contributionTargetAudience": [], "technologyFocusArea": [], "type": [],
              "sortColumn": "date", "sortDescending": True}
    )
    if resp.status_code == 401:
        print("❌ 401 Unauthorized - token expired, please refresh")
        sys.exit(1)
    data = resp.json()
    items = data.get("communityLeaderActivities") or data.get("items") or data.get("value") or []
    if not items:
        break
    all_items.extend(items)
    total = data.get("filteredCount") or data.get("totalCount") or data.get("count") or 0
    print(f"Page {page}: fetched {len(items)}, total so far {len(all_items)}/{total}")
    if len(all_items) >= total:
        break
    page += 1

# Filter by date range
start = datetime.strptime(start_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
end   = datetime.strptime(end_date, "%Y-%m-%d").replace(hour=23, minute=59, second=59, tzinfo=timezone.utc)

in_range = []
for a in all_items:
    d = a.get("date", "")[:10]
    if not d:
        continue
    try:
        dt = datetime.strptime(d, "%Y-%m-%d").replace(tzinfo=timezone.utc)
    except:
        continue
    if start <= dt <= end:
        in_range.append(a)

print(f"\nIn date range ({start_date} ~ {end_date}): {len(in_range)} activities")

# Identify AI-related
ai_related = []
for a in in_range:
    text = " ".join([
        a.get("title",""), a.get("description",""), a.get("url",""),
        a.get("activityTypeName",""), a.get("technologyFocusArea",""),
    ]).lower()
    if any(kw in text for kw in AI_KEYWORDS):
        ai_related.append(a)

print(f"AI-related: {len(ai_related)}")
for a in ai_related:
    print(f"  [{a.get('id')}] {a.get('date','')[:10]}  {a.get('title','')[:60]}")

print("\n--- Updating AI-related activities ---\n")
success, skip, fail = 0, 0, 0
for a in ai_related:
    aid = a["id"]
    # GET full activity details first (search returns partial data)
    gr = requests.get(f"https://mavenapi-prod.azurewebsites.net/api/Activities/{aid}", headers=headers)
    if gr.status_code != 200:
        print(f"  [{aid}] ❌ GET failed HTTP {gr.status_code}: {a.get('title','')[:50]}")
        fail += 1
        time.sleep(0.3)
        continue
    full = gr.json()

    if full.get("technologyFocusArea") == TARGET_FOCUS_AREA and full.get("additionalTechnologyAreas") == TARGET_ADDITIONAL:
        print(f"  [{aid}] skip (already set): {full.get('title','')[:50]}")
        skip += 1
        continue

    full["technologyFocusArea"] = TARGET_FOCUS_AREA
    full["additionalTechnologyAreas"] = TARGET_ADDITIONAL

    r = requests.put(
        "https://mavenapi-prod.azurewebsites.net/api/Activities/",
        headers=headers,
        json={"activity": full}
    )
    if r.status_code in (200, 201):
        success += 1
        print(f"  [{aid}] ✅ {full.get('title','')[:50]}")
    else:
        fail += 1
        print(f"  [{aid}] ❌ HTTP {r.status_code}: {full.get('title','')[:50]}")
        print(f"        {r.text[:200]}")
    time.sleep(0.5)

print(f"\nDone: {success} updated, {skip} skipped, {fail} failed")