合并Dev更新 - #93
Conversation
Original review guide in EnglishReviewer's GuideThis pull request migrates the database from SQLite to PostgreSQL, leveraging the File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
你好 @PackageInstaller - 我已经审查了你的更改，发现有一些需要解决的问题。
阻塞性问题：
- 似乎存在硬编码的数据库 URL。请确保这不是生产环境的 URL。(链接)
总体意见：
- 建议在
XiuxianDataManage类中添加__slots__定义，以减少内存占用，尤其是在你创建单例实例的情况下。 - 数据库初始化逻辑较为复杂，建议将其拆分为更小且命名清晰的函数，以提升可读性。
以下是我在审查过程中关注的内容
- 🟡 通用问题：发现 7 个问题
- 🔴 安全性：1 个阻塞性问题
- 🟢 测试：一切正常
- 🟡 复杂度：发现 5 个问题
- 🟢 文档：一切正常
帮我变得更有用！请点击每条评论旁的 👍 或 👎，我会根据反馈改进你的审查体验。
Original comment in English
Hey @PackageInstaller - I've reviewed your changes and found some issues that need to be addressed.
Blocking issues:
- Looks like a hardcoded database URL. Please ensure this is not a production URL. (link)
Overall Comments:
- Consider adding a
__slots__definition to theXiuxianDataManageclass to reduce memory usage, especially since you're creating a singleton instance. - The database initialization logic is complex; consider breaking it down into smaller, well-named functions to improve readability.
Here's what I looked at during the review
- 🟡 General issues: 7 issues found
- 🔴 Security: 1 blocking issue
- 🟢 Testing: all looks good
- 🟡 Complexity: 5 issues found
- 🟢 Documentation: all looks good
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| ) | ||
| from ..xiuxian_utils.item_json import Items | ||
|
|
||
| items = Items() | ||
| sql_message = XiuxianDateManage() # sql类 |
There was a problem hiding this comment.
suggestion (performance): 重复实例化 XiuxianDataManage() 可能会带来性能开销。
你在许多地方调用 'await XiuxianDataManage()' 来执行数据库操作。除非有特殊原因每次都需要新实例，否则建议使用共享实例或连接池，以减少对象实例化的开销。
Original comment in English
suggestion (performance): Repeatedly instantiating XiuxianDataManage() may incur overhead.
In many places you call 'await XiuxianDataManage()' to perform a database operation. Consider using a shared instance or connection pool to reduce object instantiation overhead, unless there is a specific reason for creating a new instance every time.
| elixir_room_level = sect_info['elixir_room_level'] # 宗门丹房等级 | ||
| if int(elixir_room_level) == len(elixir_room_level_up_config): | ||
| msg = f"宗门丹房等级已经达到最高等级，无法继续建设了！" | ||
| if XiuConfig().img: | ||
| pic = await get_msg_pic(f"@{event.sender.nickname}\n" + msg) | ||
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) |
There was a problem hiding this comment.
suggestion: 发送者昵称的兜底逻辑重复。
在多个位置都出现了在 user_info['user_name'] 和 event.sender.nickname 之间选择的模式。建议将此逻辑抽象为一个辅助函数，以提升可维护性并减少代码重复。
建议实现：
def get_sender_display_name(event, user_info):
"""
优先返回 user_info['user_name']，否则回退到 event.sender.nickname。
"""
return user_info.get('user_name') or event.sender.nickname pic = await get_msg_pic(f"@{get_sender_display_name(event, user_info)}\n" + msg)请确保将辅助函数（get_sender_display_name）放在文件合适的位置（例如顶部或其调用之前），以便所有相关代码都能访问到。同时检查其他使用相同内联兜底逻辑的地方，并替换为 get_sender_display_name() 的调用。
Original comment in English
suggestion: Repeated fallback logic for sender nickname.
The pattern to choose between user_info['user_name'] and event.sender.nickname appears in many locations. Consider abstracting this logic into a helper function to improve maintainability and reduce code duplication.
Suggested implementation:
def get_sender_display_name(event, user_info):
"""
Returns the display name by prioritizing user_info['user_name'] if it exists.
Otherwise, falls back to event.sender.nickname.
"""
return user_info.get('user_name') or event.sender.nickname pic = await get_msg_pic(f"@{get_sender_display_name(event, user_info)}\n" + msg)Make sure to place the helper function (get_sender_display_name) at an appropriate location in the file (for example, near the top or before its usage) so that it is accessible to all the code that calls it. Also, check for other parts of the code using the same inline fallback logic and replace them with calls to get_sender_display_name().
| msg = f"发生未知错误，多次尝试无果请找晓楠！" | ||
| await handle_send(bot, event, send_group_id, msg) | ||
| await impart_draw.finish() | ||
| if impart_data_draw['stone_num'] < 10: |
There was a problem hiding this comment.
suggestion: impart_draw_ 函数中分支逻辑复杂。
impart_draw_ 函数包含许多条件判断和深层嵌套逻辑。建议将部分逻辑重构为更小的辅助函数，以简化可读性和可维护性。
建议实现：
# 将以下辅助函数放在 import 语句后
async def send_error_and_finish(bot, event, send_group_id, error_msg, finish_handler):
await handle_send(bot, event, send_group_id, error_msg)
await finish_handler.finish()
async def notify_if_insufficient_stones(bot, event, send_group_id, impart_data_draw):
if impart_data_draw.get('stone_num', 0) < 10:
msg = "思恋结晶数量不足10个,无法抽卡!"
await handle_send(bot, event, send_group_id, msg)
return True
return False
# 现有代码继续 await send_error_and_finish(bot, event, send_group_id, "发生未知错误，多次尝试无果请找晓楠！", impart_draw) if await notify_if_insufficient_stones(bot, event, send_group_id, impart_data_draw):
return请确保新辅助函数放在所有调用它们的函数都能访问的位置。你可能需要调整 import 顺序或修改其他存在类似分支模式的代码。
Original comment in English
suggestion: Complex branching logic in impart_draw_ function.
The function impart_draw_ contains many conditionals with deeply nested logic. Consider refactoring parts of this function into smaller helper functions to simplify readability and maintainability.
Suggested implementation:
# Place the following helper functions (e.g., after your import statements)
async def send_error_and_finish(bot, event, send_group_id, error_msg, finish_handler):
await handle_send(bot, event, send_group_id, error_msg)
await finish_handler.finish()
async def notify_if_insufficient_stones(bot, event, send_group_id, impart_data_draw):
if impart_data_draw.get('stone_num', 0) < 10:
msg = "思恋结晶数量不足10个,无法抽卡!"
await handle_send(bot, event, send_group_id, msg)
return True
return False
# existing code continues below await send_error_and_finish(bot, event, send_group_id, "发生未知错误，多次尝试无果请找晓楠！", impart_draw) if await notify_if_insufficient_stones(bot, event, send_group_id, impart_data_draw):
returnMake sure the new helper functions are placed in a location that is accessible to the functions calling them. You may need to adjust import orders or modify other parts of the code if similar branching patterns exist.
| async def compress_img(img): | ||
| """对传入图片进行压缩""" | ||
| img_byte_arr = BytesIO() | ||
| compression_quality = max( |
There was a problem hiding this comment.
question (bug_risk): 校验图片压缩质量的取值范围。
压缩质量的计算是用 100 减去配置的限制值。请确认 XiuConfig().img_compression_limit 的取值在 0 到 100 之间，以确保压缩质量值合理。
Original comment in English
question (bug_risk): Validate image compression quality bounds.
The calculation for compression quality subtracts the configured limit from 100. Confirm that XiuConfig().img_compression_limit is within 0 to 100 to yield an appropriate quality value.
| msg = f"本群尚未开启拍卖会功能，请联系管理员开启！" | ||
| if XiuConfig().img: | ||
| pic = await get_msg_pic(f"@{event.sender.nickname}\n" + msg) | ||
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) |
There was a problem hiding this comment.
suggestion (code-quality): 用 or 替换 if 表达式 (or-if-exp-identity)
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) | |
| pic = await get_msg_pic(f"@{user_info['user_name'] or event.sender.nickname}\n" + msg) |
解释
这里我们在设置一个值时，如果它为真就用它，否则用默认值。“After” 方案更易读，也避免了 input_currency 的重复。
它的原理是先计算左侧，如果为真则赋值，否则计算右侧并赋值。
Original comment in English
suggestion (code-quality): Replace if-expression with or (or-if-exp-identity)
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) | |
| pic = await get_msg_pic(f"@{user_info['user_name'] or event.sender.nickname}\n" + msg) |
Explanation
Here we find ourselves setting a value if it evaluates toTrue, and otherwiseusing a default.
The 'After' case is a bit easier to read and avoids the duplication of
input_currency.
It works because the left-hand side is evaluated first. If it evaluates to
true then currency will be set to this and the right-hand side will not be
evaluated. If it evaluates to false the right-hand side will be evaluated and
currency will be set to DEFAULT_CURRENCY.
| msg = f"本群不存在拍卖会，请等待拍卖会开启！" | ||
| if XiuConfig().img: | ||
| pic = await get_msg_pic(f"@{event.sender.nickname}\n" + msg) | ||
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) |
There was a problem hiding this comment.
suggestion (code-quality): 用 or 替换 if 表达式 (or-if-exp-identity)
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) | |
| pic = await get_msg_pic(f"@{user_info['user_name'] or event.sender.nickname}\n" + msg) |
解释
这里我们在设置一个值时，如果它为真就用它，否则用默认值。“After” 方案更易读，也避免了 input_currency 的重复。
它的原理是先计算左侧，如果为真则赋值，否则计算右侧并赋值。
Original comment in English
suggestion (code-quality): Replace if-expression with or (or-if-exp-identity)
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) | |
| pic = await get_msg_pic(f"@{user_info['user_name'] or event.sender.nickname}\n" + msg) |
Explanation
Here we find ourselves setting a value if it evaluates toTrue, and otherwiseusing a default.
The 'After' case is a bit easier to read and avoids the duplication of
input_currency.
It works because the left-hand side is evaluated first. If it evaluates to
true then currency will be set to this and the right-hand side will not be
evaluated. If it evaluates to false the right-hand side will be evaluated and
currency will be set to DEFAULT_CURRENCY.
| msg = f"请发送正确的灵石数量" | ||
| if XiuConfig().img: | ||
| pic = await get_msg_pic(f"@{event.sender.nickname}\n" + msg) | ||
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) |
There was a problem hiding this comment.
suggestion (code-quality): 用 or 替换 if 表达式 (or-if-exp-identity)
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) | |
| pic = await get_msg_pic(f"@{user_info['user_name'] or event.sender.nickname}\n" + msg) |
解释
这里我们在设置一个值时，如果它为真就用它，否则用默认值。“After” 方案更易读，也避免了 input_currency 的重复。
它的原理是先计算左侧，如果为真则赋值，否则计算右侧并赋值。
Original comment in English
suggestion (code-quality): Replace if-expression with or (or-if-exp-identity)
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) | |
| pic = await get_msg_pic(f"@{user_info['user_name'] or event.sender.nickname}\n" + msg) |
Explanation
Here we find ourselves setting a value if it evaluates toTrue, and otherwiseusing a default.
The 'After' case is a bit easier to read and avoids the duplication of
input_currency.
It works because the left-hand side is evaluated first. If it evaluates to
true then currency will be set to this and the right-hand side will not be
evaluated. If it evaluates to false the right-hand side will be evaluated and
currency will be set to DEFAULT_CURRENCY.
| msg = f"走开走开，别捣乱！小心清空你灵石捏" | ||
| if XiuConfig().img: | ||
| pic = await get_msg_pic(f"@{event.sender.nickname}\n" + msg) | ||
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) |
There was a problem hiding this comment.
suggestion (code-quality): 用 or 替换 if 表达式 (or-if-exp-identity)
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) | |
| pic = await get_msg_pic(f"@{user_info['user_name'] or event.sender.nickname}\n" + msg) |
解释
这里我们在设置一个值时，如果它为真就用它，否则用默认值。“After” 方案更易读，也避免了 input_currency 的重复。
它的原理是先计算左侧，如果为真则赋值，否则计算右侧并赋值。
Original comment in English
suggestion (code-quality): Replace if-expression with or (or-if-exp-identity)
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) | |
| pic = await get_msg_pic(f"@{user_info['user_name'] or event.sender.nickname}\n" + msg) |
Explanation
Here we find ourselves setting a value if it evaluates toTrue, and otherwiseusing a default.
The 'After' case is a bit easier to read and avoids the duplication of
input_currency.
It works because the left-hand side is evaluated first. If it evaluates to
true then currency will be set to this and the right-hand side will not be
evaluated. If it evaluates to false the right-hand side will be evaluated and
currency will be set to DEFAULT_CURRENCY.
| msg = f"拍卖不得少于当前竞拍价的5%，目前最少加价为：{min_price}灵石，目前竞拍价为：{now_price}!" | ||
| if XiuConfig().img: | ||
| pic = await get_msg_pic(f"@{event.sender.nickname}\n" + msg) | ||
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) |
There was a problem hiding this comment.
suggestion (code-quality): 用 or 替换 if 表达式 (or-if-exp-identity)
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) | |
| pic = await get_msg_pic(f"@{user_info['user_name'] or event.sender.nickname}\n" + msg) |
解释
这里我们在设置一个值时，如果它为真就用它，否则用默认值。“After” 方案更易读，也避免了 input_currency 的重复。
它的原理是先计算左侧，如果为真则赋值，否则计算右侧并赋值。
Original comment in English
suggestion (code-quality): Replace if-expression with or (or-if-exp-identity)
| pic = await get_msg_pic(f"@{user_info['user_name'] if user_info['user_name'] else event.sender.nickname}\n" + msg) | |
| pic = await get_msg_pic(f"@{user_info['user_name'] or event.sender.nickname}\n" + msg) |
Explanation
Here we find ourselves setting a value if it evaluates toTrue, and otherwiseusing a default.
The 'After' case is a bit easier to read and avoids the duplication of
input_currency.
It works because the left-hand side is evaluated first. If it evaluates to
true then currency will be set to this and the right-hand side will not be
evaluated. If it evaluates to false the right-hand side will be evaluated and
currency will be set to DEFAULT_CURRENCY.
|
@sourcery-ai review |
There was a problem hiding this comment.
嗨 @PackageInstaller - 我已经审查了你的更改，以下是一些反馈：
总体意见：
- 建议在如
_check_data这类需要遍历多个表和列的方法中，使用asyncio.gather并发执行多个数据库查询。 final_user_data函数现在是一个async函数，请确保所有调用该函数的地方都已更新为await。
以下是我在审查过程中关注的内容
- 🟡 通用问题：发现 1 个问题
- 🟢 安全性：一切正常
- 🟢 测试：一切正常
- 🟡 复杂度：发现 5 个问题
- 🟢 文档：一切正常
帮助我变得更有用！请对每条评论点击 👍 或 👎，我会根据反馈改进你的审查体验。
Original comment in English
Hey @PackageInstaller - I've reviewed your changes - here's some feedback:
Overall Comments:
- Consider using
asyncio.gatherto run multiple database queries concurrently, especially in methods like_check_datawhere you're iterating through multiple tables and columns. - The
final_user_datafunction is now anasyncfunction, so ensure all call sites are updated toawaitit.
Here's what I looked at during the review
- 🟡 General issues: 1 issue found
- 🟢 Security: all looks good
- 🟢 Testing: all looks good
- 🟡 Complexity: 5 issues found
- 🟢 Documentation: all looks good
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| def append_draw_card_node(bot: Bot, list_tp: list, summary: str, content): | ||
| """添加节点进转发消息 | ||
|
|
||
| def build_forward_msg_list(bot: Bot, summary: str, text_msg: str, images: list = None, image_params: dict = None): |
There was a problem hiding this comment.
suggestion (bug_risk): 新增的 build_forward_msg_list 函数增加了灵活的转发支持。
该函数现在为合并转发发送提供了额外的图片参数选项。建议在图片参数缺失时进行校验，确保函数能够优雅地回退。
Original comment in English
suggestion (bug_risk): New build_forward_msg_list function adds flexible forwarding support.
The function now provides options for merged forward sending with additional image parameters. It would be useful to validate the image parameters in cases where they are missing, ensuring the function falls back gracefully.
|
|
||
|
|
||
| @impart_draw.handle(parameterless=[Cooldown(at_sender=False)]) | ||
| async def impart_draw_(bot: Bot, event: GroupMessageEvent): |
There was a problem hiding this comment.
issue (complexity): 建议将 impart_draw_ 处理器中重复且嵌套较深的部分提取为小型、命名明确的辅助函数，以提升可读性并降低复杂度。
建议将 impart_draw_ 处理器中重复且嵌套较深的部分提取为小型辅助函数。例如，可以将创建带有特殊（重复或新）卡片的随机图片列表的逻辑单独封装。这样可以减少嵌套并整合复杂步骤。例如：
def build_draw_images(time_imgs, special_card):
imgs = time_imgs[:10]
random.shuffle(imgs)
imgs[random.randint(0, 9)] = special_card
return imgs然后在主处理器中这样调用：
if await get_rank(user_id):
...
try:
reap_img = random.choice(img_list)
except Exception:
await handle_send(bot, event, send_group_id, "请检查卡图数据完整！")
await impart_draw.finish()
summary = f"道友{user_info['user_name']}的传承抽卡"
if impart_data_json.data_person_add(user_id, reap_img):
msg = (
f"检测到传承背包已经存在卡片{reap_img}\n"
"已转化为2880分钟闭关时间\n"
"累计共获得3540分钟闭关时间!\n"
"抽卡10次结果如下"
)
images = build_draw_images(time_img, reap_img)
list_tp = build_forward_msg_list(bot, summary, msg, images, image_params)
await XiuxianDataManage().add_impart_exp_day(3540, user_id)
else:
msg = (
f"累计共获得660分钟闭关时间!\n"
f"抽卡10次结果如下,获得新的传承卡片{reap_img}"
)
images = build_draw_images(time_img, reap_img)
list_tp = build_forward_msg_list(bot, summary, msg, images, image_params)
await XiuxianDataManage().add_impart_exp_day(660, user_id)
# 公共步骤：
await XiuxianDataManage().update_stone_num(10, user_id, 1)
await XiuxianDataManage().update_impart_wish(0, user_id)
await re_impart_data(user_id)
try:
await send_msg_handler(bot, event, list_tp)
except ActionFailed:
await handle_send(bot, event, send_group_id, "未知原因，抽卡失败!")
await impart_draw.finish()通过将图片构建逻辑和其他决策分支提取为辅助函数，主处理器会更易读且易于维护。
Original comment in English
issue (complexity): Consider extracting the repeated and deeply nested parts of the impart_draw_ handler into smaller, well-named helper functions to improve readability and reduce complexity.
Consider extracting the repeated and deeply nested parts of the impart_draw_ handler into small helper functions. For example, you can move the logic for creating a randomized image list with a special (repeated or new) card into its own helper. This reduces nesting and consolidates complex steps. For instance:
def build_draw_images(time_imgs, special_card):
imgs = time_imgs[:10]
random.shuffle(imgs)
imgs[random.randint(0, 9)] = special_card
return imgsThen update the main handler:
if await get_rank(user_id):
...
try:
reap_img = random.choice(img_list)
except Exception:
await handle_send(bot, event, send_group_id, "请检查卡图数据完整！")
await impart_draw.finish()
summary = f"道友{user_info['user_name']}的传承抽卡"
if impart_data_json.data_person_add(user_id, reap_img):
msg = (
f"检测到传承背包已经存在卡片{reap_img}\n"
"已转化为2880分钟闭关时间\n"
"累计共获得3540分钟闭关时间!\n"
"抽卡10次结果如下"
)
images = build_draw_images(time_img, reap_img)
list_tp = build_forward_msg_list(bot, summary, msg, images, image_params)
await XiuxianDataManage().add_impart_exp_day(3540, user_id)
else:
msg = (
f"累计共获得660分钟闭关时间!\n"
f"抽卡10次结果如下,获得新的传承卡片{reap_img}"
)
images = build_draw_images(time_img, reap_img)
list_tp = build_forward_msg_list(bot, summary, msg, images, image_params)
await XiuxianDataManage().add_impart_exp_day(660, user_id)
# Common steps:
await XiuxianDataManage().update_stone_num(10, user_id, 1)
await XiuxianDataManage().update_impart_wish(0, user_id)
await re_impart_data(user_id)
try:
await send_msg_handler(bot, event, list_tp)
except ActionFailed:
await handle_send(bot, event, send_group_id, "未知原因，抽卡失败!")
await impart_draw.finish()By isolating the image-building logic and, if needed, extracting other decision branches into helper functions, your main handler becomes easier to read and maintain.
| await handle_send(bot, event, send_group_id, msg) | ||
| await do_work.finish() | ||
|
|
||
| if mode is None: # 接取逻辑 |
There was a problem hiding this comment.
issue (complexity): 建议通过将每种模式的逻辑提取到专用的辅助函数中来重构处理器，以降低复杂度。
你可以通过将每种模式（如“刷新”、“终止”、“结算”、“接取”、“帮助”）的逻辑提取到专用的辅助函数中，减少深层嵌套和重复模式。这样主处理器只需路由动作，而无需内联所有业务逻辑。例如：
async def handle_refresh(bot: Bot, event: GroupMessageEvent, send_group_id, user_info, user_cd_message, ...):
# “刷新”模式的具体逻辑
# 可通过辅助函数移除重复的消息发送模式，例如：
msg = "..." # 处理后的消息结果
await send_response(bot, event, send_group_id, msg)
async def send_response(bot: Bot, event: GroupMessageEvent, send_group_id: str, msg: str):
if XiuConfig().img:
pic = await get_msg_pic(f"@{event.sender.nickname}\n" + msg)
await bot.send_group_msg(group_id=int(send_group_id), message=MessageSegment.image(pic))
else:
await bot.send_group_msg(group_id=int(send_group_id), message=msg)然后这样更新你的处理器：
if mode == "刷新":
await handle_refresh(bot, event, send_group_id, user_info, user_cd_message, ...)
elif mode == "终止":
await handle_terminate(bot, event, send_group_id, user_info, ...)
# 等等这种重构在保持功能不变的同时，降低了整体复杂度并提升了可读性。
Original comment in English
issue (complexity): Consider refactoring the handler by extracting each mode's logic into its own dedicated helper function to reduce complexity.
You can reduce the deep nesting and repeated patterns by extracting each mode’s logic (e.g. “刷新”, “终止”, “结算”, “接取”, “帮助”) into its own dedicated helper function. This way, the main handler simply routes the action instead of containing all business logic inline. For example:
async def handle_refresh(bot: Bot, event: GroupMessageEvent, send_group_id, user_info, user_cd_message, ...):
# Your runnable logic for "刷新"
# Remove repeated message sending pattern by using a helper, e.g.:
msg = "..." # Processed message result.
await send_response(bot, event, send_group_id, msg)
async def send_response(bot: Bot, event: GroupMessageEvent, send_group_id: str, msg: str):
if XiuConfig().img:
pic = await get_msg_pic(f"@{event.sender.nickname}\n" + msg)
await bot.send_group_msg(group_id=int(send_group_id), message=MessageSegment.image(pic))
else:
await bot.send_group_msg(group_id=int(send_group_id), message=msg)Then update your handler like this:
if mode == "刷新":
await handle_refresh(bot, event, send_group_id, user_info, user_cd_message, ...)
elif mode == "终止":
await handle_terminate(bot, event, send_group_id, user_info, ...)
# etc.This refactor keeps functionality intact while reducing overall complexity and improving readability.
|
|
||
|
|
||
|
|
||
| async def handle_send(bot, event, send_group_id, msg: str): |
There was a problem hiding this comment.
issue (complexity): 建议将常用的消息发送任务提取为辅助函数，以减少嵌套条件逻辑并提升代码清晰度。
建议将常用的消息发送任务提取为小型辅助函数，以减少嵌套条件逻辑。例如，`handle_send` 的两个分支都执行了类似的“发送”操作，将该逻辑移到专用函数中会让意图更清晰。
在 `handle_send` 中的重构示例：
```python
async def send_message(bot, event, group_id, message):
if isinstance(event, GroupMessageEvent):
await bot.send_group_msg(group_id=group_id, message=message)
else:
await bot.send_private_msg(user_id=event.user_id, message=message)
async def handle_send(bot, event, send_group_id, msg: str):
at_text = ""
if hasattr(event, "user_id"):
user_info = await XiuxianDataManage().get_user_info_with_id(event.user_id)
user_name = await get_sender_display_name(event, user_info)
at_text = f"@{user_name}\n"
full_msg = at_text + msg
if XiuConfig().img:
pic = await get_msg_pic(full_msg)
message = MessageSegment.image(pic)
else:
message = full_msg
await send_message(bot, event, int(send_group_id), message)同样，在 build_forward_msg_list 中可以将节点创建提取为辅助函数，避免重复的字典字面量：
def create_node(bot: Bot, summary: str, content):
return {
"type": "node",
"data": {
"name": summary,
"uin": bot.self_id,
"content": content,
},
}
def build_forward_msg_list(bot: Bot, summary: str, text_msg: str, images: list = None, image_params: dict = None):
use_merge_forward_send = image_params.get("use_merge_forward_send", XiuConfig().merge_forward_send) if image_params else XiuConfig().merge_forward_send
if use_merge_forward_send:
list_tp = [create_node(bot, summary, text_msg)]
if images:
img_path = image_params.get("img_path") if image_params else None
img_format = image_params.get("img_format", "png") if image_params else "png"
get_image_func = image_params.get("get_image_func") if image_params else None
for image in images:
if get_image_func:
img = get_image_func(image)
elif img_path:
img = MessageSegment.image(img_path / f"{image}.{img_format}")
else:
img = str(image)
list_tp.append(create_node(bot, summary, img))
return list_tp
else:
result_msgs = [text_msg]
if images:
result_msgs.extend(map(str, images))
return [summary, bot.self_id, result_msgs]这些更改通过减少函数内嵌套，提升了逻辑清晰度，同时保持功能不变。
<details>
<summary>Original comment in English</summary>
**issue (complexity):** Consider extracting common messaging tasks into helper functions to reduce nested conditional logic and improve code clarity.
```markdown
Consider extracting common messaging tasks into small helper functions to reduce nested conditional logic. For instance, both branches of `handle_send` perform similar “send” calls; moving that logic into a dedicated function would make the intent clearer.
Example refactoring in `handle_send`:
```python
async def send_message(bot, event, group_id, message):
if isinstance(event, GroupMessageEvent):
await bot.send_group_msg(group_id=group_id, message=message)
else:
await bot.send_private_msg(user_id=event.user_id, message=message)
async def handle_send(bot, event, send_group_id, msg: str):
at_text = ""
if hasattr(event, "user_id"):
user_info = await XiuxianDataManage().get_user_info_with_id(event.user_id)
user_name = await get_sender_display_name(event, user_info)
at_text = f"@{user_name}\n"
full_msg = at_text + msg
if XiuConfig().img:
pic = await get_msg_pic(full_msg)
message = MessageSegment.image(pic)
else:
message = full_msg
await send_message(bot, event, int(send_group_id), message)
Similarly, in build_forward_msg_list you can extract node creation into a helper function to avoid repeated dictionary literals:
def create_node(bot: Bot, summary: str, content):
return {
"type": "node",
"data": {
"name": summary,
"uin": bot.self_id,
"content": content,
},
}
def build_forward_msg_list(bot: Bot, summary: str, text_msg: str, images: list = None, image_params: dict = None):
use_merge_forward_send = image_params.get("use_merge_forward_send", XiuConfig().merge_forward_send) if image_params else XiuConfig().merge_forward_send
if use_merge_forward_send:
list_tp = [create_node(bot, summary, text_msg)]
if images:
img_path = image_params.get("img_path") if image_params else None
img_format = image_params.get("img_format", "png") if image_params else "png"
get_image_func = image_params.get("get_image_func") if image_params else None
for image in images:
if get_image_func:
img = get_image_func(image)
elif img_path:
img = MessageSegment.image(img_path / f"{image}.{img_format}")
else:
img = str(image)
list_tp.append(create_node(bot, summary, img))
return list_tp
else:
result_msgs = [text_msg]
if images:
result_msgs.extend(map(str, images))
return [summary, bot.self_id, result_msgs]These changes clarify the logic by reducing in-function nesting while keeping functionality intact.
</details>
| res = await convert_img(img) | ||
| return res | ||
|
|
||
| # 根据发送类型返回不同格式的图片 |
There was a problem hiding this comment.
issue (complexity): 建议将图片发送逻辑提取为单独函数，以减少分支复杂度并避免重复调用 compress_img。
建议将图片发送逻辑提取为独立的小函数。这样可以在保持配置检查的同时，减少主流程中的分支复杂度。例如：
async def process_image(img: Image.Image) -> str | bytes:
# 只压缩一次图片
compressed_bytes = await compress_img(img)
send_type = XiuConfig().img_send_type
if send_type == "base64":
return f"base64://{b64encode(compressed_bytes).decode()}"
# 对于 "io" 及其他默认情况
return compressed_bytes然后在主函数中用简单调用替换原有条件块：
# 替换如下代码块：
# if XiuConfig().img_send_type == "io":
# return await compress_img(img)
# elif XiuConfig().img_send_type == "base64":
# compressed_bytes = await compress_img(img)
# return f"base64://{b64encode(compressed_bytes).decode()}"
# else:
# return await compress_img(img)
# 替换为：
return await process_image(img)此重构消除了对 compress_img 的重复调用，并将配置逻辑集中到专用辅助函数中，降低了认知负担且不改变功能。
Original comment in English
issue (complexity): Consider extracting the image sending logic into a separate function to reduce branching complexity and avoid duplicate calls to compress_img.
Consider extracting the image sending logic into its own small function. This reduces the branching complexity in the main flow while keeping the configuration checks intact. For example:
async def process_image(img: Image.Image) -> str | bytes:
# Compress the image once
compressed_bytes = await compress_img(img)
send_type = XiuConfig().img_send_type
if send_type == "base64":
return f"base64://{b64encode(compressed_bytes).decode()}"
# For both "io" and any other defaults
return compressed_bytesThen, in your main function, replace the conditional block with a simple call:
# Replace this block:
# if XiuConfig().img_send_type == "io":
# return await compress_img(img)
# elif XiuConfig().img_send_type == "base64":
# compressed_bytes = await compress_img(img)
# return f"base64://{b64encode(compressed_bytes).decode()}"
# else:
# return await compress_img(img)
# With:
return await process_image(img)This refactoring removes duplicate calls to compress_img and nests the configuration logic in a dedicated helper function, lowering cognitive load without changing functionality.
|
|
||
|
|
||
| @level_up_dr.handle(parameterless=[Cooldown(stamina_cost=8, at_sender=False)]) | ||
| async def level_up_dr_(bot: Bot, event: GroupMessageEvent): |
There was a problem hiding this comment.
issue (code-quality): 我们发现了如下问题：
- 用命名表达式简化赋值和条件判断 (
use-named-expression) - 移除多余的 pass 语句 (
remove-redundant-pass) - 提取重复代码到条件语句外部 [×2] (
hoist-statement-from-if) - 用 min/max 替换比较 (
min-max-identity) - level_up_dr_ 代码质量较低 - 13% (
low-code-quality)
解释
该函数的质量分数低于 25% 的阈值。
该分数由方法长度、认知复杂度和工作记忆共同决定。
如何改进？
可以考虑将该函数重构得更短、更易读。
- 通过将功能片段提取到独立函数中，减少函数长度。这是最重要的优化——理想情况下函数应少于 10 行。
- 通过引入守卫子句提前返回，减少嵌套。
- 确保变量作用域紧凑，让相关代码聚集在一起而不是分散在函数各处。
Original comment in English
issue (code-quality): We've found these issues:
- Use named expression to simplify assignment and conditional (
use-named-expression) - Remove redundant pass statement (
remove-redundant-pass) - Hoist repeated code outside conditional statement [×2] (
hoist-statement-from-if) - Replace comparison with min/max call (
min-max-identity) - Low code quality found in level_up_dr_ - 13% (
low-code-quality)
Explanation
The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.
How can you solve this?
It might be worth refactoring this function to make it shorter and more readable.
- Reduce the function length by extracting pieces of functionality out into
their own functions. This is the most important thing you can do - ideally a
function should be less than 10 lines. - Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
sits together within the function rather than being scattered.
|
|
||
|
|
||
| @give_stone.handle(parameterless=[Cooldown(at_sender=False)]) | ||
| async def give_stone_(bot: Bot, event: GroupMessageEvent, args: Message = CommandArg()): |
There was a problem hiding this comment.
issue (code-quality): 我们发现了如下问题：
- 提取重复代码到条件语句外部 [×5] (
hoist-statement-from-if) - give_stone_ 代码质量较低 - 13% (
low-code-quality)
解释
该函数的质量分数低于 25% 的阈值。
该分数由方法长度、认知复杂度和工作记忆共同决定。
如何改进？
可以考虑将该函数重构得更短、更易读。
- 通过将功能片段提取到独立函数中，减少函数长度。这是最重要的优化——理想情况下函数应少于 10 行。
- 通过引入守卫子句提前返回，减少嵌套。
- 确保变量作用域紧凑，让相关代码聚集在一起而不是分散在函数各处。
Original comment in English
issue (code-quality): We've found these issues:
- Hoist repeated code outside conditional statement [×5] (
hoist-statement-from-if) - Low code quality found in give_stone_ - 13% (
low-code-quality)
Explanation
The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.
How can you solve this?
It might be worth refactoring this function to make it shorter and more readable.
- Reduce the function length by extracting pieces of functionality out into
their own functions. This is the most important thing you can do - ideally a
function should be less than 10 lines. - Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
sits together within the function rather than being scattered.
|
|
||
| # 偷灵石 | ||
| @steal_stone.handle(parameterless=[Cooldown(stamina_cost = 10, at_sender=False)]) | ||
| async def steal_stone_(bot: Bot, event: GroupMessageEvent, args: Message = CommandArg()): |
There was a problem hiding this comment.
issue (code-quality): 我们发现了如下问题：
- 移除不必要的 int、str、float 或 bool 类型转换 (
remove-unnecessary-cast) - 提取重复代码到条件语句外部 [×4] (
hoist-statement-from-if) - steal_stone_ 代码质量较低 - 12% (
low-code-quality)
解释
该函数的质量分数低于 25% 的阈值。
该分数由方法长度、认知复杂度和工作记忆共同决定。
如何改进？
可以考虑将该函数重构得更短、更易读。
- 通过将功能片段提取到独立函数中，减少函数长度。这是最重要的优化——理想情况下函数应少于 10 行。
- 通过引入守卫子句提前返回，减少嵌套。
- 确保变量作用域紧凑，让相关代码聚集在一起而不是分散在函数各处。
Original comment in English
issue (code-quality): We've found these issues:
- Remove unnecessary casts to int, str, float or bool (
remove-unnecessary-cast) - Hoist repeated code outside conditional statement [×4] (
hoist-statement-from-if) - Low code quality found in steal_stone_ - 12% (
low-code-quality)
Explanation
The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.
How can you solve this?
It might be worth refactoring this function to make it shorter and more readable.
- Reduce the function length by extracting pieces of functionality out into
their own functions. This is the most important thing you can do - ideally a
function should be less than 10 lines. - Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
sits together within the function rather than being scattered.
| await bot.send_group_msg(group_id=event.group_id, message=MessageSegment.image(pic)) | ||
| else: | ||
| await bot.send_group_msg(group_id=event.group_id, message=msg) | ||
| msg = f"道友已有宗门庇佑，又何必来此寻求机缘呢？" |
There was a problem hiding this comment.
issue (code-quality): 我们发现了如下问题：
- 移除无插值的 f-string (
remove-redundant-fstring) - 提取重复代码到条件语句外部 [×2] (
hoist-statement-from-if)
Original comment in English
issue (code-quality): We've found these issues:
- Replace f-string with no interpolated values with string (
remove-redundant-fstring) - Hoist repeated code outside conditional statement [×2] (
hoist-statement-from-if)
| @@ -227,62 +224,38 @@ async def boss_delete_(bot: Bot, event: GroupMessageEvent, args: Message = Comma | |||
| isInGroup = isInGroups(event) | |||
| if not isInGroup: # 不在配置表内 | |||
| msg = f"本群尚未开启世界Boss,请联系管理员开启!" | |||
There was a problem hiding this comment.
issue (code-quality): 我们发现了如下问题：
- 移除无插值的 f-string [×5] (
remove-redundant-fstring) - 用
except Exception:替代裸except:(do-not-use-bare-except)
Original comment in English
issue (code-quality): We've found these issues:
- Replace f-string with no interpolated values with string [×5] (
remove-redundant-fstring) - Use
except Exception:rather than bareexcept:(do-not-use-bare-except)
|
@sourcery-ai review |
There was a problem hiding this comment.
你好 @PackageInstaller - 我已经审查了你的更改，以下是一些反馈：
- 建议将较大的
xiuxian2_handle.py文件根据职责拆分为更小的模块（例如：数据访问、用户逻辑、常量等）。 - 这个拉取请求将一次重要的数据库迁移和异步重构与其他更改（如名称生成）合并在一起；将这些关注点拆分到不同的 PR 中可以简化审查流程。
以下是我在审查过程中关注的内容
- 🟡 通用问题：发现 1 个问题
- 🟢 安全性：一切良好
- 🟢 测试：一切良好
- 🟡 复杂度：发现 3 个问题
- 🟢 文档：一切良好
帮我变得更有用！请点击每条评论旁的 👍 或 👎，我会根据你的反馈改进审查。
Original comment in English
Hey @PackageInstaller - I've reviewed your changes - here's some feedback:
- Consider splitting the large
xiuxian2_handle.pyfile into smaller modules based on responsibility (e.g., data access, user logic, constants). - This pull request combines a significant database migration and async refactoring with other changes like name generation; separating these concerns into different PRs could simplify review.
Here's what I looked at during the review
- 🟡 General issues: 1 issue found
- 🟢 Security: all looks good
- 🟢 Testing: all looks good
- 🟡 Complexity: 3 issues found
- 🟢 Documentation: all looks good
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| await asyncio.gather( | ||
| XiuxianDataManager().updata_level(user_id, le[0]), # 更新境界 | ||
| XiuxianDataManager().update_power2(user_id), # 更新战力 | ||
| XiuxianDataManager().updata_level_cd(user_id), # 更新CD | ||
| XiuxianDataManager().update_levelrate(user_id, 0), | ||
| XiuxianDataManager().update_user_hp(user_id) # 重置用户HP，mp，atk状态 | ||
| ) |
There was a problem hiding this comment.
suggestion (performance): 建议将独立的异步调用进行批量处理以提升性能。
对于独立的数据库更新（如用户等级、战力、CD、速率），可以使用 asyncio.gather 并发执行，从而减少总等待时间。
| await asyncio.gather( | |
| XiuxianDataManager().updata_level(user_id, le[0]), # 更新境界 | |
| XiuxianDataManager().update_power2(user_id), # 更新战力 | |
| XiuxianDataManager().updata_level_cd(user_id), # 更新CD | |
| XiuxianDataManager().update_levelrate(user_id, 0), | |
| XiuxianDataManager().update_user_hp(user_id) # 重置用户HP，mp，atk状态 | |
| ) | |
| xiuxian_manager = XiuxianDataManager() | |
| await asyncio.gather( | |
| xiuxian_manager.updata_level(user_id, le[0]), # 更新境界 | |
| xiuxian_manager.update_power2(user_id), # 更新战力 | |
| xiuxian_manager.updata_level_cd(user_id), # 更新CD | |
| xiuxian_manager.update_levelrate(user_id, 0), | |
| xiuxian_manager.update_user_hp(user_id) # 重置用户HP，mp，atk状态 | |
| ) |
Original comment in English
suggestion (performance): Consider batching independent asynchronous calls to improve performance.
Use asyncio.gather for independent DB updates (user levels, power, CD, rate) to run them concurrently and reduce total wait time.
| await asyncio.gather( | |
| XiuxianDataManager().updata_level(user_id, le[0]), # 更新境界 | |
| XiuxianDataManager().update_power2(user_id), # 更新战力 | |
| XiuxianDataManager().updata_level_cd(user_id), # 更新CD | |
| XiuxianDataManager().update_levelrate(user_id, 0), | |
| XiuxianDataManager().update_user_hp(user_id) # 重置用户HP，mp，atk状态 | |
| ) | |
| xiuxian_manager = XiuxianDataManager() | |
| await asyncio.gather( | |
| xiuxian_manager.updata_level(user_id, le[0]), # 更新境界 | |
| xiuxian_manager.update_power2(user_id), # 更新战力 | |
| xiuxian_manager.updata_level_cd(user_id), # 更新CD | |
| xiuxian_manager.update_levelrate(user_id, 0), | |
| xiuxian_manager.update_user_hp(user_id) # 重置用户HP，mp，atk状态 | |
| ) |
| ) | ||
| else: | ||
| await bot.send_group_msg(group_id=int(send_group_id), message=msg) | ||
| await handle_send(bot, event, send_group_id, msg) |
There was a problem hiding this comment.
issue (complexity): 建议将重复和新卡片的处理逻辑提取为辅助函数，以减少嵌套并提升可读性。
可以将处理重复卡片和新卡片的逻辑分别提取为更小的辅助函数，这样可以减少 impart_draw_ 函数中的深层嵌套和内联消息/图片构建，同时不改变其行为。
例如，你可以将重复卡片和新卡片的处理分别提取为独立函数：
async def handle_duplicate_card(bot, event, user_info, reap_img, send_group_id):
summary = f"道友{user_info['user_name']}的传承抽卡"
msg = (
f"检测到传承背包已经存在卡片{reap_img}\n"
"已转化为2880分钟闭关时间\n"
"累计共获得3540分钟闭关时间!\n"
"抽卡10次结果如下"
)
images = await build_draw_images(time_img, reap_img)
image_params = {
"use_merge_forward_send": XiuConfig().merge_forward_send,
"img_path": img_path,
"img_format": "png",
}
list_tp = build_forward_msg_list(bot, summary, msg, images, image_params)
await XiuxianDataManager().add_impart_exp_day(3540, user_info['user_id'])
await XiuxianDataManager().update_stone_num(10, user_info['user_id'], 1)
await XiuxianDataManager().update_impart_wish(0, user_info['user_id'])
await re_impart_data(user_info['user_id'])
return list_tp
async def handle_new_card(bot, event, user_info, reap_img, send_group_id):
summary = f"道友{user_info['user_name']}的传承抽卡"
msg = (
f"累计共获得660分钟闭关时间!\n"
f"抽卡10次结果如下,获得新的传承卡片{reap_img}"
)
images = await build_draw_images(time_img, reap_img)
image_params = {
"use_merge_forward_send": XiuConfig().merge_forward_send,
"img_path": img_path,
"img_format": "png",
}
list_tp = build_forward_msg_list(bot, summary, msg, images, image_params)
await XiuxianDataManager().add_impart_exp_day(660, user_info['user_id'])
await XiuxianDataManager().update_stone_num(10, user_info['user_id'], 1)
await XiuxianDataManager().update_impart_wish(0, user_info['user_id'])
await re_impart_data(user_info['user_id'])
return list_tp然后，在 impart_draw_ 内根据条件调用这些辅助函数：
@impart_draw.handle(parameterless=[Cooldown(at_sender=False)])
async def impart_draw_(bot: Bot, event: GroupMessageEvent):
# 初始设置...
isUser, user_info, msg = await check_user(event)
if not isUser:
await handle_send(bot, event, send_group_id, msg)
await impart_draw.finish()
user_id = user_info['user_id']
impart_data_draw = await impart_check(user_id)
if impart_data_draw is None or impart_data_draw['impart_stone_quantity'] < 10:
msg = "发生未知错误或结晶不足!"
await handle_send(bot, event, send_group_id, msg)
await impart_draw.finish()
if await get_rank(user_id):
img_list = impart_data_json.data_all_keys()
try:
reap_img = random.choice(img_list)
except Exception:
msg = "请检查卡图数据完整！"
await handle_send(bot, event, send_group_id, msg)
await impart_draw.finish()
if impart_data_json.data_person_add(user_id, reap_img):
list_tp = await handle_duplicate_card(bot, event, user_info, reap_img, send_group_id)
else:
list_tp = await handle_new_card(bot, event, user_info, reap_img, send_group_id)
else:
# 处理非新卡片情况
summary = f"道友{user_info['user_name']}的传承抽卡"
msg = "累计共获得660分钟闭关时间!\n抽卡10次结果如下!"
image_params = {
"use_merge_forward_send": XiuConfig().merge_forward_send,
"img_path": img_path,
"img_format": "png",
}
list_tp = build_forward_msg_list(bot, summary, msg, time_img, image_params)
await XiuxianDataManager().add_impart_exp_day(660, user_id)
await XiuxianDataManager().update_stone_num(10, user_id, 1)
await XiuxianDataManager().add_impart_wish(10, user_id)
await re_impart_data(user_id)
try:
await send_msg_handler(bot, event, list_tp)
except ActionFailed:
msg = "未知原因，抽卡失败!"
await handle_send(bot, event, send_group_id, msg)
await impart_draw.finish()这些更改将不同的流程提取到辅助函数中，减少嵌套并提升可读性，同时保持功能不变。
Original comment in English
issue (complexity): Consider extracting the duplicate and new card handling logic into helper functions to reduce nesting and improve readability.
Consider extracting the duplicate logic for handling repeated versus new cards into smaller helper functions. This would reduce the deep nesting and inline message/image construction in the impart_draw_ function without changing its behavior.
For example, you could extract duplicate and new card handling into separate functions:
async def handle_duplicate_card(bot, event, user_info, reap_img, send_group_id):
summary = f"道友{user_info['user_name']}的传承抽卡"
msg = (
f"检测到传承背包已经存在卡片{reap_img}\n"
"已转化为2880分钟闭关时间\n"
"累计共获得3540分钟闭关时间!\n"
"抽卡10次结果如下"
)
images = await build_draw_images(time_img, reap_img)
image_params = {
"use_merge_forward_send": XiuConfig().merge_forward_send,
"img_path": img_path,
"img_format": "png",
}
list_tp = build_forward_msg_list(bot, summary, msg, images, image_params)
await XiuxianDataManager().add_impart_exp_day(3540, user_info['user_id'])
await XiuxianDataManager().update_stone_num(10, user_info['user_id'], 1)
await XiuxianDataManager().update_impart_wish(0, user_info['user_id'])
await re_impart_data(user_info['user_id'])
return list_tp
async def handle_new_card(bot, event, user_info, reap_img, send_group_id):
summary = f"道友{user_info['user_name']}的传承抽卡"
msg = (
f"累计共获得660分钟闭关时间!\n"
f"抽卡10次结果如下,获得新的传承卡片{reap_img}"
)
images = await build_draw_images(time_img, reap_img)
image_params = {
"use_merge_forward_send": XiuConfig().merge_forward_send,
"img_path": img_path,
"img_format": "png",
}
list_tp = build_forward_msg_list(bot, summary, msg, images, image_params)
await XiuxianDataManager().add_impart_exp_day(660, user_info['user_id'])
await XiuxianDataManager().update_stone_num(10, user_info['user_id'], 1)
await XiuxianDataManager().update_impart_wish(0, user_info['user_id'])
await re_impart_data(user_info['user_id'])
return list_tpThen, inside impart_draw_, delegate to these helpers based on the condition:
@impart_draw.handle(parameterless=[Cooldown(at_sender=False)])
async def impart_draw_(bot: Bot, event: GroupMessageEvent):
# Initial setup...
isUser, user_info, msg = await check_user(event)
if not isUser:
await handle_send(bot, event, send_group_id, msg)
await impart_draw.finish()
user_id = user_info['user_id']
impart_data_draw = await impart_check(user_id)
if impart_data_draw is None or impart_data_draw['impart_stone_quantity'] < 10:
msg = "发生未知错误或结晶不足!"
await handle_send(bot, event, send_group_id, msg)
await impart_draw.finish()
if await get_rank(user_id):
img_list = impart_data_json.data_all_keys()
try:
reap_img = random.choice(img_list)
except Exception:
msg = "请检查卡图数据完整！"
await handle_send(bot, event, send_group_id, msg)
await impart_draw.finish()
if impart_data_json.data_person_add(user_id, reap_img):
list_tp = await handle_duplicate_card(bot, event, user_info, reap_img, send_group_id)
else:
list_tp = await handle_new_card(bot, event, user_info, reap_img, send_group_id)
else:
# Handle non-new-card case
summary = f"道友{user_info['user_name']}的传承抽卡"
msg = "累计共获得660分钟闭关时间!\n抽卡10次结果如下!"
image_params = {
"use_merge_forward_send": XiuConfig().merge_forward_send,
"img_path": img_path,
"img_format": "png",
}
list_tp = build_forward_msg_list(bot, summary, msg, time_img, image_params)
await XiuxianDataManager().add_impart_exp_day(660, user_id)
await XiuxianDataManager().update_stone_num(10, user_id, 1)
await XiuxianDataManager().add_impart_wish(10, user_id)
await re_impart_data(user_id)
try:
await send_msg_handler(bot, event, list_tp)
except ActionFailed:
msg = "未知原因，抽卡失败!"
await handle_send(bot, event, send_group_id, msg)
await impart_draw.finish()These changes extract distinct flows into helper functions, reducing nesting and improving readability while preserving functionality.
| @@ -104,58 +104,65 @@ | |||
| # 定时任务每1小时按照宗门贡献度增加资材 | |||
| @materialsupdate.scheduled_job("cron", hour=config["发放宗门资材"]["时间"]) | |||
| async def materialsupdate_(): | |||
There was a problem hiding this comment.
issue (complexity): 建议每个函数只实例化一次 XiuxianDataManager，避免重复创建对象。
多次使用 await XiuxianDataManager().some_method(...) 可以通过在函数开头（或如果可以安全复用则在外部）创建一个实例来简化。这样可以减少重复实例化的开销并简化代码。
例如，可以将如下代码：
async def materialsupdate_():
all_sects = await XiuxianDataManager().get_all_sects_id_scale()
for s in all_sects:
await XiuxianDataManager().update_sect_materials(
sect_id=s[0],
sect_materials=s[1] * config["发放宗门资材"]["倍率"],
key=0
)重构为：
async def materialsupdate_():
data_manager = XiuxianDataManager()
all_sects = await data_manager.get_all_sects_id_scale()
for s in all_sects:
await data_manager.update_sect_materials(
sect_id=s[0],
sect_materials=s[1] * config["发放宗门资材"]["倍率"],
key=0
)在其他多次创建新实例的地方也可以采用类似模式。这样既保持了功能不变，又减少了不必要的认知负担和资源消耗。
Original comment in English
issue (complexity): Consider instantiating XiuxianDataManager once per function to avoid repeated object creation.
The repeated use of await XiuxianDataManager().some_method(...) can be simplified by creating a single instance at the beginning of the function (or even outside if it’s safe to reuse across calls). This reduces repeated instantiation overhead and simplifies the code.
For example, refactor from:
async def materialsupdate_():
all_sects = await XiuxianDataManager().get_all_sects_id_scale()
for s in all_sects:
await XiuxianDataManager().update_sect_materials(
sect_id=s[0],
sect_materials=s[1] * config["发放宗门资材"]["倍率"],
key=0
)to:
async def materialsupdate_():
data_manager = XiuxianDataManager()
all_sects = await data_manager.get_all_sects_id_scale()
for s in all_sects:
await data_manager.update_sect_materials(
sect_id=s[0],
sect_materials=s[1] * config["发放宗门资材"]["倍率"],
key=0
)Apply a similar pattern wherever a new instance is created repeatedly. This approach keeps functionality intact while reducing unnecessary cognitive overhead and resource usage.
| return user_info.get('user_name') or event.sender.nickname | ||
|
|
||
|
|
||
| async def handle_send(bot, event, send_group_id, msg: str, boss_name=""): |
There was a problem hiding this comment.
issue (complexity): 建议重构类似 handle_send 和 build_forward_msg_list 这样的多模式函数，以提升可读性并降低复杂度。
建议将辅助函数中的“多模式”流程拆分，减少每个函数中的条件判断数量。例如，在 `handle_send` 中，可以将图片和文本发送逻辑分别拆分到专用函数中。这样每个辅助函数只负责一个任务，更易于理解。
例如，可以这样重构：
```python
async def send_image(bot, event, send_group_id, img_data):
if isinstance(event, GroupMessageEvent):
await bot.send_group_msg(
group_id=int(send_group_id),
message=MessageSegment.image(img_data)
)
else:
await bot.send_private_msg(
user_id=event.user_id,
message=MessageSegment.image(img_data)
)
async def send_text(bot, event, send_group_id, msg):
await bot.send_group_msg(group_id=int(send_group_id), message=msg)
async def handle_send(bot, event, send_group_id, msg: str, boss_name=""):
at_text = ""
if event and hasattr(event, 'user_id'):
user_id = event.user_id
user_info = await XiuxianDataManager().get_user_infos_by_ids(user_id)
user_name = await get_sender_display_name(event, user_info)
at_text = f"@{user_name}\n"
if XiuConfig().img:
pic = await get_msg_pic(at_text + msg, boss_name=boss_name)
await send_image(bot, event, send_group_id, pic)
else:
await send_text(bot, event, send_group_id, msg)同样地，在 build_forward_msg_list 中可以将“node”创建逻辑单独提取为小型辅助函数。例如：
def create_node(bot, summary, content):
return {
"type": "node",
"data": {
"name": summary,
"uin": bot.self_id,
"content": content
}
}
def build_forward_msg_list(bot: Bot, summary: str, text_msg: str,
images: list = None, image_params: dict = None):
use_merge = (image_params.get("use_merge_forward_send")
if image_params else XiuConfig().merge_forward_send)
if use_merge:
nodes = [create_node(bot, summary, text_msg)]
if images:
img_path = image_params.get("img_path") if image_params else None
img_format = image_params.get("img_format", "png") if image_params else "png"
get_image_func = image_params.get("get_image_func") if image_params else None
for image in images:
if get_image_func:
img = get_image_func(image)
elif img_path:
img = MessageSegment.image(img_path / f"{image}.{img_format}")
else:
img = str(image)
nodes.append(create_node(bot, summary, img))
return nodes
else:
result_msgs = [text_msg] + [str(img) for img in images] if images else [text_msg]
return [summary, bot.self_id, result_msgs]这种方式保持了原有行为，但通过让每个辅助函数专注于单一任务，降低了每个函数的复杂度。
Original comment in English
issue (complexity): Consider refactoring multi-mode functions like handle_send and build_forward_msg_list to improve readability and reduce complexity.
Consider splitting up the “multi‐mode” flows in your helper functions to reduce the number of conditionals per function. For example, in `handle_send` you could “branch out” the image‐ and text‐sending logic into dedicated functions. This way each helper has a single responsibility and is easier to follow.
For instance, you could refactor as follows:
```python
async def send_image(bot, event, send_group_id, img_data):
if isinstance(event, GroupMessageEvent):
await bot.send_group_msg(
group_id=int(send_group_id),
message=MessageSegment.image(img_data)
)
else:
await bot.send_private_msg(
user_id=event.user_id,
message=MessageSegment.image(img_data)
)
async def send_text(bot, event, send_group_id, msg):
await bot.send_group_msg(group_id=int(send_group_id), message=msg)
async def handle_send(bot, event, send_group_id, msg: str, boss_name=""):
at_text = ""
if event and hasattr(event, 'user_id'):
user_id = event.user_id
user_info = await XiuxianDataManager().get_user_infos_by_ids(user_id)
user_name = await get_sender_display_name(event, user_info)
at_text = f"@{user_name}\n"
if XiuConfig().img:
pic = await get_msg_pic(at_text + msg, boss_name=boss_name)
await send_image(bot, event, send_group_id, pic)
else:
await send_text(bot, event, send_group_id, msg)In a similar way, in build_forward_msg_list you might isolate the logic for “node” creation into its own small helper. For example:
def create_node(bot, summary, content):
return {
"type": "node",
"data": {
"name": summary,
"uin": bot.self_id,
"content": content
}
}
def build_forward_msg_list(bot: Bot, summary: str, text_msg: str,
images: list = None, image_params: dict = None):
use_merge = (image_params.get("use_merge_forward_send")
if image_params else XiuConfig().merge_forward_send)
if use_merge:
nodes = [create_node(bot, summary, text_msg)]
if images:
img_path = image_params.get("img_path") if image_params else None
img_format = image_params.get("img_format", "png") if image_params else "png"
get_image_func = image_params.get("get_image_func") if image_params else None
for image in images:
if get_image_func:
img = get_image_func(image)
elif img_path:
img = MessageSegment.image(img_path / f"{image}.{img_format}")
else:
img = str(image)
nodes.append(create_node(bot, summary, img))
return nodes
else:
result_msgs = [text_msg] + [str(img) for img in images] if images else [text_msg]
return [summary, bot.self_id, result_msgs]This approach keeps behavior the same but reduces per-function complexity by focusing each helper on one task.
</details>
| if give_qq: | ||
| if give_qq == str(user_id): | ||
| msg = "道友不会左右互搏之术！" | ||
| if XiuConfig().img: | ||
| pic = await get_msg_pic(f"@{event.sender.nickname}\n" + msg) | ||
| await bot.send_group_msg(group_id=int(send_group_id), message=MessageSegment.image(pic)) | ||
| else: | ||
| await bot.send_group_msg(group_id=int(send_group_id), message=msg) | ||
| await handle_send(bot, event, send_group_id, msg) | ||
| await qc.finish() |
There was a problem hiding this comment.
suggestion (code-quality): 合并嵌套的 if 条件（merge-nested-ifs）
| if give_qq: | |
| if give_qq == str(user_id): | |
| msg = "道友不会左右互搏之术！" | |
| if XiuConfig().img: | |
| pic = await get_msg_pic(f"@{event.sender.nickname}\n" + msg) | |
| await bot.send_group_msg(group_id=int(send_group_id), message=MessageSegment.image(pic)) | |
| else: | |
| await bot.send_group_msg(group_id=int(send_group_id), message=msg) | |
| await handle_send(bot, event, send_group_id, msg) | |
| await qc.finish() | |
| if give_qq and give_qq == str(user_id): | |
| msg = "道友不会左右互搏之术！" | |
| await handle_send(bot, event, send_group_id, msg) | |
| await qc.finish() | |
解释
过多的嵌套会让代码难以理解，尤其是在 Python 中没有大括号来区分不同的嵌套层级。阅读深层嵌套的代码很容易混淆，因为你需要时刻记住每个条件属于哪一层。我们因此尽量减少嵌套，遇到可以用 and 合并的两个 if 条件时，这是一个很好的优化点。
Original comment in English
suggestion (code-quality): Merge nested if conditions (merge-nested-ifs)
| if give_qq: | |
| if give_qq == str(user_id): | |
| msg = "道友不会左右互搏之术！" | |
| if XiuConfig().img: | |
| pic = await get_msg_pic(f"@{event.sender.nickname}\n" + msg) | |
| await bot.send_group_msg(group_id=int(send_group_id), message=MessageSegment.image(pic)) | |
| else: | |
| await bot.send_group_msg(group_id=int(send_group_id), message=msg) | |
| await handle_send(bot, event, send_group_id, msg) | |
| await qc.finish() | |
| if give_qq and give_qq == str(user_id): | |
| msg = "道友不会左右互搏之术！" | |
| await handle_send(bot, event, send_group_id, msg) | |
| await qc.finish() | |
Explanation
Too much nesting can make code difficult to understand, and this is especiallytrue in Python, where there are no brackets to help out with the delineation of
different nesting levels.
Reading deeply nested code is confusing, since you have to keep track of which
conditions relate to which levels. We therefore strive to reduce nesting where
possible, and the situation where two if conditions can be combined using
and is an easy win.
| msg = '本群尚未开启拍卖会功能，请联系管理员开启！' | ||
| if XiuConfig().img: | ||
| pic = await get_msg_pic(f"@{event.sender.nickname}\n" + msg) | ||
| pic = await get_msg_pic(f"@{user_info['user_name'] or event.sender.nickname}\n" + msg) |
There was a problem hiding this comment.
issue (code-quality): 我们发现了以下问题：
- 用 f-string 替换字符串拼接（
use-fstring-for-concatenation） - 用普通字符串替换无插值的 f-string [×3]（
remove-redundant-fstring）
Original comment in English
issue (code-quality): We've found these issues:
- Use f-string instead of string concatenation (
use-fstring-for-concatenation) - Replace f-string with no interpolated values with string [×3] (
remove-redundant-fstring)
| await handle_send(bot, event, send_group_id, msg) | ||
| await no_use_zb.finish() | ||
| else: | ||
| msg = "装备没有被使用，无法卸载！" | ||
| if XiuConfig().img: | ||
| pic = await get_msg_pic(f"@{event.sender.nickname}\n" + msg) | ||
| await bot.send_group_msg(group_id=int(send_group_id), message=MessageSegment.image(pic)) | ||
| else: | ||
| await bot.send_group_msg(group_id=int(send_group_id), message=msg) | ||
| await handle_send(bot, event, send_group_id, msg) | ||
| await no_use_zb.finish() | ||
| else: | ||
| msg = "目前只支持卸载装备！" | ||
| if XiuConfig().img: | ||
| pic = await get_msg_pic(f"@{event.sender.nickname}\n" + msg) | ||
| await bot.send_group_msg(group_id=int(send_group_id), message=MessageSegment.image(pic)) | ||
| else: | ||
| await bot.send_group_msg(group_id=int(send_group_id), message=msg) | ||
| await handle_send(bot, event, send_group_id, msg) | ||
| await no_use_zb.finish() |
There was a problem hiding this comment.
issue (code-quality): 将重复代码提升到条件语句外部 [×4]（hoist-statement-from-if）
Original comment in English
issue (code-quality): Hoist repeated code outside conditional statement [×4] (hoist-statement-from-if)
|
|
||
| def check_equipment_can_use(user_id, goods_id): | ||
| async def check_equipment_can_use(user_id, goods_id): | ||
| """ |
There was a problem hiding this comment.
issue (code-quality): 我们发现了以下问题：
- 将赋值移到块内更靠近使用的位置（
move-assign-in-block） - 简化布尔 if 表达式（
boolean-if-exp-identity） - 将变量默认值设置移到 else 分支（
introduce-default-else） - 用 if 表达式替换 if 语句（
assign-if-exp） - 移除不必要的 int、str、float 或 bool 类型转换（
remove-unnecessary-cast）
Original comment in English
issue (code-quality): We've found these issues:
- Move assignment closer to its usage within a block (
move-assign-in-block) - Simplify boolean if expression (
boolean-if-exp-identity) - Move setting of default value for variable into
elsebranch (introduce-default-else) - Replace if statement with if expression (
assign-if-exp) - Remove unnecessary casts to int, str, float or bool (
remove-unnecessary-cast)
| # 如果还有其他装备需要卸载（对于法器和防具的情况） | ||
| if in_use_id != 0 and goods_id != in_use_id: | ||
| sql_str.append( | ||
| f"UPDATE back set update_time='{now_time}',action_time='{now_time}',state=0 WHERE user_id={user_id} and goods_id={in_use_id}") | ||
| f"UPDATE xiuxian_back set update_time='{now_time}',action_time='{now_time}',state=0 WHERE user_id={user_id} and goods_id={in_use_id}") |
There was a problem hiding this comment.
suggestion (code-quality): 将条件判断从嵌套条件中提升出来（hoist-if-from-if）
| # 如果还有其他装备需要卸载（对于法器和防具的情况） | |
| if in_use_id != 0 and goods_id != in_use_id: | |
| sql_str.append( | |
| f"UPDATE back set update_time='{now_time}',action_time='{now_time}',state=0 WHERE user_id={user_id} and goods_id={in_use_id}") | |
| f"UPDATE xiuxian_back set update_time='{now_time}',action_time='{now_time}',state=0 WHERE user_id={user_id} and goods_id={in_use_id}") | |
| # 如果还有其他装备需要卸载（对于法器和防具的情况） | |
| if in_use_id != 0 and goods_id != in_use_id: | |
| sql_str.append( | |
| f"UPDATE xiuxian_back set update_time='{now_time}',action_time='{now_time}',state=0 WHERE user_id={user_id} and goods_id={in_use_id}") |
Original comment in English
suggestion (code-quality): Hoist conditional out of nested conditional (hoist-if-from-if)
| # 如果还有其他装备需要卸载（对于法器和防具的情况） | |
| if in_use_id != 0 and goods_id != in_use_id: | |
| sql_str.append( | |
| f"UPDATE back set update_time='{now_time}',action_time='{now_time}',state=0 WHERE user_id={user_id} and goods_id={in_use_id}") | |
| f"UPDATE xiuxian_back set update_time='{now_time}',action_time='{now_time}',state=0 WHERE user_id={user_id} and goods_id={in_use_id}") | |
| # 如果还有其他装备需要卸载（对于法器和防具的情况） | |
| if in_use_id != 0 and goods_id != in_use_id: | |
| sql_str.append( | |
| f"UPDATE xiuxian_back set update_time='{now_time}',action_time='{now_time}',state=0 WHERE user_id={user_id} and goods_id={in_use_id}") |
|
|
||
|
|
||
| def get_user_main_back_msg(user_id): | ||
| async def get_user_main_back_msg(user_id): |
There was a problem hiding this comment.
issue (code-quality): get_user_main_back_msg 代码质量较低 - 14%（low-code-quality）
解释
该函数的质量分数低于 25% 的阈值。该分数综合了方法长度、认知复杂度和工作记忆。
如何改进？
可以考虑重构该函数，使其更短、更易读。
- 通过将功能片段提取到独立函数中来缩短函数长度。这是最重要的——理想情况下一个函数应少于 10 行。
- 通过引入守卫子句提前返回，减少嵌套。
- 确保变量作用域紧凑，让相关代码聚集在一起而不是分散在函数各处。
Original comment in English
issue (code-quality): Low code quality found in get_user_main_back_msg - 14% (low-code-quality)
Explanation
The quality score for this function is below the quality threshold of 25%.This score is a combination of the method length, cognitive complexity and working memory.
How can you solve this?
It might be worth refactoring this function to make it shorter and more readable.
- Reduce the function length by extracting pieces of functionality out into
their own functions. This is the most important thing you can do - ideally a
function should be less than 10 lines. - Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
sits together within the function rather than being scattered.
1 participant
Sourcery 总结
将项目从 SQLite 迁移到 PostgreSQL，引入异步数据库操作，并更新整个代码库以支持 async/await 语法。
新特性：
增强功能：
杂项：
Original summary in English
Sourcery 总结
将项目从 SQLite 迁移到 PostgreSQL，引入异步数据库操作，并更新整个代码库以支持 async/await 语法。
新特性：
增强功能：
杂项：
Original summary in English
Sourcery 总结
将项目从 SQLite 迁移到 PostgreSQL，引入异步数据库操作，并更新整个代码库以支持 async/await 语法。
新特性：
增强功能：
杂项：
Original summary in English
Summary by Sourcery
Migrate the project from SQLite to PostgreSQL, introducing asynchronous database operations and updating the entire codebase to support async/await syntax
New Features:
Enhancements:
Chores: