kxs_push_server/Util/Queue/SourceHelper.cs

using System;
using System.Data;
using System.Linq;
using System.Threading;
using Library;
using LitJson;
using MySystem.Models.Push;

namespace MySystem
{   
    public class SourceHelper
    {
        public readonly static SourceHelper Instance = new SourceHelper();
        private SourceHelper()
        {
        }

        public void Start()//启动
        {
            Thread thread = new Thread(threadStart);
            thread.IsBackground = true;
            thread.Start();
        }

        private void threadStart()
        {
            while (true)
            {
                DoSomeThing();
                Thread.Sleep(1000);
            }
        }

        //要执行的方法
        public void DoSomeThing()
        {
            WebCMSEntities db = new WebCMSEntities();
            var sources = db.SourceData.Where(m => m.Status == 1).ToList();
            foreach(var source in sources)
            {
                string Title = source.Title; //说明
                string TableName = source.TableName; //数据表
                string FieldList = source.FieldList; //原始数据字段
                string FlagField = source.TargetFieldList; //原始标记字段
                string QueryField = "";
                JsonData fieldJson = JsonMapper.ToObject(FieldList);
                for (int i = 0; i < fieldJson.Count; i++)
                {
                    JsonData item = fieldJson[i];
                    string source_name = item["source_name"].ToString();
                    if(source_name.StartsWith("v("))
                    {
                        QueryField += source_name.Replace("v(", "").Replace(")", "") + ",";
                    }
                    else
                    {
                        QueryField += source_name + ",";
                    }
                }

                //构造抓取数据
                int StartId = source.QueryId;
                string sql = "select " + QueryField.TrimEnd(',') + " from " + TableName + " where 1=1";
                var condiList = db.SourceCondition.Where(m => m.SourceId == source.Id).ToList();
                foreach(var condi in condiList)
                {
                    int QueryCondition = condi.QueryCondition;
                    if(QueryCondition == 1)
                    {
                        sql += " and " + condi.QueryField + "='" + condi.QueryVal + "'";
                    }
                    else if(QueryCondition == 2)
                    {
                        sql += " and " + condi.QueryField + " like '%" + condi.QueryVal + "%'";
                    }
                    else if(QueryCondition == 3)
                    {
                        sql += " and " + condi.QueryField + ">" + condi.QueryVal + "";
                    }
                    else if(QueryCondition == 4)
                    {
                        sql += " and " + condi.QueryField + "<" + condi.QueryVal + "";
                    }
                    else if(QueryCondition == 5)
                    {
                        string[] QueryValList = condi.QueryVal.Split('|');
                        sql += " and " + condi.QueryField + ">=" + QueryValList[0] + " and " + condi.QueryField + "<=" + QueryValList[1] + "";
                    }
                    else if(QueryCondition == 6)
                    {
                        string[] QueryValList = condi.QueryVal.Split('|');
                        sql += " and " + condi.QueryField + ">='" + QueryValList[0] + "' and " + condi.QueryField + "<='" + QueryValList[1] + "'";
                    }
                    else if(QueryCondition == 7)
                    {
                        sql += " and " + condi.QueryField + " in (" + condi.QueryVal + ")";
                    }
                    else if(QueryCondition == 8)
                    {
                        sql += " and " + condi.QueryField + " in ('" + condi.QueryVal.Replace(",", "','") + "')";
                    }
                    else if(QueryCondition == 9)
                    {
                        string[] QueryValList = condi.QueryVal.Split(',');
                        sql += " and (";
                        int index = 0;
                        foreach(string QueryVal in QueryValList)
                        {
                            index += 1;
                            sql += condi.QueryField + "=" + QueryVal;
                            if(index < QueryValList.Length)
                            {
                                sql += " or ";
                            }
                        }
                        sql += ")";
                    }
                    else
                    {
                        sql += " and " + condi.QueryField + "='" + condi.QueryVal + "'";
                    }
                }

                //创建原始数据表
                var sourceEdit = db.SourceData.FirstOrDefault(m => m.Id == source.Id && m.Version == 0);
                if(sourceEdit != null)
                {
                    CreateTable(Title, TableName, FieldList);
                    sourceEdit.Version = 1;
                    db.SaveChanges();
                }

                //插入数据
                string TargetQueryField = "";
                JsonData TargetFieldJson = JsonMapper.ToObject(FieldList);
                for (int i = 0; i < TargetFieldJson.Count; i++)
                {
                    JsonData item = TargetFieldJson[i];
                    TargetQueryField += item["name"].ToString() + ",";
                }
                string insertSql = "";
                sql = sql.Replace("${QueryId}$", StartId.ToString());
                sql += " order by Id limit 10";
                DataTable dt = CustomerSqlConn.dtable(sql, AppConfig.Base.SourceSqlConnStr);
                foreach(DataRow dr in dt.Rows)
                {
                    string values = "";
                    for (int i = 0; i < fieldJson.Count; i++)
                    {
                        JsonData item = fieldJson[i];
                        string source_name = item["source_name"].ToString();
                        string field_type = item["field_type"].ToString();
                        if(!string.IsNullOrEmpty(source_name))
                        {
                            if(source_name.StartsWith("v("))
                            {
                                values += source_name.Replace("v(", "").Replace(")", "") + ",";
                            }
                            else
                            {
                                if(field_type == "int" || field_type == "numeric")
                                {
                                    values += "" + dr[source_name].ToString() + ",";
                                }
                                else if(field_type == "datetime")
                                {
                                    values += "'" + DateTime.Parse(dr[source_name].ToString()).ToString("yyyy-MM-dd HH:mm:ss") + "',";
                                }
                                else
                                {
                                    values += "'" + dr[source_name].ToString() + "',";
                                }
                            }
                        }
                    }
                    insertSql += "insert into u_" + TableName + " (" + TargetQueryField.TrimEnd(',') + ") values (" + values.TrimEnd(',') + ");";
                    StartId = int.Parse(dr["Id"].ToString());
                    CustomerSqlConn.op("update " + TableName + " set " + FlagField + "=1 where Id=" + StartId, AppConfig.Base.RmSourceSqlConnStr);
                }
                var edit = db.SourceData.FirstOrDefault(m => m.Id == source.Id);
                if(edit != null)
                {
                    edit.QueryId = StartId;
                    db.SaveChanges();
                }
                if(!string.IsNullOrEmpty(insertSql))
                {
                    CustomerSqlConn.op(insertSql, AppConfig.Base.SqlConnStr);
                }
            }
            db.Dispose();
        }
        
