class_docstr: C4 dataset based on Common Crawl.
class_docstr_tok: ["C4","dataset","based","on","Common","Crawl","."]
docstr: Build PCollection of un-split page content.
docstr_tok: ["Build","PCollection","of","un-split","page","content","."]
code: def _get_pages_pcollection(self, pipeline, file_paths, dl_manager):
      """Build PCollection of un-split page content."""
      beam = tfds.core.lazy_imports.apache_beam
  
      def download_wet_file(path, dl_dir):
          url = f"{_DOWNLOAD_HOST}/{path}"
          out_path = f"{dl_dir}/{path}"
  
          if tf.io.gfile.exists(out_path):
              c4_utils.get_counter_inc_fn("download_wet_url")("exists")
              return out_path
  
          tmp_dir = f"{out_path}.incomplete{uuid.uuid4().hex}"
          try:
              tf.io.gfile.makedirs(tmp_dir)
              downloader = (
                  tfds.download.download_manager.get_downloader()
              )
              with downloader.tqdm():
                  # TODO(slebedev): Investigate why pytype infers Promise[Future[...]].
                  dl_path = downloader.download(url, tmp_dir).get().path  # type: ignore
              tf.io.gfile.rename(
                  os.fspath(dl_path), out_path, overwrite=True
              )
          finally:
              if tf.io.gfile.exists(tmp_dir):
                  tf.io.gfile.rmtree(tmp_dir)
  
              c4_utils.get_counter_inc_fn("download_wet_url")(
                  "downloaded"
              )
          return out_path
  
      wet_file_paths = (
          pipeline
          | "create_wet_path_urls"
          >> beam.Create(file_paths["wet_path_urls"])
          | beam.io.ReadAllFromText(
              compression_type=beam.io.filesystem.CompressionTypes.UNCOMPRESSED
          )
          # Increase parallelism.
          | beam.Reshuffle()
          | "filter_corrupt_wet_files"
          >> beam.Filter(lambda p: p not in _KNOWN_CORRUPT_WET_FILES)
          | beam.Map(
              download_wet_file,
              dl_dir=os.path.join(
                  dl_manager.download_dir, "c4_wet_files"
              ),
          )
      )
  
      # Parse WET files and filter by length.
      # Output: [PageFeatures]
      pages = (
          wet_file_paths
          | beam.FlatMap(c4_utils.split_wet_file)
          | beam.Filter(c4_utils.is_valid_length)
      )
  
      # Optionally filter for RealNews domains.
      # Output: [PageFeatures]
      if self.builder_config.realnewslike:
          with tf.io.gfile.GFile(file_paths["realnews_domains"]) as f:
              realnews_domains = json.load(f)
          pages |= beam.Filter(
              c4_utils.is_realnews_domain, realnews_domains
          )
  
      # Normalize and deduplicate by URL.
      # Output: [PageFeatures]
      pages = (
          pages
          | "normalize_url" >> beam.Map(lambda p: (p.normalized_url, p))
          | "select_newest"
          >> beam.CombinePerKey(c4_utils.select_newest_page)
          | beam.Values()
      )
  
      # Optionally filter for WebText-like URLs.
      # Output: [PageFeatures]
      if self.builder_config.webtextlike:
          webtextlike_urls = (
              pipeline
              | "read_webtextlike_urls"
              >> beam.io.ReadFromText(
                  os.path.join(
                      file_paths["openwebtext_urls_zip"],
                      _OPENWEBTEXT_URLS_FILE_PATTERN,
                  )
              )
              | "add_dummy_page"
              >> beam.Map(
                  lambda url: (
                      c4_utils.normalize_url(url),
                      PageFeatures(),
                  )
              )
          )
          pages = (
              {
                  "pages": pages
                  | beam.Map(lambda p: (p.normalized_url, p)),
                  "webtextlike_urls": webtextlike_urls,
              }
              | "group_webtextlike_urls" >> beam.CoGroupByKey()
              | beam.FlatMap(c4_utils.filter_by_webtextlike)
          )
  
      if self.builder_config.paragraph_filter:
          pages |= beam.Filter(c4_utils.paragraph_filter)
  
      if self.builder_config.clean:
          pages |= "clean_pages" >> beam.FlatMap(
              c4_utils.get_clean_page_fn()
          )
  
      if self.builder_config.dedupe:
          pages = (
              # Also removes documents with too few sentences after deduplication.
              c4_utils.remove_duplicate_text(
                  pages
              )  # pylint:disable=g-long-ternary
              if self.builder_config.clean
              else
              # If we are not cleaning, do not remove too-few-sentence documents.
              c4_utils.remove_duplicate_text(pages, min_num_sentences=0)
          )
  
      # Add detected language.
      if self.builder_config.languages == ["en"]:
          # Use langdetect for reproducibility of the original C4.
          pages |= beam.FlatMap(c4_utils.detect_english)
      else:
          pages = c4_utils.detect_languages(
              pages, valid_languages=self.builder_config.languages
          )
  
      if self.builder_config.badwords_filter_fraction > 0.0:
          # Create dictionary of badwords regex for each available language.
          badwords = collections.defaultdict(set)
          for lang, path in file_paths["badwords"].items():
              lang = lang.split("-")[0]  # remove suffix if present
              with epath.Path(path).open() as f:
                  badwords[lang].update(l.strip() for l in f)
  
          for lang, allowlist in _BADWORDS_ALLOWLIST.items():
              badwords[lang] -= allowlist
  
          pages |= beam.Filter(
              c4_utils.get_badwords_filter_fn(
                  badwords,
                  filter_fraction=self.builder_config.badwords_filter_fraction,
              )
          )
  
      return pages
  
imports: import collections
  import functools
  import json
  import os
  from typing import Optional, Sequence
  import uuid
  from absl import logging
  from etils import epath
  from tensorflow_datasets.core.utils import tree_utils
  from tensorflow_datasets.core.utils.lazy_imports_utils import tensorflow
  import tensorflow_datasets.public_api
  from tensorflow_datasets.text import c4_utils
label_desc: Use image node_id 3 for calling the C4 obj's underlying member method code with example usage: obj._get_pages_pcollection(pipeline, file_paths, dl_manager) and returns: pages, out_path, out_path