利好消息增加

src/main/scala/com/winhc/bigdata/spark/ng/chance_v9/CompanyChanceHandleV9.scala

@@ -0,0 +1,145 @@
+package com.winhc.bigdata.spark.ng.chance_v9
+
+import com.alibaba.fastjson.JSON
+import com.winhc.bigdata.spark.implicits.CaseClass2JsonHelper
+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
+import scala.collection.mutable.ListBuffer
+
+/**
+ * @Author: π
+ * @Date: 2021/9/28
+ * @Description: 案源机会
+ */
+trait CompanyChanceHandleV9 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
+  }
+
+  /**
+   * 获取变更的业务时间
+   *
+   * @param newMap
+   * @return
+   */
+   def getBizDate(newMap: Map[String, String]): String
+
+  /**
+   * 业务id
+   *
+   * @param rowkey
+   * @return
+   */
+  protected def get_biz_id(rowkey: String, new_map: Map[String, String]): String = rowkey
+
+  /**
+   * 条件过滤
+   *
+   * @return
+   */
+  def get_conditional_filter(): String = ""
+
+  /**
+   * 加入标签
+   *
+   * @return
+   */
+  def get_label(new_map: Map[String, String]): String = ""
+
+  /**
+   * 原告-被告 集合
+   *
+   * @return
+   */
+  def get_party_set(new_map: Map[String, String]): Seq[String] = Seq("", "")
+
+  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 = CompanyChanceUtils.default_filter
+
+  def flat_map: (ChangeExtract) => Seq[CompanyChanceRecord]
+
+  def getCompanyIds(new_data: Map[String, String]): Seq[String]
+
+  protected def getEntity(json: String, id_key: String): Seq[String] = {
+    if (StringUtils.isEmpty(json)) {
+      Seq.empty
+    } else {
+      val array = JSON.parseArray(json)
+      var list: mutable.Seq[String] = mutable.Seq.empty
+      for (i <- 0 until array.size()) {
+        val jSONObject = array.getJSONObject(i)
+        val keyno = jSONObject.getString(id_key)
+        list = list :+ keyno
+      }
+      list
+    }
+  }
+
+  protected def getCompanyChanceRecord(change_extract: ChangeExtract
+                                       , companyIds: Seq[String]
+                                      ): Seq[CompanyChanceRecord] = {
+
+    val tn = change_extract.tn
+    val update_type = change_extract.update_type
+    val old_data = change_extract.old_data
+    var 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 ids = companyIds.filter(_.length == 32).distinct
+    val list = ListBuffer[CompanyChanceRecord]()
+    if (ids.nonEmpty) {
+      ids.foreach(id => {
+        if (new_data.contains("case_amt")) {
+          new_data += ("case_amt" -> BaseUtil.amt_div(BaseUtil.getOrEmptyStr(new_data, "case_amt"), "1").toString)
+        }
+        if (new_data.contains("judge_date")) {
+          new_data += ("judge_date" -> BaseUtil.getOrEmptyStr(new_data, "judge_date").split(" ")(0))
+        }
+        import com.winhc.bigdata.spark.implicits.CaseClass2JsonHelper._
+
+        val record: CompanyChanceRecord = CompanyChanceRecord(
+          id = CompanyChanceUtils.generateId(rowkey, biz_date, tn)
+          , rowkey = rowkey
+          , company_id = id
+          , title = ""
+          , label = get_label(new_data)
+          , change_time = biz_date
+          , update_time = update_time
+          , update_type = update_type
+          , deleted = get_deleted(new_data)
+          , new_data.toJson
+          , get_party_set(new_data).head
+          , get_party_set(new_data).last
+        )
+        list.append(record)
+      })
+    }
+    list
+  }
+
+}

src/main/scala/com/winhc/bigdata/spark/ng/chance_v9/CompanyChanceRecord.scala

