brainsteam.co.uk/bstools/bstools.py

116 lines
3.3 KiB
Python

import click
import dotenv
import os
import requests
import ujson
import frontmatter
from urllib.parse import urlparse
@click.group()
def cli():
dotenv.load_dotenv()
pass
@cli.command()
@click.option("--folder", type=click.Path(dir_okay=True, file_okay=False), required=True)
@click.option("--old_type", type=str, required=True)
@click.option("--new_type", type=str, required=True)
def fix_post_types(folder: str, old_type: str, new_type: str):
"""Fix post type metadata"""
for root, _, files in os.walk(folder):
for file in files:
if file.endswith(".md"):
full_path = os.path.join(root,file)
data = frontmatter.load(full_path)
print(f"Analysing... {full_path}")
if 'type' not in data:
print(f"Skipping {full_path} due to incomplete frontmatter")
continue
if(data['type'] == old_type):
print(f"Update type for {full_path}: {old_type}->{new_type}")
data['type'] = new_type
with open(full_path,'wb') as f:
frontmatter.dump(data, f)
@cli.command()
@click.option("--folder", type=click.Path(dir_okay=True, file_okay=False), required=True)
@click.option("--page_meta", type=str, help="comma separated list of fields to include in page meta", required=True)
def set_page_meta(folder: str, page_meta: str):
meta = page_meta.split(",")
for root, _, files in os.walk(folder):
for file in files:
if file.endswith(".md"):
full_path = os.path.join(root,file)
data = frontmatter.load(full_path)
print(f"Update page_meta for {full_path}: {meta}")
if 'page_meta' in data:
del data['page_meta']
data['post_meta'] = meta
with open(full_path,'wb') as f:
frontmatter.dump(data, f)
@cli.command()
@click.option("--mentions-file", type=click.Path(file_okay=True), required=True)
def fetch_mentions(mentions_file: str):
"""Fetch web mentions and store as json"""
mention_ids = set()
if os.path.exists(mentions_file):
print(f"Load existing mentions from {mentions_file}")
with open(mentions_file,'r') as f:
mentions = ujson.load(f)
print(mentions.keys())
print(f"Found existing mentions for {len(mentions.keys())} urls")
else:
mentions = {}
for mentionset in mentions.values():
mention_ids.update([post['id'] for post in mentionset])
print("Requesting new mentions...")
r = requests.get(f"https://webmention.io/api/mentions.json?token={os.environ.get('WEBMENTIONSIO_API_KEY')}")
if r.json().get('error') is not None:
print(f"Failed to request webmentions: {r.json()}")
return 1
new = 0
for link in r.json()['links']:
target = urlparse(link['target']).path
if target not in mentions:
mentions[target] = []
if link['id'] not in mention_ids:
mention_ids.add(link['id'])
mentions[target].append(link)
new += 1
print(f"Found {new} new mentions")
print(f"Storing mentions at {mentions_file}")
with open(mentions_file,'w') as f:
ujson.dump(mentions, f, indent=2)
if __name__ == "__main__":
cli()