fetch_posts.py

@@ -185,10 +185,39 @@ class PostFetcher:
 			outbox_url = profile['outbox']
 
 		async with self._http.get(outbox_url) as resp: outbox = await resp.json()
-		assert outbox['type'] in {'OrderedCollection', 'OrderedCollectionPage'}
+		assert outbox['type'] == 'OrderedCollection'
 		return outbox
 
+	async def _finger_actor(self, username, instance):
+		# despite HTTP being a direct violation of the WebFinger spec, assume e.g. Tor instances do not support
+		# HTTPS-over-onion
+		finger_url = f'http://{instance}/.well-known/webfinger?resource=acct:{username}@{instance}'
+		async with self._http.get(finger_url) as resp: finger_result = await resp.json()
+		return (profile_url := self._parse_webfinger_result(username, instance, finger_result))
+
+	def _parse_webfinger_result(self, username, instance, finger_result):
+		"""given webfinger data, return profile URL for handle"""
+		def check_content_type(type, ct): return ct == type or ct.startswith(type+';')
+		check_ap = partial(check_content_type, ACTIVITYPUB_CONTENT_TYPE)
+
+		try:
+			# note: the server might decide to return multiple links
+			# so we need to decide how to prefer one.
+			# i'd put "and yarl.URL(template).host == instance" here,
+			# but some instances have no subdomain for the handle yet use a subdomain for the canonical URL.
+			# Additionally, an instance could theoretically serve profile pages over I2P and the clearnet,
+			# for example.
+			return (profile_url := next(
+				link['href']
+				for link in finger_result['links']
+				if link['rel'] == 'self' and check_ap(link['type'])
+			))
+		except StopIteration:
+			# this should never happen either
+			raise RuntimeError(f'fatal: while fingering {username}@{instance}, failed to find a profile URL')
+
 async def amain():
+	import json5 as json
 	import third_party.utils as utils
 	args = utils.arg_parser_factory(description='Fetch posts from all followed accounts').parse_args()
 	config = utils.load_config(args.cfg)