@@ -0,0 +1,106 @@
+package com.winhc.bigdata.spark.ng.chance_v9
+
+import cn.hutool.crypto.SecureUtil
+import org.apache.commons.lang3.StringUtils
+import org.apache.spark.sql.Row
+
+/**
+ * @author: π
+ * @date: 2021/9/28 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
+                     ) 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 CompanyChanceRecord(id: String,
+                               rowkey: String,
+                               company_id: String,
+                               title: String,
+                               label: String,
+                               change_time: String,
+                               update_time: String,
+                               update_type: String,
+                               deleted: String,
+                               new_data: String,
+                               plaintiff: String,
+                               defendant: String
+                              ) {
+
+  def format(): CompanyChanceRecord = {
+    if (id == null) {
+      return null
+    }
+    //剔除id为空的数据
+    if (StringUtils.isBlank(company_id)) {
+      return null
+    }
+    return CompanyChanceRecord(
+      id,
+      rowkey,
+      company_id,
+      title,
+      label,
+      change_time,
+      update_time,
+      update_type,
+      deleted,
+      new_data,
+      plaintiff,
+      defendant
+    )
+    this
+  }
+
+  def to_row(): Row = {
+    Row(SecureUtil.md5(id)
+      , rowkey
+      , company_id
+      , title
+      , label
+      , change_time
+      , update_time
+      , update_type
+      , deleted
+      , new_data
+      , plaintiff
+      , defendant
+    )
+  }
+
+}
+
+

src/main/scala/com/winhc/bigdata/spark/ng/chance_v9/CompanyChanceUtils.scala

@@ -0,0 +1,33 @@
+package com.winhc.bigdata.spark.ng.chance_v9
+
+import org.apache.commons.lang3.StringUtils
+
+/**
+ * @author: π
+ * @date: 2021/9/28 15:46
+ */
+object CompanyChanceUtils {
+
+  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/chance_v9/GoodNewsV9.scala

@@ -0,0 +1,368 @@
+package com.winhc.bigdata.spark.ng.chance_v9
+
+import com.winhc.bigdata.spark.config.EsConfig
+import com.winhc.bigdata.spark.utils.ReflectUtils.getClazz
+import com.winhc.bigdata.spark.utils.{AsyncExtract, BaseUtil, LoggingUtils, SparkUtils}
+import org.apache.commons.lang3.StringUtils
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.types.StringType
+import com.winhc.bigdata.spark.udf.{BaseFunc, CaseChanceFunc}
+
+import scala.annotation.meta.getter
+import scala.collection.immutable.ListMap
+import scala.collection.mutable
+
+/**
+ * @Author: π
+ * @Date: 2021/9/9
+ * @Description: 利好消息
+ */
+object GoodNewsV9 {
+
+  case class CompanyMonitorUtil(s: SparkSession,
+                                project: String, //表所在工程名
+                                incr: Boolean //是否增量
+                               ) extends LoggingUtils with Logging with BaseFunc with CaseChanceFunc {
+    @(transient@getter) val spark: SparkSession = s
+
+    val inc_ads_eci_debtor_relation = "winhc_ng.inc_ads_eci_debtor_relation_v9"
+    val target_ads_case_chance = "winhc_ng.ads_case_chance_good_news_v9"
+    val target_ads_case_chance_element = "winhc_ng.ads_case_chance_element_good_news_v9"
+
+    //private lazy val org_tab = if (incr) "winhc_ng.bds_change_extract" else "winhc_ng.bds_change_extract_all_v2"
+    private lazy val org_tab = "winhc_ng.bds_change_extract_v9"
+    private var ds = getLastPartitionsOrElse(org_tab, "0")
+    ds= BaseUtil.getYesterday()
+    if (!incr) {
+      ds = "0"
+    }
+
+    val relation_cols: Seq[String] = getColumns(inc_ads_eci_debtor_relation).filter(!_.equals("ds"))
+    val eci_debtor_rel_ds: String = getLastPartitionsOrElse(inc_ads_eci_debtor_relation, "0")
+
+
+    //不同name映射table
+    val tabMapping =
+      Map("company_court_announcement_v1" -> "company_court_announcement" //开庭公告-被告
+        , "wenshu_detail_v2_bg_yishen" -> "wenshu_detail_v2" //文书
+        , "wenshu_detail_v2_yg_yishen" -> "wenshu_detail_v2" //文书
+        , "wenshu_detail_v2_yg_zhongben" -> "wenshu_detail_v2" //文书
+      )
+
+    //转换字段
+    def trans(s: String): String = {
+      if (tabMapping.contains(s)) return tabMapping(s)
+      s
+    }
+
+    def calc(tableName1: String, mode: String): Unit = {
+      cleanup()
+      json_utils()
+      json_add_kv()
+      map_2_json()
+      trans_number()
+      chance_dynamic_type()
+
+      val tableName2 = trans(tableName1)
+
+      val clazz = getClazz[CompanyChanceHandleV9](s"com.winhc.bigdata.spark.ng.chance_v9.table.$tableName1", incr)
+      val conditional = clazz.get_conditional_filter()
+      val filter = clazz.filter
+      val flat_map = clazz.flat_map
+      val tn = tableName2
+
+      //-- WHERE   ds = '$ds'
+      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(",")
+        //todo 待优化
+        val company_id = r.getAs("rowkey").toString.split("_")
+        val new_data = r.getAs[Map[String, String]]("new_data")
+        val biz_date = clazz.getBizDate(new_data)
+
+        ChangeExtract(rowkey = r.getAs("rowkey")
+          , company_id = company_id(0)
+          , company_name = null
+          , tn = r.getAs("tn")
+          , update_type = r.getAs("update_type")
+          , old_data = r.getAs("old_data")
+          , new_data = r.getAs("new_data")
+          , change_fields = change_fields
+          , biz_date = biz_date
+          , update_time = BaseUtil.nowDate()
+        )
+      }).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
+        , "rowkey" -> StringType
+        , "company_id" -> StringType
+        , "title" -> StringType
+        , "label" -> StringType
+        , "change_time" -> StringType
+        , "update_time" -> StringType
+        , "update_type" -> StringType
+        , "deleted" -> StringType
+        , "new_data" -> StringType
+        , "plaintiff" -> StringType
+        , "defendant" -> StringType
+      ))
+
+      spark.createDataFrame(rdd.map(_.to_row()), schema)
+        .createOrReplaceTempView(s"good_news_debtor_relation_view$tableName1")
+
+      mode match {
+        case "0" => {
+          sql(
+            s"""
+               |SELECT  *
+               |FROM    (
+               |            SELECT  ${relation_cols.map(n => s"$n as $n").mkString(",")}
+               |            FROM    $inc_ads_eci_debtor_relation
+               |            WHERE   ds = '$eci_debtor_rel_ds'
+               |            and deleted = 0
+               |        ) AS t1
+               |JOIN (
+               |         SELECT
+               |           *
+               |         FROM good_news_debtor_relation_view$tableName1
+               |      ) AS t2
+               |ON      t1.bg_company_id = t2.company_id
+               |WHERE t1.bg_name not like concat('%','银行','%') AND t1.bg_name not like concat('%','保险','%')
+               |""".stripMargin).createOrReplaceTempView(s"good_news_debtor_relation_view${tableName1}v2")
+
+
+          sql(
+            s"""
+               |INSERT  INTO TABLE $target_ads_case_chance_element PARTITION(ds='$ds', tn='$tableName1')
+               |SELECT  md5(cleanup(CONCAT_WS('',case_chance_id,case_chance_type,type,province,city,dynamic_time))) AS id
+               |        ,CASE_CHANCE_ID
+               |        ,TYPE
+               |        ,province
+               |        ,city
+               |        ,county
+               |        ,dynamic_time
+               |        ,public_date
+               |        ,CASE_CHANCE_TYPE
+               |FROM    (
+               |            SELECT  *
+               |                    ,ROW_NUMBER() OVER(PARTITION BY CASE_CHANCE_ID,TYPE,PROVINCE,city ORDER BY CASE_CHANCE_ID) AS num
+               |            FROM    (
+               |                        SELECT   rowkey AS CASE_CHANCE_ID
+               |                                ,0 AS TYPE
+               |                                ,get_table_type('$tableName1') AS CASE_CHANCE_TYPE
+               |                                ,yg_province_code AS PROVINCE
+               |                                ,yg_city_code AS city
+               |                                ,yg_county_code AS county
+               |                                ,change_time AS dynamic_time
+               |                                ,update_time AS public_date
+               |                        FROM    good_news_debtor_relation_view${tableName1}v2
+               |                        UNION ALL
+               |                        SELECT   rowkey AS CASE_CHANCE_ID
+               |                                ,1 AS TYPE
+               |                                ,get_table_type('$tableName1') AS CASE_CHANCE_TYPE
+               |                                ,bg_province_code AS PROVINCE
+               |                                ,bg_city_code AS city
+               |                                ,bg_county_code AS county
+               |                                ,change_time AS dynamic_time
+               |                                ,update_time AS public_date
+               |                        FROM    good_news_debtor_relation_view${tableName1}v2
+               |                    )
+               |        ) AS t
+               |WHERE   t.num = 1
+               |""".stripMargin)
+
+          sql(
+            s"""
+               |
+               |INSERT INTO TABLE $target_ads_case_chance PARTITION(ds='$ds', tn='$tableName1')
+               |SELECT  rowkey AS case_chance_id
+               |        -- ,title AS title
+               |        ,bg_name as title
+               |        ,plaintiff
+               |        ,defendant
+               |        ,bg_name AS company_name
+               |        ,company_id
+               |        ,json_add_str(label,CONCAT_WS(',',get_json_kv('reg_capital',trans_number(bg_reg_capital)),get_json_kv('province_code',bg_province_code),get_json_kv('city_code',bg_city_code),get_json_kv('county_code',bg_county_code),get_json_kv('estiblish_time',bg_estiblish_time) ,get_json_kv('province_name',bg_province_name),get_json_kv('city_name',bg_city_name),get_json_kv('county_name',bg_county_name),get_json_kv('category_first',bg_category_first_name)  )) AS tags
+               |        ,rowkey AS biz_id
+               |        ,get_table_type('$tableName1') AS type
+               |        ,get_chance_dynamic_type('$tableName1') AS dynamic_type
+               |        ,new_data AS dynamic_content
+               |        ,change_time AS dynamic_time
+               |        ,update_time AS public_date
+               |FROM    (
+               |            SELECT  *
+               |                    ,ROW_NUMBER() OVER(PARTITION BY rowkey ORDER BY rowkey) AS num
+               |            FROM    good_news_debtor_relation_view${tableName1}v2
+               |        ) AS t
+               |WHERE   t.num = 1
+               |""".stripMargin)
+        }
+        case "1" => {
+
+          sql(
+            s"""
+               |SELECT  t2.*,
+               |         name,
+               |         province_code,
+               |         city_code,
+               |         county_code,
+               |         reg_capital,
+               |         estiblish_time,
+               |         phones,
+               |         cate_first_code,
+               |         cate_second_code,
+               |         cate_third_code
+               |FROM    (
+               |            SELECT  * FROM winhc_ng.tmp_xf_company_all
+               |        ) AS t1
+               |JOIN (
+               |         SELECT
+               |           *
+               |         FROM good_news_debtor_relation_view$tableName1
+               |      ) AS t2
+               |ON      t1.company_id = t2.company_id
+               |WHERE t1.name not like concat('%','银行','%') AND t1.name not like concat('%','保险','%')
+               |""".stripMargin).createOrReplaceTempView(s"good_news_debtor_relation_view${tableName1}v2")
+
+          //债务人要素表
+          sql(
+            s"""
+               |INSERT INTO TABLE $target_ads_case_chance_element  PARTITION (ds=$ds, tn='$tableName1')
+               |SELECT   md5(cleanup(CONCAT_WS('',case_chance_id,case_chance_type,type,province_code,city_code,dynamic_time))) AS id
+               |        ,case_chance_id
+               |        ,type
+               |        ,province_code
+               |        ,city_code
+               |        ,county_code
+               |        ,dynamic_time
+               |        ,public_date
+               |        ,case_chance_type
+               |FROM
+               |(
+               |select
+               |         rowkey AS case_chance_id
+               |        ,0 AS type
+               |        ,province_code
+               |        ,city_code
+               |        ,county_code
+               |        ,change_time AS dynamic_time
+               |        ,update_time AS public_date
+               |        ,get_table_type('$tableName1') AS case_chance_type
+               |from good_news_debtor_relation_view${tableName1}v2
+               |)
+               |WHERE trim(province_code) <> ''
+               |""".stripMargin)
+
+          //案源机会表
+          sql(
+            s"""
+               |INSERT INTO TABLE $target_ads_case_chance  PARTITION (ds=$ds, tn='$tableName1')
+               |SELECT
+               |     rowkey as case_chance_id
+               |    ,title
+               |    ,plaintiff
+               |    ,defendant
+               |    ,name as company_name
+               |    ,company_id
+               |    ,label as tags
+               |    ,rowkey as biz_id
+               |    ,get_table_type('$tableName1') AS type
+               |    ,get_chance_dynamic_type('$tableName1') AS dynamic_type
+               |    ,new_data AS dynamic_content
+               |    ,change_time AS dynamic_time
+               |    ,update_time AS public_date
+               |FROM good_news_debtor_relation_view${tableName1}v2
+               |""".stripMargin)
+
+        }
+      }
+    }
+  }
+
+  private val startArgs = Seq(
+    Args(tableName = "company_equity_info")
+    , Args(tableName = "company_tm")
+    , Args(tableName = "company_patent")
+    , Args(tableName = "company_copyright_works")
+    , Args(tableName = "company_copyright_reg")
+    , Args(tableName = "company_land_publicity")
+    , Args(tableName = "company_land_announcement")
+    //, Args(tableName = "company_bid")
+    , Args(tableName = "company_land_transfer")
+    , Args(tableName = "company_employment")
+    , Args(tableName = "company_certificate")
+    , Args(tableName = "company_zxr_restrict")
+    , Args(tableName = "company_zxr")
+    , Args(tableName = "company_dishonest_info")
+    , Args(tableName = "company_court_announcement")
+    , Args(tableName = "bankruptcy_open_case")
+    , Args(tableName = "wenshu_detail_v2_bg_yishen")
+    , Args(tableName = "company_court_announcement_v1", mode = "1")
+    , Args(tableName = "wenshu_detail_v2_yg_yishen", mode = "1")
+    , Args(tableName = "wenshu_detail_v2_yg_zhongben", mode = "1")
+  )
+
+  private case class Args(project: String = "winhc_ng"
+                          , tableName: String
+                          , mode: 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.mode) //通用处理
+      true
+    }))
+
+    AsyncExtract.startAndWait(spark, a)
+    spark.stop()
+  }
+}

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

