
using CsvCount_ES.ResultModel;
using Newtonsoft.Json;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.ComponentModel;
using System.Configuration;
using System.Data;
using System.Diagnostics;
using System.Drawing;
using System.IO;
using System.IO.MemoryMappedFiles;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Windows.Forms;

namespace CsvCount_ES
{
    public partial class Form1 : Form
    {

        #region 初始化、变量

        //key:newsid字段，也是itemid字段,value:int-次数累计
        private ConcurrentDictionary<string, int> totalCountDic = new ConcurrentDictionary<string, int>();

        //链式结构 key: Int64 累加(对应分块的块号)， value：这一块的ListModel的第一行和最后一行的string内容
        private ConcurrentDictionary<Int64, ListModel> listModelDic = new ConcurrentDictionary<Int64, ListModel>();

        //放在队列里面慢慢输出
        private ConcurrentQueue<ResultItem> ResultQueue = new ConcurrentQueue<ResultItem>();

        /// <summary>
        /// 消费：弹出后放入线程池中消费
        /// </summary>
        /// <remarks></remarks>
        private static ThreadPool _threadPool;

        private Stopwatch readfilesw = new Stopwatch();//读取并处理文件的执行时间。

        private Stopwatch essw = new Stopwatch();//统计全部的es调用执行时间。

        private static Int64 sn = 0;
        //警告 在 32 位 Intel 计算机上分配 64 位值不是原子操作；即该操作不是线程安全的。这意味着，如果两个人同时将一个值分配给一个静态 Int64 字段，则该字段的最终值是无法预测的。

        /// <summary>
        /// 得到编号加1的值
        /// </summary>
        /// <returns></returns>
        public static Int64 ADD()
        {
            //新建线程，线程数加一  
            return Interlocked.Increment(ref sn);
        }

        int errorCount = 0;
        int errorEsCount = 0;
        string fileName = "";
        DateTime begindt = new DateTime();

        public Form1()
        {
            InitializeComponent();
            this.listBox1.HorizontalScrollbar = true;
            this.openFileDialog1.InitialDirectory = Environment.GetFolderPath(Environment.SpecialFolder.DesktopDirectory);
            this.openFileDialog1.Filter = "数据文件|*.csv";
           

        }

        #endregion

        private void button1_Click(object sender, EventArgs e)
        {
            if (openFileDialog1.ShowDialog() == DialogResult.OK)
            {
                textBox1.Text = openFileDialog1.FileName;
                button2.Enabled = true;
                listBox1.Items.Clear();

                errorCount = 0;
                errorEsCount = 0;
                fileName = "";
                textBox_path.Text = "";
                sn = 0;
                while (ResultQueue.Count > 0)
                {
                    ResultItem ritem;
                    ResultQueue.TryDequeue(out ritem);
                }
                listModelDic.Clear();
                totalCountDic.Clear();

                label_wait.Text = "尚未开始执行！";
            }
        }

