socket可以实时推送最新内容到群里，不用api返回数据

用了群机器人接口来接收文本信息，可以不用api,用socket或者虚拟机来推送和接收消息

网页解析不好定位，标签属性一直在变，可以socket来接收网页内容发送到群里，完整代码

import re
import sqlite3
import os
import hmac
import requests
import time
import hashlib
from urllib.parse import quote_plus
import base64
import json


class Connect:
    def __init__(self, dbpath):
        self.dbpath = dbpath
        self.connect = sqlite3.connect(self.dbpath)

    def init_db(self):
        sql = '''
            create table if not exists 'notice_info'
            (
                t text,
                content text,
                href text
            )
        '''
        # self.dbpath = dbpath
        # self.connect = sqlite3.connect(self.dbpath)  # 打开或创建数据库

        c = self.connect.cursor()  # 获取游标
        c.execute(sql)  # 执行SQL语句
        self.connect.commit()  # 提交事务
        # self.connect.close()  # 关闭

    def dataExtarct(self):
        try:
            req = requests.get('https://www.jiucaigongshe.com/study_publish')
            t = re.findall('<div class="fs13-ash" data-v-0d6421ce>(.*?)</div>', req.text)[0]
            href = 'https://www.jiucaigongshe.com' + re.findall(
                '<div class="html-text h90 htmlHile" data-v-0d6421ce><a href="(.*?)" target="_blank" onclick="return false;" data-v-0d6421ce>',
                req.text)[0]

            req_href = requests.get(href)
            title = re.findall('<title>(.*?)</title>',req_href.text)[0]
            content = {
                "t": t,
                "title": title,
                "href": href
            }
            return content

        except:
            pass
    def insert_data(self):
        try:
            content = self.dataExtarct()
            sql = "insert into notice_info(t, content, href) values('" + content["t"] + "', '" + content["title"] + "', '" + \
                  content["href"] + "');"
            c = self.connect.cursor()
            c.execute(sql)
            self.connect.commit()  # 提交事务
            # self.connect.close()  # 关闭
            self.save_data()
        except:
            pass
    def save_data(self):
        try:
            content = self.dataExtarct()
            sql = "select * from notice_info where t = '" + content["t"] + "' and  content = '" + content["title"] + "' and  href= '" + content["href"] + "';"
            # print(sql)
            self.connect = sqlite3.connect(self.dbpath)  # 打开或创建数据库
            c = self.connect.cursor()  # 获取游标
            cursor = c.execute(sql)  # 执行SQL语句
            result = 0
            for row in cursor:
                print(row)
                result += 1
            if result == 0:
                self.insert_data()
                # 如果数据库中没有记录的话，那么就发送这条消息
                m = Messenger(
                    token="f73f1ea7720c877cb1a029178f6e498ba7bdbe7d535c4cda8025561f1d277652",
                    secret="SEC7b6711f3c1d685e0e975d8d0846a141bb62ef9d70639a8b24938692450d9063c"
                )
                m.send_text(content["t"]+'----'+content["title"]+'----'+content["href"])

                return 1
            else:
                return 0
        except:
            pass

class Messenger:
    def __init__(self, token=os.getenv("DD_ACCESS_TOKEN"), secret=os.getenv("DD_SECRET")):
        self.timestamp = str(round(time.time() * 1000))
        self.URL = "https://oapi.dingtalk.com/robot/send"
        self.headers = {'Content-Type': 'application/json'}
        secret = secret
        secret_enc = secret.encode('utf-8')
        string_to_sign = '{}\n{}'.format(self.timestamp, secret)
        string_to_sign_enc = string_to_sign.encode('utf-8')
        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
        self.sign = quote_plus(base64.b64encode(hmac_code))
        self.params = {'access_token': token, "sign": self.sign}

    def send_text(self, content):
        """
        发送文本
        @param content: str, 文本内容
        """
        data = {"msgtype": "text", "text": {"content": content}}
        self.params["timestamp"] = self.timestamp
        return requests.post(
            url=self.URL,
            data=json.dumps(data),
            params=self.params,
            headers=self.headers
        )

