conf_platform/server/conf_task.py

import copy
import logging
import os
import sys
import time
import requests
from flask import Flask, request
from flask import jsonify
from dotenv import load_dotenv

sys.path.append(os.path.abspath(os.path.join(os.getcwd(), "..")))
from conf.config import get_config
from common.db.mysql_help import MysqlHelper

load_dotenv(verbose=True)
env = os.getenv('env')
app = Flask(__name__)
app.config['JSON_AS_ASCII'] = False
# mysql实例
mysql_con = MysqlHelper()
conf = get_config()


@app.route("/v1/crawler/task/addlink", methods=["POST"])
def addSpiderLink():
    try:
        data = request.json
        spider_link = data['spider_link']
        task_id = data['task_id']

        sql = f'select * from crawler_author_map where spider_link={spider_link}'
        result = mysql_con.get_values(sql)
        now_time = int(time.time())
        if result:
            old_task_id = result[0]['task_id']
            if task_id == old_task_id:
                up_sql = f'update crawler_author_map set is_del=1 where spider_link="{spider_link}"'
            else:
                up_sql = f'update crawler_author_map set task_id={task_id} where spider_link="{spider_link}"'
            res = mysql_con.update_values(up_sql)

            task_sql = f'select spider_link from crawler_task where task_id ={task_id}'
            task = mysql_con.get_values(task_sql)
            spider_links = eval(task[0]['spider_link'])
            spider_links.append(spider_link)
            str_spider_links = str(spider_links)
            u_sql = f'update crawler_task set spider_link="{str_spider_links}", update_time={now_time} where task_id={task_id}'
            mysql_con.update_values(u_sql)
            return jsonify({'code': 200, 'message': '抓取名单增加成功', 'del_link': spider_link})
        else:
            sql = f'select * from crawler_task where task_id={task_id}'
            result = mysql_con.get_values(sql)
            success_list, fail_list = create_uid(result[0], task_id, spider_link=[spider_link])

            spider_links = eval(result[0]['spider_link'])
            spider_links.append(spider_link)
            str_spider_links = str(spider_links)

            u_sql = f'update crawler_task set spider_link="{str_spider_links}", update_time={now_time} where task_id={task_id}'
            mysql_con.update_values(u_sql)

            return jsonify({'code': 200, 'message': '抓取名单增加成功', 'add_link': success_list})
    except Exception as e:
        return jsonify(
            {'code': 400, 'message': '抓取名单删除失败', 'spider_link': spider_link})


@app.route("/v1/crawler/task/dellink", methods=["POST"])
def delSpiderLink():
    data = request.json
    spider_link = data['spider_link']
    task_id = data['task_id']
    up_sql = f'update crawler_author_map set is_del=0 where spider_link="{spider_link}"'
    MysqlHelper.update_values(up_sql)

    sql = f'select * from crawler_task where task_id ={task_id}'
    task = mysql_con.get_values(sql)
    spider_links = eval(task[0]['spider_link'])
    spider_links.remove(spider_link)
    now_time = int(time.time())
    u_sql = f'update crawler_task set spider_link="{spider_links}",update_time={now_time} where task_id={task_id}'
    mysql_con.update_values(u_sql)
    if spider_link:
        return jsonify({'code': 200, 'message': '抓取名单删除成功', 'del_link': spider_link})
    else:
        return jsonify(
            {'code': 400, 'message': '抓取名单删除失败', 'del_link': spider_link})


@app.route("/v1/crawler/task/getcategory", methods=["GET"])
def getCategory():
    sql = f'select id, content_category from crawler_content_category'
    result = mysql_con.get_values(sql)
    return jsonify({'code': 200, 'data': result})


@app.route("/v1/crawler/task/getboard", methods=["GET"])
def getBoard():
    sql = f'select id, mode_board from crawler_board'
    result = mysql_con.get_values(sql)
    return jsonify({'code': 200, 'data': result})


