Commit 7854701
Changed files (6)
src
src/ai/summary.py
@@ -20,7 +20,6 @@ from ai.utils import deep_merge
from config import AI, DB, DOWNLOAD_DIR, PREFIX
from database.r2 import set_cf_r2
from messages.help import social_media_help
-from messages.parser import parse_msg
from messages.sender import send2tg
from messages.utils import equal_prefix, set_reaction, startswith_prefix
from networking import download_file
@@ -71,8 +70,7 @@ async def ai_summary(client: Client, message: Message, summary_model_id: str = A
this_msg = message
if equal_prefix(message.content, PREFIX.AI_SUMMARY):
if not message.reply_to_message:
- info = parse_msg(message, use_cache=False)
- await send2tg(client, message, texts=social_media_help(info["cid"], info["ctype"]), **kwargs)
+ await send2tg(client, message, texts=social_media_help(message), **kwargs)
return
message = message.reply_to_message
models = await get_config_by_model_alias(summary_model_id, fallback_to_default=False)
src/messages/help.py
@@ -1,12 +1,14 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
+from pyrogram.types import Message
+
from config import PREFIX
-from permission import check_service
+from permission import set_permission
-def social_media_help(chat_id: int | str, ctype: str):
+def social_media_help(message:Message):
"""Get the help message for social media preview."""
- permission = check_service(cid=chat_id, ctype=ctype)
+ permission = set_permission(message)
msg = f"🔗**链接解析**: {PREFIX.SOCIAL_MEDIA}\n🔄使用 `/retry` 回复消息强制重试"
if permission["twitter"]:
msg += "\n🕊推特"
src/messages/main.py
@@ -234,8 +234,8 @@ async def preview_social_media(
if equal_prefix(this_texts, prefix=[PREFIX.SOCIAL_MEDIA, "/help", "/retry"]):
# without reply, send docs if message only contains prefix command
if not message.reply_to_message:
+ docs = social_media_help(message)
await delete_message(message)
- docs = social_media_help(info["cid"], info["ctype"])
helps = await send2tg(client, message, texts=docs, **kwargs)
await asyncio.sleep(30)
return await delete_message(helps)
src/config.py
@@ -34,52 +34,7 @@ NUM_YOUTUBE_SEARCH_RESULTS = int(os.getenv("NUM_YOUTUBE_SEARCH_RESULTS", "10"))
NUM_GOOGLE_SEARCH_RESULTS = int(os.getenv("NUM_GOOGLE_SEARCH_RESULTS", "10")) # Number of google search results
GOOGLE_SEARCH_GL = os.getenv("GOOGLE_SEARCH_GL", "cn") # "gl" parameter (Geolocation)
CLEAN_OLD_FILES_OLDER_THAN_SECONDS = int(os.getenv("CLEAN_OLD_FILES_OLDER_THAN_SECONDS", "7200"))
-
-
-class ENABLE: # see fine-grained permission in `src/permission.py`
- AI = os.getenv("ENABLE_AI", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- ASR = os.getenv("ENABLE_ASR", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- AUDIO = os.getenv("ENABLE_AUDIO", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- CRONTAB = os.getenv("ENABLE_CRONTAB", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- DOUYIN = os.getenv("ENABLE_DOUYIN", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- INSTAGRAM = os.getenv("ENABLE_INSTAGRAM", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- OCR = os.getenv("ENABLE_OCR", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- HISTORY = os.getenv("ENABLE_HISTORY", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- PRICE = os.getenv("ENABLE_PRICE", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- SEARCH_YOUTUBE = os.getenv("ENABLE_SEARCH_YOUTUBE", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- SEARCH_GOOGLE = os.getenv("ENABLE_SEARCH_GOOGLE", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- SUBTITLE = os.getenv("ENABLE_SUBTITLE", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- TIKTOK = os.getenv("ENABLE_TIKTOK", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- TWITTER = os.getenv("ENABLE_TWITTER", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- WEIBO = os.getenv("ENABLE_WEIBO", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- WECHAT = os.getenv("ENABLE_WECHAT", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- REDDIT = os.getenv("ENABLE_REDDIT", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- V2EX = os.getenv("ENABLE_V2EX", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- ARXIV = os.getenv("ENABLE_ARXIV", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- WGET = os.getenv("ENABLE_WGET", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- GITHUB = os.getenv("ENABLE_GITHUB", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- MUSIC163 = os.getenv("ENABLE_MUSIC163", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- SPOTIFY = os.getenv("ENABLE_SPOTIFY", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- XHS = os.getenv("ENABLE_XHS", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- YTDLP = os.getenv("ENABLE_YTDLP", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- YTDLP_BILIBILI = os.getenv("ENABLE_YTDLP_BILIBILI", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- YTDLP_YOUTUBE = os.getenv("ENABLE_YTDLP_YOUTUBE", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- RAW_IMG_CONVERT = os.getenv("ENABLE_RAW_IMG_CONVERT", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- GROUPS = os.getenv("ENABLE_GROUPS", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- CHANNELS = os.getenv("ENABLE_CHANNELS", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- BOTS = os.getenv("ENABLE_BOTS", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- USERS = os.getenv("ENABLE_USERS", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- SEND_AS_REPLY = os.getenv("ENABLE_SEND_AS_REPLY", "1").lower() in ["1", "y", "yes", "t", "true", "on"] # Send as a reply to the original message
- CACHE_PRICE_SYMBOLS = os.getenv("ENABLE_CACHE_PRICE_SYMBOLS", "0").lower() in ["1", "y", "yes", "t", "true", "on"]
- QUERY_DANMU = os.getenv("ENABLE_QUERY_DANMU", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- FAVORITE = os.getenv("ENABLE_FAVORITE", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- TTS = os.getenv("ENABLE_TTS", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- CONVERT_CHINESE = os.getenv("ENABLE_CONVERT_CHINESE", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- QUOTLY = os.getenv("ENABLE_QUOTLY", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- TMDB = os.getenv("ENABLE_TMDB", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- FFMPEG = os.getenv("ENABLE_FFMPEG", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- WATERMARK = os.getenv("ENABLE_WATERMARK", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
- VERSION = os.getenv("ENABLE_VERSION", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
+ENABLE_CRONTAB = os.getenv("ENABLE_CRONTAB", "1").lower() in ["1", "y", "yes", "t", "true", "on"]
class PREFIX:
src/main.py
@@ -22,15 +22,14 @@ from ai.utils import clean_anthropic_files, clean_gemini_files
from bridge.chartimg import forward_chartimg_results
from bridge.ocr import forward_ocr_results
from bridge.social import forward_social_media_results
-from config import DAILY_MESSAGES, DEVICE_NAME, ENABLE, PROXY, TOKEN, TZ, cache
+from config import DAILY_MESSAGES, DEVICE_NAME, ENABLE_CRONTAB, PROXY, TOKEN, TZ, cache
from danmu.sync import sync_livechats
from database.r2 import clean_r2_expired
from history.sync import backup_chat_history, sync_chat_history
from messages.main import process_message
from messages.parser import parse_msg
-from permission import check_permission
+from permission import set_permission
from podcast.main import summary_pods
-from price.entrypoint import match_symbol_category
from utils import cleanup_old_files, to_int
@@ -52,21 +51,21 @@ async def main():
@app.on_message(filters.group)
async def groups(client: Client, message: Message):
- permission = await check_permission(client, message)
+ permission = set_permission(message)
if permission["disabled"]:
return
await process_message(client, message, **permission)
@app.on_message(filters.channel)
async def channels(client: Client, message: Message):
- permission = await check_permission(client, message)
+ permission = set_permission(message)
if permission["disabled"]:
return
await process_message(client, message, **permission)
@app.on_message(filters.bot)
async def bots(client: Client, message: Message):
- permission = await check_permission(client, message)
+ permission = set_permission(message)
if permission["disabled"]:
return
parse_msg(message, verbose=True)
@@ -83,7 +82,7 @@ async def main():
if ctype == "BOT":
await bots(client, message) # handle bot messages
return
- permission = await check_permission(client, message)
+ permission = set_permission(message)
if permission["disabled"]:
return
parse_msg(message, verbose=True)
@@ -94,8 +93,7 @@ async def main():
@app.on_deleted_messages(group=1)
async def save_history(client: Client, message: Message | list[Message]):
await sync_chat_history(client, message)
-
- if ENABLE.CRONTAB:
+ if ENABLE_CRONTAB:
scheduler = AsyncIOScheduler(timezone=TZ)
scheduler.add_job(cron_secondly, "interval", args=[app], seconds=1)
scheduler.add_job(cron_minutely, "cron", args=[app], second=0)
@@ -125,8 +123,6 @@ async def cron_hourly(client: Client):
await clean_gemini_files()
await clean_r2_expired()
await sync_livechats()
- if ENABLE.CACHE_PRICE_SYMBOLS:
- await match_symbol_category() # to cache all supported symbols
await summary_pods(client)
src/permission.py
@@ -1,121 +1,51 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
-import contextlib
+"""Set permissions via environment variables.
+
+CTYPE: Chat Type. Enum: {GROUP, CHANNEL, BOT, PRIVATE}
+CID: Chat ID
+TID: Thread ID
+UID: User ID
+
+CONFIG_GLOBAL: json string config for global permission
+CONFIG_C{CID}: json string config for {CID}
+CONFIG_C{CID}_T{TID}: json string config for {CID}-{TID}
+CONFIG_U{UID}: json string config for {UID}
+CONFIG_C{CID}_U{UID}: json string config for {CID} and {UID}
+CONFIG_C{CID}_T{TID}_U{UID}: json string config for {CID}-{TID} and {UID}
+
+If there are multiple envvars for the same message, they will be applied in the order of:
+Global Level -> Chat Type Level -> Chat Level -> Chat_Thread Level -> User Level -> Chat_User Level -> Chat_Thread_User Level
+"""
+
+import json
import os
+from glom import glom
from loguru import logger
-from pyrogram.client import Client
from pyrogram.types import Message
-from config import ENABLE, TID, cache
+from config import cache
from messages.modify import message_modify
-from utils import i_am_bot, slim_cid, strings_list, to_int, true
+from messages.parser import get_thread_id
+from utils import slim_cid
-async def check_permission(client: Client, message: Message) -> dict:
- """Check if the user has permission to use the bot."""
+def set_permission(message: Message) -> dict:
+ """Set permissions for this message."""
message = message_modify(message)
ctype = message.chat.type.name if message.chat and message.chat.type else ""
ctype = ctype.upper().removeprefix("SUPER") # SUPERGROUP -> GROUP
-
- # check permission per category
- permission = await check_category(client, message, ctype)
-
- # check permission per service
- permission |= check_service(cid=message.chat.id, ctype=ctype)
-
- # skip for service message
- if message.service:
- permission["disabled"] = True
- return permission
-
-
-async def check_category(client: Client, message: Message, ctype: str) -> dict:
- # ruff: noqa: SIM114
- permission = {"disabled": False}
- cid = slim_cid(message.chat.id)
- if ctype == "GROUP" and not ENABLE.GROUPS:
- permission["disabled"] = True
- elif ctype == "CHANNEL" and not ENABLE.CHANNELS:
- permission["disabled"] = True
- elif ctype == "BOT" and not ENABLE.BOTS:
- permission["disabled"] = True
- elif ctype == "PRIVATE" and not ENABLE.USERS:
- permission["disabled"] = True
- is_bot = await i_am_bot(client)
- """Mark as read for these cid
- TID_MUTES=111111,234567
- TID_MUTE_111111=true
- """
- if not is_bot and (cid in [slim_cid(x) for x in strings_list(os.getenv("TID_MUTES"))] or true(os.getenv(f"TID_MUTE_{cid}"))):
- await message.read()
-
- """Skip process these chats
- TID_SKIPS=111111,234567
- TID_SKIP_111111=true
- """
- if cid in [slim_cid(x) for x in strings_list(os.getenv("TID_SKIPS"))] or true(os.getenv(f"TID_SKIP_{cid}")):
- permission["disabled"] = True
-
- """Only process these chats
- TID_ONLY_GROUPS=111111,234567,-100234567
- TID_ONLY_USERS=111111,234567,-100234567
- """
- if os.getenv(f"TID_ONLY_{ctype}S") and cid not in [slim_cid(x) for x in strings_list(os.getenv(f"TID_ONLY_{ctype}S"))]:
- permission["disabled"] = True
-
- """Whitelist mode for users, if a user is not in the whitelist, skip process
- TID_WHITELIST_MODE=true # enable whitelist mode for users
- TID_WHITELIST_USERS=111111,234567 # these are allowed chats
- TID_WHITELIST_USERS_IN_CHATS=111111,234567,-100234567 # also allow users in these chats
- """
- if ctype == "PRIVATE" and true(os.getenv("TID_WHITELIST_MODE")):
- permission["disabled"] = True
- if cid in [slim_cid(x) for x in strings_list(os.getenv("TID_WHITELIST_USERS"))]:
- permission["disabled"] = False
- # check if user is a member of these chats
- with contextlib.suppress(Exception):
- for chat_id in [int(x.strip()) for x in strings_list(os.getenv("TID_WHITELIST_USERS_IN_CHATS")) if x.strip()]:
- chat_id = to_int(f"-100{slim_cid(chat_id)}") # noqa: PLW2901
- if await client.get_chat_member(chat_id, cid):
- permission["disabled"] = False
- if permission["disabled"]:
- await message.reply_text(f"⚠️Please contact {TID.ADMIN} to use this bot\n⚠️请联系 {TID.ADMIN} 获得使用权限")
- # finally, set for specific cid: ENABLE_1234567=true
- if true(os.getenv(f"ENABLE_{cid}")):
- permission["disabled"] = False
- return permission
-
-
-@cache.memoize(ttl=0)
-def global_permissions() -> dict:
- """Set permissions for all chats.
-
- GLOBAL_YTDLP_SEND_AUDIO=0 # disable ytdlp_send_audio
- GLOBAL_TWITTER_PROVIDER=vxtwitter-fxtwitter # set twitter provider to `vxtwitter-fxtwitter`
- """
- envs = [x for x in os.environ if x.upper().startswith("GLOBAL_")]
- permission = {}
- for key in envs:
- value = os.environ[key]
- option = key.removeprefix("GLOBAL_").lower()
- permission[option] = to_bool(value)
- logger.warning(f"Set `{option}` to {to_bool(value)}")
- logger.success(f"Global permission: {permission}")
- return permission
-
-
-@cache.memoize(ttl=0)
-def check_service(cid: int | str, ctype: str) -> dict:
- if not cid or not ctype:
- return {}
- cid = slim_cid(cid)
-
- permission = {
- # default to False
- "prepend_sender_user": False,
- # default to True
- "need_prefix": True,
+ cid = glom(message, "chat.id", default=0) or 0
+ cid = slim_cid(cid) # remove -100 prefix
+ uid = glom(message, "from_user.id", default=0) or 0
+ tid = get_thread_id(message)
+
+ # Default permission
+ permissions = {
+ "disabled": False, # this is the switch for all permissions
+ "prepend_sender_user": False, # prepend: 👤[@username](tg://user?id={uid})//
+ "need_prefix": True, # need /dl for social media preview
"ai": True,
"asr": True,
"audio_extract": True,
@@ -153,102 +83,142 @@ def check_service(cid: int | str, ctype: str) -> dict:
"tmdb": True,
"ffmpeg": True,
"watermark": True,
- } | global_permissions()
-
+ }
+ # Customization
+ # skip for service message
+ if message.service:
+ permissions["disabled"] = True
if ctype == "PRIVATE":
- permission["need_prefix"] = False
-
- # global service permission
- if not ENABLE.TWITTER:
- permission["twitter"] = False
- if not ENABLE.WEIBO:
- permission["weibo"] = False
- if not ENABLE.XHS:
- permission["xhs"] = False
- if not ENABLE.GITHUB:
- permission["github"] = False
- if not ENABLE.MUSIC163:
- permission["music163"] = False
- if not ENABLE.SPOTIFY:
- permission["spotify"] = False
- if not ENABLE.DOUYIN:
- permission["douyin"] = False
- if not ENABLE.TIKTOK:
- permission["tiktok"] = False
- if not ENABLE.INSTAGRAM:
- permission["instagram"] = False
- if not ENABLE.WECHAT:
- permission["wechat"] = False
- if not ENABLE.V2EX:
- permission["v2ex"] = False
- if not ENABLE.REDDIT:
- permission["reddit"] = False
- if not ENABLE.YTDLP:
- permission["ytdlp"] = False
- if not ENABLE.YTDLP_BILIBILI:
- permission["ytdlp_bilibili"] = False
- if not ENABLE.YTDLP_YOUTUBE:
- permission["ytdlp_youtube"] = False
- if not ENABLE.ARXIV:
- permission["arxiv"] = False
- if not ENABLE.AI:
- permission["ai"] = False
- if not ENABLE.ASR:
- permission["asr"] = False
- if not ENABLE.AUDIO:
- permission["audio_extract"] = False
- if not ENABLE.SUBTITLE:
- permission["subtitle"] = False
- if not ENABLE.SEARCH_YOUTUBE:
- permission["ytb"] = False
- if not ENABLE.SEARCH_GOOGLE:
- permission["google_search"] = False
- if not ENABLE.WGET:
- permission["wget"] = False
- if not ENABLE.OCR:
- permission["ocr"] = False
- if not ENABLE.PRICE:
- permission["price"] = False
- if not ENABLE.RAW_IMG_CONVERT:
- permission["convert_img"] = False
- if not ENABLE.QUERY_DANMU:
- permission["danmu"] = False
- if not ENABLE.HISTORY:
- permission["history"] = False
- if not ENABLE.FAVORITE:
- permission["favorite"] = False
- if not ENABLE.TTS:
- permission["tts"] = False
- if not ENABLE.CONVERT_CHINESE:
- permission["convert_chinese"] = False
- if not ENABLE.QUOTLY:
- permission["quotly"] = False
- if not ENABLE.TMDB:
- permission["tmdb"] = False
- if not ENABLE.FFMPEG:
- permission["ffmpeg"] = False
- if not ENABLE.WATERMARK:
- permission["watermark"] = False
- if not ENABLE.VERSION:
- permission["version"] = False
+ permissions["need_prefix"] = False
+
+ # Set permissions via environment variables
+ permissions |= global_permissions() # CONFIG_GLOBAL
+ permissions |= chat_type_permissions(ctype) # CONFIG_GROUP/CHANNEL/BOT/PRIVATE = 0
+ permissions |= chat_permissions(cid) # CONFIG_C{CID} = {"ai": false}
+ permissions |= chat_thread_permissions(cid, tid) # CONFIG_C{CID}_T{TID} = {"ai": false}
+ permissions |= user_permissions(uid) # CONFIG_U{UID} = {"ai": false}
+
+ permissions |= chat_user_permissions(cid, uid) # CONFIG_C{CID}_U{UID} = {"ai": false}
+ permissions |= chat_thread_user_permissions(cid, tid, uid) # CONFIG_C{CID}_T{TID}_U{UID} = {"ai": false}
+ return permissions
+
+@cache.memoize(ttl=0)
+def chat_type_permissions(ctype: str) -> dict:
+ """Set chat type permissions."""
+ if not os.getenv(f"CONFIG_{ctype}"):
+ return {}
+ try:
+ permission = json.loads(os.environ[f"CONFIG_{ctype}"])
+ except json.JSONDecodeError:
+ logger.error(f"Invalid JSON in CONFIG_{ctype} config")
+ return {}
+ logger.success(f"ChatType={ctype} permissions: {permission}")
+ return permission
+
+
+@cache.memoize(ttl=0)
+def global_permissions() -> dict:
+ """Set global permissions.
+
+ CONFIG_GLOBAL={"prepend_sender_user": false, "ai": false, "twitter_provider": "vxtwitter-fxtwitter"}
+ # set prepend_sender_user to False
+ # set ai to False
+ # set twitter_provider to "vxtwitter-fxtwitter"
"""
- Set for specific chat
- SET_111111_AI=1
- SET_111111_DOUYIN=0
- SET_111111_DOUYIN_PROVIDER=tikhub
+ if not os.getenv("CONFIG_GLOBAL"):
+ return {}
+ try:
+ permission = json.loads(os.environ["CONFIG_GLOBAL"])
+ except json.JSONDecodeError:
+ logger.error("Invalid JSON in CONFIG_GLOBAL config")
+ return {}
+ logger.success(f"Global permissions: {permission}")
+ return permission
+
+
+@cache.memoize(ttl=0)
+def chat_permissions(cid: str) -> dict:
+ """Set chat permissions.
+
+ CONFIG_C{CID}={"ai": false, "twitter_provider": "vxtwitter-fxtwitter"}
+ # config for chat id: {CID}
"""
- envs = [x for x in os.environ if x.upper().startswith(f"SET_{cid}_")]
- for key in envs:
- value = os.environ[key]
- option = key.removeprefix(f"SET_{cid}_").lower()
- permission[option] = to_bool(value)
- logger.warning(f"Set `{option}` for chat={cid} to {to_bool(value)}")
- logger.success(f"Permission for chat={cid}: {permission}")
+ if not os.getenv(f"CONFIG_C{cid}"):
+ return {}
+ try:
+ permission = json.loads(os.environ[f"CONFIG_C{cid}"])
+ except json.JSONDecodeError:
+ logger.error(f"Invalid JSON in CONFIG_C{cid} config")
+ return {}
+ logger.success(f"CID={cid} permissions: {permission}")
return permission
-def to_bool(v: str) -> bool | str:
- if str(v).lower() in {"1", "true", "t", "yes", "y", "on", "0", "n", "no", "f", "false", "off"}:
- return true(v)
- return v
+@cache.memoize(ttl=0)
+def chat_thread_permissions(cid: str, tid: str) -> dict:
+ """Set chat thread permissions.
+
+ CONFIG_C{CID}_T{TID}={"ai": false, "twitter_provider": "vxtwitter-fxtwitter"}
+ # config for chat {CID}, thread id: {TID}
+ """
+ if not os.getenv(f"CONFIG_C{cid}_T{tid}"):
+ return {}
+ try:
+ permission = json.loads(os.environ[f"CONFIG_C{cid}_T{tid}"])
+ except json.JSONDecodeError:
+ logger.error(f"Invalid JSON in CONFIG_C{cid}_T{tid} config")
+ return {}
+ logger.success(f"CID={cid}_TID={tid} permissions: {permission}")
+ return permission
+
+
+def user_permissions(uid: str | int) -> dict:
+ """Set user permissions.
+
+ CONFIG_U{UID}={"ai": false, "twitter_provider": "vxtwitter-fxtwitter"}
+ # config for user {UID}
+ """
+ if not os.getenv(f"CONFIG_U{uid}"):
+ return {}
+ try:
+ permission = json.loads(os.environ[f"CONFIG_U{uid}"])
+ except json.JSONDecodeError:
+ logger.error(f"Invalid JSON in CONFIG_U{uid} config")
+ return {}
+ logger.success(f"UID={uid} permissions: {permission}")
+ return permission
+
+
+def chat_user_permissions(cid: str, uid: str | int) -> dict:
+ """Set chat user permissions.
+
+ CONFIG_C{CID}_U{UID}={"ai": false, "twitter_provider": "vxtwitter-fxtwitter"}
+ # config for chat {CID}, user {UID}
+ """
+ if not os.getenv(f"CONFIG_C{cid}_U{uid}"):
+ return {}
+ try:
+ permission = json.loads(os.environ[f"CONFIG_C{cid}_U{uid}"])
+ except json.JSONDecodeError:
+ logger.error(f"Invalid JSON in CONFIG_C{cid}_U{uid} config")
+ return {}
+ logger.success(f"CID={cid}_UID={uid} permissions: {permission}")
+ return permission
+
+
+def chat_thread_user_permissions(cid: str, tid: str | int, uid: str | int) -> dict:
+ """Set chat thread user permissions.
+
+ CONFIG_C{CID}_T{TID}_U{UID}={"ai": false, "twitter_provider": "vxtwitter-fxtwitter"}
+ # config for chat {CID}, thread id: {TID}, user {UID}
+ """
+ if not os.getenv(f"CONFIG_C{cid}_T{tid}_U{uid}"):
+ return {}
+ try:
+ permission = json.loads(os.environ[f"CONFIG_C{cid}_T{tid}_U{uid}"])
+ except json.JSONDecodeError:
+ logger.error(f"Invalid JSON in CONFIG_C{cid}_T{tid}_U{uid} config")
+ return {}
+ logger.success(f"CID={cid}_TID={tid}_UID={uid} permissions: {permission}")
+ return permission