while True:
    connect = Connect('mydata.sqlite')
    connect.init_db()
    connect.save_data()
    time.sleep(5)

dataExtarct()方法错了
第38-43行

完整代码:

import re
import sqlite3
import os
import hmac
import requests
import time
import hashlib
from urllib.parse import quote_plus
import base64
import json


class Connect:
    def __init__(self, dbpath):
        self.dbpath = dbpath
        self.connect = sqlite3.connect(self.dbpath)

    def init_db(self):
        sql = '''
            create table if not exists 'notice_info'
            (
                t text,
                content text,
                href text
            )
        '''
        # self.dbpath = dbpath
        # self.connect = sqlite3.connect(self.dbpath)  # 打开或创建数据库

        c = self.connect.cursor()  # 获取游标
        c.execute(sql)  # 执行SQL语句
        self.connect.commit()  # 提交事务
        # self.connect.close()  # 关闭

    def dataExtarct(self):
        try:
            req = requests.get('https://www.jiucaigongshe.com/study_publish')
            t = re.findall('(. *?)', req.text)[0]
            href = 'https://www.jiucaigongshe.com' + re.findall(
                '',
            req.text)[0]
            req_href = requests.get(href)
            title = re.findall('<title>(.*?)</title>', req_href.text)[0]
            content = {
                "t": t,
                "title": title,
                "href": href
            }
            return content

        except:
            pass

    def insert_data(self):
        try:
            content = self.dataExtarct()
            sql = "insert into notice_info(t, content, href) values('" + content["t"] + "', '" + content[
                "title"] + "', '" + \
                  content["href"] + "');"
            c = self.connect.cursor()
            c.execute(sql)
            self.connect.commit()  # 提交事务
            # self.connect.close()  # 关闭
            self.save_data()
        except:
            pass

    def save_data(self):
        try:
            content = self.dataExtarct()
            sql = "select * from notice_info where t = '" + content["t"] + "' and  content = '" + content[
                "title"] + "' and  href= '" + content["href"] + "';"
            # print(sql)
            self.connect = sqlite3.connect(self.dbpath)  # 打开或创建数据库
            c = self.connect.cursor()  # 获取游标
            cursor = c.execute(sql)  # 执行SQL语句
            result = 0
            for row in cursor:
                print(row)
                result += 1
            if result == 0:
                self.insert_data()
                # 如果数据库中没有记录的话，那么就发送这条消息
                m = Messenger(
                    token="f73f1ea7720c877cb1a029178f6e498ba7bdbe7d535c4cda8025561f1d277652",
                    secret="SEC7b6711f3c1d685e0e975d8d0846a141bb62ef9d70639a8b24938692450d9063c"
                )
                m.send_text(content["t"] + '----' + content["title"] + '----' + content["href"])

                return 1
            else:
                return 0
        except:
            pass


class Messenger:
    def __init__(self, token=os.getenv("DD_ACCESS_TOKEN"), secret=os.getenv("DD_SECRET")):
        self.timestamp = str(round(time.time() * 1000))
        self.URL = "https://oapi.dingtalk.com/robot/send"
        self.headers = {'Content-Type': 'application/json'}
        secret = secret
        secret_enc = secret.encode('utf-8')
        string_to_sign = '{}\n{}'.format(self.timestamp, secret)
        string_to_sign_enc = string_to_sign.encode('utf-8')
        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
        self.sign = quote_plus(base64.b64encode(hmac_code))
        self.params = {'access_token': token, "sign": self.sign}

    def send_text(self, content):
        """
        发送文本
        @param content: str, 文本内容
        """
        data = {"msgtype": "text", "text": {"content": content}}
        self.params["timestamp"] = self.timestamp
        return requests.post(
            url=self.URL,
            data=json.dumps(data),
            params=self.params,
            headers=self.headers
        )


while True:
    connect = Connect('mydata.sqlite')
    connect.init_db()
    connect.save_data()
    time.sleep(5)