107 lines
3 KiB
Python
107 lines
3 KiB
Python
"""Pygea main entry point."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from pygea.config import FeedDefinition, PygeaConfig, load_config
|
|
from pygea.pexception import PangeaServiceException
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Generate RSS feeds from Pangea")
|
|
parser.add_argument(
|
|
"-c",
|
|
"--config",
|
|
default="pygea.toml",
|
|
help="Path to runtime config TOML file",
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def _toml_string(value: str) -> str:
|
|
return json.dumps(value, ensure_ascii=False)
|
|
|
|
|
|
def render_manifest(feeds: list[dict[str, str]]) -> str:
|
|
lines: list[str] = []
|
|
for feed in feeds:
|
|
lines.extend(
|
|
[
|
|
"[[feeds]]",
|
|
f"name = {_toml_string(feed['name'])}",
|
|
f"slug = {_toml_string(feed['slug'])}",
|
|
f"url = {_toml_string(feed['url'])}",
|
|
"",
|
|
]
|
|
)
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def write_manifest(config: PygeaConfig, manifest_feeds: list[dict[str, str]]) -> None:
|
|
"""Write the feed manifest beside the generated feed output."""
|
|
if config.results.output_to_file_p is not True:
|
|
return
|
|
|
|
config.results.output_directory.mkdir(parents=True, exist_ok=True)
|
|
manifest_path = config.results.output_directory / "manifest.toml"
|
|
manifest_path.write_text(render_manifest(manifest_feeds), encoding="utf-8")
|
|
|
|
|
|
def feed_class():
|
|
from pygea.pangeafeed import PangeaFeed
|
|
|
|
return PangeaFeed
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
try:
|
|
config = load_config(args.config)
|
|
except FileNotFoundError:
|
|
print(
|
|
"Config file not found: {}".format(Path(args.config).expanduser()),
|
|
file=sys.stderr,
|
|
)
|
|
print(
|
|
"Use --config PATH or create pygea.toml in the project root",
|
|
file=sys.stderr,
|
|
)
|
|
return 2
|
|
|
|
try:
|
|
manifest_feeds: list[dict[str, str]] = []
|
|
pangea_feed_class = feed_class()
|
|
for feed in config.feeds:
|
|
pf = pangea_feed_class(config, [feed])
|
|
pf.acquire_content()
|
|
pf.generate_feed()
|
|
output_path = pf.disgorge(feed["slug"])
|
|
if output_path is not None:
|
|
manifest_feeds.append(_manifest_entry(feed, output_path))
|
|
print(
|
|
"feed for {} output to sub-directory {}".format(
|
|
feed["name"], feed["slug"]
|
|
)
|
|
)
|
|
write_manifest(config, manifest_feeds)
|
|
except PangeaServiceException as error:
|
|
print(error, file=sys.stderr)
|
|
return 1
|
|
|
|
return 0
|
|
|
|
|
|
def _manifest_entry(feed: FeedDefinition, output_path: Path) -> dict[str, str]:
|
|
return {
|
|
"name": feed["name"],
|
|
"slug": feed["slug"],
|
|
"url": output_path.resolve().as_uri(),
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|