personal-data-harvester

name	personal-data-harvester
description	Autonomously collect and structure a user's personal content history from Chinese and international platforms (豆瓣, 小红书, B站, 微信读书, 抖音, Kindle, etc.) using browser automation, local file parsing, and system share-sheet intake. Use this skill whenever the user wants to: build a personal knowledge base from their consumption history, sync reading/watching/collecting records into a local database, set up an automated pipeline to continuously harvest their data from social or reading platforms, or feed their content history into an AI agent for analysis. Trigger even if the user just says "帮我把豆瓣数据导进来", "同步我的读书记录", or "抓取我收藏的内容" — any phrase implying cross-platform personal data ingestion should use this skill.
compatibility	{"tools":["bash","computer"],"dependencies":["python3 (>=3.10)","playwright (pip install playwright && playwright install chromium)","sqlite3 (stdlib)","requests, beautifulsoup4, pydantic (pip install ...)"]}

Personal Data Harvester

Helps Claude autonomously build and maintain a local pipeline that collects a user's personal content history across platforms, stores it in a structured SQLite database, and keeps it fresh over time.

Core philosophy

• User's own data, user's own device. All collection happens under the user's authenticated session or from locally cached files. Never store credentials; reuse existing browser sessions.
• Graceful degradation. Each platform has a primary and fallback strategy. If automation breaks (platform redesign, rate-limit), fall back to file import without losing prior data.
• Privacy-first. Data stays local by default. No uploads unless the user explicitly requests it.

---

Step 0 — Assess the environment

Before writing any code, run this checklist:

python3 --version          # need >= 3.10
pip show playwright        # if missing: pip install playwright && playwright install chromium
pip show beautifulsoup4    # if missing: pip install beautifulsoup4 requests pydantic
ls ~/Library/Application\ Support/微信读书/ 2>/dev/null || echo "no local wechat-read cache"

Ask the user:

1. Which platforms to harvest first (prioritise by what they use most)?
2. Desktop or mobile primary device?
3. Acceptable harvest frequency (daily cron, manual trigger, or always-on daemon)?

Read references/platforms.md for per-platform technical details before writing any scraper.

---

Step 1 — Initialise the local database

Always create this schema first. All scrapers write to the same DB.

# scripts/init_db.py
import sqlite3, pathlib

DB = pathlib.Path.home() / ".personal-harvest" / "data.db"
DB.parent.mkdir(exist_ok=True)

with sqlite3.connect(DB) as con:
    con.executescript("""
    CREATE TABLE IF NOT EXISTS items (
        id          TEXT PRIMARY KEY,          -- platform:platform_id
        platform    TEXT NOT NULL,             -- douban | bilibili | xiaohongshu | wechatread | kindle
        type        TEXT NOT NULL,             -- book | video | note | article | post
        title       TEXT,
        url         TEXT,
        creator     TEXT,
        tags        TEXT,                      -- JSON array
        user_rating INTEGER,                   -- 1-5 or null
        user_status TEXT,                      -- want | doing | done | liked | saved
        user_note   TEXT,                      -- user's own annotation
        summary     TEXT,                      -- AI-generated or platform description
        collected_at TEXT,                     -- ISO8601
        harvested_at TEXT DEFAULT (datetime('now'))
    );
    CREATE INDEX IF NOT EXISTS idx_platform ON items(platform);
    CREATE INDEX IF NOT EXISTS idx_type     ON items(type);
    CREATE INDEX IF NOT EXISTS idx_status   ON items(user_status);
    """)
print(f"DB ready at {DB}")

---

Step 2 — Choose collection strategy per platform

Platform	Primary strategy	Fallback
豆瓣	Playwright browser automation (logged-in session)	HTML export + parse
小红书	Browser plugin DOM capture / Playwright	Manual share-sheet link intake
B 站	Playwright (favorites API endpoint visible in DevTools)	Official data export
微信读书	Local SQLite cache file	Playwright web version
抖音	iOS/Android share-sheet → link resolver	Playwright (harder, rate-limited)
Kindle	My Clippings.txt file parse	Goodreads export
豆瓣读书 App	Playwright web douban.com	Official "我的豆瓣"数据导出

See references/platforms.md for exact file paths, API endpoints, and selector patterns.

---

Step 3 — Implement scrapers

Pattern A — Playwright browser automation (豆瓣, B站, 小红书)

# scripts/scrape_douban.py
"""
Collect 豆瓣 读过/在读/想读 books and watched films.
Requires: user already logged in to douban.com in the system Chromium profile.
"""
import asyncio, json, sqlite3, pathlib
from datetime import datetime
from playwright.async_api import async_playwright

DB = pathlib.Path.home() / ".personal-harvest" / "data.db"
DOUBAN_USER_ID = "YOUR_USER_ID"  # ask the user to provide or auto-detect from profile page