@app.route("/v1/crawler/task/getmodename", methods=["GET"])
def getModeName():
    sql = f'select id, mode_name from crawler_mode'
    result = mysql_con.get_values(sql)
    return jsonify({'code': 200, 'data': result})


@app.route("/v1/crawler/task/getrecommendboard", methods=["GET"])
def getRecommendBoard():
    sql = f'select id, mode_board from crawler_recommend_board'
    result = mysql_con.get_values(sql)
    return jsonify({'code': 200, 'data': result})


@app.route("/v1/crawler/user/findmedia", methods=["GET"])
def getMediaInfo():
    data = request.args.to_dict()
    task_id = data['task_id']
    sql = f'select * from crawler_author_map where task_id={task_id} and is_del=1'
    result = mysql_con.get_values(sql)
    task_user_info = []
    for task_info in result:
        media_id = task_info['media_id']

        media_info = requests.get(url=conf['select_media_url'], params={'uid': media_id}).json()['content']
        media_name = media_info['longvideoNickName'] if media_info['longvideoNickName'] else media_info['nickName']
        nick_name = task_info['nick_name']
        spider_link = task_info['spider_link']
        create_user_time = task_info['create_user_time']
        media_data = dict(
            media_name=media_name,
            nick_name=nick_name,
            spider_link=spider_link,
            media_id={'media_id': media_id, 'media_url': conf['media_main_url'].format(media_id)},
            create_user_time=create_user_time * 1000
        )
        task_user_info.append(media_data)
    return jsonify({'code': 200, 'data': task_user_info})


@app.route("/v1/crawler/task/findtask", methods=["GET"])
def getTaskUserInfo():
    # 根据条件查找任务
    data = request.args.to_dict()
    values = ''
    for k, v in data.items():
        if isinstance(v, int):
            values += f'{k}={v} and '
        else:
            values += f'{k}="{v}" and '

    sql = f"select task_id from crawler_author_map where {values[:-4]}"  # [:-1]是为了去掉末尾的逗号
    res = mysql_con.get_values(sql)
    task_id = res['task_id']

    sql = f'select task_name, source, task_type, create_task_user, insert_time, update_task_user, update_time from crawler_task where task_id={task_id} '
    task_info = mysql_con.get_values(sql)

    return jsonify({'code': 200, 'data': task_info})


# 只接受get方法访问
@app.route("/v1/crawler/source/getall", methods=["GET"])
def getSource():
    try:
        # # 对参数进行操作
        sql = 'select * from crawler_source'
        result = mysql_con.get_values(sql)
        if not result:
            return jsonify({'code': '200', 'result': [], 'message': '没有更多数据'})
    except Exception as e:
        return jsonify({'code': '400', 'message': '获取数据源信息失败'})

    return jsonify({'code': '200', 'result': result})


@app.route("/v1/crawler/source/getasktype", methods=["GET"])
def getTaskType():
    try:
        data = request.args.to_dict()
        source = data['source']
        # # 对参数进行操作
        sql = f'select * from crawler_task_type where source="{source}"'
        result = mysql_con.get_values(sql)
        if not result:
            return jsonify({'code': '200', 'result': [], 'message': '没有更多数据'})
        else:
            task_type_list = list()
            for task_type_info in result:
                task_info = {
                    'type': task_type_info['task_type'],
                    'description': task_type_info['task_type_desc'],
                    'spider': {
                        'spider_name': task_type_info['spider_name'],
                        'description': task_type_info['spider_name_desc']
                    }
                }

                task_type_list.append(task_info)
            source_dict = {
                'task_type': task_type_list,
            }
    except Exception as e:
        return jsonify({'code': '400', 'message': '获取数据源信息失败'})

    return jsonify({'code': '200', 'result': source_dict})


