И он хорошо срабатывает при периодическом запуске например из крона скрипта с какойто периодичностью, для проверки «а закончилась ли работа предыдущего запуска?». Но дело в том, что иногда пишут скрипт которые работает с разным функционалом в зависимости от параметров запуска. Тогда в этом случае вполне допустим запуск скрипта, но с другим параметром. Для того чтобы обойти этот момент, я стал использовать несколько другой способ. А именно: опрашиваю список запущенных в текущий момент процессов и смотрю параметры их запуска. Если нахожу совпадение, то выхожу. Если нет — позволяю скрипту работать дальше. Вышло примерно так:
import psutil
def AYouRun(script_param):
pids=psutil.pids()
cnt=0
for pid in pids:
p = psutil.Process(pid)
res=p.cmdline()
if script_param in res:
cnt=cnt+1
if cnt>1:
print(f"--скрипт с параметром {script_param} уже запущен")
exit(-1)
return True
..
..
if __name__ == '__main__':
for param in sys.argv:
if param == "--telegram-news":
if AYouRun(param):
Insert2Log("Запущен мониторинг телеграм новостей",1)
client = TelegramClient("parser_data", global_config["telegram_api_id"], global_config["telegram_api_hash"])
..
..
В продолжение предыдущей статьи, появилась необходимость парсить так-же и новости в социальной сети 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..");
По сути код очень простой — получаем через 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()
Задача: необходимо просматривать несколько новостных телеграм каналов, и в случае обнаружения в новости неких стоп-слов, высылать уведомление на электронную почту.
К сожалению воспользоваться для решения этой задачи API для работы с ботами не получится, т.к. такого функционала просто нет. Выходом может служить — воспользоваться одним из многочисленных клиентов Телеграм, реализованых на PHP, Python, JavaScript (NodeJS) и т.д. В моём случае — воспользуюсь python и библиотекой telethon. К ней довольно толковая документация, в том числе и на русском
Итак, для начала нужно зайти на ресурс https://my.telegram.org/apps и получить api_id и api_hash, для того чтобы библиотека смогла создать соединение. Далее создам файл с настройками вида:
Итак, продолжаем продолжаем писать квест в консоли на языке Python. Первая часть описана здесь. В ней мы реализовали автоматическую загрузку и сохранение состояния прохождения квеста. Сейчас же займемся (начнем по крайне мере) отрисовкой локации, и реакцией на нажатые кнопки. В локации предусмотрим возможность отображения картинки из ASCII. Например json стартовой локации может выглядеть примерно так:
Создадим класс TLocation, при инициалиизации будем передавать в него инициализированный класс player. В переменной класса data — будем хранить загруженную локацию.
class TLocation:
data = {}
player={}
stdscr=curses.initscr()
scr_size=stdscr.getmaxyx()
def __init__(self,player):
self.player=player;
def load_location(self, location):
f = open("locations/" + str(location) + ".json", mode='r', encoding='utf-8')
self.data = json.load(f)
f.close()
Далее нарисуем верхнее меню, где сообщаем игроку, на какие локации он может перемещаться, и что он держит в руках:
def top_menu(self):
loc="Идти: "
if "left" in self.data["available_locations"]:
loc=loc+"влево(4) "
if "right" in self.data["available_locations"]:
loc=loc+"вправо(6) "
if "forward" in self.data["available_locations"]:
loc=loc+"вперед(8) "
if "back" in self.data["available_locations"]:
loc=loc+"назад(2) "
curses.start_color()
curses.init_pair(1, curses.COLOR_RED, curses.COLOR_BLACK)
self.stdscr.addstr(1, 1, loc,curses.color_pair(1))
curses.init_pair(1, curses.COLOR_RED, curses.COLOR_BLACK)
self.stdscr.addstr(1, 1, loc,curses.color_pair(1))
# что в руках
hands = "В руках: "
self.stdscr.addstr(2, 1, "")
for object in self.player.data["in_hands"]:
hands=hands+"["+object+"]"
if len(self.player.data["in_hands"])==0:
hands = hands+"ничего нет"
self.stdscr.addstr(2, 1, hands, curses.color_pair(1))
Ну и собственно основной код отрисовки локации, включающий бесконечный цикл ожидания нажатий клавиатуры. Предусматриваем переход на другую локацию, выход из игры и сохранение игры.
def location_view(self, location):
self.load_location(location)
self.stdscr.clear()
self.stdscr.border()
self.top_menu() # рисуем верхнее меню
# название локации
curses.init_pair(2, curses.COLOR_GREEN, curses.COLOR_BLACK)
x = int((self.scr_size[1] - len(self.data["title"])) / 2)
self.stdscr.addstr(2, x, self.data["title"],curses.color_pair(2))
# рисуем рисунок ежели он есть
y=3
if "ascii_art" in self.data:
mass_art=self.data["ascii_art"].split("\n")
i=0
while i<len(mass_art):
x=int((self.scr_size[1]-len(mass_art[i]))/2)
self.stdscr.addstr(y+i, x, mass_art[i])
i+=1
y = y + len(mass_art);
# Выводим описательную часть
curses.init_pair(3, curses.COLOR_CYAN, curses.COLOR_BLACK)
self.stdscr.addstr(y, 1, self.data["description"],curses.color_pair(3))
# Выводим нижнее меню
curses.init_pair(4, curses.COLOR_RED, curses.COLOR_BLACK)
self.stdscr.addstr(y + 1, 1, "Осмотреться вокруг [v] Применить что в руках [h]", curses.color_pair(4))
self.stdscr.addstr(y + 2, 1, "Выйти из квеста [q] Сохранить состояние [r]", curses.color_pair(4))
self.stdscr.refresh()
while True:
key=self.stdscr.getch()
print(key)
# реализация перехода с локации на локацию
if key==52 and "left" in self.data["available_locations"]:
self.player.data["location"]=self.data["available_locations"]["left"]
self.location_view(self.player.data["location"])
if key==54 and "right" in self.data["available_locations"]:
self.player.data["location"]=self.data["available_locations"]["right"]
self.location_view(self.player.data["location"])
if key==56 and "forward" in self.data["available_locations"]:
self.player.data["location"]=self.data["available_locations"]["forward"]
self.location_view(self.player.data["location"])
if key==50 and "back" in self.data["available_locations"]:
self.player.data["location"]=self.data["available_locations"]["back"]
self.location_view(self.player.data["location"])
if key == 114:
self.player.save()
#self.message("Внимание!","Состояние прохождения завершено. \nФайл находится в папке /saves")
if key==113:
curses.reset_shell_mode()
curses.endwin()
exit(0)
print(self.data)
В результате картинка (квест в консоли) на мониторе выглядит уже чуть симпатичнее:
Дело было вечером, делать было нечего (с). Ну не то чтобы совсем нечего, но выдалась свободное немножко время, поэтому для того чтобы не забыть (да уж чего там, и вспомнить уже) окончательно Python, решил сделать маленький движёк для текстовых квестов с выполняющихся в консоли (квест в консоли).
Сначала определимся что где и как:
Локации будем описывать в формате json
Локации будем складывать в папку locations. Имена файлов — номер локации.
В ходе квеста можно «сохраняться», чтобы была возможность продолжить квест
Сохранения будем хранить в папке saves
Все классы храним в папке classes
В результате у меня получилась такая структура папок и файлов:
Первым делом нарисую минимальный json стартовой локации:
Т.е. начинаем на стартовой локации (0), доступны переходы в локации 1,2,3 и 4.
Далее реализуем класс игрока, с реализацией функционала сохранения и стадии прохождения квеста:
from datetime import datetime as dt
import json
import os
class TPlayer:
data = {}
def __init__(self, name):
self.data["name"] = name # Имя пользователя
self.data["location"] = 0 # текущая локация
def load(self, filename):
"""
Загрузить состояние квеста из файла
:param filename: имя файла из папки saves
"""
f = open("saves/" + filename, mode='r', encoding='utf-8')
self.data = json.load(f)
f.close()
def list_saves(self):
"""
Показать доступные сохранения
"""
files = os.listdir("saves")
print(files)
def save(self):
"""
Сохранить текущее состояние пользователя
"""
time = dt.now()
filename = time.strftime("%d_%m_%Y_%H_%M") + ".save"
print(filename)
f = open("saves/" + filename, mode='w', encoding='utf-8')
json.dump(self.data, f)
f.close()
В главном файле (main.py), реализуем проверку аргументов командной строки и переход к началу квеста:
#!/usr/bin/env python3
# encoding: utf-8
import classes.player as tplayer
import classes.location as tlocation
import sys
player = tplayer.TPlayer("Васян")
location = tlocation.TLocation()
def start_location():
print(player.data["name"])
location.location_view(player.data["location"])
if __name__ == '__main__':
for param in sys.argv:
if param == "--load":
if len(sys.argv) == 2:
print("Ошибка: нет имени файла")
exit(0)
player.load(sys.argv[2])
start_location()
exit(0)
if param == "--new":
exit(0)
if param == "--list":
player.list_saves()
exit(0)
if len(sys.argv) == 1:
print("Для запуска квеста необходимо использовать следующие параметры:")
print(" --load - загрузить сохранение и начать квест")
print(" --list - получить список сохранений")
print(" --new <имя участника> - начать квест заново")
exit()
Разработка «квест в консоли» может быть действительно просто.. Вы можете посмотреть и другие мои статьи посвященные разработке на Python