        private void button2_Click(object sender, EventArgs e)
        {

            if (!String.IsNullOrWhiteSpace(textBox1.Text))
            {

                Int64 length = 10240;//每次块大小 512000byte  500KB  这样1个G转2010次算完
                                     //10兆字节(mb)=10485760字节(b)
                                     //UTF-8编码中，一个英文字符等于一个字节，一个中文（含繁体）等于三个字节。 本次文件中主要是英文和数字，没有中文。所以要多少字就配多大块。
                                     //50千字节(kb)=51200字节(b)
                                     //200千字节(kb)=204800字节(b)
                if (String.IsNullOrWhiteSpace(textBox_block.Text) || !Int64.TryParse(textBox_block.Text, out length) || length < 1)
                {

                    MessageBox.Show("请正确输入'分块大小'(B)");
                    return;

                }
                int tempint = 0;
                if (String.IsNullOrWhiteSpace(textBox_CPU.Text) || !int.TryParse(textBox_CPU.Text, out tempint) || tempint < 1)
                {

                    MessageBox.Show("请正确输入'单核CPU利用线程数'");
                    return;

                }
                if (String.IsNullOrWhiteSpace(textBox_output.Text) || !int.TryParse(textBox_output.Text, out tempint) || tempint < 1)
                {

                    MessageBox.Show("请正确输入'输出文件数'");
                    return;

                }
                begindt = DateTime.Now;
                this.listBox1.Items.Add("开始执行(" + begindt.ToString("yyyy-MM-dd HH:mm:ss.fff") + ")：" + textBox1.Text.ToString());

                this.listBox1.Items.Add("1.1开始： 读取并统计行数。");
                label_wait.Text = "请勿关闭,正在执行分析,耐心等待........！";

                button1.Enabled = false;
                textBox1.Enabled = false;
                button2.Enabled = false;
                textBox_block.Enabled = false;
                textBox_CPU.Enabled = false;
                textBox_output.Enabled = false;

                listBox1.Enabled = true;

                fileName = Path.GetFileNameWithoutExtension(textBox1.Text);

                Application.DoEvents();

                #region 执行内存映射方案

                string filepath = textBox1.Text;
                setup:
                try
                {
                    #region  并发用法：真正大文件时比StreamReader有用--820W行数据，439MB，转成字符串、split后，执行消耗7~8秒。 10K 为1块。

                    //读取文件
                    FileInfo fi = new FileInfo(filepath);
                    Int64 size = fi.Length;//byte长度


                    Int64 portion = (Int64)Math.Ceiling(size * 1.0 / length); //分成多少块 
                    this.listBox1.Items.Add("1.2初始化分块：单块" + length + "B，文件共分为" + portion + "块。");
                    this.listBox1.Items.Add("1.3处理中：并行分块处理此文件…………");
                    //string fp2 = @filepath;
                    readfilesw.Start();
                    //统一使用总文件大小的内存隐射对象
                    using (MemoryMappedFile mmf = MemoryMappedFile.CreateFromFile(filepath, FileMode.Open, "mapname", size))
                    {
                        //从零开始，每块单独并行处理。
                        Parallel.For(0, portion, (index, ParallelLoopState) =>
                          {
                              //块大小和 文件总大小-第几分区位置的剩余位置，取两者较小的
                              Int64 fileSize = Math.Min(length, size - length * index);
                              if (fileSize > 0)
                              {

                                  using (var accessor = mmf.CreateViewAccessor(length * index, fileSize, MemoryMappedFileAccess.Read))
                                  {
                                      List<string> list = new List<string>();



                                      byte[] byteArray = new byte[length];

                                      #region 函数参数说明
                                      //ReadArray参数
                                      //position
                                      //Type: System.Int64
                                      //访问器中的字节偏移量，从此处开始读取。
                                      //array
                                      //Type: T[]
                                      //包含从访问器读取的结构的数组。
                                      //offset
                                      //Type: System.Int32
                                      //array 中要将第一个复制的结构放置到的索引。
                                      //count
                                      //Type: System.Int32
                                      //要从访问器读取的 T 类型的结构的数目。

                                      //resultcount:读入 array 的结构数。 如果可用结构较少，则此值可能小于 count；如果到达访问器末尾，则为零。
                                      #endregion
                                      int resultcount = accessor.ReadArray(0, byteArray, 0, (int)fileSize);

                                      //测试通过：\r\n \r 空格 \n（注意：\r 空格 \n 都算1个字符,\r\n算2个字符）
                                      //规则：最后必须是\n结尾,cvs的换行都是\n结尾。
                                      string str = System.Text.Encoding.Default.GetString(byteArray);
                                      int index1 = str.LastIndexOf('\n');
                                      string[] k = str.Split('\n');
                                      string endstr = null;
                                      if (str.Length != index1)
                                      {
                                          //不相等，最后一行不能Split分析。
                                          //spilt时，将最后一行保存起来，本次处理索引-1。
                                          //头条保存起来，也不处理，本次处理索引-2。
                                          endstr = k[k.Count() - 1].Replace("\0", string.Empty).Replace("\r", string.Empty).Trim(); ;


                                      }
                                      //else
                                      //{
                                      //    //最后一行可以Split分析，不再处理
                                      //}

                                      //先简化开发，将头部和尾部都另外保存。只计算中间的。然后调用ES (索引-2：从1开始，尾部-1)
                                      //测试: 820W条记录，执行完毕 20~ 22秒。 最终结果3w2013条记录。有的直接点击3000多次。

                                      for (int j = 1; j < k.Count() - 1; j++)
                                      {

                                          try
                                          {
                                              ADD();

                                              string key = k[j].Split(',')[1].Replace("\0", string.Empty).Replace("\r", string.Empty).Trim();
                                              totalCountDic.AddOrUpdate(key, 1, (skey, scount) => { return scount = scount + 1; });

                                          }
                                          catch (Exception ext)
                                          {
                                              MessageBox.Show("某行数据第2列可能为空" + ext.ToString());

                                          }

                                      }

                                      //头尾：另外保存 
                                      ListModel listmodel = new ListModel(k[0].Replace("\0", string.Empty).Replace("\r", string.Empty).Trim(), endstr);
                                      listModelDic.TryAdd(index, listmodel);




                                      //}

                                  }
                                  //分区时，最后一个为0则忽略。

                              }

                          });

                    }

                    if (listModelDic.Count() > 0)
                    {
                        if (listModelDic.Count() == 1)
                        {

                            try
                            {
                                string key = listModelDic[0].End;
                                if (!String.IsNullOrWhiteSpace(key))
                                {
                                    ADD();

                                    totalCountDic.AddOrUpdate(listModelDic[0].End.Split(',')[1], 0, (skey, scount) => { return scount = scount + 1; });
                                }
                            }
                            catch (Exception ext)
                            {
                                MessageBox.Show("只有一块时（文件小）：第一行数据第2列可能为空" + ext.ToString());

                            }
                        }
                        else
                        {
                            //头尾单独链式处理：  从1开始。第0行的begin 是列名称，不需要分析。 
                            for (int i = 1; i < listModelDic.Count(); i++)
                            {
                                try
                                {
                                    //count = count + 1;//记录计算了多少次，与分块对应（其实就是与listModelDic的key-1对应即代表完整）

                                    string str = "";
                                    if (!String.IsNullOrWhiteSpace(listModelDic[i - 1].End))
                                    {
                                        //上一个的end+本轮的begin拼接，不需要去除/n
                                        str = listModelDic[i - 1].End + listModelDic[i].Begin;
                                    }
                                    else
                                    {
                                        //上一个的end（为空）+本轮的begin拼接，不需要去除/n
                                        str = listModelDic[i].Begin;

                                    }

                                    ADD();
                                    string key = str.Split(',')[1];
                                    totalCountDic.AddOrUpdate(key, 0, (skey, scount) => { return scount = scount + 1; });

                                }
                                catch (Exception ext)
                                {
                                    MessageBox.Show("头尾单独链式处理：某行数据第2列可能为空" + ext.ToString());

                                }
                            }


                            //最后补充最后一条的end 是没有换行了。所以不需要处理。肯定为null。 keyend.end 是空字符串。
                            //object keyend = listModelDic[listModelDic.Count()];

                        }
                    }
                    readfilesw.Stop();
                    this.listBox1.Items.Add("1.4处理完毕：");
                    this.listBox1.Items.Add("     处理行数：共" + sn + "行！与文件行数相同,标题列1行不算。");
                    this.listBox1.Items.Add("     处理完成用时：" + readfilesw.ElapsedMilliseconds + "毫秒！");
                    this.listBox1.Items.Add("     统计行数后：共" + totalCountDic.Count() + "行！");


                    begindt = DateTime.Now;

                    this.listBox1.Items.Add("2.1开始(" + DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ")： 并行调用ES，返回内容。");
                    //2.初始化4个线程。对应市场常用电脑配置。
                    int ThreadCount = 4;
                    int CPU = 2;//如果没有使用。也就是创建了1个8个线程的线程池而已。

                   
                    if (String.IsNullOrWhiteSpace(textBox_CPU.Text) || !int.TryParse(textBox_CPU.Text, out CPU))
                    {

                    }

                    ThreadCount = CPU * Environment.ProcessorCount;

                    try
                    {
                        //线程数量、线程名称
                        _threadPool = new ThreadPool(ThreadCount, "taikor");
                    }
                    catch (Exception ex)
                    {

                        //  throw;
                    }

                    this.listBox1.Items.Add("2.2初始化线程数：");
                    this.listBox1.Items.Add("     共" + ThreadCount + "个线程并读取执行！");
                    this.listBox1.Items.Add("     CPU单核心并行读取ES线程数：" + CPU + "个线程;CPU核心数：" + Environment.ProcessorCount);

                    this.listBox1.Items.Add("2.3处理中：并行调用ES………………注意本机CPU，适当增减单个CPU线程数");


                    //全部入线程池队列。
                    //totalCountDic去重了。所以必须等待去重完成后再for。
                    foreach (var temp_item in totalCountDic)
                    {
                        //入队列，工作函数和（函数的参数）数组都传参，回调函数为空。
                        _threadPool.EnqueueWorkItem(ES_Complete, temp_item);

                    }
                    essw.Start();


                    #endregion
                }
                catch (IOException ex)
                {
                    errorCount = errorCount + 1;
                    if (errorCount > 2)
                    {
                        MessageBox.Show(ex.ToString());
                        errorCount = 0;
                    }
                    else
                    {
                        //遇到csv无法文件无法使用时，通过修改扩展名实现。
                        filepath = Path.ChangeExtension(filepath, ".txt");
                        //个别CSV\ppt\word等文件类型，则必须强转为任何其他类型的文本后缀结构。否则MemoryMappedFile不识别.
                        goto setup;
                    }
                }
                catch (Exception ex)
                {

                    MessageBox.Show(ex.ToString());
                }

                #endregion


            }
            else
            {
                MessageBox.Show("请选择文件夹！");
            }




        }


     