@app.route("/v1/crawler/task/checkrepeat", methods=["POST"])
def get_repeat_list():
    data = request.json
    # 字段转换
    spider_links = data.get('spider_link')
    repeat_list = list()
    # 判断是否为重复名单
    for spider_link in spider_links:
        if isinstance(spider_link, int):
            s_sql = f"""select spider_link from crawler_author_map where spider_link={spider_link}"""
        else:
            s_sql = f"""select spider_link from crawler_author_map where spider_link='{spider_link}'"""
        result = mysql_con.get_values(s_sql)
        if result:
            repeat_list.append(spider_link)
    if repeat_list:
        return jsonify({'code': 400, 'message': '名单重复', 'repeat_list': repeat_list})
    else:

        return jsonify({'code': 200, 'message': '抓取名单校验通过', 'repeat_list': repeat_list})


@app.route("/v1/crawler/task/insert", methods=["POST"])
def insertTask():
    try:
        data = request.json
        user_data = copy.deepcopy(data)

        tag_name_list = []
        content_tag_list = []
        user_tag = data['user_tag']
        user_content_tag = data['user_content_tag']

        for tag in user_tag:
            tag_name_list.append(tag['tagName'])
        for tag in user_content_tag:
            content_tag_list.append(tag['tagName'])
        if data['min_publish_time']:
            data['min_publish_time'] = int(data['min_publish_time'] / 1000)
        else:
            data['min_publish_time'] = 0
        if not data['min_publish_day']:
            data['min_publish_day'] = 0
        data['next_time'] = int(data['next_time'] / 1000)
        data['insert_time'] = int(time.time())
        data['update_time'] = int(time.time())
        data['spider_link'] = str(data['spider_link'])
        data['spider_rule'] = str(data['spider_rule'])

        data['user_tag_info'] = str(user_tag)
        data['content_tag_info'] = str(user_content_tag)

        data['user_tag'] = ','.join(str(i) for i in tag_name_list)
        data['user_content_tag'] = ','.join(str(i) for i in content_tag_list)
        # data['crawler_interval'] = data.pop('interval')
        # 获取到一个以键且为逗号分隔的字符串，返回一个字符串
        keys = ','.join(data.keys())
        values = ','.join(['%s'] * len(data))
        sql = 'insert into {table}({keys}) VALUES({values})'.format(table='crawler_task', keys=keys, values=values)
        task_id = mysql_con.insert_values(sql, tuple(data.values()))
        if task_id:
            spider_link = user_data['spider_link']
            success_list, fail_list = create_uid(user_data, task_id, spider_link)
            return jsonify(
                {'code': 200, 'message': 'task create success', 'success_list': success_list, 'fail_list': fail_list})
    except Exception as e:
        return jsonify({'code': 500, 'message': '任务写入失败,原因:{e}'})


@app.route("/v1/crawler/task/gettask", methods=["POST"])
def getAllTask():
    try:
        get_data = request.json
        page = int(get_data.get('page', 1))
        offset = int(get_data.get('offset', 10))
        start_count = (page * offset) - offset
        end_count = page * offset
        if get_data.get('fields'):
            select_data = get_data['fields']
            values = ''
            for k, v in select_data.items():
                if isinstance(v, int):
                    values += f'{k}={v} and '
                else:
                    values += f'{k}="{v}" and '
            sql = f"select task_id from crawler_author_map where {values[:-4]} and is_del=1"  # [:-1]是为了去掉末尾的逗号
            res = mysql_con.get_values(sql)
            task_id_set = set()
            for task in res:
                task_id_set.add(task['task_id'])
            task_list = list()
            for task_id in task_id_set:
                sql = f'select * from crawler_task where task_id={task_id} order by update_time desc limit {start_count}, {end_count}'
                task_info = mysql_con.get_values(sql)[0]
                task_data = dict(
                    task_id=task_info['task_id'],
                    task_name=task_info['task_name'],
                    source=task_info['source'],
                    task_type=task_info['task_type'],
                    create_task_user=task_info['create_task_user'],
                    insert_time=task_info['insert_time'] * 1000,
                    update_task_user=task_info['update_task_user'],
                    update_time=task_info['update_time'] * 1000
                )
                task_list.append(task_data)
            return jsonify({'code': 200, 'result': task_list, 'total': len(task_list)})

        sql = f"""select * from crawler_task order by update_time desc limit {start_count}, {end_count} """
        result = mysql_con.get_values(sql)
        if not result:
            return jsonify({'code': '200', 'result': [], 'message': '没有更多任务'})
        task_list = list()
        for task_info in result:
            source = task_info['source']
            task_type = task_info['task_type']
            source_sql = f'select * from crawler_source where source="{source}"'
            source_info = mysql_con.get_values(source_sql)
            task_type_sql = f'select * from crawler_task_type where task_type="{task_type}"'
            type_info = mysql_con.get_values(task_type_sql)
            task_data = dict(
                task_id=task_info['task_id'],
                task_name=task_info['task_name'],
                source_name=source_info[0]['source_desc'],
                task_type_name=type_info[0]['task_type_desc'],
                source=task_info['source'],
                task_type=task_info['task_type'],
                create_task_user=task_info['create_task_user'],
                insert_time=task_info['insert_time'] * 1000,
                update_task_user=task_info['update_task_user'],
                update_time=task_info['update_time'] * 1000
            )
            task_list.append(task_data)

        t_sql = f"""select count(*) from crawler_task"""
        t_res = mysql_con.get_values(t_sql)
        total = t_res[0]['count(*)']
    except Exception as e:
        return jsonify({"code": "400", 'message': "任务列表获取失败"})

    return jsonify({'code': '200', 'result': task_list, 'total': total})


