333 lines
14 KiB
Python
333 lines
14 KiB
Python
"""CLI: move guides in and out of a site.
|
|
|
|
flask guides import-html PATH [PATH ...] an HTML article becomes a guide
|
|
flask guides export --out DIR guides become portable files
|
|
flask guides import PATH [PATH ...] those files become guides here
|
|
|
|
A site that adopts this app usually arrives with a folder of procedures already
|
|
written - styled HTML pages that get mailed around and opened off a share. This
|
|
imports them so they live where they are used, become searchable, tickable and
|
|
printable, and stop being a file somebody has to find.
|
|
|
|
Imports land as DRAFTS, always - both kinds. An HTML conversion is faithful but
|
|
it is not a render, and a bundle written on another site has not been read by
|
|
anybody here. Publishing is a decision somebody makes after reading.
|
|
|
|
Sites are separate databases and the useful ones are air-gapped from where a
|
|
guide gets written, so export/import is how the same procedure reaches all of
|
|
them: one JSON file per guide, images carried inside it, moved by whatever
|
|
moves files to that site.
|
|
"""
|
|
|
|
import click
|
|
from flask.cli import AppGroup, with_appcontext
|
|
|
|
from shopdb.api import db
|
|
|
|
from .models import Guide, GuideImage
|
|
from .services import authoring, bundle as bundleservice
|
|
from .services.blocks import BlockError
|
|
from .services.htmlimport import parse
|
|
|
|
|
|
# Mirrors what the upload route accepts. An import must not be able to store an
|
|
# image the editor would have refused, or the guide renders here and nowhere
|
|
# else.
|
|
ALLOWEDIMAGETYPES = ('image/png', 'image/jpeg', 'image/gif', 'image/webp')
|
|
MAXIMAGEBYTES = 8 * 1024 * 1024
|
|
|
|
|
|
def _slugify(title):
|
|
"""Same rule as the API's, kept here so the CLI does not import a route."""
|
|
import re
|
|
base = re.sub(r'[^a-z0-9]+', '-', (title or '').lower()).strip('-') or 'guide'
|
|
base = base[:180]
|
|
candidate = base
|
|
suffix = 2
|
|
while Guide.query.filter(Guide.slug == candidate).first() is not None:
|
|
candidate = '{}-{}'.format(base, suffix)
|
|
suffix += 1
|
|
return candidate
|
|
|
|
|
|
def build_cli():
|
|
guides_cli = AppGroup('guides')
|
|
|
|
@guides_cli.command('import-html')
|
|
@click.argument('paths', nargs=-1, required=True,
|
|
type=click.Path(exists=True, dir_okay=False))
|
|
@click.option('--title', default=None,
|
|
help='Override the title taken from the article.')
|
|
@click.option('--tags', default=None,
|
|
help='Comma-separated tags for every guide imported.')
|
|
@click.option('--userid', type=int, default=None,
|
|
help='Record this user as the author.')
|
|
@click.option('--replace', is_flag=True,
|
|
help='If a guide already has this slug, edit it in place '
|
|
'(a new revision) instead of creating a second one.')
|
|
@click.option('--apply/--dry-run', default=False,
|
|
help='Write to the database. The default reports only.')
|
|
@with_appcontext
|
|
def import_html(paths, title, tags, userid, replace, apply):
|
|
"""Convert HTML article(s) into draft guides."""
|
|
if title and len(paths) > 1:
|
|
raise click.UsageError('--title names one guide; pass one file.')
|
|
|
|
for path in paths:
|
|
with open(path, encoding='utf-8', errors='replace') as handle:
|
|
parsed = parse(handle.read())
|
|
|
|
name = title or parsed['title']
|
|
counts = {}
|
|
for block in parsed['blocks']:
|
|
counts[block['type']] = counts.get(block['type'], 0) + 1
|
|
|
|
click.echo('{}'.format(path))
|
|
click.echo(' title {}'.format(name))
|
|
click.echo(' blocks {} ({})'.format(
|
|
len(parsed['blocks']),
|
|
', '.join('{} {}'.format(v, k) for k, v in sorted(counts.items()))))
|
|
click.echo(' images {}'.format(len(parsed['images'])))
|
|
for note in parsed['skipped']:
|
|
click.echo(' skipped {}'.format(note))
|
|
|
|
rejected = [image for image in parsed['images']
|
|
if image['mimetype'] not in ALLOWEDIMAGETYPES
|
|
or len(image['data']) > MAXIMAGEBYTES]
|
|
for image in rejected:
|
|
click.echo(' DROPPED {} ({}, {:.1f} MB) - not an accepted image'
|
|
.format(image['token'], image['mimetype'],
|
|
len(image['data']) / 1048576))
|
|
|
|
if not apply:
|
|
click.echo(' dry run, nothing written. Pass --apply to import.')
|
|
continue
|
|
|
|
try:
|
|
guide = _import(parsed, name, tags, userid, replace, rejected)
|
|
except BlockError as error:
|
|
db.session.rollback()
|
|
click.echo(' FAILED {}'.format(error))
|
|
continue
|
|
|
|
db.session.commit()
|
|
click.echo(' imported as /guides/{} (draft, revision {})'
|
|
.format(guide.slug, guide.revision))
|
|
|
|
@guides_cli.command('export')
|
|
@click.option('--out', 'outdir', required=True,
|
|
type=click.Path(file_okay=False),
|
|
help='Directory to write the bundles into. Created if needed.')
|
|
@click.option('--slug', 'slugs', multiple=True,
|
|
help='Export just this guide. Repeatable. Default: all.')
|
|
@click.option('--published-only', is_flag=True,
|
|
help='Skip guides that have never been published.')
|
|
@with_appcontext
|
|
def export_guides(outdir, slugs, published_only):
|
|
"""Write guides out as portable .guide.json files."""
|
|
import os
|
|
|
|
query = Guide.query
|
|
if slugs:
|
|
query = query.filter(Guide.slug.in_(slugs))
|
|
if published_only:
|
|
query = query.filter(Guide.publishedrevisionid.isnot(None))
|
|
guides = query.order_by(Guide.slug).all()
|
|
|
|
missing = set(slugs) - {guide.slug for guide in guides}
|
|
for slug in sorted(missing):
|
|
click.echo(' NOT FOUND {}'.format(slug))
|
|
|
|
if not guides:
|
|
click.echo('Nothing to export.')
|
|
return
|
|
|
|
os.makedirs(outdir, exist_ok=True)
|
|
for guide in guides:
|
|
images = GuideImage.query.filter_by(guideid=guide.guideid).all()
|
|
payload = bundleservice.build(guide, images)
|
|
path = os.path.join(outdir, '{}.guide.json'.format(guide.slug))
|
|
with open(path, 'w', encoding='utf-8') as handle:
|
|
handle.write(bundleservice.dumps(payload))
|
|
click.echo(' {} {} blocks, {} image(s), {:.0f} KB{}'.format(
|
|
os.path.basename(path), len(payload['blocks']),
|
|
len(payload['images']), os.path.getsize(path) / 1024,
|
|
'' if guide.publishedrevisionid else ' [draft here]'))
|
|
click.echo('{} guide(s) -> {}'.format(len(guides), outdir))
|
|
|
|
@guides_cli.command('import')
|
|
@click.argument('paths', nargs=-1, required=True,
|
|
type=click.Path(exists=True))
|
|
@click.option('--userid', type=int, default=None,
|
|
help='Record this user as the author on this site.')
|
|
@click.option('--replace', is_flag=True,
|
|
help='If a guide already has the bundle\'s slug, edit it in '
|
|
'place (a new revision) instead of creating a second '
|
|
'one. This is how an updated guide is re-shipped.')
|
|
@click.option('--apply/--dry-run', default=False,
|
|
help='Write to the database. The default reports only.')
|
|
@with_appcontext
|
|
def import_guides(paths, userid, replace, apply):
|
|
"""Import .guide.json bundle(s) written by `guides export`."""
|
|
import os
|
|
|
|
files = []
|
|
for path in paths:
|
|
if os.path.isdir(path):
|
|
files.extend(sorted(os.path.join(path, name)
|
|
for name in os.listdir(path)
|
|
if name.endswith('.guide.json')))
|
|
else:
|
|
files.append(path)
|
|
if not files:
|
|
click.echo('No .guide.json files found.')
|
|
return
|
|
|
|
for path in files:
|
|
click.echo(os.path.basename(path))
|
|
try:
|
|
with open(path, encoding='utf-8') as handle:
|
|
payload = bundleservice.loads(handle.read())
|
|
except bundleservice.BundleError as error:
|
|
click.echo(' SKIPPED {}'.format(error))
|
|
continue
|
|
|
|
existing = Guide.query.filter(Guide.slug == payload['slug']).first()
|
|
click.echo(' title {}'.format(payload.get('title')))
|
|
click.echo(' blocks {}, images {}'.format(
|
|
len(payload['blocks']), len(payload['images'])))
|
|
if existing is not None:
|
|
click.echo(' {} already has this slug{}'.format(
|
|
'guide {}'.format(existing.guideid),
|
|
' - will be updated in place' if replace
|
|
else ' - a SECOND guide will be created; pass --replace to '
|
|
'update it instead'))
|
|
|
|
rejected = [image for image in payload['images']
|
|
if image['mimetype'] not in ALLOWEDIMAGETYPES
|
|
or len(image['data']) > MAXIMAGEBYTES]
|
|
for image in rejected:
|
|
click.echo(' DROPPED {} ({}) - not an accepted image'
|
|
.format(image['token'], image['mimetype']))
|
|
|
|
if not apply:
|
|
click.echo(' dry run, nothing written. Pass --apply to import.')
|
|
continue
|
|
|
|
try:
|
|
guide = _importbundle(payload, userid, replace, rejected)
|
|
except BlockError as error:
|
|
db.session.rollback()
|
|
click.echo(' FAILED {}'.format(error))
|
|
continue
|
|
|
|
db.session.commit()
|
|
click.echo(' imported as /guides/{} (draft, revision {})'
|
|
.format(guide.slug, guide.revision))
|
|
|
|
return [guides_cli]
|
|
|
|
|
|
def _import(parsed, name, tags, userid, replace, rejected):
|
|
"""Create or update the guide and its images. Caller commits."""
|
|
slug = None
|
|
guide = None
|
|
if replace:
|
|
# Match on the slug the title WOULD produce, which is how somebody
|
|
# re-imports a corrected article over the one they imported yesterday.
|
|
import re
|
|
wanted = re.sub(r'[^a-z0-9]+', '-', (name or '').lower()).strip('-')
|
|
guide = Guide.query.filter(Guide.slug == wanted).first()
|
|
|
|
if guide is None:
|
|
slug = _slugify(name)
|
|
guide = Guide(title=name, slug=slug, status='draft', createdby=userid)
|
|
db.session.add(guide)
|
|
db.session.flush()
|
|
|
|
# Images first: the block needs the stored id, and an id only exists once
|
|
# the row does.
|
|
droppedtokens = {image['token'] for image in rejected}
|
|
ids = {}
|
|
for image in parsed['images']:
|
|
if image['token'] in droppedtokens:
|
|
continue
|
|
row = GuideImage(guideid=guide.guideid, filename=image['filename'],
|
|
mimetype=image['mimetype'],
|
|
bytesize=len(image['data']), data=image['data'],
|
|
uploadedby=userid)
|
|
db.session.add(row)
|
|
db.session.flush()
|
|
ids[image['token']] = row.guideimageid
|
|
|
|
blocks = []
|
|
for block in parsed['blocks']:
|
|
block = dict(block)
|
|
token = block.pop('imagetoken', None)
|
|
if token is not None:
|
|
if token not in ids:
|
|
continue # its image was dropped; the caption would lie
|
|
block['imageid'] = ids[token]
|
|
blocks.append(block)
|
|
|
|
authoring.save(guide, title=name, summary=parsed['summary'] or None,
|
|
tags=tags, blocks=blocks,
|
|
note='Imported from HTML', userid=userid)
|
|
return guide
|
|
|
|
|
|
def _importbundle(payload, userid, replace, rejected):
|
|
"""Create or update a guide from a bundle. Caller commits."""
|
|
guide = None
|
|
if replace:
|
|
guide = Guide.query.filter(Guide.slug == payload['slug']).first()
|
|
|
|
if guide is None:
|
|
# The bundle's own slug when it is free, because the slug is the name
|
|
# somebody pastes into a work order and the same guide should have the
|
|
# same address on every site. A clash only suffixes.
|
|
slug = payload['slug']
|
|
if Guide.query.filter(Guide.slug == slug).first() is not None:
|
|
slug = _slugify(payload.get('title') or slug)
|
|
guide = Guide(title=payload.get('title') or slug, slug=slug,
|
|
status='draft', createdby=userid)
|
|
db.session.add(guide)
|
|
db.session.flush()
|
|
|
|
# Replacing means the incoming bundle is the whole truth for this guide.
|
|
# Its blocks will reference the images arriving now, so the ones already
|
|
# stored become orphans - and re-shipping a guide monthly would otherwise
|
|
# grow a pile of them nothing points at.
|
|
if replace:
|
|
GuideImage.query.filter_by(guideid=guide.guideid).delete()
|
|
db.session.flush()
|
|
|
|
droppedtokens = {image['token'] for image in rejected}
|
|
ids = {}
|
|
for image in payload['images']:
|
|
if image['token'] in droppedtokens:
|
|
continue
|
|
row = GuideImage(guideid=guide.guideid, filename=image['filename'],
|
|
mimetype=image['mimetype'],
|
|
bytesize=len(image['data']), data=image['data'],
|
|
uploadedby=userid)
|
|
db.session.add(row)
|
|
db.session.flush()
|
|
ids[image['token']] = row.guideimageid
|
|
|
|
blocks = []
|
|
for block in payload['blocks']:
|
|
block = dict(block)
|
|
token = block.pop('imagetoken', None)
|
|
if token is not None:
|
|
if token not in ids:
|
|
continue
|
|
block['imageid'] = ids[token]
|
|
blocks.append(block)
|
|
|
|
authoring.save(guide, title=payload.get('title'),
|
|
summary=payload.get('summary'), tags=payload.get('tags'),
|
|
blocks=blocks, note='Imported from a guide bundle',
|
|
userid=userid)
|
|
return guide
|