#!/usr/bin/env python3 """ 图片A/B测试模块 测试不同图片风格的效果 """ import os import json import random from datetime import datetime, timedelta from advanced_image_generator import AdvancedImageGenerator class ImageABTest: """图片A/B测试""" def __init__(self): self.generator = AdvancedImageGenerator() self.test_db = "/home/ubuntu/.nanobot/workspace/xiaohongshu_ab_test.db" self.styles = ['modern', 'minimal', 'gradient', 'tech'] # 初始化数据库 self._init_database() def _init_database(self): """初始化测试数据库""" import sqlite3 conn = sqlite3.connect(self.test_db) cursor = conn.cursor() # 创建测试记录表 cursor.execute(''' CREATE TABLE IF NOT EXISTS ab_tests ( id INTEGER PRIMARY KEY AUTOINCREMENT, test_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP, article_title TEXT, article_source TEXT, style_used TEXT, note_id TEXT, engagement_score REAL DEFAULT 0, likes INTEGER DEFAULT 0, collects INTEGER DEFAULT 0, comments INTEGER DEFAULT 0, shares INTEGER DEFAULT 0, check_time TIMESTAMP, notes TEXT ) ''') # 创建风格效果表 cursor.execute(''' CREATE TABLE IF NOT EXISTS style_performance ( style TEXT PRIMARY KEY, total_tests INTEGER DEFAULT 0, total_engagement REAL DEFAULT 0, avg_engagement REAL DEFAULT 0, best_score REAL DEFAULT 0, last_used TIMESTAMP ) ''') # 初始化风格数据 for style in self.styles: cursor.execute(''' INSERT OR IGNORE INTO style_performance (style, total_tests, avg_engagement) VALUES (?, 0, 0) ''', (style,)) conn.commit() conn.close() def select_best_style(self, article_title, article_source): """选择最佳图片风格""" import sqlite3 conn = sqlite3.connect(self.test_db) cursor = conn.cursor() # 获取各风格的平均表现 cursor.execute(''' SELECT style, avg_engagement, total_tests FROM style_performance ORDER BY avg_engagement DESC ''') styles_data = cursor.fetchall() conn.close() if not styles_data: # 无历史数据，随机选择 return random.choice(self.styles) # 计算选择概率（基于表现） total_engagement = sum(data[1] for data in styles_data if data[1] > 0) if total_engagement == 0: # 所有风格表现都为0，随机选择 return random.choice(self.styles) # 基于表现加权随机选择 weights = [] for style, engagement, tests in styles_data: if tests == 0: weight = 1.0 # 未测试过的风格给基础权重 else: weight = engagement + 0.1 # 加0.1避免0权重 weights.append(weight) # 归一化权重 total_weight = sum(weights) normalized_weights = [w / total_weight for w in weights] # 加权随机选择 import numpy as np selected_index = np.random.choice(len(styles_data), p=normalized_weights) selected_style = styles_data[selected_index][0] print(f"📊 A/B测试选择: {selected_style}") print(f" 候选风格: {[s[0] for s in styles_data]}") print(f" 平均互动: {[s[1] for s in styles_data]}") print(f" 测试次数: {[s[2] for s in styles_data]}") return selected_style def record_test(self, article_title, article_source, style_used, note_id): """记录A/B测试""" import sqlite3 conn = sqlite3.connect(self.test_db) cursor = conn.cursor() cursor.execute(''' INSERT INTO ab_tests (test_date, article_title, article_source, style_used, note_id, check_time) VALUES (?, ?, ?, ?, ?, ?) ''', ( datetime.now().isoformat(), article_title[:100], article_source, style_used, note_id, (datetime.now() + timedelta(hours=24)).isoformat() # 24小时后检查效果 )) test_id = cursor.lastrowid # 更新风格使用次数 cursor.execute(''' UPDATE style_performance SET total_tests = total_tests + 1, last_used = ? WHERE style = ? ''', (datetime.now().isoformat(), style_used)) conn.commit() conn.close() print(f"📝 A/B测试记录已保存 (ID: {test_id})") print(f" 风格: {style_used}") print(f" 笔记ID: {note_id}") print(f" 将在24小时后检查效果") return test_id def update_engagement(self, test_id, engagement_data): """更新互动数据""" import sqlite3 conn = sqlite3.connect(self.test_db) cursor = conn.cursor() # 计算互动分数（加权） likes = engagement_data.get('likes', 0) collects = engagement_data.get('collects', 0) comments = engagement_data.get('comments', 0) shares = engagement_data.get('shares', 0) # 权重：收藏 > 评论 > 点赞 > 分享 engagement_score = ( likes * 1.0 + collects * 3.0 + comments * 2.0 + shares * 1.5 ) cursor.execute(''' UPDATE ab_tests SET engagement_score = ?, likes = ?, collects = ?, comments = ?, shares = ?, check_time = ? WHERE id = ? ''', ( engagement_score, likes, collects, comments, shares, datetime.now().isoformat(), test_id )) # 获取测试记录 cursor.execute('SELECT style_used FROM ab_tests WHERE id = ?', (test_id,)) style_used = cursor.fetchone()[0] # 更新风格表现 cursor.execute(''' SELECT total_tests, total_engagement FROM style_performance WHERE style = ? ''', (style_used,)) result = cursor.fetchone() if result: total_tests, total_engagement = result new_total_engagement = total_engagement + engagement_score new_avg_engagement = new_total_engagement / (total_tests + 1) if total_tests > 0 else engagement_score cursor.execute(''' UPDATE style_performance SET total_engagement = ?, avg_engagement = ?, best_score = MAX(best_score, ?) WHERE style = ? ''', (new_total_engagement, new_avg_engagement, engagement_score, style_used)) conn.commit() conn.close() print(f"📈 互动数据已更新 (测试ID: {test_id})") print(f" 互动分数: {engagement_score:.2f}") print(f" 点赞: {likes}, 收藏: {collects}, 评论: {comments}, 分享: {shares}") return engagement_score def check_pending_tests(self): """检查待处理的测试""" import sqlite3 conn = sqlite3.connect(self.test_db) cursor = conn.cursor() # 查找需要检查的测试（超过24小时未检查） cursor.execute(''' SELECT id, note_id, style_used, article_title FROM ab_tests WHERE engagement_score = 0 AND check_time <= ? LIMIT 10 ''', (datetime.now().isoformat(),)) pending_tests = cursor.fetchall() conn.close() if pending_tests: print(f"🔍 发现 {len(pending_tests)} 个待检查的A/B测试") for test_id, note_id, style_used, title in pending_tests: print(f" 测试ID: {test_id}, 笔记ID: {note_id}, 风格: {style_used}") print(f" 标题: {title[:50]}...") return pending_tests def simulate_engagement(self, test_id): """模拟互动数据（用于测试）""" import random engagement_data = { 'likes': random.randint(5, 50), 'collects': random.randint(1, 20), 'comments': random.randint(0, 10), 'shares': random.randint(0, 5) } score = self.update_engagement(test_id, engagement_data) return engagement_data, score def get_style_recommendation(self): """获取风格推荐""" import sqlite3 conn = sqlite3.connect(self.test_db) cursor = conn.cursor() cursor.execute(''' SELECT style, avg_engagement, total_tests, best_score FROM style_performance ORDER BY avg_engagement DESC ''') styles = cursor.fetchall() conn.close() if not styles: return { 'recommended_style': 'modern', 'reason': '无历史数据，使用默认风格', 'all_styles': [] } # 最佳风格 best_style = styles[0][0] best_score = styles[0][1] recommendation = { 'recommended_style': best_style, 'reason': f'平均互动分数最高: {best_score:.2f}', 'confidence': min(styles[0][2] / 10, 1.0), # 置信度基于测试次数 'all_styles': [] } for style, avg_engagement, total_tests, best_score in styles: recommendation['all_styles'].append({ 'style': style, 'avg_engagement': avg_engagement, 'total_tests': total_tests, 'best_score': best_score }) return recommendation def generate_report(self, days=7): """生成A/B测试报告""" import sqlite3 from datetime import datetime, timedelta conn = sqlite3.connect(self.test_db) cursor = conn.cursor() # 时间范围 start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') # 总体统计 cursor.execute(''' SELECT COUNT(*) as total_tests, AVG(engagement_score) as avg_engagement, SUM(likes) as total_likes, SUM(collects) as total_collects, SUM(comments) as total_comments, SUM(shares) as total_shares FROM ab_tests WHERE test_date >= ? ''', (start_date,)) overall_stats = cursor.fetchone() # 风格表现 cursor.execute(''' SELECT sp.style, sp.total_tests, sp.avg_engagement, sp.best_score, COUNT(DISTINCT at.id) as recent_tests, AVG(CASE WHEN at.test_date >= ? THEN at.engagement_score ELSE NULL END) as recent_avg FROM style_performance sp LEFT JOIN ab_tests at ON sp.style = at.style_used AND at.test_date >= ? GROUP BY sp.style ORDER BY sp.avg_engagement DESC ''', (start_date, start_date)) style_stats = cursor.fetchall() conn.close() # 生成报告 report = f"📊 图片A/B测试报告\n" report += f"📅 报告周期: 最近{days}天 ({start_date} 至今)\n" report += f"⏰ 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n" if overall_stats and overall_stats[0] > 0: total_tests, avg_engagement, total_likes, total_collects, total_comments, total_shares = overall_stats report += f"📈 总体统计:\n" report += f" 总测试数: {total_tests}\n" report += f" 平均互动分数: {avg_engagement:.2f}\n" report += f" 总点赞: {total_likes or 0}\n" report += f" 总收藏: {total_collects or 0}\n" report += f" 总评论: {total_comments or 0}\n" report += f" 总分享: {total_shares or 0}\n\n" report += f"🎨 风格表现:\n" for style, total_tests, avg_engagement, best_score, recent_tests, recent_avg in style_stats: report += f" {style}:\n" report += f" 总测试: {total_tests}, 平均分数: {avg_engagement:.2f}\n" report += f" 最佳分数: {best_score:.2f}, 最近测试: {recent_tests}\n" if recent_avg: report += f" 最近平均: {recent_avg:.2f}\n" report += "\n" # 推荐 recommendation = self.get_style_recommendation() report += f"🎯 推荐风格: {recommendation['recommended_style']}\n" report += f" 理由: {recommendation['reason']}\n" report += f" 置信度: {recommendation['confidence']:.1%}\n" return report def main(): """主函数""" import argparse parser = argparse.ArgumentParser(description='图片A/B测试') subparsers = parser.add_subparsers(dest='command', help='命令') # select命令 select_parser = subparsers.add_parser('select', help='选择最佳风格') select_parser.add_argument('--title', required=True, help='文章标题') select_parser.add_argument('--source', required=True, help='文章来源') # record命令 record_parser = subparsers.add_parser('record', help='记录测试') record_parser.add_argument('--title', required=True, help='文章标题') record_parser.add_argument('--source', required=True, help='文章来源') record_parser.add_argument('--style', required=True, help='使用的风格') record_parser.add_argument('--note-id', required=True, help='小红书笔记ID') # update命令 update_parser = subparsers.add_parser('update', help='更新互动数据') update_parser.add_argument('--test-id', type=int, required=True, help='测试ID') update_parser.add_argument('--likes', type=int, default=0, help='点赞数') update_parser.add_argument('--collects', type=int, default=0, help='收藏数') update_parser.add_argument('--comments', type=int, default=0, help='评论数') update_parser.add_argument('--shares', type=int, default=0, help='分享数') # check命令 subparsers.add_parser('check', help='检查待处理测试') # report命令 report_parser = subparsers.add_parser('report', help='生成报告') report_parser.add_argument('--days', type=int, default=7, help='报告天数') # recommend命令 subparsers.add_parser('recommend', help='获取推荐') # simulate命令 simulate_parser = subparsers.add_parser('simulate', help='模拟测试') simulate_parser.add_argument('--test-id', type=int, required=True, help='测试ID') args = parser.parse_args() ab_test = ImageABTest() if args.command == 'select': style = ab_test.select_best_style(args.title, args.source) print(f"🎯 推荐风格: {style}") elif args.command == 'record': test_id = ab_test.record_test(args.title, args.source, args.style, args.note_id) print(f"✅ 测试记录已创建 (ID: {test_id})") elif args.command == 'update': engagement_data = { 'likes': args.likes, 'collects': args.collects, 'comments': args.comments, 'shares': args.shares } score = ab_test.update_engagement(args.test_id, engagement_data) print(f"✅ 互动数据已更新，分数: {score:.2f}") elif args.command == 'check': pending = ab_test.check_pending_tests() if pending: print(f"发现 {len(pending)} 个待检查测试") else: print("✅ 无待检查测试") elif args.command == 'report': report = ab_test.generate_report(args.days) print(report) elif args.command == 'recommend': recommendation = ab_test.get_style_recommendation() print(f"🎯 推荐风格: {recommendation['recommended_style']}") print(f" 理由: {recommendation['reason']}") print(f" 置信度: {recommendation['confidence']:.1%}") print("\n所有风格:") for style_info in recommendation['all_styles']: print(f" {style_info['style']}: 平均{style_info['avg_engagement']:.2f}, 测试{style_info['total_tests']}次") elif args.command == 'simulate': engagement_data, score = ab_test.simulate_engagement(args.test_id) print(f"🎭 模拟互动数据:") print(f" 点赞: {engagement_data['likes']}") print(f" 收藏: {engagement_data['collects']}") print(f" 评论: {engagement_data['comments']}") print(f" 分享: {engagement_data['shares']}") print(f" 总分: {score:.2f}") else: parser.print_help() if __name__ == "__main__": main()