added file

This commit is contained in:
Godnessli committed 2026-10-18 15:44:55 +03:00
1 parent cc433bd065
commit c07cd997c9
10 files changed
+754 -1

No files matched your search

+7
View File
@@ -0,0 +1,7 @@
*.json
*.txt
whether_token
*.sqlite
/crypto_store
/.vscode
/__pycache__
+75
View File
@@ -0,0 +1,75 @@
import psutil
import shutil
from whether import Whether
from scp import SCP
from random import randint
from unknown_unswers import (
UNKNOWN_ANSWER,
UNKNOWN_PENTA,
UNKNOWN_TRIPLE
)
unknown_count = 0
def weather_request() -> str:
global unknown_count
unknown_count = 0
wh = Whether("Санкт-Петербург")
ret = wh.get_weather()
return ret
def system_request() -> str:
global unknown_count
unknown_count = 0
ret = "Использование CPU: " + str(psutil.cpu_percent(interval=1)) + "%\n"
ram = psutil.virtual_memory()
ret += "Использование RAM: " + str(ram.percent) + "%\n"
ret += "Использовано RAM: " + str(round(ram.used / 1e9, 2)) + "ГБ\n"
total, used, free = shutil.disk_usage("/")
ret += "Использование диска:\n"
ret += "Всего: %d ГБ" % (total // (2**30)) + "\n"
ret += "Использовано: %d ГБ" % (used // (2**30)) + "\n"
ret += "Свободно: %d ГБ" % (free // (2**30)) + "\n"
return ret
def unknown_request() -> str:
global unknown_count
unknown_count += 1
print(f"[DEBUG] UNKNOWN COUNT = {unknown_count}")
if unknown_count < 3:
num = randint(0, len(UNKNOWN_ANSWER) - 1)
return UNKNOWN_ANSWER[num]
elif unknown_count < 5:
num = randint(0, len(UNKNOWN_TRIPLE) - 1)
return UNKNOWN_TRIPLE[num]
elif unknown_count >= 5:
num = randint(0, len(UNKNOWN_PENTA) - 1)
return UNKNOWN_PENTA[num]
def help_request() -> str:
ret = "\n\nМеню:\n" \
"1. Сервер — Данные об использовании сервера.\n" \
"2. Погода в <город> — Погода в городе.=(тестово).\n" \
"3. SCP — Выдача случайного объекта фонда для чтения.\n" \
"4. SCP классы — Описание классов.\n\n" \
"Можете писать слова(Пример: Сервер) или цифрой(Пример: 1)."
return ret
def scp_request() -> str:
global unknown_count
unknown_count = 0
scp = SCP()
return scp.get_text()
def scp_classed() -> str:
global unknown_count
unknown_count = 0
return SCP.get_classified()
+184
View File
@@ -0,0 +1,184 @@
import login
import answers
import sys
from nio import (
MatrixRoom,
RoomMessageText,
InviteMemberEvent,
MegolmEvent,
KeyVerificationStart,
KeyVerificationKey,
KeyVerificationMac,
KeyVerificationCancel,
LocalProtocolError,
ToDeviceError
)
WELCOME_MSG = "Добро пожаловать ко мне, УЕБОТУ! АХАХА\n" \
"И так, вот что я умею:\n" \
"1. Сервер — Данные об использовании сервера.\n" \
"2. Погода в <город> — Погода в городе.=(тестово).\n" \
"3. SCP — Выдача случайного объекта фонда для чтения.\n" \
"4. SCP классы — Описание классов.\n" \
"5. Помощь — Это меню.\n\n" \
"Можете писать слова(Пример: Сервер) или цифрой(Пример: 1)."
client = None
async def message_callback(room: MatrixRoom, event: RoomMessageText):
print(f"[DEBUG] Открытое сообщение от {event.sender}.")
if event.sender == client.user_id:
return
content = None
if event.body.lower() == "сервер" or event.body == "1":
content = answers.system_request()
elif event.body.lower() == "погода" or event.body == "2":
content = answers.weather_request()
elif event.body.lower() == "scp" or event.body == "3":
content = answers.scp_request()
elif event.body.lower() == "scp классы" or event.body == "4":
content = answers.scp_classed()
else:
content = answers.unknown_request()
content += answers.help_request()
await send_message(room, content)
async def megolm_callback(room: MatrixRoom, event: MegolmEvent):
try:
decrypted = await client.decrypt_event(event)
except Exception as e:
print(f"[ERROR] Расшифровка не удалась: {e}")
return
if isinstance(decrypted, RoomMessageText):
print(f"[DEBUG] Расшифровано: {decrypted.body}")
if decrypted.sender == client.user_id:
return
await send_message(room, decrypted.body)
else:
print(f"[DEBUG] Расшифровано, но не текст: {type(decrypted).__name__}")
async def invite_callback(room: MatrixRoom, event: InviteMemberEvent):
print(f"[INFO] Приглашение в комнату {room.room_id}")
await client.join(room.room_id)
await send_message(room, WELCOME_MSG)
async def trust_user_devices(user_id: str, device_ids=None):
"""Доверие устройствам пользователя (локально, без интерактивной верификации)."""
if user_id == client.user_id:
return
if user_id not in client.device_store:
print(f"[WARN] Устройства {user_id} пока не известны.")
return
devices = client.device_store[user_id]
trusted = 0
for device_id, olm_device in devices.items():
if device_ids and device_id not in device_ids:
continue
try:
client.verify_device(olm_device)
print(f"[OK] Доверено устройство {device_id} пользователя {user_id}.")
trusted += 1
except Exception as e:
print(f"[ERROR] Не удалось доверить {device_id}: {e}")
if trusted == 0:
print(f"[WARN] Ни одно устройство {user_id} не было доверено.")
async def send_message(room: MatrixRoom, text: str):
CHUNK = 5000
if len(text) > CHUNK:
start = 0
end = 0
while start < len(text):
end = start + CHUNK
if end < len(text):
while text[end] != ".":
end -= 1
end += 1
chunk = text[start:end]
await client.room_send(
room_id = room.room_id,
message_type = "m.room.message",
content = {"msgtype" : "m.text", "body" : chunk},
ignore_unverified_devices = True
)
start = end
print(start)
else:
await client.room_send(
room_id = room.room_id,
message_type = "m.room.message",
content = {"msgtype" : "m.text", "body" : text},
ignore_unverified_devices = True
)
async def after_first_sync():
print("[INFO] Ожидание первого синка...")
await client.synced.wait()
print("[INFO] Первый синк выполнен. Доверяем устройствам...")
for user_id in login.TRUST_USERS:
await trust_user_devices(user_id)
async def to_device_callback(event):
"""Автоматически принимает и подтверждает SAS-верификацию от доверенных пользователей."""
try:
if isinstance(event, KeyVerificationStart):
if event.sender not in login.TRUST_USERS:
print(f"[WARN] Верификация от недоверенного пользователя {event.sender}, игнорируем.")
return
if "emoji" not in event.short_authentication_string:
print(f"[WARN] Устройство {event.sender} не поддерживает эмодзи-верификацию.")
return
print(f"[INFO] Принимаем запрос на верификацию от {event.sender}...")
resp = await client.accept_key_verification(event.transaction_id)
if isinstance(resp, ToDeviceError):
print(f"[ERROR] accept_key_verification: {resp}")
return
sas = client.key_verifications[event.transaction_id]
todevice_msg = sas.share_key()
resp = await client.to_device(todevice_msg)
if isinstance(resp, ToDeviceError):
print(f"[ERROR] to_device (share_key): {resp}")
elif isinstance(event, KeyVerificationKey):
sas = client.key_verifications[event.transaction_id]
print(f"[INFO] Эмодзи: {sas.get_emoji()}")
# Автоматически подтверждаем совпадение
resp = await client.confirm_short_auth_string(event.transaction_id)
if isinstance(resp, ToDeviceError):
print(f"[ERROR] confirm_short_auth_string: {resp}")
else:
print("[INFO] SAS подтверждён автоматически.")
elif isinstance(event, KeyVerificationMac):
sas = client.key_verifications[event.transaction_id]
try:
todevice_msg = sas.get_mac()
except LocalProtocolError as e:
print(f"[WARN] Верификация прервана: {e}")
return
resp = await client.to_device(todevice_msg)
if isinstance(resp, ToDeviceError):
print(f"[ERROR] to_device (mac): {resp}")
else:
print(f"[OK] Верификация с {event.sender} завершена. "
f"Доверенные устройства: {sas.verified_devices}")
elif isinstance(event, KeyVerificationCancel):
print(f"[INFO] Верификация отменена: {event.reason}")
except Exception as e:
print(f"[ERROR] to_device_callback: {e}")
+112
View File
@@ -0,0 +1,112 @@
import getpass
import json
import os
import sys
from nio import (
AsyncClient,
AsyncClientConfig,
LoginResponse
)
CONFIG_FILE = "./credentials.json"
STORE_PATH = "./crypto_store"
TRUST_USERS = ["@tema:godnessli.ru"]
async def login_from_credentials() -> AsyncClient:
"""
Аутентификация:
- если credentials.json есть — используем сохранённые данные;
- если нет — спрашиваем у пользователя и логинимся по паролю,
после чего сохраняем данные в credentials.json.
Логика взята из официального примера matrix-nio verify_with_emoji.py.
"""
client_config = AsyncClientConfig(
max_limit_exceeded=0,
max_timeouts=0,
store_sync_tokens=True,
encryption_enabled=True,
)
if not os.path.exists(STORE_PATH):
os.makedirs(STORE_PATH)
# ===== Вариант 1: файла нет — логинимся по паролю =====
if not os.path.exists(CONFIG_FILE):
print("Файл credentials.json не найден. Входим по паролю.")
homeserver = input("Введите URL homeserver [https://matrix.example.org]: ").strip()
if not homeserver:
homeserver = "https://matrix.example.org"
if not (homeserver.startswith("https://") or homeserver.startswith("http://")):
homeserver = "https://" + homeserver
user_id = input("Введите полный user ID [@user:example.org]: ").strip()
device_name = input("Имя устройства [echo-bot]: ").strip()
if not device_name:
device_name = "echo-bot"
password = getpass.getpass("Пароль: ")
client = AsyncClient(
homeserver,
user_id,
store_path=STORE_PATH,
config=client_config,
)
resp = await client.login(password=password, device_name=device_name)
if not isinstance(resp, LoginResponse):
print(f"[FATAL] Не удалось войти: {resp}")
sys.exit(1)
# Сохраняем данные в credentials.json
with open(CONFIG_FILE, "w") as f:
json.dump(
{
"homeserver": homeserver,
"user_id": resp.user_id,
"device_id": resp.device_id,
"access_token": resp.access_token,
},
f,
indent=2,
)
print(f"[INFO] Вход выполнен. Данные сохранены в {CONFIG_FILE}.")
print(f"[INFO] user_id={resp.user_id}, device_id={resp.device_id}")
# После login нужно вызвать restore_login, чтобы установить
# user_id/device_id/access_token и загрузить хранилище E2EE
client.restore_login(
user_id=resp.user_id,
device_id=resp.device_id,
access_token=resp.access_token,
)
return client
# ===== Вариант 2: файл есть — используем сохранённые данные =====
with open(CONFIG_FILE, "r") as f:
config = json.load(f)
client = AsyncClient(
config["homeserver"],
config["user_id"],
device_id=config["device_id"],
store_path=STORE_PATH,
config=client_config,
)
client.restore_login(
user_id=config["user_id"],
device_id=config["device_id"],
access_token=config["access_token"],
)
print(
f"[INFO] Вход по сохранённым данным: "
f"{config['user_id']} / {config['device_id']}"
)
return client
+84
View File
@@ -0,0 +1,84 @@
import time
import json
class Person:
__ACCOUNTS_FILE = "accounts.json"
__DOMAIN = "godnessli.ru"
def __init__(self, account: str, admin = False):
if self.__check_person(account):
self.__read_person(account)
else:
self.__create_person(account, admin)
def get_account(self) -> str:
return self.__account
def get_unknown_answers(self) -> int:
return self.__unknown_count
def clear_answers(self):
self.__unknown_count = 0
def incr_account(self):
self.__unknown_count += 1
def __check_person(self, account: str) -> bool:
accounts = None
try:
accounts = self.__get_accounts()
except FileNotFoundError:
return False
for acc in accounts:
if acc == account:
return True
return False
def __read_person(self, account: str):
accounts = self.__get_accounts()
acc = accounts[account]
self.__nick_name = acc["nick"]
self.__admin = acc["admin"]
self.__unknown_counter = acc["u_count"]
self.__whether_counter = acc["w_count"]
self.__ideas = acc["ideas"]
def __create_person(self, account: str, admin: bool):
accounts = None
try:
accounts = self.__get_accounts()
except FileNotFoundError:
pass
with open(self.__ACCOUNTS_FILE, 'w', encoding = 'utf-8') as p_file:
begin = account.find("@")
end = account.find(self.__DOMAIN)
accounts[account] = {
"nick" : account[begin:end],
"admin" : admin,
"u_count" : 0,
"w_count" : 0,
"ideas" : []
}
p_file.write(accounts)
acc = accounts[account]
self.__nick = acc["nick"]
self.__admin = acc["admin"]
self.__unknown_counter = acc["u_count"]
self.__whether_counter = acc["w_count"]
self.__ideas = acc["ideas"]
def __get_accounts(self) -> dict:
with open(self.__ACCOUNTS_FILE, 'r', encoding = 'utf-8') as p_file:
accounts = json.load(p_file)
p_file.close()
return accounts
tema = Person("@tema:godnessli")
Submodule pythontest deleted from 5ec0246c95.
+117
View File
@@ -0,0 +1,117 @@
import os
import random
import subprocess
from bs4 import BeautifulSoup
class SCP:
__ANUBIS = os.path.expanduser("~/.nix-profile/bin/anubis-fetch")
def __init__(self, num = None):
self.__scp = {
"title" : "SCP Foundation"
}
self_num = False
if num != None:
self_num = True
while self.__scp["title"] == "SCP Foundation":
if num == None:
num = random.randint(1, 4000)
url = ""
if num == 1:
self.__scp = {
"title" : "SCP-001",
"text" : "ПРИВЕДЁННЫЕ НИЖЕ МАТЕРИАЛЫ ЗАСЕКРЕЧЕНЫ\nПРИКАЗОМ АДМИНИСТРАТОРА\nСОВЕРШЕННО СЕКРЕТНО"
}
return
if num < 10:
url = f"https://scpfoundation.net/scp-00{num}"
elif num < 100:
url = f"https://scpfoundation.net/scp-0{num}"
else:
url = f"https://scpfoundation.net/scp-{num}"
print(url)
scp_html = self.__fetch_with_anubis(url)
self.__scp = self.__parse_html(scp_html)
if self_num:
break
else:
num = None
def get_text(self) -> str:
return f"{self.__scp["title"]}\n{self.__scp["text"]}"
@staticmethod
def get_classified() -> str:
return "Основные классы SCP:\n" \
"«Безопасный» — это аномалии, которые проще всего содержать без последствий.\n" \
"«Евклид» — относятся недостаточно изученные или изначально непредсказуемые аномалии, надёжное" \
"содержание которых не всегда возможно или требует бо́льших затрат, чем на содержание «безопасных».\n" \
"Вторичные классы:\n" \
"«Кетер» — это аномалии, постоянное или надёжное содержание которых тяжело реализуемо и сопряжено со сложными процедурами сдерживания.\n" \
"«Таумиэль» применяются Фондом исключительно для содержания других аномалий.\n" \
"«Аполлион» являются аномалиями, которые или невозможно содержать, и они неминуемо нарушат" \
"условия содержания, или предполагается развитие событий по схожему сценарию.\n" \
"«Архонт» являются аномалиями, чье содержание теоретически возможно, но по каким-либо причинам их лучше не содержать.\n" \
"«Тикондерога» невозможно содержать, но в этом и нет необходимости." \
"Эти аномалии схожи с объектами класса «Архонт» в том, что их содержание не обязательно," \
"но отличны тем, что на данный момент для их содержания у Фонда недостаточно ресурсов и " \
"компетенции из-за повсеместной распространённости таких аномалий по Земле." \
"При этом их существование или аномальная природа непостижимы для широких масс. "
def __fetch_with_anubis(self, url: str) -> str:
try:
result = subprocess.run(
[self.__ANUBIS, url],
capture_output = True, text = True, check = True
)
return result.stdout
except subprocess.CalledProcessError as e:
print(f"anubis-fetch завершился с кодом {e.returncode}: {e.stderr}")
raise
except subprocess.TimeoutExpired:
print("anubis-fetch не успел завершиться за отведённое время")
raise
def __parse_html(self, text: str) -> dict:
soup = BeautifulSoup(text, "lxml")
for tag in soup(["script", "style", "nostyle"]):
tag.decompose()
content = soup.find(id = "page-content") or soup.find("div", id = "content")
parse_text = content.get_text("\n", strip = True) if content else None
if parse_text:
marker = "Класс объекта:"
indx = parse_text.find(marker)
if indx != -1:
parse_text = parse_text[indx:]
else:
marker = "Объект №:"
indx = parse_text.find(marker)
if indx != -1:
parse_text = parse_text[indx:]
marker = "«\n"
indx = parse_text.find(marker)
if indx != -1:
parse_text = parse_text[:indx]
return {
"title" : soup.title.string.strip() if soup.title else None,
#"heading" : content.find("h1").get_text(strip = True) if content and content.find("h1") else None,
"text" : parse_text
}
+52
View File
@@ -0,0 +1,52 @@
import asyncio
import sys
import traceback
import login
import callbacks
from nio import (
RoomMessageText,
InviteMemberEvent,
MegolmEvent,
KeyVerificationEvent
)
async def main():
callbacks.client = await login.login_from_credentials()
# Критично для E2EE: загружаем ключи бота на сервер,
# чтобы другие устройства могли с ним общаться
if callbacks.client.should_upload_keys:
await callbacks.client.keys_upload()
print("[INFO] Ключи E2EE загружены на сервер.")
# Регистрируем обработчики
callbacks.client.add_event_callback(callbacks.message_callback, RoomMessageText)
callbacks.client.add_event_callback(callbacks.megolm_callback, MegolmEvent)
callbacks.client.add_event_callback(callbacks.invite_callback, InviteMemberEvent)
callbacks.client.add_to_device_callback(callbacks.to_device_callback, KeyVerificationEvent)
print("[INFO] Эхо-бот запущен. Для выхода нажмите Ctrl+C.")
after_sync_task = asyncio.ensure_future(callbacks.after_first_sync())
sync_task = asyncio.ensure_future(
callbacks.client.sync_forever(timeout=30000, full_state=True)
)
try:
await asyncio.gather(after_sync_task, sync_task)
except asyncio.CancelledError:
pass
finally:
await callbacks.client.close()
print("[INFO] Клиент закрыт.")
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\n[INFO] Остановка бота.")
except Exception:
print(traceback.format_exc())
sys.exit(1)
+37
View File
@@ -0,0 +1,37 @@
UNKNOWN_ANSWER = [
"Я вас не понял. Неверный пункт меню.",
"Что это? Давай без импровизации...",
"ААААА ПАМАГИТЕ! Я НЕ ПОНИЛ! ВВАДИ ТО, ЧТО ЕСТЬ НИЖЕ",
"Я тоже умею писать что-то не понятное!\n1000100101111001100101011101",
"Да, я написан на Python. Оу, вы не это спрашивали, тогда...",
"Нет, все не так, убийца дворецкий. А, вы не этого хотели? Так и я не этого ждал...",
"Ваш запрос не имеет смысла! Давайте я помогу...",
"Стакан наполовину пуст... Кстати, я не понял, что вы хотели.",
"Стакан наполовину полон! :) И в честь этого...",
"Все что имеет начало, имеет и..."
]
UNKNOWN_TRIPLE = [
"Тебе заняться нечем?",
"Ну давай! Я подожду, пока ты соизволишь ткнуть туда, куда надо...",
"Ты издеваешься?",
"Я тебе что, шутка какая-то?",
"Слушай, опусти глаза, там меню. Ты справишься...",
"А ты упертый...",
"Пиши, пиши... я подожду тут...",
"Интересно, что будет?"
]
UNKNOWN_PENTA = [
"И так... Видимо ты очень хочешь эту кнопку...\n@tema:godnessli.ru\nЭто мой создатель, напиши ему...",
"Серьезно? Зачем ты столько раз промазал мимо меню? Это уже абсурд!",
"А я хочу, а я хочу опять... О! Ты соизволил взглянуть на текст, тогда опсти чуть глаза и нажми то, что есть в меню.",
"...ну это уже не смешно...",
"Раз ты так хочешь эту кнопку, то я скажу создателю, чтобы когда будет на нее кто-то нажимать, тебе будет присылаться ЗВУК ПУКА!",
"Я устал, босс... выбери что-то из меню...",
"АААААА, Я БЕШЕНЫЙ, БЕГИТЕ!!!",
"@tema:godnessli.ru\nС просьбой добавить кнопку к нему.",
"Я не хочу с тобой общаться...",
"Ты думаешь, это смешно?",
"Успокоился?"
]
+86
View File
@@ -0,0 +1,86 @@
from geopy import Nominatim
import requests
import json
import datetime
class Whether:
def __init__(self, city: str):
gn = Nominatim(user_agent = "kzntsvartem@gmail.com")
self.__location = gn.geocode(city)
def get_weather(self) -> str:
url = f'https://api.weather.yandex.ru/v2/forecast?lat={self.__location.latitude}&lon={self.__location.longitude}'
access_key = open("whether_token").read()
header = {
'X-Yandex-Weather-Key' : access_key
}
#response = requests.get(url, headers = header)
#with open('data.json', 'w', encoding='utf-8') as f:
# json.dump(response.json(), f, ensure_ascii=False, indent=4)
#print(json.dumps(r, indent = 4))
with open('data.json') as json_data:
d = json.load(json_data)
json_data.close()
fact = d["fact"]
print(type(fact["cloudness"]))
string = "Погода сейчас:\n"
string += self.__get_cloudness(fact["cloudness"])
string += self.__get_temp(fact["temp"], fact["feels_like"])
string += self.__get_humidity(fact["humidity"])
string += self.__get_wind(fact["wind_dir"], fact["wind_speed"], fact["wind_gust"])
string += self.__get_prec(fact["prec_type"], fact["prec_strength"])
return string
return "Что-то пошло не так :("
def __get_cloudness(self, cloudness: float) -> str:
text = ""
if cloudness >= 0.8:
text += "Пасмурно"
elif cloudness >= 0.4:
text += "Облачно с прояснениями"
elif cloudness >= 0.1:
text += "Малооблачно"
else:
text += "Ясно"
text += "\n"
return text
def __get_humidity(self, humidity: int) -> str:
text = f"Влажность: {humidity}% ("
if humidity == 100:
text += "Кто проживает на дне океана?)"
elif humidity >= 90:
text += "Очень высокая"
elif humidity >= 70:
text += "Высокая"
elif humidity >= 50:
text += "Средняя"
elif humidity >= 30:
text += "Низкая"
else:
text += "Сушь галимая"
text += ")\n"
return text
def __get_temp(self, temp: int, feels: int) -> str:
text = f"Температура: {temp}°C\n"
text += f"(Ощущается как: {feels}°C)\n"
return text
def __get_wind(self, dir: str, speed: float, gust: float) -> str:
return ""
def __get_prec(self, type: int, strength: float) -> str:
return ""