VideoToArticles/test/mysql.py

import json

import pymysql
from task.wx_search_task import wxGenerateTask

task_id = "Article_3e99f0ca-960f-4838-a109-1fcb968b6789_1720677206"
article_url = "https://mp.weixin.qq.com/s/S7-0-1eL4xo2qdDNmMSHBQ"


wgt = wxGenerateTask(mysql_client="client")
text = wgt.get_article_text(article_url)
img_list = text['data']['data']['image_url_list']
img_list = [i['image_url'] for i in img_list]
cover = img_list[0]
title = text['data']['data']['title']
res = {
    "text": text['data']['data']['body_text'],
    "title": title,
    "cover": cover,
    "img_list": img_list
}

temp = res
connection = pymysql.connect(
            host='rm-bp1159bu17li9hi94.mysql.rds.aliyuncs.com',
            port=3306,
            user='crawler',
            password='crawler123456@',
            db='piaoquan-crawler',
            charset='utf8mb4'
)

text = temp['text'].replace("\n", "")
title = temp['title']
cover = temp['cover']
img_list = temp['img_list']

sql = update_sql = f"""
UPDATE video_to_articles
SET ai_text = %s, ai_title = %s, cover = %s, img_list = %s, status_code = %s
WHERE task_id = %s;"""

cursor = connection.cursor()
cursor.execute(sql, (
    text,
    title,
    cover,
    json.dumps(img_list, ensure_ascii=False),
    3,
    task_id
))
connection.commit()