Atom feed for our EEG site
anil.recoil.org
1# /// script
2# requires-python = ">=3.11"
3# dependencies = [
4# "feedparser",
5# "beautifulsoup4",
6# "urllib3",
7# ]
8# ///
9# Do not delete the above as its needed for `uv run`
10#!/usr/bin/env python3
11
12import json
13import feedparser
14import sys
15import os
16from bs4 import BeautifulSoup
17import re
18from urllib.parse import urlparse, urljoin
19
20def extract_links_from_html(html_content, base_url=None):
21 """Extract and normalize links from HTML content"""
22 soup = BeautifulSoup(html_content, 'html.parser')
23 links = []
24
25 for a_tag in soup.find_all('a', href=True):
26 href = a_tag['href'].strip()
27
28 # Skip empty links, anchors, javascript, and mailto
29 if not href or href.startswith(('#', 'javascript:', 'mailto:')):
30 continue
31
32 # Convert relative URLs to absolute if we have a base URL
33 if base_url and not href.startswith(('http://', 'https://')):
34 href = urljoin(base_url, href)
35
36 links.append(href)
37
38 return links
39
40def normalize_url(url):
41 """Normalize URLs to consistently match them"""
42 if not url:
43 return ""
44
45 # Handle common URL shorteners or redirects (not implemented)
46
47 # Parse the URL
48 parsed = urlparse(url)
49
50 # Ensure scheme is consistent
51 scheme = parsed.scheme.lower() or 'http'
52
53 # Normalize netloc (lowercase, remove 'www.' prefix optionally)
54 netloc = parsed.netloc.lower()
55 if netloc.startswith('www.'):
56 netloc = netloc[4:]
57
58 # Remove trailing slashes and index.html/index.php
59 path = parsed.path.rstrip('/')
60 for index_file in ['/index.html', '/index.php', '/index.htm']:
61 if path.endswith(index_file):
62 path = path[:-len(index_file)]
63
64 # Remove common fragments and query parameters that don't affect content
65 # (like tracking params, utm_*, etc.)
66 query_parts = []
67 if parsed.query:
68 for param in parsed.query.split('&'):
69 if '=' in param:
70 key, value = param.split('=', 1)
71 if not key.startswith(('utm_', 'ref', 'source')):
72 query_parts.append(f"{key}={value}")
73
74 query = '&'.join(query_parts)
75
76 # Remove common hash fragments
77 fragment = ''
78
79 # Special case for common blogging platforms
80 # Medium, WordPress, Ghost, etc. may have specific URL patterns
81
82 # Reconstruct the URL
83 normalized = f"{scheme}://{netloc}{path}"
84 if query:
85 normalized += f"?{query}"
86 if fragment:
87 normalized += f"#{fragment}"
88
89 return normalized
90
91def analyze_feed():
92 # Parse the aggregated feed
93 print(f"Parsing eeg.xml...", file=sys.stderr)
94 feed_data = feedparser.parse("eeg.xml")
95
96 # Add debug info about the feed
97 print(f"Feed title: {feed_data.feed.get('title', 'Unknown')}", file=sys.stderr)
98 print(f"Feed version: {feed_data.get('version', 'Unknown')}", file=sys.stderr)
99
100 if not feed_data or not hasattr(feed_data, 'entries'):
101 print("Error: Could not parse feed or no entries found", file=sys.stderr)
102 return
103
104 print(f"Found {len(feed_data.entries)} entries in the aggregated feed", file=sys.stderr)
105
106 all_entries = []
107 entry_urls = {} # Maps normalized URLs to entry data
108
109 # First pass: collect all entries and their URLs
110 for entry in feed_data.entries:
111 # Get link
112 link = entry.get('link', '')
113 if not link:
114 continue
115
116 # Normalize the entry URL to help with matching
117 normalized_link = normalize_url(link)
118
119 # Get feed title (stored as category in the aggregated feed)
120 feed_title = "Unknown"
121 if hasattr(entry, 'tags') and entry.tags:
122 feed_title = entry.tags[0].term
123
124 # Get description/content
125 if hasattr(entry, 'content') and entry.content:
126 content = entry.content[0].value
127 else:
128 content = entry.get('summary', '')
129
130 # Extract all links from content, using the entry link as base URL for resolving relative URLs
131 content_links = extract_links_from_html(content, base_url=link)
132
133 # Get unique ID
134 entry_id = entry.get('id', link)
135
136 entry_data = {
137 'title': entry.get('title', 'No title'),
138 'link': link,
139 'normalized_link': normalized_link,
140 'feed_title': feed_title,
141 'id': entry_id,
142 'content_links': content_links,
143 'references': [], # Will be filled in the second pass
144 'referenced_by': [] # Will be filled in the second pass
145 }
146
147 all_entries.append(entry_data)
148 entry_urls[normalized_link] = entry_data
149
150 print(f"Total entries processed: {len(all_entries)}", file=sys.stderr)
151
152 # Second pass: analyze links between entries
153 for entry in all_entries:
154 # Keep track of references to avoid duplicates
155 reference_ids = set()
156 normalized_content_links = [normalize_url(link) for link in entry['content_links']]
157
158 for normalized_link in normalized_content_links:
159 if normalized_link in entry_urls and normalized_link != entry['normalized_link']:
160 # This entry links to another entry
161 referenced_entry = entry_urls[normalized_link]
162
163 # Avoid duplicate references
164 if referenced_entry['id'] in reference_ids:
165 continue
166
167 reference_ids.add(referenced_entry['id'])
168
169 # Add to the references of the current entry
170 entry['references'].append({
171 'id': referenced_entry['id'],
172 'link': referenced_entry['link'],
173 'title': referenced_entry['title'],
174 'feed_title': referenced_entry['feed_title']
175 })
176
177 # Add to the referenced_by of the referenced entry
178 # Check if this entry is already in referenced_by
179 already_referenced = any(ref['id'] == entry['id'] for ref in referenced_entry['referenced_by'])
180 if not already_referenced:
181 referenced_entry['referenced_by'].append({
182 'id': entry['id'],
183 'link': entry['link'],
184 'title': entry['title'],
185 'feed_title': entry['feed_title']
186 })
187
188 # Create the thread data structure
189 thread_data = {}
190 for entry in all_entries:
191 thread_data[entry['id']] = {
192 'id': entry['id'],
193 'title': entry['title'],
194 'link': entry['link'],
195 'feed_title': entry['feed_title'],
196 'references': entry['references'],
197 'referenced_by': entry['referenced_by']
198 }
199
200 # Write the thread data to a JSON file
201 with open('threads.json', 'w') as f:
202 json.dump(thread_data, f, indent=2)
203
204 print(f"Thread data successfully written to threads.json", file=sys.stderr)
205
206 # Generate some statistics
207 entries_with_references = sum(1 for entry in all_entries if entry['references'])
208 entries_with_referenced_by = sum(1 for entry in all_entries if entry['referenced_by'])
209
210 print(f"\nThread Analysis:", file=sys.stderr)
211 print(f"Total entries: {len(all_entries)}", file=sys.stderr)
212 print(f"Entries that reference other entries: {entries_with_references}", file=sys.stderr)
213 print(f"Entries referenced by other entries: {entries_with_referenced_by}", file=sys.stderr)
214
215if __name__ == "__main__":
216 analyze_feed()