Skip to content
Draft
60 changes: 46 additions & 14 deletions _conf_schema.json
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@
"format"
],
"default": "scrapbook",
"hint": "分析报告使用的HTML模板名称,QQ 端可以使用 `/设置模板` 查看使用指南,使用`/查看模板` 命令查看模板样式效果。其他平台不支持合并转发,可以到 [模板](https://github.com/SXP-Simon/astrbot_plugin_qq_group_daily_analysis/tree/main/assets) 文件夹中查看模板样式效果"
"hint": "分析报告使用的HTML模板名称,注意:该外部主题模板仅适用于 image(图片) 和 pdf(PDF文件) 格式,text(文本) 格式无效不会生效。QQ 端可以使用 `/设置模板` 查看使用指南,使用`/查看模板` 命令查看模板样式效果。其他平台不支持合并转发,可以到 [模板](https://github.com/SXP-Simon/astrbot_plugin_qq_group_daily_analysis/tree/main/assets) 文件夹中查看模板样式效果"
},
"debug_mode": {
"type": "bool",
Expand Down Expand Up @@ -131,6 +131,32 @@
"items": {
"type": "string"
}
},
"auto_analysis_send_report": {
"type": "bool",
"description": "分析后自动发送报告",
"default": true,
"hint": "总开关。开启后,分析定时完成后会尝试将报告推送到群聊;关闭后,仅在后台完成数据归档和本地报告生成,绝对不会发送到任何群聊。"
},
"send_report_mode": {
"description": "发送报告组限制模式",
"type": "string",
"options": [
"whitelist",
"blacklist",
"none"
],
"default": "none",
"hint": "搭配发信开关使用。whitelist: 仅允许向列表内群聊发送报告;blacklist: 拒绝向列表内群聊发送;none: 不设限制(全发)"
},
"send_report_list": {
"type": "list",
"description": "发送群聊白/黑名单列表",
"default": [],
"hint": "黑白名单模式下使用的群组列表。支持填写 AstrBot UMO (如 xxxxx:GroupMessage:123456) 或纯群号 (如 123456 将尝试匹配)。",
"items": {
"type": "string"
}
}
}
},
Expand Down Expand Up @@ -335,28 +361,34 @@
}
}
},
"pdf": {
"description": "PDF 设置",
"report_storage": {
"description": "分析报告本地存储设置",
"type": "object",
"hint": "PDF 报告输出相关配置,包括输出目录、浏览器路径和文件名格式",
"hint": "统一管理所有格式(图片、文本、PDF)分析报告的本地存档行为",
"items": {
"pdf_output_dir": {
"enable_local_storage": {
"type": "bool",
"description": "启用本地存储归档",
"default": true,
"hint": "开启后所有的报告(图片、文本、PDF)生成后都会被长久保存在统一输出目录中;关闭后报告将直接推送至群聊不再留底浪费空间(对纯净群管理员有帮助)。"
},
"report_output_dir": {
"type": "string",
"description": "PDF输出目录",
"description": "报告统一输出目录",
"default": "data/plugins/astrbot_plugin_qq_group_daily_analysis/reports",
"hint": "PDF报告文件的保存目录"
"hint": "无论是png图片、文本md文档还是pdf,生成后均会存放在此文件夹中长久归档"
},
"report_filename_format": {
"type": "string",
"description": "报告文件名格式",
"default": "群聊分析报告_{group_id}_{date}",
"hint": "报告文件名格式,无需加后缀。支持变量:{group_id}(群号)、{date}(日期)。插件会自动根据生成格式补齐 .png / .md / .pdf 后缀"
},
"browser_path": {
"type": "string",
"description": "自定义浏览器路径",
"description": "自定义浏览器路径 (生成 PDF 时使用)",
"default": "",
"hint": "要填写的话请你清楚自己在干什么。自定义浏览器的可执行文件路径(如 Chrome 或 Edge 的 .exe 文件)。提示:如果是在网页后台设置,则直接输入普通路径即可(如 C:\\Program Files\\...);如果是手动编辑 config.json 文件,请务必使用双反斜杠 '\\\\' 分隔路径。"
},
"pdf_filename_format": {
"type": "string",
"description": "PDF文件名格式",
"default": "群聊分析报告_{group_id}_{date}.pdf",
"hint": "PDF文件名格式,支持变量:{group_id}(群号)、{date}(日期)"
}
}
},
Expand Down
193 changes: 126 additions & 67 deletions src/infrastructure/config/config_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,45 @@ def get_group_list(self) -> list[str]:
"""获取群组列表(用于黑白名单)"""
return self._get_group("basic").get("group_list", [])

@staticmethod
def _match_umo_rule(rule: str, target: str) -> bool:
"""
匹配目标源(target)是否符合指定规则(rule)
支持 UMO 前缀和包含的话题会话的后段(#)提权匹配。
"""
if rule == target:
return True

# 分解目标 UMO
target_has_prefix = ":" in target
target_simple_id = target.split(":")[-1] if target_has_prefix else target
target_parent_id = target_simple_id.split("#", 1)[0] if "#" in target_simple_id else target_simple_id
target_has_topic = "#" in target_simple_id
target_prefix = target.rsplit(":", 1)[0] if target_has_prefix else ""

# 分解规则
rule_has_prefix = ":" in rule
rule_simple_id = rule.split(":")[-1] if rule_has_prefix else rule
rule_prefix = rule.rsplit(":", 1)[0] if rule_has_prefix else ""

if rule_has_prefix:
# 规则也带有平台前缀,则双方前缀必须完全一致
if not target_has_prefix or rule_prefix != target_prefix:
return False
# 允许 Telegram 等带后缀的话题会话通过“父 UMO”被包含命中
if target_has_topic and rule_simple_id == target_parent_id:
return True
return False

# 规则只是一个单独的不带前缀的纯标识
if rule == target_simple_id:
return True
# 允许单独通过群号父 ID 来命中(如 rule="123", target="telegram2:Msg:123#456")
if target_has_topic and rule == target_parent_id:
return True

return False

def is_group_allowed(self, group_id_or_umo: str) -> bool:
"""
根据配置的白/黑名单判断是否允许在该群聊中使用
Expand All @@ -64,44 +103,7 @@ def is_group_allowed(self, group_id_or_umo: str) -> bool:
glist = [str(g) for g in self.get_group_list()]
target = str(group_id_or_umo)

target_simple_id = target.split(":")[-1] if ":" in target else target
target_parent_id = (
target_simple_id.split("#", 1)[0]
if "#" in target_simple_id
else target_simple_id
)

def _is_match(
item: str,
target: str,
target_simple_id: str,
target_parent_id: str,
) -> bool:
if ":" in item:
if item == target:
return True

# 允许 Telegram 话题会话通过“父 UMO”命中,
# 例如: item=telegram2:GroupMessage:-1001
# target=telegram2:GroupMessage:-1001#2264
if "#" in target_simple_id:
if ":" not in target:
return False
item_prefix, item_tail = item.rsplit(":", 1)
target_prefix, _ = target.rsplit(":", 1)
return (
item_prefix == target_prefix and item_tail == target_parent_id
)
return False
if item == target_simple_id:
return True
# 允许 Telegram 话题会话通过父群 ID 命中简单群号白/黑名单
return "#" in target_simple_id and item == target_parent_id

is_in_list = any(
_is_match(item, target, target_simple_id, target_parent_id)
for item in glist
)
is_in_list = any(self._match_umo_rule(item, target) for item in glist)

if mode == "whitelist":
return is_in_list
Expand Down Expand Up @@ -145,6 +147,33 @@ def get_enable_auto_analysis(self) -> bool:
"""
return self.is_auto_analysis_enabled()

def get_auto_analysis_send_report(self) -> bool:
"""获取分析完成后是否自动发送报告"""
return self._get_group("auto_analysis").get("auto_analysis_send_report", True)

def get_send_report_mode(self) -> str:
"""获取发送报告限制模式 (whitelist/blacklist/none)"""
return self._get_group("auto_analysis").get("send_report_mode", "none")

def get_send_report_list(self) -> list[str]:
"""获取发送报告群组列表(用于黑白名单)"""
return self._get_group("auto_analysis").get("send_report_list", [])

def is_group_allowed_to_send_report(self, group_id_or_umo: str) -> bool:
"""根据配置的白/黑名单判断是否允许向该群发送自动分析报告"""
mode = self.get_send_report_mode().lower()
if mode not in ("whitelist", "blacklist", "none"):
mode = "none"

if mode == "none":
return True

glist = [str(g) for g in self.get_send_report_list()]
target = str(group_id_or_umo)

matched = any(self._match_umo_rule(item, target) for item in glist)
return matched if mode == "whitelist" else not matched

def get_output_format(self) -> str:
"""获取输出格式"""
return self._get_group("basic").get("output_format", "image")
Expand Down Expand Up @@ -223,18 +252,27 @@ def get_keep_original_persona(self) -> bool:
"""获取是否保持原始人格设定"""
return self._get_group("analysis_features").get("keep_original_persona", False)

def get_pdf_output_dir(self) -> str:
"""获取PDF输出目录"""
def get_enable_local_storage(self) -> bool:
"""获取是否启用本地存储归档"""
return self._get_group("report_storage").get("enable_local_storage", True)

def get_report_output_dir(self) -> str:
"""获取报告统一输出目录"""
try:
plugin_name = "astrbot_plugin_qq_group_daily_analysis"
data_path = Path(get_astrbot_data_path())
default_path = data_path / "plugin_data" / plugin_name / "reports"
return self._get_group("pdf").get("pdf_output_dir", str(default_path))

# 优先读新版配置
report_storage = self._get_group("report_storage")
if "report_output_dir" in report_storage:
return report_storage.get("report_output_dir", str(default_path))

# 兼容读取旧版配置 pdf_output_dir
pdf_group = self._get_group("pdf")
return pdf_group.get("pdf_output_dir", str(default_path))
except Exception:
return self._get_group("pdf").get(
"pdf_output_dir",
"data/plugins/astrbot_plugin_qq_group_daily_analysis/reports",
)
return "data/plugins/astrbot_plugin_qq_group_daily_analysis/reports"

def get_bot_self_ids(self) -> list:
"""获取机器人自身的 ID 列表 (兼容 bot_qq_ids)"""
Expand All @@ -244,11 +282,17 @@ def get_bot_self_ids(self) -> list:
ids = basic.get("bot_qq_ids", [])
return ids

def get_pdf_filename_format(self) -> str:
"""获取PDF文件名格式"""
return self._get_group("pdf").get(
"pdf_filename_format", "群聊分析报告_{group_id}_{date}.pdf"
)
def get_report_filename_format(self) -> str:
"""获取报告文件名格式 (无后缀)"""

report_storage = self._get_group("report_storage")
if "report_filename_format" in report_storage:
return report_storage.get("report_filename_format", "群聊分析报告_{group_id}_{date}")

old_pdf_format = self._get_group("pdf").get("pdf_filename_format", "群聊分析报告_{group_id}_{date}.pdf")
if old_pdf_format.endswith(".pdf"):
return old_pdf_format[:-4]
return old_pdf_format

def get_topic_analysis_prompt(self, style: str = "topic_prompt") -> str:
"""获取话题分析提示词模板"""
Expand Down Expand Up @@ -403,6 +447,21 @@ def set_scheduled_group_list_mode(self, mode: str):
self._ensure_group("auto_analysis")["scheduled_group_list_mode"] = mode
self.config.save_config()

def set_auto_analysis_send_report(self, enabled: bool):
"""设置是否在分析后自动发送报告"""
self._ensure_group("auto_analysis")["auto_analysis_send_report"] = enabled
self.config.save_config()

def set_send_report_mode(self, mode: str):
"""设置发送报告权限模式"""
self._ensure_group("auto_analysis")["send_report_mode"] = mode
self.config.save_config()

def set_send_report_list(self, group_list: list[str]):
"""设置发送报告黑白名单"""
self._ensure_group("auto_analysis")["send_report_list"] = group_list
self.config.save_config()

def get_scheduled_group_list(self) -> list[str]:
"""获取定时分析目标群列表"""
return self._get_group("auto_analysis").get("scheduled_group_list", [])
Expand All @@ -429,24 +488,16 @@ def is_group_in_filtered_list(
group_list = [str(x).strip() for x in group_list]
target = str(group_umo_or_id).strip()

# 兼容 UMO 匹配 (如果列表里写的是 ID,UMO 也能匹配上)
def match_umo(umo: str, item: str) -> bool:
if umo == item:
return True
if ":" in umo and umo.split(":")[-1] == item:
return True
return False

if mode == "whitelist":
if not group_list:
# 白名单为空:此级别不开启 (按需开启逻辑)
return False
return any(match_umo(target, x) for x in group_list)
return any(self._match_umo_rule(x, target) for x in group_list)
else: # blacklist
if not group_list:
# 黑名单为空:全通过
return True
return not any(match_umo(target, x) for x in group_list)
return not any(self._match_umo_rule(x, target) for x in group_list)

def set_min_messages_threshold(self, threshold: int):
"""设置最小消息阈值"""
Expand Down Expand Up @@ -492,14 +543,19 @@ def set_max_golden_quotes(self, count: int):
self._ensure_group("analysis_features")["max_golden_quotes"] = count
self.config.save_config()

def set_pdf_output_dir(self, directory: str):
"""设置PDF输出目录"""
self._ensure_group("pdf")["pdf_output_dir"] = directory
def set_enable_local_storage(self, enabled: bool):
"""设置是否启用本地存储归档"""
self._ensure_group("report_storage")["enable_local_storage"] = enabled
self.config.save_config()

def set_report_output_dir(self, directory: str):
"""设置报告产出目录"""
self._ensure_group("report_storage")["report_output_dir"] = directory
self.config.save_config()

def set_pdf_filename_format(self, format_str: str):
"""设置PDF文件名格式"""
self._ensure_group("pdf")["pdf_filename_format"] = format_str
def set_report_filename_format(self, format_str: str):
"""设置报告文件名格式"""
self._ensure_group("report_storage")["report_filename_format"] = format_str
self.config.save_config()

def get_report_template(self) -> str:
Expand Down Expand Up @@ -664,11 +720,14 @@ def _check_playwright_availability(self):

def get_browser_path(self) -> str:
"""获取自定义浏览器路径"""
report_storage = self._get_group("report_storage")
if "browser_path" in report_storage:
return report_storage.get("browser_path", "")
return self._get_group("pdf").get("browser_path", "")

def set_browser_path(self, path: str):
"""设置自定义浏览器路径"""
self._ensure_group("pdf")["browser_path"] = path
self._ensure_group("report_storage")["browser_path"] = path
self.config.save_config()

def reload_playwright(self) -> bool:
Expand Down
Loading