@app.route("/v1/crawler/task/getone", methods=["GET"])
def getOneTask():
    try:
        get_data = request.args.to_dict()
        task_id = get_data['task_id']
        sql = f'select * from crawler_task where task_id={task_id}'
        result = mysql_con.get_values(sql)

        if not result:
            return jsonify({'code': '400', 'result': [], 'message': 'no data'})

        data = result[0]
        if data['min_publish_time']:
            data['min_publish_time'] = data['min_publish_time'] * 1000
        else:
            data['min_publish_time'] = 0

        data['next_time'] = data['next_time'] * 1000
        data['spider_link'] = eval(data['spider_link'])
        data['spider_rule'] = eval(data['spider_rule'])
        #
        data['user_tag_info'] = eval(data['user_tag_info'])
        data['content_tag_info'] = eval(data['content_tag_info'])

        if not data['mode_name_id']:
            data['mode_name_id'] = ''
        if not data['mode_board_id']:
            data['mode_board_id'] = ''
        if not data['content_category_id']:
            data['content_category_id'] = ''

    except Exception as e:
        return jsonify({'code': '500', "message": "获取任务信息失败"})

    return jsonify({'code': '200', 'result': result})


@app.route("/v1/crawler/task/update", methods=["POST"])
def updateTask():
    try:
        data = request.json
        task_id = data.get('task_id')
        task_info = data.get('task_info')
        values = ''
        if task_info['min_publish_time']:
            task_info['min_publish_time'] = task_info['min_publish_time'] / 1000
        else:
            task_info['min_publish_time'] = 0
        if not task_info['min_publish_day']:
            task_info['min_publish_day'] = 0
        task_info['next_time'] = task_info['next_time'] / 1000

        user_tag = task_info['user_tag']
        user_content_tag = task_info['user_content_tag']
        tag_name_list = []
        content_tag_list = []
        for tag in user_tag:
            tag_name_list.append(tag['tagName'])
        for tag in user_content_tag:
            content_tag_list.append(tag['tagName'])
        task_info['user_tag_info'] = str(user_tag)
        task_info['content_tag_info'] = str(user_content_tag)

        task_info['user_tag'] = ','.join(str(i) for i in tag_name_list)
        task_info['user_content_tag'] = ','.join(str(i) for i in content_tag_list)

        for k, v in task_info.items():
            if isinstance(v, int):
                values += f'{k}={v},'
            else:
                values += f'{k}="{v}",'

        sql = f'update crawler_task set {values[:-1]} where task_id={task_id}'
        result = mysql_con.update_values(sql)
        if result:
            return jsonify({'code': 200, 'message': 'task update success'})
        else:
            return jsonify({'code': 400, 'message': 'task update faild'})
    except Exception as e:

        return jsonify({'code': 400, 'message': '任务更新失败'})


