30.01.2025 - ЖЗГ

Парсинг новостей групп VK

30.01.2025 Павел Грибов Python, Из жизни

В продолжение предыдущей статьи, появилась необходимость парсить так-же и новости в социальной сети vk с проверкой на наличие стоп-слов. Для этого воспользовался модулем vk на python. Так-же понадобится токен доступа полученный на https://vk.com/apps?act=manage

В итоге код получился примерно следующий:

#!/usr/bin/env python3
# encoding: utf-8
import vk
import json
import funcs

with open('config.json', 'r') as file:
    config_data = json.load(file)
    print(config_data)


api = vk.API(access_token=config_data["vk_api_token"],v='5.131')

for group in config_data["groups"]:
    chan_data = funcs.get_chan_json(group)
    skeep_after = chan_data["las_id"]
    wall_content = api.wall.get(domain=group, count=config_data["limit"])
    poz = 0
    for message in wall_content["items"]:
        #print(message)
        if poz == 0:
            chan_data["las_id"] = message["id"]
            funcs.save_chan_json(group, chan_data)
        if skeep_after == message["id"]:
            print("Все новости уже прочитаны...")
            break
        for word in config_data["alert_words"]:
          if word in message["text"]:
              print(f"--нашли слово {word}")
              funcs.SendMailVK(config_data,group, word, message)
              print(message)
        poz = poz + 1
print("all done..");

#!/usr/bin/env python3

# encoding: utf-8

import vk

import json

import funcs

with open('config.json', 'r') as file:

config_data = json.load(file)

print(config_data)

api = vk.API(access_token=config_data["vk_api_token"],v='5.131')

for group in config_data["groups"]:

chan_data = funcs.get_chan_json(group)

skeep_after = chan_data["las_id"]

wall_content = api.wall.get(domain=group, count=config_data["limit"])

poz = 0

for message in wall_content["items"]:

#print(message)

if poz == 0:

chan_data["las_id"] = message["id"]

funcs.save_chan_json(group, chan_data)

if skeep_after == message["id"]:

print("Все новости уже прочитаны...")

break

for word in config_data["alert_words"]:

if word in message["text"]:

print(f"--нашли слово {word}")

funcs.SendMailVK(config_data,group, word, message)

print(message)

poz = poz + 1

print("all done..");

По сути код очень простой — получаем через API VK все последние новости из каждой группы. Если в тексте новости находим стоп-слово, то отправляем соответствующее письмо. Так-же использую дополнительный файл функций, которые далее использую во всех парсерах:

#!/usr/bin/env python3
#encoding: utf-8
import json
from datetime import datetime
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from telethon.tl.functions.contacts import ResolveUsernameRequest
from telethon.tl.functions.channels import GetMessagesRequest
from telethon.tl.functions.messages import GetHistoryRequest, ReadHistoryRequest

def save_chan_json(chan,chan_data):
    f = open('saves/'+chan + '.json', "w+")
    json.dump(chan_data, f)
    f.close()

def get_chan_json(chan):
    # узнаём какое последнее сообщение прочитали на канале?
    chan_data = {}
    chan_data["las_id"] = 0
    try:
        with open('saves/'+chan + '.json', 'r') as file:
            chan_data = json.load(file)
            print(chan_data)
            return chan_data
    except:
        save_chan_json(chan, chan_data)
        return chan_data
    return chan_data;


def SendMail(config_data,chan,word,message):
    msg = MIMEMultipart()
    msg['Subject'] = f"Найдено слово '{word}' в новости на канале {chan} в Телеграм"
    msg.add_header('Content-Type', 'text/html')
    message.text=message.text.replace(word,"<strong>"+word+"</strong>")
    dt_pub=message.date.strftime('%d-%m-%Y %H:%M:%S')
    msg.set_payload(f"Канал: <a href='https://vc.com/{chan}'>https://t.me/{chan}</a>, опубликовано {dt_pub}<hr/>"+message.text)

    smtpObj = smtplib.SMTP(config_data["smtp_server"], config_data["smtp_port"])
    smtpObj.starttls()
    smtpObj.login(config_data["email_login"], config_data["from_password"])
    smtpObj.sendmail(config_data["email_from"], config_data["notify_email"], msg.as_string().encode('utf-8'))
    smtpObj.quit()

def SendMailVK(config_data,chan,word,message):
    msg = MIMEMultipart()
    msg['Subject'] = f"Найдено слово '{word}' в новости на группе {chan} в VK"
    msg.add_header('Content-Type', 'text/html')
    message["text"]=message["text"].replace(word,"<strong>"+word+"</strong>")
    dt_pub=datetime.utcfromtimestamp(message["date"]).strftime('%d-%m-%Y %H:%M:%S')
    msg.set_payload(f"Группа: <a href='https://vc.com/{chan}'>https://vk.com/{chan}</a>, опубликовано {dt_pub}<hr/>"+message["text"])

    smtpObj = smtplib.SMTP(config_data["smtp_server"], config_data["smtp_port"])
    smtpObj.starttls()
    smtpObj.login(config_data["email_login"], config_data["from_password"])
    smtpObj.sendmail(config_data["email_from"], config_data["notify_email"], msg.as_string().encode('utf-8'))
    smtpObj.quit()

