/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.underfs.web;

import alluxio.AlluxioURI;
import alluxio.conf.PropertyKey;
import alluxio.underfs.ConsistentUnderFileSystem;
import alluxio.underfs.UfsDirectoryStatus;
import alluxio.underfs.UfsFileStatus;
import alluxio.underfs.UfsStatus;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.UnderFileSystemConfiguration;
import alluxio.underfs.options.CreateOptions;
import alluxio.underfs.options.DeleteOptions;
import alluxio.underfs.options.FileLocationOptions;
import alluxio.underfs.options.MkdirsOptions;
import alluxio.underfs.options.OpenOptions;
import alluxio.util.UnderFileSystemUtils;
import alluxio.util.network.HttpUtils;

import com.google.common.io.ByteStreams;
import org.apache.commons.httpclient.Header;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import javax.annotation.concurrent.ThreadSafe;

/**
 * Web {@link UnderFileSystem} implementation.
 */
@ThreadSafe
public class WebUnderFileSystem extends ConsistentUnderFileSystem {
  private static final Logger LOG = LoggerFactory.getLogger(WebUnderFileSystem.class);
  private int mTimeout;
  private final String mUnsupportedMsg = "Unsupported operation for WebUnderFileSystem.";

  /**
   * Constructs a new {@link WebUnderFileSystem}.
   *
   * @param uri the {@link AlluxioURI} for this UFS
   * @param ufsConf UFS configuration
   */
  public WebUnderFileSystem(AlluxioURI uri, UnderFileSystemConfiguration ufsConf) {
    super(uri, ufsConf);
    mTimeout = (int) mUfsConf.getMs(PropertyKey.UNDERFS_WEB_CONNECTION_TIMEOUT);
  }

  @Override
  public String getUnderFSType() {
    return "web";
  }

  @Override
  public void cleanup() throws IOException {
    throw new IOException(mUnsupportedMsg);
  }

  @Override
  public void close() throws IOException {
    throw new IOException(mUnsupportedMsg);
  }

  @Override
  public OutputStream create(String path, CreateOptions options) throws IOException {
    throw new IOException(mUnsupportedMsg);
  }

  @Override
  public boolean deleteDirectory(String path, DeleteOptions options) throws IOException {
    throw new IOException(mUnsupportedMsg);
  }

  @Override
  public boolean deleteFile(String path) throws IOException {
    throw new IOException(mUnsupportedMsg);
  }

  @Override
  public boolean exists(String path) throws IOException {
    return HttpUtils.head(path, mTimeout) != null;
  }

  @Override
  public long getBlockSizeByte(String path) throws IOException {
    if (!exists(path)) {
      throw new FileNotFoundException(path);
    }

    return mUfsConf.getBytes(PropertyKey.USER_BLOCK_SIZE_BYTES_DEFAULT);
  }

  @Override
  public UfsDirectoryStatus getDirectoryStatus(String path) throws IOException {
    UfsStatus ufsStatus = getStatus(path);
    if (ufsStatus instanceof UfsDirectoryStatus) {
      return (UfsDirectoryStatus) ufsStatus;
    }

    throw new IOException("Failed to getDirectoryStatus: " + path);
  }

  @Override
  public List<String> getFileLocations(String path) throws IOException {
    return new ArrayList<>();
  }

  @Override
  public List<String> getFileLocations(String path, FileLocationOptions options)
      throws IOException {
    return getFileLocations(path);
  }

  @Override
  public UfsFileStatus getFileStatus(String path) throws IOException {
    UfsStatus ufsStatus = getStatus(path);
    if (ufsStatus instanceof UfsFileStatus) {
      return (UfsFileStatus) ufsStatus;
    }
    throw new IOException("Failed to getFileStatus: " + path);
  }

  // This call is currently only used for the web ui, where a negative value implies unknown.
  @Override
  public long getSpace(String path, SpaceType type) {
    return -1;
  }

  /**
   * Get the file status of a http url.
   *
   * @param path the http url
   * @param fileName the file name
   * @return a UfsStatus object related to the http url
   */
  private UfsStatus getStatus(String path, String fileName) throws IOException {
    long contentLength = 0;
    long lastModified = new Date().getTime();
    Header[] headers = HttpUtils.head(path, mTimeout);
    if (headers == null) {
      throw new IOException("Failed to getStatus: " + path);
    }

    for (Header header : headers) {
      String headerName = header.getName();
      if (headerName.equalsIgnoreCase("Content-Length")) {
        contentLength = Long.parseLong(header.getValue());
      } else if (headerName.equalsIgnoreCase("Last-Modified")) {
        lastModified = parseTimestamp(header.getValue(),
            mUfsConf.get(PropertyKey.UNDERFS_WEB_HEADER_LAST_MODIFIED));
      }
    }

    if (isFile(path)) {
      // Return file status.
      String contentHash = UnderFileSystemUtils.approximateContentHash(contentLength, lastModified);
      return new UfsFileStatus(fileName == null ? path : fileName, contentHash, contentLength,
          lastModified, "", "", (short) 288,
          mUfsConf.getBytes(PropertyKey.USER_BLOCK_SIZE_BYTES_DEFAULT));
    }
    // Return directory status.
    return new UfsDirectoryStatus(fileName == null ? path : fileName, "", "", (short) 800,
        lastModified);
  }