def create_uid(task, task_id, spider_link):
    if not isinstance(spider_link, list):
        spider_link = eval(spider_link)
    source = task.get('source')
    task_type = task.get('task_type')
    applets_status = task.get('applets_status')
    app_status = task.get('app_status')
    try:
        user_tag = eval(task.get('user_tag_info'))
        user_content_tag = eval(task.get('content_tag_info'))
    except Exception as e:
        user_tag = task.get('user_tag')
        user_content_tag = task.get('user_content_tag')

    mode_name_id = task.get('mode_name_id', 0)
    mode_board_id = task.get('mode_board_id', 0)
    content_category_id = task.get('content_category_id', 0)

    mn_sql = f'select * from crawler_mode where id={mode_name_id}'
    mode_name_list = mysql_con.get_values(mn_sql)
    mb_sql = f'select * from crawler_board where id={mode_board_id}'
    mode_board_list = mysql_con.get_values(mb_sql)
    cc_sql = f'select * from crawler_content_category where id={content_category_id}'
    content_category_list = mysql_con.get_values(cc_sql)

    source_sql = f'select * from crawler_source where source="{source}"'
    source_res = mysql_con.get_values(source_sql)[0]
    spider_platform = source_res['source_desc']

    if mode_name_list:
        task['mode_name_str'] = mode_name_list[0]['mode_name']
    else:
        task['mode_name_str'] = ''
    if mode_board_list:

        task['mode_board_str'] = mode_board_list[0]['mode_board']
    else:
        task['mode_board_str'] = ''
    if content_category_list:

        task['content_category_str'] = content_category_list[0]['content_category']
    else:
        task['content_category_str'] = ''

    success_list = list()
    fail_list = list()

    tag_name_list = list()
    content_tag_list = list()

    for tag in user_tag:
        tag_name_list.append(tag['tagName'])
    for tag in user_content_tag:
        content_tag_list.append(tag['tagName'])
    user_tags = ','.join(str(i) for i in tag_name_list)
    user_content_tags = ','.join(str(i) for i in content_tag_list)

    for author_url in spider_link:
        now_time = int(time.time())
        time_array = time.localtime(now_time)
        str_time = time.strftime("%Y%m%d", time_array)
        # 生成创建用户的tag

        tags = ''
        if task['task_type'] == 'author':
            spider_task = '账号'
            tags_list = ['spider', spider_task, spider_platform, user_tags, task['content_category_str'], str_time]

        elif task['task_type'] == 'search':
            spider_task = '搜索'
            tags_list = ['spider', spider_task, spider_platform, user_tags, author_url, task['content_category_str'], str_time]
        elif task['task_type'] == 'board':
            spider_task = '榜单'
            mode_tags = task['mode_board_str']

            tags_list = ['spider', spider_task, spider_platform, user_tags, mode_tags, task['content_category_str'], str_time]
        elif task['task_type'] == 'recommend':
            spider_task = '推荐'
            mode_tags = task['mode_name_str'] + task['mode_board_str']
            tags_list = ['spider', spider_task, spider_platform, user_tags, mode_tags, task['content_category_str'], str_time]
        else:
            tags_list = ['spider', spider_platform, user_tags, task['content_category_str'], str_time]
        for v in tags_list:
            if v:
                tags += str(v) + ','
        post_data = {
            # 'count': 1,     # （必须）账号个数：传1
            # 'accountType': 4,   # （必须）账号类型 ：传 4 app虚拟账号
            'pwd': '',  # 密码 默认 12346
            'nickName': '',  # 昵称  默认 vuser......
            'avatarUrl': '',
            # 头像Url  默认 http://weapppiccdn.yishihui.com/resources/images/pic_normal.png
            'tagName': tags[:-1],  # 多条数据用英文逗号分割
        }
        try:
            response = requests.post(url=conf['media_url'], params=post_data)
            media_id = response.json()['data']
            media_info = requests.get(url=conf['select_media_url'], params={'uid': media_id}).json()['content']
        except Exception as e:
            logging.warning(f'创建账户:{spider_link},失败，原因:{e}')
            fail_list.append(author_url)
            continue

        data = dict(
            spider_link=author_url,
            media_id=media_id,
            media_name=media_info['longvideoNickName'] if media_info['longvideoNickName'] else media_info['nickName'],
            source=source,
            task_type=task_type,
            applets_status=applets_status,
            app_status=app_status,
            user_tag=user_tags,
            user_content_tag=user_content_tags,
            insert_time=int(time.time()),
            update_time=int(time.time()),
            create_user_time=now_time,
            mode_name_str=task['mode_name_str'],
            mode_board_str=task['mode_board_str'],
            content_category_str=task['content_category_str'],
            # mode_value_str=mode_value_str,
            task_id=task_id,
            media_main_url=conf['media_main_url'].format(media_id)
        )
        keys = ','.join(data.keys())
        values = ','.join(['%s'] * len(data))
        table = 'crawler_author_map'
        sql = f"""insert into {table}({keys}) VALUES({values})"""
        mysql_con.insert_values(sql, tuple(data.values()))
        uer_info = dict(
            outer_id=author_url,
            uid=media_id
        )
        success_list.append(uer_info)

    return success_list, fail_list


