nix

configuration files that power my machines
Log | Files | Refs | README | LICENSE

sync (2696B)

      1 #!/usr/bin/env python
      2 
      3 import json
      4 import re
      5 import subprocess
      6 import sys
      7 import xml.etree.ElementTree as ET
      8 from pathlib import Path
      9 from typing import cast
     10 from urllib.request import Request, urlopen
     11 from urllib.error import URLError, HTTPError
     12 
     13 FEEDS_FILE = Path('feeds')
     14 CACHE_FILE = Path('hack/titles.json')
     15 
     16 MINIFLUX_URL = subprocess.check_output(
     17     ['pass', 'show', 'miniflux-url'], text=True
     18 ).strip()
     19 MINIFLUX_TOKEN = subprocess.check_output(
     20     ['pass', 'show', 'miniflux-token'], text=True
     21 ).strip()
     22 
     23 cache: dict[str, str] = (
     24     json.loads(CACHE_FILE.read_text()) if CACHE_FILE.exists() else {}
     25 )
     26 
     27 
     28 def fetch_title(url: str) -> str:
     29     if url in cache:
     30         return cache[url]
     31     try:
     32         req = Request(url, headers={'User-Agent': 'Mozilla/5.0'})
     33         with urlopen(req, timeout=10) as response:
     34             content_type = (
     35                 response.headers.get_content_charset() or 'utf-8'
     36             )
     37             html = cast(bytes, response.read()).decode(
     38                 content_type, errors='ignore'
     39             )
     40             match = re.search(r'<title>([^<]+)</title>', html, re.IGNORECASE)
     41             if match:
     42                 cache[url] = match.group(1).strip()
     43                 return cache[url]
     44     except Exception:
     45         pass
     46     cache[url] = ''
     47     return ''
     48 
     49 
     50 root = ET.Element('opml', version='1.0')
     51 body = ET.SubElement(root, 'body')
     52 outline = body
     53 
     54 for line in FEEDS_FILE.read_text().splitlines():
     55     if not (line := line.strip()):
     56         continue
     57 
     58     if line.startswith('@'):
     59         name = line.split('<')[0].replace('@', '').strip()
     60         outline = ET.SubElement(body, 'outline', text=name, title=name)
     61     elif line.startswith('http'):
     62         url, _, title = line.partition(' ')
     63         url = url.strip()
     64         title = title.strip() or fetch_title(url)
     65 
     66         if outline == body:
     67             outline = ET.SubElement(
     68                 body, 'outline', text='Uncategorized', title='Uncategorized'
     69             )
     70 
     71         elem = ET.SubElement(outline, 'outline', type='rss', xmlUrl=url)
     72         if title:
     73             elem.set('text', title)
     74             elem.set('title', title)
     75 
     76 ET.indent(root, space='  ')
     77 opml_data = cast(bytes, ET.tostring(root, encoding='utf-8', xml_declaration=True))
     78 
     79 req = Request(f'{MINIFLUX_URL}/v1/import', data=opml_data, method='POST')
     80 req.add_header('X-Auth-Token', MINIFLUX_TOKEN)
     81 req.add_header('Content-Type', 'application/xml')
     82 
     83 try:
     84     urlopen(req)
     85     print('Done')
     86 except HTTPError as e:
     87     print(f'HTTP Error {e.code}', file=sys.stderr)
     88 except URLError as e:
     89     print(f'URL Error: {e.reason}', file=sys.stderr)
     90 finally:
     91     _ = CACHE_FILE.write_text(json.dumps(cache, indent=2, ensure_ascii=False))