STATUSES = {
    "wish":    ("want",  f"https://book.douban.com/people/{DOUBAN_USER_ID}/wish"),
    "do":      ("doing", f"https://book.douban.com/people/{DOUBAN_USER_ID}/do"),
    "collect": ("done",  f"https://book.douban.com/people/{DOUBAN_USER_ID}/collect"),
}

async def scrape():
    async with async_playwright() as pw:
        # Connect to user's existing browser profile to reuse login session
        browser = await pw.chromium.launch_persistent_context(
            user_data_dir=str(pathlib.Path.home() / ".config" / "personal-harvest-browser"),
            headless=False,   # show browser so user can log in first time
            slow_mo=800,      # human-like pacing to avoid rate limits
        )
        page = await browser.new_page()

        items = []
        for status_key, (status_label, url)  STATUSES.items():
            page_num = 
             :
                 page.goto(, wait_until=)
                 page.wait_for_timeout()

                cards =  page.query_selector_all()
                  cards:
                    

                 card  cards:
                    title_el =  card.query_selector()
                    title    =  title_el.inner_text()  title_el  
                    href     =  title_el.get_attribute()  title_el  
                    item_id  = href.split()[-].strip()  href  

                    rating_el =  card.query_selector()
                    rating_cls =  rating_el.get_attribute()  rating_el  
                    
                    rating = (((c[-])  c  rating_cls.split()  c.startswith()  (c) > ), )

                    note_el =  card.query_selector()
                    note    = ( note_el.inner_text()).strip()  note_el  

                    date_el =  card.query_selector()
                    date    = ( date_el.inner_text()).strip()  date_el  

                    items.append({
                        : ,
                        : ,
                        : ,
                        : title.strip(),
                        : href,
                        : status_label,
                        : rating,
                        : note,
                        : date,
                    })

                
                next_btn =  page.query_selector()
                  next_btn:
                    
                page_num += 

        
         sqlite3.connect(DB)  con:
            con.executemany(, items)
        ()
         browser.close()

asyncio.run(scrape())

Agent instruction: After writing the scraper, run it once in non-headless mode so the user can log in. Detect login success by checking for profile avatar element. Then re-run headless.

Pattern B — Local file parse (微信读书, Kindle)

# scripts/parse_wechatread.py
"""
Parse 微信读书 local SQLite cache on macOS.
Path: ~/Library/Containers/com.tencent.WeReadMac/Data/Library/Application Support/WeRead/
"""
import sqlite3, pathlib, json

WREAD_DB_GLOB = pathlib.Path.home().glob(
    "Library/Containers/com.tencent.WeReadMac/Data/Library/Application Support/WeRead/*.db"
)
HARVEST_DB = pathlib.Path.home() / ".personal-harvest" / "data.db"

def parse():
    items = []
    for src in WREAD_DB_GLOB:
        try:
            with sqlite3.connect(f"file:{src}?mode=ro", uri=True) as con:
                # Table names vary by version — discover them first
                tables = [r[0] for r in con.execute("SELECT name FROM sqlite_master WHERE type='table'")]
                print(f"Tables in {src.name}: {tables}")

                # Common table: ZBOOK or book_info
                book_table = next((t for t in tables if "book" in t.lower()), None)
                if not book_table:
                    continue

                cols = [r[1] for r in con.execute(f"PRAGMA table_info({book_table})")]
                print(f"Columns: {cols}")

                rows = con.execute().fetchall()
                 row  rows:
                    r = ((cols, row))
                    items.append({
                        : ,
                        : ,
                        : ,
                        : r.get(, r.get(, )),
                        : r.get(, r.get(, )),
                        :   r.get(, ) <   ,
                    })
         Exception  e:
            ()

     sqlite3.connect(HARVEST_DB)  con:
        con.executemany(, items)
    ()

parse()

# scripts/parse_kindle.py
"""Parse Kindle My Clippings.txt for highlights and notes."""
import re, sqlite3, pathlib

CLIPPINGS = pathlib.Path.home() / "Documents" / "My Clippings.txt"
HARVEST_DB = pathlib.Path.home() / ".personal-harvest" / "data.db"

SEPARATOR = "=========="

def parse():
    if not CLIPPINGS.exists():
        print(f"Not found: {CLIPPINGS}"); return

    text = CLIPPINGS.read_text(encoding="utf-8-sig", errors="replace")
    entries = text.split(SEPARATOR)
    items, notes = {}, []

    for entry in entries:
        lines = [l.strip() for l in entry.strip().splitlines() if l.strip()]
        if len(lines) < 3: continue
        title_author = lines[0]
        content = "\n".join(lines[2:])

        book_id = re.sub(r"[^a-z0-9]", "", title_author.lower())[:40]
        item_id = f"kindle:{book_id}"
        if item_id not in items:
            m = re.match(r"^(.+?)\s*[\(\（](.+?)[\)\）]$", title_author)
            items[item_id] = {
                "id": item_id, "platform": "kindle", "type": "book",
                : m.group().strip()  m  title_author,
                : m.group().strip()  m  ,
                : ,
            }
        notes.append({: item_id, : content})

     sqlite3.connect(HARVEST_DB)  con:
        con.executemany(, items.values())
    ()

