<commit_before># Copyright 2019 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Common utility functions."""

from builtins import range
from future import standard_library
standard_library.install_aliases()
from past.builtins import basestring
import ast
import datetime
import functools
import gc
import hashlib
import inspect
import os
import random
import requests
import sys
import time
import urllib.parse
import urllib.request
import weakref

from base import errors
from base import memoize
from base import retry
from config import local_config
from metrics import logs
from system import environment

try:
  import psutil
except ImportError:
  psutil = None

# FIXME: Binary extensions list is still very basic.
BINARY_EXTENSIONS = [
    # Media formats.
    '.mp3',
    '.ogg',
    '.mp4',
    '.webm',
    # Image Formats.
    '.png',
    '.jpg',
    '.gif',
    # Misc.
    '.pdf',
    '.swf',
]
FUZZ_PREFIX = 'fuzz-'
TEXT_EXTENSIONS = [
    '.css', '.js', '.htm', '.html', '.svg', '.xhtml', '.xht', '.xml', '.xsl'
]
URL_REQUEST_RETRIES = 5
URL_REQUEST_FAIL_WAIT = 1
WINDOWS_PREFIX_PATH = '\\\\?\\'

# Thread pool for use in function timeouts.
THREAD_POOL = None

LOCAL_SOURCE_MANIFEST = os.path.join('src', 'appengine', 'resources',
                                     'clusterfuzz-source.manifest')


def utcnow():
  """Return datetime.datetime.utcnow(). We need this method because we can't
    mock built-in methods."""
  return datetime.datetime.utcnow()  # pragma: no cover.


def current_date_time():
  """Returns current date and time."""
  return datetime.datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC')


def utc_date_to_timestamp(date):
  """Converts a (UTC) datetime.date to a UNIX timestamp."""
  return (date - datetime.date(1970, 1, 1)).total_seconds()


def utc_datetime_to_timestamp(dt):
  """Converts a (UTC) datetime.date to a UNIX timestamp."""
  return (dt - datetime.datetime.utcfromtimestamp(0)).total_seconds()


def decode_to_unicode(obj, encoding='utf-8'):
  """Decode object to unicode encoding."""
  if isinstance(obj, basestring) and not isinstance(obj, unicode):
    try:
      obj = unicode(obj, encoding)
    except:
      obj = unicode(''.join(char for char in obj if ord(char) < 128), encoding)

  return obj


@retry.wrap(
    retries=URL_REQUEST_RETRIES,
    delay=URL_REQUEST_FAIL_WAIT,
    function='base.utils.fetch_url')
def fetch_url(url):
  """Fetch url content."""
  operations_timeout = environment.get_value('URL_BLOCKING_OPERATIONS_TIMEOUT')

  response = requests.get(url, timeout=operations_timeout)
  if response.status_code == 404:
    return None

  response.raise_for_status()
  return response.text


def fields_match(string_1,
                 string_2,
                 field_separator=':',
                 allow_empty_fields=True):
  """Match fields of two strings, separated by a |field_separator|. Empty fields
  can be ignored via |allow_empty_fields| flag."""
  if string_1 is None or string_2 is None:
    return False
  if string_1 == string_2:
    return True

  string_1_fields = string_1.split(field_separator)
  string_2_fields = string_2.split(field_separator)

  if not allow_empty_fields and len(string_1_fields) != len(string_2_fields):
    return False

  min_fields_length = min(len(string_1_fields), len(string_2_fields))
  for i in range(min_fields_length):
    if string_1_fields[i] != string_2_fields[i]:
      return False

  return True


def file_path_to_file_url(path):
  """Return a path as a file scheme url."""
  if not path:
    return ''

  path = path.lstrip(WINDOWS_PREFIX_PATH)
  # TODO(mbarbella): urljoin has several type checks for arguments. Ensure that
  # we're passing newstr on both sides while migrating to Python 3. After
  # migrating, ensure that callers pass strs to avoid this hack.
  return urllib.parse.urljoin(
      str(u'file:'), str(urllib.request.pathname2url(path)))


