mirror of
https://github.com/TheFunny/TelegramTwitterMediaBot.git
synced 2026-09-23 23:32:05 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
87f0d16028
|
||
|
|
197993e522
|
||
|
|
02abcd899e
|
||
|
|
ebb48d2bb5
|
||
|
|
be77a33e9d
|
||
|
|
0439adefa0
|
||
|
|
81e5d2dbd1
|
@@ -17,7 +17,7 @@ from utils.pixiv import ProcessPixiv
|
|||||||
from utils.telegram import Telegram
|
from utils.telegram import Telegram
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from telegram import Chat, Message, Update
|
from telegram import Message, Update
|
||||||
from telegram.ext import Application, ContextTypes
|
from telegram.ext import Application, ContextTypes
|
||||||
|
|
||||||
logger = get_logger(__name__)
|
logger = get_logger(__name__)
|
||||||
@@ -141,7 +141,7 @@ async def cmd_set_forward_channel(update: Update, context: ContextTypes.DEFAULT_
|
|||||||
return
|
return
|
||||||
channel = context.args[0]
|
channel = context.args[0]
|
||||||
try:
|
try:
|
||||||
channel: Chat = await context.bot.get_chat(channel)
|
channel = await context.bot.get_chat(channel)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
await update.effective_message.reply_text(str(e))
|
await update.effective_message.reply_text(str(e))
|
||||||
return
|
return
|
||||||
@@ -153,6 +153,11 @@ async def cmd_set_forward_channel(update: Update, context: ContextTypes.DEFAULT_
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
await update.effective_message.reply_text(str(e) + "\nPlease add the bot to the channel and set as admin")
|
await update.effective_message.reply_text(str(e) + "\nPlease add the bot to the channel and set as admin")
|
||||||
return
|
return
|
||||||
|
user = filter(lambda x: x.user.id == update.effective_user.id, channel_admin)
|
||||||
|
user = next(user, None)
|
||||||
|
if not user:
|
||||||
|
await update.effective_message.reply_text("You are not an admin of the channel.")
|
||||||
|
return
|
||||||
user_bot = filter(lambda x: x.user.id == context.bot.id, channel_admin)
|
user_bot = filter(lambda x: x.user.id == context.bot.id, channel_admin)
|
||||||
user_bot = next(user_bot, None)
|
user_bot = next(user_bot, None)
|
||||||
if user_bot.can_post_messages:
|
if user_bot.can_post_messages:
|
||||||
|
|||||||
+1
-1
@@ -1,4 +1,4 @@
|
|||||||
python-telegram-bot[webhooks]~=21.3
|
python-telegram-bot[webhooks]~=21.4
|
||||||
httpx[http2]~=0.27
|
httpx[http2]~=0.27
|
||||||
uvloop~=0.19.0; sys_platform != 'win32'
|
uvloop~=0.19.0; sys_platform != 'win32'
|
||||||
async-pixiv @ git+https://github.com/TheFunny/async-pixiv@main
|
async-pixiv @ git+https://github.com/TheFunny/async-pixiv@main
|
||||||
+14
-3
@@ -5,6 +5,7 @@ from typing import Generator, Literal, TYPE_CHECKING
|
|||||||
from uuid import uuid4
|
from uuid import uuid4
|
||||||
|
|
||||||
from async_pixiv import PixivClient
|
from async_pixiv import PixivClient
|
||||||
|
from async_pixiv.error import ApiError
|
||||||
from telegram import InlineQueryResultPhoto, InputMediaPhoto
|
from telegram import InlineQueryResultPhoto, InputMediaPhoto
|
||||||
|
|
||||||
from utils.logger import get_logger
|
from utils.logger import get_logger
|
||||||
@@ -68,7 +69,7 @@ class Pixiv:
|
|||||||
|
|
||||||
@property
|
@property
|
||||||
def author_url(self) -> str:
|
def author_url(self) -> str:
|
||||||
return str(self._illust.user.link)
|
return str(self._illust.user.link).rstrip('/')
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def description(self) -> str:
|
def description(self) -> str:
|
||||||
@@ -118,11 +119,16 @@ class ProcessPixiv:
|
|||||||
async def init_client(cls, token: str) -> None:
|
async def init_client(cls, token: str) -> None:
|
||||||
cls._client = PixivClient()
|
cls._client = PixivClient()
|
||||||
await cls._client.login_with_token(token)
|
await cls._client.login_with_token(token)
|
||||||
|
cls._token = token
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
async def close_client(cls) -> None:
|
async def close_client(cls) -> None:
|
||||||
await cls._client.close()
|
await cls._client.close()
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
async def refresh_token(cls):
|
||||||
|
await cls._client.login_with_token(cls._token)
|
||||||
|
|
||||||
def __init__(self, url: str):
|
def __init__(self, url: str):
|
||||||
self._url: str = url
|
self._url: str = url
|
||||||
|
|
||||||
@@ -135,7 +141,11 @@ class ProcessPixiv:
|
|||||||
|
|
||||||
async def _fetch_illust(self) -> Illust:
|
async def _fetch_illust(self) -> Illust:
|
||||||
illust_id = self._parse_illust_id()
|
illust_id = self._parse_illust_id()
|
||||||
return (await self._client.ILLUST.detail(illust_id)).illust
|
try:
|
||||||
|
return (await self._client.ILLUST.detail(illust_id)).illust
|
||||||
|
except ApiError:
|
||||||
|
await self.refresh_token()
|
||||||
|
return (await self._client.ILLUST.detail(illust_id)).illust # TODO use retry here
|
||||||
|
|
||||||
def _parse_illust_id(self) -> int:
|
def _parse_illust_id(self) -> int:
|
||||||
return int(self._url.split("/")[-1])
|
return int(self._url.split("/")[-1])
|
||||||
@@ -192,9 +202,10 @@ class TelegramPixiv:
|
|||||||
def message_media_generator(self) -> Generator[TypeMessageMediaResult, None, None]:
|
def message_media_generator(self) -> Generator[TypeMessageMediaResult, None, None]:
|
||||||
pixiv = self._pixiv
|
pixiv = self._pixiv
|
||||||
for media in pixiv.images:
|
for media in pixiv.images:
|
||||||
|
logger.info(str(media))
|
||||||
if pixiv.type in ("illust", "manga"):
|
if pixiv.type in ("illust", "manga"):
|
||||||
yield InputMediaPhoto(
|
yield InputMediaPhoto(
|
||||||
media=media.url,
|
media=media.large,
|
||||||
has_spoiler=pixiv.is_nsfw
|
has_spoiler=pixiv.is_nsfw
|
||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
|
|||||||
+3
-3
@@ -1,8 +1,8 @@
|
|||||||
import re
|
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)\.com/(.+)/status/(\d+)")
|
||||||
x_media_url = re.compile(r"^(?:https?://)(pbs|video)\.twimg\.com/(.*)")
|
x_media_url = re.compile(r"^(?:https?://)?(pbs|video)\.twimg\.com/(.*)")
|
||||||
x_tco_url = re.compile(r"(?:https?://)t\.co/.+$", re.M)
|
x_tco_url = re.compile(r"(?:https?://)?t\.co/.+$", re.M)
|
||||||
message_url = re.compile(r"\[.+]", re.S)
|
message_url = re.compile(r"\[.+]", re.S)
|
||||||
|
|
||||||
pixiv_url = re.compile(r"^(?:https?://)?(?:www\.)?pixiv\.net/(?:en/)?artworks/(\d+)")
|
pixiv_url = re.compile(r"^(?:https?://)?(?:www\.)?pixiv\.net/(?:en/)?(?:artworks/|i/)(\d+)")
|
||||||
|
|||||||
+1
-1
@@ -18,7 +18,7 @@ class Telegram:
|
|||||||
async with TelegramPixiv(self._url) as pixiv:
|
async with TelegramPixiv(self._url) as pixiv:
|
||||||
return pixiv
|
return pixiv
|
||||||
else:
|
else:
|
||||||
return None
|
return None # TODO add raise and catch
|
||||||
|
|
||||||
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
||||||
pass
|
pass
|
||||||
|
|||||||
Reference in New Issue
Block a user