main.go 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603
  1. package main
  2. import (
  3. "github.com/robfig/cron/v3"
  4. "go.uber.org/zap"
  5. "gorm.io/gorm"
  6. "jygit.jydev.jianyu360.cn/data_processing/common_utils"
  7. "jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
  8. "jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
  9. "strings"
  10. "time"
  11. )
  12. var (
  13. Mgo *mongodb.MongodbSim //87 环境,采集 存储的MongoDB
  14. MysqlDB *gorm.DB
  15. )
  16. func main() {
  17. Init()
  18. local, _ := time.LoadLocation("Asia/Shanghai")
  19. c := cron.New(cron.WithLocation(local), cron.WithSeconds())
  20. eid, err := c.AddFunc(GF.Cron.Spec, importData) // 处理增量专项债
  21. if err != nil {
  22. log.Info("main", zap.Any("AddFunc err", err))
  23. }
  24. log.Info("main", zap.Any("eid", eid))
  25. c.Start()
  26. defer c.Stop()
  27. select {}
  28. //importData() //导入专项债数据
  29. //dealProject() // 处理项目、债券数据
  30. //dealProjectBondRelation() //更新项目和债券的关联关系
  31. //dealProjectListAll() //处理存量项目列表数据;补充列表数据使用
  32. /*导出数据*/
  33. //exportData()
  34. }
  35. // importData 导入数据
  36. func importData() {
  37. dealProject() // 处理项目、债券数据
  38. dealProjectBondRelation() //更新项目和债券的关联关系
  39. log.Info("importData:", zap.String("count", "数据处理完毕"))
  40. }
  41. // dealProject 处理项目相关数据
  42. func dealProject() {
  43. tables := strings.Split(GF.Mongob.List, ",")
  44. detailNames := strings.Split(GF.Mongob.Detail, ",")
  45. //table := GF.Mongob.List
  46. //detailName := GF.Mongob.Detail //专项债详细表
  47. sess := Mgo.GetMgoConn()
  48. defer Mgo.DestoryMongoConn(sess)
  49. for k, table := range tables {
  50. detailName := detailNames[k]
  51. /**
  52. 这里测试用,使用 “新建成都至达州至万州铁路(南充段)(万源市)” 这个项目测试,他有变更信息
  53. */
  54. //where1 := map[string]interface{}{
  55. // //"projectName": "新建成都至达州至万州铁路(南充段)(万源市)",
  56. // "projectName": "新建成都至达州至万州铁路(南充段)(万源市)",
  57. //}
  58. //query := sess.DB("py_theme").C(table).Find(where1).Select(nil).Iter()
  59. query := sess.DB("py_theme").C(table).Find(nil).Select(nil).Iter()
  60. count := 0
  61. for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
  62. if count%100 == 0 {
  63. log.Info("current:", zap.Int("count", count), zap.Any("projectName", tmp["projectName"]), zap.String(table, detailName))
  64. }
  65. projectName := util.ObjToString(tmp["projectName"])
  66. //0.项目列表数据
  67. dealProjectList(tmp)
  68. //
  69. where := map[string]interface{}{
  70. "project.jcxx.projectName": projectName,
  71. }
  72. detail, _ := Mgo.FindOne(detailName, where)
  73. if len(*detail) == 0 {
  74. continue
  75. }
  76. //1.项目数据
  77. if project, ok := (*detail)["project"].(map[string]interface{}); ok {
  78. //1.基础信息
  79. var projectId int
  80. if jcxx, ok := project["jcxx"].(map[string]interface{}); ok {
  81. projectId = dealProjectBaseInfo(jcxx, projectName)
  82. }
  83. //2.还本付息
  84. if hbfx, ok := project["hbfx"].(map[string]interface{}); ok {
  85. dealProjectRepayment(hbfx, projectName, projectId)
  86. }
  87. //3.变更
  88. if bg, ok := project["bg"].([]interface{}); ok {
  89. if len(bg) > 0 {
  90. dealProjectChange(bg, projectName, projectId)
  91. }
  92. }
  93. //4.发行明细
  94. if fxmx, ok := project["fxmx"].([]interface{}); ok {
  95. if len(fxmx) > 0 {
  96. dealProjectIssueDetails(fxmx, projectName, projectId)
  97. }
  98. }
  99. }
  100. //2.处理债券信息
  101. if bonds, ok := (*detail)["bond"].([]interface{}); ok {
  102. dealBondInfo(bonds, projectName)
  103. }
  104. }
  105. log.Info("dealProject over ", zap.Int("total", count))
  106. }
  107. }
  108. // dealProjectBaseInfo 处理项目基本信息
  109. func dealProjectBaseInfo(jcxx map[string]interface{}, projectName string) (projectID int) {
  110. project_base_info_exist := ProjectBaseInfo{}
  111. err := MysqlDB.Where(&ProjectBaseInfo{ProjectName: projectName}).First(&project_base_info_exist).Error
  112. if err != nil && err != gorm.ErrRecordNotFound {
  113. // 处理查询错误
  114. log.Error("dealProjectBaseInfo;Error checking for existing project", zap.Error(err))
  115. }
  116. //当前项目基本信息已经存在
  117. if project_base_info_exist.ID > 0 {
  118. // 直接修改已存在记录的字段
  119. project_base_info_exist.TotalInvestment = util.Float64All(jcxx["totalInvestment"])
  120. project_base_info_exist.Area = util.ObjToString(jcxx["regionName"])
  121. project_base_info_exist.City = strings.ReplaceAll(util.ObjToString(jcxx["cityName"]), "本级", "")
  122. project_base_info_exist.District = strings.ReplaceAll(util.ObjToString(jcxx["countyName"]), "本级", "")
  123. project_base_info_exist.Capital = util.Float64All(jcxx["capital"])
  124. project_base_info_exist.ApplyTotalBonds = util.Float64All(jcxx["applyDebt"])
  125. project_base_info_exist.OtherDebtFinancing = util.Float64All(jcxx["portfolioFinancing"])
  126. project_base_info_exist.SpecialDebtCapital = util.Float64All(jcxx["specialDebtAsCapital"])
  127. project_base_info_exist.ExpectedReturn = util.Float64All(jcxx["expectedReturn"])
  128. project_base_info_exist.ProjectCost = util.IntAll(jcxx["projectCost"])
  129. project_base_info_exist.ProjectDomain = util.ObjToString(jcxx["projectTypeName3"])
  130. project_base_info_exist.ProjectOwner = util.ObjToString(jcxx["projectSubject"])
  131. project_base_info_exist.StartDate = util.ObjToString(jcxx["startDate"])
  132. project_base_info_exist.EndDate = util.ObjToString(jcxx["endDate"])
  133. project_base_info_exist.OperationStartDate = util.ObjToString(jcxx["operationStartDate"])
  134. project_base_info_exist.OperationEndDate = util.ObjToString(jcxx["operationEndDate"])
  135. project_base_info_exist.SourceIncome = util.ObjToString(jcxx["sourceIncome"])
  136. project_base_info_exist.ConstructionContent = util.ObjToString(jcxx["constructionContent"])
  137. project_base_info_exist.Remarks = util.ObjToString(jcxx["remarks"])
  138. project_base_info_exist.OtherDebtFinancingSource = util.ObjToString(jcxx["portfolioFinancingSource"])
  139. project_base_info_exist.CostIncomePercent = util.ObjToString(jcxx["costIncomePercent"])
  140. project_base_info_exist.CoverageMultiple = util.Float64All(jcxx["coverageMultiple"])
  141. project_base_info_exist.CompetentDepartment = util.ObjToString(jcxx["implementingAgency"])
  142. project_base_info_exist.AccountingFirm = util.ObjToString(jcxx["accountingFirm"])
  143. project_base_info_exist.LawFirm = util.ObjToString(jcxx["lawFirm"])
  144. project_base_info_exist.UpdateDate = util.ObjToString(jcxx["updateTime"])
  145. project_base_info_exist.CreateDate = util.ObjToString(jcxx["createTime"])
  146. // 使用 Save 来更新
  147. err = MysqlDB.Save(&project_base_info_exist).Error
  148. if err != nil {
  149. log.Info("dealProject Save (Update) ", zap.Error(err))
  150. }
  151. return project_base_info_exist.ID
  152. } else {
  153. project_base_info := ProjectBaseInfo{
  154. ProjectName: projectName,
  155. TotalInvestment: util.Float64All(jcxx["totalInvestment"]),
  156. Area: util.ObjToString(jcxx["regionName"]),
  157. City: strings.ReplaceAll(util.ObjToString(jcxx["cityName"]), "本级", ""),
  158. District: strings.ReplaceAll(util.ObjToString(jcxx["countyName"]), "本级", ""),
  159. Capital: util.Float64All(jcxx["capital"]),
  160. ApplyTotalBonds: util.Float64All(jcxx["applyDebt"]),
  161. OtherDebtFinancing: util.Float64All(jcxx["portfolioFinancing"]),
  162. SpecialDebtCapital: util.Float64All(jcxx["specialDebtAsCapital"]),
  163. ExpectedReturn: util.Float64All(jcxx["expectedReturn"]),
  164. ProjectCost: util.IntAll(jcxx["projectCost"]),
  165. ProjectDomain: util.ObjToString(jcxx["projectTypeName3"]), // 项目领域
  166. ProjectOwner: util.ObjToString(jcxx["projectSubject"]), //项目业主
  167. StartDate: util.ObjToString(jcxx["startDate"]),
  168. EndDate: util.ObjToString(jcxx["endDate"]),
  169. OperationStartDate: util.ObjToString(jcxx["operationStartDate"]),
  170. OperationEndDate: util.ObjToString(jcxx["operationEndDate"]),
  171. SourceIncome: util.ObjToString(jcxx["sourceIncome"]),
  172. ConstructionContent: util.ObjToString(jcxx["constructionContent"]),
  173. Remarks: util.ObjToString(jcxx["remarks"]),
  174. OtherDebtFinancingSource: util.ObjToString(jcxx["portfolioFinancingSource"]),
  175. CostIncomePercent: util.ObjToString(jcxx["costIncomePercent"]),
  176. CoverageMultiple: util.Float64All(jcxx["coverageMultiple"]),
  177. CompetentDepartment: util.ObjToString(jcxx["implementingAgency"]),
  178. AccountingFirm: util.ObjToString(jcxx["accountingFirm"]),
  179. LawFirm: util.ObjToString(jcxx["lawFirm"]),
  180. CreateDate: util.ObjToString(jcxx["createTime"]),
  181. UpdateDate: util.ObjToString(jcxx["updateTime"]),
  182. }
  183. err = MysqlDB.Create(&project_base_info).Error
  184. if err != nil {
  185. log.Info("dealProjectBaseInfo;dealProject Create ", zap.Error(err), zap.String("project", projectName))
  186. }
  187. return project_base_info.ID
  188. }
  189. }
  190. // dealProjectRepayment 处理项目-还本付息
  191. func dealProjectRepayment(hbfx map[string]interface{}, projectName string, projectID int) {
  192. project_repayment_exist := ProjectRepayment{}
  193. err := MysqlDB.Where(&ProjectRepayment{ProjectName: projectName, BondName: util.ObjToString(hbfx["bondName"])}).First(&project_repayment_exist).Error
  194. if err != nil && err != gorm.ErrRecordNotFound {
  195. // 处理查询错误
  196. log.Error("dealProjectRepayment;Error checking for existing project", zap.Error(err))
  197. }
  198. if project_repayment_exist.ID > 0 {
  199. project_repayment_exist.IssueTerm = util.IntAll(hbfx["issueTerm"])
  200. project_repayment_exist.PayInterestMethodName = util.ObjToString(hbfx["payInterestMethodName"])
  201. project_repayment_exist.ValueDate = util.ObjToString(hbfx["valueDate"])
  202. project_repayment_exist.InterestDate = util.ObjToString(hbfx["payInterestDate"])
  203. project_repayment_exist.LastInterestDate = util.ObjToString(hbfx["latelyPayInterestDate"])
  204. project_repayment_exist.ReminderRepayDays = util.IntAll(hbfx["days"])
  205. project_repayment_exist.MaturityDate = util.ObjToString(hbfx["expiryDate"])
  206. project_repayment_exist.DebtService = util.Float64All(hbfx["repayCapitalWithInterest"])
  207. project_repayment_exist.RedemptionMethod = util.ObjToString(hbfx["redemptionMethod"])
  208. project_repayment_exist.CumulativePayInterest = util.IntAll(hbfx["cumulativePayInterest"])
  209. project_repayment_exist.IsEarlyRepayPrincipal = util.ObjToString(hbfx["isEarlyRepayPrincipal"])
  210. project_repayment_exist.Remarks = util.ObjToString(hbfx["remarks"])
  211. // 使用 Save 来更新
  212. err = MysqlDB.Save(&project_repayment_exist).Error
  213. if err != nil {
  214. log.Info("dealProjectRepayment Save (Update) ", zap.Error(err))
  215. }
  216. } else {
  217. project_payment := ProjectRepayment{
  218. ProjectName: projectName,
  219. ProjectID: projectID,
  220. BondName: util.ObjToString(hbfx["bondName"]),
  221. IssueTerm: util.IntAll(hbfx["issueTerm"]),
  222. PayInterestMethodName: util.ObjToString(hbfx["payInterestMethodName"]),
  223. ValueDate: util.ObjToString(hbfx["valueDate"]),
  224. InterestDate: util.ObjToString(hbfx["payInterestDate"]),
  225. LastInterestDate: util.ObjToString(hbfx["latelyPayInterestDate"]),
  226. ReminderRepayDays: util.IntAll(hbfx["days"]), //提醒还款天数
  227. MaturityDate: util.ObjToString(hbfx["expiryDate"]), //到期日
  228. DebtService: util.Float64All(hbfx["repayCapitalWithInterest"]), //还本付息(万元)
  229. RedemptionMethod: util.ObjToString(hbfx["redemptionMethod"]),
  230. CumulativePayInterest: util.IntAll(hbfx["cumulativePayInterest"]),
  231. IsEarlyRepayPrincipal: util.ObjToString(hbfx["isEarlyRepayPrincipal"]),
  232. Remarks: util.ObjToString(hbfx["remarks"]),
  233. }
  234. err = MysqlDB.Create(&project_payment).Error
  235. if err != nil {
  236. log.Info("dealProjectRepayment;dealProject Create ", zap.Error(err), zap.String("project", projectName))
  237. }
  238. }
  239. }
  240. // dealProjectChange 处理项目变更
  241. func dealProjectChange(bg []interface{}, projectName string, projectId int) {
  242. for _, v := range bg {
  243. if bgda, ok := v.(map[string]interface{}); ok {
  244. project_change := ProjectChange{
  245. ProjectName: projectName,
  246. ProjectID: projectId,
  247. ChangeContent: util.ObjToString(bgda["changeContent"]),
  248. UpdateReason: util.ObjToString(bgda["updateReason"]),
  249. SubmitTime: util.ObjToString(bgda["submitTime"]),
  250. }
  251. err := MysqlDB.Create(&project_change).Error
  252. if err != nil {
  253. log.Info("dealProjectChange; Create err", zap.Error(err))
  254. }
  255. }
  256. }
  257. }
  258. // dealProjectIssueDetails 处理项目发行明细
  259. func dealProjectIssueDetails(fxmx []interface{}, projectName string, projectId int) {
  260. for _, v := range fxmx {
  261. if fx, ok := v.(map[string]interface{}); ok {
  262. project_bach_name := util.ObjToString(fx["projectBatchName"])
  263. issue_detail_exist := ProjectIssueDetails{}
  264. err := MysqlDB.Where(&ProjectIssueDetails{ProjectName: projectName, ProjectBachName: project_bach_name, BondName: util.ObjToString(fx["bondName"])}).First(&issue_detail_exist).Error
  265. if err != nil && err != gorm.ErrRecordNotFound {
  266. // 处理查询错误
  267. log.Error("dealProjectIssueDetails;Error checking for existing project", zap.Error(err))
  268. }
  269. // 存在
  270. if issue_detail_exist.ID > 0 {
  271. issue_detail_exist.FirstPublishDate = util.ObjToString(fx["firstPublishDate"])
  272. issue_detail_exist.BatchNum = util.IntAll(fx["batchNum"])
  273. issue_detail_exist.PresentIssueAmount = util.Float64All(fx["presentIssueAmount"])
  274. issue_detail_exist.IssueInterestRate = util.Float64All(fx["issueInterestRate"])
  275. issue_detail_exist.PresentAsSpecialAmount = util.Float64All(fx["presentAsSpecialAmount"])
  276. issue_detail_exist.TotalIssueAmount = util.Float64All(fx["totalIssueAmount"])
  277. issue_detail_exist.ReviseLog = util.ObjToString(fx["revise_log"])
  278. err = MysqlDB.Save(&issue_detail_exist).Error
  279. if err != nil {
  280. log.Info("dealProjectIssueDetails Save (Update) ", zap.Error(err))
  281. }
  282. } else {
  283. issue_detail := ProjectIssueDetails{
  284. ProjectName: projectName,
  285. ProjectID: projectId,
  286. ProjectBachName: project_bach_name,
  287. BondName: util.ObjToString(fx["bondName"]),
  288. FirstPublishDate: util.ObjToString(fx["firstPublishDate"]),
  289. BatchNum: util.IntAll(fx["batchNum"]),
  290. PresentIssueAmount: util.Float64All(fx["presentIssueAmount"]),
  291. IssueInterestRate: util.Float64All(fx["issueInterestRate"]),
  292. PresentAsSpecialAmount: util.Float64All(fx["presentAsSpecialAmount"]),
  293. TotalIssueAmount: util.Float64All(fx["totalIssueAmount"]),
  294. ReviseLog: util.ObjToString(fx["revise_log"]),
  295. }
  296. err := MysqlDB.Create(&issue_detail).Error
  297. if err != nil {
  298. log.Info("dealProjectIssueDetails; Create err", zap.Error(err))
  299. }
  300. }
  301. }
  302. }
  303. }
  304. // dealBondInfo 处理债券信息
  305. func dealBondInfo(bonds []interface{}, projectName string) {
  306. for _, v := range bonds {
  307. if bond, ok := v.(map[string]interface{}); ok {
  308. //1.基本信息
  309. if jbxx, ok := bond["jbxx"].(map[string]interface{}); ok {
  310. dealBondBase(jbxx, projectName)
  311. }
  312. //2.债券-修改记录;xgjl
  313. if xgjl, ok := bond["xgjl"].([]interface{}); ok && len(xgjl) > 0 {
  314. dealBondChange(xgjl)
  315. }
  316. ////3.相关小项目
  317. //if xgxxx, ok := bond["xgxxx"].([]interface{}); ok && len(xgxxx) > 0 {
  318. // dealRelationProject(xgxxx)
  319. //}
  320. }
  321. }
  322. }
  323. // dealBondBase 处理债券基本信息
  324. func dealBondBase(jbxx map[string]interface{}, projectName string) {
  325. bond_info_exist := BondInfo{}
  326. err := MysqlDB.Where(&BondInfo{BondName: util.ObjToString(jbxx["bondName"]), BondNo: util.ObjToString(jbxx["bondNo"])}).First(&bond_info_exist).Error
  327. if err != nil && err != gorm.ErrRecordNotFound {
  328. // 处理查询错误
  329. log.Error("dealBondBase;Error checking for existing project", zap.Error(err))
  330. }
  331. if bond_info_exist.ID > 0 {
  332. bond_info_exist.BondShortName = util.ObjToString(jbxx["bondName"])
  333. bond_info_exist.Area = util.ObjToString(jbxx["regionName"])
  334. bond_info_exist.BondNature = util.ObjToString(jbxx["bondType1Name"])
  335. bond_info_exist.BondType = util.ObjToString(jbxx["bondType2Name"])
  336. bond_info_exist.OfficialProjectType = util.ObjToString(jbxx["projectType1Name"])
  337. bond_info_exist.TotalAmount = util.Float64All(jbxx["totalAmount"])
  338. bond_info_exist.IssueDate = util.ObjToString(jbxx["issueDate"])
  339. bond_info_exist.IssuePlace = util.ObjToString(jbxx["issuePlaceName"])
  340. bond_info_exist.IssueTerm = util.IntAll(jbxx["issueTerm"])
  341. bond_info_exist.IssueInterestRate = util.ObjToString(jbxx["issueInterestRate"])
  342. bond_info_exist.IssuePhase = util.ObjToString(jbxx["issuePhase"])
  343. bond_info_exist.WayOfPayInterest = util.ObjToString(jbxx["payInterestMethodName"])
  344. bond_info_exist.NewBondAmount = util.Float64All(jbxx["newBondAmount"])
  345. bond_info_exist.CounterBondAmount = util.Float64All(jbxx["counterBondAmount"])
  346. bond_info_exist.RefinancingBondAmount = util.Float64All(jbxx["refinancingBondAmount"])
  347. bond_info_exist.RedemptionMethod = util.ObjToString(jbxx["redemptionMethod"])
  348. bond_info_exist.ValueDate = util.ObjToString(jbxx["valueDate"])
  349. bond_info_exist.ExpiryDate = util.ObjToString(jbxx["expiryDate"])
  350. bond_info_exist.PayInterestDate = util.ObjToString(jbxx["payInterestDate"])
  351. bond_info_exist.LatePayInterestDate = util.ObjToString(jbxx["latelyPayInterestDate"])
  352. bond_info_exist.IsEarlyRepayPrincipal = util.ObjToString(jbxx["isEarlyRepayPrincipal"])
  353. bond_info_exist.CumulativePayInterest = util.Float64All(jbxx["cumulativePayInterest"])
  354. bond_info_exist.IsCounterBond = util.ObjToString(jbxx["isCounterBond"])
  355. err = MysqlDB.Save(&bond_info_exist).Error
  356. if err != nil {
  357. log.Info("dealBondBase Save (Update) ", zap.Error(err))
  358. }
  359. } else {
  360. bond_info := BondInfo{
  361. BondName: util.ObjToString(jbxx["bondName"]),
  362. BondShortName: util.ObjToString(jbxx["bondShortName"]),
  363. BondNo: util.ObjToString(jbxx["bondNo"]),
  364. Area: util.ObjToString(jbxx["regionName"]),
  365. BondNature: util.ObjToString(jbxx["bondType1Name"]),
  366. BondType: util.ObjToString(jbxx["bondType2Name"]),
  367. OfficialProjectType: util.ObjToString(jbxx["projectType1Name"]),
  368. TotalAmount: util.Float64All(jbxx["totalAmount"]),
  369. IssueDate: util.ObjToString(jbxx["issueDate"]),
  370. IssuePlace: util.ObjToString(jbxx["issuePlaceName"]),
  371. IssueTerm: util.IntAll(jbxx["issueTerm"]),
  372. IssueInterestRate: util.ObjToString(jbxx["issueInterestRate"]),
  373. IssuePhase: util.ObjToString(jbxx["issuePhase"]),
  374. WayOfPayInterest: util.ObjToString(jbxx["payInterestMethodName"]),
  375. NewBondAmount: util.Float64All(jbxx["newBondAmount"]),
  376. CounterBondAmount: util.Float64All(jbxx["counterBondAmount"]),
  377. RefinancingBondAmount: util.Float64All(jbxx["refinancingBondAmount"]),
  378. RedemptionMethod: util.ObjToString(jbxx["redemptionMethod"]),
  379. ValueDate: util.ObjToString(jbxx["valueDate"]),
  380. ExpiryDate: util.ObjToString(jbxx["expiryDate"]),
  381. PayInterestDate: util.ObjToString(jbxx["payInterestDate"]),
  382. LatePayInterestDate: util.ObjToString(jbxx["latelyPayInterestDate"]),
  383. RemindPayDays: util.IntAll(jbxx["days"]),
  384. LastPayInterest: util.Float64All(jbxx["lastPayInterest"]),
  385. IsEarlyRepayPrincipal: util.ObjToString(jbxx["isEarlyRepayPrincipal"]),
  386. CumulativePayInterest: util.Float64All(jbxx["cumulativePayInterest"]),
  387. IsCounterBond: util.ObjToString(jbxx["isCounterBond"]),
  388. }
  389. //
  390. err = MysqlDB.Create(&bond_info).Error
  391. if err != nil {
  392. log.Info("dealBondBase;dealProject Create ", zap.Error(err), zap.String("project", projectName))
  393. }
  394. }
  395. }
  396. // dealBondChange 修改债券修改信息
  397. func dealBondChange(xgjl []interface{}) {
  398. }
  399. // dealProjectBondRelation 处理项目债券关联关系
  400. func dealProjectBondRelation() {
  401. //detailName := GF.Mongob.Detail //专项债详细表
  402. detailNames := strings.Split(GF.Mongob.Detail, ",")
  403. sess := Mgo.GetMgoConn()
  404. defer Mgo.DestoryMongoConn(sess)
  405. for _, detailName := range detailNames {
  406. query := sess.DB("py_theme").C(detailName).Find(nil).Select(nil).Iter()
  407. count := 0
  408. for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
  409. if count%100 == 0 {
  410. log.Info("current:", zap.Int("count", count))
  411. }
  412. if project, ok := tmp["project"].(map[string]interface{}); ok {
  413. //1.基础信息
  414. if jcxx, ok := project["jcxx"].(map[string]interface{}); ok {
  415. project_name := util.ObjToString(jcxx["projectName"])
  416. //fmt.Println(project_name)
  417. project_base_info_exist := ProjectBaseInfo{}
  418. err := MysqlDB.Where(&ProjectBaseInfo{ProjectName: project_name}).First(&project_base_info_exist).Error
  419. if err != nil && err != gorm.ErrRecordNotFound {
  420. // 处理查询错误
  421. log.Error("dealProjectBondRelation;Error checking for existing project", zap.Error(err))
  422. } else {
  423. if project_base_info_exist.ID > 0 {
  424. //1.更新项目的创建时间
  425. project_base_info_exist.CreateDate = util.ObjToString(jcxx["createTime"])
  426. project_base_info_exist.UpdateDate = util.ObjToString(jcxx["updateTime"])
  427. err = MysqlDB.Save(&project_base_info_exist).Error
  428. if err != nil {
  429. log.Info("dealProjectBondRelation Save (Update) ", zap.Error(err))
  430. }
  431. //2.更新项目和债券的对应关系
  432. if fxmx, ok := project["fxmx"].([]interface{}); ok {
  433. for _, v := range fxmx {
  434. if fx, ok := v.(map[string]interface{}); ok {
  435. bond_name := util.ObjToString(fx["bondName"])
  436. bond_info_exist := BondInfo{}
  437. err := MysqlDB.Where(&BondInfo{BondName: bond_name}).First(&bond_info_exist).Error
  438. if err != nil && err != gorm.ErrRecordNotFound {
  439. // 处理查询错误
  440. log.Error("dealProjectBondRelation;Error checking for existing project", zap.Error(err))
  441. } else {
  442. // 整理对应关系
  443. relation := ProjectBondRelation{
  444. ProjectID: project_base_info_exist.ID,
  445. BondID: bond_info_exist.ID,
  446. }
  447. result := MysqlDB.FirstOrCreate(&relation, ProjectBondRelation{
  448. ProjectID: relation.ProjectID,
  449. BondID: relation.BondID,
  450. })
  451. if result.Error != nil {
  452. // 处理错误
  453. log.Error("dealProjectBondRelation;Error checking for existing project", zap.Error(err), zap.String(project_name, bond_name))
  454. }
  455. }
  456. }
  457. }
  458. }
  459. }
  460. }
  461. }
  462. }
  463. }
  464. log.Info("dealProjectBondRelation", zap.Any("数据处理完毕", count))
  465. }
  466. }
  467. // dealProjectList 处理项目列表数据
  468. func dealProjectList(tmp map[string]interface{}) {
  469. projectName := util.ObjToString(tmp["projectName"])
  470. project_list_exist := ProjectListInfo{}
  471. err := MysqlDB.Where(&ProjectListInfo{ProjectName: projectName}).First(&project_list_exist).Error
  472. if err != nil && err != gorm.ErrRecordNotFound {
  473. // 处理查询错误
  474. log.Error("dealProjectList;Error checking for existing project", zap.Error(err))
  475. }
  476. if project_list_exist.ID > 0 {
  477. project_list_exist.Area = util.ObjToString(tmp["regionName"])
  478. project_list_exist.City = util.ObjToString(tmp["cityName"])
  479. project_list_exist.District = util.ObjToString(tmp["countyName"])
  480. project_list_exist.IssueTerm = util.ObjToString(tmp["issueTerm"])
  481. project_list_exist.IssueDate = util.ObjToString(tmp["issueDate"])
  482. project_list_exist.IssueInterestRate = util.ObjToString(tmp["issueInterestRate"])
  483. project_list_exist.TotalInvestment = util.Float64All(tmp["totalInvestment"])
  484. project_list_exist.PresentIssueAmount = util.Float64All(tmp["presentIssueAmount"])
  485. err = MysqlDB.Save(&project_list_exist).Error
  486. if err != nil {
  487. log.Info("dealProjectList Save (Update) ", zap.Error(err))
  488. }
  489. } else {
  490. project_list := ProjectListInfo{
  491. ProjectName: projectName,
  492. Area: util.ObjToString(tmp["regionName"]),
  493. City: util.ObjToString(tmp["cityName"]),
  494. District: util.ObjToString(tmp["countyName"]),
  495. TotalInvestment: util.Float64All(tmp["totalInvestment"]),
  496. PresentIssueAmount: util.Float64All(tmp["presentIssueAmount"]),
  497. IssueTerm: util.ObjToString(tmp["issueTerm"]),
  498. IssueDate: util.ObjToString(tmp["issueDate"]),
  499. IssueInterestRate: util.ObjToString(tmp["issueInterestRate"]),
  500. }
  501. err = MysqlDB.Create(&project_list).Error
  502. if err != nil {
  503. log.Info("dealProjectList;dealProject Create ", zap.Error(err), zap.String("project", projectName))
  504. }
  505. }
  506. }
  507. // dealProjectList 处理项目列表存量数据
  508. func dealProjectListAll() {
  509. tables := strings.Split(GF.Mongob.List, ",")
  510. sess := Mgo.GetMgoConn()
  511. defer Mgo.DestoryMongoConn(sess)
  512. for _, table := range tables {
  513. query := sess.DB("py_theme").C(table).Find(nil).Select(nil).Iter()
  514. count := 0
  515. for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
  516. if count%100 == 0 {
  517. log.Info("current:", zap.Int("count", count), zap.Any("projectName", tmp["projectName"]), zap.String("当前数据表:", table))
  518. }
  519. projectName := util.ObjToString(tmp["projectName"])
  520. project_list_exist := ProjectListInfo{}
  521. err := MysqlDB.Where(&ProjectListInfo{ProjectName: projectName}).First(&project_list_exist).Error
  522. if err != nil && err != gorm.ErrRecordNotFound {
  523. // 处理查询错误
  524. log.Error("dealProjectList;Error checking for existing project", zap.Error(err))
  525. }
  526. if project_list_exist.ID > 0 {
  527. project_list_exist.Area = util.ObjToString(tmp["regionName"])
  528. project_list_exist.City = util.ObjToString(tmp["cityName"])
  529. project_list_exist.District = util.ObjToString(tmp["countyName"])
  530. project_list_exist.IssueTerm = util.ObjToString(tmp["issueTerm"])
  531. project_list_exist.IssueDate = util.ObjToString(tmp["issueDate"])
  532. project_list_exist.IssueInterestRate = util.ObjToString(tmp["issueInterestRate"])
  533. project_list_exist.TotalInvestment = util.Float64All(tmp["totalInvestment"])
  534. project_list_exist.PresentIssueAmount = util.Float64All(tmp["presentIssueAmount"])
  535. err = MysqlDB.Save(&project_list_exist).Error
  536. if err != nil {
  537. log.Info("dealProjectList Save (Update) ", zap.Error(err))
  538. }
  539. } else {
  540. project_list := ProjectListInfo{
  541. ProjectName: projectName,
  542. Area: util.ObjToString(tmp["regionName"]),
  543. City: util.ObjToString(tmp["cityName"]),
  544. District: util.ObjToString(tmp["countyName"]),
  545. TotalInvestment: util.Float64All(tmp["totalInvestment"]),
  546. PresentIssueAmount: util.Float64All(tmp["presentIssueAmount"]),
  547. IssueTerm: util.ObjToString(tmp["issueTerm"]),
  548. IssueDate: util.ObjToString(tmp["issueDate"]),
  549. IssueInterestRate: util.ObjToString(tmp["issueInterestRate"]),
  550. }
  551. err = MysqlDB.Create(&project_list).Error
  552. if err != nil {
  553. log.Info("dealProjectList;dealProject Create ", zap.Error(err), zap.String("project", projectName))
  554. }
  555. }
  556. }
  557. }
  558. log.Info("dealProjectList", zap.String("数据处理完毕", "!!!!!"))
  559. }