+
+
+ ⚡ The next-generation emby management bot +
+ +
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
{usage}"
+ message,
+ f"参数不足，请参考用法：\n{usage}",
)
return
# 将解析好的参数传递给目标函数，避免在函数内部再调用 _parse_args
@@ -103,7 +104,9 @@ async def create_user(self, message: Message, args: list[str]):
f"✅ 创建用户成功。\n初始密码：{default_password}",
)
else:
- await self._reply_html(message, "❌ 创建用户失败，请稍后重试。")
+ await self._reply_html(
+ message, "❌ 创建用户失败，请稍后重试。"
+ )
except Exception as e:
await self._send_error(message, e, prefix="创建用户失败")
@@ -112,7 +115,9 @@ async def info(self, message: Message):
/info
如果是私聊，查看自己信息；如果群里回复某人，则查看对方信息
"""
- telegram_id = await get_user_telegram_id(self.bot_client.client, message)
+ telegram_id = await get_user_telegram_id(
+ self.bot_client.client, message
+ )
try:
user, emby_info = await self.user_service.emby_info(telegram_id)
last_active = (
@@ -124,7 +129,9 @@ async def info(self, message: Message):
emby_info.get("DateCreated", "")
)
ban_status = (
- "正常" if (user.ban_time is None or user.ban_time == 0) else "已禁用"
+ "正常"
+ if (user.ban_time is None or user.ban_time == 0)
+ else "已禁用"
)
reply_text = (
@@ -167,7 +174,9 @@ async def use_code(self, message: Message, args: list[str]):
message, "✅ 邀请码使用成功，您已获得创建账号资格"
)
else:
- await self._reply_html(message, "✅ 邀请码使用成功，您已获得白名单资格")
+ await self._reply_html(
+ message, "✅ 邀请码使用成功，您已获得白名单资格"
+ )
# 如果该邀请码在bot中记录了消息，需要删除
if self.code_to_message_id.get(code):
@@ -193,7 +202,9 @@ async def reset_emby_password(self, message: Message):
f"✅ 密码重置成功。\n新密码：{default_password}",
)
else:
- await self._reply_html(message, "❌ 密码重置失败，请稍后重试。")
+ await self._reply_html(
+ message, "❌ 密码重置失败，请稍后重试。"
+ )
except Exception as e:
await self._send_error(message, e, prefix="密码重置失败")
@@ -217,7 +228,9 @@ async def new_code(self, message: Message):
message.from_user.id, num
)
for code_obj in code_list:
- message_text = f"📌 邀请码：\n点击复制👉{code_obj.code}"
+ message_text = (
+ f"📌 邀请码：\n点击复制👉{code_obj.code}"
+ )
if message.reply_to_message is not None:
await self.bot_client.client.send_message(
chat_id=message.from_user.id,
@@ -232,7 +245,10 @@ async def new_code(self, message: Message):
await self._reply_html(message, "✅ 已发送邀请码")
else:
msg = await self._reply_html(message, message_text)
- self.code_to_message_id[code_obj.code] = (message.chat.id, msg.id)
+ self.code_to_message_id[code_obj.code] = (
+ message.chat.id,
+ msg.id,
+ )
except Exception as e:
await self._send_error(message, e, prefix="创建邀请码失败")
@@ -273,7 +289,10 @@ async def new_whitelist_code(self, message: Message):
await self._reply_html(message, "✅ 已发送邀请码")
else:
msg = await self._reply_html(message, message_text)
- self.code_to_message_id[code_obj.code] = (message.chat.id, msg.id)
+ self.code_to_message_id[code_obj.code] = (
+ message.chat.id,
+ msg.id,
+ )
except Exception as e:
await self._send_error(message, e, prefix="创建白名单邀请码失败")
@@ -285,11 +304,16 @@ async def ban_emby(self, message: Message):
reason = args[0] if args else "管理员禁用"
operator_id = message.from_user.id
- telegram_id = await get_user_telegram_id(self.bot_client.client, message)
+ telegram_id = await get_user_telegram_id(
+ self.bot_client.client, message
+ )
try:
- if await self.user_service.emby_ban(telegram_id, reason, operator_id):
+ if await self.user_service.emby_ban(
+ telegram_id, reason, operator_id
+ ):
await self._reply_html(
- message, f"✅ 已禁用用户 {telegram_id} 的Emby账号"
+ message,
+ f"✅ 已禁用用户 {telegram_id} 的Emby账号",
)
else:
await self._reply_html(message, "❌ 禁用失败，请稍后重试。")
@@ -301,11 +325,14 @@ async def unban_emby(self, message: Message):
/unban_emby (群里需回复某人或手动指定)
"""
operator_id = message.from_user.id
- telegram_id = await get_user_telegram_id(self.bot_client.client, message)
+ telegram_id = await get_user_telegram_id(
+ self.bot_client.client, message
+ )
try:
if await self.user_service.emby_unban(telegram_id, operator_id):
await self._reply_html(
- message, f"✅ 已解禁用户 {telegram_id} 的Emby账号"
+ message,
+ f"✅ 已解禁用户 {telegram_id} 的Emby账号",
)
else:
await self._reply_html(message, "❌ 解禁失败，请稍后重试。")
@@ -319,8 +346,9 @@ async def select_line(self, message: Message):
"""
try:
telegram_id = message.from_user.id
- router_list = config.router_list or await self.user_service.get_router_list(
- telegram_id
+ router_list = (
+ config.router_list
+ or await self.user_service.get_router_list(telegram_id)
)
# 缓存到 config 中，减少重复获取
if router_list and not config.router_list:
@@ -328,7 +356,9 @@ async def select_line(self, message: Message):
user_router = await self.user_service.get_user_router(telegram_id)
user_router_index = user_router.get("index", "")
- message_text = f"当前线路：{user_router_index}\n请选择线路："
+ message_text = (
+ f"当前线路：{user_router_index}\n请选择线路："
+ )
message_buttons = []
for router in router_list:
@@ -336,7 +366,9 @@ async def select_line(self, message: Message):
name = router.get("name")
# 已选线路高亮
button_text = (
- f"🔵 {name}" if index == user_router_index else f"⚪ {name}"
+ f"🔵 {name}"
+ if index == user_router_index
+ else f"⚪ {name}"
)
message_buttons.append(
[
@@ -347,7 +379,9 @@ async def select_line(self, message: Message):
)
keyboard = InlineKeyboardMarkup(message_buttons)
- await self._reply_html(message, message_text, reply_markup=keyboard)
+ await self._reply_html(
+ message, message_text, reply_markup=keyboard
+ )
except Exception as e:
await self._send_error(message, e, prefix="查询失败")
@@ -371,7 +405,9 @@ async def group_member_change_handler(self, clent, message: Message):
for new_member in message.new_chat_members:
config.group_members[new_member.id] = new_member
- async def handle_callback_query(self, client, callback_query: CallbackQuery):
+ async def handle_callback_query(
+ self, client, callback_query: CallbackQuery
+ ):
"""
回调按钮事件统一处理，如切换线路。
"""
@@ -384,7 +420,8 @@ async def handle_callback_query(self, client, callback_query: CallbackQuery):
return
selected_router = next(
- (r for r in config.router_list if r["index"] == index), None
+ (r for r in config.router_list if r["index"] == index),
+ None,
)
if not selected_router:
await callback_query.answer("线路不存在")
@@ -399,7 +436,9 @@ async def handle_callback_query(self, client, callback_query: CallbackQuery):
"生效可能会有 30 秒延迟，请耐心等候。"
)
except Exception as e:
- await callback_query.answer(f"操作失败：{str(e)}", show_alert=True)
+ await callback_query.answer(
+ f"操作失败：{str(e)}", show_alert=True
+ )
logger.error(f"Callback query failed: {e}", exc_info=True)
async def count(self, message: Message):
@@ -410,7 +449,9 @@ async def count(self, message: Message):
try:
count_data = self.user_service.emby_count()
if not count_data:
- return await self._reply_html(message, "❌ 查询失败：无法获取数据")
+ return await self._reply_html(
+ message, "❌ 查询失败：无法获取数据"
+ )
await self._reply_html(
message,
@@ -435,10 +476,13 @@ async def register_until(self, message: Message, args: list[str]):
time = datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S")
now = datetime.now()
if time < now:
- return await self._reply_html(message, "❌ 时间必须晚于当前时间")
+ return await self._reply_html(
+ message, "❌ 时间必须晚于当前时间"
+ )
await self.user_service.set_emby_config(
- message.from_user.id, register_public_time=int(time.timestamp())
+ message.from_user.id,
+ register_public_time=int(time.timestamp()),
)
await self._reply_html(
message, f"✅ 已开放注册，截止时间：{time_str}"
@@ -513,13 +557,17 @@ async def c_info(client, message):
await self.info(message)
@self.bot_client.client.on_message(
- filters.private & filters.command("use_code") & user_in_group_on_filter
+ filters.private
+ & filters.command("use_code")
+ & user_in_group_on_filter
)
async def c_use_code(client, message):
await self.use_code(message)
@self.bot_client.client.on_message(
- filters.private & filters.command("create") & user_in_group_on_filter
+ filters.private
+ & filters.command("create")
+ & user_in_group_on_filter
)
async def c_create_user(client, message):
await self.create_user(message)
diff --git a/bot/filters.py b/bot/filters.py
index 5284478..5f439ac 100644
--- a/bot/filters.py
+++ b/bot/filters.py
@@ -32,7 +32,8 @@ async def admin_user_on_filter(filter, client, update) -> bool:
return True
except Exception as e:
logger.error(
- f"Error checking admin status for user {telegram_id}: {e}", exc_info=True
+ f"Error checking admin status for user {telegram_id}: {e}",
+ exc_info=True,
)
return False
@@ -50,7 +51,8 @@ async def emby_user_on_filter(filter, client, update) -> bool:
return True
except Exception as e:
logger.error(
- f"Error checking Emby status for user {telegram_id}: {e}", exc_info=True
+ f"Error checking Emby status for user {telegram_id}: {e}",
+ exc_info=True,
)
return False
@@ -58,6 +60,8 @@ async def emby_user_on_filter(filter, client, update) -> bool:
return False
-user_in_group_on_filter = create(user_in_group_on_filter, "user_in_group_on_filter")
+user_in_group_on_filter = create(
+ user_in_group_on_filter, "user_in_group_on_filter"
+)
admin_user_on_filter = create(admin_user_on_filter, "admin_user_on_filter")
emby_user_on_filter = create(emby_user_on_filter, "emby_user_on_filter")
diff --git a/bot/message_helper.py b/bot/message_helper.py
index 4efa283..d250f2b 100644
--- a/bot/message_helper.py
+++ b/bot/message_helper.py
@@ -24,12 +24,16 @@ async def get_user_telegram_id(client, message):
# 直接提供 Telegram ID（纯数字）
if telegram_str.isdigit():
telegram_id = int(telegram_str)
- logger.debug(f"Telegram ID from arguments (numeric): {telegram_id}")
+ logger.debug(
+ f"Telegram ID from arguments (numeric): {telegram_id}"
+ )
# 使用 @username
elif telegram_str.startswith("@"):
telegram_username = telegram_str[1:] # 去掉 `@`
- logger.debug(f"Telegram username from arguments: {telegram_username}")
+ logger.debug(
+ f"Telegram username from arguments: {telegram_username}"
+ )
# 通过用户名查找 ID
if telegram_username:
@@ -46,7 +50,9 @@ async def get_user_telegram_id(client, message):
return None
except PeerIdInvalid:
error_message = f"❌ 无法获取用户 @{telegram_username} 的 ID"
- logger.warning(f"Peer ID invalid for username: {telegram_username}")
+ logger.warning(
+ f"Peer ID invalid for username: {telegram_username}"
+ )
await message.reply(error_message)
return None
except Exception as e:
diff --git a/bot/utils.py b/bot/utils.py
index 98abc64..8ac3804 100644
--- a/bot/utils.py
+++ b/bot/utils.py
@@ -14,7 +14,8 @@ def parse_iso8601(datetime_str: str):
return dt
except Exception as e:
logger.error(
- f"Error parsing ISO8601 datetime string: {datetime_str}: {e}", exc_info=True
+ f"Error parsing ISO8601 datetime string: {datetime_str}: {e}",
+ exc_info=True,
)
return None
@@ -39,5 +40,7 @@ def parse_timestamp_to_normal_date(timestamp: int):
logger.debug(f"Parsed timestamp: {timestamp}")
return dt.strftime("%Y-%m-%d %H:%M:%S")
except Exception as e:
- logger.error(f"Error parsing timestamp {timestamp}: {e}", exc_info=True)
+ logger.error(
+ f"Error parsing timestamp {timestamp}: {e}", exc_info=True
+ )
return None
diff --git a/core/emby_api.py b/core/emby_api.py
index b15ff0c..1b325c4 100644
--- a/core/emby_api.py
+++ b/core/emby_api.py
@@ -53,7 +53,11 @@ def _request(self, method: str, path: str, data=None, params=None):
)
elif method.upper() == "POST":
response = requests.post(
- url, params=params, json=data, timeout=self.timeout, headers=headers
+ url,
+ params=params,
+ json=data,
+ timeout=self.timeout,
+ headers=headers,
)
else:
raise Exception(f"暂不支持的 HTTP 方法: {method}")
@@ -64,18 +68,23 @@ def _request(self, method: str, path: str, data=None, params=None):
raise Exception("请求 Emby 服务器超时，请稍后重试或检查网络连接。")
except requests.exceptions.ConnectionError as e:
# 连接异常
- logger.error(f"Failed to connect to Emby server: {e}", exc_info=True)
+ logger.error(
+ f"Failed to connect to Emby server: {e}", exc_info=True
+ )
raise Exception(f"无法连接到 Emby 服务器: {str(e)}")
except requests.exceptions.RequestException as e:
# 其他 requests 异常
logger.error(
- f"An unknown error occurred while requesting Emby: {e}", exc_info=True
+ f"An unknown error occurred while requesting Emby: {e}",
+ exc_info=True,
)
raise Exception(f"请求 Emby 时发生未知错误: {str(e)}")
try:
response.raise_for_status()
- logger.debug(f"Request successful, status code: {response.status_code}")
+ logger.debug(
+ f"Request successful, status code: {response.status_code}"
+ )
except Exception as e:
logger.error(f"Emby API request failed: {e}", exc_info=True)
raise Exception(f"Emby API 请求失败")
@@ -93,7 +102,8 @@ def get_user(self, emby_id: str):
return self._request("GET", path)
except Exception as e:
logger.error(
- f"Failed to get user with Emby ID {emby_id}: {e}", exc_info=True
+ f"Failed to get user with Emby ID {emby_id}: {e}",
+ exc_info=True,
)
raise
@@ -109,7 +119,9 @@ def create_user(self, name: str):
try:
return self._request("POST", path, data=data)
except Exception as e:
- logger.error(f"Failed to create user with name {name}: {e}", exc_info=True)
+ logger.error(
+ f"Failed to create user with name {name}: {e}", exc_info=True
+ )
raise
def ban_user(self, emby_id: str):
@@ -147,7 +159,8 @@ def ban_user(self, emby_id: str):
return self.update_user_policy(emby_id, data)
except Exception as e:
logger.error(
- f"Failed to ban user with Emby ID {emby_id}: {e}", exc_info=True
+ f"Failed to ban user with Emby ID {emby_id}: {e}",
+ exc_info=True,
)
raise
@@ -303,7 +316,9 @@ def call_api(self, path: str):
:return: 成功时返回 JSON，失败抛出异常
"""
url = f"{self.api_url}{path}"
- headers = {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {}
+ headers = (
+ {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {}
+ )
logger.debug(f"Calling API at {url}")
try:
response = requests.get(url, headers=headers, timeout=self.timeout)
@@ -313,7 +328,9 @@ def call_api(self, path: str):
logger.error("Request to router service timed out", exc_info=True)
raise Exception("请求路由服务超时，请稍后重试或检查网络连接。")
except requests.exceptions.ConnectionError as e:
- logger.error(f"Failed to connect to router service: {e}", exc_info=True)
+ logger.error(
+ f"Failed to connect to router service: {e}", exc_info=True
+ )
raise Exception(f"无法连接到路由服务: {str(e)}")
except requests.exceptions.RequestException as e:
logger.error(
@@ -342,7 +359,8 @@ def query_user_route(self, user_id: str):
return self.call_api(f"/api/route/{user_id}")
except Exception as e:
logger.error(
- f"Failed to query user route for user ID {user_id}: {e}", exc_info=True
+ f"Failed to query user route for user ID {user_id}: {e}",
+ exc_info=True,
)
raise
@@ -350,7 +368,9 @@ def update_user_route(self, user_id: str, new_index: str):
"""
更新用户当前所使用的线路。
"""
- logger.info(f"Updating user route for user ID: {user_id} to index: {new_index}")
+ logger.info(
+ f"Updating user route for user ID: {user_id} to index: {new_index}"
+ )
try:
return self.call_api(f"/api/route/{user_id}/{new_index}")
except Exception as e:
diff --git a/models/config_model.py b/models/config_model.py
index 0b269a2..e2a7db4 100644
--- a/models/config_model.py
+++ b/models/config_model.py
@@ -10,11 +10,15 @@
class Config(BaseOrmTableWithTS):
__tablename__ = "config"
- total_register_user: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+ total_register_user: Mapped[int] = mapped_column(
+ Integer, nullable=False, default=0
+ )
register_public_user: Mapped[int] = mapped_column(
Integer, nullable=False, default=0
)
- register_public_time: Mapped[int] = mapped_column(BigInteger, nullable=True)
+ register_public_time: Mapped[int] = mapped_column(
+ BigInteger, nullable=True
+ )
class ConfigOrm(DBManager):
diff --git a/models/invite_code_model.py b/models/invite_code_model.py
index 3705697..694a65e 100644
--- a/models/invite_code_model.py
+++ b/models/invite_code_model.py
@@ -22,12 +22,18 @@ class InviteCode(BaseOrmTableWithTS):
code: Mapped[str] = mapped_column(
String(50), index=True, unique=True, nullable=False
)
- telegram_id: Mapped[int] = mapped_column(BigInteger, index=True, nullable=False)
+ telegram_id: Mapped[int] = mapped_column(
+ BigInteger, index=True, nullable=False
+ )
code_type: Mapped[InviteCodeType] = mapped_column(
Enum(InviteCodeType), nullable=False
)
- is_used: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
- used_time: Mapped[int] = mapped_column(BigInteger, default=None, nullable=True)
+ is_used: Mapped[bool] = mapped_column(
+ Boolean, default=False, nullable=False
+ )
+ used_time: Mapped[int] = mapped_column(
+ BigInteger, default=None, nullable=True
+ )
used_user_id: Mapped[int] = mapped_column(
BigInteger, default=None, nullable=True, index=True
)
diff --git a/models/user_model.py b/models/user_model.py
index 8d4069a..92acbb9 100644
--- a/models/user_model.py
+++ b/models/user_model.py
@@ -20,8 +20,12 @@ class User(BaseOrmTableWithTS):
emby_id: Mapped[str] = mapped_column(
String(50), index=True, unique=True, nullable=True
)
- is_admin: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
- is_whitelist: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
+ is_admin: Mapped[bool] = mapped_column(
+ Boolean, default=False, nullable=False
+ )
+ is_whitelist: Mapped[bool] = mapped_column(
+ Boolean, default=False, nullable=False
+ )
enable_register: Mapped[bool] = mapped_column(
Boolean, default=False, nullable=False
)
@@ -63,7 +67,9 @@ def check_use_redeem_code(self) -> None:
if self.emby_id is not None:
raise Exception("该用户已拥有 Emby 账号，无法再次使用注册邀请码。")
if self.enable_register:
- raise Exception("该用户已经具备创建 Emby 账号的资格，无需再次使用邀请码。")
+ raise Exception(
+ "该用户已经具备创建 Emby 账号的资格，无需再次使用邀请码。"
+ )
def check_use_whitelist_code(self) -> None:
"""检查是否可使用白名单邀请码。"""
diff --git a/readme.md b/readme.md
deleted file mode 100644
index 382e4d0..0000000
--- a/readme.md
+++ /dev/null
@@ -1,80 +0,0 @@
-## Emby Bot 项目
-本项目是一个基于 Pyrogram 的 Telegram Bot，用于管理 Emby 用户、发送邀请码、查询 Emby 资源等操作。同时集成了针对 Emby 路由服务的切换功能。项目主要通过 Python + SQLAlchemy + Pyrogram 实现。
-
-### 功能概述
-#### 用户管理：
-- 根据邀请码创建 Emby 用户，并分配默认密码、默认策略等。
-- 提供管理员命令禁用/解禁用户的 Emby 账号。
-- 可查看用户当前信息（白名单、管理员身份、禁用状态等）。
-#### 邀请码管理：
-- 生成普通邀请码、白名单邀请码。
-- 使用邀请码后自动更新数据库和相关标识。
-#### 线路管理：
-- 集成路由服务 API，允许用户在机器人对话中快速切换观影线路。
-#### 其他辅助功能：
-- 查看当前 Emby 影片数量。
-- 限时或限量开放注册。
-
-### 安装及运行
-```bash
-git clone https://github.com/embyplus/embyBot
-cp .env.example .env
-vim .env
-
-python3 -m pip install -r requirements.txt
-python3 app.py
-```
-
-### 配置环境变量
-
-| 变量名 | 说明 | 示例值 |
-|-------------------|---------------------------------------------------|----------------------------|
-| TIMEZONE | 时区设置 | Asia/Shanghai |
- | LOG_LEVEL | 日志级别，可选 DEBUG / INFO / WARNING / ERROR / CRITICAL | INFO |
- | BOT_TOKEN | 你的 Telegram Bot 令牌 | 123456:ABC-DEF1234ghIkl... |
- | API_ID | Telegram API ID（从 my.telegram.org 获取） | 1234567 |
- | API_HASH | Telegram API Hash | abcdef1234567890ghijklmn |
- | TELEGRAM_GROUP_ID | Bot 要监听或管理的群组 ID，支持多群可用逗号分隔 | -1001234567890 |
- | EMBY_URL | Emby 服务器 URL | https://your-emby-url |
- | EMBY_API_KEY | Emby 服务器 API Key | embyapikey123 |
- | API_URL | 路由服务 API 基础地址 | https://your-router-api |
- | API_KEY | 路由服务使用的鉴权 token，不需要则可留空 | routerapikey123 |
- | DB_HOST | 数据库主机名或 IP | 127.0.0.1 |
- | DB_PORT | 数据库端口 | 3306 |
- | DB_USER | 数据库用户名 | root |
- | DB_PASS | 数据库密码 | password |
- | DB_NAME | 数据库名 | emby_bot_db |
- | ADMIN_LIST | Bot 管理员的 Telegram ID 列表（用逗号分隔） | 123456789,987654321 |
-
-## 贡献指南
-欢迎贡献代码！为了确保项目的高质量和一致性，请遵循以下贡献规程：
-### 提交规范
-- 提交信息必须符合 Angular 提交信息规范，格式如下：
- - `type(scope): description`
- - 例如：`feat(user): 添加用户注册功能`
-### 与数据库、API 代码相关的更改
-- 请确保所有更改都经过严格的测试，并且不会引入新的错误。
-### 创建 Pull Request
-- 请确保创建 Pull Request 前进行本地测试，确保通过所有 CI/CD 测试。
-### 支持的 Type 列表
-
-| Type | 描述 |
-|----------|-------------------------------------------------------------|
-| feat | 添加新功能，比如新增用户注册、功能扩展等 |
-| fix | 修复 bug 或错误，解决问题的修改 |
-| docs | 文档相关修改，如更新说明文档、README、注释等 |
-| style | 代码格式、标点、空格等修改，不影响代码逻辑运行 |
-| refactor | 代码重构，调整代码结构而不改变功能 |
-| perf | 性能优化修改，提升效率或降低资源消耗 |
-| test | 添加或更新测试代码，保证项目稳定性 |
-| chore | 杂项维护，如依赖更新、构建脚本修改，不涉及代码逻辑 |
-| ci | 持续集成相关修改，如 GitHub Actions 工作流程优化 |
-
----
-感谢您的贡献！
-
-
-### Thanks
-- [Pyrogram](https://docs.pyrogram.org/) - Telegram API for Python
-- [SQLAlchemy](https://www.sqlalchemy.org/) - Python SQL Toolkit and Object-Relational Mapping
-- [Emby服管理bot by小草](https://github.com/xiaocao666tzh/EmbyBot)
\ No newline at end of file
diff --git a/ruff.toml b/ruff.toml
index 17bd4bc..504147a 100644
--- a/ruff.toml
+++ b/ruff.toml
@@ -28,8 +28,8 @@ exclude = [
"site-packages",
]
-# 格式化相关设置，保持与 Black 风格一致
-line-length = 88
+# 格式化相关设置，保持与 PEP8 风格一致
+line-length = 79
indent-width = 4
# 指定目标 Python 版本
@@ -39,8 +39,8 @@ target-version = "py312"
# 启用部分 pycodestyle（E4、E7、E9）、Pyflakes（F），同时增加 bugbear（B）和 flake8-quotes（Q）规则
select = ["E4", "E7", "E9", "F", "B", "Q"]
-# 如果您不希望行长检查导致过多提示，可选择忽略 E501
-ignore = ["E501"]
+# 由于必须严格遵照 PEP8，此处移除了对 E501（行长检查）的忽略
+# ignore = ["E501"]
# 默认允许自动修复所有启用的规则，但 bugbear（B）的规则风险较高，因此将其列入 unfixable，避免自动修改
fixable = ["ALL"]
@@ -59,7 +59,7 @@ dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
docstring-quotes = "double"
[format]
-# 使用双引号（与 Black 一致）
+# 使用双引号
quote-style = "double"
# 使用空格缩进，不使用制表符
indent-style = "space"
@@ -71,4 +71,4 @@ line-ending = "auto"
# Docstring 内代码示例的自动格式化（目前处于实验阶段，如不需要可保持关闭）
docstring-code-format = false
# 当 docstring 格式化启用时，动态计算代码片段的行长
-docstring-code-line-length = "dynamic"
\ No newline at end of file
+docstring-code-line-length = "dynamic"
diff --git a/services/user_service.py b/services/user_service.py
index c460761..dddeda5 100644
--- a/services/user_service.py
+++ b/services/user_service.py
@@ -28,12 +28,16 @@ def __init__(self, emby_api: EmbyApi, emby_router_api: EmbyRouterAPI):
@staticmethod
async def get_or_create_user_by_telegram_id(telegram_id: int) -> User:
"""通过 telegram_id 从数据库获取用户，如果不存在则创建一个默认用户"""
- user = await UserOrm().query_one(conds=[User.telegram_id == telegram_id])
+ user = await UserOrm().query_one(
+ conds=[User.telegram_id == telegram_id]
+ )
if not user:
default_user = User(
telegram_id=telegram_id,
is_admin=telegram_id in config.admin_list,
- telegram_name=config.group_members.get(telegram_id, {}).username
+ telegram_name=config.group_members.get(
+ telegram_id, {}
+ ).username
if config.group_members.get(telegram_id)
else None,
)
@@ -71,7 +75,9 @@ async def _emby_create_user(
user = await self.get_or_create_user_by_telegram_id(telegram_id)
emby_user = self.emby_api.create_user(username)
if not emby_user or not emby_user.get("Id"):
- raise Exception("在 Emby 系统中创建账号失败，请检查 Emby 服务是否正常。")
+ raise Exception(
+ "在 Emby 系统中创建账号失败，请检查 Emby 服务是否正常。"
+ )
emby_id = emby_user["Id"]
user.emby_id = emby_id
@@ -108,7 +114,9 @@ async def create_invite_code(
code_objs = [
InviteCode(
- code=code, telegram_id=telegram_id, code_type=InviteCodeType.REGISTER
+ code=code,
+ telegram_id=telegram_id,
+ code_type=InviteCodeType.REGISTER,
)
for code in self.gen_register_code(count)
]
@@ -124,7 +132,9 @@ async def create_whitelist_code(
code_objs = [
InviteCode(
- code=code, telegram_id=telegram_id, code_type=InviteCodeType.WHITELIST
+ code=code,
+ telegram_id=telegram_id,
+ code_type=InviteCodeType.WHITELIST,
)
for code in self.gen_whitelist_code(count)
]
@@ -147,7 +157,9 @@ async def first_or_create_emby_config(self) -> Config:
emby_config = await ConfigOrm().query_one(conds=[Config.id == 1])
if not emby_config:
emby_config = Config(
- register_public_user=0, register_public_time=0, total_register_user=0
+ register_public_user=0,
+ register_public_time=0,
+ total_register_user=0,
)
await ConfigOrm().add(emby_config)
return emby_config
@@ -158,7 +170,9 @@ async def emby_create_user(
"""创建 Emby 用户（外部调用入口），先判断各种配置是否允许注册，然后调用内部的 _emby_create_user"""
user = await self.get_or_create_user_by_telegram_id(telegram_id)
if user.has_emby_account():
- raise Exception("该 Telegram 用户已经绑定过 Emby 账号，无法重复创建。")
+ raise Exception(
+ "该 Telegram 用户已经绑定过 Emby 账号，无法重复创建。"
+ )
emby_config = await self.first_or_create_emby_config()
if not emby_config:
@@ -168,18 +182,25 @@ async def emby_create_user(
raise Exception("当前没有可用的注册权限或名额，创建账号被拒绝。")
async with ConfigOrm().transaction() as session:
- if not user.enable_register and emby_config.register_public_user > 0:
+ if (
+ not user.enable_register
+ and emby_config.register_public_user > 0
+ ):
emby_config.register_public_user -= 1
emby_config.total_register_user += 1
- new_user = await self._emby_create_user(telegram_id, username, password)
+ new_user = await self._emby_create_user(
+ telegram_id, username, password
+ )
session.add(new_user)
session.add(emby_config)
await session.commit()
return new_user
- async def _check_register_permission(self, user: User, emby_config: Config) -> bool:
+ async def _check_register_permission(
+ self, user: User, emby_config: Config
+ ) -> bool:
"""检查用户是否有权限注册 Emby 账号"""
enable_register = user.enable_register
if not enable_register and emby_config.register_public_user > 0:
@@ -207,7 +228,11 @@ async def redeem_code(self, telegram_id: int, code: str):
# 使用事务块，并通过行锁防止并发问题
async with InviteCodeOrm().transaction() as session:
# 构造 SELECT 语句，并加上 FOR UPDATE 行锁
- stmt = select(InviteCode).where(InviteCode.code == code).with_for_update()
+ stmt = (
+ select(InviteCode)
+ .where(InviteCode.code == code)
+ .with_for_update()
+ )
result = await session.execute(stmt)
valid_code = result.scalars().first()
@@ -239,7 +264,9 @@ async def redeem_code(self, telegram_id: int, code: str):
return valid_code
- async def reset_password(self, telegram_id: int, password: str = "") -> bool:
+ async def reset_password(
+ self, telegram_id: int, password: str = ""
+ ) -> bool:
"""重置用户的 Emby 密码。"""
user = await self.must_get_emby_user(telegram_id)
try:
@@ -251,7 +278,10 @@ async def reset_password(self, telegram_id: int, password: str = "") -> bool:
return False
async def emby_ban(
- self, telegram_id: int, reason: str, operator_telegram_id: Optional[int] = None
+ self,
+ telegram_id: int,
+ reason: str,
+ operator_telegram_id: Optional[int] = None,
) -> bool:
"""禁用用户"""
if operator_telegram_id is not None:
@@ -336,10 +366,14 @@ async def get_user_router(self, telegram_id: int) -> Dict:
user = await self.must_get_emby_user(telegram_id)
return self.emby_router_api.query_user_route(user.emby_id)
- async def update_user_router(self, telegram_id: int, new_index: str) -> bool:
+ async def update_user_router(
+ self, telegram_id: int, new_index: str
+ ) -> bool:
"""更新用户线路信息"""
user = await self.must_get_emby_user(telegram_id)
- return self.emby_router_api.update_user_route(str(user.emby_id), str(new_index))
+ return self.emby_router_api.update_user_route(
+ str(user.emby_id), str(new_index)
+ )
async def get_router_list(self, telegram_id: int) -> List[Dict]:
"""获取所有可用线路"""