被执行人动态

src/main/scala/com/winhc/bigdata/spark/jobs/CompanyWenshuYgYishen.scala

@@ -112,11 +112,14 @@ case class CompanyWenshuYishen(s: SparkSession, project: String, //表所在工
        |    FROM    ods_wenshu_detail
        |      LATERAL VIEW explode(split(yg_name, '\n')) tmpTable AS cname
        |    WHERE   ds = $adsListDs
-       |    AND     judge_date >= '${atMonthsBefore(1)}'
+       |    AND     judge_date >= '${atMonthsBefore(3)}'
        |    AND     is_success = '胜'
        |    AND     case_stage = '一审'
        |    AND     case_type = '民事案件'
        |    AND     length(cname) > 4
+       |    AND     cname not like '%银行%'
+       |    AND     cname not like '%保险%'
+       |    AND     case_amt > 100
         |""".stripMargin
     )
 
@@ -178,7 +181,7 @@ case class CompanyWenshuYishen(s: SparkSession, project: String, //表所在工
         val m1: Map[String, String] = queryCompany(restClient, cname)
         //动态变更内容
         val m2: Map[String, String] = YgYishenWenshuJson(r, Seq(
-          "case_reason", "case_no", "court_name", "judge_date", "link_url", "case_stage", "judge_result"))
+          "case_amt", "case_reason", "case_no", "court_name", "judge_date", "link_url", "case_stage", "judge_result"))
         val dynamic_content = Json(DefaultFormats).write(m1 ++: m2)
         val biz_date = r.getAs[Timestamp]("biz_date") //动态变更时间
         val create_time = atMonthsBefore(0, "yyyy-MM-dd HH:mm:ss") //创建时间
@@ -297,7 +300,7 @@ object CompanyWenshuYgYishen {
          |""".stripMargin)
 
     val config = mutable.Map(
-      "spark.hadoop.odps.project.name" -> "winhc_eci_dev",
+      "spark.hadoop.odps.project.name" -> "winhc_eci",
       "spark.hadoop.odps.spark.local.partition.amt" -> "100"
     )
     val spark: SparkSession = SparkUtils.InitEnv(this.getClass.getSimpleName, config)

src/main/scala/com/winhc/bigdata/spark/jobs/CompanyWenshuYgZhongben.scala