parse()

---

Step 4 — Self-healing: detect and fix breakage

After each run, check for anomalies:

# scripts/health_check.py
import sqlite3, pathlib, json
from datetime import datetime, timedelta

DB = pathlib.Path.home() / ".personal-harvest" / "data.db"

with sqlite3.connect(DB) as con:
    report = {}
    for platform in ["douban", "bilibili", "wechatread", "kindle", "xiaohongshu"]:
        count = con.execute("SELECT COUNT(*) FROM items WHERE platform=?", (platform,)).fetchone()[0]
        last  = con.execute("SELECT MAX(harvested_at) FROM items WHERE platform=?", (platform,)).fetchone()[0]
        report[platform] = {"count": count, "last_harvested": last}

    total = con.execute("SELECT COUNT(*) FROM items").fetchone()[0]

print(json.dumps(report, indent=2, ensure_ascii=False))
print(f"\nTotal items: {total}")

# Flag platforms not updated in 48h
for platform, info in report.items():
    if info["last_harvested"]:
        last_dt = datetime.fromisoformat(info["last_harvested"])
        if datetime.now() - last_dt > timedelta(hours=48):
            print(f"⚠️  {platform} stale — last harvest {info['last_harvested']}")

When a scraper fails:

1. Run health_check.py to identify which platform is stale
2. Inspect the live page: await page.screenshot(path="debug.png") to see current DOM
3. Update selectors in the scraper based on the screenshot
4. Re-run and verify count increases

---

Step 5 — Set up continuous harvest (cron)

# Add to crontab: crontab -e
# Run all harvesters at 3am daily
0 3 * * * cd ~/.personal-harvest && python3 scripts/scrape_douban.py >> logs/douban.log 2>&1
0 3 * * * cd ~/.personal-harvest && python3 scripts/parse_wechatread.py >> logs/wechatread.log 2>&1
0 3 * * * cd ~/.personal-harvest && python3 scripts/health_check.py >> logs/health.log 2>&1

Or generate a launchd plist for macOS:

<!-- ~/Library/LaunchAgents/com.personal-harvest.plist -->
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0"><dict>
  <key>Label</key><string>com.personal-harvest</string>
  <key>ProgramArguments</key>
  <array>
    <string>/usr/bin/python3</string>
    <string>/Users/YOU/.personal-harvest/scripts/run_all.py</string>
  </array>
  <key>StartCalendarInterval</key>
  <dict><key>Hour</key><integer>3</integer><key>Minute</key><integer>0</>
  StandardOutPath/Users/YOU/.personal-harvest/logs/harvest.log

---

Step 6 — Expose data to agent

Once data is in SQLite, any downstream agent can query it:

# Example: feed today's harvest summary to an AI agent
import sqlite3, pathlib

DB = pathlib.Path.home() / ".personal-harvest" / "data.db"

def get_recent_items(days=7, limit=50):
    with sqlite3.connect(DB) as con:
        rows = con.execute("""
            SELECT platform, type, title, creator, user_status, user_note, collected_at
            FROM items
            WHERE harvested_at >= datetime('now', '-? days')
            ORDER BY harvested_at DESC
            LIMIT ?
        """, (days, limit)).fetchall()
    return [dict(zip(["platform","type","title","creator","status","note","date"], r)) for r in rows]

def get_interest_profile():
    """Return aggregated interest signals for agent context."""
    with sqlite3.connect(DB) as con:
        by_status = dict(con.execute(
            "SELECT user_status, COUNT(*) FROM items GROUP BY user_status"
        ).fetchall())
        top_creators = [r[0] for r in con.execute(
            "SELECT creator, COUNT(*) c FROM items WHERE creator IS NOT NULL GROUP BY creator ORDER BY c DESC LIMIT 20"
        ).fetchall()]
    return {"counts_by_status": by_status, "top_creators": top_creators}

---

Error patterns and fixes

Symptom	Likely cause	Fix
TimeoutError on page load	Platform slow / blocked	Increase wait_until timeout; add random delay
0 cards returned	Selector changed after redesign	Screenshot page, update selector
Login redirect loop	Session expired	Re-launch non-headless, let user log in
DB locked	Two scrapers running at once	Add timeout=30 to sqlite3.connect()
微信读书 DB not found	Different macOS version path	Use glob pattern, print all found paths
Kindle clippings empty	Wrong mount path	Ask user to locate My Clippings.txt manually

---

References

• references/platforms.md — Per-platform: exact selectors, API endpoints, local file paths, rate limits
• references/anti-detection.md — Techniques to avoid bot detection (user-agent rotation, timing jitter, viewport randomisation)