修复高优先级安全和稳定性问题

1. app.py:
   - 添加user_semaphores_lock线程锁,修复get_user_semaphore竞态条件
   - 将9处裸except:改为except Exception:或具体异常类型

2. database.py:
   - 将3处裸except:改为except Exception:

3. email_service.py:
   - 添加_smtp_config_lock线程锁
   - 修复daily_sent计数竞态条件:获取配置时预增,失败时回退
   - _get_available_smtp_config和_get_next_available_smtp_config使用锁保护

🤖 Generated with Claude Code
This commit is contained in:
Yu Yon
2025-12-12 15:04:58 +08:00
parent 4a47d4ba6b
commit 63ca9265ed
3 changed files with 129 additions and 94 deletions

28
app.py
View File

@@ -162,6 +162,7 @@ IP_LOCK_DURATION = config.IP_LOCK_DURATION
max_concurrent_per_account = config.MAX_CONCURRENT_PER_ACCOUNT
max_concurrent_global = config.MAX_CONCURRENT_GLOBAL
user_semaphores = {} # {user_id: Semaphore}
user_semaphores_lock = threading.Lock() # 保护user_semaphores的线程锁
global_semaphore = threading.Semaphore(max_concurrent_global)
# 截图专用信号量:限制同时进行的截图任务数量(避免资源竞争)
@@ -1084,7 +1085,7 @@ def generate_captcha():
try:
font = ImageFont.truetype(font_path, 42)
break
except:
except Exception:
continue
if font is None:
font = ImageFont.load_default()
@@ -1937,10 +1938,11 @@ def stop_account(account_id):
def get_user_semaphore(user_id):
"""获取或创建用户的信号量"""
if user_id not in user_semaphores:
user_semaphores[user_id] = threading.Semaphore(max_concurrent_per_account)
return user_semaphores[user_id]
"""获取或创建用户的信号量(线程安全)"""
with user_semaphores_lock:
if user_id not in user_semaphores:
user_semaphores[user_id] = threading.Semaphore(max_concurrent_per_account)
return user_semaphores[user_id]
def run_task(user_id, account_id, browse_type, enable_screenshot=True, source="manual", retry_count=0):
@@ -4232,7 +4234,7 @@ def get_user_schedules_api():
for s in schedules:
try:
s['account_ids'] = json.loads(s.get('account_ids', '[]') or '[]')
except:
except (json.JSONDecodeError, TypeError):
s['account_ids'] = []
return jsonify(schedules)
@@ -4282,7 +4284,7 @@ def get_schedule_detail_api(schedule_id):
import json
try:
schedule['account_ids'] = json.loads(schedule.get('account_ids', '[]') or '[]')
except:
except (json.JSONDecodeError, TypeError):
schedule['account_ids'] = []
return jsonify(schedule)
@@ -4362,7 +4364,7 @@ def run_schedule_now_api(schedule_id):
try:
account_ids = json.loads(schedule.get('account_ids', '[]') or '[]')
except:
except (json.JSONDecodeError, TypeError):
account_ids = []
if not account_ids:
@@ -4544,7 +4546,7 @@ def cleanup_on_exit():
for user_id in user_accounts:
if account_id in user_accounts[user_id]:
user_accounts[user_id][account_id].should_stop = True
except:
except Exception:
pass
# 2. 等待所有线程完成最多等待5秒
@@ -4553,28 +4555,28 @@ def cleanup_on_exit():
try:
if thread and thread.is_alive():
thread.join(timeout=2)
except:
except Exception:
pass
# 3. 关闭浏览器工作线程池
print("- 关闭浏览器线程池...")
try:
shutdown_browser_worker_pool()
except:
except Exception:
pass
# 3.5 关闭邮件队列
print("- 关闭邮件队列...")
try:
email_service.shutdown_email_queue()
except:
except Exception:
pass
# 4. 关闭数据库连接池
print("- 关闭数据库连接池...")
try:
db_pool._pool.close_all() if db_pool._pool else None
except:
except Exception:
pass
print("✓ 资源清理完成")