poes/python-gotosocial-notif
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

ruff: fix something

Browse source
This commit is contained in:
Kusaeni 2024-10-01 11:32:29 +07:00
parent 92623b7369
commit 2f29314faa
3 changed files with 94 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
README.md
View file

@ -16,7 +16,7 @@ App ini masih dalam pengembangan dan banyak hal yang belum bisa dilakukan, berik
- [x] Boost/Reblog
- [x] Follow
- [x] Tandai data di database jika sudah terkirim
- [ ] Delay antar fungsi
- [x] Delay antar fungsi
- [ ] Cron internal
- [ ] Build

141
main.py
View file

@ -1,128 +1,160 @@
# %%
# ZED Python REPL
import requests
import os
import logging
import asyncio
import telegram
import os
import re
import json
from models import Notifikasi, Session
from sqlalchemy import and_, or_, func, update
from sqlalchemy.exc import IntegrityError
from dotenv import load_dotenv
from datetime import datetime
import requests
import telegram
from dotenv import load_dotenv
from sqlalchemy import update
from sqlalchemy.exc import IntegrityError
from models import Notifikasi, Session
# Request ke kauaku
load_dotenv()
gts_api = os.getenv('GTS_API')
gts_token = os.getenv('GTS_TOKEN')
gts_api = os.getenv("GTS_API")
gts_token = os.getenv("GTS_TOKEN")
headers = {"Authorization": f"Bearer {gts_token}"}
session = Session()
TELE_BOT = os.getenv('TELE_BOT')
TELE_CHATID = os.getenv('TELE_CHATID')
bot = telegram.Bot(token=f'{TELE_BOT}')
TELE_BOT = os.getenv("TELE_BOT")
TELE_CHATID = os.getenv("TELE_CHATID")
bot = telegram.Bot(token=f"{TELE_BOT}")
async def getNotif():
req = requests.get(f'{gts_api}?limit=5', headers=headers)
req = requests.get(f"{gts_api}?limit=5", headers=headers)
if req.status_code == 200:
data = req.json()
print(data.__class__)
try:
for notif in data:
type = notif['type']
type = notif["type"]
inreply = None
status_content = None
url = None
if 'status' in notif:
inreply = notif['status']['in_reply_to_id']
status_content = notif['status']['content']
url = notif['status']['url']
postid = notif['id']
created = datetime.strptime(notif['created_at'], "%Y-%m-%dT%H:%M:%S.%fZ")
handler = notif['account']['acct']
username = notif['account']['display_name']
kirim = 'Belum'
if "status" in notif:
inreply = notif["status"]["in_reply_to_id"]
status_content = notif["status"]["content"]
url = notif["status"]["url"]
postid = notif["id"]
created = datetime.strptime(
notif["created_at"], "%Y-%m-%dT%H:%M:%S.%fZ"
)
handler = notif["account"]["acct"]
username = notif["account"]["display_name"]
kirim = "Belum"
if type == 'follow':
insert_follow = Notifikasi(post_id=postid, created_at=created, handler=handler, display_name=username, type=type, remark=kirim)
if type == "follow":
insert_follow = Notifikasi(
post_id=postid,
created_at=created,
handler=handler,
display_name=username,
type=type,
remark=kirim,
)
session.add(insert_follow)
else:
insert_mentions = Notifikasi(post_id=postid, created_at=created, handler=handler, display_name=username, type=type, status=status_content,url=url, remark=kirim)
insert_mentions = Notifikasi(
inreplyto=inreply,
post_id=postid,
created_at=created,
handler=handler,
display_name=username,
type=type,
status=status_content,
url=url,
remark=kirim,
)
session.add(insert_mentions)
session.commit()
except IntegrityError as err:
session.rollback()
if 'post_id' in str(err.orig):
if "post_id" in str(err.orig):
print("Post Id sudah ada")
else:
print(f"Integritas data bermasalah", str(err))
print("Integritas data bermasalah", str(err))
finally:
session.close()
async def kirim_ke_tele():
cek_data = session.query(Notifikasi).filter(Notifikasi.remark.like('%elu%')).limit(10).all()
cek_data = (
session.query(Notifikasi)
.filter(Notifikasi.remark.like("%elu%"))
.limit(10)
.all()
)
for notif in cek_data:
def icon_(type_):
if type_ == 'mention':
return '💬 mention your note'
elif type_ == 'reblog':
return '🚀 boost your note'
elif type_ == 'favourite':
return '💖 falling in love with your note'
elif type_ == 'follow':
return '✋ started to follow you'
if type_ == "mention":
return "💬 mention your note"
elif type_ == "reblog":
return "🚀 boost your note"
elif type_ == "favourite":
return "💖 falling in love with your note"
elif type_ == "follow":
return "✋ started to follow you"
else:
return None
flag = icon_(notif.type)
if str(notif.type) == 'follow':
if str(notif.type) == "follow":
try:
await bot.send_message(parse_mode='html',chat_id=str(TELE_CHATID), text=f"""
await bot.send_message(
parse_mode="html",
chat_id=str(TELE_CHATID),
text=f"""
<b>{notif.display_name}</b>\n<i>{notif.handler}</i>\n{flag}\n from python
"""
""",
)
except Exception as e:
print(f'Follow: Ada masalah saat kirim ke Telebot', e)
print("Follow: Ada masalah saat kirim ke Telebot", e)
else:
html_strip = re.compile('<.*?>')
html_strip = re.compile("<.*?>")
status = str(notif.status)
status_text = re.sub(html_strip,'', status)
status_text = re.sub(html_strip, "", status)
try:
await bot.send_message(parse_mode='html',chat_id=str(TELE_CHATID), text=f"""
await bot.send_message(
parse_mode="html",
chat_id=str(TELE_CHATID),
text=f"""
<b>{notif.display_name}</b>\n<i>{notif.handler}</i>\n{flag}\n\n{status_text}\nfrom python\n{notif.url}
"""
""",
)
except Exception as e:
print(f'Mentions: Ada masalah saat kirim ke Telebot', e)
print("Mentions: Ada masalah saat kirim ke Telebot", e)
session.close()
async def tandai_notif():
try:
cek_data = session.query(Notifikasi).filter(Notifikasi.remark == 'Belum').all()
session.execute(
update(Notifikasi).
where(Notifikasi.remark == 'Belum').
values(remark='Sudah')
update(Notifikasi)
.where(Notifikasi.remark == "Belum")
.values(remark="Sudah")
)
session.commit()
except Exception as e:
print(f'Ada masalah saat update remark', e)
print("Ada masalah saat update remark", e)
finally:
session.close()
async def jalan():
await asyncio.sleep(2)
await getNotif()
@ -131,6 +163,7 @@ async def jalan():
await asyncio.sleep(5)
await tandai_notif()
#
if __name__ == "__main__":
asyncio.run(jalan())

11
models.py
View file

@ -1,20 +1,20 @@
# %%
# ZED Python REPL
from enum import unique
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Date
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Date, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# Engine
engine = create_engine('sqlite:///gotodon.db')
engine = create_engine("sqlite:///gotodon.db")
# Base
Base = declarative_base()
# Models
class Notifikasi(Base):
__tablename__ = 'notifikasi'
__tablename__ = "notifikasi"
id = Column(Integer, primary_key=True)
inreplyto = Column(String)
@ -30,6 +30,7 @@ class Notifikasi(Base):
def __repr__(self) -> str:
return f"<{self.id} : {self.handler} >"
# Buat tabel
Base.metadata.create_all(engine)