Spaces:
Runtime error
Runtime error
File size: 3,940 Bytes
985f3ee | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | import os
import sys
import json
import requests
from pathlib import Path
from dotenv import load_dotenv
load_dotenv()
BASE_DIR = Path(__file__).resolve().parent.parent
LINKEDIN_DIR = BASE_DIR / "data" / "linkedin"
POSTS_FILE = LINKEDIN_DIR / "posts.md"
APIFY_TOKEN = os.getenv("APIFY_API_TOKEN", "")
PROFILE_URL = os.getenv("LINKEDIN_PROFILE_URL", "https://www.linkedin.com/in/neuralarun/")
def fetch_linkedin_posts_apify():
if not APIFY_TOKEN or APIFY_TOKEN.startswith("your_"):
print("[ERROR] Missing APIFY_API_TOKEN in .env file.")
print("Please sign up at https://apify.com (free plan) and set APIFY_API_TOKEN in .env")
sys.exit(1)
print(f"Connecting to Apify LinkedIn Post Scraper for {PROFILE_URL}...")
# Using Apify's active LinkedIn Profile Posts Scraper actor
run_url = f"https://api.apify.com/v2/acts/harvestapi~linkedin-profile-posts/run-sync-get-dataset-items?token={APIFY_TOKEN}"
payload = {
"profileUrls": [PROFILE_URL],
"limit": 30
}
headers = {"Content-Type": "application/json"}
try:
response = requests.post(run_url, json=payload, headers=headers, timeout=120)
if response.status_code not in (200, 201):
print(f"[ERROR] Apify API returned error {response.status_code}: {response.text}")
sys.exit(1)
items = response.json()
print(f"Successfully scraped {len(items)} LinkedIn posts!")
return items
except Exception as e:
print(f"[ERROR] Failed to fetch LinkedIn posts via Apify: {e}")
sys.exit(1)
def save_posts_to_markdown(items):
LINKEDIN_DIR.mkdir(parents=True, exist_ok=True)
def get_item_date(item):
p = item.get("postedAt", {})
if isinstance(p, dict):
return p.get("date") or ""
return str(item.get("postedAtISO") or item.get("postedDate") or item.get("date") or "")
# Sort posts chronologically descending (newest post FIRST)
items.sort(key=get_item_date, reverse=True)
header = """---
type: linkedin_posts
source: harvestapi_linkedin_scraper
---
# Arun Yadav — LinkedIn Posts & Insights
This file contains public LinkedIn posts and technical insights written by Arun Yadav (@arun-yadav-768052368).
---
"""
post_blocks = []
for index, post in enumerate(items, start=1):
text = post.get("text") or post.get("content") or post.get("postText") or ""
posted_at_info = post.get("postedAt", {})
if isinstance(posted_at_info, dict):
date = posted_at_info.get("date", "N/A")[:10]
else:
date = str(post.get("postedAtISO") or post.get("postedDate") or post.get("date") or "N/A")[:10]
url = post.get("url") or post.get("postUrl") or post.get("link") or PROFILE_URL
if not text.strip():
continue
label = "MOST RECENT / LATEST LINKEDIN POST" if index == 1 else f"LinkedIn Post #{index}"
block = f"""---
## {label}
**Date:** {date}
**Post Title / Status:** {label}
> **Link:** [{url}]({url})
{text.strip()}
"""
post_blocks.append(block)
full_md = header + "\n\n".join(post_blocks)
with open(POSTS_FILE, "w", encoding="utf-8") as f:
f.write(full_md)
print(f"Updated {POSTS_FILE} with {len(post_blocks)} clean posts.")
def main():
print("=" * 60)
print("ArunCore LinkedIn Auto-Sync Pipeline")
print("=" * 60)
items = fetch_linkedin_posts_apify()
if items:
save_posts_to_markdown(items)
# Trigger auto-ingest into ChromaDB
print("\nTriggering vector ingestion into ChromaDB...")
import subprocess
subprocess.run([sys.executable, str(BASE_DIR / "scripts" / "ingest.py")], check=True)
print("\n[SUCCESS] LinkedIn posts synced and indexed in ChromaDB successfully!")
else:
print("[WARNING] No posts retrieved from Apify.")
if __name__ == "__main__":
main()
|