user_action.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243
  1. from wechat_action.sql_tools import save_human_info, delete_wechat_info, save_wechat_info
  2. from wechat_api.get_wechat_info import WechatApi
  3. from wechat_action.create_ad_layout import CreateAd
  4. from wechat_action.create_ad_plan import CreateAdPlan
  5. from wechat_action import sql_tools
  6. from sqlalchemy import Table
  7. from communication_tools import dingtalk
  8. import json
  9. import logging
  10. from datetime import datetime
  11. from wechat_action.login_ad import LogIn
  12. # 定位为:接受请求之后对应的线程处理
  13. layout_create_action = 'create_ad_layout'
  14. ad_plan_create_action = 'create_ad_plan'
  15. refresh_wechat_action = 'refresh_wechat_info'
  16. def carry_plan(user_id, ad_plan_list, log_ad, db, cookie_canuse, task_name):
  17. # TODO:失败后 ,状态为:创建中 ----进行对应删除
  18. # 创建中的数据需要进行删除
  19. sql_session = db.DBSession()
  20. try:
  21. action_record_table = Table('action_record', db.metadata,
  22. autoload=True, autoload_with=db.engine)
  23. # cookies保存
  24. if not cookie_canuse:
  25. log_ad.cookies_save(log_ad, sql_session)
  26. for _ in ad_plan_list:
  27. service_name = _['service_name']
  28. wechat_name = _['wechat_name']
  29. plan_name = _['title']
  30. try:
  31. # 现在默认layout_name在30个字符以内
  32. layout_name = _['idea']['jump_type_page_type']['layout_name']
  33. try:
  34. # 1.检查1.落地页是否创建过了
  35. log_ad.select_ad_master(service_name, wechat_name, sql_session)
  36. if CreateAd.check_sucess_api(layout_name=layout_name, log_ad=log_ad):
  37. logging.info('{},落地页已经对应进行创建'.format(plan_name))
  38. res = {'sucess': True, 'result_info': '已经创建过对应落地页'}
  39. sql_tools.action_record(res, sql_session, layout_create_action, user_id, layout_name,
  40. action_record_table,
  41. service_name, wechat_name, task_name)
  42. else:
  43. # 1.5无则创建落地页
  44. create_ad_layout = CreateAd(login_ad=log_ad, service_name=service_name, wechat_name=wechat_name)
  45. layout_typesetting_dict = sql_tools.get_layout_typesetting(sql_session, user_id,
  46. typesetting_name=layout_name)
  47. layout_typesetting_dict = json.loads(layout_typesetting_dict)
  48. res = create_ad_layout.create_layout(layout_typesetting_dict, sql_session)
  49. sql_tools.action_record(res, sql_session, layout_create_action, user_id, layout_name,
  50. action_record_table,
  51. service_name, wechat_name, task_name)
  52. except Exception as e:
  53. # raise
  54. res = {'sucess': False, 'result_info': str(e)}
  55. logging.error('layout 创建失败,{}'.format(str(e)))
  56. # 创建落地页,计划都失败
  57. sql_tools.action_record(res, sql_session, layout_create_action, user_id, layout_name,
  58. action_record_table,
  59. service_name, wechat_name, task_name)
  60. res = {'sucess': False, 'result_info': '落地页创建失败'}
  61. sql_tools.action_record(res, sql_session, ad_plan_create_action, user_id, plan_name,
  62. action_record_table,
  63. service_name, wechat_name, task_name)
  64. dingtalk.send_message('user_id:{} 落地页创建出错,进行检查\n{}'.format(user_id, str(e)))
  65. continue
  66. log_ad.refresh_driver()
  67. # 3.创建计划
  68. log_ad.select_ad_master(service_name, wechat_name, sql_session)
  69. # plan_typesetting_dict = sql_tools.get_ad_plan_typesetting(sql_session=sql_session, user_id=user_id,
  70. # typesetting_name=plan_name)
  71. create_ad_plan = CreateAdPlan(login_ad=log_ad, task=_, service_name=service_name,
  72. wechat_name=wechat_name)
  73. res = create_ad_plan.run(sql_session)
  74. # 4.更新action_record相关计划信息
  75. sql_tools.action_record(res, sql_session, ad_plan_create_action, user_id, plan_name,
  76. action_record_table,
  77. service_name, wechat_name, task_name)
  78. logging.info('创建计划任务结束')
  79. except Exception as e:
  80. # raise
  81. log_ad.driver.save_screenshot(
  82. 'user_id:{}_time_{}_plan_name:{}_plan_error.png'.format(user_id, datetime.now().strftime(
  83. "%Y-%m-%d, %H:%M:%S"), plan_name))
  84. logging.error(str(e))
  85. res = {'sucess': False, 'result_info': str(e)}
  86. sql_tools.action_record(res, sql_session, ad_plan_create_action, user_id, plan_name,
  87. action_record_table,
  88. service_name, wechat_name, task_name)
  89. finally:
  90. log_ad.refresh_driver()
  91. except Exception as e:
  92. # raise
  93. sql_tools.update_task_status_error(sql_session=sql_session, user_id=user_id, task_name=task_name)
  94. dingtalk.send_message('user_id:{} 计划执行出错,进行检查\n{}'.format(user_id, str(e)))
  95. logging.error(e)
  96. finally:
  97. # raise
  98. logging.info('创建计划,任务结束')
  99. # 每次运行微信相关操作,对微信相关信息进行刷新
  100. sql_session.commit()
  101. log_ad.driver.quit()
  102. check_task_in_hand(user_id, db)
  103. def get_human_info(user_id, log_ad, db, cookie_canuse, task_name):
  104. # 数据库
  105. action_record_table = Table('action_record', db.metadata,
  106. autoload=True, autoload_with=db.engine)
  107. human_info_table = Table('human_info', db.metadata,
  108. autoload=True, autoload_with=db.engine)
  109. wechat_info_table = Table('wechat_info', db.metadata,
  110. autoload=True, autoload_with=db.engine)
  111. sql_session = db.DBSession()
  112. # 行为记录对应参数
  113. object_name = ''
  114. service_name = ''
  115. wechat_name = ''
  116. try:
  117. # 1.cookies保存
  118. if not cookie_canuse:
  119. log_ad.cookies_save(log_ad, sql_session)
  120. # 动作存放
  121. object_name = ''
  122. action_info = {'user_id': user_id, 'service_name': service_name,
  123. 'wechat_name': wechat_name,
  124. 'action_type': refresh_wechat_action, 'object_name': object_name, 'task_name': task_name,
  125. 'status': 'todo'}
  126. record_insert = sql_tools.save_action_record(action_record_info=action_info,
  127. table_action_record=action_record_table)
  128. sql_session.execute(record_insert)
  129. sql_session.commit()
  130. # wechat_info.每次都删除掉前面全部数据,进行更新
  131. # human_info 进行全局更新
  132. w_api = WechatApi(log_ad=log_ad)
  133. res_info = w_api.get_human_info(sql_session)
  134. if not res_info['sucess']:
  135. dingtalk.send_message('user_id:{} 获取微信数据为空,进行检查'.format(user_id))
  136. sql_tools.action_record(res_info, sql_session, refresh_wechat_action, user_id,
  137. object_name, action_record_table,
  138. service_name, wechat_name, task_name)
  139. return
  140. # wechat info进行数据更新
  141. # 1.删除所有数据
  142. delete_wechat_info(sql_session=sql_session, user_id=user_id)
  143. # 2.重新添加一遍相关数据
  144. for _ in res_info['result_list']:
  145. service_name_wechat = _['service_name']
  146. wechat_name_wechat = _['wechat_name']
  147. appid = _['appid']
  148. wxname = _['wxname']
  149. wechat_info = {'appid': appid, 'wxname': wxname, 'service_name': service_name_wechat,
  150. 'wechat_name': wechat_name_wechat,
  151. 'user_id': user_id}
  152. wechat_insert = save_wechat_info(wechat_info=wechat_info,
  153. table_wechat=wechat_info_table)
  154. sql_session.execute(wechat_insert)
  155. sql_session.commit()
  156. logging.info('update wechat info')
  157. # human info 相关数据进行更新
  158. for _ in res_info['result_list']:
  159. service_name_human = _['service_name']
  160. wechat_name_human = _['wechat_name']
  161. human_info = _['data']['list']
  162. update_res = human_info_table.update() \
  163. .where(human_info_table.c.service_name == service_name_human) \
  164. .where(human_info_table.c.wechat_name == wechat_name_human) \
  165. .values(human_info=human_info)
  166. update_res = sql_session.execute(update_res)
  167. sql_session.commit()
  168. if update_res.rowcount == 0:
  169. human_info = {'service_name': service_name_human, 'wechat_name': wechat_name_human,
  170. 'human_info': human_info}
  171. human_insert = save_human_info(human_info=human_info,
  172. table_human=human_info_table)
  173. sql_session.execute(human_insert)
  174. sql_session.commit()
  175. logging.info('update human info')
  176. sql_tools.action_record(res_info, sql_session, refresh_wechat_action,
  177. user_id, object_name, action_record_table,
  178. service_name, wechat_name, task_name)
  179. log_ad.refresh_driver()
  180. except Exception as e:
  181. res_info = {'sucess': False, 'result_info': str(e)}
  182. sql_tools.action_record(res_info, sql_session, refresh_wechat_action,
  183. user_id, object_name, action_record_table,
  184. service_name, wechat_name, task_name)
  185. log_ad.driver.save_screenshot(
  186. 'user_id:{}_time_{}_wechat_info_error.png'.format(user_id, datetime.now().strftime("%Y-%m-%d, %H:%M:%S")))
  187. dingtalk.send_message('user_id:{} 获取微信数据出错,进行检查\n{}'.format(user_id, str(e)))
  188. log_ad.refresh_driver()
  189. logging.error(e)
  190. finally:
  191. logging.info('刷新用户,任务结束')
  192. sql_session.commit()
  193. log_ad.driver.quit()
  194. check_task_in_hand(user_id=user_id, db=db)
  195. def check_task_in_hand(user_id, db):
  196. sql_session = db.DBSession()
  197. lines = sql_tools.get_task_in_hand_limit_one(user_id=user_id, sql_session=sql_session)
  198. task_name = ''
  199. ad_plan_list = []
  200. action_type = ''
  201. for _ in lines:
  202. typesetting, wechat_name, service_name, action_type, task_name = _
  203. ad_plan_list.append(json.loads(typesetting))
  204. if action_type == refresh_wechat_action:
  205. # 刷新log_ad,以免log_ad超过生命周期
  206. log_ad = LogIn(user_id=user_id)
  207. get_human_info(user_id, log_ad, db, True, task_name)
  208. if action_type == ad_plan_create_action:
  209. log_ad = LogIn(user_id=user_id)
  210. carry_plan(user_id, ad_plan_list, log_ad, db, True, task_name)
  211. if __name__ == "__main__":
  212. pass