Compare commits

..
10 Commits
8 changed files with 43 additions and 11 deletions
+1 -1
View File
@@ -1,4 +1,4 @@
FROM python:3.12-slim-bullseye
FROM python:3.13-slim-bullseye
LABEL maintainer="admin@yoursfunny.top"
+1 -1
View File
@@ -13,7 +13,7 @@ ADMIN = [int(i) for i in os.getenv("BOT_ADMIN", "").split(",") if i]
PIXIV_REFRESH_TOKEN = os.getenv("PIXIV_REFRESH_TOKEN")
WEBHOOK = os.getenv("WEBHOOK", False)
WEBHOOK = os.getenv("WEBHOOK").strip().lower() in ("true", "yes", "1")
if WEBHOOK:
WEBHOOK_LISTEN = os.getenv("WEBHOOK_LISTEN", "0.0.0.0")
WEBHOOK_PORT = int(os.getenv("WEBHOOK_PORT", 8443))
+17
View File
@@ -162,6 +162,22 @@ async def query_template(update: Update, context: CustomContext) -> None:
await _edit_message.forward[0].edit_caption(context.chat_data.template[name])
@send_action(ChatAction.TYPING)
async def cmd_start(update: Update, context: CustomContext) -> None:
await update.effective_message.reply_text(
"Welcome to the Twitter Fetcher Bot!\n"
"You can use this bot to fetch tweets from Twitter and forward them to a channel.\n"
"Use /set_forward_channel to set a channel to forward tweets.\n"
"Use /remove_forward_channel to remove the channel.\n"
"Use /edit_before_forward to enable or disable edit before forward.\n"
"Use /set_template to set a template for the forwarded message.\n"
"Use /bot_dict to see the bot's data.\n"
"Use /clear_edit_message to clear the edit message cache.\n"
"You can also reply to a message with a tweet URL to fetch the tweet and forward it to the channel.\n"
"You can also use inline query to search for tweets."
)
@send_action(ChatAction.TYPING)
async def cmd_set_forward_channel(update: Update, context: CustomContext) -> None:
if not context.args:
@@ -289,6 +305,7 @@ def main():
MessageHandler((filters.Regex(regex.x_url) | filters.Regex(regex.pixiv_url)) | filters.Regex(
regex.bsky_url) & filters.ChatType.PRIVATE,
handel_url_media),
CommandHandler("start", cmd_start),
CommandHandler("set_forward_channel", cmd_set_forward_channel),
CommandHandler("remove_forward_channel", cmd_remove_forward_channel),
CommandHandler("edit_before_forward", cmd_edit_before_forward),
+2 -2
View File
@@ -1,4 +1,4 @@
python-telegram-bot[webhooks]~=21.7
python-telegram-bot[webhooks]~=22.5
httpx[http2]~=0.27
uvloop~=0.19.0; sys_platform != 'win32'
uvloop~=0.22; sys_platform != 'win32'
async-pixiv @ git+https://github.com/TheFunny/async-pixiv@main
+11 -2
View File
@@ -11,6 +11,7 @@ if TYPE_CHECKING:
bsky_api_url = 'https://public.api.bsky.app/xrpc/app.bsky.feed.getPostThread'
SENSITIVE_TAG = {'sexual', 'nudity', 'porn', 'graphic-media'}
class BskyMedia:
__slots__ = ('_url', '_thumb', '_type', '__dict__')
@@ -101,7 +102,7 @@ class ProcessBsky:
author_id=self._bsky['author']['handle'],
text=self._bsky['record']['text'],
media=self._bsky_media,
sensitive=False # TODO: implement sensitive
sensitive=self._sensitive
)
async def __aexit__(self, exc_type, exc_val, exc_tb):
@@ -151,4 +152,12 @@ class ProcessBsky:
)
]
case _:
raise NotImplementedError(f"Unknown Bsky embed type: {embed['type']}")
raise NotImplementedError(f"Unknown Bsky embed type: {embed['$type']}")
@property
def _sensitive(self) -> bool:
return any(
tag in label['val']
for label in self._bsky['labels']
for tag in SENSITIVE_TAG
)
+3 -2
View File
@@ -1,10 +1,11 @@
import re
x_url = re.compile(r"^(?:https?://)?(?:www\.|mobile\.)?(?:x|twitter|fixvx|vxtwitter)\.com/(.+)/status/(\d+)")
x_url = re.compile(
r"^(?:https?://)?(?:www\.|mobile\.)?(?:x|twitter|fixvx|vxtwitter|fixupx|fxtwitter)\.com/(.+)/status/(\d+)")
x_media_url = re.compile(r"^(?:https?://)?(pbs|video)\.twimg\.com/(.*)")
x_tco_url = re.compile(r"(?:https?://)?t\.co/.+$", re.M)
message_url = re.compile(r"\[.+]", re.S)
pixiv_url = re.compile(r"^(?:https?://)?(?:www\.)?pixiv\.net/(?:en/)?(?:artworks/|i/)(\d+)")
bsky_url = re.compile(r"^(?:https?://)?(?:bsky)\.app/profile/(.+)/post/(.+)")
bsky_url = re.compile(r"^(?:https?://)?bsky\.app/profile/(.+)/post/(.+)")
+3 -3
View File
@@ -84,7 +84,7 @@ class Tweet:
@cached_property
def url(self) -> str:
return f"https://twitter.com/{self._author_id}/status/{self._id}"
return f"https://x.com/{self._author_id}/status/{self._id}"
@property
def author(self) -> str:
@@ -92,7 +92,7 @@ class Tweet:
@cached_property
def author_url(self) -> str:
return f"https://twitter.com/{self._author_id}"
return f"https://x.com/{self._author_id}"
@property
def text(self) -> str:
@@ -136,7 +136,7 @@ class ProcessTweet:
@property
def _tweet_text(self) -> str:
match = x_tco_url.search(self._tweet['text'])
return self._tweet['text'][:match.start()].strip(" ") if match else self._tweet['text']
return self._tweet['text'][:match.start()].strip() if match else self._tweet['text']
@property
def _tweet_media(self) -> list[TweetMedia]:
+5
View File
@@ -31,6 +31,7 @@ class BskyPost(TypedDict):
author: BskyAuthor
record: BskyPostRecord
embed: BskyEmbedImages | BskyEmbedVideo | BskyEmbedExternal
labels: list[BskyLabel]
class BskyAuthor(TypedDict):
@@ -42,6 +43,10 @@ class BskyPostRecord(TypedDict):
text: str
class BskyLabel(TypedDict):
val: str
class BskyEmbedImage(TypedDict):
thumb: str
fullsize: str