        //创建表
        public void CreateTable(string Title, string tableName, string fieldList)
        {
            string sql = "CREATE TABLE `u_" + tableName + "` (\n";
            JsonData fieldJson = JsonMapper.ToObject(fieldList);
            string prikey = "";
            for (int i = 0; i < fieldJson.Count; i++)
            {
                JsonData item = fieldJson[i];
                string nullCondition = "";
                string fieldType = item["field_type"].ToString();
                string defaultValue = item["default"].ToString();
                if(!string.IsNullOrEmpty(defaultValue))
                {
                    nullCondition = " NOT NULL DEFAULT " + defaultValue;
                }
                if(item["pri"].ToString() == "1" && item["name"].ToString() == "id")
                {
                    sql += "`" + item["name"].ToString() + "` int NOT NULL AUTO_INCREMENT COMMENT '" + item["title"].ToString() + "',\n";
                }
                else
                {
                    if(fieldType == "int")
                    {
                        sql += "`" + item["name"].ToString() + "` int NOT NULL DEFAULT " + defaultValue + " COMMENT '" + item["title"].ToString() + "',\n";
                    }
                    else if(fieldType == "numeric")
                    {
                        sql += "`" + item["name"].ToString() + "` numeric(18,2) NOT NULL DEFAULT " + defaultValue + " COMMENT '" + item["title"].ToString() + "',\n";
                    }
                    else if(fieldType == "datetime")
                    {
                        sql += "`" + item["name"].ToString() + "` datetime COMMENT '" + item["title"].ToString() + "',\n";
                    }
                    else
                    {
                        sql += "`" + item["name"].ToString() + "` " + fieldType + "(" + item["len"].ToString() + ")" + nullCondition + " COMMENT '" + item["title"].ToString() + "',\n";
                    }
                }
                if(item["pri"].ToString() == "1")
                {
                    prikey = item["name"].ToString();
                }
            }
            if(!string.IsNullOrEmpty(prikey))
            {
                sql += "PRIMARY KEY (`" + prikey + "`)\n";
            }
            sql += ") ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='" + Title + "';";
            CustomerSqlConn.op(sql, AppConfig.Base.SqlConnStr);
        }
    }
}