From 2327a88228d9055f40151a8c43f4357f90f58c00 Mon Sep 17 00:00:00 2001 From: luc Date: Mon, 31 Aug 2026 17:24:10 +0200 Subject: [PATCH] dataset-retrieval/retrieving-bluesky-datasets.ipynb: update --- .../retrieving-bluesky-datasets.ipynb | 308 ++++++++++-------- 1 file changed, 166 insertions(+), 142 deletions(-) diff --git a/dataset-retrieval/retrieving-bluesky-datasets.ipynb b/dataset-retrieval/retrieving-bluesky-datasets.ipynb index 9a31618..81d8918 100644 --- a/dataset-retrieval/retrieving-bluesky-datasets.ipynb +++ b/dataset-retrieval/retrieving-bluesky-datasets.ipynb @@ -98,14 +98,16 @@ "metadata": {}, "outputs": [], "source": [ + "import re, random, time\n", "import pandas as pd\n", "import numpy as np\n", - "import re\n", "from tqdm import tqdm\n", "from more_itertools import chunked\n", "\n", "# Remove URLs, hashtags, mentions, emojis and whitespaces\n", "def clean_text(text):\n", + " if not isinstance(text, str):\n", + " return np.nan\n", " text = re.sub(r'http\\S+|www\\S+|https\\S+', '', text, flags=re.MULTILINE)\n", " text = re.sub(r'#\\S+', '', text)\n", " text = re.sub(r'@\\S+', '', text)\n", @@ -113,110 +115,163 @@ " text = re.sub(r'\\s+', ' ', text).strip()\n", " return text\n", "\n", - "# Search for posts containing a specific keyword\n", - "def get_posts_by_keyword(keyword, since, until, limit=300):\n", + "# Retry helper\n", + "def call_with_retry(fn, *args, max_retries=5, base_delay=2, **kwargs):\n", + " last_exc = None\n", + " for attempt in range(1, max_retries + 1):\n", + " try:\n", + " return fn(*args, **kwargs)\n", + " except Exception as e:\n", + " last_exc = e\n", + " wait = base_delay * (2 ** (attempt - 1)) + random.uniform(0, 1)\n", + " print(f' [retry {attempt}/{max_retries}] {type(e).__name__}: {e} - retrying in {wait:.1f}s')\n", + " time.sleep(wait)\n", + " raise last_exc\n", + "\n", + "# Post field extraction\n", + "def extract_post(post):\n", " try:\n", - " num_posts = 0\n", - " post_data = []\n", - "\n", - " progress_bar = tqdm(total=limit, desc=\"Fetching posts\")\n", - "\n", - " while num_posts < limit:\n", - " \n", - " search_results = client.app.bsky.feed.search_posts({'q': keyword, 'since': since, 'until': until, 'lang': 'en', 'limit': 100})\n", - " posts = search_results.posts\n", - "\n", - " for post in posts:\n", - " # Extract post details\n", - " text = clean_text(post.record.text) if hasattr(post.record, 'text') else np.nan\n", - " iso_date = post.record.created_at if hasattr(post.record, 'created_at') else np.nan\n", - " user = post.author.handle if hasattr(post.author, 'handle') else np.nan\n", - "\n", - " post_id = post.uri if hasattr(post, 'uri') else np.nan\n", - " parent_id = post.record.reply.parent.uri if hasattr(post.record, 'reply') and hasattr(post.record.reply, 'parent') else 'no_parent'\n", - "\n", - " # Format date-time\n", - " if iso_date != np.nan:\n", - " date_obj = datetime.fromisoformat(iso_date.replace('Z', '+00:00'))\n", - " date_str = date_obj.strftime('%Y-%m-%d %H:%M:%S')\n", - " else:\n", - " date_str = np.nan\n", - "\n", - " if post_id != np.nan and user != np.nan:\n", - " post_data.append({\n", - " 'date': date_str,\n", - " 'text': text,\n", - " 'user': user,\n", - " 'id': post_id,\n", - " 'parentid': parent_id\n", - " })\n", - " else:\n", - " continue\n", - " \n", - " until = iso_date\n", - " num_posts += len(posts)\n", - " progress_bar.update(len(posts))\n", - " progress_bar.set_postfix_str(f'now at: {date_str}')\n", - "\n", - " if date_obj < (start_timestamp + timedelta(hours=1)):\n", - " break\n", - "\n", - " progress_bar.close()\n", - " df = pd.DataFrame(post_data)\n", - " return df\n", - "\n", + " text = clean_text(post.record.text) if hasattr(post.record, 'text') else None\n", + " iso_date = post.record.created_at if hasattr(post.record, 'created_at') else None\n", + " user = post.author.handle if hasattr(post.author, 'handle') else None\n", + " post_id = post.uri if hasattr(post, 'uri') else None\n", + " parent_id = (\n", + " post.record.reply.parent.uri\n", + " if hasattr(post.record, 'reply') and hasattr(post.record.reply, 'parent')\n", + " else 'no_parent'\n", + " )\n", + " \n", + " if iso_date is None or post_id is None or user is None:\n", + " return None\n", + " \n", + " date_obj = datetime.fromisoformat(iso_date.replace('Z', '+00:00'))\n", + " date_str = date_obj.strftime('%Y-%m-%d %H:%M:%S')\n", + " \n", + " return {\n", + " 'date': date_str,\n", + " 'text': text if text is not None else np.nan,\n", + " 'user': user,\n", + " 'id': post_id,\n", + " 'parentid': parent_id,\n", + " }, date_obj\n", " except Exception as e:\n", - " print(f\"Error fetching posts: {e}\")\n", + " print(f\" Skipping malformed post: {e}\")\n", + " return None\n", + "\n", + "# Cleanup\n", + "def _dedupe(rows):\n", + " if not rows:\n", " return pd.DataFrame()\n", + " df = pd.DataFrame(rows)\n", + " before = len(df)\n", + " df = df.drop_duplicates(subset='id')\n", + " df = df.dropna(subset=['text', 'id', 'user'])\n", + " df = df[df['text'].astype(str).str.strip() != '']\n", + " after = len(df)\n", + " if before != after:\n", + " print(f\"(Filtered {before - after} duplicate/empty rows out of {before}.)\")\n", + " return df.reset_index(drop=True)\n", "\n", - "# Obtain posts from uri list\n", - "def get_posts(uris):\n", - " try:\n", - " post_data = []\n", - "\n", - " progress_bar = tqdm(total=len(uris), desc=\"Fetching posts\")\n", - "\n", - " for uribit in list(chunked(uris,20)):\n", - "\n", - " search_results = client.app.bsky.feed.get_posts({'uris': uribit})\n", - " posts = search_results.posts\n", - "\n", - " for post in posts: \n", - " # Extract post details\n", - " text = clean_text(post.record.text) if hasattr(post.record, 'text') else np.nan\n", - " iso_date = post.record.created_at if hasattr(post.record, 'created_at') else np.nan\n", - " user = post.author.handle if hasattr(post.author, 'handle') else np.nan\n", - "\n", - " post_id = post.uri if hasattr(post, 'uri') else np.nan\n", - " parent_id = post.record.reply.parent.uri if hasattr(post.record, 'reply') and hasattr(post.record.reply, 'parent') else 'no_parent'\n", - "\n", - " # Format date-time\n", - " if iso_date != np.nan:\n", - " date_obj = datetime.fromisoformat(iso_date.replace('Z', '+00:00'))\n", - " date_str = date_obj.strftime('%Y-%m-%d %H:%M:%S')\n", - " else:\n", - " date_str = np.nan\n", - "\n", - " if post_id != np.nan and user != np.nan:\n", - " post_data.append({\n", - " 'date': date_str,\n", - " 'text': text,\n", - " 'user': user,\n", - " 'id': post_id,\n", - " 'parentid': parent_id\n", - " })\n", - " else:\n", - " continue\n", " \n", + "# Search for posts containing a specific keyword\n", + "def get_posts_by_keyword(keyword, since, until, limit=300, page_size=100, max_retries=5):\n", + "\n", + " state = _STATE.setdefault(keyword, {'until': until, 'rows': []})\n", + " \n", + " if state['until'] != until and state['rows']:\n", + " print(f\"Resuming '{keyword}' from until={state['until']}, {len(state['rows'])} posts already fetched this session.\")\n", + " \n", + " since_dt = datetime.fromisoformat(since.replace('Z', '+00:00'))\n", + " progress_bar = tqdm(total=limit, initial=min(len(state['rows']), limit), desc=\"Fetching posts\")\n", + "\n", + " try:\n", + " while len(state['rows']) < limit:\n", + " try:\n", + " search_results = call_with_retry(client.app.bsky.feed.search_posts, {'q': keyword, 'since': since, 'until': state['until'], 'lang': 'en', 'limit': page_size}, max_retries=max_retries)\n", + " except Exception as e:\n", + " print(f\"Giving up on this page after {max_retries} retries: {e}\")\n", + " print(\"Progress so far is kept — just call this again to continue.\")\n", + " break\n", + " \n", + " posts = search_results.posts\n", + " if not posts:\n", + " print(\"No more posts returned — reached the end of available results.\")\n", + " break\n", + " \n", + " oldest_date_obj = None\n", + " for post in posts:\n", + " result = extract_post(post)\n", + " if result is None:\n", + " continue\n", + " row, date_obj = result\n", + " state['rows'].append(row)\n", + " if oldest_date_obj is None or date_obj < oldest_date_obj:\n", + " oldest_date_obj = date_obj\n", + " \n", " progress_bar.update(len(posts))\n", " \n", + " if oldest_date_obj is not None:\n", + " state['until'] = oldest_date_obj.isoformat().replace('+00:00', 'Z')\n", + " progress_bar.set_postfix_str(f'now at: {state[\"until\"]}')\n", + " \n", + " if oldest_date_obj < since_dt + timedelta(hours=1):\n", + " print(\"Reached the `since` boundary.\")\n", + " break\n", + " \n", + " finally:\n", " progress_bar.close()\n", - " df = pd.DataFrame(post_data)\n", - " return df\n", + " \n", + " print(f\"{len(state['rows'])} posts fetched in total this session for '{keyword}'. Call again with the same checkpoint_name to keep going.\")\n", + " return _dedupe(state['rows'])\n", "\n", - " except Exception as e:\n", - " print(f\"Error fetching posts: {e}\")\n", - " return pd.DataFrame()" + "# Obtain posts from uri list\n", + "def get_posts(uris, chunk_size=20, max_retries=5):\n", + " state = _STATE.setdefault('_by_uri', {'done_uris': set(), 'rows': []})\n", + " \n", + " remaining = [u for u in uris if u not in state['done_uris']]\n", + " chunks = list(chunked(remaining, chunk_size))\n", + " \n", + " progress_bar = tqdm(total=len(uris), initial=len(state['done_uris']), desc=\"Fetching posts\")\n", + " \n", + " try:\n", + " for uribit in chunks:\n", + " try:\n", + " search_results = call_with_retry(client.app.bsky.feed.get_posts, {'uris': uribit}, max_retries=max_retries)\n", + " except Exception as e:\n", + " print(f\"Skipping this chunk of {len(uribit)} URIs after {max_retries} retries: {e}\")\n", + " continue\n", + " \n", + " for post in search_results.posts:\n", + " result = extract_post(post)\n", + " if result is None:\n", + " continue\n", + " row, _ = result\n", + " state['rows'].append(row)\n", + " \n", + " state['done_uris'].update(uribit)\n", + " progress_bar.update(len(uribit))\n", + " \n", + " finally:\n", + " progress_bar.close()\n", + " \n", + " print(f\"{len(state['done_uris'])}/{len(uris)} URIs processed in total.\")\n", + " return _dedupe(state['rows'])" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "Define state" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "_STATE = {}" ] }, { @@ -233,7 +288,7 @@ "metadata": {}, "outputs": [], "source": [ - "raw_df_data = get_posts_by_keyword('...', iso_start_timestamp, iso_end_timestamp, 500000)\n", + "raw_df_data = get_posts_by_keyword('...', iso_start_timestamp, iso_end_timestamp, 100000)\n", "raw_df_data.info()" ] }, @@ -251,13 +306,19 @@ "metadata": {}, "outputs": [], "source": [ + "_STATE = {}\n", "id_list = raw_df_data['id'].to_list()\n", "id_list.append('no_parent')\n", - "\n", - "absent_parents = raw_df_data.loc[~raw_df_data['parentid'].isin(id_list), 'parentid'].tolist()\n", - "\n", + "absent_parents = raw_df_data.loc[~raw_df_data['parentid'].isin(id_list), 'parentid'].tolist()" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ "df_absent_data = get_posts(absent_parents)\n", - "\n", "df_absent_data.info()" ] }, @@ -275,44 +336,7 @@ "metadata": {}, "outputs": [], "source": [ - "df_data = pd.concat(raw_df_data, df_absent_data, ignore_index=True)\n", - "\n", - "df_data.info()" - ] - }, - { - "attachments": {}, - "cell_type": "markdown", - "metadata": {}, - "source": [ - "Drop duplicates." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "df_data = df_data.drop_duplicates()\n", - "df_data.info()" - ] - }, - { - "attachments": {}, - "cell_type": "markdown", - "metadata": {}, - "source": [ - "Remove empty entries." - ] - }, - { - "cell_type": "code", - "execution_count": null, - "metadata": {}, - "outputs": [], - "source": [ - "df_data.dropna(inplace=True,subset=['text'])\n", + "df_data = pd.concat([raw_df_data, df_absent_data], ignore_index=True)\n", "df_data.info()" ] }, @@ -356,7 +380,7 @@ "metadata": {}, "outputs": [], "source": [ - "df_data.to_csv(f'../datasets/{case}.csv', index=False)" + "df_data.to_csv(f'{case}.csv', index=False)" ] }, { @@ -458,7 +482,7 @@ "metadata": {}, "outputs": [], "source": [ - "closed_df_reduced_data.to_csv(f'../datasets/{case}-social-closed.csv', index=False)" + "closed_df_reduced_data.to_csv(f'{case}-social-closed.csv', index=False)" ] }, { @@ -475,7 +499,7 @@ "metadata": {}, "outputs": [], "source": [ - "closed_df_data.to_csv(f'../datasets/{case}-closed.csv', index=False)" + "closed_df_data.to_csv(f'{case}-closed.csv', index=False)" ] } ],