@app.route("/v1/crawler/author/create", methods=["POST"])
def createUser():
    spider_link = request.json.get('spider_link')
    source = request.json.get('source')
    task_type = request.json.get('task_type')
    applets_status = request.json.get('applets_status')
    app_status = request.json.get('app_status')
    user_tag = request.json.get('user_tag')
    user_content_tag = request.json.get('user_content_tag')
    success_list = list()
    fail_list = list()
    for author_url in spider_link:
        try:
            f_sql = f"""select spider_link from crawler_author_map where spider_link="{author_url}" """
            result = mysql_con.get_values(f_sql)
            if result:
                success_list.append(author_url)
                continue
            else:
                tag_name_list = []
                content_tag_list = []
                for tag in user_tag:
                    tag_name_list.append(tag['tagName'])
                for tag in user_content_tag:
                    content_tag_list.append(tag['tagName'])
                user_tags = ','.join(str(i) for i in tag_name_list)
                user_content_tags = ','.join(str(i) for i in content_tag_list)
                post_data = {
                    # 'count': 1,     # （必须）账号个数：传1
                    # 'accountType': 4,   # （必须）账号类型 ：传 4 app虚拟账号
                    'pwd': '',  # 密码 默认 12346
                    'nickName': '',  # 昵称  默认 vuser......
                    'avatarUrl': '',
                    # 头像Url  默认 http://weapppiccdn.yishihui.com/resources/images/pic_normal.png
                    'tagName': user_tags,  # 多条数据用英文逗号分割
                }
                response = requests.post(url=conf['media_url'], params=post_data)
                media_id = response.json()['data']

                data = dict(
                    spider_link=author_url,
                    media_id=media_id,
                    source=source,
                    task_type=task_type,
                    applets_status=applets_status,
                    app_status=app_status,
                    user_tag=user_tags,
                    user_content_tag=user_content_tags,
                    insert_time=int(time.time()),
                    update_time=int(time.time())
                )
                keys = ','.join(data.keys())
                values = ','.join(['%s'] * len(data))
                table = 'crawler_author_map'
                sql = f"""insert into {table}({keys}) VALUES({values})"""
                result = mysql_con.insert_values(sql, tuple(data.values()))
                if not result:
                    fail_list.append(author_url)
                else:
                    success_list.append(author_url)
        except Exception as e:
            fail_list.append(author_url)
            continue
    return jsonify({'code': 200, 'result': {'success': success_list, 'fail': fail_list}})


if __name__ == "__main__":
    app.run(debug=True, port=5050)