def SendMailNews(config_data,url,word,message):
    msg = MIMEMultipart()
    msg['Subject'] = f"Найдено слово '{word}' в новости на сайте {url}"
    msg.add_header('Content-Type', 'text/html')
    if message.get("href")!=None:
     message.string="<a href='"+message["href"]+"'>"+message.string.replace(word,"<strong>"+word+"</strong>")+"</a>"
    else:
        message.string = message.string.replace(word,"<strong>" + word + "</strong>")
    msg.set_payload(message.string)

    smtpObj = smtplib.SMTP(config_data["smtp_server"], config_data["smtp_port"])
    smtpObj.starttls()
    smtpObj.login(config_data["email_login"], config_data["from_password"])
    smtpObj.sendmail(config_data["email_from"], config_data["notify_email"], msg.as_string().encode('utf-8'))
    smtpObj.quit()

#!/usr/bin/env python3

#encoding: utf-8

import json

from datetime import datetime

import smtplib

from email.mime.multipart import MIMEMultipart

from email.mime.text import MIMEText

from telethon.tl.functions.contacts import ResolveUsernameRequest

from telethon.tl.functions.channels import GetMessagesRequest

from telethon.tl.functions.messages import GetHistoryRequest, ReadHistoryRequest

def save_chan_json(chan,chan_data):

f = open('saves/'+chan + '.json', "w+")

json.dump(chan_data, f)

f.close()

def get_chan_json(chan):

# узнаём какое последнее сообщение прочитали на канале?

chan_data = {}

chan_data["las_id"] = 0

try:

with open('saves/'+chan + '.json', 'r') as file:

chan_data = json.load(file)

print(chan_data)

return chan_data

except:

save_chan_json(chan, chan_data)

return chan_data

return chan_data;

def SendMail(config_data,chan,word,message):

msg = MIMEMultipart()

msg['Subject'] = f"Найдено слово '{word}' в новости на канале {chan} в Телеграм"

msg.add_header('Content-Type', 'text/html')

message.text=message.text.replace(word,""+word+"")

dt_pub=message.date.strftime('%d-%m-%Y %H:%M:%S')

msg.set_payload(f"Канал: <a href='https://vc.com/{chan}'>https://t.me/{chan}</a>, опубликовано {dt_pub}<hr/>"+message.text)

smtpObj = smtplib.SMTP(config_data["smtp_server"], config_data["smtp_port"])

smtpObj.starttls()

smtpObj.login(config_data["email_login"], config_data["from_password"])

smtpObj.sendmail(config_data["email_from"], config_data["notify_email"], msg.as_string().encode('utf-8'))

smtpObj.quit()

def SendMailVK(config_data,chan,word,message):

msg = MIMEMultipart()

msg['Subject'] = f"Найдено слово '{word}' в новости на группе {chan} в VK"

msg.add_header('Content-Type', 'text/html')

message["text"]=message["text"].replace(word,""+word+"")

dt_pub=datetime.utcfromtimestamp(message["date"]).strftime('%d-%m-%Y %H:%M:%S')

msg.set_payload(f"Группа: <a href='https://vc.com/{chan}'>https://vk.com/{chan}</a>, опубликовано {dt_pub}<hr/>"+message["text"])

smtpObj = smtplib.SMTP(config_data["smtp_server"], config_data["smtp_port"])

smtpObj.starttls()

smtpObj.login(config_data["email_login"], config_data["from_password"])

smtpObj.sendmail(config_data["email_from"], config_data["notify_email"], msg.as_string().encode('utf-8'))

smtpObj.quit()

def SendMailNews(config_data,url,word,message):

msg = MIMEMultipart()

msg['Subject'] = f"Найдено слово '{word}' в новости на сайте {url}"

msg.add_header('Content-Type', 'text/html')

if message.get("href")!=None:

message.string="<a href='"+message["href"]+"'>"+message.string.replace(word,""+word+"")+"</a>"

else:

message.string = message.string.replace(word,"" + word + "")

msg.set_payload(message.string)

smtpObj = smtplib.SMTP(config_data["smtp_server"], config_data["smtp_port"])

smtpObj.starttls()

smtpObj.login(config_data["email_login"], config_data["from_password"])

smtpObj.sendmail(config_data["email_from"], config_data["notify_email"], msg.as_string().encode('utf-8'))

smtpObj.quit()

Оставить комментарий python, vk, парсинг новостей групп

Пн	Вт	Ср	Чт	Пт	Сб	Вс
		1	2	3	4	5
6	7	8	9	10	11	12
13	14	15	16	17	18	19
20	21	22	23	24	25	26
27	28	29	30	31