Release 1.0.0

.env.example

@@ -0,0 +1,7 @@
+CACHE_CHANNEL_LIMIT=500
+CACHE_TIMEOUT_7TV=300
+CACHE_TIMEOUT_BTTV=300
+CACHE_TIMEOUT_FFZ=300
+CACHE_TIMEOUT_TWITCH=300
+TWITCH_CLIENT_ID=
+TWITCH_SECRET=

Dockerfile

@@ -0,0 +1,6 @@
+FROM python:3.7.13-alpine
+COPY ./ /app
+WORKDIR /app
+RUN ls -a
+RUN pip3 install -r requirements.txt
+CMD ["gunicorn", "--preload", "--bind", "0.0.0.0:8000", "app:app"]

README.md

@@ -0,0 +1,65 @@
+# emotes-api
+Fetches Twitch emotes from various providers (Twitch, 7TV, BTTV, FFZ).
+
+[![Clang analysis](https://status.crippled.dev/api/badge/5/status)](https://emotes.crippled.dev)
+
+# Endpoints
+- `/v1/channel/<username>/<provider>`
+- `/v1/global/<provider>`
+
+Available providers: `twitch`, `7tv`, `bttv`, `ffz`, `all`.
+
+## Output
+```json
+[
+ {
+ "animated": false,
+ "code": "FeelsDankMan",
+ "provider": 1,
+ "urls": [
+ {
+ "size": "1x",
+ "url": "https://cdn.7tv.app/emote/63071bb9464de28875c52531/1x.webp"
+ },
+ {
+ "size": "2x",
+ "url": "https://cdn.7tv.app/emote/63071bb9464de28875c52531/2x.webp"
+ },
+ {
+ "size": "3x",
+ "url": "https://cdn.7tv.app/emote/63071bb9464de28875c52531/3x.webp"
+ },
+ {
+ "size": "4x",
+ "url": "https://cdn.7tv.app/emote/63071bb9464de28875c52531/4x.webp"
+ }
+ ],
+ "zero_width": false
+ },
+]
+```
+
+### Providers
+| id | Provider |
+| --- | --- |
+| 0 | Twitch |
+| 1 | 7TV |
+| 2 | BTTV |
+| 3 | FFZ |
+
+# Setup
+1. Register an app on [dev.twitch.tv](https://dev.twitch.tv/console/apps/create) and enter the client ID and secret in [.env](.env.example).
+2. Install requirements.
+```python
+pip3 install -r requirements.txt
+```
+3. Run the server.
+```python
+gunicorn --preload --bind 0.0.0.0:8000 app:app
+```
+
+# Development
+- Unit tests
+```python
+python3 -m unittest
+```

app.py

@@ -4,91 +4,18 @@
 from flask_limiter import Limiter
 from flask_limiter.util import get_remote_address
 import logging
-from twitchAPI.twitch import Twitch
-import os
+from cachetools import LRUCache, cached
 from dotenv import load_dotenv
-import asyncio
-import time
+import os
+from models.channel import Channel
 
 # Providers
 # 0: Twitch
 # 1: 7TV
 # 2: BTTV
 # 3: FFZ
 
-CACHE_TIMEOUT = 300 # Emote cache duration in seconds
-
-load_dotenv()
-TWITCH_CLIENT_ID = os.getenv('TWITCH_CLIENT_ID')
-TWITCH_SECRET = os.getenv('TWITCH_SECRET')
-
-global_emotes_twitch = []
-global_emotes_twitch_updated = 0
-
-class EmoteUrl:
- def __init__(self, size, url):
- self.size = size
- self.url = url
-
- def toDict(self):
- return dict(size=self.size, url=self.url)
-
-class Emote:
- def __init__(self, code, provider, zero_width):
- self.code = code
- self.provider = provider
- self.zero_width = zero_width
- self.urls = []
-
- def toDict(self):
- return dict(
- code=self.code,
- provider=self.provider,
- zero_width=self.zero_width,
- urls=[u.toDict() for u in self.urls]
- )
-
- def addUrl(self, size, url):
- self.urls.append(EmoteUrl(size, url))
-
-def parseTwitchEmote(e, template):
- emote = Emote(
- code=e.name,
- provider=0,
- zero_width=False
- )
-
- for scale in e.scale:
- values = {
- 'id': e.id,
- 'format': 'default',
- 'theme_mode': e.theme_mode[0],
- 'scale': scale,
- }
- url = template.replace('{{', '{').replace('}}', '}').format(**values)
- size = f"{int(float(scale))}x"
- emote.addUrl(size, url)
-
- return emote
-
-# Twitch API
-async def update_twitch_global_emotes():
- global global_emotes_twitch_updated
- if (time.time() - global_emotes_twitch_updated) < CACHE_TIMEOUT:
- print('Using cached global twitch emotes.')
- return
-
- print('Updating global twitch emotes..')
- global_emotes_twitch.clear()
-
- twitch = await Twitch(TWITCH_CLIENT_ID, TWITCH_SECRET)
- emotes = await twitch.get_global_emotes()
- for e in emotes:
- emote = parseTwitchEmote(e, emotes.template)
- global_emotes_twitch.append(emote)
-
- global_emotes_twitch_updated = time.time()
- print(f'Updated global twitch emotes: {len(global_emotes_twitch)} emotes.')
+channels = {}
 
 print("Starting server..")
 
@@ -98,29 +25,66 @@ async def update_twitch_global_emotes():
 Compress(app)
 limiter = Limiter(app, key_func=get_remote_address)
 
+load_dotenv()
+CACHE_CHANNEL_LIMIT = int(os.getenv('CACHE_CHANNEL_LIMIT', 500))
+
+cache = LRUCache(maxsize=CACHE_CHANNEL_LIMIT)
+
 def custom_error(status_code, message):
  response = jsonify({
  'status_code': status_code,
  'message': message
  })
  response.status_code = status_code
- app.logger.warning('Returning error:', status_code, message)
+ app.logger.warning('Returning error: %s %s', status_code, message)
  return response
 
+@app.errorhandler(404)
+def not_found(error):
+ return custom_error(404, 'Page not found.')
+
 @app.errorhandler(429)
 def rate_limit_exceeded(error):
  return custom_error(429, 'Rate limit exceeded. Please slow down (60 requests per minute).')
 
-@app.route('/global/<provider>')
-@limiter.limit("60/minute")
-def global_emotes(provider):
+@app.route('/')
+def index():
+ return jsonify({
+ 'global': '/v1/global/<provider>',
+ 'channel': '/v1/channel/<username>/<provider>',
+ 'providers': ['twitch', '7tv', 'bttv', 'ffz', 'all'],
+ })
+
+@cached(cache)
+def get_channel(login):
+ return Channel(login)
+
+def get_emotes(login, provider):
+ channel = get_channel(login)
+
  if provider == 'twitch':
- print('Twitch emotes requested.')
- asyncio.run(update_twitch_global_emotes())
- return jsonify([e.toDict() for e in global_emotes_twitch])
+ return jsonify(channel.getTwitchEmotes())
+ elif provider == '7tv':
+ return jsonify(channel.getSevenTVEmotes())
+ elif provider == 'bttv':
+ return jsonify(channel.getBTTVEmotes())
+ elif provider == 'ffz':
+ return jsonify(channel.getFFZEmotes())
+ elif provider == 'all':
+ return jsonify(channel.getEmotes())
 
  return custom_error(404, 'Provider not found.')
 
+@app.route('/v1/global/<provider>')
+@limiter.limit("60/minute")
+def global_emotes(provider):
+ return get_emotes('_global', provider)
+
+@app.route('/v1/channel/<login>/<provider>')
+@limiter.limit("60/minute")
+def channel_emotes(login, provider):
+ return get_emotes(login, provider)
+
 if __name__ == '__main__':
  gunicorn_logger = logging.getLogger('gunicorn.error')
  app.logger.handlers = gunicorn_logger.handlers

captain-definition

@@ -1,12 +1,4 @@
 {
- "schemaVersion" :2 ,
- "dockerfileLines" :[
- "FROM python:3.7.13-alpine",
- "COPY ./ /app",
- "WORKDIR /app",
- "RUN ls -a",
- "RUN pip3 install -r requirements.txt",
- "CMD [ \"gunicorn\", \"--preload\", \"--bind\", \"0.0.0.0:8000\", \"app:app\" ]"
- ]
+ "schemaVersion": 2,
+ "dockerfilePath": "./Dockerfile"
 }
-

models/channel.py

@@ -0,0 +1,57 @@
+from dotenv import load_dotenv
+from twitchAPI.twitch import Twitch
+import asyncio
+import os
+import threading
+import providers.twitch
+import providers.seventv
+import providers.bttv
+import providers.ffz
+
+load_dotenv()
+TWITCH_CLIENT_ID = os.getenv('TWITCH_CLIENT_ID')
+TWITCH_SECRET = os.getenv('TWITCH_SECRET')
+
+class Channel:
+ async def getUserId(self):
+ twitch = await Twitch(TWITCH_CLIENT_ID, TWITCH_SECRET)
+ async for user in twitch.get_users(logins=self.login):
+ self.user_id = user.id
+ print(f'[{self.login}] Found user: {self.user_id}')
+
+ def __init__(self, login):
+ self.twitch_emotes = []
+ self.twitch_emotes_updated = 0
+ self.twitch_lock = threading.Lock()
+ self.seventv_emotes = []
+ self.seventv_emotes_updated = 0
+ self.seventv_lock = threading.Lock()
+ self.bttv_emotes = []
+ self.bttv_emotes_updated = 0
+ self.bttv_lock = threading.Lock()
+ self.ffz_emotes = []
+ self.ffz_emotes_updated = 0
+ self.ffz_lock = threading.Lock()
+ self.login = login
+
+ if login != '_global':
+ asyncio.run(self.getUserId())
+
+ def getTwitchEmotes(self):
+ return providers.twitch.getEmotes(self)
+
+ def getSevenTVEmotes(self):
+ return providers.seventv.getEmotes(self)
+
+ def getBTTVEmotes(self):
+ return providers.bttv.getEmotes(self)
+
+ def getFFZEmotes(self):
+ return providers.ffz.getEmotes(self)
+
+ def getEmotes(self):
+ twitch_emotes = self.getTwitchEmotes()
+ seventv_emotes = self.getSevenTVEmotes()
+ bttv_emotes = self.getBTTVEmotes()
+ ffz_emotes = self.getFFZEmotes()
+ return twitch_emotes + seventv_emotes + bttv_emotes + ffz_emotes

models/emotes.py

@@ -0,0 +1,27 @@
+class EmoteUrl:
+ def __init__(self, size, url):
+ self.size = size
+ self.url = url
+
+ def toDict(self):
+ return dict(size=self.size, url=self.url)
+
+class Emote:
+ def __init__(self, code, provider, zero_width, animated):
+ self.code = code
+ self.provider = provider
+ self.zero_width = zero_width
+ self.animated = animated
+ self.urls = []
+
+ def toDict(self):
+ return dict(
+ code=self.code,
+ provider=self.provider,
+ zero_width=self.zero_width,
+ animated = self.animated,
+ urls=[u.toDict() for u in self.urls]
+ )
+
+ def addUrl(self, size, url):
+ self.urls.append(EmoteUrl(size, url))