  @Override
  public UfsStatus getStatus(String path) throws IOException {
    return getStatus(path, null);
  }

  @Override
  public boolean isDirectory(String path) throws IOException {
    String contentType = "";
    Header[] headers = HttpUtils.head(path, mTimeout);
    if (headers == null) {
      return false;
    }

    for (Header header : headers) {
      if (header.getName().equalsIgnoreCase("Content-Type")) {
        contentType = header.getValue();
        break;
      }
    }

    if (contentType.contains("text/html")) {
      Elements titleElements = Jsoup.connect(path).get().select("title");
      if (titleElements.size() > 0) {
        String title = titleElements.get(0).text();
        List<String> titles = mUfsConf.getList(PropertyKey.UNDERFS_WEB_TITLES, ",");
        for (final String t : titles) {
          if (title.contains(t)) {
            return true;
          }
        }
      }
    }
    return false;
  }

  @Override
  public boolean isFile(String path) throws IOException {
    return !isDirectory(path);
  }

  /**
   * Parse the string to a unix timestamp.
   *
   * @param datetime a date time string
   * @param format the pattern related to the datetime string
   * @return the unixstamp for the datetime string
   */
  private long parseTimestamp(String datetime, String format) {
    long timestamp = new Date().getTime();
    if (datetime.length() > 0 && datetime.compareTo("-") != 0) {
      try {
        SimpleDateFormat dateFormat = new SimpleDateFormat(format);
        Date parsedDate = dateFormat.parse(datetime);
        timestamp = parsedDate.getTime();
      } catch (Exception e) {
        LOG.error("Failed to format {} to a timestamp, {}", datetime, e.getMessage());
      }
    }
    return timestamp;
  }

  @Override
  public UfsStatus[] listStatus(String path) throws IOException {
    Document doc;
    try {
      doc = Jsoup.connect(path).get();
    } catch (Exception e) {
      LOG.error("Failed to get content from URL {}, {}", path, e.getMessage());
      return null;
    }

    Elements bodyElements = doc.select("body");
    if (bodyElements.size() == 0) {
      return null;
    }

    Element bodyElement = bodyElements.first();
    Elements bodyChildNodes = bodyElement.children();
    Elements listElements = null;
    for (Element bodyChildNode : bodyChildNodes) {
      listElements = bodyChildNode.select("a");
      if (listElements.size() > 0) {
        break;
      }
    }

    if (listElements == null || listElements.size() == 0) {
      return null;
    }

    List<String> parentNames = mUfsConf.getList(PropertyKey.UNDERFS_WEB_PARENT_NAMES, ",");
    int flagIndex = -1;
    for (int i = 0; flagIndex == -1 && i < listElements.size(); i++) {
      for (final String flag : parentNames) {
        if (listElements.get(i).text().equalsIgnoreCase(flag)) {
          flagIndex = i;
          break;
        }
      }
    }

    List<UfsStatus> statusList = new ArrayList<>();
    for (int i = (flagIndex == -1 ? 0 : flagIndex); i < listElements.size(); i++) {
      Element listElement = listElements.get(i);
      String href = listElement.attr("href");
      String fileName = listElement.text();
      if (!new WebUnderFileSystemFactory().supportsPath(href)) {
        href = path + "/" + href;
      }

      try {
        UfsStatus ufsStatus = getStatus(href, fileName);
        statusList.add(ufsStatus);
      } catch (IOException e) {
        LOG.error("Failed to get status for url: {}", href, e);
      }
    }

    UfsStatus[] rtn = new UfsStatus[statusList.size()];
    return statusList.toArray(rtn);
  }

  @Override
  public boolean mkdirs(String path, MkdirsOptions options) throws IOException {
    throw new IOException(mUnsupportedMsg);
  }

  @Override
  public InputStream open(String path, OpenOptions options) throws IOException {
    InputStream inputStream = HttpUtils.getInputStream(path, mTimeout);
    try {
      ByteStreams.skipFully(inputStream, options.getOffset());
    } catch (IOException e) {
      inputStream.close();
      throw e;
    }
    return inputStream;
  }

  @Override
  public boolean renameDirectory(String src, String dst) throws IOException {
    throw new IOException(mUnsupportedMsg);
  }

  @Override
  public boolean renameFile(String src, String dst) throws IOException {
    throw new IOException(mUnsupportedMsg);
  }

  @Override
  public void setOwner(String path, String user, String group) throws IOException {
    throw new IOException(mUnsupportedMsg);
  }

  @Override
  public void setMode(String path, short mode) throws IOException {
    throw new IOException(mUnsupportedMsg);
  }

  @Override
  public void connectFromMaster(String hostname) throws IOException {
    throw new IOException(mUnsupportedMsg);
  }

  @Override
  public void connectFromWorker(String hostname) throws IOException {
    throw new IOException(mUnsupportedMsg);
  }

  @Override
  public boolean supportsFlush() throws IOException {
    throw new IOException(mUnsupportedMsg);
  }
}
