conf_task.py 9.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243
  1. import os
  2. import sys
  3. import time
  4. import requests
  5. from flask import Flask, request
  6. from flask import jsonify
  7. sys.path.append(os.path.abspath(os.path.join(os.getcwd(), "..")))
  8. from common.db.mysql_help import MysqlHelper
  9. from user_spider.user_info import *
  10. app = Flask(__name__)
  11. app.config['JSON_AS_ASCII'] = False
  12. # 只接受get方法访问
  13. @app.route("/v1/crawler/source/getall", methods=["GET"])
  14. def getSource():
  15. try:
  16. # 获取传入的params参数
  17. get_data = request.args.to_dict()
  18. # # 对参数进行操作
  19. sql = 'select source, task_type, spider_name, machine, source_desc, task_type_desc, spider_name_desc from crawler_source'
  20. result = MysqlHelper.get_values(sql)
  21. if not result:
  22. return jsonify({'code': '200', 'result': [], 'message': '没有更多数据'})
  23. source_list = list()
  24. for source, task_type, spider_name, machine, source_desc, task_type_desc, spider_name_desc in result:
  25. source_dict = {
  26. 'task_type': [
  27. {
  28. 'type': task_type,
  29. 'description': task_type_desc,
  30. 'spider': {
  31. 'spider_name': spider_name,
  32. 'description': spider_name_desc
  33. }
  34. }
  35. ],
  36. 'description': source_desc,
  37. 'source': source,
  38. 'machine': machine
  39. }
  40. source_list.append(source_dict)
  41. except Exception as e:
  42. return jsonify({'code': '400', 'message': '获取数据源信息失败'})
  43. return jsonify({'code': '200', 'result': source_list})
  44. @app.route("/v1/crawler/task/insert", methods=["POST"])
  45. def insertTask():
  46. try:
  47. data = request.json
  48. outer_info = data.get('spider_link')
  49. source = data.get('source')
  50. exist_outer_info = list()
  51. for link in outer_info:
  52. s_sql = f"""select spider_link from crawler_task where source={source}"""
  53. result = MysqlHelper.get_values(s_sql)
  54. if not result:
  55. continue
  56. if link in eval(result[0]):
  57. exist_outer_info.append(link)
  58. if exist_outer_info:
  59. return jsonify({'code': 200, 'message': '名单重复', 'repeat_list': exist_outer_info})
  60. # 字段转换
  61. tag_name_list = []
  62. content_tag_list = []
  63. user_tag = data['user_tag']
  64. user_content_tag = data['user_content_tag']
  65. for tag in user_tag:
  66. tag_name_list.append(tag['tagName'])
  67. for tag in user_content_tag:
  68. content_tag_list.append(tag['tagName'])
  69. data['min_publish_time'] = int(data['min_publish_time'] / 1000)
  70. data['next_time'] = int(data['next_time'] / 1000)
  71. data['insert_time'] = int(time.time())
  72. data['update_time'] = int(time.time())
  73. data['spider_link'] = str(data['spider_link'])
  74. data['spider_rule'] = str(data['spider_rule'])
  75. data['user_tag'] = ','.join(str(i) for i in tag_name_list)
  76. data['user_content_tag'] = ','.join(str(i) for i in content_tag_list)
  77. # data['crawler_interval'] = data.pop('interval')
  78. # 获取到一个以键且为逗号分隔的字符串,返回一个字符串
  79. keys = ','.join(data.keys())
  80. values = ','.join(['%s'] * len(data))
  81. sql = 'insert into {table}({keys}) VALUES({values})'.format(table='crawler_task', keys=keys, values=values)
  82. MysqlHelper.insert_values(sql, tuple(data.values()))
  83. except Exception as e:
  84. return jsonify({'code': 400, 'message': '任务写入失败'})
  85. return jsonify({'code': 200, 'message': 'task create success'})
  86. @app.route("/v1/crawler/task/gettask", methods=["GET"])
  87. def getAllTask():
  88. try:
  89. get_data = request.args.to_dict()
  90. page = int(get_data.get('page', 1))
  91. offset = int(get_data.get('offset', 10))
  92. start_count = (page * offset) - offset
  93. end_count = page * offset
  94. sql = f"""select task_id, task_name from crawler_task limit {start_count}, {end_count}"""
  95. result = MysqlHelper.get_values(sql)
  96. if not result:
  97. return jsonify({'code': '200', 'result': [], 'message': 'no data'})
  98. source_list = list()
  99. for task_id, task_name in result:
  100. data = dict(
  101. task_id=task_id,
  102. task_name=task_name,
  103. )
  104. source_list.append(data)
  105. t_sql = f"""select count(*) from crawler_task"""
  106. t_res = MysqlHelper.get_values(t_sql)
  107. total = t_res[0]
  108. except Exception as e:
  109. return jsonify({"code": "400", 'message': "任务列表获取失败"})
  110. return jsonify({'code': '200', 'result': source_list, 'total': total})
  111. @app.route("/v1/crawler/task/getone", methods=["GET"])
  112. def getOneTask():
  113. try:
  114. get_data = request.args.to_dict()
  115. task_id = get_data['task_id']
  116. sql = f'select task_id, spider_link from crawler_task where task_id={task_id}'
  117. result = MysqlHelper.get_values(sql)
  118. if not result:
  119. return jsonify({'code': '200', 'result': [], 'message': 'no data'})
  120. for task_id, spider_link in result:
  121. data = dict(
  122. task_id=task_id,
  123. spider_link=spider_link,
  124. )
  125. except Exception as e:
  126. return jsonify({'code': '400', "message": "获取任务信息失败"})
  127. return jsonify({'code': '200', 'result': data})
  128. @app.route("/v1/crawler/task/update", methods=["POST"])
  129. def updateTask():
  130. try:
  131. task_id = request.json.get('task_id')
  132. spider_link = request.json.get('spider_link')
  133. print(spider_link, task_id)
  134. sql = f"""UPDATE crawler_task SET spider_link='{spider_link}' where task_id = {task_id}"""
  135. print(sql)
  136. result = MysqlHelper.update_values(sql)
  137. if result:
  138. return jsonify({'code': 200, 'message': 'task update success'})
  139. else:
  140. return jsonify({'code': 400, 'message': 'task update faild'})
  141. except Exception as e:
  142. return jsonify({'code': 400, 'message': '任务更新失败'})
  143. def get_user_info(source):
  144. source_spider = {
  145. 'xigua': xigua_user_info
  146. }
  147. return source_spider.get(source)
  148. @app.route("/v1/crawler/author/create", methods=["POST"])
  149. def createUser():
  150. get_media_url = 'http://videotest-internal.yishihui.com/longvideoapi/user/virtual/crawler/registerVirtualUser'
  151. spider_link = request.json.get('spider_link')
  152. source = request.json.get('source')
  153. task_type = request.json.get('task_type')
  154. applets_status = request.json.get('applets_status')
  155. app_status = request.json.get('app_status')
  156. user_tag = request.json.get('user_tag')
  157. user_content_tag = request.json.get('user_content_tag')
  158. success_list = list()
  159. fail_list = list()
  160. for author_url in spider_link:
  161. try:
  162. post_data = {
  163. # 'count': 1, # (必须)账号个数:传1
  164. # 'accountType': 4, # (必须)账号类型 :传 4 app虚拟账号
  165. 'pwd': '', # 密码 默认 12346
  166. 'nickName': '', # 昵称 默认 vuser......
  167. 'avatarUrl': '',
  168. # 头像Url 默认 http://weapppiccdn.yishihui.com/resources/images/pic_normal.png
  169. 'tagName': user_tag, # 多条数据用英文逗号分割
  170. }
  171. response = requests.post(url=get_media_url, params=post_data)
  172. media_id = response.json()['data']
  173. f_sql = f"""select spider_link from crawler_author_map where spider_link="{author_url}" """
  174. result = MysqlHelper.get_values(f_sql)
  175. if result:
  176. success_list.append(author_url)
  177. continue
  178. else:
  179. tag_name_list = []
  180. content_tag_list = []
  181. for tag in user_tag:
  182. tag_name_list.append(tag['tagName'])
  183. for tag in user_content_tag:
  184. content_tag_list.append(tag['tagName'])
  185. data = dict(
  186. spider_link=author_url,
  187. media_id=media_id,
  188. source=source,
  189. task_type=task_type,
  190. applets_status=applets_status,
  191. app_status=app_status,
  192. user_tag=','.join(str(i) for i in tag_name_list),
  193. user_content_tag=','.join(str(i) for i in content_tag_list),
  194. insert_time=int(time.time()),
  195. update_time=int(time.time())
  196. )
  197. keys = ','.join(data.keys())
  198. values = ','.join(['%s'] * len(data))
  199. table = 'crawler_author_map'
  200. sql = f"""insert into {table}({keys}) VALUES({values})"""
  201. result = MysqlHelper.insert_values(sql, tuple(data.values()))
  202. if not result:
  203. fail_list.append(author_url)
  204. else:
  205. success_list.append(author_url)
  206. except Exception as e:
  207. fail_list.append(author_url)
  208. continue
  209. return jsonify({'code': 200, 'result': {'success': success_list, 'fail': fail_list}})
  210. if __name__ == "__main__":
  211. app.run(debug=True, port=5050)