        /// <summary>
        /// 
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="CountItem"></param>
        private void ES_Complete(KeyValuePair<string, int> CountItem)
        {

            try
            {
                Item item_obj = ESItemAccess.SearchByItemID(CountItem.Key);
                ResultItem result = new ResultItem();
                //如果为空，也要返回itemid 和 点击次数。
                result.ItemID = CountItem.Key;
                result.ClickCount = CountItem.Value;
                if (item_obj != null)
                {
                    #region 能从ES中查找到值

                    result.ItemID = item_obj.ItemID;
                    result.Url = item_obj.Url;
                    result.Url = item_obj.Url;
                    result.CleanTitle = item_obj.CleanTitle;
                    result.CleanText = item_obj.CleanText;
                    result.PubDate = item_obj.PubDate == null ? "" : item_obj.PubDate.ToString("yyyy-MM-dd HH:mm:ss.fff");
                    result.MediaName = item_obj.MediaName;
                    result.DuplicationID = item_obj.DuplicationID;


                    //赋值  ResultItemAnalyzeData
                    ResultItemAnalyzeData newdata = new ResultItemAnalyzeData();
                    foreach (ItemAnalyzeData adata in item_obj.AnalyzeData)
                    {
                        if ((!String.IsNullOrWhiteSpace(adata.IssueID)) &&
                            (adata.IssueID.Equals("Common") || adata.IssueID.Equals("Reader")))
                        {

                            //本次有值
                            if (null != adata.StockIDs)
                            {
                                //内容不需叠加
                                if (null == newdata.StockIDs)
                                {
                                    newdata.StockIDs = adata.StockIDs;
                                }
                                //else
                                //{
                                //    //内容需叠加 
                                //    newdata.StockIDs = newdata.StockIDs.Union(adata.StockIDs).ToArray<string>(); //剔除重复项 

                                //}
                            }

                            if (null != adata.MarketIDs)
                            {

                                if (null == newdata.MarketIDs)
                                {
                                    newdata.MarketIDs = adata.MarketIDs;
                                }
                                //else
                                //{

                                //    newdata.MarketIDs = newdata.MarketIDs.Union(adata.MarketIDs).ToArray<string>(); //剔除重复项 

                                //}
                            }

                            if (null != adata.BlockIDs)
                            {
                                //内容不需叠加
                                if (null == newdata.BlockIDs)
                                {
                                    newdata.BlockIDs = adata.BlockIDs;
                                }
                                //else
                                //{
                                //    //内容需叠加 
                                //    newdata.BlockIDs = newdata.BlockIDs.Union(adata.BlockIDs).ToArray<string>(); //剔除重复项 

                                //}
                            }

                            if (null != adata.IG)
                            {
                                if (null == newdata.IG)
                                {
                                    newdata.IG = adata.IG;
                                }

                            }
                        }
                    }//end for

                    result.AnalyzeData = newdata;

                    #endregion

                }
                else
                {
                    //找不到ES的数量
                    Interlocked.Increment(ref errorEsCount);

                }

                //将结果写入到队列中
                ResultQueue.Enqueue(result);

                //totalCountDic已经去重了。所以可以count相等。
                if (ResultQueue.Count == totalCountDic.Count)
                {
                    essw.Stop();

                    this.Invoke(new MethodInvoker(() =>
                    {
                        this.listBox1.Items.Add("2.4处理完毕：");
                        this.listBox1.Items.Add("     ES全部处理完成用时：" + essw.ElapsedMilliseconds + "毫秒！");
                        this.listBox1.Items.Add("     ES中找到的条数有：" + (totalCountDic.Count() - errorEsCount) + "条");
                        this.listBox1.Items.Add("     ES中找不到的条数有：" + errorEsCount + "条");

                        this.listBox1.Items.Add("3.1开始(" + DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff") + ")：写入JSON文件：");
                        this.listBox1.Items.Add("3.2初始化文件数量： " + textBox_output.Text + "个。");
                        this.listBox1.Items.Add("3.3并行写入文件中。");
                    }
                    ));

                    Stopwatch writesw = new Stopwatch();
                    writesw.Start();
                    Task[] tasks = new Task[int.Parse(textBox_output.Text)];

                    for (int i = 0; i < int.Parse(textBox_output.Text); i++)
                    {
                        int a = i;//注意：传递引用变量容易内部被更改。
                        tasks[i] = Task.Factory.StartNew(() => writeFile(a));
                    }
                    Task.WaitAll(tasks);
                    writesw.Stop();
                    this.Invoke(new MethodInvoker(() =>
                    {
                        this.listBox1.Items.Add("3.4写入文件完毕：");
                        this.listBox1.Items.Add("     写入文件完成用时：" + writesw.ElapsedMilliseconds + "毫秒！");
                    }));


                    errorCount = 0;
                    while (ResultQueue.Count > 0)
                    {
                        ResultItem ritem;
                        ResultQueue.TryDequeue(out ritem);
                    }
                    listModelDic.Clear();
                    totalCountDic.Clear();

                    this.Invoke(new MethodInvoker(() =>
                    {

                        this.listBox1.Items.Add("完毕：程序执行完成。总用时:" + (DateTime.Now - begindt).TotalMilliseconds + "毫秒");
                        this.label_wait.Text = "此文件已经执行完毕！可以继续选择其他文件！";
                        MessageBox.Show("此文件已经执行完毕！可以继续选择其他文件！");


                        foreach (var item in this.listBox1.Items)
                        {
                            LogService.WriteInfo(item.ToString());

                        }
                        this.textBox_path.Text = Environment.CurrentDirectory + "\\OutPut\\";

                        this.textBox1.Enabled = true;
                        this.button1.Enabled = true;
                        this.textBox_block.Enabled = true;
                        this.textBox_CPU.Enabled = true;
                        this.textBox_output.Enabled = true;

                    }));

                }

            }
            catch (Exception ex)
            {

                MessageBox.Show(ex.ToString());
            }


        }


