Merge remote-tracking branch 'origin/master'

src/main/scala/com/winhc/bigdata/spark/ng/change/NgChangeExtractAll.scala

@@ -25,7 +25,7 @@ case class NgChangeExtractAll(s: SparkSession,
                              ) extends LoggingUtils with Logging {
   @(transient@getter) val spark: SparkSession = s
 
-  val target_tab = "bds_change_extract_all"
+  val target_tab = "bds_change_extract_all_v2"
 
   init()
 
@@ -102,7 +102,7 @@ case class NgChangeExtractAll(s: SparkSession,
          |FROM
          |    tmp_change_extract_view_$tableName
          |WHERE
-         |    update_type = 'insert'
+         |--    update_type = 'insert'
          |""".stripMargin)
   }
 }

src/main/scala/com/winhc/bigdata/spark/ng/change/NgChangeExtractArgs.scala

@@ -66,6 +66,9 @@ object NgChangeExtractArgs {
       ,NgChangeExtractArgs(tableName = "company_zxr_restrict",primaryFields = "pid")    //限高
       ,NgChangeExtractArgs(tableName = "restrictions_on_exit",primaryFields = "executed_person_keyno,limited_person_pid")    //限制出镜
       ,NgChangeExtractArgs(tableName = "zxr_evaluate_results",primaryFields = "keyno")    //询价评估结果
+      ,NgChangeExtractArgs(tableName = "increase_registered_capital_info")    //增资记录
+      ,NgChangeExtractArgs(tableName = "zxr_evaluate")    //询价评估-选定评估机构
+      ,NgChangeExtractArgs(tableName = "company_bid")    //招投标
     )
 
 

src/main/scala/com/winhc/bigdata/spark/ng/change/table/company_bid.scala

@@ -0,0 +1,11 @@
+
+
+package com.winhc.bigdata.spark.ng.change.table
+
+import com.winhc.bigdata.spark.ng.change.NgCompanyChangeHandle
+import com.winhc.bigdata.spark.utils.DateUtils
+
+case class company_bid(equCols: Seq[String], is_inc:Boolean) extends NgCompanyChangeHandle {
+
+  override protected def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("publish_time"))
+}

src/main/scala/com/winhc/bigdata/spark/ng/change/table/company_license.scala

@@ -9,5 +9,5 @@ import com.winhc.bigdata.spark.utils.DateUtils
 case class company_license(equCols: Seq[String],is_inc:Boolean) extends NgCompanyChangeHandle {
 
 
-  override protected def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("issue_date"))
+  override protected def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("start_date"))
 }

src/main/scala/com/winhc/bigdata/spark/ng/change/table/increase_registered_capital_info.scala

@@ -0,0 +1,12 @@
+
+
+package com.winhc.bigdata.spark.ng.change.table
+
+import com.winhc.bigdata.spark.ng.change.NgCompanyChangeHandle
+import com.winhc.bigdata.spark.utils.DateUtils
+
+case class increase_registered_capital_info(equCols: Seq[String], is_inc:Boolean) extends NgCompanyChangeHandle {
+
+
+  override protected def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("change_time"))
+}

src/main/scala/com/winhc/bigdata/spark/ng/change/table/zxr_evaluate.scala

@@ -0,0 +1,12 @@
+
+
+package com.winhc.bigdata.spark.ng.change.table
+
+import com.winhc.bigdata.spark.ng.change.NgCompanyChangeHandle
+import com.winhc.bigdata.spark.utils.DateUtils
+
+case class zxr_evaluate(equCols: Seq[String], is_inc:Boolean) extends NgCompanyChangeHandle {
+
+
+  override protected def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("insert_time"))
+}

src/main/scala/com/winhc/bigdata/spark/ng/jobs/args_company_job.scala

@@ -430,6 +430,133 @@ object args_company_job {
       , is_super_filter = false
     )
 
+    , args_company_job("company_wechat", Seq("company_id", "public_num")
+      , rowkey_udf = "md5(cleanup(concat_ws('', company_id, public_num)))"
+      , is_super_filter = false
+    )
+
+    , args_company_job("company_bid", Seq("title", "link", "publish_time")
+      , rowkey_udf = "md5(cleanup(concat_ws('', title, link, split_date(cast(publish_time as String)))))"
+      , is_super_filter = false
+      , where = " is_json_str(purchaser_info) and is_json_str(supplier_info)"
+      , explode_args = Seq(
+        explode_args("purchaser_info", "$.keyno", "purchaser_info_id_explode")
+        , explode_args("supplier_info", "$.keyno", "supplier_info_id_explode")
+      )
+    )
+
+    , args_company_job("company_annual_report", Seq("company_id", "report_year")
+      , rowkey_udf = "md5(cleanup(concat_ws('', company_id, report_year)))"
+      , id_user_defined_rowkey = true
+      , is_super_filter = false
+    )
+
+    , args_company_job("company_annual_report_out_investment", Seq("main_id", "out_investment_name")
+      , rowkey_udf = "concat_ws('_', main_id, md5(cleanup(concat_ws('',main_id, out_investment_name))) )"
+      , is_super_filter = false
+    )
+
+    , args_company_job("company_annual_report_out_guarantee", Seq("main_id", "creditor", "obligor")
+      , rowkey_udf = "concat_ws('_', main_id, md5(cleanup(concat_ws('',main_id, creditor, obligor))) )"
+      , is_super_filter = false
+    )
+
+    , args_company_job("company_annual_report_change", Seq("main_id", "change_item", "change_time")
+      , rowkey_udf = "concat_ws('_', main_id, md5(cleanup(concat_ws('',main_id, change_item, split_date(cast(change_time as String))))) )"
+      , is_super_filter = false
+    )
+
+    , args_company_job("company_annual_report_equity_change", Seq("main_id", "change_time")
+      , rowkey_udf = "concat_ws('_', main_id, md5(cleanup(concat_ws('',main_id, split_date(cast(change_time as String))))) )"
+      , is_super_filter = false
+    )
+    , args_company_job("company_annual_report_holder", Seq("main_id", "investor_name")
+      , rowkey_udf = "concat_ws('_', main_id, md5(cleanup(concat_ws('',main_id, investor_name))) )"
+      , is_super_filter = false
+    )
+
+    , args_company_job("company_annual_report_webinfo", Seq("main_id", "website")
+      , rowkey_udf = "concat_ws('_', main_id, md5(cleanup(concat_ws('',main_id, website))) )"
+      , is_super_filter = false
+    )
+
+    , args_company_job("company_annual_report_social_security", Seq("main_id")
+      , rowkey_udf = "concat_ws('_', main_id, md5(cleanup(concat_ws('',main_id))) )"
+      , is_super_filter = false
+    )
+//融资
+    , args_company_job("accounts_receivablae", Seq("company_name", "notice_date", "pledgee_info", "assignee_info", "deadline")
+      , rowkey_udf = "md5(cleanup(concat_ws('', company_name, split_date(cast(notice_date as String)), get_text_from_json(pledgee_info, 'pledgee'), get_text_from_json(assignee_info, 'assignee'), deadline)))"
+      , is_super_filter = false
+    )
+
+    , args_company_job("dcm_registration_limit", Seq("company_id", "fileno", "bond_lassify", "start_date")
+        , rowkey_udf = "md5(cleanup(concat_ws('', company_id, fileno, bond_lassify, split_date(cast(start_date as String)))))"
+        , is_super_filter = false
+    )
+
+    , args_company_job("bond_financing", Seq("company_id", "abbreviation", "symbol")
+      , rowkey_udf = "md5(cleanup(concat_ws('', company_id, abbreviation, symbol)))"
+      , is_super_filter = false
+    )
+
+
+    , args_company_job("overseas_bond_financing", Seq("company_id", "bond_simple_name", "issue_date")
+      , rowkey_udf = "md5(cleanup(concat_ws('', company_id, bond_simple_name, split_date(cast(issue_date as String)))))"
+      , is_super_filter = false
+    )
+
+    , args_company_job("bank_loan", Seq("company_id", "creditor", "amount_financed", "start_date")
+      , rowkey_udf = "md5(cleanup(concat_ws('', company_id, creditor, amount_financed, split_date(cast(start_date as String)))))"
+      , is_super_filter = false
+    )
+
+    , args_company_job("credit_line", Seq("company_id", "end_date", "discloure_date", "credit_line")
+      , rowkey_udf = "md5(cleanup(concat_ws('', company_id, split_date(cast(end_date as String)), split_date(cast(discloure_date as String)), credit_line)))"
+      , is_super_filter = false
+    )
+
+
+    , args_company_job("lease_financing", Seq("company_id", "leaser_info", "equity", "start_date")
+      , rowkey_udf = "md5(cleanup(concat_ws('', company_id, get_text_from_json(leaser_info, 'company_name'), equity, split_date(cast(start_date as String)))))"
+      , is_super_filter = false
+    )
+
+    , args_company_job("trust_financing", Seq("company_id", "consignee_name_list", "notice_date", "start_date")
+      , rowkey_udf = "md5(cleanup(concat_ws('', company_id, split_date(cast(notice_date as String)), get_text_from_json(consignee_name_list, 'company_name'), split_date(cast(start_date as String)))))"
+      , is_super_filter = false
+    )
+
+    , args_company_job("other_financing", Seq("company_id", "financing_amount", "declare_date")
+      , rowkey_udf = "md5(cleanup(concat_ws('', company_id, financing_amount, split_date(cast(declare_date as String)))))"
+      , is_super_filter = false
+    )
+
+    , args_company_job("litigation_mediation", Seq("case_no", "filing_date")
+      , rowkey_udf = "md5(cleanup(concat_ws('',case_no,split_date(cast(filing_date as String)) )))"
+      , is_super_filter = false
+      , where = "is_json_str(litigant_info) and is_json_str(defendant_info) and is_json_str(plaintiff_info)"
+      , explode_args = Seq(
+        explode_args("plaintiff_info", "$.litigant_id", "plaintiff_info_id_explode")
+        , explode_args("defendant_info", "$.litigant_id", "defendant_info_id_explode")
+      )
+    )
+
+
+    , args_company_job("construction_qualification", Seq("company_id", "cert_type", "cert_id", "cert_name")
+      , rowkey_udf = "md5(cleanup(concat_ws('', company_id, cert_type, cert_id, cert_name)))"
+      , is_super_filter = false
+    )
+
+    , args_company_job("construction_person", Seq("company_id", "person_name", "zczy_name", "seal_code")
+      , rowkey_udf = "md5(cleanup(concat_ws('', company_id, person_name, zczy_name, seal_code)))"
+      , is_super_filter = false
+    )
+
+    , args_company_job("construction_project", Seq("company_name", "prj_name", "prj_num")
+      , rowkey_udf = "md5(cleanup(concat_ws('', company_name, prj_name, prj_num)))"
+      , is_super_filter = false
+    )
   )
 
   def get_args_company_job(tn: String): args_company_job = {

src/main/scala/com/winhc/bigdata/spark/ng/monitor/AbstractDailyHandle.scala

@@ -0,0 +1,23 @@
+package com.winhc.bigdata.spark.ng.monitor
+
+import scala.collection.mutable
+
+/**
+ * @author: XuJiakai
+ * @date: 2021/6/29 14:29
+ */
+abstract class AbstractDailyHandle() extends CompanyMonitorHandleV2 {
+  override def filter: (String, String, Seq[String], Map[String, String], Map[String, String]) => Boolean = (update_type: String, biz_date: String, change_fields: Seq[String], old_data: Map[String, String], new_data: Map[String, String]) => {
+    CompanyMonitorUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)
+  }
+
+  def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo]
+
+  override def flat_map: (ChangeExtract) => Seq[CompanyMonitorRecord] = (change_extract: ChangeExtract) => {
+    val new_data = change_extract.new_data
+    var list: mutable.Seq[CompanyMonitorRecord] = mutable.Seq.empty
+    list = list :+ getCompanyMonitorRecord(change_extract, getEntityInfo(new_data))
+    list
+  }
+
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/CompanyMonitorHandleV2.scala

@@ -0,0 +1,156 @@
+package com.winhc.bigdata.spark.ng.monitor
+
+import com.alibaba.fastjson.JSON
+import com.winhc.bigdata.spark.utils.BaseUtil
+import org.apache.commons.lang3.StringUtils
+import org.apache.spark.internal.Logging
+
+import scala.annotation.meta.{getter, setter}
+import scala.collection.mutable
+
+/**
+ * @Author: π
+ * @Date: 2021/9/9
+ * @Description: 企业财产监控v8版本
+ */
+trait CompanyMonitorHandleV2 extends Logging {
+
+  @getter
+  @setter
+  protected val is_inc: Boolean //false 为存量
+
+  /**
+   * 变更时间
+   *
+   * @param new_map
+   * @return
+   */
+  protected def get_change_time(bizDate: String, new_map: Map[String, String]): String = {
+    var res = bizDate
+    if (bizDate.length == 10) {
+      res = res.concat(" 00:00:00")
+    }
+    res
+  }
+
+  /**
+   * 业务id
+   *
+   * @param rowkey
+   * @return
+   */
+  protected def get_biz_id(rowkey: String, new_map: Map[String, String]): String = rowkey
+
+  /**
+   * 条件过滤
+   *
+   * @return
+   */
+  def get_conditional_filter(): String = ""
+
+
+  protected def get_money(new_map: Map[String, String]): Double = 0D
+
+  protected def processMoney(amt: String): String = {
+    var amtstr = amt
+    if (amtstr == null) {
+      "0"
+    } else {
+      amtstr = amtstr.replaceAll("[^\\d.]", "")
+      if (amtstr.equals("") || amtstr.split("\\.").length > 2 || amtstr.startsWith(".") || amtstr.endsWith(".")) {
+        amtstr = "0"
+      }
+      try {
+        amtstr.toDouble
+      } catch {
+        case ex: Exception => {
+          amtstr = "0"
+        }
+      }
+      amtstr
+    }
+  }
+
+  protected def get_deleted(new_map: Map[String, String], key: String = "deleted"): String = {
+    if (new_map == null) return "-1"
+    val deleted = new_map.getOrElse(key, "-1")
+    deleted
+  }
+
+  def filter: (String, String, Seq[String], Map[String, String], Map[String, String]) => Boolean = CompanyMonitorUtils.default_filter
+
+  def flat_map: (ChangeExtract) => Seq[CompanyMonitorRecord]
+
+
+  protected def getCompanyMonitorRecord(change_extract: ChangeExtract
+                                        , entity_info: Seq[EntityInfo]
+                                       ): CompanyMonitorRecord = {
+    val tn = change_extract.tn
+    val update_type = change_extract.update_type
+    val old_data = change_extract.old_data
+    val new_data = change_extract.new_data
+    val biz_date = change_extract.biz_date
+    val rowkey = change_extract.rowkey
+    val update_time = change_extract.update_time
+    //val date = if (is_inc) update_time else biz_date
+    if(entity_info.isEmpty){
+      CompanyMonitorRecord(
+        //todo 待验证
+        id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn)
+        , entity_info = entity_info
+        , dimension_type = ""
+        , flow_type = ""
+        , rta_desc = ""
+        , change_time = biz_date
+        , biz_id = rowkey
+        , info_risk_level = ""
+        , `type` = ""
+        , create_time = update_time
+        , amt = get_money(new_data)
+        , update_type = update_type
+        , deleted = get_deleted(new_data)
+      )
+    }else{
+      val type_id = entity_info.map(_.type_id).head
+      val args = CompanyMonitorMappings.get_args(`type_id`)
+      CompanyMonitorRecord(
+        //todo 待验证
+        id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn)
+        , entity_info = entity_info
+        , dimension_type = args.dimension_type
+        , flow_type = args.flow_type
+        , rta_desc = ""
+        , change_time = biz_date
+        , biz_id = rowkey
+        , info_risk_level = args.info_risk_level
+        , `type` = type_id
+        , create_time = update_time
+        , amt = get_money(new_data)
+        , update_type = update_type
+        , deleted = get_deleted(new_data)
+      )
+    }
+
+  }
+
+
+  protected def getEntity(json: String, id_key: String, name_key: String, type_id: String
+                         ): Seq[EntityInfo] = {
+    if (StringUtils.isEmpty(json)) {
+      Seq.empty
+    } else {
+      val array = JSON.parseArray(json)
+      var list: mutable.Seq[EntityInfo] = mutable.Seq.empty
+      for (i <- 0 until array.size()) {
+        val jSONObject = array.getJSONObject(i)
+        val keyno = jSONObject.getString(id_key)
+        val name = jSONObject.getString(name_key)
+        list = list :+ EntityInfo(keyno = keyno, name = name, type_id = type_id)
+      }
+      list
+    }
+  }
+
+
+
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/CompanyMonitorMappings.scala

@@ -0,0 +1,57 @@
+package com.winhc.bigdata.spark.ng.monitor
+
+/**
+ * @author: π
+ * @date: 2021/9/9 15:46
+ */
+
+case class NgCompanyMonitoArgs(type_id: String //线索id
+                               , type_name: String //线索名称
+                               , dimension_table: String //维度id
+                               , dimension_type: String //维度表名
+                               , flow_type: String //流向id 0 -> 流入 1 -> 流出  2->流向未知
+                               , info_risk_level: String //线索等级id 1->低级 2->中级 3->高级
+                              )
+
+object CompanyMonitorMappings {
+  val args: Seq[NgCompanyMonitoArgs] =
+    Seq(
+      NgCompanyMonitoArgs(type_id = "20", type_name = "新增胜诉案件", dimension_table = "wenshu_detail2", dimension_type = "18", flow_type = "0", info_risk_level = "2")
+      , NgCompanyMonitoArgs(type_id = "19", type_name = "有恢复执行案件", dimension_table = "wenshu_detail2", dimension_type = "18", flow_type = "1", info_risk_level = "1")
+      , NgCompanyMonitoArgs(type_id = "27", type_name = "新增财产保全案件", dimension_table = "wenshu_detail2", dimension_type = "18", flow_type = "1", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "7", type_name = "失信信息移除", dimension_table = "NG_COMPANY_DISHONEST_INFO", dimension_type = "7", flow_type = "1", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "32", type_name = "新增一条失信", dimension_table = "NG_COMPANY_DISHONEST_INFO", dimension_type = "7", flow_type = "2", info_risk_level = "2")
+      , NgCompanyMonitoArgs(type_id = "18", type_name = "限制高消费移除", dimension_table = "NG_COMPANY_ZXR_RESTRICT", dimension_type = "17", flow_type = "1", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "6", type_name = "有恢复执行案件即将开庭", dimension_table = "NG_COMPANY_COURT_OPEN_ANNOUNCEMENT", dimension_type = "6", flow_type = "1", info_risk_level = "2")
+      , NgCompanyMonitoArgs(type_id = "22", type_name = "有资产即将被司法拍卖", dimension_table = "NG_AUCTION_TRACKING", dimension_type = "19", flow_type = "0", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "8", type_name = "出质了持有股权", dimension_table = "NG_COMPANY_EQUITY_INFO", dimension_type = "8", flow_type = "0", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "33", type_name = "出质的股权变更为无效", dimension_table = "NG_COMPANY_EQUITY_INFO", dimension_type = "8", flow_type = "1", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "13", type_name = "抵押了公司土地", dimension_table = "NG_COMPANY_LAND_MORTGAGE", dimension_type = "12", flow_type = "0", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "21", type_name = "接受了他人土地抵押", dimension_table = "NG_COMPANY_LAND_MORTGAGE", dimension_type = "12", flow_type = "1", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "12", type_name = "购买了新的土地", dimension_table = "NG_COMPANY_LAND_ANNOUNCEMENT", dimension_type = "11", flow_type = "1", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "10", type_name = "新增了股东", dimension_table = "NG_COMPANY_HOLDER", dimension_type = "10", flow_type = "0", info_risk_level = "2")//无全量
+      , NgCompanyMonitoArgs(type_id = "11", type_name = "新增对外投资", dimension_table = "NG_COMPANY_HOLDER", dimension_type = "10", flow_type = "1", info_risk_level = "3")//无全量
+      , NgCompanyMonitoArgs(type_id = "34", type_name = "投资的公司，所占股份下降", dimension_table = "NG_COMPANY_HOLDER", dimension_type = "10", flow_type = "0", info_risk_level = "2")//无全量
+      , NgCompanyMonitoArgs(type_id = "2", type_name = "注册资本上升", dimension_table = "NG_INCREASE_REGISTERED_CAPITAL_INFO", dimension_type = "2", flow_type = "0", info_risk_level = "2")
+      , NgCompanyMonitoArgs(type_id = "9", type_name = "有新的融资", dimension_table = "NG_COMPANY_FINANCE", dimension_type = "9", flow_type = "0", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "1", type_name = "新增破产公告", dimension_table = "NG_BANKRUPTCY_OPEN_CASE", dimension_type = "1", flow_type = "1", info_risk_level = "2")
+      , NgCompanyMonitoArgs(type_id = "3", type_name = "新增招标信息", dimension_table = "NG_COMPANY_BID", dimension_type = "3", flow_type = "0", info_risk_level = "3")//待处理
+      , NgCompanyMonitoArgs(type_id = "14", type_name = "资产被抵押", dimension_table = "NG_COMPANY_MORTGAGE_INFO", dimension_type = "13", flow_type = "0", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "16", type_name = "新增商标信息", dimension_table = "NG_COMPANY_TM", dimension_type = "15", flow_type = "0", info_risk_level = "1")
+      , NgCompanyMonitoArgs(type_id = "15", type_name = "新增专利信息", dimension_table = "NG_COMPANY_PATENT", dimension_type = "14", flow_type = "0", info_risk_level = "2")
+      , NgCompanyMonitoArgs(type_id = "4", type_name = "新增著作权信息", dimension_table = "NG_COMPANY_COPYRIGHT_REG", dimension_type = "4", flow_type = "0", info_risk_level = "1")
+      , NgCompanyMonitoArgs(type_id = "5", type_name = "新增作品著作权", dimension_table = "NG_COMPANY_COPYRIGHT_WORKS", dimension_type = "5", flow_type = "0", info_risk_level = "1")
+      , NgCompanyMonitoArgs(type_id = "23", type_name = "有资产选定询价评估机构", dimension_table = "NG_ZXR_EVALUATE", dimension_type = "20", flow_type = "0", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "24", type_name = "有资产完成询价评估", dimension_table = "NG_ZXR_EVALUATE_RESULTS", dimension_type = "21", flow_type = "0", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "26", type_name = "新增行政许可", dimension_table = "NG_COMPANY_LICENSE", dimension_type = "22", flow_type = "0", info_risk_level = "2")
+      , NgCompanyMonitoArgs(type_id = "28", type_name = "在外持有股权被冻结", dimension_table = "NG_COMPANY_JUDICIAL_ASSISTANCE", dimension_type = "23", flow_type = "1", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "29", type_name = "有股东持有的股权被冻结", dimension_table = "NG_COMPANY_JUDICIAL_ASSISTANCE", dimension_type = "23", flow_type = "1", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "30", type_name = "有股权解除冻结", dimension_table = "NG_COMPANY_JUDICIAL_ASSISTANCE", dimension_type = "23", flow_type = "1", info_risk_level = "3")
+      , NgCompanyMonitoArgs(type_id = "31", type_name = "有股东持有的股权解除冻结", dimension_table = "NG_COMPANY_JUDICIAL_ASSISTANCE", dimension_type = "23", flow_type = "1", info_risk_level = "3")
+    )
+
+
+  def get_args(type_id: String): NgCompanyMonitoArgs = {
+    args.find(p => type_id.equals(p.type_id)).getOrElse(throw new NullPointerException("type_id is not found !!!"))
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/CompanyMonitorRecord.scala

@@ -0,0 +1,113 @@
+package com.winhc.bigdata.spark.ng.monitor
+
+import cn.hutool.crypto.SecureUtil
+import com.winhc.bigdata.spark.implicits.CaseClass2JsonHelper._
+import org.apache.commons.lang3.StringUtils
+import org.apache.spark.sql.Row
+
+/**
+ * @author: π
+ * @date: 2021/9/10 13:49
+ */
+case class ChangeExtract(rowkey: String
+                         , company_id: String
+                         , company_name: String
+                         , tn: String
+                         , update_type: String
+                         , old_data: Map[String, String]
+                         , new_data: Map[String, String]
+                         , change_fields: Seq[String]
+                         , biz_date: String
+                         , update_time: String
+                        )
+
+
+case class EntityInfo(keyno: String
+                      , name: String
+                      , type_id: String
+                     ) extends Comparable[EntityInfo] {
+
+
+  override def hashCode(): Int = s"$name".hashCode()
+
+  override def equals(obj: Any): Boolean =
+    obj match {
+      case d: EntityInfo =>
+        s"$name".equals(s"${d.name}")
+      case _ =>
+        false
+    }
+
+  override def compareTo(o: EntityInfo): Int = s"$name".compareTo(s"${o.name}")
+}
+
+case class RowkeyInfo(rowkey: String, tn: String) {
+  def toStr(): String = {
+    s"$tn@@$rowkey"
+  }
+}
+
+
+case class CompanyMonitorRecord(id: String,
+                                entity_info: Seq[EntityInfo],
+                                dimension_type: String,
+                                flow_type: String,
+                                rta_desc: String,
+                                change_time: String,
+                                biz_id: String,
+                                info_risk_level: String,
+                                `type`: String,
+                                create_time: String,
+                                amt: Double,
+                                update_type: String,
+                                deleted: String
+                               ) {
+
+  def format(): CompanyMonitorRecord = {
+    if (id == null) {
+      return null
+    }
+    if (entity_info == null || entity_info.isEmpty) {
+      return null
+    }
+    //剔除id为空的数据
+    val rec = entity_info.filter(r => StringUtils.isNotEmpty(r.keyno))
+    if (rec.isEmpty) return null
+
+    if (rec.length != entity_info.length)
+      return CompanyMonitorRecord(id,
+        rec,
+        dimension_type,
+        flow_type,
+        rta_desc,
+        change_time,
+        biz_id,
+        info_risk_level,
+        `type`,
+        create_time,
+        amt,
+        update_type,
+        deleted
+      )
+    this
+  }
+
+  def to_row(): Row = {
+    Row(SecureUtil.md5(id)
+      , entity_info.toJson()
+      , dimension_type
+      , flow_type
+      , rta_desc
+      , change_time
+      , biz_id
+      , info_risk_level
+      , `type`
+      , create_time
+      , amt
+      , update_type
+      , deleted
+    )
+  }
+}
+
+

src/main/scala/com/winhc/bigdata/spark/ng/monitor/CompanyMonitorUtils.scala

@@ -0,0 +1,33 @@
+package com.winhc.bigdata.spark.ng.monitor
+
+import org.apache.commons.lang3.StringUtils
+
+/**
+ * @author: π
+ * @date: 2021/9/9 15:46
+ */
+object CompanyMonitorUtils {
+
+  def default_filter(update_type: String, biz_date: String, change_fields: Seq[String], old_data: Map[String, String], new_data: Map[String, String]): Boolean = {
+    if (biz_date == null) return false
+    //if (update_type.equals("remove") || update_type.equals("other")) return false
+    if (update_type.equals("update") && change_fields.isEmpty) return false
+    true
+  }
+
+  def generateId(rowkey: String, biz_date: String, tn: String, random_num: String = null): String = {
+    if (StringUtils.isEmpty(random_num)) {
+      s"$rowkey@$biz_date@$tn"
+    } else {
+      s"$rowkey@$biz_date@$tn@$random_num"
+    }
+  }
+
+  def formatDate(date: String): String = {
+    if (StringUtils.isEmpty(date)) {
+      null
+    } else {
+      if (date.contains(" ")) date.split(" ")(0) else date
+    }
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/CompanyMonitorV2.scala

@@ -0,0 +1,221 @@
+package com.winhc.bigdata.spark.ng.monitor
+
+import com.winhc.bigdata.spark.config.EsConfig
+import com.winhc.bigdata.spark.jobs.dynamic.CompanyDynamicHandleUtils
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+import com.winhc.bigdata.spark.udf.BaseFunc
+import com.winhc.bigdata.spark.utils.BaseUtil.isWindows
+import com.winhc.bigdata.spark.utils.ReflectUtils.getClazz
+import com.winhc.bigdata.spark.utils.{AsyncExtract, LoggingUtils, SparkUtils}
+import org.apache.commons.lang3.StringUtils
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.types.{DoubleType, StringType}
+import org.apache.spark.sql.{Row, SparkSession}
+
+import scala.annotation.meta.getter
+import scala.collection.immutable.ListMap
+import scala.collection.mutable
+
+/**
+ * @Author: π
+ * @Date: 2021/9/9
+ * @Description: 企业财产监控v8版本
+ */
+object CompanyMonitorV2 {
+  val targetTab = "winhc_ng.ads_company_monitor"
+
+  case class CompanyMonitorUtil(s: SparkSession,
+                                project: String, //表所在工程名
+                                incr: Boolean //是否增量
+                               ) extends LoggingUtils with Logging with BaseFunc {
+    @(transient@getter) val spark: SparkSession = s
+
+    private lazy val org_tab = if (incr) "winhc_ng.bds_change_extract" else "winhc_ng.bds_change_extract_all_v2"
+    private val ds = getLastPartitionsOrElse(org_tab, "0")
+    private val ads_judicial_case_relation_case_amt = "winhc_ng.ads_judicial_case_relation_case_amt"
+
+    def calc(tableName: String, needMoney: String): Unit = {
+      val clazz = getClazz[CompanyMonitorHandleV2](s"com.winhc.bigdata.spark.ng.monitor.table.$tableName", incr)
+      val conditional = clazz.get_conditional_filter()
+      val filter = clazz.filter
+      val flat_map = clazz.flat_map
+      val tn = tableName
+
+      val rdd = sql(
+        s"""
+           |SELECT  *
+           |FROM    $org_tab
+           |WHERE   ds = '$ds'
+           |AND     tn = '$tn'
+           |$conditional
+           |""".stripMargin)
+        .rdd.map(r => {
+        val value = r.getAs[String]("change_fields")
+        val change_fields: Seq[String] = if (StringUtils.isEmpty(value)) Seq.empty else value.split(",")
+        ChangeExtract(rowkey = r.getAs("rowkey")
+          , company_id = r.getAs("company_id")
+          , company_name = null
+          , tn = r.getAs("table_name")
+          , update_type = r.getAs("update_type")
+          , old_data = r.getAs("old_data")
+          , new_data = r.getAs("new_data")
+          , change_fields = change_fields
+          , biz_date = r.getAs("biz_date")
+          , update_time = r.getAs("update_time")
+        )
+      }).filter(r => {
+        if (filter == null) {
+          true
+        } else {
+          filter(r.update_type, r.biz_date, r.change_fields, r.old_data, r.new_data)
+        }
+      }).flatMap(flat_map)
+        .map(_.format())
+        .filter(_ != null)
+
+      val schema = getSchema(ListMap(
+        "id" -> StringType
+        , "entity_info" -> StringType
+        , "dimension_type" -> StringType
+        , "flow_type" -> StringType
+        , "rta_desc" -> StringType
+        , "change_time" -> StringType
+        , "biz_id" -> StringType
+        , "info_risk_level" -> StringType
+        , "type" -> StringType
+        , "create_time" -> StringType
+        , "amt" -> DoubleType
+        , "update_type" -> StringType
+        , "deleted" -> StringType
+      ))
+
+      spark.createDataFrame(rdd.map(_.to_row()), schema)
+        .createOrReplaceTempView(s"company_monitor_tmp1_$tableName")
+
+      needMoney match {
+        case "0" => {
+          sql(
+            s"""
+               |INSERT ${if (isWindows) "INTO" else "OVERWRITE"} TABLE $targetTab PARTITION(ds='$ds',tn='$tableName')
+               |SELECT  id
+               |        , entity_info
+               |        , dimension_type
+               |        , flow_type
+               |        , rta_desc
+               |        , change_time
+               |        , biz_id
+               |        , info_risk_level
+               |        , type
+               |        , create_time
+               |        , amt
+               |        , update_type
+               |        , deleted
+               |FROM
+               |    company_monitor_tmp1_$tableName
+               |WHERE id IS NOT NULL
+               |""".stripMargin)
+        }
+        case "1" => {
+          sql(
+            s"""
+               |INSERT ${if (isWindows) "INTO" else "OVERWRITE"} TABLE $targetTab PARTITION(ds='$ds',tn='$tableName')
+               |SELECT  id
+               |        , entity_info
+               |        , dimension_type
+               |        , flow_type
+               |        , rta_desc
+               |        , change_time
+               |        , biz_id
+               |        , info_risk_level
+               |        , type
+               |        , create_time
+               |        , case_amt
+               |        , update_type
+               |        , deleted
+               |FROM
+               |    (
+               |        SELECT a.*, b.case_amt
+               |        FROM  company_monitor_tmp1_$tableName a
+               |        LEFT JOIN $ads_judicial_case_relation_case_amt b  ON a.biz_id = b.detail_id and b.tn = '$tableName'
+               |    )
+               |WHERE id IS NOT NULL
+               |""".stripMargin)
+        }
+      }
+    }
+  }
+
+  private val startArgs = Seq(
+    Args(tableName = "wenshu_detail_v2")
+    , Args(tableName = "company_dishonest_info", needMoney = "1") //todo 金额提取
+    //, Args(tableName = "company_zxr")
+    , Args(tableName = "company_zxr_restrict", needMoney = "1") //todo 金额提取
+    , Args(tableName = "company_court_open_announcement")
+    , Args(tableName = "company_equity_info")
+    , Args(tableName = "company_land_mortgage")
+    , Args(tableName = "company_land_announcement")
+    , Args(tableName = "company_finance")
+    , Args(tableName = "bankruptcy_open_case")
+    , Args(tableName = "company_bid") //todo 待处理
+    , Args(tableName = "company_mortgage_info")
+    , Args(tableName = "company_tm")
+    , Args(tableName = "company_patent")
+    , Args(tableName = "company_copyright_reg")
+    , Args(tableName = "company_copyright_works")
+    , Args(tableName = "company_holder")
+    //, Args(tableName = "company", bName = 1)
+    , Args(tableName = "auction_tracking")
+    , Args(tableName = "increase_registered_capital_info")
+    , Args(tableName = "zxr_evaluate")
+    , Args(tableName = "zxr_evaluate_results")
+    , Args(tableName = "company_license")
+    //, Args(tableName = "company_judicial_assistance")
+
+  )
+
+  private case class Args(project: String = "winhc_ng"
+                          , tableName: String
+                          , needMoney: String = "0")
+
+  def main(args: Array[String]): Unit = {
+
+    if (args.length != 3) {
+      println(
+        s"""
+           |Please enter the legal parameters !
+           |<project> <tableNames> <incr>
+           |""".stripMargin)
+      sys.exit(-99)
+    }
+
+    val Array(project, tableNames, incr) = args
+
+    println(
+      s"""
+         |project: $project
+         |tableNames: $tableNames
+         |incr: $incr
+         |""".stripMargin)
+
+    val config = EsConfig.getEsConfigMap ++ mutable.Map(
+      "spark.hadoop.odps.project.name" -> project,
+      "spark.hadoop.odps.spark.local.partition.amt" -> "1000"
+    )
+    val spark = SparkUtils.InitEnv("CompanyMonitorV2", config)
+    val cd = CompanyMonitorUtil(spark, project, if ("incr".equals(incr)) true else false)
+
+    var start = startArgs
+    if (!tableNames.equals("all")) {
+      val set = tableNames.split(",").toSet
+      start = start.filter(a => set.contains(a.tableName))
+    }
+
+    val a = start.map(e => (e.tableName, () => {
+      cd.calc(e.tableName, e.needMoney) //通用处理
+      true
+    }))
+
+    AsyncExtract.startAndWait(spark, a)
+    spark.stop()
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/auction_tracking.scala

@@ -0,0 +1,22 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.utils.BaseUtil
+
+/**
+ * @Date: 2021/9/9
+ * @Description: 司法拍卖
+ */
+case class auction_tracking(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override protected def get_money(new_map: Map[String, String]): Double = {
+    BaseUtil.amt_div(processMoney(new_map("query_price")), "10000")
+  }
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    getEntity(new_data.getOrEmptyStr("company_info"), "company_id", "company_name", "22")
+
+  }
+
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/bankruptcy_open_case.scala

@@ -0,0 +1,14 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 新增破产公告
+ */
+case class bankruptcy_open_case(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    getEntity(new_data("respondent_info"), id_key = "litigant_id", name_key = "name", "1")
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_bid.scala

@@ -0,0 +1,15 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 招投标
+ */
+case class company_bid(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    getEntity(new_data.getOrEmptyStr("purchaser_info"), "keyno", "company_name", "3")
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_copyright_reg.scala

@@ -0,0 +1,14 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 软件著作权
+ */
+case class company_copyright_reg(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    getEntity(new_data.getOrEmptyStr("author_nationality_info"), "keyno", "name","4")
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_copyright_works.scala

@@ -0,0 +1,14 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 著作权
+ */
+case class company_copyright_works(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    getEntity(new_data.getOrEmptyStr("author_info"), "keyno", "name","5")
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_court_open_announcement.scala

@@ -0,0 +1,30 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import scala.collection.mutable
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 有恢复执行案件即将开庭(被告)
+ */
+case class company_court_open_announcement(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override def get_conditional_filter(): String = {
+    "AND  new_data['case_no'] like concat('%','恢','%')"
+  }
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    try {
+      var list: mutable.Seq[EntityInfo] = mutable.Seq.empty
+      val company_info = new_data.getOrEmptyStr("defendant_info")
+      list = list ++ getEntity(company_info, "litigant_id", "name","6")
+      list
+    } catch {
+      case e: Exception => {
+        logError(e.getMessage, e)
+        Seq.empty
+      }
+    }
+  }
+
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_dishonest_info.scala

@@ -0,0 +1,74 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, ChangeExtract, CompanyMonitorMappings, CompanyMonitorRecord, CompanyMonitorUtils, EntityInfo, NgCompanyMonitoArgs}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+
+import scala.collection.mutable
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 失信信息移除
+ *               新增一条失信
+ */
+case class company_dishonest_info(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    Seq.empty[EntityInfo]
+  }
+
+  override def flat_map: (ChangeExtract) => Seq[CompanyMonitorRecord] = (change_extract: ChangeExtract) => {
+    var list: mutable.Seq[CompanyMonitorRecord] = mutable.Seq.empty
+
+    val tn = change_extract.tn
+    val update_type = change_extract.update_type
+    val new_data = change_extract.new_data
+    val biz_date = change_extract.biz_date
+    val rowkey = change_extract.rowkey
+    val update_time = change_extract.update_time
+
+    if (new_data.getOrEmptyStr("deleted").equals("1")) {
+      //失信信息移除
+      val `type` = "7"
+      val args = CompanyMonitorMappings.get_args(`type`)
+      list = list :+ CompanyMonitorRecord(
+        id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type`)
+        , entity_info = Seq(EntityInfo(keyno = new_data.getOrEmptyStr("keyno"), name = new_data.getOrEmptyStr("name"), type_id = `type`))
+        , dimension_type = args.dimension_type
+        , flow_type = args.flow_type
+        , rta_desc = ""
+        , change_time = biz_date
+        , biz_id = rowkey
+        , info_risk_level = args.info_risk_level
+        , `type` = `type`
+        , create_time = update_time
+        , amt = get_money(new_data)
+        , update_type = update_type
+        , deleted = get_deleted(new_data)
+      )
+    } else if (new_data.getOrEmptyStr("deleted").equals("0")
+      && (new_data.getOrEmptyStr("action_content").contains("有履行能力")
+      || new_data.getOrEmptyStr("action_content").contains("转移财产")
+      || new_data.getOrEmptyStr("action_content").contains("违反财产报告制度")
+      )) {
+      val `type` = "32"
+      val args = CompanyMonitorMappings.get_args(`type`)
+      //新增一条失信
+      list = list :+ CompanyMonitorRecord(
+        id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type`)
+        , entity_info = Seq(EntityInfo(keyno = new_data.getOrEmptyStr("keyno"), name = new_data.getOrEmptyStr("name"), type_id = `type`))
+        , dimension_type = args.dimension_type
+        , flow_type = args.flow_type
+        , rta_desc = ""
+        , change_time = biz_date
+        , biz_id = rowkey
+        , info_risk_level = args.info_risk_level
+        , `type` = `type`
+        , create_time = update_time
+        , amt = get_money(new_data)
+        , update_type = update_type
+        , deleted = get_deleted(new_data)
+      )
+    }
+    list
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_equity_info.scala

@@ -0,0 +1,78 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, ChangeExtract, CompanyMonitorMappings, CompanyMonitorRecord, CompanyMonitorUtils, EntityInfo}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.utils.BaseUtil
+
+import scala.collection.mutable
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 出质了持有股权(非历史忽略)
+ *               出质的股权变更为无效
+ */
+case class company_equity_info(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override protected def get_money(new_map: Map[String, String]): Double = {
+    BaseUtil.amt_div(processMoney(new_map("equity_amount")),"1")
+  }
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    Seq.empty[EntityInfo]
+  }
+
+  override def flat_map: (ChangeExtract) => Seq[CompanyMonitorRecord] = (change_extract: ChangeExtract) => {
+    var list: mutable.Seq[CompanyMonitorRecord] = mutable.Seq.empty
+
+    val tn = change_extract.tn
+    val update_type = change_extract.update_type
+    val new_data = change_extract.new_data
+    val biz_date = change_extract.biz_date
+    val rowkey = change_extract.rowkey
+    val update_time = change_extract.update_time
+    val change_fields = change_extract.change_fields
+
+    //出质了持有股权(非历史忽略)
+    val `type1` = "8"
+    val args1 = CompanyMonitorMappings.get_args(`type1`)
+    list = list :+ CompanyMonitorRecord(
+      id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type1`)
+      , entity_info = getEntity(new_data.getOrEmptyStr("pledgor_info"), "pledgor_id", "pledgor",type_id=`type1`)
+      , dimension_type = args1.dimension_type
+      , flow_type = args1.flow_type
+      , rta_desc = ""
+      , change_time = biz_date
+      , biz_id = rowkey
+      , info_risk_level = args1.info_risk_level
+      , `type` = `type1`
+      , create_time = update_time
+      , amt = get_money(new_data)
+      , update_type = update_type
+      , deleted = get_deleted(new_data)
+    )
+
+    //出质的股权变更为无效
+    if (new_data.getOrEmptyStr("state").equals("无效")) {
+      val `type2` = "33"
+      val args2 = CompanyMonitorMappings.get_args(`type2`)
+      list = list :+ CompanyMonitorRecord(
+        id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type2`)
+        , entity_info = getEntity(new_data.getOrEmptyStr("pledgor_info"), "pledgor_id", "pledgor",type_id=`type2`)
+        , dimension_type = args2.dimension_type
+        , flow_type = args2.flow_type
+        , rta_desc = ""
+        , change_time = biz_date
+        , biz_id = rowkey
+        , info_risk_level = args2.info_risk_level
+        , `type` = `type2`
+        , create_time = update_time
+        , amt = get_money(new_data)
+        , update_type = update_type
+        , deleted = get_deleted(new_data)
+      )
+    }
+
+
+    list
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_finance.scala

@@ -0,0 +1,15 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 融资信息
+ */
+case class company_finance(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    Seq(EntityInfo(keyno = new_data.getOrEmptyStr("company_id"), name = new_data.getOrEmptyStr("company_name"), type_id = "9"))
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_holder.scala

@@ -0,0 +1,104 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, ChangeExtract, CompanyMonitorMappings, CompanyMonitorRecord, CompanyMonitorUtils, EntityInfo}
+import com.winhc.bigdata.spark.utils.BaseUtil
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import org.apache.commons.lang3.StringUtils
+
+import scala.collection.mutable
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 新增了股东
+ *               新增对外投资
+ *               投资的公司，所占股份下降
+ */
+case class company_holder(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override protected def get_money(new_map: Map[String, String]): Double = {
+    BaseUtil.amt_div(processMoney(new_map("amount")), "1")
+  }
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    Seq.empty[EntityInfo]
+  }
+
+  override def flat_map: (ChangeExtract) => Seq[CompanyMonitorRecord] = (change_extract: ChangeExtract) => {
+    var list: mutable.Seq[CompanyMonitorRecord] = mutable.Seq.empty
+
+    val tn = change_extract.tn
+    val update_type = change_extract.update_type
+    val new_data = change_extract.new_data
+    val old_data = change_extract.old_data
+    val biz_date = change_extract.biz_date
+    val rowkey = change_extract.rowkey
+    val update_time = change_extract.update_time
+    val change_fields = change_extract.change_fields
+
+    //新增了股东
+    val `type1` = "10"
+    val args1 = CompanyMonitorMappings.get_args(`type1`)
+    list = list :+ CompanyMonitorRecord(
+      id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type1`)
+      , entity_info = Seq(EntityInfo(keyno = new_data.getOrEmptyStr("company_id"), name = new_data.getOrEmptyStr("company_name"), type_id = `type1`))
+      , dimension_type = args1.dimension_type
+      , flow_type = args1.flow_type
+      , rta_desc = ""
+      , change_time = CompanyMonitorUtils.formatDate(biz_date)
+      , biz_id = rowkey
+      , info_risk_level = args1.info_risk_level
+      , `type` = `type1`
+      , create_time = CompanyMonitorUtils.formatDate(update_time)
+      , amt = get_money(new_data)
+      , update_type = update_type
+      , deleted = get_deleted(new_data)
+    )
+
+    //新增对外投资
+    val `type2` = "11"
+    val args2 = CompanyMonitorMappings.get_args(`type2`)
+    list = list :+ CompanyMonitorRecord(
+      id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type2`)
+      , entity_info = Seq(EntityInfo(keyno = new_data.getOrEmptyStr("holder_id"), name = new_data.getOrEmptyStr("holder_name"), type_id = `type2`))
+      , dimension_type = args2.dimension_type
+      , flow_type = args2.flow_type
+      , rta_desc = ""
+      , change_time = CompanyMonitorUtils.formatDate(biz_date)
+      , biz_id = rowkey
+      , info_risk_level = args2.info_risk_level
+      , `type` = `type2`
+      , create_time = CompanyMonitorUtils.formatDate(update_time)
+      , amt = get_money(new_data)
+      , update_type = update_type
+      , deleted = get_deleted(new_data)
+    )
+
+    //投资的公司，所占股份下降
+    if (new_data != null && old_data != null) {
+      val new_percent = new_data.getOrEmptyStr("percent")
+      val old_percent = old_data.getOrEmptyStr("percent")
+      if (change_fields != null && change_fields.contains("percent") && StringUtils.isNotBlank(new_percent) && StringUtils.isNotBlank(old_percent) && new_percent.toDouble < old_percent.toDouble) {
+        val `type3` = "34"
+        val args3 = CompanyMonitorMappings.get_args(`type3`)
+        list = list :+ CompanyMonitorRecord(
+          id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type3`)
+          , entity_info = Seq(EntityInfo(keyno = new_data.getOrEmptyStr("holder_id"), name = new_data.getOrEmptyStr("holder_name"), type_id = `type3`))
+          , dimension_type = args3.dimension_type
+          , flow_type = args3.flow_type
+          , rta_desc = ""
+          , change_time = CompanyMonitorUtils.formatDate(biz_date)
+          , biz_id = rowkey
+          , info_risk_level = args3.info_risk_level
+          , `type` = `type3`
+          , create_time = CompanyMonitorUtils.formatDate(update_time)
+          , amt = get_money(new_data)
+          , update_type = update_type
+          , deleted = get_deleted(new_data)
+        )
+      }
+    }
+    list
+  }
+}
+
+

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_judicial_assistance.scala

@@ -0,0 +1,113 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.monitor._
+import com.winhc.bigdata.spark.utils.BaseUtil
+import org.apache.commons.lang3.StringUtils
+
+import scala.collection.mutable
+
+/**
+ * @Date: 2020/9/13 11:44
+ * @Description: 在外持有股权被冻结
+ *               有股东持有的股权被冻结
+ *               有股权解除冻结
+ *               有股东持有的股权解除冻结
+ */
+case class company_judicial_assistance(is_inc: Boolean) extends AbstractDailyHandle {
+
+
+  override protected def get_money(new_map: Map[String, String]): Double = {
+    BaseUtil.amt_div(processMoney(new_map("equity_amount")), "1")
+  }
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    Seq.empty[EntityInfo]
+  }
+
+  override def flat_map: (ChangeExtract) => Seq[CompanyMonitorRecord] = (change_extract: ChangeExtract) => {
+    var list: mutable.Seq[CompanyMonitorRecord] = mutable.Seq.empty
+
+    val tn = change_extract.tn
+    val update_type = change_extract.update_type
+    val new_data = change_extract.new_data
+    val biz_date = change_extract.biz_date
+    val rowkey = change_extract.rowkey
+    val update_time = change_extract.update_time
+
+    if (StringUtils.isNotBlank(new_data.getOrEmptyStr("fz_execute_order_num"))) {
+      //冻结信息
+      val `type1` = "28"
+      val args1 = CompanyMonitorMappings.get_args(`type1`)
+      list = list :+ CompanyMonitorRecord(
+        id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type1`)
+        , entity_info = Seq(EntityInfo(keyno = new_data.getOrEmptyStr("executed_person_id"), name = new_data.getOrEmptyStr("executed_person"), type_id = `type1`))
+        , dimension_type = args1.dimension_type
+        , flow_type = args1.flow_type
+        , rta_desc = ""
+        , change_time = biz_date
+        , biz_id = rowkey
+        , info_risk_level = args1.info_risk_level
+        , `type` = `type1`
+        , create_time = update_time
+        , amt = get_money(new_data)
+        , update_type = update_type
+        , deleted = get_deleted(new_data)
+      )
+      val `type2` = "29"
+      val args2 = CompanyMonitorMappings.get_args(`type2`)
+      list = list :+ CompanyMonitorRecord(
+        id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type2`)
+        , entity_info = Seq(EntityInfo(keyno = new_data.getOrEmptyStr("company_id"), name = new_data.getOrEmptyStr("company_name"), type_id = `type2`))
+        , dimension_type = args2.dimension_type
+        , flow_type = args2.flow_type
+        , rta_desc = ""
+        , change_time = biz_date
+        , biz_id = rowkey
+        , info_risk_level = args2.info_risk_level
+        , `type` = `type2`
+        , create_time = update_time
+        , amt = get_money(new_data)
+        , update_type = update_type
+        , deleted = get_deleted(new_data)
+      )
+    } else if (StringUtils.isNotBlank(new_data.getOrEmptyStr("lf_execute_order_num"))) {
+      //解冻信息
+      val `type3` = "30"
+      val args3 = CompanyMonitorMappings.get_args(`type3`)
+      list = list :+ CompanyMonitorRecord(
+        id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type3`)
+        , entity_info = Seq(EntityInfo(keyno = new_data.getOrEmptyStr("executed_person_id"), name = new_data.getOrEmptyStr("executed_person"), type_id = `type3`))
+        , dimension_type = args3.dimension_type
+        , flow_type = args3.flow_type
+        , rta_desc = ""
+        , change_time = biz_date
+        , biz_id = rowkey
+        , info_risk_level = args3.info_risk_level
+        , `type` = `type3`
+        , create_time = update_time
+        , amt = get_money(new_data)
+        , update_type = update_type
+        , deleted = get_deleted(new_data)
+      )
+      val `type4` = "31"
+      val args4 = CompanyMonitorMappings.get_args(`type4`)
+      list = list :+ CompanyMonitorRecord(
+        id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type4`)
+        , entity_info = Seq(EntityInfo(keyno = new_data.getOrEmptyStr("company_id"), name = new_data.getOrEmptyStr("company_name"), type_id = `type4`))
+        , dimension_type = args4.dimension_type
+        , flow_type = args4.flow_type
+        , rta_desc = ""
+        , change_time = biz_date
+        , biz_id = rowkey
+        , info_risk_level = args4.info_risk_level
+        , `type` = `type4`
+        , create_time = update_time
+        , amt = get_money(new_data)
+        , update_type = update_type
+        , deleted = get_deleted(new_data)
+      )
+    }
+    list
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_land_announcement.scala

@@ -0,0 +1,23 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+import com.winhc.bigdata.spark.utils.BaseUtil
+import com.winhc.bigdata.spark.implicits.MapHelper._
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 购买了新的土地-购地信息
+ */
+case class company_land_announcement(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override protected def get_money(new_map: Map[String, String]): Double = {
+    BaseUtil.amt_div(processMoney(new_map("tran_price")), "1")
+  }
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+
+    Seq(
+      EntityInfo(keyno = new_data.getOrEmptyStr("company_id"), name = new_data.getOrEmptyStr("application_name"), type_id = "12")
+    )
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_land_mortgage.scala

@@ -0,0 +1,72 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, ChangeExtract, CompanyMonitorMappings, CompanyMonitorRecord, CompanyMonitorUtils, EntityInfo}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.utils.BaseUtil
+
+import scala.collection.mutable
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 抵押了公司土地-抵押人
+ *               接受了他人土地抵押-抵押权人
+ */
+case class company_land_mortgage(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override protected def get_money(new_map: Map[String, String]): Double = {
+    BaseUtil.amt_div(processMoney(new_map("mortgage_amount")), "1")
+  }
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    Seq.empty[EntityInfo]
+  }
+
+  override def flat_map: (ChangeExtract) => Seq[CompanyMonitorRecord] = (change_extract: ChangeExtract) => {
+    var list: mutable.Seq[CompanyMonitorRecord] = mutable.Seq.empty
+
+    val tn = change_extract.tn
+    val update_type = change_extract.update_type
+    val new_data = change_extract.new_data
+    val biz_date = change_extract.biz_date
+    val rowkey = change_extract.rowkey
+    val update_time = change_extract.update_time
+
+    //抵押了公司土地
+    val `type1` = "13"
+    val args1 = CompanyMonitorMappings.get_args(`type1`)
+    list = list :+ CompanyMonitorRecord(
+      id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type1`)
+      , entity_info = Seq(EntityInfo(keyno = new_data.getOrEmptyStr("mortgagor_company_id"), name = new_data.getOrEmptyStr("mortgagor"), type_id = `type1`))
+      , dimension_type = args1.dimension_type
+      , flow_type = args1.flow_type
+      , rta_desc = ""
+      , change_time = biz_date
+      , biz_id = rowkey
+      , info_risk_level = args1.info_risk_level
+      , `type` = `type1`
+      , create_time = update_time
+      , amt = get_money(new_data)
+      , update_type = update_type
+      , deleted = get_deleted(new_data)
+    )
+    val `type2` = "21"
+    val args2 = CompanyMonitorMappings.get_args(`type2`)
+    //接受了他人土地抵押
+    list = list :+ CompanyMonitorRecord(
+      id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type2`)
+      , entity_info = Seq(EntityInfo(keyno = new_data.getOrEmptyStr("mortgagee_company_id"), name = new_data.getOrEmptyStr("mortgagee"), type_id = `type2`))
+      , dimension_type = args2.dimension_type
+      , flow_type = args2.flow_type
+      , rta_desc = ""
+      , change_time = biz_date
+      , biz_id = rowkey
+      , info_risk_level = args2.info_risk_level
+      , `type` = `type2`
+      , create_time = update_time
+      , amt = get_money(new_data)
+      , update_type = update_type
+      , deleted = get_deleted(new_data)
+    )
+    list
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_license.scala

@@ -0,0 +1,17 @@
+
+
+package com.winhc.bigdata.spark.ng.monitor.table
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+
+/**
+ * @author: π
+ * @date: 2021/9/15 16:44
+ */
+case class company_license(is_inc:Boolean) extends AbstractDailyHandle {
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    Seq(EntityInfo(keyno = new_data.getOrEmptyStr("company_id"), name = new_data.getOrEmptyStr("company_name"),type_id = "26"))
+  }
+
+}
+

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_mortgage_info.scala

@@ -0,0 +1,20 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+import com.winhc.bigdata.spark.utils.BaseUtil
+import com.winhc.bigdata.spark.implicits.MapHelper._
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 资产被抵押-动产抵押
+ */
+case class company_mortgage_info(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override protected def get_money(new_map: Map[String, String]): Double = {
+    BaseUtil.amt_div(processMoney(new_map("amount")), "1")
+  }
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    Seq(EntityInfo(keyno = new_data.getOrEmptyStr("company_id"), name = new_data.getOrEmptyStr("company_name"), type_id = "14"))
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_patent.scala

@@ -0,0 +1,14 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 新增专利信息
+ */
+case class company_patent(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    getEntity(new_data.getOrEmptyStr("applicant_name_info"), "keyno", "name","15")
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_tm.scala

@@ -0,0 +1,16 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 新增商标信息
+ */
+case class company_tm(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    Seq(EntityInfo(keyno = new_data.getOrEmptyStr("keyno"), name = new_data.getOrEmptyStr("applicant_cn"), type_id = "16"))
+  }
+
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_zxr.scala

@@ -0,0 +1,22 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 企业被执
+ */
+case class company_zxr(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override protected def get_money(new_map: Map[String, String]): Double = {
+
+    var amtstr = new_map("exec_money")
+    amtstr = processMoney(amtstr)
+    amtstr = (amtstr.toDouble/10000 ).formatted("%.2f")
+    amtstr.toDouble
+
+
+  }
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = ???
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/company_zxr_restrict.scala

@@ -0,0 +1,51 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, ChangeExtract, CompanyMonitorMappings, CompanyMonitorRecord, CompanyMonitorUtils, EntityInfo}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+
+import scala.collection.mutable
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 限制高消费移除
+ */
+case class company_zxr_restrict(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    Seq.empty[EntityInfo]
+  }
+
+  override def flat_map: (ChangeExtract) => Seq[CompanyMonitorRecord] = (change_extract: ChangeExtract) => {
+    var list: mutable.Seq[CompanyMonitorRecord] = mutable.Seq.empty
+
+    val tn = change_extract.tn
+    val update_type = change_extract.update_type
+    val new_data = change_extract.new_data
+    val biz_date = change_extract.biz_date
+    val rowkey = change_extract.rowkey
+    val update_time = change_extract.update_time
+
+    if (new_data.getOrEmptyStr("deleted").equals("1")) {
+      //限制高消费移除
+      val `type` = "18"
+      val args = CompanyMonitorMappings.get_args(`type`)
+      list = list :+ CompanyMonitorRecord(
+        id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type`)
+        , entity_info = Seq(EntityInfo(keyno = new_data.getOrEmptyStr("company_id"), name = new_data.getOrEmptyStr("company_name"), `type_id` = `type`)
+          , EntityInfo(keyno = new_data.getOrEmptyStr("pid"), name = new_data.getOrEmptyStr("person_name"), `type_id` = `type`))
+        , dimension_type = args.dimension_type
+        , flow_type = args.flow_type
+        , rta_desc = ""
+        , change_time = biz_date
+        , biz_id = rowkey
+        , info_risk_level = args.info_risk_level
+        , `type` = `type`
+        , create_time = update_time
+        , amt = get_money(new_data)
+        , update_type = update_type
+        , deleted = get_deleted(new_data)
+      )
+    }
+    list
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/increase_registered_capital_info.scala

@@ -0,0 +1,20 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.utils.BaseUtil
+
+/**
+ * @Date: 2020/12/14
+ * @Description: 注册资本上升-增资记录
+ */
+case class increase_registered_capital_info(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override protected def get_money(new_map: Map[String, String]): Double = {
+    BaseUtil.amt_div(processMoney(new_map("content_after")), "1")
+  }
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    Seq(EntityInfo(keyno = new_data.getOrEmptyStr("company_id"), name = new_data.getOrEmptyStr("company_name"),type_id = "2"))
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/wenshu_detail_v2.scala

@@ -0,0 +1,98 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, ChangeExtract, CompanyMonitorMappings, CompanyMonitorRecord, CompanyMonitorUtils, EntityInfo}
+import com.winhc.bigdata.spark.utils.BaseUtil
+import com.winhc.bigdata.spark.implicits.MapHelper._
+
+import scala.collection.mutable
+
+/**
+ * @Date: 2020/9/13 11:44
+ * @Description: 新增胜诉案件(原告) | AND data['is_success'] = '胜'  AND data['case_stage']= '一审'  AND  data['case_type'] = '民事案件' AND  data['name_type'] = 'y'
+ *               有恢复执行案件(被告) | AND  data['case_no'] like concat('%','恢','%')
+ *               新增财产保全案件(被告) | AND  data['case_no'] like concat('%','恢','%')
+ */
+case class wenshu_detail_v2(is_inc: Boolean) extends AbstractDailyHandle {
+
+
+  override protected def get_money(new_map: Map[String, String]): Double = {
+    BaseUtil.amt_div(processMoney(new_map("judge_amt")), "1")
+  }
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    Seq.empty[EntityInfo]
+  }
+
+  override def flat_map: (ChangeExtract) => Seq[CompanyMonitorRecord] = (change_extract: ChangeExtract) => {
+    var list: mutable.Seq[CompanyMonitorRecord] = mutable.Seq.empty
+
+    val tn = change_extract.tn
+    val update_type = change_extract.update_type
+    val new_data = change_extract.new_data
+    val biz_date = change_extract.biz_date
+    val rowkey = change_extract.rowkey
+    val update_time = change_extract.update_time
+
+    if (new_data.getOrEmptyStr("is_success").equals("胜")
+      && new_data.getOrEmptyStr("case_stage").equals("一审")
+      && new_data.getOrEmptyStr("case_type").equals("民事案件")) {
+      //新增胜诉案件
+      val `type` = "20"
+      val args1 = CompanyMonitorMappings.get_args(`type`)
+      list = list :+ CompanyMonitorRecord(
+        id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type`)
+        , entity_info = getEntity(new_data.getOrEmptyStr("plaintiff_info"), "litigant_id", "name",type_id = `type`)
+        , dimension_type = args1.dimension_type
+        , flow_type = args1.flow_type
+        , rta_desc = ""
+        , change_time = biz_date
+        , biz_id = rowkey
+        , info_risk_level = args1.info_risk_level
+        , `type` = `type`
+        , create_time = update_time
+        , amt = get_money(new_data)
+        , update_type = update_type
+        , deleted = get_deleted(new_data)
+      )
+    } else if (new_data.getOrEmptyStr("case_no").contains("恢")) {
+      val `type` = "19"
+      val args2 = CompanyMonitorMappings.get_args(`type`)
+      //有恢复执行案件
+      list = list :+ CompanyMonitorRecord(
+        id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type`)
+        , entity_info = getEntity(new_data.getOrEmptyStr("defendant_info"), "litigant_id", "name",type_id = `type`)
+        , dimension_type = args2.dimension_type
+        , flow_type = args2.flow_type
+        , rta_desc = ""
+        , change_time = biz_date
+        , biz_id = rowkey
+        , info_risk_level = args2.info_risk_level
+        , `type` = `type`
+        , create_time = update_time
+        , amt = get_money(new_data)
+        , update_type = update_type
+        , deleted = get_deleted(new_data)
+      )
+    } else if (new_data.getOrEmptyStr("case_no").contains("保")) {
+      val `type` = "27"
+      val args3 = CompanyMonitorMappings.get_args(`type`)
+      //新增财产保全案件
+      list = list :+ CompanyMonitorRecord(
+        id = CompanyMonitorUtils.generateId(rowkey, biz_date, tn, `type`)
+        , entity_info = getEntity(new_data.getOrEmptyStr("defendant_info"), "litigant_id", "name",type_id = `type`)
+        , dimension_type = args3.dimension_type
+        , flow_type = args3.flow_type
+        , rta_desc = ""
+        , change_time = biz_date
+        , biz_id = rowkey
+        , info_risk_level = args3.info_risk_level
+        , `type` = `type`
+        , create_time = update_time
+        , amt = get_money(new_data)
+        , update_type = update_type
+        , deleted = get_deleted(new_data)
+      )
+    }
+    list
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/zxr_evaluate.scala

@@ -0,0 +1,14 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+/**
+ * @Date: 2021/1/07 18:50
+ * @Description: 询价评估
+ */
+case class zxr_evaluate(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    Seq(EntityInfo(keyno = new_data.getOrEmptyStr("keyno"), name = new_data.getOrEmptyStr("name"),type_id = "23"))
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/monitor/table/zxr_evaluate_results.scala

@@ -0,0 +1,38 @@
+package com.winhc.bigdata.spark.ng.monitor.table
+
+import com.alibaba.fastjson.JSONArray
+import com.winhc.bigdata.spark.ng.monitor.{AbstractDailyHandle, EntityInfo}
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.utils.BaseUtil._
+import org.apache.commons.lang3.StringUtils
+import com.alibaba.fastjson.JSON
+import scala.collection.mutable.ListBuffer
+
+/**
+ * @Date: 2021/1/07 18:50
+ * @Description: 询价评估结果
+ */
+case class zxr_evaluate_results(is_inc: Boolean) extends AbstractDailyHandle {
+
+  override protected def get_money(new_map: Map[String, String]): Double = {
+    val money = new_map("money")
+    if (StringUtils.isBlank(money)) {
+      0D
+    } else {
+      val array: JSONArray = JSON.parseArray(money)
+      if (array.size() == 0) {
+        0D
+      } else {
+        val list = ListBuffer[Double]()
+        for (i <- 0 until array.size()) {
+          list.append(array.getString(i).toDouble)
+        }
+        amt_div(amt_div(list.sum.toString, list.size.toString).toString, "10000")
+      }
+    }
+  }
+
+  override def getEntityInfo(new_data: Map[String, String]): Seq[EntityInfo] = {
+    Seq(EntityInfo(keyno = new_data.getOrEmptyStr("keyno"), name = new_data.getOrEmptyStr("name"),type_id = "24"))
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/relation/lookup_tab_pid.scala

@@ -4,7 +4,7 @@ import com.alibaba.fastjson.{JSON, JSONArray, JSONPath}
 import com.winhc.bigdata.spark.config.EsConfig
 import com.winhc.bigdata.spark.udf.{BaseFunc, CompanyMapping}
 import com.winhc.bigdata.spark.utils.BaseUtil.{isWindows, is_json_str}
-import com.winhc.bigdata.spark.utils.{BaseUtil, CompanyRelationUtils, LoggingUtils, SparkUtils}
+import com.winhc.bigdata.spark.utils.{AsyncExtract, BaseUtil, CompanyRelationUtils, LoggingUtils, SparkUtils}
 import org.apache.commons.lang3.StringUtils
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.functions.{col, struct, to_json}
@@ -65,10 +65,10 @@ object args_job {
       , explode = "LATERAL VIEW explode( split(concat_ws('\u0001', cg_assignee_id, cg_executed_person_id, gn_executed_person_id, lf_executed_person_id, fz_executed_person_id, executed_person_id),'\u0001') ) key AS key_no"
       , keyno = "key_no")
 
-//    , args_job(tableName = "company_lawsuit"
-//      , explode = "LATERAL VIEW explode(jsonall_2_array('$.litigant_id', concat_ws('\u0001', defendant_info, plaintiff_info, litigant_info)) ) key AS key_no"
-//      , keyno = "key_no"
-//    )
+    //    , args_job(tableName = "company_lawsuit"
+    //      , explode = "LATERAL VIEW explode(jsonall_2_array('$.litigant_id', concat_ws('\u0001', defendant_info, plaintiff_info, litigant_info)) ) key AS key_no"
+    //      , keyno = "key_no"
+    //    )
 
     , args_job(tableName = "company"
       , rowkey = "company_id"
@@ -198,7 +198,7 @@ case class lookup_tab_pid(s: SparkSession
          |FROM    $tab_back_deleted
          |WHERE   ds > $ds
          |GROUP BY person_id
-         |""".stripMargin).createOrReplaceTempView("mapping")
+         |""".stripMargin).createOrReplaceTempView(s"mapping$tn")
 
     sql(
       s"""
@@ -217,18 +217,17 @@ case class lookup_tab_pid(s: SparkSession
          |                    ) AS t1
          |        ) AS t2
          |WHERE   t2.num = 1
