mail.go 9.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335
  1. package pcsg
  2. import (
  3. "encoding/json"
  4. "eta/eta_email_analysis/global"
  5. "eta/eta_email_analysis/models/eta"
  6. "eta/eta_email_analysis/models/report"
  7. "eta/eta_email_analysis/services/oss"
  8. "eta/eta_email_analysis/utils"
  9. "eta/eta_email_analysis/utils/mail"
  10. "fmt"
  11. "github.com/google/uuid"
  12. "html"
  13. "io/fs"
  14. "log"
  15. "os"
  16. "path"
  17. "strings"
  18. "sync"
  19. "time"
  20. )
  21. // 同步用户锁,防止重复同步,不管是全量还是增量,都是同一时间只能一个同步
  22. var lockListenEmail sync.Mutex
  23. // 邮件改名规则
  24. var ruleList []report.MailRule
  25. func ListenMail() {
  26. var err error
  27. defer func() {
  28. if err != nil {
  29. global.FILE_LOG.Errorf("监听邮件失败:%s", err.Error())
  30. }
  31. }()
  32. //Fix()
  33. //return
  34. lockListenEmail.Lock()
  35. // 目录创建
  36. _ = ensureDirExists(fmt.Sprintf("%s%s", global.CONFIG.Serve.StaticDir, `file`))
  37. mailMessageChan := make(chan mail.MailMessage, 5) // 创建一个通道,用于接收邮件消息
  38. mailMessageDoneChan := make(chan bool, 1) // 创建一个通道,用于接收邮件消息
  39. //fmt.Println(len(mailMessageChan))
  40. // 获取规则
  41. ruleList = []report.MailRule{}
  42. ruleJson, e := eta.GetBusinessConfByKey("MailCheckRule")
  43. if e != nil {
  44. err = e
  45. global.FILE_LOG.Error("获取规则配置失败:%v", err)
  46. return
  47. }
  48. err = json.Unmarshal([]byte(ruleJson.ConfVal), &ruleList)
  49. if err != nil {
  50. global.FILE_LOG.Error("解析规则配置失败:%v", err)
  51. return
  52. }
  53. // 邮件监听后的处理函数
  54. go afterByListen(mailMessageChan, mailMessageDoneChan)
  55. fmt.Println("开始监听邮件")
  56. emailMessageUID := global.CONFIG.Email.StartReadIndex
  57. if emailMessageUID <= 0 {
  58. emailMessageUID, err = report.GetMaxOutsideReportByEmailMessageId()
  59. // 查找失败了,那么就记录
  60. if err != nil {
  61. global.FILE_LOG.Errorf("获取已入库的最大邮件id失败:%s", err.Error())
  62. }
  63. }
  64. {
  65. idList, err := report.GetAllOutsideReportByEmailMessageId()
  66. // 已经存在了,那么就返回
  67. if err != nil {
  68. global.FILE_LOG.Errorf("获取已入库的所有邮件id失败:%s", err.Error())
  69. } else {
  70. idMap := make(map[int]bool)
  71. for _, v := range idList {
  72. idMap[v] = true
  73. }
  74. mail.IsHandleMessageIdMap = idMap
  75. }
  76. }
  77. mailAddress := fmt.Sprintf("%s:%d", global.CONFIG.Email.Host, global.CONFIG.Email.Port)
  78. mail.ListenMail(mailAddress, global.CONFIG.Email.Folder, global.CONFIG.Email.UserName, global.CONFIG.Email.Password, global.CONFIG.Email.ReadBatchSize, emailMessageUID, mailMessageChan, mailMessageDoneChan)
  79. return
  80. }
  81. func afterByListen(mailMessageChan chan mail.MailMessage, mailMessageDoneChan chan bool) {
  82. defer func() {
  83. log.Println("监听读取结束")
  84. lockListenEmail.Unlock()
  85. }()
  86. for {
  87. select {
  88. case emailMessage := <-mailMessageChan:
  89. //fmt.Println("读取成功")
  90. //fmt.Println(emailMessage.Title)
  91. handleMailMessage(emailMessage)
  92. case <-time.After(10 * time.Second):
  93. //fmt.Println("监听超时了")
  94. break
  95. case <-mailMessageDoneChan:
  96. //fmt.Println("读取完成一轮了")
  97. for len(mailMessageChan) > 0 {
  98. emailMessage := <-mailMessageChan
  99. handleMailMessage(emailMessage)
  100. }
  101. //fmt.Println("结束了")
  102. return
  103. }
  104. }
  105. }
  106. func handleMailMessage(emailMessage mail.MailMessage) (err error) {
  107. defer func() {
  108. if err != nil {
  109. global.FILE_LOG.Errorf("邮件处理失败,邮件标题:%s,错误原因:%v", emailMessage.Title, err)
  110. }
  111. for _, v := range emailMessage.Resources {
  112. os.Remove(v)
  113. }
  114. for _, v := range emailMessage.Attachment {
  115. os.Remove(v)
  116. }
  117. }()
  118. //rootPath := `C:\Users\123\go\src\eta\eta_email_analysis\static\`
  119. ossClient := oss.NewOssClient()
  120. //fmt.Println(emailMessage.Title)
  121. outsideReportAttachmentList := make([]*report.OutsideReportAttachment, 0)
  122. emailMessageUID := int(emailMessage.Uid)
  123. outReport, err := report.GetOutsideReportByEmailMessageId(emailMessageUID)
  124. // 已经存在了,那么就返回
  125. if err == nil {
  126. if outReport.ReportUpdateTime != emailMessage.Date {
  127. // 如果更新时间不一样,那么就更新收件时间
  128. err = outReport.Update([]string{"ReportUpdateTime"})
  129. if err != nil {
  130. global.FILE_LOG.Errorf("更新收件时间失败:%s", err.Error())
  131. return
  132. }
  133. global.FILE_LOG.Debugf("更新收件时间,名称:%s, 原时间:%s, 新收件时间: %s", outReport.Title, outReport.ReportUpdateTime, emailMessage.Date)
  134. }
  135. global.FILE_LOG.Debugf("已存在,就不处理了,报告标题:%s;;邮件下标:%d", outReport.Title, emailMessage.Uid)
  136. return
  137. }
  138. fmt.Println("开始处理邮件,标题:", emailMessage.Title, ";邮件下标:", emailMessage.Uid)
  139. // sql报错,那么就返回
  140. if err != nil && !utils.IsErrNoRow(err) {
  141. return
  142. }
  143. // 已经存在了,那么就返回
  144. if outReport.OutsideReportID > 0 {
  145. return
  146. }
  147. title := emailMessage.Title
  148. var abstract string
  149. var classifyId int
  150. var classifyName string
  151. var sysUserId int
  152. var sysUserName string
  153. sysUserName = emailMessage.FromAddress
  154. // 查找用户
  155. if emailMessage.From != `` {
  156. tmpEmailStrList := strings.Split(sysUserName, "<")
  157. if len(tmpEmailStrList) >= 2 {
  158. sysUserName = tmpEmailStrList[1]
  159. tmpEmailStrList = strings.Split(sysUserName, ">")
  160. if len(tmpEmailStrList) >= 2 {
  161. sysUserName = tmpEmailStrList[0]
  162. }
  163. }
  164. sysUser, tmpErr := eta.GetSysUserByEmail(sysUserName)
  165. if tmpErr != nil && !utils.IsErrNoRow(tmpErr) {
  166. err = tmpErr
  167. return
  168. }
  169. if tmpErr == nil {
  170. sysUserId = sysUser.AdminId
  171. sysUserName = sysUser.RealName
  172. }
  173. }
  174. title = strings.Replace(title, "[PCI ANALYTICS CLOUD PLATFORM(00) Fwd] - ", "", -1)
  175. for _, v := range ruleList {
  176. tmpTitle := strings.ToLower(emailMessage.Title)
  177. rule := strings.ToLower(v.Rule)
  178. if strings.Contains(tmpTitle, rule) {
  179. title = v.Title
  180. abstract = v.Abstract
  181. classifyId = v.ClassifyId
  182. classify, e := report.GetClassifyByClassifyId(v.ClassifyId)
  183. if e != nil {
  184. err = e
  185. global.FILE_LOG.Error("获取分类失败:", err)
  186. return
  187. }
  188. classifyName = classify.ClassifyName
  189. // 特殊规则
  190. //if strings.Contains(title, "海外分公司市场信息月报") || strings.Contains(title, "区域市场信息月报") {
  191. // if emailMessage.From == "report.pcanalyst00@petrochina-usa.com" {
  192. // sysUserName = "PCI"
  193. // break
  194. // } else {
  195. // v.Author = emailMessage.From
  196. // }
  197. //}
  198. if v.Author != `` {
  199. if strings.Contains(v.Author, "@") {
  200. sysUser, tmpErr := eta.GetSysUserByEmail(v.Author)
  201. if tmpErr != nil && !utils.IsErrNoRow(tmpErr) {
  202. err = tmpErr
  203. return
  204. }
  205. if sysUser.AdminId > 0 {
  206. sysUserId = sysUser.AdminId
  207. sysUserName = sysUser.RealName
  208. } else {
  209. sysUserName = v.Author
  210. }
  211. } else {
  212. sysUserName = v.Author
  213. }
  214. }
  215. break
  216. }
  217. }
  218. for k, v := range emailMessage.Resources {
  219. randStr := utils.GetRandStringNoSpecialChar(28)
  220. ext := path.Ext(v)
  221. fileName := randStr + ext
  222. resourceUrl, tmpErr := ossClient.UploadFile(fileName, v, "")
  223. //os.Remove(v)
  224. if tmpErr != nil {
  225. global.FILE_LOG.Error(emailMessage.Title, "- 文件上传文件失败:", tmpErr)
  226. continue
  227. }
  228. tmpK := strings.Replace(k, "<", "cid:", -1)
  229. tmpK = strings.Replace(tmpK, ">", "", -1)
  230. emailMessage.Content = strings.Replace(emailMessage.Content, fmt.Sprint(tmpK, `"`), fmt.Sprint(resourceUrl, `"`), -1)
  231. }
  232. for name, v := range emailMessage.Attachment {
  233. var fileSize int64
  234. fileInfo, tmpErr := os.Stat(v)
  235. if tmpErr != nil {
  236. continue
  237. }
  238. fileSize = fileInfo.Size()
  239. randStr := utils.GetRandStringNoSpecialChar(28)
  240. ext := path.Ext(v)
  241. fileName := randStr + ext
  242. resourceUrl, tmpErr := ossClient.UploadFile(fileName, v, "")
  243. //defer os.Remove(v)
  244. if tmpErr != nil {
  245. global.FILE_LOG.Error(emailMessage.Title, " - 文件上传文件失败:", tmpErr)
  246. continue
  247. }
  248. outsideReportAttachmentList = append(outsideReportAttachmentList, &report.OutsideReportAttachment{
  249. OutsideReportAttachmentID: 0,
  250. OutsideReportID: 0,
  251. Title: name,
  252. URL: resourceUrl,
  253. CreateTime: time.Now(),
  254. FileSize: fileSize,
  255. })
  256. }
  257. //fileName := fmt.Sprintf("%s%s.%s", rootPath, emailMessage.Title, "html")
  258. htmlEscapeBody := html.EscapeString(emailMessage.Content)
  259. //fmt.Println(htmlEscapeBody)
  260. //err = utils.SaveToFile([]byte(emailMessage.Content), fileName)
  261. //if err != nil {
  262. // fmt.Println(fileName, "生成失败;err:", err)
  263. //}
  264. reportUpdateTime := emailMessage.Date
  265. if reportUpdateTime.IsZero() {
  266. reportUpdateTime = time.Now()
  267. }
  268. reportInfo := &report.OutsideReport{
  269. OutsideReportID: 0,
  270. Source: 3,
  271. Title: title,
  272. Abstract: abstract,
  273. ClassifyID: classifyId,
  274. ClassifyName: classifyName,
  275. Content: htmlEscapeBody,
  276. SysUserID: sysUserId,
  277. SysUserName: sysUserName,
  278. EmailMessageUID: emailMessageUID,
  279. ReportUpdateTime: reportUpdateTime,
  280. ModifyTime: time.Now(),
  281. CreateTime: time.Now(),
  282. ReportCode: uuid.New().String(),
  283. }
  284. err = report.CreateOutsideReport(reportInfo, outsideReportAttachmentList)
  285. return
  286. }
  287. func ensureDirExists(dirPath string) error {
  288. info, err := os.Stat(dirPath)
  289. if err == nil {
  290. if info.IsDir() {
  291. return nil // 目录已存在
  292. }
  293. return fmt.Errorf("path '%s' exists but is not a directory", dirPath)
  294. }
  295. if os.IsNotExist(err) {
  296. return os.MkdirAll(dirPath, fs.ModePerm)
  297. }
  298. return err
  299. }