@@ -0,0 +1,55 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+import org.json4s.DefaultFormats
+import org.json4s.jackson.Json
+
+
+/**
+ * @Date: 2021/9/29
+ * @Description: 破产公告
+ */
+case class bankruptcy_open_case(is_inc: Boolean) extends CompanyChanceHandleV9 {
+
+  override def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("public_date"))
+
+  override def get_conditional_filter(): String = {
+    ""
+  }
+
+  override def get_label(new_data: Map[String, String]): String = {
+    val m: Map[String, String] = Map(
+      "agency_court" -> new_data.getOrEmptyStr("agency_court"),
+      "case_type" -> new_data.getOrEmptyStr("case_type")
+    )
+    Json(DefaultFormats).write(m)
+  }
+
+  override def get_party_set(new_data: Map[String, String]): Seq[String] = {
+    Seq(
+      new_data.getOrEmptyStr("applicant"),
+      new_data.getOrEmptyStr("respondent")
+    )
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    getEntity(new_data.getOrEmptyStr("respondent_info"), "litigant_id")
+  }
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

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

@@ -0,0 +1,37 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description:招投标
+ */
+case class company_bid(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("publish_time"))
+
+  override def get_conditional_filter(): String = {
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    getEntity(new_data.getOrEmptyStr("purchaser_info"), "keyno")
+  }
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+
+
+}