def filter_file_list(file_list):
  """Filters file list by removing duplicates, non-existent files
  and directories."""
  filtered_file_list = []
  for file_path in file_list:
    if not os.path.exists(file_path):
      continue

    if os.path.isdir(file_path):
      continue

    # Do a os specific case normalization before comparison.
    if (os.path.normcase(file_path) in list(
        map(os.path.normcase, filtered_file_list))):
      continue

    filtered_file_list.append(file_path)

  if len(filtered_file_list) != len(file_list):
    logs.log('Filtered file list (%s) from (%s).' % (str(filtered_file_list),
                                                     str(file_list)))

  return filtered_file_list<commit_msg>Use newstr in file_path_to_file_url. (#1188)<commit_after># Copyright 2019 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Common utility functions."""

from builtins import range
from future import standard_library
standard_library.install_aliases()
from past.builtins import basestring
import ast
import datetime
import functools
import gc
import hashlib
import inspect
import os
import random
import requests
import sys
import time
import urllib.parse
import urllib.request
import weakref

from base import errors
from base import memoize
from base import retry
from config import local_config
from metrics import logs
from system import environment

try:
  import psutil
except ImportError:
  psutil = None

# FIXME: Binary extensions list is still very basic.
BINARY_EXTENSIONS = [
    # Media formats.
    '.mp3',
    '.ogg',
    '.mp4',
    '.webm',
    # Image Formats.
    '.png',
    '.jpg',
    '.gif',
    # Misc.
    '.pdf',
    '.swf',
]
FUZZ_PREFIX = 'fuzz-'
TEXT_EXTENSIONS = [
    '.css', '.js', '.htm', '.html', '.svg', '.xhtml', '.xht', '.xml', '.xsl'
]
URL_REQUEST_RETRIES = 5
URL_REQUEST_FAIL_WAIT = 1
WINDOWS_PREFIX_PATH = '\\\\?\\'

# Thread pool for use in function timeouts.
THREAD_POOL = None

LOCAL_SOURCE_MANIFEST = os.path.join('src', 'appengine', 'resources',
                                     'clusterfuzz-source.manifest')


def utcnow():
  """Return datetime.datetime.utcnow(). We need this method because we can't
    mock built-in methods."""
  return datetime.datetime.utcnow()  # pragma: no cover.


def current_date_time():
  """Returns current date and time."""
  return datetime.datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC')


def utc_date_to_timestamp(date):
  """Converts a (UTC) datetime.date to a UNIX timestamp."""
  return (date - datetime.date(1970, 1, 1)).total_seconds()


def utc_datetime_to_timestamp(dt):
  """Converts a (UTC) datetime.date to a UNIX timestamp."""
  return (dt - datetime.datetime.utcfromtimestamp(0)).total_seconds()


def decode_to_unicode(obj, encoding='utf-8'):
  """Decode object to unicode encoding."""
  if isinstance(obj, basestring) and not isinstance(obj, unicode):
    try:
      obj = unicode(obj, encoding)
    except:
      obj = unicode(''.join(char for char in obj if ord(char) < 128), encoding)

  return obj


@retry.wrap(
    retries=URL_REQUEST_RETRIES,
    delay=URL_REQUEST_FAIL_WAIT,
    function='base.utils.fetch_url')
def fetch_url(url):
  """Fetch url content."""
  operations_timeout = environment.get_value('URL_BLOCKING_OPERATIONS_TIMEOUT')

  response = requests.get(url, timeout=operations_timeout)
  if response.status_code == 404:
    return None

  response.raise_for_status()
  return response.text


def fields_match(string_1,
                 string_2,
                 field_separator=':',
                 allow_empty_fields=True):
  """Match fields of two strings, separated by a |field_separator|. Empty fields
  can be ignored via |allow_empty_fields| flag."""
  if string_1 is None or string_2 is None:
    return False
  if string_1 == string_2:
    return True

  string_1_fields = string_1.split(field_separator)
  string_2_fields = string_2.split(field_separator)

  if not allow_empty_fields and len(string_1_fields) != len(string_2_fields):
    return False

  min_fields_length = min(len(string_1_fields), len(string_2_fields))
  for i in range(min_fields_length):
    if string_1_fields[i] != string_2_fields[i]:
      return False

  return True


def file_path_to_file_url(path):
  """Return a path as a file scheme url."""
  if not path:
    return ''

  path = path.lstrip(WINDOWS_PREFIX_PATH)
  # TODO(mbarbella): urljoin has several type checks for arguments. Ensure that
  # we're passing newstr on both sides while migrating to Python 3. After
  # migrating, ensure that callers pass strs to avoid this hack.
  from builtins import str
  return urllib.parse.urljoin(
      str(u'file:'), str(urllib.request.pathname2url(path)))


def filter_file_list(file_list):
  """Filters file list by removing duplicates, non-existent files
  and directories."""
  filtered_file_list = []
  for file_path in file_list:
    if not os.path.exists(file_path):
      continue

    if os.path.isdir(file_path):
      continue

    # Do a os specific case normalization before comparison.
    if (os.path.normcase(file_path) in list(
        map(os.path.normcase, filtered_file_list))):
      continue

    filtered_file_list.append(file_path)

  if len(filtered_file_list) != len(file_list):
    logs.log('Filtered file list (%s) from (%s).' % (str(filtered_file_list),
                                                     str(file_list)))

  return filtered_file_list
{'fraction_non_alphanumeric': 0.10458148076551324, 'fraction_numerical': 0.008892325536439203, 'mean_word_length': 3.6608108108108106, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6685448', 'n_tokens_mistral': 3262, 'n_tokens_neox': 3117, 'n_words': 991}