-         |""".stripMargin).createOrReplaceTempView("tab_tmp")
+         |""".stripMargin).createOrReplaceTempView(s"tab_tmp$tn")
 
     sql(
       s"""
-         |INSERT OVERWRITE TABLE $tar_tab PARTITION(ds='$lastDs',tn='$tn')
+         |INSERT ${if (isWindows) "INTO" else "OVERWRITE"} TABLE $tar_tab PARTITION(ds='$lastDs',tn='$tn')
          |SELECT  ${tar_cols.mkString(",")},
-         |        -- get_table_message(${tar_cols.mkString(",")}, '$tn') message
          |        ${to_json(tar_cols ++ Seq("tn"))}
-         |FROM    mapping a
+         |FROM    mapping$tn a
          |JOIN    (
          |            SELECT  *,'$tn' as tn
-         |            FROM  tab_tmp
+         |            FROM  tab_tmp$tn
          |            ${args_job.explode}
          |        ) b
          |ON      a.person_id = b.$keyno
@@ -241,11 +240,11 @@ case class lookup_tab_pid(s: SparkSession
 
   }
 
-  private def to_json(seq:Seq[String]): String = {
+  private def to_json(seq: Seq[String]): String = {
     val r1 = seq.map(x => {
       s"'$x',$x"
     }).mkString(",")
-   s"to_json(map($r1)) message"
+    s"to_json(map($r1)) message"
   }
 
   private def get_partition_order_by(): String = {
@@ -265,15 +264,24 @@ object lookup_tab_pid {
       println("please set project tn.")
       sys.exit(-1)
     }
-    val Array(project, tn) = args
+    val Array(project, tableName) = args
     val config = EsConfig.getEsConfigMap ++ mutable.Map(
       "spark.hadoop.odps.project.name" -> project,
       "spark.debug.maxToStringFields" -> "200",
       "spark.hadoop.odps.spark.local.partition.amt" -> "1000"
     )
     val spark = SparkUtils.InitEnv(this.getClass.getSimpleName, config)
-    val re = lookup_tab_pid(s = spark, project = project, args_job.get_args_company_job(tn))
-    re.calc()
+    var start = args_job.tab_args
+    if (!tableName.equals("all")) {
+      val set = tableName.split(",").toSet
+      start = start.filter(a => set.contains(a.tableName))
+    }
+    val a = start.map(e => (e.tableName, () => {
+      val re = lookup_tab_pid(s = spark, project = project, args_job.get_args_company_job(e.tableName))
+      re.calc()
+      true
+    }))
+    AsyncExtract.startAndWait(spark, a)
     spark.stop()
   }
 }

src/main/scala/com/winhc/bigdata/spark/ng/utils/CompanySummaryNg_new.scala

@@ -64,6 +64,9 @@ object CompanySummaryNg_new {
     , get_default_summary_args("increase_registered_capital_info", "company_id")
     , get_default_summary_args("reduction_registered_capital_info", "company_id")
 
+    , get_default_summary_args("construction_qualification", "company_id")
+    , get_default_summary_args("construction_person", "company_id")
+    , get_default_summary_args("construction_project", "company_id")
 
     , SummaryArgs(table_name = "company_court_open_announcement_explode"
       , companyIdField = "plaintiff_info_id_explode"
@@ -282,6 +285,26 @@ object CompanySummaryNg_new {
       ))
     )
 
+
+    , SummaryArgs(table_name = "litigation_mediation_explode"
+      , companyIdField = "plaintiff_info_id_explode"
+      , distinctField = "rowkey,plaintiff_info_id_explode"
+      , where = "plaintiff_info_id_explode is not null and length(plaintiff_info_id_explode) = 32 "
+      , groupByInfo = GroupByInfoNg(field = "deleted", value_alias = Seq(
+        ("0", s"litigation_mediation_del_0_plaintiff")
+        , ("1", s"litigation_mediation_del_1_plaintiff")
+      ))
+    )
+
+    , SummaryArgs(table_name = "litigation_mediation_explode"
+      , companyIdField = "defendant_info_id_explode"
+      , distinctField = "rowkey,defendant_info_id_explode"
+      , where = "defendant_info_id_explode is not null and length(defendant_info_id_explode) = 32 "
+      , groupByInfo = GroupByInfoNg(field = "deleted", value_alias = Seq(
+        ("0", s"litigation_mediation_del_0_defendant")
+        , ("1", s"litigation_mediation_del_1_defendant")
+      ))
+    )
     // ====================================================
 
     , SummaryArgs(table_name = "bankruptcy_open_case_explode"
@@ -349,6 +372,21 @@ object CompanySummaryNg_new {
     , get_default_summary_args("company_mortgage_info", "company_id")
     , get_default_summary_args("company_brief_cancel_announcement", "company_id")
     , get_default_summary_args("company_double_random_check_info", "company_id")
+    , get_default_summary_args("company_wechat", "company_id")
+    , get_default_summary_args("company_annual_report", "company_id")
+
+
+
+    , get_default_summary_args("accounts_receivablae", "company_id")
+    , get_default_summary_args("dcm_registration_limit", "company_id")
+    , get_default_summary_args("bond_financing", "company_id")
+    , get_default_summary_args("overseas_bond_financing", "company_id")
+    , get_default_summary_args("bank_loan", "company_id")
+    , get_default_summary_args("credit_line", "company_id")
+    , get_default_summary_args("lease_financing", "company_id")
+    , get_default_summary_args("trust_financing", "company_id")
+    , get_default_summary_args("other_financing", "company_id")
+
 
     , SummaryArgs(table_name = "auction_tracking_explode"
       , companyIdField = "company_info_id_explode"
@@ -466,6 +504,30 @@ object CompanySummaryNg_new {
       ))
     )
 
+
+
+    , SummaryArgs(table_name = "company_bid_explode"
+      , companyIdField = "purchaser_info_id_explode"
+      , distinctField = "rowkey,purchaser_info_id_explode"
+      , where = "purchaser_info_id_explode is not null and length(purchaser_info_id_explode) = 32 "
+      , groupByInfo = GroupByInfoNg(field = "deleted", value_alias = Seq(
+        ("0", s"company_bid_del_0_purchaser")
+        , ("1", s"company_bid_del_1_purchaser")
+      ))
+    )
+
+    , SummaryArgs(table_name = "company_bid_explode"
+      , companyIdField = "supplier_info_id_explode"
+      , distinctField = "rowkey,supplier_info_id_explode"
+      , where = "supplier_info_id_explode is not null and length(supplier_info_id_explode) = 32 "
+      , groupByInfo = GroupByInfoNg(field = "deleted", value_alias = Seq(
+        ("0", s"company_bid_del_0_supplier")
+        , ("1", s"company_bid_del_1_supplier")
+      ))
+    )
+
+
+
   )
 
 

src/main/scala/com/winhc/bigdata/spark/ng/utils/PersonSummaryNg_new.scala

@@ -121,6 +121,30 @@ object PersonSummaryNg_new {
         , ("1", s"company_court_register_del_1_defendant")
       ))
     )
+
+
+
+    , SummaryArgs(table_name = "litigation_mediation_explode"
+      , companyIdField = "plaintiff_info_id_explode"
+      , distinctField = "rowkey,plaintiff_info_id_explode"
+      , where = "plaintiff_info_id_explode is not null and length(plaintiff_info_id_explode) = 33 "
+      , groupByInfo = GroupByInfoNg(field = "deleted", value_alias = Seq(
+        ("0", s"litigation_mediation_del_0_plaintiff")
+        , ("1", s"litigation_mediation_del_1_plaintiff")
+      ))
+    )
+
+    , SummaryArgs(table_name = "litigation_mediation_explode"
+      , companyIdField = "defendant_info_id_explode"
+      , distinctField = "rowkey,defendant_info_id_explode"
+      , where = "defendant_info_id_explode is not null and length(defendant_info_id_explode) = 33 "
+      , groupByInfo = GroupByInfoNg(field = "deleted", value_alias = Seq(
+        ("0", s"litigation_mediation_del_0_defendant")
+        , ("1", s"litigation_mediation_del_1_defendant")
+      ))
+    )
+
+
     , SummaryArgs(table_name = "company_send_announcement_explode"
       , companyIdField = "plaintiff_info_id_explode"
       , distinctField = "rowkey,plaintiff_info_id_explode"
@@ -169,6 +193,29 @@ object PersonSummaryNg_new {
       ))
     )
 
+    , SummaryArgs(table_name = "wenshu_detail_v2_explode"
+      , companyIdField = "plaintiff_info_id_explode"
+      , distinctField = "rowkey,plaintiff_info_id_explode"
+      , where = "plaintiff_info_id_explode is not null and length(plaintiff_info_id_explode) = 33 "
+      , groupByInfo = GroupByInfoNg(field = "deleted", value_alias = Seq(
+        ("0", s"wenshu_detail_v2_del_0_plaintiff")
+        , ("1", s"wenshu_detail_v2_del_1_plaintiff")
+      ))
+    )
+
+    , SummaryArgs(table_name = "wenshu_detail_v2_explode"
+      , companyIdField = "defendant_info_id_explode"
+      , distinctField = "rowkey,defendant_info_id_explode"
+      , where = "defendant_info_id_explode is not null and length(defendant_info_id_explode) = 33 "
+      , groupByInfo = GroupByInfoNg(field = "deleted", value_alias = Seq(
+        ("0", s"wenshu_detail_v2_del_0_defendant")
+        , ("1", s"wenshu_detail_v2_del_1_defendant")
+      ))
+    )
+
+
+
+
   )
 
 

src/main/scala/com/winhc/bigdata/spark/ng/utils/export_company_index_2_es.scala

@@ -213,7 +213,8 @@ object export_company_index_2_es {
       , "rowkey,defendant_info,plaintiff_info,court,start_date,title,case_no,deleted".split(","))
     , export_2_es_args("company_court_register"
       , "rowkey,case_no,case_reason,defendant_info,plaintiff_info,filing_date,deleted".split(","))
-
+    , export_2_es_args("litigation_mediation"
+      , "rowkey,case_no,case_reason,defendant_info,plaintiff_info,filing_date,deleted".split(","))
     , export_2_es_args("company_zxr_final_case"
       , "rowkey,name,keyno,court_name,case_final_time,case_create_time,case_no,exec_amount,no_exec_amount,status,deleted".split(",")
       , handles = Seq(field_handle(field_name = "exec_amount", handle = "parse_double(exec_amount)"),field_handle(field_name = "no_exec_amount", handle = "parse_double(no_exec_amount)")))
@@ -312,13 +313,39 @@ object export_company_index_2_es {
 
     , export_2_es_args("company_employment"
       , "rowkey,company_id,title,city,district,company_name,ori_salary,ori_salary_small,ori_salary_big,start_date,end_date,source,education,experience,location,create_time,update_time,deleted".split(","))
-//    , export_2_es_args("wenshu_detail_v2"
-//      , ("rowkey,case_id,doc_id,case_type,case_no,doc_type,case_reason_level1,case_reason_level2,case_reason_level3,case_reason_level4,case_reason,case_reason_levelnum," +
-//        "case_stage,new_case_stage,case_amt,judge_amt,accept_amt,province_court_name,city_court_name,court_name,court_province_code,court_city_code,court_level,party_info," +
-//        "people_info,yg_info,yg_type,plaintiff_info,yg_wtdlr,yg_faren,yg_lawyer,bg_info,bg_type,defendant_info,bg_wtdlr,bg_faren,bg_lawyer,third_party,danbao,fact,court_view," +
-//        "judge,judge_name,clerk,clerk_name,judge_date_cn,judge_date,judge_year,judge_result,is_success,url,title,addenda,keywords,public_date,update_date,sample_type,judge_main," +
-//        "all_connect_case_no,connect_case_no,litigant_info,party_title,first_case_yg,first_case_bg,all_lawyer,trial_verify,debate_focus,judge_result_point,associate_party,argue_info," +
-//        "lawsuit_point,plead_point,quote_legislation,deleted").split(","))
+    , export_2_es_args("company_wechat"
+      , "rowkey,company_id,title,img_url,public_num,auth,q_r_img_url,q_r_oss_path,recommend,create_time,update_time,deleted".split(","))
+    , export_2_es_args("company_bid"
+      , "rowkey,title,publish_time,purchaser_info,supplier_info,province,type,items,create_time,update_time,deleted".split(","))
+    , export_2_es_args("company_annual_report"
+      , "rowkey,company_id,report_year,create_time,update_time,deleted".split(","))
+//    融资
+    , export_2_es_args("accounts_receivablae"
+      , "rowkey,company_id,company_name,notice_date,pledgee_info,assignee_info,finance_orig,finance_type,deadline,start_date,end_date,pledge_asset,source,create_time,update_time,deleted".split(","))
+    , export_2_es_args("dcm_registration_limit"
+      , "rowkey,company_id,company_name,bond_lassify,init_regist,sumissue_scale,sumocc_limit,start_date,fileno,underwriter_info,source,create_time,update_time,deleted".split(","))
+    , export_2_es_args("bond_financing"
+      , "rowkey,company_id,company_name,abbreviation,symbol,typedesc_1,issued_value,coupon,maturity_date,begin_date,source,create_time,update_time,deleted".split(","))
+    , export_2_es_args("overseas_bond_financing"
+      , "rowkey,company_id,company_name,bond_simple_name,teminal_amount,issue_volume,coupons_will,residual_maturity,face_value,currency,source,create_time,update_time,deleted".split(","))
+    , export_2_es_args("bank_loan"
+      , "rowkey,company_id,company_name,declare_date,creditor,amount_financed,annual_interest_rate,expire_date,start_date,end_date,file_url,source,create_time,update_time,deleted".split(","))
+    , export_2_es_args("credit_line"
+      , "rowkey,company_id,company_name,end_date,discloure_date,credit_line,credit_line_used,credit_line_unused,credit_entity_count,source,create_time,update_time,deleted".split(","))
+    , export_2_es_args("lease_financing"
+      , "rowkey,company_id,company_name,notice_date,leaser_info,guaranteelabel,finance,equity,rate,deadline,start_date,end_date,leasing_method,source,create_time,update_time,deleted".split(","))
+    , export_2_es_args("trust_financing"
+      , "rowkey,company_id,company_name,notice_date,consignee_name_list,guarantor_name_list,finance_orig,annual_rate,deadline,start_date,end_date,file_url,source,create_time,update_time,deleted".split(","))
+    , export_2_es_args("other_financing"
+      , "rowkey,company_id,company_name,declare_date,financing_type,financing_amount,interest_rate,client_info,begin_date,end_date,file_url,source,create_time,update_time,deleted".split(","))
+
+    , export_2_es_args("construction_qualification"
+      , "rowkey,company_id,company_name,cert_type,cert_id,cert_name,organ_date,end_date,organ_name,source,create_time,update_time,deleted".split(","))
+    , export_2_es_args("construction_person"
+      , "rowkey,company_id,company_name,person_name,id_card,regtype_name,zczy_name,seal_code,reg_edate,source,create_time,update_time,deleted".split(","))
+    , export_2_es_args("construction_project"
+      , "rowkey,company_id,company_name,prj_name,county_num,prj_type,build_corp_name,money,create_time,update_time,deleted".split(","))
+
   )
 
 

src/main/scala/com/winhc/bigdata/spark/utils/AsyncExtract.scala

@@ -35,7 +35,7 @@ object AsyncExtract {
       .filter(_ != null)
       .map(i => i.status()).exists(i => JobExecutionStatus.FAILED.equals(i))
     if (failed) {
-      println("There are failed jobs !!!")
+      println("There are failed jobs -999 !!!")
       sys.exit(-999)
     }
   }
@@ -54,6 +54,7 @@ case class AsyncExtract(seq: Seq[(String, () => Boolean)]) extends Logging {
           println(s"---${e._1}---------$res-------------")
         } catch {
           case ex: Exception => {
+            println(s"---error-job: ${e._1}---------${ex.getMessage}-------------")
             error_count.incrementAndGet()
             ex.printStackTrace()
           }
@@ -64,7 +65,8 @@ case class AsyncExtract(seq: Seq[(String, () => Boolean)]) extends Logging {
     })
     latch.await()
     if (error_count.get() != 0) {
-      println("There are failed jobs !!!")
+      println("error_count: "+error_count.get())
+      println("There are failed jobs -998 !!!")
       sys.exit(-998)
     }
   }