import {
  createReadStream, createWriteStream, existsSync, mkdirSync, rmSync,
} from 'fs';
import { basename, join, extname } from 'path';
import es from 'event-stream';
import moment from 'moment';
import {
  print, printError, printSuccess, PrintTime,
} from '../utils/print';
import {
  csvStr2arr, isCsvRow,
} from '../utils/csv';
import { getProgressBarInstance, promiseQueue } from '../utils/common';
import { getFileRowsNumber } from '../utils/shell';
import { getFileStats } from '../utils/file';

declare module 'event-stream'{
  export function through(fn1: (() => void) | null, fn2: (() => void)): MapStream;
}

export interface HandleCsv {
  inputFilePath: string;
  successOutputFilePath: string;
  errorOutputFilePath: string;
  columnCount: number;
  dateIndexes?: number[];
}

export interface HandleCsvs {
  outputDirPath: string;
  files: {
    filePath: string;
    columnCount: number;
    dateIndexes?: number[];
  }[];
}

export async function handleCsv({
  inputFilePath,
  successOutputFilePath,
  errorOutputFilePath,
  columnCount,
  dateIndexes = [],
}: HandleCsv): Promise<void> {
  const parseTimeStr = `[${basename(inputFilePath)}] 解析耗时`;
  PrintTime.start(parseTimeStr);
  let successCount = 0;
  let errorCount = 0;

  const total = await getFileRowsNumber(inputFilePath);
  const readStream = createReadStream(inputFilePath, { encoding: 'utf-8' });
  const succWriteStream = createWriteStream(successOutputFilePath);
  const errWriteStream = createWriteStream(errorOutputFilePath);
  const bar = getProgressBarInstance(`${basename(inputFilePath)}: `, total);

  const removeEmptyFile = () => {
    if (successCount === 0) {
      rmSync(successOutputFilePath);
    }
    if (errorCount === 0) {
      rmSync(errorOutputFilePath);
    }
  };

  const handleComplete = async () => {
    removeEmptyFile();
    print('');
    print(`[${basename(inputFilePath)}] 解析完成，文件大小 [${(await getFileStats(inputFilePath)).size}]`);
    print(`[${basename(inputFilePath)}] 解析完成，共计 [${successCount + errorCount}] 行`);
    PrintTime.end(parseTimeStr);
    successCount && printSuccess(`正确 [${successCount}] 行，已输出到 [${successOutputFilePath}]`);
    errorCount && printError(`错误 [${errorCount}] 行，已经输出到 [${errorOutputFilePath}]`);
    print('-----------------------------------------------------');
    print('');
  };

  function writeSucc(str: string): Promise<void> {
    return new Promise((resolve, reject) => {
      succWriteStream.write(str + '\n', () => {
        resolve();
      });
    });
  }

  function writeErr(str: string): Promise<void> {
    return new Promise((resolve, reject) => {
      errWriteStream.write(str + '\n', () => {
        resolve();
      });
    });
  }

  async function handleLine(line: string, cb: (unknown: unknown, line: string) => void) {
    if (line === '') {
      cb(null, line);
      return;
    }
    let str = '';
    let isCsv = false;

    // 如果需要标准化时间，则判断和转换解析的数据
    if (dateIndexes.length > 0) {
      const csvArr = await csvStr2arr(line);
      const [rowArr] = csvArr;

      if (rowArr?.length === columnCount) {
        dateIndexes.forEach((dateIndex) => {
          const dateValue = rowArr[dateIndex];

          const m = moment(dateValue);
          if (m.isValid()) {
            isCsv = true;
            rowArr[dateIndex] = m.format('YYYY-MM-DD HH:mm:ss');
          } else {
            isCsv = false;
          }
        });
        str = rowArr.join(',');
      } else {
        isCsv = false;
      }
    } else {
      isCsv = await isCsvRow(line, columnCount);
      str = line;
    }

    // 写文件
    if (isCsv) {
      successCount += 1;
      await writeSucc(str);
    } else {
      errorCount += 1;
      await writeErr(line);
    }

    bar.tick();
    cb(null, line);
  }

  return new Promise((resolve) => {
    readStream
      .pipe(es.split())
      .pipe(es.map((handleLine)))
      .pipe(es.through(null, () => {
        handleComplete();
        resolve();
      }));
  });
}

export function handleCsvs(conf: HandleCsvs): void {
  const {
    outputDirPath,
    files,
  } = conf;
  promiseQueue(files.map((file) => {
    const { filePath, columnCount, dateIndexes } = file;
    const ext = extname(filePath);
    const plainFilePath = basename(filePath).replace(ext, '');

    const successOutputDir = join(outputDirPath, 'success');
    const errorOutputDir = join(outputDirPath, 'error');

    if (!existsSync(successOutputDir)) {
      mkdirSync(successOutputDir);
    }
    if (!existsSync(errorOutputDir)) {
      mkdirSync(errorOutputDir);
    }

    return () => handleCsv({
      inputFilePath: filePath,
      successOutputFilePath: join(successOutputDir, `${plainFilePath}${ext}`),
      errorOutputFilePath: join(errorOutputDir, `${plainFilePath}${ext}`),
      columnCount,
      dateIndexes,
    });
  }));
}