Use a new backend for instagram

cogs/configuration.py

@@ -6,9 +6,9 @@
 
 import discord
 from discord.ext import commands
-from modules.misobot import MisoBot
 
 from modules import emoji_literals, exceptions, queries, util
+from modules.misobot import MisoBot
 
 
 class ChannelSetting(commands.TextChannelConverter):
@@ -352,7 +352,7 @@ async def starboard_emoji(self, ctx: commands.Context, emoji):
 
  if emoji[0] == "<":
  # is custom emoji
- if not await queries.is_donator(ctx, ctx.author, 2):
+ if not await queries.is_donator(ctx.bot, ctx.author, 2):
  raise exceptions.CommandInfo(
  "You have to be a [donator](https://misobot.xyz/donate) "
  "to use custom emojis with the starboard!"

cogs/errorhandler.py

@@ -8,12 +8,12 @@
 import discord
 from discord.ext import commands
 from loguru import logger
+
+from modules import emojis, exceptions, queries, util
 from modules.instagram import InstagramError
 from modules.misobot import MisoBot
 from modules.tiktok import TiktokError
 
-from modules import emojis, exceptions, queries, util
-
 
 @dataclass
 class ErrorMessages:
@@ -174,7 +174,7 @@ async def handle_cooldown(
  ):
  if (
  ctx.author.id == ctx.bot.owner_id
- or await queries.is_donator(ctx, ctx.author, 2)
+ or await queries.is_donator(self.bot, ctx.author, 2)
  or await queries.is_vip(self.bot, ctx.author)
  ):
  await self.reinvoke_command(ctx)

cogs/events.py

@@ -9,11 +9,11 @@
 import discord
 from discord.ext import commands, tasks
 from loguru import logger
+
+from modules import emoji_literals, exceptions, queries, util
 from modules.media_embedders import BaseEmbedder, InstagramEmbedder, TikTokEmbedder
 from modules.misobot import MisoBot
 
-from modules import emoji_literals, queries, util
-
 
 class Events(commands.Cog):
  """Event handlers for various discord events"""
@@ -65,7 +65,7 @@ async def on_command_completion(self, ctx: commands.Context):
  await queries.save_command_usage(ctx)
 
  if random.randint(1, 69) == 1 and not await queries.is_donator(
- ctx, ctx.author
+ ctx.bot, ctx.author
  ):
  logger.info("Sending donation beg message")
  await util.send_donation_beg(ctx.channel)
@@ -360,7 +360,12 @@ async def parse_media_auto_embed(
  embedder = InstagramEmbedder(self.bot)
  posts = embedder.extract_links(message.content, include_shortcodes=False)
  if posts:
- await self.embed_posts(posts, message, embedder)
+ if await util.user_is_donator(message.author, self.bot):
+ await self.embed_posts(posts, message, embedder)
+ else:
+ raise exceptions.CommandInfo(
+ "Only [donators](https://misobot.xyz/donate) can embed instagram posts!"
+ )
 
  if media_settings["tiktok"]:
  embedder = TikTokEmbedder(self.bot)

cogs/notifications.py

@@ -9,9 +9,9 @@
 import regex
 from discord.ext import commands
 from loguru import logger
-from modules.misobot import MisoBot
 
 from modules import emojis, exceptions, queries, util
+from modules.misobot import MisoBot
 
 
 class Notifications(commands.Cog):
@@ -174,7 +174,7 @@ async def notification_add(self, ctx: commands.Context, *, keyword: str):
  "Global notifications have been removed for performance reasons."
  )
 
- if not await queries.is_donator(ctx, ctx.author, 2):
+ if not await queries.is_donator(ctx.bot, ctx.author, 2):
  amount = await self.bot.db.fetch_value(
  "SELECT COUNT(*) FROM notification WHERE user_id = %s",
  ctx.author.id,

cogs/user.py

@@ -10,9 +10,9 @@
 import discord
 import humanize
 from discord.ext import commands
-from modules.misobot import MisoBot
 
 from modules import emojis, exceptions, queries, util
+from modules.misobot import MisoBot
 
 
 class User(commands.Cog):
@@ -558,7 +558,7 @@ def make_badge(classname):
  if user.bot:
  badges.append(make_badge(badge_classes["bot"]))
 
- if await queries.is_donator(ctx, user):
+ if await queries.is_donator(ctx.bot, user):
  badges.append(make_badge(badge_classes["patreon"]))
 
  user_settings = await self.bot.db.fetch_row(

modules/instagram.py

@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: MPL-2.0
 # https://git.joinemm.dev/miso-bot
 
+import asyncio
 from dataclasses import dataclass
 from enum import Enum
 from typing import TYPE_CHECKING, Optional
@@ -12,6 +13,7 @@
 import arrow
 import orjson
 import redis
+from bs4 import BeautifulSoup
 from loguru import logger
 
 if TYPE_CHECKING:
@@ -48,17 +50,17 @@ class IgMedia:
 
 @dataclass
 class IgUser:
- id: int
  username: str
- avatar_url: str
+ id: int | None = None
+ avatar_url: str | None = None
 
 
 @dataclass
 class IgPost:
  url: str
  user: IgUser
  media: list[IgMedia]
- timestamp: int
+ timestamp: int | None = None
  caption: str = ""
 
 
@@ -91,6 +93,79 @@ def decode(shortcode, alphabet=ENCODING_CHARS):
  )
 
 
+class InstaFix:
+ BASE_URL = "https://www.ddinstagram.com"
+
+ def __init__(self, session: aiohttp.ClientSession):
+ self.session = session
+
+ async def request(self, url: str):
+ tries = 0
+ while tries < 3:
+ try:
+ async with self.session.get(
+ url, allow_redirects=False, headers={"User-Agent": "bot"}
+ ) as response:
+ text = await response.text()
+ return text
+
+ except aiohttp.ClientConnectorError as e:
+ logger.warning(e)
+ tries += 1
+ await asyncio.sleep(tries)
+
+ async def try_media(self, shortcode: str) -> list:
+ media = []
+ for i in range(1, 11):
+ text = await self.request(f"{self.BASE_URL}/p/{shortcode}/{i}")
+ soup = BeautifulSoup(text, "lxml")
+ imagetag = soup.find("meta", {"property": "og:image"})
+ videotag = soup.find("meta", {"property": "og:video"})
+
+ if imagetag:
+ media.append(
+ IgMedia(
+ url=self.BASE_URL + imagetag.attrs["content"],
+ media_type=MediaType.PHOTO,
+ )
+ )
+ elif videotag:
+ media.append(
+ IgMedia(
+ url=self.BASE_URL + videotag.attrs["content"],
+ media_type=MediaType.VIDEO,
+ )
+ )
+ else:
+ break
+
+ return media
+
+ async def get_post(self, shortcode: str):
+ text = await self.request(f"{self.BASE_URL}/p/{shortcode}")
+ soup = BeautifulSoup(text, "lxml")
+ metadata = {
+ "url": soup.find("a").attrs["href"],
+ "description": soup.find("meta", {"property": "og:description"}).attrs[
+ "content"
+ ],
+ "username": soup.find("meta", {"name": "twitter:title"}).attrs["content"],
+ }
+
+ media = await self.try_media(shortcode)
+
+ return IgPost(
+ url=metadata["url"],
+ user=IgUser(username=metadata["username"].strip("@")),
+ caption=metadata["description"],
+ media=media,
+ timestamp=None,
+ )
+
+ async def get_story(self, username: str, story_pk: str):
+ raise InstagramError("Instagram stories are not supported at the moment.")
+
+
 class Datalama:
  BASE_URL = "https://api.datalikers.com"
 
@@ -113,10 +188,12 @@ async def api_request_with_cache(
  ) -> tuple[dict, bool, str]:
  cache_key = self.make_cache_key(endpoint, params)
 
+ was_cached = False
  data = await self.try_cache(cache_key)
- was_cached = data is not None
- if not was_cached:
+ if data is None:
  data = await self.api_request(endpoint, params)
+ else:
+ was_cached = True
 
  return data, was_cached, cache_key
 
@@ -132,6 +209,7 @@ async def try_cache(self, cache_key: str) -> dict | None:
  if prom := self.bot.get_cog("Prometheus"):
  await prom.increment_instagram_cache_hits() # type: ignore
  return orjson.loads(cached_response)
+
  return None
 
  async def save_cache(self, cache_key: str, data: dict, lifetime: int):
@@ -333,10 +411,10 @@ def __init__(
 
  if use_proxy:
  proxy_url: str = bot.keychain.PROXY_URL
- self.proxy = proxy_url
  proxy_user: str = bot.keychain.PROXY_USER
  proxy_pass: str = bot.keychain.PROXY_PASS
 
+ self.proxy: str | None = proxy_url
  self.proxy_auth = aiohttp.BasicAuth(proxy_user, proxy_pass)
  else:
  self.proxy = None

modules/media_embedders.py

@@ -6,6 +6,7 @@
 import io
 from typing import TYPE_CHECKING, Any
 
+import aiohttp
 import arrow
 import discord
 import regex
@@ -17,6 +18,7 @@
 from loguru import logger
 
 from modules import emojis, exceptions, instagram, util
+from modules.instagram import InstaFix
 from modules.tiktok import TikTok
 
 if TYPE_CHECKING:
@@ -343,24 +345,28 @@ async def create_message(
  instagram_asset: InstagramPost | InstagramStory,
  options: Options | None = None,
  ):
- if isinstance(instagram_asset, InstagramPost):
- post = await self.bot.datalama.get_post_v1(instagram_asset.shortcode)
- identifier = instagram_asset.shortcode
- elif isinstance(instagram_asset, InstagramStory):
- post = await self.bot.datalama.get_story_v1(
- instagram_asset.username, instagram_asset.story_pk
- )
- identifier = instagram_asset.story_pk
+ async with aiohttp.ClientSession(read_timeout=3) as session:
+ instafix = InstaFix(session)
+ if isinstance(instagram_asset, InstagramPost):
+ post = await instafix.get_post(instagram_asset.shortcode)
+ identifier = instagram_asset.shortcode
+ elif isinstance(instagram_asset, InstagramStory):
+ post = await instafix.get_story(
+ instagram_asset.username, instagram_asset.story_pk
+ )
+ identifier = instagram_asset.story_pk
 
  username = discord.utils.escape_markdown(post.user.username)
- caption = f"{self.EMOJI} **@{username}** <t:{post.timestamp}:d>"
+ caption = f"{self.EMOJI} **@{username}**"
+ if post.timestamp:
+ caption += f" <t:{post.timestamp}:d>"
  if options and options.captions:
  caption += f"\n>>> {post.caption}"
  tasks = []
  for n, media in enumerate(post.media, start=1):
  ext = "mp4" if media.media_type == instagram.MediaType.VIDEO else "jpg"
- dateformat = arrow.get(post.timestamp).format("YYMMDD")
- filename = f"{dateformat}-@{post.user.username}-{identifier}-{n}.{ext}"
+ # dateformat = arrow.get(post.timestamp).format("YYMMDD")
+ filename = f"@{post.user.username}-{identifier}-{n}.{ext}"
  tasks.append(
  self.download_media(
  media.url,

modules/misobot.py

@@ -18,7 +18,6 @@
 
 from modules import cache, maria, util
 from modules.help import EmbedHelpCommand
-from modules.instagram import Datalama
 from modules.keychain import Keychain
 from modules.reddit import Reddit
 from modules.redis import Redis
@@ -101,7 +100,6 @@ def __init__(
  self.version = "5.1"
  self.extensions_loaded = False
  self.redis: Redis = Redis()
- self.datalama = Datalama(self)
  self.boot_up_time: float | None = None
  self.session: aiohttp.ClientSession
  self.reddit_client = Reddit(self)

modules/queries.py

@@ -7,7 +7,6 @@
 from typing import TYPE_CHECKING
 
 import discord
-from discord.ext import commands
 
 from modules import exceptions
 
@@ -45,14 +44,11 @@ async def update_setting(ctx, table, setting, new_value):
 
 
 async def is_donator(
- ctx: commands.Context,
+ bot: "MisoBot",
  user: discord.User | discord.Member,
  unlock_tier: int | None = None,
 ):
- if user.id == ctx.bot.owner_id:
- return True
-
- data = await ctx.bot.db.fetch_row(
+ data = await bot.db.fetch_row(
  """
  SELECT donation_tier, currently_active FROM donator
  WHERE user_id = %s

modules/util.py

@@ -859,12 +859,18 @@ async def send_tasks_result_list(
  await send_as_pages(ctx, content, rows, maxrows=20)
 
 
-async def patron_check(ctx):
- if ctx.author.id == ctx.bot.owner_id:
+async def user_is_donator(user: discord.User, bot: "MisoBot") -> bool:
+ # if user.id == bot.owner_id:
+ # return True
+ if await queries.is_donator(bot, user):
  return True
- if await queries.is_donator(ctx, ctx.author):
+ if await queries.is_vip(bot, user):
  return True
- if await queries.is_vip(ctx.bot, ctx.author):
+ return False
+
+
+async def patron_check(ctx):
+ if user_is_donator(ctx.author, ctx.bot):
  return True
  raise PatronCheckFailure