feedfetcher/feedfetcher.py
File Type: text/x-script.python
#!/usr/bin/env python3
"""Fetch the feeds listed in a text file and render an HTML page.
Reads feed URLs (one per line; blank lines and # comments are skipped),
keeps the 5 most recent non-short entries from each feed, and writes an
HTML document. A line wrapped in === markers, such as ===Retro Tech===,
starts a category. Entries published in the last 7 days are highlighted.
Usage:
./feedfetcher.py [--feeds FILE] [--output FILE] [--copy DEST]
[--items N] [--new-interval DURATION]
[--cache FILE] [--from-cache] [--open]
With no arguments, reads feeds.txt and writes feeds.html under
$XDG_CACHE_HOME/feedfetcher (or ~/.cache/feedfetcher), next to the
fetched-feed cache. A --copy destination containing ":" is sent with
scp. --from-cache rebuilds the page from the cache without fetching.
"""
from __future__ import annotations
import argparse
import html
import json
import os
import re
import shutil
import subprocess
import sys
import webbrowser
import urllib.error
import urllib.request
import xml.etree.ElementTree as ET
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta, timezone
from email.utils import parsedate_to_datetime
from pathlib import Path
USER_AGENT = "feedfetcher/1.0 (+https://github.com/local/feedfetcher)"
RECENT_WINDOW = timedelta(days=7)
MAX_ITEMS = 5
REQUEST_TIMEOUT = 30
YOUTUBE_CHANNEL_RE = re.compile(
r"(?:youtube\.com|youtu\.be)/(?:feeds/videos\.xml\?channel_id=|channel/)(UC[\w-]{22})"
)
# YouTube's channel feed only carries the 15 newest uploads. Channels that
# post mostly Shorts can have fewer than 5 long videos in that window, so
# those feeds are read from the uploads playlist with Shorts removed.
YOUTUBE_LONGFORM_PLAYLIST = "https://www.youtube.com/feeds/videos.xml?playlist_id=UULF{}"
SHORTS_RE = re.compile(r"/shorts/", re.IGNORECASE)
CATEGORY_RE = re.compile(r"^===\s*(.+?)\s*===$")
def local_name(tag: str) -> str:
return tag.rsplit("}", 1)[-1]
def child_text(element: ET.Element, name: str) -> str:
"""Text of a direct child. Entry titles live inside each entry, so a
channel title has to come from the channel element itself."""
for child in element:
if local_name(child.tag) == name and child.text:
return child.text.strip()
return ""
def find_child(element: ET.Element, name: str) -> ET.Element | None:
for child in element:
if local_name(child.tag) == name:
return child
return None
def parse_date(value: str) -> datetime | None:
if not value:
return None
value = value.strip()
try:
parsed = parsedate_to_datetime(value)
except (TypeError, ValueError, IndexError):
parsed = None
if parsed is None:
try:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
except ValueError:
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc)
def fetch(url: str, accept: str | None = None) -> bytes:
headers = {"User-Agent": USER_AGENT}
if accept:
headers["Accept"] = accept
request = urllib.request.Request(url, headers=headers)
with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT) as response:
return response.read()
def atom_link(element: ET.Element, rel: str) -> str:
for child in element:
if local_name(child.tag) != "link":
continue
child_rel = child.get("rel") or "alternate"
if child_rel == rel and child.get("href"):
return child.get("href", "").strip()
return ""
def thumbnail_url(element: ET.Element) -> str:
"""Best image for an entry or channel, from common feed extensions."""
for child in element:
name = local_name(child.tag)
if name == "thumbnail" and child.get("url"):
return child.get("url", "").strip()
if name == "image":
href = child.get("href") or child_text(child, "url")
if href:
return href.strip()
if name == "group":
nested = thumbnail_url(child)
if nested:
return nested
if name in {"content", "enclosure"}:
url = child.get("url", "")
mime = child.get("type", "")
if url and (mime.startswith("image/") or re.search(r"\.(jpe?g|png|gif|webp)(\?|$)", url, re.I)):
return url.strip()
return ""
def entry_link(element: ET.Element, kind: str) -> str:
if kind == "channel":
link = child_text(element, "link")
if link:
return link
guid = find_child(element, "guid")
if guid is not None and guid.get("isPermaLink", "true").lower() != "false":
return (guid.text or "").strip()
return ""
return atom_link(element, "alternate") or atom_link(element, "self")
def entry_date(element: ET.Element, kind: str) -> str:
if kind == "channel":
return child_text(element, "pubDate") or child_text(element, "date")
return (
child_text(element, "published")
or child_text(element, "updated")
or child_text(element, "issued")
)
def feed_title(channel: ET.Element, kind: str) -> str:
"""The channel's own title, which is a direct child of the feed element.
A title nested inside an entry belongs to that entry, even when the
feed lists entries before its own title.
"""
for child in channel:
if local_name(child.tag) == "title" and child.text and child.text.strip():
return child.text.strip()
return ""
def item_matches(title: str, text_filter: str) -> bool:
"""A leading ! keeps entries that do not contain the text."""
if text_filter.startswith("!"):
return text_filter[1:].casefold() not in title.casefold()
return text_filter.casefold() in title.casefold()
def parse_feed(xml_bytes: bytes, max_items: int = MAX_ITEMS) -> tuple[str, str, str, list[dict]]:
"""Return title, site URL, image URL, and entries from a channel or Atom feed."""
root = ET.fromstring(xml_bytes)
kind = "atom" if local_name(root.tag) == "feed" else "channel"
channel = find_child(root, "channel") if kind == "channel" else root
if channel is None:
raise ValueError("feed has no channel")
channel_title = feed_title(channel, kind)
site = atom_link(channel, "alternate") if kind == "atom" else child_text(channel, "link")
image = thumbnail_url(channel)
entries = []
item_name = "item" if kind == "channel" else "entry"
for element in channel:
if local_name(element.tag) != item_name:
continue
link = entry_link(element, kind)
if not link or SHORTS_RE.search(link):
continue
title = child_text(element, "title") or link
published_raw = entry_date(element, kind)
entries.append(
{
"title": title,
"link": link,
"published": parse_date(published_raw),
"published_raw": published_raw,
"thumbnail": thumbnail_url(element),
}
)
if len(entries) >= max_items:
break
return channel_title, site, image, entries
def youtube_channel_id(feed_url: str, site_url: str) -> str:
for candidate in (feed_url, site_url):
match = YOUTUBE_CHANNEL_RE.search(candidate or "")
if match:
return match.group(1)
return ""
def channel_avatar(feed_url: str, site_url: str, feed_image: str) -> str:
"""Prefer a real channel portrait; otherwise the feed's own image."""
channel_id = youtube_channel_id(feed_url, site_url)
if not channel_id:
return feed_image
page_url = f"https://www.youtube.com/channel/{channel_id}"
try:
page = fetch(page_url, accept="text/html").decode("utf-8", "replace")
except (urllib.error.URLError, TimeoutError, OSError):
return feed_image
match = re.search(
r'<meta property="og:image" content="([^"]+)"',
page,
)
if match:
return html.unescape(match.group(1))
return feed_image
def read_feed_list(path: Path) -> list[dict]:
"""Return category markers and feed URLs in file order.
A category applies to every feed URL until the next category line.
"""
entries = []
category = ""
for line in path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
match = CATEGORY_RE.match(line)
if match:
category = match.group(1).strip()
entries.append({"kind": "category", "name": category})
continue
url, text_filter = split_feed_line(line)
entries.append({"kind": "feed", "url": url, "category": category, "filter": text_filter})
return entries
def split_feed_line(line: str) -> tuple[str, str]:
"""Split a feed URL from an optional "; filter" suffix.
The filter may be quoted to keep spaces, and a leading ! excludes
entries whose titles contain the text.
"""
if ";" not in line:
return line, ""
url, _, raw = line.partition(";")
raw = raw.strip()
negated = raw.startswith("!")
body = raw[1:].strip() if negated else raw
if len(body) >= 2 and body[0] == body[-1] and body[0] in {'"', "'"}:
body = body[1:-1]
return url.strip(), ("!" + body) if negated else body
def format_date(when: datetime | None, raw: str) -> str:
# Match the bash script, which prints the date in the local timezone.
if when is not None:
return when.astimezone().strftime("%Y-%m-%d")
return raw
def render_item(item: dict, now: datetime, recent: timedelta = RECENT_WINDOW) -> str:
"""Render one entry. item is a parsed entry, now is the generation time, and recent is the NEW window. Returns an HTML list item."""
published = item["published"]
is_new = is_recent(item, now, recent)
is_today = published_today(item, now)
date_text = format_date(published, item["published_raw"])
# A today entry is also new, so the new filter keeps it.
classes = "item today new" if is_today else "item new" if is_new else "item"
badge = '<span class="badge">TODAY</span>' if is_today else '<span class="badge">NEW</span>' if is_new else ""
if published is not None:
stamp = published.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
date_html = f'<time datetime="{stamp}">{html.escape(date_text)}</time>'
elif date_text:
date_html = f"<time>{html.escape(date_text)}</time>"
else:
date_html = ""
thumb = item["thumbnail"]
thumb_html = (
f'<img class="thumb" src="{html.escape(thumb, quote=True)}" alt="" loading="lazy">'
if thumb
else '<span class="thumb thumb-empty" aria-hidden="true"></span>'
)
title = html.escape(item["title"])
link = html.escape(item["link"], quote=True)
haystack = html.escape(item["title"].casefold(), quote=True)
return (
f'<li class="{classes}" data-text="{haystack}">'
f'<a href="{link}" target="_blank" rel="noopener">'
f"{thumb_html}"
f'<span class="item-body"><span class="item-title">{title}</span>'
f'<span class="item-meta">{date_html}{badge}</span></span>'
f"</a></li>"
)
def render_category(name: str) -> str:
"""Render a category heading. name is the heading text. Returns an HTML heading."""
return (
f'<h2 class="category" data-category="{html.escape(name.casefold(), quote=True)}">'
f"{html.escape(name)}</h2>"
)
def render_feed(
title: str,
site: str,
image: str,
items: list[dict],
now: datetime,
category: str = "",
recent: timedelta = RECENT_WINDOW,
) -> str:
"""Render one feed. title, site, and image describe the feed, items are the entries to show, now is the generation time, category is the heading it belongs to, and recent is the NEW window. Returns a section."""
safe_title = html.escape(title)
heading = safe_title
if site:
heading = f'<a href="{html.escape(site, quote=True)}" target="_blank" rel="noopener">{safe_title}</a>'
if image:
avatar = (
f'<img class="avatar" src="{html.escape(image, quote=True)}" '
f'alt="" width="72" height="72" loading="lazy">'
)
else:
letter = html.escape((title[:1] or "?").upper())
avatar = f'<span class="avatar avatar-fallback" aria-hidden="true">{letter}</span>'
fresh = [render_item(item, now, recent) for item in items if is_recent(item, now, recent) or published_today(item, now)]
older = [render_item(item, now, recent) for item in items if not is_recent(item, now, recent) and not published_today(item, now)]
parts = []
if fresh:
parts.append('<ul class="fresh">\n' + "\n".join(fresh) + "\n</ul>")
if older:
parts.append('<ul class="archive">\n' + "\n".join(older) + "\n</ul>")
return (
f'<section class="channel" data-channel="{html.escape(title.casefold(), quote=True)}" '
f'data-category="{html.escape(category.casefold(), quote=True)}">\n'
f'<header class="channel-head">{avatar}<h2>{heading}</h2></header>\n'
+ "\n".join(parts)
+ "\n</section>"
)
def is_recent(item: dict, now: datetime, recent: timedelta = RECENT_WINDOW) -> bool:
"""True when item was published within recent of now. Both datetimes are timezone-aware."""
published = item["published"]
return published is not None and (now - published) <= recent
def published_today(item: dict, now: datetime) -> bool:
"""True when item was published within 24 hours of now."""
published = item["published"]
return published is not None and timedelta(0) <= (now - published) <= timedelta(hours=24)
PAGE_TEMPLATE = "page.html"
PAGE_STYLE = "page.css"
PAGE_SCRIPT = "page.js"
def load_asset(name: str) -> str:
"""Read a page asset from the script directory. name is the filename. Returns its text."""
path = script_dir() / name
try:
return path.read_text(encoding="utf-8")
except OSError as exc:
raise SystemExit(f"Page asset not found: {path}") from exc
def render_category_filters(names: list[str]) -> str:
"""Render the category chips. names are the category headings. Returns HTML, or an empty string when there are none."""
if not names:
return ""
buttons = [
'<button type="button" data-category="" aria-pressed="true">all</button>'
]
for name in names:
buttons.append(
f'<button type="button" data-category="{html.escape(name.casefold(), quote=True)}">'
f"{html.escape(name)}</button>"
)
return (
'<div class="category-scroller">'
'<button type="button" class="caret caret-left" data-dir="-1" aria-label="Scroll categories left" tabindex="-1">‹</button>'
'<div class="categories" id="categories" role="group" aria-label="Categories">'
+ "".join(buttons)
+ '</div><button type="button" class="caret caret-right" data-dir="1" aria-label="Scroll categories right" tabindex="-1">›</button></div>'
)
def render_page(sections: list[str], now: datetime, failures: int, categories: list[str] | None = None) -> str:
"""Fill the page template with the generated feeds. sections are feed HTML, now is the generation time, failures is unused, and categories are the chip names. Returns the HTML document."""
stamp = now.astimezone().strftime("%Y-%m-%d %H:%M %Z").strip()
body = "\n".join(sections) if sections else '<p class="generated">No feed entries found.</p>'
page = load_asset(PAGE_TEMPLATE)
replacements = {
"/*PAGE_STYLE*/": load_asset(PAGE_STYLE).strip("\n"),
"/*PAGE_SCRIPT*/": load_asset(PAGE_SCRIPT).strip("\n"),
"<!--GENERATED-->": html.escape(stamp),
"<!--CATEGORIES-->": render_category_filters(categories or []),
"<!--FEEDS-->": body,
}
for marker, value in replacements.items():
if marker not in page:
raise SystemExit(f"Page template is missing {marker}")
page = page.replace(marker, value, 1)
return page
def feed_to_json(feed: dict) -> dict:
return {
"url": feed["url"],
"title": feed["title"],
"site": feed["site"],
"image": feed["image"],
"items": [
{
"title": item["title"],
"link": item["link"],
"published": item["published"].isoformat() if item["published"] else None,
"published_raw": item["published_raw"],
"thumbnail": item["thumbnail"],
}
for item in feed["items"]
],
}
def feed_from_json(feed: dict) -> dict:
return {
"url": feed["url"],
"title": feed["title"],
"site": feed.get("site", ""),
"image": feed.get("image", ""),
"items": [
{
"title": item["title"],
"link": item["link"],
"published": parse_date(item["published"]) if item.get("published") else None,
"published_raw": item.get("published_raw", ""),
"thumbnail": item.get("thumbnail", ""),
}
for item in feed.get("items", [])
],
}
def write_cache(path: Path, loaded: dict[str, dict]) -> None:
payload = {
"fetched_at": datetime.now(timezone.utc).isoformat(),
"feeds": [feed_to_json(loaded[url]) for url in loaded],
}
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
print(f"Cache written: {path}")
def read_cache(path: Path) -> dict[str, dict]:
payload = json.loads(path.read_text(encoding="utf-8"))
feeds = payload["feeds"] if isinstance(payload, dict) else payload
return {feed["url"]: feed_from_json(feed) for feed in feeds}
def select_items(feed: dict, text_filter: str, max_items: int) -> dict:
"""Choose the entries to render. The stored feed keeps its full list."""
items = feed["items"]
if text_filter:
items = [item for item in items if item_matches(item["title"], text_filter)]
return {**feed, "items": items[:max_items]}
def load_cached(path: Path) -> dict[str, dict]:
"""Return cached feeds, or an empty mapping when no usable cache exists."""
if not path.is_file():
return {}
try:
return read_cache(path)
except (OSError, json.JSONDecodeError, KeyError, TypeError) as exc:
print(f"Could not read cache {path}: {exc}", file=sys.stderr)
return {}
# Keep more than the page shows so a title filter can still fill the page
# from entries the feed listed beyond the first few matches.
CACHE_ITEMS = 50
def fetch_parsed(url: str, max_items: int) -> tuple[str, str, str, list[dict]]:
accept = "application/atom+xml, application/rss+xml, application/xml, text/xml"
payload = fetch(url, accept=accept)
return parse_feed(payload, max_items)
def relevant_count(items: list[dict], text_filter: str) -> int:
if not text_filter:
return len(items)
return sum(item_matches(item["title"], text_filter) for item in items)
def load_feed(url: str, max_items: int = MAX_ITEMS, text_filter: str = "") -> dict | None:
print(f"Processing feed: {url}", flush=True)
try:
title, site, image, items = fetch_parsed(url, max(max_items, CACHE_ITEMS))
except (urllib.error.URLError, TimeoutError, OSError, ET.ParseError, ValueError) as exc:
print(f"No items found or parse error for {url} ({exc})", file=sys.stderr)
return None
channel_id = youtube_channel_id(url, site)
if channel_id and relevant_count(items, text_filter) < max_items:
# The channel feed only holds 15 uploads. When Shorts fill that
# window, the long-form playlist still has the older videos.
try:
_, _, _, longform = fetch_parsed(
YOUTUBE_LONGFORM_PLAYLIST.format(channel_id[2:]), max(max_items, CACHE_ITEMS)
)
except (urllib.error.URLError, TimeoutError, OSError, ET.ParseError, ValueError):
longform = []
if len(longform) > len(items):
items = longform
if not items:
print(f"No items found or parse error for {url}", file=sys.stderr)
return None
return {
"url": url,
"title": title or url,
"site": site,
"image": channel_avatar(url, site, image),
"items": items,
}
def script_dir() -> Path:
return Path(__file__).resolve().parent
def cache_dir() -> Path:
"""$XDG_CACHE_HOME/feedfetcher, or ~/.cache/feedfetcher when unset."""
base = os.environ.get("XDG_CACHE_HOME")
root = Path(base).expanduser() if base else Path.home() / ".cache"
return root / "feedfetcher"
def default_cache_path() -> Path:
"""JSON cache of fetched feeds, inside the cache directory."""
return cache_dir() / "feeds-cache.json"
def default_output_path() -> Path:
"""Generated HTML page, inside the cache directory."""
return cache_dir() / "feeds.html"
DURATION_RE = re.compile(r"^(\d+)([dhm])$")
DURATION_UNITS = {"d": "days", "h": "hours", "m": "minutes"}
def parse_duration(value: str) -> timedelta:
"""Parse a duration such as 7d, 12h, or 30m."""
match = DURATION_RE.match(value.strip())
if not match or int(match.group(1)) < 1:
raise argparse.ArgumentTypeError("expected a duration like 7d, 12h, or 30m")
return timedelta(**{DURATION_UNITS[match.group(2)]: int(match.group(1))})
def parse_args(argv: list[str]) -> argparse.Namespace:
here = script_dir()
parser = argparse.ArgumentParser(
prog=Path(argv[0]).name,
description="Fetch feeds and write an HTML page of recent entries.",
)
parser.add_argument("--feeds", default=str(here / "feeds.txt"), metavar="FILE", help="feed list (default: feeds.txt next to this script)")
parser.add_argument("--output", default=str(default_output_path()), metavar="FILE", help=f"HTML file to write (default: {default_output_path()})")
parser.add_argument(
"--copy",
default=str(Path.home() / "Videos" / "Internet-Shows"),
metavar="DEST",
help="also copy the HTML here; a DEST containing ':' is sent with scp (default: ~/Videos/Internet-Shows when that directory exists)",
)
parser.add_argument("--no-copy", action="store_true", help="do not copy the HTML file after writing it")
parser.add_argument("--items", type=int, default=MAX_ITEMS, metavar="N", help=f"entries to keep from each feed (default: {MAX_ITEMS})")
parser.add_argument("--new-interval", type=parse_duration, default=RECENT_WINDOW, metavar="DURATION", help="mark entries newer than this as NEW; 7d, 12h, or 30m (default: 7d)")
parser.add_argument("--cache", default=str(default_cache_path()), metavar="FILE", help=f"JSON cache of fetched feeds (default: {default_cache_path()})")
parser.add_argument("--from-cache", action="store_true", help="build the page from the cache without fetching feeds")
parser.add_argument("--open", action="store_true", help="open the generated HTML file in the default browser")
args = parser.parse_args(argv[1:])
if args.items < 1:
parser.error("--items must be at least 1")
return args
def open_output(output_path: Path) -> None:
"""Open the HTML file in the default browser. output_path is the file just written."""
if not webbrowser.open(output_path.resolve().as_uri()):
raise OSError(f"could not open {output_path}")
def copy_output(output_path: Path, dest: str) -> None:
"""Copy the HTML file locally, or with scp when dest contains ':'."""
if ":" in dest:
target = dest if dest.endswith("/") else dest + "/"
subprocess.run(["scp", str(output_path), target + output_path.name], check=True)
print(f"Copied to {target}{output_path.name}")
return
directory = Path(dest).expanduser()
if not directory.is_dir():
print(f"Copy destination is not a directory, skipping: {directory}", file=sys.stderr)
return
shutil.copy2(output_path, directory / output_path.name)
print(f"Copied to {directory / output_path.name}")
def main(argv: list[str]) -> int:
args = parse_args(argv)
input_path = Path(args.feeds).expanduser()
output_path = Path(args.output).expanduser()
cache_path = Path(args.cache).expanduser()
max_items = args.items
recent = args.new_interval
if not input_path.is_file():
print(f"Input file not found: {input_path}", file=sys.stderr)
return 1
now = datetime.now(timezone.utc)
feed_list = read_feed_list(input_path)
feeds = [entry for entry in feed_list if entry["kind"] == "feed"]
urls = [entry["url"] for entry in feeds]
filters = {entry["url"]: entry["filter"] for entry in feeds}
categories = [entry["name"] for entry in feed_list if entry["kind"] == "category"]
loaded: dict[str, dict] = {}
failures = 0
if args.from_cache:
if not cache_path.is_file():
print(f"Cache file not found: {cache_path}", file=sys.stderr)
return 1
try:
cached = read_cache(cache_path)
except (OSError, json.JSONDecodeError, KeyError, TypeError) as exc:
print(f"Could not read cache {cache_path}: {exc}", file=sys.stderr)
return 1
print(f"Using cache: {cache_path}")
for url in urls:
feed = cached.get(url)
if feed is None or not feed["items"]:
print(f"No cached items for {url}", file=sys.stderr)
failures += 1
else:
loaded[url] = feed
else:
cached = load_cached(cache_path)
workers = min(8, max(1, len(urls)))
with ThreadPoolExecutor(max_workers=workers) as pool:
futures = {pool.submit(load_feed, url, max_items, filters.get(url, "")): url for url in urls}
for future in as_completed(futures):
url = futures[future]
try:
result = future.result()
except Exception as exc: # a worker bug should not drop the other feeds
print(f"No items found or parse error for {url} ({exc})", file=sys.stderr)
result = None
if result is None:
fallback = cached.get(url)
if fallback and fallback["items"]:
print(f"Using cached copy of {url}", file=sys.stderr)
loaded[url] = fallback
else:
failures += 1
else:
loaded[url] = result
try:
write_cache(cache_path, loaded)
except OSError as exc:
print(f"Could not write cache {cache_path}: {exc}", file=sys.stderr)
return 1
sections = []
open_group = False
for entry in feed_list:
if entry["kind"] == "category":
if open_group:
sections.append("</div>")
sections.append(f'<div class="group">\n{render_category(entry["name"])}')
open_group = True
continue
feed = loaded.get(entry["url"])
if feed is None:
continue
shown = select_items(feed, entry.get("filter", ""), max_items)
if not shown["items"]:
continue
sections.append(
render_feed(shown["title"], shown["site"], shown["image"], shown["items"], now, entry["category"], recent)
)
if open_group:
sections.append("</div>")
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(render_page(sections, now, failures, categories), encoding="utf-8")
print(f"HTML document generated: {output_path}")
if not args.no_copy:
try:
copy_output(output_path, args.copy)
except (OSError, subprocess.CalledProcessError) as exc:
print(f"Copy failed: {exc}", file=sys.stderr)
return 1
if args.open:
try:
open_output(output_path)
except OSError as exc:
print(f"Could not open {output_path}: {exc}", file=sys.stderr)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main(sys.argv))