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')

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')