        /// <summary>
        /// 写入文件
        /// </summary>
        /// <param name="index"></param>
        private void writeFile(int index)
        {
          

            //命名结构：原filename_begintime的时_分_index_。
            string filepath = Environment.CurrentDirectory + "\\OutPut\\" + fileName + "_" + begindt.Hour + "_" + begindt.Minute + "_" + index + ".json";

            //输出到JSON文件中。
            using (FileStream fs = new FileStream(filepath, FileMode.Append))
            {
                using (StreamWriter sw = new StreamWriter(fs))
                {
                    sw.Write("[");
                    while (ResultQueue.Count > 0)
                    {
                        try
                        {
                            ResultItem tempitem = new ResultItem();

                            ResultQueue.TryDequeue(out tempitem);
                            string returnstr;
                            //将result结果放入到集合中。统一转为json字符串，写入文件。
                            if (ResultQueue.Count > 0)
                            {
                                returnstr = JsonConvert.SerializeObject(tempitem) + ",";
                            }
                            else
                            {
                                returnstr = JsonConvert.SerializeObject(tempitem);

                            }
                            sw.Write(returnstr);
                            sw.Flush();


                        }
                        catch (Exception exm)
                        {
                            MessageBox.Show(exm.ToString());
                        }

                    }
                    sw.Write("]");

                }
            }
         

     
          }


