技术实现

# main.py - 主程序入口
import argparse
import logging
from src.utils.config import load_config
from src.ai_content_generator.content_generator import ContentGenerator
from src.video_generator.video_generator import VideoGenerator
from src.video_editor.video_editor import VideoEditor
from src.product_selector.product_selector import ProductSelector
from src.link_manager.link_manager import LinkManager
from src.video_publisher.video_publisher import VideoPublisher
from src.traffic_optimizer.traffic_optimizer import TrafficOptimizer

def main():
    # 解析命令行参数
    parser = argparse.ArgumentParser(description='快手AI短视频带货软件')
    parser.add_argument('--mode', choices=['auto', 'interactive'], default='interactive',
                      help='运行模式: auto(自动) 或 interactive(交互)')
    parser.add_argument('--product', type=str, help='产品关键词或ID')
    parser.add_argument('--template', type=str, help='视频模板名称')
    parser.add_argument('--publish', action='store_true', help='是否自动发布')
    parser.add_argument('--optimize', action='store_true', help='是否自动优化投放')
    args = parser.parse_args()
    
    # 加载配置
    config = load_config()
    
    # 初始化日志
    setup_logging(config['paths']['log_dir'])
    
    if args.mode == 'auto':
        run_auto_mode(args, config)
    else:
        run_interactive_mode(config)

def run_auto_mode(args, config):
    """自动模式：一键完成从选品到发布的全流程"""
    logging.info("启动自动模式")
    
    # 初始化各模块
    product_selector = ProductSelector(config['api']['kuaishou_api_key'])
    content_generator = ContentGenerator(config['api']['openai_api_key'])
    video_generator = VideoGenerator(config['ai']['stable_diffusion_model_id'])
    video_editor = VideoEditor()
    link_manager = LinkManager(config['api']['kuaishou_api_key'])
    publisher = VideoPublisher(config['api']['kuaishou_api_key'], config['api']['kuaishou_api_secret'])
    optimizer = TrafficOptimizer(config['api']['kuaishou_api_key'])
    
    # 1. 选品
    logging.info(f"开始选品: {args.product}")
    products = product_selector.search_products(keywords=args.product, limit=1)
    if not products:
        logging.error(f"未找到相关产品: {args.product}")
        return
    product = products[0]
    logging.info(f"已选择产品: {product['name']}")
    
    # 2. 生成文案
    logging.info("开始生成文案")
    script = content_generator.generate_script(
        product_name=product['name'],
        product_features=product['features'],
        target_audience=product['target_audience'],
        style="专业"
    )
    logging.info(f"文案生成完成: {len(script)} 字符")
    
    # 3. 生成视频
    logging.info("开始生成视频")
    video_path = video_generator.generate_video(
        script=script,
        product_images=[product['image_url']],
        style=args.template or "modern",
        duration=30,
        resolution=config['video']['default_resolution']
    )
    logging.info(f"视频生成完成: {video_path}")
    
    # 4. 剪辑视频
    logging.info("开始剪辑视频")
    edited_video = video_editor.edit_video(
        video_path=video_path,
        template=args.template or "product_showcase",
        add_music=True,
        add_captions=True
    )
    logging.info(f"视频剪辑完成: {edited_video}")
    
    # 5. 创建带货链接
    logging.info("开始创建带货链接")
    links = link_manager.create_links(
        product_ids=[product['id']],
        display_times=[(5, 25)]
    )
    logging.info(f"带货链接创建完成: {links}")
    
    # 6. 发布视频
    if args.publish:
        logging.info("开始发布视频")
        publish_result = publisher.publish_video(
            video_path=edited_video,
            title=f"{product['name']} - {script.split('.')[0]}",
            description=script[:100] + "...",
            tags=[args.product, "带货", "推荐"],
            product_links=links
        )
        logging.info(f"视频发布完成: {publish_result}")
        
        # 7. 优化投放
        if args.optimize and publish_result.get('status') == 'success':
            logging.info("开始优化投放")
            video_id = publish_result.get('video_id')
            campaign = optimizer.create_campaign(
                video_id=video_id,
                budget=1000,
                daily_budget=100,
                target_audience={"interests": [args.product]}
            )
            logging.info(f"投放优化完成: {campaign}")
    
    logging.info("自动模式执行完成")

def run_interactive_mode(config):
    """交互模式：提供更灵活的操作选项"""
    logging.info("启动交互模式")
    
    # 实现交互式菜单
    # ...

if __name__ == "__main__":
    main()
                        

# src/utils/config.py - 配置管理
import os
import json
import logging
from dotenv import load_dotenv

def load_config(config_path=None):
    """加载配置文件"""
    # 首先加载环境变量
    load_dotenv()
    
    # 默认配置路径
    if not config_path:
        config_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), 'config.json')
    
    # 如果配置文件不存在，创建默认配置
    if not os.path.exists(config_path):
        return create_default_config(config_path)
    
    # 读取配置文件
    try:
        with open(config_path, 'r', encoding='utf-8') as f:
            config = json.load(f)
        
        # 用环境变量覆盖配置
        if 'KUAISHOU_API_KEY' in os.environ:
            config['api']['kuaishou_api_key'] = os.environ['KUAISHOU_API_KEY']
        if 'KUAISHOU_API_SECRET' in os.environ:
            config['api']['kuaishou_api_secret'] = os.environ['KUAISHOU_API_SECRET']
        if 'OPENAI_API_KEY' in os.environ:
            config['api']['openai_api_key'] = os.environ['OPENAI_API_KEY']
        
        return config
    except Exception as e:
        logging.error(f"加载配置文件失败: {e}")
        return create_default_config(config_path)

def create_default_config(config_path):
    """创建默认配置"""
    config = {
        "api": {
            "kuaishou_api_key": os.environ.get('KUAISHOU_API_KEY', ''),
            "kuaishou_api_secret": os.environ.get('KUAISHOU_API_SECRET', ''),
            "openai_api_key": os.environ.get('OPENAI_API_KEY', '')
        },
        "paths": {
            "temp_dir": "/tmp/kuaishou_ai",
            "data_dir": "./data",
            "log_dir": "./logs"
        },
        "database": {
            "type": "sqlite",
            "path": "./data/kuaishou_ai.db"
        },
        "video": {
            "default_resolution": "1080p",
            "default_fps": 30,
            "default_duration": 30
        },
        "ai": {
            "stable_diffusion_model_id": "stabilityai/stable-diffusion-xl-base-1.0",
            "text_model": "gpt-4"
        }
    }
    
    # 确保目录存在
    os.makedirs(os.path.dirname(config_path), exist_ok=True)
    
    # 保存默认配置
    try:
        with open(config_path, 'w', encoding='utf-8') as f:
            json.dump(config, f, indent=4, ensure_ascii=False)
    except Exception as e:
        logging.error(f"保存默认配置失败: {e}")
    
    return config

def setup_logging(log_dir):
    """设置日志"""
    os.makedirs(log_dir, exist_ok=True)
    log_file = os.path.join(log_dir, 'kuaishou_ai.log')
    
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler(log_file),
            logging.StreamHandler()
        ]
    )