@@ -104,6 +104,7 @@ case class CompanyWenshu(s: SparkSession, project: String, //表所在工程名
        |            ,ROW_NUMBER() OVER(PARTITION BY cname,bg_name,case_no,court_name,judge_date ORDER BY judge_date) RN
        |    ,court_name
        |    ,judge_date AS biz_date
+       |    ,judge_result
        |    ,url AS link_url
        |    ,is_success
        |    FROM    $ods_wenshu_detail
@@ -166,7 +167,7 @@ case class CompanyWenshu(s: SparkSession, project: String, //表所在工程名
         val business_type_name = "0" //动态类型name
         val m1: Map[String, String] = queryCompany(restClient, cname)
         //动态变更内容
-        val m2: Map[String, String] = WenshuJson(r, Seq("case_reason", "case_no", "court_name", "judge_date", "link_url", "case_stage", "is_success"))
+        val m2: Map[String, String] = WenshuJson(r, Seq("case_amt", "case_reason", "case_no", "court_name", "judge_date", "link_url", "case_stage", "judge_result"))
         val dynamic_content = Json(DefaultFormats).write(m1 ++: m2)
         val biz_date = r.getAs[Timestamp]("biz_date") //动态变更时间
         val create_time = atMonthsBefore(0, "yyyy-MM-dd HH:mm:ss") //创建时间
@@ -285,7 +286,7 @@ object CompanyWenshuYgZhongben {
          |""".stripMargin)
 
     val config = mutable.Map(
-      "spark.hadoop.odps.project.name" -> "winhc_eci_dev",
+      "spark.hadoop.odps.project.name" -> "winhc_eci",
       "spark.hadoop.odps.spark.local.partition.amt" -> "100"
     )
     val spark: SparkSession = SparkUtils.InitEnv(this.getClass.getSimpleName, config)

src/main/scala/com/winhc/bigdata/spark/jobs/chance/ChangeExtract.scala

@@ -25,7 +25,7 @@ object ChangeExtract {
   }
 
   def getHandleClazz(tableName: String, equCols: Seq[String]): {def handle(rowkey: String, oldMap: Map[String, String], newMap: Map[String, String]): (String, String, String, Map[String, String], String, String, String, String, Map[String, String])} = {
-    val clazz = s"com.winhc.bigdata.spark.jobs.chance.$tableName"
+    val clazz = s"com.winhc.bigdata.spark.jobs.chance.table.$tableName"
     val foo = Class.forName(clazz)
       .getConstructors.head.newInstance(equCols)
       .asInstanceOf[ {
@@ -291,6 +291,8 @@ object ChangeExtract {
     , Args(tableName = "company_land_publicity", primaryFields = "title,location,use_for")
     , Args(tableName = "company_land_announcement", primaryFields = "e_number,project_name")
     , Args(tableName = "company_bid_list", primaryFields = "title")
+    , Args(tableName = "company_zxr_list", primaryFields = "case_no,exec_money")
+
     , Args(tableName = "company_land_transfer", primaryFields = "num,location")
     , Args(tableName = "company_employment", primaryFields = "source")
     , Args(tableName = "company_env_punishment", primaryFields = "punish_number")

src/main/scala/com/winhc/bigdata/spark/jobs/chance/CompanyChangeHandle.scala

@@ -1,6 +1,7 @@
 package com.winhc.bigdata.spark.jobs.chance
 
 import com.winhc.bigdata.spark.utils.BaseUtil.cleanup
+import com.winhc.bigdata.spark.utils.ChangeExtractUtils
 import org.apache.commons.lang3.StringUtils
 import org.apache.spark.internal.Logging
 
@@ -70,3 +71,185 @@ trait CompanyChangeHandle extends Serializable with Logging {
     }
   }
 }
+
+//土地公示
+case class company_land_publicity(equCols: Seq[String]) extends CompanyChangeHandle with Serializable {
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = {
+    val str = ChangeExtractUtils.getTags(newMap, "地块公示", Array("project_name", "location", "application_name", "area", "final_price", "land_num"))
+    str
+  }
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("publication_start_date")
+
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("project_name"), s"${newMap("project_name")}地块公示发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("project_name"), s"新增${newMap("project_name")}地块公示")
+}
+
+
+case class company(equCols: Seq[String]) extends CompanyChangeHandle with Serializable {
+  override def getCid(rowkey: String, newMap: Map[String, String]): String = rowkey
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = "1"
+
+  override def getBizTime(newMap: Map[String, String]): String = "业务时间"
+
+  override def getUpdateTitle(newMap: Map[String, String]): String = "更新一家公司"
+
+  override def getInsertTitle(newMap: Map[String, String]): String = "新增一家公司"
+}
+
+//商标
+case class company_tm(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("tm_name"), s"${newMap("tm_name")}商标发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("tm_name"), s"新增${newMap("tm_name")}商标")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("商标", newMap("tm_name"), newMap("app_date"), newMap("reg_no"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("app_date")
+}
+
+//专利
+case class company_patent_list(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"${newMap("title")}专利发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"新增${newMap("title")}专利")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("专利", newMap("title"), newMap("app_date"), newMap("app_number"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("app_date")
+}
+
+//资质证书
+case class company_certificate(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("cert_no"), s"${newMap("cert_no")}资质证书发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("cert_no"), s"新增${newMap("cert_no")}资质证书")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("资质证书", newMap("cert_no"), newMap("start_date"), newMap("type"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("start_date")
+}
+
+//作品著作权
+case class company_copyright_works_list(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"${newMap("name")}作品著作权发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"新增${newMap("name")}作品著作权")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("作品著作权", newMap("name"), newMap("reg_time"), newMap("reg_num"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("reg_time")
+}
+
+//软件著作权
+case class company_copyright_reg_list(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("full_name"), s"${newMap("full_name")}软件著作权发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("full_name"), s"新增${newMap("full_name")}软件著作权")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("软件著作权", newMap("full_name"), newMap("reg_time"), newMap("reg_num"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("reg_time")
+}
+
+//网站
+case class company_icp(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("domain"), s"${newMap("domain")}网站备案发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("domain"), s"新增${newMap("domain")}网站备案")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("网站备案", newMap("domain"), newMap("examine_date"), newMap("liscense"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("examine_date")
+}
+
+
+//购地信息
+case class company_land_announcement(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("project_name"), s"${newMap("project_name")}购地信息发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("project_name"), s"新增${newMap("project_name")}购地信息")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "购地信息", Array("project_name", "project_loc", "area", "tran_price", "e_number"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("contract_date")
+}
+
+//招聘
+case class company_employment(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"${newMap("title")}招聘信息发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"新增${newMap("title")}招聘信息")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "招聘", Array("title", "city->employment_city", "employ_num", "start_date"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("start_date")
+}
+
+//招投标
+case class company_bid_list(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"${newMap("title")}招投标信息发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("title"), s"新增${newMap("title")}招投标信息")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "招投标", Array("publish_time", "title", "purchaser", "province", "abs"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("publish_time")
+}
+
+//土地转让
+case class company_land_transfer(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("location"), s"${newMap("title")}土地转让信息发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("location"), s"新增${newMap("location")}土地转让信息")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "土地转让", Array("merchandise_time", "user_pre", "user_now", "location", "area", "merchandise_price", "aministrative_area"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("merchandise_time")
+}
+
+//环保处罚
+case class company_env_punishment(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("punish_number"), s"${newMap("title")}环保处罚信息发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("punish_number"), s"新增${newMap("punish_number")}环保处罚信息")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "环保处罚", Array("name", "department", "punish_number", "punish_basis", "law_break", "reason", "content", "deleted"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("publish_time")
+}
+
+case class company_abnormal_info(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = "经营异常发生变更"
+
+  override def getInsertTitle(newMap: Map[String, String]): String = "新增一条经营异常"
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap, "经营异常", Array("put_department", "remove_department", "put_reason", "put_date", "remove_date", "remove_reason"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("put_date")
+}
+
+//欠税
+case class company_own_tax(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"${newMap("name")}欠税公告发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"新增${newMap("name")}欠税公告")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.get_ip_tags("欠税公告", newMap("name"), newMap("publish_date"), newMap("tax_num"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("publish_date")
+}
+
+//被执行人
+case class company_zxr_list(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"${newMap("name")}被执行人发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("name"), s"新增${newMap("name")}被执行人")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap,"成为被执行人", Array("case_create_time", "case_no", "exec_money"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("case_create_time")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/chance/table/company_zxr_list.scala

@@ -0,0 +1,15 @@
+package com.winhc.bigdata.spark.jobs.chance.table
+
+import com.winhc.bigdata.spark.jobs.chance.CompanyChangeHandle
+import com.winhc.bigdata.spark.utils.ChangeExtractUtils
+
+//被执行人
+case class company_zxr_list(equCols: Seq[String]) extends CompanyChangeHandle {
+  override def getUpdateTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("cname"), s"${newMap("cname")}被执行人发生变更")
+
+  override def getInsertTitle(newMap: Map[String, String]): String = getValueOrNull(newMap("cname"), s"新增${newMap("cname")}被执行人")
+
+  override def getLabel(oldMap: Map[String, String], newMap: Map[String, String]): String = ChangeExtractUtils.getTags(newMap,"成为被执行人", Array("case_create_time", "case_no", "exec_money"))
+
+  override def getBizTime(newMap: Map[String, String]): String = newMap("case_create_time")
+}

src/main/scala/com/winhc/bigdata/spark/jobs/dynamic/CompanyDynamic.scala

@@ -181,6 +181,7 @@ object CompanyDynamic {
     , Args(tableName = "company_equity_info")
     , Args(tableName = "company_staff", bName = 0)
     , Args(tableName = "company", bName = 0)
+    , Args(tableName = "company_zxr_list", bName = 0)
     , Args(tableName = "bankruptcy_open_case", bName = 1)
   )
 

src/main/scala/com/winhc/bigdata/spark/jobs/dynamic/CompanyDynamicHandle.scala

@@ -16,7 +16,7 @@ trait CompanyDynamicHandle {
     , "company_land_mortgage" -> "land_mortgage" //土地抵押
     , "company_bid_list" -> "tender_es" //中标信息ES
     , "" -> "enterprise_shixin" //失信
-    , "" -> "enterprise_zhixing" //被执
+    , "company_zxr_list" -> "enterprise_zhixing" //被执
     , "" -> "shareholder_shixin" //股东失信
     , "" -> "shareholder_zhixing" //股东被执
     , "" -> "tender_qichacha" //中标信息企查查
@@ -90,7 +90,7 @@ trait CompanyDynamicHandle {
     , "" -> "34" // 行政许可
     , "" -> "35" // 双随机抽查
     , "" -> "36" // 限制高消费
-    , "" -> "37" // 被执行人
+    , "company_zxr_list" -> "37" // 被执行人
     , "" -> "38" // 送达报告
     , "bankruptcy_open_case" -> "39" // 破产重整
   )

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

@@ -0,0 +1,61 @@
+package com.winhc.bigdata.spark.jobs.dynamic.tables
+import com.winhc.bigdata.spark.implicits.MapHelper._
+
+import com.winhc.bigdata.spark.jobs.dynamic.CompanyDynamicHandle
+
+case class company_zxr_list() extends CompanyDynamicHandle{
+  /**
+   * 信息描述
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_rta_desc(old_map: Map[String, String], new_map: Map[String, String]): String = new_map("case_no")
+
+  /**
+   * 变更内容
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_change_content(old_map: Map[String, String], new_map: Map[String, String], cname: String = null): String = new_map.toJson(Seq("case_create_time->立案日期","case_no->案号","exec_money->执行标的"))
+
+  /**
+   * 风险等级
+   *
+   * @param old_map
+   * @param new_map
+   * @return
+   */
+  override protected def get_info_risk_level(old_map: Map[String, String], new_map: Map[String, String]): String = "高风险"
+
+  /**
+   *
+   * @param rowkey
+   * @param cid
+   * @param change_fields
+   * @param old_map
+   * @param new_map
+   * @return cid
+   *         cname
+   *         info_type
+   *         rta_desc
+   *         change_content
+   *         change_time
+   *         biz_id
+   *         sub_info_type
+   *         info_risk_level
+   *         winhc_suggest
+   */
+  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, suggestion: String): Seq[(String, String, String, String, String, String, String, String, String, String)] = {
+    if ("1".equals(new_map("deleted"))) {
+      return Seq.empty
+    }
+
+    super.handle(rowkey, bizDate, cid, change_fields, old_map, new_map, cname, "该企业是被执行人")
+
+
+  }
+}