def execute()

in src/python/pants/backend/docgen/tasks/markdown_to_html.py [0:0]

• 66 lines of code
• 20 McCabe index (conditional complexity)

  def execute(self):
    # TODO(John Sirois): consider adding change detection

    outdir = os.path.join(self.get_options().pants_distdir, 'markdown')
    css_path = os.path.join(outdir, 'css', 'codehighlight.css')
    css = util().emit_codehighlight_css(css_path, self.code_style)
    if css:
      self.context.log.info('Emitted {}'.format(css))

    def is_page(target):
      return isinstance(target, Page)

    roots = set()
    interior_nodes = set()
    if self.open:
      dependencies_by_page = self.context.dependents(on_predicate=is_page, from_predicate=is_page)
      roots.update(dependencies_by_page.keys())
      for dependencies in dependencies_by_page.values():
        interior_nodes.update(dependencies)
        roots.difference_update(dependencies)
      for page in self.context.targets(is_page):
        # There are no in or out edges so we need to show show this isolated page.
        if not page.dependencies and page not in interior_nodes:
          roots.add(page)

    with self.context.new_workunit(name='render', labels=[WorkUnitLabel.MULTITOOL]):
      plaingenmap = self.context.products.get(self.MARKDOWN_HTML_PRODUCT)
      wikigenmap = self.context.products.get(self.WIKI_HTML_PRODUCT)
      show = []
      for page in self.context.targets(is_page):
        def process_page(key, outdir, url_builder, genmap, fragment=False):
          if page.format == 'rst':
            with self.context.new_workunit(name='rst') as workunit:
              html_path = self.process_rst(
                workunit,
                page,
                os.path.join(outdir, util().page_to_html_path(page)),
                os.path.join(page.payload.sources.rel_path, page.source),
                self.fragment or fragment,
              )
          else:
            with self.context.new_workunit(name='md'):
              html_path = self.process_md(
                os.path.join(outdir, util().page_to_html_path(page)),
                os.path.join(page.payload.sources.rel_path, page.source),
                self.fragment or fragment,
                url_builder,
                css=css,
              )
          self.context.log.info('Processed {} to {}'.format(page.source, html_path))
          relpath = os.path.relpath(html_path, outdir)
          genmap.add(key, outdir, [relpath])
          return html_path

        def url_builder(linked_page):
          dest = util().page_to_html_path(linked_page)
          src_dir = os.path.dirname(util().page_to_html_path(page))
          return linked_page.name, os.path.relpath(dest, src_dir)

        page_path = os.path.join(outdir, 'html')
        html = process_page(page, page_path, url_builder, plaingenmap)
        if css and not self.fragment:
          plaingenmap.add(page, self.workdir, list(css_path))
        if self.open and page in roots:
          show.append(html)

        if page.provides:
          for wiki in page.provides:
            basedir = os.path.join(self.workdir, str(hash(wiki)))
            process_page((wiki, page), basedir, wiki.wiki.url_builder, wikigenmap, fragment=True)

    if show:
      try:
        desktop.ui_open(*show)
      except desktop.OpenError as e:
        raise TaskError(e)