财产监控

pom.xml

@@ -313,6 +313,11 @@
             <artifactId>spark-graphx_${scala.binary.version}</artifactId>
             <version>2.4.6</version>
         </dependency>
+        <dependency>
+            <groupId>com.belerweb</groupId>
+            <artifactId>pinyin4j</artifactId>
+            <version>2.5.1</version>
+        </dependency>
 
     </dependencies>
 

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/CompanyMonitor.scala

@@ -0,0 +1,255 @@
+package com.winhc.bigdata.spark.jobs.monitor
+
+import java.util.Date
+
+import com.winhc.bigdata.spark.config.EsConfig
+import com.winhc.bigdata.spark.jobs.dynamic.{CompanyDynamicForDayCount, CompanyDynamicHandle, CompanyDynamicHandleUtils}
+import com.winhc.bigdata.spark.jobs.message.IntellectualMessage
+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.time.DateFormatUtils
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.types.StringType
+import org.apache.spark.sql.{Row, SparkSession}
+
+import scala.annotation.meta.getter
+import scala.collection.immutable.ListMap
+import scala.collection.mutable
+
+/**
+ * @Author: π
+ * @Date: 2020/12/8
+ * @Description: 企业财产监控
+ */
+object CompanyMonitor {
+  val env = "dev"
+  val targetTab = "ads_company_monitor"
+
+  case class CompanyMonitorUtil(s: SparkSession,
+                                project: String, //表所在工程名
+                                ds: String //此维度主键
+                               ) extends LoggingUtils with Logging with BaseFunc {
+    @(transient@getter) val spark: SparkSession = s
+
+
+    def init(): Unit = {
+      sql(
+        s"""
+           |CREATE TABLE IF NOT EXISTS ${getEnvProjectName(env, project)}.$targetTab
+           |(
+           |    id              STRING COMMENT '唯一标示',
+           |    cid             STRING COMMENT '公司id',
+           |    cname           STRING COMMENT '公司name',
+           |    info_type       STRING COMMENT '变更分类，大类',
+           |    flow_type       STRING COMMENT '财产流向',
+           |    rta_desc        STRING COMMENT '变更信息描述，变更标题',
+           |    change_time     STRING COMMENT '变更时间',
+           |    biz_id          STRING COMMENT '业务id，数据行id',
+           |    info_risk_level STRING COMMENT '变更风险等级',
+           |    create_time     STRING COMMENT '创建时间'
+           |)
+           |COMMENT '企业财务监控输出表'
+           |PARTITIONED BY
+           |(
+           |    ds              STRING COMMENT '分区',
+           |    tn              STRING COMMENT '表名'
+           |)
+           |LIFECYCLE 30
+           |""".stripMargin)
+    }
+
+    val tabMapping =
+      Map("wenshu_detail_combine_v2" -> "wenshu_detail_combine" //胜诉案件
+        , "company_land_mortgage_v2" -> "company_land_mortgage" //土地抵押权人
+      )
+
+    //转换字段
+    def trans(s: String): String = {
+      var res = s
+      if (tabMapping.contains(s)) {
+        res = tabMapping(s)
+      }
+      res
+    }
+
+    //表名（不加前后辍）
+    def calc(tableName: String
+             , bName: Int = 0 //是否补充cname字段
+            ): Unit = {
+      val handle = getClazz[CompanyMonitorHandle](s"com.winhc.bigdata.spark.jobs.monitor.tables.$tableName")
+
+      val types = handle.org_type()
+      val conditional = handle.get_conditional_filter()
+      val tn = trans(tableName)
+
+      val rdd = sql(
+        bName match {
+          //默认:无需补全cname字段
+          case 0 =>
+            s"""
+               |SELECT  *,null AS cname
+               |FROM    ${project}.ads_change_extract
+               |WHERE   ds = '$ds'
+               |AND     tn = '$tn'
+               |AND     TYPE in (${types.map("'" + _ + "'").mkString(",")})
+               |$conditional
+               |""".stripMargin
+          //需根据cid补全cname字段数据
+          case 1 =>
+            s"""
+               |SELECT A.*,B.cname AS cname
+               |FROM(
+               |  SELECT  *
+               |  FROM    ${project}.ads_change_extract
+               |  WHERE   ds = '$ds'
+               |  AND     tn = '$tn'
+               |  AND     TYPE in (${types.map("'" + _ + "'").mkString(",")})
+               |  $conditional
+               |) AS A
+               |LEFT JOIN (
+               |    SELECT cid,cname FROM  $project.base_company_mapping
+               |    WHERE ds = '${getLastPartitionsOrElse(project + ".base_company_mapping", "0")}'
+               |) AS B
+               |ON A.cid = B.cid
+               |""".stripMargin
+
+        })
+        .rdd.flatMap(r => {
+        val rowkey = r.getAs[String]("rowkey")
+        val cid = r.getAs[String]("cid")
+        val new_data = r.getAs[Map[String, String]]("data")
+        val old_data = r.getAs[Map[String, String]]("old_data")
+        val biz_date = r.getAs[String]("biz_date")
+        val fields = r.getAs[String]("fields")
+        val cname = r.getAs[String]("cname")
+        if (biz_date == null)
+          None
+        val result = handle.handle(rowkey, biz_date, cid, if (fields == null) null else fields.split(","), old_data, new_data, cname)
+        if (result == null) {
+          None
+        }
+        else {
+          result.map(res => Row(CompanyDynamicHandleUtils.getDynamicId(res._1, res._5, res._7, res._6),
+            res._1, res._2, res._3, res._4,
+            res._5.replaceAll("null", ""), res._6, res._7, res._8, res._9,
+            DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss")))
+        }
+      })
+
+      val schema = getSchema(ListMap(
+        "id" -> StringType
+        , "cid" -> StringType
+        , "cname" -> StringType
+        , "table_type" -> StringType
+        , "flow_type" -> StringType
+        , "rta_desc" -> StringType
+        , "change_time" -> StringType
+        , "biz_id" -> StringType
+        , "info_risk_level" -> StringType
+        , "type" -> StringType
+        , "create_time" -> StringType
+      ))
+      spark.createDataFrame(rdd, schema)
+        .createOrReplaceTempView(s"company_monitor_tmp_$tableName")
+
+      unescapeHtml4()
+      sql(
+        s"""
+           |INSERT ${if (isWindows) "INTO" else "OVERWRITE"} TABLE ${getEnvProjectName(env, project)}.$targetTab PARTITION(ds='$ds',tn='$tableName')
+           |SELECT  id
+           |        , cid
+           |        , cname
+           |        , table_type
+           |        , flow_type
+           |        , unescapeHtml4(rta_desc) rta_desc
+           |        , change_time
+           |        , biz_id
+           |        , info_risk_level
+           |        , type
+           |        , create_time
+           |FROM
+           |    company_monitor_tmp_$tableName
+           |WHERE id IS NOT NULL
+           |AND   to_timestamp(change_time) <= now()
+           |""".stripMargin)
+    }
+  }
+
+  private val startArgs = Seq(
+    Args(tableName = "wenshu_detail_combine_v2", bName = 1)
+    , Args(tableName = "company_dishonest_info", bName = 1)
+    , Args(tableName = "company_zxr_list", bName = 1)
+    , Args(tableName = "company_zxr_restrict", bName = 1)
+    , Args(tableName = "company_court_open_announcement_list", bName = 1)
+    , Args(tableName = "wenshu_detail_combine", bName = 1)
+    , Args(tableName = "company_equity_info_list", bName = 1)
+    , Args(tableName = "company_land_mortgage", bName = 1)
+    , Args(tableName = "company_land_announcement", bName = 1)
+    , Args(tableName = "company_finance", bName = 1)
+    , Args(tableName = "bankruptcy_open_case", bName = 1)
+    , Args(tableName = "company_bid_list", bName = 1)
+    , Args(tableName = "company_mortgage_info", bName = 1)
+    , Args(tableName = "company_tm", bName = 1)
+    , Args(tableName = "company_patent_list", bName = 1)
+    , Args(tableName = "company_copyright_reg_list", bName = 1)
+    , Args(tableName = "company_copyright_works_list", bName = 1)
+    , Args(tableName = "company_holder_v2", bName = 1)
+    //, Args(tableName = "company", bName = 1)
+    , Args(tableName = "company_land_mortgage_v2", bName = 1)
+    , Args(tableName = "auction_tracking_list", bName = 1)
+
+  )
+
+  private case class Args(project: String = "winhc_eci_dev"
+                          , tableName: String
+                          , bName: Int = 1
+                          , aggs: Int = 0)
+
+  def main(args: Array[String]): Unit = {
+
+
+    if (args.length != 3) {
+      println(
+        s"""
+           |Please enter the legal parameters !
+           |<project> <tableNames> <ds>
+           |""".stripMargin)
+      sys.exit(-99)
+    }
+
+    val Array(project, tableNames, ds) = args
+
+    println(
+      s"""
+         |project: $project
+         |tableNames: $tableNames
+         |ds: $ds
+         |""".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("CompanyMonitor", config)
+    val cd = CompanyMonitorUtil(spark, project, ds)
+    //cd.init()
+
+    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, () => {
+      e.aggs match {
+        case _ => cd.calc(e.tableName, e.bName) //通用处理
+      }
+      true
+    }))
+
+    AsyncExtract.startAndWait(spark, a)
+    spark.stop()
+  }
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/CompanyMonitorHandle.scala

@@ -0,0 +1,238 @@
+package com.winhc.bigdata.spark.jobs.monitor
+
+import org.apache.spark.internal.Logging
+
+/**
+ * @Description: 财产监控
+ * @author π
+ * @date 2020/12/8 14:48
+ */
+trait CompanyMonitorHandle extends Logging {
+
+  //风险等级映射 1->低级 2->中级 3->高级
+  private val info_risk_level_map = Map(
+    "wenshu_detail_combine_v2" -> "2" //新增胜诉案件
+    , "company_dishonest_info" -> "3" //失信移除
+    , "company_zxr_list" -> "3" //被执行人移除
+    , "company_zxr_restrict" -> "3" //限制高消费移除
+    , "company_court_open_announcement_list" -> "2" //有恢复执行案件即将开庭
+    , "wenshu_detail_combine" -> "1" //有恢复执行案件-文书
+    , "company_equity_info_list" -> "3" //股权出质(出质人)
+    , "company_land_mortgage" -> "3" //土地抵押（抵押人）
+    , "company_land_mortgage_v2" -> "3" //土地抵押（抵押权人）
+    , "company_land_announcement" -> "3" //购地信息
+    , "company_finance" -> "3" //融资信息
+    , "bankruptcy_open_case" -> "2" //破产公告
+    , "company_bid_list" -> "3" //招投标
+    , "company_mortgage_info" -> "3" //动产抵押
+    , "company_tm" -> "1" //商标
+    , "company_patent_list" -> "2" //专利
+    , "company_copyright_reg_list" -> "1" //软件著作权
+    , "company_copyright_works_list" -> "1" //作品著作权
+    , "company_holder_add" -> "2" //股东新增
+    , "company_holder_deleted" -> "1" //股东移除
+    , "company" -> "2" //工商信息
+    , "auction_tracking_list" -> "3" //司法拍卖
+  )
+
+  //财产流向 0 -> 流入 1 -> 流出
+  private val flow_type_map = Map(
+    "wenshu_detail_combine_v2" -> "0" //新增胜诉案件
+    , "company_dishonest_info" -> "1" //失信移除
+    , "company_zxr_list" -> "1" //被执行人移除
+    , "company_zxr_restrict" -> "1" //被执行人移除
+    , "company_court_open_announcement_list" -> "1" //有恢复执行案件即将开庭
+    , "wenshu_detail_combine" -> "1" //有恢复执行案件-文书
+    , "company_equity_info_list" -> "0" //股权出质(出质人)
+    , "company_land_mortgage" -> "0" //土地抵押（抵押人）
+    , "company_land_mortgage_v2" -> "0" //土地抵押（抵押权人）
+    , "company_land_announcement" -> "1" //购地信息
+    , "company_finance" -> "0" //融资信息
+    , "bankruptcy_open_case" -> "1" //破产公告
+    , "company_bid_list" -> "0" //招投标
+    , "company_mortgage_info" -> "0" //动产抵押
+    , "company_tm" -> "0" //商标
+    , "company_patent_list" -> "0" //专利
+    , "company_copyright_reg_list" -> "0" //软件著作权
+    , "company_copyright_works_list" -> "0" //作品著作权
+    , "company_holder_add" -> "0" //股东新增
+    , "company_holder_deleted" -> "1" //股东移除
+    , "company" -> "0" //工商信息
+    , "auction_tracking_list" -> "0" //司法拍卖
+  )
+
+
+  //维度映射
+  private val table_type_map = Map(
+    "bankruptcy_open_case" -> "1" //BANKRUPTCY_OPEN_CASE//HBase
+    , "company" -> "2" //COMPANY//HBase
+    , "company_bid_list" -> "3" //COMPANY_BID_LIST//HBase
+    , "company_copyright_reg_list" -> "4" //COMPANY_COPYRIGHT_REG_LIST//HBase
+    , "company_copyright_works_list" -> "5" //COMPANY_COPYRIGHT_WORKS_LIST//HBase
+    , "company_court_open_announcement_list" -> "6" //COMPANY_COURT_OPEN_ANNOUNCEMENT_LIST//HBase
+    , "company_dishonest_info" -> "7" //COMPANY_DISHONEST_INFO//HBase
+    , "company_equity_info_list" -> "8" //COMPANY_EQUITY_INFO_LIST//HBase
+    , "company_finance" -> "9" //COMPANY_FINANCE//HBase
+    , "company_holder_v2" -> "10" //COMPANY_HOLDER//HBase
+    , "company_land_announcement" -> "11" //COMPANY_LAND_ANNOUNCEMENT//HBase
+    , "company_land_mortgage" -> "12" //COMPANY_LAND_MORTGAGE//HBase
+    , "company_land_mortgage_v2" -> "12" //COMPANY_LAND_MORTGAGE//HBase
+    , "company_mortgage_info" -> "13" //COMPANY_MORTGAGE_INFO//HBase
+    , "company_patent_list" -> "14" //COMPANY_PATENT_LIST//HBase
+    , "company_tm" -> "15" //COMPANY_TM//HBase
+    , "company_zxr_list" -> "16" //COMPANY_ZXR_LIST//HBase
+    , "company_zxr_restrict" -> "17" //COMPANY_ZXR_RESTRICT//HBase
+    , "wenshu_detail_combine" -> "18" //WENSHU_DETAIL//ES
+    , "wenshu_detail_combine_v2" -> "18" //WENSHU_DETAIL//ES
+    , "auction_tracking_list" -> "19" //AUCTION_TRACKING_LIST//HBase
+  )
+
+  //类别
+  private val type_map = Map(
+    "bankruptcy_open_case" -> "1" //新增破产公告
+    , "company" -> "2" //注册资本上升
+    , "company_bid_list" -> "3" //新增招标信息
+    , "company_copyright_reg_list" -> "4" //软件著作权
+    , "company_copyright_works_list" -> "5" //作品著作权
+    , "company_court_open_announcement_list" -> "6" //有恢复执行案件即将开庭
+    , "company_dishonest_info" -> "7" //失信信息移除
+    , "company_equity_info_list" -> "8" //出质了持有股权
+    , "company_finance" -> "9" //有新的融资
+    , "company_holder_add" -> "10" //新增了股东
+    , "company_holder_remove" -> "11" //股东退出
+    , "company_land_announcement" -> "12" //购买了新的土地
+    , "company_land_mortgage" -> "13" //抵押了公司土地
+    , "company_mortgage_info" -> "14" //资产被抵押
+    , "company_patent_list" -> "15" //新增专利信息
+    , "company_tm" -> "16" //新增商标信息
+    , "company_zxr_list" -> "17" //被执行人信息移除
+    , "company_zxr_restrict" -> "18" //限制高消费移除
+    , "wenshu_detail_combine" -> "19" //有恢复执行案件
+    , "wenshu_detail_combine_v2" -> "20" //新增胜诉案件
+    , "company_land_mortgage_v2" -> "21" //接受了他人土地抵押
+    , "auction_tracking_list" -> "22" //有资产即将被司法拍卖
+  )
+
+  /**
+   *
+   * @param rowkey
+   * @param cid
+   * @param change_fields
+   * @param old_map
+   * @param new_map
+   * @return
+   * cid
+   * cname
+   * info_type
+   * flow_type
+   * rta_desc
+   * change_time
+   * biz_id
+   * info_risk_level
+   * type
+   */
+  def handle(rowkey: String, bizDate: String, cid: String, change_fields: Seq[String], old_map: Map[String, String], new_map: Map[String, String], cname: String = null): Seq[(String, String, String, String, String, String, String, String, String)] = {
+    val rta_desc = get_rta_desc(old_map, new_map)
+    if (rta_desc == null) {
+      return Seq.empty
+    }
+    try {
+      Seq((cid
+        , cname
+        , get_table_type()
+        , get_flow_type()
+        , rta_desc
+        , get_change_time(bizDate, new_map)
+        , get_biz_id(rowkey, new_map)
+        , get_info_risk_level(old_map, new_map)
+        , get_type()
+      ))
+    } catch {
+      case e: Exception => {
+        logError(e.getMessage, e)
+      }
+        Seq.empty
+    }
+
+  }
+
+  /**
+   * 来源表的变更类型，insert or update
+   *
+   * @return
+   */
+  def org_type(): Seq[String] = Seq("insert")
+
+
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String
+
+  /**
+   * 类别映射
+   *
+   * @return
+   */
+  protected def get_type(): String = type_map(getClass.getSimpleName)
+
+  /**
+   * 维度映射
+   *
+   * @return
+   */
+  protected def get_table_type(): String = table_type_map(getClass.getSimpleName)
+
+  /**
+   * 风险等级
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  protected def get_info_risk_level(old_map: Map[String, String], new_map: Map[String, String]): String = info_risk_level_map(getClass.getSimpleName)
+
+  /**
+   * 财产流向
+   * 0 流入 1 流出
+   *
+   * @return
+   */
+  protected def get_flow_type(): String = flow_type_map(getClass.getSimpleName)
+
+
+  /**
+   * 变更时间
+   *
+   * @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 = ""
+
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/auction_tracking_list.scala

@@ -0,0 +1,23 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/12/14
+ * @Description: 司法拍卖
+ */
+case class auction_tracking_list() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""标题：${new_map("title")}
+       |起拍价：${new_map("applicant")}
+       |拍卖时间：${new_map("public_date")}""".stripMargin
+
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/bankruptcy_open_case.scala

@@ -0,0 +1,23 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 破产公告
+ */
+case class bankruptcy_open_case() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""案号：${new_map("case_no")}
+       |申请人：${new_map("applicant")}
+       |公开日期：${new_map("public_date")}""".stripMargin
+
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company.scala

@@ -0,0 +1,76 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+import com.winhc.bigdata.spark.utils.RegCapitalAmount
+import org.apache.commons.lang3.StringUtils
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 工商数据
+ */
+case class company() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""注册资本：${new_map("reg_capital")}
+       |变更时间：${new_map("update_time")}""".stripMargin
+
+  //注册资本上升
+  override def handle(rowkey: String, bizDate: String, cid: String, change_fields: Seq[String], old_map: Map[String, String], new_map: Map[String, String], cname: String = null): Seq[(String, String, String, String, String, String, String, String, String)] = {
+    if (!compareAmount(old_map, new_map)) {
+      return Seq.empty
+    }
+    try {
+      Seq((cid
+        , cname
+        , get_table_type()
+        , get_flow_type()
+        , get_rta_desc(old_map, new_map)
+        , get_change_time(bizDate, new_map)
+        , get_biz_id(rowkey, new_map)
+        , get_info_risk_level(old_map, new_map)
+        , get_type()
+      ))
+    } catch {
+      case e: Exception => {
+        logError(e.getMessage, e)
+      }
+        Seq.empty
+    }
+  }
+
+  /**
+   * 来源表的变更类型，insert or update
+   *
+   * @return
+   */
+  override def org_type(): Seq[String] = {
+    Seq("update")
+  }
+
+  /**
+   * 注册资本比较
+   */
+  def compareAmount(old_map: Map[String, String], new_map: Map[String, String]): Boolean = {
+    if (old_map == null || new_map == null) {
+      return false
+    }
+    val old = old_map("reg_capital")
+    val now = new_map("reg_capital")
+    if (StringUtils.isBlank(old) || StringUtils.isBlank(now)) {
+      return false
+    }
+    val old_num = RegCapitalAmount.getAmount(old)
+    val now_num = RegCapitalAmount.getAmount(now)
+    if (StringUtils.isBlank(old_num) || StringUtils.isBlank(now_num)) {
+      return false
+    }
+    old_num.toDouble < now_num.toDouble
+  }
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_bid_list.scala

@@ -0,0 +1,21 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 招投标
+ */
+case class company_bid_list() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""标题：${new_map("title")}""".stripMargin
+
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_copyright_reg_list.scala

@@ -0,0 +1,24 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 软件著作权
+ */
+case class company_copyright_reg_list() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""登记号：${new_map("reg_num")}
+       |软件名称：${new_map("full_name")}
+       |软件简称：${new_map("simple_name")}""".stripMargin
+
+
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_copyright_works_list.scala

@@ -0,0 +1,24 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 著作权
+ */
+case class company_copyright_works_list() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""作品名称：${new_map("name")}
+       |作品类别：${new_map("type")}
+       |登记日期：${new_map("reg_time")}""".stripMargin
+
+
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_court_open_announcement_list.scala

@@ -0,0 +1,29 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 有恢复执行案件即将开庭
+ */
+case class company_court_open_announcement_list() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String = {
+    s"""上诉人/原告：${new_map("plaintiff")}
+       |被上诉人/被告：${new_map("defendant")}
+       |案号：${new_map("case_no")}
+       |案由：${new_map("case_reason")}""".stripMargin
+  }
+
+  override def get_conditional_filter(): String = {
+    "AND  data['case_no'] like concat('%','恢','%')"
+  }
+
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_dishonest_info.scala

@@ -0,0 +1,26 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 企业失信
+ */
+case class company_dishonest_info() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""案号：${new_map("case_no")}
+       |履行情况：${new_map("performance")}
+       |发布日期：${new_map("pub_date")}
+       |失信行为：${new_map("action_content")}""".stripMargin
+
+  override def org_type() = {
+    Seq("update")
+  }
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_equity_info_list.scala

@@ -0,0 +1,27 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 股权出质人
+ */
+case class company_equity_info_list() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""股权出质登记日期：${new_map("reg_date")}
+       |质权人：${new_map("pledgee")}
+       |出质股权标的企业：${new_map("target")}
+       |出质股权数额：${new_map("equity_amount")}""".stripMargin
+
+  override def get_conditional_filter(): String = {
+    "AND data['type'] = '1'"
+  }
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_finance.scala

@@ -0,0 +1,23 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 融资信息
+ */
+case class company_finance() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""融资金额：${new_map("money")}
+       |轮次：${new_map("round")}
+       |投资人：${new_map("inverstors")}""".stripMargin
+
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_holder_v2.scala

@@ -0,0 +1,67 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 股东
+ */
+case class company_holder_v2() extends CompanyMonitorHandle {
+
+  /**
+   * 来源表的变更类型，insert or update
+   *
+   * @return
+   */
+  override def org_type(): Seq[String] = Seq("insert", "update")
+
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   * cid
+   * cname
+   * table_type
+   * flow_type
+   * rta_desc
+   * change_time
+   * biz_id
+   * info_risk_level
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""|认缴金额：${new_map("amount")}
+        |认缴出资时间：${new_map("update_time")}""".stripMargin
+
+  override def handle(rowkey: String, bizDate: String, cid: String, change_fields: Seq[String], old_map: Map[String, String], new_map: Map[String, String], cname: String = null): Seq[(String, String, String, String, String, String, String, String, String)] = {
+    val rta_desc = get_rta_desc(old_map, new_map)
+    if (rta_desc == null) {
+      return Seq.empty
+    }
+
+    try {
+      var flag = false
+      if (change_fields == null) {
+        flag = true
+      }
+      Seq((cid
+        , cname
+        //, if (flag) "company_holder_add" else "company_holder_deleted"
+        , if (flag) "10" else "10"
+        , if (flag) "0" else "1"
+        , rta_desc
+        , get_change_time(bizDate, new_map)
+        , get_biz_id(rowkey, new_map)
+        , if (flag) "2" else "1"
+        , if (flag) "10" else "11"
+      ))
+    } catch {
+      case e: Exception => {
+        logError(e.getMessage, e)
+      }
+        Seq.empty
+    }
+  }
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_land_announcement.scala

@@ -0,0 +1,24 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 购地信息
+ */
+case class company_land_announcement() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""项目名称：${new_map("project_name")}
+       |面积(公顷)：${new_map("area")}
+       |成交价格(万元)：${new_map("tran_price")}
+       |合同签订日期：${new_map("contract_date")}""".stripMargin
+
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_land_mortgage.scala

@@ -0,0 +1,28 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 土地抵押-抵押人
+ */
+case class company_land_mortgage() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""所在行政区：${new_map("land_aministrative_area")}
+       |土地面积(公顷)：${new_map("land_area")}
+       |抵押面积(公顷)：${new_map("area")}
+       |评估金额(万元)：${new_map("evaluate_amount")}
+       |抵押金额(万元)：${new_map("mortgage_amount")}""".stripMargin
+
+
+  override def get_conditional_filter(): String = {
+    "AND  data['type'] in ('bothone','mortgagor') "
+  }
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_land_mortgage_v2.scala

@@ -0,0 +1,28 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 土地抵押-抵押权人
+ */
+case class company_land_mortgage_v2() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""所在行政区：${new_map("land_aministrative_area")}
+       |土地面积(公顷)：${new_map("land_area")}
+       |抵押面积(公顷)：${new_map("area")}
+       |评估金额(万元)：${new_map("evaluate_amount")}
+       |抵押金额(万元)：${new_map("mortgage_amount")}""".stripMargin
+
+
+  override def get_conditional_filter(): String = {
+    "AND  data['type'] in ('bothtwo','mortgagee') "
+  }
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_mortgage_info.scala

@@ -0,0 +1,24 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 动产抵押
+ */
+case class company_mortgage_info() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""登记编号：${new_map("reg_num")}
+       |被担保债权数额：${new_map("amount")}
+       |登记机关：${new_map("reg_department")}
+       |状态：${new_map("status")}""".stripMargin
+
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_patent_list.scala

@@ -0,0 +1,25 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 专利
+ */
+case class company_patent_list() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""申请号：${new_map("app_number")}
+       |公开公告号：${new_map("pub_number")}
+       |专利名称：${new_map("title")}
+       |公开公告日：${new_map("pub_date")}""".stripMargin
+
+
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_tm.scala

@@ -0,0 +1,25 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 商标
+ */
+case class company_tm() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""商标名：${new_map("tm_name")}
+       |申请日期：${new_map("app_date")}
+       |注册号：${new_map("reg_no")}
+       |申请人：${new_map("applicant_cn")}""".stripMargin
+
+
+
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_zxr_list.scala

@@ -0,0 +1,27 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 企业被执
+ */
+case class company_zxr_list() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""更新日期：${new_map("update_time")}
+       |执行标的：${new_map("exec_money")}
+       |案号：${new_map("case_no")}
+       |立案日期：${new_map("case_create_time")}""".stripMargin
+
+
+  override def org_type() = {
+    Seq("update")
+  }
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/company_zxr_restrict.scala

@@ -0,0 +1,27 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 限制消费令
+ */
+case class company_zxr_restrict() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String =
+    s"""案号：${new_map("case_no")}
+       |企业名称：${new_map("company_name")}
+       |限制消费人名称：${new_map("name")}
+       |执行法院名称：${new_map("court_name")}
+       |立案日期：${new_map("case_create_time")}""".stripMargin
+
+  override def org_type() = {
+    Seq("update")
+  }
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/wenshu_detail_combine.scala

@@ -0,0 +1,37 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 有恢复执行案件-文书
+ */
+case class wenshu_detail_combine() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String = {
+    s"""案件名称：${new_map("title")}
+       |案号：${new_map("case_no")}
+       |法院：${new_map("court_name")}
+       |判决日期：${new_map("judge_date")}""".stripMargin
+  }
+
+  override def get_conditional_filter(): String = {
+    "AND  data['case_no'] like concat('%','恢','%')"
+  }
+
+  /**
+   * 业务id
+   *
+   * @param rowkey
+   * @return
+   */
+  override def get_biz_id(rowkey: String, new_map: Map[String, String]): String = {
+    new_map("case_id")
+  }
+}

src/main/scala/com/winhc/bigdata/spark/jobs/monitor/tables/wenshu_detail_combine_v2.scala

@@ -0,0 +1,39 @@
+package com.winhc.bigdata.spark.jobs.monitor.tables
+
+import com.winhc.bigdata.spark.jobs.monitor.CompanyMonitorHandle
+
+/**
+ * @Date: 2020/8/12 18:50
+ * @Description: 新增胜诉案件
+ */
+case class wenshu_detail_combine_v2() extends CompanyMonitorHandle {
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String = {
+    s"""案件名称：${new_map("title")}
+       |案号：${new_map("case_no")}
+       |法院：${new_map("court_name")}
+       |判决日期：${new_map("judge_date")}""".stripMargin
+  }
+
+  override def get_conditional_filter(): String = {
+    "AND data['is_success'] = '胜'  AND data['case_stage']= '一审'  AND  data['case_type'] = '民事案件'"
+  }
+
+
+  /**
+   * 业务id
+   *
+   * @param rowkey
+   * @return
+   */
+  override def get_biz_id(rowkey: String, new_map: Map[String, String]): String = {
+    new_map("case_id")
+  }
+
+}