nezumi/astrbot_plugin_qq_group_daily_analysis
1
0
Fork 0
mirror of https://github.com/Nezumi-2711/astrbot_plugin_qq_group_daily_analysis.git synced 2026-09-22 13:38:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: 更新脚本

Browse Source
This commit is contained in:
SXP-Simon
2026-02-11 14:09:17 +08:00
parent c817bb2e2c
commit 5bf391594c
3 changed files with 290 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3 -1
View File
@@ -10,4 +10,6 @@ debug_output.html
data/cmd_config.json
data/t2i_templates/astrbot_powershell.html
data/t2i_templates/base.html
.sisyphus/
.sisyphus/
data/plugin_data/astrbot_plugin_qq_group_daily_analysis/cache/avatars/123456789_40.jpg
data/plugin_data/astrbot_plugin_qq_group_daily_analysis/cache/avatars/987654321_40.jpg
scripts/debug_render.py
+105 -54
View File
@@ -2,23 +2,54 @@ import argparse
import asyncio
import os
import sys
import types
from pathlib import Path
# ==========================================
# 1. Environment Setup
# ==========================================
# Add src to path so we can import our modules
sys.path.append(os.path.join(os.path.dirname(__file__), ".."))
# Assuming we are in scripts/
current_dir = os.path.dirname(os.path.abspath(__file__))
plugin_root = os.path.abspath(os.path.join(current_dir, ".."))
sys.path.insert(0, plugin_root)
# Mock astrbot.api before importing our modules
import types
astrbot_api = types.ModuleType("astrbot.api")
astrbot_api.logger = types.ModuleType("logger")
astrbot_api.logger.info = lambda x, *args, **kwargs: print(f"[INFO] {x}")
astrbot_api.logger.error = lambda x, *args, **kwargs: print(f"[ERROR] {x}")
astrbot_api.logger.warning = lambda x, *args, **kwargs: print(f"[WARN] {x}")
class MockLogger:
def info(self, msg, *args, **kwargs):
print(f"[INFO] {msg}")
def error(self, msg, *args, **kwargs):
print(f"[ERROR] {msg}")
def warning(self, msg, *args, **kwargs):
print(f"[WARN] {msg}")
def debug(self, msg, *args, **kwargs):
print(f"[DEBUG] {msg}")
def log(self, level, msg, *args, **kwargs):
print(f"[LOG {level}] {msg}")
def isEnabledFor(self, level):
return True
astrbot_api.logger = MockLogger()
astrbot_api.AstrBotConfig = dict
sys.modules["astrbot.api"] = astrbot_api
from src.models.data_models import ( # noqa: E402
# Mock astrbot.core.utils.astrbot_path
astrbot_core_utils = types.ModuleType("astrbot.core.utils")
astrbot_path = types.ModuleType("astrbot.core.utils.astrbot_path")
astrbot_path.get_astrbot_data_path = lambda: Path(".")
sys.modules["astrbot.core.utils"] = astrbot_core_utils
sys.modules["astrbot.core.utils.astrbot_path"] = astrbot_path
from src.domain.entities.analysis_result import ( # noqa: E402
ActivityVisualization,
EmojiStatistics,
GoldenQuote,
@@ -27,12 +58,12 @@ from src.models.data_models import ( # noqa: E402
TokenUsage,
UserTitle,
)
from src.reports.generators import ReportGenerator # noqa: E402
from src.reports.templates import HTMLTemplates # noqa: E402
from src.infrastructure.reporting.generators import ReportGenerator # noqa: E402
from src.infrastructure.reporting.templates import HTMLTemplates # noqa: E402
class MockConfigManager:
def __init__(self, template_name: str = "format") -> None:
def __init__(self, template_name: str = "scrapbook") -> None:
self.template_name = template_name
def get_report_template(self) -> str:
@@ -53,11 +84,20 @@ class MockConfigManager:
def get_pdf_filename_format(self) -> str:
return "report_{group_id}_{date}.pdf"
def get_enable_user_card(self) -> bool:
return True
async def mock_get_user_avatar(user_id: int) -> str:
# Return a placeholder or a real base64 if needed
# For debugging, a simple colored square or a known avatar is fine
return "https://q4.qlogo.cn/headimg_dl?dst_uin=123456789&spec=640"
@property
def playwright_available(self) -> bool:
return True
def get_browser_path(self) -> str:
return ""
async def mock_get_user_avatar(user_id: str) -> str:
# Return a known avatar URL for testing
return f"https://q4.qlogo.cn/headimg_dl?dst_uin={user_id}&spec=640"
async def debug_render(
@@ -66,32 +106,12 @@ async def debug_render(
# 1. Setup Mock Data
config_manager = MockConfigManager(template_name)
# Mock Analysis Result
# 2. Mock Analysis Result using Entities
stats = GroupStatistics(
message_count=1250,
total_characters=45000,
participant_count=42,
most_active_period="20:00 - 22:00",
golden_quotes=[
GoldenQuote(
content="代码写得好，下班走得早。",
sender="张三",
reason="深刻揭示了程序员的生存法则",
qq=123456789,
),
GoldenQuote(
content="这个Bug我不修，它就是个Feature。",
sender="李四",
reason="经典的开发辩解",
qq=987654321,
),
GoldenQuote(
content="PHP是世界上最好的语言！",
sender="王五",
reason="引发了长达3小时的群聊大讨论",
qq=112233445,
),
],
emoji_count=156,
emoji_statistics=EmojiStatistics(face_count=100, mface_count=56),
activity_visualization=ActivityVisualization(
@@ -99,9 +119,6 @@ async def debug_render(
i: (10 + i * 5 if i < 12 else 100 - i * 2) for i in range(24)
}
),
token_usage=TokenUsage(
prompt_tokens=1500, completion_tokens=800, total_tokens=2300
),
)
topics = [
@@ -125,58 +142,92 @@ async def debug_render(
user_titles = [
UserTitle(
name="张三",
qq=123456789,
user_id="123456789",
title="代码收割机",
mbti="INTJ",
reason="在短短一小时内提交了10个PR，效率惊人。",
),
UserTitle(
name="李四",
qq=987654321,
user_id="987654321",
title="群聊气氛组",
mbti="ENFP",
reason="总能精准接住每一个冷笑话，让群里充满快活的气息。",
),
UserTitle(
name="https://www.example.com/very/long/url/that/might/overflow/the/container/if/word/break/is/not/set/correctly/and/it/keeps/going/and/going/forever",
qq=112233445,
name="潜水员",
user_id="112233445",
title="深夜潜水员",
mbti="INFP",
reason="总是在凌晨三点出没，留下几句深奥的话语后消失。",
),
]
golden_quotes = [
GoldenQuote(
content="代码写得好，下班走得早。",
sender="张三",
reason="深刻揭示了程序员的生存法则",
user_id="123456789",
),
GoldenQuote(
content="这个Bug我不修，它就是个Feature。",
sender="李四",
reason="经典的开发辩解",
user_id="987654321",
),
GoldenQuote(
content="PHP是世界上最好的语言！",
sender="王五",
reason="引发了长达3小时的群聊大讨论",
user_id="112233445",
),
]
stats.golden_quotes = golden_quotes
stats.token_usage = TokenUsage(
prompt_tokens=1500, completion_tokens=800, total_tokens=2300
)
analysis_result = {
"statistics": stats,
"topics": topics,
"user_titles": user_titles,
"analysis_date": "2026年02月11日",
"group_id": "123456",
"group_name": "测试群组",
}
# 2. Initialize Generator
# 3. Initialize Generator
generator = ReportGenerator(config_manager)
# Override _get_user_avatar to avoid real network calls if desired,
# but here we'll just let it use the mock URL
generator._get_user_avatar = mock_get_user_avatar
# Override internal methods to facilitate debugging without real dependencies
generator._get_user_avatar_data = mock_get_user_avatar
# 3. Render Data
# 4. Prepare Render Data
# Note: _prepare_render_data handles converting Entities to template-friendly dicts
render_payload = await generator._prepare_render_data(analysis_result)
# 4. Render Main Template
# We'll test the image template
# 5. Render Main Template
html_templates = HTMLTemplates(config_manager)
# Get image template string
raw_template = html_templates.get_image_template()
if not raw_template:
print(f"[ERROR] Failed to load template for '{template_name}'")
return
# Use generator's internal renderer
final_html = generator._render_html_template(raw_template, render_payload)
# 5. Save to file
# 6. Save to file
output_path = Path(output_file)
output_path.write_text(final_html, encoding="utf-8")
print(
f"Successfully rendered template '{template_name}' to {output_path.absolute()}"
)
print("You can now open this file with VS Code Live Server to debug your HTML/CSS.")
print("You can now open this file with your browser to debug your HTML/CSS.")
def main() -> None:
@@ -187,8 +238,8 @@ def main() -> None:
"-t",
"--template",
type=str,
default="retro_futurism",
help="Template name to render (default: retro_futurism)",
default="scrapbook",
help="Template name to render (default: scrapbook)",
)
parser.add_argument(
"-o",
scripts/mock_pdf_gen.py
+182 -179
View File
@@ -1,52 +1,74 @@
import asyncio
import os
import sys
from datetime import datetime
import types
from pathlib import Path
# ==========================================
# 1. Environment Setup (Critical for Imports)
# 1. Environment Setup
# ==========================================
# Add project root to sys.path so we can import 'astrbot' and plugin modules
# In docker environment:
# /AstrBot
# /AstrBot/data/plugins/astrbot_plugin_qq_group_daily_analysis
current_dir = os.path.dirname(os.path.abspath(__file__))
# Assuming structure: .../data/plugins/astrbot_plugin_qq_group_daily_analysis/scripts/mock_pdf_gen.py
# We need to go up 4 levels to reach 'AstrBot-master' root which contains the 'astrbot' package
# data/plugins/astrbot_plugin_qq_group_daily_analysis/scripts -> ... -> AstrBot-master
project_root = os.path.abspath(os.path.join(current_dir, "../../../../"))
sys.path.insert(0, project_root)
plugin_root = os.path.abspath(os.path.join(current_dir, ".."))
sys.path.insert(0, plugin_root)
print(f"Project Root: {project_root}")
# Mock astrbot.api
astrbot_api = types.ModuleType("astrbot.api")
# Mock logger before importing anything that uses it
# Mock logger
from data.plugins.astrbot_plugin_qq_group_daily_analysis.src.utils import ( # noqa: E402
logger as plugin_logger_module,
)
plugin_logger_module.logger.info = lambda msg, *args, **kwargs: print(f"[INFO] {msg}")
plugin_logger_module.logger.error = lambda msg, *args, **kwargs: print(f"[ERROR] {msg}")
plugin_logger_module.logger.warning = lambda msg, *args, **kwargs: print(
f"[WARN] {msg}"
)
class MockLogger:
def info(self, msg, *args, **kwargs):
print(f"[INFO] {msg}")
def error(self, msg, *args, **kwargs):
print(f"[ERROR] {msg}")
def warning(self, msg, *args, **kwargs):
print(f"[WARN] {msg}")
def debug(self, msg, *args, **kwargs):
print(f"[DEBUG] {msg}")
def log(self, level, msg, *args, **kwargs):
print(f"[LOG {level}] {msg}")
def isEnabledFor(self, level):
return True
astrbot_api.logger = MockLogger()
astrbot_api.AstrBotConfig = dict
sys.modules["astrbot.api"] = astrbot_api
# Mock astrbot.core.utils.astrbot_path
astrbot_core_utils = types.ModuleType("astrbot.core.utils")
astrbot_path = types.ModuleType("astrbot.core.utils.astrbot_path")
astrbot_path.get_astrbot_data_path = lambda: Path(".")
sys.modules["astrbot.core.utils"] = astrbot_core_utils
sys.modules["astrbot.core.utils.astrbot_path"] = astrbot_path
# Now import plugin modules
try:
from data.plugins.astrbot_plugin_qq_group_daily_analysis.src.core.config import (
ConfigManager,
)
from data.plugins.astrbot_plugin_qq_group_daily_analysis.src.reports.generators import (
ReportGenerator,
)
except ImportError as e:
print(f"Import Error: {e}")
sys.exit(1)
from src.domain.entities.analysis_result import ( # noqa: E402
ActivityVisualization,
EmojiStatistics,
GoldenQuote,
GroupStatistics,
SummaryTopic,
TokenUsage,
UserTitle,
)
from src.infrastructure.reporting.generators import ReportGenerator # noqa: E402
# ==========================================
# 2. Mocks
# ==========================================
class MockConfig:
def get(self, key, default=None):
return default
class MockConfigManager:
def __init__(self):
pass
def get_pdf_output_dir(self):
# Output to the scripts directory for easy access
@@ -59,11 +81,17 @@ class MockConfig:
return 5
def get_max_user_titles(self):
return 5
return 8
def get_max_golden_quotes(self):
return 5
def get_report_template(self):
return "scrapbook"
def get_enable_user_card(self):
return True
@property
def playwright_available(self):
return True
@@ -72,162 +100,132 @@ class MockConfig:
return ""
async def mock_get_user_avatar(user_id: str) -> str:
# Return a known avatar URL for testing
return f"https://q4.qlogo.cn/headimg_dl?dst_uin={user_id}&spec=640"
# ==========================================
# 3. Main Execution
# ==========================================
async def main():
print("Initializing ReportGenerator...")
config_manager = ConfigManager(MockConfig())
config_manager = MockConfigManager()
generator = ReportGenerator(config_manager)
# Mock Data (Rich data to test layout)
# Override avatar fetching to avoid real network calls during testing
generator._get_user_avatar_data = mock_get_user_avatar
# 1. Mock Analysis Result using Entities
stats = GroupStatistics(
message_count=1280,
total_characters=8500,
participant_count=42,
most_active_period="20:00-22:00",
emoji_count=156,
emoji_statistics=EmojiStatistics(face_count=100, mface_count=56),
activity_visualization=ActivityVisualization(
hourly_activity={i: (i * 5) % 65 for i in range(24)},
daily_activity={"2026-02-11": 1280},
),
)
topics = [
SummaryTopic(
topic="AstrBot新功能",
detail="大家对PDF生成功能的讨论非常热烈，提出了很多优化建议。",
contributors=["开发者", "测试员"],
),
SummaryTopic(
topic="周末计划",
detail="有人提议去爬山，也有人想在家打游戏。",
contributors=["旅行家", "宅男"],
),
SummaryTopic(
topic="代码调试",
detail="关于Python异步编程的深入探讨。",
contributors=["小白", "大神"],
),
SummaryTopic(
topic="美食分享",
detail="深夜放毒，发了很多火锅和烧烤的照片。",
contributors=["吃货A", "吃货B"],
),
SummaryTopic(
topic="模组推荐",
detail="推荐了一些好用的Minecraft模组。",
contributors=["MC玩家"],
),
]
user_titles = [
UserTitle(
name="极客",
user_id="10001",
title="代码魔术师",
mbti="INTJ",
reason="总是能用一行代码解决复杂问题。",
),
UserTitle(
name="社牛",
user_id="10002",
title="气氛组组长",
mbti="ENFP",
reason="群里冷场时总能第一时间活跃气氛。",
),
UserTitle(
name="百科",
user_id="10003",
title="移动维基",
mbti="ISTJ",
reason="不管问什么问题，他都知道答案。",
),
UserTitle(
name="潜水",
user_id="10004",
title="深海幽灵",
mbti="INTP",
reason="虽然很少说话，但每次发言都直击要害。",
),
UserTitle(
name="欧皇",
user_id="10005",
title="天选之子",
mbti="ESFJ",
reason="抽卡次次出金，让人羡慕嫉妒恨。",
),
]
stats.golden_quotes = [
GoldenQuote(
sender="大佬",
content="这代码能跑就行，别动它！",
reason="至理名言，动了就崩。",
user_id="20001",
),
GoldenQuote(
sender="萌新",
content="为什么我的报错和你不一？",
reason="经典的灵魂发问。",
user_id="20002",
),
GoldenQuote(
sender="群主",
content="再发涩图全部禁言！",
reason="来自管理层的威慑。",
user_id="888888",
),
]
stats.token_usage = TokenUsage(
prompt_tokens=2000, completion_tokens=1000, total_tokens=3000
)
analysis_result = {
"date": datetime.now().strftime("%Y年%m月%d日"),
"statistics": {
"total_messages": 1280,
"active_users": 42,
"emoji_count": 156,
"total_chars": 8500,
"message_count": 1280,
"participant_count": 42,
"total_characters": 8500,
"most_active_period": "20:00-22:00",
"token_usage": type(
"obj",
(object,),
{"total_tokens": 500, "prompt_tokens": 200, "completion_tokens": 300},
),
"activity_visualization": type(
"obj",
(object,),
{
"hourly_activity": {
i: (i * 5) % 60 for i in range(24)
}, # Fake activity data
"heatmap_data": [],
},
),
},
"highlight_time": {
"period": "21:00-22:00",
"reason": "夜深人静，群里却热闹非凡，大家都在讨论新的游戏活动。",
},
"topics": [
{
"topic": "AstrBot新功能",
"detail": "大家对PDF生成功能的讨论非常热烈，提出了很多优化建议。",
"contributors": ["开发者", "测试员"],
},
{
"topic": "周末计划",
"detail": "有人提议去爬山，也有人想在家打游戏。",
"contributors": ["旅行家", "宅男"],
},
{
"topic": "代码调试",
"detail": "关于Python异步编程的深入探讨。",
"contributors": ["小白", "大神"],
},
{
"topic": "美食分享",
"detail": "深夜放毒，发了很多火锅和烧烤的照片。",
"contributors": ["吃货A", "吃货B"],
},
{
"topic": "模组推荐",
"detail": "推荐了一些好用的Minecraft模组。",
"contributors": ["MC玩家"],
},
],
"user_titles": [
{
"name": "极客",
"title": "代码魔术师",
"mbti": "INTJ",
"reason": "总是能用一行代码解决复杂问题。",
"qq": "10001",
},
{
"name": "社牛",
"title": "气氛组组长",
"mbti": "ENFP",
"reason": "群里冷场时总能第一时间活跃气氛。",
"qq": "10002",
},
{
"name": "百科",
"title": "移动维基",
"mbti": "ISTJ",
"reason": "不管问什么问题，他都知道答案。",
"qq": "10003",
},
{
"name": "潜水",
"title": "深海幽灵",
"mbti": "INTP",
"reason": "虽然很少说话，但每次发言都直击要害。",
"qq": "10004",
},
{
"name": "欧皇",
"title": "天选之子",
"mbti": "ESFJ",
"reason": "抽卡次次出金，让人羡慕嫉妒恨。",
"qq": "10005",
},
],
"statistics": stats,
"topics": topics,
"user_titles": user_titles,
}
analysis_result["statistics"]["golden_quotes"] = [
{
"sender": "大佬",
"content": "这代码能跑就行，别动它！",
"reason": "至理名言，动了就崩。",
"qq": "20001",
},
{
"sender": "萌新",
"content": "为什么我的报错和你不一？",
"reason": "经典的灵魂发问。",
"qq": "20002",
},
{
"sender": "群主",
"content": "再发黄色图全部禁言！",
"reason": "来自管理层的威慑。",
"qq": "888888",
},
]
# Helper wrapper for dot notation access needed by template
class DictWrapper:
def __init__(self, data):
self._data = data
for k, v in data.items():
if isinstance(v, list):
setattr(
self,
k,
[DictWrapper(i) if isinstance(i, dict) else i for i in v],
)
elif isinstance(v, dict):
setattr(self, k, DictWrapper(v))
else:
setattr(self, k, v)
def __getitem__(self, key):
return self._data[key]
def get(self, key, default=None):
return self._data.get(key, default)
# Wrap sections that need dot access
analysis_result["statistics"] = DictWrapper(analysis_result["statistics"])
analysis_result["topics"] = [DictWrapper(t) for t in analysis_result["topics"]]
analysis_result["user_titles"] = [
DictWrapper(t) for t in analysis_result["user_titles"]
]
print("Generating PDF Report...")
group_id = "test_group_mock"
@@ -236,7 +234,12 @@ async def main():
if pdf_path:
print(f"\n[SUCCESS] PDF Generated Successfully: {pdf_path}")
print(f"File Size: {os.path.getsize(pdf_path) / 1024:.2f} KB")
if os.path.exists(pdf_path):
print(f"File Size: {os.path.getsize(pdf_path) / 1024:.2f} KB")
else:
print(
"[WARN] Generated path returned but file not found on local disk (might be in container)."
)
else:
print("\n[FAILURE] PDF Generation Failed.")