src/main/scala/com/winhc/bigdata/spark/ng/chance_v9/table/company_certificate.scala

@@ -0,0 +1,36 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description:资质证书
+ */
+case class company_certificate(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("start_date"))
+
+  override def get_conditional_filter(): String = {
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    Seq(new_data.getOrEmptyStr("company_id"))
+  }
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

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

@@ -0,0 +1,35 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description:软件著作权
+ */
+case class company_copyright_reg(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("reg_time"))
+
+  override def get_conditional_filter(): String = {
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    getEntity(new_data.getOrEmptyStr("author_nationality_info"), "keyno")
+  }
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

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

@@ -0,0 +1,36 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description:作品著作权
+ */
+case class company_copyright_works(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("reg_time"))
+
+  override def get_conditional_filter(): String = {
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    getEntity(new_data.getOrEmptyStr("author_info"), "keyno")
+  }
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/chance_v9/table/company_court_announcement.scala

@@ -0,0 +1,57 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+import org.json4s.{DefaultFormats, Formats, NoTypeHints}
+import org.json4s.jackson.{Json, Serialization}
+import org.json4s.jackson.Serialization.write
+
+
+/**
+ * @Date: 2021/9/29
+ * @Description: 法院公告-原告
+ */
+case class company_court_announcement(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String =  DateUtils.getBizDate(newMap("publish_date"))
+
+  override def get_conditional_filter(): String = {
+    "AND  new_data['announcement_type'] = '起诉状副本及开庭传票' "
+  }
+
+  override def get_label(new_data: Map[String, String]): String = {
+    val m: Map[String, String] = Map(
+      "announcement_type" -> new_data.getOrEmptyStr("announcement_type"),
+      "publish_date" -> new_data.getOrEmptyStr("publish_date")
+    )
+    Json(DefaultFormats).write(m)
+  }
+
+  override def get_party_set(new_data: Map[String, String]): Seq[String] = {
+    Seq(
+      new_data.getOrEmptyStr("plaintiff"),
+      new_data.getOrEmptyStr("litigant")
+    )
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    //Seq(new_data.getOrEmptyStr("related_company_id")) ++
+    getEntity(new_data.getOrEmptyStr("plaintiff_info"), "litigant_id") //++
+    //getEntity(new_data.getOrEmptyStr("pledgee_info"), "pledgee_id")
+  }
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/chance_v9/table/company_court_announcement_v1.scala

@@ -0,0 +1,54 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+import org.json4s.DefaultFormats
+import org.json4s.jackson.Json
+
+
+/**
+ * @Date: 2021/9/29
+ * @Description: 法院公告-被告
+ */
+case class company_court_announcement_v1(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String =  DateUtils.getBizDate(newMap("publish_date"))
+
+  override def get_conditional_filter(): String = {
+    "AND  new_data['announcement_type'] = '起诉状副本及开庭传票' "
+  }
+
+  override def get_label(new_data: Map[String, String]): String = {
+    val m: Map[String, String] = Map(
+      "announcement_type" -> new_data.getOrEmptyStr("announcement_type"),
+      "publish_date" -> new_data.getOrEmptyStr("publish_date")
+    )
+    Json(DefaultFormats).write(m)
+  }
+
+  override def get_party_set(new_data: Map[String, String]): Seq[String] = {
+    Seq(
+      new_data.getOrEmptyStr("plaintiff"),
+      new_data.getOrEmptyStr("litigant")
+    )
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    getEntity(new_data.getOrEmptyStr("litigant_info"), "litigant_id")
+  }
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

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

@@ -0,0 +1,37 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description: 开庭公告
+ */
+case class company_court_open_announcement(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String =  DateUtils.getBizDate(newMap("start_date"))
+
+  override def get_conditional_filter(): String ={
+    "AND  new_data['case_no'] like concat('%','恢','%')"
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    //Seq(new_data.getOrEmptyStr("related_company_id")) ++
+      getEntity(new_data.getOrEmptyStr("defendant_info"), "litigant_id") //++
+      //getEntity(new_data.getOrEmptyStr("pledgee_info"), "pledgee_id")
+  }
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

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

@@ -0,0 +1,35 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description:失信人
+ */
+case class company_dishonest_info(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String =  DateUtils.getBizDate(newMap("pub_date"))
+
+  override def get_conditional_filter(): String = {
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("deleted")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    Seq(new_data.getOrEmptyStr("keyno"))
+  }
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/chance_v9/table/company_employment.scala

@@ -0,0 +1,36 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description:招聘
+ */
+case class company_employment(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("start_date"))
+
+  override def get_conditional_filter(): String = {
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    Seq(new_data.getOrEmptyStr("company_id"))
+  }
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

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

@@ -0,0 +1,38 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description: 股权出质-出质人
+ */
+case class company_equity_info(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("reg_date"))
+
+  override def get_conditional_filter(): String ={
+    //"AND  new_data['case_no'] like concat('%','恢','%')"
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    //Seq(new_data.getOrEmptyStr("related_company_id")) ++
+      getEntity(new_data.getOrEmptyStr("pledgor_info"), "pledgor_id") //++
+      //getEntity(new_data.getOrEmptyStr("pledgee_info"), "pledgee_id")
+  }
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

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

@@ -0,0 +1,34 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+
+/**
+ * @Date: 2021/9/29
+ * @Description:股东
+ */
+case class company_holder(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String = newMap("update_time")
+
+  override def get_conditional_filter(): String = {
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert") || update_type.equals("create")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    Seq(new_data.getOrEmptyStr("company_id"))
+  }
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

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

@@ -0,0 +1,36 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description:购地信息
+ */
+case class company_land_announcement(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String =  DateUtils.getBizDate(newMap("contract_date"))
+
+  override def get_conditional_filter(): String = {
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    Seq(new_data.getOrEmptyStr("company_id"))
+  }
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/chance_v9/table/company_land_publicity.scala

@@ -0,0 +1,36 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description:地块公示
+ */
+case class company_land_publicity(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override  def getBizDate(newMap: Map[String, String]): String =  DateUtils.getBizDate(newMap("publication_date"), newMap("publication_start_date"))
+
+  override def get_conditional_filter(): String = {
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    Seq(new_data.getOrEmptyStr("company_id"))
+  }
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/chance_v9/table/company_land_transfer.scala

@@ -0,0 +1,35 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description:土地转让
+ */
+case class company_land_transfer(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String =  DateUtils.getBizDate(newMap("merchandise_time"))
+
+  override def get_conditional_filter(): String = {
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    Seq(new_data.getOrEmptyStr("pre_keyno"))
+  }
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

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

@@ -0,0 +1,35 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description:专利
+ */
+case class company_patent(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("app_date"))
+
+  override def get_conditional_filter(): String ={
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    getEntity(new_data.getOrEmptyStr("applicant_name_info"), "keyno")
+  }
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

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

@@ -0,0 +1,35 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description:商标
+ */
+case class company_tm(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("app_date"), newMap("update_time"))
+
+  override def get_conditional_filter(): String = {
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    Seq(new_data.getOrEmptyStr("keyno"))
+  }
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

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

@@ -0,0 +1,35 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description:被执
+ */
+case class company_zxr(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String =  DateUtils.getBizDate(newMap("case_create_time"))
+
+  override def get_conditional_filter(): String = {
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("deleted")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    Seq(new_data.getOrEmptyStr("keyno"))
+  }
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

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

@@ -0,0 +1,35 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description:限高
+ */
+case class company_zxr_restrict(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String =  DateUtils.getBizDate(newMap("case_create_time"))
+
+  override def get_conditional_filter(): String = {
+    ""
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("deleted")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    Seq(new_data.getOrEmptyStr("company_id"))
+  }
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

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

@@ -0,0 +1,37 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.DateUtils
+
+/**
+ * @Date: 2021/9/29
+ * @Description: 裁判文书-恢复执行
+ */
+case class wenshu_detail_v2(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("publish_date"),newMap("judge_date"))
+
+  override def get_conditional_filter(): String ={
+    "AND  new_data['case_no'] like concat('%','恢','%')"
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    //Seq(new_data.getOrEmptyStr("related_company_id")) ++
+      getEntity(new_data.getOrEmptyStr("defendant_info"), "litigant_id") //++
+      //getEntity(new_data.getOrEmptyStr("pledgee_info"), "pledgee_id")
+  }
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/chance_v9/table/wenshu_detail_v2_bg_yishen.scala

@@ -0,0 +1,55 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.{BaseUtil, DateUtils}
+import org.json4s.DefaultFormats
+import org.json4s.jackson.Json
+
+/**
+ * @Date: 2021/9/29
+ * @Description: 裁判文书-被告-一审
+ */
+case class wenshu_detail_v2_bg_yishen(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("publish_date"),newMap("judge_date"))
+
+  override def get_conditional_filter(): String = {
+    " AND  new_data['is_success'] = '胜' AND new_data['case_stage'] = '一审'  AND new_data['case_type'] = '民事案件' "
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def get_label(new_data: Map[String, String]): String = {
+    BaseUtil.getTags(new_data,"case_amt","judge_date","4")
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    //Seq(new_data.getOrEmptyStr("related_company_id")) ++
+    getEntity(new_data.getOrEmptyStr("defendant_info"), "litigant_id") //++
+    //getEntity(new_data.getOrEmptyStr("pledgee_info"), "pledgee_id")
+  }
+
+  /**
+   * 原告-被告 集合
+   *
+   * @return
+   */
+  override def get_party_set(new_map: Map[String, String]): Seq[String] = {
+    Seq(getEntity(new_map.getOrEmptyStr("plaintiff_info"), "name").mkString(",")
+      , getEntity(new_map.getOrEmptyStr("defendant_info"), "name").mkString(","))
+  }
+
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/chance_v9/table/wenshu_detail_v2_yg_yishen.scala

@@ -0,0 +1,50 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.{BaseUtil, DateUtils}
+import org.json4s.DefaultFormats
+import org.json4s.jackson.Json
+
+
+/**
+ * @Date: 2021/9/29
+ * @Description: 裁判文书 - 原告 - 一审
+ */
+case class wenshu_detail_v2_yg_yishen(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("publish_date"),newMap("judge_date"))
+
+  override def get_conditional_filter(): String = {
+    "AND  new_data['is_success'] = '胜' AND new_data['case_stage'] = '一审'  AND new_data['case_type'] = '民事案件'"
+  }
+
+  override def get_label(new_data: Map[String, String]): String = {
+    BaseUtil.getTags(new_data,"case_amt","judge_date","5")
+  }
+
+  override def get_party_set(new_data: Map[String, String]): Seq[String] = {
+    Seq(
+      getEntity(new_data.getOrEmptyStr("plaintiff_info"), "name").mkString(","),
+      getEntity(new_data.getOrEmptyStr("defendant_info"), "name").mkString(",")
+    )
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    getEntity(new_data.getOrEmptyStr("plaintiff_info"), "litigant_id")
+  }
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

src/main/scala/com/winhc/bigdata/spark/ng/chance_v9/table/wenshu_detail_v2_yg_zhongben.scala

@@ -0,0 +1,55 @@
+package com.winhc.bigdata.spark.ng.chance_v9.table
+
+import com.winhc.bigdata.spark.implicits.MapHelper._
+import com.winhc.bigdata.spark.ng.chance_v9.{ChangeExtract, CompanyChanceRecord, CompanyChanceUtils}
+import com.winhc.bigdata.spark.ng.chance_v9.CompanyChanceHandleV9
+import com.winhc.bigdata.spark.utils.{BaseUtil, DateUtils}
+import org.json4s.DefaultFormats
+import org.json4s.jackson.Json
+
+
+/**
+ * @Date: 2021/9/29
+ * @Description: 裁判文书 - 原告 - 终本
+ */
+case class wenshu_detail_v2_yg_zhongben(is_inc: Boolean) extends CompanyChanceHandleV9 {
+  override def getBizDate(newMap: Map[String, String]): String = DateUtils.getBizDate(newMap("publish_date"),newMap("judge_date"))
+
+  override def get_conditional_filter(): String = {
+    "AND  new_data['judge_result'] like '%终结本次执行程序%' AND new_data['case_stage'] = '一审'  AND new_data['case_type'] = '民事案件'"
+  }
+
+  override def get_label(new_data: Map[String, String]): String = {
+    BaseUtil.getTags(new_data,"case_amt","judge_date","6")
+//    val m: Map[String, String] = Map(
+//      "case_amt" -> BaseUtil.amt_div(new_data.getOrEmptyStr("case_amt"), "1",2).toString,
+//      "judge_date" -> new_data.getOrEmptyStr("judge_date").split(" ")(0)
+//    )
+//    Json(DefaultFormats).write(m)
+  }
+
+  override def get_party_set(new_data: Map[String, String]): Seq[String] = {
+    Seq(
+      getEntity(new_data.getOrEmptyStr("plaintiff_info"), "name").mkString(","),
+      getEntity(new_data.getOrEmptyStr("defendant_info"), "name").mkString(",")
+    )
+  }
+
+  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]) => {
+    if (CompanyChanceUtils.default_filter(update_type, biz_date, change_fields, old_data, new_data)) {
+      if (update_type.equals("insert")) true
+      else
+        false
+    } else {
+      false
+    }
+  }
+
+  override def getCompanyIds(new_data: Map[String, String]): Seq[String] = {
+    getEntity(new_data.getOrEmptyStr("plaintiff_info"), "litigant_id")
+  }
+
+  override def flat_map: ChangeExtract => Seq[CompanyChanceRecord] = (change_extract: ChangeExtract) => {
+    getCompanyChanceRecord(change_extract, getCompanyIds(change_extract.new_data))
+  }
+}

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

@@ -29,13 +29,13 @@ object CompanyMonitorMappings {
       , 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 = "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 = "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")
@@ -54,4 +54,10 @@ object CompanyMonitorMappings {
   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 !!!"))
   }
+
+  def main(a: Array[String]): Unit = {
+    args.sortBy(y=>y.type_id.toLong).foreach(x => {
+      println(x.type_id + " ----- " + x.type_name)
+    })
+  }
 }

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

@@ -35,7 +35,7 @@ object CompanyMonitorV2 {
     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 clazz: CompanyMonitorHandleV2 = 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

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

@@ -901,6 +901,7 @@ object BaseUtil {
 
 
   def main(args: Array[String]): Unit = {
+    println(getYesterday())
     println(transNumber("2000.105000万人民币"))
     println(transNumber(null))
     println(transNumber(""))