        public static long ConvertDataTimeToLong(DateTime dt)
        {
            DateTime dtStart = TimeZone.CurrentTimeZone.ToLocalTime(new DateTime(1970, 1, 1));
            TimeSpan toNow = dt.Subtract(dtStart);
            long timeStamp = toNow.Ticks;
            //timeStamp = long.Parse(timeStamp.ToString().Substring(0, timeStamp.ToString().Length - 4));
            return timeStamp;
        }




        private void listBox1_SelectedIndexChanged(object sender, EventArgs e)
        {

            ListControl ls = (ListControl)sender;

            MessageBox.Show(ls.Text);
        }

        private void GroupRangeTips_Load(object sender, EventArgs e)
        {
            this.listBox1.DrawMode = DrawMode.OwnerDrawVariable;
            this.listBox1.DrawItem += new DrawItemEventHandler(ListBoxGroupRange_DrawItem);
        }

        //自绘Item，使其视觉效果更好
        private void ListBoxGroupRange_DrawItem(object sender, DrawItemEventArgs e)
        {
            e.Graphics.FillRectangle(new SolidBrush(e.BackColor), e.Bounds);
            if (e.Index >= 0)
            {
                StringFormat sStringFormat = new StringFormat();
                sStringFormat.LineAlignment = StringAlignment.Center;
                e.Graphics.DrawString(((ListBox)sender).Items[e.Index].ToString(), e.Font, new SolidBrush(e.ForeColor), e.Bounds, sStringFormat);
            }
            e.DrawFocusRectangle();
        }

        private void ListBoxGroupRange_MeasureItem(object sender, MeasureItemEventArgs e)
        {
            e.ItemHeight = e.ItemHeight + 12;
        }
    }
}
